symfonic.agent.receipt_reconciliation

receipt_reconciliation

Explicit, read-only recovery evidence for a receipt-backed public Agent.

The host must persist the opaque handle before invoking the Agent. A host-owned authority authenticates the current principal and declares executable versions. This module never resumes a turn, returns an AgentResult, or authorizes replay.

ReceiptReconciliation dataclass

ReceiptReconciliation(state: str, completion: CompletedTurn | None = None)

A durable proof or uncertainty; IN_FLIGHT needs future lease evidence.

abandon_receipt_handle async

abandon_receipt_handle(agent: Any, handle: str, authority: Any) -> None

Host-authorized cancellation only before a durable claim has begun.

A source created earlier retains the pin object; its preclaim hook sees ABANDONED and refuses rather than running after cancellation.

Source code in src/symfonic/agent/receipt_reconciliation.py

async def abandon_receipt_handle(agent: Any, handle: str, authority: Any) -> None:
    """Host-authorized cancellation only before a durable claim has begun.

    A source created earlier retains the pin object; its preclaim hook sees
    ABANDONED and refuses rather than running after cancellation.
    """
    from symfonic.agent.facade import Agent
    from symfonic.agent.facade_args import receipt_capability

    journal, scope, principal, versions = _authority(authority)
    document = _open(handle, journal)
    if (type(agent) is not Agent or document["principal"] != principal
            or document["scope"] != scope.path
            or document["executables"] != [list(item) for item in versions]):
        _refuse()
    memory = receipt_capability(agent._composed_capabilities)
    if memory is None or memory.scope != scope or memory.journal is not journal:
        _refuse()
    pins = agent._receipt_plan_cache.get("receipt_request_pins", {})
    invocation, digest = document["invocation"], document["request"]
    pin = pins.get(invocation)
    if (type(pin) is not _ReceiptPin or pin.digest != digest
            or pin.issuance != document["issuance"] or pin.state != "PENDING"):
        _refuse()
    try:
        if await journal.has_invocation(invocation, scope, digest):
            _refuse()
    except ContractViolationError:
        raise
    except Exception:
        _refuse()
    if pins.get(invocation) is not pin or pin.state != "PENDING":
        _refuse()
    pin.state = "ABANDONED"
    del pins[invocation]

prepare_receipt_handle async

prepare_receipt_handle(agent: Any, prompt: str, *, invocation: Any, authority: Any, attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = (), state: Any = None, session_id: str = '', output_type: Any = None) -> str

Issue before effects; the next run/stream must use this exact request.

The host persists the returned value before calling run/stream. Its authority must be created after application authentication; the library has no default. Every provider/tool executable must have a host-declared version key. A majority read refuses IDs already claimed, including across fresh processes.

Source code in src/symfonic/agent/receipt_reconciliation.py

async def prepare_receipt_handle(
    agent: Any, prompt: str, *, invocation: Any, authority: Any,
    attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = (),
    state: Any = None, session_id: str = "", output_type: Any = None,
) -> str:
    """Issue before effects; the next run/stream must use this exact request.

    The host persists the returned value before calling run/stream. Its authority
    must be created after application authentication; the library has no default.
    Every provider/tool executable must have a host-declared version key. A
    majority read refuses IDs already claimed, including across fresh processes.
    """
    from symfonic.agent.facade import Agent
    from symfonic.agent.facade_args import receipt_capability
    from symfonic.agent.facade_prepare import ReceiptPlanReservation
    from symfonic.agent.facade_turn import turn_for
    from symfonic.agent.invocation import AgentInvocation
    from symfonic.agent.receipt_history import snapshot_receipt_history
    from symfonic.capabilities.memory._receipt_journal import _ReceiptJournal
    from symfonic.kernel._receipt_invocation import _request_identity

    journal, scope, principal, versions = _authority(authority)
    if type(agent) is not Agent or type(invocation) is not AgentInvocation:
        _refuse()
    memory = receipt_capability(agent._composed_capabilities)
    if memory is None or memory.scope != scope or memory.journal is not journal:
        _refuse()
    cache = agent._receipt_plan_cache
    if cache.get("receipt_unhandled_dispatch_seen"):
        _refuse()  # opt into handle mode before the first receipt dispatch
    expected = {"provider"} | {f"tool:{tool.name}" for tool in agent._plans.tools.tools}
    if {name for name, _ in versions} != expected:
        _refuse()
    reservation = agent._prepare(prompt, output_type)
    if type(reservation) is not ReceiptPlanReservation:
        _refuse()
    request = turn_for(prompt, attachments, history, state, session_id=session_id,
                       invocation=invocation)
    from dataclasses import replace

    scoped, history_identity = snapshot_receipt_history(replace(request, scope=scope))
    _, digest = _request_identity(reservation.plan_id, scoped,
                                  scope, _ReceiptJournal(journal, scope),
                                  history_identity=history_identity)
    pins = cache.setdefault("receipt_request_pins", {})
    prior_pin = pins.get(invocation.invocation_id)
    observed_dispatches = cache.get("receipt_dispatch_epoch", 0)
    try:
        if await journal.has_invocation(invocation.invocation_id, scope, digest):
            _refuse()
    except ContractViolationError:
        raise
    except Exception:
        _refuse()
    if (cache.get("receipt_dispatch_epoch", 0) != observed_dispatches
            or cache.get("receipt_unhandled_dispatch_seen")
            or pins.get(invocation.invocation_id) is not prior_pin):
        _refuse()  # a same-Agent dispatch raced the read-only preflight
    existing = pins.get(invocation.invocation_id)
    if existing is not None:
        # Never supersede an unconsumed source's authority with a new handle.
        _refuse()
    if len(pins) >= _MAX_OUTSTANDING:
        _refuse()  # explicit bound on outstanding pre-issued handles; no eviction
    issuance = secrets.token_hex(16)
    document = {"schema": _SCHEMA, "invocation": invocation.invocation_id,
                "scope": scope.path, "version": 0, "request": digest,
                "plan": reservation.plan_id, "principal": principal,
                "executables": versions, "issuance": issuance}
    handle = _seal(journal, document)
    pins[invocation.invocation_id] = _ReceiptPin(digest, issuance)
    cache["receipt_handle_mode"] = True
    return handle

reconcile_receipt async

reconcile_receipt(handle: str, authority: Any) -> ReceiptReconciliation

Read exact Mongo evidence; absence or uncertainty never licenses replay.

Source code in src/symfonic/agent/receipt_reconciliation.py

async def reconcile_receipt(handle: str, authority: Any) -> ReceiptReconciliation:
    """Read exact Mongo evidence; absence or uncertainty never licenses replay."""
    journal, scope, principal, versions = _authority(authority)
    document = _open(handle, journal)
    if (document["principal"] != principal or document["scope"] != scope.path
            or document["executables"] != [list(item) for item in versions]
            or type(document["invocation"]) is not str
            or type(document["request"]) is not str
            or type(document["plan"]) is not str):
        _refuse()
    invocation, digest = document["invocation"], document["request"]
    try:
        result = await journal.read_completion(invocation, scope, digest)
        if type(result) is CompletedTurn:
            return ReceiptReconciliation("COMPLETED", result)
        if type(result) is JournalUnknown:
            return ReceiptReconciliation("UNKNOWN")
        if type(result) is CompletionMissing:
            # A claimed row may be abandoned: this journal has no durable
            # liveness lease. Do not infer IN_FLIGHT or safe replay from it.
            return ReceiptReconciliation("UNKNOWN")
        _refuse()
    except MemoryContractError:
        _refuse()
    except ContractViolationError:
        raise
    except Exception:
        return ReceiptReconciliation("UNKNOWN")