symfonic.agent.postgres_receipt_reconciliation

postgres_receipt_reconciliation

Host-authenticated PG v1 receipt evidence, issued before public Agent effects.

Persist the opaque handle before run/stream. Reconciliation never resumes a turn or returns an AgentResult. The host authority supplies authenticated_principal, scope, executable_versions, journal, receipt_domain and require_access(principal, scope). Its exact domain must be the journal's owned domain. Fresh processes must use the same PostgreSQL URI with explicit host, port, role and database; options must match and environment-derived defaults are refused. Userinfo password rotation is permitted. URI aliases deliberately refuse rather than guess identity.

prepare_postgres_receipt_handle async

prepare_postgres_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 and pin the exact next public run/stream request.

The preflight performs only a read; an existing claim refuses. The host must persist the result before dispatch. UNKNOWN never authorizes a retry.

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

async def prepare_postgres_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 and pin the exact next public run/stream request.

    The preflight performs only a read; an existing claim refuses. The host must
    persist the result before dispatch. UNKNOWN never authorizes a retry.
    """
    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.capabilities.memory.graph_receipt import domain_of
    from symfonic.capabilities.memory.graph_store import GraphBackedHms
    from symfonic.kernel._receipt_invocation import _request_identity
    from symfonic.memory.backends.postgres_receipt_views import receipt_pair_domain
    from symfonic.memory.backends.postgres_receipt_write import PostgresReceiptWriter

    journal, scope, principal, versions, domain = _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()
    writer = memory.writer
    if (type(writer) is not PostgresReceiptWriter
            or writer._domain is not authority.receipt_domain
            or type(writer._store) is not GraphBackedHms
            or domain_of(writer._store) is not authority.receipt_domain):
        _refuse()
    try:
        if receipt_pair_domain(writer._store._graph, writer._store._vectors) is not writer._domain:
            _refuse()
    except MemoryContractError:
        _refuse()
    cache = agent._receipt_plan_cache
    if cache.get("receipt_unhandled_dispatch_seen"):
        _refuse()
    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)
    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)
    epoch = 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) != epoch
            or cache.get("receipt_unhandled_dispatch_seen")
            or pins.get(invocation.invocation_id) is not prior_pin
            or prior_pin is not None or len(pins) >= _MAX_OUTSTANDING):
        _refuse()
    issuance = secrets.token_hex(16)
    document = {"schema": _SCHEMA, "invocation": invocation.invocation_id,
                "scope": scope.path, "version": 1, "request": digest,
                "plan": reservation.plan_id, "principal": principal,
                "executables": versions, "issuance": issuance, "domain": domain}
    handle = _seal(journal, document)
    # Reuse the dispatch protocol, not Mongo authority or parsing.
    pins[invocation.invocation_id] = _ReceiptPin(digest, issuance)
    cache["receipt_handle_mode"] = True
    return handle

reconcile_postgres_receipt async

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

Read accepted PG completion metadata; publication alone remains UNKNOWN.

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

async def reconcile_postgres_receipt(handle: str, authority: Any) -> ReceiptReconciliation:
    """Read accepted PG completion metadata; publication alone remains UNKNOWN."""
    journal, scope, principal, versions, domain = _authority(authority)
    document = _open(handle, journal)
    if (document["principal"] != principal or document["scope"] != scope.path
            or document["domain"] != domain
            or document["executables"] != [list(item) for item in versions]):
        _refuse()
    invocation, digest = document["invocation"], document["request"]
    try:
        result = await journal.read_completion(invocation, scope, digest)
        if type(result) is CompletedTurn:
            if (result.invocation_id != invocation or result.full_scope != scope
                    or result.request_digest != digest or result.acceptance_sequence is None):
                _refuse()
            return ReceiptReconciliation("COMPLETED", result)
        if type(result) in (JournalUnknown, CompletionMissing):
            return ReceiptReconciliation("UNKNOWN")
        _refuse()
    except MemoryContractError:
        _refuse()
    except ContractViolationError:
        raise
    except Exception:
        return ReceiptReconciliation("UNKNOWN")