coord/coord

#!/usr/bin/env python3
"""Transactional cross-harness coordination CLI.

Runtime state lives outside the repository in a private SQLite database. The
repository contains only this implementation and a stable project identifier.
Conversation IDs survive resume; runtime IDs identify one live attachment and
are the only identities allowed to mutate coordination state.
"""

from __future__ import annotations

import argparse
import contextlib
import importlib.util
import io
import json
import os
import pathlib
import re
import shlex
import socket
import subprocess
import sys
import time
from typing import Any

HERE = pathlib.Path(__file__).resolve().parent
ROOT = HERE.parent if HERE.name == ".coord" else HERE
SPEC = importlib.util.spec_from_file_location("coord_store", HERE / "store.py")
storelib = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = storelib
SPEC.loader.exec_module(storelib)
WORK_SPEC = importlib.util.spec_from_file_location("coord_workspace", HERE / "workspace.py")
worklib = importlib.util.module_from_spec(WORK_SPEC)
assert WORK_SPEC.loader is not None
sys.modules[WORK_SPEC.name] = worklib
WORK_SPEC.loader.exec_module(worklib)


def project_id() -> str:
    override = os.environ.get("COORD_PROJECT_ID")
    value = override
    if not value:
        marker = HERE / "project-id"
        value = marker.read_text(encoding="utf-8").strip()
    if (
        not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", value)
        or ".." in value
    ):
        raise ValueError("COORD_PROJECT_ID/project-id must be one safe filename slug")
    return value


def db_path() -> pathlib.Path:
    override = os.environ.get("COORD_DB")
    if override:
        return pathlib.Path(override)
    state = pathlib.Path(
        os.environ.get("XDG_STATE_HOME", pathlib.Path.home() / ".local" / "state")
    )
    return state / "fleet-agent-coord" / f"{project_id()}.sqlite3"


def store() -> Any:
    return storelib.CoordStore(db_path())


def identity(args: argparse.Namespace) -> str:
    supplied = (
        getattr(args, "identity_arg", "")
        or getattr(args, "runtime", "")
        or os.environ.get("COORD_INSTANCE_ID", "")
        or getattr(args, "sid", "")
        or os.environ.get("CODEX_THREAD_ID", "")
        or os.environ.get("CLAUDE_SESSION_ID", "")
    )
    if not supplied:
        raise ValueError(
            "no identity: use a managed shell or pass -i HANDLE; "
            "full runtime/conversation IDs remain accepted for recovery"
        )
    return supplied


def runtime(args: argparse.Namespace, db: Any | None = None) -> str:
    return (db or store()).resolve_runtime(identity(args))


def runtime_handle(runtime_id: str) -> str:
    return storelib.friendly_handle(runtime_id, "runtime")


def work_handle(instance_id: str) -> str:
    return storelib.friendly_handle(instance_id, "work")


def peer_label(item: dict[str, Any]) -> str:
    label = str(item.get("label") or "")
    if label.startswith("subagent:"):
        return "subagent"
    return label or pathlib.Path(str(item.get("cwd") or "")).name or "?"


def command_resources(
    args: argparse.Namespace, *, required: bool = True
) -> list[str] | None:
    positional = list(getattr(args, "resources", ()) or ())
    legacy = list(getattr(args, "legacy_resources", ()) or ())
    if positional and legacy:
        raise ValueError("supply resources positionally, not twice")
    resources = positional or legacy
    if required and not resources:
        raise ValueError("at least one resource is required")
    return resources or None


def command_resource(args: argparse.Namespace) -> str:
    positional = getattr(args, "resource", "") or ""
    legacy = getattr(args, "legacy_resource", "") or ""
    if positional and legacy:
        raise ValueError("supply the resource positionally, not twice")
    resource = positional or legacy
    if not resource:
        raise ValueError("one resource is required")
    return resource


def command_task(args: argparse.Namespace) -> str:
    positional = getattr(args, "task", "") or ""
    legacy = getattr(args, "legacy_task", "") or ""
    if positional and legacy:
        raise ValueError("supply the task positionally, not twice")
    task = positional or legacy
    if not task:
        raise ValueError("a task name is required")
    return task


def command_body(args: argparse.Namespace) -> str:
    positional = getattr(args, "body", "") or ""
    legacy = getattr(args, "legacy_body", "") or ""
    if positional and legacy:
        raise ValueError("supply the message body positionally, not twice")
    body = positional or legacy
    if not body:
        raise ValueError("a message body is required")
    return body


def public_record(record: dict[str, Any], *, include_secret: bool = False) -> dict[str, Any]:
    result = dict(record)
    result["handle"] = work_handle(str(record["instance_id"]))
    if not include_secret:
        result.pop("ownership_token", None)
    return result


def filtered_work(
    records: list[dict[str, Any]], args: argparse.Namespace, db: Any
) -> list[dict[str, Any]]:
    if getattr(args, "active_only", False):
        records = [
            item
            for item in records
            if item.get("lifecycle_state") not in {"removed", "archived"}
        ]
    states = set(getattr(args, "state", ()) or ())
    if states:
        records = [
            item
            for item in records
            if states
            & {
                str(item.get("lifecycle_state") or ""),
                str(item.get("gate_state") or ""),
                str(item.get("integration_state") or ""),
            }
        ]
    owner = getattr(args, "owner", "") or ""
    if owner:
        owner = db.resolve_runtime(owner)
        records = [
            item for item in records if str(item.get("runtime_id") or "") == owner
        ]
    since = getattr(args, "since", "") or ""
    if since:
        if not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d(?::\d\d)?Z", since):
            raise ValueError(
                "--since must be an exact UTC timestamp like 2026-07-31T02:00Z"
            )
        if len(since) == 17:
            since = since[:-1] + ":00Z"
        records = [
            item for item in records if str(item.get("updated_at") or "") >= since
        ]
    return records


def cmd_hello(args: argparse.Namespace) -> int:
    db = store()
    conversation = (
        args.conversation
        or args.sid
        or os.environ.get("CODEX_THREAD_ID", "")
        or os.environ.get("CLAUDE_SESSION_ID", "")
    )
    if not conversation:
        raise ValueError("hello requires --conversation or a full --sid")
    requested_runtime = args.runtime or os.environ.get("COORD_INSTANCE_ID") or None
    runtime_id = db.register_instance(
        conversation,
        runtime_id=requested_runtime,
        harness=args.harness,
        host=socket.gethostname(),
        cwd=str(pathlib.Path(args.cwd or os.getcwd()).resolve()),
        label=args.label,
        ttl_seconds=args.ttl,
    )
    print(runtime_id)
    return 0


def cmd_heartbeat(args: argparse.Namespace) -> int:
    db = store()
    db.heartbeat(
        runtime(args, db),
        ttl_seconds=args.ttl,
        renew_lease_ttl_seconds=args.lease_ttl,
    )
    return 0


def cmd_bye(args: argparse.Namespace) -> int:
    db = store()
    db.end_instance(runtime(args, db), reason=args.reason)
    return 0


def cmd_claim(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    resources = command_resources(args)
    try:
        db.claim_resources(
            owner,
            resources,
            mode=args.mode,
            purpose=args.purpose,
            ttl_seconds=args.ttl,
        )
    except storelib.ClaimRefused as exc:
        for conflict in exc.conflicts:
            holder = runtime_handle(conflict.runtime_id)
            remaining = max(0, int(conflict.expires_at - time.time()))
            print(
                f"coord: REFUSED {conflict.path}: {conflict.mode} lease by "
                f"{holder} for {conflict.purpose or 'unstated'} "
                f"(expires in {remaining}s). Ask with: "
                f"coord send \"lease request: {conflict.path}\" --to {holder}",
                file=sys.stderr,
            )
        return 1
    print(f"claimed {', '.join(resources)} ({args.mode})")
    return 0


def cmd_release(args: argparse.Namespace) -> int:
    db = store()
    released = db.release_resources(runtime(args, db), command_resources(args))
    print("released " + (", ".join(released) if released else "nothing"))
    return 0


def cmd_renew(args: argparse.Namespace) -> int:
    db = store()
    renewed = db.renew_resources(
        runtime(args, db),
        resources=command_resources(args, required=False),
        ttl_seconds=args.ttl,
    )
    print("renewed " + (", ".join(renewed) if renewed else "nothing"))
    return 0


def cmd_list(args: argparse.Namespace) -> int:
    db = store()
    instances = db.live_instances()
    leases = db.active_leases()
    work = filtered_work(db.workspace_records(), args, db)
    if args.json:
        print(
            json.dumps(
                {
                    "instances": instances,
                    "leases": leases,
                    "work": [
                        public_record(item, include_secret=args.show_token)
                        for item in work
                    ],
                },
                sort_keys=True,
                separators=(",", ":"),
            )
        )
        return 0
    print(f"agents {len(instances)}")
    for item in instances:
        label = peer_label(item)
        line = f"  {runtime_handle(item['runtime_id'])} {item['harness']} {label}"
        if args.verbose:
            line += (
                f" runtime={item['runtime_id']} "
                f"conversation={item['conversation_id']} cwd={item['cwd']}"
            )
        print(line)
    print(f"leases {len(leases)}")
    for lease in leases:
        line = (
            f"  {lease['path']} {lease['mode']} "
            f"{runtime_handle(lease['runtime_id'])} {lease['purpose']}"
        )
        if args.verbose:
            line += f" runtime={lease['runtime_id']}"
        print(line)
    if args.work:
        print(f"work {len(work)}")
        for item in work:
            line = (
                f"  {work_handle(item['instance_id'])} "
                f"{item.get('lifecycle_state', '?')}/{item.get('gate_state', '?')} "
                f"{item.get('task', '?')}"
            )
            if args.verbose:
                line += (
                    f" id={item['instance_id']} "
                    f"path={item.get('actual_path') or item.get('planned_path')}"
                )
            print(line)
    else:
        print(f"work {len(work)} (use `coord work list`)")
    return 0


def cmd_send(args: argparse.Namespace) -> int:
    db = store()
    recipient = args.to
    if recipient != "*":
        recipient = db.resolve_runtime(recipient)
    message_id = db.send_message(
        runtime(args, db),
        recipient=recipient,
        kind=args.kind,
        body=command_body(args),
        refs=args.ref,
    )
    print(f"coord: queued message {message_id}")
    return 0


def cmd_inbox(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    messages = db.inbox(owner, limit=args.limit, mark_delivered=not args.peek)
    for message in messages:
        print(
            f"[{message.message_id}] {runtime_handle(message.sender_runtime_id)} "
            f"({message.kind}): {message.body}"
        )
        for ref in message.refs:
            print(f"    ref: {ref}")
    if not messages:
        print("coord: no pending messages")
    return 0


def cmd_ack(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    ids = args.message_ids
    if not ids:
        ids = [item.message_id for item in db.inbox(owner, mark_delivered=False)]
    acked = db.ack(owner, ids)
    print(f"coord: acknowledged {', '.join(map(str, acked)) or 'nothing'}")
    return 0


def cmd_check(args: argparse.Namespace) -> int:
    """Recognized-write guard. Only an actual conflict returns one."""
    try:
        db = store()
        owner = runtime(args, db)
        db.heartbeat(
            owner,
            ttl_seconds=args.instance_ttl,
            renew_lease_ttl_seconds=args.lease_ttl,
        )
        resource = command_resource(args)
        for lease in db.active_leases():
            if lease["runtime_id"] == owner or lease["mode"] != "exclusive":
                continue
            if db.resources_overlap(resource, lease["path"]):
                print(
                    f"{resource} is leased exclusively by "
                    f"{runtime_handle(lease['runtime_id'])} for "
                    f"{lease['purpose'] or 'unstated'}",
                    file=sys.stderr,
                )
                return 1
    except Exception as exc:  # fail open for the hook boundary
        print(f"coord check failed open: {exc}", file=sys.stderr)
    return 0


def cmd_audit(args: argparse.Namespace) -> int:
    if args.output:
        fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as out:
            store().export_audit_jsonl(
                out, after_id=args.after, include_secrets=args.include_secrets
            )
    else:
        store().export_audit_jsonl(
            sys.stdout,
            after_id=args.after,
            include_secrets=args.include_secrets,
        )
    return 0


BATCH_TOP_LEVEL = frozenset(
    {"claim", "release", "renew", "list", "send", "inbox", "ack", "check", "remote"}
)
BATCH_WORK = frozenset({"list", "gc", "inspect", "park", "resume", "archive", "gate"})
BATCH_IDENTITY_OPTIONS = ("-i", "--identity", "--runtime", "--sid")


def _batch_steps(tokens: list[str]) -> list[list[str]]:
    steps: list[list[str]] = [[]]
    for token in tokens:
        if token == ":":
            if not steps[-1]:
                raise ValueError("batch contains an empty action")
            steps.append([])
        else:
            steps[-1].append(token)
    if not steps[-1]:
        raise ValueError("batch contains an empty action")
    if not steps[0]:
        raise ValueError("batch requires at least one action")
    return steps


def _batch_action(parsed: argparse.Namespace) -> str:
    if parsed.command == "work":
        action = f"work {parsed.work_command}"
        if parsed.work_command not in BATCH_WORK:
            raise ValueError(
                f"{action} is not batchable; run it explicitly so its target "
                "and side effects remain visible"
            )
        return action
    if parsed.command not in BATCH_TOP_LEVEL:
        raise ValueError(
            f"{parsed.command} is not batchable; run it explicitly"
        )
    return str(parsed.command)


def _reject_batch_identity_override(tokens: list[str]) -> None:
    token = tokens[0]
    override = token in BATCH_IDENTITY_OPTIONS or (
        token.startswith("-i") and token != "-i"
    )
    override = override or any(
        token.startswith(f"{option}=")
        for option in BATCH_IDENTITY_OPTIONS
        if option.startswith("--")
    )
    if override:
        raise ValueError(
            "batch actions inherit one exact identity; nested identity "
            "options are forbidden"
        )


def _execute(parsed: argparse.Namespace) -> int:
    try:
        return int(parsed.func(parsed) or 0)
    except (KeyError, ValueError, storelib.ClaimRefused, worklib.WorkspaceError) as exc:
        print(f"coord: {exc}", file=sys.stderr)
        return 2
    except subprocess.CalledProcessError as exc:
        detail = (exc.stderr or exc.stdout or "").strip()
        command = shlex.join(str(part) for part in exc.cmd)
        print(
            f"coord: {command} failed: {detail or f'exit {exc.returncode}'}",
            file=sys.stderr,
        )
        return 2


def cmd_batch(args: argparse.Namespace) -> int:
    """Run explicitly delimited coordinator actions, stopping on failure."""
    owner = runtime(args)
    steps = _batch_steps(args.actions)
    results: list[dict[str, Any]] = []
    final_code = 0
    for index, tokens in enumerate(steps, start=1):
        stdout = io.StringIO()
        stderr = io.StringIO()
        action = "invalid"
        code = 2
        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
            try:
                _reject_batch_identity_override(tokens)
                nested = build_parser().parse_args(["-i", owner, *tokens])
                if runtime(nested) != owner:
                    raise ValueError(
                        "batch action identity does not match its inherited owner"
                    )
                action = _batch_action(nested)
                code = _execute(nested)
            except SystemExit as exc:
                code = int(exc.code or 0)
            except ValueError as exc:
                print(f"coord: {exc}", file=sys.stderr)
        result = {
            "step": index,
            "action": action,
            "code": code,
            "stdout": stdout.getvalue(),
            "stderr": stderr.getvalue(),
        }
        results.append(result)
        if not args.json:
            print(f"{index}/{len(steps)} {action}")
            if result["stdout"]:
                print(result["stdout"], end="")
            if result["stderr"]:
                print(result["stderr"], end="", file=sys.stderr)
        if code:
            final_code = code
            if not args.json:
                print(
                    f"batch stopped at {index}/{len(steps)} {action}",
                    file=sys.stderr,
                )
            break
    if args.json:
        print(
            json.dumps(
                {
                    "ok": final_code == 0,
                    "completed": len(results),
                    "total": len(steps),
                    "steps": results,
                },
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    return final_code


def _command_output(argv: list[str]) -> str:
    try:
        return subprocess.run(
            argv,
            check=True,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ).stdout.strip()
    except subprocess.CalledProcessError as exc:
        detail = (exc.stderr or exc.stdout or "").strip()
        raise ValueError(
            f"{shlex.join(argv)} failed: {detail or f'exit {exc.returncode}'}"
        ) from exc


def _authoritative_repository(db: Any, requested: str) -> pathlib.Path:
    if requested:
        return pathlib.Path(requested).expanduser().resolve()
    cwd = pathlib.Path.cwd().resolve()
    for record in db.workspace_records():
        actual = record.get("actual_path")
        if not actual:
            continue
        try:
            cwd.relative_to(pathlib.Path(str(actual)).resolve())
        except ValueError:
            continue
        return pathlib.Path(str(record["source"])).resolve()
    return ROOT


def _jj_commit(repository: pathlib.Path, revision: str) -> str:
    return _command_output(
        [
            "jj",
            "--ignore-working-copy",
            "-R",
            str(repository),
            "log",
            "--no-graph",
            "-r",
            revision,
            "-T",
            "commit_id",
        ]
    )


def _jj_is_ancestor(repository: pathlib.Path, ancestor: str, descendant: str) -> bool:
    observed = _command_output(
        [
            "jj",
            "--ignore-working-copy",
            "-R",
            str(repository),
            "log",
            "--no-graph",
            "-r",
            f"{ancestor} & ::{descendant}",
            "-T",
            "commit_id",
        ]
    )
    return observed == ancestor


def cmd_remote_verify(args: argparse.Namespace) -> int:
    """Compare local state with an authoritative, non-cached remote read."""
    db = store()
    repository = _authoritative_repository(db, args.repository)
    remotes = _command_output(
        [
            "jj",
            "--ignore-working-copy",
            "-R",
            str(repository),
            "git",
            "remote",
            "list",
        ]
    ).splitlines()
    candidates = []
    for line in remotes:
        fields = line.split(maxsplit=1)
        if len(fields) == 2 and fields[0] == args.remote:
            candidates.append(fields[1])
    if len(candidates) != 1:
        raise ValueError(f"remote {args.remote!r} is missing or ambiguous in {repository}")
    url = candidates[0]
    local = _jj_commit(repository, args.bookmark)
    tracking = _jj_commit(repository, f"{args.bookmark}@{args.remote}")
    remote_output = _command_output(
        ["git", "ls-remote", "--refs", url, f"refs/heads/{args.bookmark}"]
    )
    rows = [line.split() for line in remote_output.splitlines() if line.strip()]
    if len(rows) != 1 or len(rows[0]) < 2:
        raise ValueError(
            f"authoritative remote has no unique refs/heads/{args.bookmark}"
        )
    authoritative = rows[0][0]
    matched = local == tracking == authoritative
    if local == tracking:
        relation = "equal"
    elif _jj_is_ancestor(repository, tracking, local):
        relation = "local-ahead"
    elif _jj_is_ancestor(repository, local, tracking):
        relation = "local-behind"
    else:
        relation = "diverged"
    result = {
        "bookmark": args.bookmark,
        "remote": args.remote,
        "url": url,
        "repository": str(repository),
        "local": local,
        "tracking": tracking,
        "authoritative": authoritative,
        "local_tracking_relation": relation,
        "matched": matched,
    }
    if args.json:
        print(json.dumps(result, sort_keys=True, separators=(",", ":")))
    else:
        state = "match" if matched else "DIVERGED"
        print(
            f"{args.bookmark} {state} local={local[:12]} "
            f"tracking={tracking[:12]} remote={authoritative[:12]} "
            f"relation={relation}"
        )
        print(f"url {url}")
    return 0 if matched else 1


def workspace_registry(db: Any):
    return lambda event, record: db.save_workspace_record(event, record)


def cmd_work_create(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    instance = next(
        item for item in db.live_instances() if item["runtime_id"] == owner
    )
    record = worklib.create_workspace(
        args.source or ROOT,
        task=command_task(args),
        session_id=instance["conversation_id"],
        runtime_id=owner,
        conversation_id=instance["conversation_id"],
        scratch_root=args.scratch_root,
        destination=args.destination,
        base_revision=args.base,
        dependencies=args.depends_on,
        native_workspace=args.native_workspace,
        registry=workspace_registry(db),
    )
    if args.json:
        print(
            json.dumps(
                public_record(record, include_secret=args.show_token),
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    else:
        print(
            f"created {work_handle(record['instance_id'])} "
            f"{record['actual_path']}"
        )
        print(f"source {record['source']} base {args.base}")
    return 0


def cmd_work_adopt(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    candidate = pathlib.Path(args.path).expanduser().resolve()
    for existing in db.workspace_records():
        actual = existing.get("actual_path")
        if (
            actual
            and pathlib.Path(str(actual)).resolve() == candidate
            and existing.get("lifecycle_state") not in {"removed"}
        ):
            raise ValueError(
                f"path already belongs to workspace {existing['instance_id']}; "
                "inspect or resume that record instead"
            )
    record = worklib.adopt_workspace(
        candidate,
        task=command_task(args),
        session_id=args.conversation or identity(args),
        runtime_id=owner,
        conversation_id=args.conversation or identity(args),
        source=args.source or ROOT,
        scratch_root=args.scratch_root,
        base_revision=args.base,
        dependencies=args.depends_on,
        prior_ownership_token=args.prior_ownership_token,
        recovery_override=args.recovery_override,
        registry=workspace_registry(db),
    )
    if args.json:
        print(
            json.dumps(
                public_record(record, include_secret=args.show_token),
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    else:
        print(f"adopted {work_handle(record['instance_id'])} {record['actual_path']}")
    return 0


def cmd_work_list(args: argparse.Namespace) -> int:
    db = store()
    records = filtered_work(db.workspace_records(), args, db)
    for item in records:
        line = (
            f"{work_handle(item['instance_id'])} "
            f"{item.get('lifecycle_state', '?')}/{item.get('gate_state', '?')} "
            f"{item.get('task', '?')}"
        )
        if args.verbose:
            line += (
                f" id={item['instance_id']} "
                f"path={item.get('actual_path') or item.get('planned_path')}"
            )
        print(line)
    return 0


def cmd_work_gc(args: argparse.Namespace) -> int:
    """Inventory registered cleanup/recovery candidates without deleting."""
    records = store().workspace_records()
    candidates: list[tuple[dict[str, Any], str]] = []
    retained = 0
    for item in records:
        state = str(item.get("lifecycle_state") or "")
        path = pathlib.Path(
            str(item.get("actual_path") or item.get("planned_path") or "")
        )
        if state == "removed":
            continue
        if not path.is_dir():
            candidates.append((item, "missing-record"))
        elif state == "orphan-candidate":
            candidates.append((item, "recovery"))
        elif state == "archived":
            candidates.append((item, "cleanup-review"))
        else:
            retained += 1
    print(f"candidates {len(candidates)} retained {retained}")
    for item, reason in candidates:
        handle = work_handle(str(item["instance_id"]))
        path = pathlib.Path(
            str(item.get("actual_path") or item.get("planned_path") or "")
        )
        print(f"  {handle} {reason} {item.get('task', '?')} path={path.name}")
        if args.verbose:
            print(
                f"    state={item.get('lifecycle_state', '?')}/"
                f"{item.get('gate_state', '?')} id={item['instance_id']} path={path}"
            )
    return 0


def work_record(args: argparse.Namespace, db: Any) -> dict[str, Any]:
    explicit = getattr(args, "instance", "") or ""
    instance = explicit or os.environ.get("COORD_WORK_ID", "")
    if not instance:
        raise ValueError("a workspace handle is required")
    record = db.workspace_record(instance)
    if not explicit:
        actual = pathlib.Path(str(record.get("actual_path") or "")).resolve()
        cwd = pathlib.Path.cwd().resolve()
        try:
            cwd.relative_to(actual)
        except ValueError as exc:
            raise ValueError(
                "managed workspace default is only valid inside its registered path"
            ) from exc
        if str(record.get("runtime_id") or "") != runtime(args, db):
            raise ValueError(
                "managed workspace default does not belong to this live runtime"
            )
    return record


def work_gate_state(args: argparse.Namespace) -> str:
    state = getattr(args, "state", "") or ""
    if state:
        return state
    candidate = getattr(args, "instance", "") or ""
    if candidate in {"pending", "running", "passed", "failed", "waived"}:
        args.instance = ""
        return candidate
    raise ValueError("gate state is required")


def cmd_work_inspect(args: argparse.Namespace) -> int:
    db = store()
    record = worklib.inspect_workspace(
        work_record(args, db), registry=workspace_registry(db)
    )
    if args.json:
        print(
            json.dumps(
                public_record(record, include_secret=args.show_token),
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    else:
        print(
            f"{work_handle(record['instance_id'])} "
            f"{record.get('lifecycle_state', '?')}/{record.get('gate_state', '?')} "
            f"{record.get('task', '?')}\n"
            f"path {record.get('actual_path') or record.get('planned_path')}\n"
            f"dirty {record.get('dirty')} advanced {record.get('advanced')} "
            f"integration {record.get('integration_state', '?')}"
        )
    return 0


def cmd_work_state(args: argparse.Namespace) -> int:
    db = store()
    gate_state = work_gate_state(args) if args.work_command == "gate" else ""
    record = work_record(args, db)
    registry = workspace_registry(db)
    if args.work_command == "park":
        updated = worklib.park_workspace(record, reason=args.reason, registry=registry)
    elif args.work_command == "resume":
        updated = worklib.resume_workspace(
            record, runtime_id=runtime(args, db), registry=registry
        )
    elif args.work_command == "archive":
        updated = worklib.archive_workspace(record, note=args.note, registry=registry)
    else:
        updated = worklib.set_gate_state(record, gate_state, registry=registry)
    if getattr(args, "json", False):
        print(
            json.dumps(
                public_record(
                    updated, include_secret=getattr(args, "show_token", False)
                ),
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    else:
        print(
            f"{work_handle(updated['instance_id'])} "
            f"{updated.get('lifecycle_state', '?')}/{updated.get('gate_state', '?')}"
        )
    return 0


def cmd_work_remove(args: argparse.Namespace) -> int:
    db = store()
    record = work_record(args, db)
    handle = work_handle(record["instance_id"])
    if args.confirm:
        if args.confirm != handle:
            raise ValueError(f"confirmation must exactly match workspace handle {handle}")
        ownership_token = str(record.get("ownership_token") or "")
    else:
        ownership_token = args.ownership_token
    if not ownership_token:
        raise ValueError(f"confirm removal with --confirm {handle}")
    updated = worklib.remove_workspace(
        record,
        ownership_token=ownership_token,
        automatic=args.automatic,
        allow_ungated=args.allow_ungated,
        registry=workspace_registry(db),
    )
    print(f"coord: removed {updated['actual_path']}")
    return 0


def cmd_work_start(args: argparse.Namespace) -> int:
    db = store()
    record = worklib.start_agent(
        work_record(args, db),
        harness=args.harness,
        conversation_id=args.conversation,
        registry=workspace_registry(db),
    )
    print(
        f"coord: started {record['tmux']['window_name']} in "
        f"{record['tmux']['session_id']} ({record['tmux']['window_id']})"
    )
    print(shlex.join(worklib.attach_command(record)))
    return 0


def cmd_work_attach(args: argparse.Namespace) -> int:
    command = worklib.attach_command(store().workspace_record(args.instance))
    if args.execute:
        os.execvp(command[0], command)
    print(shlex.join(command))
    return 0


def cmd_work_close(args: argparse.Namespace) -> int:
    db = store()
    record = work_record(args, db)
    updated = worklib.close_agent(record, registry=workspace_registry(db))
    runtime_id = str(record.get("runtime_id") or "")
    if runtime_id:
        try:
            db.end_instance(db.resolve_runtime(runtime_id), reason="window-closed")
        except KeyError:
            pass
    print(f"closed managed window for {work_handle(updated['instance_id'])}")
    return 0


def cmd_work_queue(args: argparse.Namespace) -> int:
    db = store()
    owner = runtime(args, db)
    record = work_record(args, db)
    if args.action == "integrate":
        updated = worklib.prove_integrated(
            record,
            target_revision=args.target,
            registry=workspace_registry(db),
        )
        print(
            f"integrated {updated['integration_release_evidence']['source_commit'][:12]} "
            f"into {updated['integration_release_evidence']['target_commit'][:12]}"
        )
        return 0
    if args.action == "release":
        if record.get("integration_owner_runtime") != owner:
            raise ValueError(
                "only the runtime that acquired integration may release it"
            )
        if record.get("integration_resource") != args.resource:
            raise ValueError("release resource does not match acquisition evidence")

    def queue(action: str, _record: dict[str, Any]) -> None:
        if action == "acquire":
            db.claim_resources(
                owner,
                [args.resource],
                mode="exclusive",
                purpose=f"integrate workspace {args.instance}",
            )
        elif action in {"release", "cancel"}:
            if action == "release":
                released = db.release_resources(owner, [args.resource])
                if args.resource not in released:
                    raise ValueError(
                        "integration resource was not held; refusing false release"
                    )

    updated = worklib.enqueue_integration(
        record,
        queue=queue,
        action=args.action,
        registry=workspace_registry(db),
    )
    if args.action == "acquire":
        updated["integration_owner_runtime"] = owner
        updated["integration_resource"] = args.resource
    elif args.action == "release":
        updated["integration_lock_release_evidence"] = {
            "runtime_id": owner,
            "resource": args.resource,
        }
    db.save_workspace_record("workspace-integration-evidence", updated)
    print(f"coord: integration state {updated['integration_state']}")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="coord")
    parser.add_argument(
        "-i",
        "--identity",
        dest="identity_arg",
        default="",
        help="exact friendly runtime handle (normally inferred from the environment)",
    )
    parser.add_argument("--runtime", default="", help=argparse.SUPPRESS)
    parser.add_argument(
        "--sid",
        default="",
        help=argparse.SUPPRESS,
    )
    sub = parser.add_subparsers(dest="command", required=True)

    hello = sub.add_parser("hello")
    hello.add_argument("--conversation", default="")
    hello.add_argument("--harness", default="unknown")
    hello.add_argument("--label", default="")
    hello.add_argument("--cwd", default="")
    hello.add_argument("--ttl", type=float, default=storelib.DEFAULT_INSTANCE_TTL)
    hello.set_defaults(func=cmd_hello)

    heartbeat = sub.add_parser("heartbeat")
    heartbeat.add_argument("--ttl", type=float, default=storelib.DEFAULT_INSTANCE_TTL)
    heartbeat.add_argument("--lease-ttl", type=float, default=storelib.DEFAULT_LEASE_TTL)
    heartbeat.set_defaults(func=cmd_heartbeat)

    bye = sub.add_parser("bye")
    bye.add_argument("--reason", default="clean")
    bye.set_defaults(func=cmd_bye)

    for name, handler in (("claim", cmd_claim), ("release", cmd_release)):
        command = sub.add_parser(name)
        command.add_argument("resources", nargs="*")
        command.add_argument(
            "--resources",
            "--paths",
            nargs="+",
            dest="legacy_resources",
            help=argparse.SUPPRESS,
        )
        if name == "claim":
            command.add_argument(
                "--mode", choices=tuple(storelib.VALID_LEASE_MODES), default="exclusive"
            )
            command.add_argument("-p", "--purpose", default="")
            command.add_argument("--ttl", type=float, default=storelib.DEFAULT_LEASE_TTL)
        command.set_defaults(func=handler)

    renew = sub.add_parser("renew")
    renew.add_argument("resources", nargs="*")
    renew.add_argument(
        "--resources",
        "--paths",
        nargs="+",
        dest="legacy_resources",
        help=argparse.SUPPRESS,
    )
    renew.add_argument("--ttl", type=float, default=storelib.DEFAULT_LEASE_TTL)
    renew.set_defaults(func=cmd_renew)

    listing = sub.add_parser("list")
    listing.add_argument("--active-only", action="store_true")
    listing.add_argument("-v", "--verbose", action="store_true")
    listing.add_argument("--json", action="store_true")
    listing.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
    listing.add_argument("--work", action="store_true", help="include workspace rows")
    listing.add_argument("--state", action="append", default=[])
    listing.add_argument("--owner", default="")
    listing.add_argument("--since", default="")
    listing.set_defaults(func=cmd_list)

    send = sub.add_parser("send")
    send.add_argument("body", nargs="?")
    send.add_argument("--to", default="*")
    send.add_argument("--kind", default="fyi")
    send.add_argument("--body", dest="legacy_body", help=argparse.SUPPRESS)
    send.add_argument("--ref", action="append", default=[])
    send.set_defaults(func=cmd_send)

    inbox = sub.add_parser("inbox")
    inbox.add_argument("--limit", type=int, default=100)
    inbox.add_argument("--peek", action="store_true")
    inbox.set_defaults(func=cmd_inbox)

    ack = sub.add_parser("ack")
    ack.add_argument("message_ids", type=int, nargs="*")
    ack.set_defaults(func=cmd_ack)

    check = sub.add_parser("check")
    check.add_argument("resource", nargs="?")
    check.add_argument(
        "--resource",
        "--path",
        dest="legacy_resource",
        help=argparse.SUPPRESS,
    )
    check.add_argument("--instance-ttl", type=float, default=storelib.DEFAULT_INSTANCE_TTL)
    check.add_argument("--lease-ttl", type=float, default=storelib.DEFAULT_LEASE_TTL)
    check.set_defaults(func=cmd_check)

    audit = sub.add_parser("audit")
    audit.add_argument("--after", type=int, default=0)
    audit.add_argument("--output")
    audit.add_argument("--include-secrets", action="store_true")
    audit.set_defaults(func=cmd_audit)

    batch = sub.add_parser(
        "batch",
        help="compose guarded coordinator actions separated by ':'",
    )
    batch.add_argument("--json", action="store_true")
    batch.add_argument("actions", nargs=argparse.REMAINDER)
    batch.set_defaults(func=cmd_batch)

    remote = sub.add_parser("remote", help="authoritative remote-state checks")
    remote_sub = remote.add_subparsers(dest="remote_command", required=True)
    verify = remote_sub.add_parser("verify")
    verify.add_argument("bookmark")
    verify.add_argument("--remote", default="origin")
    verify.add_argument("--repository", default="")
    verify.add_argument("--json", action="store_true")
    verify.set_defaults(func=cmd_remote_verify)

    work = sub.add_parser("work", help="managed isolated jj clones and tmux agents")
    work_sub = work.add_subparsers(dest="work_command", required=True)
    create = work_sub.add_parser("create")
    create.add_argument("task", nargs="?")
    create.add_argument("--task", dest="legacy_task", help=argparse.SUPPRESS)
    create.add_argument("--source", default="")
    create.add_argument("--scratch-root", default=str(worklib.DEFAULT_SCRATCH_ROOT))
    create.add_argument("--destination")
    create.add_argument("--base", default="master@origin")
    create.add_argument("--depends-on", action="append", default=[])
    create.add_argument("--native-workspace", action="store_true")
    create.add_argument("--json", action="store_true")
    create.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
    create.set_defaults(func=cmd_work_create)
    adopt = work_sub.add_parser(
        "adopt", help="register an existing independent scratch clone"
    )
    adopt.add_argument("path")
    adopt.add_argument("task", nargs="?")
    adopt.add_argument("--task", dest="legacy_task", help=argparse.SUPPRESS)
    adopt.add_argument("--source", default="")
    adopt.add_argument("--scratch-root", default=str(worklib.DEFAULT_SCRATCH_ROOT))
    adopt.add_argument("--base", default="master@origin")
    adopt.add_argument("--depends-on", action="append", default=[])
    adopt.add_argument("--conversation", default="")
    adopt.add_argument("--prior-ownership-token", default="")
    adopt.add_argument("--recovery-override", action="store_true")
    adopt.add_argument("--json", action="store_true")
    adopt.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
    adopt.set_defaults(func=cmd_work_adopt)
    work_list = work_sub.add_parser("list")
    work_list.add_argument("--active-only", action="store_true")
    work_list.add_argument("-v", "--verbose", action="store_true")
    work_list.add_argument("--state", action="append", default=[])
    work_list.add_argument("--owner", default="")
    work_list.add_argument("--since", default="")
    work_list.set_defaults(func=cmd_work_list)
    gc = work_sub.add_parser(
        "gc", help="report registered cleanup and recovery candidates"
    )
    gc.add_argument("-v", "--verbose", action="store_true")
    gc.set_defaults(func=cmd_work_gc)
    inspect = work_sub.add_parser("inspect")
    inspect.add_argument("instance", nargs="?")
    inspect.add_argument("--json", action="store_true")
    inspect.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
    inspect.set_defaults(func=cmd_work_inspect)
    for name in ("park", "resume", "archive"):
        state = work_sub.add_parser(name)
        state.add_argument("instance")
        if name == "park":
            state.add_argument("--reason", default="")
        elif name == "archive":
            state.add_argument("--note", default="")
        state.add_argument("--json", action="store_true")
        state.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
        state.set_defaults(func=cmd_work_state)
    gate = work_sub.add_parser("gate")
    gate.add_argument("instance", nargs="?")
    gate.add_argument("state", nargs="?")
    gate.add_argument("--json", action="store_true")
    gate.add_argument("--show-token", action="store_true", help=argparse.SUPPRESS)
    gate.set_defaults(func=cmd_work_state)
    remove = work_sub.add_parser("remove")
    remove.add_argument("instance")
    remove.add_argument("--confirm", default="")
    remove.add_argument("--ownership-token", default="", help=argparse.SUPPRESS)
    remove.add_argument("--automatic", action="store_true")
    remove.add_argument("--allow-ungated", action="store_true")
    remove.set_defaults(func=cmd_work_remove)
    start = work_sub.add_parser("start")
    start.add_argument("instance")
    start.add_argument("--harness", choices=("codex", "claude"), required=True)
    start.add_argument("--conversation", default=None)
    start.set_defaults(func=cmd_work_start)
    attach = work_sub.add_parser("attach")
    attach.add_argument("instance")
    attach.add_argument("--execute", action="store_true")
    attach.set_defaults(func=cmd_work_attach)
    close = work_sub.add_parser(
        "close", help="safely close the exact managed tmux window"
    )
    close.add_argument("instance")
    close.set_defaults(func=cmd_work_close)
    queue = work_sub.add_parser("queue")
    queue.add_argument("instance")
    queue.add_argument(
        "action", choices=("enqueue", "acquire", "release", "integrate", "cancel")
    )
    queue.add_argument("--resource", default="integration/master")
    queue.add_argument("--target", default="master")
    queue.set_defaults(func=cmd_work_queue)
    return parser


def main() -> int:
    args = build_parser().parse_args()
    return _execute(args)


if __name__ == "__main__":
    raise SystemExit(main())