Add workflow evidence export utility for research notes

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`

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",
+]

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

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())

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()