feat(content): extend anti-soup guard to pitch title, open_session topic, pr_update, note narratives

The universal guard covered say/dm/note-text/progress/notify/pitch
problem+solution. Close the remaining content-tool fields the agent
authors: pitch title, open_session topic, pr_update title/body (when
supplied), and the decision/reflect narrative sub-fields of note
(rationale/context/what_done/...). Narrative fields are only checked
when the agent fills them — an omitted field keeps its tolerant
'(not provided)' placeholder so a thin note still records and never
trips the do-server circuit breaker. Fold pr_update's no-fields + soup
checks into one helper to stay under the return-count bound.

roboco/services/gateway/content_actions.py

@@ -354,6 +354,50 @@ class ContentActions:
             )
         return None
 
+    @classmethod
+    def _reject_structured_soup(
+        cls, scope: str, structured: dict[str, Any] | None
+    ) -> Envelope | None:
+        """Soup-guard the scope's narrative sub-fields when the agent fills them.
+
+        Only *provided, non-empty* fields are checked — an omitted narrative
+        field keeps its tolerant ``(not provided)`` placeholder default (so a
+        thin note is never hard-rejected, preserving the do-server breaker
+        contract), but ``rationale='asdf'`` is soup and lands nowhere.
+        """
+        for field in _SCOPE_NARRATIVE_FIELDS.get(scope, ()):
+            value = (structured or {}).get(field)
+            if not (value and str(value).strip()):
+                continue
+            if rej := cls._reject_soup(str(value), field=field, min_chars=4):
+                return rej
+        return None
+
+    @classmethod
+    def _pr_update_input_check(
+        cls, title: str | None, body: str | None, reviewers: list[str] | None
+    ) -> Envelope | None:
+        """At-least-one-field + anti-soup gate for ``pr_update`` inputs.
+
+        Folds the no-op guard and the title/body soup guard into one call so
+        the verb body keeps its return count under the complexity bound.
+        """
+        if title is None and body is None and reviewers is None:
+            return Envelope.invalid_state(
+                message="no fields to update",
+                remediate=(
+                    "provide at least one of title, body, or reviewers; "
+                    "passing all None has no effect"
+                ),
+                context_briefing={},
+            )
+        for _pf, _pv, _min in (("title", title, 8), ("body", body, 15)):
+            if _pv is not None and (
+                rej := cls._reject_soup(_pv, field=_pf, min_chars=_min)
+            ):
+                return rej
+        return None
+
     async def commit(
         self,
         *,
@@ -516,6 +560,8 @@ class ContentActions:
                 remediate=f"scope must be one of: {sorted(_VALID_NOTE_SCOPES)}",
                 context_briefing={},
             )
+        if rej := self._reject_structured_soup(scope, structured):
+            return rej
         if task_id is not None:
             if reject := await self._verify_explicit_task_ownership(agent_id, task_id):
                 return reject
@@ -562,11 +608,12 @@ class ContentActions:
         proposal. On CEO approval the system provisions a repo per target cell,
         registers the projects, and seeds the first Main-PM task.
         """
-        for _pf, _pv in (
-            ("problem", problem),
-            ("proposed_solution", proposed_solution),
+        for _pf, _pv, _min in (
+            ("title", title, 5),
+            ("problem", problem, 15),
+            ("proposed_solution", proposed_solution, 15),
         ):
-            if rej := self._reject_soup(_pv, field=_pf, min_chars=15):
+            if rej := self._reject_soup(_pv, field=_pf, min_chars=_min):
                 return rej
         from pydantic import ValidationError as PydanticValidationError
 
@@ -1066,6 +1113,8 @@ class ContentActions:
         already has a primary session in the same channel, it reuses
         that session instead of opening a new one.
         """
+        if rej := self._reject_soup(topic, field="topic", min_chars=5):
+            return rej
         from roboco.models.session import (
             SessionForTasksCreate,
             SessionTaskRelationshipType,
@@ -1320,15 +1369,8 @@ class ContentActions:
             invalid_state — schema-level check is the first line of
             defense; this guard catches direct gateway calls)
         """
-        if title is None and body is None and reviewers is None:
-            return Envelope.invalid_state(
-                message="no fields to update",
-                remediate=(
-                    "provide at least one of title, body, or reviewers; "
-                    "passing all None has no effect"
-                ),
-                context_briefing={},
-            )
+        if rej := self._pr_update_input_check(title, body, reviewers):
+            return rej
         t = await self.task.get(task_id)
         if t is None:
             return Envelope.not_found(message=f"task {task_id} not found")

tests/unit/gateway/test_content_actions.py

@@ -874,3 +874,105 @@ async def test_commit_gate_reads_settings_banned_words(
 
     env = await ca.commit(agent_id=uuid4(), message="bananaword")
     assert env.as_dict()["error"] == "invalid_state"
+
+
+# ---------------------------------------------------------------------------
+# Universal anti-soup guard — content-tool fields beyond say/dm/note/notify
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_pitch_placeholder_title_rejected_before_role_check() -> None:
+    """A soupy pitch title lands nowhere — rejected before the Board role gate."""
+    deps = _make_deps()
+    ca = ContentActions(deps)
+
+    env = await ca.pitch(
+        agent_id=uuid4(),
+        title="wip",
+        slug="my-product",
+        problem="Users cannot reset their password without contacting support.",
+        proposed_solution="Add a self-service password reset flow with email tokens.",
+        target_cells=["backend"],
+    )
+
+    assert env.as_dict()["error"] == "invalid_state"
+    deps.task.agent_for.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_open_session_placeholder_topic_rejected() -> None:
+    """``open_session`` topic must be substantive, not 'tbd'."""
+    deps = _make_deps()
+    ca = ContentActions(deps)
+
+    env = await ca.open_session(
+        agent_id=uuid4(), task_id=uuid4(), channel="backend-cell", topic="tbd"
+    )
+
+    assert env.as_dict()["error"] == "invalid_state"
+    deps.task.agent_for.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_pr_update_placeholder_body_rejected() -> None:
+    """``pr_update`` body, when supplied, must not be filler."""
+    deps = _make_deps()
+    ca = ContentActions(deps)
+
+    env = await ca.pr_update(agent_id=uuid4(), task_id=uuid4(), body="wip wip")
+
+    assert env.as_dict()["error"] == "invalid_state"
+    deps.task.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_pr_update_substantive_body_passes_soup_guard() -> None:
+    """A real body clears the soup guard and proceeds to the (missing) PR check."""
+    task = AsyncMock()
+    task.get.return_value = None  # not_found short-circuits after the soup guard
+    deps = _make_deps(task=task)
+    ca = ContentActions(deps)
+
+    env = await ca.pr_update(
+        agent_id=uuid4(),
+        task_id=uuid4(),
+        body="Rebased onto master and resolved the migration conflict.",
+    )
+
+    assert env.as_dict()["error"] == "not_found"
+    deps.task.get.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_note_decision_rationale_soup_rejected() -> None:
+    """A provided-but-soupy decision narrative field is rejected."""
+    deps = _make_deps()
+    ca = ContentActions(deps)
+
+    env = await ca.note(
+        agent_id=uuid4(),
+        text="Chose asyncpg over psycopg for the connection pool.",
+        scope="decision",
+        structured={"rationale": "asdf"},
+    )
+
+    assert env.as_dict()["error"] == "invalid_state"
+    deps.journal.write_entry.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_note_decision_omitted_narrative_still_records() -> None:
+    """An omitted narrative field keeps its tolerant placeholder — note records."""
+    deps = _make_deps()
+    ca = ContentActions(deps)
+
+    env = await ca.note(
+        agent_id=uuid4(),
+        text="Chose asyncpg over psycopg for the connection pool.",
+        scope="decision",
+        structured={"context": "Need an async driver for the new pool."},
+    )
+
+    assert env.as_dict()["error"] is None
+    deps.journal.write_entry.assert_awaited()