feat(content): obligate role note sections like journals

Completes the note(scope='handoff') write-path (WIP 23e6ee57): every role
with a dedicated note section is now obligated to populate it, the same way
journals are obligated.

Obligations (foundation.policy.tracing):
- DEV_NOTES / PR_REVIEWER_NOTES / QUICK_CONTEXT_MIN_CHARS requirements +
  checkers, wired onto i_am_done (dev_notes), delegate (quick_context), and
  pr_pass / pr_fail / post_pr_review (pr_reviewer_notes).
- Fixes a latent bug: the docs-notes checker read dev_notes instead of
  doc_notes (the documenter's section); the i_documented shim now feeds
  doc_notes to match.

Auditor: a session-scoped note obligation on i_am_idle — the auditor owns no
delivery task and has no delivery verb, so it must have recorded an
observation within the window before going idle (JournalService.has_recent_entry).

Write-then-gate: persisted sections (dev_notes / quick_context) are
pre-written by the agent's note(scope='handoff') before the gated verb;
argument-borne sections (doc_notes / pr_reviewer_notes) are checked through a
SimpleNamespace shim, the same pattern qa_notes already uses.

Config: dev/pr_reviewer/quick_context min-chars (40/40/30), panel-tunable.
Plus per-gap remediation hints and full coverage (write-path routing,
ownership, validation->remediation, each obligation, the doc_notes fix).

Full make-quality green: 9777 passed, 95.6% coverage.

roboco/config.py

@@ -777,6 +777,21 @@ class Settings(BaseSettings):
         ge=1,
         description="Minimum characters for docs notes",
     )
+    dev_notes_min_chars: int = Field(
+        default=40,
+        ge=1,
+        description="Minimum characters for a developer's dev_notes section",
+    )
+    pr_reviewer_notes_min_chars: int = Field(
+        default=40,
+        ge=1,
+        description="Minimum characters for a PR reviewer's pr_reviewer_notes section",
+    )
+    quick_context_min_chars: int = Field(
+        default=30,
+        ge=1,
+        description="Minimum characters for a PM's quick_context resumption section",
+    )
 
     # Commit-validator thresholds (wired into the gateway commit() gate)
     commit_subject_min_chars: int = Field(

roboco/foundation/policy/tracing.py

@@ -40,6 +40,15 @@ class Requirement(StrEnum):
     SELF_VERIFIED = "self_verified"
     NOTES_MIN_CHARS = "notes>=min"
     SUBTASKS_TERMINAL = "subtasks_terminal"
+    # Role note-section obligations (parity with the journal requirements): a
+    # role with a dedicated note section must populate it via
+    # note(scope='handoff') the same way journals are obligated. The developer's
+    # dev_notes, the PR reviewer's pr_reviewer_notes, and the PM's quick_context
+    # had no agent write-path before — these obligate the section now that one
+    # exists.
+    DEV_NOTES_MIN_CHARS = "dev_notes>=min"
+    PR_REVIEWER_NOTES_MIN_CHARS = "pr_reviewer_notes>=min"
+    QUICK_CONTEXT_MIN_CHARS = "quick_context>=min"
 
 
 @dataclass(frozen=True)
@@ -55,6 +64,9 @@ class GateContext:
     qa_notes_min_chars: int = 80
     docs_notes_min_chars: int = 20
     notes_min_chars: int = 20
+    dev_notes_min_chars: int = 40
+    pr_reviewer_notes_min_chars: int = 40
+    quick_context_min_chars: int = 30
 
 
 @dataclass(frozen=True)
@@ -149,10 +161,29 @@ def _check_qa_evidence_inspected(task: Any, _ctx: GateContext) -> list[str]:
 
 
 def _check_docs_notes_min_chars(task: Any, ctx: GateContext) -> list[str]:
-    notes = getattr(task, "dev_notes", "") or ""
+    notes = getattr(task, "doc_notes", "") or ""
     return [] if len(notes) >= ctx.docs_notes_min_chars else ["docs_notes>=min"]
 
 
+def _check_dev_notes_min_chars(task: Any, ctx: GateContext) -> list[str]:
+    notes = getattr(task, "dev_notes", "") or ""
+    return [] if len(notes) >= ctx.dev_notes_min_chars else ["dev_notes>=min"]
+
+
+def _check_pr_reviewer_notes_min_chars(task: Any, ctx: GateContext) -> list[str]:
+    notes = getattr(task, "pr_reviewer_notes", "") or ""
+    return (
+        []
+        if len(notes) >= ctx.pr_reviewer_notes_min_chars
+        else ["pr_reviewer_notes>=min"]
+    )
+
+
+def _check_quick_context_min_chars(task: Any, ctx: GateContext) -> list[str]:
+    notes = getattr(task, "quick_context", "") or ""
+    return [] if len(notes) >= ctx.quick_context_min_chars else ["quick_context>=min"]
+
+
 def _check_docs_files_non_empty(task: Any, _ctx: GateContext) -> list[str]:
     docs = getattr(task, "documents", None) or []
     return [] if len(docs) >= 1 else ["docs_files_non_empty"]
@@ -195,6 +226,9 @@ _CHECKERS: dict[Requirement, Checker] = {
     Requirement.SELF_VERIFIED: _check_self_verified,
     Requirement.NOTES_MIN_CHARS: _check_notes_min_chars,
     Requirement.SUBTASKS_TERMINAL: _check_subtasks_terminal,
+    Requirement.DEV_NOTES_MIN_CHARS: _check_dev_notes_min_chars,
+    Requirement.PR_REVIEWER_NOTES_MIN_CHARS: _check_pr_reviewer_notes_min_chars,
+    Requirement.QUICK_CONTEXT_MIN_CHARS: _check_quick_context_min_chars,
 }
 
 
@@ -232,8 +266,13 @@ VERB_REQUIREMENTS: dict[str, frozenset[Requirement]] = {
             Requirement.JOURNAL_DECISION_AT_CLAIM,
         }
     ),
-    # PM delegate — pre-gateway PM.md required journal:decision before each delegate.
-    "delegate": frozenset({Requirement.JOURNAL_DECISION}),
+    # PM delegate — pre-gateway PM.md required journal:decision before each
+    # delegate. QUICK_CONTEXT_MIN_CHARS obligates the PM's resumption section
+    # (quick_context) on the parent: satisfiable because the PM pre-writes it via
+    # note(scope='handoff', section={done, next, ...}) before delegate.
+    "delegate": frozenset(
+        {Requirement.JOURNAL_DECISION, Requirement.QUICK_CONTEXT_MIN_CHARS}
+    ),
     # Developer submit — adds JOURNAL_DURING_WORK_AT_LEAST_ONE for mid-flight cadence.
     # SELF_VERIFIED is set by the auto-run in_progress→verifying transition; it
     # stays in the required-set as a defense-in-depth backstop.
@@ -246,6 +285,10 @@ VERB_REQUIREMENTS: dict[str, frozenset[Requirement]] = {
             Requirement.JOURNAL_REFLECT,
             Requirement.JOURNAL_DURING_WORK_AT_LEAST_ONE,
             Requirement.ACCEPTANCE_CRITERIA_ADDRESSED,
+            # The developer's dedicated section: obligated like the journal.
+            # Satisfiable because the dev pre-writes dev_notes via
+            # note(scope='handoff') before i_am_done (write-then-gate).
+            Requirement.DEV_NOTES_MIN_CHARS,
         }
     ),
     # QA pass/fail.
@@ -263,12 +306,23 @@ VERB_REQUIREMENTS: dict[str, frozenset[Requirement]] = {
             Requirement.JOURNAL_LEARNING,
         }
     ),
-    # PR reviewer posts its change-request — must record a learning entry first.
-    "post_pr_review": frozenset({Requirement.JOURNAL_LEARNING}),
+    # PR reviewer posts its change-request — must record a learning entry first,
+    # and fill its dedicated pr_reviewer_notes section. The section note is the
+    # verb's own argument (review body), so it is checked via a SimpleNamespace
+    # shim at the call site, not the persisted task (write-then-gate: the arg
+    # isn't on the task yet — same pattern as qa_notes).
+    "post_pr_review": frozenset(
+        {Requirement.JOURNAL_LEARNING, Requirement.PR_REVIEWER_NOTES_MIN_CHARS}
+    ),
     # In-path PR-review gate: the reviewer records a learning entry before
-    # passing or failing the assembled PR (parity with post_pr_review).
-    "pr_pass": frozenset({Requirement.JOURNAL_LEARNING}),
-    "pr_fail": frozenset({Requirement.JOURNAL_LEARNING}),
+    # passing or failing the assembled PR (parity with post_pr_review), and
+    # fills pr_reviewer_notes (the verb's notes/issues argument, shimmed).
+    "pr_pass": frozenset(
+        {Requirement.JOURNAL_LEARNING, Requirement.PR_REVIEWER_NOTES_MIN_CHARS}
+    ),
+    "pr_fail": frozenset(
+        {Requirement.JOURNAL_LEARNING, Requirement.PR_REVIEWER_NOTES_MIN_CHARS}
+    ),
     # Doc submit.
     "i_documented": frozenset(
         {

roboco/services/gateway/choreographer/_impl.py

@@ -45,7 +45,10 @@ from roboco.services.gateway.remediation import (
     hint_for_missing_progress,
     hint_for_missing_qa_notes,
     hint_for_missing_reflect,
+    hint_for_short_dev_notes,
     hint_for_short_doc_notes,
+    hint_for_short_pr_reviewer_notes,
+    hint_for_short_quick_context,
     hint_for_unaddressed_acceptance_criteria,
 )
 
@@ -1876,6 +1879,7 @@ class Choreographer:
         auto-verify path. It is filtered here and re-asserted by the spec
         action's own preconditions.
         """
+        from roboco.config import settings as _settings
         from roboco.foundation.policy import tracing as _tr
 
         has_reflect = await self.journal.has_reflect_for_task(agent_id, task_id)
@@ -1890,6 +1894,9 @@ class Choreographer:
             journal_learning_present=has_learning,
             journal_struggle_present=has_struggle,
             journal_during_work_count=during_work_count,
+            # DEV_NOTES_MIN_CHARS reads the persisted task.dev_notes — the dev
+            # pre-writes it via note(scope='handoff') before i_am_done.
+            dev_notes_min_chars=_settings.dev_notes_min_chars,
         )
         requirements: list[_tr.Requirement] = [
             r
@@ -2004,7 +2011,15 @@ class Choreographer:
             and (datetime.now(UTC) - latest).total_seconds() <= window_seconds
         )
 
-        ctx = _tr.GateContext(journal_decision_present=fresh)
+        # QUICK_CONTEXT_MIN_CHARS applies only to ``delegate`` (its
+        # required-set is the only one carrying it); the PM pre-writes the
+        # parent's quick_context via note(scope='handoff') before delegating.
+        # Setting the threshold here is inert for unblock / escalate, which do
+        # not require it.
+        ctx = _tr.GateContext(
+            journal_decision_present=fresh,
+            quick_context_min_chars=_settings.quick_context_min_chars,
+        )
         result = _tr.check_requirements(
             task=t,
             requirements=list(_tr.requirements_for(verb)),
@@ -2189,6 +2204,17 @@ class Choreographer:
                 min_chars=_roboco_settings.docs_notes_min_chars
             ),
             "docs_files_non_empty": hint_for_missing_doc_files(),
+            "dev_notes>=min": hint_for_short_dev_notes(
+                min_chars=getattr(_roboco_settings, "dev_notes_min_chars", 40),
+                task_id=tid,
+            ),
+            "pr_reviewer_notes>=min": hint_for_short_pr_reviewer_notes(
+                min_chars=getattr(_roboco_settings, "pr_reviewer_notes_min_chars", 40),
+            ),
+            "quick_context>=min": hint_for_short_quick_context(
+                min_chars=getattr(_roboco_settings, "quick_context_min_chars", 30),
+                task_id=tid,
+            ),
             "journal:note_at_claim": (
                 "pre-gateway parity P1: write a journal:note at claim. "
                 f"Call note(scope='note', task_id='{tid}', "
@@ -2969,22 +2995,21 @@ class Choreographer:
                 ),
                 context_briefing=briefing,
             )
-        if guard := await self._pending_assignment_guard(agent_id, briefing):
-            return await self._emit_rejection(
-                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
-            )
-        if guard := await self._pm_unfinished_review_guard(agent_id, briefing):
-            return await self._emit_rejection(
-                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
-            )
-        if guard := await self._pm_uncovered_decomposition_guard(agent_id, briefing):
-            return await self._emit_rejection(
-                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
-            )
-        if guard := await self._pm_uncovered_required_cells_guard(agent_id, briefing):
-            return await self._emit_rejection(
-                guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
-            )
+        # Pre-idle guards, evaluated in order — the first that returns an
+        # Envelope short-circuits to a rejection (kept as a loop so adding a
+        # guard doesn't push this verb over the return-count bound).
+        idle_guards = (
+            self._pending_assignment_guard,
+            self._pm_unfinished_review_guard,
+            self._pm_uncovered_decomposition_guard,
+            self._pm_uncovered_required_cells_guard,
+            self._auditor_note_guard,
+        )
+        for guard_fn in idle_guards:
+            if guard := await guard_fn(agent_id, briefing):
+                return await self._emit_rejection(
+                    guard, agent_id=agent_id, task_id=None, verb="i_am_idle"
+                )
         paused_ids = await self._auto_pause_in_progress_tasks(agent_id)
         await self.task.mark_agent_idle(agent_id)
         if paused_ids:
@@ -3108,6 +3133,47 @@ class Choreographer:
             context_briefing=briefing,
         )
 
+    # The auditor must have recorded an observation within this window before
+    # it may go idle (session-scoped note obligation; see _auditor_note_guard).
+    _AUDITOR_IDLE_NOTE_WINDOW_SECONDS: int = 3600
+
+    async def _auditor_note_guard(
+        self, agent_id: UUID, briefing: dict[str, Any]
+    ) -> Envelope | None:
+        """Refuse i_am_idle when the auditor has not recorded an observation.
+
+        Every role with a dedicated note section is obligated to populate it,
+        the same way journals are obligated. The auditor's section is
+        ``auditor_notes``, but it owns no delivery task and has no delivery
+        verb to hang the obligation on — so the obligation is session-scoped:
+        before going idle the auditor must have recorded an observation within
+        the last ``_AUDITOR_IDLE_NOTE_WINDOW_SECONDS`` (a reflect journal entry,
+        or a note(scope='handoff') section write — either leaves a journal
+        entry). Inert for every other role.
+        """
+        agent = await self.task.agent_for(agent_id)
+        if agent is None or str(agent.role) != "auditor":
+            return None
+        if await self.journal.has_recent_entry(
+            agent_id, self._AUDITOR_IDLE_NOTE_WINDOW_SECONDS
+        ):
+            return None
+        return Envelope.invalid_state(
+            message=(
+                "as the auditor you must record an observation before going "
+                "idle — your auditor_notes section is the artifact the company "
+                "reviews, so it cannot be left empty."
+            ),
+            remediate=(
+                "call note(scope='reflect', text='<what you observed and any "
+                "concern>') for a journal observation, or "
+                "note(scope='handoff', task_id='<task>', section={'summary': "
+                "'<observation>', 'severity': 'info'|'watch'|'risk'}) to fill a "
+                "task's auditor_notes — then retry i_am_idle()."
+            ),
+            context_briefing=briefing,
+        )
+
     async def _pm_uncovered_decomposition_guard(
         self, agent_id: UUID, briefing: dict[str, Any]
     ) -> Envelope | None:

roboco/services/gateway/choreographer/doc.py

@@ -272,13 +272,13 @@ class DocMixin(_Base):
         persisted to the task yet (the spec runner / verb body writes
         them via the atomic action / pre-dispatch stamp), so we thread
         them through a SimpleNamespace shim with the minimal attributes
-        the foundation checkers read off the task object (dev_notes +
+        the foundation checkers read off the task object (doc_notes +
         documents — see foundation.policy.tracing._check_docs_notes_min_chars
         and _check_docs_files_non_empty).
         """
         has_reflect = await self.journal.has_reflect_for_task(doc_agent_id, task_id)
         task_view = SimpleNamespace(
-            dev_notes=notes,
+            doc_notes=notes,
             documents=list(files),
         )
         ctx = _tr.GateContext(

roboco/services/gateway/choreographer/pr_gate.py

@@ -229,7 +229,9 @@ class PRGateMixin(_Base):
         if isinstance(pre, Envelope):
             return pre
         t, agent, role_str, briefing, spec_ctx = pre
-        gate = await self._gate_tracing(reviewer_agent_id, task_id, t, role_str, verb)
+        gate = await self._gate_tracing(
+            reviewer_agent_id, task_id, t, role_str, verb, notes=notes
+        )
         if gate is not None:
             return gate
         runner = self._verb_runner()
@@ -332,14 +334,28 @@ class PRGateMixin(_Base):
         t: Any,
         role_str: str,
         verb: str,
+        *,
+        notes: str,
     ) -> Envelope | None:
-        """pr_pass / pr_fail require a journal:learning entry (parity with QA)."""
+        """pr_pass / pr_fail require a journal:learning entry (parity with QA)
+        plus a substantive pr_reviewer_notes section.
+
+        The section note is the verb's own ``notes`` argument (the review
+        verdict / issues), not yet persisted to the task, so it is threaded
+        through a SimpleNamespace shim (the foundation checker reads
+        ``task.pr_reviewer_notes`` — same write-then-gate pattern as qa_notes).
+        """
+        from roboco.config import settings as _settings
+
         has_learning = await self.journal.has_learning_for_task(
             reviewer_agent_id, task_id
         )
-        ctx = _tr.GateContext(journal_learning_present=has_learning)
+        ctx = _tr.GateContext(
+            journal_learning_present=has_learning,
+            pr_reviewer_notes_min_chars=_settings.pr_reviewer_notes_min_chars,
+        )
         result = _tr.check_requirements(
-            task=SimpleNamespace(),
+            task=SimpleNamespace(pr_reviewer_notes=notes),
             requirements=list(_tr.requirements_for(verb)),
             ctx=ctx,
         )

roboco/services/gateway/choreographer/pr_review.py

@@ -293,7 +293,7 @@ class PRReviewerMixin(_Base):
                 verb="post_pr_review",
             )
         gate = await self._pr_review_tracing_gate(
-            reviewer_agent_id, task_id, t, role_str
+            reviewer_agent_id, task_id, t, role_str, body=body
         )
         if gate is not None:
             return gate
@@ -359,15 +359,33 @@ class PRReviewerMixin(_Base):
         }
 
     async def _pr_review_tracing_gate(
-        self, reviewer_agent_id: UUID, task_id: UUID, t: Any, role_str: str
+        self,
+        reviewer_agent_id: UUID,
+        task_id: UUID,
+        t: Any,
+        role_str: str,
+        *,
+        body: str,
     ) -> Envelope | None:
-        """post_pr_review requires a journal:learning entry (parity with QA)."""
+        """post_pr_review requires a journal:learning entry (parity with QA)
+        plus a substantive pr_reviewer_notes section.
+
+        The section note is the verb's own ``body`` argument (the change
+        request), not yet persisted to the task, so it is threaded through a
+        SimpleNamespace shim (the foundation checker reads
+        ``task.pr_reviewer_notes`` — same write-then-gate pattern as qa_notes).
+        """
+        from roboco.config import settings as _settings
+
         has_learning = await self.journal.has_learning_for_task(
             reviewer_agent_id, task_id
         )
-        ctx = _tr.GateContext(journal_learning_present=has_learning)
+        ctx = _tr.GateContext(
+            journal_learning_present=has_learning,
+            pr_reviewer_notes_min_chars=_settings.pr_reviewer_notes_min_chars,
+        )
         result = _tr.check_requirements(
-            task=SimpleNamespace(),
+            task=SimpleNamespace(pr_reviewer_notes=body),
             requirements=list(_tr.requirements_for("post_pr_review")),
             ctx=ctx,
         )

roboco/services/gateway/remediation.py

@@ -71,3 +71,29 @@ def hint_for_missing_doc_files() -> str:
         "i_documented(files=[...]) requires the list of doc-file paths you "
         "committed; pass at least one path"
     )
+
+
+def hint_for_short_dev_notes(*, min_chars: int, task_id: str) -> str:
+    return (
+        f"your dev_notes section is empty or under {min_chars} chars. Before "
+        f"i_am_done, call note(scope='handoff', task_id='{task_id}', "
+        "section={'summary': '<what you built, key changes, risks>'}) to fill "
+        "it, then retry."
+    )
+
+
+def hint_for_short_pr_reviewer_notes(*, min_chars: int) -> str:
+    return (
+        f"your review note must be at least {min_chars} chars stating what you "
+        "checked and the verdict rationale; pass a longer `body` (post_pr_review)"
+        " / `notes` (pr_pass) / `issues` (pr_fail)."
+    )
+
+
+def hint_for_short_quick_context(*, min_chars: int, task_id: str) -> str:
+    return (
+        f"your quick_context section is empty or under {min_chars} chars. Before "
+        f"delegate, call note(scope='handoff', task_id='{task_id}', "
+        "section={'done': '<state so far>', 'next': '<what the cell should do>'})"
+        " to leave a resumption handoff, then retry."
+    )

roboco/services/journal.py

@@ -6,7 +6,7 @@ Each agent has their own journal with entries tied to tasks and sessions.
 Integrates with the Optimal API for RAG indexing of entries.
 """
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any, ClassVar
 from uuid import UUID
 
@@ -861,6 +861,28 @@ class JournalService(BaseService):
             agent_id, task_id, JournalEntryType.STRUGGLE
         )
 
+    async def has_recent_entry(self, agent_id: UUID, within_seconds: int) -> bool:
+        """True iff the agent wrote any journal entry within the last
+        ``within_seconds`` (of any type, any task).
+
+        Backs the auditor's i_am_idle note obligation: the auditor has no
+        delivery verb and no single owned task, so its section-fill obligation
+        is session-scoped — it must have recorded an observation (a reflect
+        journal entry, or a note(scope='handoff') section write, which also
+        drops a journal trail entry) recently before going idle.
+        """
+        cutoff = datetime.now(UTC) - timedelta(seconds=within_seconds)
+        query = (
+            select(func.count(JournalEntryTable.id))
+            .join(JournalTable, JournalEntryTable.journal_id == JournalTable.id)
+            .where(
+                JournalTable.agent_id == agent_id,
+                JournalEntryTable.timestamp >= cutoff,
+            )
+        )
+        result = await self.session.execute(query)
+        return (result.scalar() or 0) > 0
+
     async def write_struggle(
         self,
         *,

tests/foundation/test_tracing.py

@@ -31,6 +31,9 @@ def test_requirement_enum_has_canonical_values() -> None:
         "self_verified",
         "notes>=min",
         "subtasks_terminal",
+        "dev_notes>=min",
+        "pr_reviewer_notes>=min",
+        "quick_context>=min",
     }
     actual = {r.value for r in tracing.Requirement}
     assert actual == expected, f"Requirement drift: {actual ^ expected}"
@@ -52,6 +55,51 @@ def test_gate_result_has_passed_and_missing() -> None:
     assert result.missing == []
 
 
+def test_docs_notes_checker_reads_doc_notes_not_dev_notes() -> None:
+    """Regression: _check_docs_notes_min_chars must read ``doc_notes`` (the
+    documenter's section), not ``dev_notes`` (the developer's)."""
+    ctx = tracing.GateContext(docs_notes_min_chars=20)
+    # A long dev_notes must NOT satisfy the docs requirement.
+    only_dev = SimpleNamespace(dev_notes="x" * 50, doc_notes="")
+    assert tracing._check_docs_notes_min_chars(only_dev, ctx) == ["docs_notes>=min"]
+    # A long doc_notes satisfies it.
+    has_doc = SimpleNamespace(dev_notes="", doc_notes="y" * 25)
+    assert tracing._check_docs_notes_min_chars(has_doc, ctx) == []
+
+
+def test_note_section_checkers_read_their_own_fields() -> None:
+    ctx = tracing.GateContext()  # defaults: dev 40, pr_reviewer 40, quick_context 30
+    assert (
+        tracing._check_dev_notes_min_chars(SimpleNamespace(dev_notes="z" * 40), ctx)
+        == []
+    )
+    assert tracing._check_dev_notes_min_chars(
+        SimpleNamespace(dev_notes="z" * 39), ctx
+    ) == ["dev_notes>=min"]
+    assert (
+        tracing._check_pr_reviewer_notes_min_chars(
+            SimpleNamespace(pr_reviewer_notes="z" * 40), ctx
+        )
+        == []
+    )
+    assert tracing._check_quick_context_min_chars(
+        SimpleNamespace(quick_context="z" * 29), ctx
+    ) == ["quick_context>=min"]
+
+
+def test_note_section_obligations_wired_to_verbs() -> None:
+    assert tracing.Requirement.DEV_NOTES_MIN_CHARS in tracing.requirements_for(
+        "i_am_done"
+    )
+    assert tracing.Requirement.QUICK_CONTEXT_MIN_CHARS in tracing.requirements_for(
+        "delegate"
+    )
+    for verb in ("pr_pass", "pr_fail", "post_pr_review"):
+        assert tracing.Requirement.PR_REVIEWER_NOTES_MIN_CHARS in (
+            tracing.requirements_for(verb)
+        )
+
+
 def test_check_requirements_passes_when_all_satisfied() -> None:
     task = SimpleNamespace(
         plan={"x": 1},

tests/integration/test_full_lifecycle_real_db.py

@@ -424,6 +424,14 @@ async def test_dev_full_chain_through_awaiting_qa(
     assert refreshed is not None
     assert refreshed.pr_number == _PR_NUMBER, "PR recorded on task"
 
+    # i_am_done obligates the developer's dev_notes section — the agent fills
+    # it via note(scope='handoff') first; record_section_note is that write.
+    await task_service.record_section_note(
+        task.id,
+        "developer",
+        {"summary": "Implemented /healthz and added a test for the happy path."},
+    )
+
     # 4. i_am_done — auto-runs in_progress → verifying → awaiting_qa.
     env = await c.i_am_done(dev_agent.id, task.id, "tests pass; route works")
     assert env.error is None, f"i_am_done failed: {env.message}"
@@ -482,6 +490,12 @@ async def test_full_chain_through_doc_handoff(
     )
     await task_service.add_progress(task.id, dev_agent.id, "implemented /healthz")
     await c.open_pr(dev_agent.id, task.id)
+    # i_am_done obligates the developer's dev_notes section (note(scope='handoff')).
+    await task_service.record_section_note(
+        task.id,
+        "developer",
+        {"summary": "Implemented /healthz and added a test for the happy path."},
+    )
     env = await c.i_am_done(dev_agent.id, task.id, "tests pass; route works")
     assert env.error is None
     assert env.status == "awaiting_qa"

tests/integration/test_journal_service.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import uuid
+from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any
 from unittest.mock import AsyncMock as _AsyncMock
 from unittest.mock import MagicMock as _MagicMock
@@ -11,7 +12,13 @@ from uuid import uuid4 as _u
 
 import pytest
 import pytest_asyncio
-from roboco.db.tables import AgentTable, JournalTable, ProjectTable, TaskTable
+from roboco.db.tables import (
+    AgentTable,
+    JournalEntryTable,
+    JournalTable,
+    ProjectTable,
+    TaskTable,
+)
 from roboco.models import AgentRole, AgentStatus, Team
 from roboco.models.base import (
     JournalEntryType,
@@ -29,7 +36,7 @@ from roboco.models.journal import (
     TaskReflectionParams,
 )
 from roboco.services.journal import JournalService
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError as _IE
 
 if TYPE_CHECKING:
@@ -730,3 +737,52 @@ async def test_search_entries_swallows_exception(
 
     results = await svc.search_entries(aid, "query")
     assert results == []
+
+
+@pytest.mark.asyncio
+async def test_has_recent_entry_false_then_true(journal_setup: dict) -> None:
+    """No entries → False; a freshly written entry is within the window."""
+    svc: JournalService = journal_setup["svc"]
+    agent_id = journal_setup["agent_id"]
+    task_id = journal_setup["task_id"]
+
+    assert await svc.has_recent_entry(agent_id, 3600) is False
+
+    await svc.write_entry(
+        agent_id=agent_id,
+        title="observation",
+        content="watching the seam between FE and BE",
+        scope="reflect",
+        task_id=task_id,
+    )
+    assert await svc.has_recent_entry(agent_id, 3600) is True
+
+
+@pytest.mark.asyncio
+async def test_has_recent_entry_excludes_entries_outside_window(
+    journal_setup: dict, db_session: AsyncSession
+) -> None:
+    """An entry older than the window does not count as recent."""
+    svc: JournalService = journal_setup["svc"]
+    agent_id = journal_setup["agent_id"]
+    task_id = journal_setup["task_id"]
+
+    await svc.write_entry(
+        agent_id=agent_id,
+        title="stale observation",
+        content="recorded two hours ago",
+        scope="reflect",
+        task_id=task_id,
+    )
+    # Backdate every entry on this agent's journal to two hours ago.
+    journal = await svc.get_journal_by_agent(agent_id)
+    assert journal is not None
+    await db_session.execute(
+        update(JournalEntryTable)
+        .where(JournalEntryTable.journal_id == journal.id)
+        .values(timestamp=datetime.now(UTC) - timedelta(hours=2))
+    )
+    await db_session.flush()
+
+    assert await svc.has_recent_entry(agent_id, 3600) is False
+    assert await svc.has_recent_entry(agent_id, 3 * 3600) is True

tests/integration/test_lifecycle_real_db.py

@@ -437,6 +437,15 @@ async def test_dev_full_chain_through_awaiting_qa(
     env = await c.open_pr(dev_agent.id, task.id)
     assert env.error is None, f"open_pr failed: {env.message}"
 
+    # i_am_done now obligates the developer's dev_notes section — the agent
+    # fills it via note(scope='handoff') first. record_section_note is the
+    # service call that write-path makes.
+    await task_service.record_section_note(
+        task.id,
+        "developer",
+        {"summary": "Implemented /healthz and added a test for the happy path."},
+    )
+
     env = await c.i_am_done(dev_agent.id, task.id, "tests pass; route works")
     assert env.error is None, f"i_am_done failed: {env.message}"
     assert env.status == Status.AWAITING_QA.value

tests/integration/v1/test_full_pending_to_completed.py

@@ -173,18 +173,14 @@ class _MockContentActions:
         *,
         agent_id: object,
         text: object,
-        scope: str = "note",
-        task_id: object = None,
-        structured: object = None,
+        **_kwargs: object,
     ) -> Envelope:
-        # `structured` mirrors the Wave 2 G4 production signature (panel
-        # decision/reflect fields). The mock ignores it — the test asserts
-        # lifecycle transitions, not journal-entry rendering.
+        # The route forwards scope / task_id / structured / section as keyword
+        # args; the stub absorbs them via **_kwargs (it asserts lifecycle
+        # transitions, not journal-entry / section rendering). ``section``
+        # mirrors the note(scope='handoff') write-path signature.
         _ = agent_id
         _ = text
-        _ = scope
-        _ = task_id
-        _ = structured
         return Envelope.ok(status="noted", task_id=None, next="continue")
 
 

tests/unit/gateway/test_acceptance_criteria_status_writer.py

@@ -75,7 +75,8 @@ def _ready_task(task_id: Any, agent_id: Any) -> MagicMock:
         acceptance_criteria_status=[],
         commits=[{"sha": "deadbeef"}],
         documents=[],
-        dev_notes="",
+        # i_am_done obligates the developer's dev_notes section (>=40 chars).
+        dev_notes="Implemented the change and added tests covering the new path.",
         quick_context=None,
     )
 

tests/unit/gateway/test_auditor_idle_note_guard.py

@@ -0,0 +1,81 @@
+"""Tests for the auditor's i_am_idle note obligation (_auditor_note_guard).
+
+Every role with a dedicated note section is obligated to populate it, like
+journals. The auditor owns no delivery task and has no delivery verb, so its
+obligation is session-scoped: it must have recorded an observation within the
+window before it may go idle. Inert for every other role.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    return ChoreographerDeps(**base)
+
+
+@pytest.mark.asyncio
+async def test_auditor_idle_blocked_without_recent_observation() -> None:
+    """An auditor with no recent journal entry is refused idle."""
+    auditor_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.agent_for.return_value = MagicMock(role="auditor", team="board")
+    journal = AsyncMock()
+    journal.has_recent_entry.return_value = False
+    c = Choreographer(_make_deps(task=task_svc, journal=journal))
+
+    guard = await c._auditor_note_guard(auditor_id, briefing={})
+
+    assert guard is not None
+    body = guard.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "auditor_notes" in body["message"]
+    assert "note(scope='reflect'" in body["remediate"]
+    # The window query was actually consulted.
+    journal.has_recent_entry.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_auditor_idle_allowed_with_recent_observation() -> None:
+    """An auditor that recorded an observation recently may idle."""
+    auditor_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.agent_for.return_value = MagicMock(role="auditor", team="board")
+    journal = AsyncMock()
+    journal.has_recent_entry.return_value = True
+    c = Choreographer(_make_deps(task=task_svc, journal=journal))
+
+    guard = await c._auditor_note_guard(auditor_id, briefing={})
+
+    assert guard is None
+
+
+@pytest.mark.asyncio
+async def test_idle_note_guard_inert_for_non_auditor() -> None:
+    """A developer never trips the auditor guard (no journal lookup at all)."""
+    dev_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.agent_for.return_value = MagicMock(role="developer", team="backend")
+    journal = AsyncMock()
+    c = Choreographer(_make_deps(task=task_svc, journal=journal))
+
+    guard = await c._auditor_note_guard(dev_id, briefing={})
+
+    assert guard is None
+    journal.has_recent_entry.assert_not_awaited()

tests/unit/gateway/test_choreographer_delegate_guards.py

@@ -107,6 +107,7 @@ async def test_delegate_blocks_when_parent_assigned_to_other_agent() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=other_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -131,6 +132,7 @@ async def test_delegate_allows_when_parent_in_progress_and_owned() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -158,6 +160,7 @@ async def test_delegate_blocks_when_subtask_cap_exceeded() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     too_many = [MagicMock(id=uuid4()) for _ in range(13)]
     task_svc = AsyncMock()
@@ -184,6 +187,7 @@ async def test_delegate_allows_when_subtask_cap_within_soft_zone() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     many = [MagicMock(id=uuid4()) for _ in range(10)]
     new_task = MagicMock(id=uuid4())
@@ -211,6 +215,7 @@ async def test_delegate_allows_at_zero_subtasks() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -236,6 +241,7 @@ async def test_delegate_blocks_at_exact_cap_plus_one() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     # Already 12 children — adding the 13th must be blocked.
     twelve = [MagicMock(id=uuid4()) for _ in range(12)]
@@ -250,3 +256,31 @@ async def test_delegate_blocks_at_exact_cap_plus_one() -> None:
     body = env.as_dict()
     assert body["error"] == "invalid_state"
     task_svc.create_subtask.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_delegate_blocks_when_parent_quick_context_empty() -> None:
+    """delegate obligates the PM's quick_context resumption section on the
+    parent; an empty quick_context (PM never called note(scope='handoff'))
+    fails the tracing gate before any subtask is created."""
+    pm_id = uuid4()
+    parent_id = uuid4()
+    parent = MagicMock(
+        id=parent_id,
+        project_id=uuid4(),
+        status="in_progress",
+        assigned_to=pm_id,
+        quick_context="",
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = parent
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.get_subtasks.return_value = []
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.delegate(pm_id, parent_id, _delegate_inputs())
+    body = env.as_dict()
+    assert body["error"] == "tracing_gap"
+    assert "quick_context>=min" in body["missing"]
+    task_svc.create_subtask.assert_not_awaited()

tests/unit/gateway/test_choreographer_dev.py

@@ -428,7 +428,7 @@ async def test_i_am_done_blocks_when_acceptance_criteria_unaddressed() -> None:
         pr_url="https://x/pr/8",
         team="backend",
         documents=[],
-        dev_notes="",
+        dev_notes="Implemented the change and added tests covering the new path.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = t
@@ -480,7 +480,7 @@ async def test_i_am_done_reflect_note_addresses_acceptance_criteria() -> None:
         pr_url="https://x/pr/8",
         team="backend",
         documents=[],
-        dev_notes="",
+        dev_notes="Implemented the change and added tests covering the new path.",
         qa_notes="",
     )
     task_svc = AsyncMock()
@@ -534,7 +534,7 @@ async def test_i_am_done_blocks_when_journal_reflect_missing() -> None:
         pr_url="https://x/pr/8",
         team="backend",
         documents=[],
-        dev_notes="",
+        dev_notes="Implemented the change and added tests covering the new path.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = t
@@ -717,7 +717,7 @@ def _passing_i_am_done_task(agent_id: Any, task_id: Any) -> Any:
         pr_url="https://x/pr/8",
         team="backend",
         documents=[],
-        dev_notes="",
+        dev_notes="Implemented the change and added tests covering the new path.",
         qa_notes="",
     )
 

tests/unit/gateway/test_choreographer_impl_branches.py

@@ -618,6 +618,9 @@ async def test_delegate_parent_no_project_rejected() -> None:
         project_id=None,
         product_id=None,
         title="p",
+        # delegate obligates the PM's quick_context; supply it so the
+        # no-project guard is the load-bearing rejection.
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent

tests/unit/gateway/test_choreographer_pm_extras.py

@@ -672,6 +672,7 @@ async def test_delegate_main_pm_to_cell_pm_creates_subtask() -> None:
         project_id=project_id,
         status="in_progress",
         assigned_to=main_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -713,6 +714,7 @@ async def test_delegate_cell_pm_to_team_dev_creates_subtask() -> None:
         project_id=project_id,
         status="in_progress",
         assigned_to=cell_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()
@@ -749,6 +751,7 @@ async def test_delegate_main_pm_to_dev_is_rejected() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=main_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -784,6 +787,7 @@ async def test_delegate_cell_pm_to_other_pm_rejected() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=cell_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -818,6 +822,7 @@ async def test_delegate_unknown_assignee_returns_invalid_state() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -852,6 +857,7 @@ async def test_delegate_invalid_team_enum_rejected() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -1128,6 +1134,7 @@ async def test_delegate_main_pm_to_cell_pm_rejects_code_typed_subtask() -> None:
         project_id=uuid4(),
         status="in_progress",
         assigned_to=main_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     task_svc = AsyncMock()
     task_svc.get.return_value = parent
@@ -1166,6 +1173,7 @@ async def test_delegate_main_pm_to_cell_pm_accepts_planning_subtask() -> None:
         project_id=project_id,
         status="in_progress",
         assigned_to=main_pm_id,
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
     new_task = MagicMock(id=uuid4())
     task_svc = AsyncMock()

tests/unit/gateway/test_choreographer_reassignment.py

@@ -97,7 +97,8 @@ async def test_i_am_done_reassigns_task_to_qa_agent() -> None:
         # Gate Set E requires non-empty commits before submit_qa.
         commits=[{"sha": "abc"}],
         documents=[],
-        dev_notes="",
+        # i_am_done obligates the developer's dev_notes section (>=40 chars).
+        dev_notes="Implemented the change and added tests covering the new path.",
     )
     after_verify = MagicMock(
         **{**initial.__dict__, "status": "verifying", "self_verified": True},
@@ -162,7 +163,8 @@ async def test_i_am_done_skips_reassign_when_no_qa_agent() -> None:
         # Gate Set E requires non-empty commits before submit_qa.
         commits=[{"sha": "abc"}],
         documents=[],
-        dev_notes="",
+        # i_am_done obligates the developer's dev_notes section (>=40 chars).
+        dev_notes="Implemented the change and added tests covering the new path.",
     )
     after_verify = MagicMock(
         **{**initial.__dict__, "status": "verifying", "self_verified": True},

tests/unit/gateway/test_choreographer_submit_qa_gates.py

@@ -92,7 +92,8 @@ def _ready_task(task_id: Any, agent_id: Any) -> MagicMock:
         ],
         commits=[{"sha": "abc"}],
         documents=[],
-        dev_notes="",
+        # i_am_done obligates the developer's dev_notes section (>=40 chars).
+        dev_notes="Implemented the change and added tests covering the new path.",
     )
 
 
@@ -265,6 +266,41 @@ async def test_i_am_done_blocks_when_no_progress() -> None:
     task_svc.submit_qa.assert_not_awaited()
 
 
+# ---------------------------------------------------------------------------
+# Role note-section obligation: dev_notes must be filled (note(scope='handoff'))
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_i_am_done_blocks_when_dev_notes_empty() -> None:
+    """i_am_done obligates the developer's dev_notes section; an empty
+    dev_notes (the dev never called note(scope='handoff')) fails the gate."""
+    agent_id = uuid4()
+    task_id = uuid4()
+    t = _ready_task(task_id, agent_id)
+    t.dev_notes = ""
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(
+        id=agent_id, role="developer", team="backend", slug=None
+    )
+    journal_svc = AsyncMock()
+    journal_svc.has_reflect_for_task.return_value = True
+    journal_svc.has_decision_for_task.return_value = True
+    journal_svc.latest_decision_at.return_value = datetime.now(UTC)
+    journal_svc.has_learning_for_task.return_value = False
+    journal_svc.has_struggle_for_task.return_value = False
+    deps = _make_deps(task=task_svc, journal=journal_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_am_done(agent_id, task_id, "done")
+    body = env.as_dict()
+    assert body["error"] == "tracing_gap"
+    assert "dev_notes>=min" in body["missing"]
+    assert "scope='handoff'" in body["remediate"]
+    task_svc.submit_qa.assert_not_awaited()
+
+
 # ---------------------------------------------------------------------------
 # E.5 happy path: all gates pass → submit_qa runs (NO catch-up)
 # ---------------------------------------------------------------------------

tests/unit/gateway/test_delegate_incomplete_input.py

@@ -62,6 +62,8 @@ def _parent_in_progress(pm_id: Any) -> MagicMock:
         status="in_progress",
         assigned_to=pm_id,
         priority=2,
+        # delegate obligates the PM's quick_context resumption section.
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
 
 

tests/unit/gateway/test_delegate_project_routing.py

@@ -48,6 +48,8 @@ def _parent(pm_id: Any, product_id: Any = None, project_id: Any = None) -> Magic
         product_id=product_id,
         status="in_progress",
         assigned_to=pm_id,
+        # delegate obligates the PM's quick_context resumption section.
+        quick_context="Decomposition planned; cells implement their slice next.",
     )
 
 

tests/unit/gateway/test_note_handoff.py

@@ -0,0 +1,167 @@
+"""Tests for note(scope='handoff') — the role note-section write-path.
+
+``note()`` only ever wrote the JOURNAL; ``scope='handoff'`` is how an agent
+authors its dedicated SECTION (dev_notes / quick_context / auditor_notes …)
+through the ``apply_structured_note`` chokepoint. These cover the routing,
+ownership, validation→remediation, and journal-trail behaviours.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.foundation.policy.content import ContentValidationError
+from roboco.services.gateway.content_actions import ContentActions, ContentActionsDeps
+
+
+def _make_deps(**overrides: object) -> ContentActionsDeps:
+    task = overrides.get("task") or AsyncMock()
+    return ContentActionsDeps(
+        task=task,
+        git=overrides.get("git") or AsyncMock(),
+        messaging=overrides.get("messaging") or AsyncMock(),
+        a2a=overrides.get("a2a") or AsyncMock(),
+        journal=overrides.get("journal") or AsyncMock(),
+        workspace=overrides.get("workspace") or AsyncMock(),
+        notifications=overrides.get("notifications") or AsyncMock(),
+        notification_delivery=overrides.get("notification_delivery") or AsyncMock(),
+        evidence_repo=overrides.get("evidence_repo") or AsyncMock(),
+    )
+
+
+def _dev_task_svc(task_id: object, role: str = "developer") -> AsyncMock:
+    """A task service whose active/context task is owned by the caller."""
+    svc = AsyncMock()
+    svc.agent_for.return_value = MagicMock(role=role)
+    svc.get_journal_context_task_for_agent.return_value = MagicMock(id=task_id)
+    svc.record_section_note.return_value = None
+    return svc
+
+
+@pytest.mark.asyncio
+async def test_handoff_developer_writes_dev_notes_from_text() -> None:
+    """A developer handoff routes to the 'developer' content type, defaulting
+    the payload to {'summary': text} when no explicit section is given."""
+    agent_id, task_id = uuid4(), uuid4()
+    svc = _dev_task_svc(task_id)
+    ca = ContentActions(_make_deps(task=svc))
+
+    summary = "Implemented the endpoint and added happy-path tests."
+    env = await ca.note(agent_id=agent_id, text=summary, scope="handoff")
+
+    assert env.as_dict()["error"] is None
+    svc.record_section_note.assert_awaited_once()
+    called_task_id, content_type, payload = svc.record_section_note.call_args.args
+    assert called_task_id == task_id
+    assert content_type == "developer"
+    assert payload == {"summary": summary}
+
+
+@pytest.mark.asyncio
+async def test_handoff_passes_explicit_section_through() -> None:
+    """An explicit ``section`` dict is the payload (e.g. PM resumption)."""
+    agent_id, task_id = uuid4(), uuid4()
+    svc = _dev_task_svc(task_id, role="cell_pm")
+    ca = ContentActions(_make_deps(task=svc))
+
+    section = {"done": "Planned the decomposition.", "next": "Cells implement."}
+    env = await ca.note(
+        agent_id=agent_id, text="handoff", scope="handoff", section=section
+    )
+
+    assert env.as_dict()["error"] is None
+    _tid, content_type, payload = svc.record_section_note.call_args.args
+    assert content_type == "resumption"
+    assert payload == section
+
+
+@pytest.mark.asyncio
+async def test_handoff_also_writes_journal_trail_entry() -> None:
+    """The section write drops a journal trail entry so it shows in the log."""
+    agent_id, task_id = uuid4(), uuid4()
+    svc = _dev_task_svc(task_id)
+    journal = AsyncMock()
+    ca = ContentActions(_make_deps(task=svc, journal=journal))
+
+    await ca.note(agent_id=agent_id, text="Did the thing thoroughly.", scope="handoff")
+
+    journal.write_entry.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_handoff_role_without_section_is_rejected() -> None:
+    """A role with no dedicated section (board/advisory) cannot handoff."""
+    agent_id = uuid4()
+    svc = AsyncMock()
+    svc.agent_for.return_value = MagicMock(role="product_owner")
+    ca = ContentActions(_make_deps(task=svc))
+
+    env = await ca.note(agent_id=agent_id, text="observation", scope="handoff")
+    body = env.as_dict()
+
+    assert body["error"] == "invalid_state"
+    assert "no dedicated note section" in body["message"]
+    svc.record_section_note.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handoff_validation_error_returns_remediation_not_422() -> None:
+    """A malformed section payload becomes a remediation Envelope, never a 422
+    (a raw 422 would trip the do-server circuit breaker)."""
+    agent_id, task_id = uuid4(), uuid4()
+    svc = _dev_task_svc(task_id, role="auditor")
+    svc.record_section_note.side_effect = ContentValidationError(
+        "severity", "field required"
+    )
+    ca = ContentActions(_make_deps(task=svc))
+
+    env = await ca.note(
+        agent_id=agent_id,
+        text="risk spotted",
+        scope="handoff",
+        section={"summary": "x"},
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "invalid_state"
+    assert "severity" in body["message"]
+    assert "auditor" in body["remediate"]
+
+
+@pytest.mark.asyncio
+async def test_handoff_no_task_to_attach_is_rejected() -> None:
+    """With no active/context task and no explicit task_id, handoff refuses."""
+    agent_id = uuid4()
+    svc = AsyncMock()
+    svc.agent_for.return_value = MagicMock(role="developer")
+    svc.get_journal_context_task_for_agent.return_value = None
+    ca = ContentActions(_make_deps(task=svc))
+
+    env = await ca.note(agent_id=agent_id, text="orphan note", scope="handoff")
+    body = env.as_dict()
+
+    assert body["error"] == "invalid_state"
+    assert "task_id" in body["remediate"]
+    svc.record_section_note.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handoff_explicit_task_not_owned_is_rejected() -> None:
+    """An explicit task_id the caller does not own is an ownership violation."""
+    agent_id, task_id = uuid4(), uuid4()
+    svc = AsyncMock()
+    svc.agent_for.return_value = MagicMock(role="developer")
+    svc.get.return_value = MagicMock(
+        id=task_id, assigned_to=uuid4(), project_id=uuid4(), product_id=None
+    )
+    ca = ContentActions(_make_deps(task=svc))
+
+    env = await ca.note(
+        agent_id=agent_id, text="poking another task", scope="handoff", task_id=task_id
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    svc.record_section_note.assert_not_awaited()

tests/unit/gateway/test_pm_decision_window.py

@@ -89,10 +89,16 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
 def _make_task(task_id: Any) -> Any:
     """A task stub that the tracing gate accepts as-is.
 
-    `_check_pm_decision_required` only consults the (agent, task) journal
-    lookup — the task object itself is opaque to that check.
+    `_check_pm_decision_required` consults the (agent, task) journal lookup
+    for the decision window and, for ``delegate``, the persisted
+    ``quick_context`` resumption section — so the stub carries a substantive
+    quick_context to satisfy that obligation.
     """
-    return MagicMock(id=task_id, status="in_progress")
+    return MagicMock(
+        id=task_id,
+        status="in_progress",
+        quick_context="Decomposition planned; cells implement their slice next.",
+    )
 
 
 # ---------------------------------------------------------------------------

tests/unit/gateway/test_pr_reviewer_notes_obligation.py

@@ -0,0 +1,86 @@
+"""The PR reviewer's pr_reviewer_notes section is obligated on its verbs.
+
+pr_pass / pr_fail (in-path gate) and post_pr_review (inbound external PR) each
+require a substantive review note. The note is the verb's own argument (not yet
+persisted), so it is checked through a SimpleNamespace shim against the
+``pr_reviewer_notes`` field — these cover both the short-circuit (too short) and
+the pass-through (long enough) at the tracing-gate helpers.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+
+def _make_choreographer(*, has_learning: bool = True) -> Choreographer:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base["journal"].has_learning_for_task.return_value = has_learning
+    return Choreographer(ChoreographerDeps(**base))
+
+
+_LONG = "Reviewed the assembled diff end to end; the seam contract holds."
+_SHORT = "looks ok"
+
+
+@pytest.mark.asyncio
+async def test_gate_tracing_blocks_on_short_notes() -> None:
+    c = _make_choreographer()
+    env = await c._gate_tracing(
+        uuid4(), uuid4(), MagicMock(), "pr_reviewer", "pr_pass", notes=_SHORT
+    )
+    assert env is not None
+    body = env.as_dict()
+    assert body["error"] == "tracing_gap"
+    assert "pr_reviewer_notes>=min" in body["missing"]
+
+
+@pytest.mark.asyncio
+async def test_gate_tracing_passes_on_substantive_notes() -> None:
+    c = _make_choreographer()
+    env = await c._gate_tracing(
+        uuid4(), uuid4(), MagicMock(), "pr_reviewer", "pr_fail", notes=_LONG
+    )
+    assert env is None
+
+
+@pytest.mark.asyncio
+async def test_post_pr_review_tracing_blocks_on_short_body() -> None:
+    c = _make_choreographer()
+    env = await c._pr_review_tracing_gate(
+        uuid4(), uuid4(), MagicMock(), "pr_reviewer", body=_SHORT
+    )
+    assert env is not None
+    assert "pr_reviewer_notes>=min" in env.as_dict()["missing"]
+
+
+@pytest.mark.asyncio
+async def test_post_pr_review_tracing_passes_on_substantive_body() -> None:
+    c = _make_choreographer()
+    env = await c._pr_review_tracing_gate(
+        uuid4(), uuid4(), MagicMock(), "pr_reviewer", body=_LONG
+    )
+    assert env is None
+
+
+@pytest.mark.asyncio
+async def test_missing_learning_still_blocks_even_with_long_notes() -> None:
+    """The journal:learning requirement remains independent of the note."""
+    c = _make_choreographer(has_learning=False)
+    env = await c._gate_tracing(
+        uuid4(), uuid4(), MagicMock(), "pr_reviewer", "pr_pass", notes=_LONG
+    )
+    assert env is not None
+    assert "journal:learning" in env.as_dict()["missing"]

tests/unit/services/test_content_notes.py

@@ -6,7 +6,7 @@ from types import SimpleNamespace
 
 import pytest
 from roboco.foundation.policy.content import ContentValidationError, PrReviewContent
-from roboco.services.content_notes import apply_structured_note
+from roboco.services.content_notes import apply_structured_note, content_type_for_role
 
 
 def _task() -> SimpleNamespace:
@@ -95,3 +95,22 @@ def test_notes_structured_reassigned_for_dirty_tracking() -> None:
     assert t.notes_structured is not before  # new dict object
     assert "developer" in t.notes_structured  # prior entry preserved
     assert "doc" in t.notes_structured
+
+
+def test_content_type_for_role_maps_section_roles() -> None:
+    """Each role with a dedicated section routes to its content type."""
+    assert content_type_for_role("developer") == "developer"
+    assert content_type_for_role("qa") == "qa"
+    assert content_type_for_role("documenter") == "doc"
+    assert content_type_for_role("pr_reviewer") == "pr_review"
+    assert content_type_for_role("auditor") == "auditor"
+    assert content_type_for_role("cell_pm") == "resumption"
+    assert content_type_for_role("main_pm") == "resumption"
+
+
+def test_content_type_for_role_none_for_sectionless_roles() -> None:
+    """Board / advisory / on-demand roles have no dedicated section."""
+    assert content_type_for_role("product_owner") is None
+    assert content_type_for_role("head_marketing") is None
+    assert content_type_for_role("ceo") is None
+    assert content_type_for_role("prompter") is None