From ae1f21dbafe4be676d679546bcd6a5abee55467b Mon Sep 17 00:00:00 2001 From: Adam Gohain <68021524+akgohain@users.noreply.github.com> Date: Mon, 13 Apr 2026 15:56:37 -0400 Subject: [PATCH] Add workflow evidence export utility for paper notes --- docs/research/workflow-evidence-export.md | 33 +++++ server_api/workflow/__init__.py | 13 ++ server_api/workflow/data_access.py | 40 ++++++ server_api/workflow/evidence_export.py | 149 ++++++++++++++++++++++ tests/test_workflow_evidence_export.py | 106 +++++++++++++++ 5 files changed, 341 insertions(+) create mode 100644 docs/research/workflow-evidence-export.md create mode 100644 server_api/workflow/__init__.py create mode 100644 server_api/workflow/data_access.py create mode 100644 server_api/workflow/evidence_export.py create mode 100644 tests/test_workflow_evidence_export.py diff --git a/docs/research/workflow-evidence-export.md b/docs/research/workflow-evidence-export.md new file mode 100644 index 00000000..eb59fcd3 --- /dev/null +++ b/docs/research/workflow-evidence-export.md @@ -0,0 +1,33 @@ +# Workflow Evidence Export + +Use this utility to create a stable JSON summary for paper/research notes from a seeded workflow fixture. + +## Output schema + +The exporter writes a JSON document with a stable top-level version key: + +- `version` — currently `1.0` +- `workflow_id` +- `stage_progression_summary` +- `agent_proposal_approval_summary` +- `key_event_timeline_snippet` (machine-readable list) + +## Run the export + +```bash +uv run python -m server_api.workflow.evidence_export \ + --workflow-json path/to/workflow-fixture.json \ + --output docs/research/exports/workflow-evidence.json +``` + +## Where output is written + +The `--output` path is written exactly as provided. Parent directories are created automatically. + +For research notes, keep exports under: + +- `docs/research/exports/` + +Example output target: + +- `docs/research/exports/workflow-evidence.json` diff --git a/server_api/workflow/__init__.py b/server_api/workflow/__init__.py new file mode 100644 index 00000000..efb047a6 --- /dev/null +++ b/server_api/workflow/__init__.py @@ -0,0 +1,13 @@ +"""Workflow utilities and export helpers.""" + +from .evidence_export import ( + EVIDENCE_EXPORT_VERSION, + build_workflow_evidence_export, + export_workflow_evidence, +) + +__all__ = [ + "EVIDENCE_EXPORT_VERSION", + "build_workflow_evidence_export", + "export_workflow_evidence", +] diff --git a/server_api/workflow/data_access.py b/server_api/workflow/data_access.py new file mode 100644 index 00000000..11392076 --- /dev/null +++ b/server_api/workflow/data_access.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List + + +class WorkflowDataError(ValueError): + """Raised when workflow payloads cannot be parsed.""" + + +def load_workflow_payload(workflow_path: str | Path) -> Dict[str, Any]: + """Load a workflow JSON payload from disk. + + The returned value is always a dictionary with an ``events`` key. + """ + + path = Path(workflow_path) + if not path.exists(): + raise WorkflowDataError(f"Workflow file does not exist: {path}") + + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise WorkflowDataError(f"Workflow file is not valid JSON: {path}") from exc + + if not isinstance(payload, dict): + raise WorkflowDataError("Workflow payload must be a JSON object") + + events = payload.get("events", []) + if not isinstance(events, list): + raise WorkflowDataError("Workflow payload 'events' field must be a list") + + normalized_events: List[Dict[str, Any]] = [] + for event in events: + if isinstance(event, dict): + normalized_events.append(event) + + payload["events"] = normalized_events + return payload diff --git a/server_api/workflow/evidence_export.py b/server_api/workflow/evidence_export.py new file mode 100644 index 00000000..c0932d3b --- /dev/null +++ b/server_api/workflow/evidence_export.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import argparse +import json +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any, Dict, List + +from .data_access import WorkflowDataError, load_workflow_payload + +EVIDENCE_EXPORT_VERSION = "1.0" + + +def _stage_progression(payload: Dict[str, Any]) -> Dict[str, Any]: + stages = payload.get("stages") if isinstance(payload.get("stages"), list) else [] + events = payload["events"] + + observed: List[str] = [] + entered_at: Dict[str, str] = {} + transition_count = 0 + + for event in events: + event_type = event.get("type") + stage = event.get("stage") + if event_type in {"stage_entered", "stage_changed"} and isinstance(stage, str): + transition_count += 1 + if stage not in observed: + observed.append(stage) + if stage not in entered_at and isinstance(event.get("timestamp"), str): + entered_at[stage] = event["timestamp"] + + completed = [stage for stage in stages if stage in observed] + pending = [stage for stage in stages if stage not in observed] + + return { + "declared_stages": stages, + "observed_stages": observed, + "completed_stages": completed, + "pending_stages": pending, + "entered_at": entered_at, + "transition_count": transition_count, + } + + +def _proposal_approval_summary(events: List[Dict[str, Any]]) -> Dict[str, Any]: + proposals_by_agent: Counter[str] = Counter() + approvals_by_agent: Counter[str] = Counter() + decisions_by_proposal: Dict[str, Dict[str, Any]] = defaultdict(dict) + + for event in events: + event_type = event.get("type") + proposal_id = str(event.get("proposal_id")) if event.get("proposal_id") is not None else None + agent = event.get("agent") if isinstance(event.get("agent"), str) else "unknown" + + if event_type == "proposal_submitted": + proposals_by_agent[agent] += 1 + if proposal_id: + decisions_by_proposal[proposal_id]["submitted_by"] = agent + decisions_by_proposal[proposal_id]["submitted_at"] = event.get("timestamp") + elif event_type in {"proposal_approved", "proposal_rejected"}: + approvals_by_agent[agent] += 1 + if proposal_id: + decisions_by_proposal[proposal_id]["decision"] = ( + "approved" if event_type == "proposal_approved" else "rejected" + ) + decisions_by_proposal[proposal_id]["decided_by"] = agent + decisions_by_proposal[proposal_id]["decided_at"] = event.get("timestamp") + + approved_count = sum( + 1 for details in decisions_by_proposal.values() if details.get("decision") == "approved" + ) + rejected_count = sum( + 1 for details in decisions_by_proposal.values() if details.get("decision") == "rejected" + ) + + return { + "proposal_count": int(sum(proposals_by_agent.values())), + "approved_count": approved_count, + "rejected_count": rejected_count, + "proposals_by_agent": dict(proposals_by_agent), + "decisions_by_agent": dict(approvals_by_agent), + "proposals": dict(decisions_by_proposal), + } + + +def _timeline_snippet(events: List[Dict[str, Any]], max_events: int = 20) -> List[Dict[str, Any]]: + timeline: List[Dict[str, Any]] = [] + for event in events[:max_events]: + timeline.append( + { + "timestamp": event.get("timestamp"), + "event_type": event.get("type"), + "stage": event.get("stage"), + "agent": event.get("agent"), + "proposal_id": event.get("proposal_id"), + "status": event.get("status"), + } + ) + return timeline + + +def build_workflow_evidence_export(payload: Dict[str, Any]) -> Dict[str, Any]: + events = payload.get("events", []) + return { + "version": EVIDENCE_EXPORT_VERSION, + "workflow_id": payload.get("workflow_id"), + "stage_progression_summary": _stage_progression(payload), + "agent_proposal_approval_summary": _proposal_approval_summary(events), + "key_event_timeline_snippet": _timeline_snippet(events), + } + + +def export_workflow_evidence(workflow_path: str | Path, output_path: str | Path) -> Path: + payload = load_workflow_payload(workflow_path) + export_payload = build_workflow_evidence_export(payload) + + output = Path(output_path) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(export_payload, indent=2), encoding="utf-8") + return output + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Export workflow evidence summaries for paper-writing notes." + ) + parser.add_argument("--workflow-json", required=True, help="Path to workflow JSON input") + parser.add_argument( + "--output", + required=True, + help="Path to write export JSON (for example docs/research/exports/workflow-evidence.json)", + ) + return parser + + +def main() -> int: + args = _parser().parse_args() + try: + output = export_workflow_evidence(args.workflow_json, args.output) + except WorkflowDataError as exc: + print(f"workflow evidence export failed: {exc}") + return 2 + + print(f"workflow evidence export written to {output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_workflow_evidence_export.py b/tests/test_workflow_evidence_export.py new file mode 100644 index 00000000..a2cf8dd9 --- /dev/null +++ b/tests/test_workflow_evidence_export.py @@ -0,0 +1,106 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from server_api.workflow.evidence_export import ( + EVIDENCE_EXPORT_VERSION, + build_workflow_evidence_export, + export_workflow_evidence, +) + + +class WorkflowEvidenceExportTests(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.temp_path = Path(self.temp_dir.name) + self.fixture_path = self.temp_path / "seeded-workflow.json" + self.output_path = self.temp_path / "exports" / "workflow-evidence.json" + + fixture = { + "workflow_id": "wf-seeded-001", + "stages": ["intake", "analysis", "draft", "review"], + "events": [ + { + "timestamp": "2026-03-01T10:00:00Z", + "type": "stage_entered", + "stage": "intake", + "agent": "orchestrator", + }, + { + "timestamp": "2026-03-01T10:02:00Z", + "type": "proposal_submitted", + "stage": "analysis", + "proposal_id": "p-1", + "agent": "agent-alpha", + }, + { + "timestamp": "2026-03-01T10:03:00Z", + "type": "proposal_approved", + "stage": "analysis", + "proposal_id": "p-1", + "agent": "reviewer-1", + "status": "approved", + }, + { + "timestamp": "2026-03-01T10:05:00Z", + "type": "stage_changed", + "stage": "analysis", + "agent": "orchestrator", + }, + { + "timestamp": "2026-03-01T10:08:00Z", + "type": "proposal_submitted", + "stage": "draft", + "proposal_id": "p-2", + "agent": "agent-beta", + }, + { + "timestamp": "2026-03-01T10:09:00Z", + "type": "proposal_rejected", + "stage": "draft", + "proposal_id": "p-2", + "agent": "reviewer-2", + "status": "rejected", + }, + ], + } + self.fixture_path.write_text(json.dumps(fixture), encoding="utf-8") + + def tearDown(self): + self.temp_dir.cleanup() + + def test_build_export_contains_required_sections(self): + payload = json.loads(self.fixture_path.read_text(encoding="utf-8")) + evidence = build_workflow_evidence_export(payload) + + self.assertEqual(evidence["version"], EVIDENCE_EXPORT_VERSION) + self.assertIn("stage_progression_summary", evidence) + self.assertIn("agent_proposal_approval_summary", evidence) + self.assertIn("key_event_timeline_snippet", evidence) + + stage_summary = evidence["stage_progression_summary"] + self.assertEqual(stage_summary["completed_stages"], ["intake", "analysis"]) + self.assertIn("draft", stage_summary["pending_stages"]) + + proposal_summary = evidence["agent_proposal_approval_summary"] + self.assertEqual(proposal_summary["proposal_count"], 2) + self.assertEqual(proposal_summary["approved_count"], 1) + self.assertEqual(proposal_summary["rejected_count"], 1) + + timeline = evidence["key_event_timeline_snippet"] + self.assertGreaterEqual(len(timeline), 3) + self.assertIn("event_type", timeline[0]) + + def test_export_writes_output_file_for_seeded_fixture(self): + result_path = export_workflow_evidence(self.fixture_path, self.output_path) + self.assertEqual(result_path, self.output_path) + self.assertTrue(self.output_path.exists()) + + written = json.loads(self.output_path.read_text(encoding="utf-8")) + self.assertEqual(written["workflow_id"], "wf-seeded-001") + self.assertEqual(written["version"], EVIDENCE_EXPORT_VERSION) + + +if __name__ == "__main__": + unittest.main()