test(ci): require retained-history projection delivery

tests/control_plane/test_research_history_capacity.py

@@ -40,9 +40,11 @@ def observation(node: str) -> dict:
 
 
 def todo(todo_id: str, index: int, **fields) -> dict:
+    # Seed the same lossless priority display that native Todo authoring produces.
+    priority = fields.get("priority", "P2")
     return {"schema_version": "todo_item_v0", "todo_id": todo_id, "index": index, "role": "agent",
-            "status": "open", "done": False, "text": f"Synthetic {todo_id}.", "archive_state": "active",
-            "source_section": "Agent Todo", "priority": "P2", "claimed_by": AGENT,
+            "status": "open", "done": False, "text": f"[{priority}] Synthetic {todo_id}.", "archive_state": "active",
+            "source_section": "Agent Todo", "priority": priority, "claimed_by": AGENT,
             "task_class": "advancement_task", **fields}
 
 
@@ -86,9 +88,12 @@ def complete(registry: Path, runtime: Path, todo_id: str, key: str, **fields) ->
     lease = acquire_task_lease(registry_path=registry, runtime_root=runtime, goal_id=GOAL, todo_id=todo_id,
                                owner=AGENT, idempotency_key=key, ttl_seconds=300)
     assert lease["ok"], lease
-    return complete_goal_todo(registry_path=registry, goal_id=GOAL, todo_id=todo_id, agent_id=AGENT,
+    result = complete_goal_todo(registry_path=registry, goal_id=GOAL, todo_id=todo_id, agent_id=AGENT,
         claimed_by=AGENT, no_followup=True, evidence="Synthetic closeout.", task_lease_idempotency_key=key,
         task_lease_expected_version=lease["lease"]["version"], **fields)
+    assert result["projection_delivery"] == "delivered", result.get("projection_outbox")
+    assert result["projection_outbox"]["parse_render_parity"] is True
+    return result
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
@@ -141,10 +146,14 @@ def close_joint(evidence: str) -> dict:
 
         accepted = close_joint("Typed synthetic result.")
         assert accepted["completed"]
+        assert accepted["projection_delivery"] == "delivered", accepted.get("projection_outbox")
+        assert accepted["projection_outbox"]["parse_render_parity"] is True
+        assert accepted["projection_outbox"]["todo_count"] == HISTORY + 2
         assert accepted["capability_completion_evidence"]["experiment_node_id"] == "joint"
         revision = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL)["provider_revision"]
         replay = close_joint("Same identity retry.")
         assert replay["provider_status"] == "replayed"
+        assert replay["projection_delivery"] == "current", replay.get("projection_outbox")
         assert replay["capability_completion_evidence"] == accepted["capability_completion_evidence"]
         assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL)["provider_revision"] == revision
     finally: