fix: reject broken runtime launchers and preserve failed proof evidence

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e387-534d-70e3-ad53-4ea05676db8c

tests/test_metered_queue_reconciliation.py

@@ -0,0 +1,46 @@
+"""The one-row repair cannot widen or synthesize mutation authority."""
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+from copy import deepcopy
+
+import pytest
+
+spec = importlib.util.spec_from_file_location("repair", Path(__file__).resolve().parents[1] / "scripts/reconcile-metered-queue-grant.py")
+repair = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(repair)
+IMAGE = "sha256:713bddad10a41950f446100a8b370fca8ccdd0b8969cccae751e3870c9c63ccd"
+
+
+@pytest.fixture
+def records():
+    action = dict(repository_grant=deepcopy(repair.GRANT), target_repo="hfact-glas-proof", harness_profile_ref="harness.agent-dev-local@1.1.1", labels=["hfact-metered"], description="bounded task", task_template="proof")
+    definition = SimpleNamespace(id=repair.DEFINITION, enabled=False, version=1, rules_json=[dict(id="execute-hfact-glas-metered-proof", condition="True", action=action)])
+    row = SimpleNamespace(id=repair.RUN, activity_definition_id=repair.DEFINITION, state="open", attempt=0, claim_owner=None, lease_until=None, target_repo="hfact-glas-proof", harness_profile_ref="harness.agent-dev-local@1.1.1", triggering_event_id=repair.TRIGGER, source_type="rule", source_id="execute-hfact-glas-metered-proof", repository_grant=None, description="bounded task", title="proof", labels=["hfact-metered"])
+    return definition, row
+
+
+def test_copies_only_typed_existing_authority(records):
+    definition, row = records
+    assert repair.validate(definition, row, IMAGE) is definition.rules_json[0]["action"]["repository_grant"]
+    assert row.repository_grant is None
+
+
+@pytest.mark.parametrize("field,value", [("state", "claimed"), ("attempt", 1), ("claim_owner", "worker"), ("repository_grant", repair.GRANT), ("triggering_event_id", "another-trigger"), ("description", "changed task")])
+def test_changed_or_used_row_refused(records, field, value):
+    definition, row = records
+    setattr(row, field, value)
+    with pytest.raises(ValueError):
+        repair.validate(definition, row, IMAGE)
+
+
+def test_changed_definition_cannot_grant_more(records):
+    definition, row = records
+    definition.rules_json[0]["action"]["repository_grant"]["allowed_paths"] = ["**"]
+    with pytest.raises(ValueError, match="typed_authority_drift"):
+        repair.validate(definition, row, IMAGE)
+
+
+def test_stale_worker_blocks_repair(records):
+    with pytest.raises(ValueError, match="grant_aware_worker_required"):
+        repair.validate(*records, "sha256:old")