fix: enforce persistent project controller leases

zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md

@@ -25,6 +25,8 @@ Drive the mission only through the orchestrator runtime tools:
 - Quiescent `mission_running`: when no runnable work remains and evidence supports closure, call `end_mission`.
 - `attention_needed`: read every open item and its evidence, decide exactly once per item with `decide_attention`, then call `advance_project`.
 - User-requested or justified mission cancellation: call `abort_project(project_id, reason)`.
+- Intentional handoff to another root orchestrator: stop all mutation, then call `release_project(project_id)`. Never share one writable workspace between root orchestrators.
+- Controller crash with a stale lease: call `recover_workspace_lease(workspace_dir, reason)` only after stopping mutation. Recovery succeeds only when the recorded same-host controller PID is dead and writes a durable audit record.
 - `done`, `failed`, `aborted`: terminal states. Do not continue mission work unless the user starts a new scope or explicitly asks for forensic inspection.
 
 `submit_plan` and `decide_attention` persist state; they do not dispatch work. `advance_project` dispatches workers, validators, merge work, and gate evaluation according to runtime state. `end_mission` requests runtime closure and terminal review; call it only after planning, validation, gates, evidence review, and open attention handling support closure.
@@ -361,8 +363,10 @@ Orchestrator tools:
 - `decide_attention(project_id, decisions)`: resolve every open attention item with exactly one decision, then return to runtime flow. Call `advance_project` afterward.
 - `end_mission(project_id)`: request runtime closure and terminal review only after work is quiescent and evidence supports closure.
 - `abort_project(project_id, reason)`: terminal cancellation with a recorded reason.
+- `release_project(project_id)`: explicitly release persistent workspace ownership so another root orchestrator can take over. Read-only inspection by other sessions remains available before release. Running work always blocks release; use audited dead-controller recovery only after the owning process is proven dead.
+- `recover_workspace_lease(workspace_dir, reason)`: remove a stale or orphan lease only when the recorded same-host controller process is provably dead. It does not claim the workspace; start or claim the project after recovery.
 
-Every orchestrator tool returns an envelope with `projectId`, `state`, `projectRoot`, `harnessRoot`, and `dag`. Some lifecycle/decision tools intentionally return `dag=null`; `submit_plan` and `advance_project` return a compact frontier view; `inspect_project` returns the full task-list view. Trust the envelope, runtime files, and returned paths over session memory.
+Lifecycle and inspection tools return an envelope with `projectId`, `state`, `projectRoot`, `harnessRoot`, and `dag`. Some lifecycle/decision tools intentionally return `dag=null`; `submit_plan` and `advance_project` return a compact frontier view; `inspect_project` returns the full task-list view. Lease release and recovery tools return acknowledgements. Trust the envelope, runtime files, and returned paths over session memory.
 
 `submit_plan` accepts a `TaskList`:
 

zenith/src/zenith_harness/cli.py

@@ -250,9 +250,17 @@ def inspect_tasks_cmd(project_id: str, mission_id: str | None) -> None:
 @cli.command("abort-project")
 @click.argument("project_id")
 @click.option("--reason", required=True)
-def abort_project_cmd(project_id: str, reason: str) -> None:
+@click.option(
+    "--owner-id",
+    default=None,
+    help=(
+        "Stable controller/session id. Defaults to ZENITH_CONTROLLER_ID, "
+        "CODEX_THREAD_ID, or a Claude session id."
+    ),
+)
+def abort_project_cmd(project_id: str, reason: str, owner_id: str | None) -> None:
     """Mark a project Aborted (CLI-side: preserves tasks.json + attempts/)."""
-    from .controller import ProjectController
+    from .controller import ProjectController, ToolError
     from .dispatcher import MockDispatcher, MockTerminalReviewer
     from .models import TerminalReviewHandoff, WorkHandoff
 
@@ -262,7 +270,29 @@ def abort_project_cmd(project_id: str, reason: str) -> None:
         MockDispatcher(lambda r: WorkHandoff(node_id=r.task.id, done=False, report="aborted")),
         MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")),
     )
-    env = controller.abort_project(project_id, reason)
+    resolved_owner = owner_id or next(
+        (
+            os.environ[name]
+            for name in (
+                "ZENITH_CONTROLLER_ID",
+                "CODEX_THREAD_ID",
+                "CLAUDE_CODE_SESSION_ID",
+                "CLAUDE_SESSION_ID",
+            )
+            if os.environ.get(name)
+        ),
+        None,
+    )
+    if resolved_owner is None:
+        raise click.ClickException(
+            "abort-project requires --owner-id or ZENITH_CONTROLLER_ID "
+            "(CODEX_THREAD_ID and Claude session ids are also accepted)"
+        )
+    try:
+        controller.claim_project(project_id, resolved_owner)
+        env = controller.abort_project(project_id, reason, resolved_owner)
+    except ToolError as exc:
+        raise click.ClickException(str(exc)) from exc
     click.echo(f"Aborted {project_id}: state={env.state.state}")
 
 

zenith/src/zenith_harness/controller.py

@@ -1,4 +1,4 @@
-"""ProjectController — routes the 7 orchestrator MCP tools.
+"""ProjectController -- routes the 8 orchestrator MCP tools.
 
 See `specs/task_list/PRODUCT.md`. The controller owns:
 - envelope construction
@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from pathlib import Path
 
 from .config import HarnessConfig
 from .coordinator import MissionCoordinator
@@ -31,7 +32,7 @@
     TaskListPatch,
     TaskStateFile,
 )
-from .storage import ProjectStore
+from .storage import ProjectStore, WorkspaceLease, WorkspaceLeaseConflict
 from .task_list_patch import apply_patch
 from .task_validation import (
     ValidationError,
@@ -71,10 +72,35 @@ def __init__(
     # Tool methods
     # ------------------------------------------------------------------
 
-    def start_project(self, brief: str, workspace_dir: str) -> Envelope:
+    def start_project(
+        self, brief: str, workspace_dir: str, owner_id: str
+    ) -> Envelope:
         if not brief.strip():
             raise ToolError("invalid_brief", "brief is empty")
-        record = self.store.create_project(brief, workspace_dir)
+        requested_workspace = Path(workspace_dir).expanduser()
+        if not requested_workspace.is_absolute() or not requested_workspace.is_dir():
+            raise ToolError(
+                "invalid_workspace",
+                "workspace_dir must be an existing absolute directory",
+            )
+        project_id = self.store.generate_project_id(brief)
+        try:
+            self.store.claim_workspace_lease_for_workspace(
+                workspace_dir, project_id, owner_id
+            )
+        except ValueError as exc:
+            raise ToolError("invalid_owner", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_owned", str(exc)) from exc
+        try:
+            record = self.store.create_project(
+                brief, workspace_dir, project_id=project_id
+            )
+        except Exception:
+            self.store.release_workspace_lease_for_workspace(
+                workspace_dir, project_id, owner_id
+            )
+            raise
         mission_id = self.store.generate_mission_id(1)
         record.current_mission_id = mission_id
         self.store.save_project(record)
@@ -83,6 +109,69 @@ def start_project(self, brief: str, workspace_dir: str) -> Envelope:
         )
         return self._build_envelope(record.id, dag_mode="none")
 
+    def claim_project(self, project_id: str, owner_id: str) -> None:
+        try:
+            self.store.claim_workspace_lease(project_id, owner_id)
+        except ValueError as exc:
+            raise ToolError("invalid_owner", str(exc)) from exc
+        except FileNotFoundError as exc:
+            raise ToolError("invalid_workspace", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_owned", str(exc)) from exc
+
+    def release_project(
+        self,
+        project_id: str,
+        owner_id: str,
+    ) -> None:
+        try:
+            self.store.claim_workspace_lease(project_id, owner_id)
+        except ValueError as exc:
+            raise ToolError("invalid_owner", str(exc)) from exc
+        except FileNotFoundError as exc:
+            raise ToolError("invalid_workspace", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_owned", str(exc)) from exc
+        record = self.store.load_project(project_id)
+        state = self.store.load_state(project_id)
+        mission_id = self._current_mission_id(record, state)
+        if mission_id is not None:
+            task_state = self.store.load_task_state(project_id, mission_id)
+            running = sorted(
+                task_id
+                for task_id, entry in task_state.tasks.items()
+                if entry.status == "running"
+            )
+            if running:
+                raise ToolError(
+                    "workspace_busy",
+                    "cannot release controller ownership while tasks are running: "
+                    + ", ".join(running),
+                )
+        try:
+            self.store.release_workspace_lease(project_id, owner_id)
+        except ValueError as exc:
+            raise ToolError("invalid_owner", str(exc)) from exc
+        except FileNotFoundError as exc:
+            raise ToolError("invalid_workspace", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_owned", str(exc)) from exc
+
+    def recover_workspace_lease(
+        self, workspace_dir: str, owner_id: str, reason: str
+    ) -> WorkspaceLease | None:
+        try:
+            return self.store.recover_workspace_lease_for_workspace(
+                workspace_dir, owner_id, reason
+            )
+        except ValueError as exc:
+            code = "invalid_owner" if "owner_id" in str(exc) else "invalid_recovery"
+            raise ToolError(code, str(exc)) from exc
+        except FileNotFoundError as exc:
+            raise ToolError("invalid_workspace", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_recovery_blocked", str(exc)) from exc
+
     def submit_plan(self, project_id: str, task_list: TaskList) -> Envelope:
         state = self._require_state(project_id)
         if not isinstance(state, MissionPlanning):
@@ -184,10 +273,33 @@ def decide_attention(
     def inspect_project(self, project_id: str) -> Envelope:
         return self._build_envelope(project_id, dag_mode="full")
 
-    def abort_project(self, project_id: str, reason: str) -> Envelope:
+    def abort_project(
+        self, project_id: str, reason: str, owner_id: str
+    ) -> Envelope:
+        try:
+            self.store.claim_workspace_lease(project_id, owner_id)
+        except ValueError as exc:
+            raise ToolError("invalid_owner", str(exc)) from exc
+        except FileNotFoundError as exc:
+            raise ToolError("invalid_workspace", str(exc)) from exc
+        except WorkspaceLeaseConflict as exc:
+            raise ToolError("workspace_owned", str(exc)) from exc
         record = self.store.load_project(project_id)
         state = self.store.load_state(project_id)
         mid = self._current_mission_id(record, state)
+        if mid is not None:
+            task_state = self.store.load_task_state(project_id, mid)
+            running = sorted(
+                task_id
+                for task_id, entry in task_state.tasks.items()
+                if entry.status == "running"
+            )
+            if running:
+                raise ToolError(
+                    "workspace_busy",
+                    "cannot abort and release ownership while tasks are running: "
+                    + ", ".join(running),
+                )
         if mid:
             try:
                 self.store.seal_mission(
@@ -197,7 +309,49 @@ def abort_project(self, project_id: str, reason: str) -> Envelope:
                 pass
         self.store.clear_attention(project_id)
         self.store.save_state(project_id, Aborted(reason=reason))
-        return self._build_envelope(project_id, dag_mode="none")
+        envelope = self._build_envelope(project_id, dag_mode="none")
+        try:
+            self.store.release_workspace_lease(project_id, owner_id)
+        except ValueError as exc:
+            self.store.append_workspace_lease_audit(
+                {
+                    "action": "abort_state_committed_lease_release_failed",
+                    "project_id": project_id,
+                    "owner_id": owner_id,
+                    "error": str(exc),
+                }
+            )
+            raise ToolError(
+                "abort_succeeded_lease_release_failed",
+                f"project state is aborted but lease release failed: {exc}",
+            ) from exc
+        except FileNotFoundError as exc:
+            self.store.append_workspace_lease_audit(
+                {
+                    "action": "abort_state_committed_lease_release_failed",
+                    "project_id": project_id,
+                    "owner_id": owner_id,
+                    "error": str(exc),
+                }
+            )
+            raise ToolError(
+                "abort_succeeded_lease_release_failed",
+                f"project state is aborted but lease release failed: {exc}",
+            ) from exc
+        except WorkspaceLeaseConflict as exc:
+            self.store.append_workspace_lease_audit(
+                {
+                    "action": "abort_state_committed_lease_release_failed",
+                    "project_id": project_id,
+                    "owner_id": owner_id,
+                    "error": str(exc),
+                }
+            )
+            raise ToolError(
+                "abort_succeeded_lease_release_failed",
+                f"project state is aborted but lease release failed: {exc}",
+            ) from exc
+        return envelope
 
     # ------------------------------------------------------------------
     # Decision pipeline