#!/usr/bin/env python3 """Isolated Jujutsu workspace and managed tmux lifecycle primitives. This module deliberately has no dependency on coord's eventual registry/store implementation. Callers supply narrow append-only callbacks: registry(event: str, record: dict) -> None integration_queue(action: str, record: dict) -> None Independent, colocated ``jj git clone`` instances are the safe default. Native ``jj workspace add`` instances are available only when explicitly requested; their metadata states that they share an operation log and object store with the source repository and therefore are not a session-isolation boundary. No lifecycle operation silently deletes work. Removal requires an ownership token, a clean Jujutsu working copy, and (for automatic cleanup) a passed or explicitly waived gate. """ from __future__ import annotations import argparse import copy import datetime as dt import hashlib import json import os from pathlib import Path import re import secrets import shlex import shutil import subprocess import sys from typing import Any, Callable, Iterable, Mapping, MutableMapping, Sequence import uuid DEFAULT_SCRATCH_ROOT = Path("/var/tmp/fleet-audit") SCHEMA = 1 Registry = Callable[[str, dict[str, Any]], None] Queue = Callable[[str, dict[str, Any]], None] Runner = Callable[..., subprocess.CompletedProcess[str]] class WorkspaceError(RuntimeError): """Base class for lifecycle failures.""" class UnsafeRemoval(WorkspaceError): """The requested cleanup could discard or misidentify work.""" def _utc_now() -> str: return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _run( argv: Sequence[str], *, cwd: str | os.PathLike[str] | None = None, runner: Runner = subprocess.run, ) -> subprocess.CompletedProcess[str]: """Run one argv-only command and retain stdout for exact-id parsing.""" return runner( [str(arg) for arg in argv], cwd=None if cwd is None else str(cwd), check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def _emit(registry: Registry | None, event: str, record: Mapping[str, Any]) -> None: if registry is not None: registry(event, copy.deepcopy(dict(record))) def slugify(value: str, *, limit: int = 22) -> str: """Return a short tmux/path-safe task slug.""" slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") slug = slug[:limit].rstrip("-") return slug or "task" def _jj_workspace_identity(task: str, instance_id: str) -> dict[str, str]: """Return unique, presentation-safe jj metadata for one managed agent.""" task_slug = slugify(task) return { "workspace_name": f"agent-{instance_id}", "bookmark": f"agent/{task_slug}/{instance_id}", "description": f"coord work {instance_id}: {task.strip()}", } def task_window_name( task: str, session_id: str, existing: Iterable[str] = (), ) -> str: """Allocate ``-`` without ever targeting an existing window. The four-character suffix is derived from the full session id, not from its display prefix. If the first digest chunk collides, subsequent chunks are tried while retaining the human-facing shape. """ used = set(existing) digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest() slug = slugify(task) for offset in range(0, len(digest) - 3, 4): candidate = f"{slug}-{digest[offset:offset + 4]}" if candidate not in used: return candidate raise WorkspaceError("could not allocate a unique four-character tmux window suffix") def _safe_destination(root: Path, destination: Path) -> Path: root = root.expanduser().resolve() destination = destination.expanduser() if not destination.is_absolute(): destination = root / destination destination = destination.resolve() try: destination.relative_to(root) except ValueError as exc: raise WorkspaceError(f"workspace destination must remain under scratch root {root}") from exc if destination == root: raise WorkspaceError("scratch root itself cannot be a workspace destination") return destination def _jj_ids( repo: Path, revision: str, *, jj: str, runner: Runner, ) -> dict[str, str]: """Read full change and commit ids without snapshotting a working copy.""" result = _run( [ jj, "--ignore-working-copy", "-R", str(repo), "log", "--no-graph", "-r", revision, "-T", 'change_id.normal_hex() ++ "\\n" ++ commit_id ++ "\\n"', ], runner=runner, ) values = [line.strip() for line in result.stdout.splitlines() if line.strip()] if len(values) != 2 or not all(re.fullmatch(r"[0-9a-fA-F]+", value) for value in values): raise WorkspaceError(f"jj returned malformed exact ids for {revision!r}: {values!r}") return {"change_id": values[0].lower(), "commit_id": values[1].lower()} def _is_ancestor( repo: Path, ancestor_commit: str, descendant_commit: str, *, jj: str, runner: Runner, ) -> bool: result = _run( [ jj, "--ignore-working-copy", "-R", str(repo), "log", "--no-graph", "-r", f"{ancestor_commit} & ::{descendant_commit}", "-T", 'commit_id ++ "\\n"', ], runner=runner, ) observed = [line.strip().lower() for line in result.stdout.splitlines() if line.strip()] return observed == [ancestor_commit.lower()] def _owner_marker(path: Path) -> Path: return path / ".jj" / "fleet-coord-owner.json" def _write_owner_marker(path: Path, *, instance_id: str, token: str) -> None: marker = _owner_marker(path) marker.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps( {"schema": SCHEMA, "instance_id": instance_id, "ownership_token": token}, separators=(",", ":"), sort_keys=True, ) fd = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: os.write(fd, (payload + "\n").encode("utf-8")) finally: os.close(fd) def _replace_owner_marker(path: Path, *, instance_id: str, token: str) -> None: """Atomically rotate an ownership marker during explicit orphan adoption.""" marker = _owner_marker(path) marker.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps( {"schema": SCHEMA, "instance_id": instance_id, "ownership_token": token}, separators=(",", ":"), sort_keys=True, ) temporary = marker.with_name(f"{marker.name}.new-{uuid.uuid4().hex}") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: os.write(fd, (payload + "\n").encode("utf-8")) os.fsync(fd) finally: os.close(fd) os.replace(temporary, marker) def _base_record( *, instance_id: str, task: str, session_id: str, runtime_id: str, conversation_id: str, source: Path, scratch_root: Path, planned_path: Path, dependencies: Iterable[str], mode: str, ownership_token: str, ) -> dict[str, Any]: now = _utc_now() limitations: list[str] = [] isolation = "independent-clone" if mode == "native-workspace": isolation = "shared-jj-repository" limitations = [ "shares the source repository's operation log and object store", "is not an isolation boundary for Jujutsu history operations", "must not be used when an independent clone is required", ] return { "schema": SCHEMA, "instance_id": instance_id, "task": task, "session_id": session_id, "runtime_id": runtime_id, "conversation_id": conversation_id, "mode": mode, "isolation": isolation, "limitations": limitations, "source": str(source), "scratch_root": str(scratch_root), "planned_path": str(planned_path), "actual_path": "", "dependencies": list(dependencies), "planned_resources": [], "actual_resources": [], "base": {}, "change": {}, "lifecycle_state": "creating", "gate_state": "pending", "integration_state": "not-queued", "ownership_token": ownership_token, "tmux": {}, "jj_workspace_name": "", "jj_bookmark": "", "jj_change_description": "", "created_at": now, "updated_at": now, } def create_workspace( source: str | os.PathLike[str], *, task: str, session_id: str, runtime_id: str = "", conversation_id: str = "", scratch_root: str | os.PathLike[str] = DEFAULT_SCRATCH_ROOT, destination: str | os.PathLike[str] | None = None, base_revision: str = "master@origin", dependencies: Iterable[str] = (), native_workspace: bool = False, registry: Registry | None = None, jj: str = "jj", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Create and register an independent clone (or explicit native workspace).""" if not task.strip() or not session_id.strip(): raise WorkspaceError("task and full session_id are required") source_path = Path(source).expanduser().resolve() root = Path(scratch_root).expanduser().resolve() root.mkdir(parents=True, exist_ok=True) instance_id = uuid.uuid4().hex default_name = f"{slugify(task)}-{instance_id[:12]}" dest = _safe_destination(root, Path(destination) if destination else Path(default_name)) if dest.exists(): raise WorkspaceError(f"workspace destination already exists: {dest}") mode = "native-workspace" if native_workspace else "clone" token = secrets.token_hex(32) record = _base_record( instance_id=instance_id, task=task, session_id=session_id, runtime_id=runtime_id, conversation_id=conversation_id, source=source_path, scratch_root=root, planned_path=dest, dependencies=dependencies, mode=mode, ownership_token=token, ) jj_identity = _jj_workspace_identity(task, instance_id) record["jj_workspace_name"] = jj_identity["workspace_name"] record["jj_bookmark"] = jj_identity["bookmark"] record["jj_change_description"] = jj_identity["description"] _emit(registry, "workspace-planned", record) try: if native_workspace: _run( [ jj, "--ignore-working-copy", "-R", str(source_path), "workspace", "add", "--name", jj_identity["workspace_name"], "-r", base_revision, str(dest), ], runner=runner, ) else: _run( [jj, "git", "clone", "--colocate", str(source_path), str(dest)], runner=runner, ) # Clone chooses its default branch as @'s parent. Record the # requested exact base and start a fresh change there when a # non-default base was explicitly selected. if base_revision not in ("", "master"): _run([jj, "-R", str(dest), "new", base_revision], runner=runner) _run( [jj, "-R", str(dest), "workspace", "rename", jj_identity["workspace_name"]], runner=runner, ) # A named workspace, described working-copy change, and namespaced # bookmark make concurrent agent heads legible without granting any # authority to their display names. Full IDs remain canonical evidence. _run( [jj, "-R", str(dest), "describe", "-m", jj_identity["description"]], runner=runner, ) _run( [jj, "-R", str(dest), "bookmark", "create", jj_identity["bookmark"], "-r", "@"], runner=runner, ) _write_owner_marker(dest, instance_id=instance_id, token=token) record["actual_path"] = str(dest.resolve()) record["base"] = _jj_ids(dest, base_revision, jj=jj, runner=runner) record["change"] = _jj_ids(dest, "@", jj=jj, runner=runner) record["initial_change"] = dict(record["change"]) record["lifecycle_state"] = "active" record["updated_at"] = _utc_now() _emit(registry, "workspace-created", record) return record except Exception as exc: # Creation succeeded but metadata did not. Preserve the directory: # an incomplete workspace is a recovery candidate, never trash. record["actual_path"] = str(dest) record["lifecycle_state"] = "orphan-candidate" record["updated_at"] = _utc_now() _emit(registry, "workspace-create-incomplete", record) if isinstance(exc, subprocess.CalledProcessError): detail = (exc.stderr or exc.stdout or "").strip() command = shlex.join(str(part) for part in exc.cmd) raise WorkspaceError( f"workspace creation failed: {command}: " f"{detail or f'exit {exc.returncode}'}; " f"preserved recovery candidate {dest}" ) from exc raise def adopt_workspace( path: str | os.PathLike[str], *, task: str, session_id: str, runtime_id: str = "", conversation_id: str = "", source: str | os.PathLike[str], scratch_root: str | os.PathLike[str] = DEFAULT_SCRATCH_ROOT, base_revision: str = "master@origin", dependencies: Iterable[str] = (), prior_ownership_token: str = "", recovery_override: bool = False, registry: Registry | None = None, jj: str = "jj", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Adopt an existing independent colocated clone after explicit validation. This is the recovery path after coordinator-state loss or for an older independently-created clone. It never discovers or adopts by directory name alone, and it does not accept native/shared Jujutsu workspaces. """ if not task.strip() or not session_id.strip(): raise WorkspaceError("task and full session_id are required") root = Path(scratch_root).expanduser().resolve() candidate = _safe_destination(root, Path(path)) source_path = Path(source).expanduser().resolve() if not candidate.is_dir(): raise WorkspaceError(f"adoption candidate is not a directory: {candidate}") if not (candidate / ".jj").is_dir() or not (candidate / ".git").is_dir(): raise WorkspaceError( "adoption requires an independent colocated clone with both .jj/ and .git/" ) if candidate == source_path: raise WorkspaceError("the canonical source checkout cannot be adopted as scratch work") marker_path = _owner_marker(candidate) if marker_path.exists(): try: prior_marker = json.loads(marker_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise WorkspaceError( "existing ownership marker is unreadable; explicit recovery is required" ) from exc prior_token = str(prior_marker.get("ownership_token") or "") if not recovery_override and ( not prior_ownership_token or not secrets.compare_digest(prior_ownership_token, prior_token) ): raise WorkspaceError( "candidate already has an ownership marker; supply its prior ownership " "token or use the explicit recovery override after checking live records" ) # Exact-id reads prove this is a usable jj repository before any marker is # changed. A dirty clone remains adoptable because adoption preserves it. base = _jj_ids(candidate, base_revision, jj=jj, runner=runner) change = _jj_ids(candidate, "@", jj=jj, runner=runner) dirty, dirty_summary = _snapshot_dirty(candidate, jj=jj, runner=runner) instance_id = uuid.uuid4().hex token = secrets.token_hex(32) record = _base_record( instance_id=instance_id, task=task, session_id=session_id, runtime_id=runtime_id, conversation_id=conversation_id, source=source_path, scratch_root=root, planned_path=candidate, dependencies=dependencies, mode="clone", ownership_token=token, ) record.update( { "actual_path": str(candidate), "base": base, "change": change, "initial_change": dict(change), "lifecycle_state": "active", "adopted": True, "adopted_at": _utc_now(), "dirty_at_adoption": dirty, "dirty_summary_at_adoption": dirty_summary, } ) _emit(registry, "workspace-adoption-planned", record) _replace_owner_marker(candidate, instance_id=instance_id, token=token) record["updated_at"] = _utc_now() _emit(registry, "workspace-adopted", record) return record def _snapshot_dirty( path: Path, *, jj: str, runner: Runner, ) -> tuple[bool, str]: """Snapshot only this isolated workspace and return its current diff.""" result = _run([jj, "-R", str(path), "diff", "--summary"], runner=runner) summary = result.stdout return bool(summary.strip()), summary def inspect_workspace( record: Mapping[str, Any], *, registry: Registry | None = None, jj: str = "jj", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Inspect a candidate without mutating any repository except its own snapshot.""" updated = copy.deepcopy(dict(record)) path = Path(str(updated.get("actual_path") or updated.get("planned_path"))) updated["exists"] = path.is_dir() updated["dirty"] = None updated["dirty_summary"] = "" updated["ownership_verified"] = False if path.is_dir(): try: marker = json.loads(_owner_marker(path).read_text(encoding="utf-8")) updated["ownership_verified"] = ( marker.get("instance_id") == updated.get("instance_id") and secrets.compare_digest( str(marker.get("ownership_token", "")), str(updated.get("ownership_token", "")), ) ) except (FileNotFoundError, json.JSONDecodeError, OSError): updated["ownership_verified"] = False dirty, summary = _snapshot_dirty(path, jj=jj, runner=runner) updated["dirty"] = dirty updated["dirty_summary"] = summary initial_change = dict(updated.get("initial_change") or updated.get("change") or {}) current_change = _jj_ids(path, "@", jj=jj, runner=runner) updated["change"] = current_change updated["advanced"] = bool( initial_change and ( initial_change.get("change_id") != current_change.get("change_id") or initial_change.get("commit_id") != current_change.get("commit_id") ) ) updated["inspected_at"] = _utc_now() updated["updated_at"] = updated["inspected_at"] _emit(registry, "workspace-inspected", updated) return updated def list_workspaces( records: Iterable[Mapping[str, Any]], *, include_removed: bool = False, ) -> list[dict[str, Any]]: """Return registry records in stable lifecycle/task/path order. Discovery belongs to the registry fold, not to an unsafe filesystem scan: an unregistered directory is an orphan candidate and must never become an automatic deletion target merely because its name resembles ours. """ selected = [ copy.deepcopy(dict(record)) for record in records if include_removed or record.get("lifecycle_state") != "removed" ] return sorted( selected, key=lambda record: ( str(record.get("lifecycle_state", "")), str(record.get("task", "")), str(record.get("actual_path") or record.get("planned_path") or ""), ), ) def set_gate_state( record: Mapping[str, Any], state: str, *, registry: Registry | None = None, ) -> dict[str, Any]: if state not in {"pending", "running", "passed", "failed", "waived"}: raise WorkspaceError(f"invalid gate state: {state}") updated = copy.deepcopy(dict(record)) updated["gate_state"] = state updated["updated_at"] = _utc_now() _emit(registry, "workspace-gate", updated) return updated def park_workspace( record: Mapping[str, Any], *, reason: str = "", registry: Registry | None = None, ) -> dict[str, Any]: updated = copy.deepcopy(dict(record)) updated["lifecycle_state"] = "parked" updated["park_reason"] = reason updated["parked_at"] = _utc_now() updated["updated_at"] = updated["parked_at"] _emit(registry, "workspace-parked", updated) return updated def resume_workspace( record: Mapping[str, Any], *, runtime_id: str | None = None, registry: Registry | None = None, ) -> dict[str, Any]: path = Path(str(record.get("actual_path") or "")) if not path.is_dir(): raise WorkspaceError(f"parked workspace is unavailable: {path}") updated = copy.deepcopy(dict(record)) updated["lifecycle_state"] = "active" if runtime_id is not None: updated["runtime_id"] = runtime_id updated["resumed_at"] = _utc_now() updated["updated_at"] = updated["resumed_at"] _emit(registry, "workspace-resumed", updated) return updated def archive_workspace( record: Mapping[str, Any], *, note: str = "", registry: Registry | None = None, ) -> dict[str, Any]: """Mark a workspace archival candidate while preserving all bytes in place.""" updated = copy.deepcopy(dict(record)) updated["lifecycle_state"] = "archived" updated["archive_note"] = note updated["archived_at"] = _utc_now() updated["updated_at"] = updated["archived_at"] _emit(registry, "workspace-archived", updated) return updated def remove_workspace( record: Mapping[str, Any], *, ownership_token: str, automatic: bool = False, allow_ungated: bool = False, registry: Registry | None = None, jj: str = "jj", tmux: str = "tmux", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Remove a verified clean instance; refuse every ambiguous case.""" try: if _live_tmux_identity(record, tmux=tmux, runner=runner): raise UnsafeRemoval( "workspace still has a live, ownership-verified tmux agent; " "stop or park it before removal" ) except WorkspaceError as exc: raise UnsafeRemoval(str(exc)) from exc inspected = inspect_workspace(record, jj=jj, runner=runner) if not inspected.get("exists"): raise UnsafeRemoval("workspace path does not exist; record retained as an orphan candidate") if not ownership_token or not secrets.compare_digest( ownership_token, str(inspected.get("ownership_token", "")) ): raise UnsafeRemoval("ownership token does not match registry record") if not inspected.get("ownership_verified"): raise UnsafeRemoval("on-disk ownership marker does not match registry record") if inspected.get("dirty") is not False: raise UnsafeRemoval("workspace is dirty or its cleanliness is unknown; refusing removal") release_evidence = inspected.get("integration_release_evidence") if inspected.get("advanced") and ( inspected.get("integration_state") != "integrated" or not isinstance(release_evidence, dict) or not release_evidence.get("runtime_id") or not release_evidence.get("resource") or not release_evidence.get("source_commit") or not release_evidence.get("target_commit") ): raise UnsafeRemoval( "workspace contains an advanced change without verified integration release evidence; " "refusing removal" ) if inspected.get("advanced"): current_commit = str((inspected.get("change") or {}).get("commit_id") or "") integrated_commit = str((release_evidence or {}).get("source_commit") or "") if current_commit.lower() != integrated_commit.lower() and not _is_ancestor( repo=Path(str(inspected["actual_path"])), ancestor_commit=integrated_commit, descendant_commit=current_commit, jj=jj, runner=runner, ): raise UnsafeRemoval( "workspace moved away from its integrated candidate; refusing removal" ) gated = inspected.get("gate_state") in {"passed", "waived"} if automatic and not gated: raise UnsafeRemoval("automatic cleanup never removes ungated work") if not automatic and not gated and not allow_ungated: raise UnsafeRemoval("ungated work requires explicit allow_ungated=True") path = Path(str(inspected["actual_path"])).resolve() root = Path(str(inspected["scratch_root"])).resolve() _safe_destination(root, path) before = path.stat(follow_symlinks=False) quarantine = root / ( f".fleet-coord-delete-{inspected['instance_id']}-{uuid.uuid4().hex}" ) os.rename(path, quarantine) try: after = quarantine.stat(follow_symlinks=False) marker = json.loads(_owner_marker(quarantine).read_text(encoding="utf-8")) marker_ok = ( marker.get("instance_id") == inspected.get("instance_id") and secrets.compare_digest( str(marker.get("ownership_token") or ""), str(inspected.get("ownership_token") or ""), ) ) quarantine_dirty, _summary = _snapshot_dirty( quarantine, jj=jj, runner=runner ) quarantine_change = _jj_ids(quarantine, "@", jj=jj, runner=runner) if ( (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) or not marker_ok or quarantine_dirty or quarantine_change != inspected.get("change") ): raise UnsafeRemoval( "workspace identity or contents changed before quarantine; " "refusing deletion" ) if inspected.get("mode") == "native-workspace": workspace_name = str(inspected.get("jj_workspace_name") or "") source = str(inspected.get("source") or "") if not workspace_name or not source: raise UnsafeRemoval( "native workspace metadata is incomplete; refusing removal" ) _run( [ jj, "--ignore-working-copy", "-R", source, "workspace", "forget", workspace_name, ], runner=runner, ) except Exception: if quarantine.exists() and not path.exists(): os.rename(quarantine, path) raise shutil.rmtree(quarantine) inspected["lifecycle_state"] = "removed" inspected["removed_at"] = _utc_now() inspected["updated_at"] = inspected["removed_at"] _emit(registry, "workspace-removed", inspected) return inspected def enqueue_integration( record: Mapping[str, Any], *, queue: Queue, action: str = "enqueue", registry: Registry | None = None, ) -> dict[str, Any]: """Hand one record to the sole integration/push queue or resource lock.""" if action not in {"enqueue", "acquire", "release", "cancel"}: raise WorkspaceError(f"unsupported integration queue action: {action}") updated = copy.deepcopy(dict(record)) current = str(updated.get("integration_state") or "not-queued") allowed_from = { "enqueue": {"not-queued", "cancelled"}, "acquire": {"queued"}, "release": {"integrating"}, "cancel": {"queued"}, } if current not in allowed_from[action]: raise WorkspaceError( f"cannot {action} integration from state {current!r}; " f"expected one of {sorted(allowed_from[action])}" ) queue(action, copy.deepcopy(updated)) updated["integration_state"] = { "enqueue": "queued", "acquire": "integrating", "release": "lock-released", "cancel": "cancelled", }[action] updated["updated_at"] = _utc_now() _emit(registry, "workspace-integration", updated) return updated def freeze_integration_candidate( record: Mapping[str, Any], *, registry: Registry | None = None, jj: str = "jj", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Freeze contentful @, or @- when jj has created a clean empty child.""" dirty = record.get("dirty") if dirty is True: change = dict(record.get("change") or {}) elif dirty is False: path = Path(str(record.get("actual_path") or "")).resolve() change = _jj_ids(path, "@-", jj=jj, runner=runner) else: raise WorkspaceError("integration acquisition requires inspected cleanliness") if not change.get("change_id") or not change.get("commit_id"): raise WorkspaceError("integration acquisition requires an inspected exact change") updated = copy.deepcopy(dict(record)) updated["integration_candidate"] = { "change_id": str(change["change_id"]).lower(), "commit_id": str(change["commit_id"]).lower(), "captured_at": _utc_now(), } updated["updated_at"] = updated["integration_candidate"]["captured_at"] _emit(registry, "workspace-integration-candidate", updated) return updated def prove_integrated( record: Mapping[str, Any], *, target_revision: str, registry: Registry | None = None, jj: str = "jj", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Prove the exact workspace commit is an ancestor of the target revision.""" if record.get("integration_state") != "lock-released": raise WorkspaceError("integration proof requires a released integration lock") if record.get("gate_state") not in {"passed", "waived"}: raise WorkspaceError("integration proof requires a passed or explicitly waived gate") owner = str(record.get("integration_owner_runtime") or "") resource = str(record.get("integration_resource") or "") release = dict(record.get("integration_lock_release_evidence") or {}) if not owner or not resource or release.get("runtime_id") != owner or release.get( "resource" ) != resource: raise WorkspaceError( "integration proof requires exact owner/resource lock-release evidence" ) path = Path(str(record.get("actual_path") or "")).resolve() candidate = dict(record.get("integration_candidate") or {}) source_commit = str(candidate.get("commit_id", "")) if not source_commit: raise WorkspaceError("workspace record has no frozen integration candidate") target = _jj_ids(path, target_revision, jj=jj, runner=runner) if not _is_ancestor( path, source_commit, target["commit_id"], jj=jj, runner=runner, ): raise WorkspaceError( f"source commit {source_commit} is not an ancestor of target " f"{target['commit_id']}" ) updated = copy.deepcopy(dict(record)) updated["integration_state"] = "integrated" updated["integration_release_evidence"] = { "runtime_id": record.get("integration_owner_runtime"), "resource": record.get("integration_resource"), "source_commit": source_commit.lower(), "target_commit": target["commit_id"], "target_revision": target_revision, } updated["updated_at"] = _utc_now() _emit(registry, "workspace-integrated", updated) return updated def _tmux( args: Sequence[str], *, tmux: str, runner: Runner, ) -> subprocess.CompletedProcess[str]: return _run([tmux, *args], runner=runner) def _managed_session( project: Path, *, tmux: str, runner: Runner, ) -> str | None: try: sessions = _tmux(["list-sessions", "-F", "#{session_id}"], tmux=tmux, runner=runner) except subprocess.CalledProcessError: return None for session_id in sessions.stdout.splitlines(): session_id = session_id.strip() if not session_id.startswith("$"): continue try: value = _tmux( ["display-message", "-p", "-t", session_id, "#{@fleet_ai_project}"], tmux=tmux, runner=runner, ).stdout.strip() except subprocess.CalledProcessError: continue if value == str(project): return session_id return None def _live_tmux_identity( record: Mapping[str, Any], *, tmux: str, runner: Runner, ) -> bool: """Verify recorded immutable IDs still belong to this managed workspace.""" metadata = dict(record.get("tmux") or {}) session_id = str(metadata.get("session_id", "")) window_id = str(metadata.get("window_id", "")) pane_id = str(metadata.get("pane_id", "")) if not ( session_id.startswith("$") and window_id.startswith("@") and pane_id.startswith("%") ): return False expected = { session_id: ("#{@fleet_ai_project}", str(Path(str(record["actual_path"])).resolve())), window_id: ("#{@fleet_ai_instance}", str(record.get("instance_id", ""))), pane_id: ("#{@fleet_ai_instance}", str(record.get("instance_id", ""))), } for target, (template, wanted) in expected.items(): try: actual = _tmux( ["display-message", "-p", "-t", target, template], tmux=tmux, runner=runner, ).stdout.strip() except subprocess.CalledProcessError: return False if actual != wanted: raise WorkspaceError( f"tmux id {target} exists but ownership metadata mismatches; " "refusing name/id reuse" ) return True def _agent_command( harness: str, *, conversation_id: str, direct_binary: str | None, ) -> list[str]: if harness not in {"codex", "claude"}: raise WorkspaceError("harness must be 'codex' or 'claude'") binary = direct_binary or f"{harness}-direct" if not conversation_id: return [binary] if harness == "codex": return [binary, "resume", conversation_id] return [binary, "--resume", conversation_id] def start_agent( record: Mapping[str, Any], *, harness: str, conversation_id: str | None = None, direct_binary: str | None = None, registry: Registry | None = None, tmux: str = "tmux", runner: Runner = subprocess.run, ) -> dict[str, Any]: """Start/resume an agent in a managed project session. Existing tmux objects are discovered by user options and then addressed only by immutable ``$session``, ``@window`` and ``%pane`` ids. Names are presentation labels, never lookup targets. """ updated = copy.deepcopy(dict(record)) project = Path(str(updated.get("actual_path") or "")).resolve() if not project.is_dir(): raise WorkspaceError(f"workspace does not exist: {project}") if _live_tmux_identity(updated, tmux=tmux, runner=runner): _emit(registry, "workspace-agent-reused", updated) return updated conv = conversation_id if conversation_id is not None else str(updated.get("conversation_id", "")) command = _agent_command(harness, conversation_id=conv, direct_binary=direct_binary) attachment_runtime = str(uuid.uuid4()) updated["runtime_id"] = attachment_runtime launch_command = [ "env", f"COORD_INSTANCE_ID={attachment_runtime}", f"COORD_WORK_ID={updated['instance_id']}", f"COORD_WORKSPACE_PATH={project}", f"COORD_WORKSPACE_MODE={str(updated.get('mode') or '')}", f"COORD_JJ_WORKSPACE={str(updated.get('jj_workspace_name') or '')}", f"COORD_JJ_BOOKMARK={str(updated.get('jj_bookmark') or '')}", f"FLEET_AI_TASK={str(updated.get('task') or 'task')}", *command, ] session_id = _managed_session(project, tmux=tmux, runner=runner) existing_names: list[str] = [] if session_id: windows = _tmux( ["list-windows", "-t", session_id, "-F", "#{window_name}"], tmux=tmux, runner=runner, ) existing_names = [line for line in windows.stdout.splitlines() if line] window_name = task_window_name( str(updated.get("task") or "task"), str(updated.get("session_id") or updated.get("instance_id")), existing_names, ) agent_shell = shlex.join(launch_command) heartbeat_shell = shlex.join( [ str(project / ".claude" / "coord" / "coord"), "--runtime", attachment_runtime, "heartbeat", ] ) shell_command = ( f"( sleep 60; while {heartbeat_shell} >/dev/null 2>&1; " "do sleep 60; done ) /dev/null || true; " "wait \"$coord_hb\" 2>/dev/null || true' EXIT; " f"{agent_shell}; rc=$?; " "kill \"$coord_hb\" 2>/dev/null || true; " "wait \"$coord_hb\" 2>/dev/null || true; " "trap - EXIT; exit \"$rc\"" ) if session_id is None: presentation_name = "ai-" + hashlib.sha256(str(project).encode()).hexdigest()[:12] created = _tmux( [ "new-session", "-d", "-P", "-F", "#{session_id}\t#{window_id}\t#{pane_id}", "-s", presentation_name, "-n", window_name, "-c", str(project), shell_command, ], tmux=tmux, runner=runner, ) else: created = _tmux( [ "new-window", "-d", "-P", "-F", "#{session_id}\t#{window_id}\t#{pane_id}", "-t", session_id, "-n", window_name, "-c", str(project), shell_command, ], tmux=tmux, runner=runner, ) fields = created.stdout.strip().split("\t") if len(fields) != 3 or not ( fields[0].startswith("$") and fields[1].startswith("@") and fields[2].startswith("%") ): raise WorkspaceError(f"tmux did not return immutable object ids: {created.stdout!r}") session_id, window_id, pane_id = fields # Every mutation below targets immutable ids captured from tmux itself. _tmux(["set-option", "-q", "-t", session_id, "@fleet_ai_managed", "1"], tmux=tmux, runner=runner) _tmux( ["set-option", "-q", "-t", session_id, "@fleet_ai_project", str(project)], tmux=tmux, runner=runner, ) _tmux(["set-option", "-qw", "-t", window_id, "automatic-rename", "off"], tmux=tmux, runner=runner) options = { "@fleet_ai_instance": str(updated.get("instance_id", "")), "@fleet_ai_task": str(updated.get("task", "")), "@fleet_ai_workspace": str(project), "@fleet_ai_runtime": str(updated.get("runtime_id", "")), "@fleet_ai_conversation": conv, } for key, value in options.items(): _tmux(["set-option", "-qw", "-t", window_id, key, value], tmux=tmux, runner=runner) _tmux( ["set-option", "-qp", "-t", pane_id, "@fleet_ai_instance", str(updated.get("instance_id", ""))], tmux=tmux, runner=runner, ) updated["conversation_id"] = conv updated["tmux"] = { "session_id": session_id, "window_id": window_id, "pane_id": pane_id, "window_name": window_name, "command": command, } updated["lifecycle_state"] = "active" updated["updated_at"] = _utc_now() _emit(registry, "workspace-agent-started", updated) return updated def attach_command( record: Mapping[str, Any], *, tmux: str = "tmux", runner: Runner = subprocess.run, ) -> list[str]: """Return an attach argv after revalidating immutable ownership metadata.""" if not _live_tmux_identity(record, tmux=tmux, runner=runner): raise WorkspaceError("recorded tmux agent is not live") session_id = str((record.get("tmux") or {}).get("session_id", "")) if not session_id.startswith("$"): raise WorkspaceError("record has no immutable tmux session id") return ["tmux", "attach-session", "-t", session_id] def close_agent( record: Mapping[str, Any], *, registry: Registry | None = None, tmux: str = "tmux", runner: Runner = subprocess.run, environ: Mapping[str, str] = os.environ, ) -> dict[str, Any]: """Close one verified window without ejecting its attached client.""" if not _live_tmux_identity(record, tmux=tmux, runner=runner): raise WorkspaceError("recorded tmux agent is not live") updated = copy.deepcopy(dict(record)) metadata = dict(updated.get("tmux") or {}) target_session = str(metadata["session_id"]) target_window = str(metadata["window_id"]) pane = environ.get("TMUX_PANE", "") if pane: current_window = _tmux( ["display-message", "-p", "-t", pane, "#{window_id}"], tmux=tmux, runner=runner, ).stdout.strip() if current_window == target_window: raise WorkspaceError( "refusing to close the caller's current managed window because " "tmux could terminate the registry update; run work close from " "the landing page or another window/session" ) current_session = _tmux( ["display-message", "-p", "-t", pane, "#{session_id}"], tmux=tmux, runner=runner, ).stdout.strip() if current_session == target_session: windows = _tmux( ["list-windows", "-t", target_session, "-F", "#{window_id}"], tmux=tmux, runner=runner, ).stdout.splitlines() if len([item for item in windows if item.strip()]) <= 1: sessions = _tmux( [ "list-sessions", "-F", "#{session_id}\t#{@fleet_ai_managed}", ], tmux=tmux, runner=runner, ).stdout.splitlines() landing = next( ( line.split("\t", 1)[0] for line in sessions if "\t" in line and line.split("\t", 1)[0] != target_session and line.split("\t", 1)[1] == "1" ), "", ) if not landing: raise WorkspaceError( "this is the current session's last window and no other " "managed session exists; create or attach another managed " "workspace before closing it" ) _tmux(["switch-client", "-t", landing], tmux=tmux, runner=runner) _tmux(["kill-window", "-t", target_window], tmux=tmux, runner=runner) updated["tmux"] = {} updated["agent_closed_at"] = _utc_now() updated["updated_at"] = updated["agent_closed_at"] _emit(registry, "workspace-agent-closed", updated) return updated def _json_record(path: str) -> dict[str, Any]: return json.loads(Path(path).read_text(encoding="utf-8")) def main(argv: Sequence[str] | None = None) -> int: """Small CLI for operators; registry integration remains callback-driven.""" parser = argparse.ArgumentParser(prog="coord-workspace") sub = parser.add_subparsers(dest="command", required=True) create = sub.add_parser("create") create.add_argument("source") create.add_argument("--task", required=True) create.add_argument("--sid", required=True) create.add_argument("--runtime-id", default="") create.add_argument("--conversation-id", default="") create.add_argument("--scratch-root", default=str(DEFAULT_SCRATCH_ROOT)) create.add_argument("--destination") create.add_argument("--base", default="master") create.add_argument("--native-workspace", action="store_true") inspect = sub.add_parser("inspect") inspect.add_argument("record") park = sub.add_parser("park") park.add_argument("record") park.add_argument("--reason", default="") resume = sub.add_parser("resume") resume.add_argument("record") archive = sub.add_parser("archive") archive.add_argument("record") archive.add_argument("--note", default="") args = parser.parse_args(argv) if args.command == "create": out = create_workspace( args.source, task=args.task, session_id=args.sid, runtime_id=args.runtime_id, conversation_id=args.conversation_id, scratch_root=args.scratch_root, destination=args.destination, base_revision=args.base, native_workspace=args.native_workspace, ) else: record = _json_record(args.record) if args.command == "inspect": out = inspect_workspace(record) elif args.command == "park": out = park_workspace(record, reason=args.reason) elif args.command == "resume": out = resume_workspace(record) else: out = archive_workspace(record, note=args.note) json.dump(out, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except WorkspaceError as exc: print(f"coord-workspace: {exc}", file=sys.stderr) raise SystemExit(2)