diff --git a/roboco/api/deps.py b/roboco/api/deps.py index cf43f8ba..05d591fb 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -502,7 +502,7 @@ def require_task_action( async def get_choreographer( db_session: DbSession, ) -> Choreographer: - """Build a Choreographer with all 7 service dependencies wired up.""" + """Build a Choreographer with all service dependencies wired up.""" return Choreographer( ChoreographerDeps( task=TaskService(db_session), @@ -512,6 +512,7 @@ async def get_choreographer( journal=JournalService(db_session), audit=get_audit_service(), evidence_repo=EvidenceRepo(db_session), + messaging=MessagingService(db_session), ) ) diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 57c0bbfd..ba83fd70 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -11,6 +11,7 @@ injection so later phases just fill in the bodies. from __future__ import annotations +import contextlib from dataclasses import dataclass from datetime import UTC, datetime from typing import Any, ClassVar @@ -173,6 +174,10 @@ class ChoreographerDeps: journal: Any audit: Any evidence_repo: Any + # Task #156: messaging is optional so existing callsites + tests that + # don't exercise session propagation don't have to plumb it in. The + # delegate() path uses it to thread parent sessions onto new subtasks. + messaging: Any = None @dataclass(frozen=True) @@ -305,11 +310,39 @@ class Choreographer: def evidence_repo(self) -> Any: return self._deps.evidence_repo + @property + def messaging(self) -> Any: + return self._deps.messaging + async def _touch(self, task_id: UUID | None) -> None: """Best-effort heartbeat write; silent on missing task.""" if task_id is not None: await self.task.heartbeat(task_id) + async def _record_milestone_progress( + self, + task_id: UUID, + agent_id: UUID, + message: str, + percentage: int | None = None, + ) -> None: + """Append a server-emitted progress entry on a lifecycle milestone. + + Task #155: agents call ``progress()`` inconsistently. Server-side + auto-emit on natural milestones (open_pr, i_am_done) guarantees + the panel + audit view always have entries at the major + transitions, regardless of how chatty the agent is. Best-effort: + a missing task_id or write failure must not break the verb path + — progress is observability, not correctness. + """ + with contextlib.suppress(Exception): + await self.task.add_progress( + task_id=task_id, + agent_id=agent_id, + message=message, + percentage=percentage, + ) + @staticmethod def _reassigned_rejection( ctx: _ReassignedCtx, @@ -1034,12 +1067,34 @@ class Choreographer: task_id=task_id, verb="open_pr", ) - # Re-fetch: git_service.create_pr writes pr_number / pr_url onto - # the task row. The runner doesn't bubble that update back, so a - # fresh load is the simplest way to surface the new fields in the - # OK envelope's next-hint and introspection block. + return await self._open_pr_success_envelope( + agent_id, task_id, t, briefing, role_str + ) + + async def _open_pr_success_envelope( + self, + agent_id: UUID, + task_id: UUID, + t: Any, + briefing: dict[str, Any], + role_str: str, + ) -> Envelope: + """Refresh the task, auto-emit milestone progress, build the OK envelope. + + git_service.create_pr writes pr_number / pr_url onto the task row; + the runner doesn't bubble that update back so we re-fetch. Task + #155 milestone progress fires server-side so the panel + audit + log always show "opened PR #N" regardless of agent chattiness. + """ refreshed = await self.task.get(task_id) t = refreshed if refreshed is not None else t + if t.pr_number is not None: + await self._record_milestone_progress( + task_id, + agent_id, + f"opened PR #{t.pr_number}", + percentage=70, + ) return Envelope.ok( status=str(t.status), task_id=str(task_id), @@ -1266,6 +1321,14 @@ class Choreographer: ) await self._notify_qa(ctx.agent_id, ctx.task_id, t) await self._touch(ctx.task_id) + # Task #155: server-side milestone progress so the panel always + # records the QA handoff regardless of agent's progress() habits. + await self._record_milestone_progress( + ctx.task_id, + ctx.agent_id, + "submitted for QA review", + percentage=90, + ) return await self._build_i_am_done_ok(ctx.agent_id, ctx.task_id, t) async def _i_am_done_resume_from_verifying(self, ctx: _IAmDoneContext) -> Envelope: @@ -2886,7 +2949,20 @@ class Choreographer: nature=nature_enum, estimated_complexity=complexity_enum, ) - return await self.task.create_subtask(req) + new_task = await self.task.create_subtask(req) + # Task #156: thread the parent's existing session links onto the + # new subtask so the assigned agent (dev/qa/doc) lands in the + # group chat the PM has already been talking in. Pre-gateway + # parity — sessions were wired to the whole tree at creation + # time; the gateway path created subtasks one-by-one and forgot + # this step. Idempotent on re-runs; no-op when no parent session. + if self.messaging is not None: + await self.messaging.propagate_sessions_to_subtask( + parent_task_id=parent_task_id, + subtask_id=new_task.id, + added_by=pm_agent_id, + ) + return new_task @staticmethod def _validate_delegation_chain(pm_role: str, target_slug: str) -> str | None: diff --git a/roboco/services/messaging.py b/roboco/services/messaging.py index 9eacc0d4..455d085a 100644 --- a/roboco/services/messaging.py +++ b/roboco/services/messaging.py @@ -1008,6 +1008,44 @@ class MessagingService(BaseService): ) return list(result.scalars().all()) + async def propagate_sessions_to_subtask( + self, + parent_task_id: UUID, + subtask_id: UUID, + added_by: UUID, + ) -> list[SessionTaskTable]: + """Link every session attached to ``parent_task_id`` onto ``subtask_id``. + + Task #156: pre-gateway flow created a session with the whole task + tree at once, so subtasks were visible in the parent's group chat + the moment they existed. The gateway creates subtasks one at a + time via ``delegate()``, so this step re-attaches every existing + parent session link to the new child. + + ``link_session_to_task`` is idempotent on duplicate (session, task) + pairs, so re-runs are no-ops. Primary status is NOT propagated — + each subtask owns its own primary slot, and a primary on the + parent should not auto-claim the subtask's primary too. + """ + parent_links = await self.get_sessions_for_task(parent_task_id) + propagated: list[SessionTaskTable] = [] + for parent_link in parent_links: + session_id = cast("UUID", parent_link.session_id) + rel_raw = parent_link.relationship_type + try: + rel = SessionTaskRelationshipType(rel_raw) + except (TypeError, ValueError): + rel = SessionTaskRelationshipType.DISCUSSION + link = await self.link_session_to_task( + session_id=session_id, + task_id=subtask_id, + added_by=added_by, + is_primary=False, + relationship_type=rel, + ) + propagated.append(link) + return propagated + async def _walk_task_ancestors(self, task_id: UUID) -> list["TaskTable"]: # type: ignore[name-defined] # noqa: F821 """Return [parent, grandparent, ..., root] for a task, empty if none. diff --git a/tests/unit/gateway/test_delegate_propagates_sessions.py b/tests/unit/gateway/test_delegate_propagates_sessions.py new file mode 100644 index 00000000..2b26d304 --- /dev/null +++ b/tests/unit/gateway/test_delegate_propagates_sessions.py @@ -0,0 +1,138 @@ +"""Task #156: delegate() must thread parent sessions onto the new subtask. + +Pre-gateway flow created sessions for whole task trees at once, so subtasks +were visible in the group chat the PM was already using. The gateway's +delegate() creates subtasks one-by-one — without this step the new agent +spawns into an empty channel and can't see the PM's prior discussion. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps +from roboco.services.gateway.choreographer._impl import DelegateInputs + + +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(), + "messaging": AsyncMock(), + } + base.update(overrides) + task = base["task"] + task.session = MagicMock() + task.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + _ldef = base["journal"].latest_decision_at.return_value + if type(_ldef).__name__ in ("MagicMock", "AsyncMock"): + base["journal"].latest_decision_at.return_value = datetime.now(UTC) + return ChoreographerDeps(**base) + + +@pytest.mark.asyncio +async def test_create_subtask_propagates_parent_sessions() -> None: + """When the choreographer creates a subtask, the parent's session + links are auto-attached to it via MessagingService.propagate_sessions_to_subtask. + """ + pm_id = uuid4() + parent_id = uuid4() + new_task_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + team="backend", + status="in_progress", + task_type="planning", + sequence=0, + assigned_to=pm_id, + ) + new_task = MagicMock(id=new_task_id, status="pending") + task_svc = AsyncMock() + task_svc.create_subtask.return_value = new_task + messaging = AsyncMock() + messaging.propagate_sessions_to_subtask.return_value = [] + deps = _make_deps(task=task_svc, messaging=messaging) + c = Choreographer(deps) + + inputs = DelegateInputs( + title="Backend slice", + description="API + DB", + acceptance_criteria=["api works", "schema migrated"], + assigned_to="be-dev-1", + team="backend", + task_type="code", + nature="technical", + estimated_complexity="medium", + ) + result = await c._create_subtask_from_inputs(pm_id, parent_id, parent, inputs) + assert result is new_task + + messaging.propagate_sessions_to_subtask.assert_awaited_once_with( + parent_task_id=parent_id, + subtask_id=new_task_id, + added_by=pm_id, + ) + + +@pytest.mark.asyncio +async def test_create_subtask_no_messaging_skips_propagation() -> None: + """When messaging dep is None (e.g. lightweight test wiring), the + subtask is still created — propagation is a soft enhancement, not a + hard requirement.""" + pm_id = uuid4() + parent_id = uuid4() + new_task_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + team="backend", + status="in_progress", + task_type="planning", + sequence=0, + assigned_to=pm_id, + ) + new_task = MagicMock(id=new_task_id, status="pending") + task_svc = AsyncMock() + task_svc.create_subtask.return_value = new_task + deps = _make_deps(task=task_svc, messaging=None) + c = Choreographer(deps) + + inputs = DelegateInputs( + title="Backend slice", + description="API + DB", + acceptance_criteria=["api works"], + assigned_to="be-dev-1", + team="backend", + task_type="code", + nature="technical", + estimated_complexity="medium", + ) + # Must not raise even though messaging is None. + result = await c._create_subtask_from_inputs(pm_id, parent_id, parent, inputs) + assert result is new_task diff --git a/tests/unit/gateway/test_milestone_progress.py b/tests/unit/gateway/test_milestone_progress.py new file mode 100644 index 00000000..036b4b10 --- /dev/null +++ b/tests/unit/gateway/test_milestone_progress.py @@ -0,0 +1,68 @@ +"""Task #155: server-side auto-progress on lifecycle milestones. + +Smoke-9 ended with zero progress entries because the dev never called +progress() explicitly. The fix: server emits progress entries at +deterministic lifecycle milestones (open_pr → "opened PR #N", +i_am_done → "submitted for QA review") so the panel + audit have entries +regardless of agent chattiness. Progress is observability — write failures +must not break the verb path. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + + +@pytest.mark.asyncio +async def test_record_milestone_progress_calls_add_progress() -> None: + """The helper proxies to TaskService.add_progress with the right args.""" + task_svc = AsyncMock() + deps = ChoreographerDeps( + task=task_svc, + work_session=AsyncMock(), + git=AsyncMock(), + a2a=AsyncMock(), + journal=AsyncMock(), + audit=AsyncMock(), + evidence_repo=AsyncMock(), + messaging=AsyncMock(), + ) + c = Choreographer(deps) + + task_id = uuid4() + agent_id = uuid4() + await c._record_milestone_progress(task_id, agent_id, "opened PR #20", 70) + + task_svc.add_progress.assert_awaited_once_with( + task_id=task_id, + agent_id=agent_id, + message="opened PR #20", + percentage=70, + ) + + +@pytest.mark.asyncio +async def test_record_milestone_progress_swallows_errors() -> None: + """Progress is observability. A failing add_progress must not raise + so the verb body's main flow is unaffected.""" + task_svc = AsyncMock() + task_svc.add_progress.side_effect = RuntimeError("db lock contention") + deps = ChoreographerDeps( + task=task_svc, + work_session=AsyncMock(), + git=AsyncMock(), + a2a=AsyncMock(), + journal=AsyncMock(), + audit=AsyncMock(), + evidence_repo=AsyncMock(), + messaging=AsyncMock(), + ) + c = Choreographer(deps) + + # Must NOT raise. + await c._record_milestone_progress(uuid4(), uuid4(), "submitted for QA review", 90) + task_svc.add_progress.assert_awaited_once() diff --git a/tests/unit/services/test_propagate_sessions_to_subtask.py b/tests/unit/services/test_propagate_sessions_to_subtask.py new file mode 100644 index 00000000..456ae7ac --- /dev/null +++ b/tests/unit/services/test_propagate_sessions_to_subtask.py @@ -0,0 +1,98 @@ +"""Task #156: MessagingService.propagate_sessions_to_subtask. + +Tests the helper's call shape and idempotency contract without going +through the DB. Integration coverage (real DB, real linking) lives in +``tests/integration/test_messaging_service.py``. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.models.session import SessionTaskRelationshipType +from roboco.services.messaging import MessagingService + + +def _link(session_id: object, relationship_type: str) -> MagicMock: + link = MagicMock() + link.session_id = session_id + link.relationship_type = relationship_type + return link + + +@pytest.mark.asyncio +async def test_propagate_links_every_parent_session_to_subtask() -> None: + """Every link on the parent gets re-attached to the new subtask.""" + svc = MessagingService.__new__(MessagingService) + parent_session = uuid4() + review_session = uuid4() + svc.get_sessions_for_task = AsyncMock( # type: ignore[method-assign] + return_value=[ + _link(parent_session, "discussion"), + _link(review_session, "review"), + ] + ) + + calls: list[dict[str, Any]] = [] + + async def fake_link(**kwargs: Any) -> Any: + calls.append(kwargs) + link = MagicMock() + link.session_id = kwargs["session_id"] + link.task_id = kwargs["task_id"] + return link + + svc.link_session_to_task = fake_link # type: ignore[method-assign] + + parent_id = uuid4() + subtask_id = uuid4() + added_by = uuid4() + expected_sessions = {parent_session, review_session} + out = await svc.propagate_sessions_to_subtask(parent_id, subtask_id, added_by) + assert len(out) == len(expected_sessions) + assert {c["session_id"] for c in calls} == expected_sessions + # Every propagated link must be non-primary — primary is the subtask's + # own slot, never inherited from the parent. + assert all(c["is_primary"] is False for c in calls) + # Every call must target the new subtask (not the parent). + assert {c["task_id"] for c in calls} == {subtask_id} + # Relationship types preserved. + rels = {c["relationship_type"] for c in calls} + assert SessionTaskRelationshipType.DISCUSSION in rels + assert SessionTaskRelationshipType.REVIEW in rels + + +@pytest.mark.asyncio +async def test_propagate_no_parent_sessions_returns_empty() -> None: + """When the parent has no session links, propagation is a no-op.""" + svc = MessagingService.__new__(MessagingService) + svc.get_sessions_for_task = AsyncMock(return_value=[]) # type: ignore[method-assign] + svc.link_session_to_task = AsyncMock() # type: ignore[method-assign] + + out = await svc.propagate_sessions_to_subtask(uuid4(), uuid4(), uuid4()) + assert out == [] + svc.link_session_to_task.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_propagate_unknown_relationship_type_defaults_to_discussion() -> None: + """Garbage relationship_type on the parent link doesn't crash; it + defaults to DISCUSSION so the subtask is still linked.""" + svc = MessagingService.__new__(MessagingService) + svc.get_sessions_for_task = AsyncMock( # type: ignore[method-assign] + return_value=[_link(uuid4(), "definitely-not-a-real-type")] + ) + calls: list[dict[str, Any]] = [] + + async def fake_link(**kwargs: Any) -> Any: + calls.append(kwargs) + return MagicMock() + + svc.link_session_to_task = fake_link # type: ignore[method-assign] + + await svc.propagate_sessions_to_subtask(uuid4(), uuid4(), uuid4()) + assert len(calls) == 1 + assert calls[0]["relationship_type"] == SessionTaskRelationshipType.DISCUSSION