Skip to content

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.

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 
50
51
52
53
54
55
56
57
@contextmanager
def bind_run_identity(identity: RunIdentity) -> Iterator[RunIdentity]:
    """Bind one identity for exactly the invocation that owns it."""
    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 
35
36
37
def current_run_identity() -> RunIdentity | None:
    """Return the invocation active in this async context, if any."""
    return _CURRENT.get()

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 
40
41
42
def push_run_identity(identity: RunIdentity):
    """Bind ``identity`` and return the token its lifecycle owner resets."""
    return _CURRENT.set(identity)

reset_run_identity 

reset_run_identity(token) -> None

Reset a token returned by :func:push_run_identity.

Source code in src/symfonic/kernel/contracts/run_identity.py 
45
46
47
def reset_run_identity(token) -> None:
    """Reset a token returned by :func:`push_run_identity`."""
    _CURRENT.reset(token)