Skip to content

Commit 49de2ce

Browse files
authored
feat: add AI workflow snapshot command
Adds a minimal deterministic AI workflow snapshot command that combines safe_pr_gate and agent_artifact_bundle evidence with optional MCP context output references and focused tests.
1 parent 35cfc2f commit 49de2ce

2 files changed

Lines changed: 233 additions & 0 deletions

File tree

scripts/ai_workflow_snapshot.py

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
#!/usr/bin/env python3
2+
"""Build a deterministic local AI workflow evidence snapshot."""
3+
4+
from __future__ import annotations
5+
6+
import argparse
7+
import json
8+
import sys
9+
from pathlib import Path
10+
from typing import Any
11+
12+
REPO_ROOT = Path(__file__).resolve().parents[1]
13+
if str(REPO_ROOT) not in sys.path:
14+
sys.path.insert(0, str(REPO_ROOT))
15+
16+
from scripts.agent_artifact_bundle import build_agent_artifact_bundle
17+
from scripts.safe_pr_gate import GateState, collect_gate_state
18+
19+
20+
def _parse_args(argv: list[str]) -> argparse.Namespace:
21+
parser = argparse.ArgumentParser(description="Build a deterministic local AI workflow evidence snapshot.")
22+
parser.add_argument(
23+
"--validation-command",
24+
action="append",
25+
default=[],
26+
help="Validation command executed for the snapshot evidence. May be repeated.",
27+
)
28+
parser.add_argument(
29+
"--validation-result",
30+
action="append",
31+
default=[],
32+
help="Validation result corresponding to each command. May be repeated.",
33+
)
34+
parser.add_argument(
35+
"--mcp-context-output-ref",
36+
help="Optional reference to a previously generated MCP context output.",
37+
)
38+
return parser.parse_args(argv)
39+
40+
41+
def _error_response(exc: RuntimeError) -> dict[str, Any]:
42+
return {
43+
"error": {
44+
"message": str(exc),
45+
"type": exc.__class__.__name__,
46+
},
47+
"ok": False,
48+
"result": "ERROR",
49+
}
50+
51+
52+
def build_ai_workflow_snapshot(
53+
state: GateState,
54+
*,
55+
validation_commands: list[str],
56+
validation_results: list[str],
57+
mcp_context_output_ref: str | None = None,
58+
) -> dict[str, Any]:
59+
agent_artifact_bundle = build_agent_artifact_bundle(
60+
state,
61+
allow_main=True,
62+
validation_commands=validation_commands,
63+
validation_results=validation_results,
64+
mcp_context_output_ref=mcp_context_output_ref,
65+
)
66+
snapshot: dict[str, Any] = {
67+
"agent_artifact_bundle": agent_artifact_bundle,
68+
"ok": agent_artifact_bundle["ok"],
69+
"result": agent_artifact_bundle["result"],
70+
"safe_pr_gate": agent_artifact_bundle["safe_pr_gate"],
71+
"validation_evidence": agent_artifact_bundle["validation_evidence"],
72+
}
73+
if mcp_context_output_ref is not None:
74+
snapshot["mcp_context_output_ref"] = mcp_context_output_ref
75+
return snapshot
76+
77+
78+
def _emit_json(payload: dict[str, Any]) -> None:
79+
sys.stdout.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n")
80+
81+
82+
def main(argv: list[str] | None = None) -> int:
83+
args = _parse_args(sys.argv[1:] if argv is None else argv)
84+
try:
85+
snapshot = build_ai_workflow_snapshot(
86+
collect_gate_state(),
87+
validation_commands=list(args.validation_command),
88+
validation_results=list(args.validation_result),
89+
mcp_context_output_ref=args.mcp_context_output_ref,
90+
)
91+
_emit_json(snapshot)
92+
return 0 if snapshot["ok"] else 1
93+
except RuntimeError as exc:
94+
_emit_json(_error_response(exc))
95+
return 1
96+
97+
98+
if __name__ == "__main__":
99+
raise SystemExit(main())

tests/test_ai_workflow_snapshot.py

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
from __future__ import annotations
2+
3+
import json
4+
5+
import pytest
6+
7+
import scripts.ai_workflow_snapshot as ai_workflow_snapshot
8+
from scripts.safe_pr_gate import GateState
9+
10+
11+
def test_build_ai_workflow_snapshot_is_deterministic_and_includes_requested_evidence() -> None:
12+
state = GateState(
13+
branch="feat/ai-workflow-snapshot",
14+
status_short=(),
15+
changed_paths=("scripts/ai_workflow_snapshot.py",),
16+
)
17+
18+
snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot(
19+
state,
20+
validation_commands=["python -m compileall -q scripts/ai_workflow_snapshot.py"],
21+
validation_results=["pass"],
22+
mcp_context_output_ref="artifacts/mcp_context_layer_example.json",
23+
)
24+
25+
assert snapshot["ok"] is True
26+
assert snapshot["result"] == "PASS"
27+
assert snapshot["safe_pr_gate"] == snapshot["agent_artifact_bundle"]["safe_pr_gate"]
28+
assert snapshot["validation_evidence"] == snapshot["agent_artifact_bundle"]["validation_evidence"]
29+
assert snapshot["mcp_context_output_ref"] == "artifacts/mcp_context_layer_example.json"
30+
assert snapshot["agent_artifact_bundle"]["mcp_context_output_ref"] == "artifacts/mcp_context_layer_example.json"
31+
32+
first = json.dumps(snapshot, separators=(",", ":"), sort_keys=True)
33+
second = json.dumps(snapshot, separators=(",", ":"), sort_keys=True)
34+
assert first == second
35+
36+
37+
def test_build_ai_workflow_snapshot_reflects_safe_gate_failure_without_main_error() -> None:
38+
state = GateState(branch="main", status_short=(), changed_paths=())
39+
40+
snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot(
41+
state,
42+
validation_commands=[],
43+
validation_results=[],
44+
)
45+
46+
assert snapshot["ok"] is False
47+
assert snapshot["result"] == "FAIL"
48+
assert snapshot["safe_pr_gate"]["ok"] is False
49+
assert snapshot["safe_pr_gate"]["result"] == "FAIL"
50+
assert snapshot["safe_pr_gate"]["problems"] == ["on_main_branch"]
51+
52+
53+
def test_build_ai_workflow_snapshot_omits_optional_mcp_reference() -> None:
54+
state = GateState(branch="feat/ai-workflow-snapshot", status_short=(), changed_paths=())
55+
56+
snapshot = ai_workflow_snapshot.build_ai_workflow_snapshot(
57+
state,
58+
validation_commands=[],
59+
validation_results=[],
60+
)
61+
62+
assert "mcp_context_output_ref" not in snapshot
63+
assert "mcp_context_output_ref" not in snapshot["agent_artifact_bundle"]
64+
65+
66+
def test_main_emits_compact_deterministic_json(
67+
monkeypatch: pytest.MonkeyPatch,
68+
capsys: pytest.CaptureFixture[str],
69+
) -> None:
70+
state = GateState(
71+
branch="feat/ai-workflow-snapshot",
72+
status_short=(),
73+
changed_paths=("scripts/ai_workflow_snapshot.py",),
74+
)
75+
monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state)
76+
77+
exit_code = ai_workflow_snapshot.main(
78+
[
79+
"--validation-command",
80+
"python -m compileall -q scripts/ai_workflow_snapshot.py",
81+
"--validation-result",
82+
"pass",
83+
]
84+
)
85+
captured = capsys.readouterr()
86+
output = json.loads(captured.out)
87+
88+
assert exit_code == 0
89+
assert captured.err == ""
90+
assert captured.out == json.dumps(output, separators=(",", ":"), sort_keys=True) + "\n"
91+
assert output["ok"] is True
92+
assert output["result"] == "PASS"
93+
assert "mcp_context_output_ref" not in output
94+
95+
96+
def test_main_returns_failure_exit_code_when_safe_gate_fails(
97+
monkeypatch: pytest.MonkeyPatch,
98+
capsys: pytest.CaptureFixture[str],
99+
) -> None:
100+
state = GateState(
101+
branch="feat/ai-workflow-snapshot",
102+
status_short=(" M docs/example.md",),
103+
changed_paths=("docs/example.md",),
104+
)
105+
monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state)
106+
107+
exit_code = ai_workflow_snapshot.main([])
108+
output = json.loads(capsys.readouterr().out)
109+
110+
assert exit_code == 1
111+
assert output["ok"] is False
112+
assert output["result"] == "FAIL"
113+
assert output["safe_pr_gate"]["problems"] == ["dirty_working_tree"]
114+
115+
116+
def test_main_reports_validation_mismatch_as_deterministic_error_json(
117+
monkeypatch: pytest.MonkeyPatch,
118+
capsys: pytest.CaptureFixture[str],
119+
) -> None:
120+
state = GateState(branch="feat/ai-workflow-snapshot", status_short=(), changed_paths=())
121+
monkeypatch.setattr(ai_workflow_snapshot, "collect_gate_state", lambda: state)
122+
123+
exit_code = ai_workflow_snapshot.main(["--validation-command", "pytest tests/test_ai_workflow_snapshot.py -q"])
124+
output = json.loads(capsys.readouterr().out)
125+
126+
assert exit_code == 1
127+
assert output == {
128+
"error": {
129+
"message": "validation command/result count mismatch: 1 command(s), 0 result(s)",
130+
"type": "RuntimeError",
131+
},
132+
"ok": False,
133+
"result": "ERROR",
134+
}

0 commit comments

Comments
 (0)