Capture one Agent turn's erase admission before it reads (CORE-199 P3-4).
Raw graph-only PostgreSQL v1 memory, alone or with an exact
PostgresConversationLog (CORE-199 P3-4, CORE-194). Before retrieval and
the transcript read, an admission stage observes the erase generations
covering the turn's scope on every authority (erase_admission). The
composition's writer and lifecycle present that observation, as a value, to
every mutation they make; the store's mutation fence re-proves it after the
tenant lock, and the final transcript append re-proves it under the cursor
lock. A turn that read before an admin forget therefore cannot write what it
read after the forget completed.
The observation is recorded per kernel run, keyed weakly on the lineage the
kernel issues it (RunLineage), and a run's first capture is final:
nothing releases or replaces it, and binding another identity issues no
lineage, so a stale turn cannot refresh it. A run inherits the captures of
every run it was started inside -- through runs that captured nothing, and
after their own identities are gone -- so a delegated child or a worker is
held to its ancestors' tokens, proved before it reads (turn_lineage).
A public tool obtains the running turn's admission with :func:turn_writer.
Only memory_capabilities mints a registry, and its store's pool then
requires a run's admission for every raw mutation.
GuardedMemory
GuardedMemory(store: Any, generations: TurnGenerations, path: str)
The composition's writer and lifecycle; nothing else is forwarded.
Each call presents the running turn's admission, so a turn that never
captured refuses before the store is reached.
Source code in src/symfonic/capabilities/memory/guarded_memory.py
| def __init__(self, store: Any, generations: TurnGenerations, path: str) -> None:
self._store = store
self._generations = generations
self._path = path
|
TurnGenerations
TurnGenerations(log: Any, pool: Any = None, *, _mint: object = None)
The admission a composition captured for each live run; fail-closed.
The first capture of a run is its token for good: a later round, a
re-capture after a forget, or a second capture under another path never
replaces it.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def __init__(self, log: Any, pool: Any = None, *, _mint: object = None) -> None:
if _mint is not _MINT:
raise MemoryContractError(UNMINTED)
self._log = log
self._pool = pool if pool is not None else log.pool
|
capture
async
capture(path: str, *, recovered: bool = False, history: bool = False) -> None
Observe every authority before the turn reads; a carried transcript refuses.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| async def capture(self, path: str, *, recovered: bool = False, history: bool = False) -> None:
"""Observe every authority before the turn reads; a carried transcript refuses."""
if recovered or history: # history: in the stage too, so a wrapper cannot hide it
raise MemoryContractError(RECOVERED if recovered else HISTORY)
lineage = _lineage()
await prove_ancestry() # an inherited capture a forget covered refuses here
captures = _RUNS.setdefault(lineage, {})
if self not in captures:
observed = await observe_all(self._pool, path)
captures.setdefault(self, observed)
self._own(path)
|
token
token(path: str) -> tuple[tuple[str, int], ...]
The conversation log's token, which its final append re-proves.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def token(self, path: str) -> tuple[tuple[str, int], ...]:
"""The conversation log's token, which its final append re-proves."""
log_pool = getattr(self._log, "pool", None)
for fence in self._own(path):
if fence.pool is log_pool and not fence.memory:
return fence.token
raise MemoryContractError(NO_TOKEN)
|
ambient_fences
ambient_fences() -> tuple[Fence, ...]
Every capture of the running kernel run and the runs it descends from.
Source code in src/symfonic/capabilities/memory/turn_lineage.py
| def ambient_fences() -> tuple[Fence, ...]:
"""Every capture of the running kernel run and the runs it descends from."""
from symfonic.kernel.contracts.run_identity import current_run_lineage, run_ancestry
lineage = current_run_lineage()
return () if lineage is None else _fences(run_ancestry(lineage))
|
bind_turn_generations
bind_turn_generations(store: Any, scope: Any, conversation: Any) -> tuple[Any, Any]
(conversation, registry) for raw graph-only PostgreSQL v1 memory.
An exact history over an exact log carries the registry; memory alone gets
one of its own; other memory leaves an anchored log a transcript-only one.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def bind_turn_generations(store: Any, scope: Any, conversation: Any) -> tuple[Any, Any]:
"""``(conversation, registry)`` for raw graph-only PostgreSQL v1 memory.
An exact history over an exact log carries the registry; memory alone gets
one of its own; other memory leaves an anchored log a transcript-only one.
"""
from symfonic.capabilities.memory.conversation_history import NativeConversationHistory
if isinstance(conversation, NativeConversationHistory) and conversation.generations is not None:
raise MemoryContractError(UNMINTED)
from symfonic.capabilities.memory.binding import composition_scope
from symfonic.capabilities.memory.graph_native_admission import admit_raw_pg_mutation
from symfonic.capabilities.memory.graph_store import GraphBackedHms
if not (isinstance(store, GraphBackedHms)
and admit_raw_pg_mutation(composition_scope(scope), store, None)):
from symfonic.capabilities.memory.conversation_generation import bind_conversation
return bind_conversation(conversation), None
pool = memory_pool(store)
if conversation is None:
return None, _mint(None, pool)
if not _exact_history(conversation):
return conversation, None
from dataclasses import replace
registry = _mint(conversation.store, pool)
return replace(conversation, generations=registry), registry
|
fenced
fenced(conversation: Any) -> bool
The history carries a registry bound to its own exact log.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def fenced(conversation: Any) -> bool:
"""The history carries a registry bound to its own exact log."""
generations = getattr(conversation, "generations", None)
return (
_exact_history(conversation)
and type(generations) is TurnGenerations
and generations in _MINTED
and generations.log is conversation.store
)
|
fenced_composition
fenced_composition(capabilities: Any) -> bool
Whether any composed capability makes its turns capture an admission.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def fenced_composition(capabilities: Any) -> bool:
"""Whether any composed capability makes its turns capture an admission."""
from symfonic.capabilities.memory.conversation_history import NativeConversationHistory
from symfonic.capabilities.memory.turn_admission_stage import MemoryTurnAdmission
# The real type's ancestry: a subclass is still an admission, and an
# overridden __class__ never stands in for one.
return any(
issubclass(type(capability), MemoryTurnAdmission)
or (isinstance(capability, NativeConversationHistory)
and capability.generations is not None)
for capability in capabilities
)
|
guarded_writer
guarded_writer(store: Any, conversation: Any, scope: Any, generations: Any = None) -> Any
Wrap the composition's store when its turns capture an admission.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def guarded_writer(store: Any, conversation: Any, scope: Any, generations: Any = None) -> Any:
"""Wrap the composition's store when its turns capture an admission."""
if fenced(conversation):
return GuardedMemory(store, conversation.generations, scope.path)
if (
conversation is None
and type(generations) is TurnGenerations
and generations in _MINTED
and generations.log is None
and generations._pool is memory_pool(store) # noqa: SLF001
):
return GuardedMemory(store, generations, scope.path)
return store
|
history_stage
history_stage(conversation: Any, generations: Any = None, scope: Any = None) -> tuple
The history's own stages join the bundle; memory alone gets its admission.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def history_stage(conversation: Any, generations: Any = None, scope: Any = None) -> tuple:
"""The history's own stages join the bundle; memory alone gets its admission."""
from symfonic.capabilities.memory.conversation_history import NativeConversationHistory
if isinstance(conversation, NativeConversationHistory):
return (conversation,)
if conversation is None and generations is not None:
from symfonic.capabilities.memory.turn_admission_stage import MemoryTurnAdmission
return (MemoryTurnAdmission(generations, scope.path),)
return ()
|
minted
minted(generations: Any) -> bool
generations is a registry memory_capabilities minted.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def minted(generations: Any) -> bool:
"""``generations`` is a registry ``memory_capabilities`` minted."""
return type(generations) is TurnGenerations and generations in _MINTED
|
refuse_history
refuse_history(capabilities: Any) -> None
Refuse caller-supplied history on such a composition (CORE-194).
history is the recovered transcript by another door, with no
provenance at all: nothing says before which forget it was read, and a
fresh capture proves only the present.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def refuse_history(capabilities: Any) -> None:
"""Refuse caller-supplied history on such a composition (CORE-194).
``history`` is the recovered transcript by another door, with no
provenance at all: nothing says before which forget it was read, and a
fresh capture proves only the present.
"""
if fenced_composition(capabilities):
raise MemoryContractError(HISTORY)
|
refuse_recovery
refuse_recovery(capabilities: Any) -> None
Refuse a recovered turn over a composition whose turns capture an admission.
Its checkpoint carries a transcript no persisted erase generation covers,
so until one is stored with the checkpoint the recovered turn refuses
before it runs, and a host asks before it claims the pause.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def refuse_recovery(capabilities: Any) -> None:
"""Refuse a recovered turn over a composition whose turns capture an admission.
Its checkpoint carries a transcript no persisted erase generation covers,
so until one is stored with the checkpoint the recovered turn refuses
before it runs, and a host asks before it claims the pause.
"""
if fenced_composition(capabilities):
raise MemoryContractError(RECOVERED)
|
turn_writer
turn_writer(store: Any) -> AdmittedMemory
store's mutations under the running turn's admission, held as a value.
For a tool or work the turn spawns: the admission is taken now, inside the
run, and survives a stripped context or the run's end -- as the turn's
old observation, which refuses once a forget has covered it.
Source code in src/symfonic/capabilities/memory/turn_generation.py
| def turn_writer(store: Any) -> AdmittedMemory:
"""``store``'s mutations under the running turn's admission, held as a value.
For a tool or work the turn spawns: the admission is taken now, inside the
run, and survives a stripped context or the run's end -- as the turn's
old observation, which refuses once a forget has covered it.
"""
_lineage()
fences = ambient_fences()
if not fences:
raise MemoryContractError(NO_TOKEN)
return AdmittedMemory(store, _mint_admission(fences))
|