commit 9c47ef5fde573e0580202412d3ffac7adc6221bf Author: sid Date: Thu Jul 30 22:56:38 2026 -0600 feat: standalone coordinator and repository installer diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9ed964d --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +# Legacy pre-SQLite runtime residue. Current state lives in the private, +# host-local SQLite database outside the repository; never version either form. +state/ +__pycache__/ +*.pyc diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..c0bcb1f --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,237 @@ +# Agent coordination design + +Status: implemented. Packaging and installation: [`README.md`](README.md). + +## Boundary that actually isolates work + +The shared Jujutsu working copy is also `@`. Every `jj` command snapshots all peer filesystem +edits into whichever change currently owns that checkout. File leases cannot prevent this because +the absorbing session did not write those files. This caused duplicate work, foreign features +inside unrelated commits, and one temporary loss/recovery incident. + +Writable main agents therefore use independent colocated `jj git clone` instances. They have +independent operation logs, working-copy commits and Git backends, and ordinary Git-aware Nix +flake discovery works in them. The canonical checkout is for read-only inspection and one +explicit integration owner. + +Native `jj workspace add` remains available behind `--native-workspace`. It is useful when sharing +one repository is intentional, but it shares the operation log/object store and is not the +default isolation boundary. In this repository its secondary directory also lacks an ordinary +`.git`, so `nix flake metadata .` does not discover the colocated Git flake normally. + +Managed clones live under `/var/tmp/fleet-audit`, the large non-snapshotted scratch dataset. +`/tmp` and `/home` are on the nearly-full snapshotted dataset and must not receive agent clones. + +## Identity + +- A conversation ID is durable across resume. +- A runtime ID is a fresh UUID for one live attachment to that conversation. +- Human/LLM-facing runtime and workspace identities are exact five-word handles derived from the + full opaque ID with a domain-separated SHA-256 mapping over 128 common words (35 display bits). + Exact handle matches carry authority; prefixes never do, and an unlikely collision is refused + rather than guessed. +- Full UUIDs remain canonical in SQLite, tmux metadata, launcher environments and deterministic + audit exports. Handles are aliases, not secrets and not replacements for internal entropy. +- Managed launches export the exact workspace record as `COORD_WORK_ID`. It may default only + low-blast-radius workspace-local operations after proving that the live runtime owns the record + and `cwd` is inside its registered path. Resource scopes and destructive targets stay explicit. +- The launcher stores full IDs in tmux user options. Window names are presentation only: + `-`. + +The distinction prevents a stale pre-resume process from renewing or releasing the resumed +process's resources. + +## Transactional state + +Runtime state is a private SQLite WAL database at +`$XDG_STATE_HOME/fleet-agent-coord/.sqlite3` (override with `COORD_DB`). It is outside +every agent-writable clone, directory mode 0700 and database/WAL mode 0600. + +This database is deliberately host-local. It coordinates Claude and Codex processes sharing one +machine; it does not serialize work performed from different fleet nodes. Cross-host deployment +still needs an explicit human/integration owner (or a future authenticated service). + +All conflict-check-plus-claim operations use `BEGIN IMMEDIATE`. Resource names are exact, +repository-relative hierarchical names. Equal names and ancestor/descendant names overlap. +The compatibility matrix is: + +| Existing | Requested append | Requested exclusive | +|---|---:|---:| +| append | allowed | refused | +| exclusive | refused | refused | + +Leases cover files and global operations alike, for example: + +- `tests/checks.nix` +- `integration/master` +- `push/origin/master` +- `activate/home/tpp15s` +- `deploy/system/nix-control` +- `external/fleet-dotfiles` + +Heartbeats renew both runtime presence and held leases. Expired/dead runtime leases are reaped +transactionally. Partial release removes only the named resources. + +Messages have monotonic integer IDs. A broadcast snapshots its recipient set at send time. +Delivery transitions `queued → delivered → acked`. Delivered context repeats at hook boundaries +until the receiving agent explicitly runs `coord ack `. The first delivery carries the +body; later boundaries carry the message ID, sender handle, kind and body digest plus an explicit +`coord inbox --peek` recovery command. A rejected/lost harness response therefore cannot silently +consume it, while a long body is not reinjected on every tool call. + +Audit events remain append-like in SQLite and can be exported deterministically as JSONL with +`coord audit`. + +## Action composition + +`coord batch ACTION : ACTION...` reduces agent/tool round trips without inventing a second command +language: every action is parsed by the ordinary CLI parser and executed by its ordinary handler. +It runs sequentially, stops on the first nonzero result, and can return compact human output or +one JSON envelope. The separator is a standalone token, so quoted message punctuation is not +special. + +A batch is intentionally not atomic and performs no rollback. Each state-changing action retains +its own transaction, validation and audit event. The allowlist excludes clone creation/removal, +tmux process replacement and integration queueing. Generic composition must not become a way to +hide destructive scope or ambiguous external side effects. + +Every nested action inherits the outer batch's already-resolved exact runtime. Identity flags are +rejected inside action token streams and the resolved nested runtime is asserted equal before +dispatch, preventing one batch from mixing or impersonating audit/lease authority. + +Remote publication, integration, activation and deployment need purpose-built workflows with +explicit resources and expected object IDs. In particular, a future publish workflow must acquire +the exact push lease, authoritatively read the network remote, refuse an unexpected base, avoid +force, reread the new remote head, and retain the lease with recovery instructions after an +ambiguous mutation result. + +## Enforcement and degradation + +The shared hook recognizes Claude `Edit`/`Write` paths and every Codex `apply_patch` source and +`Move to:` destination. It canonicalizes against the hook payload's working directory and blocks +a peer's exclusive overlapping lease. + +This is scoped enforcement, not a shell parser. Shell writers, formatters, generators and VCS +operations remain advisory. Isolation is what makes that honest limitation safe. A private clone +can continue working if the coordinator is unavailable. + +The hook and lifecycle bridge fail open on load, parse or database failure. A positively +recognized conflicting write fails closed. Denial and `additionalContext` are separate branches +because Codex rejects a response containing both. + +Claude subagents receive their own deterministic runtime UUID, derived from the full parent +conversation ID and hook `agent_id`. They therefore conflict, message, and lease independently +instead of inheriting the parent's authority. Writable subagents should still use their own +managed clone; the distinct identity makes accidental writes in the integration checkout fail +closed unless that subagent explicitly owns the resource. + +## Workspace lifecycle and cleanup + +Each record contains the full task/conversation/runtime identity, source and actual path, exact +base/current Jujutsu IDs, dependencies, planned/actual resources, gate/integration/lifecycle +states, immutable tmux IDs and an ownership token. + +The default CLI view deliberately omits those machine fields. It prints handles, task/state, and +only the path needed to enter a newly created clone. `--verbose`, `--json`, and audit export are +explicit diagnostic/machine surfaces. Cleanup tokens remain private by default; normal removal +requires repeating the exact work handle and then compares the registry token with the on-disk +marker. + +Lifecycle states support active, parked, resumed, archived, queued/integrating/lock-released/ +integrated and removed work. Lock release is not integration evidence: a separate proof verifies +that the exact source commit is an ancestor of the exact target commit. Cleanup never guesses: + +The current “queue” is an integration lock plus an auditable state machine, not a fair scheduler: +enqueue order and recorded dependencies are visible but do not automatically grant the next turn. +The integration owner adjudicates readiness and stale bases. + +- path must remain below the recorded scratch root; +- private registry token and on-disk `.jj` marker must agree; +- working copy must be clean; +- an advanced change must be released/integrated; +- automatic cleanup requires a passed or explicitly waived gate. + +Missing, dirty, ungated or identity-mismatched directories remain recovery candidates. + +Known independent clones can be recovered after database loss with explicit `work adopt`. Adoption +requires both `.jj/` and `.git/`, a path beneath the configured scratch root, and a source distinct +from the candidate. It validates exact Jujutsu IDs before rotating the ownership marker. There is +deliberately no directory-name scan or automatic adoption. + +If older shared-checkout work was lost, recover read-only with +`jj --at-op= --ignore-working-copy file show `, then merge the recovered content into a +fresh isolated clone. Never restore a whole stale file over newer work. + +## Tmux rules + +Managed sessions discover project ownership through tmux user options and mutate objects only by +immutable `$session`, `@window` and `%pane` IDs returned by tmux. Names are never lookup targets. +Automatic window renaming is disabled. Codex resumes with `codex-direct resume `; +Claude resumes with `claude-direct --resume `. + +`work close` revalidates those IDs and kills the exact window. It refuses when called from any +pane in the target window, because killing that window would terminate the coordinator before its +registry update. If the operator is attached elsewhere in the target session's last window, it +first switches to another managed session and otherwise refuses, preventing the surprising drop +to a plain outer shell. + +The ordinary `codex`/`claude` helper also persists interactive resume invocations and forwards +their arguments to the session runner. Automation (`codex exec`, `claude -p`, pipes and non-TTY +calls) remains direct and status-faithful. + +## Proved failure modes + +Behavioral checks cover: + +- eight-way exclusive race: exactly one winner; +- six simultaneous messages: no loss and monotonic IDs; +- exclusive/append matrix, hierarchical conflicts, renewal/expiry and partial release; +- truncated identity refusal; +- `apply_patch` move-destination blocking; +- repeated delivery until explicit acknowledgement; +- independent clone metadata and explicit native-workspace limitations; +- dirty, ungated, advanced and ownership-mismatched cleanup refusal; +- tmux name collision handling and immutable-ID-only mutations; +- distinct Claude subagent authority, explicit orphan adoption and exact-window close; +- real disposable `jj git clone --colocate` create/inspect/remove. + +## Implementation language and versioning + +Python plus the standard-library SQLite driver remains the current best fit: hooks can run before +a Nix activation or build bootstrap, deployment has no compiled-artifact handoff, and the +transaction boundary lives in SQLite rather than process memory. A Go rewrite would improve +single-binary distribution, startup predictability and static typing, but would not improve the +isolation or locking model by itself. Those benefits do not currently justify a second +implementation and migration surface. + +## Interface and context budget + +Coordinator output crosses an unusually expensive boundary: lifecycle context and command output +are repeatedly injected into model context. The interface therefore follows these rules: + +- managed environments infer caller identity; ordinary commands do not repeat it; +- required operands are positional (`claim integration/master`, `work create cache-review`); +- readable long options remain for genuinely optional semantics; +- default listings use exact word handles and suppress full IDs, conversations, paths and + workspace rows not requested by the caller; +- JSON is compact and explicit, and ownership secrets are redacted unless a private recovery + export opts in; +- full Jujutsu/Git IDs are stored as evidence but presentation output uses a sufficient display + prefix only where no later command consumes it as authority. + +This is a larger saving than removing the two dash characters from every option. Cryptic flags +can increase correction turns and negate their tiny lexical saving. + +The executable hook source is cooperative infrastructure, not a security boundary: a writable +clone contains a writable copy, and hook trust is hash-pinned by each harness. Before deploying +incompatible database schemas, add explicit versioned migrations with backup/rollback and package +one immutable coordinator runtime through Home Manager. That packaging can use Python or Go; it +does not require a language rewrite. + +## Third-party adjudication + +The current system is intentionally small and local. MCP Agent Mail may later add a searchable +mail UI, and `tmux-agent-status` may add an operator sidebar, but neither is authoritative for +leases. Beads adds a useful task DAG at the cost of Dolt; Gas Town and Agent of Empires are +Git-worktree-oriented and too invasive as the coordination substrate. Any pilot must compose with, +not replace, the transactional resource and isolated-clone boundary. diff --git a/README.md b/README.md new file mode 100644 index 0000000..b467676 --- /dev/null +++ b/README.md @@ -0,0 +1,25 @@ +# coord + +Harness-neutral coordination for Claude Code, Codex, and other repository agents. + +The standalone project is the source of truth. Install a vendored copy into another repository: + +```sh +./install /path/to/repository --project-id my-logical-project +``` + +An installation provides: + +- `.coord/` — canonical CLI, hooks, runtime modules, installer, design, and immutable project ID. +- `.claude/coord/` — compatibility shims for existing sessions and old instructions. +- merged `.claude/settings.json` and `.codex/hooks.json` entries pointing at `.coord/`. + +Upgrade an existing installation with the same command. Existing project IDs are preserved and a +conflicting requested ID is refused. Check drift without writing: + +```sh +./install /path/to/repository --check +``` + +Runtime state remains outside repositories in the private host-local SQLite database described in +[`DESIGN.md`](DESIGN.md). Installing or upgrading never copies, deletes, or rewrites that state. diff --git a/coord b/coord new file mode 100755 index 0000000..507a4a0 --- /dev/null +++ b/coord @@ -0,0 +1,1274 @@ +#!/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()) diff --git a/hook.py b/hook.py new file mode 100755 index 0000000..ad1957a --- /dev/null +++ b/hook.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Fail-open recognized-write guard and transactional peer-message delivery.""" + +from __future__ import annotations + +import importlib.util +import importlib.machinery +import hashlib +import json +import os +import pathlib +import re +import socket +import sys +import uuid + +HERE = pathlib.Path(__file__).resolve().parent +CLI_SPEC = importlib.util.spec_from_loader( + "coord_cli", + importlib.machinery.SourceFileLoader("coord_cli", str(HERE / "coord")), +) +try: + cli = importlib.util.module_from_spec(CLI_SPEC) + assert CLI_SPEC.loader is not None + CLI_SPEC.loader.exec_module(cli) +except Exception as exc: # the hook boundary must fail open, including import + print(f"coord-hook: cannot load coordinator; failing open: {exc}", file=sys.stderr) + raise SystemExit(0) + +PATCH_PATH = re.compile( + r"^\*\*\* (?:Add|Update|Delete) File: (.+)$|^\*\*\* Move to: (.+)$", + re.MULTILINE, +) +MAX_MESSAGES = 5 + + +def allow() -> None: + raise SystemExit(0) + + +def runtime_for(db: object, event: dict) -> str: + agent_id = str(event.get("agent_id") or "") + conversation = str(event.get("session_id") or "") + if agent_id and conversation: + sub_conversation = f"{conversation}:agent:{agent_id}" + exact = str( + uuid.uuid5( + uuid.NAMESPACE_URL, f"fleet-agent-coord:{sub_conversation}" + ) + ) + try: + return db.resolve_runtime(exact) + except KeyError: + return db.register_instance( + sub_conversation, + runtime_id=exact, + harness="claude-code-subagent", + host=socket.gethostname(), + cwd=str(project_root()), + label=f"subagent:{agent_id}", + ) + exact = os.environ.get("COORD_INSTANCE_ID", "") + return db.resolve_runtime(exact or event.get("session_id", "")) + + +def project_root() -> pathlib.Path: + """Return the checkout root; the override exists for packaged tests/tools.""" + override = os.environ.get("COORD_PROJECT_ROOT", "") + return pathlib.Path(override).resolve() if override else cli.ROOT + + +def relative_path(raw: str, cwd: pathlib.Path, root: pathlib.Path) -> str | None: + candidate = pathlib.Path(raw) + if not candidate.is_absolute(): + candidate = cwd / candidate + try: + return candidate.resolve(strict=False).relative_to(root).as_posix() + except ValueError: + return None + + +def paths_from(event: dict) -> list[str]: + tool_input = event.get("tool_input") or {} + cwd = pathlib.Path(event.get("cwd") or os.getcwd()).resolve() + root = project_root() + raw_paths: list[str] = [] + direct = tool_input.get("file_path") or tool_input.get("path") + if isinstance(direct, str): + raw_paths.append(direct) + body = tool_input.get("command") or tool_input.get("input") or "" + if isinstance(body, str): + for match in PATCH_PATH.finditer(body): + raw_paths.append(match.group(1) or match.group(2)) + normalized = (relative_path(path.strip(), cwd, root) for path in raw_paths) + return list(dict.fromkeys(path for path in normalized if path)) + + +def output_context(messages: list[object]) -> None: + lines = [] + for item in messages: + sender = cli.runtime_handle(item.sender_runtime_id) + if item.state == "delivered": + digest = hashlib.sha256(item.body.encode("utf-8")).hexdigest()[:10] + lines.append( + f"[{item.message_id}] {sender} pending again " + f"({item.kind}, sha256:{digest}); read with `coord inbox --peek`" + ) + else: + lines.append( + f"[{item.message_id}] {sender} ({item.kind}): {item.body}" + ) + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "additionalContext": ( + "Messages from peer agent runtimes:\n" + + "\n".join(lines) + + "\nDelivery repeats until explicitly acknowledged with " + "`coord ack `." + ), + } + } + ) + ) + + +def main() -> None: + try: + event = json.loads(sys.stdin.read()) + except Exception: + allow() + db = cli.store() + owner = runtime_for(db, event) + db.heartbeat( + owner, + ttl_seconds=cli.storelib.DEFAULT_INSTANCE_TTL, + renew_lease_ttl_seconds=cli.storelib.DEFAULT_LEASE_TTL, + ) + + managed_workspace = db.is_managed_workspace(project_root()) + leases = db.active_leases() + for path in paths_from(event): + owner_has_lease = False + for lease in leases: + if ( + lease["runtime_id"] == owner + and lease["mode"] == "exclusive" + and db.resources_overlap(path, lease["path"]) + ): + owner_has_lease = True + if lease["runtime_id"] == owner or lease["mode"] != "exclusive": + continue + if not db.resources_overlap(path, lease["path"]): + continue + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": ( + f"{path} is leased exclusively by " + f"{cli.runtime_handle(lease['runtime_id'])} for " + f"{lease['purpose'] or 'unstated'}. Coordinate before editing." + ), + } + } + ) + ) + return + if not managed_workspace and not owner_has_lease: + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": ( + f"{path} is in the canonical integration checkout. " + "Writable work must use a managed isolated clone, or " + "the integration owner must claim this exact resource." + ), + } + } + ) + ) + return + + pending = db.inbox(owner, limit=MAX_MESSAGES, mark_delivered=True) + if pending: + output_context(pending) + + +if __name__ == "__main__": + try: + main() + except SystemExit: + raise + except Exception as exc: # fail open is load-bearing + print(f"coord-hook: failing open: {exc}", file=sys.stderr) + raise SystemExit(0) diff --git a/install b/install new file mode 100755 index 0000000..936ffe7 --- /dev/null +++ b/install @@ -0,0 +1,303 @@ +#!/usr/bin/env python3 +"""Install or verify a vendored, harness-neutral coordinator.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import re +import secrets +import stat +import sys +from typing import Any + + +HERE = Path(__file__).resolve().parent +CORE_FILES = ( + ".gitignore", + "DESIGN.md", + "README.md", + "coord", + "hook.py", + "install", + "session.py", + "store.py", + "workspace.py", +) +EXECUTABLES = {"coord", "install"} +PROJECT_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") + +CLAUDE_COMMANDS = { + "SessionStart": ('python3 "$CLAUDE_PROJECT_DIR/.coord/session.py"', 10), + "SessionEnd": ('python3 "$CLAUDE_PROJECT_DIR/.coord/session.py"', 5), + "PreToolUse": ('python3 "$CLAUDE_PROJECT_DIR/.coord/hook.py"', 10), +} +CODEX_DISCOVERY = ( + 'root="$PWD"; while [ "$root" != / ] && [ ! -f ' + '"$root/.coord/{name}.py" ]; do root="${{root%/*}}"; ' + '[ -n "$root" ] || root=/; done; ' + '[ -f "$root/.coord/{name}.py" ] || exit 0; ' + 'python3 "$root/.coord/{name}.py"' +) + + +def legacy_wrapper(name: str) -> str: + if name == "coord": + return """#!/usr/bin/env python3 +import os +from pathlib import Path +import sys + +target = Path(__file__).resolve().parents[2] / ".coord" / "coord" +os.execv(str(target), [str(target), *sys.argv[1:]]) +""" + if name in {"hook.py", "session.py"}: + return f"""#!/usr/bin/env python3 +from pathlib import Path +import runpy + +target = Path(__file__).resolve().parents[2] / ".coord" / "{name}" +runpy.run_path(str(target), run_name="__main__") +""" + if name in {"store.py", "workspace.py"}: + module = f"_coord_{name.removesuffix('.py')}_compat" + return f"""from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path +import sys + +target = Path(__file__).resolve().parents[2] / ".coord" / "{name}" +spec = spec_from_file_location("{module}", target) +assert spec is not None and spec.loader is not None +implementation = module_from_spec(spec) +sys.modules[spec.name] = implementation +spec.loader.exec_module(implementation) +globals().update({{ + key: value for key, value in vars(implementation).items() + if key not in {{"__name__", "__loader__", "__package__", "__spec__"}} +}}) +""" + raise ValueError(f"unsupported compatibility wrapper: {name}") + + +def read_json(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError(f"refusing invalid JSON at {path}: {exc}") from exc + if not isinstance(value, dict): + raise ValueError(f"refusing non-object JSON at {path}") + return value + + +def is_coord_hook(item: object) -> bool: + return isinstance(item, dict) and isinstance(item.get("command"), str) and ( + ".coord/" in item["command"] or ".claude/coord/" in item["command"] + ) + + +def merged_hooks( + current: dict[str, Any], desired: dict[str, tuple[str, int]] +) -> dict[str, Any]: + result = json.loads(json.dumps(current)) + hooks = result.setdefault("hooks", {}) + if not isinstance(hooks, dict): + raise ValueError("refusing hook configuration whose 'hooks' value is not an object") + for event, (command, timeout) in desired.items(): + existing = hooks.get(event, []) + if not isinstance(existing, list): + raise ValueError(f"refusing hook event {event!r} whose value is not a list") + retained = [] + for group in existing: + if not isinstance(group, dict): + retained.append(group) + continue + entries = group.get("hooks") + if not isinstance(entries, list): + retained.append(group) + continue + kept_entries = [entry for entry in entries if not is_coord_hook(entry)] + if kept_entries: + kept_group = dict(group) + kept_group["hooks"] = kept_entries + retained.append(kept_group) + installed: dict[str, Any] = { + "hooks": [{"type": "command", "command": command, "timeout": timeout}] + } + if event == "PreToolUse": + installed["matcher"] = ".*" + retained.append(installed) + hooks[event] = retained + return result + + +def codex_hooks(current: dict[str, Any]) -> dict[str, Any]: + desired = { + event: ( + CODEX_DISCOVERY.format( + name="hook" if event == "PreToolUse" else "session" + ), + timeout, + ) + for event, (_command, timeout) in CLAUDE_COMMANDS.items() + } + return merged_hooks(current, desired) + + +def write_if_changed( + path: Path, + content: bytes, + *, + executable: bool = False, + check: bool, + drift: list[str], +) -> None: + observed = path.read_bytes() if path.is_file() else None + expected_mode = 0o755 if executable else 0o644 + observed_executable = path.is_file() and bool( + stat.S_IMODE(path.stat().st_mode) & stat.S_IXUSR + ) + mode_ok = path.is_file() and observed_executable == executable + if observed == content and mode_ok: + return + drift.append(str(path)) + if check: + return + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.coord-new-{secrets.token_hex(6)}") + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, expected_mode) + try: + os.write(fd, content) + os.fsync(fd) + finally: + os.close(fd) + os.replace(temporary, path) + os.chmod(path, expected_mode) + + +def project_id(target: Path, requested: str, *, check: bool) -> str: + canonical = target / ".coord" / "project-id" + legacy = target / ".claude" / "coord" / "project-id" + existing = [] + for path in (canonical, legacy): + if path.is_file(): + existing.append((path, path.read_text(encoding="utf-8").strip())) + values = {value for _path, value in existing} + if len(values) > 1: + detail = ", ".join(f"{path}={value!r}" for path, value in existing) + raise ValueError(f"refusing ambiguous existing project IDs: {detail}") + selected = next(iter(values), "") + if requested and selected and requested != selected: + raise ValueError( + f"refusing project ID change from {selected!r} to {requested!r}" + ) + selected = selected or requested + if not selected: + if check: + raise ValueError("installation has no project ID") + raise ValueError("new installation requires --project-id") + if not PROJECT_ID_RE.fullmatch(selected) or ".." in selected: + raise ValueError("project ID must be one safe immutable slug") + return selected + + +def install(target: Path, *, requested_id: str, check: bool) -> list[str]: + target = target.expanduser().resolve() + if not target.is_dir(): + raise ValueError(f"target is not a directory: {target}") + selected_id = project_id(target, requested_id, check=check) + drift: list[str] = [] + destination = target / ".coord" + for name in CORE_FILES: + write_if_changed( + destination / name, + (HERE / name).read_bytes(), + executable=name in EXECUTABLES, + check=check, + drift=drift, + ) + write_if_changed( + destination / "project-id", + f"{selected_id}\n".encode(), + check=check, + drift=drift, + ) + + legacy = target / ".claude" / "coord" + for name in ("coord", "hook.py", "session.py", "store.py", "workspace.py"): + write_if_changed( + legacy / name, + legacy_wrapper(name).encode(), + executable=name == "coord", + check=check, + drift=drift, + ) + write_if_changed( + legacy / "project-id", + f"{selected_id}\n".encode(), + check=check, + drift=drift, + ) + write_if_changed( + legacy / "DESIGN.md", + b"Canonical design: [`../../.coord/DESIGN.md`](../../.coord/DESIGN.md).\\n", + check=check, + drift=drift, + ) + write_if_changed( + legacy / ".gitignore", + b"state/\\n__pycache__/\\n", + check=check, + drift=drift, + ) + + settings = target / ".claude" / "settings.json" + expected_claude = ( + json.dumps( + merged_hooks(read_json(settings), CLAUDE_COMMANDS), + indent=2, + sort_keys=True, + ) + + "\n" + ).encode() + write_if_changed(settings, expected_claude, check=check, drift=drift) + + codex = target / ".codex" / "hooks.json" + expected_codex = ( + json.dumps(codex_hooks(read_json(codex)), indent=2, sort_keys=True) + "\n" + ).encode() + write_if_changed(codex, expected_codex, check=check, drift=drift) + return drift + + +def main() -> int: + parser = argparse.ArgumentParser(prog="coord-install") + parser.add_argument("target", nargs="?", default=".") + parser.add_argument("--project-id", default="") + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + try: + drift = install( + Path(args.target), requested_id=args.project_id, check=args.check + ) + except ValueError as exc: + print(f"coord-install: {exc}", file=sys.stderr) + return 2 + if args.check: + if drift: + print(f"coord-install: drift ({len(drift)} files)") + for path in drift: + print(f" {path}") + return 1 + print("coord-install: installation matches") + return 0 + print(f"coord-install: {'updated' if drift else 'already current'}") + print(f"target {Path(args.target).expanduser().resolve()}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/project-id b/project-id new file mode 100644 index 0000000..6642169 --- /dev/null +++ b/project-id @@ -0,0 +1 @@ +coord-tooling diff --git a/session.py b/session.py new file mode 100755 index 0000000..d30d423 --- /dev/null +++ b/session.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Best-effort session lifecycle bridge for Claude Code and Codex hooks.""" + +from __future__ import annotations + +import importlib.util +import importlib.machinery +import hashlib +import json +import os +import pathlib +import socket +import sqlite3 +import subprocess +import sys +import uuid + +HERE = pathlib.Path(__file__).resolve().parent +SPEC = importlib.util.spec_from_loader( + "coord_cli", + importlib.machinery.SourceFileLoader("coord_cli", str(HERE / "coord")), +) +cli = importlib.util.module_from_spec(SPEC) +assert SPEC.loader is not None +SPEC.loader.exec_module(cli) + + +def annotate_tmux(runtime_id: str, conversation_id: str) -> None: + pane = os.environ.get("TMUX_PANE") + if not pane: + return + for key, value in ( + ("@coord_instance", runtime_id), + ("@coord_conversation", conversation_id), + ("@coord_task", os.environ.get("FLEET_AI_TASK", "")), + ): + subprocess.run( + ["tmux", "set-option", "-p", "-q", "-t", pane, key, value], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def subagent_identity(conversation_id: str, agent_id: str) -> tuple[str, str]: + """Derive the stable conversation/runtime pair shared by all hook processes.""" + conversation = f"{conversation_id}:agent:{agent_id}" + runtime = str(uuid.uuid5(uuid.NAMESPACE_URL, f"fleet-agent-coord:{conversation}")) + return conversation, runtime + + +def main() -> None: + event = json.loads(sys.stdin.read()) + conversation = event.get("session_id") or "" + if not conversation: + return + agent_id = str(event.get("agent_id") or "") + db = cli.store() + name = event.get("hook_event_name") or "" + requested = os.environ.get("COORD_INSTANCE_ID", "") + if agent_id: + conversation, requested = subagent_identity(conversation, agent_id) + harness = ( + "codex" + if os.environ.get("CODEX_THREAD_ID") + else "claude-code" + ) + + if name == "SessionEnd": + # Without a launcher-provided runtime UUID, conversation identity is + # insufficient: a stale process could otherwise end a newer resume. + if not requested: + return + try: + db.end_instance(db.resolve_runtime(requested), reason="clean") + except KeyError: + pass + return + + try: + # A direct, unwrapped harness cannot communicate a newly generated UUID + # to later hook processes. Reuse its one live conversation attachment; + # managed launchers always provide a fresh exact runtime. + if not requested: + runtime_id = db.resolve_runtime(conversation) + db.heartbeat( + runtime_id, + renew_lease_ttl_seconds=cli.storelib.DEFAULT_LEASE_TTL, + ) + else: + runtime_id = db.register_instance( + conversation, + runtime_id=requested, + harness=harness, + host=socket.gethostname(), + cwd=str(pathlib.Path(event.get("cwd") or os.getcwd()).resolve()), + label=os.environ.get("FLEET_AI_TASK", ""), + ) + except (KeyError, sqlite3.IntegrityError): + if not requested: + runtime_id = db.register_instance( + conversation, + harness=harness, + host=socket.gethostname(), + cwd=str(pathlib.Path(event.get("cwd") or os.getcwd()).resolve()), + label=os.environ.get("FLEET_AI_TASK", ""), + ) + else: + runtime_id = db.resolve_runtime(requested) + db.heartbeat( + runtime_id, + renew_lease_ttl_seconds=cli.storelib.DEFAULT_LEASE_TTL, + ) + annotate_tmux(runtime_id, conversation) + + peers = [item for item in db.live_instances() if item["runtime_id"] != runtime_id] + leases = [item for item in db.active_leases() if item["runtime_id"] != runtime_id] + pending = db.inbox(runtime_id, limit=5, mark_delivered=True) + if not peers and not pending: + return + lines: list[str] = [] + if peers: + lines.append(f"Live peer agents: {len(peers)}") + for peer in peers: + lines.append( + f" · {cli.runtime_handle(peer['runtime_id'])} " + f"{peer['harness']} task={cli.peer_label(peer)}" + ) + if leases: + lines.append("Active peer resource leases:") + for lease in leases: + lines.append( + f" · {lease['path']} [{lease['mode']}] by " + f"{cli.runtime_handle(lease['runtime_id'])} — {lease['purpose']}" + ) + if pending: + lines.append("Pending durable peer messages:") + for message in pending: + sender = cli.runtime_handle(message.sender_runtime_id) + if message.state == "delivered": + digest = hashlib.sha256(message.body.encode("utf-8")).hexdigest()[:10] + lines.append( + f" · [{message.message_id}] {sender} pending again " + f"({message.kind}, sha256:{digest}); `coord inbox --peek`" + ) + else: + lines.append( + f" · [{message.message_id}] {sender} " + f"({message.kind}): {message.body}" + ) + lines.append( + "Use an isolated managed clone for writable work; claim shared integration, " + "push, activation, and deployment resources before those operations." + ) + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "SessionStart", + "additionalContext": "\n".join(lines), + } + } + ) + ) + + +if __name__ == "__main__": + try: + main() + except Exception as exc: # lifecycle hooks must fail open + print(f"coord-session: failing open: {exc}", file=sys.stderr) + raise SystemExit(0) diff --git a/store.py b/store.py new file mode 100755 index 0000000..c07a37c --- /dev/null +++ b/store.py @@ -0,0 +1,1139 @@ +#!/usr/bin/env python3 +"""Transactional storage for inter-session coordination. + +This module is deliberately independent of the ``coord`` command so it can be +integrated incrementally by the CLI, lifecycle hook, and write-veto hook. + +The important identity distinction is: + +``conversation_id`` + The durable Claude/Codex conversation or thread identifier. A resumed + conversation retains this identifier. + +``runtime_id`` + A UUID for one live process attachment to that conversation. Every resume + registers a fresh runtime, so a stale process cannot impersonate the + resumed one or accidentally renew its leases. + +Lease keys are generic resource names, not necessarily filesystem paths. For +example, callers may claim ``integration/master``, ``push/origin/master``, +``activate/home/tpp15s``, or ``deploy/system/nix-control`` with the same atomic +semantics as ``tests/checks.nix``. + +All read/check/write decisions that must be atomic use ``BEGIN IMMEDIATE``. +SQLite serializes those transactions even when callers are separate processes. +WAL mode lets readers continue while a writer commits. Runtime state is local: +the containing directory is forced to 0700 and database files to 0600. +""" + +from __future__ import annotations + +import contextlib +import dataclasses +import hashlib +import hmac +import json +import os +import pathlib +import sqlite3 +import time +import uuid +from collections.abc import Callable, Iterable, Iterator, Sequence +from typing import Any, TextIO + + +SCHEMA_VERSION = 1 +DEFAULT_INSTANCE_TTL = 900.0 +DEFAULT_LEASE_TTL = 3600.0 +VALID_LEASE_MODES = frozenset(("exclusive", "append")) + +# Human-facing handles deliberately have much less entropy than the UUIDs they +# represent: they are exact, collision-checked aliases, not security tokens. +# Five 7-bit words provide a 35-bit display namespace while normally costing +# far fewer model tokens than a 36-character UUID. Full IDs remain canonical +# in SQLite, tmux metadata, hook environment variables, and audit exports. +HANDLE_WORDS = ( + "acorn", "alpine", "amber", "apple", "arrow", "ash", "atlas", "aurora", + "autumn", "badger", "bamboo", "bay", "beacon", "bear", "birch", "blue", + "breeze", "brook", "cedar", "cloud", "coast", "coral", "crane", "creek", + "dawn", "delta", "dune", "eagle", "earth", "ember", "fern", "field", + "finch", "flame", "flora", "fog", "forest", "fox", "frost", "garden", + "glade", "gold", "grove", "gull", "harbor", "hare", "hawk", "hazel", + "hill", "ice", "iris", "island", "ivy", "jade", "jay", "lake", + "lark", "leaf", "lemon", "light", "lily", "lion", "lotus", "maple", + "marsh", "meadow", "mint", "moon", "moss", "north", "oak", "ocean", + "olive", "orchid", "otter", "owl", "panda", "path", "peach", "pear", + "pine", "plum", "pond", "poppy", "rain", "raven", "reed", "ridge", + "river", "robin", "rock", "rose", "ruby", "sage", "sand", "seal", + "shadow", "shore", "silver", "sky", "snow", "south", "sparrow", "spring", + "star", "stone", "storm", "sun", "swift", "tide", "tiger", "trail", + "tree", "tulip", "valley", "violet", "wave", "west", "whale", "willow", + "wind", "wing", "winter", "wolf", "wood", "wren", "yarrow", "zenith", +) + + +def friendly_handle(value: str, namespace: str) -> str: + """Return a stable five-word exact alias for an opaque internal ID.""" + if not value: + raise ValueError("cannot derive a handle from an empty identity") + if len(HANDLE_WORDS) != 128 or len(set(HANDLE_WORDS)) != 128: + raise RuntimeError("friendly handle vocabulary must contain 128 unique words") + digest = hashlib.sha256(f"{namespace}\0{value}".encode("utf-8")).digest() + bits = int.from_bytes(digest[:5], "big") >> 5 + indices = [((bits >> shift) & 0x7F) for shift in (28, 21, 14, 7, 0)] + return "-".join(HANDLE_WORDS[index] for index in indices) + + +@dataclasses.dataclass(frozen=True) +class LeaseConflict: + path: str + runtime_id: str + mode: str + purpose: str + expires_at: float + + +class ClaimRefused(RuntimeError): + """An atomic lease claim conflicted; no path in the request was claimed.""" + + def __init__(self, conflicts: Sequence[LeaseConflict]): + self.conflicts = tuple(conflicts) + detail = ", ".join( + f"{item.path} ({item.mode} by {item.runtime_id})" for item in conflicts + ) + super().__init__(f"lease claim refused: {detail}") + + +@dataclasses.dataclass(frozen=True) +class Delivery: + message_id: int + sender_runtime_id: str + kind: str + body: str + refs: tuple[str, ...] + created_at: float + state: str + + +class CoordStore: + """Small process-safe coordination database. + + A store object holds no long-lived connection and is safe to share between + threads. Each operation opens its own connection. Independent processes may + construct their own objects pointed at the same database. + """ + + def __init__( + self, + db_path: str | os.PathLike[str], + *, + clock: Callable[[], float] = time.time, + timeout: float = 30.0, + ) -> None: + self.path = pathlib.Path(db_path) + self.clock = clock + self.timeout = timeout + self._secure_parent() + self._initialize() + + def _secure_parent(self) -> None: + self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(self.path.parent, 0o700) + + def _secure_database_files(self) -> None: + for suffix in ("", "-wal", "-shm"): + candidate = pathlib.Path(f"{self.path}{suffix}") + with contextlib.suppress(FileNotFoundError): + os.chmod(candidate, 0o600) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect( + self.path, + timeout=self.timeout, + isolation_level=None, + ) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("PRAGMA busy_timeout = %d" % int(self.timeout * 1000)) + self._secure_database_files() + return conn + + @contextlib.contextmanager + def _transaction(self) -> Iterator[sqlite3.Connection]: + conn = self._connect() + try: + conn.execute("BEGIN IMMEDIATE") + yield conn + conn.commit() + except BaseException: + conn.rollback() + raise + finally: + conn.close() + self._secure_database_files() + + def _initialize(self) -> None: + # journal_mode cannot be changed while a transaction is active. Set it + # once before creating the schema; the mode persists in the database. + conn = self._connect() + try: + mode = conn.execute("PRAGMA journal_mode = WAL").fetchone()[0] + if str(mode).lower() != "wal": + raise RuntimeError(f"could not enable SQLite WAL mode (got {mode!r})") + conn.execute("PRAGMA synchronous = FULL") + finally: + conn.close() + self._secure_database_files() + with self._transaction() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS metadata ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS instances ( + runtime_id TEXT PRIMARY KEY, + conversation_id TEXT NOT NULL, + harness TEXT NOT NULL, + host TEXT NOT NULL, + cwd TEXT NOT NULL, + label TEXT NOT NULL, + registered_at REAL NOT NULL, + heartbeat_at REAL NOT NULL, + heartbeat_expires_at REAL NOT NULL, + ended_at REAL + ); + CREATE INDEX IF NOT EXISTS instances_conversation + ON instances(conversation_id); + + CREATE TABLE IF NOT EXISTS workspace_records ( + instance_id TEXT PRIMARY KEY, + record_json TEXT NOT NULL, + updated_at REAL NOT NULL + ); + + CREATE TABLE IF NOT EXISTS leases ( + path TEXT NOT NULL, + runtime_id TEXT NOT NULL REFERENCES instances(runtime_id), + mode TEXT NOT NULL CHECK(mode IN ('exclusive', 'append')), + purpose TEXT NOT NULL, + claimed_at REAL NOT NULL, + renewed_at REAL NOT NULL, + expires_at REAL NOT NULL, + PRIMARY KEY(path, runtime_id) + ); + CREATE INDEX IF NOT EXISTS leases_expiry ON leases(expires_at); + + CREATE TABLE IF NOT EXISTS messages ( + message_id INTEGER PRIMARY KEY AUTOINCREMENT, + sender_runtime_id TEXT NOT NULL REFERENCES instances(runtime_id), + requested_recipient TEXT NOT NULL, + kind TEXT NOT NULL, + body TEXT NOT NULL, + refs_json TEXT NOT NULL, + created_at REAL NOT NULL + ); + + CREATE TABLE IF NOT EXISTS message_deliveries ( + message_id INTEGER NOT NULL REFERENCES messages(message_id), + recipient_runtime_id TEXT NOT NULL REFERENCES instances(runtime_id), + state TEXT NOT NULL CHECK(state IN ('queued', 'delivered', 'acked')), + delivered_at REAL, + acked_at REAL, + PRIMARY KEY(message_id, recipient_runtime_id) + ); + CREATE INDEX IF NOT EXISTS deliveries_inbox + ON message_deliveries(recipient_runtime_id, state, message_id); + + CREATE TABLE IF NOT EXISTS audit_events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + occurred_at REAL NOT NULL, + event TEXT NOT NULL, + runtime_id TEXT, + details_json TEXT NOT NULL + ); + """ + ) + row = conn.execute( + "SELECT value FROM metadata WHERE key = 'schema_version'" + ).fetchone() + if row is None: + conn.execute( + "INSERT INTO metadata(key, value) VALUES('schema_version', ?)", + (str(SCHEMA_VERSION),), + ) + elif int(row["value"]) != SCHEMA_VERSION: + raise RuntimeError( + f"unsupported coordination schema {row['value']}; " + f"expected {SCHEMA_VERSION}" + ) + + @staticmethod + def _normalize_paths(paths: Iterable[str]) -> tuple[str, ...]: + normalized = tuple( + dict.fromkeys(os.path.normpath(path.strip()) for path in paths if path.strip()) + ) + if not normalized: + raise ValueError("at least one non-empty path is required") + if any(path == "." or os.path.isabs(path) or path.startswith("../") for path in normalized): + raise ValueError("lease paths must be repository-relative and may not escape it") + return normalized + + @staticmethod + def resources_overlap(left: str, right: str) -> bool: + """Whether two exact resource names are equal or one contains the other.""" + return ( + left == right + or left.startswith(right.rstrip("/") + "/") + or right.startswith(left.rstrip("/") + "/") + ) + + @staticmethod + def _audit( + conn: sqlite3.Connection, + now: float, + event: str, + runtime_id: str | None, + details: dict[str, Any], + ) -> None: + conn.execute( + """ + INSERT INTO audit_events(occurred_at, event, runtime_id, details_json) + VALUES(?, ?, ?, ?) + """, + (now, event, runtime_id, json.dumps(details, sort_keys=True, separators=(",", ":"))), + ) + + @staticmethod + def _require_live( + conn: sqlite3.Connection, runtime_id: str, now: float + ) -> sqlite3.Row: + row = conn.execute( + """ + SELECT * FROM instances + WHERE runtime_id = ? AND ended_at IS NULL AND heartbeat_expires_at > ? + """, + (runtime_id, now), + ).fetchone() + if row is None: + raise KeyError(f"runtime instance is not registered and live: {runtime_id}") + return row + + @staticmethod + def _reap(conn: sqlite3.Connection, now: float) -> int: + cursor = conn.execute( + """ + DELETE FROM leases + WHERE expires_at <= ? + OR runtime_id IN ( + SELECT runtime_id FROM instances + WHERE ended_at IS NOT NULL OR heartbeat_expires_at <= ? + ) + """, + (now, now), + ) + return cursor.rowcount + + def register_instance( + self, + conversation_id: str, + *, + runtime_id: str | None = None, + harness: str = "", + host: str = "", + cwd: str = "", + label: str = "", + ttl_seconds: float = DEFAULT_INSTANCE_TTL, + ) -> str: + if not conversation_id: + raise ValueError("conversation_id is required") + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + runtime_id = runtime_id or str(uuid.uuid4()) + if runtime_id == conversation_id: + raise ValueError("runtime_id must be distinct from conversation_id") + now = self.clock() + with self._transaction() as conn: + conn.execute( + """ + INSERT INTO instances( + runtime_id, conversation_id, harness, host, cwd, label, + registered_at, heartbeat_at, heartbeat_expires_at, ended_at + ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, NULL) + """, + ( + runtime_id, + conversation_id, + harness, + host, + cwd, + label, + now, + now, + now + ttl_seconds, + ), + ) + # Carry undelivered work across a genuine resume. Only dead/expired + # attachments of the same conversation are migrated; concurrently + # live forks retain independent inboxes. + stale_deliveries = conn.execute( + """ + SELECT d.message_id, d.recipient_runtime_id + FROM message_deliveries AS d + JOIN instances AS old ON old.runtime_id = d.recipient_runtime_id + WHERE old.conversation_id = ? AND old.runtime_id != ? + AND (old.ended_at IS NOT NULL OR old.heartbeat_expires_at <= ?) + AND d.state != 'acked' + ORDER BY d.message_id, d.recipient_runtime_id + """, + (conversation_id, runtime_id, now), + ).fetchall() + for delivery in stale_deliveries: + conn.execute( + """ + INSERT INTO message_deliveries( + message_id, recipient_runtime_id, state + ) VALUES(?, ?, 'queued') + ON CONFLICT(message_id, recipient_runtime_id) DO NOTHING + """, + (delivery["message_id"], runtime_id), + ) + conn.execute( + """ + DELETE FROM message_deliveries + WHERE message_id = ? AND recipient_runtime_id = ? + """, + (delivery["message_id"], delivery["recipient_runtime_id"]), + ) + self._audit( + conn, + now, + "instance.registered", + runtime_id, + { + "conversation_id": conversation_id, + "ttl_seconds": ttl_seconds, + "deliveries_migrated": len(stale_deliveries), + }, + ) + return runtime_id + + def heartbeat( + self, + runtime_id: str, + *, + ttl_seconds: float = DEFAULT_INSTANCE_TTL, + renew_lease_ttl_seconds: float | None = None, + ) -> None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + if renew_lease_ttl_seconds is not None and renew_lease_ttl_seconds <= 0: + raise ValueError("renew_lease_ttl_seconds must be positive") + now = self.clock() + with self._transaction() as conn: + row = conn.execute( + """ + SELECT ended_at, heartbeat_expires_at + FROM instances WHERE runtime_id = ? + """, + (runtime_id,), + ).fetchone() + if ( + row is None + or row["ended_at"] is not None + or row["heartbeat_expires_at"] <= now + ): + raise KeyError(f"runtime instance is not active: {runtime_id}") + conn.execute( + """ + UPDATE instances + SET heartbeat_at = ?, heartbeat_expires_at = ? + WHERE runtime_id = ? + """, + (now, now + ttl_seconds, runtime_id), + ) + renewed = 0 + if renew_lease_ttl_seconds is not None: + renewed = conn.execute( + """ + UPDATE leases SET renewed_at = ?, expires_at = ? + WHERE runtime_id = ? AND expires_at > ? + """, + (now, now + renew_lease_ttl_seconds, runtime_id, now), + ).rowcount + self._audit( + conn, + now, + "instance.heartbeat", + runtime_id, + {"ttl_seconds": ttl_seconds, "leases_renewed": renewed}, + ) + + def end_instance(self, runtime_id: str, *, reason: str = "clean") -> None: + now = self.clock() + with self._transaction() as conn: + cursor = conn.execute( + """ + UPDATE instances SET ended_at = ? + WHERE runtime_id = ? AND ended_at IS NULL + """, + (now, runtime_id), + ) + if cursor.rowcount != 1: + raise KeyError(f"runtime instance is not active: {runtime_id}") + released = conn.execute( + "DELETE FROM leases WHERE runtime_id = ?", (runtime_id,) + ).rowcount + self._audit( + conn, + now, + "instance.ended", + runtime_id, + {"reason": reason, "leases_released": released}, + ) + + def claim( + self, + runtime_id: str, + paths: Iterable[str], + *, + mode: str, + purpose: str = "", + ttl_seconds: float = DEFAULT_LEASE_TTL, + ) -> None: + paths = self._normalize_paths(paths) + if mode not in VALID_LEASE_MODES: + raise ValueError(f"invalid lease mode: {mode}") + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, runtime_id, now) + self._reap(conn, now) + rows = conn.execute( + """ + SELECT path, runtime_id, mode, purpose, expires_at + FROM leases + WHERE runtime_id != ? + AND (? = 'exclusive' OR mode = 'exclusive') + ORDER BY path, runtime_id + """, + (runtime_id, mode), + ).fetchall() + conflicts = [ + row + for row in rows + if any(self.resources_overlap(path, row["path"]) for path in paths) + ] + if conflicts: + raise ClaimRefused( + [ + LeaseConflict( + row["path"], + row["runtime_id"], + row["mode"], + row["purpose"], + row["expires_at"], + ) + for row in conflicts + ] + ) + for path in paths: + conn.execute( + """ + INSERT INTO leases( + path, runtime_id, mode, purpose, + claimed_at, renewed_at, expires_at + ) VALUES(?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(path, runtime_id) DO UPDATE SET + mode = excluded.mode, + purpose = excluded.purpose, + renewed_at = excluded.renewed_at, + expires_at = excluded.expires_at + """, + (path, runtime_id, mode, purpose, now, now, now + ttl_seconds), + ) + self._audit( + conn, + now, + "lease.claimed", + runtime_id, + { + "paths": paths, + "mode": mode, + "purpose": purpose, + "ttl_seconds": ttl_seconds, + }, + ) + + def release(self, runtime_id: str, paths: Iterable[str]) -> tuple[str, ...]: + paths = self._normalize_paths(paths) + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, runtime_id, now) + placeholders = ",".join("?" for _ in paths) + rows = conn.execute( + f""" + SELECT path FROM leases + WHERE runtime_id = ? AND path IN ({placeholders}) + ORDER BY path + """, + (runtime_id, *paths), + ).fetchall() + released = tuple(row["path"] for row in rows) + conn.execute( + f""" + DELETE FROM leases + WHERE runtime_id = ? AND path IN ({placeholders}) + """, + (runtime_id, *paths), + ) + self._audit( + conn, now, "lease.released", runtime_id, {"paths": released} + ) + return released + + def renew_leases( + self, + runtime_id: str, + *, + paths: Iterable[str] | None = None, + ttl_seconds: float = DEFAULT_LEASE_TTL, + ) -> tuple[str, ...]: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + normalized = self._normalize_paths(paths) if paths is not None else None + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, runtime_id, now) + self._reap(conn, now) + parameters: list[Any] = [now, now + ttl_seconds, runtime_id] + suffix = "" + if normalized is not None: + suffix = " AND path IN (%s)" % ",".join("?" for _ in normalized) + parameters.extend(normalized) + conn.execute( + f""" + UPDATE leases SET renewed_at = ?, expires_at = ? + WHERE runtime_id = ?{suffix} + """, + parameters, + ) + rows = conn.execute( + f""" + SELECT path FROM leases + WHERE runtime_id = ?{suffix} + ORDER BY path + """, + (runtime_id, *(normalized or ())), + ).fetchall() + renewed = tuple(row["path"] for row in rows) + self._audit( + conn, + now, + "lease.renewed", + runtime_id, + {"paths": renewed, "ttl_seconds": ttl_seconds}, + ) + return renewed + + # Resource-named aliases make it explicit that leases cover global operations + # and external systems as well as repository files. + def claim_resources( + self, + runtime_id: str, + resources: Iterable[str], + *, + mode: str, + purpose: str = "", + ttl_seconds: float = DEFAULT_LEASE_TTL, + ) -> None: + self.claim( + runtime_id, + resources, + mode=mode, + purpose=purpose, + ttl_seconds=ttl_seconds, + ) + + def release_resources( + self, runtime_id: str, resources: Iterable[str] + ) -> tuple[str, ...]: + return self.release(runtime_id, resources) + + def renew_resources( + self, + runtime_id: str, + *, + resources: Iterable[str] | None = None, + ttl_seconds: float = DEFAULT_LEASE_TTL, + ) -> tuple[str, ...]: + return self.renew_leases( + runtime_id, paths=resources, ttl_seconds=ttl_seconds + ) + + def active_leases(self) -> list[dict[str, Any]]: + now = self.clock() + with self._transaction() as conn: + self._reap(conn, now) + return [ + dict(row) + for row in conn.execute( + """ + SELECT l.* FROM leases AS l + JOIN instances AS i USING(runtime_id) + WHERE l.expires_at > ? AND i.ended_at IS NULL + AND i.heartbeat_expires_at > ? + ORDER BY l.path, l.runtime_id + """, + (now, now), + ) + ] + + def live_instances(self) -> list[dict[str, Any]]: + """Return live runtime attachments with their full conversation IDs.""" + now = self.clock() + conn = self._connect() + try: + return [ + dict(row) + for row in conn.execute( + """ + SELECT * FROM instances + WHERE ended_at IS NULL AND heartbeat_expires_at > ? + ORDER BY registered_at, runtime_id + """, + (now,), + ) + ] + finally: + conn.close() + + def resolve_runtime(self, identity: str) -> str: + """Resolve an exact runtime, friendly handle, or live conversation. + + Prefix matching is intentionally forbidden. Friendly handles are + derived from the full runtime ID and must match exactly; the extremely + unlikely ambiguous handle is refused rather than guessed. + """ + now = self.clock() + conn = self._connect() + try: + exact = conn.execute( + """ + SELECT runtime_id FROM instances + WHERE runtime_id = ? AND ended_at IS NULL + AND heartbeat_expires_at > ? + """, + (identity, now), + ).fetchone() + if exact is not None: + return str(exact["runtime_id"]) + handle_matches = [ + str(row["runtime_id"]) + for row in conn.execute( + """ + SELECT runtime_id FROM instances + WHERE ended_at IS NULL AND heartbeat_expires_at > ? + ORDER BY runtime_id + """, + (now,), + ) + if hmac.compare_digest( + friendly_handle(str(row["runtime_id"]), "runtime"), identity + ) + ] + if len(handle_matches) == 1: + return handle_matches[0] + if len(handle_matches) > 1: + raise KeyError( + f"ambiguous runtime handle: {identity}; use a full runtime ID" + ) + latest = conn.execute( + """ + SELECT runtime_id FROM instances + WHERE conversation_id = ? AND ended_at IS NULL + AND heartbeat_expires_at > ? + ORDER BY registered_at DESC, runtime_id DESC + LIMIT 1 + """, + (identity, now), + ).fetchone() + if latest is None: + raise KeyError(f"no live runtime for exact identity: {identity}") + return str(latest["runtime_id"]) + finally: + conn.close() + + def save_workspace_record( + self, + event: str, + record: dict[str, Any], + ) -> None: + """Transactionally retain the latest full lifecycle record.""" + instance_id = str(record.get("instance_id", "")) + if not instance_id: + raise ValueError("workspace record requires instance_id") + now = self.clock() + payload = json.dumps(record, sort_keys=True, separators=(",", ":")) + with self._transaction() as conn: + conn.execute( + """ + INSERT INTO workspace_records(instance_id, record_json, updated_at) + VALUES(?, ?, ?) + ON CONFLICT(instance_id) DO UPDATE SET + record_json = excluded.record_json, + updated_at = excluded.updated_at + """, + (instance_id, payload, now), + ) + self._audit( + conn, + now, + event, + str(record.get("runtime_id") or "") or None, + {"instance_id": instance_id, "record": record}, + ) + + def workspace_record(self, instance_id: str) -> dict[str, Any]: + conn = self._connect() + try: + row = conn.execute( + "SELECT record_json FROM workspace_records WHERE instance_id = ?", + (instance_id,), + ).fetchone() + if row is not None: + return dict(json.loads(row["record_json"])) + matches = [] + for candidate in conn.execute( + "SELECT instance_id, record_json FROM workspace_records ORDER BY instance_id" + ): + if hmac.compare_digest( + friendly_handle(str(candidate["instance_id"]), "work"), + instance_id, + ): + matches.append(dict(json.loads(candidate["record_json"]))) + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + raise KeyError( + f"ambiguous workspace handle: {instance_id}; use a full instance ID" + ) + raise KeyError(f"unknown workspace instance: {instance_id}") + finally: + conn.close() + + def workspace_records(self) -> list[dict[str, Any]]: + conn = self._connect() + try: + return [ + dict(json.loads(row["record_json"])) + for row in conn.execute( + "SELECT record_json FROM workspace_records ORDER BY updated_at, instance_id" + ) + ] + finally: + conn.close() + + def is_managed_workspace(self, path: str | os.PathLike[str]) -> bool: + candidate = pathlib.Path(path).resolve() + for record in self.workspace_records(): + actual = record.get("actual_path") + if ( + not actual + or record.get("lifecycle_state") != "active" + or pathlib.Path(str(actual)).resolve() != candidate + ): + continue + try: + marker = json.loads( + (candidate / ".jj" / "fleet-coord-owner.json").read_text( + encoding="utf-8" + ) + ) + except (FileNotFoundError, OSError, json.JSONDecodeError): + return False + return ( + marker.get("instance_id") == record.get("instance_id") + and hmac.compare_digest( + str(marker.get("ownership_token", "")), + str(record.get("ownership_token", "")), + ) + ) + return False + + def record_event( + self, + event: str, + *, + runtime_id: str | None = None, + details: dict[str, Any] | None = None, + ) -> int: + """Append a generic structured event for integrations layered above this API.""" + if not event: + raise ValueError("event is required") + now = self.clock() + with self._transaction() as conn: + if runtime_id is not None: + self._require_live(conn, runtime_id, now) + self._audit(conn, now, event, runtime_id, details or {}) + return int(conn.execute("SELECT last_insert_rowid()").fetchone()[0]) + + def send_message( + self, + sender_runtime_id: str, + *, + recipient: str | Iterable[str] = "*", + kind: str = "fyi", + body: str, + refs: Iterable[str] = (), + ) -> int: + if not body: + raise ValueError("message body is required") + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, sender_runtime_id, now) + if recipient == "*": + requested = "*" + recipients = [ + row["runtime_id"] + for row in conn.execute( + """ + SELECT runtime_id FROM instances + WHERE runtime_id != ? AND ended_at IS NULL + AND heartbeat_expires_at > ? + ORDER BY runtime_id + """, + (sender_runtime_id, now), + ) + ] + else: + recipients = ( + (recipient,) + if isinstance(recipient, str) + else tuple(dict.fromkeys(recipient)) + ) + requested = ",".join(recipients) + if not recipients: + raise ValueError("at least one recipient is required") + placeholders = ",".join("?" for _ in recipients) + found = { + row["runtime_id"] + for row in conn.execute( + f""" + SELECT runtime_id FROM instances + WHERE runtime_id IN ({placeholders}) + AND ended_at IS NULL AND heartbeat_expires_at > ? + """, + (*recipients, now), + ) + } + missing = set(recipients) - found + if missing: + raise KeyError(f"recipient runtime is not live: {sorted(missing)}") + cursor = conn.execute( + """ + INSERT INTO messages( + sender_runtime_id, requested_recipient, kind, + body, refs_json, created_at + ) VALUES(?, ?, ?, ?, ?, ?) + """, + ( + sender_runtime_id, + requested, + kind, + body, + json.dumps(tuple(refs), separators=(",", ":")), + now, + ), + ) + message_id = int(cursor.lastrowid) + conn.executemany( + """ + INSERT INTO message_deliveries( + message_id, recipient_runtime_id, state + ) VALUES(?, ?, 'queued') + """, + ((message_id, target) for target in recipients), + ) + self._audit( + conn, + now, + "message.queued", + sender_runtime_id, + { + "message_id": message_id, + "requested_recipient": requested, + "recipient_snapshot": recipients, + }, + ) + return message_id + + def inbox( + self, + runtime_id: str, + *, + limit: int = 100, + mark_delivered: bool = True, + ) -> list[Delivery]: + if limit <= 0: + return [] + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, runtime_id, now) + rows = conn.execute( + """ + SELECT m.*, d.state + FROM message_deliveries AS d + JOIN messages AS m USING(message_id) + WHERE d.recipient_runtime_id = ? AND d.state != 'acked' + ORDER BY m.message_id + LIMIT ? + """, + (runtime_id, limit), + ).fetchall() + ids = [row["message_id"] for row in rows] + if mark_delivered and ids: + placeholders = ",".join("?" for _ in ids) + conn.execute( + f""" + UPDATE message_deliveries + SET state = 'delivered', delivered_at = COALESCE(delivered_at, ?) + WHERE recipient_runtime_id = ? + AND message_id IN ({placeholders}) AND state = 'queued' + """, + (now, runtime_id, *ids), + ) + self._audit( + conn, + now, + "message.delivered", + runtime_id, + {"message_ids": ids}, + ) + return [ + Delivery( + message_id=row["message_id"], + sender_runtime_id=row["sender_runtime_id"], + kind=row["kind"], + body=row["body"], + refs=tuple(json.loads(row["refs_json"])), + created_at=row["created_at"], + # Report the state observed by this read. This transaction + # updates queued rows after reading them, while the hook + # needs to distinguish first delivery from a reminder. + state=row["state"], + ) + for row in rows + ] + + def ack(self, runtime_id: str, message_ids: Iterable[int]) -> tuple[int, ...]: + ids = tuple(dict.fromkeys(int(item) for item in message_ids)) + if not ids: + return () + now = self.clock() + with self._transaction() as conn: + self._require_live(conn, runtime_id, now) + placeholders = ",".join("?" for _ in ids) + rows = conn.execute( + f""" + SELECT message_id FROM message_deliveries + WHERE recipient_runtime_id = ? + AND message_id IN ({placeholders}) AND state != 'acked' + ORDER BY message_id + """, + (runtime_id, *ids), + ).fetchall() + acked = tuple(row["message_id"] for row in rows) + conn.execute( + f""" + UPDATE message_deliveries + SET state = 'acked', acked_at = ? + WHERE recipient_runtime_id = ? + AND message_id IN ({placeholders}) AND state != 'acked' + """, + (now, runtime_id, *ids), + ) + self._audit( + conn, now, "message.acked", runtime_id, {"message_ids": acked} + ) + return acked + + def delivery_state(self, message_id: int, runtime_id: str) -> str | None: + conn = self._connect() + try: + row = conn.execute( + """ + SELECT state FROM message_deliveries + WHERE message_id = ? AND recipient_runtime_id = ? + """, + (message_id, runtime_id), + ).fetchone() + return None if row is None else str(row["state"]) + finally: + conn.close() + + def audit_events(self, *, after_id: int = 0) -> list[dict[str, Any]]: + conn = self._connect() + try: + rows = conn.execute( + """ + SELECT * FROM audit_events + WHERE event_id > ? ORDER BY event_id + """, + (after_id,), + ).fetchall() + return [ + { + "event_id": row["event_id"], + "occurred_at": row["occurred_at"], + "event": row["event"], + "runtime_id": row["runtime_id"], + "details": json.loads(row["details_json"]), + } + for row in rows + ] + finally: + conn.close() + + def export_audit_jsonl( + self, + destination: str | os.PathLike[str] | TextIO, + *, + after_id: int = 0, + include_secrets: bool = False, + ) -> int: + def without_secrets(value: Any) -> Any: + if isinstance(value, dict): + return { + key: without_secrets(item) + for key, item in value.items() + if key != "ownership_token" + } + if isinstance(value, list): + return [without_secrets(item) for item in value] + return value + + events = self.audit_events(after_id=after_id) + close = False + if hasattr(destination, "write"): + stream = destination + else: + target = pathlib.Path(destination) + target.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + os.fchmod(fd, 0o600) + stream = os.fdopen(fd, "w", encoding="utf-8") + close = True + try: + for event in events: + exported = event if include_secrets else without_secrets(event) + stream.write( + json.dumps(exported, sort_keys=True, separators=(",", ":")) + + "\n" + ) + finally: + if close: + stream.close() + return len(events) diff --git a/tests/test_install.py b/tests/test_install.py new file mode 100755 index 0000000..b1aad01 --- /dev/null +++ b/tests/test_install.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Behavioral tests for fresh install and legacy in-place migration.""" + +from __future__ import annotations + +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +INSTALL = ROOT / "install" + + +class InstallBehavior(unittest.TestCase): + def run_install( + self, target: Path, *args: str + ) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(INSTALL), str(target), *args], + cwd=ROOT, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + + def test_fresh_install_requires_explicit_immutable_project_id(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + target = Path(temporary) + missing = self.run_install(target) + self.assertEqual(missing.returncode, 2) + self.assertIn("--project-id", missing.stderr) + installed = self.run_install(target, "--project-id", "fresh-project") + self.assertEqual(installed.returncode, 0, installed.stderr) + self.assertEqual( + (target / ".coord/project-id").read_text().strip(), + "fresh-project", + ) + checked = self.run_install(target, "--check") + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + + def test_legacy_upgrade_preserves_hooks_identity_and_both_entrypoints(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + target = Path(temporary) + legacy = target / ".claude/coord" + legacy.mkdir(parents=True) + (legacy / "project-id").write_text("legacy-project\n") + (target / ".claude/settings.json").write_text( + json.dumps( + { + "effortLevel": "high", + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [ + { + "type": "command", + "command": "printf unrelated", + "timeout": 2, + }, + { + "type": "command", + "command": "python3 .claude/coord/hook.py", + "timeout": 10, + }, + ], + } + ] + }, + } + ) + ) + (target / ".codex").mkdir() + (target / ".codex/hooks.json").write_text( + json.dumps( + { + "hooks": { + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 .claude/coord/session.py", + "timeout": 10, + } + ] + } + ] + } + } + ) + ) + + installed = self.run_install(target) + self.assertEqual(installed.returncode, 0, installed.stderr) + self.assertEqual( + (target / ".coord/project-id").read_text().strip(), + "legacy-project", + ) + self.assertEqual( + (target / ".claude/coord/project-id").read_text().strip(), + "legacy-project", + ) + settings = json.loads((target / ".claude/settings.json").read_text()) + self.assertEqual(settings["effortLevel"], "high") + commands = [ + hook["command"] + for group in settings["hooks"]["PreToolUse"] + for hook in group["hooks"] + ] + self.assertIn("printf unrelated", commands) + self.assertEqual( + sum("/.coord/hook.py" in command for command in commands), 1 + ) + self.assertFalse(any(".claude/coord/" in command for command in commands)) + + canonical = subprocess.run( + [str(target / ".coord/coord"), "--help"], + cwd="/", + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + compatibility = subprocess.run( + [str(target / ".claude/coord/coord"), "--help"], + cwd="/", + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + self.assertEqual(canonical.returncode, 0, canonical.stderr) + self.assertEqual(compatibility.returncode, canonical.returncode) + self.assertEqual(compatibility.stdout, canonical.stdout) + + store_path = target / ".claude/coord/store.py" + spec = importlib.util.spec_from_file_location("legacy_store_test", store_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + self.assertEqual( + module.friendly_handle("identity", "runtime").count("-"), 4 + ) + + conflicting = self.run_install( + target, "--project-id", "different-project" + ) + self.assertEqual(conflicting.returncode, 2) + self.assertIn("refusing project ID change", conflicting.stderr) + + (target / ".coord/coord").write_text("# drift\n") + drift = self.run_install(target, "--check") + self.assertEqual(drift.returncode, 1) + self.assertIn("drift", drift.stdout) + repaired = self.run_install(target) + self.assertEqual(repaired.returncode, 0, repaired.stderr) + final = self.run_install(target, "--check") + self.assertEqual(final.returncode, 0, final.stdout + final.stderr) + + def test_installed_cli_smokes_hello_claim_and_work_create(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = root / "project" + subprocess.run( + ["jj", "git", "init", "--colocate", str(target)], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + subprocess.run( + ["jj", "-R", str(target), "bookmark", "create", "master", "-r", "@"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + installed = self.run_install( + target, "--project-id", "smoke-project" + ) + self.assertEqual(installed.returncode, 0, installed.stderr) + cli = target / ".coord/coord" + env = { + **os.environ, + "COORD_DB": str(root / "state/coord.sqlite3"), + "COORD_PROJECT_ID": "smoke-project", + } + hello = subprocess.run( + [ + str(cli), + "hello", + "--conversation", + "smoke-conversation", + "--harness", + "test", + "--cwd", + str(target), + ], + cwd=target, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + self.assertEqual(hello.returncode, 0, hello.stderr) + runtime = hello.stdout.strip() + claimed = subprocess.run( + [str(cli), "-i", runtime, "claim", "smoke/resource"], + cwd="/", + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + self.assertEqual(claimed.returncode, 0, claimed.stderr) + created = subprocess.run( + [ + str(target / ".claude/coord/coord"), + "-i", + runtime, + "work", + "create", + "smoke-work", + "--source", + str(target), + "--scratch-root", + str(root / "scratch"), + "--json", + ], + cwd="/", + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + self.assertEqual(created.returncode, 0, created.stderr) + record = json.loads(created.stdout) + self.assertEqual(record["task"], "smoke-work") + self.assertTrue(Path(record["actual_path"]).is_dir()) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/workspace.py b/workspace.py new file mode 100755 index 0000000..d9d29db --- /dev/null +++ b/workspace.py @@ -0,0 +1,1186 @@ +#!/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 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 _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": "", + "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, + ) + _emit(registry, "workspace-planned", record) + + try: + if native_workspace: + workspace_name = f"agent-{instance_id[:12]}" + _run( + [ + jj, + "--ignore-working-copy", + "-R", + str(source_path), + "workspace", + "add", + "--name", + workspace_name, + "-r", + base_revision, + str(dest), + ], + runner=runner, + ) + record["jj_workspace_name"] = workspace_name + 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) + + _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" + ) + 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 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") + path = Path(str(record.get("actual_path") or "")).resolve() + source_commit = str((record.get("change") or {}).get("commit_id", "")) + if not source_commit: + raise WorkspaceError("workspace record has no exact source commit") + target = _jj_ids(path, target_revision, jj=jj, runner=runner) + result = _run( + [ + jj, + "--ignore-working-copy", + "-R", + str(path), + "log", + "--no-graph", + "-r", + f"{source_commit} & ::{target['commit_id']}", + "-T", + 'commit_id ++ "\\n"', + ], + runner=runner, + ) + observed = [line.strip().lower() for line in result.stdout.splitlines() if line.strip()] + if observed != [source_commit.lower()]: + 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"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)