diff --git a/src/k1link/perception/streaming_lifecycle.py b/src/k1link/perception/streaming_lifecycle.py
new file mode 100644
index 0000000..a2c146b
--- /dev/null
+++ b/src/k1link/perception/streaming_lifecycle.py
@@ -0,0 +1,257 @@
+"""Lifecycle/fencing for one subprocess-backed full perception profile.
+
+Uses existing GraphState, StreamStart and bounded scheduler primitives. The
+trusted controller supplies child commands and one Worker-wide lease directory;
+neither is accepted from stream payloads. No backend queue or model is created
+here. External-client inventory and real network authentication remain separate.
+"""
+
+from __future__ import annotations
+
+import os
+import signal
+import subprocess
+import threading
+import time
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager, suppress
+from pathlib import Path
+from typing import Any, Literal
+
+from .graph_contracts import GraphState
+from .realtime_contract import StreamStart
+from .realtime_scene import _wire_integer
+from .streaming_queue import StreamBundle, StreamMailbox
+from .worker_lease import WorkerLease, WorkerLeaseError
+
+
+def _group_exists(pgid: int) -> bool:
+    try:
+        os.killpg(pgid, 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True  # Unknown is not a release proof.
+
+
+class StreamingLifecycle:
+    def __init__(
+        self,
+        start: StreamStart,
+        lease_root: Path,
+        mailbox: StreamMailbox,
+        stop: threading.Event,
+        *,
+        ttl_seconds: float = 2.0,
+        clock_ns: Callable[[], int] = time.monotonic_ns,
+    ) -> None:
+        self.start, self.mailbox, self.stop_event = start, mailbox, stop
+        self._lock = threading.RLock()
+        self._cleanup_lock = threading.Lock()
+        self._children: list[subprocess.Popen[bytes]] = []
+        self._threads: list[threading.Thread] = []
+        self._active = {"gpu": 0, "cpu": 0}
+        self.state = GraphState.CREATED
+        self.reason: str | None = None
+        self.stop_requested_ns: int | None = None
+        self.retired_ns: int | None = None
+        self.lease = WorkerLease(lease_root, start, ttl_seconds=ttl_seconds, clock_ns=clock_ns)
+        self.state = GraphState.STARTING
+        self._watchdog_stop = threading.Event()
+        self._watchdog = threading.Thread(
+            target=self._watch, name="perception-lease-watchdog", daemon=True
+        )
+        self._watchdog.start()
+
+    def _check(self, start: StreamStart, *, starting: bool = False) -> None:
+        allowed = (GraphState.STARTING, GraphState.RUNNING) if starting else (GraphState.RUNNING,)
+        if self.state not in allowed or self.stop_event.is_set():
+            raise WorkerLeaseError("runtime is not accepting work")
+        try:
+            self.lease.check(start)
+        except WorkerLeaseError:
+            # A stale client is rejected but cannot cancel the current owner.
+            if start == self.start:
+                self.request_stop("lease-lost")
+            raise
+
+    def renew(self, start: StreamStart) -> None:
+        with self._lock:
+            self._check(start, starting=True)
+            self.lease.renew(start)
+
+    def ready(self) -> None:
+        with self._lock:
+            self._check(self.start, starting=True)
+            if self.state != GraphState.STARTING:
+                raise WorkerLeaseError("warmup completion already consumed")
+            if any(p.poll() is not None for p in self._children):
+                self.request_stop("child-exited")
+                raise WorkerLeaseError("owned child exited during warmup")
+            self.state = GraphState.RUNNING
+
+    def spawn(self, factory: Callable[[], subprocess.Popen[bytes]]) -> subprocess.Popen[bytes]:
+        """Trusted, bounded process creation + registration, atomic with stop."""
+        with self._lock:
+            self._check(self.start, starting=True)
+            if self.state != GraphState.STARTING or len(self._children) >= 8:
+                raise WorkerLeaseError("model processes may start only once during warmup")
+            process = factory()
+            self._children.append(process)
+            if process.poll() is None and os.getpgid(process.pid) != process.pid:
+                self.request_stop("invalid-process-group")
+                raise WorkerLeaseError("runtime children need dedicated process groups")
+            return process
+
+    def track_thread(self, thread: threading.Thread) -> None:
+        with self._lock:
+            self._check(self.start, starting=True)
+            if len(self._threads) >= 8 or thread in self._threads:
+                raise WorkerLeaseError("runtime thread registration outside bound")
+            self._threads.append(thread)
+
+    def admit(self, start: StreamStart, bundle: StreamBundle) -> bool:
+        with self._lock:
+            self._check(start)
+            return self.mailbox.put(bundle)
+
+    @contextmanager
+    def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]:
+        with self._lock:
+            self._check(start)
+            if lane not in self._active or self._active[lane]:
+                raise WorkerLeaseError("only one active call per compute lane")
+            self._active[lane] += 1
+        try:
+            yield
+            self.check_current(start)
+        except BaseException:
+            self.request_stop("failed")
+            raise
+        finally:
+            with self._lock:
+                self._active[lane] -= 1
+
+    def check_current(self, start: StreamStart) -> None:
+        with self._lock:
+            self._check(start)
+
+    def validate_result_binding(self, value: object) -> None:
+        """Recheck at receipt/use; accepting a hash alone cannot renew authority."""
+        start = StreamStart.from_dict(value)
+        self.check_current(start)
+
+    def request_stop(self, reason: str = "cancelled") -> None:
+        if reason not in (
+            "completed",
+            "cancelled",
+            "lease-lost",
+            "child-exited",
+            "invalid-process-group",
+            "failed",
+        ):
+            raise ValueError("unknown runtime stop reason")
+        with self._lock:
+            if self.state in (GraphState.STOPPED, GraphState.CANCELLED, GraphState.FAILED):
+                return
+            self.reason = self.reason or reason
+            if self.stop_requested_ns is None:
+                self.stop_requested_ns = time.monotonic_ns()
+            self.state = GraphState.STOPPING
+            self.stop_event.set()
+            self.mailbox.cancel()
+
+    def _watch(self) -> None:
+        while not self._watchdog_stop.wait(0.05):
+            try:
+                with self._lock:
+                    if self.state in (GraphState.STARTING, GraphState.RUNNING):
+                        self._check(self.start, starting=True)
+                        if any(p.poll() is not None for p in self._children):
+                            self.request_stop("child-exited")
+                    stopping = self.state == GraphState.STOPPING
+                if stopping:
+                    self.stop_children()
+                    return
+            except (WorkerLeaseError, OSError):
+                self.request_stop("lease-lost")
+                self.stop_children()
+                return
+
+    def stop_children(self) -> bool:
+        """Stop only owned groups. No Boolean resource-release attestation input."""
+        with self._cleanup_lock:
+            with self._lock:
+                children = tuple(self._children)
+            for sent_signal, grace in ((signal.SIGTERM, 4.0), (signal.SIGKILL, 1.0)):
+                for process in reversed(children):
+                    if _group_exists(process.pid):
+                        with suppress(ProcessLookupError):
+                            os.killpg(process.pid, sent_signal)
+                deadline = time.monotonic() + grace
+                while True:
+                    complete = all(
+                        p.poll() is not None and not _group_exists(p.pid) for p in children
+                    )
+                    if complete or time.monotonic() >= deadline:
+                        break
+                    time.sleep(0.01)
+                if complete:
+                    return True
+            return False
+
+    def close(self, reason: str = "completed") -> bool:
+        self.request_stop(reason)
+        self._watchdog_stop.set()
+        children_stopped = self.stop_children()
+        if self._watchdog is not threading.current_thread():
+            self._watchdog.join(timeout=0.1)
+        with self._lock:
+            if self.lease.released:
+                return True
+            # Caller joins/finishes its adapters; active callbacks and payloads
+            # continue to fence even if all GPU subprocesses have died already.
+            if (
+                not children_stopped
+                or any(self._active.values())
+                or any(t.is_alive() for t in self._threads)
+                or self._watchdog.is_alive()
+                or not self.mailbox.quiescent
+            ):
+                return False
+            self.lease._retire_after_verified_stop()
+            self.retired_ns = time.monotonic_ns()
+            self.state = (
+                GraphState.STOPPED
+                if self.reason == "completed"
+                else GraphState.CANCELLED
+                if self.reason == "cancelled"
+                else GraphState.FAILED
+            )
+            return True
+
+    def snapshot(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "state": self.state.value,
+                "reason": self.reason,
+                "start": self.start.to_dict(),
+                "active_calls": dict(self._active),
+                "lease_released": self.lease.released,
+                "input_payloads_released": self.mailbox.quiescent,
+                "owned_children": len(self._children),
+                "live_children": sum(p.poll() is None for p in self._children),
+                "actuation_allowed": False,
+                "lease_renewals": self.lease.renewals,
+                "lease_deadline_monotonic_ns": _wire_integer(self.lease.deadline_ns),
+                "stop_requested_monotonic_ns": (
+                    _wire_integer(self.stop_requested_ns)
+                    if self.stop_requested_ns is not None
+                    else None
+                ),
+                "retired_monotonic_ns": (
+                    _wire_integer(self.retired_ns) if self.retired_ns is not None else None
+                ),
+                "event_clock": "worker-process-monotonic",
+            }
diff --git a/src/k1link/perception/worker_lease.py b/src/k1link/perception/worker_lease.py
new file mode 100644
index 0000000..914df39
--- /dev/null
+++ b/src/k1link/perception/worker_lease.py
@@ -0,0 +1,162 @@
+"""Worker-local cooperative fence for managed perception runtimes (POSIX).
+
+The controller selects one fixed private directory per physical Worker, shared
+by ALL its managed profile containers. It must not come from a job/stream.
+An OS lock serializes processes; a durable active record quarantines a crashed
+owner even after the kernel releases its lock. Time expiry never frees the GPU.
+This does not police unmanaged GPU clients or replace backend job claims.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import stat
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+from k1link.artifacts import write_json_atomic
+
+from .realtime_contract import StreamStart, _integer, _number
+
+LEASE_SCHEMA = "missioncore.perception-worker-lease/v1"
+
+
+class WorkerLeaseError(RuntimeError):
+    """Unknown, occupied, expired or quarantined Worker ownership."""
+
+
+class WorkerLease:
+    """Owned by the trusted local supervisor, never deserialized from a client.
+
+    Heartbeats are bounded local receipt-time renewals. A late heartbeat cannot
+    resurrect authority. Only the supervisor may retire after checking actual
+    process/thread/input release. There is deliberately no force/takeover API.
+    """
+
+    def __init__(
+        self,
+        root: Path,
+        start: StreamStart,
+        *,
+        ttl_seconds: float = 2.0,
+        clock_ns: Callable[[], int] = time.monotonic_ns,
+    ) -> None:
+        _number(ttl_seconds, "lease ttl", minimum=0.1)
+        if ttl_seconds > 30:
+            raise WorkerLeaseError("lease ttl exceeds 30 seconds")
+        if root.is_symlink():
+            raise WorkerLeaseError("lease directory must not be a symlink")
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.root = root.resolve()
+        metadata = self.root.stat()
+        if metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022:
+            raise WorkerLeaseError(
+                "lease directory must be controller-owned and not writable by peers"
+            )
+        self.start, self.clock_ns = start, clock_ns
+        self.ttl_ns = int(ttl_seconds * 1e9)
+        self.path = self.root / "worker.lock"
+        self.record_path = self.root / "owner.json"
+        self.descriptor = os.open(
+            self.path, os.O_RDWR | os.O_CREAT | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600
+        )
+        self.released = False
+        self.fenced = False
+        self.renewals = 0
+        self.pid = os.getpid()
+        try:
+            opened = os.fstat(self.descriptor)
+            if not stat.S_ISREG(opened.st_mode) or opened.st_nlink != 1:
+                raise WorkerLeaseError("lease lock must be one regular file")
+            if opened.st_uid != os.geteuid() or opened.st_mode & 0o077:
+                raise WorkerLeaseError("lease lock must be private")
+            self.identity = (opened.st_dev, opened.st_ino)
+            self._check_path()
+            try:
+                fcntl.flock(self.descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError as exc:
+                raise WorkerLeaseError("worker already owned") from exc
+            self._check_path()
+            self._check_previous()
+            self.last_ns = self.clock_ns()
+            _integer(self.last_ns, "worker monotonic time")
+            self.deadline_ns = self.last_ns + self.ttl_ns
+            write_json_atomic(self.record_path, self._record("active"))
+        except BaseException:
+            os.close(self.descriptor)
+            self.released = self.fenced = True
+            raise
+
+    def _check_path(self) -> None:
+        current = self.path.lstat()
+        if not stat.S_ISREG(current.st_mode) or (current.st_dev, current.st_ino) != self.identity:
+            raise WorkerLeaseError("worker lock identity changed")
+
+    def _check_previous(self) -> None:
+        try:
+            metadata = self.record_path.lstat()
+        except FileNotFoundError:
+            return
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 8192:
+            raise WorkerLeaseError("invalid worker ownership record")
+        if metadata.st_uid != os.geteuid() or metadata.st_mode & 0o077 or metadata.st_nlink != 1:
+            raise WorkerLeaseError("worker ownership record must be private")
+        try:
+            record = json.loads(self.record_path.read_bytes())
+            if (
+                set(record) != {"schema_version", "state", "start"}
+                or record["schema_version"] != LEASE_SCHEMA
+            ):
+                raise ValueError("schema")
+            previous = StreamStart.from_dict(record["start"])
+        except (ValueError, TypeError, KeyError) as exc:
+            raise WorkerLeaseError("corrupt worker ownership record") from exc
+        if previous.worker_id != self.start.worker_id:
+            raise WorkerLeaseError("controller directory belongs to another worker")
+        if record["state"] != "released":
+            raise WorkerLeaseError("previous owner unretired; resource-release recovery required")
+        if self.start.lease_generation <= previous.lease_generation:
+            raise WorkerLeaseError("lease generation must strictly increase")
+        if self.start.epoch_id == previous.epoch_id:
+            raise WorkerLeaseError("a new activation needs a new epoch")
+
+    def _record(self, state: str) -> dict[str, object]:
+        return {"schema_version": LEASE_SCHEMA, "state": state, "start": self.start.to_dict()}
+
+    def check(self, start: StreamStart) -> None:
+        if start != self.start:
+            raise WorkerLeaseError("stream identity or lease generation mismatch")
+        if self.released or self.fenced or os.getpid() != self.pid:
+            raise WorkerLeaseError("worker authority is fenced")
+        try:
+            self._check_path()
+            now = self.clock_ns()
+            _integer(now, "worker monotonic time")
+            if now < self.last_ns or now >= self.deadline_ns:
+                raise WorkerLeaseError("worker lease expired or clock moved backwards")
+            self.last_ns = now
+        except (OSError, ValueError, WorkerLeaseError) as exc:
+            self.fenced = True
+            raise WorkerLeaseError("worker lease no longer current") from exc
+
+    def renew(self, start: StreamStart) -> None:
+        self.check(start)
+        self.deadline_ns = self.last_ns + self.ttl_ns
+        self.renewals += 1
+
+    def _retire_after_verified_stop(self) -> None:
+        """Private supervisor boundary, not a caller-supplied release boolean."""
+        if self.released:
+            return
+        self.fenced = True
+        if os.getpid() != self.pid:
+            raise WorkerLeaseError("only the owning controller can retire")
+        self._check_path()
+        write_json_atomic(self.record_path, self._record("released"))
+        # Keep the stable file: unlink/recreate would split the lock domain.
+        fcntl.flock(self.descriptor, fcntl.LOCK_UN)
+        os.close(self.descriptor)
+        self.released = True
diff --git a/tests/test_perception_streaming_lifecycle.py b/tests/test_perception_streaming_lifecycle.py
new file mode 100644
index 0000000..18e4512
--- /dev/null
+++ b/tests/test_perception_streaming_lifecycle.py
@@ -0,0 +1,257 @@
+"""Small synthetic ownership/lifecycle checks; no model or GPU workload."""
+
+import json
+import os
+import subprocess
+import sys
+import threading
+from dataclasses import replace
+
+import pytest
+
+from k1link.perception.graph_contracts import GraphState
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.worker_lease import WorkerLeaseError
+
+
+def start(generation=1):
+    return StreamStart(
+        run_id=f"run-{generation}",
+        source_id="fixture",
+        worker_id="worker-006",
+        epoch_id=f"epoch-{generation}",
+        lease_generation=generation,
+        profile_sha256="a" * 64,
+        image_sha256="b" * 64,
+        effective_config_sha256="c" * 64,
+        calibration_sha256="d" * 64,
+        clock_domain_id="source-clock",
+        input_mode="live",
+    )
+
+
+def runtime(tmp_path, *, generation=1, clock=None):
+    return StreamingLifecycle(
+        start(generation),
+        tmp_path,
+        StreamMailbox(),
+        threading.Event(),
+        **({"clock_ns": clock} if clock is not None else {}),
+    )
+
+
+def test_only_ready_admits_and_clean_stop_requires_new_generation_and_epoch(tmp_path):
+    run = runtime(tmp_path)
+    packet = {"sequence": 0, "payload_bytes": 20}
+    try:
+        assert run.state == GraphState.STARTING
+        with pytest.raises(WorkerLeaseError, match="accepting"):
+            run.admit(run.start, packet)
+        run.ready()
+        assert run.admit(run.start, packet)
+        assert run.mailbox.take() is packet
+        with (
+            run.work(run.start, "gpu"),
+            pytest.raises(WorkerLeaseError, match="one active"),
+            run.work(run.start, "gpu"),
+        ):
+            pytest.fail("second GPU callback admitted")
+        run.validate_result_binding(run.start.to_dict())
+        assert not run.close()  # The CPU consumer still owns the input.
+        assert not run.lease.released and run.state == GraphState.STOPPING
+        with pytest.raises(WorkerLeaseError, match="owned"):
+            runtime(tmp_path, generation=2)
+        run.mailbox.release(packet)
+    finally:
+        assert run.close()
+    assert run.state == GraphState.STOPPED
+    with pytest.raises(WorkerLeaseError, match="generation"):
+        runtime(tmp_path)
+    next_run = runtime(tmp_path, generation=2)
+    assert next_run.close()
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("epoch_id", "old"),
+        ("run_id", "old"),
+        ("worker_id", "other-worker"),
+        ("source_id", "other-source"),
+        ("profile_sha256", "e" * 64),
+        ("image_sha256", "e" * 64),
+        ("effective_config_sha256", "e" * 64),
+        ("calibration_sha256", "e" * 64),
+        ("clock_domain_id", "other-clock"),
+        ("input_mode", "recorded-source-paced"),
+        ("lease_generation", 2),
+    ],
+)
+def test_stale_client_cannot_compute_publish_or_cancel_current_owner(tmp_path, field, value):
+    run = runtime(tmp_path)
+    try:
+        run.ready()
+        old = replace(run.start, **{field: value})
+        with pytest.raises(WorkerLeaseError, match="mismatch"):
+            run.renew(old)
+        with pytest.raises(WorkerLeaseError, match="mismatch"):
+            run.admit(old, {"sequence": 0, "payload_bytes": 1})
+        with pytest.raises(WorkerLeaseError, match="mismatch"):
+            run.validate_result_binding(old.to_dict())
+        assert not run.stop_event.is_set() and run.state == GraphState.RUNNING
+        run.check_current(run.start)
+    finally:
+        assert run.close()
+
+
+def test_expiry_cannot_be_renewed_and_never_releases_an_active_callback(tmp_path):
+    clock = [1_000_000_000]
+    run = runtime(tmp_path, clock=lambda: clock[0])
+    run.ready()
+    try:
+        with pytest.raises(WorkerLeaseError), run.work(run.start, "gpu"):
+            clock[0] += 2_000_000_000
+            with pytest.raises(WorkerLeaseError):
+                run.renew(run.start)
+            assert run.stop_event.is_set()
+            assert not run.close()
+            with pytest.raises(WorkerLeaseError, match="owned"):
+                runtime(tmp_path, generation=2)
+        assert run.state == GraphState.STOPPING and run.reason == "lease-lost"
+    finally:
+        assert run.close()
+    assert run.state == GraphState.FAILED
+
+
+def test_heartbeat_uses_receipt_time_and_backwards_clock_fences(tmp_path):
+    clock = [1_000_000_000]
+    run = runtime(tmp_path, clock=lambda: clock[0])
+    try:
+        clock[0] += 1_900_000_000
+        run.renew(run.start)
+        clock[0] += 1_900_000_000
+        run.ready()
+        clock[0] -= 1
+        with pytest.raises(WorkerLeaseError):
+            run.check_current(run.start)
+        assert run.reason == "lease-lost"
+    finally:
+        assert run.close()
+
+
+def test_watchdog_stops_a_real_owned_child_without_waiting_for_another_frame(tmp_path):
+    clock = [1_000_000_000]
+    run = runtime(tmp_path, clock=lambda: clock[0])
+    child = run.spawn(
+        lambda: subprocess.Popen(
+            [sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(10)"],
+            stdout=subprocess.PIPE,
+            start_new_session=True,
+        )
+    )
+    try:
+        assert child.stdout.readline() == b"ready\n"
+        run.ready()
+        clock[0] += 2_000_000_000
+        assert run.stop_event.wait(1)
+        child.wait(timeout=2)
+        with pytest.raises(WorkerLeaseError):
+            run.validate_result_binding(run.start.to_dict())
+        assert not run.lease.released  # Expiry is not retirement.
+    finally:
+        assert run.close()
+        child.stdout.close()
+
+
+def test_provider_failure_fences_new_work_and_pending_input(tmp_path):
+    run = runtime(tmp_path)
+    run.ready()
+    run.admit(run.start, {"sequence": 0, "payload_bytes": 1})
+    try:
+        with pytest.raises(ValueError, match="provider"), run.work(run.start, "cpu"):
+            raise ValueError("provider failed")
+        assert run.mailbox.quiescent and run.reason == "failed"
+        with pytest.raises(WorkerLeaseError):
+            run.check_current(run.start)
+    finally:
+        assert run.close()
+
+
+def test_live_registered_thread_prevents_retirement(tmp_path):
+    run = runtime(tmp_path)
+    release = threading.Event()
+    thread = threading.Thread(target=release.wait)
+    run.track_thread(thread)
+    thread.start()
+    try:
+        assert not run.close()
+        assert not run.lease.released
+    finally:
+        release.set()
+        thread.join(timeout=1)
+        assert run.close()
+
+
+def test_crashed_controller_remains_quarantined_after_os_lock_is_gone(tmp_path):
+    program = """
+import json, os, sys
+from pathlib import Path
+from k1link.perception.worker_lease import WorkerLease
+from k1link.perception.realtime_contract import StreamStart
+lease = WorkerLease(Path(sys.argv[1]), StreamStart.from_dict(json.loads(sys.argv[2])))
+os._exit(17)
+"""
+    child = subprocess.run(
+        [sys.executable, "-c", program, str(tmp_path), json.dumps(start().to_dict())],
+        timeout=3,
+        capture_output=True,
+    )
+    assert child.returncode == 17, child.stderr.decode()
+    with pytest.raises(WorkerLeaseError, match="unretired"):
+        runtime(tmp_path, generation=2)
+    assert json.loads((tmp_path / "owner.json").read_text())["state"] == "active"
+
+
+def test_cross_process_contender_is_denied_without_mutating_owner_record(tmp_path):
+    run = runtime(tmp_path)
+    before = (tmp_path / "owner.json").read_bytes()
+    program = """
+import json, sys
+from pathlib import Path
+from k1link.perception.worker_lease import WorkerLease, WorkerLeaseError
+from k1link.perception.realtime_contract import StreamStart
+try:
+    WorkerLease(Path(sys.argv[1]), StreamStart.from_dict(json.loads(sys.argv[2])))
+except WorkerLeaseError as exc:
+    print(str(exc)); sys.exit(21)
+sys.exit(99)
+"""
+    try:
+        child = subprocess.run(
+            [sys.executable, "-c", program, str(tmp_path), json.dumps(start(2).to_dict())],
+            capture_output=True,
+            timeout=3,
+        )
+        assert child.returncode == 21 and b"already owned" in child.stdout
+        assert (tmp_path / "owner.json").read_bytes() == before
+    finally:
+        assert run.close()
+
+
+def test_unknown_or_replaced_lock_cannot_grant_authority(tmp_path):
+    run = runtime(tmp_path)
+    run.ready()
+    original = tmp_path / "original.lock"
+    os.rename(tmp_path / "worker.lock", original)
+    (tmp_path / "worker.lock").write_bytes(b"")
+    try:
+        with pytest.raises(WorkerLeaseError):
+            run.check_current(run.start)
+        with pytest.raises(WorkerLeaseError, match="identity"):
+            run.close()
+    finally:
+        # Synthetic fixture only: restore exact inode so cleanup can retire.
+        os.replace(original, tmp_path / "worker.lock")
+        assert run.close()