symfonic.platform.native_entry_journal

native_entry_journal

Opt-in, host-owned evidence that an invocation reached the kernel.

NativeEntryJournal

NativeEntryJournal(path: str | Path, *, release_version: str, deployment_id: str, entrypoint: str)

Write metadata once for each kernel bind_run observer hook.

Supply this callable through ObservabilityServices.event_sinks and telemetry. Labels are bounded, non-sensitive operator identifiers, never user content. The host provisions the local directory and owns capacity and rotation. No run, lineage, tenant or session identifier is persisted or manufactured. Entry proves kernel reachability, not pre-kernel admission or population coverage. Each call uses a POSIX lock and synchronous fsync; remote file durability is not attested. A host-owned snapshot counts write failures even when kernel observers swallow them. Event delivery is a no-op.

Source code in src/symfonic/platform/native_entry_journal.py

def __init__(
    self, path: str | Path, *, release_version: str, deployment_id: str,
    entrypoint: str,
) -> None:
    if any(_label(value) == "unknown" for value in
           (release_version, deployment_id, entrypoint)):
        raise ValueError(
            "attribution requires bounded non-sensitive version/deployment/entrypoint labels"
        )
    try:
        import fcntl
    except ImportError as exc:
        raise RuntimeError("NativeEntryJournal requires POSIX local file locking") from exc
    self._locking = fcntl
    self._path = Path(path)
    self._release_version = release_version
    self._deployment_id = deployment_id
    self._entrypoint = entrypoint
    self._epoch = uuid4().hex
    self._failures = 0
    self._in_flight = 0
    self._lock = Lock()

capture_snapshot property

capture_snapshot: tuple[int, int, str]

Atomically read failed writes, active writes and this journal's epoch.

bind_run

bind_run(run_id: Any, **kwargs: Any) -> None

Accept observer lineage while projecting only identity presence.

Source code in src/symfonic/platform/native_entry_journal.py

def bind_run(self, run_id: Any, **kwargs: Any) -> None:
    """Accept observer lineage while projecting only identity presence."""
    with self._lock:
        self._in_flight += 1
    failed = True
    try:
        _append_row(self._path, {
            "kind": "kernel_native_entry.v1",
            "outcome": "native_entry",
            "release_version": self._release_version,
            "deployment_id": self._deployment_id,
            "entrypoint": self._entrypoint,
            "observed_at": datetime.now(UTC).isoformat(),
            "identity_known": bool(run_id),
        }, self._locking, os)
        failed = False
    finally:
        with self._lock:
            if failed:
                self._failures += 1
            self._in_flight -= 1

observe_run

observe_run(run_id: Any, **kwargs: Any) -> None

Use the host's best-effort observation channel with counted writes.

Source code in src/symfonic/platform/native_entry_journal.py

def observe_run(self, run_id: Any, **kwargs: Any) -> None:
    """Use the host's best-effort observation channel with counted writes."""
    self.bind_run(run_id, **kwargs)