Skills › Compound Engineering › ce-work › unit_workspace_integration.py
unit_workspace_integration.py
677 lines · skills/ce-work/scripts/unit_workspace_integration.py at 3449230
1"""Canonical integration, locking, wave sequencing, and exact restoration."""
3from __future__ import annotations
5import json
6import os
7import re
8import secrets
9import shutil
10from pathlib import Path
12from unit_workspace_state import *
13from unit_workspace_jobs import find_attempt, parse_diff_paths, scope_expansion_pending
16def integration_lock_path(doc: dict) -> str:
17 ident = doc["repository"]["identity_digest"] + "\0" + doc["branch"]["ref"]
18 # Anchor to the root this run actually lives under (run_dir searches both
19 # candidate roots), never the current invocation's preferred creation root:
20 # a fallback-root run must validate and release the lock it recorded.
21 run_root = os.path.dirname(run_dir(doc["run_id"]))
22 return os.path.join(run_root, ".locks", f"integration-{digest_bytes(ident.encode())}.json")
25def read_integration_lock(path: str) -> dict:
26 return read_private_json(path)
29def validated_lock_nonce(doc: dict, unit_id: str, lock: dict) -> str:
30 expected = {
31 "run_id": doc["run_id"],
32 "unit_id": unit_id,
33 "repository": doc["repository"]["identity_digest"],
34 "branch_ref": doc["branch"]["ref"],
35 }
36 if any(lock.get(k) != v for k, v in expected.items()):
37 raise Operational("BLOCKED", "integration lock identity mismatch")
38 nonce = lock.get("nonce")
39 if not isinstance(nonce, str) or not re.fullmatch(r"[0-9a-f]{48}", nonce):
40 raise TrustFailure("integration lock nonce is malformed")
41 return nonce
44def validate_lock(doc: dict, unit_id: str, token: str) -> tuple[str, dict]:
45 path = integration_lock_path(doc)
46 try:
47 lock = read_integration_lock(path)
48 except TrustFailure as exc:
49 if not os.path.lexists(path):
50 raise Operational("BLOCKED", "integration lock is missing") from exc
51 raise
52 nonce = validated_lock_nonce(doc, unit_id, lock)
53 if nonce != token:
54 raise Operational("BLOCKED", "integration lock token or identity mismatch")
55 return path, lock
58def cmd_integration_acquire(args) -> tuple[str, dict]:
59 with locked_manifest(args.run_id) as doc:
60 validate_repo(doc)
61 unit = doc["units"].get(args.unit_id)
62 plan_verification = bool(getattr(args, "plan_verification", False))
63 recover_only = bool(getattr(args, "recover_only", False))
64 allowed_states = INTEGRATABLE_STATES | {"preserved", "committed", "cleaned"}
65 if plan_verification or (recover_only and unit and unit.get("state") == "native-completed"):
66 allowed_states.add("native-completed")
67 if not unit or unit["state"] not in allowed_states:
68 raise Operational("REFUSED", "unit is not ready for integration")
69 if not plan_verification and unit["state"] != "native-completed":
70 validate_wave_order(doc, unit)
71 path = integration_lock_path(doc)
72 existing = doc.get("integration_lock")
73 if existing:
74 if not args.resume:
75 raise Operational("REFUSED", "integration claim already exists; pass --resume to recover the same claim")
76 if existing.get("phase", "held") != "held":
77 raise Operational("REFUSED", "integration claim is releasing; resume or retry release before acquisition")
78 validate_lock(doc, args.unit_id, existing["nonce"])
79 return "ACQUIRED", {"lock_token": existing["nonce"], "resumed": True, "path": path}
80 nonce = secrets.token_hex(24)
81 resumed = False
82 if recover_only:
83 lock = read_integration_lock(path)
84 nonce = validated_lock_nonce(doc, args.unit_id, lock)
85 resumed = True
86 else:
87 payload = {"run_id": args.run_id, "unit_id": args.unit_id, "nonce": nonce, "repository": doc["repository"]["identity_digest"], "branch_ref": doc["branch"]["ref"], "created_at": now_iso()}
88 try:
89 create_private(path, (json.dumps(payload, sort_keys=True) + "\n").encode())
90 test_fault("integration-lock-after-create")
91 except Operational as exc:
92 if exc.word == "INTERRUPTED":
93 raise
94 lock = read_integration_lock(path)
95 if lock.get("run_id") == args.run_id and lock.get("unit_id") == args.unit_id:
96 if not args.resume:
97 raise Operational("REFUSED", "integration lock file already exists; pass --resume to recover its claim")
98 nonce = validated_lock_nonce(doc, args.unit_id, lock)
99 resumed = True
100 else:
101 raise Operational("BLOCKED", "another run/unit owns canonical integration", {"owner_run": lock.get("run_id"), "owner_unit": lock.get("unit_id")})
102 with locked_manifest(args.run_id, write=True) as doc:
103 doc["integration_lock"] = {"unit_id": args.unit_id, "nonce": nonce, "path": path, "phase": "held"}
104 event(doc, "integration-lock-acquired", args.unit_id, {"resumed": resumed})
105 return "ACQUIRED", {"lock_token": nonce, "resumed": resumed, "path": path}
108def semantic_snapshot(repo: str) -> dict:
109 head = git_text(repo, "rev-parse", "HEAD")
110 head_tree = git_text(repo, "rev-parse", "HEAD^{tree}")
111 index_tree = git_text(repo, "write-tree")
112 raw = git(repo, "status", "--porcelain=v2", "-z", "--untracked-files=all")
113 worktree_index = git(repo, "diff", "--name-only", "-z")
114 return {
115 "head": head,
116 "branch_ref": git_text(repo, "symbolic-ref", "-q", "HEAD", check=False),
117 "head_tree": head_tree,
118 "index_tree": index_tree,
119 "status_sha256": digest_bytes(raw),
120 "status_empty": not bool(raw),
121 "worktree_index_empty": not bool(worktree_index),
122 }
125def expected_apply_snapshot(repo: str, pre_head: str, unit: dict) -> dict:
126 transport = unit["transport"]
127 if pre_head == transport["base"]:
128 tree = transport["tree"]
129 else:
130 # Compute the same semantic three-way result as applying the
131 # base-parented transport commit, without touching the canonical index.
132 merged = git_text(repo, "merge-tree", "--write-tree", pre_head, transport["commit"])
133 tree = merged.splitlines()[0] if merged else ""
134 if not tree:
135 raise Operational("BLOCKED", "could not derive expected canonical apply tree")
136 raw = git(repo, "diff-tree", "-r", "-M", "--name-status", "-z", pre_head, tree)
137 return {"index_tree": tree, "changed_paths": parse_diff_paths(raw)}
140def matches_expected_apply(repo: str, unit: dict, snap: dict | None = None) -> bool:
141 snap = snap or semantic_snapshot(repo)
142 pre = unit.get("integration", {}).get("pre_fold")
143 expected = unit.get("integration", {}).get("expected_apply")
144 if not pre or not expected:
145 return False
146 return (
147 snap["head"] == pre["head"]
148 and snap["index_tree"] == expected["index_tree"]
149 and snap["worktree_index_empty"]
150 and status_paths(repo) == set(expected["changed_paths"])
151 )
154def wave_members(doc: dict, unit: dict) -> list[dict]:
155 wave = unit.get("wave", {})
156 wave_id = wave.get("id")
157 if not wave_id:
158 return []
159 base = wave.get("base")
160 members = [
161 candidate for candidate in doc.get("units", {}).values()
162 if candidate.get("wave", {}).get("id") == wave_id
163 ]
164 positions = [candidate.get("wave", {}).get("position") for candidate in members]
165 if any(candidate.get("wave", {}).get("base") != base for candidate in members):
166 raise Operational("BLOCKED", "wave members do not share one recorded base")
167 if len(set(positions)) != len(positions):
168 raise Operational("BLOCKED", "wave positions are not unique")
169 return sorted(members, key=lambda candidate: candidate["wave"]["position"])
172def validate_wave_order(doc: dict, unit: dict) -> None:
173 members = wave_members(doc, unit)
174 earlier_unresolved = [
175 candidate["unit_id"] for candidate in members
176 if candidate["wave"]["position"] < unit["wave"]["position"]
177 and not (
178 candidate.get("state") in {"committed", "preserved", "cleaned"}
179 or (
180 candidate.get("state") == "native-completed"
181 and unit_accepted_commit(candidate) is not None
182 )
183 )
184 ]
185 if earlier_unresolved:
186 raise Operational(
187 "BLOCKED",
188 "earlier wave units must be committed or preserved before this fold-in",
189 {"reason": "earlier wave unit not resolved", "units": earlier_unresolved},
190 )
193def wave_member_changed_paths(candidate: dict) -> set[str] | None:
194 transport = candidate.get("transport", {})
195 if transport.get("commit"):
196 paths = transport.get("changed_paths")
197 elif candidate.get("state") == "native-completed":
198 if unit_accepted_commit(candidate) is None:
199 raise TrustFailure("native wave completion evidence is malformed")
200 completion = find_attempt(candidate).get("fallback", {}).get("completed")
201 paths = completion.get("changed_paths") if isinstance(completion, dict) else None
202 else:
203 return None
204 if not isinstance(paths, list) or any(not isinstance(path, str) for path in paths):
205 raise TrustFailure("wave member changed paths are malformed")
206 return set(paths)
209def validate_wave_collisions(
210 doc: dict,
211 unit: dict,
212 overrides: dict[str, set[str]] | None = None,
213 require_complete: bool = True,
214) -> None:
215 members = wave_members(doc, unit)
216 if not members:
217 return
218 overrides = overrides or {}
219 changed_by_unit: dict[str, set[str]] = {}
220 unterminated: list[str] = []
221 for candidate in members:
222 unit_id = candidate["unit_id"]
223 paths = overrides.get(unit_id)
224 if paths is None:
225 paths = wave_member_changed_paths(candidate)
226 if paths is None:
227 unterminated.append(unit_id)
228 else:
229 changed_by_unit[unit_id] = paths
230 if unterminated:
231 if require_complete:
232 raise Operational(
233 "BLOCKED",
234 "every wave worker must terminalize before the first fold-in",
235 {"reason": "wave not fully terminalized", "units": unterminated},
236 )
237 collisions: dict[str, list[str]] = {}
238 for index, left in enumerate(members):
239 for right in members[index + 1:]:
240 left_paths = changed_by_unit.get(left["unit_id"])
241 right_paths = changed_by_unit.get(right["unit_id"])
242 if left_paths is None or right_paths is None:
243 continue
244 overlap = sorted(left_paths & right_paths)
245 if overlap:
246 collisions[f'{left["unit_id"]}:{right["unit_id"]}'] = overlap
247 if collisions:
248 raise Operational(
249 "BLOCKED",
250 "wave transports have a changed-path collision",
251 {"reason": "changed-path collision", "collisions": collisions},
252 )
255def validate_wave_ready(doc: dict, unit: dict) -> None:
256 members = wave_members(doc, unit)
257 if not members:
258 return
259 validate_wave_order(doc, unit)
260 validate_wave_collisions(doc, unit)
263def validate_wave_advancement(members: list[dict], unit: dict, parent: str, canonical: str) -> list[str]:
264 position = unit["wave"]["position"]
265 targets = [candidate for candidate in members if candidate["wave"]["position"] > position]
266 for candidate in targets:
267 allowed = candidate["wave"].get("allowed_heads", [])
268 if canonical in allowed:
269 continue
270 if not allowed or allowed[-1] != parent:
271 raise Operational("BLOCKED", "wave advancement is not the exact recorded canonical chain")
272 return [candidate["unit_id"] for candidate in targets]
275def advance_wave_allowed_heads(members: list[dict], position: int, canonical: str) -> list[str]:
276 advanced: list[str] = []
277 for candidate in members:
278 if candidate["wave"]["position"] <= position:
279 continue
280 allowed = candidate["wave"].setdefault("allowed_heads", [])
281 if canonical not in allowed:
282 allowed.append(canonical)
283 advanced.append(candidate["unit_id"])
284 return advanced
287def validate_dependencies_ready(doc: dict, unit: dict) -> None:
288 missing: list[str] = []
289 unaccepted: list[str] = []
290 for dependency_id in unit.get("dependencies", []):
291 dependency = doc.get("units", {}).get(dependency_id)
292 if dependency is None:
293 missing.append(dependency_id)
294 continue
295 if unit_accepted_commit(dependency) is None:
296 unaccepted.append(dependency_id)
297 if missing or unaccepted:
298 raise Operational(
299 "BLOCKED",
300 "unit dependencies must have controller-accepted canonical commits before preflight",
301 {
302 "unit_id": unit["unit_id"],
303 "missing_dependencies": missing,
304 "unaccepted_dependencies": unaccepted,
305 },
306 )
309def dependency_advanced_head(doc: dict, unit: dict, head: str) -> bool:
310 dependency_commits = [
311 unit_accepted_commit(doc["units"][dependency_id])
312 for dependency_id in unit.get("dependencies", [])
313 ]
314 if any(commit is None for commit in dependency_commits):
315 return False
316 accepted_heads = {
317 commit
318 for unit_id, candidate in doc.get("units", {}).items()
319 if unit_id != unit.get("unit_id")
320 if (commit := unit_accepted_commit(candidate)) is not None
321 }
322 if head not in accepted_heads:
323 return False
324 allowed_heads = unit.get("wave", {}).get("allowed_heads", [])
325 required_ancestors = {
326 *accepted_heads,
327 *dependency_commits,
328 *allowed_heads,
329 unit.get("workspace", {}).get("base"),
330 }
331 required_ancestors.discard(None)
332 repo = doc["repository"]["toplevel"]
333 return all(
334 git_text(repo, "merge-base", commit, head, check=False) == commit
335 for commit in required_ancestors
336 )
339def validate_preflight_ancestry(doc: dict, unit: dict, heads: set[str]) -> None:
340 required = {
341 commit
342 for unit_id, candidate in doc.get("units", {}).items()
343 if unit_id != unit["unit_id"]
344 and (commit := unit_accepted_commit(candidate)) is not None
345 }
346 wave = unit.get("wave", {})
347 required.update(head for head in wave.get("allowed_heads", []) if head != wave.get("base"))
349 repo = doc["repository"]["toplevel"]
350 missing = {
351 head: sorted(
352 commit for commit in required
353 if git_text(repo, "merge-base", commit, head, check=False) != commit
354 )
355 for head in sorted(heads)
356 }
357 missing = {head: commits for head, commits in missing.items() if commits}
358 if missing:
359 raise Operational(
360 "BLOCKED",
361 "preflight HEAD omits controller-accepted prerequisite commits",
362 {"unit_id": unit["unit_id"], "missing_ancestry": missing},
363 )
366def cmd_preflight(args) -> tuple[str, dict]:
367 with locked_manifest(args.run_id) as doc:
368 info = validate_repo(doc)
369 unit = doc["units"].get(args.unit_id)
370 if not unit or unit["state"] not in {"integration-pending", "preserved"}:
371 raise Operational("REFUSED", "unit is not integration-pending")
372 validate_lock(doc, args.unit_id, args.lock_token)
373 validate_dependencies_ready(doc, unit)
374 if scope_expansion_pending(unit):
375 raise Operational(
376 "BLOCKED",
377 "worker requested scope expansion; inspect the retained result and transport, then resolve or re-dispatch explicitly",
378 {
379 "unit_id": args.unit_id,
380 "terminal_status": "scope_expansion",
381 "transport": unit["transport"],
382 "recovery_path": unit["recovery_path"],
383 },
384 )
385 validate_wave_ready(doc, unit)
386 allowed = set(unit["wave"].get("allowed_heads", []))
387 requested: set[str] = set()
388 if args.allowed_head:
389 requested = {git_text(info["toplevel"], "rev-parse", f"{h}^{{commit}}") for h in args.allowed_head}
390 if any(head not in allowed and not dependency_advanced_head(doc, unit, head) for head in requested):
391 raise Operational("BLOCKED", "unrecorded same-wave HEAD allowance")
392 if info["head"] not in allowed and not dependency_advanced_head(doc, unit, info["head"]):
393 raise Operational("BLOCKED", "canonical HEAD advanced outside the recorded wave")
394 validate_preflight_ancestry(doc, unit, requested | {info["head"]})
395 snap = semantic_snapshot(info["toplevel"])
396 if not snap["status_empty"] or snap["index_tree"] != snap["head_tree"]:
397 raise Operational("BLOCKED", "canonical checkout is not clean at preflight")
398 expected = expected_apply_snapshot(info["toplevel"], snap["head"], unit)
399 intent_revision = doc["revision"] + 1
400 with locked_manifest(args.run_id, write=True) as doc:
401 unit = doc["units"][args.unit_id]
402 unit["state"] = "integration-pending"
403 unit["integration"]["intent_revision"] = intent_revision
404 unit["integration"]["pre_fold"] = snap
405 unit["integration"]["expected_apply"] = expected
406 event(doc, "canonical-apply-intent", args.unit_id, {"transport": unit["transport"]["commit"], "pre_head": snap["head"]})
407 return "PREFLIGHT_OK", {"unit_id": args.unit_id, "pre_fold": snap, "transport": unit["transport"]}
410def cmd_mark_applied(args) -> tuple[str, dict]:
411 with locked_manifest(args.run_id) as doc:
412 validate_lock(doc, args.unit_id, args.lock_token)
413 unit = doc["units"].get(args.unit_id)
414 if not unit or unit["state"] not in {"integration-pending", "integrated"} or not unit["integration"].get("pre_fold"):
415 raise Operational("REFUSED", "no recorded preflight intent")
416 repo = validate_repo(doc)["toplevel"]
417 snap = semantic_snapshot(repo)
418 if snap["head"] != unit["integration"]["pre_fold"]["head"]:
419 raise Operational("BLOCKED", "canonical HEAD moved before apply was recorded")
420 if not matches_expected_apply(repo, unit, snap):
421 raise Operational("BLOCKED", "canonical state does not match the expected transport application")
422 test_fault("after-apply-observed")
423 with locked_manifest(args.run_id, write=True) as doc:
424 unit = doc["units"][args.unit_id]
425 unit["state"] = "integrated"
426 unit["integration"]["applied"] = {"at": now_iso(), "post_index_tree": snap["index_tree"], "status_sha256": snap["status_sha256"]}
427 event(doc, "transport-applied", args.unit_id, {"post_index_tree": snap["index_tree"]})
428 return "APPLIED", {"unit_id": args.unit_id, "post_index_tree": snap["index_tree"]}
431def cmd_mark_verified(args) -> tuple[str, dict]:
432 with locked_manifest(args.run_id, write=True) as doc:
433 validate_lock(doc, args.unit_id, args.lock_token)
434 unit = doc["units"].get(args.unit_id)
435 if not unit or unit["state"] not in {"integrated", "verified"}:
436 raise Operational("REFUSED", "unit is not applied")
437 repo = validate_repo(doc)["toplevel"]
438 if not matches_expected_apply(repo, unit):
439 raise Operational(
440 "BLOCKED",
441 "canonical state changed after the recorded transport application",
442 {
443 "unit_id": args.unit_id,
444 "reason": "canonical state no longer matches the expected transport application",
445 },
446 )
447 evidence = {"at": now_iso(), "digest": args.evidence_digest, "summary": args.summary}
448 ignored_state = getattr(args, "ignored_state", None)
449 if isinstance(ignored_state, str):
450 ignored_state = parse_json_arg(ignored_state, "ignored-state")
451 if ignored_state is not None:
452 evidence["ignored_state"] = ignored_state
453 unit["integration"]["verification"] = evidence
454 unit["state"] = "verified"
455 event(doc, "canonical-verification-passed", args.unit_id, {"digest": args.evidence_digest})
456 return "VERIFIED", {"unit_id": args.unit_id, "verification": evidence}
459def reconcile_commit(doc: dict, unit: dict) -> dict | None:
460 repo = doc["repository"]["toplevel"]
461 head = git_text(repo, "rev-parse", "HEAD")
462 parents = git_text(repo, "rev-list", "--parents", "-n", "1", head).split()
463 expected_parent = unit["integration"]["pre_fold"]["head"]
464 expected_tree = unit["integration"]["applied"]["post_index_tree"]
465 actual_tree = git_text(repo, "rev-parse", "HEAD^{tree}")
466 if parents == [head, expected_parent] and actual_tree == expected_tree and not status_paths(repo):
467 return {"commit": head, "parent": expected_parent, "tree": actual_tree, "at": now_iso()}
468 return None
471def cmd_mark_committed(args) -> tuple[str, dict]:
472 with locked_manifest(args.run_id) as doc:
473 validate_lock(doc, args.unit_id, args.lock_token)
474 unit = doc["units"].get(args.unit_id)
475 if not unit or unit["state"] not in {"verified", "committed"}:
476 raise Operational("REFUSED", "unit has not passed canonical verification")
477 commit = reconcile_commit(doc, unit)
478 if not commit:
479 raise Operational("BLOCKED", "canonical commit parent/tree/cleanliness do not match recorded integration")
480 with locked_manifest(args.run_id, write=True) as doc:
481 unit = doc["units"][args.unit_id]
482 unit["integration"]["canonical_commit"] = commit
483 unit["state"] = "committed"
484 event(doc, "canonical-commit-confirmed", args.unit_id, {"commit": commit["commit"]})
485 return "COMMITTED", {"unit_id": args.unit_id, "canonical_commit": commit}
488def cmd_wave_advance(args) -> tuple[str, dict]:
489 with locked_manifest(args.run_id) as doc:
490 info = validate_repo(doc)
491 unit = doc["units"].get(args.unit_id)
492 if not unit or unit.get("state") != "committed":
493 raise Operational("REFUSED", "only a committed wave unit can advance its siblings")
494 validate_lock(doc, args.unit_id, args.lock_token)
495 members = wave_members(doc, unit)
496 if not members:
497 raise Operational("REFUSED", "unit does not belong to a parallel wave")
498 validate_wave_ready(doc, unit)
499 canonical = git_text(info["toplevel"], "rev-parse", f"{args.canonical_commit}^{{commit}}")
500 recorded = unit.get("integration", {}).get("canonical_commit", {})
501 if recorded.get("commit") != canonical or info["head"] != canonical:
502 raise Operational("BLOCKED", "canonical wave commit does not match manifest and HEAD")
503 parent = unit.get("integration", {}).get("pre_fold", {}).get("head")
504 if recorded.get("parent") != parent:
505 raise Operational("BLOCKED", "canonical wave commit parent is not the recorded pre-fold HEAD")
506 validate_wave_advancement(members, unit, parent, canonical)
507 with locked_manifest(args.run_id, write=True) as doc:
508 unit = doc["units"][args.unit_id]
509 position = unit["wave"]["position"]
510 advanced = advance_wave_allowed_heads(wave_members(doc, unit), position, canonical)
511 event(doc, "wave-advanced", args.unit_id, {"canonical_commit": canonical, "eligible_siblings": advanced})
512 return "WAVE_ADVANCED", {"unit_id": args.unit_id, "canonical_commit": canonical, "eligible_siblings": advanced}
515def path_in_tree(repo: str, treeish: str, rel: str) -> bool:
516 out = git(repo, "ls-tree", "-z", "--full-tree", treeish, "--", rel)
517 return bool(out)
520def remove_introduced_paths(repo: str, unit: dict) -> None:
521 pre = unit["integration"]["pre_fold"]["head"]
522 base = unit["transport"]["base"]
523 commit = unit["transport"]["commit"]
524 raw = git(repo, "diff-tree", "-r", "-M", "--name-status", "-z", base, commit)
525 for rel in parse_diff_paths(raw):
526 if path_in_tree(repo, pre, rel):
527 continue
528 target = os.path.abspath(os.path.join(repo, rel))
529 if os.path.commonpath([repo, target]) != repo:
530 raise Operational("BLOCKED", "transport path escaped canonical repository")
531 if os.path.islink(target) or os.path.isfile(target):
532 os.unlink(target)
533 elif os.path.isdir(target):
534 shutil.rmtree(target)
535 parent = os.path.dirname(target)
536 while parent != repo and os.path.commonpath([repo, parent]) == repo:
537 try:
538 os.rmdir(parent)
539 except OSError:
540 break
541 parent = os.path.dirname(parent)
544def restore(run_id: str, unit_id: str, lock_token: str) -> bool:
545 with locked_manifest(run_id) as doc:
546 validate_lock(doc, unit_id, lock_token)
547 unit = doc["units"].get(unit_id)
548 if not unit or not unit["integration"].get("pre_fold"):
549 raise Operational("REFUSED", "unit has no pre-fold snapshot")
550 repo = doc["repository"]["toplevel"]
551 pre = dict(unit["integration"]["pre_fold"])
552 git_dir = git_text(repo, "rev-parse", "--path-format=absolute", "--absolute-git-dir")
553 cherry_pick_head = os.path.join(git_dir, "CHERRY_PICK_HEAD")
554 expected_conflict = False
555 if os.path.isfile(cherry_pick_head) and git_text(repo, "rev-parse", "HEAD") == pre["head"]:
556 expected_conflict = Path(cherry_pick_head).read_text().strip() == unit["transport"]["commit"]
557 current = None if expected_conflict else semantic_snapshot(repo)
558 already_exact = current == pre if current else False
559 expected_apply = matches_expected_apply(repo, unit, current) if current else False
560 partial_reset = bool(current) and (
561 unit.get("state") == "restoring"
562 and current["head"] == pre["head"]
563 and current["index_tree"] == pre["index_tree"]
564 and current["worktree_index_empty"]
565 and status_paths(repo).issubset(set(unit["integration"]["expected_apply"]["changed_paths"]))
566 )
567 if not (already_exact or expected_apply or partial_reset or expected_conflict):
568 raise Operational("BLOCKED", "canonical state is not a proven in-flight transport state; refusing destructive restoration")
569 with locked_manifest(run_id, write=True) as doc:
570 unit = doc["units"][unit_id]
571 unit["state"] = "restoring"
572 event(doc, "restore-intent", unit_id)
573 if not already_exact:
574 git(repo, "cherry-pick", "--abort", check=False)
575 git(repo, "reset", "--hard", pre["head"])
576 test_fault("restore-after-reset")
577 with locked_manifest(run_id) as doc:
578 unit = doc["units"][unit_id]
579 remove_introduced_paths(repo, unit)
580 test_fault("restore-after-path-removal")
581 actual = semantic_snapshot(repo)
582 exact = actual == pre
583 with locked_manifest(run_id, write=True) as doc:
584 unit = doc["units"][unit_id]
585 unit["integration"]["restore"] = {
586 "at": now_iso(),
587 "exact": exact,
588 "already_exact": already_exact,
589 "snapshot": actual,
590 }
591 if exact:
592 unit["state"] = "preserved"
593 event(doc, "canonical-restored", unit_id)
594 else:
595 blocker = {"at": now_iso(), "unit_id": unit_id, "reason": "exact pre-fold restoration could not be proven"}
596 doc["blockers"].append(blocker)
597 event(doc, "restore-blocked", unit_id)
598 return exact
601def cmd_restore(args) -> tuple[str, dict]:
602 exact = restore(args.run_id, args.unit_id, args.lock_token)
603 if not exact:
604 raise Operational("BLOCKED", "exact pre-fold restoration could not be proven")
605 return "PRESERVED", {"unit_id": args.unit_id, "recovery_path": os.path.join(run_dir(args.run_id), "units", args.unit_id)}
610def release_lock_is_owned(doc: dict, unit_id: str, lock_token: str, lock: dict) -> bool:
611 expected = {
612 "run_id": doc["run_id"],
613 "unit_id": unit_id,
614 "repository": doc["repository"]["identity_digest"],
615 "branch_ref": doc["branch"]["ref"],
616 }
617 if any(lock.get(key) != value for key, value in expected.items()):
618 return False
619 nonce = lock.get("nonce")
620 if not isinstance(nonce, str) or not re.fullmatch(r"[0-9a-f]{48}", nonce):
621 return False
622 return nonce == lock_token
625def integration_release(run_id: str, unit_id: str, lock_token: str) -> None:
626 with locked_manifest(run_id, write=True) as doc:
627 held = doc.get("integration_lock")
628 if not held or held.get("unit_id") != unit_id or held.get("nonce") != lock_token:
629 raise Operational("REFUSED", "integration lock token or identity mismatch")
630 unit = doc["units"].get(unit_id)
631 pre_apply = bool(
632 unit
633 and unit.get("state") == "integration-pending"
634 and not unit.get("integration", {}).get("pre_fold")
635 )
636 if not unit or (unit["state"] not in {"committed", "preserved", "cleaned", "native-completed"} and not pre_apply):
637 raise Operational("REFUSED", "integration lock releases only before preflight or after accepted completion")
638 path = held.get("path")
639 if path != integration_lock_path(doc):
640 raise Operational("BLOCKED", "manifest integration lock path changed")
641 phase = held.get("phase", "held")
642 if phase == "held":
643 validate_lock(doc, unit_id, lock_token)
644 held["phase"] = "releasing"
645 held["release_started_at"] = now_iso()
646 event(doc, "integration-lock-release-intent", unit_id)
647 elif phase != "releasing":
648 raise Operational("BLOCKED", "manifest integration claim has an unknown phase")
649 with locked_manifest(run_id, write=True) as doc:
650 held = doc.get("integration_lock")
651 if not held or held.get("unit_id") != unit_id or held.get("nonce") != lock_token or held.get("phase") != "releasing":
652 raise Operational("BLOCKED", "manifest integration claim changed")
653 if held.get("path") != path or path != integration_lock_path(doc):
654 raise Operational("BLOCKED", "manifest integration lock path changed")
655 current = None
656 try:
657 current = read_integration_lock(path)
658 except TrustFailure:
659 if os.path.lexists(path):
660 raise
661 if current is not None and release_lock_is_owned(doc, unit_id, lock_token, current):
662 try:
663 os.unlink(path)
664 except FileNotFoundError:
665 pass
666 test_fault("integration-release-after-unlink")
667 if os.path.lexists(path):
668 current = read_integration_lock(path)
669 if release_lock_is_owned(doc, unit_id, lock_token, current):
670 raise Operational("BLOCKED", "integration lock file remained after release")
671 doc["integration_lock"] = None
672 event(doc, "integration-lock-released", unit_id)
675def cmd_integration_release(args) -> tuple[str, dict]:
676 integration_release(args.run_id, args.unit_id, args.lock_token)
677 return "RELEASED", {"unit_id": args.unit_id}