zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
602 lines
21 KiB
Python
Executable File
602 lines
21 KiB
Python
Executable File
"""Incremental checkpoints use isolated local repositories, never product or remote state."""
|
|
|
|
from argparse import Namespace
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
|
|
from govoplan_devkit import runner
|
|
from govoplan_devkit.checkpoints import Checkpoints
|
|
from govoplan_devkit.common import atomic_json, read_json, resource_lock, state_root
|
|
|
|
|
|
@pytest.fixture
|
|
def workspace(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "xdg"))
|
|
root = tmp_path / "workspace"
|
|
repos = {}
|
|
for name in ("alpha", "beta"):
|
|
repo = root / name
|
|
repo.mkdir(parents=True)
|
|
subprocess.run(["git", "init", "-q", str(repo)], check=True)
|
|
(repo / "source.txt").write_text(name + " original\n")
|
|
subprocess.run(["git", "-C", str(repo), "add", "source.txt"], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(repo),
|
|
"-c",
|
|
"user.name=Fixture",
|
|
"-c",
|
|
"user.email=fixture@example.invalid",
|
|
"commit",
|
|
"-qm",
|
|
"fixture",
|
|
],
|
|
check=True,
|
|
)
|
|
repos[name] = repo
|
|
config = tmp_path / "project.json"
|
|
config.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"name": "Incremental fixture",
|
|
"repositories": [{"name": name, "path": name} for name in repos],
|
|
"checks": [],
|
|
"profiles": {},
|
|
}
|
|
)
|
|
)
|
|
traces = tmp_path / "traces"
|
|
traces.mkdir()
|
|
args = Namespace(
|
|
workspace_root=root,
|
|
project=config,
|
|
state_dir=tmp_path / "state",
|
|
dry_run=False,
|
|
jobs=2,
|
|
profile="quick",
|
|
resume=None,
|
|
repo=[],
|
|
changed=False,
|
|
)
|
|
return args, repos, traces
|
|
|
|
|
|
def stage(workspace, identity, repo="alpha", *, inputs=True, body="", **extra):
|
|
_, repos, traces = workspace
|
|
counter = traces / identity
|
|
code = (
|
|
"from pathlib import Path; "
|
|
f"counter=Path({str(counter)!r}); "
|
|
"counter.write_text(str(int(counter.read_text())+1 if counter.exists() else 1)); "
|
|
f"print({identity!r},flush=True); " + body
|
|
)
|
|
result = {
|
|
"id": identity,
|
|
"title": identity,
|
|
"argv": [sys.executable, "-c", code],
|
|
"cwd": str(repos[repo]),
|
|
"timeout_seconds": 5,
|
|
**extra,
|
|
}
|
|
if inputs:
|
|
result["inputs"] = {"repos": [repo]}
|
|
return result
|
|
|
|
|
|
def run(workspace, stages, *, environment="e" * 64):
|
|
args, _, _ = workspace
|
|
with patch.object(runner, "environment_fingerprint", return_value=environment):
|
|
return runner.run_checks(args, stages)
|
|
|
|
|
|
def count(workspace, identity):
|
|
path = workspace[2] / identity
|
|
return int(path.read_text()) if path.exists() else 0
|
|
|
|
|
|
def stages_by_id(result):
|
|
return {item["id"]: item for item in result["stages"]}
|
|
|
|
|
|
def test_checkpoint_probes_and_durable_persistence_hold_stage_resource_lock(workspace):
|
|
args, _, _ = workspace
|
|
locks = state_root(args.workspace_root) / "resource-locks"
|
|
original_identity, original_write = Checkpoints.identity, runner.atomic_json
|
|
probes, persisted = [], []
|
|
|
|
def assert_held():
|
|
with pytest.raises(RuntimeError, match="busy"):
|
|
with resource_lock(locks, "checkpoint-fixture"):
|
|
pass
|
|
|
|
def identity(self, selected):
|
|
assert_held()
|
|
probes.append(selected["id"])
|
|
return original_identity(self, selected)
|
|
|
|
def write(path, payload):
|
|
original_write(path, payload)
|
|
if (
|
|
payload.get("phase") == "checking"
|
|
and payload["stages"][0].get("checkpoint_verified") is True
|
|
and not persisted
|
|
):
|
|
assert_held()
|
|
assert read_json(path)["stages"][0]["checkpoint_verified"] is True
|
|
persisted.append(path)
|
|
|
|
with (
|
|
patch.object(Checkpoints, "identity", identity),
|
|
patch.object(runner, "atomic_json", side_effect=write),
|
|
):
|
|
result = run(
|
|
workspace, [stage(workspace, "a", resources=["checkpoint-fixture"])]
|
|
)
|
|
assert result["status"] == "passed"
|
|
assert len(probes) >= 2 and len(persisted) == 1
|
|
|
|
|
|
def test_checkpoint_save_failure_never_releases_dependent_execution(workspace):
|
|
args, _, _ = workspace
|
|
original_write, original_command, original_wait = (
|
|
runner.atomic_json,
|
|
runner._execute_stage_command,
|
|
runner.wait,
|
|
)
|
|
scheduler_waiting = threading.Event()
|
|
checkpoint_stalled = threading.Event()
|
|
scheduler_rechecked = threading.Event()
|
|
release_failure = threading.Event()
|
|
failed_write_paths, results, errors = [], [], []
|
|
|
|
def command(selected, *values, **kwargs):
|
|
if selected["id"] == "producer":
|
|
# Let the scheduler finish its initial running-state save and enter
|
|
# its wait loop before the command publishes a passing checkpoint.
|
|
assert scheduler_waiting.wait(timeout=5)
|
|
return original_command(selected, *values, **kwargs)
|
|
|
|
def wait(*values, **kwargs):
|
|
scheduler_waiting.set()
|
|
result = original_wait(*values, **kwargs)
|
|
if checkpoint_stalled.is_set():
|
|
scheduler_rechecked.set()
|
|
return result
|
|
|
|
def write(path, payload):
|
|
producer = next(
|
|
(item for item in payload.get("stages", []) if item["id"] == "producer"),
|
|
None,
|
|
)
|
|
if (
|
|
producer
|
|
and producer.get("checkpoint_verified") is True
|
|
and not failed_write_paths
|
|
):
|
|
failed_write_paths.append(path)
|
|
checkpoint_stalled.set()
|
|
assert release_failure.wait(timeout=5)
|
|
raise OSError("fixture checkpoint persistence failed")
|
|
return original_write(path, payload)
|
|
|
|
def execute():
|
|
try:
|
|
results.append(
|
|
run(
|
|
workspace,
|
|
[
|
|
stage(workspace, "producer"),
|
|
stage(workspace, "dependent", "beta", deps=["producer"]),
|
|
],
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
worker = threading.Thread(target=execute)
|
|
with (
|
|
patch.object(runner, "atomic_json", side_effect=write),
|
|
patch.object(runner, "_execute_stage_command", side_effect=command),
|
|
patch.object(runner, "wait", side_effect=wait),
|
|
):
|
|
worker.start()
|
|
try:
|
|
assert checkpoint_stalled.wait(timeout=5)
|
|
assert scheduler_rechecked.wait(timeout=5)
|
|
# Keep persistence blocked across a scheduling turn: the in-memory
|
|
# producer status must not grant authority to start a consumer.
|
|
time.sleep(0.15)
|
|
assert count(workspace, "dependent") == 0
|
|
durable = read_json(failed_write_paths[0])
|
|
assert (
|
|
stages_by_id(durable)["producer"].get("checkpoint_verified") is not True
|
|
)
|
|
finally:
|
|
release_failure.set()
|
|
worker.join(timeout=8)
|
|
assert not worker.is_alive() and not errors
|
|
assert len(results) == 1 and results[0]["status"] == "failed"
|
|
assert count(workspace, "producer") == 1 and count(workspace, "dependent") == 0
|
|
by_id = stages_by_id(results[0])
|
|
assert by_id["producer"]["status"] == "failed"
|
|
assert by_id["producer"]["checkpoint_verified"] is False
|
|
assert "fixture checkpoint persistence failed" in by_id["producer"]["error"]
|
|
assert by_id["dependent"]["status"] == "skipped"
|
|
persisted = runner.read_receipt(
|
|
args.workspace_root, args.state_dir, results[0]["run_id"]
|
|
)
|
|
assert stages_by_id(persisted)["producer"]["checkpoint_verified"] is False
|
|
|
|
|
|
def test_verified_checkpoint_reuses_unchanged_stage_after_other_repo_changes(workspace):
|
|
args, repos, _ = workspace
|
|
plan = [stage(workspace, "a"), stage(workspace, "b", "beta")]
|
|
first = run(workspace, plan)
|
|
old_bytes = Path(first["receipt_path"]).read_bytes()
|
|
(repos["beta"] / "source.txt").write_text("beta changed\n")
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
by_id = stages_by_id(second)
|
|
assert second["status"] == "passed" and second["snapshot_verified"] is True
|
|
assert count(workspace, "a") == 1 and count(workspace, "b") == 2
|
|
assert by_id["a"]["reused_from"] == first["run_id"]
|
|
assert "reused_from" not in by_id["b"]
|
|
assert by_id["a"]["checkpoint_verified"] is True
|
|
assert by_id["a"]["checkpoint_version"] == 1
|
|
assert isinstance(by_id["a"]["cache_key"], str) and by_id["a"]["cache_key"]
|
|
assert second["source_fingerprint"] != first["source_fingerprint"]
|
|
assert Path(first["receipt_path"]).read_bytes() == old_bytes
|
|
|
|
|
|
def test_failed_post_execution_probe_preserves_actual_log_without_certifying_it(
|
|
workspace,
|
|
):
|
|
from govoplan_devkit.checkpoints import Checkpoints
|
|
|
|
identity = Checkpoints.identity
|
|
calls = 0
|
|
|
|
def probe(self, selected):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
raise ValueError("fixture input became unreadable")
|
|
return identity(self, selected)
|
|
|
|
with patch.object(Checkpoints, "identity", probe):
|
|
result = run(workspace, [stage(workspace, "a")])
|
|
selected = result["stages"][0]
|
|
assert result["status"] == "stale"
|
|
assert selected["status"] == "stale" and selected["exit_code"] == 0
|
|
assert selected["checkpoint_verified"] is False
|
|
assert Path(selected["log_path"]).read_text() == "a\n"
|
|
assert "fixture input became unreadable" in selected["error"]
|
|
|
|
|
|
def test_changed_and_new_commands_run_without_discarding_unrelated_checkpoint(
|
|
workspace,
|
|
):
|
|
args, _, _ = workspace
|
|
first = run(workspace, [stage(workspace, "a"), stage(workspace, "b", "beta")])
|
|
args.resume = first["run_id"]
|
|
second = run(
|
|
workspace,
|
|
[
|
|
stage(workspace, "a"),
|
|
stage(workspace, "b", "beta", body="print('changed command')"),
|
|
stage(workspace, "new", "beta"),
|
|
],
|
|
)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 1
|
|
assert count(workspace, "b") == 2
|
|
assert count(workspace, "new") == 1
|
|
assert first["plan_fingerprint"] != second["plan_fingerprint"]
|
|
|
|
|
|
def test_unrelated_profile_edit_does_not_change_existing_stage_execution_identity(
|
|
workspace,
|
|
):
|
|
args, _, _ = workspace
|
|
plan = [stage(workspace, "a")]
|
|
first = run(workspace, plan)
|
|
config = json.loads(args.project.read_text())
|
|
config["checks"] = [
|
|
{
|
|
"id": "extra",
|
|
"argv": [sys.executable, "-c", "print('extra')"],
|
|
"cwd": "beta",
|
|
"repos": ["beta"],
|
|
}
|
|
]
|
|
config["profiles"] = {"backend": ["extra"]}
|
|
args.project.write_text(json.dumps(config))
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 1
|
|
|
|
|
|
def test_added_dependency_edge_invalidates_consumer_even_when_both_repo_bytes_match(
|
|
workspace,
|
|
):
|
|
args, _, _ = workspace
|
|
a, b = stage(workspace, "a"), stage(workspace, "b", "beta")
|
|
first = run(workspace, [a, b])
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, [a, {**b, "deps": ["a"]}])
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 1 and count(workspace, "b") == 2
|
|
|
|
|
|
def test_failed_run_reuses_successful_phase_but_retries_failed_phase(workspace):
|
|
args, _, traces = workspace
|
|
ready = traces / "ready"
|
|
plan = [
|
|
stage(workspace, "a"),
|
|
stage(
|
|
workspace,
|
|
"b",
|
|
"beta",
|
|
after=["a"],
|
|
body=f"raise SystemExit(0 if Path({str(ready)!r}).exists() else 3)",
|
|
),
|
|
]
|
|
first = run(workspace, plan)
|
|
assert first["status"] == "failed"
|
|
assert stages_by_id(first)["a"]["checkpoint_verified"] is True
|
|
ready.touch()
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 1 and count(workspace, "b") == 2
|
|
|
|
|
|
def test_changed_dependency_invalidates_transitive_consumers_but_not_independent_stage(
|
|
workspace,
|
|
):
|
|
args, repos, _ = workspace
|
|
plan = [
|
|
stage(workspace, "a"),
|
|
stage(workspace, "b", "beta", deps=["a"]),
|
|
stage(workspace, "c", "beta", deps=["b"]),
|
|
stage(workspace, "independent", "beta"),
|
|
]
|
|
first = run(workspace, plan)
|
|
(repos["alpha"] / "source.txt").write_text("alpha changed\n")
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert {
|
|
identity: count(workspace, identity)
|
|
for identity in ("a", "b", "c", "independent")
|
|
} == {"a": 2, "b": 2, "c": 2, "independent": 1}
|
|
|
|
|
|
def test_order_only_predecessor_change_does_not_invalidate_independent_stage(workspace):
|
|
args, repos, _ = workspace
|
|
plan = [stage(workspace, "a"), stage(workspace, "b", "beta", after=["a"])]
|
|
first = run(workspace, plan)
|
|
(repos["alpha"] / "source.txt").write_text("alpha changed\n")
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 2 and count(workspace, "b") == 1
|
|
assert stages_by_id(second)["b"]["reused_from"] == first["run_id"]
|
|
|
|
|
|
def test_order_only_failure_prevents_new_downstream_execution(workspace):
|
|
plan = [
|
|
stage(workspace, "a", body="raise SystemExit(8)"),
|
|
stage(workspace, "b", "beta", after=["a"]),
|
|
]
|
|
result = run(workspace, plan)
|
|
assert result["status"] == "failed"
|
|
assert count(workspace, "a") == 1 and count(workspace, "b") == 0
|
|
assert stages_by_id(result)["b"]["status"] == "skipped"
|
|
|
|
|
|
def test_never_reused_stage_and_actual_consumers_rerun_but_order_only_stage_can_reuse(
|
|
workspace,
|
|
):
|
|
args, _, _ = workspace
|
|
plan = [
|
|
stage(workspace, "producer", reuse="never"),
|
|
stage(workspace, "consumer", "beta", deps=["producer"]),
|
|
stage(workspace, "independent", "beta", after=["producer"]),
|
|
]
|
|
first = run(workspace, plan)
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "producer") == 2
|
|
assert count(workspace, "consumer") == 2
|
|
assert count(workspace, "independent") == 1
|
|
|
|
|
|
def test_unspecified_input_scope_stays_conservatively_workspace_wide(workspace):
|
|
args, repos, _ = workspace
|
|
plan = [stage(workspace, "broad", "beta", inputs=False)]
|
|
first = run(workspace, plan)
|
|
(repos["alpha"] / "source.txt").write_text("alpha changed\n")
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "broad") == 2
|
|
|
|
|
|
def test_global_environment_change_invalidates_all_scoped_checkpoints(workspace):
|
|
args, _, _ = workspace
|
|
plan = [stage(workspace, "a"), stage(workspace, "b", "beta")]
|
|
first = run(workspace, plan)
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan, environment="f" * 64)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 2 and count(workspace, "b") == 2
|
|
assert second["environment_fingerprint"] != first["environment_fingerprint"]
|
|
|
|
|
|
def test_changed_declared_scope_prevents_same_id_reuse(workspace):
|
|
args, _, _ = workspace
|
|
selected = stage(workspace, "a")
|
|
first = run(workspace, [selected])
|
|
args.resume = first["run_id"]
|
|
selected = {**selected, "inputs": {"repos": ["alpha", "beta"]}}
|
|
second = run(workspace, [selected])
|
|
assert second["status"] == "passed" and count(workspace, "a") == 2
|
|
|
|
|
|
def test_input_mutation_during_passing_command_never_creates_reusable_checkpoint(
|
|
workspace,
|
|
):
|
|
args, repos, _ = workspace
|
|
plan = [
|
|
stage(
|
|
workspace,
|
|
"mutates",
|
|
body="Path('source.txt').write_text('mutated during stage\\n')",
|
|
)
|
|
]
|
|
first = run(workspace, plan)
|
|
assert first["status"] != "passed"
|
|
assert stages_by_id(first)["mutates"].get("checkpoint_verified") is not True
|
|
args.resume = first["run_id"]
|
|
# The same command is now stable against the already-mutated current bytes;
|
|
# its unverified first result still cannot be skipped.
|
|
second = run(workspace, plan)
|
|
assert count(workspace, "mutates") == 2
|
|
assert second["status"] == "passed"
|
|
assert (repos["alpha"] / "source.txt").read_text() == "mutated during stage\n"
|
|
|
|
|
|
def test_later_restoration_of_bytes_cannot_turn_invalid_phase_into_overall_pass(
|
|
workspace,
|
|
):
|
|
_, repos, _ = workspace
|
|
source = repos["alpha"] / "source.txt"
|
|
original = source.read_text()
|
|
plan = [
|
|
stage(
|
|
workspace,
|
|
"mutates",
|
|
body="Path('source.txt').write_text('temporary change\\n')",
|
|
),
|
|
stage(
|
|
workspace,
|
|
"restores",
|
|
"beta",
|
|
after=["mutates"],
|
|
body=f"Path({str(source)!r}).write_text({original!r})",
|
|
),
|
|
]
|
|
result = run(workspace, plan)
|
|
assert result["status"] != "passed"
|
|
assert stages_by_id(result)["mutates"].get("checkpoint_verified") is not True
|
|
|
|
|
|
def test_recovered_interruption_reuses_only_verified_durable_checkpoint(workspace):
|
|
args, _, _ = workspace
|
|
plan = [stage(workspace, "a"), stage(workspace, "b", "beta", after=["a"])]
|
|
first = run(workspace, plan)
|
|
receipt = read_json(Path(first["receipt_path"]))
|
|
receipt.update(
|
|
status="running", phase="checking", snapshot_verified=False, finished_at=None
|
|
)
|
|
unfinished = stages_by_id(receipt)["b"]
|
|
unfinished.update(status="running", exit_code=None, checkpoint_verified=False)
|
|
atomic_json(Path(first["receipt_path"]), runner._seal(receipt))
|
|
parser = argparse.ArgumentParser()
|
|
runner.register(parser.add_subparsers(dest="command", required=True))
|
|
recovered = parser.parse_args(
|
|
["recover", first["run_id"], "--apply", "--confirm-processes-stopped"],
|
|
namespace=Namespace(**vars(args)),
|
|
)
|
|
recovery = recovered.handler(recovered)
|
|
assert recovery["status"] == "interrupted"
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 1 and count(workspace, "b") == 2
|
|
assert stages_by_id(second)["a"]["reused_from"] == first["run_id"]
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["tampered", "missing"])
|
|
def test_cached_log_tamper_or_loss_is_not_accepted_as_verified_evidence(
|
|
workspace, change
|
|
):
|
|
args, _, _ = workspace
|
|
plan = [stage(workspace, "a")]
|
|
first = run(workspace, plan)
|
|
path = Path(first["stages"][0]["log_path"])
|
|
if change == "missing":
|
|
path.unlink()
|
|
else:
|
|
path.write_text("tampered evidence\n")
|
|
args.resume = first["run_id"]
|
|
result = run(workspace, plan)
|
|
assert result["status"] == "failed"
|
|
assert (
|
|
result["snapshot_verified"] is not True
|
|
or result["stages"][0]["status"] == "failed"
|
|
)
|
|
assert result["stages"][0].get("checkpoint_verified") is not True
|
|
assert "reused_from" not in result["stages"][0]
|
|
assert "error" in result["stages"][0]
|
|
assert count(workspace, "a") == 1
|
|
|
|
|
|
def test_stale_run_donates_only_checkpoints_matching_final_current_inputs(workspace):
|
|
args, repos, _ = workspace
|
|
source = repos["alpha"] / "source.txt"
|
|
plan = [
|
|
stage(workspace, "a"),
|
|
stage(
|
|
workspace,
|
|
"changes-alpha",
|
|
"beta",
|
|
after=["a"],
|
|
body=f"Path({str(source)!r}).write_text('new alpha bytes\\n')",
|
|
),
|
|
]
|
|
first = run(workspace, plan)
|
|
assert first["status"] != "passed"
|
|
args.resume = first["run_id"]
|
|
second = run(workspace, plan)
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 2
|
|
# This order-only phase did not consume alpha and its own scoped inputs are
|
|
# still identical; it may retain its independently verified checkpoint.
|
|
assert count(workspace, "changes-alpha") == 1
|
|
|
|
|
|
def test_legacy_checkpointless_receipt_does_not_gain_incremental_authority(workspace):
|
|
args, repos, _ = workspace
|
|
plan = [stage(workspace, "a")]
|
|
first = run(workspace, plan)
|
|
receipt = read_json(Path(first["receipt_path"]))
|
|
receipt.pop("fingerprint_version", None)
|
|
for item in receipt["stages"]:
|
|
for field in ("checkpoint_version", "checkpoint_verified", "cache_key"):
|
|
item.pop(field, None)
|
|
atomic_json(Path(first["receipt_path"]), runner._seal(receipt))
|
|
(repos["beta"] / "source.txt").write_text("unrelated changed source\n")
|
|
args.resume = first["run_id"]
|
|
try:
|
|
second = run(workspace, plan)
|
|
except ValueError:
|
|
return # Rejecting legacy incremental reuse is also safely fail-closed.
|
|
assert second["status"] == "passed"
|
|
assert count(workspace, "a") == 2
|