unit_workspace_lifecycle.py · Compound Engineering

1"""Resume, fallback, reap, and finalized-artifact lifecycle operations."""

2

3from __future__ import annotations

4

5import os

6import re

7import shutil

8import subprocess

9import sys

10from types import SimpleNamespace

11

12from unit_workspace_state import *

13from unit_workspace_jobs import *

14from unit_workspace_integration import *

15

16

17def cmd_status(args) -> tuple[str, dict]:

18 with locked_manifest(args.run_id) as doc:

19 validate_repo(doc)

20 source = doc.get("source") or {"kind": "plan", **doc.get("plan", {})}

21 if args.unit_id:

22 unit = doc["units"].get(args.unit_id)

23 if not unit:

24 raise Operational("REFUSED", "unknown unit")

25 body = {"run_id": args.run_id, "revision": doc["revision"], "source": source, "unit": unit, "integration_lock": doc.get("integration_lock"), "verifications": doc.get("verifications", []), "blockers": doc.get("blockers", [])}

26 else:

27 body = {"run_id": args.run_id, "revision": doc["revision"], "source": source, "units": doc["units"], "integration_lock": doc.get("integration_lock"), "verifications": doc.get("verifications", []), "blockers": doc.get("blockers", []), "recovery_path": run_dir(args.run_id)}

28 return "STATUS", body

29

30

31def unfinished_run(doc: dict, canonical_head: str) -> bool:

32 units = doc.get("units")

33 if not isinstance(units, dict):

34 raise TrustFailure("manifest units are malformed")

35 if not units:

36 return True

37 states: list[str] = []

38 for uid, unit in units.items():

39 if not isinstance(uid, str) or not SAFE_ID.fullmatch(uid) or not isinstance(unit, dict):

40 raise TrustFailure("manifest unit identity or record is malformed")

41 state = unit.get("state")

42 if state not in UNIT_STATES:

43 raise TrustFailure(f"manifest unit state is invalid: {uid}")

44 states.append(state)

45 terminal_states = {"cleaned", "native-completed"}

46 if any(state not in terminal_states for state in states):

47 return True

48 for uid, unit in doc["units"].items():

49 if unit.get("state") != "native-completed":

50 continue

51 attempt = find_attempt(unit)

52 fallback = attempt.get("fallback", {})

53 claim = fallback.get("claimed") if isinstance(fallback, dict) else None

54 completion = fallback.get("completed") if isinstance(fallback, dict) else None

55 claim_valid = isinstance(claim, dict) and claim.get("mode") in {"prefer", "require"}

56 if not (

57 claim_valid

58 and isinstance(completion, dict)

59 and completion.get("claim") == claim

60 and isinstance(completion.get("at"), str)

61 and completion.get("at")

62 and isinstance(completion.get("summary"), str)

63 and completion.get("summary")

64 and isinstance(completion.get("evidence_digest"), str)

65 and re.fullmatch(r"[0-9a-f]{64}", completion["evidence_digest"])

66 and isinstance(completion.get("accepted_head"), str)

67 and re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", completion["accepted_head"])

68 and completion.get("base") == unit.get("workspace", {}).get("base")

69 and isinstance(completion.get("snapshot"), dict)

70 and completion["snapshot"].get("head") == completion["accepted_head"]

71 and completion["snapshot"].get("branch_ref") == doc.get("branch", {}).get("ref")

72 and completion["snapshot"].get("status_empty") is True

73 and completion["snapshot"].get("worktree_index_empty") is True

74 and completion["snapshot"].get("head_tree") == completion["snapshot"].get("index_tree")

75 and completion["snapshot"].get("status_sha256") == digest_bytes(b"")

76 ):

77 raise TrustFailure(f"native fallback completion receipt is malformed: {uid}")

78 receipts = doc.get("verifications", [])

79 if not isinstance(receipts, list) or any(not isinstance(receipt, dict) for receipt in receipts):

80 raise TrustFailure("manifest verification receipts are malformed")

81 accepted_units = accepted_unit_commit_snapshot(units)

82 if accepted_units is None:

83 return True

84 return doc.get("integration_lock") is not None or not any(

85 receipt.get("verification_exit") == 0

86 and receipt.get("accepted_units") == accepted_units

87 and receipt.get("canonical_head") == canonical_head

88 and canonical_head in accepted_units.values()

89 for receipt in receipts

90 )

91

92

93def discover_resume_run(repo: str, plan_digest: str) -> tuple[str, list[dict]]:

94 if not re.fullmatch(r"[0-9a-f]{64}", plan_digest):

95 raise Operational("REFUSED", "plan digest must be a lowercase SHA-256 hex value")

96 ensure_root()

97 info = repo_info(repo)

98 candidates: list[dict] = []

99 # A run recorded under the other candidate root (sandboxed vs unsandboxed

100 # session) must still be discoverable; scan every candidate that exists.

101 # Read-only: repairing a root this session cannot write (a leftover /tmp

102 # tree under the sandbox) would abort discovery before the writable one.

103 entries = []

104 for root in candidate_runs_roots():

105 if not os.path.isdir(root) or os.path.islink(root):

106 continue

107 entries.extend(os.scandir(root))

108 for entry in sorted(entries, key=lambda row: row.path):

109 if entry.name == ".locks":

110 continue

111 if not entry.is_dir(follow_symlinks=False):

112 raise TrustFailure(f"unexpected non-directory entry in run root: {entry.path}")

113 if not SAFE_ID.fullmatch(entry.name) or not entry.name.strip("."):

114 raise TrustFailure(f"unsafe run entry name: {entry.path}")

115 validate_private_dir(entry.path)

116 doc = read_private_json(os.path.join(entry.path, "manifest.json"))

117 if doc.get("schema_version") != SCHEMA_VERSION or doc.get("run_id") != entry.name:

118 raise TrustFailure(f"manifest schema or run identity mismatch: {entry.path}")

119 repository = doc.get("repository")

120 branch = doc.get("branch")

121 plan = doc.get("plan")

122 source = doc.get("source")

123 if not isinstance(repository, dict) or not isinstance(branch, dict) or not isinstance(plan, dict):

124 raise TrustFailure(f"manifest repository, branch, or plan record is malformed: {entry.path}")

125 if (

126 repository.get("identity_digest") != info["identity_digest"]

127 or repository.get("toplevel") != info["toplevel"]

128 or repository.get("git_dir") != info["git_dir"]

129 or branch.get("ref") != info["branch_ref"]

130 ):

131 continue

132 if source is not None and not isinstance(source, dict):

133 raise TrustFailure(f"manifest source record is malformed: {entry.path}")

134 source_kind = source.get("kind") if isinstance(source, dict) else plan.get("kind", "plan")

135 source_digest = source.get("digest") if isinstance(source, dict) else plan.get("digest")

136 if source_kind != "plan" or source_digest != plan_digest:

137 continue

138 validate_source(doc)

139 if unfinished_run(doc, info["head"]):

140 candidates.append({

141 "run_id": entry.name,

142 "updated_at": doc.get("updated_at"),

143 "recovery_path": entry.path,

144 "unit_states": {uid: unit.get("state") for uid, unit in doc["units"].items()},

145 })

146 if not candidates:

147 raise Operational("NOT_FOUND", "no unfinished run matches repository, branch, and plan digest", {"candidates": []})

148 if len(candidates) > 1:

149 raise Operational("AMBIGUOUS", "multiple unfinished runs match; pass --run-id", {"candidates": candidates})

150 return candidates[0]["run_id"], candidates

151

152

153def resolve_resume_run(args) -> str:

154 if args.run_id:

155 if args.repo or args.plan_digest:

156 raise Operational("REFUSED", "resume accepts --run-id alone or both --repo and --plan-digest")

157 return safe_id(args.run_id, "run id")

158 if not args.repo or not args.plan_digest:

159 raise Operational("REFUSED", "resume requires --run-id or both --repo and --plan-digest")

160 run_id, _ = discover_resume_run(args.repo, args.plan_digest)

161 return run_id

162

163

164def retained_worker_blocker(run_id: str, unit_id: str, error: Operational) -> dict | None:

165 if error.word != "BLOCKED" or str(error) != "worker returned a host-resolvable blocker":

166 return None

167 with locked_manifest(run_id) as doc:

168 unit = doc["units"].get(unit_id)

169 if not unit or unit.get("state") != "authored":

170 return None

171 attempt = find_attempt(unit)

172 receipt = attempt.get("terminal_receipt")

173 if (

174 attempt.get("process_state") != "done"

175 or attempt.get("terminal_validation_failure") is not None

176 or not isinstance(receipt, dict)

177 or receipt.get("terminal_status") != "blocked"

178 ):

179 return None

180 blocker = {

181 "unit_id": unit_id,

182 "terminal_status": "blocked",

183 "summary": receipt.get("summary", ""),

184 "terminal_receipt": receipt,

185 "recovery_path": os.path.join(run_dir(run_id), "units", unit_id),

186 }

187 return blocker if error.detail == blocker else None

188

189

190def resume_terminalize(run_id: str, unit_id: str) -> list[dict]:

191 try:

192 transport = terminalize(run_id, unit_id)

193 except Operational as exc:

194 blocker = retained_worker_blocker(run_id, unit_id, exc)

195 if blocker is None:

196 raise

197 return [{

198 "unit_id": unit_id,

199 "action": "worker-blocker-retained",

200 "terminal_status": blocker["terminal_status"],

201 "summary": blocker["summary"],

202 "recovery_path": blocker["recovery_path"],

203 }]

204 return [{"unit_id": unit_id, "action": "terminalized", "transport": transport["commit"]}]

205

206

207def resume_monitor(run_id: str, unit_id: str) -> list[dict]:

208 evidence = sync_job(run_id, unit_id)

209 actions = [{"unit_id": unit_id, "action": "monitored", "process_state": evidence["process_state"]}]

210 if evidence["process_state"] == "done":

211 actions.extend(resume_terminalize(run_id, unit_id))

212 return actions

213

214

215def resolve_unit_recovery_blockers(run_id: str, unit_id: str, reason: str | None = None) -> None:

216 with locked_manifest(run_id, write=True) as doc:

217 resolved = 0

218 for blocker in doc.get("blockers", []):

219 if (

220 blocker.get("unit_id") == unit_id

221 and blocker.get("retain_integration_lock") is True

222 and (reason is None or blocker.get("reason") == reason)

223 and not blocker.get("resolved_at")

224 ):

225 blocker["resolved_at"] = now_iso()

226 blocker["resolved_by"] = "resume"

227 resolved += 1

228 if resolved:

229 event(doc, "recovery-blockers-resolved", unit_id, {"count": resolved})

230

231

232def plan_wide_verification_attempts(doc: dict) -> list[dict]:

233 attempts = doc.get("verification_attempts", [])

234 if not isinstance(attempts, list) or any(not isinstance(attempt, dict) for attempt in attempts):

235 raise TrustFailure("manifest plan-wide verification attempts are malformed")

236 for attempt in attempts:

237 if (

238 not isinstance(attempt.get("attempt_id"), str)

239 or not re.fullmatch(r"[0-9a-f]{32}", attempt["attempt_id"])

240 or attempt.get("status") not in {"pending", "receipt-recorded"}

241 or not isinstance(attempt.get("integration_lock_nonce"), str)

242 or not re.fullmatch(r"[0-9a-f]{48}", attempt["integration_lock_nonce"])

243 or not isinstance(attempt.get("lock_unit_id"), str)

244 or not SAFE_ID.fullmatch(attempt["lock_unit_id"])

245 ):

246 raise TrustFailure("manifest plan-wide verification attempt identity or state is malformed")

247 return attempts

248

249

250def pending_plan_wide_verification(doc: dict, lock: dict) -> dict | None:

251 attempts = plan_wide_verification_attempts(doc)

252 pending = [

253 attempt for attempt in attempts

254 if attempt.get("status") == "pending"

255 and attempt.get("integration_lock_nonce") == lock.get("nonce")

256 and attempt.get("lock_unit_id") == lock.get("unit_id")

257 ]

258 if len(pending) > 1:

259 raise TrustFailure("multiple pending plan-wide verification attempts share one integration lock")

260 return pending[0] if pending else None

261

262

263def receipted_plan_wide_verification(doc: dict, lock: dict) -> dict | None:

264 attempts = plan_wide_verification_attempts(doc)

265 recorded = [

266 attempt for attempt in attempts

267 if attempt.get("status") == "receipt-recorded"

268 and attempt.get("integration_lock_nonce") == lock.get("nonce")

269 and attempt.get("lock_unit_id") == lock.get("unit_id")

270 ]

271 if len(recorded) > 1:

272 raise TrustFailure("multiple receipted plan-wide verification attempts share one integration lock")

273 if not recorded:

274 return None

275 evidence_digest = recorded[0].get("evidence_digest")

276 if not isinstance(evidence_digest, str) or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):

277 raise TrustFailure("plan-wide verification attempt receipt digest is malformed")

278 verifications = doc.get("verifications", [])

279 if not isinstance(verifications, list) or any(not isinstance(receipt, dict) for receipt in verifications):

280 raise TrustFailure("manifest verification receipts are malformed")

281 receipts = [

282 receipt for receipt in verifications

283 if receipt.get("evidence_digest") == evidence_digest

284 ]

285 if len(receipts) != 1:

286 raise TrustFailure("plan-wide verification receipt is missing or duplicated")

287 return recorded[0]

288

289

290def plan_wide_blocker_retains_lock(doc: dict, lock: dict) -> bool:

291 return any(

292 blocker.get("unit_id") is None

293 and blocker.get("retain_integration_lock") is True

294 and blocker.get("integration_lock_nonce") == lock.get("nonce")

295 and not blocker.get("resolved_at")

296 for blocker in doc.get("blockers", [])

297 )

298

299

300def resume_finalize_committed(run_id: str, unit_id: str) -> list[dict]:

301 with locked_manifest(run_id) as doc:

302 unit = doc["units"][unit_id]

303 state = unit["state"]

304 lock = doc.get("integration_lock")

305 pending_plan_verification = pending_plan_wide_verification(doc, lock) if lock else None

306 receipted_plan_verification = receipted_plan_wide_verification(doc, lock) if lock else None

307 retained_plan_lock = bool(lock and plan_wide_blocker_retains_lock(doc, lock))

308 wave_id = unit.get("wave", {}).get("id")

309 cleanup = unit.get("cleanup") or {}

310 artifact_cleanup_complete = cleanup.get("artifact_cleanup", {}).get("complete") is True

311 if state in {"cleaned", "native-completed"}:

312 actions: list[dict] = []

313 if (state == "cleaned" or cleanup) and not artifact_cleanup_complete:

314 cmd_cleanup(SimpleNamespace(

315 run_id=run_id,

316 unit_id=unit_id,

317 abandon=False,

318 expect_transport=None,

319 expect_job=None,

320 ))

321 actions.append({"unit_id": unit_id, "action": "artifact-cleanup-reconciled"})

322 if lock and lock.get("unit_id") == unit_id:

323 if pending_plan_verification:

324 raise Operational(

325 "BLOCKED",

326 "pending plan-wide verification retains the canonical integration lock",

327 {

328 "unit_id": unit_id,

329 "verification_attempt_id": pending_plan_verification.get("attempt_id"),

330 "retain_integration_lock": True,

331 },

332 )

333 if retained_plan_lock:

334 raise Operational(

335 "BLOCKED",

336 "plan-wide verification blocker retains the canonical integration lock",

337 {"unit_id": unit_id, "retain_integration_lock": True},

338 )

339 if state == "native-completed" and not receipted_plan_verification:

340 raise Operational(

341 "BLOCKED",

342 "native-completed unit retains the canonical integration lock without a plan-wide verification receipt",

343 {"unit_id": unit_id, "retain_integration_lock": True},

344 )

345 cmd_integration_release(SimpleNamespace(run_id=run_id, unit_id=unit_id, lock_token=lock["nonce"]))

346 actions.append({"unit_id": unit_id, "action": "integration-release-reconciled"})

347 return actions

348 canonical_record = unit.get("integration", {}).get("canonical_commit")

349 canonical = canonical_record.get("commit") if isinstance(canonical_record, dict) else None

350 if state != "committed" or not canonical:

351 raise Operational("BLOCKED", "committed-unit recovery lacks an accepted canonical commit")

352 if lock is None:

353 lock_token = cmd_integration_acquire(SimpleNamespace(run_id=run_id, unit_id=unit_id, resume=False))[1]["lock_token"]

354 elif lock.get("unit_id") == unit_id:

355 lock_token = lock["nonce"]

356 with locked_manifest(run_id) as doc:

357 validate_lock(doc, unit_id, lock_token)

358 else:

359 raise Operational("BLOCKED", "another unit holds the canonical integration lock")

360 actions: list[dict] = []

361 if wave_id:

362 cmd_wave_advance(SimpleNamespace(

363 run_id=run_id,

364 unit_id=unit_id,

365 lock_token=lock_token,

366 canonical_commit=canonical,

367 ))

368 actions.append({"unit_id": unit_id, "action": "wave-advance-reconciled", "commit": canonical})

369 cmd_cleanup(SimpleNamespace(

370 run_id=run_id,

371 unit_id=unit_id,

372 abandon=False,

373 expect_transport=None,

374 expect_job=None,

375 ))

376 cmd_integration_release(SimpleNamespace(run_id=run_id, unit_id=unit_id, lock_token=lock_token))

377 resolve_unit_recovery_blockers(run_id, unit_id)

378 actions.append({"unit_id": unit_id, "action": "committed-unit-finalized", "commit": canonical})

379 return actions

380

381

382def cmd_resume(args) -> tuple[str, dict]:

383 run_id = resolve_resume_run(args)

384 actions: list[dict] = []

385 with locked_manifest(run_id) as doc:

386 validate_repo(doc)

387 unit_ids = list(doc["units"])

388 claim = doc.get("integration_lock")

389 releasing = dict(claim) if isinstance(claim, dict) and claim.get("phase") == "releasing" else None

390 orphan_unit = None

391 orphan_path = integration_lock_path(doc)

392 if claim is None and os.path.lexists(orphan_path):

393 orphan = read_integration_lock(orphan_path)

394 candidate = orphan.get("unit_id")

395 if orphan.get("run_id") != run_id or not isinstance(candidate, str) or candidate not in doc["units"]:

396 raise Operational(

397 "BLOCKED",

398 "external integration lock does not belong to this run/unit",

399 {"owner_run": orphan.get("run_id"), "owner_unit": candidate},

400 )

401 validated_lock_nonce(doc, candidate, orphan)

402 orphan_unit = candidate

403 if releasing:

404 integration_release(run_id, releasing["unit_id"], releasing["nonce"])

405 actions.append({"unit_id": releasing["unit_id"], "action": "integration-release-reconciled"})

406 if orphan_unit:

407 cmd_integration_acquire(SimpleNamespace(

408 run_id=run_id,

409 unit_id=orphan_unit,

410 resume=True,

411 recover_only=True,

412 ))

413 actions.append({"unit_id": orphan_unit, "action": "integration-lock-adopted"})

414 for uid in unit_ids:

415 with locked_manifest(run_id) as doc:

416 unit = doc["units"][uid]

417 state = unit["state"]

418 attempt = find_attempt(unit)

419 lock = doc.get("integration_lock")

420 if state == "queued" and not attempt.get("job_id"):

421 matches = matching_runner_jobs(run_id, unit)

422 if len(matches) > 1:

423 raise Operational("AMBIGUOUS", f"multiple runner jobs match queued unit {uid}")

424 if len(matches) == 1:

425 with locked_manifest(run_id, write=True) as current:

426 current_unit = current["units"][uid]

427 current_attempt = find_attempt(current_unit)

428 if current_attempt.get("job_id") not in (None, matches[0]):

429 raise Operational("AMBIGUOUS", "attempt was concurrently bound")

430 current_attempt["job_id"] = matches[0]

431 current_unit["state"] = "authoring"

432 event(current, "job-adopted", uid, {"job_id": matches[0]})

433 actions.append({"unit_id": uid, "action": "job-adopted", "job_id": matches[0]})

434 actions.extend(resume_monitor(run_id, uid))

435 elif state == "authoring" and attempt.get("job_id"):

436 actions.extend(resume_monitor(run_id, uid))

437 elif state == "authored":

438 actions.extend(resume_terminalize(run_id, uid))

439 elif state == "restoring" and lock and lock.get("unit_id") == uid:

440 exact = restore(run_id, uid, lock["nonce"])

441 if not exact:

442 raise Operational("BLOCKED", "exact pre-fold preservation could not be proven")

443 integration_release(run_id, uid, lock["nonce"])

444 resolve_unit_recovery_blockers(

445 run_id,

446 uid,

447 reason="integration failed and exact restoration could not be proven",

448 )

449 actions.append({

450 "unit_id": uid,

451 "action": "restored",

452 "canonical_preserved": True,

453 "integration_lock_released": True,

454 })

455 elif state == "preserved" and lock and lock.get("unit_id") == uid:

456 validate_lock(doc, uid, lock["nonce"])

457 restore_evidence = unit.get("integration", {}).get("restore")

458 pre_fold = unit.get("integration", {}).get("pre_fold")

459 if (

460 not isinstance(restore_evidence, dict)

461 or restore_evidence.get("exact") is not True

462 or not isinstance(pre_fold, dict)

463 or restore_evidence.get("snapshot") != pre_fold

464 ):

465 raise Operational("BLOCKED", "preserved-unit recovery lacks exact restoration evidence")

466 if semantic_snapshot(doc["repository"]["toplevel"]) != pre_fold:

467 raise Operational("BLOCKED", "canonical checkout no longer matches the exact restored snapshot")

468 integration_release(run_id, uid, lock["nonce"])

469 resolve_unit_recovery_blockers(

470 run_id,

471 uid,

472 reason="integration failed after exact restoration but lock release failed",

473 )

474 actions.append({"unit_id": uid, "action": "integration-release-reconciled"})

475 elif state == "integration-pending" and not unit["integration"].get("pre_fold") and lock and lock.get("unit_id") == uid:

476 validate_lock(doc, uid, lock["nonce"])

477 integration_release(run_id, uid, lock["nonce"])

478 actions.append({"unit_id": uid, "action": "preflight-lock-released"})

479 elif state == "integration-pending" and unit["integration"].get("pre_fold") and lock and lock.get("unit_id") == uid:

480 validate_lock(doc, uid, lock["nonce"])

481 repo = doc["repository"]["toplevel"]

482 snap = semantic_snapshot(repo)

483 if snap == unit["integration"]["pre_fold"]:

484 exact = restore(run_id, uid, lock["nonce"])

485 if not exact:

486 raise Operational("BLOCKED", "exact pre-fold preservation could not be proven")

487 integration_release(run_id, uid, lock["nonce"])

488 actions.append({

489 "unit_id": uid,

490 "action": "preflight-exact-state-recovered",

491 "canonical_preserved": True,

492 "integration_lock_released": True,

493 })

494 else:

495 if not matches_expected_apply(repo, unit, snap):

496 raise Operational("BLOCKED", "canonical dirt does not match the expected in-flight transport; preserved for recovery")

497 with locked_manifest(run_id, write=True) as current:

498 current_unit = current["units"][uid]

499 current_unit["state"] = "integrated"

500 current_unit["integration"]["applied"] = {

501 "at": now_iso(),

502 "post_index_tree": snap["index_tree"],

503 "status_sha256": snap["status_sha256"],

504 "reconciled": True,

505 }

506 event(current, "transport-apply-reconciled", uid, {"post_index_tree": snap["index_tree"]})

507 actions.append({"unit_id": uid, "action": "apply-reconciled"})

508 elif state in {"integrated", "verified"} and lock and lock.get("unit_id") == uid:

509 validate_lock(doc, uid, lock["nonce"])

510 commit = None

511 if state == "verified":

512 with locked_manifest(run_id) as current:

513 commit = reconcile_commit(current, current["units"][uid])

514 if commit:

515 with locked_manifest(run_id, write=True) as current:

516 current["units"][uid]["integration"]["canonical_commit"] = commit

517 current["units"][uid]["state"] = "committed"

518 event(current, "canonical-commit-reconciled", uid, {"commit": commit["commit"]})

519 actions.append({"unit_id": uid, "action": "commit-reconciled", "commit": commit["commit"]})

520 actions.extend(resume_finalize_committed(run_id, uid))

521 else:

522 exact = restore(run_id, uid, lock["nonce"])

523 if not exact:

524 raise Operational("BLOCKED", "exact pre-fold preservation could not be proven")

525 integration_release(run_id, uid, lock["nonce"])

526 actions.append({

527 "unit_id": uid,

528 "action": "pre-commit-integration-restored",

529 "interrupted_state": state,

530 "canonical_preserved": True,

531 "integration_lock_released": True,

532 })

533 elif state in {"committed", "cleaned", "native-completed"}:

534 actions.extend(resume_finalize_committed(run_id, uid))

535 return "RESUMED", {"run_id": run_id, "actions": actions, "redispatched": False, "applied": False}

536

537

538def fallback_basis(doc: dict, unit: dict) -> tuple[str, dict]:

539 if unit.get("state") == "integration-pending" and unit.get("transport", {}).get("commit"):

540 raise Operational("REFUSED", "pinned worker transport must be reconciled rather than bypassed by fallback")

541 attempt = find_attempt(unit)

542 process_state = attempt.get("process_state")

543 if process_state == "done" and attempt.get("terminal_validation_failure"):

544 validate_terminal_validation_failure(doc["run_id"], unit, attempt)

545 snap = semantic_snapshot(doc["repository"]["toplevel"])

546 allowed_heads = set(unit.get("wave", {}).get("allowed_heads", []))

547 if (

548 snap["head"] not in allowed_heads

549 and not dependency_advanced_head(doc, unit, snap["head"])

550 ) or not snap["status_empty"] or snap["index_tree"] != snap["head_tree"]:

551 raise Operational("BLOCKED", "canonical checkout diverged or is dirty; native fallback is not safe")

552 return "terminal-validation-failure", attempt

553 if process_state in TERMINAL_PROCESS - {"done"} or (process_state == "never-started" and attempt.get("job_id")):

554 snap = semantic_snapshot(doc["repository"]["toplevel"])

555 allowed_heads = set(unit.get("wave", {}).get("allowed_heads", []))

556 if (

557 snap["head"] not in allowed_heads

558 and not dependency_advanced_head(doc, unit, snap["head"])

559 ) or not snap["status_empty"] or snap["index_tree"] != snap["head_tree"]:

560 raise Operational("BLOCKED", "canonical checkout diverged or is dirty; native fallback is not safe")

561 recorded = attempt.get("terminal_receipt")

562 if process_state == "failed" and isinstance(recorded, dict) and recorded.get("terminal_status") in {"unavailable", "failed"}:

563 reader = (

564 unavailable_terminal_receipt

565 if recorded["terminal_status"] == "unavailable"

566 else launched_failure_terminal_receipt

567 )

568 observed = reader(doc["run_id"], unit, attempt)

569 if observed != recorded:

570 raise Operational("BLOCKED", "recorded failed receipt evidence changed")

571 return observed["failure_reason"], attempt

572 return str(attempt.get("fallback", {}).get("reason") or process_state), attempt

573 restore_evidence = unit.get("integration", {}).get("restore")

574 if unit.get("state") == "preserved" and restore_evidence and restore_evidence.get("exact") is True:

575 if doc.get("integration_lock"):

576 raise Operational("REFUSED", "release the integration lock after exact restoration before fallback")

577 actual = semantic_snapshot(doc["repository"]["toplevel"])

578 expected = unit["integration"].get("pre_fold")

579 if actual != expected:

580 raise Operational("BLOCKED", "canonical checkout no longer matches the exact restored snapshot")

581 return "canonical-attempt-preserved", attempt

582 if process_state == "running":

583 raise Operational("REFUSED", "a live attempt still owns implementation; fallback is not authorized")

584 if process_state == "done":

585 raise Operational("REFUSED", "successful worker output must be reconciled rather than bypassed by fallback")

586 raise Operational("REFUSED", "no authoritative terminal or exactly restored attempt authorizes fallback")

587

588

589def cmd_claim_fallback(args) -> tuple[str, dict]:

590 # Refresh runner evidence first. A stale manifest cannot authorize native

591 # work while a detached attempt may still be live.

592 with locked_manifest(args.run_id) as doc:

593 validate_repo(doc)

594 unit = doc["units"].get(args.unit_id)

595 if not unit:

596 raise Operational("REFUSED", "unknown unit")

597 attempt = find_attempt(unit)

598 should_sync = bool(attempt.get("job_id")) and unit.get("state") == "authoring"

599 if should_sync:

600 sync_job(args.run_id, args.unit_id)

601

602 with locked_manifest(args.run_id, write=True) as doc:

603 validate_repo(doc)

604 unit = doc["units"].get(args.unit_id)

605 if not unit:

606 raise Operational("REFUSED", "unknown unit")

607 attempt = find_attempt(unit)

608 fallback = attempt.setdefault("fallback", {})

609 fallback.setdefault("completed", None)

610 claimed = fallback.get("claimed")

611 if claimed:

612 return "FALLBACK_ALREADY_AUTHORIZED", {

613 "unit_id": args.unit_id,

614 "start_native": False,

615 "reason": claimed["reason"],

616 "claim": claimed,

617 }

618 validate_dependencies_ready(doc, unit)

619 reason, attempt = fallback_basis(doc, unit)

620 claim_snapshot = semantic_snapshot(doc["repository"]["toplevel"])

621 wave = unit.get("wave", {})

622 if wave.get("id"):

623 validate_wave_order(doc, unit)

624 allowed_heads = wave.get("allowed_heads", [])

625 if not allowed_heads or (

626 claim_snapshot["head"] != allowed_heads[-1]

627 and not dependency_advanced_head(doc, unit, claim_snapshot["head"])

628 ):

629 raise Operational("BLOCKED", "native fallback must start from the latest recorded wave head")

630 mode = doc.get("binding", {}).get("mode")

631 if mode not in {"prefer", "require"}:

632 raise Operational("REFUSED", f"binding mode {mode!r} does not authorize native fallback")

633 claim = {

634 "at": now_iso(),

635 "reason": reason,

636 "caller_mode": args.caller_mode,

637 "mode": mode,

638 "canonical_head": claim_snapshot["head"],

639 }

640 fallback.update({"eligible": False, "reason": reason, "claimed": claim})

641 event(doc, "native-fallback-authorized", args.unit_id, {"reason": reason, "mode": mode, "caller_mode": args.caller_mode})

642 return "FALLBACK_AUTHORIZED", {"unit_id": args.unit_id, "start_native": True, "reason": reason, "claim": claim}

643

644

645def validate_fallback_ancestry(doc: dict, unit: dict, accepted_head: str) -> None:

646 required: list[dict] = []

647 dependency_ids = set(unit.get("dependencies", []))

648 for dependency_id in unit.get("dependencies", []):

649 dependency = doc.get("units", {}).get(dependency_id)

650 accepted_commit = unit_accepted_commit(dependency) if isinstance(dependency, dict) else None

651 if accepted_commit is None:

652 raise Operational(

653 "BLOCKED",

654 "unit dependency completion evidence changed before native fallback completion",

655 {"unit_id": unit["unit_id"], "dependency_id": dependency_id},

656 )

657 required.append({"kind": "dependency", "unit_id": dependency_id, "commit": accepted_commit})

658

659 wave = unit.get("wave", {})

660 if wave.get("id"):

661 wave_members(doc, unit)

662 base = wave.get("base")

663 for head in wave.get("allowed_heads", []):

664 if head != base:

665 required.append({"kind": "wave-head", "commit": head})

666

667 represented_commits = {item["commit"] for item in required}

668 for accepted_unit_id in sorted(doc.get("units", {})):

669 if accepted_unit_id == unit["unit_id"] or accepted_unit_id in dependency_ids:

670 continue

671 accepted_unit = doc["units"][accepted_unit_id]

672 accepted_commit = unit_accepted_commit(accepted_unit)

673 if accepted_commit is None or accepted_commit in represented_commits:

674 continue

675 required.append({"kind": "accepted-unit", "unit_id": accepted_unit_id, "commit": accepted_commit})

676 represented_commits.add(accepted_commit)

677

678 missing = [

679 item for item in required

680 if git_text(

681 doc["repository"]["toplevel"], "merge-base", item["commit"], accepted_head, check=False,

682 ) != item["commit"]

683 ]

684 if missing:

685 raise Operational(

686 "BLOCKED",

687 "accepted native fallback head does not contain every controller-accepted prerequisite",

688 {"unit_id": unit["unit_id"], "accepted_head": accepted_head, "missing_ancestry": missing},

689 )

690

691

692def cmd_complete_fallback(args) -> tuple[str, dict]:

693 if not re.fullmatch(r"[0-9a-f]{64}", args.evidence_digest):

694 raise Operational("REFUSED", "native fallback evidence digest must be lowercase SHA-256 hex")

695 summary = args.summary.strip()

696 if not summary or "\0" in summary or len(summary.encode()) > 1024:

697 raise Operational("REFUSED", "native fallback summary must be non-empty and at most 1024 bytes")

698 if not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", args.accepted_head):

699 raise Operational("REFUSED", "native fallback accepted head must be a Git object id")

700

701 with locked_manifest(args.run_id, write=True) as doc:

702 validate_repo(doc)

703 unit = doc["units"].get(args.unit_id)

704 if not unit:

705 raise Operational("REFUSED", "unknown unit")

706 attempt = find_attempt(unit)

707 fallback = attempt.get("fallback")

708 claim = fallback.get("claimed") if isinstance(fallback, dict) else None

709 if not isinstance(claim, dict):

710 raise Operational("REFUSED", "native fallback completion requires an existing claim")

711 if fallback.get("completed") is not None or unit.get("state") == "native-completed":

712 raise Operational("REFUSED", "native fallback completion was already recorded")

713 claim_mode = claim.get("mode")

714 if claim_mode not in {"prefer", "require"}:

715 raise Operational("REFUSED", "native fallback completion requires an authorized prefer or require claim")

716 if doc.get("integration_lock") is not None:

717 raise Operational("REFUSED", "release the integration lock before completing native fallback")

718

719 repo = doc["repository"]["toplevel"]

720 snapshot = semantic_snapshot(repo)

721 if snapshot.get("branch_ref") != doc["branch"]["ref"]:

722 raise Operational("BLOCKED", "canonical branch changed before native fallback completion")

723 if snapshot.get("status_empty") is not True:

724 raise Operational("BLOCKED", "commit or restore canonical changes before completing native fallback")

725 accepted_commit = git_text(repo, "rev-parse", "--verify", f"{args.accepted_head}^{{commit}}", check=False)

726 if accepted_commit != args.accepted_head or snapshot.get("head") != args.accepted_head:

727 raise Operational("BLOCKED", "accepted native fallback head does not match canonical HEAD")

728 base = unit.get("workspace", {}).get("base")

729 if not isinstance(base, str) or git_text(repo, "merge-base", base, args.accepted_head, check=False) != base:

730 raise Operational("BLOCKED", "accepted native fallback head does not descend from the recorded unit base")

731 validate_fallback_ancestry(doc, unit, args.accepted_head)

732

733 wave = unit.get("wave", {})

734 changed_paths: list[str] = []

735 advanced: list[str] = []

736 if wave.get("id"):

737 validate_wave_order(doc, unit)

738 claim_head = claim.get("canonical_head")

739 allowed_heads = wave.get("allowed_heads", [])

740 if (

741 not isinstance(claim_head, str)

742 or not allowed_heads

743 or claim_head != allowed_heads[-1]

744 or git_text(repo, "merge-base", claim_head, args.accepted_head, check=False) != claim_head

745 ):

746 raise Operational("BLOCKED", "native fallback completion does not extend the latest recorded wave head")

747 raw = git(repo, "diff-tree", "-r", "-M", "--name-status", "-z", claim_head, args.accepted_head)

748 changed_paths = parse_diff_paths(raw)

749 validate_wave_collisions(

750 doc,

751 unit,

752 overrides={unit["unit_id"]: set(changed_paths)},

753 require_complete=False,

754 )

755 members = wave_members(doc, unit)

756 validate_wave_advancement(members, unit, claim_head, args.accepted_head)

757

758 receipt = {

759 "at": now_iso(),

760 "base": base,

761 "accepted_head": args.accepted_head,

762 "evidence_digest": args.evidence_digest,

763 "summary": summary,

764 "snapshot": snapshot,

765 "claim": dict(claim),

766 "changed_paths": changed_paths,

767 }

768 fallback["completed"] = receipt

769 unit["state"] = "native-completed"

770 if wave.get("id"):

771 advanced = advance_wave_allowed_heads(members, wave["position"], args.accepted_head)

772 event(doc, "wave-advanced", args.unit_id, {

773 "canonical_commit": args.accepted_head,

774 "eligible_siblings": advanced,

775 })

776 event(doc, "native-fallback-completed", args.unit_id, {

777 "accepted_head": args.accepted_head,

778 "evidence_digest": args.evidence_digest,

779 })

780 return "FALLBACK_COMPLETED", {

781 "unit_id": args.unit_id,

782 "completion": receipt,

783 "eligible_siblings": advanced,

784 }

785

786

787def cmd_reap(args) -> tuple[str, dict]:

788 with locked_manifest(args.run_id) as doc:

789 validate_repo(doc)

790 unit = doc["units"].get(args.unit_id)

791 if not unit:

792 raise Operational("REFUSED", "unknown unit")

793 attempt = find_attempt(unit)

794 if not attempt.get("job_id"):

795 return "REAPED", {"unit_id": args.unit_id, "process_state": "never-started"}

796 job_dir = runner_job_dir(args.run_id, attempt["job_id"])

797 runner = os.path.join(os.path.dirname(__file__), "peer-job-runner.py")

798 proc = subprocess.run([sys.executable, runner, "reap", job_dir], capture_output=True, check=False)

799 if proc.returncode not in (0,):

800 raise Operational("BLOCKED", f"runner reap failed: {proc.stderr.decode('utf-8', 'replace').strip()}")

801 evidence = sync_job(args.run_id, args.unit_id)

802 return "REAPED", {"unit_id": args.unit_id, **evidence, "recovery_path": os.path.join(run_dir(args.run_id), "units", args.unit_id)}

803

804

805def remove_finalized_artifacts(run_id: str, unit_id: str) -> None:

806 """Prune bulky controller-owned artifacts only after a unit is finalized."""

807 with locked_manifest(run_id) as doc:

808 unit = doc["units"].get(unit_id)

809 cleanup = unit.get("cleanup") if unit else None

810 if (

811 not unit

812 or unit.get("state") not in {"cleaned", "native-completed"}

813 or not isinstance(cleanup, dict)

814 ):

815 raise Operational("REFUSED", "artifact pruning requires a finalized unit with recorded cleanup")

816 attempt_job_ids = [attempt.get("job_id") for attempt in unit.get("attempts", []) if attempt.get("job_id")]

817 authorization_paths = [attempt.get("authorization_path") for attempt in unit.get("attempts", []) if attempt.get("authorization_path")]

818 packet_path = unit.get("packet", {}).get("path")

819 result_dir = os.path.join(os.path.dirname(unit["workspace"]["path"]), "result")

820 root = run_dir(run_id)

821 paths = [(packet_path, "file"), (result_dir, "dir")]

822 paths.extend((authorization_path, "file") for authorization_path in authorization_paths)

823 paths.extend((runner_job_dir(run_id, job_id), "dir") for job_id in attempt_job_ids)

824 for candidate, kind in paths:

825 if not candidate or not os.path.lexists(candidate):

826 continue

827 absolute = os.path.abspath(candidate)

828 if os.path.commonpath([root, absolute]) != root or absolute == root:

829 raise Operational("BLOCKED", "finalized artifact path escaped the owned run")

830 if kind == "file":

831 read_private(absolute, MAX_PACKET_BYTES)

832 os.unlink(absolute)

833 else:

834 validate_private_dir(absolute)

835 shutil.rmtree(absolute)

836 with locked_manifest(run_id, write=True) as doc:

837 unit = doc["units"][unit_id]

838 unit["packet"]["retained"] = False

839 for attempt in unit.get("attempts", []):

840 attempt["bulky_artifacts_retained"] = False

841 attempt["authorization_retained"] = False

842 unit["cleanup"]["artifact_cleanup"] = {"at": now_iso(), "complete": True}

843 event(doc, "finalized-artifacts-pruned", unit_id, {"job_count": len(attempt_job_ids)})

844

845

846def retained_blocked_abandonment_receipt(run_id: str, unit: dict, attempt: dict) -> dict:

847 recorded = attempt.get("terminal_receipt")

848 if (

849 unit.get("state") != "authored"

850 or attempt.get("process_state") != "done"

851 or attempt.get("terminal_validation_failure") is not None

852 or not isinstance(recorded, dict)

853 or recorded.get("terminal_status") != "blocked"

854 ):

855 raise Operational("REFUSED", "done output is not an exactly retained worker blocker")

856 observed_process = process_evidence(runner_job_dir(run_id, attempt["job_id"]))["process_state"]

857 if observed_process != "done":

858 raise Operational("BLOCKED", "retained worker-blocker job evidence changed")

859 observed_receipt = terminal_receipt(unit, attempt)

860 if observed_receipt != recorded:

861 raise Operational("BLOCKED", "retained worker-blocker receipt evidence changed")

862 return {

863 "kind": "retained-worker-blocker",

864 "value": attempt["job_id"],

865 "process_state": observed_process,

866 "terminal_status": recorded["terminal_status"],

867 "result_sha256": recorded["result_sha256"],

868 "raw_log_sha256": recorded["raw_log_sha256"],

869 }

870

871

872def owned_workspace_path(run_id: str, unit_id: str, recorded_workspace: str) -> str:

873 unit_root = os.path.join(run_dir(run_id), "units", unit_id)

874 expected_workspace = os.path.join(unit_root, "workspace")

875 if os.path.abspath(recorded_workspace) != expected_workspace:

876 raise TrustFailure("manifest workspace path does not match the controller-owned unit workspace")

877 validate_private_dir(unit_root)

878 return expected_workspace

879

880

881def remove_unregistered_owned_workspace(run_id: str, unit_id: str, recorded_workspace: str) -> None:

882 expected_workspace = owned_workspace_path(run_id, unit_id, recorded_workspace)

883 if not os.path.lexists(expected_workspace):

884 return

885 entry = os.lstat(expected_workspace)

886 uid_getter = getattr(os, "geteuid", None) or getattr(os, "getuid", None)

887 effective_uid = uid_getter() if uid_getter is not None else None

888 if not stat.S_ISDIR(entry.st_mode) or stat.S_ISLNK(entry.st_mode):

889 raise Operational("BLOCKED", "unregistered workspace path is not a real directory")

890 if effective_uid is not None and entry.st_uid != effective_uid:

891 raise Operational("BLOCKED", "unregistered workspace is not owned by the current user")

892 shutil.rmtree(expected_workspace)

893 if os.path.lexists(expected_workspace):

894 raise Operational("BLOCKED", "unregistered workspace remained after cleanup")

895

896

897def cmd_cleanup(args) -> tuple[str, dict]:

898 with locked_manifest(args.run_id) as doc:

899 validate_repo(doc)

900 unit = doc["units"].get(args.unit_id)

901 if not unit:

902 raise Operational("REFUSED", "unknown unit")

903 cleanup_recorded = isinstance(unit.get("cleanup"), dict)

904 if unit["state"] == "cleaned" or (unit["state"] == "native-completed" and cleanup_recorded):

905 if not unit.get("cleanup", {}).get("artifact_cleanup", {}).get("complete"):

906 pass

907 else:

908 return "CLEANED", {"unit_id": args.unit_id, "resumed": True}

909 if unit["state"] == "cleaned" or (unit["state"] == "native-completed" and cleanup_recorded):

910 remove_finalized_artifacts(args.run_id, args.unit_id)

911 return "CLEANED", {"unit_id": args.unit_id, "resumed": True}

912 with locked_manifest(args.run_id) as doc:

913 unit = doc["units"][args.unit_id]

914 attempt = find_attempt(unit)

915 if attempt.get("process_state") == "running":

916 raise Operational("REFUSED", "cannot cleanup a live worker")

917 commit = unit["transport"].get("commit")

918 abandonment_receipt = None

919 if args.abandon:

920 if commit:

921 if args.expect_transport != commit:

922 raise Operational("REFUSED", "abandon cleanup requires exact transport SHA")

923 abandonment_receipt = {"kind": "transport", "value": commit}

924 else:

925 terminal_failures = TERMINAL_PROCESS - {"done"}

926 validation_failure = attempt.get("process_state") == "done" and attempt.get("terminal_validation_failure")

927 if not attempt.get("job_id"):

928 raise Operational("REFUSED", "transport-free cleanup requires an authoritative failed or reaped job")

929 if args.expect_job != attempt["job_id"]:

930 raise Operational("REFUSED", "transport-free cleanup requires the exact terminal job id")

931 if validation_failure:

932 validate_terminal_validation_failure(args.run_id, unit, attempt)

933 abandonment_receipt = {"kind": "terminal-validation-failure", "value": attempt["job_id"], "process_state": "done"}

934 elif attempt.get("process_state") == "done":

935 abandonment_receipt = retained_blocked_abandonment_receipt(args.run_id, unit, attempt)

936 else:

937 if attempt.get("process_state") not in terminal_failures:

938 raise Operational("REFUSED", "transport-free cleanup requires an authoritative failed or reaped job")

939 observed = process_evidence(runner_job_dir(args.run_id, attempt["job_id"]))["process_state"]

940 if observed != attempt["process_state"] or observed not in terminal_failures:

941 raise Operational("BLOCKED", "terminal job evidence changed; refusing cleanup")

942 abandonment_receipt = {"kind": "terminal-job", "value": attempt["job_id"], "process_state": observed}

943 elif unit["state"] != "committed":

944 raise Operational("REFUSED", "uncommitted output is retained unless explicitly abandoned")

945 workspace = unit["workspace"]["path"]

946 ref = unit["transport"].get("ref")

947 repo = doc["repository"]["toplevel"]

948 common = doc["repository"]["common_dir"]

949 workspace = owned_workspace_path(args.run_id, args.unit_id, workspace)

950 with locked_manifest(args.run_id, write=True) as doc:

951 event(doc, "cleanup-intent", args.unit_id, {"workspace": workspace, "ref": ref, "abandonment_receipt": abandonment_receipt})

952 with admin_lock(common):

953 present = [r for r in worktree_rows(repo) if os.path.realpath(str(r.get("worktree", ""))) == os.path.realpath(workspace)]

954 if present:

955 git(repo, "worktree", "remove", "--force", workspace)

956 test_fault("cleanup-after-worktree-remove")

957 if any(os.path.realpath(str(r.get("worktree", ""))) == os.path.realpath(workspace) for r in worktree_rows(repo)):

958 raise Operational("BLOCKED", "worktree remained registered after cleanup")

959 remove_unregistered_owned_workspace(args.run_id, args.unit_id, workspace)

960 if ref and commit:

961 current = git_text(repo, "rev-parse", "-q", "--verify", ref, check=False)

962 if current and current != commit:

963 raise Operational("BLOCKED", "transport ref changed; refusing cleanup")

964 if current:

965 git(repo, "update-ref", "-d", ref, commit)

966 with locked_manifest(args.run_id, write=True) as doc:

967 unit = doc["units"][args.unit_id]

968 finalized_state = "native-completed" if unit["state"] == "native-completed" else "cleaned"

969 unit["cleanup"] = {

970 "at": now_iso(),

971 "workspace_removed": True,

972 "ref_removed": True,

973 "abandoned": bool(args.abandon),

974 "abandonment_receipt": abandonment_receipt,

975 "artifact_cleanup": {"at": None, "complete": False},

976 }

977 unit["state"] = finalized_state

978 event(doc, "unit-cleaned", args.unit_id)

979 test_fault("cleanup-before-artifact-prune")

980 remove_finalized_artifacts(args.run_id, args.unit_id)

981 return "CLEANED", {"unit_id": args.unit_id, "resumed": False}