feat(copilot): FixContext JSON (de)serialization

api/services/workflow_copilot/serde.py

@@ -0,0 +1,65 @@
+"""Pure JSON (de)serialization for the domain ``FixContext``.
+
+``FixContext`` (see ``core.workflow_copilot.models``) is the per-session
+working state persisted, one row per version, as the ``context`` JSONB
+column on ``workflow_copilot_session_commits`` (P3a Task 4). This module
+has no I/O and no SQLAlchemy dependency -- it only converts between the
+dataclass and a JSON-safe ``dict``, so the SQL repository can hand the dict
+straight to an ``AdjustedJSON`` column and get it back unchanged.
+"""
+
+from dataclasses import asdict
+from typing import Any
+
+from core.workflow_copilot.models import (
+    ChangeSet,
+    ChecklistError,
+    Diagnosis,
+    FixContext,
+    MutationIntent,
+    Risk,
+)
+
+
+def fix_context_to_dict(fc: FixContext) -> dict[str, Any]:
+    """Serialize a ``FixContext`` to a JSON-safe dict.
+
+    ``dataclasses.asdict`` recurses into the nested dataclasses
+    (``Diagnosis``, ``MutationIntent``, ``Risk``, ``ChangeSet``,
+    ``ChecklistError``), turning each into a plain dict, and every field on
+    ``FixContext`` and its nested dataclasses is already a
+    str/int/bool/list/dict/None -- so the result is JSON-safe as-is.
+    """
+    return asdict(fc)
+
+
+def fix_context_from_dict(d: dict[str, Any]) -> FixContext:
+    """Rebuild a ``FixContext`` from a dict produced by ``fix_context_to_dict``.
+
+    Reconstructs the nested dataclasses explicitly rather than splatting
+    ``d`` into ``FixContext(**d)`` -- the nested values are plain dicts (and
+    lists of dicts) after a JSON round-trip and must become the right
+    dataclass instances. Every lookup is defensive (``.get`` with a default
+    matching the ``FixContext``/nested-dataclass field default) so an
+    older or partial dict still yields a valid ``FixContext``.
+    """
+    diagnosis = Diagnosis(**d["diagnosis"]) if d.get("diagnosis") else None
+    staged_repair = [MutationIntent(**m) for m in d.get("staged_repair", [])]
+    risk = Risk(**d["risk"]) if d.get("risk") else None
+    change_set = ChangeSet(**d["change_set"]) if d.get("change_set") else None
+    checklist_errors = [ChecklistError(**e) for e in d.get("checklist_errors", [])]
+
+    return FixContext(
+        failed_run_id=d.get("failed_run_id", ""),
+        diagnosis=diagnosis,
+        staged_repair=staged_repair,
+        risk=risk,
+        change_set=change_set,
+        checkpoint_id=d.get("checkpoint_id", ""),
+        verify_run_id=d.get("verify_run_id", ""),
+        test_input_ref=d.get("test_input_ref", ""),
+        last_snapshot_hash=d.get("last_snapshot_hash", ""),
+        next_seq=d.get("next_seq", 0),
+        source=d.get("source", ""),
+        checklist_errors=checklist_errors,
+    )

api/tests/unit_tests/services/workflow_copilot/test_serde.py

@@ -0,0 +1,88 @@
+"""Round-trip tests for the ``FixContext`` JSON (de)serializer.
+
+``FixContext`` is the per-session working state persisted as the
+``context`` JSONB column on ``workflow_copilot_session_commits`` (P3a Task 4).
+This module verifies the pure (de)serializer is lossless and produces a
+JSON-safe dict, independent of any database concern.
+"""
+
+import json
+
+from core.workflow_copilot.models import (
+    ChangeSet,
+    ChecklistError,
+    Diagnosis,
+    FixContext,
+    MutationIntent,
+    Risk,
+)
+from services.workflow_copilot.serde import fix_context_from_dict, fix_context_to_dict
+
+
+def _full_fix_context() -> FixContext:
+    return FixContext(
+        failed_run_id="TR-1",
+        diagnosis=Diagnosis(
+            culprit_node_id="n1",
+            root_cause="TypeError in code node",
+            severity="high",
+        ),
+        staged_repair=[
+            MutationIntent(
+                op="set_node_config",
+                args={"node_id": "n1", "path": "code", "value": "return {}"},
+            )
+        ],
+        risk=Risk(level="low", reason="config-only change", has_external_side_effect=False),
+        change_set=ChangeSet(changed_nodes=["n1"], diff="- old\n+ new"),
+        checkpoint_id="cp-1",
+        verify_run_id="TR-2",
+        test_input_ref="ti-1",
+        last_snapshot_hash="abc123",
+        next_seq=3,
+        source="checklist",
+        checklist_errors=[
+            ChecklistError(
+                node_id="n2",
+                node_type="code",
+                title="Code",
+                messages=["missing required field"],
+                unconnected=False,
+                plugin_missing=False,
+            )
+        ],
+    )
+
+
+def test_fully_populated_fix_context_round_trips():
+    fc = _full_fix_context()
+
+    result = fix_context_from_dict(fix_context_to_dict(fc))
+
+    assert result == fc
+
+
+def test_empty_fix_context_round_trips():
+    fc = FixContext()
+
+    result = fix_context_from_dict(fix_context_to_dict(fc))
+
+    assert result == fc
+
+
+def test_to_dict_output_is_json_serializable():
+    fc = _full_fix_context()
+
+    d = fix_context_to_dict(fc)
+
+    # Must not raise -- this is what actually lands in the JSONB column.
+    json.dumps(d)
+
+
+def test_from_dict_tolerates_missing_optional_keys():
+    # An older/partial persisted dict -- only a subset of keys present.
+    d = {"failed_run_id": "TR-9"}
+
+    fc = fix_context_from_dict(d)
+
+    assert fc == FixContext(failed_run_id="TR-9")