diff --git a/src/backend/orchestration/orchestration_manager.py b/src/backend/orchestration/orchestration_manager.py index df43f8d47..831cf9649 100644 --- a/src/backend/orchestration/orchestration_manager.py +++ b/src/backend/orchestration/orchestration_manager.py @@ -1,6 +1,7 @@ """Orchestration manager (agent_framework version) handling multi-agent Magentic workflow creation and execution.""" import asyncio +import json import logging import uuid import re @@ -198,6 +199,13 @@ async def init_orchestration( len(participant_list), ) + # Attach context needed for the pre-planning team-scope gate + # (see run_orchestration → _evaluate_team_scope). Stored on the workflow + # so the gate can classify a request against this team's agents/data + # without rebuilding a chat client. + workflow._team_config = team_config + workflow._manager_chat_client = manager_chat_client + return workflow # --------------------------- @@ -364,6 +372,25 @@ async def run_orchestration(self, user_id: str, input_task) -> None: task_text = getattr(input_task, "description", str(input_task)) self.logger.debug("Task: %s", task_text) + # ---- Team-scope gate (generic, team-agnostic) ------------------- + + scope = await self._evaluate_team_scope(workflow, task_text) + if scope is not None and not scope.get("in_scope", True): + self.logger.info( + "Request judged OUT OF SCOPE for team; presenting single " + "MagenticManager out-of-scope step (job='%s')", job_id, + ) + + team_agent_names = self._get_team_agent_names(workflow) + await self._handle_out_of_scope( + user_id=user_id, + task_text=task_text, + out_of_scope_message=scope.get("message", ""), + team_agent_names=team_agent_names, + ) + await self._cleanup_workflow_mcp(user_id) + return + try: final_output_ref: list = [None] orchestrator_chunks: list[str] = [] @@ -534,6 +561,244 @@ async def _cleanup_workflow_mcp(self, user_id: str) -> None: # Mark workflow as terminated so next request creates a fresh one workflow._terminated = True + # --------------------------- + # Team-scope gate + # --------------------------- + async def _evaluate_team_scope(self, workflow, task_text: str) -> Optional[dict]: + """Classify whether ``task_text`` is within the current team's scope. + + The decision is made by a focused single-purpose classifier call using + the manager chat client, given the team's purpose, its agents (and the + data/knowledge each works on), and representative example tasks. This is + deliberately separate from the planning prompt so the strict scope + decision is not diluted by the "include every agent" planning rules. + + Returns: + ``{"in_scope": bool, "message": str}`` when the classification + succeeds, or ``None`` when it cannot be evaluated (missing context or + an error) — in which case the caller proceeds normally (fail-open). + """ + team_config = getattr(workflow, "_team_config", None) + chat_client = getattr(workflow, "_manager_chat_client", None) + if team_config is None or chat_client is None or not task_text: + return None + + try: + agent_lines = [] + for ag in getattr(team_config, "agents", []) or []: + name = getattr(ag, "name", "") or "" + desc = getattr(ag, "description", "") or "" + data = getattr(ag, "knowledge_base_name", "") or "" + line = f"- {name}: {desc}" + if data: + line += f" (works on data: {data})" + agent_lines.append(line) + agents_block = "\n".join(agent_lines) or "- (no agents listed)" + + example_lines = [] + for t in getattr(team_config, "starting_tasks", []) or []: + tname = getattr(t, "name", "") or "" + tprompt = getattr(t, "prompt", "") or "" + example_lines.append(f"- {tname}: {tprompt}".strip()) + examples_block = "\n".join(example_lines) or "- (none provided)" + + system_prompt = ( + "You are a strict scope classifier for a specialized multi-agent " + "team. Decide whether a user's request falls within THIS team's " + "specialization.\n\n" + "A team is defined ENTIRELY by its stated purpose, the specific " + "agents it has and what each does, the data/knowledge those agents " + "work with, and its representative example tasks.\n\n" + "Rules:\n" + "- IN SCOPE only if the request clearly matches this team's " + "specialization and could be fulfilled by these agents using their " + "data.\n" + "- OUT OF SCOPE if the request belongs to a DIFFERENT " + "specialization, even when superficially related or in a broadly " + "similar field (e.g. drafting a product press release is NOT the " + "same specialization as generating retail social-media content; HR " + "onboarding is NOT product marketing; contract/NDA compliance is " + "NOT RFP evaluation).\n" + "- If the request is genuinely ambiguous or a reasonable subset of " + "the example tasks, treat it as IN SCOPE.\n\n" + "Respond with ONLY a compact JSON object and nothing else:\n" + '{"in_scope": true|false, "reason": "", ' + '"message": ""}' + ) + user_prompt = ( + f"TEAM NAME: {getattr(team_config, 'name', '')}\n" + f"TEAM PURPOSE: {getattr(team_config, 'description', '')}\n\n" + f"AGENTS:\n{agents_block}\n\n" + f"EXAMPLE IN-SCOPE TASKS:\n{examples_block}\n\n" + f"USER REQUEST:\n{task_text}" + ) + + response = await chat_client.get_response( + [Message("system", [system_prompt]), + Message("user", [user_prompt])] + ) + raw = (getattr(response, "text", "") or "").strip() + self.logger.info("[SCOPE-GATE] classifier raw response: %s", raw[:500]) + + parsed = self._parse_scope_json(raw) + if parsed is None: + self.logger.warning( + "[SCOPE-GATE] Could not parse classifier output — proceeding " + "normally (fail-open)." + ) + return None + + in_scope = bool(parsed.get("in_scope", True)) + message = str(parsed.get("message", "") or "").strip() + if not in_scope and not message: + message = ( + "This request appears to be outside the scope of the selected " + "team, so it cannot be handled reliably here. Please switch to " + "the appropriate team and try again." + ) + self.logger.info( + "[SCOPE-GATE] in_scope=%s reason=%s", + in_scope, parsed.get("reason", ""), + ) + return {"in_scope": in_scope, "message": message} + except Exception as e: # fail-open: never block a task on classifier error + self.logger.warning( + "[SCOPE-GATE] Scope evaluation failed (%s) — proceeding normally.", e + ) + return None + + @staticmethod + def _parse_scope_json(text: str) -> Optional[dict]: + """Extract the first JSON object from a classifier response.""" + if not text: + return None + cleaned = text.strip() + if cleaned.startswith("```"): + cleaned = "\n".join( + ln for ln in cleaned.splitlines() if not ln.strip().startswith("```") + ).strip() + try: + return json.loads(cleaned) + except (json.JSONDecodeError, ValueError): + pass + m = re.search(r"\{.*\}", cleaned, re.DOTALL) + if m: + try: + return json.loads(m.group(0)) + except (json.JSONDecodeError, ValueError): + return None + return None + + @staticmethod + def _get_team_agent_names(workflow) -> list[str]: + """Return the plan ``team`` roster for the frontend "Agent Team" panel. + + Mirrors the normal plan exactly: ``run_orchestration`` builds its + ``participant_names`` from ``workflow.get_executors_list()`` executor ids + (which include the ``magentic_orchestrator`` shown as "Magentic + Orchestrator"). Using the same source keeps the out-of-scope Agent Team + identical to a normal plan's. Falls back to the stored team config's + agent names when executors are unavailable. + """ + try: + names = [ + executor.id + for executor in workflow.get_executors_list() + if getattr(executor, "id", "") + ] + if names: + return names + except Exception: + pass + + team_config = getattr(workflow, "_team_config", None) + return [ + getattr(ag, "name", "") + for ag in getattr(team_config, "agents", []) or [] + if getattr(ag, "name", "") + ] + + async def _handle_out_of_scope( + self, + *, + user_id: str, + task_text: str, + out_of_scope_message: str, + team_agent_names: Optional[list[str]] = None, + ) -> None: + """Present a single MagenticManager out-of-scope step for approval, then + deliver the out-of-scope notice as the final answer (no agents run).""" + from models.plan_models import MPlan, MStep + + message = out_of_scope_message or ( + "This request appears to be outside the scope of the selected team. " + "Please switch to the appropriate team and try again." + ) + + team = list(team_agent_names) if team_agent_names else ["MagenticManager"] + + mplan = MPlan() + mplan.user_id = user_id + mplan.user_request = task_text + mplan.team = team + mplan.steps = [ + MStep( + agent="MagenticManager", + action=( + "Inform the user that this request is out of scope for the " + "selected team and suggest a suitable team." + ), + ) + ] + + try: + orchestration_config.plans[mplan.id] = mplan + except Exception as e: + self.logger.error("Error storing out-of-scope plan: %s", e) + + approval_message = messages.PlanApprovalRequest( + plan=mplan, + status="PENDING_APPROVAL", # type: ignore[arg-type] + context={"task": task_text, "out_of_scope": True}, + ) + await connection_config.send_status_update_async( + message=approval_message, + user_id=user_id, + message_type=WebsocketMessageType.PLAN_APPROVAL_REQUEST, + ) + + approval_response = await wait_for_plan_approval(mplan.id, user_id) + + if approval_response and approval_response.approved: + self.logger.info("Out-of-scope plan approved — sending final notice.") + await asyncio.sleep(1.5) + await connection_config.send_status_update_async( + { + "type": WebsocketMessageType.FINAL_RESULT_MESSAGE, + "data": { + "content": message, + "status": "completed", + "timestamp": asyncio.get_event_loop().time(), + }, + }, + user_id, + message_type=WebsocketMessageType.FINAL_RESULT_MESSAGE, + ) + else: + self.logger.info("Out-of-scope plan rejected by user.") + await connection_config.send_status_update_async( + { + "type": WebsocketMessageType.PLAN_APPROVAL_RESPONSE, + "data": approval_response, + }, + user_id=user_id, + message_type=WebsocketMessageType.PLAN_APPROVAL_RESPONSE, + ) + # --------------------------- # Plan review handling # --------------------------- diff --git a/src/backend/orchestration/plan_review_helpers.py b/src/backend/orchestration/plan_review_helpers.py index 1339e3312..e8ff9d4b4 100644 --- a/src/backend/orchestration/plan_review_helpers.py +++ b/src/backend/orchestration/plan_review_helpers.py @@ -84,7 +84,9 @@ def get_magentic_prompt_kwargs( agent_lines = "\n".join(f"- {name}" for name in participant_names) mandatory_block = ( "\n\nMANDATORY AGENTS (CRITICAL — NON-NEGOTIABLE):\n" - "Every plan you generate MUST include EVERY ONE of the following agents,\n" + "UNLESS the request is out of scope (see TEAM SCOPE POLICY above — in\n" + "which case the single MagenticManager step is the entire plan),\n" + "every plan you generate MUST include EVERY ONE of the following agents,\n" "each as its own distinct step, each invoked at least once:\n" + agent_lines + "\nDo NOT omit any of these agents, even if a step seems optional,\n" @@ -93,7 +95,28 @@ def get_magentic_prompt_kwargs( "is INVALID — regenerate it\nuntil every listed agent appears as a step.\n" ) - plan_append = """ + + scope_policy = """ + +TEAM SCOPE POLICY (CRITICAL — EVALUATE THIS FIRST, BEFORE ANY OTHER RULE): +This team can ONLY handle requests that fall within the collective expertise of +its listed agents. Each agent's description above tells you its domain and the +data/knowledge it works on — that, and nothing else, defines the team's scope. + +Judge whether the user's request is covered by at least one listed agent's domain: +- IN SCOPE: at least one agent's expertise/data is clearly relevant → plan normally. +- OUT OF SCOPE: NO listed agent's domain covers the request (e.g. a contract / + compliance request given to a marketing or RFP team, or an HR / onboarding + request given to a product-marketing team). In this case you MUST NOT invoke any + domain agent. Output a plan with EXACTLY ONE step and nothing else: + [{{"agent": "MagenticManager", "action": "Inform the user that this request is out of scope for this team and that they should switch to the appropriate team and try again, without naming any specific team."}}] + +When out of scope, the mandatory-inclusion rule below does NOT apply — the lone +MagenticManager step IS the complete, valid plan. Only take this path when the +request is genuinely outside every agent's domain; when in doubt, plan normally. +""" + + plan_append = scope_policy + """ PLAN RULES: - Steps are HIGH-LEVEL task assignments — one step per agent. Do NOT prescribe @@ -137,6 +160,11 @@ def get_magentic_prompt_kwargs( final_append = """ FINAL ANSWER RULES: +- If the approved plan was a single out-of-scope MagenticManager step (no domain + agents ran), your final answer is a brief, polite message that states the + request is out of scope for this team and that the user should switch to the + appropriate team and try again. Do NOT name, recommend, or guess any specific + team, and do NOT attempt to answer the out-of-scope request itself. - Compile ONLY from messages agents actually produced. Quote verbatim where appropriate. - Do NOT fabricate URLs, results, or content that no agent produced. - If a required agent step did not run, state it plainly — do not pretend it did. @@ -183,6 +211,10 @@ def get_magentic_prompt_kwargs( work agents. It only routes tasks and compiles the final output. COMPLETION CHECK (CRITICAL): +- OUT-OF-SCOPE PLAN: If the approved plan is a single MagenticManager out-of-scope + step (no domain agents), there is nothing to route. Set is_request_satisfied to + true immediately and produce the out-of-scope final answer — do NOT select any + next_speaker and do NOT invoke domain agents. Before setting is_request_satisfied to true, you MUST verify: 1. Review the conversation history and list every agent that has actually produced a substantive response (meaningful output — calling tools and returning results