symfonic.kernel.contracts.run_identity

run_identity

Run lineage shared by the kernel and delegation capabilities.

Each execution owns a fresh run_id. Delegation preserves the root id and names the immediate parent; it never reuses the parent's id for a child.

RunLineage

RunLineage(run_id: str, parents: tuple[RunLineage, ...], *, unresolved: bool = False, loop: Any = None)

One kernel run's place among the runs it was started inside.

Issued only by the kernel's run lifecycle, never by binding an identity, so code that binds a hand-built identity keeps the lineage of the run it is executing in, and code that strips its context has none -- until it starts a run, which descends from the run that spawned its task. Holds its parents strongly and no identity: a descendant keeps every ancestor's lineage -- and state keyed weakly on it -- alive after the ancestor's identity is gone (CORE-199 review F4).

unresolved: the run, or a run it descends from, named a parent or root that is not among its issued ancestors -- one that ended and was released, or was never issued here. Its lineage cannot be proved, so an erase fence holds it to every capture it may have lost (CORE-194).

memory: stamped by the kernel at run start from the compiled plan (CORE-194). A run that touched memory marks every ancestor, monotonically, so a durable root proof is never sealed above it.

loop: the event loop the run was issued on, weakly, or None. A capture keyed on the run lives on stores bound to that loop, so a descendant on another loop proves it there (run_admission).

Source code in src/symfonic/kernel/contracts/run_identity.py

def __init__(
    self,
    run_id: str,
    parents: tuple[RunLineage, ...],
    *,
    unresolved: bool = False,
    loop: Any = None,
) -> None:
    self.run_id = run_id
    self.parents = parents
    self.unresolved = unresolved or any(parent.unresolved for parent in parents)
    self._loop = None if loop is None else weakref.ref(loop)
    #: ``None`` unknown, ``False`` stamped memory-free, ``True`` touched.
    self.memory: bool | None = None

loop property

loop: Any

The loop this run was issued on, while it exists.

note_memory

note_memory(touched: bool) -> None

Stamp this run; True reaches every ancestor and is never cleared.

Source code in src/symfonic/kernel/contracts/run_identity.py

def note_memory(self, touched: bool) -> None:
    """Stamp this run; ``True`` reaches every ancestor and is never cleared."""
    if not touched:
        if self.memory is None:
            self.memory = False
        return
    for node in run_ancestry(self):
        node.memory = True

bind_run_identity

bind_run_identity(identity: RunIdentity) -> Iterator[RunIdentity]

Bind one identity for exactly the invocation that owns it.

Source code in src/symfonic/kernel/contracts/run_identity.py

@contextmanager
def bind_run_identity(identity: RunIdentity) -> Iterator[RunIdentity]:
    """Bind one identity for exactly the invocation that owns it."""
    _remember(identity)
    token = _CURRENT.set(identity)
    try:
        yield identity
    finally:
        _CURRENT.reset(token)

current_run_identity

current_run_identity() -> RunIdentity | None

Return the invocation active in this async context, if any.

Source code in src/symfonic/kernel/contracts/run_identity.py

def current_run_identity() -> RunIdentity | None:
    """Return the invocation active in this async context, if any."""
    return _CURRENT.get()

current_run_lineage

current_run_lineage() -> RunLineage | None

The lineage of the kernel run this context executes in, if any.

Source code in src/symfonic/kernel/contracts/run_identity.py

def current_run_lineage() -> RunLineage | None:
    """The lineage of the kernel run this context executes in, if any."""
    return _LINEAGE.get()

live_run_identity

live_run_identity(run_id: str | None) -> RunIdentity | None

The first identity bound under run_id while it is still alive.

Source code in src/symfonic/kernel/contracts/run_identity.py

def live_run_identity(run_id: str | None) -> RunIdentity | None:
    """The first identity bound under ``run_id`` while it is still alive."""
    return _LIVE.get(run_id) if run_id else None

push_run_identity

push_run_identity(identity: RunIdentity)

Bind identity and return the token its lifecycle owner resets.

Source code in src/symfonic/kernel/contracts/run_identity.py

def push_run_identity(identity: RunIdentity):
    """Bind ``identity`` and return the token its lifecycle owner resets."""
    _remember(identity)
    return _CURRENT.set(identity)

reset_run_identity

reset_run_identity(token) -> None

Reset a token returned by :func:push_run_identity or the kernel's issue.

Source code in src/symfonic/kernel/contracts/run_identity.py

def reset_run_identity(token) -> None:
    """Reset a token returned by :func:`push_run_identity` or the kernel's issue."""
    if type(token) is _Issued:
        from symfonic.kernel.contracts.run_spawn import exit_issuing

        exit_issuing(token.task, token.node)
        _LINEAGE.reset(token.lineage)
        _CURRENT.reset(token.identity)
        return
    _CURRENT.reset(token)

run_ancestry

run_ancestry(lineage: RunLineage) -> tuple[RunLineage, ...]

lineage and every run it descends from, each once.

Source code in src/symfonic/kernel/contracts/run_identity.py

def run_ancestry(lineage: RunLineage) -> tuple[RunLineage, ...]:
    """``lineage`` and every run it descends from, each once."""
    found: list[RunLineage] = []
    known: set[int] = set()  # by identity; every node is alive in ``found``
    pending = [lineage]
    while pending:
        node = pending.pop()
        if id(node) not in known:
            known.add(id(node))
            found.append(node)
            pending.extend(node.parents)
    return tuple(found)