symfonic.kernel.contracts.lifecycle

lifecycle

The narrow ports a run's lifecycle is expressed through (RCX-8…RCX-11).

Everything an invocation opened is released through one of the three shapes here — a resource, a checkpointer, or a registered finalizer — and everything that happened while it unwound is reported in one value. That is the whole design: teardown is not a list of special cases scattered across the call sites that opened things, it is one unwinding of one stack, and the record it leaves is what makes "the run ended cleanly" a claim a test can check rather than an absence of complaints.

Ports are Protocols so an adopter's pool, saver, or socket satisfies them by shape (CON-S, dependency inversion). Standard library only, by rule.

AdapterPressure

Bases: Protocol

The BP-12 numbers an adapter exports for one run.

Structural on purpose: kernel.backpressure.AdapterMetrics satisfies it without importing this module, so the contract package keeps its rule of depending on nothing.

BackgroundEntry dataclass

BackgroundEntry(owner: str, purpose: str, deadline_seconds: float | None = None)

One registered unit of run-owned work (R7).

An entry, not a bare task: a task set records that something is running, while owner/purpose/deadline record who to ask when it is still running at teardown — which is the question a leaked task always raises.

BackgroundFailure dataclass

BackgroundFailure(owner: str, purpose: str, error: str)

Owned work that raised, attributed to the run that spawned it (RCX-11).

Without this, a task that finished badly inside the grace window was indistinguishable from one that finished well: the drain never retrieved the exception, so the only trace was asyncio's "Task exception was never retrieved" at some later collection — a traceback belonging to no run. A task teardown cancelled is not a failure; that is tasks_cancelled.

CheckpointerPort

Bases: Protocol

Durable run state: made ready before effects, flushed before close.

ensure_ready is separate from construction because readiness can open external resources (a pool, a schema migration) and must therefore be lazy, idempotent, and — critically — registered for teardown at the moment it succeeds, not at the call site that happened to trigger it.

close async

close() -> None

Release the durable handle, flushed or not.

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

async def close(self) -> None:
    """Release the durable handle, flushed or not."""

ensure_ready async

ensure_ready() -> None

Open whatever durable state this run needs. Idempotent per run.

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

async def ensure_ready(self) -> None:
    """Open whatever durable state this run needs. Idempotent per run."""

flush async

flush() -> None

Push buffered writes. Failure is reported; it never blocks close.

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

async def flush(self) -> None:
    """Push buffered writes. Failure is reported; it never blocks close."""

FinalizerFailure dataclass

FinalizerFailure(name: str, error: str)

A finalizer that raised or overran its budget, named so it can be fixed.

PressureRecord dataclass

PressureRecord(adapter: str, high_watermark: int = 0, byte_high_watermark: int = 0, blocked_seconds: float = 0.0, events_shed: Mapping[str, int] = (lambda: MappingProxyType({}))(), terminal_delivery_failed: bool = False, abandoned: bool = False)

One adapter's per-run pressure, frozen into the report (BP-12).

ResourcePort

Bases: Protocol

Anything a run acquires and must give back — a pool, a socket, a lease.

One method. A resource that also needed flushing is a checkpointer, and a resource that needed a health check is a resource plus a stage; widening this port would let every future collaborator negotiate its own teardown, which is precisely the sprawl this task removes.

name property

name: str

A stable label, used when the release is reported as having failed.

release async

release() -> None

Give the resource back. Called exactly once, on every exit path.

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

async def release(self) -> None:
    """Give the resource back. Called exactly once, on every exit path."""

TeardownReport dataclass

TeardownReport(run_id: str, reason: TeardownReason, elapsed_ms: float = 0.0, finalizers_run: int = 0, finalizer_failures: tuple[FinalizerFailure, ...] = (), tasks_awaited: int = 0, tasks_cancelled: int = 0, background_failures: tuple[BackgroundFailure, ...] = (), terminal_kind: str | None = None, terminal_delivered: bool = False, pressure: tuple[PressureRecord, ...] = ())

What teardown did, recorded once (RCX-11).

A run that dropped events or cancelled background work without saying so is indistinguishable from a healthy one. This value is the difference. It carries no exception objects and no live handles — only facts safe to log (ERR-5) — so it can cross a diagnostics or metrics boundary unchanged.

clean property

clean: bool

True when nothing had to be forced and nothing owned by the run failed.

A terminal-delivery failure counts. BP-10 case 3 — an attached consumer that never received done — is the loudest thing a run can do wrong, and a report that called such a run clean because the task bookkeeping happened to balance would be exactly the indistinguishability RCX-11 exists to prevent. Case 2 (the consumer left) is not a failure and is not recorded as one by any adapter.

explain

explain() -> str

Render the record as safe-to-log text (ERR-5).

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

def explain(self) -> str:
    """Render the record as safe-to-log text (ERR-5)."""
    lines = [
        f"run {self.run_id} torn down: reason={self.reason} "
        f"elapsed_ms={self.elapsed_ms:.3f} finalizers={self.finalizers_run} "
        f"awaited={self.tasks_awaited} cancelled={self.tasks_cancelled} "
        f"terminal={self.terminal_kind} delivered={self.terminal_delivered}"
    ]
    lines.extend(
        f"finalizer failed: {failure.name} — {failure.error}"
        for failure in self.finalizer_failures
    )
    lines.extend(
        f"background failed: {failure.owner}:{failure.purpose} — {failure.error}"
        for failure in self.background_failures
    )
    lines.extend(record.render() for record in self.pressure)
    return "\n".join(lines)