diff --git a/workspace/backend/app/config.py b/workspace/backend/app/config.py index f6b135a71..e2ed0832e 100644 --- a/workspace/backend/app/config.py +++ b/workspace/backend/app/config.py @@ -77,6 +77,9 @@ class Config: # Cloud agents CLOUD_AGENT_MAX_CONTEXT_MESSAGES: int = int(os.environ.get("CLOUD_AGENT_MAX_CONTEXT_MESSAGES", "10")) CLOUD_AGENT_MAX_DEPTH: int = int(os.environ.get("CLOUD_AGENT_MAX_DEPTH", "3")) + # Max cloud agents invoked concurrently for one message — bounds DB + # connections and provider calls when a message fans out to many agents. + CLOUD_AGENT_MAX_CONCURRENCY: int = int(os.environ.get("CLOUD_AGENT_MAX_CONCURRENCY", "4")) # Google OAuth (for "Sign in with Google" Gemini integration) GOOGLE_OAUTH_CLIENT_ID: str = os.environ.get("GOOGLE_OAUTH_CLIENT_ID", "") diff --git a/workspace/backend/app/mods/workspace_mod.py b/workspace/backend/app/mods/workspace_mod.py index dc1de3d41..e0eb77abd 100644 --- a/workspace/backend/app/mods/workspace_mod.py +++ b/workspace/backend/app/mods/workspace_mod.py @@ -582,10 +582,19 @@ def _fallback_targets(event, channel, mentions: List[str], online_names: set = N Priority: explicit @mentions → master (for human/member msgs) → online participant → any participant. When ``online_names`` is provided, an online participant is chosen over an offline one so messages aren't stranded on a - dead agent. An explicit @mention is always honored as-is (the user chose it). + dead agent. Explicit @mentions are always honored as-is (the sender chose + them) — ALL of them, so "@a do X @b do Y" fans out and the mentioned + agents work in parallel instead of only the first one being targeted. """ if mentions: - return [mentions[0]] + targets = list(dict.fromkeys(mentions)) # dedupe, keep order + if event.source.startswith("openagents:"): + # An agent mentioning itself must not self-trigger. + sender = event.source[len("openagents:"):] + targets = [t for t in targets if t != sender] + # All mentions were self-mentions → nobody should respond. Return + # here (not fall through) so a self-note never re-routes to master. + return targets if channel.master_agent: if event.source.startswith("openagents:"): sender = event.source[len("openagents:"):] @@ -602,6 +611,55 @@ def _fallback_targets(event, channel, mentions: List[str], online_names: set = N return [participants[0]] if participants else [] +# How far back to look for the human message that assigned the current task. +# Bounded so a long thread doesn't scan the whole channel on every agent turn. +_ASSIGNMENT_LOOKBACK = 20 + + +def _human_assignment_holds(db, workspace, channel, sender: str, + known_agents: List[str]) -> bool: + """True when a human directly @assigned ``sender`` and that assignment stands. + + Walks the channel's recent human messages newest → oldest: + • one that @mentions ``sender`` → the assignment holds + • one with no @mention at all → free-form chat reopened routing + • one that @mentions only others → keep looking further back (parallel + assignments arrive as separate messages, one per agent) + + Callers use this to keep a directly-addressed agent's own output from + being handed to a peer. When the human picked the agent, its progress + notes and results belong to the human — not to whichever bystander the + LLM router happens to choose. With several agents working in parallel + that misrouting is not an edge case: every worker's "starting now…" + note becomes a spurious turn for every other worker, which then answers + a task it was never given (and only after its own 60s job drains). + """ + from app.models import EventRecord + + rows = db.execute( + select(EventRecord) + .where( + EventRecord.network_id == workspace.id, + EventRecord.target == f"channel/{channel.name}", + EventRecord.type == "workspace.message.posted", + EventRecord.source.like("human:%"), + ) + .order_by(EventRecord.timestamp.desc()) + .limit(_ASSIGNMENT_LOOKBACK) + ).scalars().all() + + for evt in rows: + payload = evt.payload or {} + if payload.get("message_type", "chat") in ("thinking", "status", "todos"): + continue + mentions = _extract_mentions(payload.get("content") or "", known_agents) + if not mentions: + return False + if sender in mentions: + return True + return False + + def _master_targets(event, channel, mentions: List[str]) -> List[str]: """Deterministic routing for "master" orchestration mode (star topology). @@ -625,10 +683,12 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: sender = source[len("openagents:"):] if sender == master: # Master is delegating. Route to any mentioned sub-agents - # (never itself); no mention → the master answered, so stop. + # (never itself, deduped); no mention → the master answered, + # so stop. participants = {p.agent_name for p in (channel.participants or [])} delegated = [ - m for m in mentions if m != master and m in participants + m for m in dict.fromkeys(mentions) + if m != master and m in participants ] return delegated # A sub-agent spoke → return control to the master hub. @@ -662,9 +722,14 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: pick the addressed agent, not the mentioned one. B. If the LATEST message is from a HUMAN: - - Always pick exactly one agent. Humans expect a reply — never output \ + - Always pick at least one agent. Humans expect a reply — never output \ "stop" for a human message. - Prefer whoever is directly addressed. + - If the message assigns separate, independent tasks to SEVERAL agents \ +("@alice do X and @bob do Y"), pick ALL of them (comma-separated) so they \ +work in parallel. Pick several agents only for genuinely independent \ +tasks — if one task depends on another's result, pick only the agent \ +whose task comes first. - If nobody is directly addressed, check CONVERSATIONAL CONTINUITY: \ if the user was just conversing with a specific agent (the last agent \ reply was from agent X, or X asked the user a question that this message \ @@ -673,19 +738,27 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: fall back to the master agent. C. If the LATEST message is from an AGENT: - - If it delegates or hands off to another agent ("@Alice please do X", \ -"Alice, could you check X"), route to that agent. + - If it delegates or hands off to other agents ("@Alice please do X", \ +"Alice, could you check X"), route to them — ALL of the delegated agents \ +(comma-separated) when it hands independent tasks to several at once. - If it reports back to the master or asks the master to decide, route to the master. - If it is a FINAL answer to the previous human question or an \ acknowledgement ("done", "saved", "sounds good"), output "stop". + - If it is progress narration about work the sender is doing right now \ +("running the command now", "on it", "this will take about a minute"), \ +output "stop". Nobody else should act on it — several agents often work \ +in parallel, and each one's progress note must not become a turn for the \ +others. - Never route back to the same agent that just spoke (no self-loops). - When unsure, prefer "stop" to avoid infinite agent-to-agent loops. EXAMPLES: Human: "@alice what's the status?" → next:alice + Human: "@alice check the logs, @bob fix the tests" → next:alice,bob (independent tasks, parallel) Human: "check @alice's notes, @bob" → next:bob (bob is addressed) Human: "how about julia?" (julia is not an agent) → next: (who owns that topic) Agent alice: "@bob can you verify?" → next:bob + Agent alice: "@bob run the tests and @carol update the docs" → next:bob,carol Agent alice: "Done — results attached." → stop Agent bob (master): "Here's the final answer ..." → stop @@ -696,8 +769,9 @@ def _master_targets(event, channel, mentions: List[str]) -> List[str]: alice: "I pulled these results: [...]." Human: "thanks, can you also check Y?" → next:alice (follow-up to alice) -Output EXACTLY one line, lowercase, no punctuation or explanation: +Output EXACTLY one line, lowercase, no spaces or explanation: next: + next:, (several agents, comma-separated) stop""" @@ -864,14 +938,14 @@ async def _route_with_llm( if provider == "openai": response = client.chat.completions.create( model=model, - max_tokens=30, + max_tokens=64, messages=[{"role": "user", "content": prompt}], ) raw_result = response.choices[0].message.content.strip() else: response = client.messages.create( model=model, - max_tokens=30, + max_tokens=64, messages=[{"role": "user", "content": prompt}], ) raw_result = response.content[0].text.strip() @@ -884,9 +958,15 @@ async def _route_with_llm( logger.info("LLM router decision: %s (channel=%s, sender=%s, provider=%s)", raw_result, channel.name, sender, provider) if result.startswith("next:"): - # Preserve the original case from the model output so we can - # match against participant names, which ARE case-sensitive. - agent_name = raw_result[len("next:"):].strip().split(",")[0].strip() + # The router may name several agents ("next: a, b") — honor all + # of them so independent tasks run in parallel. Preserve the + # original case from the model output so we can match against + # participant names, which ARE case-sensitive. + requested = [ + n.strip() + for n in raw_result[len("next:"):].split(",") + if n.strip() + ] # Case-insensitive participant lookup, then canonicalize to # the stored case. # Validate against the candidate set (online participants when any @@ -895,30 +975,32 @@ async def _route_with_llm( participants_by_lower = { name.lower(): name for name in candidate_names } - canonical = participants_by_lower.get(agent_name.lower()) - if canonical is None: - logger.warning( - "LLM router returned unknown agent: %r (valid: %s)", - agent_name, list(participants_by_lower.values()), - ) - # For human senders, fall through to the safety net below - # so the user always gets a reply. - if not (new_event.source or "").startswith("human:"): - return [] - agent_name = None - else: - agent_name = canonical + sender_name = None + if (new_event.source or "").startswith("openagents:"): + sender_name = new_event.source[len("openagents:"):] + targets: List[str] = [] + for name in requested: + canonical = participants_by_lower.get(name.lower()) + if canonical is None: + logger.warning( + "LLM router returned unknown agent: %r (valid: %s)", + name, list(participants_by_lower.values()), + ) + continue # Reject self-loops — router sometimes picks the agent # who just spoke. Sender's adapter skips own messages but # legacy clients would still see the target and retry. - if (new_event.source or "").startswith("openagents:"): - sender = new_event.source[len("openagents:"):] - if agent_name == sender: - logger.info("LLM router self-loop rejected: %s", sender) - return [] - return [agent_name] - else: - agent_name = None # "stop" or unrecognized + if canonical == sender_name: + logger.info("LLM router self-loop rejected: %s", sender_name) + continue + if canonical not in targets: + targets.append(canonical) + if targets: + return targets + # No valid target survived — for human senders fall through to + # the safety net below so the user always gets a reply. + if not (new_event.source or "").startswith("human:"): + return [] # Safety net: humans ALWAYS get a response. If the router said # "stop" (or returned an invalid agent) for a human message, @@ -1121,6 +1203,12 @@ async def _handle_message_posted(event: Event, ctx: PipelineContext) -> Optional from app.config import config mode = (getattr(channel, "orchestration_mode", None) or "dynamic").lower() + sender_agent = ( + event.source[len("openagents:"):] + if event.source.startswith("openagents:") else None + ) + peer_mentions = [m for m in mentions if m != sender_agent] + if mode == "master": # Deterministic star topology — no LLM. If the channel somehow # has no master, fall back to the generic mention/online logic @@ -1129,6 +1217,32 @@ async def _handle_message_posted(event: Event, ctx: PipelineContext) -> Optional targets = _master_targets(event, channel, mentions) else: targets = _fallback_targets(event, channel, mentions, online_names) + elif mentions and event.source.startswith("human:"): + # A human explicitly @mentioned agents — honor the mentions + # as-is (all of them) without consulting the LLM router. The + # user already chose the targets, and the router used to pick a + # single agent for "@a task1 @b task2", collapsing parallel work + # onto one agent whose queue then serialized the tasks. + targets = _fallback_targets(event, channel, mentions, online_names) + elif ( + mode != "workflow" + and sender_agent + and not peer_mentions + and _human_assignment_holds( + db, workspace, channel, sender_agent, known_agents, + ) + ): + # The sender is working a task a human handed it by name, and it + # is not handing off to anyone (no @mention of a peer). Its + # output goes back to the human — do not consult the router, + # which has no way to tell "I'm starting the job now" from a + # handoff and would wake a bystander agent for someone else's + # task. The mirror image of the human-mention bypass above. + logger.info( + "Direct assignment holds for %s — not routing its message to a peer", + sender_agent, + ) + targets = [] elif mode == "workflow" and config.ROUTER_LLM_ENABLED and _get_router_api_key(): # LLM router steered by the user's natural-language plan. targets = await _route_with_llm( diff --git a/workspace/backend/app/services/cloud_agent.py b/workspace/backend/app/services/cloud_agent.py index a52517cec..ae5697b93 100644 --- a/workspace/backend/app/services/cloud_agent.py +++ b/workspace/backend/app/services/cloud_agent.py @@ -43,67 +43,91 @@ async def invoke_cloud_agents(workspace_id: str, event_data: dict) -> None: logger.warning("cloud_agent: max depth %d reached, skipping", depth) return + # Dedupe (order preserved) — duplicate mentions must not invoke the + # same cloud agent twice with duplicate, billable provider requests. + names = list(dict.fromkeys(n for n in target_agents if n != "__no_response__")) + if not names: + return + + # Invoke the targeted cloud agents concurrently — a message like + # "@a task1 @b task2" must not make b wait for a's API round-trip. + # The semaphore bounds the fan-out so a message mentioning many agents + # can't exhaust the DB connection pool or hammer provider APIs. + semaphore = asyncio.Semaphore(max(1, config.CLOUD_AGENT_MAX_CONCURRENCY)) + + async def _bounded(name: str) -> None: + async with semaphore: + await _invoke_guarded(workspace_id, event_data, name, depth) + + await asyncio.gather(*(_bounded(name) for name in names)) + + +async def _invoke_guarded( + workspace_id: str, event_data: dict, agent_name: str, depth: int, +) -> None: + """Look up one cloud agent's config and invoke it, posting an error + message to the channel on failure. DB sessions are short-lived — the + config lookup releases its connection before the (slow) provider call + so concurrent invocations don't pin pool connections while waiting on + external APIs.""" db = SessionLocal() try: - for agent_name in target_agents: - if agent_name == "__no_response__": - continue - - cloud_config = db.execute( - select(CloudAgentConfig).where( - CloudAgentConfig.workspace_id == workspace_id, - CloudAgentConfig.agent_name == agent_name, - CloudAgentConfig.status == "active", - ) - ).scalar_one_or_none() - - if not cloud_config: - continue - - try: - await _invoke_single(db, workspace_id, event_data, cloud_config, depth) - except Exception as exc: - logger.exception( - "cloud_agent: failed to invoke %s (%s/%s)", - agent_name, cloud_config.provider, cloud_config.model, - ) - error_detail = str(exc)[:200] if str(exc) else "Unknown error" - # Use a fresh DB session for error posting — the original - # session may be stale after a long async API call. - await _post_error_message( - workspace_id, event_data, agent_name, - f"Failed to get a response from {cloud_config.provider}/{cloud_config.model}: " - f"{error_detail}", - ) + cloud_config = db.execute( + select(CloudAgentConfig).where( + CloudAgentConfig.workspace_id == workspace_id, + CloudAgentConfig.agent_name == agent_name, + CloudAgentConfig.status == "active", + ) + ).scalar_one_or_none() finally: db.close() + if not cloud_config: + return + + try: + await _invoke_single(workspace_id, event_data, cloud_config, depth) + except Exception as exc: + logger.exception( + "cloud_agent: failed to invoke %s (%s/%s)", + agent_name, cloud_config.provider, cloud_config.model, + ) + error_detail = str(exc)[:200] if str(exc) else "Unknown error" + # _post_error_message opens its own fresh DB session. + await _post_error_message( + workspace_id, event_data, agent_name, + f"Failed to get a response from {cloud_config.provider}/{cloud_config.model}: " + f"{error_detail}", + ) + async def _invoke_single( - db, workspace_id: str, event_data: dict, + workspace_id: str, event_data: dict, cloud_config: CloudAgentConfig, depth: int, ) -> None: """Invoke a single cloud agent and post the response.""" - channel_target = event_data.get("target", "") - agent_name = cloud_config.agent_name - if cloud_config.category == "image": - await _invoke_image_agent(db, workspace_id, event_data, cloud_config) + await _invoke_image_agent(workspace_id, event_data, cloud_config) elif cloud_config.category == "audio": - await _invoke_audio_agent(db, workspace_id, event_data, cloud_config) + await _invoke_audio_agent(workspace_id, event_data, cloud_config) else: - await _invoke_chat_agent(db, workspace_id, event_data, cloud_config, depth) + await _invoke_chat_agent(workspace_id, event_data, cloud_config, depth) async def _invoke_chat_agent( - db, workspace_id: str, event_data: dict, + workspace_id: str, event_data: dict, cloud_config: CloudAgentConfig, depth: int, ) -> None: """Invoke a chat cloud agent.""" channel_target = event_data.get("target", "") agent_name = cloud_config.agent_name - messages = _build_conversation_context(db, workspace_id, channel_target, agent_name) + # Short session for the context read — released before the provider call. + db = SessionLocal() + try: + messages = _build_conversation_context(db, workspace_id, channel_target, agent_name) + finally: + db.close() content = event_data.get("payload", {}).get("content", "") if content: @@ -127,14 +151,18 @@ async def _invoke_chat_agent( base_url=cloud_config.base_url, ) - await _post_response( - db, workspace_id, channel_target, agent_name, - response_text, depth, - ) + db = SessionLocal() + try: + await _post_response( + db, workspace_id, channel_target, agent_name, + response_text, depth, + ) + finally: + db.close() async def _invoke_image_agent( - db, workspace_id: str, event_data: dict, + workspace_id: str, event_data: dict, cloud_config: CloudAgentConfig, ) -> None: """Invoke an image generation cloud agent.""" @@ -151,7 +179,7 @@ async def _invoke_image_agent( # concrete prompt using recent channel history. Falls back to the raw # instruction when no router LLM / no context is available. prompt = await _compose_image_prompt( - db, workspace_id, channel_target, agent_name, instruction, + workspace_id, channel_target, agent_name, instruction, ) if not prompt: return @@ -169,30 +197,35 @@ async def _invoke_image_agent( base_url=cloud_config.base_url, ) - file_id = await _upload_image( - db, workspace_id, channel_target, agent_name, - image_bytes, image_format, prompt, - ) + # Session opened only after the (slow) generation call — upload and + # response posting share it so the FileRecord commits with the message. + db = SessionLocal() + try: + file_id = await _upload_image( + db, workspace_id, channel_target, agent_name, + image_bytes, image_format, prompt, + ) - channel_name = channel_target.replace("channel/", "") if channel_target.startswith("channel/") else None - content_type = f"image/{image_format}" - filename = f"generated_{file_id[:8]}.{image_format}" - - await _post_response( - db, workspace_id, channel_target, agent_name, - f"Here's the generated image for: *{instruction[:100]}*", - depth=0, - attachments=[{ - "file_id": file_id, - "filename": filename, - "content_type": content_type, - "size": len(image_bytes), - }], - ) + content_type = f"image/{image_format}" + filename = f"generated_{file_id[:8]}.{image_format}" + + await _post_response( + db, workspace_id, channel_target, agent_name, + f"Here's the generated image for: *{instruction[:100]}*", + depth=0, + attachments=[{ + "file_id": file_id, + "filename": filename, + "content_type": content_type, + "size": len(image_bytes), + }], + ) + finally: + db.close() async def _invoke_audio_agent( - db, workspace_id: str, event_data: dict, + workspace_id: str, event_data: dict, cloud_config: CloudAgentConfig, ) -> None: """Invoke a text-to-speech cloud agent.""" @@ -215,24 +248,28 @@ async def _invoke_audio_agent( text=text, ) - file_id = await _upload_image( - db, workspace_id, channel_target, agent_name, - audio_bytes, audio_format, text, - ) + db = SessionLocal() + try: + file_id = await _upload_image( + db, workspace_id, channel_target, agent_name, + audio_bytes, audio_format, text, + ) - filename = f"speech_{file_id[:8]}.{audio_format}" - - await _post_response( - db, workspace_id, channel_target, agent_name, - f"Generated speech for: *{text[:100]}*", - depth=0, - attachments=[{ - "file_id": file_id, - "filename": filename, - "content_type": f"audio/{audio_format}", - "size": len(audio_bytes), - }], - ) + filename = f"speech_{file_id[:8]}.{audio_format}" + + await _post_response( + db, workspace_id, channel_target, agent_name, + f"Generated speech for: *{text[:100]}*", + depth=0, + attachments=[{ + "file_id": file_id, + "filename": filename, + "content_type": f"audio/{audio_format}", + "size": len(audio_bytes), + }], + ) + finally: + db.close() def _build_conversation_context( @@ -277,7 +314,7 @@ def _build_conversation_context( async def _compose_image_prompt( - db, workspace_id: str, channel_target: str, agent_name: str, instruction: str, + workspace_id: str, channel_target: str, agent_name: str, instruction: str, ) -> str: """Turn a (possibly referential) instruction like "make an image of cherie's brief above" into a concrete, self-contained image prompt by @@ -297,7 +334,12 @@ async def _compose_image_prompt( if not (config.ROUTER_LLM_ENABLED and api_key): return instruction - context = _build_conversation_context(db, workspace_id, channel_target, agent_name) + # Short session for the context read — released before the LLM call. + db = SessionLocal() + try: + context = _build_conversation_context(db, workspace_id, channel_target, agent_name) + finally: + db.close() if not context: return instruction diff --git a/workspace/backend/tests/test_cloud_agent.py b/workspace/backend/tests/test_cloud_agent.py new file mode 100644 index 000000000..7adfd5685 --- /dev/null +++ b/workspace/backend/tests/test_cloud_agent.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +""" +Tests for concurrent cloud agent invocation. + +invoke_cloud_agents fans out to every targeted agent, but must dedupe +duplicate targets (duplicate mentions would double-invoke a billable +provider API) and bound the concurrency so a large mention list can't +exhaust the DB connection pool. +""" + +import asyncio +from unittest.mock import MagicMock, patch + +from app.config import config +from app.services import cloud_agent + + +def _run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +def _event(targets, depth=0): + metadata = {"target_agents": targets} + if depth: + metadata["cloud_agent_depth"] = depth + return { + "target": "channel/general", + "payload": {"content": "hi"}, + "metadata": metadata, + } + + +class TestInvokeCloudAgents: + + def test_duplicate_targets_invoked_once(self): + calls = [] + + async def fake(workspace_id, event_data, name, depth): + calls.append(name) + + with patch.object(cloud_agent, "_invoke_guarded", side_effect=fake): + _run(cloud_agent.invoke_cloud_agents( + "ws1", _event(["agent-a", "agent-b", "agent-a", "agent-a"]), + )) + assert calls == ["agent-a", "agent-b"] + + def test_sentinel_only_no_invocations(self): + mock = MagicMock() + with patch.object(cloud_agent, "_invoke_guarded", mock): + _run(cloud_agent.invoke_cloud_agents("ws1", _event(["__no_response__"]))) + mock.assert_not_called() + + def test_sentinel_mixed_with_real_targets_is_dropped(self): + calls = [] + + async def fake(workspace_id, event_data, name, depth): + calls.append(name) + + with patch.object(cloud_agent, "_invoke_guarded", side_effect=fake): + _run(cloud_agent.invoke_cloud_agents( + "ws1", _event(["agent-a", "__no_response__"]), + )) + assert calls == ["agent-a"] + + def test_concurrency_is_capped(self): + current = 0 + peak = 0 + calls = [] + + async def fake(workspace_id, event_data, name, depth): + nonlocal current, peak + current += 1 + peak = max(peak, current) + await asyncio.sleep(0.005) + current -= 1 + calls.append(name) + + names = [f"agent-{i}" for i in range(6)] + with patch.object(config, "CLOUD_AGENT_MAX_CONCURRENCY", 2), \ + patch.object(cloud_agent, "_invoke_guarded", side_effect=fake): + _run(cloud_agent.invoke_cloud_agents("ws1", _event(names))) + + assert sorted(calls) == sorted(names), "every target must still be invoked" + assert peak <= 2, f"concurrency exceeded the cap (peak={peak})" + + def test_depth_limit_skips_all(self): + mock = MagicMock() + with patch.object(cloud_agent, "_invoke_guarded", mock): + _run(cloud_agent.invoke_cloud_agents( + "ws1", _event(["agent-a"], depth=config.CLOUD_AGENT_MAX_DEPTH), + )) + mock.assert_not_called() diff --git a/workspace/backend/tests/test_llm_router.py b/workspace/backend/tests/test_llm_router.py index 9f4965534..fe70b0b3e 100644 --- a/workspace/backend/tests/test_llm_router.py +++ b/workspace/backend/tests/test_llm_router.py @@ -10,7 +10,12 @@ from unittest.mock import patch, MagicMock from app.models import Channel, ChannelMember, WorkspaceMember, Workspace -from app.mods.workspace_mod import _route_with_llm, _master_targets, _handle_message_posted +from app.mods.workspace_mod import ( + _fallback_targets, + _handle_message_posted, + _master_targets, + _route_with_llm, +) from openagents.core.onm_events import Event from openagents.core.onm_mods import PipelineContext @@ -111,19 +116,37 @@ def test_router_stop(self, _mock_model, _mock_key, mock_get_client, db, multi_ag @patch("app.mods.workspace_mod._get_llm_client") @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") - def test_router_multiple_agents_picks_first(self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace): - """When the LLM returns comma-separated agents, only the first is used.""" + def test_router_multiple_agents_all_targeted(self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace): + """When the LLM returns comma-separated agents, ALL of them are + targeted so their tasks run in parallel.""" mock_client = MagicMock() mock_client.messages.create.return_value = _mock_anthropic_response("next:agent-master,agent-worker") mock_get_client.return_value = (mock_client, "anthropic") ws = multi_agent_workspace["workspace"] ch = multi_agent_workspace["channel"] - # Human sender so the first of the comma-list isn't a self-loop. + # Human sender so no name in the comma-list is a self-loop. event = _make_event("human:user", "channel/session-test", "Both agents need to act") result = _run(_route_with_llm(ch, event, db, ws)) - assert result == ["agent-master"] + assert result == ["agent-master", "agent-worker"] + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_router_multiple_agents_self_loop_dropped_others_kept(self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace): + """A multi-name router answer that includes the sender keeps the + other agents and only drops the self-loop.""" + mock_client = MagicMock() + mock_client.messages.create.return_value = _mock_anthropic_response("next:agent-master,agent-worker") + mock_get_client.return_value = (mock_client, "anthropic") + + ws = multi_agent_workspace["workspace"] + ch = multi_agent_workspace["channel"] + event = _make_event("openagents:agent-master", "channel/session-test", "delegating work") + + result = _run(_route_with_llm(ch, event, db, ws)) + assert result == ["agent-worker"] @patch("app.mods.workspace_mod._get_llm_client") @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") @@ -235,6 +258,12 @@ def test_master_delegates_via_mention(self, db, multi_agent_workspace): "@agent-worker please handle this") assert _master_targets(event, ch, ["agent-worker"]) == ["agent-worker"] + def test_master_duplicate_delegation_deduped(self, db, multi_agent_workspace): + ch = multi_agent_workspace["channel"] + event = _make_event("openagents:agent-master", "channel/session-test", + "@agent-worker do X, and @agent-worker also do Y") + assert _master_targets(event, ch, ["agent-worker", "agent-worker"]) == ["agent-worker"] + def test_master_no_mention_stops(self, db, multi_agent_workspace): ch = multi_agent_workspace["channel"] # Master answered the human directly (no delegation) → stop @@ -374,6 +403,222 @@ def test_openai_router_stop(self, _mock_model, _mock_key, mock_get_client, db, m assert result == [] +class TestParallelMentionRouting: + """Explicit @mentions fan out to ALL mentioned agents so tasks assigned + to different agents in one channel run in parallel instead of being + funneled through the single-target LLM router and queued on one agent.""" + + def test_fallback_returns_all_mentions(self, db, multi_agent_workspace): + ch = multi_agent_workspace["channel"] + event = _make_event("human:user", "channel/session-test", + "@agent-master do X and @agent-worker do Y") + result = _fallback_targets(event, ch, ["agent-master", "agent-worker"]) + assert result == ["agent-master", "agent-worker"] + + def test_fallback_dedupes_repeated_mentions(self, db, multi_agent_workspace): + ch = multi_agent_workspace["channel"] + event = _make_event("human:user", "channel/session-test", + "@agent-worker do X, @agent-worker also do Y") + result = _fallback_targets(event, ch, ["agent-worker", "agent-worker"]) + assert result == ["agent-worker"] + + def test_fallback_filters_agent_self_mention(self, db, multi_agent_workspace): + ch = multi_agent_workspace["channel"] + event = _make_event("openagents:agent-master", "channel/session-test", + "@agent-master note to self, @agent-worker take over") + result = _fallback_targets(event, ch, ["agent-master", "agent-worker"]) + assert result == ["agent-worker"] + + def test_fallback_all_self_mentions_returns_empty(self, db, multi_agent_workspace): + ch = multi_agent_workspace["channel"] + event = _make_event("openagents:agent-master", "channel/session-test", + "@agent-master note to self") + result = _fallback_targets(event, ch, ["agent-master"]) + assert result == [] + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_human_mentions_bypass_llm_router( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + """Dynamic mode, human message with explicit @mentions → all + mentioned agents targeted directly; the LLM router is never called.""" + mock_client = MagicMock() + mock_get_client.return_value = (mock_client, "anthropic") + + ws = multi_agent_workspace["workspace"] + event = _make_event("human:user", "channel/session-test", + "@agent-master analyze the logs @agent-worker fix the tests") + ctx = PipelineContext(network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws) + + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata.get("target_agents") == ["agent-master", "agent-worker"] + mock_client.messages.create.assert_not_called() + + def test_master_mode_still_routes_human_to_master(self, db, multi_agent_workspace): + """Master mode keeps its star topology — a human @mentioning a + sub-agent still goes through the master hub.""" + ws = multi_agent_workspace["workspace"] + ch = multi_agent_workspace["channel"] + ch.orchestration_mode = "master" + db.flush() + event = _make_event("human:user", "channel/session-test", + "@agent-worker please handle this") + ctx = PipelineContext(network_id=str(ws.id), agent_address="human:user", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + assert out.metadata.get("target_agents") == ["agent-master"] + + +class TestDirectAssignmentNotReroutedToPeers: + """While agents work in parallel on tasks a human handed them by name, + their own progress notes and results must not be routed to a peer. + + Without this, each worker's "running the command now" note is handed to + the LLM router, which cannot tell it from a handoff and wakes a + bystander agent — who then answers a task it was never given, once its + own job drains off the queue. + """ + + def _record_human(self, db, ws, content, ts): + from app.models import EventRecord + db.add(EventRecord( + id=f"evt-{ts}", + network_id=ws.id, + type="workspace.message.posted", + source="human:user", + target="channel/session-test", + payload={"content": content, "message_type": "chat"}, + metadata_={}, + timestamp=ts, + )) + db.flush() + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_assigned_agents_progress_note_stops( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + mock_client = MagicMock() + mock_get_client.return_value = (mock_client, "anthropic") + ws = multi_agent_workspace["workspace"] + self._record_human(db, ws, "@agent-worker run task B", 1000) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "Running the command now, back in a minute.") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["__no_response__"] + mock_client.messages.create.assert_not_called() + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_assignment_survives_a_later_task_for_another_agent( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + """Parallel assignments arrive as separate messages, one per agent. + A newer "@agent-master do X" must not cancel agent-worker's own + assignment — that is exactly the case that produced the misroute.""" + mock_client = MagicMock() + mock_get_client.return_value = (mock_client, "anthropic") + ws = multi_agent_workspace["workspace"] + self._record_human(db, ws, "@agent-worker run task B", 1000) + self._record_human(db, ws, "@agent-master run task A", 1001) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "Task B done: 60.00s.") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["__no_response__"] + mock_client.messages.create.assert_not_called() + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_explicit_handoff_still_routes( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + """An assigned agent that @mentions a peer is delegating — the + router still decides, so handoffs keep working.""" + mock_client = MagicMock() + mock_client.messages.create.return_value = _mock_anthropic_response("next:agent-master") + mock_get_client.return_value = (mock_client, "anthropic") + ws = multi_agent_workspace["workspace"] + self._record_human(db, ws, "@agent-worker run task B", 1000) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "@agent-master can you review this?") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["agent-master"] + mock_client.messages.create.assert_called_once() + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_unaddressed_human_message_reopens_routing( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + """Free-form human chat (no @mention) ends the direct-assignment + window — agent replies go back through the router.""" + mock_client = MagicMock() + mock_client.messages.create.return_value = _mock_anthropic_response("next:agent-master") + mock_get_client.return_value = (mock_client, "anthropic") + ws = multi_agent_workspace["workspace"] + self._record_human(db, ws, "@agent-worker run task B", 1000) + self._record_human(db, ws, "what does everyone think?", 1002) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "I think we should ship it.") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["agent-master"] + mock_client.messages.create.assert_called_once() + + @patch("app.mods.workspace_mod._get_llm_client") + @patch("app.mods.workspace_mod._get_router_api_key", return_value="test-key") + @patch("app.mods.workspace_mod._get_router_model", return_value="claude-haiku-4-5-20251001") + def test_never_assigned_agent_still_routes( + self, _mock_model, _mock_key, mock_get_client, db, multi_agent_workspace, + ): + """An agent the human never addressed by name is unaffected.""" + mock_client = MagicMock() + mock_client.messages.create.return_value = _mock_anthropic_response("next:agent-master") + mock_get_client.return_value = (mock_client, "anthropic") + ws = multi_agent_workspace["workspace"] + self._record_human(db, ws, "@agent-master run task A", 1000) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "Here are my findings.") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["agent-master"] + mock_client.messages.create.assert_called_once() + + def test_master_mode_unaffected(self, db, multi_agent_workspace): + """Master mode's star topology still returns sub-agent output to + the hub — the assignment rule only applies to dynamic mode.""" + ws = multi_agent_workspace["workspace"] + ch = multi_agent_workspace["channel"] + ch.orchestration_mode = "master" + db.flush() + self._record_human(db, ws, "@agent-worker run task B", 1000) + + event = _make_event("openagents:agent-worker", "channel/session-test", + "Task B done.") + ctx = PipelineContext(network_id=str(ws.id), agent_address="openagents:agent-worker", db=db, workspace=ws) + out = _run(_handle_message_posted(event, ctx)) + + assert out.metadata.get("target_agents") == ["agent-master"] + + class TestMessagePostedTargetAgents: """_handle_message_posted must ALWAYS set target_agents, even when routing decides nobody should respond. Otherwise legacy clients