diff --git a/backend/app/api/enterprise.py b/backend/app/api/enterprise.py index f506efdae..84cdd7d06 100644 --- a/backend/app/api/enterprise.py +++ b/backend/app/api/enterprise.py @@ -936,7 +936,6 @@ async def _runtime_model_settings_payload(db: AsyncSession, *, tenant_id: uuid.U or_(LLMModel.tenant_id.is_(None), LLMModel.tenant_id == tenant_id), LLMModel.enabled.is_(True), LLMModel.deleted_at.is_(None), - LLMModel.supports_tool_calling.is_(True), ) .order_by(LLMModel.created_at.desc()) ) @@ -1000,12 +999,6 @@ async def update_runtime_model_settings( raise HTTPException(status_code=422, detail=f"Model {model_id} belongs to another tenant") if not model.enabled: raise HTTPException(status_code=422, detail=f"Model {model_id} is disabled") - if model.supports_tool_calling is not True: - raise HTTPException( - status_code=422, - detail=f"Model {model_id} has not passed the native tool-calling test", - ) - result = await db.execute( select(SystemSetting).where( SystemSetting.key == runtime_model_setting_key(resolved_tenant_id) diff --git a/backend/app/api/groups.py b/backend/app/api/groups.py index 55af6aa5e..02a5870d0 100644 --- a/backend/app/api/groups.py +++ b/backend/app/api/groups.py @@ -1596,6 +1596,14 @@ async def download_group_workspace_file( } if allow_inline and media_type == "image/svg+xml": headers["Content-Security-Policy"] = "sandbox; default-src 'none'; style-src 'unsafe-inline'" + _stage_audit( + db, + current_user=current_user, + action="group:workspace_download", + tenant_id=tenant_id, + group_id=group_id, + details={"path": value.path, "inline": allow_inline}, + ) return Response(content=value.content, media_type=media_type, headers=headers) diff --git a/backend/app/api/websocket.py b/backend/app/api/websocket.py index 93850cd51..2ced670b8 100644 --- a/backend/app/api/websocket.py +++ b/backend/app/api/websocket.py @@ -162,6 +162,14 @@ async def is_user_viewing_session(self, agent_id: str, session_id: str, user_id: manager = ConnectionManager() +def _websocket_content_log_summary(content: object) -> str: + """Return payload-free metadata for one inbound WebSocket message.""" + if not isinstance(content, str): + return f"content_type={type(content).__name__}" + image_count = content.count("[image_data:data:image/") + return f"content_chars={len(content)} image_count={image_count}" + + def _runtime_error_packet( *, code: str, @@ -680,7 +688,7 @@ async def _accept_client_message( override_model_id = data.get("model_id") is_onboarding_trigger = data.get("kind") == "onboarding_trigger" logger.info( - f"[WS] Received: {str(content)[:50]}" + f"[WS] Received: {_websocket_content_log_summary(content)}" + (" [onboarding]" if is_onboarding_trigger else "") ) if not isinstance(content, str) or (not content and not is_onboarding_trigger): diff --git a/backend/app/services/agent_runtime/adapter.py b/backend/app/services/agent_runtime/adapter.py index 371e103e3..38a573add 100644 --- a/backend/app/services/agent_runtime/adapter.py +++ b/backend/app/services/agent_runtime/adapter.py @@ -20,10 +20,6 @@ StartRunCommand, ) from app.services.agent_runtime.graph import RuntimeGraphIdentity -from app.services.agent_runtime.model_capabilities import ( - ModelCapabilityError, - ModelCapabilityResolver, -) from app.services.agent_runtime.persistence import ( RunRegistration, enqueue_cancel, @@ -156,7 +152,7 @@ async def _require_agent_runtime_model( command: StartRunCommand, agent: Agent | None, ) -> uuid.UUID | None: - """Reject new tool-driven Agent Runs before any durable Run is created.""" + """Pin an active tenant-valid model before a durable Run is created.""" if command.run_kind == "orchestration": return command.model_id model = await load_active_model( @@ -165,20 +161,12 @@ async def _require_agent_runtime_model( tenant_id=command.tenant_id, ) if model is None and agent is not None: - model = await resolve_active_agent_model( - self._db, - agent, - require_tool_calling=True, - ) + model = await resolve_active_agent_model(self._db, agent) if model is None: raise RuntimeAdapterError( "model_unavailable", - "Agent Runtime has no active tool-capable model in the command tenant", + "Agent Runtime has no active model in the command tenant", ) - try: - ModelCapabilityResolver.require_native_tool_calling(model) - except ModelCapabilityError as exc: - raise RuntimeAdapterError(exc.code, str(exc)) from exc return model.id @staticmethod diff --git a/backend/app/services/agent_runtime/checkpoint_side_effects.py b/backend/app/services/agent_runtime/checkpoint_side_effects.py index ffd3d8d0a..6d7024776 100644 --- a/backend/app/services/agent_runtime/checkpoint_side_effects.py +++ b/backend/app/services/agent_runtime/checkpoint_side_effects.py @@ -467,6 +467,21 @@ async def _record_lifecycle_events( agent_id = uuid.UUID(run.agent_id) if run.agent_id is not None else None events: list[tuple[str, str, dict, str, str | None]] = [] if checkpoint is None: + terminal_result = await db.execute( + select(AgentRunEvent.id) + .where( + AgentRunEvent.tenant_id == run.tenant_id, + AgentRunEvent.run_id == run.run_id, + AgentRunEvent.event_type.in_( + ("run_completed", "run_failed", "run_cancelled") + ), + ) + .limit(1) + ) + if terminal_result.scalar_one_or_none() is not None: + # A later cancel command may release control resources, but it must + # not append a second, contradictory terminal outcome. + return events.append( ( "run_cancelled", diff --git a/backend/app/services/agent_runtime/delivery.py b/backend/app/services/agent_runtime/delivery.py index ebe3cc466..fd4ca77b5 100644 --- a/backend/app/services/agent_runtime/delivery.py +++ b/backend/app/services/agent_runtime/delivery.py @@ -29,6 +29,9 @@ from app.services.participant_identity import get_or_create_agent_participant +# ``ack`` remains readable only for historical delivery receipts created before +# native Group start acknowledgements were retired. New requests accept only +# waiting and terminal delivery kinds. DeliveryKind = Literal["ack", "waiting", "terminal"] DeliveryLifecycleStatus = Literal[ "waiting_user", @@ -85,8 +88,6 @@ class DeliveryRequest: @property def idempotency_key(self) -> str: - if self.kind == "ack": - return f"run:{self.run_id}:ack" if self.kind == "waiting": return f"run:{self.run_id}:waiting:{self.interrupt_id}" return f"run:{self.run_id}:terminal:{self.lifecycle_status}" @@ -152,7 +153,7 @@ def _require_text( def _validate_request(request: DeliveryRequest) -> None: - if request.kind not in {"ack", "waiting", "terminal"}: + if request.kind not in {"waiting", "terminal"}: raise DeliveryServiceError( "invalid_delivery_request", f"unsupported delivery kind: {request.kind!r}", @@ -162,22 +163,6 @@ def _validate_request(request: DeliveryRequest) -> None: "invalid_delivery_request", "original_target_outcome must be not_attempted or unknown", ) - if request.kind == "ack": - _require_text(request.content, field="content") - if ( - request.checkpoint_id is not None - or request.lifecycle_status is not None - or request.interrupt_id is not None - or request.group_handoff_intent is not None - or request.failure_code is not None - or request.failure_message is not None - ): - raise DeliveryServiceError( - "invalid_delivery_request", - "ack delivery cannot carry checkpoint lifecycle fields", - ) - return - _require_text(request.checkpoint_id, field="checkpoint_id", max_length=255) if request.kind == "waiting": _require_text(request.content, field="content") diff --git a/backend/app/services/agent_runtime/group_acknowledgement.py b/backend/app/services/agent_runtime/group_acknowledgement.py deleted file mode 100644 index 4d537f104..000000000 --- a/backend/app/services/agent_runtime/group_acknowledgement.py +++ /dev/null @@ -1,84 +0,0 @@ -"""Durable public acknowledgement for accepted group mention Runs.""" - -from __future__ import annotations - -from collections.abc import Mapping -import uuid - -from loguru import logger -from sqlalchemy import select - -from app.models.agent_run import AgentRun -from app.services.agent_runtime.command_worker import ( - CheckpointObservation, - RuntimeCommandRecord, - RuntimeRunRecord, - RuntimeSessionFactory, -) -from app.services.agent_runtime.delivery import ( - DeliveryReceipt, - DeliveryRequest, - deliver_runtime_message, -) -from app.services.group_realtime import publish_stored_group_message - - -_ACK_CONTENT = "收到，我开始处理。" - - -class RuntimeGroupStartAcknowledgementHandler: - """Write one ordinary Agent group message only after intake is committed.""" - - def __init__(self, *, session_factory: RuntimeSessionFactory) -> None: - self._session_factory = session_factory - - async def handle( - self, - *, - run: RuntimeRunRecord, - command: RuntimeCommandRecord, - checkpoint: CheckpointObservation | None, - ) -> None: - del checkpoint - if command.command_type != "start" or run.run_kind == "orchestration": - return - - receipt: DeliveryReceipt | None = None - async with self._session_factory() as db: - async with db.begin(): - result = await db.execute( - select(AgentRun.delivery_target).where( - AgentRun.tenant_id == run.tenant_id, - AgentRun.id == run.run_id, - ) - ) - target = result.scalar_one_or_none() - if not isinstance(target, Mapping) or target.get("kind") != "group": - return - receipt = await deliver_runtime_message( - db, - DeliveryRequest( - tenant_id=run.tenant_id, - run_id=run.run_id, - kind="ack", - content=_ACK_CONTENT, - ), - ) - if ( - receipt is not None - and receipt.status == "delivered" - and isinstance(receipt.actual_session_id, uuid.UUID) - and isinstance(receipt.message_id, uuid.UUID) - ): - try: - await publish_stored_group_message( - self._session_factory, - tenant_id=run.tenant_id, - session_id=receipt.actual_session_id, - message_id=receipt.message_id, - ) - except Exception as exc: - logger.warning(f"[GroupRealtime] ACK publish lookup failed: {exc}") - - -__all__ = ["RuntimeGroupStartAcknowledgementHandler"] diff --git a/backend/app/services/agent_runtime/group_handoff.py b/backend/app/services/agent_runtime/group_handoff.py index 8f6f72cd3..e508d536a 100644 --- a/backend/app/services/agent_runtime/group_handoff.py +++ b/backend/app/services/agent_runtime/group_handoff.py @@ -395,8 +395,8 @@ def _source_run_matches( or source_run.system_role is not None or source_run.runtime_type != "langgraph" or source_run.runtime_thread_id != str(source_run.id) - # Group start acknowledgement may already be delivered while the Run - # remains active. This projection is not a Runtime lifecycle state. + # Historical Runs may already have delivered the retired start ACK; + # current Runs remain pending until waiting or terminal delivery. or source_run.delivery_status not in {"pending", "delivered"} ): raise GroupAgentHandoffError( @@ -517,6 +517,17 @@ async def _validate_targets( "Group mention resolution did not preserve the frozen participant order", repairable=True, ) + self_targets = [ + mention.participant_id + for mention in resolved + if mention.agent is not None and mention.agent.id == source_agent_id + ] + if self_targets: + raise GroupAgentHandoffError( + "group_handoff_self_target", + "An Agent cannot create a public handoff to itself", + repairable=True, + ) for mention in resolved: assert mention.agent is not None if not _target_budget_available(mention.agent, now=clock): diff --git a/backend/app/services/agent_runtime/model_capabilities.py b/backend/app/services/agent_runtime/model_capabilities.py index 69bdba1e4..8966cca1e 100644 --- a/backend/app/services/agent_runtime/model_capabilities.py +++ b/backend/app/services/agent_runtime/model_capabilities.py @@ -76,28 +76,6 @@ def _legacy_output_limit(value: int | None) -> int | None: class ModelCapabilityResolver: """Resolve model semantics without performing provider I/O.""" - @staticmethod - def require_native_tool_calling(model: LLMModel) -> None: - """Require a concrete model row to be safe for an Agent tool Runtime.""" - model_label = getattr(model, "label", None) or model.model - if model.supports_tool_calling is True: - return - if model.supports_tool_calling is False: - raise ModelCapabilityError( - "model_tool_calling_unsupported", - ( - f"Model {model_label!r} did not produce a valid " - "native tool call during its capability test." - ), - ) - raise ModelCapabilityError( - "model_tool_calling_unverified", - ( - f"Model {model_label!r} has not passed the native " - "tool-calling capability test required by Agent Runtime." - ), - ) - @staticmethod def capabilities( model: LLMModel, diff --git a/backend/app/services/agent_runtime/model_step_service.py b/backend/app/services/agent_runtime/model_step_service.py index aedfec3db..a9c42fe5b 100644 --- a/backend/app/services/agent_runtime/model_step_service.py +++ b/backend/app/services/agent_runtime/model_step_service.py @@ -224,6 +224,8 @@ def _retry_http_status(error: Exception) -> str: - For a chained request such as "wake A and ask A to wake B", this Run should mention A only and give A the concrete instruction to wake B. Do not wake B from this Run unless the user also asked you to contact B directly. - Runtime publishes the `finish.content` as your public group reply and starts one child Run per mentioned participant so each target can reply publicly in this same group session. After `finish`, you cannot add another mention target from this Run. For multiple mentions, verify that the array contains every intended recipient before calling `finish`. - `send_message_to_agent` is private A2A. Use it only when you need private advice or facts and the target does not need to reply publicly in the group. It is never a substitute for `finish.mention_participant_ids` when the user asks you to `@` an Agent or have them respond in the group. +- A planned group transition must remain in this group session. When `group_context.planning_hint` assigns a later responsibility to another current-group Agent, never call `send_message_to_agent` for that transition under any `msg_type`; publish your completed part with `finish`, mention that Agent through `mention_participant_ids`, and state exactly what they must do and reply with publicly. +- Do not perform another Agent's assigned responsibility, wait for its private delegated result, merge that private result into your answer, or claim that Agent completed work on your behalf. A private A2A result is not that Agent's public group reply. - A textual `@name` or display name in `finish.content` is only text and never routes or wakes an Agent. Never infer participant IDs from display names. If no other Agent needs to join and reply publicly, omit `mention_participant_ids`. - If this Run was started because another Agent mentioned you, answer only the part addressed to you in `current_responsibility`, using your own role and voice, and normally finish without mentioning anyone. Do not repeat the source Agent's message, answer on behalf of other mentioned participants, describe its mention operation as your own action, or mention the source/co-mentioned Agents merely to reciprocate a greeting or acknowledgment. Mention another Agent only for a new concrete question, request, or responsibility that genuinely requires another public reply. - When several Agents were already woken by the same source message, each has its own Run. Address them by plain display name if useful, but do not `@` them just to make them greet or acknowledge one another again. @@ -730,15 +732,40 @@ def _repair( repair_code: str | None = None, repair_tool_name: str | None = None, ) -> ModelStepResult: + assistant_message = _assistant_message(state, context, step) + if ( + not str(assistant_message.get("content") or "").strip() + and not assistant_message.get("tool_calls") + ): + # Invalid/truncated tool calls cannot be replayed in provider history. + # Persist only the user-role repair instruction; an empty assistant + # message is rejected by providers such as Cohere. + assistant_message = None return ModelStepResult( intent="text", - assistant_message=_assistant_message(state, context, step), + assistant_message=assistant_message, repair_instruction=instruction, repair_code=repair_code, repair_tool_name=repair_tool_name, ) +def _safe_provider_failure_message(error: Exception) -> str: + """Return bounded user-facing provider diagnostics; raw bodies stay in logs.""" + match = re.search( + r"(? LLMModel | None: async with self._session_factory() as db: - candidates = await active_agent_model_candidates( - db, - agent, - require_tool_calling=True, - ) + candidates = await active_agent_model_candidates(db, agent) return next((model for model in candidates if model.id != primary_model.id), None) async def compact_inputs( @@ -1530,7 +1548,7 @@ async def complete_once( ) raise RuntimeModelCallError( "model_call_failed", - str(primary_error) or type(primary_error).__name__, + _safe_provider_failure_message(primary_error), ) from primary_error tenant_id = uuid.UUID(context.tenant_id) fallback = await self._fallback_model( @@ -1620,7 +1638,7 @@ async def complete_once( ) raise RuntimeModelCallError( "model_failover_failed", - str(fallback_error) or type(fallback_error).__name__, + _safe_provider_failure_message(fallback_error), ) from fallback_error actual_model = fallback failed_over_from = model @@ -1724,7 +1742,7 @@ async def complete_once( ) return _error( "model_call_failed", - str(exc) or type(exc).__name__, + "The model call failed.", ) diff --git a/backend/app/services/agent_runtime/planning.py b/backend/app/services/agent_runtime/planning.py index 8438d33f6..5db4dff5d 100644 --- a/backend/app/services/agent_runtime/planning.py +++ b/backend/app/services/agent_runtime/planning.py @@ -5,6 +5,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass import json +import re from typing import Protocol, cast import uuid @@ -40,6 +41,30 @@ _MAX_ENTRY_STEPS = 50 _PLAN_FIELDS = frozenset({"version", "mode", "goal", "plan_prompt", "entry_steps"}) _ENTRY_FIELDS = frozenset({"agent_id", "instruction"}) +_SIMPLE_CHECK_INS = frozenset( + { + "在吗", + "在嘛", + "在么", + "在不在", + "都在吗", + "都在嘛", + "你们在吗", + "你们在嘛", + "有人吗", + "你好", + "你好呀", + "你们好", + "大家好", + "嗨", + "哈喽", + "哈啰", + "hi", + "hello", + "hey", + } +) +_CHECK_IN_PUNCTUATION = re.compile(r"[\s,，.!！?？。:：;；~～、]+") _SYSTEM_PROMPT = """You are Clawith's internal multi-Agent planning component. Return exactly one JSON object and no Markdown. Never call tools and never do the work yourself. @@ -58,7 +83,16 @@ ] } Set mode to enforced only when the human explicitly specified workflow constraints such as Agent assignments, order, rounds, dependencies, branches, or completion conditions. Otherwise use advisory. +Use the simplest plan that satisfies the human's actual request. Do not invent analysis, synthesis, status reporting, review, or collaboration merely because several Agents were mentioned. +Before arranging any work, silently rewrite user_goal into clear directives in the original mention order. For each directive, fix the exact Agent, action, input, expected public output, and any dependency or next Agent. Then build goal, plan_prompt, and entry_steps only from those normalized directives; do not return the rewrite as an extra field. +Bind an instruction after an @mentioned Agent to that Agent until the next @mention, unless the human explicitly says otherwise. Never swap or reassign that work based on candidate order, Agent name, role_description, or perceived capability. For example, "@A write a poem @B then translate it" means A writes the poem, publicly hands that poem to B, and B translates it; it never means B writes and A translates. +When wording is vague, make the smallest literal interpretation explicit in goal, plan_prompt, and entry instructions before scheduling. Preserve every unambiguous Agent-to-responsibility binding, and never resolve ambiguity by moving work to a different Agent. +When repairing an invalid previous output, repeat this normalization from the original user_goal and verify every Agent-to-responsibility binding before returning corrected JSON. Do not merely repair JSON syntax or preserve a semantically wrong assignment from previous_output. +For a greeting or check-in, start the addressed Agents in parallel and tell each one to reply briefly as itself. Do not ask one Agent to report another Agent's status, unify their greetings, or exchange public handoffs. entry_steps starts only the first Agent or first parallel Agents. It may be a subset of candidates. Do not describe a DAG, step IDs, dependencies, progress, or later scheduling fields. Later collaboration proceeds through public Agent handoffs. +Create a public handoff only when a different Agent must provide a new reply for the task to proceed. Never create a handoff from an Agent to itself. +Each assigned Agent must author its own public group reply. Never route a planned group transition through private A2A, never ask an entry Agent to wait for a private result, and never ask one Agent to perform or claim another Agent's assigned work. +For every sequential transition, plan_prompt and the responsible entry instruction must say exactly which different Agent to wake publicly next, what concrete result to pass in the public group message, and what that Agent must reply with in the group. plan_prompt must be complete enough for every later participating Agent to receive unchanged. Preserve the human's explicit constraints, but do not repeat platform rules or invent mandatory constraints.""" @@ -171,6 +205,71 @@ def _candidate_agent_ids(state: RuntimeGraphState) -> frozenset[uuid.UUID]: return frozenset(resolved) +def _simple_check_in_plan( + state: RuntimeGraphState, + *, + goal: str, + candidate_agent_ids: frozenset[uuid.UUID], +) -> JsonObject | None: + """Return a deterministic one-reply-per-Agent plan for exact greetings.""" + raw_candidates = state["snapshots"].initial_input.get("candidate_agents") + if not isinstance(raw_candidates, Sequence) or isinstance( + raw_candidates, + (str, bytes, bytearray), + ): + return None + + candidates: list[tuple[uuid.UUID, str]] = [] + for raw_candidate in raw_candidates: + if not isinstance(raw_candidate, Mapping): + return None + try: + agent_id = uuid.UUID(str(raw_candidate.get("agent_id"))) + except (TypeError, ValueError): + return None + raw_name = raw_candidate.get("name") + if ( + agent_id not in candidate_agent_ids + or not isinstance(raw_name, str) + or not raw_name.strip() + ): + return None + candidates.append((agent_id, raw_name.strip())) + + remaining = goal + for _, name in sorted(candidates, key=lambda candidate: len(candidate[1]), reverse=True): + remaining = remaining.replace(f"@{name}", " ") + normalized = _CHECK_IN_PUNCTUATION.sub("", remaining).casefold() + if normalized not in _SIMPLE_CHECK_INS: + return None + + plan: JsonObject = { + "version": _PLAN_VERSION, + "mode": "advisory", + "goal": "Each mentioned Agent replies briefly to the user's greeting or check-in as itself.", + "plan_prompt": ( + "This is a simple greeting or check-in. Every entry Agent replies once, " + "briefly, and only as itself. Do not report another Agent's status, do not " + "ask another Agent to reply, and do not create a public handoff." + ), + "entry_steps": [ + { + "agent_id": str(agent_id), + "instruction": ( + f"Reply briefly to the user's greeting or check-in as {name} only. " + "Do not report another Agent's status and do not mention or hand off " + "to another Agent." + ), + } + for agent_id, name in candidates + ], + } + return validate_planning_output( + plan, + candidate_agent_ids=candidate_agent_ids, + ) + + def validate_planning_output( raw: object, *, @@ -327,6 +426,20 @@ async def complete_once( ) -> PlanningModelResult: try: candidates = _candidate_agent_ids(state) + except PlanningContractError as exc: + return PlanningModelResult( + error_code=exc.code, + error_message=str(exc), + retryable=False, + ) + simple_plan = _simple_check_in_plan( + state, + goal=context.goal, + candidate_agent_ids=candidates, + ) + if simple_plan is not None: + return PlanningModelResult(plan=simple_plan) + try: model = await self._load_model(context) except PlanningContractError as exc: return PlanningModelResult( diff --git a/backend/app/services/agent_runtime/planning_scheduler.py b/backend/app/services/agent_runtime/planning_scheduler.py index b82eeaa0f..b0d135173 100644 --- a/backend/app/services/agent_runtime/planning_scheduler.py +++ b/backend/app/services/agent_runtime/planning_scheduler.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence +import logging import uuid from sqlalchemy import select @@ -17,6 +18,7 @@ RuntimeSessionFactory, ) from app.services.agent_runtime.contracts import StartRunCommand +from app.services.agent_runtime.delivery import DeliveryRequest, deliver_runtime_message from app.services.agent_runtime.planning import checkpoint_plan from app.services.group_message_service import ( GroupMessageServiceError, @@ -25,9 +27,11 @@ _load_sender_scope, _resolve_mentions, ) +from app.services.group_realtime import publish_stored_group_message _PLANNING_ROLE = "group_planning" +logger = logging.getLogger(__name__) class PlanningSchedulingError(RuntimeError): @@ -278,7 +282,6 @@ def _validate_entry_targets( or target.agent is None or target.agent.id != agent_id or target.model is None - or target.agent.primary_model_id != target.model.id ): raise PlanningSchedulingError( "planning_entry_unavailable", @@ -379,6 +382,68 @@ async def handle( return if checkpoint.state["lifecycle"]["status"] != "completed": return + try: + await self._schedule_completed(run=run, checkpoint=checkpoint) + except PlanningSchedulingError as exc: + await self._deliver_terminal_failure( + run=run, + checkpoint=checkpoint, + error=exc, + ) + + async def _deliver_terminal_failure( + self, + *, + run: RuntimeRunRecord, + checkpoint: CheckpointObservation, + error: PlanningSchedulingError, + ) -> None: + """Project a deterministic scheduling rejection instead of retrying forever.""" + logger.warning( + "Planning entry scheduling failed permanently: run_id=%s code=%s error=%s", + run.run_id, + error.code, + error, + ) + async with self._session_factory() as db: + async with db.begin(): + receipt = await deliver_runtime_message( + db, + DeliveryRequest( + tenant_id=run.tenant_id, + run_id=run.run_id, + kind="terminal", + content="", + checkpoint_id=checkpoint.checkpoint_id, + lifecycle_status="failed", + failure_code=error.code, + failure_message=str(error), + ), + ) + if ( + receipt.status == "delivered" + and receipt.actual_session_id is not None + and receipt.message_id is not None + ): + try: + await publish_stored_group_message( + self._session_factory, + tenant_id=run.tenant_id, + session_id=receipt.actual_session_id, + message_id=receipt.message_id, + ) + except Exception as exc: + logger.warning( + "Planning failure realtime publish lookup failed: %s", + exc, + ) + + async def _schedule_completed( + self, + *, + run: RuntimeRunRecord, + checkpoint: CheckpointObservation, + ) -> None: plan = checkpoint_plan(checkpoint.state) raw_entries = plan["entry_steps"] diff --git a/backend/app/services/agent_runtime/run_compactor.py b/backend/app/services/agent_runtime/run_compactor.py index 90f6515ed..e4ea401c9 100644 --- a/backend/app/services/agent_runtime/run_compactor.py +++ b/backend/app/services/agent_runtime/run_compactor.py @@ -10,7 +10,10 @@ from app.config import Settings, get_settings from app.models.llm import LLMModel -from app.services.agent_runtime.model_capabilities import ModelCapabilityResolver +from app.services.agent_runtime.model_capabilities import ( + ModelCapabilityError, + ModelCapabilityResolver, +) from app.services.agent_runtime.node_executor import RunCompactResult from app.services.agent_runtime.state import ( JsonObject, @@ -535,15 +538,18 @@ def _budget(self, model: LLMModel): model.model, model.max_output_tokens, ) - return ModelCapabilityResolver.runtime_budget( - model, - requested_max_output_tokens=requested_output, - static_prompt_tokens=_estimate_tokens(_SYSTEM_PROMPT), - tool_schema_tokens=_estimate_tokens(_COMPACT_TOOL), - reserved_runtime_tokens=2048, - safety_margin_tokens=256, - settings=self._settings, - ) + try: + return ModelCapabilityResolver.runtime_budget( + model, + requested_max_output_tokens=requested_output, + static_prompt_tokens=_estimate_tokens(_SYSTEM_PROMPT), + tool_schema_tokens=_estimate_tokens(_COMPACT_TOOL), + reserved_runtime_tokens=2048, + safety_margin_tokens=256, + settings=self._settings, + ) + except ModelCapabilityError as exc: + raise RunCompactorError(exc.code, str(exc)) from exc async def _compact_batches( self, @@ -621,7 +627,10 @@ async def compact_if_needed( messages = _thread_messages(state, current_run_id=context.run_id) if not messages: return RunCompactResult() - inputs = await self._input_loader(state, context) + try: + inputs = await self._input_loader(state, context) + except ModelCapabilityError as exc: + raise RunCompactorError(exc.code, str(exc)) from exc if not _should_compact(inputs): return RunCompactResult() diff --git a/backend/app/services/agent_runtime/runtime_model_settings.py b/backend/app/services/agent_runtime/runtime_model_settings.py index e6f105bd9..aa1fad906 100644 --- a/backend/app/services/agent_runtime/runtime_model_settings.py +++ b/backend/app/services/agent_runtime/runtime_model_settings.py @@ -5,9 +5,10 @@ import uuid from dataclasses import dataclass -from sqlalchemy import select +from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession +from app.models.llm import LLMModel from app.models.system_settings import SystemSetting @@ -66,9 +67,52 @@ async def resolve_runtime_model_settings( value.get("compact_model_id"), setting_name="compact_model_id", ) + requested_ids = { + model_id + for model_id in ( + configured_planning, + configured_compact, + environment_planning_model_id, + environment_compact_model_id, + ) + if model_id is not None + } + eligible_ids: set[uuid.UUID] = set() + if requested_ids: + eligible_result = await db.execute( + select(LLMModel.id).where( + LLMModel.id.in_(requested_ids), + or_(LLMModel.tenant_id.is_(None), LLMModel.tenant_id == tenant_id), + LLMModel.enabled.is_(True), + LLMModel.deleted_at.is_(None), + ) + ) + eligible_ids = { + value.id if isinstance(value, LLMModel) else value + for value in eligible_result.scalars().all() + } + + def resolve_one( + configured_id: uuid.UUID | None, + environment_id: uuid.UUID | None, + ) -> tuple[uuid.UUID | None, str]: + if configured_id in eligible_ids: + return configured_id, "database" + if environment_id in eligible_ids: + return environment_id, "environment" + return None, "unavailable" + + planning_id, planning_source = resolve_one( + configured_planning, + environment_planning_model_id, + ) + compact_id, compact_source = resolve_one( + configured_compact, + environment_compact_model_id, + ) return RuntimeModelSettings( - planning_model_id=configured_planning or environment_planning_model_id, - compact_model_id=configured_compact or environment_compact_model_id, - planning_source="database" if configured_planning else "environment", - compact_source="database" if configured_compact else "environment", + planning_model_id=planning_id, + compact_model_id=compact_id, + planning_source=planning_source, + compact_source=compact_source, ) diff --git a/backend/app/services/agent_runtime/session_context_background.py b/backend/app/services/agent_runtime/session_context_background.py index 3ca70f606..a99632b92 100644 --- a/backend/app/services/agent_runtime/session_context_background.py +++ b/backend/app/services/agent_runtime/session_context_background.py @@ -410,10 +410,38 @@ async def scan_once(self) -> int: async with self._session_factory() as db: statement = ( select(ChatSession.tenant_id, ChatSession.id) + .join( + Group, + (Group.id == ChatSession.group_id) + & (Group.tenant_id == ChatSession.tenant_id), + ) .where( ChatSession.deleted_at.is_(None), ChatSession.last_message_at.is_not(None), ChatSession.session_type == "group", + Group.deleted_at.is_(None), + sa.exists( + select(1) + .select_from(GroupMember) + .join( + Participant, + Participant.id == GroupMember.participant_id, + ) + .join( + Agent, + (Participant.type == "agent") + & (Participant.ref_id == Agent.id), + ) + .where( + GroupMember.group_id == ChatSession.group_id, + GroupMember.removed_at.is_(None), + Agent.tenant_id == ChatSession.tenant_id, + Agent.status.in_(_ACTIVE_AGENT_STATUSES), + Agent.is_expired.is_(False), + Agent.access_mode != "private", + Agent.deleted_at.is_(None), + ) + ), ) .order_by(ChatSession.id) .limit(self._settings.AGENT_RUNTIME_SESSION_COMPACT_SCAN_BATCH_SIZE) diff --git a/backend/app/services/agent_runtime/worker_service.py b/backend/app/services/agent_runtime/worker_service.py index 43ac5aa6b..9dd2c567d 100644 --- a/backend/app/services/agent_runtime/worker_service.py +++ b/backend/app/services/agent_runtime/worker_service.py @@ -48,9 +48,6 @@ RuntimeGraphIdentity, build_agent_runtime_graph, ) -from app.services.agent_runtime.group_acknowledgement import ( - RuntimeGroupStartAcknowledgementHandler, -) from app.services.agent_runtime.heartbeat_completion import ( HeartbeatRuntimeCompletionHandler, ) @@ -321,9 +318,7 @@ def build_runtime_worker_components( lock_engine=lock_engine, checkpoint_reader=driver, command_executor=driver, - pre_command_handler=RuntimeGroupStartAcknowledgementHandler( - session_factory=session_factory, - ), + pre_command_handler=None, post_checkpoint_handler=post_checkpoint_handler, rejection_handler=post_checkpoint_handler, claimant=resolved_claimant, diff --git a/backend/app/services/group_file_service.py b/backend/app/services/group_file_service.py index c852e74c3..b56b7a323 100644 --- a/backend/app/services/group_file_service.py +++ b/backend/app/services/group_file_service.py @@ -8,6 +8,8 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import PurePosixPath +from urllib.parse import unquote import uuid from sqlalchemy import select @@ -43,6 +45,9 @@ def __init__(self, code: str, message: str) -> None: self.code = code +_FORBIDDEN_WORKSPACE_EXTENSIONS = frozenset({".exe"}) + + @dataclass(frozen=True, slots=True) class GroupTextFile: """Business-level view of one group text file.""" @@ -123,7 +128,18 @@ def _memory_key(group_id: uuid.UUID, agent_id: uuid.UUID) -> str: def _normalize_workspace_relative(path: str, *, allow_empty: bool) -> str: raw = (path or "").replace("\\", "/").strip() - if raw.startswith("/") or ".." in raw.split("/"): + decoded = raw + for _ in range(2): + next_decoded = unquote(decoded).replace("\\", "/") + if next_decoded == decoded: + break + decoded = next_decoded + if ( + raw.startswith("/") + or ".." in raw.split("/") + or decoded.startswith("/") + or ".." in decoded.split("/") + ): raise GroupFileServiceError( "group_workspace_path_invalid", "Group workspace paths must be relative and cannot contain '..'", @@ -137,6 +153,29 @@ def _normalize_workspace_relative(path: str, *, allow_empty: bool) -> str: return normalized +def _validate_workspace_write(path: str, content: bytes | None = None) -> None: + suffix = PurePosixPath(path).suffix.lower() + if suffix in _FORBIDDEN_WORKSPACE_EXTENSIONS: + raise GroupFileServiceError( + "group_workspace_file_type_forbidden", + f"Files with the {suffix} extension are not allowed in Group workspace", + ) + if content is None or suffix in BINARY_REVISION_EXTENSIONS: + return + if b"\x00" in content: + raise GroupFileServiceError( + "group_file_content_invalid", + "Group text files cannot contain NUL bytes", + ) + try: + content.decode("utf-8") + except UnicodeDecodeError as exc: + raise GroupFileServiceError( + "group_file_content_invalid", + "Group text files must contain valid UTF-8", + ) from exc + + def _workspace_key(group_id: uuid.UUID, path: str, *, allow_empty: bool) -> tuple[str, str]: normalized = _normalize_workspace_relative(path, allow_empty=allow_empty) root = normalize_storage_key(f"{_group_root(group_id)}/workspace") @@ -270,6 +309,8 @@ async def _prepare_runtime_workspace_operation( actor_participant_id=actor_participant_id, ) normalized, key = _workspace_key(group_id, path, allow_empty=False) + if operation == "write": + _validate_workspace_write(normalized) storage = get_storage_backend() current = await storage.get_version(key) if current.is_dir: @@ -997,6 +1038,7 @@ async def write_workspace_file( actor_participant_id=actor_participant_id, ) normalized, key = _workspace_key(group_id, path, allow_empty=False) + _validate_workspace_write(normalized) return await _write_text( db, group_id=group_id, @@ -1032,6 +1074,7 @@ async def write_workspace_binary_file( actor_participant_id=actor_participant_id, ) normalized, key = _workspace_key(group_id, path, allow_empty=False) + _validate_workspace_write(normalized, content) if require_absent and expected_version_token is not None: raise GroupFileServiceError( "group_file_write_condition_invalid", diff --git a/backend/app/services/group_message_service.py b/backend/app/services/group_message_service.py index 28e64109b..4b74f505b 100644 --- a/backend/app/services/group_message_service.py +++ b/backend/app/services/group_message_service.py @@ -331,7 +331,6 @@ async def _resolve_mentions( LLMModel.id.in_(model_ids), LLMModel.deleted_at.is_(None), LLMModel.enabled.is_(True), - LLMModel.supports_tool_calling.is_(True), ) ) models = { diff --git a/backend/app/services/heartbeat.py b/backend/app/services/heartbeat.py index a42e8cd8c..c930356bd 100644 --- a/backend/app/services/heartbeat.py +++ b/backend/app/services/heartbeat.py @@ -220,6 +220,11 @@ async def _heartbeat_tick(): triggered = 0 for agent in agents: + # Capture diagnostic identity before a nested transaction can + # roll back and expire ORM attributes. Exception handlers must + # never lazy-load from an expired async ORM instance. + agent_id = agent.id + agent_name = agent.name # Skip expired agents if agent.is_expired: continue @@ -250,7 +255,7 @@ async def _heartbeat_tick(): if not runtime_decision.use_v2: logger.error( "Heartbeat for {} remains due because Runtime is disabled ({})", - agent.name, + agent_name, runtime_decision.reason, ) continue @@ -295,7 +300,7 @@ async def _heartbeat_tick(): except HeartbeatRuntimeIntakeError as exc: logger.error( "Heartbeat Runtime intake failed for {} ({}): {}", - agent.name, + agent_name, exc.code, exc, ) @@ -303,30 +308,30 @@ async def _heartbeat_tick(): except Exception as exc: logger.exception( "Heartbeat claim failed for {}: {}", - agent.name, + agent_name, exc, ) continue logger.info( "💓 Queued heartbeat for {} as Runtime Run {}", - agent.name, + agent_name, runtime_handle.run_id, ) try: await write_audit_log( "heartbeat_fire", { - "agent_name": agent.name, + "agent_name": agent_name, "runtime_type": runtime_handle.runtime_type, "run_id": str(runtime_handle.run_id), }, - agent_id=agent.id, + agent_id=agent_id, ) except Exception as exc: logger.warning( "Failed to write heartbeat_fire audit log for {}: {}", - agent.name, + agent_name, exc, ) triggered += 1 diff --git a/backend/app/services/llm/caller.py b/backend/app/services/llm/caller.py index 5e4ad21d1..742aadf0d 100644 --- a/backend/app/services/llm/caller.py +++ b/backend/app/services/llm/caller.py @@ -491,14 +491,6 @@ async def call_llm( _max_tool_rounds, _token_limit_msg = await _get_agent_config(agent_id) if _token_limit_msg: return _token_limit_msg - from app.services.agent_runtime.model_capabilities import ( - ModelCapabilityError, - ModelCapabilityResolver, - ) - try: - ModelCapabilityResolver.require_native_tool_calling(model) - except ModelCapabilityError as exc: - return f"[Error] {exc.code}: {exc}" if max_tool_rounds_override and max_tool_rounds_override < _max_tool_rounds: _max_tool_rounds = max_tool_rounds_override diff --git a/backend/app/services/llm/model_resolution.py b/backend/app/services/llm/model_resolution.py index eddcfca26..ae5fb5cde 100644 --- a/backend/app/services/llm/model_resolution.py +++ b/backend/app/services/llm/model_resolution.py @@ -16,14 +16,11 @@ def _is_usable( model: LLMModel, *, tenant_id: uuid.UUID | None, - require_tool_calling: bool, ) -> bool: if getattr(model, "deleted_at", None) is not None or not model.enabled: return False if model.tenant_id not in {None, tenant_id}: return False - if require_tool_calling and model.supports_tool_calling is not True: - return False return True @@ -32,7 +29,6 @@ async def load_active_model( *, model_id: uuid.UUID | None, tenant_id: uuid.UUID | None, - require_tool_calling: bool = False, ) -> LLMModel | None: """Load one enabled, non-deleted model valid for the requested tenant.""" if model_id is None: @@ -49,7 +45,6 @@ async def load_active_model( if model is None or not _is_usable( model, tenant_id=tenant_id, - require_tool_calling=require_tool_calling, ): return None return model @@ -58,8 +53,6 @@ async def load_active_model( async def active_agent_model_candidates( db: AsyncSession, agent: Agent, - *, - require_tool_calling: bool = False, ) -> tuple[LLMModel, ...]: """Resolve primary, fallback, then tenant default without rewriting stored IDs.""" if getattr(agent, "deleted_at", None) is not None: @@ -102,7 +95,6 @@ async def active_agent_model_candidates( and _is_usable( model, tenant_id=agent.tenant_id, - require_tool_calling=require_tool_calling, ) ) @@ -110,14 +102,8 @@ async def active_agent_model_candidates( async def resolve_active_agent_model( db: AsyncSession, agent: Agent, - *, - require_tool_calling: bool = False, ) -> LLMModel | None: - candidates = await active_agent_model_candidates( - db, - agent, - require_tool_calling=require_tool_calling, - ) + candidates = await active_agent_model_candidates(db, agent) return candidates[0] if candidates else None diff --git a/backend/app/services/sandbox/local/subprocess_backend.py b/backend/app/services/sandbox/local/subprocess_backend.py index a3bf9d6ce..fd738cf23 100644 --- a/backend/app/services/sandbox/local/subprocess_backend.py +++ b/backend/app/services/sandbox/local/subprocess_backend.py @@ -15,6 +15,7 @@ MAX_STDOUT_CAPTURE_BYTES = 1_000_000 MAX_STDERR_CAPTURE_BYTES = 500_000 +VENV_CREATION_TIMEOUT_SECONDS = 120 # Security patterns - reused from agent_tools.py @@ -153,18 +154,48 @@ def _bind_if_exists(self, host_path: str, guest_path: str | None = None, *, read bind_flag = "--ro-bind" if read_only else "--bind" return [bind_flag, str(host), target] - def _ensure_workspace_venv(self, venv_path: Path) -> None: + async def _ensure_workspace_venv(self, venv_path: Path) -> None: venv_python = venv_path / "bin" / "python" if not venv_python.exists(): - import subprocess - # Use uv to create the virtual environment for extreme speed # --seed ensures pip is still present in the venv - subprocess.run( - ["uv", "venv", "--seed", str(venv_path)], - check=True, + proc = await asyncio.create_subprocess_exec( + "uv", + "venv", + "--seed", + str(venv_path), cwd=str(venv_path.parent), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + start_new_session=True, ) + try: + _, stderr = await asyncio.wait_for( + proc.communicate(), + timeout=VENV_CREATION_TIMEOUT_SECONDS, + ) + except (asyncio.TimeoutError, asyncio.CancelledError) as exc: + if proc.returncode is None: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) + except (ProcessLookupError, PermissionError): + proc.kill() + try: + await asyncio.wait_for(proc.wait(), timeout=5) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + if isinstance(exc, asyncio.CancelledError): + raise + raise RuntimeError( + "Timed out while creating the execute_code virtual environment" + ) from exc + if proc.returncode != 0: + detail = stderr.decode("utf-8", errors="replace").strip()[:500] + raise RuntimeError( + "Failed to create the execute_code virtual environment" + + (f": {detail}" if detail else "") + ) # Fix shebang lines in pip scripts to use bwrap-visible path # venv creates scripts with absolute paths to the host Python, @@ -174,10 +205,8 @@ def _ensure_workspace_venv(self, venv_path: Path) -> None: def _fix_pip_shebangs(self, venv_path: Path) -> None: """Replace pip with a bash wrapper that delegates to uv pip for extreme performance.""" venv_bin = venv_path / "bin" - sandbox_python = "/workspace/.venv/bin/python" - - wrapper_script = f"#!/bin/bash\nexec uv pip \"$@\"\n" - + wrapper_script = '#!/bin/bash\nexec uv pip "$@"\n' + for pip_cmd in ["pip", "pip3", "pip3.12"]: pip_path = venv_bin / pip_cmd if pip_path.parent.exists(): @@ -392,7 +421,7 @@ async def execute( script_path = work_path / f"_exec_tmp{ext}" try: - self._ensure_workspace_venv(venv_path) + await self._ensure_workspace_venv(venv_path) script_path.write_text(code, encoding="utf-8") sandbox_command = self._build_command(language, f"/workspace/{script_path.name}") @@ -492,7 +521,7 @@ async def read_stream(stream, out, label="stdout"): except Exception as e: duration_ms = int((time.time() - start_time) * 1000) - logger.exception(f"[Subprocess] Execution error") + logger.exception("[Subprocess] Execution error") return ExecutionResult( success=False, stdout="", diff --git a/backend/app/services/storage_runtime/facade.py b/backend/app/services/storage_runtime/facade.py index 00133a217..8b25a2dd2 100644 --- a/backend/app/services/storage_runtime/facade.py +++ b/backend/app/services/storage_runtime/facade.py @@ -16,7 +16,19 @@ tenant_storage_prefix, ) +__all__ = [ + "agent_storage_prefix", + "ensure_local_path", + "get_storage_backend", + "guess_content_type", + "normalize_storage_key", + "tenant_storage_prefix", +] + _storage_backend: StorageBackend | None = None +_CONTENT_TYPE_OVERRIDES = { + ".webp": "image/webp", +} def get_storage_backend() -> StorageBackend: @@ -57,4 +69,9 @@ async def ensure_local_path(key: str) -> Path: def guess_content_type(filename: str) -> str: - return mimetypes.guess_type(filename)[0] or "application/octet-stream" + suffix = Path(filename).suffix.lower() + return ( + _CONTENT_TYPE_OVERRIDES.get(suffix) + or mimetypes.guess_type(filename)[0] + or "application/octet-stream" + ) diff --git a/backend/tests/test_agent_runtime_adapter.py b/backend/tests/test_agent_runtime_adapter.py index 333ce7784..74537bfc3 100644 --- a/backend/tests/test_agent_runtime_adapter.py +++ b/backend/tests/test_agent_runtime_adapter.py @@ -272,16 +272,9 @@ async def test_missing_or_invalid_agent_budget_fails_without_runtime_fallback( @pytest.mark.asyncio -@pytest.mark.parametrize( - ("supports_tool_calling", "error_code"), - [ - (None, "model_tool_calling_unverified"), - (False, "model_tool_calling_unsupported"), - ], -) -async def test_new_agent_run_fails_before_persistence_without_verified_tool_calling( +@pytest.mark.parametrize("supports_tool_calling", [None, False]) +async def test_new_agent_run_accepts_saved_model_without_verified_tool_calling( supports_tool_calling: bool | None, - error_code: str, ) -> None: tenant_id = uuid.uuid4() agent = _agent(tenant_id) @@ -292,19 +285,25 @@ async def test_new_agent_run_fails_before_persistence_without_verified_tool_call supports_tool_calling=supports_tool_calling, ) db = _session(agent, model) + run = _run( + tenant_id=tenant_id, + agent_id=agent.id, + model_turn_limit=50, + ) with patch( "app.services.agent_runtime.adapter.register_run_with_start", - new=AsyncMock(), + new=AsyncMock( + return_value=RegisteredRun(run, _stored_command(run, "start"), True) + ), ) as persist: - with pytest.raises(RuntimeAdapterError) as raised: - await RuntimeCommandIntake( - db, - settings=_settings(enabled=True), - ).start_run(command) + await RuntimeCommandIntake( + db, + settings=_settings(enabled=True), + ).start_run(command) - assert raised.value.code == error_code - persist.assert_not_awaited() + persist.assert_awaited_once() + assert persist.await_args.args[1].model_id == model.id @pytest.mark.asyncio diff --git a/backend/tests/test_agent_runtime_checkpoint_side_effects.py b/backend/tests/test_agent_runtime_checkpoint_side_effects.py index aa6fa9303..004f49f9a 100644 --- a/backend/tests/test_agent_runtime_checkpoint_side_effects.py +++ b/backend/tests/test_agent_runtime_checkpoint_side_effects.py @@ -46,8 +46,9 @@ class _StoredRun: class _Session: - def __init__(self, value: object) -> None: + def __init__(self, value: object, *, terminal_event: object | None = None) -> None: self.value = value + self.terminal_event = terminal_event self.flush_count = 0 self.statements = [] @@ -62,6 +63,8 @@ def begin(self) -> _Transaction: async def execute(self, statement) -> _ScalarResult: self.statements.append(statement) + if "FROM agent_run_events" in str(statement): + return _ScalarResult(self.terminal_event) return _ScalarResult(self.value) async def flush(self) -> None: @@ -512,6 +515,30 @@ async def test_cancel_before_start_releases_lane_without_fabricating_checkpoint( assert stored.lane_held is False assert stored.lane_claimed_at is None assert sessions.sessions[0].flush_count == 1 + event = sessions.sessions[0].statements[-1].compile( + dialect=postgresql.dialect() + ).params + assert event["event_type"] == "run_cancelled" + + +@pytest.mark.asyncio +async def test_cancel_after_rejected_start_does_not_append_second_terminal_event() -> None: + run, command, _ = _records(command_type="cancel") + stored = _StoredRun() + session = _Session(stored, terminal_event=uuid.uuid4()) + handler = RuntimeCheckpointSideEffects( + session_factory=_SessionFactory(), # type: ignore[arg-type] + ) + handler._session_factory = lambda: session # type: ignore[method-assign] + + await handler.handle(run=run, command=command, checkpoint=None) + + assert stored.lane_held is False + assert stored.lane_claimed_at is None + assert not any( + "INSERT INTO agent_run_events" in str(statement) + for statement in session.statements + ) def test_waiting_delivery_uses_correlation_id_and_prompt() -> None: diff --git a/backend/tests/test_agent_runtime_delivery.py b/backend/tests/test_agent_runtime_delivery.py index 8df895076..92be49557 100644 --- a/backend/tests/test_agent_runtime_delivery.py +++ b/backend/tests/test_agent_runtime_delivery.py @@ -209,12 +209,6 @@ def test_delivery_request_uses_the_documented_stable_keys() -> None: run_id = uuid.uuid4() tenant_id = uuid.uuid4() - ack = DeliveryRequest( - tenant_id=tenant_id, - run_id=run_id, - kind="ack", - content="Accepted", - ) waiting = DeliveryRequest( tenant_id=tenant_id, run_id=run_id, @@ -233,11 +227,30 @@ def test_delivery_request_uses_the_documented_stable_keys() -> None: lifecycle_status="completed", ) - assert ack.idempotency_key == f"run:{run_id}:ack" assert waiting.idempotency_key == f"run:{run_id}:waiting:interrupt-7" assert terminal.idempotency_key == f"run:{run_id}:terminal:completed" +@pytest.mark.asyncio +async def test_new_ack_delivery_requests_are_rejected() -> None: + run_id = uuid.uuid4() + tenant_id = uuid.uuid4() + db = _RecordingDB() + + with pytest.raises(DeliveryServiceError) as exc_info: + await deliver_runtime_message( + db, + DeliveryRequest( + tenant_id=tenant_id, + run_id=run_id, + kind="ack", + content="Accepted", + ), + ) + + assert exc_info.value.code == "invalid_delivery_request" + + @pytest.mark.asyncio async def test_direct_delivery_accepts_the_session_scoped_langgraph_thread() -> None: tenant_id = uuid.uuid4() diff --git a/backend/tests/test_agent_runtime_group_handoff.py b/backend/tests/test_agent_runtime_group_handoff.py index cc56bb6c0..6fbe4bdb0 100644 --- a/backend/tests/test_agent_runtime_group_handoff.py +++ b/backend/tests/test_agent_runtime_group_handoff.py @@ -217,6 +217,7 @@ def _target( *, tenant_id: uuid.UUID, participant_id: uuid.UUID | None = None, + agent_id: uuid.UUID | None = None, name: str = "Target Agent", ) -> ResolvedGroupMention: model = LLMModel( @@ -229,7 +230,7 @@ def _target( enabled=True, ) agent = Agent( - id=uuid.uuid4(), + id=agent_id or uuid.uuid4(), tenant_id=tenant_id, creator_id=uuid.uuid4(), name=name, @@ -431,6 +432,51 @@ async def test_multi_target_preflight_failure_is_all_or_none_and_repairable() -> assert ensure.await_count == 0 +@pytest.mark.asyncio +async def test_self_handoff_fails_preflight_before_cycle_checks() -> None: + source_run, scope, context, state = _records() + self_target = _target( + tenant_id=source_run.tenant_id, + participant_id=scope.participant.id, + agent_id=source_run.agent_id, + name=scope.participant.display_name, + ) + ensure = AsyncMock(return_value=_cycle_check()) + + with ( + patch( + "app.services.agent_runtime.group_handoff._load_source_run", + new=AsyncMock(return_value=source_run), + ), + patch( + "app.services.agent_runtime.group_handoff._load_sender_scope", + new=AsyncMock(return_value=scope), + ), + patch( + "app.services.agent_runtime.group_handoff._resolve_mentions", + new=AsyncMock(return_value=(self_target,)), + ), + patch( + "app.services.agent_runtime.group_handoff.AgentCycleGuard.ensure_delegation_allowed", + new=ensure, + ), + ): + with pytest.raises(GroupAgentHandoffError) as raised: + await preflight_group_agent_handoff( + _DB(), # type: ignore[arg-type] + state=state, + context=context, + content="@Source Agent please answer again", + mention_participant_ids=(str(self_target.participant_id),), + settings=_settings(), + clock=lambda: NOW, + ) + + assert raised.value.code == "group_handoff_self_target" + assert raised.value.repairable is True + assert ensure.await_count == 0 + + @pytest.mark.asyncio async def test_cycle_limit_fails_preflight_before_terminal() -> None: source_run, scope, context, state = _records() diff --git a/backend/tests/test_agent_runtime_group_scheduling.py b/backend/tests/test_agent_runtime_group_scheduling.py index 03104733c..c318bd379 100644 --- a/backend/tests/test_agent_runtime_group_scheduling.py +++ b/backend/tests/test_agent_runtime_group_scheduling.py @@ -1,9 +1,8 @@ -"""Group acknowledgement and checkpoint-authoritative lane release tests.""" +"""Checkpoint-authoritative Group scheduling-lane release tests.""" from __future__ import annotations from datetime import UTC, datetime -from unittest.mock import AsyncMock, patch import uuid import pytest @@ -14,10 +13,6 @@ RuntimeCommandRecord, RuntimeRunRecord, ) -from app.services.agent_runtime.delivery import DeliveryReceipt, DeliveryRequest -from app.services.agent_runtime.group_acknowledgement import ( - RuntimeGroupStartAcknowledgementHandler, -) from app.services.agent_runtime.scheduling_lane import SchedulingLaneCompletionHandler from app.services.agent_runtime.state import ( RunInputSnapshots, @@ -161,105 +156,6 @@ def _checkpoint(run: RuntimeRunRecord, *, status: str) -> CheckpointObservation: ) -@pytest.mark.asyncio -async def test_group_ack_is_an_idempotent_delivery_before_graph_execution() -> None: - run_record, command, run = _records() - session = _Session(run.delivery_target) - handler = RuntimeGroupStartAcknowledgementHandler( - session_factory=_SessionFactory(session), # type: ignore[arg-type] - ) - - with patch( - "app.services.agent_runtime.group_acknowledgement.deliver_runtime_message", - new=AsyncMock(), - ) as deliver: - await handler.handle(run=run_record, command=command, checkpoint=None) - - request = deliver.await_args.args[1] - assert isinstance(request, DeliveryRequest) - assert request.run_id == run.id - assert request.kind == "ack" - assert request.content == "收到，我开始处理。" - assert request.idempotency_key == f"run:{run.id}:ack" - - -@pytest.mark.asyncio -async def test_non_group_start_does_not_emit_group_ack() -> None: - run_record, command, run = _records(target_kind="direct") - handler = RuntimeGroupStartAcknowledgementHandler( - session_factory=_SessionFactory(_Session(run.delivery_target)), # type: ignore[arg-type] - ) - - with patch( - "app.services.agent_runtime.group_acknowledgement.deliver_runtime_message", - new=AsyncMock(), - ) as deliver: - await handler.handle(run=run_record, command=command, checkpoint=None) - - deliver.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_group_ack_realtime_publish_runs_after_delivery_commit() -> None: - run_record, command, run = _records() - events: list[str] = [] - - class _OrderedTransaction: - async def __aenter__(self): - events.append("begin") - return self - - async def __aexit__(self, exc_type, exc, traceback): - events.append("commit") - return False - - class _OrderedSession(_Session): - def begin(self): - return _OrderedTransaction() - - session = _OrderedSession(run.delivery_target) - handler = RuntimeGroupStartAcknowledgementHandler( - session_factory=_SessionFactory(session), # type: ignore[arg-type] - ) - message_id = uuid.uuid4() - receipt = DeliveryReceipt( - tenant_id=run_record.tenant_id, - run_id=run_record.run_id, - idempotency_key=f"run:{run_record.run_id}:ack", - status="delivered", - delivery_kind="ack", - checkpoint_id=None, - message_id=message_id, - requested_session_id=uuid.UUID(run_record.session_id), - actual_session_id=uuid.UUID(run_record.session_id), - fallback_reason=None, - error_code=None, - ) - - async def fake_deliver(*_args, **_kwargs): - events.append("deliver") - return receipt - - async def fake_publish(*_args, **_kwargs): - assert events == ["begin", "deliver", "commit"] - events.append("publish") - return True - - with ( - patch( - "app.services.agent_runtime.group_acknowledgement.deliver_runtime_message", - new=fake_deliver, - ), - patch( - "app.services.agent_runtime.group_acknowledgement.publish_stored_group_message", - new=fake_publish, - ), - ): - await handler.handle(run=run_record, command=command, checkpoint=None) - - assert events == ["begin", "deliver", "commit", "publish"] - - @pytest.mark.asyncio async def test_terminal_checkpoint_releases_lane_without_reading_projection() -> None: run_record, _, run = _records() diff --git a/backend/tests/test_agent_runtime_model_step_service.py b/backend/tests/test_agent_runtime_model_step_service.py index ef2b9afba..e64320cd9 100644 --- a/backend/tests/test_agent_runtime_model_step_service.py +++ b/backend/tests/test_agent_runtime_model_step_service.py @@ -370,23 +370,24 @@ def _failover_service( @pytest.mark.asyncio -@pytest.mark.parametrize( - ("supports_tool_calling", "error_code"), - [ - (None, "model_tool_calling_unverified"), - (False, "model_tool_calling_unsupported"), - ], -) -async def test_agent_model_step_fails_closed_before_provider_call( +@pytest.mark.parametrize("supports_tool_calling", [None, False]) +async def test_agent_model_step_calls_saved_model_without_verified_tool_calling( supports_tool_calling: bool | None, - error_code: str, ) -> None: tenant_id = uuid.uuid4() model = _model(tenant_id) model.supports_tool_calling = supports_tool_calling agent = _agent(tenant_id) state = _state(tenant_id, model, agent) - completion = AsyncMock() + completion = AsyncMock( + return_value=LLMCompletionStep( + content="Completed with the saved model.", + tool_calls=(), + reasoning_content=None, + retry_instruction=None, + usage=TokenUsage(total_tokens=20), + ) + ) result = await _service( model, @@ -395,9 +396,9 @@ async def test_agent_model_step_fails_closed_before_provider_call( completion, ).complete_once(state, _context(state)) - assert result.intent == "error" - assert result.error["code"] == error_code - completion.assert_not_awaited() + assert result.intent == "finish" + assert result.finish_content == "Completed with the saved model." + completion.assert_awaited_once() @pytest.mark.asyncio @@ -483,6 +484,7 @@ async def complete(*args, **kwargs): assert result.intent == "text" assert result.repair_code == "invalid_tool_call" assert result.repair_tool_name == "write_file" + assert result.assistant_message is None @pytest.mark.asyncio @@ -1428,6 +1430,10 @@ async def group_application_tools(agent_id: uuid.UUID) -> list[dict]: assert "every intended recipient" in group_system_prompt assert "`send_message_to_agent` is private A2A" in group_system_prompt assert "never a substitute for `finish.mention_participant_ids`" in group_system_prompt + assert "A planned group transition must remain in this group session" in group_system_prompt + assert "under any `msg_type`" in group_system_prompt + assert "Do not perform another Agent's assigned responsibility" in group_system_prompt + assert "A private A2A result is not that Agent's public group reply" in group_system_prompt assert "textual `@name` or display name" in group_system_prompt assert "omit `mention_participant_ids`" in group_system_prompt assert "using your own role and voice" in group_system_prompt @@ -2197,8 +2203,7 @@ async def complete(*args, **kwargs): assert result.tool_calls == () assert result.repair_instruction is not None assert "only tool call" in result.repair_instruction - assert result.assistant_message is not None - assert "tool_calls" not in result.assistant_message + assert result.assistant_message is None @pytest.mark.asyncio @@ -2400,9 +2405,39 @@ async def complete(*args, **kwargs): assert result.intent == "error" assert result.error is not None assert result.error["code"] == "model_call_failed" + assert result.error["message"] == "Model provider request failed." + assert "invalid API key" not in result.error["message"] assert calls == 1 +@pytest.mark.asyncio +async def test_provider_validation_error_is_redacted_from_runtime_delivery() -> None: + tenant_id = uuid.uuid4() + model = _model(tenant_id) + agent = _agent(tenant_id) + state = _state(tenant_id, model, agent) + + async def complete(*args, **kwargs): + del args, kwargs + raise RuntimeError( + 'HTTP 400: {"error":{"metadata":{"provider_name":"Cohere"},' + '"user_id":"private-user-id","message":"invalid request"}}' + ) + + result = await _service( + model, + agent, + _ContextBuilder(_build()), + complete, + ).complete_once(state, _context(state)) + + assert result.intent == "error" + assert result.error == { + "code": "model_call_failed", + "message": "Model provider rejected the request (HTTP 400).", + } + + @pytest.mark.asyncio async def test_retryable_primary_error_without_fallback_pauses_for_resume() -> None: tenant_id = uuid.uuid4() diff --git a/backend/tests/test_agent_runtime_planning.py b/backend/tests/test_agent_runtime_planning.py index e4b41f914..5b0123b57 100644 --- a/backend/tests/test_agent_runtime_planning.py +++ b/backend/tests/test_agent_runtime_planning.py @@ -66,13 +66,14 @@ def _context( model_id: uuid.UUID | None = None, run_id: uuid.UUID | None = None, tenant_id: uuid.UUID | None = None, + goal: str = "Research the topic, then write the answer", ) -> RuntimeContext: return RuntimeContext( tenant_id=str(tenant_id or uuid.uuid4()), run_id=str(run_id or uuid.uuid4()), command_id=str(uuid.uuid4()), executor=cast(RuntimeNodeExecutor, object()), - goal="Research the topic, then write the answer", + goal=goal, run_kind="orchestration", source_type="chat", model_id=str(model_id or uuid.uuid4()), @@ -270,6 +271,129 @@ async def complete(model_arg, messages, **kwargs): assert "digital employee in Clawith" not in planning_prompt assert "call `finish`" not in planning_prompt assert "call `wait`" not in planning_prompt + assert "Use the simplest plan" in planning_prompt + assert "silently rewrite user_goal into clear directives" in planning_prompt + assert "Bind an instruction after an @mentioned Agent to that Agent" in planning_prompt + assert '"@A write a poem @B then translate it"' in planning_prompt + assert "never resolve ambiguity by moving work to a different Agent" in planning_prompt + assert "repeat this normalization from the original user_goal" in planning_prompt + assert "Do not merely repair JSON syntax" in planning_prompt + assert "greeting or check-in" in planning_prompt + assert "Never create a handoff from an Agent to itself" in planning_prompt + assert "Each assigned Agent must author its own public group reply" in planning_prompt + assert "Never route a planned group transition through private A2A" in planning_prompt + assert "must say exactly which different Agent to wake publicly next" in planning_prompt + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "goal", + [ + "@Agent 1 @Agent 2 在嘛", + "@Agent 1 @Agent 2 你们好！", + "@Agent 1 @Agent 2 hello?", + ], +) +async def test_simple_multi_agent_check_in_returns_a_fast_plan_without_calling_the_model( + goal: str, +) -> None: + first, second = uuid.uuid4(), uuid.uuid4() + model = LLMModel( + id=uuid.uuid4(), + tenant_id=None, + provider="openai", + model="planning-model", + api_key_encrypted="encrypted", + label="Planning", + enabled=True, + max_output_tokens=2048, + max_input_tokens=64_000, + ) + calls = [] + + async def complete(*args, **kwargs): + calls.append((args, kwargs)) + raise AssertionError("simple check-ins must not call the Planning model") + + result = await PlanningModelService( + session_factory=_session_factory(model), # type: ignore[arg-type] + completion=complete, + ).complete_once( + _state((first, second)), + _context(model_id=model.id, goal=goal), + ) + + assert result.error_code is None + assert result.plan == { + "version": 2, + "mode": "advisory", + "goal": "Each mentioned Agent replies briefly to the user's greeting or check-in as itself.", + "plan_prompt": ( + "This is a simple greeting or check-in. Every entry Agent replies once, " + "briefly, and only as itself. Do not report another Agent's status, do not " + "ask another Agent to reply, and do not create a public handoff." + ), + "entry_steps": [ + { + "agent_id": str(first), + "instruction": ( + "Reply briefly to the user's greeting or check-in as Agent 1 only. " + "Do not report another Agent's status and do not mention or hand off " + "to another Agent." + ), + }, + { + "agent_id": str(second), + "instruction": ( + "Reply briefly to the user's greeting or check-in as Agent 2 only. " + "Do not report another Agent's status and do not mention or hand off " + "to another Agent." + ), + }, + ], + } + assert calls == [] + + +@pytest.mark.asyncio +async def test_greeting_with_a_real_task_still_uses_the_planning_model() -> None: + first, second = uuid.uuid4(), uuid.uuid4() + model = LLMModel( + id=uuid.uuid4(), + tenant_id=None, + provider="openai", + model="planning-model", + api_key_encrypted="encrypted", + label="Planning", + enabled=True, + max_output_tokens=2048, + max_input_tokens=64_000, + ) + calls = [] + + async def complete(model_arg, messages, **kwargs): + calls.append((model_arg, messages, kwargs)) + return LLMCompletionStep( + content=json.dumps(_plan(first)), + tool_calls=(), + reasoning_content=None, + retry_instruction=None, + usage=TokenUsage(), + ) + + result = await PlanningModelService( + session_factory=_session_factory(model), # type: ignore[arg-type] + completion=complete, + ).complete_once( + _state((first, second)), + _context( + model_id=model.id, + goal="@Agent 1 @Agent 2 你好，请分析本周交付风险", + ), + ) + + assert result.plan == _plan(first) + assert len(calls) == 1 @pytest.mark.asyncio diff --git a/backend/tests/test_agent_runtime_planning_scheduler.py b/backend/tests/test_agent_runtime_planning_scheduler.py index 7a7afcad8..90a704b92 100644 --- a/backend/tests/test_agent_runtime_planning_scheduler.py +++ b/backend/tests/test_agent_runtime_planning_scheduler.py @@ -22,10 +22,10 @@ RuntimeRunRecord, ) from app.services.agent_runtime.contracts import RunHandle, StartRunCommand +from app.services.agent_runtime.delivery import DeliveryReceipt from app.services.agent_runtime.planning import validate_planning_output from app.services.agent_runtime.planning_scheduler import ( PlanningCheckpointScheduler, - PlanningSchedulingError, ) from app.services.agent_runtime.state import RunInputSnapshots, RuntimeGraphState from app.services.group_message_service import ResolvedGroupMention, _SenderScope @@ -424,7 +424,46 @@ async def start_run(command: StartRunCommand) -> RunHandle: @pytest.mark.asyncio -async def test_entry_revalidation_failure_creates_no_partial_child() -> None: +async def test_completed_plan_uses_the_resolved_active_fallback_model() -> None: + run, checkpoint, root, message, scope, candidates, _ = _records() + first, second, _ = candidates + first.agent.primary_model_id = uuid.uuid4() + second.agent.primary_model_id = uuid.uuid4() + db = _Session(root, message) + start = AsyncMock(side_effect=(_handle(run.tenant_id), _handle(run.tenant_id))) + + with ( + patch( + "app.services.agent_runtime.planning_scheduler._load_sender_scope", + new=AsyncMock(return_value=scope), + ), + patch( + "app.services.agent_runtime.planning_scheduler._resolve_mentions", + new=AsyncMock(return_value=(first, second)), + ), + patch( + "app.services.agent_runtime.planning_scheduler.RuntimeCommandIntake.start_run", + new=start, + ), + ): + await PlanningCheckpointScheduler( + session_factory=_SessionFactory(db), # type: ignore[arg-type] + settings=_settings(), + ).handle(run=run, checkpoint=checkpoint) + + commands = [call.args[0] for call in start.await_args_list] + assert [command.model_id for command in commands] == [ + first.model.id, + second.model.id, + ] + assert all( + command.model_id != target.agent.primary_model_id + for command, target in zip(commands, (first, second), strict=True) + ) + + +@pytest.mark.asyncio +async def test_entry_revalidation_failure_rolls_back_and_delivers_terminal_error() -> None: run, checkpoint, root, message, scope, candidates, _ = _records() first, second, _ = candidates invalid = ResolvedGroupMention( @@ -437,7 +476,22 @@ async def test_entry_revalidation_failure_creates_no_partial_child() -> None: reason="agent_unavailable", ) db = _Session(root, message) + delivery_db = _Session(root) + factory = _SessionFactory(db, delivery_db) start = AsyncMock() + receipt = DeliveryReceipt( + tenant_id=run.tenant_id, + run_id=run.run_id, + idempotency_key=f"run:{run.run_id}:terminal:failed", + status="delivered", + delivery_kind="terminal", + checkpoint_id=checkpoint.checkpoint_id, + message_id=uuid.uuid4(), + requested_session_id=uuid.UUID(run.session_id), + actual_session_id=uuid.UUID(run.session_id), + fallback_reason=None, + error_code=None, + ) with ( patch( @@ -452,17 +506,37 @@ async def test_entry_revalidation_failure_creates_no_partial_child() -> None: "app.services.agent_runtime.planning_scheduler.RuntimeCommandIntake.start_run", new=start, ), + patch( + "app.services.agent_runtime.planning_scheduler.deliver_runtime_message", + new=AsyncMock(return_value=receipt), + ) as deliver, + patch( + "app.services.agent_runtime.planning_scheduler.publish_stored_group_message", + new=AsyncMock(), + ) as publish, ): - with pytest.raises(PlanningSchedulingError) as raised: - await PlanningCheckpointScheduler( - session_factory=_SessionFactory(db), # type: ignore[arg-type] - settings=_settings(), - ).handle(run=run, checkpoint=checkpoint) + await PlanningCheckpointScheduler( + session_factory=factory, # type: ignore[arg-type] + settings=_settings(), + ).handle(run=run, checkpoint=checkpoint) - assert raised.value.code == "planning_entry_unavailable" start.assert_not_awaited() assert root.delivery_status == "pending" assert db.transaction is not None and db.transaction.rolled_back is True + request = deliver.await_args.args[1] + assert request.run_id == run.run_id + assert request.lifecycle_status == "failed" + assert request.failure_code == "planning_entry_unavailable" + assert request.failure_message == ( + "A Planning entry Agent is no longer an available Group target" + ) + assert request.checkpoint_id == checkpoint.checkpoint_id + publish.assert_awaited_once_with( + factory, + tenant_id=run.tenant_id, + session_id=uuid.UUID(run.session_id), + message_id=receipt.message_id, + ) @pytest.mark.asyncio diff --git a/backend/tests/test_agent_runtime_run_compactor.py b/backend/tests/test_agent_runtime_run_compactor.py index 437d10fd3..1f1d2d9ca 100644 --- a/backend/tests/test_agent_runtime_run_compactor.py +++ b/backend/tests/test_agent_runtime_run_compactor.py @@ -10,6 +10,7 @@ from app.config import Settings from app.models.llm import LLMModel +from app.services.agent_runtime.model_capabilities import ModelCapabilityError from app.services.agent_runtime.run_compactor import ( RunCompactInputs, RunCompactorError, @@ -249,6 +250,59 @@ async def forbidden(*_args, **_kwargs): assert raised.value.code == "missing_request_budget" +@pytest.mark.asyncio +async def test_invalid_request_budget_from_input_loader_is_deterministic() -> None: + state, context, _tenant_id = _state([_normal("current")]) + + async def load( + _state: RuntimeGraphState, + _context: RuntimeContext, + ) -> RunCompactInputs: + raise ModelCapabilityError( + "invalid_request_budget", + "requested output tokens leave no room in the shared context window", + ) + + async def forbidden(*_args, **_kwargs): + raise AssertionError("invalid request budget must fail before model use") + + service = RuntimeRunCompactorService( + settings=_settings(), + completion=forbidden, + input_loader=load, + ) + + with pytest.raises(RunCompactorError) as raised: + await service.compact_if_needed(state, context) + + assert raised.value.code == "invalid_request_budget" + assert raised.value.is_deterministic_compact_error is True + + +@pytest.mark.asyncio +async def test_invalid_compact_model_budget_is_a_deterministic_runtime_error() -> None: + state, context, tenant_id = _state( + [_normal("old", "old " * 300), _normal("current")] + ) + model = _model(tenant_id) + model.max_input_tokens = None + model.context_window_tokens_override = model.max_output_tokens + + async def forbidden(*_args, **_kwargs): + raise AssertionError("invalid compact budget must fail before model use") + + with pytest.raises(RunCompactorError) as raised: + await _service( + model=model, + completion=forbidden, + effective_budget=1_000, + current_tokens=800, + ).compact_if_needed(state, context) + + assert raised.value.code == "invalid_request_budget" + assert raised.value.is_deterministic_compact_error is True + + @pytest.mark.asyncio async def test_at_eighty_percent_compacts_prefix_and_keeps_current_input_exact() -> None: messages = [ diff --git a/backend/tests/test_agent_runtime_session_context_background.py b/backend/tests/test_agent_runtime_session_context_background.py index d2d57b325..095700a3d 100644 --- a/backend/tests/test_agent_runtime_session_context_background.py +++ b/backend/tests/test_agent_runtime_session_context_background.py @@ -166,7 +166,7 @@ async def test_direct_session_has_no_second_session_compact_policy() -> None: @pytest.mark.asyncio -async def test_background_scanner_only_selects_group_sessions() -> None: +async def test_background_scanner_only_selects_live_groups_with_active_agents() -> None: captured = [] class _ScannerDB: @@ -189,7 +189,13 @@ async def execute(self, statement): assert await scanner.scan_once() == 0 assert len(captured) == 1 compiled = captured[0].compile() - assert "session_type" in str(compiled) + sql = str(compiled) + assert "session_type" in sql + assert "groups.deleted_at IS NULL" in sql + assert "EXISTS" in sql + assert "group_members.removed_at IS NULL" in sql + assert "agents.deleted_at IS NULL" in sql + assert "agents.status IN" in sql assert "group" in compiled.params.values() diff --git a/backend/tests/test_agent_runtime_worker_service.py b/backend/tests/test_agent_runtime_worker_service.py index 851e00009..f2aa6be51 100644 --- a/backend/tests/test_agent_runtime_worker_service.py +++ b/backend/tests/test_agent_runtime_worker_service.py @@ -19,9 +19,6 @@ from app.services.agent_runtime.heartbeat_completion import ( HeartbeatRuntimeCompletionHandler, ) -from app.services.agent_runtime.group_acknowledgement import ( - RuntimeGroupStartAcknowledgementHandler, -) from app.services.agent_runtime.onboarding_completion import ( OnboardingRuntimeCompletionHandler, ) @@ -316,10 +313,7 @@ def test_component_builder_installs_current_agent_and_planning_graphs() -> None: ) assert components.channel_delivery_worker._claimant == "worker-test" assert components.async_tool_poll_scheduler._session_factory is not None - assert isinstance( - components.worker._pre_command_handler, - RuntimeGroupStartAcknowledgementHandler, - ) + assert components.worker._pre_command_handler is None terminal_handlers = components.worker._post_checkpoint_handler._terminal_handlers assert [type(handler) for handler in terminal_handlers] == [ SessionContextCompletionHandler, diff --git a/backend/tests/test_finish_protocol.py b/backend/tests/test_finish_protocol.py index ada20f9fc..de38a2790 100644 --- a/backend/tests/test_finish_protocol.py +++ b/backend/tests/test_finish_protocol.py @@ -361,28 +361,34 @@ async def test_call_llm_requires_finish_tool_to_stop(monkeypatch): @pytest.mark.asyncio -@pytest.mark.parametrize( - ("supports_tool_calling", "error_code"), - [ - (None, "model_tool_calling_unverified"), - (False, "model_tool_calling_unsupported"), - ], -) -async def test_legacy_tool_loop_fails_closed_before_provider_call( +@pytest.mark.parametrize("supports_tool_calling", [None, False]) +async def test_legacy_tool_loop_calls_saved_model_without_verified_tool_calling( monkeypatch, supports_tool_calling, - error_code, ): from app.services.llm import caller model = _model() model.supports_tool_calling = supports_tool_calling - - def create_client(**_kwargs): - raise AssertionError("provider must not be called") + fake_client = FakeStreamClient([_finish_response("Final answer.")]) monkeypatch.setattr(caller, "_get_agent_config", lambda _agent_id: _async_return((50, None))) - monkeypatch.setattr(caller, "create_llm_client", create_client) + monkeypatch.setattr(caller, "_get_user_name", lambda _user_id: _async_return("Ray")) + monkeypatch.setattr( + "app.services.agent_context.build_agent_context", + lambda *_args, **_kwargs: _async_return(("static", "dynamic")), + ) + monkeypatch.setattr( + caller, + "get_agent_tools_for_llm", + lambda _agent_id: _async_return([]), + ) + monkeypatch.setattr(caller, "create_llm_client", lambda **_kwargs: fake_client) + monkeypatch.setattr( + caller, + "record_token_usage", + lambda *_args, **_kwargs: _async_return(None), + ) result = await caller.call_llm( model, @@ -390,9 +396,11 @@ def create_client(**_kwargs): "Agent", "", agent_id=uuid.uuid4(), + user_id=uuid.uuid4(), ) - assert result.startswith(f"[Error] {error_code}:") + assert result == "Final answer." + assert len(fake_client.messages_seen) == 1 @pytest.mark.asyncio diff --git a/backend/tests/test_group_api.py b/backend/tests/test_group_api.py index 321f8d492..58fb2dcb2 100644 --- a/backend/tests/test_group_api.py +++ b/backend/tests/test_group_api.py @@ -380,6 +380,10 @@ async def fake_read(_db, **kwargs): assert response.body == b"\x89PNG\r\n" assert response.media_type == "image/png" assert response.headers["content-disposition"].startswith("inline;") + audit = next(value for value in db.added if isinstance(value, AuditLog)) + assert audit.action == "group:workspace_download" + assert audit.details["path"] == "images/chart.png" + assert audit.details["inline"] is True assert calls == [ { "tenant_id": tenant_id, @@ -390,6 +394,40 @@ async def fake_read(_db, **kwargs): ] +@pytest.mark.asyncio +async def test_workspace_download_uses_portable_webp_content_type(monkeypatch) -> None: + tenant_id = uuid.uuid4() + user = _user(tenant_id) + participant = _participant(user) + group = _group(tenant_id, participant.id) + db = _RecordingDB() + + async def fake_download_user(**_kwargs): + return user + + async def fake_participant(_db, _user): + return participant + + async def fake_read(_db, **_kwargs): + return SimpleNamespace(path="images/chart.webp", content=b"RIFFpayloadWEBP") + + monkeypatch.setattr(groups_api, "_download_user", fake_download_user) + monkeypatch.setattr(groups_api, "_current_participant", fake_participant) + monkeypatch.setattr(groups_api.group_file_service, "read_workspace_binary_file", fake_read) + + response = await groups_api.download_group_workspace_file( + group.id, + path="images/chart.webp", + token="download-token", + inline=True, + credentials=None, + db=db, + ) + + assert response.media_type == "image/webp" + assert response.headers["content-disposition"].startswith("inline;") + + @pytest.mark.asyncio async def test_create_group_stages_domain_change_and_audit_in_one_transaction(monkeypatch) -> None: tenant_id = uuid.uuid4() diff --git a/backend/tests/test_group_file_service.py b/backend/tests/test_group_file_service.py index b15cdd6f0..0aa40e5a9 100644 --- a/backend/tests/test_group_file_service.py +++ b/backend/tests/test_group_file_service.py @@ -184,6 +184,41 @@ async def test_group_workspace_rejects_traversal_and_stale_writes( ) assert path_error.value.code == "group_workspace_path_invalid" + with pytest.raises(group_file_service.GroupFileServiceError) as encoded_path: + await group_file_service.write_workspace_file( + db, + tenant_id=tenant_id, + group_id=group_id, + actor_participant_id=actor.id, + path="%2e%2e/system/announcement.md", + content="escape", + ) + assert encoded_path.value.code == "group_workspace_path_invalid" + + with pytest.raises(group_file_service.GroupFileServiceError) as executable: + await group_file_service.write_workspace_binary_file( + db, + tenant_id=tenant_id, + group_id=group_id, + actor_participant_id=actor.id, + path="payload.exe", + content=b"MZ", + content_type="application/octet-stream", + ) + assert executable.value.code == "group_workspace_file_type_forbidden" + + with pytest.raises(group_file_service.GroupFileServiceError) as invalid_text: + await group_file_service.write_workspace_binary_file( + db, + tenant_id=tenant_id, + group_id=group_id, + actor_participant_id=actor.id, + path="invalid.txt", + content=b"\xff\xfe", + content_type="text/plain", + ) + assert invalid_text.value.code == "group_file_content_invalid" + current = await group_file_service.write_workspace_file( db, tenant_id=tenant_id, diff --git a/backend/tests/test_group_message_service.py b/backend/tests/test_group_message_service.py index 14d8109a3..6127669fd 100644 --- a/backend/tests/test_group_message_service.py +++ b/backend/tests/test_group_message_service.py @@ -191,6 +191,7 @@ def test_mentions_are_deduplicated_in_client_order() -> None: @pytest.mark.asyncio async def test_mention_resolution_only_exposes_active_group_members() -> None: tenant_id, user, scope, target, mention = _records() + mention.model.supports_tool_calling = False human_target = Participant( id=uuid.uuid4(), type="user", @@ -241,6 +242,8 @@ async def test_mention_resolution_only_exposes_active_group_members() -> None: assert resolved[0].valid is True and resolved[0].triggers_agent is True assert resolved[0].agent is mention.agent + assert resolved[0].model is mention.model + assert "llm_models.supports_tool_calling IS true" not in str(db.statements[-1]) assert resolved[1].valid is True and resolved[1].triggers_agent is False assert resolved[1].participant_type == "user" assert resolved[2].valid is False diff --git a/backend/tests/test_heartbeat_runtime.py b/backend/tests/test_heartbeat_runtime.py index 92467a82c..aa3d591db 100644 --- a/backend/tests/test_heartbeat_runtime.py +++ b/backend/tests/test_heartbeat_runtime.py @@ -2,7 +2,9 @@ from __future__ import annotations +from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta, timezone +from types import SimpleNamespace from unittest.mock import AsyncMock, patch import uuid @@ -124,6 +126,100 @@ def test_default_heartbeat_prompt_does_not_advertise_hardcoded_tools() -> None: assert hardcoded_tool not in prompt +@pytest.mark.asyncio +async def test_heartbeat_intake_error_does_not_read_expired_agent_identity() -> None: + class _ExpiringAgent: + def __init__(self) -> None: + self._id = uuid.uuid4() + self._name = "Heartbeat Agent" + self.expired = False + self.name_reads = 0 + self.tenant_id = None + self.is_expired = False + self.expires_at = None + self.heartbeat_active_hours = "00:00-23:59" + self.heartbeat_interval_minutes = 1 + self.last_heartbeat_at = None + + @property + def id(self): + return self._id + + @property + def name(self): + self.name_reads += 1 + if self.expired: + raise RuntimeError("expired ORM attribute was accessed") + return self._name + + agent = _ExpiringAgent() + + class _Result: + rowcount = 1 + + def scalars(self): + return self + + def all(self): + return [agent] + + class _NestedTransaction: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback): + if exc_type is not None: + agent.expired = True + return False + + class _HeartbeatSession: + async def execute(self, _statement): + return _Result() + + def begin_nested(self): + return _NestedTransaction() + + async def commit(self): + return None + + @asynccontextmanager + async def fake_session(): + yield _HeartbeatSession() + + async def fail_intake(*_args, **_kwargs): + raise HeartbeatRuntimeIntakeError( + "model_unavailable", + "Heartbeat Agent has no primary model", + ) + + audit = AsyncMock() + with ( + patch("app.database.async_session", new=fake_session), + patch( + "app.services.agent_runtime.config.decide_runtime_v2", + return_value=SimpleNamespace(use_v2=True, reason="enabled"), + ), + patch( + "app.services.timezone_utils.get_agent_timezone_sync", + return_value="UTC", + ), + patch( + "app.services.heartbeat._build_heartbeat_instruction", + new=AsyncMock(return_value=("Review", {})), + ), + patch( + "app.services.heartbeat_runtime.enqueue_heartbeat_runtime", + new=fail_intake, + ), + patch("app.services.audit_logger.write_audit_log", new=audit), + ): + await heartbeat_service._heartbeat_tick() + + assert agent.expired is True + assert agent.name_reads == 1 + audit.assert_not_awaited() + + @pytest.mark.asyncio async def test_disabled_heartbeat_rollout_leaves_claim_for_legacy_execution() -> None: agent = _agent() diff --git a/backend/tests/test_model_capabilities.py b/backend/tests/test_model_capabilities.py index 3d0e553ae..d431159bd 100644 --- a/backend/tests/test_model_capabilities.py +++ b/backend/tests/test_model_capabilities.py @@ -87,31 +87,6 @@ def test_llm_capability_columns_and_checks_are_declared() -> None: assert source in capability_source_check -@pytest.mark.parametrize( - ("supports_tool_calling", "error_code"), - [ - (None, "model_tool_calling_unverified"), - (False, "model_tool_calling_unsupported"), - ], -) -def test_agent_runtime_requires_verified_native_tool_calling( - supports_tool_calling: bool | None, - error_code: str, -) -> None: - model = _model(supports_tool_calling=supports_tool_calling) - - with pytest.raises(ModelCapabilityError) as exc_info: - ModelCapabilityResolver.require_native_tool_calling(model) - - assert exc_info.value.code == error_code - - -def test_verified_native_tool_calling_is_accepted() -> None: - ModelCapabilityResolver.require_native_tool_calling( - _model(supports_tool_calling=True) - ) - - def test_matching_overrides_win_without_changing_limit_semantics() -> None: model = _model( context_window_tokens=100_000, @@ -284,8 +259,13 @@ async def test_platform_model_resolution_rejects_unusable_models( async def test_global_runtime_model_resolvers_accept_only_enabled_platform_models() -> None: tenant_id = uuid.uuid4() compact_id = uuid.uuid4() - planning_id = uuid.uuid4() - model = _model(tenant_id=None, enabled=True) + planning_id = compact_id + model = _model( + id=compact_id, + tenant_id=None, + enabled=True, + supports_tool_calling=True, + ) settings = Settings( _env_file=None, MULTI_AGENT_COMPACT_MODEL_ID=compact_id, @@ -295,7 +275,9 @@ async def test_global_runtime_model_resolvers_accept_only_enabled_platform_model assert await resolve_multi_agent_compact_model(session, settings, tenant_id=tenant_id) is model # type: ignore[arg-type] assert await resolve_multi_agent_planning_model(session, settings, tenant_id=tenant_id) is model # type: ignore[arg-type] - assert len(session.statements) == 4 + # Each resolver reads settings, validates the configured ID is currently + # runnable, then loads the selected model. + assert len(session.statements) == 6 assert settings.AGENT_RUNTIME_SUMMARY_THRESHOLD_RATIO == 0.85 assert settings.AGENT_RUNTIME_MODEL_CAPABILITY_REFRESH_SECONDS == 86400 assert settings.MULTI_AGENT_COMPACT_MODEL_ID == compact_id diff --git a/backend/tests/test_model_resolution.py b/backend/tests/test_model_resolution.py index 026f4b6b9..8c2767c90 100644 --- a/backend/tests/test_model_resolution.py +++ b/backend/tests/test_model_resolution.py @@ -92,7 +92,7 @@ async def test_candidates_skip_deleted_disabled_cross_tenant_and_duplicate_model @pytest.mark.asyncio -async def test_tool_calling_requirement_filters_incapable_candidate(): +async def test_tool_calling_diagnostic_does_not_filter_saved_candidate(): from app.services.llm.model_resolution import active_agent_model_candidates tenant_id = uuid.uuid4() @@ -110,13 +110,9 @@ async def test_tool_calling_requirement_filters_incapable_candidate(): DummyResult([primary, fallback]), ]) - candidates = await active_agent_model_candidates( - db, - agent, - require_tool_calling=True, - ) + candidates = await active_agent_model_candidates(db, agent) - assert candidates == (fallback,) + assert candidates == (primary, fallback) @pytest.mark.asyncio diff --git a/backend/tests/test_runtime_model_settings_api.py b/backend/tests/test_runtime_model_settings_api.py index 06ed4223e..502c65450 100644 --- a/backend/tests/test_runtime_model_settings_api.py +++ b/backend/tests/test_runtime_model_settings_api.py @@ -2,6 +2,7 @@ import uuid from types import SimpleNamespace +from unittest.mock import AsyncMock, patch import pytest from fastapi import HTTPException @@ -24,16 +25,23 @@ def scalars(self) -> "_ModelResult": def all(self) -> list[LLMModel]: return self.models + def scalar_one_or_none(self) -> LLMModel | None: + return self.models[0] if self.models else None + class _ModelSession: def __init__(self, models: list[LLMModel]) -> None: self.models = models self.execute_count = 0 + self.commit = AsyncMock() async def execute(self, _statement: object) -> _ModelResult: self.execute_count += 1 return _ModelResult(self.models) + def add(self, _value: object) -> None: + pass + def _model(model_id: uuid.UUID, **overrides: object) -> LLMModel: values: dict[str, object] = { @@ -76,7 +84,6 @@ async def test_runtime_model_settings_are_company_admin_only() -> None: [ {"tenant_id": uuid.uuid4()}, {"enabled": False}, - {"supports_tool_calling": False}, ], ) async def test_runtime_model_settings_reject_ineligible_models( @@ -103,3 +110,37 @@ async def test_runtime_model_settings_reject_ineligible_models( assert exc_info.value.status_code == 422 assert session.execute_count == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("supports_tool_calling", [None, False]) +async def test_runtime_model_settings_accept_saved_model_without_verified_tools( + supports_tool_calling: bool | None, +) -> None: + model_id = uuid.uuid4() + tenant_id = uuid.uuid4() + session = _ModelSession( + [_model(model_id, supports_tool_calling=supports_tool_calling)] + ) + expected = {"planning_model_id": str(model_id)} + + with patch( + "app.api.enterprise._runtime_model_settings_payload", + new=AsyncMock(return_value=expected), + ): + result = await update_runtime_model_settings( + RuntimeModelSettingsUpdate( + planning_model_id=model_id, + compact_model_id=model_id, + ), + tenant_id=str(tenant_id), + current_user=SimpleNamespace( + role="platform_admin", + identity=None, + tenant_id=tenant_id, + ), # type: ignore[arg-type] + db=session, # type: ignore[arg-type] + ) + + assert result == expected + session.commit.assert_awaited_once() diff --git a/backend/tests/test_runtime_model_settings_resolution.py b/backend/tests/test_runtime_model_settings_resolution.py new file mode 100644 index 000000000..69448f8b7 --- /dev/null +++ b/backend/tests/test_runtime_model_settings_resolution.py @@ -0,0 +1,81 @@ +"""Runtime model settings must never resolve stale or unrunnable model IDs.""" + +from types import SimpleNamespace +import uuid + +import pytest + +from app.services.agent_runtime.runtime_model_settings import ( + resolve_runtime_model_settings, + runtime_model_setting_key, +) + + +class _Result: + def __init__(self, values: list[object]) -> None: + self._values = values + + def scalars(self): + return self + + def all(self) -> list[object]: + return self._values + + +class _Session: + def __init__(self, *results: _Result) -> None: + self._results = iter(results) + self.statements = [] + + async def execute(self, statement): + self.statements.append(statement) + return next(self._results) + + +@pytest.mark.asyncio +async def test_deleted_configured_model_falls_back_only_to_eligible_environment_model() -> None: + tenant_id = uuid.uuid4() + deleted_id = uuid.uuid4() + environment_id = uuid.uuid4() + setting = SimpleNamespace( + key=runtime_model_setting_key(tenant_id), + value={ + "planning_model_id": str(deleted_id), + "compact_model_id": str(deleted_id), + }, + ) + db = _Session( + _Result([setting]), + _Result([environment_id]), + ) + + resolved = await resolve_runtime_model_settings( + db, # type: ignore[arg-type] + tenant_id=tenant_id, + environment_planning_model_id=environment_id, + environment_compact_model_id=None, + ) + + assert resolved.planning_model_id == environment_id + assert resolved.planning_source == "environment" + assert resolved.compact_model_id is None + assert resolved.compact_source == "unavailable" + assert "supports_tool_calling" not in str(db.statements[1]) + + +@pytest.mark.asyncio +async def test_no_configured_models_requires_no_model_lookup() -> None: + tenant_id = uuid.uuid4() + db = _Session(_Result([])) + + resolved = await resolve_runtime_model_settings( + db, # type: ignore[arg-type] + tenant_id=tenant_id, + environment_planning_model_id=None, + environment_compact_model_id=None, + ) + + assert resolved.planning_model_id is None + assert resolved.compact_model_id is None + assert resolved.planning_source == "unavailable" + assert resolved.compact_source == "unavailable" diff --git a/backend/tests/test_sandbox_subprocess_backend.py b/backend/tests/test_sandbox_subprocess_backend.py new file mode 100644 index 000000000..260b9ac0a --- /dev/null +++ b/backend/tests/test_sandbox_subprocess_backend.py @@ -0,0 +1,75 @@ +"""Local sandbox bootstrap must not block the Backend event loop.""" + +import asyncio +from pathlib import Path + +import pytest + +from app.services.sandbox.config import SandboxConfig +from app.services.sandbox.local import subprocess_backend +from app.services.sandbox.local.subprocess_backend import SubprocessBackend + + +@pytest.mark.asyncio +async def test_workspace_venv_uses_async_subprocess(monkeypatch, tmp_path: Path) -> None: + venv_path = tmp_path / ".venv" + calls: list[tuple[object, ...]] = [] + + class _Process: + returncode = 0 + pid = 123 + + async def communicate(self): + await asyncio.sleep(0) + (venv_path / "bin").mkdir(parents=True) + (venv_path / "bin" / "python").write_text("", encoding="utf-8") + return b"", b"" + + async def fake_create(*args, **kwargs): + calls.append((*args, kwargs)) + return _Process() + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create) + backend = SubprocessBackend(SandboxConfig()) + + await backend._ensure_workspace_venv(venv_path) + + assert calls + assert calls[0][:3] == ("uv", "venv", "--seed") + + +@pytest.mark.asyncio +async def test_workspace_venv_timeout_terminates_child(monkeypatch, tmp_path: Path) -> None: + terminated: list[int] = [] + + class _Process: + returncode = None + pid = 456 + + async def communicate(self): + await asyncio.Event().wait() + + async def wait(self): + self.returncode = -15 + return self.returncode + + def kill(self): + self.returncode = -9 + + async def fake_create(*_args, **_kwargs): + return _Process() + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create) + monkeypatch.setattr(subprocess_backend, "VENV_CREATION_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(subprocess_backend.os, "getpgid", lambda pid: pid) + monkeypatch.setattr( + subprocess_backend.os, + "killpg", + lambda pid, _signal: terminated.append(pid), + ) + backend = SubprocessBackend(SandboxConfig()) + + with pytest.raises(RuntimeError, match="Timed out"): + await backend._ensure_workspace_venv(tmp_path / ".venv") + + assert terminated == [456] diff --git a/backend/tests/test_websocket_runtime_chat.py b/backend/tests/test_websocket_runtime_chat.py index 052517698..570b86dc5 100644 --- a/backend/tests/test_websocket_runtime_chat.py +++ b/backend/tests/test_websocket_runtime_chat.py @@ -16,6 +16,7 @@ AcceptedWebChatMessage, WebChatRuntimeIntake, WebSocketChatHandler, + _websocket_content_log_summary, ) from app.models.agent_run import AgentRun from app.models.agent_run_command import AgentRunCommand @@ -51,6 +52,20 @@ async def close(self, code: int) -> None: self.closed_code = code +def test_websocket_log_summary_never_includes_message_or_image_payload() -> None: + payload = ( + "[image_data:data:image/png;base64,SECRET_IMAGE_PAYLOAD]" + "[image_data:data:image/jpeg;base64,SECOND_SECRET_PAYLOAD]" + " user secret" + ) + + summary = _websocket_content_log_summary(payload) + + assert summary == f"content_chars={len(payload)} image_count=2" + assert "SECRET" not in summary + assert "data:image" not in summary + + class _Transaction: async def __aenter__(self): return self diff --git a/frontend/src/components/ModelSwitcher.tsx b/frontend/src/components/ModelSwitcher.tsx index ac5b1f3a3..5b106aff5 100644 --- a/frontend/src/components/ModelSwitcher.tsx +++ b/frontend/src/components/ModelSwitcher.tsx @@ -4,6 +4,7 @@ import { useQuery } from '@tanstack/react-query'; import { useTranslation } from 'react-i18next'; import { IconChevronDown, IconCheck } from '@tabler/icons-react'; import { enterpriseApi } from '../services/api'; +import { subscribeModelCacheInvalidation } from '../services/modelCacheEvents'; interface Model { id: string; @@ -11,6 +12,7 @@ interface Model { model: string; label?: string; enabled?: boolean; + supports_tool_calling?: boolean | null; } interface Props { @@ -39,11 +41,18 @@ export default function ModelSwitcher({ value, onChange, tenantDefaultId, disabl { top: number; bottom: number; left: number; width: number; placement: 'above' | 'below'; maxHeight: number } | null >(null); - const { data: models = [] } = useQuery({ + const { data: models = [], refetch: refetchModels } = useQuery({ queryKey: ['llm-models'], queryFn: enterpriseApi.llmModels, }); + useEffect( + () => subscribeModelCacheInvalidation(() => { + void refetchModels(); + }), + [refetchModels], + ); + const enabled = (models as Model[]).filter(m => m.enabled !== false); const selected = enabled.find(m => m.id === value) || enabled.find(m => tenantDefaultId && m.id === tenantDefaultId) diff --git a/frontend/src/pages/agent-detail/AgentDetailPage.tsx b/frontend/src/pages/agent-detail/AgentDetailPage.tsx index 4a356cf46..ad75ec881 100644 --- a/frontend/src/pages/agent-detail/AgentDetailPage.tsx +++ b/frontend/src/pages/agent-detail/AgentDetailPage.tsx @@ -3189,8 +3189,18 @@ export default function AgentDetailPage() { const qMatch = msg.content.match(/\nQuestion: ([\s\S]+)$/); parsed = { ...msg, fileName, content: qMatch ? qMatch[1].trim() : '' }; } - // If file is an image and no imageUrl yet, build download URL for preview - if (parsed.fileName && !parsed.imageUrl && id) { + // A multi-image message stores all names in the legacy file_name field. + // Never turn that comma-joined label into one invalid download path; + // ChatMessageItem renders each persisted image_data marker instead. + const inlineImageMarkerCount = ( + parsed.content.match(/\[image_data:data:image\//g) || [] + ).length; + if ( + parsed.fileName + && !parsed.imageUrl + && inlineImageMarkerCount <= 1 + && id + ) { const ext = parsed.fileName.split('.').pop()?.toLowerCase() || ''; if (IMAGE_EXTS.includes(ext)) { parsed.imageUrl = `/api/agents/${id}/files/download?path=workspace/uploads/${encodeURIComponent(parsed.fileName)}&token=${token}`; @@ -4676,9 +4686,14 @@ export default function AgentDetailPage() { () => (llmModels as any[]).filter((m: any) => m.enabled), [llmModels], ); - const effectiveChatModelId = overrideModelId - || agent?.primary_model_id - || myTenant?.default_model_id + const effectiveChatModelId = [ + overrideModelId, + agent?.primary_model_id, + myTenant?.default_model_id, + ].find( + (candidate): candidate is string => Boolean(candidate) + && enabledLlmModels.some((model: any) => model.id === candidate), + ) || enabledLlmModels[0]?.id || null; diff --git a/frontend/src/pages/enterprise-settings/tabs/LlmTab.tsx b/frontend/src/pages/enterprise-settings/tabs/LlmTab.tsx index 2eeb12665..64b03f2ac 100644 --- a/frontend/src/pages/enterprise-settings/tabs/LlmTab.tsx +++ b/frontend/src/pages/enterprise-settings/tabs/LlmTab.tsx @@ -5,6 +5,7 @@ import { IconEdit } from '@tabler/icons-react'; import { useDialog } from '../../../components/Dialog/DialogProvider'; import { useToast } from '../../../components/Toast/ToastProvider'; import { useAuthStore } from '../../../stores'; +import { notifyModelCacheInvalidated } from '../../../services/modelCacheEvents'; import { fetchJson } from '../utils/fetchJson'; interface LLMModel { @@ -40,8 +41,8 @@ interface RuntimeModelSettings { tenant_id: string; planning_model_id: string | null; compact_model_id: string | null; - planning_source: 'database' | 'environment'; - compact_source: 'database' | 'environment'; + planning_source: 'database' | 'environment' | 'unavailable'; + compact_source: 'database' | 'environment' | 'unavailable'; candidates: Array>; } @@ -98,6 +99,7 @@ export default function LlmTab({ selectedTenantId }: LlmTabProps) { qc.invalidateQueries({ queryKey: ['tenant-default-model'] }); qc.invalidateQueries({ queryKey: ['agents'] }); qc.invalidateQueries({ queryKey: ['agent'] }); + notifyModelCacheInvalidated(); }; const { data: models = [] } = useQuery({ @@ -310,7 +312,7 @@ export default function LlmTab({ selectedTenantId }: LlmTabProps) { {t('enterprise.llm.runtimeModelsTitle', '多智能体运行时模型')}
- {t('enterprise.llm.runtimeModelsHint', '可使用当前公司的模型或平台模型；候选模型必须已启用并通过原生工具调用测试。保存后立即生效。')} + {t('enterprise.llm.runtimeModelsHint', '可使用当前公司或平台已保存且启用的模型；工具测试结果仅作诊断，不影响选择。保存后立即生效。')}
{runtimeModelSettings.candidates.length === 0 ? ( @@ -582,7 +584,7 @@ export default function LlmTab({ selectedTenantId }: LlmTabProps) { ) : m.supports_tool_calling === false ? ( {t('enterprise.llm.toolsUnavailable', 'Tools unavailable')} @@ -590,7 +592,7 @@ export default function LlmTab({ selectedTenantId }: LlmTabProps) { ) : ( {t('enterprise.llm.toolsUnverified', 'Tools unverified')} diff --git a/frontend/src/pages/groups/GroupsPage.tsx b/frontend/src/pages/groups/GroupsPage.tsx index f574bcdbd..e8347da47 100644 --- a/frontend/src/pages/groups/GroupsPage.tsx +++ b/frontend/src/pages/groups/GroupsPage.tsx @@ -41,6 +41,7 @@ import './groups.css'; type RenameTarget = { groupId: string; sessionId: string; current: string }; const HISTORY_PAGE_SIZE = 30; +const ACTIVE_RUN_TRANSITION_GRACE_MS = 5000; const readFlag = (key: string, fallback: boolean) => { const stored = localStorage.getItem(key); @@ -81,6 +82,7 @@ export default function GroupsPage() { const [hasMore, setHasMore] = useState(false); const [loadingMore, setLoadingMore] = useState(false); const [cancellingRuns, setCancellingRuns] = useState(false); + const [awaitingPlannedRuns, setAwaitingPlannedRuns] = useState(false); // One nav rail now: a tree of groups with their sessions nested underneath. It collapses to a // stub, and the side panel stays out of the way until asked for. const [groupsCollapsed, setGroupsCollapsed] = useState( @@ -105,6 +107,7 @@ export default function GroupsPage() { const latestRealtimeMessageBySessionRef = useRef>(new Map()); const readRequestsRef = useRef>(new Set()); const lastReadMessageBySessionRef = useRef>(new Map()); + const planningTransitionUntilRef = useRef(0); const { data: groups = [], @@ -176,16 +179,28 @@ export default function GroupsPage() { queryFn: () => groupApi.activeRuns(groupId!, sessionId!), enabled: Boolean(activeGroup && activeSession), retry: false, - refetchInterval: (query) => ( - query.state.data?.some((run) => run.can_cancel) ? 1000 : false - ), + refetchInterval: (query) => { + const runs = query.state.data ?? []; + const hasPlanningRun = runs.some( + (run) => run.can_cancel && run.system_role === 'group_planning', + ); + if (hasPlanningRun) { + planningTransitionUntilRef.current = Date.now() + ACTIVE_RUN_TRANSITION_GRACE_MS; + } + if (runs.some((run) => run.can_cancel)) return 1000; + return Date.now() < planningTransitionUntilRef.current ? 250 : false; + }, }); const activeRunIds = activeRunStates .filter((run) => run.can_cancel) .map((run) => run.run_id); - const isPlanning = activeRunStates.some( + const planningRunVisible = activeRunStates.some( (run) => run.can_cancel && run.system_role === 'group_planning', ); + const agentRunVisible = activeRunStates.some( + (run) => run.can_cancel && Boolean(run.agent_id), + ); + const isPlanning = planningRunVisible || (awaitingPlannedRuns && !agentRunVisible); const runningAgents = useMemo(() => { const membersByAgentId = new Map( members @@ -211,6 +226,31 @@ export default function GroupsPage() { ); const isManager = me?.role === 'manager'; + useEffect(() => { + planningTransitionUntilRef.current = 0; + setAwaitingPlannedRuns(false); + }, [groupId, sessionId]); + + useEffect(() => { + if (planningRunVisible) { + planningTransitionUntilRef.current = Date.now() + ACTIVE_RUN_TRANSITION_GRACE_MS; + setAwaitingPlannedRuns(true); + return; + } + if (agentRunVisible) { + setAwaitingPlannedRuns(false); + return; + } + if (!awaitingPlannedRuns) return; + const remaining = planningTransitionUntilRef.current - Date.now(); + if (remaining <= 0) { + setAwaitingPlannedRuns(false); + return; + } + const timer = setTimeout(() => setAwaitingPlannedRuns(false), remaining); + return () => clearTimeout(timer); + }, [agentRunVisible, awaitingPlannedRuns, planningRunVisible]); + // Header facts line: the group's makeup, which does not change as the session switches. const memberCounts = useMemo(() => ({ agents: members.filter((member) => member.participant_type === 'agent').length, @@ -461,6 +501,10 @@ export default function GroupsPage() { message_id: createRandomUUID(), }); setMessages((previous) => mergeMessages(previous, [intake.message])); + if (intake.dispatch_kind === 'planning') { + planningTransitionUntilRef.current = Date.now() + ACTIVE_RUN_TRANSITION_GRACE_MS; + setAwaitingPlannedRuns(true); + } if (intake.run_ids.length > 0) void refetchActiveRuns(); // Planning can fail before any agent starts — say so instead of leaving a silent gap. diff --git a/frontend/src/services/modelCacheEvents.ts b/frontend/src/services/modelCacheEvents.ts new file mode 100644 index 000000000..ae2c5de6d --- /dev/null +++ b/frontend/src/services/modelCacheEvents.ts @@ -0,0 +1,21 @@ +const MODEL_CACHE_EVENT = 'clawith:model-cache-invalidated'; +const MODEL_CACHE_STORAGE_KEY = 'clawith_model_cache_version'; + +export function notifyModelCacheInvalidated(): void { + const version = `${Date.now()}:${Math.random()}`; + localStorage.setItem(MODEL_CACHE_STORAGE_KEY, version); + window.dispatchEvent(new Event(MODEL_CACHE_EVENT)); +} + +export function subscribeModelCacheInvalidation(callback: () => void): () => void { + const onLocal = () => callback(); + const onStorage = (event: StorageEvent) => { + if (event.key === MODEL_CACHE_STORAGE_KEY) callback(); + }; + window.addEventListener(MODEL_CACHE_EVENT, onLocal); + window.addEventListener('storage', onStorage); + return () => { + window.removeEventListener(MODEL_CACHE_EVENT, onLocal); + window.removeEventListener('storage', onStorage); + }; +} diff --git a/frontend/tests/chatMultimodalContract.test.mjs b/frontend/tests/chatMultimodalContract.test.mjs new file mode 100644 index 000000000..aae8a9c60 --- /dev/null +++ b/frontend/tests/chatMultimodalContract.test.mjs @@ -0,0 +1,17 @@ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; + +const source = readFileSync( + new URL('../src/pages/agent-detail/AgentDetailPage.tsx', import.meta.url), + 'utf8', +); + +test('persisted multi-image chat renders individual image markers', () => { + assert.match( + source, + /const inlineImageMarkerCount = \(\s*parsed\.content\.match\(\/\\\[image_data:data:image\\\/\/g\) \|\| \[\]\s*\)\.length/, + ); + assert.match(source, /inlineImageMarkerCount <= 1/); + assert.match(source, /inlineImages\.map\(\(url, idx\) =>/); +}); diff --git a/frontend/tests/groupInteractionContract.test.mjs b/frontend/tests/groupInteractionContract.test.mjs index 8bf7bc5f2..b46bc6cab 100644 --- a/frontend/tests/groupInteractionContract.test.mjs +++ b/frontend/tests/groupInteractionContract.test.mjs @@ -18,6 +18,10 @@ const groupUnread = readFileSync( new URL('../src/hooks/useGroupUnread.ts', import.meta.url), 'utf8', ); +const messageStream = readFileSync( + new URL('../src/pages/groups/MessageStream.tsx', import.meta.url), + 'utf8', +); test('new group sessions may use the backend default title while group names stay required', () => { assert.match(promptModal, /allowEmpty\?: boolean/); @@ -84,6 +88,29 @@ test('group composer and stream use session-wide active runs', () => { assert.match(groupsPage, /runningAgents=\{runningAgents\}/); }); +test('planning-to-entry transition keeps polling and preserves the typing indicator', () => { + assert.match(groupsPage, /ACTIVE_RUN_TRANSITION_GRACE_MS/); + assert.match(groupsPage, /planningTransitionUntilRef/); + assert.match(groupsPage, /setAwaitingPlannedRuns\(true\)/); + assert.match(groupsPage, /Date\.now\(\) < planningTransitionUntilRef\.current \? 250 : false/); + assert.match(groupsPage, /planningRunVisible \|\| \(awaitingPlannedRuns && !agentRunVisible\)/); +}); + +test('planning and running agents keep the single transient typing indicator', () => { + assert.match(messageStream, /\{isPlanning && \(/); + assert.match(messageStream, /groups\.taskPlanning/); + assert.match(messageStream, /\{runningAgents\.map\(\(agent\) => \(/); + assert.match(messageStream, /\{agent\.name\}/); + assert.equal( + (messageStream.match(/group-run-indicator-bubble/g) ?? []).length, + 2, + ); + assert.equal( + (messageStream.match(//g) ?? []).length, + 2, + ); +}); + test('group planning failures preserve the backend message and diagnostics', () => { assert.match(groupsPage, /intake\.error\?\.message/); assert.match(groupsPage, /intake\.error\?\.trace_id/); diff --git a/frontend/tests/runtimeModelSettingsContract.test.mjs b/frontend/tests/runtimeModelSettingsContract.test.mjs index 9713c15cd..9639d66ef 100644 --- a/frontend/tests/runtimeModelSettingsContract.test.mjs +++ b/frontend/tests/runtimeModelSettingsContract.test.mjs @@ -6,6 +6,18 @@ const source = readFileSync( new URL('../src/pages/enterprise-settings/tabs/LlmTab.tsx', import.meta.url), 'utf8', ); +const modelSwitcher = readFileSync( + new URL('../src/components/ModelSwitcher.tsx', import.meta.url), + 'utf8', +); +const modelCacheEvents = readFileSync( + new URL('../src/services/modelCacheEvents.ts', import.meta.url), + 'utf8', +); +const agentDetail = readFileSync( + new URL('../src/pages/agent-detail/AgentDetailPage.tsx', import.meta.url), + 'utf8', +); test('company admins can select planning and group context models', () => { assert.match(source, /\/enterprise\/runtime-model-settings/); @@ -19,8 +31,45 @@ test('company admins can select planning and group context models', () => { assert.match(source, /群聊上下文模型/); }); -test('runtime model choices are restricted to tenant-safe backend candidates', () => { +test('runtime model choices use tenant-safe candidates without treating tool probes as a gate', () => { assert.match(source, /runtimeModelSettings\.candidates\.map/); - assert.match(source, /当前公司的模型或平台模型/); + assert.match(source, /当前公司或平台已保存且启用的模型/); + assert.match(source, /工具测试结果仅作诊断，不影响选择/); + assert.doesNotMatch(source, /候选模型必须已启用并通过原生工具调用测试/); assert.match(source, /保存后立即生效/); }); + +test('stale runtime model ids stay unselected instead of selecting the first option', () => { + assert.match(source, /planning_source: 'database' \| 'environment' \| 'unavailable'/); + assert.match(source, /planning_model_id: runtimeModelSettings\.planning_model_id \|\| ''/); + assert.match(source, /compact_model_id: runtimeModelSettings\.compact_model_id \|\| ''/); + assert.match(source, /