Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/execution-plan.md
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,9 @@ Schema 25 stores append-only revisions in `cycle_execution_plans` in the history
database. The version-1 metadata projection removes prompts, criterion prose,
reasoning, commands and output. Commands become SHA-256 bindings; criteria become
stable `AC-n` identities. Check histories retain HEAD, results, closed cause
tokens and evidence fingerprints. Canonical JSON uses sorted keys, compact
tokens and evidence fingerprints; the raw reported `attempts` kept as
`attempts_normalized` and any evidence entry that is not an object are not
persisted. Canonical JSON uses sorted keys, compact
separators and ASCII escaping; its SHA-256 is `digest`. Every stage receives
`plan_id`, `revision` and `digest` for its schema-2 receipt. A receipt supplements
the existing attempt, head and `REV-*` contracts; it grants no execution rights.
Expand Down
14 changes: 11 additions & 3 deletions scripts/cycle_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,7 +259,8 @@ def clean(value, key=""):
if key in {"id", "executor", "profile", "context"} and isinstance(value, str):
return persisted_token(value)
if key == "items":
return [f"AC-{i}" for i in range(1, len(value) + 1)]
return ([f"AC-{i}" for i in range(1, len(value) + 1)]
if isinstance(value, (list, tuple)) else [])
if key == "cause":
if value in {"defect", "blocked", "omitted", "missing_tool", "inconclusive",
"environment", "stale", "unverified", "time_limit", ""}:
Expand All @@ -278,12 +279,19 @@ def clean(value, key=""):
conclusion = value.get("verification_conclusion")
return ({"verification_conclusion": conclusion} if conclusion in {
"verified", "verified_with_reservations", "not_verified", "failed"} else {})
if key in {"attempts", "review"}:
if key in {"attempts", "attempts_normalized", "review"}:
# attempts_normalized keeps the raw reported attempts, with their
# prose and output; the projection never carries either.
return [] if key == "attempts" else {}
if key == "evidence":
# Only structured entries carry projectable fields; a prose entry
# (a reviewer may report evidence as plain strings) has none.
if not isinstance(value, (list, tuple)):
return []
return [{k: clean(v, k) for k, v in item.items()
if k in {"command", "level", "exit_code", "executed", "test_sha256",
"output_sha256", "context_sha256", "limit_seconds"}} for item in value]
"output_sha256", "context_sha256", "limit_seconds"}}
for item in value if isinstance(item, dict)]
if isinstance(value, dict):
return {k: clean(v, k) for k, v in value.items()}
if isinstance(value, (list, tuple)):
Expand Down
39 changes: 33 additions & 6 deletions tests/test_fast_path.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
from __future__ import annotations

import contextlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
Expand All @@ -15,6 +17,29 @@
import fast_path # noqa: E402


@contextlib.contextmanager
def tracked_snapshot():
"""A copy of the files Git tracks, so a scan sees what CI sees.

The privacy scan walks every file under its root. A developer's working tree
also holds untracked notes and ignored worktrees that CI never has, and they
must not decide whether the scan itself works. Without Git (an unpacked
archive) the root itself is the snapshot.
"""
listed = subprocess.run(["git", "ls-files", "-z"], cwd=ROOT, capture_output=True, check=False)
if listed.returncode != 0:
yield ROOT
return
with tempfile.TemporaryDirectory() as directory:
for name in listed.stdout.decode("utf-8").split("\0"):
source = ROOT / name
if name and source.is_file():
target = Path(directory) / name
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
yield Path(directory)


def git(directory: Path, *arguments: str) -> str:
# Identity is set per command: a CI runner has no global one.
result = subprocess.run(
Expand Down Expand Up @@ -430,13 +455,14 @@ def test_every_job_has_a_label_and_validation_lists(self) -> None:


class CommandTests(unittest.TestCase):
def run_script(self, *arguments: str, **environment: str) -> subprocess.CompletedProcess:
def run_script(self, *arguments: str, root: Path = ROOT,
**environment: str) -> subprocess.CompletedProcess:
env = {key: value for key, value in os.environ.items()
if not key.startswith(("GITHUB_", "FAST_PATH_"))}
env.update(environment)
return subprocess.run(
[sys.executable, str(ROOT / "scripts" / "fast_path.py"), *arguments],
capture_output=True, text=True, env=env, cwd=ROOT,
[sys.executable, str(root / "scripts" / "fast_path.py"), *arguments],
capture_output=True, text=True, env=env, cwd=root,
)

def test_classify_fails_closed_and_still_exits_successfully(self) -> None:
Expand All @@ -459,11 +485,11 @@ def test_run_fast_rejects_a_test_name_that_is_not_a_module(self) -> None:
self.assertNotEqual(0, result.returncode)

def test_run_fast_runs_the_scan_and_the_mapped_tests(self) -> None:
with tempfile.TemporaryDirectory() as directory:
with tempfile.TemporaryDirectory() as directory, tracked_snapshot() as snapshot:
summary = Path(directory) / "summary.md"

result = self.run_script(
"run-fast", "--job", "validate", "--scan",
"run-fast", "--job", "validate", "--scan", root=snapshot,
FAST_PATH_TESTS="test_spike_fixture_runner", GITHUB_STEP_SUMMARY=str(summary),
)

Expand Down Expand Up @@ -524,7 +550,8 @@ def test_the_scan_is_reusable_and_reports_what_the_validator_reports(self) -> No
any("possible private data in docs/guide.md" in error for error in normalized_errors)
)
self.assertTrue(any("metadata sidecar" in error for error in normalized_errors))
self.assertEqual([], module.scan_for_private_data(ROOT))
with tracked_snapshot() as snapshot:
self.assertEqual([], module.scan_for_private_data(snapshot))


if __name__ == "__main__":
Expand Down
3 changes: 2 additions & 1 deletion tests/test_validate_package.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,8 @@ def assert_error_contains(self, errors: list[str], expected: str) -> None:
)

def test_current_package_is_valid(self) -> None:
self.assertEqual([], VALIDATOR.validate_package(ROOT))
# A tracked-file copy: untracked local files are not part of the package.
self.assertEqual([], VALIDATOR.validate_package(self.copy_package()))

def test_authorization_contract_rejects_missing_or_expanded_actions(self) -> None:
package = self.copy_package()
Expand Down
23 changes: 23 additions & 0 deletions tests/test_verification_decisions.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,29 @@ def test_timeout_records_command_limit_and_evidence(self):
self.assertEqual(restored.verification.results[-1].cause, "time_limit")
self.assertEqual(restored.verification.results[-1].evidence[0]["limit_seconds"], 600)

def test_projection_ignores_prose_evidence_entries(self):
entry = {"command": self.check.command, "exit_code": 1, "executed": 1}
payload = {"verification": {"results": [
{"id": "x", "evidence": ["reviewer prose", entry, 3]},
{"id": "y", "evidence": "plain text"}]}}
projected = plan_projection(SimpleNamespace(to_dict=lambda: payload))
results = projected["plan"]["verification"]["results"]
self.assertEqual(len(results[0]["evidence"]), 1)
self.assertEqual(results[0]["evidence"][0]["exit_code"], 1)
self.assertEqual(results[1]["evidence"], [])
malformed = {"criteria": {"items": 3}, "x": {"items": ["a", "b"]}}
projected = plan_projection(SimpleNamespace(to_dict=lambda: malformed))["plan"]
self.assertEqual((projected["criteria"]["items"], projected["x"]["items"]),
([], ["AC-1", "AC-2"]))

def test_projection_drops_raw_normalized_attempts(self):
raw = {"type": "array", "value": [{"command": "pytest", "evidence": "raw output"}]}
payload = {"verification": {"results": [{"id": "x", "attempts": [{"note": "n"}],
"attempts_normalized": raw}]}}
result = plan_projection(SimpleNamespace(to_dict=lambda: payload)
)["plan"]["verification"]["results"][0]
self.assertEqual((result["attempts"], result["attempts_normalized"]), ([], {}))

def test_terminal_status_is_non_successful_and_serializable(self):
self.assertEqual(telemetry.SCHEMA_VERSION, 29)
self.assertEqual(cycle_status._result_icon("VERIFICATION_INCOMPLETE"), "⚠️")
Expand Down
Loading