Skip to content

symfonic.capabilities.memory.mongo_run_lineage

mongo_run_lineage

A native Mongo v1 Agent turn's erase admission, captured per kernel run (CORE-199/171).

A native Mongo turn's journal claim records an erase witness: every covering key's erase generation for the turn's scope, read before the turn reads (mongo_operation_journal.claim_invocation). Its WRITE and FLUSH re-prove that witness inside the tenant guard. This module carries the same witness to every run started inside that turn: the claimed row's own stored witness is recorded on the kernel run's issued RunLineage, keyed weakly, together with the database the turn claimed on. Nothing replaces or refreshes it.

The claim runs in its caller's context, before the kernel issues the turn's own run: a turn nested in another run claims, and checks its first effects, inside that caller's run. The witness waits for the run issued after the claim -- the first effect bound to another run identity than the claim's -- so it lands on the turn's own lineage, never its caller's. A run the caller started before the turn is not the turn's descendant (CORE-194 review P2).

The kernel proves every inherited capture when a run starts, through the port in kernel.contracts.run_admission: the witness of each ancestor is compared on that ancestor's own database, on its own loop, with the erase records now. A forget of the turn's scope or an ancestor committed since refuses the run before any stage, provider call, tool or effect -- whether it composes native Mongo memory at another scope, other memory, or none. A descendant's own fresh witness never stands in for its ancestor's.

Scope of the proof: captures live in this process. A run that cannot reach its ancestor's database, or reads a row with no witness, refuses; nothing is recorded for a run that was not issued here, and no history is reconstructed.

claimed async 

claimed(turn: Any, journal: Any, owner: Any) -> None

Hold the witness owner's claim stored, until turn's run reads.

Source code in src/symfonic/capabilities/memory/mongo_run_lineage.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def claimed(turn: Any, journal: Any, owner: Any) -> None:
    """Hold the witness ``owner``'s claim stored, until ``turn``'s run reads."""
    from symfonic.memory.backends import mongo_operation_journal_codec as codec
    from symfonic.memory.backends.mongo_native_erased import WITNESS_FIELD

    expected = codec.identity(owner.invocation_id, owner.full_scope, owner.request_digest)
    from symfonic.kernel.contracts.run_identity import current_run_identity

    row = await journal._rows.find_one({"_id": expected["_id"]}, {WITNESS_FIELD: 1})
    caller = current_run_identity()
    _CLAIMED[turn] = (
        (journal._rows.database, owner.full_scope, _witness_of(row)),
        None if caller is None else weakref.ref(caller),
    )

claiming 

claiming(turn: Any, journal: Any) -> Any

journal.claim_invocation for turn, holding the witness it stored.

The receipt dispatch proves the caller's inherited ancestry before this claim (agent.receipt_dispatch), since the claim runs where the caller's run is not visible; the kernel proves it again when the run starts.

Source code in src/symfonic/capabilities/memory/mongo_run_lineage.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def claiming(turn: Any, journal: Any) -> Any:
    """``journal.claim_invocation`` for ``turn``, holding the witness it stored.

    The receipt dispatch proves the caller's inherited ancestry before this
    claim (``agent.receipt_dispatch``), since the claim runs where the caller's
    run is not visible; the kernel proves it again when the run starts.
    """

    async def claim(invocation_id: str, scope: Any, digest: str) -> Any:
        from symfonic.capabilities.memory.operation_journal import ClaimAcquired

        result = await journal.claim_invocation(invocation_id, scope, digest)
        if type(result) is ClaimAcquired:
            await claimed(turn, journal, result.owner)
        return result

    return claim

prove_before_claim async 

prove_before_claim() -> None

Refuse a native Mongo claim whose inherited captures a forget covered.

A receipt Agent claims before its kernel run starts, in a task where the caller's run is not visible, so the ancestry is resolved here as the kernel will resolve the run's parent and proved through the kernel's port: each ancestor on its own loop, with its own clients, also when this job was handed to a thread running another loop (review P1-X1).

Source code in src/symfonic/capabilities/memory/mongo_run_lineage.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
async def prove_before_claim() -> None:
    """Refuse a native Mongo claim whose inherited captures a forget covered.

    A receipt Agent claims before its kernel run starts, in a task where the
    caller's run is not visible, so the ancestry is resolved here as the kernel
    will resolve the run's parent and proved through the kernel's port: each
    ancestor on its own loop, with its own clients, also when this job was
    handed to a thread running another loop (review P1-X1).
    """
    from symfonic.kernel.contracts.run_admission import prove_ancestry_of
    from symfonic.kernel.contracts.run_identity import current_run_lineage
    from symfonic.kernel.contracts.run_spawn import issuing_lineage, spawned_lineage

    await prove_ancestry_of(current_run_lineage() or issuing_lineage() or spawned_lineage())

prove_runs async 

prove_runs(runs: tuple[Any, ...]) -> None

The kernel's run-start preflight over native Mongo captures of runs.

Source code in src/symfonic/capabilities/memory/mongo_run_lineage.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def prove_runs(runs: tuple[Any, ...]) -> None:
    """The kernel's run-start preflight over native Mongo captures of ``runs``."""
    from symfonic.memory.backends.mongo_native_erased import ERASED_COLLECTION, witness

    proved: set[tuple[int, str]] = set()
    for run in runs:
        for database, scope, captured in _RUNS.get(run, ()):
            key = (id(database), scope.path)
            if key in proved:
                continue
            try:
                current = await witness(database[ERASED_COLLECTION], scope)
            except Exception as exc:  # an unreachable ancestor store cannot prove it
                raise MemoryContractError(UNPROVABLE) from exc
            if current != captured:
                raise MemoryContractError(INHERITED_ERASED)
            proved.add(key)

record 

record(turn: Any) -> None

Record turn's claimed witness on the turn's own kernel run, once.

The claim precedes the kernel run; its first effect, before any read, runs inside it. An effect still bound to the identity the claim ran under is the caller's, before the turn's run issued: the witness stays pending. A turn effecting outside any run leaves nothing to inherit.

Source code in src/symfonic/capabilities/memory/mongo_run_lineage.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def record(turn: Any) -> None:
    """Record ``turn``'s claimed witness on the turn's own kernel run, once.

    The claim precedes the kernel run; its first effect, before any read, runs
    inside it. An effect still bound to the identity the claim ran under is
    the caller's, before the turn's run issued: the witness stays pending. A
    turn effecting outside any run leaves nothing to inherit.
    """
    from symfonic.kernel.contracts.run_identity import current_run_identity, current_run_lineage

    lineage, pending = current_run_lineage(), _CLAIMED.get(turn)
    if lineage is None or pending is None:
        return
    entry, caller = pending
    claimed_under = None if caller is None else caller()
    if claimed_under is not None and claimed_under is current_run_identity():
        return  # still the caller's run: the turn's own has not issued
    del _CLAIMED[turn]
    _RUNS[lineage] = (*_RUNS.get(lineage, ()), entry)