symfonic.agent.receipt_adoption

receipt_adoption

Authenticated preclaim adoption under an explicit HOST-ATTESTED contract.

Only an exact majority no-row read admits a fresh Agent. Admission is not a claim: the existing atomic journal claim arbitrates concurrent adopters. A claimed/ambiguous invocation stays UNKNOWN; accepted turns expose proof only. v1 handles remain reconciliation-only and cannot be adopted. The host must durably register the returned identity before dispatch. Active-issuance checks are observations, not an atomic fence against concurrent host revocation.

ReceiptAdoption dataclass

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

ADOPTED permits a subsequent claim attempt; never promises execution.

adopt_receipt_handle async

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

Restore exact identity in a never-dispatched, unreserved fresh Agent.

Authenticated host attestation is mandatory. The framework proves equality to that declaration, not independent truth of opaque executable state.

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

async def adopt_receipt_handle(
    agent: Any, handle: str, prompt: str, *, authority: Any,
    attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = (),
    state: Any = None, session_id: str = "", output_type: Any = None,
) -> ReceiptAdoption:
    """Restore exact identity in a never-dispatched, unreserved fresh Agent.

    Authenticated host attestation is mandatory. The framework proves equality
    to that declaration, not independent truth of opaque executable state.
    """
    from symfonic.agent.facade import Agent
    from symfonic.agent.facade_args import receipt_capability
    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)
    document = _open(handle, journal, schema=_SCHEMA, extra_fields={"configuration"})
    if (type(agent) is not Agent or agent._closed
            or agent._receipt_plan_cache is None or agent._receipt_plan_cache
            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()
    fingerprint, config = configuration_fingerprint(agent, authority, versions, output_type)
    if fingerprint != document["configuration"]:
        _refuse()
    _active(authority, document, scope, principal)
    invocation = AgentInvocation(document["invocation"])
    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(document["plan"], scoped,
                                  scope, _ReceiptJournal(journal, scope),
                                  history_identity=history_identity)
    if digest != document["request"]:
        _refuse()
    try:
        present = await journal.has_invocation(invocation.invocation_id, scope, digest)
        if type(present) is not bool:
            return ReceiptAdoption("UNKNOWN")
        if present:
            completion = await journal.read_completion(invocation.invocation_id, scope, digest)
            if type(completion) is CompletedTurn:
                return ReceiptAdoption("COMPLETED", completion)
            return ReceiptAdoption("UNKNOWN")
    except (ContractViolationError, MemoryContractError):
        _refuse()
    except Exception:
        return ReceiptAdoption("UNKNOWN")
    if agent._receipt_plan_cache or agent._closed:
        _refuse()  # Same-Agent operation raced the awaited majority preflight.
    cache = agent._receipt_plan_cache
    cache.update(plan_id=document["plan"], output_type=output_type,
                 receipt_handle_mode=True, receipt_explicit_config=config)
    guard = _guard(agent, authority, document, output_type)
    try:
        guard()
    except BaseException:
        cache["poisoned"] = True
        raise
    cache["receipt_request_pins"] = {
        invocation.invocation_id: _ReceiptPin(digest, document["issuance"], before_claim=guard)}
    return ReceiptAdoption("ADOPTED")

adoptable_receipt_identity

adoptable_receipt_identity(handle: str, authority: Any) -> ReceiptHandleIdentity

Authenticated host registration data; exposes no request payload.

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

def adoptable_receipt_identity(handle: str, authority: Any) -> ReceiptHandleIdentity:
    """Authenticated host registration data; exposes no request payload."""
    journal, scope, principal, versions = _authority(authority)
    document = _open(handle, journal, schema=_SCHEMA, extra_fields={"configuration"})
    if (document["scope"] != scope.path or document["principal"] != principal
            or document["executables"] != [list(item) for item in versions]):
        _refuse()
    return ReceiptHandleIdentity(document["invocation"], document["issuance"], scope, principal)

prepare_adoptable_receipt_handle async

prepare_adoptable_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

Seal host declaration and explicit config; host registers before run.

Use adoptable_receipt_identity to register the exact identity in a durable host registry. authority.require_active_receipt must synchronously return exact True at adoption and preclaim, or refuse. Missing, revoked, async or unavailable registry refuses. Concurrent revocation is not atomically fenced with the Mongo claim; hosts needing that must not use this API.

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

async def prepare_adoptable_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:
    """Seal host declaration and explicit config; host registers before run.

    Use adoptable_receipt_identity to register the exact identity in a durable
    host registry. authority.require_active_receipt must synchronously return
    exact True at adoption and preclaim, or refuse. Missing, revoked, async or
    unavailable registry refuses. Concurrent revocation is not atomically
    fenced with the Mongo claim; hosts needing that must not use this API.
    """
    from symfonic.agent.facade import Agent

    journal, _, _, versions = _authority(authority)
    if type(agent) is not Agent or agent._receipt_plan_cache is None:
        _refuse()
    cache = agent._receipt_plan_cache
    prior_handle_mode = cache.get("receipt_handle_mode")
    prior_dispatch_epoch = cache.get("receipt_dispatch_epoch", 0)
    fingerprint, config = configuration_fingerprint(agent, authority, versions, output_type)
    handle = await prepare_receipt_handle(
        agent, prompt, invocation=invocation, authority=authority,
        attachments=attachments, history=history, state=state,
        session_id=session_id, output_type=output_type)
    pin = cache["receipt_request_pins"][invocation.invocation_id]
    try:
        document = _open(handle, journal)
        document.update(schema=_SCHEMA, configuration=fingerprint)
        _guard(agent, authority, document, output_type, require_active=False)()
        guard = _guard(agent, authority, document, output_type)
        handle = _seal(journal, document)
        pin.before_claim = guard
        cache["receipt_explicit_config"] = config
        return handle
    except BaseException:
        pins = cache["receipt_request_pins"]
        if pins.get(invocation.invocation_id) is pin and pin.state == "PENDING":
            pin.state = "ABANDONED"
            del pins[invocation.invocation_id]
        if (not prior_handle_mode and not pins
                and cache.get("receipt_dispatch_epoch", 0) == prior_dispatch_epoch):
            cache.pop("receipt_handle_mode", None)
        raise