Pi runtime compatibility baseline

.fo-dispatch/pi-ci-git-identity-validation.txt

@@ -0,0 +1,9 @@
+Validation of commit c499809a (test: bootstrap git identity for temp repos): PASSED
+
+Evidence:
+- Reviewed create_test_project(): it now sets repo-local git config user.name=Spacedock Test and user.email=spacedock-test@example.invalid before the initial empty commit.
+- Reviewed regression test: tests/test_test_lib_helpers.py::test_create_test_project_bootstraps_repo_local_git_identity isolates HOME and XDG_CONFIG_HOME, unsets GIT_AUTHOR_* and GIT_COMMITTER_* variables, calls create_test_project(), verifies local config, and verifies HEAD exists.
+- Focused validation passed: unset CLAUDECODE && uv run pytest tests/test_test_lib_helpers.py -k create_test_project_bootstraps_repo_local_git_identity -v -> 1 passed, 19 deselected.
+- Static validation passed: unset CLAUDECODE && uv run pytest tests/ --ignore=tests/fixtures -m "not live_claude and not live_codex and not live_pi" -q -> 541 passed, 31 deselected, 10 subtests passed.
+
+Recommendation: PASSED.

Makefile

@@ -10,7 +10,7 @@ OPUS_MODEL ?= opus
 
 test-static:
 	unset CLAUDECODE && uv run pytest tests/ --ignore=tests/fixtures \
-	  -m "not live_claude and not live_codex" -q
+	  -m "not live_claude and not live_codex and not live_pi" -q
 
 # Single-file live override — pass TEST=tests/<file>.py RUNTIME=claude|codex.
 # Replaces the old test-e2e-commission target: `make test-e2e TEST=tests/test_commission.py`.

pyproject.toml

@@ -11,9 +11,17 @@ dev = [
 ]
 
 [tool.pytest.ini_options]
+testpaths = [
+    "tests",
+]
+norecursedirs = [
+    "tests/fixtures",
+    "plugins",
+]
 markers = [
     "live_claude: spawns a live Claude runtime (pipe or PTY)",
     "live_codex: spawns a live Codex runtime",
+    "live_pi: spawns a live Pi runtime",
     "serial: must run serially (PTY, stubbed git/gh, or explicit sequencing)",
     "teams_mode: requires CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1 (teams dispatch path)",
     "bare_mode: requires CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS unset (bare dispatch path)",

scripts/pi_session_registry.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+"""Thin Spacedock mapping over Pi's built-in session persistence.
+
+Pi session ids / session files are the canonical worker handle. This module does
+not try to replace Pi's own session system or emulate Claude-style team state.
+It only records the workflow-specific association between an FO-owned worker
+label and the Pi session that currently backs it, plus the minimum extra state
+Spacedock needs for reuse/shutdown bookkeeping.
+"""
+
+import json
+import os
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+
+@dataclass
+class WorkerSessionRecord:
+    worker_label: str
+    dispatch_agent_id: str
+    worker_key: str
+    session_id: str
+    session_file: str
+    cwd: str
+    entity_slug: str
+    stage_name: str
+    state: str
+    completion_epoch: int
+
+
+class PiSessionRegistry:
+    """Persist a minimal worker-label -> Pi-session mapping for Spacedock."""
+
+    def __init__(self, path: Path):
+        self.path = Path(path)
+
+    def _load(self) -> dict[str, WorkerSessionRecord]:
+        if not self.path.exists():
+            return {}
+        try:
+            data = json.loads(self.path.read_text())
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(f"Pi session registry is corrupt: {self.path}") from exc
+        return {
+            worker_label: WorkerSessionRecord(**record)
+            for worker_label, record in data.items()
+        }
+
+    def _save(self, records: dict[str, WorkerSessionRecord]) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            worker_label: asdict(record)
+            for worker_label, record in records.items()
+        }
+        tmp_path = self.path.with_name(f".{self.path.name}.tmp-{os.getpid()}")
+        tmp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+        tmp_path.replace(self.path)
+
+    def upsert(self, record: WorkerSessionRecord) -> WorkerSessionRecord:
+        records = self._load()
+        records[record.worker_label] = record
+        self._save(records)
+        return record
+
+    def get(self, worker_label: str) -> WorkerSessionRecord | None:
+        return self._load().get(worker_label)
+
+    def mark_active_again(self, worker_label: str) -> WorkerSessionRecord:
+        records = self._load()
+        record = records[worker_label]
+        if record.state == "active":
+            raise RuntimeError(f"Pi worker {worker_label} is already active")
+        if record.state == "shutdown":
+            raise RuntimeError(f"Pi worker {worker_label} is shutdown")
+        record.state = "active"
+        record.completion_epoch += 1
+        self._save(records)
+        return record
+
+    def mark_shutdown(self, worker_label: str) -> WorkerSessionRecord:
+        records = self._load()
+        record = records[worker_label]
+        record.state = "shutdown"
+        self._save(records)
+        return record
+
+    def routable(self, worker_label: str) -> bool:
+        record = self.get(worker_label)
+        return bool(record is not None and record.state == "completed")