From bb151802551ca146d226828d823d57166d8326a9 Mon Sep 17 00:00:00 2001 From: UntitledPhases Date: Sun, 26 Jul 2026 22:20:00 -0400 Subject: [PATCH] Add workflow visualization export --- README.md | 26 + docs/v1-spec.md | 4 + docs/visualization.md | 77 +++ .../artifacts.jsonl | 8 + .../visualize-readiness-workflow/events.jsonl | 6 + .../handoffs.jsonl | 1 + .../ownership_claims.jsonl | 1 + .../privacy_boundaries.jsonl | 1 + .../returns.jsonl | 1 + .../review_gates.jsonl | 5 + .../visualize-readiness-workflow/tasks.jsonl | 1 + .../visualize-readiness-workflow/workflow.svg | 165 +++++ src/agentic_sdlc/cli.py | 24 +- src/agentic_sdlc/visualize.py | 614 ++++++++++++++++++ tests/test_visualize.py | 113 ++++ 15 files changed, 1045 insertions(+), 2 deletions(-) create mode 100644 docs/visualization.md create mode 100644 evidence/visualize-readiness-workflow/artifacts.jsonl create mode 100644 evidence/visualize-readiness-workflow/events.jsonl create mode 100644 evidence/visualize-readiness-workflow/handoffs.jsonl create mode 100644 evidence/visualize-readiness-workflow/ownership_claims.jsonl create mode 100644 evidence/visualize-readiness-workflow/privacy_boundaries.jsonl create mode 100644 evidence/visualize-readiness-workflow/returns.jsonl create mode 100644 evidence/visualize-readiness-workflow/review_gates.jsonl create mode 100644 evidence/visualize-readiness-workflow/tasks.jsonl create mode 100644 evidence/visualize-readiness-workflow/workflow.svg create mode 100644 src/agentic_sdlc/visualize.py create mode 100644 tests/test_visualize.py diff --git a/README.md b/README.md index 6f5b98b..ee160c2 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,8 @@ python -m agentic_sdlc validate fixtures/valid_workflow python -m agentic_sdlc validate fixtures/privacy_violation python -m agentic_sdlc status fixtures/ownership_conflict python -m agentic_sdlc review-plan fixtures/missing_review_evidence +python -m agentic_sdlc visualize fixtures/valid_workflow +python -m agentic_sdlc visualize fixtures/valid_workflow --format svg --output workflow.svg ``` Machine-readable output: @@ -57,8 +59,31 @@ Machine-readable output: ```powershell $env:PYTHONPATH = "src" python -m agentic_sdlc validate fixtures/stale_resume --json +python -m agentic_sdlc visualize fixtures/missing_review_evidence --json ``` +## Workflow Visualization + +`visualize` turns the same validated JSONL records into a deterministic workflow +graph. The default output is Mermaid, so the result can be pasted into Markdown +or rendered by tools that support Mermaid diagrams. + +```powershell +$env:PYTHONPATH = "src" +python -m agentic_sdlc visualize fixtures/valid_workflow +python -m agentic_sdlc visualize fixtures/valid_workflow --format svg --output workflow.svg +python -m agentic_sdlc visualize fixtures/missing_review_evidence --json +``` + +The graph is not a runtime trace. It is a read-only view of the workflow state: +tasks, handoffs, events, resume packets, review gates, evidence, ownership, and +validation findings. Invalid workspaces still render, but the command exits with +`1` and labels failing records with stable `ASDLC###` rule IDs. + +SVG output is the minimal visual view: tasks are boxes, events are circles, +review gates are diamonds, evidence artifacts are ellipses, and failures use red +outlines with compact rule labels. + Run tests: ```powershell @@ -99,4 +124,5 @@ Adapters can come later. First the workflow records need to be trustworthy. - [V1 spec](docs/v1-spec.md) - [Rule catalog](docs/rule-catalog.md) +- [Visualization](docs/visualization.md) - [JSON Schemas](schemas/README.md) diff --git a/docs/v1-spec.md b/docs/v1-spec.md index 93738cc..c90e5f9 100644 --- a/docs/v1-spec.md +++ b/docs/v1-spec.md @@ -51,6 +51,7 @@ Published schema files live in `schemas/`. agentic-sdlc validate [--json] agentic-sdlc status [--json] agentic-sdlc review-plan [--json] +agentic-sdlc visualize [--json] [--format mermaid|json|svg] [--output PATH] ``` All commands are offline and read-only. @@ -79,6 +80,8 @@ V1 is acceptable when: - `validate` fails every invalid fixture with stable rule IDs. - `status` summarizes task counts, review gates, handoff readiness, stale resumes, and ownership conflicts. - `review-plan` explains what evidence is missing for review gates. +- `visualize` renders the workflow as deterministic JSON, Mermaid, or SVG, + including validation state and stable `ASDLC###` findings. - output is available as human text and JSON. - test coverage includes positive and negative fixtures for every advertised invariant. - no command runs agents, writes to GitHub, starts watchers, promotes memory, or scans broad folders. @@ -90,6 +93,7 @@ V1 is acceptable when: - MCP integration - scheduled watchers - dashboard UI +- live runtime progress visualization - vector memory - automatic durable memory promotion - cross-device bus automation diff --git a/docs/visualization.md b/docs/visualization.md new file mode 100644 index 0000000..6f4eac5 --- /dev/null +++ b/docs/visualization.md @@ -0,0 +1,77 @@ +# Visualization + +`agentic-sdlc visualize` is a read-only projection of workflow records into a +graph. It does not run agents or infer hidden state. It renders what the bounded +workflow directory already proves. + +## Command + +```text +agentic-sdlc visualize [--json] [--format mermaid|json|svg] [--output PATH] +``` + +Defaults: + +- view: `readiness` +- format: `mermaid` +- output: stdout + +Exit codes match `validate`: + +- `0`: no error findings +- `1`: validation error findings or unreadable input + +Invalid workspaces still render so the graph can show where the workflow is +blocked. + +## Readiness View + +The readiness graph shows the evidence chain that determines whether work can +continue: + +- task dependencies +- handoffs into tasks +- task events +- return/resume packets +- review gates +- evidence artifacts and events +- ownership claims +- privacy boundaries +- validation findings + +Mermaid output is intended for human inspection: + +```powershell +agentic-sdlc visualize fixtures/valid_workflow +``` + +JSON output is the canonical machine-readable graph: + +```powershell +agentic-sdlc visualize fixtures/missing_review_evidence --json +``` + +The JSON graph includes nodes, edges, counts, validation status, and redacted +finding messages. It does not include raw record payloads or loader-private +fields. + +SVG output is the minimal visual form: + +```powershell +agentic-sdlc visualize fixtures/valid_workflow --format svg --output workflow.svg +``` + +The SVG uses shape language instead of verbose labels: + +- task: square +- event: circle +- review gate: diamond +- artifact: ellipse +- missing evidence: dashed octagon +- invalid record: red outline with compact `ASDLC###` label + +## Boundary + +Visualization is not a dashboard, watcher, scheduler, runtime trace, or agent +orchestrator. Live agent progress such as `builder working 4m 16s` belongs to a +future runtime adapter after the offline evidence model is stable. diff --git a/evidence/visualize-readiness-workflow/artifacts.jsonl b/evidence/visualize-readiness-workflow/artifacts.jsonl new file mode 100644 index 0000000..7f402c5 --- /dev/null +++ b/evidence/visualize-readiness-workflow/artifacts.jsonl @@ -0,0 +1,8 @@ +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_research","created_at":"2026-05-20T18:03:30Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"analysis","status":"passed","required":true,"summary":"Inspected CLI, validator, fixtures, schemas, tests, README, V1 spec, and external Mermaid syntax before implementation."} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_code","created_at":"2026-05-20T18:04:30Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"implementation","status":"passed","required":true,"subject_paths":["src/agentic_sdlc/visualize.py","src/agentic_sdlc/cli.py","tests/test_visualize.py","README.md","docs/v1-spec.md","docs/visualization.md","evidence/visualize-readiness-workflow/**"],"summary":"Added normalized graph model, Mermaid renderer, JSON renderer, visualize CLI, docs, tests, and real workflow evidence records."} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_tests","created_at":"2026-05-20T18:05:30Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"test_result","status":"passed","required":true,"command":"python -m unittest discover -s tests -v","observed_result":"unit tests passed"} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_valid_demo","created_at":"2026-05-20T18:06:30Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"cli_output","status":"passed","required":true,"command":"agentic-sdlc visualize fixtures/valid_workflow --json","observed_result":"valid workflow rendered with PASS state and evidence-chain edges"} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_invalid_demo","created_at":"2026-05-20T18:07:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"cli_output","status":"passed","required":true,"command":"agentic-sdlc visualize fixtures/missing_review_evidence --json","observed_result":"invalid workflow rendered with ASDLC010 finding and nonzero validation exit"} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_svg_demo","created_at":"2026-05-20T18:07:15Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"visual_output","status":"passed","required":true,"command":"agentic-sdlc visualize evidence/visualize-readiness-workflow --format svg --output evidence/visualize-readiness-workflow/workflow.svg","observed_result":"actual SVG diagram rendered with boxes, circles, diamonds, ellipses, arrows, and minimal labels"} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_self_validation","created_at":"2026-05-20T18:07:30Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"validation_result","status":"passed","required":true,"command":"agentic-sdlc validate evidence/visualize-readiness-workflow","observed_result":"real workflow evidence validated successfully"} +{"schema_version":"1.0","artifact_type":"artifact","id":"artifact_visualize_privacy_scan","created_at":"2026-05-20T18:08:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","artifact_kind":"privacy_check","status":"passed","required":true,"command":"agentic-sdlc validate evidence/visualize-readiness-workflow","observed_result":"validator privacy checks passed for the persisted workflow evidence"} diff --git a/evidence/visualize-readiness-workflow/events.jsonl b/evidence/visualize-readiness-workflow/events.jsonl new file mode 100644 index 0000000..480914a --- /dev/null +++ b/evidence/visualize-readiness-workflow/events.jsonl @@ -0,0 +1,6 @@ +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_research","created_at":"2026-05-20T18:01:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"research_and_breadth_review","task_id":"task_visualize_readiness","sequence":1} +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_system_analysis","created_at":"2026-05-20T18:02:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"repo_system_analysis","task_id":"task_visualize_readiness","sequence":2} +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_reverse_pass","created_at":"2026-05-20T18:03:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"plan_reverse_pass_resolved","task_id":"task_visualize_readiness","sequence":3} +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_implementation","created_at":"2026-05-20T18:04:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"implementation_completed","task_id":"task_visualize_readiness","sequence":4} +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_tests_passed","created_at":"2026-05-20T18:05:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"tests_passed","task_id":"task_visualize_readiness","sequence":5} +{"schema_version":"1.0","artifact_type":"event","id":"event_visualize_self_validation","created_at":"2026-05-20T18:06:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","event_type":"self_validation_passed","task_id":"task_visualize_readiness","sequence":6} diff --git a/evidence/visualize-readiness-workflow/handoffs.jsonl b/evidence/visualize-readiness-workflow/handoffs.jsonl new file mode 100644 index 0000000..11f5ebb --- /dev/null +++ b/evidence/visualize-readiness-workflow/handoffs.jsonl @@ -0,0 +1 @@ +{"schema_version":"1.0","artifact_type":"handoff","id":"handoff_visualize_readiness","created_at":"2026-05-20T18:00:30Z","provenance":{"source":"actual_repo_change","actor":"orchestrator"},"privacy_classification":"internal","task_id":"task_visualize_readiness","objective":"Add a read-only visualization command that renders workflow records and validation state as deterministic JSON or Mermaid.","allowed_paths":["src/agentic_sdlc/visualize.py","src/agentic_sdlc/cli.py","tests/test_visualize.py","README.md","docs/v1-spec.md","docs/visualization.md","evidence/visualize-readiness-workflow/**"],"return_contract":"Return changed files, commands run, validation outputs, and remaining risks without raw transcripts or secrets.","evidence_required":["repo_inspection","implementation_summary","unit_tests","cli_visualization_demo","self_validation","privacy_scan"]} diff --git a/evidence/visualize-readiness-workflow/ownership_claims.jsonl b/evidence/visualize-readiness-workflow/ownership_claims.jsonl new file mode 100644 index 0000000..e3151b9 --- /dev/null +++ b/evidence/visualize-readiness-workflow/ownership_claims.jsonl @@ -0,0 +1 @@ +{"schema_version":"1.0","artifact_type":"ownership_claim","id":"ownership_visualize_main","created_at":"2026-05-20T18:00:45Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","task_id":"task_visualize_readiness","owner":"codex-main-thread","paths":["src/agentic_sdlc/visualize.py","src/agentic_sdlc/cli.py","tests/test_visualize.py","README.md","docs/v1-spec.md","docs/visualization.md","evidence/visualize-readiness-workflow/**"],"coordination_id":"visualize-readiness-workflow"} diff --git a/evidence/visualize-readiness-workflow/privacy_boundaries.jsonl b/evidence/visualize-readiness-workflow/privacy_boundaries.jsonl new file mode 100644 index 0000000..164daf7 --- /dev/null +++ b/evidence/visualize-readiness-workflow/privacy_boundaries.jsonl @@ -0,0 +1 @@ +{"schema_version":"1.0","artifact_type":"privacy_boundary","id":"privacy_visualize_records","created_at":"2026-05-20T18:01:15Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","task_id":"task_visualize_readiness","scope":"repo-local workflow evidence","policy":"Persist concise artifact summaries and command names only; do not persist secrets, raw transcripts, private absolute paths, or bulky logs."} diff --git a/evidence/visualize-readiness-workflow/returns.jsonl b/evidence/visualize-readiness-workflow/returns.jsonl new file mode 100644 index 0000000..9c38dd3 --- /dev/null +++ b/evidence/visualize-readiness-workflow/returns.jsonl @@ -0,0 +1 @@ +{"schema_version":"1.0","artifact_type":"return_resume","id":"resume_visualize_readiness_done","created_at":"2026-05-20T18:12:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","task_id":"task_visualize_readiness","handoff_id":"handoff_visualize_readiness","last_seen_event_id":"event_visualize_self_validation","next_safe_action":"Use the visualize command to inspect real workflow records, then harden SOP-specific adherence rules in a separate slice.","commands_run":["python -m unittest discover -s tests -v","agentic-sdlc visualize fixtures/valid_workflow --json","agentic-sdlc visualize fixtures/missing_review_evidence --json","agentic-sdlc visualize evidence/visualize-readiness-workflow --format svg --output evidence/visualize-readiness-workflow/workflow.svg","agentic-sdlc validate evidence/visualize-readiness-workflow"],"checks":[{"name":"unit_tests","status":"passed","evidence_id":"artifact_visualize_tests"},{"name":"valid_visualization_demo","status":"passed","evidence_id":"artifact_visualize_valid_demo"},{"name":"invalid_visualization_demo","status":"passed","evidence_id":"artifact_visualize_invalid_demo"},{"name":"svg_visualization_demo","status":"passed","evidence_id":"artifact_visualize_svg_demo"},{"name":"self_validation","status":"passed","evidence_id":"artifact_visualize_self_validation"}],"touched_paths":["src/agentic_sdlc/visualize.py","src/agentic_sdlc/cli.py","tests/test_visualize.py","README.md","docs/v1-spec.md","docs/visualization.md","evidence/visualize-readiness-workflow/**"],"open_decisions":["Define a dedicated SOP adherence rule only after the evidence model stabilizes."],"blocked_actions":[],"requires_human_approval":[]} diff --git a/evidence/visualize-readiness-workflow/review_gates.jsonl b/evidence/visualize-readiness-workflow/review_gates.jsonl new file mode 100644 index 0000000..4749b55 --- /dev/null +++ b/evidence/visualize-readiness-workflow/review_gates.jsonl @@ -0,0 +1,5 @@ +{"schema_version":"1.0","artifact_type":"review_gate","id":"gate_visualize_research","created_at":"2026-05-20T18:09:00Z","provenance":{"source":"actual_repo_change","actor":"reviewer"},"privacy_classification":"internal","task_id":"task_visualize_readiness","gate_type":"research_system_analysis","status":"passed","required_evidence":["analysis"],"evidence":["artifact_visualize_research"]} +{"schema_version":"1.0","artifact_type":"review_gate","id":"gate_visualize_implementation","created_at":"2026-05-20T18:09:30Z","provenance":{"source":"actual_repo_change","actor":"reviewer"},"privacy_classification":"internal","task_id":"task_visualize_readiness","gate_type":"implementation_review","status":"passed","required_evidence":["implementation"],"evidence":["artifact_visualize_code"]} +{"schema_version":"1.0","artifact_type":"review_gate","id":"gate_visualize_tests","created_at":"2026-05-20T18:10:00Z","provenance":{"source":"actual_repo_change","actor":"reviewer"},"privacy_classification":"internal","task_id":"task_visualize_readiness","gate_type":"tests","status":"passed","required_evidence":["test_result"],"evidence":["artifact_visualize_tests"]} +{"schema_version":"1.0","artifact_type":"review_gate","id":"gate_visualize_demo","created_at":"2026-05-20T18:10:30Z","provenance":{"source":"actual_repo_change","actor":"reviewer"},"privacy_classification":"internal","task_id":"task_visualize_readiness","gate_type":"demo_and_self_validation","status":"passed","required_evidence":["cli_output","visual_output","validation_result"],"evidence":["artifact_visualize_valid_demo","artifact_visualize_invalid_demo","artifact_visualize_svg_demo","artifact_visualize_self_validation"]} +{"schema_version":"1.0","artifact_type":"review_gate","id":"gate_visualize_privacy","created_at":"2026-05-20T18:11:00Z","provenance":{"source":"actual_repo_change","actor":"reviewer"},"privacy_classification":"internal","task_id":"task_visualize_readiness","gate_type":"privacy","status":"passed","required_evidence":["privacy_check"],"evidence":["artifact_visualize_privacy_scan"]} diff --git a/evidence/visualize-readiness-workflow/tasks.jsonl b/evidence/visualize-readiness-workflow/tasks.jsonl new file mode 100644 index 0000000..2ed38d7 --- /dev/null +++ b/evidence/visualize-readiness-workflow/tasks.jsonl @@ -0,0 +1 @@ +{"schema_version":"1.0","artifact_type":"task","id":"task_visualize_readiness","created_at":"2026-05-20T18:00:00Z","provenance":{"source":"actual_repo_change","actor":"codex-main-thread"},"privacy_classification":"internal","title":"Add deterministic workflow visualization","status":"done","depends_on":[],"review_gate_ids":["gate_visualize_research","gate_visualize_implementation","gate_visualize_tests","gate_visualize_demo","gate_visualize_privacy"],"current_resume_id":"resume_visualize_readiness_done"} diff --git a/evidence/visualize-readiness-workflow/workflow.svg b/evidence/visualize-readiness-workflow/workflow.svg new file mode 100644 index 0000000..8f2689e --- /dev/null +++ b/evidence/visualize-readiness-workflow/workflow.svg @@ -0,0 +1,165 @@ + + Agentic SDLC readiness graph: PASS + + + + + + + PASS + evidence\visualize-readiness-workflow + + + + + + + + + + + + + + + + + + + + + + + + + + + + task | task_visualize_readiness | Add deterministic workflow visualization + + T + + + handoff | handoff_visualize_readiness | Add a read-only visualization command that renders workflow records and valid... + + H + + + event | event_visualize_implementation | #4 implementation_completed + + E4 + + + event | event_visualize_research | #1 research_and_breadth_review + + E1 + + + event | event_visualize_reverse_pass | #3 plan_reverse_pass_resolved + + E3 + + + event | event_visualize_self_validation | #6 self_validation_passed + + E6 + + + event | event_visualize_system_analysis | #2 repo_system_analysis + + E2 + + + event | event_visualize_tests_passed | #5 tests_passed + + E5 + + + return_resume | resume_visualize_readiness_done | Use the visualize command to inspect real workflow records, then harden SOP-s... + + R + + + review_gate | gate_visualize_demo | demo_and_self_validation passed + + G + + + review_gate | gate_visualize_implementation | implementation_review passed + + G + + + review_gate | gate_visualize_privacy | privacy passed + + G + + + review_gate | gate_visualize_research | research_system_analysis passed + + G + + + review_gate | gate_visualize_tests | tests passed + + G + + + artifact | artifact_visualize_code | implementation passed + + A + + + artifact | artifact_visualize_invalid_demo | cli_output passed + + A + + + artifact | artifact_visualize_privacy_scan | privacy_check passed + + A + + + artifact | artifact_visualize_research | analysis passed + + A + + + artifact | artifact_visualize_self_validation | validation_result passed + + A + + + artifact | artifact_visualize_svg_demo | visual_output passed + + A + + + artifact | artifact_visualize_tests | test_result passed + + A + + + artifact | artifact_visualize_valid_demo | cli_output passed + + A + + + ownership_claim | ownership_visualize_main | codex-main-thread + + O + + + privacy_boundary | privacy_visualize_records | repo-local workflow evidence + + P + + + valid + + error + + event + + gate + diff --git a/src/agentic_sdlc/cli.py b/src/agentic_sdlc/cli.py index 67c4731..c2335af 100644 --- a/src/agentic_sdlc/cli.py +++ b/src/agentic_sdlc/cli.py @@ -6,6 +6,7 @@ from typing import Sequence from .validator import ValidationReport, validate_workspace +from .visualize import build_workflow_graph, render_json, render_mermaid, render_svg def _print_human(report: ValidationReport) -> None: @@ -53,6 +54,12 @@ def build_parser() -> argparse.ArgumentParser: cmd = sub.add_parser(command) cmd.add_argument("workspace", type=Path) cmd.add_argument("--json", action="store_true", dest="as_json") + visualize = sub.add_parser("visualize") + visualize.add_argument("workspace", type=Path) + visualize.add_argument("--json", action="store_true", dest="as_json") + visualize.add_argument("--format", choices=("mermaid", "json", "svg"), default="mermaid") + visualize.add_argument("--view", choices=("readiness",), default="readiness") + visualize.add_argument("--output", type=Path) return parser @@ -60,7 +67,21 @@ def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) report = validate_workspace(args.workspace) - if args.command == "review-plan": + if args.command == "visualize": + graph = build_workflow_graph(report, view=args.view) + output_format = "json" if args.as_json else args.format + if output_format == "json": + output = render_json(graph) + elif output_format == "svg": + output = render_svg(graph) + else: + output = render_mermaid(graph) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(output, encoding="utf-8") + else: + print(output, end="") + elif args.command == "review-plan": _print_review_plan(report, args.as_json) elif args.as_json: print(json.dumps(report.to_dict(), indent=2)) @@ -68,4 +89,3 @@ def main(argv: Sequence[str] | None = None) -> int: _print_human(report) return 0 if report.valid else 1 - diff --git a/src/agentic_sdlc/visualize.py b/src/agentic_sdlc/visualize.py new file mode 100644 index 0000000..ad46031 --- /dev/null +++ b/src/agentic_sdlc/visualize.py @@ -0,0 +1,614 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass +from html import escape +import json +import re +from typing import Any, Iterable + +from .validator import Finding, ValidationReport + + +NODE_TYPE_ORDER = { + "task": 0, + "handoff": 1, + "event": 2, + "return_resume": 3, + "review_gate": 4, + "artifact": 5, + "ownership_claim": 6, + "privacy_boundary": 7, + "memory_candidate": 8, + "missing": 9, +} + +SAFE_METADATA_FIELDS = { + "privacy_classification", + "status", + "artifact_kind", + "gate_type", + "event_type", + "sequence", + "task_id", + "depends_on", + "review_gate_ids", + "current_resume_id", + "handoff_id", + "last_seen_event_id", + "required_evidence", + "evidence", + "missing_evidence", + "required", + "command", + "owner", + "allowed_paths", + "evidence_required", + "return_contract", + "touched_paths", + "commands_run", + "checks", + "subject_paths", + "summary", + "observed_result", +} + +SECRET_PATTERNS = [ + re.compile(r"sk-[A-Za-z0-9_-]{8,}"), + re.compile(r"ghp_[A-Za-z0-9_]{8,}"), + re.compile(r"AKIA[0-9A-Z]{8,}"), + re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), +] + +PRIVATE_PATH_PATTERNS = [ + re.compile(r"C:\\Users\\[^\\]+\\(?:Desktop|Documents|Downloads|AppData|\.ssh)", re.IGNORECASE), + re.compile(r"C:/Users/[^/]+/(?:Desktop|Documents|Downloads|AppData|\.ssh)", re.IGNORECASE), + re.compile(r"/home/[^/]+/\.ssh", re.IGNORECASE), +] + + +@dataclass(frozen=True) +class GraphNode: + id: str + record_id: str | None + type: str + label: str + status: str | None + valid: bool + findings: list[str] + metadata: dict[str, Any] + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class GraphEdge: + source: str + target: str + kind: str + label: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class WorkflowGraph: + root: str + valid: bool + view: str + counts: dict[str, int] + nodes: list[GraphNode] + edges: list[GraphEdge] + findings: list[dict[str, Any]] + + def to_dict(self) -> dict[str, Any]: + return { + "root": self.root, + "valid": self.valid, + "view": self.view, + "counts": dict(sorted(self.counts.items())), + "nodes": [node.to_dict() for node in self.nodes], + "edges": [edge.to_dict() for edge in self.edges], + "findings": self.findings, + } + + +def build_workflow_graph(report: ValidationReport, view: str = "readiness") -> WorkflowGraph: + if view != "readiness": + raise ValueError("only the readiness view is implemented") + + findings_by_record = _findings_by_record(report.findings) + review_plan = {gate["id"]: gate for gate in report.review_plan()} + nodes: list[GraphNode] = [] + edges: list[GraphEdge] = [] + edge_keys: set[tuple[str, str, str]] = set() + record_node_ids: dict[str, str] = {} + missing_node_ids: dict[tuple[str, str], str] = {} + used_node_ids: set[str] = set() + + for record_type in sorted(report.records, key=lambda item: NODE_TYPE_ORDER.get(item, 99)): + for index, record in enumerate(sorted(report.records[record_type], key=_record_sort_key)): + raw_id = str(record.get("id") or f"anonymous_{record_type}_{index + 1}") + node_id = _unique_node_id(raw_id, used_node_ids) + used_node_ids.add(node_id) + record_node_ids.setdefault(raw_id, node_id) + record_findings = findings_by_record.get(raw_id, []) + metadata = _safe_metadata(record) + if record_type == "review_gate": + metadata.update(_safe_metadata(review_plan.get(raw_id, {}))) + nodes.append( + GraphNode( + id=node_id, + record_id=raw_id, + type=record_type, + label=_record_label(record_type, record, record_findings), + status=_record_status(record), + valid=not any(finding.severity == "error" for finding in record_findings), + findings=sorted({finding.rule_id for finding in record_findings}), + metadata=metadata, + ) + ) + + def ensure_missing(reference_id: str, expected_type: str) -> str: + key = (expected_type, reference_id) + existing = missing_node_ids.get(key) + if existing: + return existing + node_id = _unique_node_id(f"missing_{expected_type}_{reference_id}", used_node_ids) + used_node_ids.add(node_id) + missing_node_ids[key] = node_id + nodes.append( + GraphNode( + id=node_id, + record_id=reference_id, + type="missing", + label=f"missing {expected_type}: {_compact(reference_id, 60)}", + status="missing", + valid=False, + findings=[], + metadata={"expected_type": expected_type}, + ) + ) + return node_id + + def resolve(reference_id: Any, expected_type: str) -> str | None: + if reference_id is None: + return None + raw = str(reference_id) + return record_node_ids.get(raw) or ensure_missing(raw, expected_type) + + def add_edge(source_ref: Any, target_ref: Any, kind: str, source_type: str, target_type: str) -> None: + source = resolve(source_ref, source_type) + target = resolve(target_ref, target_type) + if not source or not target: + return + key = (source, target, kind) + if key in edge_keys: + return + edge_keys.add(key) + edges.append(GraphEdge(source=source, target=target, kind=kind, label=kind.replace("_", " "))) + + for task in report.records.get("task", []): + task_id = task.get("id") + for dep_id in _as_list(task.get("depends_on")): + add_edge(dep_id, task_id, "depends_on", "task", "task") + for gate_id in _as_list(task.get("review_gate_ids")): + add_edge(task_id, gate_id, "review_gate", "task", "review_gate") + if task.get("current_resume_id"): + add_edge(task_id, task.get("current_resume_id"), "current_resume", "task", "return_resume") + + for event in report.records.get("event", []): + if event.get("task_id"): + add_edge(event.get("task_id"), event.get("id"), "task_event", "task", "event") + + for artifact in report.records.get("artifact", []): + if artifact.get("task_id"): + add_edge(artifact.get("task_id"), artifact.get("id"), "task_artifact", "task", "artifact") + if artifact.get("produced_by_event_id"): + add_edge(artifact.get("produced_by_event_id"), artifact.get("id"), "produced", "event", "artifact") + + for handoff in report.records.get("handoff", []): + if handoff.get("task_id"): + add_edge(handoff.get("id"), handoff.get("task_id"), "handoff_for", "handoff", "task") + + for resume in report.records.get("return_resume", []): + if resume.get("task_id"): + add_edge(resume.get("task_id"), resume.get("id"), "return_resume", "task", "return_resume") + if resume.get("handoff_id"): + add_edge(resume.get("handoff_id"), resume.get("id"), "handoff_return", "handoff", "return_resume") + if resume.get("last_seen_event_id"): + add_edge(resume.get("last_seen_event_id"), resume.get("id"), "last_seen", "event", "return_resume") + + for gate in report.records.get("review_gate", []): + if gate.get("task_id"): + add_edge(gate.get("task_id"), gate.get("id"), "review_gate", "task", "review_gate") + for evidence_id in _as_list(gate.get("evidence")): + add_edge(evidence_id, gate.get("id"), "evidence_for", "artifact", "review_gate") + plan = review_plan.get(gate.get("id"), {}) + for missing in _as_list(plan.get("missing_evidence")): + add_edge(missing, gate.get("id"), "missing_evidence", "evidence", "review_gate") + + for claim in report.records.get("ownership_claim", []): + if claim.get("task_id"): + add_edge(claim.get("id"), claim.get("task_id"), "ownership_for", "ownership_claim", "task") + + for boundary in report.records.get("privacy_boundary", []): + if boundary.get("task_id"): + add_edge(boundary.get("id"), boundary.get("task_id"), "privacy_for", "privacy_boundary", "task") + + for candidate in report.records.get("memory_candidate", []): + if candidate.get("task_id"): + add_edge(candidate.get("task_id"), candidate.get("id"), "memory_candidate", "task", "memory_candidate") + + return WorkflowGraph( + root=report.root, + valid=report.valid, + view=view, + counts=report.counts, + nodes=sorted(nodes, key=_node_sort_key), + edges=sorted(edges, key=lambda edge: (edge.source, edge.kind, edge.target)), + findings=[_finding_to_dict(finding) for finding in report.findings], + ) + + +def graph_to_dict(graph: WorkflowGraph) -> dict[str, Any]: + return graph.to_dict() + + +def render_json(graph: WorkflowGraph) -> str: + return json.dumps(graph.to_dict(), indent=2, sort_keys=True) + + +def render_mermaid(graph: WorkflowGraph) -> str: + lines = [ + "flowchart LR", + f" %% root: {_mermaid_comment(graph.root)}", + f" %% status: {'PASS' if graph.valid else 'FAIL'}", + ] + symbol_by_node = {node.id: f"n{index}" for index, node in enumerate(graph.nodes)} + for node in graph.nodes: + symbol = symbol_by_node[node.id] + label = _mermaid_node_label(node) + lines.append(f" {symbol}[{json.dumps(label)}]") + for edge in graph.edges: + source = symbol_by_node.get(edge.source) + target = symbol_by_node.get(edge.target) + if not source or not target: + continue + lines.append(f" {source} -->|{_mermaid_edge_label(edge.label)}| {target}") + + if graph.findings: + lines.append(" subgraph validation_findings[Validation findings]") + for index, finding in enumerate(graph.findings): + symbol = f"f{index}" + label = f"{finding['rule_id']} {finding['severity']}" + if finding.get("record_id"): + label += f"\\n{finding['record_id']}" + label += f"\\n{_compact(str(finding.get('message', '')), 80)}" + lines.append(f" {symbol}[{json.dumps(label)}]") + lines.append(f" class {symbol} error") + lines.append(" end") + + lines.extend( + [ + " classDef pass fill:#e8f5e9,stroke:#2e7d32,color:#111111", + " classDef error fill:#ffebee,stroke:#c62828,color:#111111", + " classDef missing fill:#fff3e0,stroke:#ef6c00,color:#111111,stroke-dasharray: 5 5", + ] + ) + for node in graph.nodes: + class_name = "missing" if node.type == "missing" else "pass" if node.valid else "error" + lines.append(f" class {symbol_by_node[node.id]} {class_name}") + return "\n".join(lines) + "\n" + + +def render_svg(graph: WorkflowGraph) -> str: + positions = _svg_positions(graph.nodes) + width = max((position[0] for position in positions.values()), default=160) + 160 + height = max((position[1] for position in positions.values()), default=160) + 160 + lines = [ + f'', + f" {escape('Agentic SDLC readiness graph: ' + ('PASS' if graph.valid else 'FAIL'))}", + " ", + ' ', + ' ', + " ", + " ", + ' ', + f' {escape("PASS" if graph.valid else "FAIL")}', + f' {escape(_compact(graph.root, 80))}', + ] + if graph.findings: + rules = ", ".join(sorted({finding["rule_id"] for finding in graph.findings})) + lines.append( + f' {escape(rules)}' + ) + + for edge in graph.edges: + start = positions.get(edge.source) + end = positions.get(edge.target) + if not start or not end: + continue + x1, y1 = start + x2, y2 = end + mid_x = (x1 + x2) / 2 + lines.append( + ' ' + ) + + for index, node in enumerate(graph.nodes): + x, y = positions[node.id] + lines.extend(_svg_node(node, index, x, y)) + + legend_y = height - 44 + lines.extend( + [ + f' ', + f' valid', + f' ', + f' error', + f' ', + f' event', + f' ', + f' gate', + ] + ) + lines.append("") + return "\n".join(lines) + "\n" + + +def _findings_by_record(findings: Iterable[Finding]) -> dict[str, list[Finding]]: + grouped: dict[str, list[Finding]] = {} + for finding in findings: + if finding.record_id: + grouped.setdefault(finding.record_id, []).append(finding) + return grouped + + +def _record_sort_key(record: dict[str, Any]) -> tuple[str, str]: + return (str(record.get("id", "")), str(record.get("_source_file", ""))) + + +def _node_sort_key(node: GraphNode) -> tuple[int, str]: + return (NODE_TYPE_ORDER.get(node.type, 99), node.id) + + +def _unique_node_id(raw_id: str, used: set[str]) -> str: + base = _safe_graph_id(raw_id) + candidate = base + index = 2 + while candidate in used: + candidate = f"{base}_{index}" + index += 1 + return candidate + + +def _safe_graph_id(raw_id: str) -> str: + value = re.sub(r"[^A-Za-z0-9_]+", "_", raw_id).strip("_") + return value or "node" + + +def _record_status(record: dict[str, Any]) -> str | None: + for field in ("status", "event_type", "artifact_kind", "gate_type"): + value = record.get(field) + if value: + return str(value) + return None + + +def _record_label(record_type: str, record: dict[str, Any], findings: list[Finding]) -> str: + record_id = str(record.get("id") or "anonymous") + detail = "" + if record_type == "task": + detail = str(record.get("title") or record.get("status") or "") + elif record_type == "event": + sequence = record.get("sequence") + event_type = record.get("event_type", "") + detail = f"#{sequence} {event_type}" if sequence is not None else str(event_type) + elif record_type == "artifact": + detail = " ".join(str(record.get(field, "")) for field in ("artifact_kind", "status")).strip() + elif record_type == "review_gate": + detail = " ".join(str(record.get(field, "")) for field in ("gate_type", "status")).strip() + elif record_type == "handoff": + detail = str(record.get("objective") or "handoff") + elif record_type == "return_resume": + detail = str(record.get("next_safe_action") or "return/resume") + elif record_type == "ownership_claim": + detail = str(record.get("owner") or "ownership") + elif record_type == "memory_candidate": + detail = str(record.get("summary") or record.get("title") or "memory candidate") + elif record_type == "privacy_boundary": + detail = str(record.get("scope") or "privacy boundary") + + parts = [record_type, record_id] + if detail: + parts.append(_compact(_redact_text(detail), 80)) + if findings: + parts.append("ERROR " + ", ".join(sorted({finding.rule_id for finding in findings}))) + return "\n".join(parts) + + +def _safe_metadata(record: dict[str, Any]) -> dict[str, Any]: + return { + key: _redact_value(value) + for key, value in sorted(record.items()) + if key in SAFE_METADATA_FIELDS and value not in (None, "", []) + } + + +def _redact_value(value: Any) -> Any: + if isinstance(value, str): + return _redact_text(value) + if isinstance(value, list): + return [_redact_value(item) for item in value] + if isinstance(value, dict): + return {str(key): _redact_value(item) for key, item in sorted(value.items())} + return value + + +def _redact_text(text: str) -> str: + redacted = text + for pattern in SECRET_PATTERNS: + redacted = pattern.sub("[redacted-secret]", redacted) + for pattern in PRIVATE_PATH_PATTERNS: + redacted = pattern.sub("[redacted-private-path]", redacted) + return redacted + + +def _finding_to_dict(finding: Finding) -> dict[str, Any]: + return { + "rule_id": finding.rule_id, + "severity": finding.severity, + "message": _redact_text(finding.message), + "record_id": finding.record_id, + "path": _redact_text(finding.path) if finding.path else None, + "hint": _redact_text(finding.hint) if finding.hint else None, + } + + +def _mermaid_node_label(node: GraphNode) -> str: + return _redact_text(node.label) + + +def _mermaid_edge_label(label: str) -> str: + value = re.sub(r"[^A-Za-z0-9 _-]+", "", label).strip() + return value or "link" + + +def _mermaid_comment(value: str) -> str: + return value.replace("\n", " ").replace("\r", " ").replace("--", "-") + + +def _compact(value: str, limit: int) -> str: + cleaned = " ".join(value.replace("\n", " ").replace("\r", " ").split()) + if len(cleaned) <= limit: + return cleaned + return cleaned[: max(0, limit - 3)].rstrip() + "..." + + +def _as_list(value: Any) -> list[Any]: + if value is None: + return [] + if isinstance(value, list): + return value + return [value] + + +def _svg_positions(nodes: list[GraphNode]) -> dict[str, tuple[float, float]]: + lanes = { + "handoff": 140, + "ownership_claim": 140, + "privacy_boundary": 140, + "task": 340, + "event": 540, + "return_resume": 740, + "artifact": 740, + "missing": 740, + "review_gate": 940, + "memory_candidate": 1140, + } + grouped: dict[str, list[GraphNode]] = {} + for node in nodes: + grouped.setdefault(node.type, []).append(node) + + positions: dict[str, tuple[float, float]] = {} + base_y = 140 + gap_y = 110 + for node_type in sorted(grouped, key=lambda item: (lanes.get(item, 1240), NODE_TYPE_ORDER.get(item, 99))): + group = sorted(grouped[node_type], key=lambda node: node.id) + x = lanes.get(node_type, 1240) + for index, node in enumerate(group): + positions[node.id] = (float(x), float(base_y + index * gap_y)) + return positions + + +def _svg_node(node: GraphNode, index: int, x: float, y: float) -> list[str]: + style = _svg_style(node) + code = escape(_svg_node_code(node, index)) + title = escape(_compact(node.label.replace("\n", " | "), 240)) + if node.findings: + title += escape(" | " + ", ".join(node.findings)) + text_fill = "#111827" + lines = [f" ", f" {title}"] + if node.type == "event": + lines.append( + f' ' + ) + elif node.type == "review_gate": + points = f"{x:.1f},{y - 36:.1f} {x + 42:.1f},{y:.1f} {x:.1f},{y + 36:.1f} {x - 42:.1f},{y:.1f}" + lines.append( + f' ' + ) + elif node.type == "artifact": + lines.append( + f' ' + ) + elif node.type == "missing": + points = _svg_octagon_points(x, y, 38) + lines.append( + f' ' + ) + else: + lines.append( + f' ' + ) + lines.append( + f' {code}' + ) + if node.findings: + rule = escape(",".join(node.findings)) + lines.append( + f' {rule}' + ) + lines.append(" ") + return lines + + +def _svg_style(node: GraphNode) -> dict[str, str]: + if node.type == "missing": + return {"fill": "#fff7ed", "stroke": "#ea580c", "stroke_width": "2.2", "dash": ' stroke-dasharray="5 4"'} + if not node.valid: + return {"fill": "#fee2e2", "stroke": "#dc2626", "stroke_width": "2.4", "dash": ""} + if node.type == "event": + return {"fill": "#eef2ff", "stroke": "#4f46e5", "stroke_width": "2", "dash": ""} + if node.type == "review_gate": + return {"fill": "#fef9c3", "stroke": "#a16207", "stroke_width": "2", "dash": ""} + if node.type == "artifact": + return {"fill": "#ecfeff", "stroke": "#0891b2", "stroke_width": "2", "dash": ""} + return {"fill": "#e8f5e9", "stroke": "#2e7d32", "stroke_width": "2", "dash": ""} + + +def _svg_node_code(node: GraphNode, index: int) -> str: + if node.type == "event": + sequence = node.metadata.get("sequence") + return f"E{sequence}" if sequence is not None else "E" + codes = { + "task": "T", + "handoff": "H", + "return_resume": "R", + "review_gate": "G", + "artifact": "A", + "ownership_claim": "O", + "privacy_boundary": "P", + "memory_candidate": "M", + "missing": "?", + } + return codes.get(node.type, f"N{index}") + + +def _svg_octagon_points(x: float, y: float, radius: float) -> str: + inset = radius * 0.42 + points = [ + (x - inset, y - radius), + (x + inset, y - radius), + (x + radius, y - inset), + (x + radius, y + inset), + (x + inset, y + radius), + (x - inset, y + radius), + (x - radius, y + inset), + (x - radius, y - inset), + ] + return " ".join(f"{px:.1f},{py:.1f}" for px, py in points) diff --git a/tests/test_visualize.py b/tests/test_visualize.py new file mode 100644 index 0000000..039c450 --- /dev/null +++ b/tests/test_visualize.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path +import json +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from agentic_sdlc.cli import main +from agentic_sdlc.validator import validate_workspace +from agentic_sdlc.visualize import build_workflow_graph, render_json, render_mermaid, render_svg + + +class VisualizationTests(unittest.TestCase): + def test_valid_workflow_graph_contains_evidence_chain(self) -> None: + report = validate_workspace(ROOT / "fixtures" / "valid_workflow") + graph = build_workflow_graph(report) + data = graph.to_dict() + + node_ids = {node["id"] for node in data["nodes"]} + self.assertIn("task_valid_build", node_ids) + self.assertIn("gate_valid_tests", node_ids) + self.assertIn("artifact_tests_passed", node_ids) + + edges = {(edge["source"], edge["target"], edge["kind"]) for edge in data["edges"]} + self.assertIn(("task_valid_build", "gate_valid_tests", "review_gate"), edges) + self.assertIn(("artifact_tests_passed", "gate_valid_tests", "evidence_for"), edges) + self.assertTrue(data["valid"]) + + def test_missing_review_evidence_graph_exposes_rule_and_missing_node(self) -> None: + report = validate_workspace(ROOT / "fixtures" / "missing_review_evidence") + graph = build_workflow_graph(report) + data = graph.to_dict() + + self.assertFalse(data["valid"]) + self.assertIn("ASDLC010", {finding["rule_id"] for finding in data["findings"]}) + gate = next(node for node in data["nodes"] if node["id"] == "gate_missing_evidence") + self.assertIn("ASDLC010", gate["findings"]) + missing_nodes = [node for node in data["nodes"] if node["type"] == "missing"] + self.assertTrue(any(node["record_id"] == "test_result" for node in missing_nodes)) + + def test_json_output_is_deterministic_and_excludes_loader_private_fields(self) -> None: + report = validate_workspace(ROOT / "fixtures" / "valid_workflow") + graph = build_workflow_graph(report) + first = render_json(graph) + second = render_json(graph) + + self.assertEqual(first, second) + parsed = json.loads(first) + self.assertNotIn("_source_file", first) + self.assertNotIn("_expected_type", first) + self.assertEqual(parsed["view"], "readiness") + + def test_mermaid_output_contains_validation_state(self) -> None: + report = validate_workspace(ROOT / "fixtures" / "missing_review_evidence") + graph = build_workflow_graph(report) + output = render_mermaid(graph) + + self.assertIn("flowchart LR", output) + self.assertIn("status: FAIL", output) + self.assertIn("ASDLC010", output) + self.assertIn("Validation findings", output) + + def test_svg_output_uses_shapes_and_minimal_labels(self) -> None: + report = validate_workspace(ROOT / "fixtures" / "missing_review_evidence") + graph = build_workflow_graph(report) + output = render_svg(graph) + + self.assertIn("T<", output) + self.assertIn(">G<", output) + + def test_cli_visualize_json_uses_validation_exit_code(self) -> None: + buffer = StringIO() + with redirect_stdout(buffer): + exit_code = main(["visualize", str(ROOT / "fixtures" / "missing_review_evidence"), "--json"]) + + self.assertEqual(exit_code, 1) + parsed = json.loads(buffer.getvalue()) + self.assertFalse(parsed["valid"]) + self.assertIn("ASDLC010", {finding["rule_id"] for finding in parsed["findings"]}) + + def test_cli_visualize_output_file_writes_without_stdout(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + output_path = Path(tmp) / "workflow.svg" + buffer = StringIO() + with redirect_stdout(buffer): + exit_code = main( + [ + "visualize", + str(ROOT / "fixtures" / "valid_workflow"), + "--format", + "svg", + "--output", + str(output_path), + ] + ) + + self.assertEqual(exit_code, 0) + self.assertEqual(buffer.getvalue(), "") + self.assertIn("