fix(content): keep coordination notes off dev_notes / quick_context

Two agent-authored fields were leaking non-developer content into the
human note columns the panel renders:

- apply_escalation appended '[ESCALATED] From X to Y\nReason: ...' to
  dev_notes (the developer's space). On a re-escalation loop a stuck cell
  PM grew one task's dev_notes to ~8KB across 5 escalations. It now writes
  a structured orchestration_markers['escalation'] record; the target
  still learns the reason from the escalate notification.
- approve_and_start string-packed 'approve_and_start_notes:<text>' into
  quick_context (raw key:value soup). It now writes
  orchestration_markers['approve_and_start_notes'], leaving quick_context
  for the human ResumptionNote only.

Adds typed marker accessors (get/set_escalation, get/set_approve_and_start_notes)
and refactors _record_pr_review under the complexity bound by extracting
_compose_review_body. Documents update_task_with_message as the legacy
A2A-protocol log (dev_notes is intentional there, not pollution).

roboco/foundation/policy/content/markers.py

@@ -30,6 +30,8 @@ EXTERNAL_PR_HEAD = "external_pr_head"
 EXTERNAL_PR_SUPERSEDE = "external_pr_supersede"
 SELF_HEAL_FP = "self_heal_fp"
 DISMISSED = "dismissed"
+ESCALATION = "escalation"
+APPROVE_AND_START_NOTES = "approve_and_start_notes"
 
 
 def get_marker(task: HasMarkers, key: str, default: Any = None) -> Any:
@@ -136,3 +138,36 @@ def is_dismissed(task: HasMarkers) -> bool:
 
 def mark_dismissed(task: HasMarkers) -> None:
     set_marker(task, DISMISSED, True)
+
+
+# --- escalation ------------------------------------------------------------ #
+# A coordination event, NOT a developer note. It used to be appended to
+# ``dev_notes`` (polluting the developer's space and growing unboundedly on a
+# re-escalation loop); it lives here as the latest structured record instead.
+# Delivery of the reason to the target is handled by the escalate notification.
+
+
+def get_escalation(task: HasMarkers) -> dict[str, str] | None:
+    val = get_marker(task, ESCALATION)
+    return val if isinstance(val, dict) else None
+
+
+def set_escalation(
+    task: HasMarkers, *, from_slug: str, to_slug: str, reason: str
+) -> None:
+    set_marker(task, ESCALATION, {"from": from_slug, "to": to_slug, "reason": reason})
+
+
+# --- approve-and-start notes ----------------------------------------------- #
+# The CEO's note when approving a board-reviewed coordination root. Used to be
+# string-packed into ``quick_context`` as ``approve_and_start_notes:<text>``;
+# kept here so ``quick_context`` carries only the human ResumptionNote.
+
+
+def get_approve_and_start_notes(task: HasMarkers) -> str | None:
+    val = get_marker(task, APPROVE_AND_START_NOTES)
+    return str(val) if val else None
+
+
+def set_approve_and_start_notes(task: HasMarkers, notes: str) -> None:
+    set_marker(task, APPROVE_AND_START_NOTES, notes)

roboco/services/a2a.py

@@ -539,7 +539,14 @@ class A2AService:
 
     @staticmethod
     def update_task_with_message(task: TaskTable, message: A2AMessage) -> None:
-        """Update an existing task's dev_notes with new message content."""
+        """Append A2A-protocol message text to the task's A2A log (dev_notes).
+
+        NOTE: this is the *legacy A2A-protocol* message store — A2A-protocol
+        tasks carry their request/response thread in ``dev_notes`` (keyed by the
+        ``"A2A Request"`` marker that ``_notify_original_requester`` checks).
+        It is NOT the gateway agent flow (those use the A2AConversation tables),
+        so it does not pollute normal delivery tasks' developer notes.
+        """
         text_parts = [p for p in message.parts if p.type == "text"]
         if not text_parts:
             return

roboco/services/task.py

@@ -199,6 +199,17 @@ def _append_capped(existing: str | None, addition: str) -> str:
     return _TRUNCATION_MARKER + joined[-keep:]
 
 
+def _compose_review_body(summary: str | None, issues: list[str] | None) -> str:
+    """Combine a PR-review summary with issue bullets into one body string."""
+    body = (summary or "").strip()
+    if not issues:
+        return body
+    bullets = "\n".join(f"- {i}" for i in issues if i and i.strip())
+    if not bullets:
+        return body
+    return f"{body}\n\n{bullets}".strip() if body else bullets
+
+
 @dataclass
 class _CompletionSnapshot:
     """Fields copied off a TaskTable before its session detaches.
@@ -3655,10 +3666,7 @@ class TaskService(BaseService):
         """
         if (task.notes_structured or {}).get("pr_review"):
             return
-        body = (summary or "").strip()
-        if issues:
-            bullets = "\n".join(f"- {i}" for i in issues if i and i.strip())
-            body = f"{body}\n\n{bullets}".strip() if body else bullets
+        body = _compose_review_body(summary, issues)
         if not body:
             return
         try:
@@ -4176,11 +4184,14 @@ class TaskService(BaseService):
         task.assigned_to = cast("Any", target_agent_id)
         task.claimed_by = cast("Any", target_agent_id)
         task.status = TaskStatus.BLOCKED
-        existing_notes = task.dev_notes or ""
-        escalation_note = (
-            f"\n\n[ESCALATED] From {escalator_slug} to {target_slug}\nReason: {reason}"
+        # Record the escalation as a structured marker — NOT appended to
+        # dev_notes (the developer's space). The old append polluted dev_notes
+        # and, on a re-escalation loop, grew it unboundedly (a cell PM stuck on
+        # needs_revision escalated 5x → 8KB of [ESCALATED] blocks). The target
+        # learns the reason from the escalate notification (escalate_and_notify).
+        markers.set_escalation(
+            task, from_slug=escalator_slug, to_slug=target_slug, reason=reason
         )
-        task.dev_notes = existing_notes + escalation_note
         await self.session.flush()
         # This path sets BLOCKED directly (bypassing the strict transition
         # validator), so emit the task.blocked audit explicitly — no status
@@ -4430,9 +4441,10 @@ class TaskService(BaseService):
             task.confirmed_by_human = True
 
         if notes:
-            existing = task.quick_context or ""
-            entry = f"approve_and_start_notes:{notes}"
-            task.quick_context = f"{existing}\n{entry}".strip() if existing else entry
+            # Coordination metadata, not a human handoff — store as a marker so
+            # quick_context carries only the structured ResumptionNote (no raw
+            # `approve_and_start_notes:<text>` soup leaking into the panel).
+            markers.set_approve_and_start_notes(task, notes)
 
         await self.session.flush()
         await self._emit_task_event(

tests/integration/test_approve_and_start_service.py

@@ -91,22 +91,25 @@ async def test_reassigns_to_main_pm_and_keeps_pending(start_setup: dict) -> None
     assert out is not None
     assert out.assigned_to == start_setup["main_pm"].id
     assert out.status == TaskStatus.PENDING
-    assert out.quick_context is not None
-    assert "approve_and_start_notes:" in out.quick_context
-    assert note in out.quick_context
+    # The CEO note is a structured marker, not raw soup in quick_context.
+    assert (out.orchestration_markers or {})["approve_and_start_notes"] == note
+    assert "approve_and_start_notes:" not in (out.quick_context or "")
 
 
 @pytest.mark.asyncio
-async def test_audit_note_appends_to_existing_context(start_setup: dict) -> None:
+async def test_audit_note_preserves_context_and_stores_marker(
+    start_setup: dict,
+) -> None:
     task = start_setup["mk"]()
     task.quick_context = "prior context"
     await start_setup["db"].flush()
     note = "Board signed off; ship it."
     out = await start_setup["svc"].approve_and_start(task.id, note)
     assert out is not None
-    assert out.quick_context is not None
-    assert "prior context" in out.quick_context
-    assert f"approve_and_start_notes:{note}" in out.quick_context
+    # An existing human ResumptionNote is preserved untouched...
+    assert out.quick_context == "prior context"
+    # ...and the CEO note lands in the markers, not appended to the context.
+    assert (out.orchestration_markers or {})["approve_and_start_notes"] == note
 
 
 @pytest.mark.asyncio

tests/integration/test_task_service_transitions.py

@@ -1448,7 +1448,10 @@ async def test_apply_escalation_reassigns_and_blocks(
     assert task.status == TaskStatus.BLOCKED
     assert task.assigned_to == target.id
     assert task.blocker_raised_by == task_setup["agent_id"]
-    assert "[ESCALATED]" in (task.dev_notes or "")
+    # The escalation is a structured marker, NOT a developer note.
+    assert not (task.dev_notes or "")
+    esc = (task.orchestration_markers or {})["escalation"]
+    assert esc == {"from": "dev-1", "to": "cell-pm", "reason": "external blocker"}
 
 
 @pytest.mark.asyncio

tests/unit/foundation/policy/content/test_markers.py

@@ -50,6 +50,24 @@ def test_clear_marker_nulls_when_empty() -> None:
     assert t.orchestration_markers is None
 
 
+def test_escalation_roundtrip() -> None:
+    t = _task()
+    assert m.get_escalation(t) is None
+    m.set_escalation(t, from_slug="be-pm", to_slug="main-pm", reason="re-open please")
+    assert m.get_escalation(t) == {
+        "from": "be-pm",
+        "to": "main-pm",
+        "reason": "re-open please",
+    }
+
+
+def test_approve_and_start_notes_roundtrip() -> None:
+    t = _task()
+    assert m.get_approve_and_start_notes(t) is None
+    m.set_approve_and_start_notes(t, "Board approved; build it.")
+    assert m.get_approve_and_start_notes(t) == "Board approved; build it."
+
+
 def test_documenter_self_heal_head_supersede() -> None:
     t = _task()
     m.set_documenter(t, "doc-uuid")