Skill Tracer
Skills › Compound Engineering › ce-work › unit_workspace_jobs.py

unit_workspace_jobs.py

1161 lines · skills/ce-work/scripts/unit_workspace_jobs.py at 3449230

Open on GitHub
1"""Unit preparation, runner evidence, and complete-tree transport lifecycle."""
2
3from __future__ import annotations
4
5import base64
6import json
7import os
8import re
9import stat
11from unit_workspace_state import *
14def _valid_retry_commit_id(value: object) -> bool:
15 return isinstance(value, str) and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", value) is not None
18def _validate_retry_base(doc: dict, unit: dict, requested_base: str) -> None:
19 wave = unit.get("wave", {})
20 original_base = wave.get("base")
21 allowed_heads = wave.get("allowed_heads", [])
22 if not _valid_retry_commit_id(original_base):
23 raise TrustFailure("recorded retry base is malformed")
24 if not isinstance(allowed_heads, list) or any(not _valid_retry_commit_id(head) for head in allowed_heads):
25 raise TrustFailure("recorded retry HEAD allowances are malformed")
27 accepted_heads = {
28 commit
29 for candidate in doc.get("units", {}).values()
30 if (commit := unit_accepted_commit(candidate)) is not None
31 }
32 latest_allowed = allowed_heads[-1] if allowed_heads else original_base
33 if requested_base != original_base and requested_base not in accepted_heads:
34 raise Operational(
35 "BLOCKED",
36 "retry base is not a controller-accepted canonical head",
37 {"requested_base": requested_base, "latest_allowed_head": latest_allowed},
38 )
39 repo = doc["repository"]["toplevel"]
40 required = accepted_heads | {original_base, *allowed_heads}
41 missing = sorted(
42 commit for commit in required
43 if git_text(repo, "merge-base", commit, requested_base, check=False) != commit
44 )
45 if missing:
46 raise Operational(
47 "BLOCKED",
48 "retry base omits controller-accepted canonical history",
49 {
50 "requested_base": requested_base,
51 "latest_allowed_head": latest_allowed,
52 "missing_ancestry": missing,
53 },
54 )
57def _record_retry_base(doc: dict, unit: dict, requested_base: str) -> None:
58 wave = unit["wave"]
59 position = wave.get("position")
60 if not isinstance(position, int):
61 raise TrustFailure("recorded retry wave position is malformed")
62 targets = [unit]
63 if wave.get("id"):
64 for candidate in doc.get("units", {}).values():
65 candidate_wave = candidate.get("wave", {})
66 if candidate is unit or candidate_wave.get("id") != wave["id"]:
67 continue
68 candidate_position = candidate_wave.get("position")
69 if not isinstance(candidate_position, int):
70 raise TrustFailure("recorded wave position is malformed")
71 if candidate_position > position:
72 targets.append(candidate)
73 for candidate in targets:
74 candidate_wave = candidate.get("wave", {})
75 if candidate_wave.get("base") != wave.get("base"):
76 raise Operational("BLOCKED", "wave members do not share one recorded base")
77 allowed_heads = candidate_wave.setdefault("allowed_heads", [])
78 if not isinstance(allowed_heads, list) or any(not _valid_retry_commit_id(head) for head in allowed_heads):
79 raise TrustFailure("recorded retry HEAD allowances are malformed")
80 if requested_base not in allowed_heads:
81 allowed_heads.append(requested_base)
84def cmd_prepare(args) -> tuple[str, dict]:
85 uid = safe_id(args.unit_id, "unit id")
86 attempt_id = safe_id(args.attempt_id, "attempt id")
87 packet_bytes = read_external_packet(args.packet)
88 packet_digest = digest_bytes(packet_bytes)
89 with locked_manifest(args.run_id) as doc:
90 info = validate_repo(doc)
91 repo = info["toplevel"]
92 base = git_text(repo, "rev-parse", f"{args.base}^{{commit}}")
93 if info["head"] != base:
94 raise Operational("BLOCKED", "canonical HEAD does not equal requested unit base")
95 if status_paths(repo):
96 raise Operational("BLOCKED", "canonical checkout is dirty; external workspace unavailable")
97 existing = doc["units"].get(uid)
98 unit_root = os.path.join(run_dir(args.run_id), "units", uid)
99 workspace = os.path.join(unit_root, "workspace")
100 packet_path = os.path.join(unit_root, "packet.md")
101 authorization_path = os.path.join(unit_root, "authorization.json")
102 authorization = attempt_authorization(doc, args.activity_posture, uid, attempt_id, packet_digest)
103 authorization_bytes = (json.dumps(authorization, sort_keys=True, separators=(",", ":")) + "\n").encode()
104 authorization_digest = digest_bytes(authorization_bytes)
105 contract_wave_base = existing.get("wave", {}).get("base") if existing else base
106 expected_contract = {
107 "dependencies": list(args.dependency),
108 "wave": {"id": args.wave_id, "base": contract_wave_base, "position": args.wave_position},
109 "packet_digest": packet_digest,
110 "attempt_id": attempt_id,
111 "authorization": authorization,
112 "authorization_path": authorization_path,
113 "authorization_digest": authorization_digest,
114 }
115 retrying = False
116 if existing:
117 matching_attempts = [attempt for attempt in existing.get("attempts", []) if attempt.get("attempt_id") == attempt_id]
118 if not matching_attempts:
119 cleanup = existing.get("cleanup")
120 if (
121 existing.get("state") != "cleaned"
122 or not isinstance(cleanup, dict)
123 or cleanup.get("abandoned") is not True
124 or cleanup.get("artifact_cleanup", {}).get("complete") is not True
125 ):
126 raise Operational("REFUSED", "a fresh attempt requires an exactly abandoned and fully cleaned prior attempt")
127 if doc.get("integration_lock"):
128 raise Operational("REFUSED", "release the prior integration lock before preparing a retry")
129 if existing.get("dependencies") != list(args.dependency):
130 raise Operational("BLOCKED", "retry dependencies differ from the recorded unit")
131 prior_wave = existing.get("wave", {})
132 if {
133 "id": prior_wave.get("id"),
134 "position": prior_wave.get("position"),
135 } != {"id": args.wave_id, "position": args.wave_position}:
136 raise Operational("BLOCKED", "retry wave identity/position differs from the recorded unit")
137 _validate_retry_base(doc, existing, base)
138 retrying = True
139 else:
140 attempt = find_attempt(existing, attempt_id)
141 if existing and not retrying and (
142 existing.get("workspace", {}).get("path") != workspace
143 or existing.get("workspace", {}).get("base") != base
144 ):
145 raise Operational("BLOCKED", "duplicate unit id has a different workspace contract")
146 if existing and not retrying:
147 if existing.get("state") == "cleaned" or existing.get("cleanup"):
148 raise Operational(
149 "REFUSED",
150 "cleaned unit cannot reuse a recorded attempt id; supply a fresh --attempt-id after exact abandonment cleanup and lock release",
151 )
152 observed_contract = {
153 "dependencies": existing.get("dependencies"),
154 "wave": {key: existing.get("wave", {}).get(key) for key in ("id", "base", "position")},
155 "packet_digest": existing.get("packet_digest"),
156 "attempt_id": attempt.get("attempt_id"),
157 "authorization": attempt.get("authorization"),
158 "authorization_path": attempt.get("authorization_path"),
159 "authorization_digest": attempt.get("authorization_digest"),
160 }
161 if observed_contract != expected_contract or existing.get("packet", {}).get("path") != packet_path:
162 raise Operational("BLOCKED", "resumed prepare contract differs from the recorded unit")
163 if read_private(packet_path, MAX_PACKET_BYTES) != packet_bytes:
164 raise Operational("BLOCKED", "controller-owned unit packet no longer matches supplied bytes")
165 if read_private(authorization_path, MAX_JSON_BYTES) != authorization_bytes:
166 raise Operational("BLOCKED", "controller-owned authorization no longer matches the recorded attempt")
167 result_fd, _ = open_recorded_result_dir(existing)
168 os.close(result_fd)
169 if existing and not retrying and existing["workspace"].get("registered"):
170 if existing.get("state") == "queued":
171 validate_pristine_unit_base(doc, existing)
172 else:
173 validate_workspace(doc, existing)
174 return "PREPARED", {
175 "unit_id": uid, "attempt_id": attempt_id,
176 "workspace": workspace, "result_dir": os.path.join(unit_root, "result"),
177 "packet_path": packet_path, "packet_digest": packet_digest,
178 "authorization_path": authorization_path, "authorization_digest": authorization_digest,
179 "adapter": attempt["adapter"],
180 "base": base, "resumed": True,
181 }
182 ensure_private_dir(unit_root)
183 result_dir = os.path.join(unit_root, "result")
184 ensure_private_dir(result_dir)
185 result_dir_identity = private_result_dir_identity(result_dir)
186 if os.path.lexists(packet_path):
187 if read_private(packet_path, MAX_PACKET_BYTES) != packet_bytes:
188 raise Operational("BLOCKED", "controller-owned packet path contains different bytes")
189 else:
190 create_private(packet_path, packet_bytes)
191 if os.path.lexists(authorization_path):
192 if read_private(authorization_path, MAX_JSON_BYTES) != authorization_bytes:
193 raise Operational("BLOCKED", "controller-owned authorization path contains different bytes")
194 else:
195 create_private(authorization_path, authorization_bytes)
196 attempt_record = {
197 "attempt_id": attempt_id,
198 "job_id": None,
199 "dispatch_authorization_receipt": None,
200 "process_state": "never-started",
201 "activity": {"posture": args.activity_posture, "latest_at": None},
202 "fallback": {"eligible": False, "reason": None, "claimed": None},
203 "authorization": authorization,
204 "authorization_path": authorization_path,
205 "authorization_digest": authorization_digest,
206 "authorization_retained": True,
207 "adapter": os.path.realpath(os.path.join(os.path.dirname(__file__), "cross-model-work.sh")),
208 "terminal_receipt": None,
209 }
210 if not existing:
211 unit = {
212 "unit_id": uid,
213 "state": "queued",
214 "dependencies": list(args.dependency),
215 "wave": {"id": args.wave_id, "base": base, "position": args.wave_position, "allowed_heads": [base]},
216 "packet_digest": packet_digest,
217 "packet": {"path": packet_path, "digest": packet_digest, "bytes": len(packet_bytes), "retained": True},
218 "workspace": {"path": workspace, "base": base, "registered": False},
219 "result_dir_identity": result_dir_identity,
220 "attempts": [attempt_record],
221 "transport": {"base": None, "tree": None, "commit": None, "ref": None, "digest": None, "changed_paths": []},
222 "integration": {"intent_revision": None, "pre_fold": None, "expected_apply": None, "applied": None, "verification": None, "canonical_commit": None, "restore": None},
223 "cleanup": None,
224 "recovery_path": unit_root,
225 }
226 with locked_manifest(args.run_id, write=True) as doc:
227 if uid in doc["units"]:
228 raise Operational("BLOCKED", "unit was concurrently claimed")
229 doc["units"][uid] = unit
230 event(doc, "worktree-add-intent", uid, {"path": workspace, "base": base})
231 elif retrying:
232 with locked_manifest(args.run_id, write=True) as doc:
233 unit = doc["units"].get(uid)
234 cleanup = unit.get("cleanup") if unit else None
235 if (
236 not unit
237 or unit.get("state") != "cleaned"
238 or not isinstance(cleanup, dict)
239 or cleanup.get("abandoned") is not True
240 or cleanup.get("artifact_cleanup", {}).get("complete") is not True
241 or doc.get("integration_lock")
242 ):
243 raise Operational("BLOCKED", "unit retry eligibility changed while it was being prepared")
244 if any(attempt.get("attempt_id") == attempt_id for attempt in unit.get("attempts", [])):
245 raise Operational("BLOCKED", "retry attempt id was concurrently claimed")
246 info = validate_repo(doc)
247 if info["head"] != base:
248 raise Operational("BLOCKED", "canonical HEAD changed while retry was being prepared")
249 if unit.get("dependencies") != list(args.dependency):
250 raise Operational("BLOCKED", "retry dependencies differ from the recorded unit")
251 prior_wave = unit.get("wave", {})
252 if {
253 "id": prior_wave.get("id"),
254 "position": prior_wave.get("position"),
255 } != {"id": args.wave_id, "position": args.wave_position}:
256 raise Operational("BLOCKED", "retry wave identity/position differs from the recorded unit")
257 _validate_retry_base(doc, unit, base)
258 previous = find_attempt(unit)
259 previous["cleanup_receipt"] = dict(cleanup)
260 restore = unit.get("integration", {}).get("restore")
261 if restore is not None:
262 previous["restore_receipt"] = json.loads(json.dumps(restore))
263 unit["state"] = "queued"
264 unit["packet_digest"] = packet_digest
265 unit["packet"] = {"path": packet_path, "digest": packet_digest, "bytes": len(packet_bytes), "retained": True}
266 unit["workspace"] = {"path": workspace, "base": base, "registered": False}
267 unit["result_dir_identity"] = result_dir_identity
268 _record_retry_base(doc, unit, base)
269 unit["attempts"].append(attempt_record)
270 unit["transport"] = {"base": None, "tree": None, "commit": None, "ref": None, "digest": None, "changed_paths": []}
271 unit["integration"] = {"intent_revision": None, "pre_fold": None, "expected_apply": None, "applied": None, "verification": None, "canonical_commit": None, "restore": None}
272 unit["cleanup"] = None
273 unit["recovery_path"] = unit_root
274 event(doc, "unit-retry-prepared", uid, {"attempt_id": attempt_id, "base": base})
275 event(doc, "worktree-add-intent", uid, {"path": workspace, "base": base})
276 with locked_manifest(args.run_id) as doc:
277 common = doc["repository"]["common_dir"]
278 repo = doc["repository"]["toplevel"]
279 with admin_lock(common):
280 if not os.path.exists(workspace):
281 git(repo, "worktree", "add", "--detach", workspace, base)
282 test_fault("after-worktree-add")
283 with locked_manifest(args.run_id) as doc:
284 unit = doc["units"][uid]
285 validate_pristine_unit_base(doc, unit)
286 with locked_manifest(args.run_id, write=True) as doc:
287 unit = doc["units"][uid]
288 unit["workspace"]["registered"] = True
289 event(doc, "worktree-prepared", uid, {"path": workspace, "base": base})
290 return "PREPARED", {
291 "unit_id": uid, "attempt_id": attempt_id,
292 "workspace": workspace, "result_dir": os.path.join(unit_root, "result"),
293 "packet_path": packet_path, "packet_digest": packet_digest,
294 "authorization_path": authorization_path, "authorization_digest": authorization_digest,
295 "adapter": attempt_record["adapter"],
296 "base": base, "resumed": False,
297 }
300def runner_job_dir(run_id: str, job_id: str) -> str:
301 return os.path.join(run_dir(run_id), "jobs", safe_id(job_id, "job id"))
304def process_evidence(job_dir: str) -> dict:
305 validate_private_dir(job_dir)
306 status_path = os.path.join(job_dir, "status")
307 if os.path.lexists(status_path):
308 word = read_private(status_path, 256).decode("ascii", "strict").strip()
309 if word not in TERMINAL_PROCESS:
310 raise TrustFailure("runner terminal state is invalid")
311 elif os.path.lexists(os.path.join(job_dir, "pid")):
312 read_private_json(os.path.join(job_dir, "pid"))
313 word = "running"
314 else:
315 word = "never-started"
316 failure_reason = None
317 reason_path = os.path.join(job_dir, "reason")
318 if word in TERMINAL_PROCESS and os.path.lexists(reason_path):
319 failure_reason = read_private(reason_path, 4096).decode("utf-8", "strict").strip() or None
320 activity = {"latest_at": None, "log_bytes": 0}
321 log = os.path.join(job_dir, "out.log")
322 if os.path.lexists(log):
323 st = stat_private_file(log)
324 activity = {"latest_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(st.st_mtime)), "log_bytes": st.st_size}
325 return {"process_state": word, "failure_reason": failure_reason, "activity": activity}
328HOST_RECEIPT_FIELDS = (
329 "requested_route", "actual_route", "target", "harness", "intermediaries",
330 "model_requested", "model_actual", "model_receipt_status", "effort_requested", "activity_posture",
331 "restriction_posture", "failure_reason", "raw_log", "packet_digest",
333MAX_RESULT_BYTES = 5 * 1024 * 1024
334MAX_REPORTED_CHANGED_FILES = 1000
337def _validate_private_dir_fd(fd: int, path: str) -> os.stat_result:
338 st = os.fstat(fd)
339 effective_uid = os.geteuid() if hasattr(os, "geteuid") else None
340 if not stat.S_ISDIR(st.st_mode):
341 raise TrustFailure(f"not a real directory: {path}")
342 if effective_uid is not None and st.st_uid != effective_uid:
343 raise TrustFailure(f"directory is not owned by current user: {path}")
344 mode = stat.S_IMODE(st.st_mode)
345 if mode != 0o700:
346 raise TrustFailure(f"directory mode is {mode:04o}, expected 0700: {path}")
347 return st
350def private_result_dir_identity(path: str) -> dict:
351 try:
352 fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | O_NOFOLLOW)
353 except OSError as exc:
354 raise TrustFailure(f"cannot safely open result directory {path}: {exc}") from exc
355 try:
356 st = _validate_private_dir_fd(fd, path)
357 return {"dev": st.st_dev, "ino": st.st_ino}
358 finally:
359 os.close(fd)
362def open_recorded_result_dir(unit: dict) -> tuple[int, str]:
363 result_dir = os.path.join(os.path.dirname(unit["workspace"]["path"]), "result")
364 identity = unit.get("result_dir_identity")
365 if (
366 not isinstance(identity, dict)
367 or set(identity) != {"dev", "ino"}
368 or any(not isinstance(identity.get(key), int) or isinstance(identity.get(key), bool) for key in ("dev", "ino"))
369 ):
370 raise TrustFailure("unit has no valid controller-recorded result directory identity")
371 try:
372 fd = os.open(result_dir, os.O_RDONLY | os.O_DIRECTORY | O_NOFOLLOW)
373 except OSError as exc:
374 raise TrustFailure(f"cannot safely open result directory {result_dir}: {exc}") from exc
375 try:
376 st = _validate_private_dir_fd(fd, result_dir)
377 if (st.st_dev, st.st_ino) != (identity["dev"], identity["ino"]):
378 raise TrustFailure("controller result directory identity changed")
379 return fd, result_dir
380 except Exception:
381 os.close(fd)
382 raise
385def read_private_at(dir_fd: int, name: str, cap: int, display_path: str) -> bytes:
386 if os.path.basename(name) != name or name in {"", ".", ".."}:
387 raise TrustFailure(f"unsafe state file name: {name!r}")
388 try:
389 fd = os.open(name, os.O_RDONLY | O_NOFOLLOW, dir_fd=dir_fd)
390 except OSError as exc:
391 raise TrustFailure(f"cannot safely open state file {display_path}: {exc}") from exc
392 try:
393 st = os.fstat(fd)
394 effective_uid = os.geteuid() if hasattr(os, "geteuid") else None
395 if not stat.S_ISREG(st.st_mode):
396 raise TrustFailure(f"state is not a regular file: {display_path}")
397 if effective_uid is not None and st.st_uid != effective_uid:
398 raise TrustFailure(f"state is not owned by current user: {display_path}")
399 mode = stat.S_IMODE(st.st_mode)
400 if mode != 0o600:
401 raise TrustFailure(f"state mode is {mode:04o}, expected 0600: {display_path}")
402 if st.st_size > cap:
403 raise TrustFailure(f"state exceeds {cap}-byte limit: {display_path}")
404 out = bytearray()
405 while len(out) <= cap:
406 part = os.read(fd, min(65536, cap + 1 - len(out)))
407 if not part:
408 break
409 out.extend(part)
410 if len(out) > cap:
411 raise TrustFailure(f"state grew beyond {cap}-byte limit: {display_path}")
412 return bytes(out)
413 finally:
414 os.close(fd)
417def stat_private_at(
418 dir_fd: int,
419 name: str,
420 display_path: str,
421 *,
422 missing_ok: bool = False,
423) -> os.stat_result | None:
424 if os.path.basename(name) != name or name in {"", ".", ".."}:
425 raise TrustFailure(f"unsafe state file name: {name!r}")
426 try:
427 fd = os.open(name, os.O_RDONLY | O_NOFOLLOW, dir_fd=dir_fd)
428 except FileNotFoundError:
429 if missing_ok:
430 return None
431 raise TrustFailure(f"cannot safely open state file {display_path}: file is missing")
432 except OSError as exc:
433 raise TrustFailure(f"cannot safely open state file {display_path}: {exc}") from exc
434 try:
435 st = os.fstat(fd)
436 effective_uid = os.geteuid() if hasattr(os, "geteuid") else None
437 if not stat.S_ISREG(st.st_mode):
438 raise TrustFailure(f"state is not a regular file: {display_path}")
439 if effective_uid is not None and st.st_uid != effective_uid:
440 raise TrustFailure(f"state is not owned by current user: {display_path}")
441 mode = stat.S_IMODE(st.st_mode)
442 if mode != 0o600:
443 raise TrustFailure(f"state mode is {mode:04o}, expected 0600: {display_path}")
444 return st
445 finally:
446 os.close(fd)
449def read_recorded_result_file(unit: dict, name: str, cap: int) -> bytes:
450 result_fd, result_dir = open_recorded_result_dir(unit)
451 try:
452 return read_private_at(
453 result_fd,
454 name,
455 cap,
456 os.path.join(result_dir, name),
457 )
458 finally:
459 os.close(result_fd)
462def read_recorded_result_json(unit: dict) -> tuple[dict, bytes]:
463 result_path = os.path.join(os.path.dirname(unit["workspace"]["path"]), "result", "implementation-result.json")
464 raw = read_recorded_result_file(unit, "implementation-result.json", MAX_RESULT_BYTES)
465 try:
466 value = json.loads(raw)
467 except (ValueError, UnicodeDecodeError) as exc:
468 raise TrustFailure(f"malformed JSON state: {result_path}") from exc
469 if not isinstance(value, dict):
470 raise TrustFailure(f"JSON state is not an object: {result_path}")
471 return value, raw
474def terminal_receipt(
475 unit: dict,
476 attempt: dict,
477 *,
478 unavailable: bool = False,
479 launched_failure: bool = False,
480) -> dict:
481 result_dir = os.path.join(os.path.dirname(unit["workspace"]["path"]), "result")
482 receipt, result_bytes = read_recorded_result_json(unit)
483 authorization = attempt.get("authorization")
484 if not isinstance(authorization, dict):
485 raise Operational("BLOCKED", "attempt has no controller-issued route authorization")
486 expected = {
487 "requested_route": authorization["route"],
488 "actual_route": None if unavailable else authorization["route"],
489 "target": authorization["target"],
490 "harness": authorization["harness"],
491 "intermediaries": authorization["intermediaries"],
492 "model_requested": authorization["model_requested"],
493 "effort_requested": authorization.get("effort_requested"),
494 "restriction_posture": authorization["restriction_posture"],
495 "packet_digest": unit["packet_digest"],
496 }
497 if unavailable or launched_failure:
498 expected["activity_posture"] = authorization["activity_posture"]
499 mismatches = {key: {"expected": value, "actual": receipt.get(key)} for key, value in expected.items() if receipt.get(key) != value}
500 if mismatches:
501 raise Operational("BLOCKED", "adapter terminal receipt does not match controller authorization", {"mismatches": mismatches})
502 terminal_status = receipt.get("terminal_status")
503 if unavailable:
504 neutral = {
505 "schema_version": 1,
506 "terminal_status": "unavailable",
507 "summary": "External route unavailable",
508 "changed_files": [],
509 "evidence": [],
510 "scope_expansion": None,
511 "model_actual": "unverified",
512 "model_receipt_status": "unverified",
513 }
514 invalid = {key: {"expected": value, "actual": receipt.get(key)} for key, value in neutral.items() if receipt.get(key) != value}
515 failure_reason = receipt.get("failure_reason")
516 if invalid or not isinstance(failure_reason, str) or not failure_reason or len(failure_reason.encode()) > 4096:
517 raise Operational(
518 "BLOCKED",
519 "failed runner did not publish a bounded neutral unavailable receipt",
520 {"mismatches": invalid},
521 )
522 elif launched_failure:
523 neutral = {
524 "schema_version": 1,
525 "terminal_status": "failed",
526 "changed_files": [],
527 "evidence": [],
528 "scope_expansion": None,
529 }
530 invalid = {key: {"expected": value, "actual": receipt.get(key)} for key, value in neutral.items() if receipt.get(key) != value}
531 failure_reason = receipt.get("failure_reason")
532 summary = receipt.get("summary")
533 if (
534 invalid
535 or not isinstance(failure_reason, str)
536 or not failure_reason
537 or len(failure_reason.encode()) > 4096
538 or not isinstance(summary, str)
539 or not summary
540 or len(summary.encode()) > 4096
541 ):
542 raise Operational(
543 "BLOCKED",
544 "failed runner did not publish a bounded neutral launched-route receipt",
545 {"mismatches": invalid},
546 )
547 else:
548 if terminal_status not in {"completed", "blocked", "scope_expansion"}:
549 raise Operational("BLOCKED", "successful runner did not publish a host-resolvable adapter result")
550 if terminal_status == "scope_expansion" and not isinstance(receipt.get("scope_expansion"), dict):
551 raise Operational("BLOCKED", "scope-expansion adapter result has no expansion receipt")
552 changed_files = receipt.get("changed_files")
553 if (
554 not isinstance(changed_files, list)
555 or len(changed_files) > MAX_REPORTED_CHANGED_FILES
556 or any(not isinstance(path, str) or not path for path in changed_files)
557 ):
558 raise Operational("BLOCKED", "adapter terminal receipt has invalid changed-files evidence")
559 raw_log = receipt.get("raw_log")
560 expected_log = os.path.join(result_dir, "adapter.log")
561 if not isinstance(raw_log, str) or os.path.abspath(raw_log) != expected_log:
562 raise Operational("BLOCKED", "adapter raw-log receipt escaped the controller result directory")
563 log_bytes = read_recorded_result_file(unit, "adapter.log", 10 * 1024 * 1024)
564 return {key: receipt.get(key) for key in HOST_RECEIPT_FIELDS} | {
565 "terminal_status": receipt["terminal_status"],
566 "summary": str(receipt.get("summary", ""))[:4096],
567 "changed_files": changed_files,
568 "changed_file_count": len(changed_files),
569 "evidence_count": len(receipt.get("evidence", [])),
570 "scope_expansion_requested": receipt.get("scope_expansion") is not None,
571 "result_sha256": digest_bytes(result_bytes),
572 "raw_log_sha256": digest_bytes(log_bytes),
573 "raw_log_bytes": len(log_bytes),
574 }
577def _validate_authorized_failed_job(
578 run_id: str,
579 unit: dict,
580 attempt: dict,
581) -> None:
582 job_id = attempt.get("job_id")
583 if not isinstance(job_id, str):
584 raise Operational("BLOCKED", "failed receipt has no bound runner job")
585 job_dir = runner_job_dir(run_id, job_id)
586 if process_evidence(job_dir)["process_state"] != "failed":
587 raise Operational("BLOCKED", "failed receipt requires authoritative failed runner evidence")
588 meta = read_private_json(os.path.join(job_dir, "meta.json"))
589 if meta.get("job_id") != job_id:
590 raise Operational("BLOCKED", "runner job metadata identity mismatch")
591 validate_runner_contract(run_id, unit, meta)
592 expected_result_dir = os.path.join(os.path.dirname(unit["workspace"]["path"]), "result")
593 expected_dispatch = {
594 "attempt_id": attempt.get("attempt_id"),
595 "job_id": job_id,
596 "authorization_path": attempt.get("authorization_path"),
597 "authorization_digest": attempt.get("authorization_digest"),
598 "workspace": unit["workspace"]["path"],
599 "packet_path": unit["packet"]["path"],
600 "packet_digest": unit["packet_digest"],
601 "result_dir": expected_result_dir,
602 "result_dir_identity": unit.get("result_dir_identity"),
603 }
604 if attempt.get("dispatch_authorization_receipt") != expected_dispatch:
605 raise Operational("BLOCKED", "failed receipt is not bound to the exact authorized dispatch")
608def _authorized_failed_terminal_receipt(
609 run_id: str,
610 unit: dict,
611 attempt: dict,
612 *,
613 unavailable: bool,
614) -> dict:
615 _validate_authorized_failed_job(run_id, unit, attempt)
616 return terminal_receipt(
617 unit,
618 attempt,
619 unavailable=unavailable,
620 launched_failure=not unavailable,
621 )
624def unavailable_terminal_receipt(run_id: str, unit: dict, attempt: dict) -> dict:
625 return _authorized_failed_terminal_receipt(run_id, unit, attempt, unavailable=True)
628def launched_failure_terminal_receipt(run_id: str, unit: dict, attempt: dict) -> dict:
629 return _authorized_failed_terminal_receipt(run_id, unit, attempt, unavailable=False)
632def record_terminal_validation_failure(run_id: str, unit_id: str, error: Operational) -> None:
633 if isinstance(error, TrustFailure):
634 raise error
635 with locked_manifest(run_id) as doc:
636 unit = doc["units"][unit_id]
637 result_digest = digest_bytes(read_recorded_result_file(unit, "implementation-result.json", MAX_RESULT_BYTES))
638 with locked_manifest(run_id, write=True) as doc:
639 attempt = find_attempt(doc["units"][unit_id])
640 failure = {
641 "at": now_iso(),
642 "word": error.word,
643 "reason": str(error),
644 "detail": error.detail,
645 "job_id": attempt.get("job_id"),
646 "result_sha256": result_digest,
647 }
648 attempt["terminal_validation_failure"] = failure
649 fallback = attempt.setdefault("fallback", {})
650 fallback.setdefault("claimed", None)
651 fallback["eligible"] = fallback.get("claimed") is None
652 fallback["reason"] = "terminal-validation-failure"
653 event(doc, "terminal-validation-failed", unit_id, failure)
656def validate_terminal_validation_failure(run_id: str, unit: dict, attempt: dict) -> dict:
657 failure = attempt.get("terminal_validation_failure")
658 if not isinstance(failure, dict) or failure.get("job_id") != attempt.get("job_id"):
659 raise Operational("REFUSED", "attempt has no exact terminal-validation failure")
660 observed = process_evidence(runner_job_dir(run_id, attempt["job_id"]))["process_state"]
661 if observed != "done":
662 raise Operational("BLOCKED", "terminal-validation job evidence changed")
663 if digest_bytes(read_recorded_result_file(unit, "implementation-result.json", MAX_RESULT_BYTES)) != failure.get("result_sha256"):
664 raise Operational("BLOCKED", "terminal-validation result evidence changed")
665 return failure
668def retire_terminal_validation_failure(unit: dict) -> None:
669 attempt = find_attempt(unit)
670 failure = attempt.get("terminal_validation_failure")
671 claimed = attempt.get("fallback", {}).get("claimed")
672 if failure is not None and not claimed:
673 attempt.pop("terminal_validation_failure")
674 attempt["fallback"] = {"eligible": False, "reason": None, "claimed": None}
677def validate_runner_contract(run_id: str, unit: dict, meta: dict) -> None:
678 unit_id = unit["unit_id"]
679 expected_result_dir = os.path.join(run_dir(run_id), "units", unit_id, "result")
680 expected_result_file = os.path.join(expected_result_dir, "implementation-result.json")
681 if meta.get("skill") != "ce-work":
682 raise Operational("BLOCKED", "runner skill must be 'ce-work'")
683 if meta.get("run_id") != run_id:
684 raise Operational("BLOCKED", f"runner run id must equal the controller run id exactly: expected {run_id!r}")
685 if meta.get("label") != unit_id:
686 raise Operational(
687 "BLOCKED",
688 f"runner label must equal unit id exactly: expected {unit_id!r}, got {meta.get('label')!r}",
689 )
690 if meta.get("input_digest") != unit["packet_digest"]:
691 raise Operational("BLOCKED", "runner input digest must equal the controller packet digest")
692 if not isinstance(meta.get("result_path"), str) or os.path.abspath(meta["result_path"]) != expected_result_file:
693 raise Operational(
694 "BLOCKED",
695 f"runner result path must be the controller result file: {expected_result_file}",
696 )
697 attempt = find_attempt(unit)
698 authorization = attempt.get("authorization")
699 authorization_path = attempt.get("authorization_path")
700 authorization_digest = attempt.get("authorization_digest")
701 if not isinstance(authorization, dict) or not isinstance(authorization_path, str) or not isinstance(authorization_digest, str):
702 raise Operational("BLOCKED", "attempt has no controller-issued authorization artifact")
703 authorization_bytes = read_private(authorization_path, MAX_JSON_BYTES)
704 try:
705 observed_authorization = json.loads(authorization_bytes)
706 except (ValueError, UnicodeDecodeError) as exc:
707 raise TrustFailure("controller authorization artifact is malformed") from exc
708 if observed_authorization != authorization or digest_bytes(authorization_bytes) != authorization_digest:
709 raise Operational("BLOCKED", "controller authorization artifact no longer matches the recorded attempt")
710 expected_argv = [
711 attempt.get("adapter"), authorization_path, unit["workspace"]["path"],
712 unit["packet"]["path"], unit["packet_digest"], expected_result_dir,
713 ]
714 if meta.get("worker_argv") != expected_argv:
715 raise Operational(
716 "BLOCKED", "runner worker argv does not match the controller-issued fixed-route contract",
717 {"expected_argv": expected_argv, "actual_argv": meta.get("worker_argv")},
718 )
721def cmd_authorize_dispatch(args) -> tuple[str, dict]:
722 run_id = safe_id(args.run_id, "run id")
723 unit_id = safe_id(args.unit_id, "unit id")
724 attempt_id = safe_id(args.attempt_id, "attempt id")
725 job_id = safe_id(args.job_id, "job id")
726 if not re.fullmatch(r"[0-9a-f]{64}", args.authorization_digest):
727 raise Operational("REFUSED", "observed authorization digest must be lowercase SHA-256")
728 if not re.fullmatch(r"[0-9a-f]{64}", args.packet_digest):
729 raise Operational("REFUSED", "observed packet digest must be lowercase SHA-256")
730 with locked_manifest(run_id, write=True) as doc:
731 validate_repo(doc)
732 unit = doc["units"].get(unit_id)
733 if not unit:
734 raise Operational("REFUSED", "unknown unit")
735 attempt = find_attempt(unit, attempt_id)
736 if unit.get("state") not in {"queued", "authoring"}:
737 raise Operational("REFUSED", "dispatch authorization is available only before worker completion")
738 bound_job = attempt.get("job_id")
739 if bound_job not in (None, job_id):
740 raise Operational("AMBIGUOUS", "attempt is already bound to another job")
741 job_dir = os.path.join(run_dir(run_id), "jobs", job_id)
742 validate_private_dir(job_dir)
743 meta = read_private_json(os.path.join(job_dir, "meta.json"))
744 if meta.get("job_id") != job_id:
745 raise Operational("BLOCKED", "runner job metadata identity mismatch")
746 validate_runner_contract(run_id, unit, meta)
748 expected_authorization_path = attempt.get("authorization_path")
749 expected_authorization_digest = attempt.get("authorization_digest")
750 if os.path.abspath(args.authorization) != expected_authorization_path:
751 raise Operational("BLOCKED", "authorization path does not match the recorded attempt")
752 if args.authorization_digest != expected_authorization_digest:
753 raise Operational("BLOCKED", "observed authorization digest does not match the recorded attempt")
754 authorization_bytes = read_private(expected_authorization_path, MAX_JSON_BYTES)
755 if digest_bytes(authorization_bytes) != expected_authorization_digest:
756 raise Operational("BLOCKED", "controller authorization bytes no longer match the recorded digest")
757 try:
758 authorization = json.loads(authorization_bytes)
759 except (ValueError, UnicodeDecodeError) as exc:
760 raise TrustFailure("controller authorization artifact is malformed") from exc
761 if authorization != attempt.get("authorization"):
762 raise Operational("BLOCKED", "controller authorization object no longer matches the recorded attempt")
763 if (
764 authorization.get("run_id") != run_id
765 or authorization.get("unit_id") != unit_id
766 or authorization.get("attempt_id") != attempt_id
767 ):
768 raise Operational("BLOCKED", "authorization run/unit/attempt identity mismatch")
770 expected_workspace = unit["workspace"]["path"]
771 if os.path.abspath(args.workspace) != expected_workspace:
772 raise Operational("BLOCKED", "workspace path does not match the recorded unit")
773 expected_dispatch_authorization_receipt = {
774 "attempt_id": attempt_id,
775 "job_id": job_id,
776 "authorization_path": expected_authorization_path,
777 "authorization_digest": expected_authorization_digest,
778 "workspace": expected_workspace,
779 "packet_path": unit["packet"]["path"],
780 "packet_digest": unit["packet_digest"],
781 "result_dir": os.path.join(os.path.dirname(expected_workspace), "result"),
782 "result_dir_identity": unit.get("result_dir_identity"),
783 }
784 recorded_dispatch_authorization_receipt = attempt.get("dispatch_authorization_receipt")
785 if recorded_dispatch_authorization_receipt is not None and (
786 bound_job != job_id
787 or recorded_dispatch_authorization_receipt != expected_dispatch_authorization_receipt
788 ):
789 raise Operational("BLOCKED", "recorded dispatch authorization does not match the exact request")
790 resumed = recorded_dispatch_authorization_receipt == expected_dispatch_authorization_receipt
791 if resumed:
792 validate_workspace(doc, unit)
793 else:
794 validate_pristine_unit_base(doc, unit)
796 expected_packet = unit["packet"]["path"]
797 if os.path.abspath(args.packet) != expected_packet:
798 raise Operational("BLOCKED", "packet path does not match the controller-owned unit packet")
799 if args.packet_digest != unit["packet_digest"] or authorization.get("packet_digest") != unit["packet_digest"]:
800 raise Operational("BLOCKED", "packet digest does not match the recorded authorization")
801 packet_bytes = read_private(expected_packet, MAX_PACKET_BYTES)
802 if digest_bytes(packet_bytes) != unit["packet_digest"]:
803 raise Operational("BLOCKED", "controller-owned packet bytes no longer match the recorded digest")
805 expected_result_dir = os.path.join(os.path.dirname(expected_workspace), "result")
806 if os.path.abspath(args.result_dir) != expected_result_dir:
807 raise Operational("BLOCKED", "result directory does not match the recorded unit")
808 result_fd, _ = open_recorded_result_dir(unit)
809 os.close(result_fd)
810 if not resumed:
811 attempt["job_id"] = job_id
812 attempt["dispatch_authorization_receipt"] = expected_dispatch_authorization_receipt
813 unit["state"] = "authoring"
814 event(doc, "job-bound", unit_id, {
815 "attempt_id": attempt_id,
816 "job_id": job_id,
817 "source": "authorize-dispatch",
818 })
819 return "AUTHORIZED", {
820 "run_id": run_id,
821 "unit_id": unit_id,
822 "attempt_id": attempt_id,
823 "job_id": job_id,
824 "resumed": resumed,
825 "authorization_digest": expected_authorization_digest,
826 "packet_digest": unit["packet_digest"],
827 }
830def matching_runner_jobs(run_id: str, unit: dict) -> list[str]:
831 jobs = os.path.join(run_dir(run_id), "jobs")
832 validate_private_dir(jobs)
833 matches: list[str] = []
834 for entry in os.scandir(jobs):
835 if not entry.is_dir(follow_symlinks=False):
836 continue
837 safe_id(entry.name, "job id")
838 validate_private_dir(entry.path)
839 meta = read_private_json(os.path.join(entry.path, "meta.json"))
840 if (
841 meta.get("skill") == "ce-work"
842 and meta.get("run_id") == run_id
843 and meta.get("label") == unit["unit_id"]
844 and meta.get("input_digest") == unit["packet_digest"]
845 ):
846 validate_runner_contract(run_id, unit, meta)
847 matches.append(entry.name)
848 return sorted(matches)
851def find_attempt(unit: dict, attempt_id: str | None = None) -> dict:
852 attempts = unit.get("attempts", [])
853 if attempt_id:
854 matches = [a for a in attempts if a.get("attempt_id") == attempt_id]
855 else:
856 matches = attempts[-1:]
857 if len(matches) != 1:
858 raise Operational("AMBIGUOUS", "attempt could not be identified exactly")
859 return matches[0]
862def scope_expansion_pending(unit: dict) -> bool:
863 """Return whether the current authored result still requires host resolution."""
864 receipt = find_attempt(unit).get("terminal_receipt")
865 return isinstance(receipt, dict) and receipt.get("terminal_status") == "scope_expansion"
868def cmd_record_job(args) -> tuple[str, dict]:
869 with locked_manifest(args.run_id) as doc:
870 unit = doc["units"].get(args.unit_id)
871 if not unit:
872 raise Operational("REFUSED", "unknown unit")
873 attempt = find_attempt(unit, args.attempt_id)
874 if attempt.get("job_id"):
875 if attempt["job_id"] != args.job_id:
876 raise Operational("AMBIGUOUS", "attempt is already bound to another job")
877 return "AUTHORING", {
878 "unit_id": args.unit_id,
879 "job_id": args.job_id,
880 "resumed": True,
881 "unit_state": unit["state"],
882 }
883 job_dir = runner_job_dir(args.run_id, args.job_id)
884 meta = read_private_json(os.path.join(job_dir, "meta.json"))
885 validate_runner_contract(args.run_id, unit, meta)
886 with locked_manifest(args.run_id, write=True) as doc:
887 unit = doc["units"][args.unit_id]
888 attempt = find_attempt(unit, args.attempt_id)
889 bound_job = attempt.get("job_id")
890 if bound_job == args.job_id:
891 return "AUTHORING", {
892 "unit_id": args.unit_id,
893 "job_id": args.job_id,
894 "resumed": True,
895 "unit_state": unit["state"],
896 }
897 if bound_job is not None:
898 raise Operational("AMBIGUOUS", "attempt was concurrently bound")
899 if unit.get("state") != "queued":
900 raise Operational("REFUSED", "an unbound job can be recorded only while the unit is queued")
901 attempt["job_id"] = args.job_id
902 unit["state"] = "authoring"
903 event(doc, "job-bound", args.unit_id, {"attempt_id": args.attempt_id, "job_id": args.job_id})
904 return "AUTHORING", {"unit_id": args.unit_id, "job_id": args.job_id, "resumed": False}
907def sync_job(run_id: str, unit_id: str) -> dict:
908 with locked_manifest(run_id) as doc:
909 unit = doc["units"].get(unit_id)
910 if not unit:
911 raise Operational("REFUSED", "unknown unit")
912 attempt = find_attempt(unit)
913 if not attempt.get("job_id"):
914 return {"process_state": "never-started", "activity": attempt["activity"]}
915 evidence = process_evidence(runner_job_dir(run_id, attempt["job_id"]))
916 failure_receipt = None
917 oversized_result_failure = False
918 if evidence["process_state"] == "failed":
919 result_fd, result_dir = open_recorded_result_dir(unit)
920 try:
921 result_stat = stat_private_at(
922 result_fd,
923 "implementation-result.json",
924 os.path.join(result_dir, "implementation-result.json"),
925 missing_ok=True,
926 )
927 finally:
928 os.close(result_fd)
929 if result_stat is not None and result_stat.st_size > MAX_RESULT_BYTES:
930 _validate_authorized_failed_job(run_id, unit, attempt)
931 oversized_result_failure = True
932 else:
933 for reader in (unavailable_terminal_receipt, launched_failure_terminal_receipt):
934 try:
935 failure_receipt = reader(run_id, unit, attempt)
936 break
937 except TrustFailure:
938 raise
939 except Operational:
940 continue
941 with locked_manifest(run_id, write=True) as doc:
942 attempt = find_attempt(doc["units"][unit_id])
943 prior_state = attempt.get("process_state")
944 prior_activity = dict(attempt["activity"])
945 prior_fallback = dict(attempt.get("fallback", {}))
946 prior_receipt = attempt.get("terminal_receipt")
947 attempt["process_state"] = evidence["process_state"]
948 attempt["activity"].update(evidence["activity"])
949 if failure_receipt is not None:
950 attempt["terminal_receipt"] = failure_receipt
951 authoritative_failure = evidence["process_state"] in TERMINAL_PROCESS - {"done"} or (
952 evidence["process_state"] == "never-started" and bool(attempt.get("job_id"))
953 )
954 effective_failure_reason = None
955 if authoritative_failure:
956 effective_failure_reason = (
957 failure_receipt["failure_reason"]
958 if failure_receipt is not None
959 else evidence["failure_reason"]
960 if oversized_result_failure and evidence["failure_reason"]
961 else evidence["process_state"]
962 )
963 fallback = attempt.setdefault("fallback", {})
964 fallback.setdefault("claimed", None)
965 fallback["eligible"] = fallback.get("claimed") is None
966 fallback["reason"] = effective_failure_reason
967 changed = (
968 prior_state != evidence["process_state"]
969 or prior_activity != attempt["activity"]
970 or prior_fallback != attempt.get("fallback", {})
971 or prior_receipt != attempt.get("terminal_receipt")
972 )
973 if changed:
974 event(doc, "job-synced", unit_id, {"process_state": evidence["process_state"]})
975 if prior_state != evidence["process_state"] and evidence["process_state"] in TERMINAL_PROCESS:
976 event(doc, "job-terminal", unit_id, {"process_state": evidence["process_state"]})
977 if failure_receipt is not None and prior_receipt != failure_receipt:
978 receipt_event = "route-unavailable" if failure_receipt["terminal_status"] == "unavailable" else "route-failed"
979 event(doc, receipt_event, unit_id, {"failure_reason": failure_receipt["failure_reason"]})
980 activity = dict(attempt["activity"])
981 return {
982 "process_state": evidence["process_state"],
983 "failure_reason": effective_failure_reason,
984 "activity": activity,
985 }
988def cmd_sync_job(args) -> tuple[str, dict]:
989 evidence = sync_job(args.run_id, args.unit_id)
990 return "SYNCED", {"unit_id": args.unit_id, **evidence}
993def transport_ref(run_id: str, unit_id: str) -> str:
994 return f"refs/ce-work/{digest_bytes(run_id.encode())[:20]}/{digest_bytes(unit_id.encode())[:20]}"
997def no_sequencer(workspace: str) -> None:
998 git_dir = git_text(workspace, "rev-parse", "--path-format=absolute", "--absolute-git-dir")
999 for name in ("MERGE_HEAD", "CHERRY_PICK_HEAD", "REVERT_HEAD", "rebase-merge", "rebase-apply"):
1000 if os.path.exists(os.path.join(git_dir, name)):
1001 raise Operational("BLOCKED", f"worker workspace has unresolved Git operation: {name}")
1004def parse_diff_paths(raw: bytes) -> list[str]:
1005 parts = raw.split(b"\0")
1006 paths: list[str] = []
1007 expect_paths = 0
1008 for part in parts:
1009 if not part:
1010 continue
1011 text = part.decode("utf-8", "surrogateescape")
1012 if expect_paths:
1013 paths.append(text)
1014 expect_paths -= 1
1015 else:
1016 expect_paths = 2 if text.startswith(("R", "C")) else 1
1017 if expect_paths:
1018 raise Operational("BLOCKED", "incomplete NUL-delimited transport inventory")
1019 return paths
1022def diff_changes_gitlink(raw: bytes) -> bool:
1023 for record in raw.split(b"\0"):
1024 if not record.startswith(b":"):
1025 continue
1026 fields = record[1:].split(b" ", 4)
1027 if len(fields) >= 2 and b"160000" in fields[:2]:
1028 return True
1029 return False
1032def terminalize(run_id: str, unit_id: str) -> dict:
1033 evidence = sync_job(run_id, unit_id)
1034 if evidence["process_state"] != "done":
1035 detail = {}
1036 if evidence["process_state"] == "failed":
1037 with locked_manifest(run_id) as doc:
1038 attempt = find_attempt(doc["units"][unit_id])
1039 receipt = attempt.get("terminal_receipt")
1040 if isinstance(receipt, dict) and receipt.get("terminal_status") == "unavailable":
1041 detail = {"terminal_receipt": receipt, "failure_reason": receipt["failure_reason"]}
1042 raise Operational(
1043 "BLOCKED",
1044 f"worker is not authoritatively done ({evidence['process_state']})",
1045 detail,
1047 try:
1048 with locked_manifest(run_id) as doc:
1049 unit = doc["units"].get(unit_id)
1050 if not unit:
1051 raise Operational("REFUSED", "unknown unit")
1052 receipt = terminal_receipt(unit, find_attempt(unit))
1053 if receipt.get("model_receipt_status") == "mismatch":
1054 raise Operational("BLOCKED", "adapter reported a served-model mismatch")
1055 except Operational as exc:
1056 record_terminal_validation_failure(run_id, unit_id, exc)
1057 raise
1058 with locked_manifest(run_id, write=True) as doc:
1059 unit = doc["units"].get(unit_id)
1060 if unit and unit["state"] == "authoring":
1061 find_attempt(unit)["terminal_receipt"] = receipt
1062 unit["state"] = "authored"
1063 event(doc, "worker-output-authored", unit_id, {"route": receipt["actual_route"], "model": receipt["model_actual"]})
1064 if receipt["terminal_status"] == "blocked":
1065 raise Operational(
1066 "BLOCKED",
1067 "worker returned a host-resolvable blocker",
1069 "unit_id": unit_id,
1070 "terminal_status": "blocked",
1071 "summary": receipt["summary"],
1072 "terminal_receipt": receipt,
1073 "recovery_path": os.path.join(run_dir(run_id), "units", unit_id),
1074 },
1076 with locked_manifest(run_id, write=True) as doc:
1077 unit = doc["units"].get(unit_id)
1078 if not unit:
1079 raise Operational("REFUSED", "unknown unit")
1080 if unit["state"] == "integration-pending" and unit["transport"].get("commit"):
1081 retire_terminal_validation_failure(unit)
1082 return unit["transport"]
1083 if unit["state"] != "authored":
1084 raise Operational("BLOCKED", f"unit cannot terminalize from {unit['state']}")
1085 if find_attempt(unit).get("fallback", {}).get("claimed"):
1086 raise Operational(
1087 "REFUSED",
1088 "native fallback already owns implementation; worker output cannot be terminalized",
1090 validate_workspace(doc, unit)
1091 workspace = unit["workspace"]["path"]
1092 base = unit["workspace"]["base"]
1093 repo = doc["repository"]["toplevel"]
1094 try:
1095 no_sequencer(workspace)
1096 ignored_raw = git(workspace, "ls-files", "--others", "--ignored", "--exclude-standard", "-z")
1097 ignored_paths = [
1098 part.decode("utf-8", "surrogateescape")
1099 for part in ignored_raw.split(b"\0")
1100 if part
1102 if ignored_paths:
1103 preview = json.dumps(ignored_paths[:20], ensure_ascii=True)
1104 suffix = f" and {len(ignored_paths) - 20} more" if len(ignored_paths) > 20 else ""
1105 raise Operational(
1106 "BLOCKED",
1107 f"worker workspace contains ignored untracked output that cannot enter the transport: {preview}{suffix}",
1108 {"ignored_paths": ignored_paths[:100], "ignored_path_count": len(ignored_paths)},
1110 git(workspace, "add", "-A", "--", ".")
1111 tree = git_text(workspace, "write-tree")
1112 mode_diff = git(repo, "diff-tree", "-r", "--raw", "-z", "--no-renames", base, tree)
1113 if diff_changes_gitlink(mode_diff):
1114 raise Operational("BLOCKED", "submodule state cannot be transported implicitly")
1115 except Operational as exc:
1116 record_terminal_validation_failure(run_id, unit_id, exc)
1117 raise
1118 ref = transport_ref(run_id, unit_id)
1119 existing = git_text(repo, "rev-parse", "-q", "--verify", ref, check=False)
1120 if existing:
1121 parents = git_text(repo, "rev-list", "--parents", "-n", "1", existing).split()
1122 existing_tree = git_text(repo, "rev-parse", f"{existing}^{{tree}}")
1123 if parents != [existing, base] or existing_tree != tree:
1124 raise Operational("BLOCKED", "preexisting transport ref does not match final tree/base")
1125 commit = existing
1126 else:
1127 env = {
1128 "GIT_AUTHOR_NAME": "ce-work transport",
1129 "GIT_AUTHOR_EMAIL": "ce-work@localhost",
1130 "GIT_COMMITTER_NAME": "ce-work transport",
1131 "GIT_COMMITTER_EMAIL": "ce-work@localhost",
1133 commit = git(repo, "commit-tree", tree, "-p", base, input_data=f"ce-work transport {run_id}/{unit_id}\n".encode(), env=env).decode().strip()
1134 zero = "0" * len(commit)
1135 git(repo, "update-ref", ref, commit, zero)
1136 test_fault("after-transport-ref")
1137 raw_diff = git(repo, "diff-tree", "-r", "-M", "--name-status", "-z", base, commit)
1138 paths = parse_diff_paths(raw_diff)
1139 tdigest = digest_bytes(base.encode() + b"\0" + tree.encode() + b"\0" + commit.encode() + b"\0" + raw_diff)
1140 transport = {
1141 "base": base, "tree": tree, "commit": commit, "ref": ref,
1142 "digest": tdigest, "changed_paths": paths,
1143 "inventory_b64": base64.b64encode(raw_diff).decode(),
1145 # Make successful cleanup non-destructive: after F is pinned, normalize the
1146 # retained inspection worktree to the exact transported tree.
1147 git(workspace, "reset", "--hard", commit)
1148 with locked_manifest(run_id, write=True) as doc:
1149 unit = doc["units"][unit_id]
1150 if unit["state"] not in ("authored", "integration-pending"):
1151 raise Operational("BLOCKED", "unit state changed during terminalization")
1152 retire_terminal_validation_failure(unit)
1153 unit["state"] = "integration-pending"
1154 unit["transport"] = transport
1155 event(doc, "transport-pinned", unit_id, {"commit": commit, "ref": ref, "digest": tdigest})
1156 return transport
1159def cmd_terminalize(args) -> tuple[str, dict]:
1160 transport = terminalize(args.run_id, args.unit_id)
1161 return "INTEGRATION_PENDING", {"unit_id": args.unit_id, "transport": transport}