Prevent model protocol envelopes from leaking into chat

backend/app/api/enterprise.py

@@ -31,7 +31,6 @@
 from app.services.autonomy_service import autonomy_service
 from app.services.enterprise_sync import enterprise_sync_service
 from app.services.llm import get_provider_manifest, get_model_api_key, create_llm_client, LLMMessage
-from app.services.llm.finish import FINISH_TOOL_DEFINITION, find_finish_call
 from app.services.platform_service import platform_service
 from app.services.sso_service import sso_service
 from app.services.agent_runtime.runtime_model_settings import (
@@ -43,6 +42,41 @@
 router = APIRouter(prefix="/enterprise", tags=["enterprise"])
 settings = get_settings()
 
+_CAPABILITY_PROBE_TOOL_DEFINITION = {
+    "type": "function",
+    "function": {
+        "name": "capability_probe",
+        "description": "Return the fixed value through a native structured tool call.",
+        "parameters": {
+            "type": "object",
+            "properties": {"value": {"type": "string", "enum": ["ok"]}},
+            "required": ["value"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+def _has_valid_capability_probe(tool_calls: list[dict]) -> bool:
+    for call in tool_calls:
+        function = call.get("function")
+        if not isinstance(function, dict) or function.get("name") != "capability_probe":
+            continue
+        raw_arguments = function.get("arguments", "{}")
+        try:
+            arguments = (
+                json.loads(raw_arguments)
+                if isinstance(raw_arguments, str)
+                else dict(raw_arguments)
+                if isinstance(raw_arguments, dict)
+                else None
+            )
+        except (TypeError, ValueError, json.JSONDecodeError):
+            continue
+        if arguments == {"value": "ok"}:
+            return True
+    return False
+
 
 def _is_platform_admin_user(user: User) -> bool:
     """Return true for tenant-role or identity-level platform admins."""
@@ -215,7 +249,7 @@ async def test_llm_model(
     data: LLMTestRequest,
     current_user: User = Depends(get_current_admin),
 ):
-    """Test connectivity and native ``finish`` tool calling independently."""
+    """Test connectivity and native structured tool calling independently."""
     import time
 
     start = time.time()
@@ -269,32 +303,27 @@ async def test_llm_model(
                         role="system",
                         content=(
                             "This is a native tool-calling protocol test. Call the "
-                            "provided finish tool exactly once and do not answer in text."
+                            "provided capability_probe tool with value set to ok."
                         ),
                     ),
                     LLMMessage(
                         role="user",
-                        content="Call finish now with content set to ok.",
+                        content="Call capability_probe now with value set to ok.",
                     ),
                 ],
-                tools=[FINISH_TOOL_DEFINITION],
+                tools=[_CAPABILITY_PROBE_TOOL_DEFINITION],
                 max_tokens=128,
             )
             tool_calls = list(tool_response.tool_calls or [])
-            finish_call = find_finish_call(tool_calls)
-            tool_supported = bool(
-                len(tool_calls) == 1
-                and finish_call is not None
-                and finish_call.valid
-            )
+            tool_supported = _has_valid_capability_probe(tool_calls)
             if not tool_supported:
                 tool_error = (
                     "Model returned plain text or an invalid tool call instead of "
-                    "exactly one valid finish tool call."
+                    "a valid capability_probe(value=ok) tool call."
                 )
         except Exception as exc:
             tool_supported = None
-            tool_error = f"Native finish tool probe failed: {type(exc).__name__}: {exc}"[:500]
+            tool_error = f"Native tool probe failed: {type(exc).__name__}: {exc}"[:500]
         tool_latency_ms = int((time.time() - tool_start) * 1000)
         capability_recorded = await _record_llm_tool_capability(
             target,

backend/app/api/tools.py

@@ -470,7 +470,7 @@ async def update_agent_tools(
         if not tool_obj:
             raise HTTPException(status_code=404, detail="Tool not found")
 
-        # System-category tools (e.g. finish) are protocol-level and
+        # System-category tools are protocol-level and
         # must always remain enabled — reject any attempt to disable them.
         if tool_obj.category == "system" and not u.enabled:
             continue

backend/app/services/agent_context.py

@@ -326,8 +326,8 @@ async def _load_company_information(db, agent_id: uuid.UUID) -> str:
 
 # Runtime Protocol
 
-- When the task is complete, call `finish` with the exact final answer for the user.
-- Do not call `finish` with another tool or while required work is incomplete.
+- When the task is complete, return the exact final answer as normal Assistant content.
+- Do not return a final answer while required work or Tool Calls are still incomplete.
 - When progress genuinely requires user input, approval, another Agent result, or
   an external event, call `wait` with a concise reason.
 - Do not simulate Runtime control tools in plain text.
@@ -360,7 +360,7 @@ async def _load_company_information(db, agent_id: uuid.UUID) -> str:
 
 # Verification
 
-Before calling `finish`, verify that:
+Before returning the final Assistant response, verify that:
 - Every material user requirement has been addressed.
 - Required tool actions actually succeeded.
 - Required files, records, messages, or other artifacts exist.

backend/app/services/agent_runtime/checkpoint_side_effects.py

@@ -694,6 +694,7 @@ async def handle(
                 "reason": command.payload.get("reason") or "cancelled_by_command",
                 "waiting_request": None,
             }
+            lifecycle.pop("pending_group_at", None)
             product_checkpoint = replace(
                 checkpoint,
                 state={**checkpoint.state, "lifecycle": lifecycle},

backend/app/services/agent_runtime/group_at.py

@@ -0,0 +1,87 @@
+"""Group-only structured mention intent used before a natural final response."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from copy import deepcopy
+from typing import Any
+import uuid
+
+
+AT_TOOL_NAME = "at"
+MAX_GROUP_AT_PARTICIPANTS = 100
+
+AT_TOOL_DEFINITION: dict[str, Any] = {
+    "type": "function",
+    "function": {
+        "name": AT_TOOL_NAME,
+        "description": (
+            "Set the complete list of Group Agents that must be visibly mentioned "
+            "and woken by the next final public reply. This only stages routing and "
+            "does not send a message or finish the Run."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "participant_ids": {
+                    "type": "array",
+                    "items": {"type": "string", "format": "uuid"},
+                    "maxItems": MAX_GROUP_AT_PARTICIPANTS,
+                    "uniqueItems": True,
+                }
+            },
+            "required": ["participant_ids"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+class GroupAtArgumentsError(ValueError):
+    """The model supplied an invalid group ``at`` target set."""
+
+
+def group_at_tool_definition() -> dict[str, Any]:
+    return deepcopy(AT_TOOL_DEFINITION)
+
+
+def parse_group_at_participant_ids(arguments: Mapping[str, object]) -> tuple[str, ...]:
+    unsupported = set(arguments) - {"participant_ids"}
+    if unsupported:
+        raise GroupAtArgumentsError(
+            "`at` contains unsupported fields: "
+            + ", ".join(sorted(str(field) for field in unsupported))
+        )
+    raw_ids = arguments.get("participant_ids")
+    if not isinstance(raw_ids, list):
+        raise GroupAtArgumentsError("`at.participant_ids` must be an array")
+    if len(raw_ids) > MAX_GROUP_AT_PARTICIPANTS:
+        raise GroupAtArgumentsError(
+            f"`at.participant_ids` may contain at most {MAX_GROUP_AT_PARTICIPANTS} entries"
+        )
+    normalized: list[str] = []
+    for raw_id in raw_ids:
+        if not isinstance(raw_id, str):
+            raise GroupAtArgumentsError(
+                "`at.participant_ids` must contain only UUID strings"
+            )
+        try:
+            participant_id = str(uuid.UUID(raw_id))
+        except ValueError as exc:
+            raise GroupAtArgumentsError(
+                "`at.participant_ids` must contain only valid UUID strings"
+            ) from exc
+        if participant_id in normalized:
+            raise GroupAtArgumentsError("`at.participant_ids` must contain unique UUIDs")
+        normalized.append(participant_id)
+    return tuple(normalized)
+
+
+__all__ = [
+    "AT_TOOL_DEFINITION",
+    "AT_TOOL_NAME",
+    "GroupAtArgumentsError",
+    "MAX_GROUP_AT_PARTICIPANTS",
+    "group_at_tool_definition",
+    "parse_group_at_participant_ids",
+]

backend/app/services/agent_runtime/group_handoff.py

@@ -1,7 +1,7 @@
 """Terminal public-mention handoff for native Group Agent Runs.
 
-The model submits participant IDs through the shared ``finish`` tool.  This
-module validates every target before the source Run can become terminal, freezes
+The model stages participant IDs through the Group-only ``at`` tool. This
+module validates the staged targets and final Assistant response, then freezes
 one immutable delivery intent, and later applies that exact intent inside the
 ordinary Runtime delivery transaction.
 """
@@ -609,7 +609,7 @@ async def preflight_group_agent_handoff(
     if state["lifecycle"].get("status") != "running":
         raise GroupAgentHandoffError(
             "group_handoff_source_invalid",
-            "A handoff finish may be submitted only by a running Group Agent Run",
+            "A Group handoff may be submitted only by a running Group Agent Run",
             repairable=False,
         )
     tenant_id = _context_uuid(context.tenant_id, field="tenant_id")