probity-effects

Keep one effect per agent tool call when its reply is lost: the guard asks the service that owns the effect before anything is sent again, records each operation as an uncertain-effect/v1 record, and tests your tools against eight lost-reply scenarios in CI.

It's for teams whose LangGraph, OpenAI Agents SDK or AG2 agents create orders, move money or open tickets, and who cannot afford a retry policy, a resumed thread or a model calling again to repeat the write.

Install the released package: pip install probity-effects==0.2.0.

Quick start

Python 3.11 or later. This example needs no network: the stand-in owner commits each write and loses the reply.

from probity_effects import EffectGuard, EffectOwner, Lookup

ledger = {}

def send(call):
    ledger[call.key] = call.args          # the write commits ...
    raise TimeoutError("reply lost")      # ... and its reply never arrives

owner = EffectOwner("orders", lookup=lambda key: Lookup(present=key in ledger))
guard = EffectGuard(owner)
for _ in range(3):                         # a retry, a resumed run, a model calling again
    result = guard.perform("create_order", {"sku": "A-1", "quantity": 1}, send=send)
    print(result.outcome, result.sent)
print(len(ledger), "effect")

It prints:

committed 1
committed 0
committed 0
1 effect

The first call loses its reply, asks the owner, and finds the order. The next two read the journal and send nothing.

In a LangGraph agent

from probity_effects import EffectCall, EffectOwner, Journal
from probity_effects.langgraph import effect_tool

orders = EffectOwner("orders", lookup=find_order_by_key)   # your lookup, by idempotency key

@effect_tool(orders, journal=Journal.open("effects.jsonl"))
def create_order(sku: str, quantity: int, effect: EffectCall) -> dict:
    """Create an order."""
    reply = httpx.post(ORDERS, json={"sku": sku, "quantity": quantity},
                       headers={"Idempotency-Key": effect.key}, timeout=5.0)
    reply.raise_for_status()
    return reply.json()

Pass create_order to ToolNode or create_react_agent as usual, and give the graph a thread_id. A lost reply never reaches LangGraph's retry machinery as an exception; an outcome the owner cannot confirm comes back to the model as outcome unknown, and calling again sends nothing until the owner answers. A call with no thread_id and no key_from is refused and sends nothing.

In an OpenAI Agents SDK agent

from agents import Agent, Runner
from probity_effects.openai_agents import effect_function_tool

@effect_function_tool(orders, journal=Journal.open("effects.jsonl"))
def create_order(sku: str, quantity: int, effect: EffectCall) -> dict:
    """Create an order."""
    ...                                    # the same body as above

agent = Agent(name="orders", tools=[create_order])
result = await Runner.run(agent, "Order one A-1", context={"thread_id": conversation_id})

Use the same thread_id for every run of one conversation.

In an AG2 agent

from probity_effects.ag2 import effect_tool

@effect_tool(orders, journal=Journal.open("effects.jsonl"))
def create_order(sku: str, quantity: int, effect: EffectCall) -> dict:
    """Create an order."""
    ...                                    # the same body as above

agent = Agent("orders", config=config, tools=[create_order])
await agent.ask("Order one A-1", stream=stream, variables={"thread_id": conversation_id})

Asking again on the same stream is the same conversation. Pass a thread_id to resume in a new process. AG2 turns any tool exception into a tool error for the model, so on_unknown="raise" does not stop an AG2 run.

In CI

from probity_effects.testing import LANGGRAPH_VARIANTS, run_langgraph_scenario

@pytest.mark.parametrize("variant", LANGGRAPH_VARIANTS)
def test_survives_lost_replies(lost_reply_scenario, variant):
    verdict = run_langgraph_scenario(make_tools, lost_reply_scenario, variant=variant)
    assert verdict.passed, verdict.problems

pip install probity-effects[testing] adds the scenarios from agent-evidence-vectors. make_tools(owner, journal) builds your tools; each scenario runs on four LangGraph paths and is judged by the owner's counts, the reference reader's verdict on the record, and the journal's agreement with the owner. For the Agents SDK use OPENAI_AGENTS_VARIANTS and run_openai_agents_scenario, and for AG2 AG2_VARIANTS and run_ag2_scenario, with three paths each.

Status

Version 0.2.0, alpha, with adapters for LangGraph, the OpenAI Agents SDK and AG2. Version 0.1.0 told a second caller with the same arguments that a write was committed when it was never sent, if the tool had no thread and no key_from; 0.2.0 refuses such a call, so upgrade.

Documentation

License

Apache-2.0.