diff --git a/.superpowers/sdd/evidence-task-4-report.md b/.superpowers/sdd/evidence-task-4-report.md index bd18ce30..8cfd32bb 100644 --- a/.superpowers/sdd/evidence-task-4-report.md +++ b/.superpowers/sdd/evidence-task-4-report.md @@ -85,3 +85,23 @@ Task 4 scoped Ruff: All checks passed ``` Repository-wide Ruff continues to report the same 34 unrelated pre-existing legacy-test findings. + +## Final resume-integrity fix + +Resume is now read-only until the source checkpoint proves trustworthy. The runner loads the source +before allocating a new run ID or directory, validates the exact stage state/timestamp/error ledger, +rejects duplicate stage identifiers, and recomputes compatibility from every persisted compatibility +field plus the exact ordered persisted stage IDs. It first requires the stored fingerprint to match +that recomputation, then compares the trusted recomputation with the requested job fingerprint. + +Valid-JSON tampering tests cover removed, inserted/duplicated, reordered, and substituted stages; +input-field and stored-fingerprint changes; and invalid stage-state shapes. Every rejection occurs +before stage execution and asserts that the runs directory contains no orphan allocation. + +Final verification: + +```text +focused job/lock suite: 34 passed in 0.56s +full harness suite: 620 passed, 5 deselected, 17 warnings in 27.42s +Task 4 scoped Ruff: All checks passed +``` diff --git a/harness/tests/test_job_runner.py b/harness/tests/test_job_runner.py index d0821271..76d6ff69 100644 --- a/harness/tests/test_job_runner.py +++ b/harness/tests/test_job_runner.py @@ -233,3 +233,59 @@ def test_run_directories_are_private_and_fsynced_before_atomic_replace(tmp_path, first_replace = events.index("replace") assert "fsync" in events[:first_replace] assert "fsync" in events[first_replace + 1 :] + + +def _tamper_checkpoint(tmp_path, report, transform): + path = tmp_path / ".tht-jobs" / report.job_type / "runs" / report.run_id / "checkpoint.json" + payload = json.loads(path.read_text()) + transform(payload) + path.write_text(json.dumps(payload)) + + +@pytest.mark.parametrize( + "transform", + [ + lambda payload: payload["stages"].pop(), + lambda payload: payload["stages"].reverse(), + lambda payload: payload["stages"].append(dict(payload["stages"][0])), + lambda payload: payload["stages"][0].update(name="substitute"), + lambda payload: payload.update(input_fingerprint="sha256:" + "f" * 64), + lambda payload: payload["stages"][0].update(status="pending"), + ], +) +def test_semantically_tampered_checkpoint_fails_without_orphan_run(tmp_path, transform): + def one(_context): + pass + + def two(_context): + pass + + spec = _spec(tmp_path, stage_ids=("one", "two")) + first = run_job(spec, [one, two]) + runs = tmp_path / ".tht-jobs" / "evidence" / "runs" + before = {path.name for path in runs.iterdir()} + _tamper_checkpoint(tmp_path, first, transform) + called = False + + def forbidden(_context): + nonlocal called + called = True + + with pytest.raises(CorruptCheckpointError, match="invalid|incompatible"): + run_job(spec.with_resume(first.run_id), [forbidden, forbidden]) + assert called is False + assert {path.name for path in runs.iterdir()} == before + + +def test_stored_compatibility_fingerprint_tamper_fails_without_orphan(tmp_path): + first = run_job(_spec(tmp_path), [lambda _context: None]) + runs = tmp_path / ".tht-jobs" / "evidence" / "runs" + before = {path.name for path in runs.iterdir()} + _tamper_checkpoint( + tmp_path, + first, + lambda payload: payload.update(compatibility_fingerprint="sha256:" + "0" * 64), + ) + with pytest.raises(CorruptCheckpointError, match="invalid"): + run_job(_spec(tmp_path, resume_run_id=first.run_id), [lambda _context: None]) + assert {path.name for path in runs.iterdir()} == before diff --git a/harness/tht/jobs/models.py b/harness/tht/jobs/models.py index 0572ec08..d4c9b70f 100644 --- a/harness/tht/jobs/models.py +++ b/harness/tht/jobs/models.py @@ -7,7 +7,7 @@ from datetime import UTC, datetime from pathlib import Path from typing import Literal, Self -from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator, model_validator _JOB_KEY = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") @@ -113,6 +113,26 @@ class StageRun(_FrozenModel): _name_key = field_validator("name")(_validate_job_key) + @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) + ): + raise ValueError("pending stage cannot contain timestamps or error") + if self.status == "running" and ( + self.started_at is None or self.finished_at is not None or self.error is not None + ): + raise ValueError("running stage requires only started_at") + if self.status == "succeeded" and ( + self.started_at is None or self.finished_at is None or self.error is not None + ): + raise ValueError("succeeded stage requires timestamps and no error") + if self.status == "failed" and ( + 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") + return self + class JobRun(_FrozenModel): """Durable checkpoint, persisted after every state transition.""" @@ -144,6 +164,32 @@ class JobRun(_FrozenModel): _persisted_versions = field_validator("spec_version", "pipeline_version")(_validate_job_key) _resumed_from = field_validator("resumed_from")(_validate_run_id) + @model_validator(mode="after") + def ledger_shape(self) -> "JobRun": + names = [stage.name for stage in self.stages] + if len(names) != len(set(names)): + raise ValueError("stage identifiers must be unique") + statuses = [stage.status for stage in self.stages] + first_incomplete = next( + (index for index, status in enumerate(statuses) if status != "succeeded"), + len(statuses), + ) + if any(status != "pending" for status in statuses[first_incomplete + 1 :]): + raise ValueError("stage ledger must be an ordered execution prefix") + if self.status == "succeeded" and ( + self.finished_at is None or any(status != "succeeded" for status in statuses) + ): + raise ValueError("succeeded job requires a complete succeeded ledger") + if self.status == "failed" and ( + self.finished_at is None + or first_incomplete == len(statuses) + or statuses[first_incomplete] != "failed" + ): + raise ValueError("failed job requires the first incomplete stage to be failed") + if self.status == "running" and self.finished_at is not None: + raise ValueError("running job cannot have finished_at") + return self + class JobReport(JobRun): """Public machine-readable terminal report (contains no paths or stage outputs).""" diff --git a/harness/tht/jobs/runner.py b/harness/tht/jobs/runner.py index 19056370..1e05a869 100644 --- a/harness/tht/jobs/runner.py +++ b/harness/tht/jobs/runner.py @@ -94,8 +94,11 @@ def _resume_run(spec: JobSpec, run_id: str, stages: Sequence[Stage], source: Job requested_names = list(spec.stage_ids) if len(requested_names) != len(stages) or len(requested_names) != len(set(requested_names)): raise CorruptCheckpointError("resume checkpoint is incompatible with requested stages") + source_fingerprint = _source_compatibility_fingerprint(source) + if source.compatibility_fingerprint != source_fingerprint: + raise CorruptCheckpointError("resume checkpoint compatibility fingerprint is invalid") expected = _compatibility_fingerprint(spec, requested_names) - if source.compatibility_fingerprint != expected: + if source_fingerprint != expected or [stage.name for stage in source.stages] != requested_names: raise CorruptCheckpointError( "resume checkpoint is incompatible; start an intentional new run without resume" ) @@ -129,12 +132,8 @@ def run_job(spec: JobSpec, stages: Sequence[Stage]) -> JobReport: """Run stages once, returning a terminal report instead of leaking stage exceptions.""" with WorkspaceJobLock(spec.workspace_root, spec.workspace_id, spec.job_type): jobs_root = spec.workspace_root / ".tht-jobs" / spec.job_type / "runs" - run_id = uuid.uuid4().hex - run_dir = jobs_root / run_id - _prepare_run_directory(spec.workspace_root, spec.job_type, run_id) - checkpoint_path = run_dir / "checkpoint.json" if spec.resume_run_id is None: - run = _new_run(spec, run_id, stages) + source = None else: source_path = jobs_root / spec.resume_run_id / "checkpoint.json" if not source_path.exists(): @@ -146,6 +145,15 @@ def run_job(spec: JobSpec, stages: Sequence[Stage]) -> JobReport: if len(matches) == 1: source_path = matches[0] source = _load_checkpoint(source_path) + _validate_resume_source(spec, stages, source) + + run_id = uuid.uuid4().hex + run_dir = jobs_root / run_id + _prepare_run_directory(spec.workspace_root, spec.job_type, run_id) + checkpoint_path = run_dir / "checkpoint.json" + if source is None: + run = _new_run(spec, run_id, stages) + else: run = _resume_run(spec, run_id, stages, source) _persist(checkpoint_path, run) context = JobContext(run_id, spec.job_type, spec.dry_run, spec.workspace_root, run_dir) @@ -213,6 +221,26 @@ def _compatibility_fingerprint(spec: JobSpec, stage_ids: list[str]) -> str: return _value_fingerprint(canonical) +def _source_compatibility_fingerprint(source: JobRun) -> str: + payload = { + "schema_version": source.schema_version, + "workspace": source.workspace_fingerprint, + "job_type": source.job_type, + "dry_run": source.dry_run, + "spec_version": source.spec_version, + "pipeline_version": source.pipeline_version, + "config_fingerprint": source.config_fingerprint, + "input_fingerprint": source.input_fingerprint, + "stage_ids": [stage.name for stage in source.stages], + } + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return _value_fingerprint(canonical) + + +def _validate_resume_source(spec: JobSpec, stages: Sequence[Stage], source: JobRun) -> None: + _resume_run(spec, "0" * 32, stages, source) + + def _prepare_run_directory(workspace_root: Path, job_type: str, run_id: str) -> None: parent_fd = os.open(workspace_root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: