Wave 2 features: A2A live view (CEO chime-in + reply budget) and prompter memory (#297)

* feat(a2a): live view — watch fleet conversations, CEO chime-in, reply budget

A2A_MESSAGE_SENT published from A2AService.send (excerpt-capped) and
fanned through the existing /ws/system bridge; CEO-only admin REST for
conversations/messages + a reply route on the publish-bearing send path;
panel /a2a page with live transcript and a composer gated on task-linked
conversations. The matrix gains its one asymmetric rule: CEO may message
anyone, nobody may target the CEO — and agent replies inside a
CEO-opened conversation are hard-budgeted to one per CEO message
(per conversation, per agent), rejected with wait-don't-retry guidance.
Built subagent-driven (Sonnet 5), reviewed; v1 seams documented in the
map delta.

* feat(prompter): intake remembers the task history

Intake spawns now carry a per-project chronological digest of recent
tasks (capped: 15 lines/project, 4000 chars total — ~300-1000 tokens)
merged into the ambient layer, and the interviewer gets a bounded
search_past_tasks tool (one shared implementation behind the grok MCP
tool and the Claude SDK in-process tool) to check precedent
mid-conversation. Informational memory only — the sequencing analyzer
keeps ownership of ordering. Built subagent-driven (Sonnet 5), reviewed;
pre-existing conventions-ambient MegaTask-scope gap flagged, untouched.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/unit/api/test_websocket_bridge.py

@@ -13,6 +13,7 @@ from uuid import uuid4
 
 import pytest
 from roboco.api.websocket_bridge import (
+    _handle_a2a_message_event,
     _handle_agent_event,
     _handle_message_event,
     _handle_notification_sent,
@@ -429,6 +430,44 @@ async def test_handle_usage_snapshot_broadcasts_to_system() -> None:
     assert len(msg["by_agent"]) == 1
 
 
+# ---------------------------------------------------------------------------
+# _handle_a2a_message_event
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_handle_a2a_message_event_broadcasts_to_system() -> None:
+    """An A2A_MESSAGE_SENT event is forwarded to /ws/system as an
+    `a2a.message` frame — the CEO's live view of every agent-to-agent chat."""
+    event = _evt(
+        EventType.A2A_MESSAGE_SENT,
+        {
+            "conversation_id": "conv-1",
+            "message_id": "msg-1",
+            "task_id": "task-1",
+            "from_agent": "be-dev-1",
+            "to_agent": "be-qa",
+            "skill": "code_review",
+            "body_excerpt": "please review",
+            "timestamp": "2026-07-02T00:00:00+00:00",
+        },
+    )
+    with patch("roboco.api.websocket_bridge.manager") as mgr:
+        mgr.broadcast_system = AsyncMock()
+        await _handle_a2a_message_event(event)
+    mgr.broadcast_system.assert_awaited_once()
+    msg = mgr.broadcast_system.await_args.args[0]
+    assert msg["type"] == "a2a.message"
+    assert msg["conversation_id"] == "conv-1"
+    assert msg["message_id"] == "msg-1"
+    assert msg["task_id"] == "task-1"
+    assert msg["from_agent"] == "be-dev-1"
+    assert msg["to_agent"] == "be-qa"
+    assert msg["skill"] == "code_review"
+    assert msg["body_excerpt"] == "please review"
+    assert msg["timestamp"] == "2026-07-02T00:00:00+00:00"
+
+
 # ---------------------------------------------------------------------------
 # Registration + start
 # ---------------------------------------------------------------------------
@@ -465,6 +504,8 @@ def test_register_websocket_bridge_handlers_subscribes_all_event_types() -> None
     assert EventType.USAGE_SNAPSHOT in types
     # Message delivery forwarded to /ws/channels + /ws/sessions.
     assert EventType.MESSAGE_SENT in types
+    # A2A live chat forwarded to /ws/system (CEO live view).
+    assert EventType.A2A_MESSAGE_SENT in types
 
 
 @pytest.mark.asyncio

tests/unit/enforcement/test_a2a_access.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import pytest
+from roboco.agents_config import can_a2a_direct
 from roboco.enforcement.a2a_access import (
     A2AAccessDeniedError,
     get_a2a_allowed_targets,
@@ -50,3 +51,46 @@ def test_get_a2a_allowed_targets_excludes_self() -> None:
     targets = get_a2a_allowed_targets("be-dev-1", ["be-dev-1", "be-qa"])
     # Self should be filtered.
     assert "be-dev-1" not in targets or "be-qa" in targets
+
+
+# ---------------------------------------------------------------------------
+# CEO-initiated A2A — the one asymmetric rule: CEO may send, nobody may
+# target CEO (the block above must still hold).
+# ---------------------------------------------------------------------------
+
+
+def test_validate_a2a_access_ceo_to_agent_allowed() -> None:
+    result = validate_a2a_access("ceo", "be-dev-1")
+    assert result is True
+
+
+def test_can_a2a_direct_ceo_to_main_pm() -> None:
+    assert can_a2a_direct("ceo", "main-pm") == (True, None)
+
+
+def test_can_a2a_direct_ceo_to_board_member() -> None:
+    """Board members are normally unreachable via direct A2A for everyone
+    else (routed through main-pm) — CEO is exempt from that restriction."""
+    assert can_a2a_direct("ceo", "product-owner") == (True, None)
+
+
+def test_validate_a2a_to_ceo_still_denied_with_ceo_send_rule() -> None:
+    """Regression: allowing CEO-initiated A2A must not loosen the inbound
+    block — nobody may target the CEO."""
+    with pytest.raises(A2AAccessDeniedError):
+        validate_a2a_access("be-dev-1", "ceo")
+
+
+def test_get_a2a_allowed_targets_ceo_includes_all_roles() -> None:
+    targets = get_a2a_allowed_targets("ceo", ["be-dev-1", "be-qa", "main-pm"])
+    assert set(targets) == {"be-dev-1", "be-qa", "main-pm"}
+
+
+def test_can_a2a_direct_to_ceo_message_explains_reply_only() -> None:
+    """An agent can never INITIATE with the CEO (only reply inside a
+    conversation the CEO opened) — the matrix denial message must say so,
+    not point at the old blanket 'use notify()' framing."""
+    allowed, reason = can_a2a_direct("be-dev-1", "ceo")
+    assert allowed is False
+    assert reason is not None
+    assert "reply" in reason.lower()

tests/unit/mcp_servers/test_intake_server.py

@@ -316,3 +316,166 @@ async def test_propose_draft_reports_relay_failure(
     msg = await intake_server.propose_draft({"title": "X"})
     assert "Could not submit the draft" in msg
     assert "http_503" in msg
+
+
+# ---------------------------------------------------------------------------
+# search_past_tasks — the intake's mid-conversation "have we done this before?"
+# tool (grok-CLI path).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_success_sends_q_and_limit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ROBOCO_API_URL", "http://orch:8000")
+    seen: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["params"] = dict(request.url.params)
+        return httpx.Response(
+            200,
+            json=[
+                {
+                    "id": "abcdef12-3456-7890-abcd-ef1234567890",
+                    "title": "Fix login bug",
+                    "status": "completed",
+                    "team": "backend",
+                    "date": "2026-01-01",
+                }
+            ],
+        )
+
+    async with _client(handler) as client:
+        result = await intake_server.query_past_tasks(
+            "sess-1", "login", limit=5, client=client
+        )
+
+    assert seen["url"].startswith(
+        "http://orch:8000/api/prompter/live/sess-1/search-tasks"
+    )
+    assert seen["params"]["q"] == "login"
+    assert seen["params"]["limit"] == "5"
+    assert result["results"][0]["title"] == "Fix login bug"
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_too_short_query_never_calls_http() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        raise AssertionError("must not call the relay for a too-short query")
+
+    async with _client(handler) as client:
+        result = await intake_server.query_past_tasks("sess-1", "a", client=client)
+
+    assert result == {"error": "query_too_short", "results": []}
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_http_error_shape() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(503)
+
+    async with _client(handler) as client:
+        result = await intake_server.query_past_tasks("sess-1", "login", client=client)
+
+    assert result["error"] == "http_503"
+    assert result["results"] == []
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_request_failure() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("boom")
+
+    async with _client(handler) as client:
+        result = await intake_server.query_past_tasks("sess-1", "login", client=client)
+
+    assert result["error"] == "request_failed"
+    assert "boom" in result["detail"]
+    assert result["results"] == []
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_clamps_limit_above_max() -> None:
+    seen: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["limit"] = dict(request.url.params)["limit"]
+        return httpx.Response(200, json=[])
+
+    async with _client(handler) as client:
+        await intake_server.query_past_tasks(
+            "sess-1", "login", limit=999, client=client
+        )
+
+    assert seen["limit"] == "10"
+
+
+@pytest.mark.asyncio
+async def test_query_past_tasks_clamps_limit_below_min() -> None:
+    seen: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["limit"] = dict(request.url.params)["limit"]
+        return httpx.Response(200, json=[])
+
+    async with _client(handler) as client:
+        await intake_server.query_past_tasks("sess-1", "login", limit=0, client=client)
+
+    assert seen["limit"] == "1"
+
+
+def test_format_search_results_error_dict() -> None:
+    msg = intake_server.format_search_results({"error": "http_503", "results": []})
+    assert "Could not search past tasks" in msg
+    assert "http_503" in msg
+
+
+def test_format_search_results_empty_list() -> None:
+    msg = intake_server.format_search_results({"results": []})
+    assert "No past tasks matched" in msg
+
+
+def test_format_search_results_renders_lines() -> None:
+    result = {
+        "results": [
+            {
+                "id": "abcdef1234567890",
+                "title": "Fix login bug",
+                "status": "completed",
+                "team": "backend",
+                "date": "2026-01-01",
+            }
+        ]
+    }
+    msg = intake_server.format_search_results(result)
+    assert "`abcdef12`" in msg  # short id truncated to 8 chars
+    assert "Fix login bug" in msg
+    assert "completed" in msg
+    assert "backend" in msg
+    assert "2026-01-01" in msg
+
+
+@pytest.mark.asyncio
+async def test_search_past_tasks_requires_a_live_session(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("ROBOCO_PROMPTER_SESSION_ID", raising=False)
+    msg = await intake_server.search_past_tasks("login")
+    assert "No live session id" in msg
+
+
+@pytest.mark.asyncio
+async def test_search_past_tasks_success_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ROBOCO_PROMPTER_SESSION_ID", "sess-1")
+    stub_result = {
+        "results": [{"id": "x", "title": "T", "status": "s", "team": "t", "date": "d"}]
+    }
+
+    async def _stub(_session_id: str, _query: str, **_kwargs: Any) -> dict[str, Any]:
+        return stub_result
+
+    monkeypatch.setattr(intake_server, "query_past_tasks", _stub)
+    msg = await intake_server.search_past_tasks("login")
+    assert msg == intake_server.format_search_results(stub_result)

tests/unit/runtime/test_intake_spawn.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import patch
-from uuid import UUID
+from uuid import UUID, uuid4
 
 import pytest
 from roboco.runtime.orchestrator import (
@@ -263,6 +263,118 @@ class TestIntakeScopeSlugs:
             )
 
 
+# ---------------------------------------------------------------------------
+# _resolve_history_digest_projects — the prompter-memory ambient's project scope
+# (covers all three intake scopes, unlike the conventions ambient resolver).
+# ---------------------------------------------------------------------------
+
+
+class TestResolveHistoryDigestProjects:
+    @pytest.mark.asyncio
+    async def test_project_slug_branch_resolves_single_project(self) -> None:
+        class _FakeProjectSvc:
+            async def get_by_slug(self, slug: str) -> Any:
+                return SimpleNamespace(slug=slug, id=uuid4())
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await AgentOrchestrator._resolve_history_digest_projects(
+                object(), project_slug="roboco", product_id=None, project_ids=None
+            )
+        assert [p.slug for p in projects] == ["roboco"]
+
+    @pytest.mark.asyncio
+    async def test_project_slug_missing_returns_empty(self) -> None:
+        class _FakeProjectSvc:
+            async def get_by_slug(self, _slug: str) -> Any:
+                return None
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await AgentOrchestrator._resolve_history_digest_projects(
+                object(), project_slug="ghost", product_id=None, project_ids=None
+            )
+        assert projects == []
+
+    @pytest.mark.asyncio
+    async def test_product_id_branch_delegates_to_ambient_product_projects(
+        self,
+    ) -> None:
+        sentinel = [SimpleNamespace(slug="p1", id=uuid4())]
+
+        async def _fake_product_projects(_db: Any, product_id: str) -> list[Any]:
+            assert product_id == "prod-1"
+            return sentinel
+
+        with patch.object(
+            AgentOrchestrator, "_ambient_product_projects", _fake_product_projects
+        ):
+            projects = await AgentOrchestrator._resolve_history_digest_projects(
+                object(), project_slug=None, product_id="prod-1", project_ids=None
+            )
+        assert projects is sentinel
+
+    @pytest.mark.asyncio
+    async def test_project_ids_branch_preserves_order_and_skips_missing(
+        self,
+    ) -> None:
+        good1 = "11111111-1111-1111-1111-111111111111"
+        missing = "22222222-2222-2222-2222-222222222222"
+        good2 = "33333333-3333-3333-3333-333333333333"
+
+        class _FakeProjectSvc:
+            async def get(self, pid: Any) -> Any:
+                if str(pid) == missing:
+                    return None
+                return SimpleNamespace(slug=f"proj-{str(pid)[0]}", id=pid)
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await AgentOrchestrator._resolve_history_digest_projects(
+                object(),
+                project_slug=None,
+                product_id=None,
+                project_ids=[good1, missing, good2],
+            )
+        # Order preserved; the unresolvable id is skipped, not raised — this is
+        # a best-effort ambient resolver, not the hard clone-scope resolver.
+        assert [p.slug for p in projects] == ["proj-1", "proj-3"]
+
+    @pytest.mark.asyncio
+    async def test_no_scope_given_returns_empty(self) -> None:
+        projects = await AgentOrchestrator._resolve_history_digest_projects(
+            object(), project_slug=None, product_id=None, project_ids=None
+        )
+        assert projects == []
+
+
+# ---------------------------------------------------------------------------
+# _resolve_history_digest_ambient — best-effort: any failure returns None.
+# ---------------------------------------------------------------------------
+
+
+class TestResolveHistoryDigestAmbient:
+    @pytest.mark.asyncio
+    async def test_failure_returns_none_not_raises(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        orch = _make_minimal_orchestrator()
+
+        def _boom() -> Any:
+            raise RuntimeError("db unavailable")
+
+        monkeypatch.setattr("roboco.db.base.get_session_factory", _boom)
+
+        result = await orch._resolve_history_digest_ambient("roboco")
+        assert result is None
+
+
 # ---------------------------------------------------------------------------
 # spawn_intake_session / reap_intake_session — orchestration (docker mocked).
 # ---------------------------------------------------------------------------
@@ -389,6 +501,63 @@ class TestSpawnIntakeSession:
         await orch.spawn_intake_session("sess-2", project_slug="roboco")
         assert stopped == [INTAKE_AGENT_ID]  # the old one was reaped first
 
+    @pytest.mark.asyncio
+    async def test_spawn_merges_conventions_and_history_ambient(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """The composed prompt's ambient is the conventions + history-digest
+        blocks joined with compose_prompt's own layer separator."""
+        orch = _make_minimal_orchestrator()
+        run_calls: list[list[str]] = []
+        _wire_spawn_mocks(monkeypatch, orch, run_calls)
+
+        async def _conventions(*_a: Any, **_k: Any) -> str | None:
+            return "CONVENTIONS BLOCK"
+
+        async def _history(*_a: Any, **_k: Any) -> str | None:
+            return "HISTORY BLOCK"
+
+        monkeypatch.setattr(orch, "_resolve_conventions_ambient", _conventions)
+        monkeypatch.setattr(orch, "_resolve_history_digest_ambient", _history)
+
+        captured: dict[str, Any] = {}
+
+        def _spy_prompt(*_args: Any, **kwargs: Any) -> Path:
+            captured["ambient"] = kwargs.get("ambient")
+            return Path("/tmp/intake-1-prompt.md")
+
+        monkeypatch.setattr(orch, "_generate_composed_prompt", _spy_prompt)
+
+        await orch.spawn_intake_session("sess-merge", project_slug="roboco")
+
+        assert captured["ambient"] == "CONVENTIONS BLOCK\n\n---\n\nHISTORY BLOCK"
+
+    @pytest.mark.asyncio
+    async def test_spawn_ambient_none_when_both_resolvers_empty(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        orch = _make_minimal_orchestrator()
+        run_calls: list[list[str]] = []
+        _wire_spawn_mocks(monkeypatch, orch, run_calls)
+
+        async def _none(*_a: Any, **_k: Any) -> str | None:
+            return None
+
+        monkeypatch.setattr(orch, "_resolve_conventions_ambient", _none)
+        monkeypatch.setattr(orch, "_resolve_history_digest_ambient", _none)
+
+        captured: dict[str, Any] = {}
+
+        def _spy_prompt(*_args: Any, **kwargs: Any) -> Path:
+            captured["ambient"] = kwargs.get("ambient")
+            return Path("/tmp/intake-1-prompt.md")
+
+        monkeypatch.setattr(orch, "_generate_composed_prompt", _spy_prompt)
+
+        await orch.spawn_intake_session("sess-no-ambient", project_slug="roboco")
+
+        assert captured["ambient"] is None
+
     @pytest.mark.asyncio
     async def test_initial_message_is_scheduled(
         self, monkeypatch: pytest.MonkeyPatch

tests/unit/services/test_prompter.py

@@ -8,10 +8,15 @@ conftest fixtures.
 
 from __future__ import annotations
 
-from typing import Any, cast
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, cast
 from uuid import UUID, uuid4
 
 import pytest
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
 from roboco.db.tables import (
     AgentTable,
     ProductTable,
@@ -28,15 +33,23 @@ from roboco.models.base import (
     Team,
 )
 from roboco.seeds.initial_data import AGENT_UUIDS
+from roboco.services import prompter as prompter_module
 from roboco.services.base import ServiceError, ValidationError
 from roboco.services.prompter import (
+    _HISTORY_DIGEST_PER_PROJECT_LIMIT,
+    _HISTORY_TITLE_EXCERPT_CAP,
     PrompterService,
     _cell_teams,
     _clean_list,
     _draft_cell_map,
+    _task_activity_date,
+    _title_excerpt,
+    build_history_digest,
+    compact_task_rows,
     compose_description,
     derive_scale,
     get_prompter_service,
+    history_digest_layer,
     parse_readiness,
 )
 
@@ -1052,3 +1065,216 @@ async def test_create_task_from_draft_does_not_mutate_caller_draft(
     assert draft["the_work"][0]["items"] == original_items
     # ...and the top-level acceptance_criteria was NOT replaced.
     assert draft["acceptance_criteria"] == ["done"]
+
+
+# =============================================================================
+# Prompter memory v1 — history digest + compact search rows (pure, no DB)
+# =============================================================================
+
+
+def _task(title: str, **overrides: Any) -> TaskTable:
+    """An unattached TaskTable instance — plain attribute assignment, no session.
+
+    Defaults to a completed backend task with no dates; pass ``completed_at`` /
+    ``updated_at`` / ``created_at`` / ``status`` / ``team`` to override.
+    """
+    fields: dict[str, Any] = {
+        "id": uuid4(),
+        "title": title,
+        "status": TaskStatus.COMPLETED,
+        "team": Team.BACKEND,
+        "completed_at": None,
+        "updated_at": None,
+        "created_at": None,
+    }
+    fields.update(overrides)
+    return TaskTable(**fields)
+
+
+def test_task_activity_date_prefers_completed_at() -> None:
+    now = datetime.now(UTC)
+    task = _task(
+        "t",
+        completed_at=now,
+        updated_at=now - timedelta(days=1),
+        created_at=now - timedelta(days=2),
+    )
+    assert _task_activity_date(task) == now
+
+
+def test_task_activity_date_falls_back_to_updated_at() -> None:
+    now = datetime.now(UTC)
+    task = _task(
+        "t", completed_at=None, updated_at=now, created_at=now - timedelta(days=1)
+    )
+    assert _task_activity_date(task) == now
+
+
+def test_task_activity_date_falls_back_to_created_at() -> None:
+    now = datetime.now(UTC)
+    task = _task("t", completed_at=None, updated_at=None, created_at=now)
+    assert _task_activity_date(task) == now
+
+
+def test_title_excerpt_leaves_short_titles_untouched() -> None:
+    assert _title_excerpt("Fix login bug") == "Fix login bug"
+
+
+def test_title_excerpt_truncates_long_titles_with_ellipsis() -> None:
+    long_title = "A" * 100
+    excerpt = _title_excerpt(long_title)
+    assert len(excerpt) == _HISTORY_TITLE_EXCERPT_CAP
+    assert excerpt.endswith("…")
+
+
+def test_build_history_digest_empty_is_blank() -> None:
+    assert build_history_digest([]) == ""
+
+
+def test_build_history_digest_caps_at_limit_keeps_most_recent() -> None:
+    now = datetime.now(UTC)
+    # t0 oldest ... t19 newest.
+    ascending = [
+        _task(f"t{i}", created_at=now + timedelta(days=i), updated_at=None)
+        for i in range(20)
+    ]
+    # Mimic the DB's most-recent-first ordering.
+    most_recent_first = list(reversed(ascending))
+
+    digest = build_history_digest(most_recent_first)
+
+    lines = digest.splitlines()
+    assert len(lines) == _HISTORY_DIGEST_PER_PROJECT_LIMIT
+    for i in range(5):  # the 5 oldest are excluded
+        assert f"`{str(ascending[i].id)[:8]}`" not in digest
+    for i in range(5, 20):  # the 15 most recent are present
+        assert f"`{str(ascending[i].id)[:8]}`" in digest
+
+
+def test_build_history_digest_renders_oldest_first() -> None:
+    now = datetime.now(UTC)
+    a = _task("Task A", created_at=now - timedelta(days=2), updated_at=None)
+    b = _task("Task B", created_at=now - timedelta(days=1), updated_at=None)
+    c = _task("Task C", created_at=now, updated_at=None)
+
+    # DB order is most-recent-first: C, B, A.
+    digest = build_history_digest([c, b, a])
+
+    idx_a = digest.index("Task A")
+    idx_b = digest.index("Task B")
+    idx_c = digest.index("Task C")
+    assert idx_a < idx_b < idx_c
+
+
+def test_compact_task_rows_shape() -> None:
+    now = datetime.now(UTC)
+    task = _task(
+        "Fix login bug",
+        status=TaskStatus.COMPLETED,
+        team=Team.BACKEND,
+        completed_at=now,
+    )
+    rows = compact_task_rows([task])
+    assert len(rows) == 1
+    row = rows[0]
+    assert set(row.keys()) == {"id", "title", "status", "team", "date"}
+    assert row["id"] == str(task.id)
+    assert row["title"] == "Fix login bug"
+    assert row["status"] == "completed"
+    assert row["team"] == "backend"
+    assert row["date"] == now.date().isoformat()
+
+
+def test_compact_task_rows_preserves_none_team() -> None:
+    task = _task("No team", team=None, created_at=datetime.now(UTC))
+    rows = compact_task_rows([task])
+    assert rows[0]["team"] is None
+
+
+# -----------------------------------------------------------------------------
+# history_digest_layer — ambient-block assembly (project_history_digest stubbed)
+# -----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_history_digest_layer_empty_projects_returns_none() -> None:
+    assert await history_digest_layer(cast("AsyncSession", object()), []) is None
+
+
+@pytest.mark.asyncio
+async def test_history_digest_layer_single_project_has_no_header(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def _fake(_session: Any, _project: Any, *, _limit: int = 15) -> str | None:
+        return "- `abc12345` Some task (completed, 2026-01-01)"
+
+    monkeypatch.setattr(prompter_module, "project_history_digest", _fake)
+    project = SimpleNamespace(slug="roboco", id=uuid4())
+
+    text = await history_digest_layer(cast("AsyncSession", object()), [project])
+
+    assert text is not None
+    assert text.startswith("## Task History\n\n### Recent tasks\n")
+    assert "### Recent tasks —" not in text
+
+
+@pytest.mark.asyncio
+async def test_history_digest_layer_multi_project_headers_by_slug(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    projects = [
+        SimpleNamespace(slug="backend-svc", id=uuid4()),
+        SimpleNamespace(slug="frontend-app", id=uuid4()),
+    ]
+
+    async def _fake(_session: Any, project: Any, *, _limit: int = 15) -> str | None:
+        return f"- `deadbeef` Task for {project.slug} (completed, 2026-01-01)"
+
+    monkeypatch.setattr(prompter_module, "project_history_digest", _fake)
+
+    text = await history_digest_layer(cast("AsyncSession", object()), projects)
+
+    assert text is not None
+    assert "### Recent tasks — `backend-svc`" in text
+    assert "### Recent tasks — `frontend-app`" in text
+
+
+@pytest.mark.asyncio
+async def test_history_digest_layer_skips_projects_with_no_tasks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    has_tasks = SimpleNamespace(slug="has-tasks", id=uuid4())
+    no_tasks = SimpleNamespace(slug="empty-proj", id=uuid4())
+
+    async def _fake(_session: Any, project: Any, *, _limit: int = 15) -> str | None:
+        return (
+            "- `deadbeef` A task (completed, 2026-01-01)"
+            if project is has_tasks
+            else None
+        )
+
+    monkeypatch.setattr(prompter_module, "project_history_digest", _fake)
+
+    text = await history_digest_layer(
+        cast("AsyncSession", object()), [has_tasks, no_tasks]
+    )
+
+    assert text is not None
+    assert "has-tasks" in text
+    assert "empty-proj" not in text
+
+
+@pytest.mark.asyncio
+async def test_history_digest_layer_all_empty_returns_none(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def _fake(_session: Any, _project: Any, *, _limit: int = 15) -> str | None:
+        return None
+
+    monkeypatch.setattr(prompter_module, "project_history_digest", _fake)
+    projects = [
+        SimpleNamespace(slug="a", id=uuid4()),
+        SimpleNamespace(slug="b", id=uuid4()),
+    ]
+
+    assert await history_digest_layer(cast("AsyncSession", object()), projects) is None