symfonic.kernel.lifecycle

lifecycle

The one owner of everything a run opened (RCX-10, RCX-11, BP-8…BP-12).

Before this module, "clean up" was a habit: each call site that acquired something remembered — or forgot — to release it, on the paths it happened to think about. Success was usually handled, error sometimes, cancellation almost never, and nothing anywhere recorded what had been skipped.

:class:RunLifecycle replaces the habit with a stack. Resources, checkpointer readiness, background work, terminal delivery, and arbitrary finalizers all register with the same owner, unwind in one order on every exit path, and leave one :class:TeardownReport behind. The report is the point: a run that cancelled work or lost a terminal event now says so, instead of looking exactly like a healthy one.

RunLifecycle

RunLifecycle(run_id: str, *, deadline_seconds: float | None = None, grace_seconds: float = DEFAULT_TEARDOWN_GRACE_SECONDS)

Owns one run's releasable state and unwinds it exactly once.

Source code in src/symfonic/kernel/lifecycle.py

def __init__(
    self,
    run_id: str,
    *,
    deadline_seconds: float | None = None,
    grace_seconds: float = DEFAULT_TEARDOWN_GRACE_SECONDS,
) -> None:
    self.run_id = run_id
    self._deadline = RunDeadline(run_id, deadline_seconds)
    self._grace = grace_seconds
    self._registry = BackgroundRegistry(run_id, run_deadline_seconds=deadline_seconds)
    self._finalizers = FinalizerStack(grace_seconds=grace_seconds)
    self._pressure: list[PressureRecord] = []
    self._ready = False
    self._closed = False
    self._report: TeardownReport | None = None
    self._terminal_kind: str | None = None
    self._terminal_delivered = False

deadline property

deadline: RunDeadline

The run's clock: how much time is left, and what running out means.

report property

report: TeardownReport | None

The teardown record, or None while the run is still open.

acquire

acquire(resource: ResourcePort) -> ResourcePort

Take a resource and register its release in the same expression.

Returning the resource is what makes the acquisition and the release impossible to separate in a diff: there is no way to write the first without the second. The release lands in post_drain: background work this run spawned may still be writing through the resource, and giving it back first is a use-after-release the drain would then hide.

Source code in src/symfonic/kernel/lifecycle.py

def acquire(self, resource: ResourcePort) -> ResourcePort:
    """Take a resource and register its release in the same expression.

    Returning the resource is what makes the acquisition and the release
    impossible to separate in a diff: there is no way to write the first
    without the second. The release lands in ``post_drain``: background work
    this run spawned may still be writing through the resource, and giving
    it back first is a use-after-release the drain would then hide.
    """
    self.require_open("acquiring a resource")
    self._finalizers.push(resource.release, name=f"resource:{resource.name}")
    return resource

bounded async

bounded(step: Awaitable[Any], *, doing: str) -> Any

Await one step of forward progress inside the run's remaining budget.

This is how a deadline reaches a consumer-paced entry point, and it is the reason the budget is not left to :meth:scope alone: a scope can only bound a call one task both enters and leaves, so before this existed deadline_seconds was enforced on run() and silently ignored on every streaming projection — one plan value with two meanings, which is the entry-point drift the kernel exists to remove.

Source code in src/symfonic/kernel/lifecycle.py

async def bounded(self, step: Awaitable[Any], *, doing: str) -> Any:
    """Await one step of forward progress inside the run's remaining budget.

    This is how a deadline reaches a *consumer-paced* entry point, and it
    is the reason the budget is not left to :meth:`scope` alone: a scope
    can only bound a call one task both enters and leaves, so before this
    existed ``deadline_seconds`` was enforced on ``run()`` and silently
    ignored on every streaming projection — one plan value with two
    meanings, which is the entry-point drift the kernel exists to remove.
    """
    return await self._deadline.bounded(step, doing=doing)

deliver_terminal async

deliver_terminal(event: KernelEvent, deliver: Callable[[KernelEvent], Awaitable[Any]] | None = None) -> bool

Record the run's terminal event and attempt its delivery (BP-10).

The attempt is bounded and unshielded, and that is deliberate. On the cancellation path the delivering task is already cancelled, so a shielded attempt would keep a dead consumer's write alive past the run that owed it; failing fast and recording BP-10 case 3 is the honest outcome. Buffer pressure never reaches here — capacity for a terminal event is reserved or preempted upstream (BP-4).

A grace of 0 means "cancel owned work immediately", not "skip teardown"; the delivery it bounds is left unbounded rather than cut before it can write a byte.

Source code in src/symfonic/kernel/lifecycle.py

async def deliver_terminal(
    self,
    event: KernelEvent,
    deliver: Callable[[KernelEvent], Awaitable[Any]] | None = None,
) -> bool:
    """Record the run's terminal event and attempt its delivery (BP-10).

    The attempt is bounded and unshielded, and that is deliberate. On the
    cancellation path the delivering task is already cancelled, so a
    shielded attempt would keep a dead consumer's write alive past the run
    that owed it; failing fast and recording BP-10 case 3 is the honest
    outcome. Buffer pressure never reaches here — capacity for a terminal
    event is reserved or preempted upstream (BP-4).

    A grace of ``0`` means "cancel owned work immediately", not "skip
    teardown"; the delivery it bounds is left unbounded rather than cut
    before it can write a byte.
    """
    self._terminal_kind = event.kind
    if deliver is None:
        self._terminal_delivered = True
        return True
    try:
        async with teardown_budget(self._grace):
            await deliver(event)
    except asyncio.CancelledError:
        self._terminal_delivered = False
        raise  # CXL-3: cancellation is recorded, never swallowed
    except Exception:  # noqa: BLE001 — a failed delivery is counted, not raised
        self._terminal_delivered = False
        return False
    self._terminal_delivered = True
    return True

ensure_ready async

ensure_ready(checkpointer: CheckpointerPort | None) -> None

Make durable state ready once, registering its teardown on success.

Registration happens here rather than at construction because a checkpointer that never became ready has nothing to flush, and a checkpointer that did must be flushed no matter which of the eight call sites happened to trigger readiness first.

Both land in post_drain: a spawned checkpoint writer flushed before it was drained would have its writes silently dropped.

Source code in src/symfonic/kernel/lifecycle.py

async def ensure_ready(self, checkpointer: CheckpointerPort | None) -> None:
    """Make durable state ready once, registering its teardown on success.

    Registration happens *here* rather than at construction because a
    checkpointer that never became ready has nothing to flush, and a
    checkpointer that did must be flushed no matter which of the eight call
    sites happened to trigger readiness first.

    Both land in ``post_drain``: a spawned checkpoint writer flushed before
    it was drained would have its writes silently dropped.
    """
    if checkpointer is None or self._ready:
        return
    self.require_open("preparing the checkpointer")
    await checkpointer.ensure_ready()
    self._ready = True
    self._finalizers.push(checkpointer.close, name="checkpointer.close")
    self._finalizers.push(checkpointer.flush, name="checkpointer.flush")

owns_timeout

owns_timeout(error: BaseException) -> bool

True only for the timeout this run's own deadline raised (CXL-6).

Source code in src/symfonic/kernel/lifecycle.py

def owns_timeout(self, error: BaseException) -> bool:
    """True only for the timeout this run's own deadline raised (CXL-6)."""
    return self._deadline.owns(error)

push_finalizer

push_finalizer(finalizer: Callable[[], Any], *, name: str | None = None, phase: FinalizerPhase = 'post_drain') -> None

Register cleanup to run in reverse registration order, within its phase.

The default phase runs after owned work is drained, which is what keeps whatever the finalizer releases alive while the run's own tasks may still be using it. Pass phase="pre_drain" only for the inverse dependency: a finalizer the drain itself is waiting on.

Source code in src/symfonic/kernel/lifecycle.py

def push_finalizer(
    self,
    finalizer: Callable[[], Any],
    *,
    name: str | None = None,
    phase: FinalizerPhase = "post_drain",
) -> None:
    """Register cleanup to run in reverse registration order, within its phase.

    The default phase runs *after* owned work is drained, which is what
    keeps whatever the finalizer releases alive while the run's own tasks
    may still be using it. Pass ``phase="pre_drain"`` only for the inverse
    dependency: a finalizer the drain itself is waiting on.
    """
    self.require_open("registering a finalizer")
    label = name or getattr(finalizer, "__name__", "finalizer")
    self._finalizers.push(finalizer, name=label, phase=phase)

record_pressure

record_pressure(adapter: str, metrics: AdapterPressure) -> None

Freeze one adapter's BP-12 numbers into the run's record.

Source code in src/symfonic/kernel/lifecycle.py

def record_pressure(self, adapter: str, metrics: AdapterPressure) -> None:
    """Freeze one adapter's BP-12 numbers into the run's record."""
    self._pressure.append(
        PressureRecord(
            adapter=adapter,
            high_watermark=metrics.high_watermark,
            byte_high_watermark=getattr(metrics, "byte_high_watermark", 0),
            blocked_seconds=metrics.blocked_seconds,
            events_shed=MappingProxyType(dict(metrics.events_shed)),
            terminal_delivery_failed=metrics.terminal_delivery_failed,
            abandoned=metrics.abandoned,
        )
    )

require_open

require_open(action: str = 'write') -> None

Refuse post-close mutation and emission (RCX-10, BP-9).

Reads stay legal — diagnostics about a finished run are the reason the object survives its teardown at all.

Source code in src/symfonic/kernel/lifecycle.py

def require_open(self, action: str = "write") -> None:
    """Refuse post-close mutation and emission (RCX-10, BP-9).

    Reads stay legal — diagnostics about a finished run are the reason the
    object survives its teardown at all.
    """
    if self._closed:
        raise ContractViolationError(
            f"run {self.run_id} is closed; {action} after teardown is a "
            "defect, not a late arrival to accommodate (BP-9)."
        )

scope async

scope() -> AsyncIterator[None]

The run's cancellation scope: a deadline is an error, not a cancel.

asyncio.timeout implements an elapsed deadline by cancelling the body, so without this translation the two situations CXL-6 insists on separating would reach the caller as the same exception. External cancellation passes straight through, unswallowed (CXL-2/CXL-3).

Enter it only in a task that both enters and leaves it. The kernel's own entry points do not: they bill the same clock through :meth:bounded, one step at a time, which is legal from a generator that may be resumed by a different task than the one that suspended it.

Source code in src/symfonic/kernel/lifecycle.py

@asynccontextmanager
async def scope(self) -> AsyncIterator[None]:
    """The run's cancellation scope: a deadline is an error, not a cancel.

    ``asyncio.timeout`` implements an elapsed deadline by cancelling the
    body, so without this translation the two situations CXL-6 insists on
    separating would reach the caller as the same exception. External
    cancellation passes straight through, unswallowed (CXL-2/CXL-3).

    Enter it only in a task that both enters and leaves it. The kernel's own
    entry points do not: they bill the same clock through :meth:`bounded`,
    one step at a time, which is legal from a generator that may be resumed
    by a different task than the one that suspended it.
    """
    async with self._deadline.scope():
        yield

spawn

spawn(work: Any, *, owner: str, purpose: str, deadline_seconds: float | None = None) -> asyncio.Task[Any]

Create run-owned background work (RCX-8/BP-8); never fire-and-forget.

Source code in src/symfonic/kernel/lifecycle.py

def spawn(
    self,
    work: Any,
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> asyncio.Task[Any]:
    """Create run-owned background work (RCX-8/BP-8); never fire-and-forget."""
    return self._registry.spawn(
        work, owner=owner, purpose=purpose, deadline_seconds=deadline_seconds
    )

teardown async

teardown(reason: TeardownReason = 'completed') -> TeardownReport

Unwind everything, once, and return the record of having done it.

Idempotent because it runs from a finally that several exit paths can reach: a second call returns the first call's report rather than re-running finalizers against state they already released.

Source code in src/symfonic/kernel/lifecycle.py

async def teardown(self, reason: TeardownReason = "completed") -> TeardownReport:
    """Unwind everything, once, and return the record of having done it.

    Idempotent because it runs from a ``finally`` that several exit paths
    can reach: a second call returns the first call's report rather than
    re-running finalizers against state they already released.
    """
    if self._report is not None:
        return self._report
    started = time.perf_counter()
    ran, failures = await self._finalizers.unwind("pre_drain")
    drained = await self._registry.drain(grace_seconds=self._grace)
    after_ran, after_failures = await self._finalizers.unwind("post_drain")
    self._closed = True
    self._report = TeardownReport(
        run_id=self.run_id,
        reason=reason,
        elapsed_ms=(time.perf_counter() - started) * 1000.0,
        finalizers_run=ran + after_ran,
        finalizer_failures=failures + after_failures,
        tasks_awaited=drained.awaited,
        tasks_cancelled=drained.cancelled,
        background_failures=drained.failures,
        terminal_kind=self._terminal_kind,
        terminal_delivered=self._terminal_delivered,
        pressure=tuple(self._pressure),
    )
    return self._report