Skills › Compound Engineering › ce-work › unit_workspace_transaction.py
unit_workspace_transaction.py
612 lines · skills/ce-work/scripts/unit_workspace_transaction.py at 3449230
1"""Fail-stop canonical integration for one terminalized external unit."""
3from __future__ import annotations
5import hashlib
6import json
7import os
8import secrets
9import shutil
10import subprocess
11from pathlib import Path
12from types import SimpleNamespace
14from unit_workspace_state import *
15from unit_workspace_integration import (
16 cmd_integration_acquire,
17 cmd_integration_release,
18 cmd_mark_applied,
19 cmd_mark_committed,
20 cmd_mark_verified,
21 cmd_preflight,
22 cmd_restore,
23 cmd_wave_advance,
24 matches_expected_apply,
25 remove_introduced_paths,
26 semantic_snapshot,
27 validate_lock,
28)
29from unit_workspace_lifecycle import (
30 cmd_cleanup,
31 pending_plan_wide_verification,
32 plan_wide_verification_attempts,
33 receipted_plan_wide_verification,
34)
35from unit_workspace_ignored import diff_ignored_state, inventory_ignored_state
38def _args(**values):
39 return SimpleNamespace(**values)
42def _verification_command(args, operation: str = "integrate") -> list[str]:
43 command = list(args.verification_command)
44 if command and command[0] == "--":
45 command.pop(0)
46 if not command or any(not value or "\0" in value for value in command):
47 raise Operational("REFUSED", f"{operation} requires a non-empty verification command after --")
48 return command
51def _remove_owned_new_paths(repo: str, paths: set[str], pre_head: str) -> None:
52 for rel in sorted(paths, key=lambda value: (value.count("/"), value), reverse=True):
53 if git(repo, "ls-tree", "-z", "--full-tree", pre_head, "--", rel):
54 continue
55 target = os.path.abspath(os.path.join(repo, rel))
56 if os.path.commonpath([repo, target]) != repo:
57 raise Operational("BLOCKED", "verification artifact path escaped canonical repository")
58 if os.path.islink(target) or os.path.isfile(target):
59 os.unlink(target)
60 elif os.path.isdir(target):
61 shutil.rmtree(target)
64def _restore_owned_verification(
65 run_id: str,
66 unit_id: str,
67 token: str,
68 before: dict,
69 before_paths: set[str],
70 after_paths: set[str],
71) -> None:
72 with locked_manifest(run_id) as doc:
73 validate_repo(doc)
74 unit = doc["units"].get(unit_id)
75 if not unit or not unit.get("integration", {}).get("pre_fold"):
76 raise Operational("BLOCKED", "owned verification restoration lacks pre-fold evidence")
77 repo = doc["repository"]["toplevel"]
78 pre = dict(unit["integration"]["pre_fold"])
79 expected = unit["integration"]["expected_apply"]
80 if not (
81 before["head"] == pre["head"]
82 and before["index_tree"] == expected["index_tree"]
83 and before["worktree_index_empty"]
84 and before_paths == set(expected["changed_paths"])
85 ):
86 raise Operational("BLOCKED", "owned verification did not start from the expected transport application")
87 if git_text(repo, "rev-parse", "HEAD") != pre["head"]:
88 raise Operational("BLOCKED", "verification changed canonical HEAD; refusing automatic restoration")
89 verification_paths = after_paths - before_paths
90 with locked_manifest(run_id, write=True) as doc:
91 doc["units"][unit_id]["state"] = "restoring"
92 event(doc, "restore-intent", unit_id, {"source": "controller-owned-verification"})
93 git(repo, "reset", "--hard", pre["head"])
94 with locked_manifest(run_id) as doc:
95 remove_introduced_paths(repo, doc["units"][unit_id])
96 _remove_owned_new_paths(repo, verification_paths, pre["head"])
97 actual = semantic_snapshot(repo)
98 exact = actual == pre
99 with locked_manifest(run_id, write=True) as doc:
100 unit = doc["units"][unit_id]
101 unit["integration"]["restore"] = {"at": now_iso(), "exact": exact, "snapshot": actual}
102 if exact:
103 unit["state"] = "preserved"
104 event(doc, "canonical-restored", unit_id, {"source": "controller-owned-verification"})
105 else:
106 blocker = {"at": now_iso(), "unit_id": unit_id, "reason": "exact pre-fold restoration could not be proven"}
107 doc["blockers"].append(blocker)
108 event(doc, "restore-blocked", unit_id, {"source": "controller-owned-verification"})
109 if not exact:
110 raise Operational("BLOCKED", "exact pre-fold restoration could not be proven")
113def _verification_log(run_id: str, unit_id: str) -> tuple[str, object]:
114 parent = os.path.join(run_dir(run_id), "units", unit_id, "result")
115 validate_private_dir(parent)
116 path = os.path.join(parent, f"host-verification-{secrets.token_hex(6)}.log")
117 fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_NOFOLLOW, 0o600)
118 return path, os.fdopen(fd, "wb")
121def _run_verification_log(run_id: str) -> tuple[str, object]:
122 parent = os.path.join(run_dir(run_id), "jobs")
123 validate_private_dir(parent)
124 path = os.path.join(parent, f"run-verification-{secrets.token_hex(6)}.log")
125 fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_NOFOLLOW, 0o600)
126 return path, os.fdopen(fd, "wb")
129def _validate_accepted_run_head(repo: str, units: dict, current_head: str) -> None:
130 """Require HEAD to be the accepted commit that contains every completed unit."""
131 commits: set[str] = set()
132 for unit in units.values():
133 commit = unit_accepted_commit(unit)
134 if commit is None:
135 raise Operational("BLOCKED", "unit completion evidence changed before plan-wide verification")
136 base = unit.get("workspace", {}).get("base")
137 if not isinstance(base, str) or git_text(repo, "merge-base", base, commit, check=False) != base:
138 raise Operational(
139 "BLOCKED",
140 "controller-accepted unit commit does not descend from its recorded base",
141 {"unit_id": unit.get("unit_id"), "base": base, "accepted_commit": commit},
142 )
143 if commit in commits:
144 raise Operational("BLOCKED", "unit completion evidence contains duplicate accepted commits")
145 commits.add(commit)
147 if current_head not in commits:
148 raise Operational(
149 "BLOCKED",
150 "canonical HEAD no longer matches the final controller-accepted unit commit",
151 {"accepted_heads": sorted(commits), "actual_head": current_head},
152 )
153 if any(git_text(repo, "merge-base", commit, current_head, check=False) != commit for commit in commits):
154 raise Operational(
155 "BLOCKED",
156 "canonical HEAD does not contain every controller-accepted unit",
157 {"accepted_heads": sorted(commits), "actual_head": current_head},
158 )
161def _record_run_verification_attempt(
162 args,
163 attempt_id: str,
164 lock_unit: str,
165 lock_token: str,
166 command: list[str],
167 before: dict,
168 verification_log: str,
169) -> None:
170 with locked_manifest(args.run_id, write=True) as doc:
171 validate_lock(doc, lock_unit, lock_token)
172 doc.setdefault("verification_attempts", [])
173 attempts = plan_wide_verification_attempts(doc)
174 if any(attempt.get("attempt_id") == attempt_id for attempt in attempts):
175 raise TrustFailure("plan-wide verification attempt identity is duplicated")
176 attempts.append({
177 "attempt_id": attempt_id,
178 "started_at": now_iso(),
179 "status": "pending",
180 "integration_lock_nonce": lock_token,
181 "lock_unit_id": lock_unit,
182 "argv": command,
183 "summary": args.verification_summary,
184 "canonical_snapshot": before,
185 "verification_log": verification_log,
186 })
187 event(doc, "run-verification-started", None, {"attempt_id": attempt_id})
190def _record_run_verification_receipt(args, attempt_id: str, lock_token: str, receipt: dict) -> None:
191 with locked_manifest(args.run_id, write=True) as doc:
192 attempts = plan_wide_verification_attempts(doc)
193 matches = [attempt for attempt in attempts if attempt.get("attempt_id") == attempt_id]
194 if len(matches) != 1:
195 raise TrustFailure("plan-wide verification attempt identity is missing or duplicated")
196 attempt = matches[0]
197 if attempt.get("status") != "pending" or attempt.get("integration_lock_nonce") != lock_token:
198 raise TrustFailure("plan-wide verification attempt state or lock identity changed")
199 validate_lock(doc, attempt["lock_unit_id"], lock_token)
200 doc.setdefault("verifications", []).append(receipt)
201 attempt.update({
202 "status": "receipt-recorded",
203 "completed_at": now_iso(),
204 "evidence_digest": receipt["evidence_digest"],
205 })
206 event(doc, "run-verification-passed" if receipt["verification_exit"] == 0 else "run-verification-failed", None, {
207 "attempt_id": attempt_id,
208 "evidence_digest": receipt["evidence_digest"],
209 "verification_exit": receipt["verification_exit"],
210 })
211 if receipt["verification_exit"] != 0:
212 doc["blockers"].append({
213 "at": now_iso(),
214 "unit_id": None,
215 "reason": "plan-wide verification failed",
216 "evidence_digest": receipt["evidence_digest"],
217 })
220def _verify_run_locked(
221 args,
222 repo: str,
223 command: list[str],
224 units: dict,
225 attempt_id: str,
226 lock_unit: str,
227 lock_token: str,
228) -> tuple[str, dict]:
229 before = semantic_snapshot(repo)
230 before_paths = status_paths(repo)
231 if not before["status_empty"] or before_paths:
232 raise Operational("BLOCKED", "verify-run requires a clean canonical checkout")
233 _validate_accepted_run_head(repo, units, before["head"])
234 accepted_units = accepted_unit_commit_snapshot(units)
235 if accepted_units is None:
236 raise Operational("BLOCKED", "unit completion evidence changed before plan-wide verification")
237 before_ignored = inventory_ignored_state(repo)
239 verification_log, stream = _run_verification_log(args.run_id)
240 with stream:
241 _record_run_verification_attempt(
242 args,
243 attempt_id,
244 lock_unit,
245 lock_token,
246 command,
247 before,
248 verification_log,
249 )
250 try:
251 proc = subprocess.run(
252 command,
253 cwd=repo,
254 stdin=subprocess.DEVNULL,
255 stdout=stream,
256 stderr=subprocess.STDOUT,
257 env=sanitized_git_environment({"PYTHONDONTWRITEBYTECODE": "1"}),
258 check=False,
259 )
260 verification_exit = proc.returncode
261 except OSError as exc:
262 stream.write(f"verification launch failed: {exc}\n".encode("utf-8", "replace"))
263 verification_exit = 127
264 test_fault("verify-run-before-receipt")
266 after = semantic_snapshot(repo)
267 after_paths = status_paths(repo)
268 ignored_state = diff_ignored_state(before_ignored, inventory_ignored_state(repo))
269 cleaned_paths: list[str] = []
270 if after != before:
271 if after["branch_ref"] != before["branch_ref"] or after["head"] != before["head"]:
272 with locked_manifest(args.run_id, write=True) as doc:
273 lock = doc.get("integration_lock") or {}
274 blocker = {
275 "at": now_iso(),
276 "unit_id": None,
277 "reason": "plan-wide verification changed canonical branch or HEAD",
278 "retain_integration_lock": True,
279 "integration_lock_nonce": lock.get("nonce"),
280 }
281 doc["blockers"].append(blocker)
282 event(doc, "run-verification-restore-blocked", None, {"verification_exit": verification_exit})
283 raise Operational(
284 "BLOCKED",
285 "plan-wide verification changed canonical branch or HEAD; automatic restoration refused",
286 {
287 "verification_exit": verification_exit,
288 "verification_log": verification_log,
289 "cleaned_paths": cleaned_paths,
290 "ignored_state": ignored_state,
291 "retain_integration_lock": True,
292 },
293 )
294 deletion_paths = after_paths - before_paths
295 cleaned_paths = sorted(deletion_paths)
296 git(repo, "reset", "--hard", before["head"])
297 _remove_owned_new_paths(repo, deletion_paths, before["head"])
298 restored = semantic_snapshot(repo)
299 if restored != before:
300 with locked_manifest(args.run_id, write=True) as doc:
301 lock = doc.get("integration_lock") or {}
302 blocker = {
303 "at": now_iso(),
304 "unit_id": None,
305 "reason": "plan-wide verification restoration could not be proven",
306 "retain_integration_lock": True,
307 "integration_lock_nonce": lock.get("nonce"),
308 }
309 doc["blockers"].append(blocker)
310 event(doc, "run-verification-restore-blocked", None, {"verification_exit": verification_exit})
311 raise Operational(
312 "BLOCKED",
313 "plan-wide verification restoration could not be proven",
314 {
315 "verification_exit": verification_exit,
316 "verification_log": verification_log,
317 "cleaned_paths": cleaned_paths,
318 "ignored_state": ignored_state,
319 "retain_integration_lock": True,
320 },
321 )
323 log_digest = hashlib.sha256(Path(verification_log).read_bytes()).hexdigest()
324 receipt = {
325 "attempt_id": attempt_id,
326 "at": now_iso(),
327 "argv": command,
328 "summary": args.verification_summary,
329 "verification_exit": verification_exit,
330 "log_sha256": log_digest,
331 "canonical_head": before["head"],
332 "accepted_units": accepted_units,
333 "canonical_state_changed": after != before,
334 "cleaned_paths": cleaned_paths,
335 "ignored_state": ignored_state,
336 "verification_log": verification_log if verification_exit != 0 else None,
337 "verification_log_retained": verification_exit != 0,
338 }
339 receipt["evidence_digest"] = digest_bytes(json.dumps(receipt, sort_keys=True, separators=(",", ":")).encode())
340 _record_run_verification_receipt(args, attempt_id, lock_token, receipt)
341 if verification_exit != 0:
342 raise Operational(
343 "BLOCKED",
344 "plan-wide authoritative verification failed",
345 {
346 "verification_exit": verification_exit,
347 "verification_log": verification_log,
348 "evidence_digest": receipt["evidence_digest"],
349 "cleaned_paths": cleaned_paths,
350 "ignored_state": ignored_state,
351 },
352 )
353 os.unlink(verification_log)
354 return "RUN_VERIFIED", {
355 "verification_exit": 0,
356 "evidence_digest": receipt["evidence_digest"],
357 "canonical_head": before["head"],
358 "cleaned_paths": cleaned_paths,
359 "ignored_state": ignored_state,
360 "verification_log_retained": False,
361 }
364def cmd_verify_run(args) -> tuple[str, dict]:
365 """Run a plan-wide gate while holding the canonical integration lock."""
366 command = _verification_command(args, "verify-run")
367 with locked_manifest(args.run_id) as doc:
368 info = validate_repo(doc)
369 units = doc.get("units", {})
370 if not units or any(not unit_ready_for_run_verification(unit) for unit in units.values()):
371 raise Operational(
372 "REFUSED",
373 "verify-run requires every unit to be terminal with an accepted canonical commit",
374 )
375 if doc.get("integration_lock") is not None:
376 raise Operational("BLOCKED", "verify-run requires no active integration lock")
377 repo = info["toplevel"]
378 lock_unit = sorted(units)[-1]
379 acquired = cmd_integration_acquire(_args(
380 run_id=args.run_id,
381 unit_id=lock_unit,
382 resume=False,
383 plan_verification=True,
384 ))[1]
385 token = acquired["lock_token"]
386 attempt_id = secrets.token_hex(16)
387 try:
388 with locked_manifest(args.run_id) as doc:
389 validate_repo(doc)
390 units = doc.get("units", {})
391 if not units or any(not unit_ready_for_run_verification(unit) for unit in units.values()):
392 raise Operational("BLOCKED", "external unit completion evidence changed before plan-wide verification")
393 accepted_units = dict(units)
394 result = _verify_run_locked(
395 args,
396 repo,
397 command,
398 accepted_units,
399 attempt_id,
400 lock_unit,
401 token,
402 )
403 except Operational as exc:
404 with locked_manifest(args.run_id) as doc:
405 lock = doc.get("integration_lock")
406 pending = pending_plan_wide_verification(doc, lock) if isinstance(lock, dict) else None
407 receipted = receipted_plan_wide_verification(doc, lock) if isinstance(lock, dict) else None
408 if not exc.detail.get("retain_integration_lock") and not (
409 pending and pending.get("attempt_id") == attempt_id
410 ):
411 if receipted and receipted.get("attempt_id") == attempt_id:
412 test_fault("verify-run-after-receipt")
413 cmd_integration_release(_args(run_id=args.run_id, unit_id=lock_unit, lock_token=token))
414 raise
415 test_fault("verify-run-after-receipt")
416 cmd_integration_release(_args(run_id=args.run_id, unit_id=lock_unit, lock_token=token))
417 return result
420def _integration_recovery_failure(args, original: Operational, failure: Operational, phase: str) -> Operational:
421 if phase == "restore":
422 reason = "integration failed and exact restoration could not be proven"
423 event_name = "integration-restore-blocked"
424 else:
425 reason = "integration failed after exact restoration but lock release failed"
426 event_name = "integration-release-blocked"
427 detail = {
428 "reason": reason,
429 "unit_id": args.unit_id,
430 "original_failure": str(original),
431 "original_word": original.word,
432 f"{phase}_failure": str(failure),
433 f"{phase}_word": failure.word,
434 "retain_integration_lock": True,
435 "recovery_path": os.path.join(run_dir(args.run_id), "units", args.unit_id),
436 }
437 if "ignored_state" in original.detail:
438 detail["ignored_state"] = original.detail["ignored_state"]
439 with locked_manifest(args.run_id, write=True) as doc:
440 doc["blockers"].append({"at": now_iso(), **detail})
441 event(doc, event_name, args.unit_id, {
442 "original_word": original.word,
443 f"{phase}_word": failure.word,
444 })
445 return Operational("BLOCKED", reason, detail)
448def cmd_integrate(args) -> tuple[str, dict]:
449 command = _verification_command(args)
450 if not args.commit_message.strip() or len(args.commit_message.encode()) > 1024:
451 raise Operational("REFUSED", "commit message must be non-empty and at most 1024 bytes")
453 token = None
454 before = None
455 verification_log = None
456 committed = False
457 ignored_state = None
458 try:
459 acquired = cmd_integration_acquire(_args(run_id=args.run_id, unit_id=args.unit_id, resume=False))[1]
460 token = acquired["lock_token"]
461 cmd_preflight(_args(
462 run_id=args.run_id,
463 unit_id=args.unit_id,
464 lock_token=token,
465 allowed_head=args.allowed_head,
466 ))
467 with locked_manifest(args.run_id) as doc:
468 repo = doc["repository"]["toplevel"]
469 transport = doc["units"][args.unit_id]["transport"]["commit"]
470 git(repo, "cherry-pick", "--no-commit", transport)
471 cmd_mark_applied(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
472 with locked_manifest(args.run_id) as doc:
473 unit = doc["units"][args.unit_id]
474 if not matches_expected_apply(repo, unit):
475 raise Operational("BLOCKED", "canonical apply changed before verification")
476 before = semantic_snapshot(repo)
477 before_paths = status_paths(repo)
478 before_ignored = inventory_ignored_state(repo)
480 verification_log, stream = _verification_log(args.run_id, args.unit_id)
481 with stream:
482 try:
483 proc = subprocess.run(
484 command,
485 cwd=repo,
486 stdin=subprocess.DEVNULL,
487 stdout=stream,
488 stderr=subprocess.STDOUT,
489 env=sanitized_git_environment({"PYTHONDONTWRITEBYTECODE": "1"}),
490 check=False,
491 )
492 verification_exit = proc.returncode
493 except OSError as exc:
494 stream.write(f"verification launch failed: {exc}\n".encode("utf-8", "replace"))
495 verification_exit = 127
496 after = semantic_snapshot(repo)
497 after_paths = status_paths(repo)
498 ignored_state = diff_ignored_state(before_ignored, inventory_ignored_state(repo))
499 verification_failed = verification_exit != 0 or after != before
500 if verification_failed:
501 _restore_owned_verification(args.run_id, args.unit_id, token, before, before_paths, after_paths)
502 cleaned_paths = sorted(after_paths - before_paths)
503 log_digest = hashlib.sha256(Path(verification_log).read_bytes()).hexdigest()
504 if verification_failed:
505 cmd_integration_release(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
506 token = None
507 raise Operational(
508 "BLOCKED",
509 "authoritative verification failed or changed canonical state",
510 {
511 "unit_id": args.unit_id,
512 "verification_exit": verification_exit,
513 "verification_log": verification_log,
514 "canonical_state_changed": after != before,
515 "cleaned_paths": cleaned_paths,
516 "ignored_state": ignored_state,
517 },
518 )
519 evidence = digest_bytes(json.dumps({
520 "argv": command,
521 "exit": verification_exit,
522 "log_sha256": log_digest,
523 "before": before,
524 "after": after,
525 "cleaned_paths": cleaned_paths,
526 "ignored_state": ignored_state,
527 }, sort_keys=True, separators=(",", ":")).encode())
528 cmd_mark_verified(_args(
529 run_id=args.run_id,
530 unit_id=args.unit_id,
531 lock_token=token,
532 evidence_digest=evidence,
533 summary=args.verification_summary,
534 ignored_state=ignored_state,
535 ))
536 test_fault("before-canonical-commit")
537 commit_index_tree(repo, args.commit_message)
538 committed_body = cmd_mark_committed(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))[1]
539 committed = True
540 canonical = committed_body["canonical_commit"]["commit"]
541 test_fault("after-canonical-commit-confirmed")
542 with locked_manifest(args.run_id) as doc:
543 wave_id = doc["units"][args.unit_id].get("wave", {}).get("id")
544 if wave_id:
545 cmd_wave_advance(_args(
546 run_id=args.run_id,
547 unit_id=args.unit_id,
548 lock_token=token,
549 canonical_commit=canonical,
550 ))
551 cmd_cleanup(_args(
552 run_id=args.run_id,
553 unit_id=args.unit_id,
554 abandon=False,
555 expect_transport=None,
556 expect_job=None,
557 ))
558 cmd_integration_release(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
559 token = None
560 return "UNIT_COMMITTED", {
561 "unit_id": args.unit_id,
562 "canonical_commit": canonical,
563 "verification_digest": evidence,
564 "verification_log_retained": False,
565 "cleaned_paths": cleaned_paths,
566 "ignored_state": ignored_state,
567 }
568 except (Operational, TrustFailure) as original:
569 if ignored_state is not None:
570 original.detail.setdefault("ignored_state", ignored_state)
571 if token is not None and committed:
572 detail = {
573 "reason": "canonical commit accepted but post-commit finalization is incomplete",
574 "unit_id": args.unit_id,
575 "canonical_commit": canonical,
576 "original_failure": str(original),
577 "original_word": original.word,
578 "retain_integration_lock": True,
579 "recovery_path": os.path.join(run_dir(args.run_id), "units", args.unit_id),
580 "ignored_state": ignored_state,
581 }
582 with locked_manifest(args.run_id, write=True) as doc:
583 doc["blockers"].append({"at": now_iso(), **detail})
584 event(doc, "post-commit-finalization-blocked", args.unit_id, {
585 "canonical_commit": canonical,
586 "original_word": original.word,
587 })
588 raise Operational(
589 "BLOCKED",
590 "canonical commit accepted but post-commit finalization is incomplete",
591 detail,
592 ) from original
593 if token is not None and original.detail.get("retain_integration_lock"):
594 raise
595 if token is not None:
596 with locked_manifest(args.run_id) as doc:
597 unit = doc["units"].get(args.unit_id)
598 pre_fold = unit.get("integration", {}).get("pre_fold") if unit else None
599 if not pre_fold:
600 cmd_integration_release(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
601 token = None
602 raise
603 try:
604 cmd_restore(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
605 except (Operational, TrustFailure) as restore_failure:
606 raise _integration_recovery_failure(args, original, restore_failure, "restore") from restore_failure
607 try:
608 cmd_integration_release(_args(run_id=args.run_id, unit_id=args.unit_id, lock_token=token))
609 token = None
610 except (Operational, TrustFailure) as release_failure:
611 raise _integration_recovery_failure(args, original, release_failure, "release") from release_failure
612 raise