From b6a52995ae3d8ea5453dd9efcd48377e2d93b31e Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 04:54:00 +0200 Subject: [PATCH] fix(jobs): bind completed effects to checkpoints --- .superpowers/sdd/evidence-task-5b-report.md | 21 +++++++ harness/tests/test_corpus_pipeline.py | 48 +++++++++++++++ harness/tests/test_job_runner.py | 28 +++++++++ harness/tht/jobs/models.py | 16 ++++- harness/tht/jobs/runner.py | 68 +++++++++++++++++++-- 5 files changed, 174 insertions(+), 7 deletions(-) diff --git a/.superpowers/sdd/evidence-task-5b-report.md b/.superpowers/sdd/evidence-task-5b-report.md index 77c10f94..ad0fcfdf 100644 --- a/.superpowers/sdd/evidence-task-5b-report.md +++ b/.superpowers/sdd/evidence-task-5b-report.md @@ -71,3 +71,24 @@ Fresh fix-wave verification: - Available harness suite (same sandbox exclusions described above): `579 passed, 5 deselected, 31 warnings`. - Scoped Ruff and `git diff --check`: clean. + +## Final P1 fix — effect state and checkpoint-bound manifest roots + +- Stage checkpoints now distinguish `intent` from `completed`. Vector intent is atomically sealed + and checkpointed before upsert. A process-level `BaseException` after a partial multi-record + write leaves the stage `running/intent`; resume never promotes it and instead reconciles + `existing_hashes`, writing only the missing records. The completed state is persisted only after + reconciliation returns successfully. +- Every stage now persists its completed artifact state while still `running`, before the + post-call fault hook. The checkpoint binds the SHA-256 of canonical `artifact-manifest.json`, + effect state, exact producer stage, and exact required-file mapping. Resume validates this root + and all bindings before promotion or copying. +- Added process-interruption coverage proving the already-written vector record is not submitted + twice, remaining records are written, and publish completes only after reconciliation. Added + coordinated artifact/manifest, spec-binding, and producer-binding tamper rejection tests. + +Fresh verification: + +- Focused jobs/corpus/CLI/search suite: `86 passed, 18 warnings`. +- Available broad harness suite: `583 passed, 5 deselected, 32 warnings`. +- Scoped Ruff and `git diff --check`: clean. diff --git a/harness/tests/test_corpus_pipeline.py b/harness/tests/test_corpus_pipeline.py index a9383260..58cd25f4 100644 --- a/harness/tests/test_corpus_pipeline.py +++ b/harness/tests/test_corpus_pipeline.py @@ -62,6 +62,22 @@ class Vectors: return 0 +class InterruptingVectors(Vectors): + def __init__(self): + super().__init__() + self.batches = [] + self.interrupt = True + + def upsert(self, collection, records): + self.batches.append([value.record.id for value in records]) + if self.interrupt: + self.interrupt = False + self.records.append(records[0]) + raise KeyboardInterrupt("process interruption after partial write") + self.records.extend(records) + return len(records) + + def item(name, fingerprint): return SourceObject( source_id=f"fs:{name}", uri=f"file:///safe/{name}.md", fingerprint=f"sha256:{fingerprint}" @@ -309,3 +325,35 @@ def test_job_pipeline_rejects_corrupt_required_artifacts_before_resume( input_fingerprint="sha256:" + "2" * 64, resume_run_id=crashed.name, ) + + +def test_vector_intent_is_reconciled_after_process_interruption_without_duplicate_upsert(tmp_path): + one = item("one", "a") + vectors = InterruptingVectors() + candidate = pipeline( + tmp_path, Source([(one, "a" * 250)]), vectors=vectors, + policy=ChunkPolicy(version="chunk-v1", max_chars=100), + ) + with pytest.raises(KeyboardInterrupt): + candidate.run_as_job( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint="sha256:" + "1" * 64, + input_fingerprint="sha256:" + "2" * 64, + ) + runs = tmp_path / ".tht-jobs" / "evidence" / "runs" + interrupted = next(runs.iterdir()) + checkpoint = __import__("json").loads((interrupted / "checkpoint.json").read_text()) + vector_stage = checkpoint["stages"][3] + assert vector_stage["status"] == "running" + assert vector_stage["effect_state"] == "intent" + first_written = vectors.batches[0][0] + + result = candidate.run_as_job( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint="sha256:" + "1" * 64, + input_fingerprint="sha256:" + "2" * 64, + resume_run_id=interrupted.name, + ) + assert result.status == "succeeded" and result.published is True + assert first_written not in vectors.batches[1] + assert len(vectors.records) == 3 diff --git a/harness/tests/test_job_runner.py b/harness/tests/test_job_runner.py index 05298d35..0874b047 100644 --- a/harness/tests/test_job_runner.py +++ b/harness/tests/test_job_runner.py @@ -145,6 +145,34 @@ def test_nonexistent_well_formed_resume_run_id_is_rejected(tmp_path): run_job(_spec(tmp_path).with_resume("a" * 32), [lambda _context: None]) +@pytest.mark.parametrize("tamper", ["artifact_and_manifest", "spec", "producer"]) +def test_resume_rejects_manifest_root_or_binding_tamper(tmp_path, tamper): + def stage(context): + artifacts = context.run_dir / "artifacts" + artifacts.mkdir() + (artifacts / "effect.json").write_text("ok") + return StageArtifacts(("effect.json",)) + + report = run_job(_spec(tmp_path), [stage]) + artifacts = tmp_path / ".tht-jobs" / "evidence" / "runs" / report.run_id / "artifacts" + manifest_path = artifacts / "artifact-manifest.json" + manifest = json.loads(manifest_path.read_text()) + if tamper == "artifact_and_manifest": + (artifacts / "effect.json").write_text("evil") + digest = __import__("hashlib").sha256(b"evil").hexdigest() + manifest["stages"]["stage"]["files"]["effect.json"] = { + "sha256": digest, "size": 4, + } + elif tamper == "spec": + manifest["spec_fingerprint"] = "sha256:" + "0" * 64 + else: + manifest["stages"]["other"] = manifest["stages"].pop("stage") + manifest_path.write_text(json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n") + + with pytest.raises(CorruptCheckpointError, match="artifact"): + run_job(_spec(tmp_path).with_resume(report.run_id), [stage]) + + def test_successful_job_is_idempotently_resumable(tmp_path): calls = [] diff --git a/harness/tht/jobs/models.py b/harness/tht/jobs/models.py index d4c9b70f..e85cd8a1 100644 --- a/harness/tht/jobs/models.py +++ b/harness/tht/jobs/models.py @@ -15,6 +15,7 @@ _RUN_ID = re.compile(r"^[0-9a-f]{32}$") _FINGERPRINT = re.compile(r"^sha256:[0-9a-f]{64}$") JobStatus = Literal["pending", "running", "succeeded", "failed"] StageStatus = Literal["pending", "running", "succeeded", "failed"] +EffectState = Literal["intent", "completed"] def utc_now() -> datetime: @@ -110,13 +111,22 @@ class StageRun(_FrozenModel): started_at: datetime | None = None finished_at: datetime | None = None error: StageError | None = None + effect_state: EffectState | None = None + artifact_manifest_digest: str | None = None + artifact_files: tuple[str, ...] = () _name_key = field_validator("name")(_validate_job_key) + _artifact_digest = field_validator("artifact_manifest_digest")( + lambda value: value if value is None or _FINGERPRINT.fullmatch(value) else _invalid_fingerprint() + ) @model_validator(mode="after") def state_shape(self) -> "StageRun": if self.status == "pending" and any( - value is not None for value in (self.started_at, self.finished_at, self.error) + value is not None for value in ( + self.started_at, self.finished_at, self.error, self.effect_state, + self.artifact_manifest_digest, + ) ): raise ValueError("pending stage cannot contain timestamps or error") if self.status == "running" and ( @@ -131,6 +141,10 @@ class StageRun(_FrozenModel): self.started_at is None or self.finished_at is None or self.error is None ): raise ValueError("failed stage requires timestamps and safe error") + if (self.effect_state is None) != (self.artifact_manifest_digest is None): + raise ValueError("effect state and artifact manifest digest must be persisted together") + if self.artifact_files and self.effect_state is None: + raise ValueError("artifact files require a persisted effect state") return self diff --git a/harness/tht/jobs/runner.py b/harness/tht/jobs/runner.py index 8ace8104..5c052760 100644 --- a/harness/tht/jobs/runner.py +++ b/harness/tht/jobs/runner.py @@ -35,6 +35,14 @@ class JobContext: dry_run: bool workspace_root: Path run_dir: Path + _record_artifacts: Callable[[str, tuple[str, ...], str], None] | None = None + + def record_artifacts( + self, stage: str, required: tuple[str, ...], effect_state: str, + ) -> None: + if self._record_artifacts is None: + raise RuntimeError("artifact recorder is unavailable") + self._record_artifacts(stage, required, effect_state) Stage = Callable[[JobContext], Any] @@ -45,7 +53,7 @@ def _artifact_digest(path: Path) -> dict[str, Any]: return {"sha256": hashlib.sha256(payload).hexdigest(), "size": len(payload)} -def _seal_artifacts(context: JobContext, stage: str, result: Any, spec: JobSpec) -> None: +def _seal_artifacts(context: JobContext, stage: str, result: Any, spec: JobSpec) -> str: required = result.required if isinstance(result, StageArtifacts) else () root = context.run_dir / "artifacts" root.mkdir(exist_ok=True) @@ -68,14 +76,16 @@ def _seal_artifacts(context: JobContext, stage: str, result: Any, spec: JobSpec) prior["required"].remove(relative) prior["files"].pop(relative, None) manifest["stages"][stage] = {"required": list(required), "files": files} - _atomic_write(manifest_path, json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n") + canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n" + _atomic_write(manifest_path, canonical) + return _value_fingerprint(canonical) def seal_stage_artifacts( context: JobContext, stage: str, required: tuple[str, ...], spec: JobSpec, ) -> None: """Durably record external-effect intent before a stage performs that effect.""" - _seal_artifacts(context, stage, StageArtifacts(required), spec) + context.record_artifacts(stage, required, "intent") def _validate_artifacts(run_dir: Path, spec: JobSpec, source: JobRun) -> set[str]: @@ -86,6 +96,7 @@ def _validate_artifacts(run_dir: Path, spec: JobSpec, source: JobRun) -> set[str return set() try: manifest = json.loads(manifest_path.read_text()) + root_digest = _value_fingerprint(manifest_path.read_text()) if manifest["spec_fingerprint"] != _compatibility_fingerprint(spec, list(spec.stage_ids)): raise ValueError sealed = set(manifest["stages"]) @@ -100,6 +111,16 @@ def _validate_artifacts(run_dir: Path, spec: JobSpec, source: JobRun) -> set[str allowed.add(relative) if not successful.issubset(sealed): raise ValueError + for stage in source.stages: + if stage.effect_state is None: + continue + record = manifest["stages"].get(stage.name) + if ( + stage.artifact_manifest_digest != root_digest + or record is None + or tuple(record["required"]) != stage.artifact_files + ): + raise ValueError entries = list(root.iterdir()) if any(path.is_symlink() or not path.is_file() for path in entries): raise ValueError @@ -113,7 +134,10 @@ def _validate_artifacts(run_dir: Path, spec: JobSpec, source: JobRun) -> set[str allowed.add("compensated.json") if actual != allowed: raise ValueError - return sealed + return { + stage.name for stage in source.stages + if stage.effect_state == "completed" + } except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error: raise CorruptCheckpointError("resume artifact manifest is invalid") from error @@ -199,6 +223,9 @@ def _resume_run( resumed_stages.append(StageRun( name=name, status="succeeded", started_at=previous.started_at or utc_now(), finished_at=utc_now(), + effect_state="completed", + artifact_manifest_digest=previous.artifact_manifest_digest, + artifact_files=previous.artifact_files, )) else: resumed_stages.append(StageRun(name=name)) @@ -254,9 +281,36 @@ def run_job( if source_artifacts.exists(): shutil.copytree(source_artifacts, run_dir / "artifacts") _persist(checkpoint_path, run) - context = JobContext(run_id, spec.job_type, spec.dry_run, spec.workspace_root, run_dir) + current_index = -1 + + def record_artifacts(stage_name: str, required: tuple[str, ...], effect_state: str) -> None: + nonlocal run + if current_index < 0 or run.stages[current_index].name != stage_name: + raise CorruptCheckpointError("artifact producer does not match running stage") + digest = _seal_artifacts(context, stage_name, StageArtifacts(required), spec) + manifest = json.loads((run_dir / "artifacts" / "artifact-manifest.json").read_text()) + updated = [] + for position, value in enumerate(run.stages): + record = manifest["stages"].get(value.name) + if record is not None and (value.status == "succeeded" or position == current_index): + state = effect_state if position == current_index else value.effect_state + updated.append(value.model_copy(update={ + "effect_state": state, + "artifact_manifest_digest": digest, + "artifact_files": tuple(record["required"]), + })) + else: + updated.append(value) + run = run.model_copy(update={"stages": tuple(updated)}) + _persist(checkpoint_path, run) + + context = JobContext( + run_id, spec.job_type, spec.dry_run, spec.workspace_root, run_dir, + record_artifacts, + ) for index, stage_callable in enumerate(stages): + current_index = index if run.stages[index].status == "succeeded": continue stage = run.stages[index].model_copy( @@ -285,7 +339,9 @@ def run_job( ) _persist(checkpoint_path, run) break - _seal_artifacts(context, stage.name, stage_result, spec) + required = stage_result.required if isinstance(stage_result, StageArtifacts) else () + context.record_artifacts(stage.name, required, "completed") + stage = run.stages[index] if after_stage_return is not None: after_stage_return(context, stage.name) succeeded = stage.model_copy(update={"status": "succeeded", "finished_at": utc_now()})