[F010] notification: never dedup informational notifications (knowledge-share data loss)

The purpose-based dedup suppressed a same-purpose (same sender/type/task,
overlapping recipients) notification while a prior one was unacked. For
informational types (KNOWLEDGE_SHARE / MENTION / A2A_REQUEST / BROADCAST +
the pickup-proves-receipt triad) each send carries DISTINCT content (a new
learning, a new mention) and acking is voluntary, so a recipient who never
acks the prior one let the dedup permanently suppress every subsequent
same-sender broadcast - silent learning-broadcast data loss.

The dedup's anti-loop rationale (stop unacked-set inflation soft-blocking
i_am_idle) only holds for action-required signals. Gate the dedup on
ACK_REQUIRED_BY_TYPE.get(type, True): action-required types still dedup,
informational types always create. Unmapped types default True (dedup on).

TDD red->green; ruff + mypy clean; notification + dedup suites green (20).

roboco/services/notification.py

@@ -492,29 +492,41 @@ class NotificationService:
             # type, a different task, a different sender, or a recipient who has
             # already acked all go through. Body text is NOT compared, so
             # rewording cannot defeat the guard.
+            #
+            # F010: the dedup only applies to ACTION-REQUIRED types
+            # (ACK_REQUIRED_BY_TYPE -> True). Informational types
+            # (KNOWLEDGE_SHARE / MENTION / A2A_REQUEST / BROADCAST / the
+            # pickup-proves-receipt triad) carry distinct content per send — a
+            # new learning, a new mention — and acking them is voluntary, so a
+            # recipient who never acks would let the dedup permanently suppress
+            # every subsequent same-sender broadcast (silent learning-broadcast
+            # data loss). The anti-loop rationale only holds for ack-required
+            # signals; informational ones are not deduped.
             related = params.related_task_id
-            dup_q = (
-                select(NotificationTable.id)
-                .where(NotificationTable.from_agent == from_agent_uuid)
-                .where(NotificationTable.type == params.notification_type)
-                .where(NotificationTable.to_agents.overlap(to_agents_uuids))
-                .where(~NotificationTable.acked_by.contains(to_agents_uuids))
-                .where(
-                    NotificationTable.related_task_id == related
-                    if related is not None
-                    else NotificationTable.related_task_id.is_(None)
+            is_ack_required = ACK_REQUIRED_BY_TYPE.get(params.notification_type, True)
+            if is_ack_required:
+                dup_q = (
+                    select(NotificationTable.id)
+                    .where(NotificationTable.from_agent == from_agent_uuid)
+                    .where(NotificationTable.type == params.notification_type)
+                    .where(NotificationTable.to_agents.overlap(to_agents_uuids))
+                    .where(~NotificationTable.acked_by.contains(to_agents_uuids))
+                    .where(
+                        NotificationTable.related_task_id == related
+                        if related is not None
+                        else NotificationTable.related_task_id.is_(None)
+                    )
+                    .limit(1)
                 )
-                .limit(1)
-            )
-            if await db.scalar(dup_q) is not None:
-                logger.info(
-                    "Suppressed duplicate notification (same purpose, unacked)",
-                    from_agent=str(from_agent_uuid),
-                    type=params.notification_type.value,
-                    related_task_id=str(related) if related is not None else None,
-                    to_agents=[str(a) for a in to_agents_uuids],
-                )
-                return
+                if await db.scalar(dup_q) is not None:
+                    logger.info(
+                        "Suppressed duplicate notification (same purpose, unacked)",
+                        from_agent=str(from_agent_uuid),
+                        type=params.notification_type.value,
+                        related_task_id=str(related) if related is not None else None,
+                        to_agents=[str(a) for a in to_agents_uuids],
+                    )
+                    return
             notification = NotificationTable(
                 type=params.notification_type,
                 priority=params.priority,

tests/unit/test_notification_dedup.py

@@ -69,3 +69,53 @@ async def test_create_notification_suppresses_same_purpose_duplicate() -> None:
     db.add.assert_not_called()
     db.commit.assert_not_called()
     db.scalar.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_informational_knowledge_share_not_deduped() -> None:
+    """F010: KNOWLEDGE_SHARE (informational, requires_ack=False) must NOT be
+    deduped. Each learning broadcast carries distinct content (a new learning);
+    a recipient who never acks the prior one (acking is voluntary for
+    informational types) would permanently suppress every subsequent
+    knowledge-share from the same sender → silent learning-broadcast data loss.
+    The dedup's anti-loop rationale only applies to action-required types."""
+    db = MagicMock()
+    # A same-purpose unacked KNOWLEDGE_SHARE prior exists — but it must NOT
+    # suppress the new one.
+    db.scalar = AsyncMock(return_value=uuid4())
+    # ``db.add`` must give the row an id — the delivery path calls
+    # ``require_uuid(notification.id)``.
+    db.add = MagicMock(side_effect=lambda obj: setattr(obj, "id", uuid4()))
+    db.flush = AsyncMock()
+    db.commit = AsyncMock()
+
+    svc = NotificationService()
+    svc._resolve_recipients = AsyncMock(return_value=[uuid4()])  # type: ignore[method-assign]
+    params = CreateNotificationParams(
+        notification_type=NotificationType.KNOWLEDGE_SHARE,
+        priority=NotificationPriority.NORMAL,
+        from_agent="from-1",
+        to_agents=["to-1"],
+        subject="New Learning: bug",
+        body="a fresh learning the recipient has not seen",
+        related_task_id=None,
+    )
+    with (
+        patch(
+            "roboco.services.notification.get_db_context",
+            return_value=_FakeDBCtx(db),
+        ),
+        patch(
+            "roboco.services.notification._resolve_agent_uuid",
+            AsyncMock(return_value=uuid4()),
+        ),
+        patch(
+            "roboco.services.notification_delivery.get_notification_delivery_service",
+            lambda _db: MagicMock(deliver=AsyncMock(return_value=None)),
+        ),
+    ):
+        await svc._create_notification(params)
+
+    # Informational ⇒ NOT suppressed: a row was created + committed.
+    db.add.assert_called_once()
+    db.commit.assert_awaited_once()