coord/workspace.py

#!/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 ``<slug>-<sid4>`` 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 & coord_hb=$!; "
        "trap 'kill \"$coord_hb\" 2>/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)