devtools/coord
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: harden concurrent jj workspace handoff

Browse source
This commit is contained in:
sid 2026-08-08 13:30:59 -06:00
parent c118c433c6
commit d408119701
11 changed files with 366 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

1
AGENTS.md
View file

@ -26,6 +26,7 @@ jj st
- Preserve unrelated hooks, skills, settings, `AGENTS.md`, and `CLAUDE.md` in target repositories.
- Keep runtime SQLite state outside repositories and preserve immutable project IDs.
- Use independent `jj git clone --colocate` instances for concurrent writable work.
- Keep agent bookmarks namespaced; only the integration owner moves shared bookmarks.
- Treat full IDs as authority; short names and five-word handles are presentation/resolution aids.
- Never guess during workspace cleanup or target tmux objects by their names.

8
CLAUDE.md
View file

@ -62,8 +62,9 @@ Preserve these properties in every change:
7. Writes use a complete write loop, fsync, and atomic replacement. Do not leave partial files.
8. Full opaque IDs remain canonical evidence. Exact five-word handles may resolve identities;
prefixes and display-only shortened VCS IDs never carry authority.
9. Managed writable work defaults to independent colocated `jj git clone` instances. A native
`jj workspace` is opt-in because it shares an operation log and object store.
9. Managed writable work defaults to independent colocated `jj git clone` instances. Each gets a
unique jj workspace name, described working-copy change, and namespaced agent bookmark. A
native `jj workspace` is opt-in because it shares an operation log and object store.
10. Workspace deletion never guesses. Require registry/on-disk ownership agreement, clean state,
gate evidence, integration evidence where needed, and exact confirmation.
11. Tmux names are presentation only. Create concise `<task>-<sid4>` names, but inspect and mutate
@ -74,7 +75,8 @@ Preserve these properties in every change:
conflicting write fails closed. Do not claim shell-command parsing or security isolation the
hook does not provide.
14. Publishing, integration, activation, deployment, and other external side effects require
explicit resource scopes and authoritative before/after evidence. Never force.
explicit resource scopes and authoritative before/after evidence. Acquisition snapshots and
freezes the exact candidate commit; only the recorded owner may release it. Never force.
## Change workflow

18
DESIGN.md
View file

@ -15,6 +15,13 @@ independent operation logs, working-copy commits and Git backends, and ordinary
flake discovery works in them. The canonical checkout is for read-only inspection and one
explicit integration owner.
Every managed checkout receives a unique `agent-<full-instance-id>` jj workspace name, a
namespaced `agent/<task>/<full-instance-id>` bookmark, and a working-copy description carrying
the full coordinator instance ID. These make parallel heads legible and handoffs explicit; they
remain presentation/navigation metadata, while stored full change and commit IDs are authority.
Agents do not move shared bookmarks. The integration owner alone does so under an exact resource
lease.
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
@ -54,6 +61,11 @@ This database is deliberately host-local. It coordinates Claude and Codex proces
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).
Filesystem isolation also does not isolate external mutable state. Concurrent agents must use
distinct service ports, database/schema names, containers, and temporary/output directories when
their tools would otherwise share them. Dependency download caches may be shared only when their
cache protocol is concurrency-safe; mutable build trees must remain workspace-local.
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:
@ -148,6 +160,12 @@ The current “queue” is an integration lock plus an auditable state machine,
enqueue order and recorded dependencies are visible but do not automatically grant the next turn.
The integration owner adjudicates readiness and stale bases.
On acquisition, coord snapshots the isolated checkout and freezes its exact change and commit IDs
as the integration candidate. Release records the same full runtime and resource that acquired the
lock. Integration proof is rejected without matching owner/resource evidence and proves the frozen
candidate—not a stale create-time record—is an ancestor of the exact target. Cleanup also refuses
a workspace changed after that candidate was integrated.
- path must remain below the recorded scratch root;
- private registry token and on-disk `.jj` marker must agree;
- working copy must be clean;

10
README.md
View file

@ -127,6 +127,16 @@ operations:
.coord/coord work inspect <work-handle>
```
Each managed clone has a unique jj workspace name, a namespaced `agent/...` bookmark, and a
described working-copy change. Agents may rewrite their own change but must not move shared
bookmarks such as `master`; one integration owner acquires `integration/<bookmark>`, handles the
frozen candidate commit, and records the exact target.
Separate checkouts isolate repository files and build trees, not external services. Assign
workspace-specific ports, database/schema names, containers, and temporary/output directories
when concurrent agents run mutable services. Share dependency caches only when the cache itself is
safe for concurrent writers.
Managed windows have short human-readable names, while mutations target immutable tmux IDs.
`work close` safely closes the recorded window without dropping the operator to an outer shell.
`work remove` is conservative: inspect its refusal, integration, gate, and confirmation

2
VERSION
View file

@ -1 +1 @@
0.1.0
0.2.0

10
coord
View file

@ -1008,6 +1008,13 @@ def cmd_work_queue(args: argparse.Namespace) -> int:
)
if record.get("integration_resource") != args.resource:
raise ValueError("release resource does not match acquisition evidence")
elif args.action == "acquire":
# Snapshot this isolated checkout now. Integration must be tied to the
# exact commit observed at acquisition, not the initial create record
# or a possibly stale prior inspection.
record = worklib.inspect_workspace(
record, registry=workspace_registry(db)
)
def queue(action: str, _record: dict[str, Any]) -> None:
if action == "acquire":
@ -1032,6 +1039,9 @@ def cmd_work_queue(args: argparse.Namespace) -> int:
registry=workspace_registry(db),
)
if args.action == "acquire":
updated = worklib.freeze_integration_candidate(
updated, registry=workspace_registry(db)
)
updated["integration_owner_runtime"] = owner
updated["integration_resource"] = args.resource
elif args.action == "release":

8
skills/coord/SKILL.md
View file

@ -12,6 +12,8 @@ shell; use `-i <exact-five-word-handle>` only for manual recovery or selection.
- For ordinary writable implementation, create an isolated clone with
`.coord/coord work create <short-task>`. Work inside the returned path.
- Keep work on its generated `agent/...` bookmark. Do not move shared bookmarks; the integration
owner handles those under an exact resource lease.
- For a shared external boundary, claim an explicit resource such as `integration/master`,
`push/origin/master`, or a deployment target.
- Claim files in a shared checkout only when isolation is unavailable. Use `--mode append` solely
@ -61,6 +63,9 @@ Use exact five-word work handles returned by the CLI:
Use `work close`, not raw pane/window killing, for a managed agent. Window names are human labels;
the coordinator mutates only recorded immutable tmux IDs.
The checkout isolates files, not ports, databases, containers, or other external mutable state.
Give concurrent agents distinct instances of those resources and keep build outputs local.
Before removal, inspect the work record. Never bypass dirty-tree, ownership, gate, or integration
refusals. Use `work adopt` only for a known independent clone with explicit recovery evidence.
@ -68,7 +73,8 @@ refusals. Use `work adopt` only for a known independent clone with explicit reco
Acquire the exact resource before integration, push, activation, deployment, or another external
mutation. Verify the authoritative remote or target before and after the action. Never force-push.
Lock release alone is not integration proof; record the exact integrated target with
Acquisition freezes the inspected candidate commit. Lock release alone is not integration proof;
record the exact integrated target with
`work queue <handle> integrate --target <revision>`.
For architecture, failure modes, and recovery rules, read the repository’s `.coord/DESIGN.md`.

2
skills/coord/agents/openai.yaml
View file

@ -1,4 +1,4 @@
interface:
display_name: "Agent Coordinator"
short_description: "Coordinate agents, leases, workspaces, and tmux"
short_description: "Isolate agent work and coordinate safe integration"
default_prompt: "Use $coord to organize concurrent agent work safely."

82
tests/test_install.py
View file

@ -344,6 +344,88 @@ class InstallBehavior(unittest.TestCase):
record = json.loads(created.stdout)
self.assertEqual(record["task"], "smoke-work")
self.assertTrue(Path(record["actual_path"]).is_dir())
self.assertEqual(
record["jj_workspace_name"], f"agent-{record['instance_id']}"
)
self.assertEqual(
record["jj_bookmark"],
f"agent/smoke-work/{record['instance_id']}",
)
workspace_list = subprocess.run(
["jj", "-R", record["actual_path"], "workspace", "list"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
)
self.assertIn(record["jj_workspace_name"], workspace_list.stdout)
described = subprocess.run(
[
"jj",
"-R",
record["actual_path"],
"log",
"--no-graph",
"-r",
record["jj_bookmark"],
"-T",
'description.first_line() ++ "\\n"',
],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
)
self.assertEqual(
described.stdout.strip(), record["jj_change_description"]
)
(Path(record["actual_path"]) / "agent-change.txt").write_text(
"frozen candidate\n", encoding="utf-8"
)
for action in ("enqueue", "acquire", "release"):
queued = subprocess.run(
[
str(cli),
"-i",
runtime,
"work",
"queue",
record["instance_id"],
action,
],
cwd=target,
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
self.assertEqual(queued.returncode, 0, queued.stderr)
inspected = subprocess.run(
[
str(cli),
"-i",
runtime,
"work",
"inspect",
record["instance_id"],
"--json",
],
cwd=target,
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
)
acquired = json.loads(inspected.stdout)
self.assertEqual(
acquired["integration_candidate"]["commit_id"],
acquired["change"]["commit_id"],
)
self.assertEqual(
acquired["integration_lock_release_evidence"]["resource"],
"integration/master",
)
def test_invalid_hook_json_refuses_before_any_target_replacement(self) -> None:
with tempfile.TemporaryDirectory() as temporary:

107
tests/test_workspace.py Normal file
View file

@ -0,0 +1,107 @@
#!/usr/bin/env python3
"""Behavioral checks for exact integration handoff evidence."""
from __future__ import annotations
from pathlib import Path
import subprocess
import tempfile
import unittest
import workspace
class IntegrationEvidence(unittest.TestCase):
def test_freeze_captures_the_current_exact_change(self) -> None:
record = {
"change": {"change_id": "A1", "commit_id": "B2"},
"dirty": True,
}
frozen = workspace.freeze_integration_candidate(record)
self.assertEqual(
frozen["integration_candidate"],
{
"change_id": "a1",
"commit_id": "b2",
"captured_at": frozen["integration_candidate"]["captured_at"],
},
)
self.assertNotIn("integration_candidate", record)
def test_freeze_uses_parent_of_clean_jj_working_copy(self) -> None:
def runner(
argv: list[str], **_kwargs: object
) -> subprocess.CompletedProcess[str]:
self.assertIn("@-", argv)
return subprocess.CompletedProcess(argv, 0, "a1\nb2\n", "")
with tempfile.TemporaryDirectory() as temporary:
frozen = workspace.freeze_integration_candidate(
{
"actual_path": temporary,
"change": {"change_id": "c3", "commit_id": "d4"},
"dirty": False,
},
runner=runner,
)
self.assertEqual(frozen["integration_candidate"]["change_id"], "a1")
self.assertEqual(frozen["integration_candidate"]["commit_id"], "b2")
def test_integration_proves_frozen_candidate_not_stale_record_change(self) -> None:
source_commit = "a" * 40
target_commit = "c" * 40
target_change = "b" * 32
observed: list[list[str]] = []
def runner(
argv: list[str], **_kwargs: object
) -> subprocess.CompletedProcess[str]:
observed.append(argv)
if "change_id.normal_hex()" in argv[-1]:
stdout = f"{target_change}\n{target_commit}\n"
else:
stdout = f"{source_commit}\n"
return subprocess.CompletedProcess(argv, 0, stdout, "")
with tempfile.TemporaryDirectory() as temporary:
record = {
"actual_path": str(Path(temporary)),
"gate_state": "passed",
"integration_state": "lock-released",
"change": {"change_id": "d" * 32, "commit_id": "d" * 40},
"integration_candidate": {
"change_id": "e" * 32,
"commit_id": source_commit,
},
"integration_owner_runtime": "runtime-1",
"integration_resource": "integration/master",
"integration_lock_release_evidence": {
"runtime_id": "runtime-1",
"resource": "integration/master",
},
}
integrated = workspace.prove_integrated(
record,
target_revision="master",
runner=runner,
)
evidence = integrated["integration_release_evidence"]
self.assertEqual(evidence["source_commit"], source_commit)
self.assertEqual(evidence["target_commit"], target_commit)
self.assertTrue(any(source_commit in part for call in observed for part in call))
def test_integration_refuses_missing_owner_release_evidence(self) -> None:
with self.assertRaisesRegex(workspace.WorkspaceError, "owner/resource"):
workspace.prove_integrated(
{
"gate_state": "passed",
"integration_state": "lock-released",
"integration_candidate": {"commit_id": "a" * 40},
},
target_revision="master",
)
if __name__ == "__main__":
unittest.main(verbosity=2)

145
workspace.py
View file

@ -84,6 +84,16 @@ def slugify(value: str, *, limit: int = 22) -> str:
return slug or "task"
def _jj_workspace_identity(task: str, instance_id: str) -> dict[str, str]:
"""Return unique, presentation-safe jj metadata for one managed agent."""
task_slug = slugify(task)
return {
"workspace_name": f"agent-{instance_id}",
"bookmark": f"agent/{task_slug}/{instance_id}",
"description": f"coord work {instance_id}: {task.strip()}",
}
def task_window_name(
task: str,
session_id: str,
@ -149,6 +159,33 @@ def _jj_ids(
return {"change_id": values[0].lower(), "commit_id": values[1].lower()}
def _is_ancestor(
repo: Path,
ancestor_commit: str,
descendant_commit: str,
*,
jj: str,
runner: Runner,
) -> bool:
result = _run(
[
jj,
"--ignore-working-copy",
"-R",
str(repo),
"log",
"--no-graph",
"-r",
f"{ancestor_commit} & ::{descendant_commit}",
"-T",
'commit_id ++ "\\n"',
],
runner=runner,
)
observed = [line.strip().lower() for line in result.stdout.splitlines() if line.strip()]
return observed == [ancestor_commit.lower()]
def _owner_marker(path: Path) -> Path:
return path / ".jj" / "fleet-coord-owner.json"
@ -236,6 +273,8 @@ def _base_record(
"ownership_token": ownership_token,
"tmux": {},
"jj_workspace_name": "",
"jj_bookmark": "",
"jj_change_description": "",
"created_at": now,
"updated_at": now,
}
@ -284,11 +323,14 @@ def create_workspace(
mode=mode,
ownership_token=token,
)
jj_identity = _jj_workspace_identity(task, instance_id)
record["jj_workspace_name"] = jj_identity["workspace_name"]
record["jj_bookmark"] = jj_identity["bookmark"]
record["jj_change_description"] = jj_identity["description"]
_emit(registry, "workspace-planned", record)
try:
if native_workspace:
workspace_name = f"agent-{instance_id[:12]}"
_run(
[
jj,
@ -298,14 +340,13 @@ def create_workspace(
"workspace",
"add",
"--name",
workspace_name,
jj_identity["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)],
@ -316,6 +357,22 @@ def create_workspace(
# non-default base was explicitly selected.
if base_revision not in ("", "master"):
_run([jj, "-R", str(dest), "new", base_revision], runner=runner)
_run(
[jj, "-R", str(dest), "workspace", "rename", jj_identity["workspace_name"]],
runner=runner,
)
# A named workspace, described working-copy change, and namespaced
# bookmark make concurrent agent heads legible without granting any
# authority to their display names. Full IDs remain canonical evidence.
_run(
[jj, "-R", str(dest), "describe", "-m", jj_identity["description"]],
runner=runner,
)
_run(
[jj, "-R", str(dest), "bookmark", "create", jj_identity["bookmark"], "-r", "@"],
runner=runner,
)
_write_owner_marker(dest, instance_id=instance_id, token=token)
record["actual_path"] = str(dest.resolve())
@ -631,6 +688,19 @@ def remove_workspace(
"workspace contains an advanced change without verified integration release evidence; "
"refusing removal"
)
if inspected.get("advanced"):
current_commit = str((inspected.get("change") or {}).get("commit_id") or "")
integrated_commit = str((release_evidence or {}).get("source_commit") or "")
if current_commit.lower() != integrated_commit.lower() and not _is_ancestor(
repo=Path(str(inspected["actual_path"])),
ancestor_commit=integrated_commit,
descendant_commit=current_commit,
jj=jj,
runner=runner,
):
raise UnsafeRemoval(
"workspace moved away from its integrated candidate; refusing removal"
)
gated = inspected.get("gate_state") in {"passed", "waived"}
if automatic and not gated:
raise UnsafeRemoval("automatic cleanup never removes ungated work")
@ -735,6 +805,35 @@ def enqueue_integration(
return updated
def freeze_integration_candidate(
record: Mapping[str, Any],
*,
registry: Registry | None = None,
jj: str = "jj",
runner: Runner = subprocess.run,
) -> dict[str, Any]:
"""Freeze contentful @, or @- when jj has created a clean empty child."""
dirty = record.get("dirty")
if dirty is True:
change = dict(record.get("change") or {})
elif dirty is False:
path = Path(str(record.get("actual_path") or "")).resolve()
change = _jj_ids(path, "@-", jj=jj, runner=runner)
else:
raise WorkspaceError("integration acquisition requires inspected cleanliness")
if not change.get("change_id") or not change.get("commit_id"):
raise WorkspaceError("integration acquisition requires an inspected exact change")
updated = copy.deepcopy(dict(record))
updated["integration_candidate"] = {
"change_id": str(change["change_id"]).lower(),
"commit_id": str(change["commit_id"]).lower(),
"captured_at": _utc_now(),
}
updated["updated_at"] = updated["integration_candidate"]["captured_at"]
_emit(registry, "workspace-integration-candidate", updated)
return updated
def prove_integrated(
record: Mapping[str, Any],
*,
@ -748,28 +847,28 @@ def prove_integrated(
raise WorkspaceError("integration proof requires a released integration lock")
if record.get("gate_state") not in {"passed", "waived"}:
raise WorkspaceError("integration proof requires a passed or explicitly waived gate")
owner = str(record.get("integration_owner_runtime") or "")
resource = str(record.get("integration_resource") or "")
release = dict(record.get("integration_lock_release_evidence") or {})
if not owner or not resource or release.get("runtime_id") != owner or release.get(
"resource"
) != resource:
raise WorkspaceError(
"integration proof requires exact owner/resource lock-release evidence"
)
path = Path(str(record.get("actual_path") or "")).resolve()
source_commit = str((record.get("change") or {}).get("commit_id", ""))
candidate = dict(record.get("integration_candidate") or {})
source_commit = str(candidate.get("commit_id", ""))
if not source_commit:
raise WorkspaceError("workspace record has no exact source commit")
raise WorkspaceError("workspace record has no frozen integration candidate")
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"',
],
if not _is_ancestor(
path,
source_commit,
target["commit_id"],
jj=jj,
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']}"
@ -910,6 +1009,10 @@ def start_agent(
"env",
f"COORD_INSTANCE_ID={attachment_runtime}",
f"COORD_WORK_ID={updated['instance_id']}",
f"COORD_WORKSPACE_PATH={project}",
f"COORD_WORKSPACE_MODE={str(updated.get('mode') or '')}",
f"COORD_JJ_WORKSPACE={str(updated.get('jj_workspace_name') or '')}",
f"COORD_JJ_BOOKMARK={str(updated.get('jj_bookmark') or '')}",
f"FLEET_AI_TASK={str(updated.get('task') or 'task')}",
*command,
]