From 05bdeb12cdd3327a99cbe63491a66e93a57a4281 Mon Sep 17 00:00:00 2001 From: ProfRandom92 Date: Fri, 22 May 2026 21:47:40 +0200 Subject: [PATCH] feat: add AI workflow snapshot command --- scripts/ai_workflow_snapshot.py | 99 +++++++++++++++++++++ tests/test_ai_workflow_snapshot.py | 134 +++++++++++++++++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 scripts/ai_workflow_snapshot.py create mode 100644 tests/test_ai_workflow_snapshot.py diff --git a/scripts/ai_workflow_snapshot.py b/scripts/ai_workflow_snapshot.py new file mode 100644 index 0000000..e9b63f0 --- /dev/null +++ b/scripts/ai_workflow_snapshot.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Build a deterministic local AI workflow evidence snapshot.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from scripts.agent_artifact_bundle import build_agent_artifact_bundle +from scripts.safe_pr_gate import GateState, collect_gate_state + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Build a deterministic local AI workflow evidence snapshot.") + parser.add_argument( + "--validation-command", + action="append", + default=[], + help="Validation command executed for the snapshot evidence. May be repeated.", + ) + parser.add_argument( + "--validation-result", + action="append", + default=[], + help="Validation result corresponding to each command. May be repeated.", + ) + parser.add_argument( + "--mcp-context-output-ref", + help="Optional reference to a previously generated MCP context output.", + ) + return parser.parse_args(argv) + + +def _error_response(exc: RuntimeError) -> dict[str, Any]: + return { + "error": { + "message": str(exc), + "type": exc.__class__.__name__, + }, + "ok": False, + "result": "ERROR", + } + + +def build_ai_workflow_snapshot( + state: GateState, + *, + validation_commands: list[str], + validation_results: list[str], + mcp_context_output_ref: str | None = None, +) -> dict[str, Any]: + agent_artifact_bundle = build_agent_artifact_bundle( + state, + allow_main=True, + validation_commands=validation_commands, + validation_results=validation_results, + mcp_context_output_ref=mcp_context_output_ref, + ) + snapshot: dict[str, Any] = { + "agent_artifact_bundle": agent_artifact_bundle, + "ok": agent_artifact_bundle["ok"], + "result": agent_artifact_bundle["result"], + "safe_pr_gate": agent_artifact_bundle["safe_pr_gate"], + "validation_evidence": agent_artifact_bundle["validation_evidence"], + } + if mcp_context_output_ref is not None: + snapshot["mcp_context_output_ref"] = mcp_context_output_ref + return snapshot + + +def _emit_json(payload: dict[str, Any]) -> None: + sys.stdout.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n") + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(sys.argv[1:] if argv is None else argv) + try: + snapshot = build_ai_workflow_snapshot( + collect_gate_state(), + validation_commands=list(args.validation_command), + validation_results=list(args.validation_result), + mcp_context_output_ref=args.mcp_context_output_ref, + ) + _emit_json(snapshot) + return 0 if snapshot["ok"] else 1 + except RuntimeError as exc: + _emit_json(_error_response(exc)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_ai_workflow_snapshot.py b/tests/test_ai_workflow_snapshot.py new file mode 100644 index 0000000..3dd7c5b --- /dev/null +++ b/tests/test_ai_workflow_snapshot.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import json + +import pytest + +import scripts.ai_workflow_snapshot as ai_workflow_snapshot +from scripts.safe_pr_gate import GateState + + +def test_build_ai_workflow_snapshot_is_deterministic_and_includes_requested_evidence() -> None: + state = GateState( + branch="feat/ai-workflow-snapshot", + status_short=(), + changed_paths=("scripts/ai_workflow_snapshot.py",), + ) + + snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot( + state, + validation_commands=["python -m compileall -q scripts/ai_workflow_snapshot.py"], + validation_results=["pass"], + mcp_context_output_ref="artifacts/mcp_context_layer_example.json", + ) + + assert snapshot["ok"] is True + assert snapshot["result"] == "PASS" + assert snapshot["safe_pr_gate"] == snapshot["agent_artifact_bundle"]["safe_pr_gate"] + assert snapshot["validation_evidence"] == snapshot["agent_artifact_bundle"]["validation_evidence"] + assert snapshot["mcp_context_output_ref"] == "artifacts/mcp_context_layer_example.json" + assert snapshot["agent_artifact_bundle"]["mcp_context_output_ref"] == "artifacts/mcp_context_layer_example.json" + + first = json.dumps(snapshot, separators=(",", ":"), sort_keys=True) + second = json.dumps(snapshot, separators=(",", ":"), sort_keys=True) + assert first == second + + +def test_build_ai_workflow_snapshot_reflects_safe_gate_failure_without_main_error() -> None: + state = GateState(branch="main", status_short=(), changed_paths=()) + + snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot( + state, + validation_commands=[], + validation_results=[], + ) + + assert snapshot["ok"] is False + assert snapshot["result"] == "FAIL" + assert snapshot["safe_pr_gate"]["ok"] is False + assert snapshot["safe_pr_gate"]["result"] == "FAIL" + assert snapshot["safe_pr_gate"]["problems"] == ["on_main_branch"] + + +def test_build_ai_workflow_snapshot_omits_optional_mcp_reference() -> None: + state = GateState(branch="feat/ai-workflow-snapshot", status_short=(), changed_paths=()) + + snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot( + state, + validation_commands=[], + validation_results=[], + ) + + assert "mcp_context_output_ref" not in snapshot + assert "mcp_context_output_ref" not in snapshot["agent_artifact_bundle"] + + +def test_main_emits_compact_deterministic_json( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + state = GateState( + branch="feat/ai-workflow-snapshot", + status_short=(), + changed_paths=("scripts/ai_workflow_snapshot.py",), + ) + monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state) + + exit_code = ai_workflow_snapshot.main( + [ + "--validation-command", + "python -m compileall -q scripts/ai_workflow_snapshot.py", + "--validation-result", + "pass", + ] + ) + captured = capsys.readouterr() + output = json.loads(captured.out) + + assert exit_code == 0 + assert captured.err == "" + assert captured.out == json.dumps(output, separators=(",", ":"), sort_keys=True) + "\n" + assert output["ok"] is True + assert output["result"] == "PASS" + assert "mcp_context_output_ref" not in output + + +def test_main_returns_failure_exit_code_when_safe_gate_fails( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + state = GateState( + branch="feat/ai-workflow-snapshot", + status_short=(" M docs/example.md",), + changed_paths=("docs/example.md",), + ) + monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state) + + exit_code = ai_workflow_snapshot.main([]) + output = json.loads(capsys.readouterr().out) + + assert exit_code == 1 + assert output["ok"] is False + assert output["result"] == "FAIL" + assert output["safe_pr_gate"]["problems"] == ["dirty_working_tree"] + + +def test_main_reports_validation_mismatch_as_deterministic_error_json( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + state = GateState(branch="feat/ai-workflow-snapshot", status_short=(), changed_paths=()) + monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state) + + exit_code = ai_workflow_snapshot.main(["--validation-command", "pytest tests/test_ai_workflow_snapshot.py -q"]) + output = json.loads(capsys.readouterr().out) + + assert exit_code == 1 + assert output == { + "error": { + "message": "validation command/result count mismatch: 1 command(s), 0 result(s)", + "type": "RuntimeError", + }, + "ok": False, + "result": "ERROR", + }