WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-03 08:18:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] agents: publicize cross-mixin members (p_run_agent_loop, p_build_mcp_servers, etc.) + safe_resp_text so the decomposition passes p-private

Browse Source
This commit is contained in:
ciregenz
2026-06-23 16:44:41 -07:00
parent 1cc60b2875
commit 751bb77032
23 changed files with 119 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agent_manager.py
+16 -16
View File
@@ -81,7 +81,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# Live mirror of the in-flight streamed assistant text per session, so a
# stop can persist the partial reply instantly instead of waiting out the
# multi-second SDK teardown the cancel handler sits behind.
self.p_live_partial: Dict[str, LivePartial] = {}
self.live_partial: Dict[str, LivePartial] = {}
@@ -106,7 +106,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
@typechecked
async def p_run_agent_loop(self, session_id: str, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, fork_session: bool = False, selected_browser_ids: Optional[List[str]] = None, selected_app_output_ids: Optional[List[str]] = None, selected_setting_ids: Optional[List[str]] = None):
async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, fork_session: bool = False, selected_browser_ids: Optional[List[str]] = None, selected_app_output_ids: Optional[List[str]] = None, selected_setting_ids: Optional[List[str]] = None):
"""Run the Claude Agent SDK query loop for a session."""
session = self.sessions.get(session_id)
if not session:
@@ -114,7 +114,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
from backend.apps.agents.providers.registry import get_api_type as p_get_api_type
p_api = p_get_api_type(session.model)
prompt_content = self.p_build_prompt_content(
prompt_content = self.build_prompt_content(
prompt, images, context_paths, forced_tools, attached_skills,
api_type=p_api, model=session.model,
)
@@ -129,7 +129,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
)
except ImportError:
logger.warning("claude_agent_sdk not installed, running in mock mode")
await self.p_run_mock_agent(session_id, prompt)
await self.run_mock_agent(session_id, prompt)
return
session.status = "running"
@@ -181,7 +181,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# Reconcile active_mcps against currently-enabled tools (Phase 3).
# If the user toggled a server off in the Tools page mid-session,
# drop it from active_mcps automatically so the model isn't told
# "X is active" while p_build_mcp_servers silently filters it out.
# "X is active" while build_mcp_servers silently filters it out.
# Emit a context_status event so the model and UI both know.
try:
p_enabled = {
@@ -230,8 +230,8 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# Pass session.active_mcps as the activation filter. Empty list ⇒
# no MCP tools shipped to the SDK; the model must MCPSearch and
# MCPActivate first. The product invariant lives here at the
# dispatch layer (see p_build_mcp_servers docstring).
mcp_servers = await self.p_build_mcp_servers(session.allowed_tools, session.active_mcps)
# dispatch layer (see build_mcp_servers docstring).
mcp_servers = await self.build_mcp_servers(session.allowed_tools, session.active_mcps)
browser_delegation_tools, invoke_agent_tools = register_builtin_mcp_servers(
mcp_servers, session, builtin_perms, selected_browser_ids, os.path.dirname(__file__)
@@ -338,7 +338,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
logger.info(f"[MCP-DEBUG] effective_disallowed: {effective_disallowed}")
# `p_router_model_id` and `p_api_type_for_session` were resolved
# at the top of p_run_agent_loop (before any closures were
# at the top of run_agent_loop (before any closures were
# defined) so analytics closures could tag events with them.
# Reuse those values here and keep session.provider in sync.
resolved_model = p_router_model_id
@@ -519,14 +519,14 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# programmatic summarization (no aux LLM call) so this adds
# zero latency on the user's turn.
try:
if self.p_maybe_compact(session):
if self.maybe_compact(session):
new_input = estimate_post_compact_input(session)
await ws_manager.send_to_session(session_id, "agent:context_status", {
"session_id": session_id,
"reason": "compacted",
"compacted_through_msg_id": session.compacted_through_msg_id,
})
await self.p_emit_context_update(
await self.emit_context_update(
session_id,
session,
input_tokens=new_input,
@@ -741,12 +741,12 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
if isinstance(message, StreamEvent):
await stream_event.handle_stream_event(
message, session, session_id, turn, thinking, self.p_live_partial
message, session, session_id, turn, thinking, self.live_partial
)
elif isinstance(message, AssistantMessage):
await assistant_message.handle_assistant_message(
message, session, session_id, turn, thinking, self.p_live_partial, self.sessions
message, session, session_id, turn, thinking, self.live_partial, self.sessions
)
elif isinstance(message, ResultMessage):
await result_message.handle_result_message(
@@ -797,7 +797,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
})
turn.stream_text_msg_id = None
turn.stream_text_accum = ""
self.p_live_partial.pop(session_id, None)
self.live_partial.pop(session_id, None)
for p_tool_msg_id in turn.stream_tool_msg_ids_ordered:
await ws_manager.send_to_session(session_id, "agent:stream_end", {
"session_id": session_id,
@@ -820,7 +820,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# analogous flow) flagged pending_continuation during this
# turn, kick off a follow-up turn immediately with the
# captured prompt. We dispatch as a fire-and-forget task so
# the current p_run_agent_loop frame can unwind cleanly
# the current run_agent_loop frame can unwind cleanly
# before the next turn's options + history rebuild kicks in.
# The follow-up is `hidden=True` so it doesn't add a user
# bubble to the visible chat; the model sees it as a
@@ -854,7 +854,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
session.needs_fresh_session = True
# Persist whatever streamed before the cancel (edit / branch
# switch paths; the user-stop path already did this in stop_agent).
await self.p_commit_partial_now(session)
await self.commit_partial_now(session)
turn.stream_text_msg_id = None
turn.stream_text_accum = ""
except Exception as e:
@@ -1106,7 +1106,7 @@ class AgentManager(SessionLifecycleMixin, SessionPersistenceMixin, MessagingMixi
# snapshot the live turn is writing.
p_is_live_task = self.tasks.get(session_id) is asyncio.current_task()
if p_is_live_task:
self.p_live_partial.pop(session_id, None)
self.live_partial.pop(session_id, None)
if session_id in self.sessions and p_is_live_task:
# For canvas-launched App Builder sessions, the workspace
# folder IS the session_id (see launch_agent), so meta.json
backend/apps/agents/agents.py
+3 -3
View File
@@ -306,7 +306,7 @@ async def compact_session(session_id: str):
session = agent_manager.sessions.get(session_id)
if not session:
raise HTTPException(status_code=404, detail="session not found")
fired = agent_manager.p_maybe_compact(session, force=True)
fired = agent_manager.maybe_compact(session, force=True)
if fired:
from backend.apps.agents.core.ws_manager import ws_manager
try:
@@ -315,7 +315,7 @@ async def compact_session(session_id: str):
"reason": "compacted",
"compacted_through_msg_id": session.compacted_through_msg_id,
})
await agent_manager.p_emit_context_update(
await agent_manager.emit_context_update(
session_id,
session,
input_tokens=estimate_post_compact_input(session),
@@ -346,7 +346,7 @@ async def clear_session(session_id: str):
"status": session.status,
"session": session.model_dump(mode="json"),
})
await agent_manager.p_emit_context_update(
await agent_manager.emit_context_update(
session_id,
session,
input_tokens=0,
backend/apps/agents/browser/browser_agent.py
+2 -2
View File
@@ -848,7 +848,7 @@ async def run_browser_agent(
except Exception:
pass
session.status = "completed"
agent_manager.p_sync_session_close(session)
agent_manager.sync_session_close(session)
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id, "status": "completed",
"session": session.model_dump(mode="json"),
@@ -2149,7 +2149,7 @@ async def run_browser_agent(
except Exception as e:
logger.warning(f"[browser-agent {session_id}] keep_open persist failed: {e}")
agent_manager.p_sync_session_close(session)
agent_manager.sync_session_close(session)
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id,
"status": final_status,
backend/apps/agents/browser/browser_fast_path.py
+2 -2
View File
@@ -242,8 +242,8 @@ async def classify_and_brief(prompt: str, settings, primary_api: str | None) ->
),
timeout=8.0,
)
from backend.apps.agents.core.aux_llm import _safe_resp_text
verdict, brief = _parse_verdict_and_brief(_safe_resp_text(resp))
from backend.apps.agents.core.aux_llm import safe_resp_text
verdict, brief = _parse_verdict_and_brief(safe_resp_text(resp))
logger.info(
f"[browser-fast-path] classifier: {verdict.upper()} brief={len(brief)}ch "
f"model={aux_model} in {int((time.monotonic() - t0) * 1000)}ms"
backend/apps/agents/browser/browser_fast_read.py
+2 -2
View File
@@ -63,7 +63,7 @@ async def try_fast_read(prompt: str, brief: str, settings, primary_api: str | No
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.providers.registry import resolve_aux_model
from backend.apps.agents.core.aux_llm import _safe_resp_text
from backend.apps.agents.core.aux_llm import safe_resp_text
aux_model, _ = await resolve_aux_model(
settings, preferred_tier="haiku", primary_api=primary_api,
@@ -83,7 +83,7 @@ async def try_fast_read(prompt: str, brief: str, settings, primary_api: str | No
),
timeout=15.0,
)
answer = _safe_resp_text(resp).strip()
answer = safe_resp_text(resp).strip()
answer_ms = int((time.monotonic() - t1) * 1000)
if not answer or answer.upper().startswith("INSUFFICIENT"):
logger.info(f"[browser-fast-read] aux found page insufficient ({answer_ms}ms); browser fallback")
backend/apps/agents/core/aux_llm.py
+1 -1
View File
@@ -26,7 +26,7 @@ def aux_max_tokens_for(model: str | None, base: int = 100) -> int:
return base
def _safe_resp_text(resp) -> str:
def safe_resp_text(resp) -> str:
"""Extract text from an Anthropic-shape response, tolerating Gemini/OpenAI
edge cases. Gemini through 9Router occasionally returns `content=[]` (e.g.
safety stop, function-call-only turn) which makes `resp.content[0].text`
backend/apps/agents/manager/AgentLaunchMixin.py
+2 -2
View File
@@ -1,6 +1,6 @@
"""Agent run entry points for AgentManager: launch a new top-level run and the staticmethod
invoke_agent helper (fork-and-send a sub-agent). The no-SDK mock fallback lives in MockAgentMixin.
Split into a mixin to keep the manager file under the size ceiling; self.p_run_agent_loop /
Split into a mixin to keep the manager file under the size ceiling; self.run_agent_loop /
self.sessions resolve across the MRO exactly as before."""
import logging
@@ -230,7 +230,7 @@ class AgentLaunchMixin:
"message": user_msg.model_dump(mode="json"),
})
await self.p_run_agent_loop(fork.id, message, fork_session=True)
await self.run_agent_loop(fork.id, message, fork_session=True)
last_assistant = None
for msg in reversed(fork.messages):
backend/apps/agents/manager/MessagingMixin.py
+2 -2
View File
@@ -154,7 +154,7 @@ class MessagingMixin:
if fast_verdict != "no":
task = asyncio.create_task(browser_dispatch.run_browser_fast_path(session, session_id, prompt, selected_browser_ids, fast_brief, fast_verdict))
else:
task = asyncio.create_task(self.p_run_agent_loop(session_id, prompt, images=images, context_paths=context_paths, forced_tools=forced_tools, attached_skills=attached_skills, selected_browser_ids=selected_browser_ids, selected_app_output_ids=selected_app_output_ids, selected_setting_ids=selected_setting_ids))
task = asyncio.create_task(self.run_agent_loop(session_id, prompt, images=images, context_paths=context_paths, forced_tools=forced_tools, attached_skills=attached_skills, selected_browser_ids=selected_browser_ids, selected_app_output_ids=selected_app_output_ids, selected_setting_ids=selected_setting_ids))
self.tasks[session_id] = task
@typechecked
@@ -234,7 +234,7 @@ class MessagingMixin:
"session": session.model_dump(mode="json"),
})
task = asyncio.create_task(self.p_run_agent_loop(
task = asyncio.create_task(self.run_agent_loop(
session_id, new_content,
images=target_msg.images,
context_paths=target_msg.context_paths,
backend/apps/agents/manager/MockAgentMixin.py
+3 -3
View File
@@ -19,7 +19,7 @@ logger = logging.getLogger(__name__)
class MockAgentMixin:
@typechecked
async def p_run_mock_agent(self, session_id: str, prompt: str):
async def run_mock_agent(self, session_id: str, prompt: str):
"""Mock agent loop for development without claude_agent_sdk installed."""
session = self.sessions.get(session_id)
if not session:
@@ -55,7 +55,7 @@ class MockAgentMixin:
tool_input_content = {"tool": "Bash", "input": {"command": f"echo 'Processing: {prompt}'"}, "approved": decision.get("behavior") == "allow"}
tool_msg_id = uuid4().hex
await self.p_stream_tool_input(
await self.stream_tool_input(
session_id, tool_msg_id, "Bash",
json.dumps(tool_input_content["input"], indent=2),
)
@@ -85,7 +85,7 @@ class MockAgentMixin:
f"The agent was configured with:\n- Model: {session.model}\n- Mode: {session.mode}"
)
asst_msg_id = uuid4().hex
await self.p_stream_text(session_id, asst_msg_id, asst_text)
await self.stream_text(session_id, asst_msg_id, asst_text)
asst_msg = Message(id=asst_msg_id, role="assistant", content=asst_text, branch_id=session.active_branch_id)
session.messages.append(asst_msg)
backend/apps/agents/manager/RunSupportMixin.py
+11 -11
View File
@@ -1,7 +1,7 @@
"""Per-run support methods for AgentManager: build the gated MCP server set, warm the prompt
cache, stream-emit helpers, commit/drain a stopped turn, context-update broadcast, and the aux
metadata + prompt/attachment delegators. Split into a mixin to keep the manager file under the
size ceiling; self.sessions / self.tasks / self.p_live_partial resolve across the MRO as before."""
size ceiling; self.sessions / self.tasks / self.live_partial resolve across the MRO as before."""
import asyncio
import logging
@@ -39,7 +39,7 @@ logger = logging.getLogger(__name__)
class RunSupportMixin:
@typechecked
async def p_build_mcp_servers(
async def build_mcp_servers(
self,
allowed_tools: List[str],
active_mcps: Optional[List[str]] = None,
@@ -118,11 +118,11 @@ class RunSupportMixin:
return build_dir_tree(root, max_depth, prefix)
@typechecked
def p_maybe_compact(self, session: AgentSession, force: bool = False) -> bool:
def maybe_compact(self, session: AgentSession, force: bool = False) -> bool:
return context_budget.maybe_compact(session, force)
@typechecked
async def p_emit_context_update(
async def emit_context_update(
self,
session_id: str,
session: AgentSession,
@@ -139,11 +139,11 @@ class RunSupportMixin:
)
@typechecked
def p_build_prompt_content(self, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, api_type: str = "anthropic", model: str = ""):
def build_prompt_content(self, prompt: str, images: Optional[List] = None, context_paths: Optional[List] = None, forced_tools: Optional[List[str]] = None, attached_skills: Optional[List] = None, api_type: str = "anthropic", model: str = ""):
return build_prompt_content(prompt, images, context_paths, forced_tools, attached_skills, api_type, model)
@typechecked
def p_resolve_attachments(self, context_paths: Optional[List], api_type: str, model: str) -> Tuple[str, List[dict], List[str]]:
def resolve_attachments(self, context_paths: Optional[List], api_type: str, model: str) -> Tuple[str, List[dict], List[str]]:
return resolve_attachments(context_paths, api_type, model)
@typechecked
@@ -151,7 +151,7 @@ class RunSupportMixin:
return resolve_context_paths(context_paths)
@typechecked
async def p_stream_text(self, session_id: str, msg_id: str, text: str, delay: float = 0.03):
async def stream_text(self, session_id: str, msg_id: str, text: str, delay: float = 0.03):
"""Emit stream_start, word-by-word deltas, and stream_end for a text message."""
await ws_manager.send_to_session(session_id, "agent:stream_start", {
"session_id": session_id,
@@ -173,7 +173,7 @@ class RunSupportMixin:
})
@typechecked
async def p_stream_tool_input(self, session_id: str, msg_id: str, tool_name: str, input_json: str, delay: float = 0.02):
async def stream_tool_input(self, session_id: str, msg_id: str, tool_name: str, input_json: str, delay: float = 0.02):
"""Emit stream_start, chunked deltas, and stream_end for a tool_call input."""
await ws_manager.send_to_session(session_id, "agent:stream_start", {
"session_id": session_id,
@@ -195,12 +195,12 @@ class RunSupportMixin:
})
@typechecked
async def p_commit_partial_now(self, session) -> bool:
async def commit_partial_now(self, session) -> bool:
"""Persist the in-flight streamed assistant text as a real message and
push it to the client, idempotently. Lets a stop show the partial
instantly instead of waiting out the SDK teardown the cancel handler
sits behind. Returns True if it committed something."""
live = self.p_live_partial.pop(session.id, None)
live = self.live_partial.pop(session.id, None)
if not live:
return False
text = live.text or ""
@@ -230,7 +230,7 @@ class RunSupportMixin:
return True
@typechecked
async def p_drain_task(self, task) -> None:
async def drain_task(self, task) -> None:
"""Await a cancelled task's (possibly slow) teardown off the hot path."""
try:
await task
backend/apps/agents/manager/SessionControlMixin.py
+2 -2
View File
@@ -47,7 +47,7 @@ class SessionControlMixin:
# teardown, which can take several seconds; doing it here means the
# streamed text stays put the instant Stop is pressed instead of
# blinking out and reappearing once teardown finishes.
await self.p_commit_partial_now(session)
await self.commit_partial_now(session)
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id,
"status": "stopped",
@@ -68,7 +68,7 @@ class SessionControlMixin:
task = self.tasks.pop(session_id, None)
if task and not task.done():
task.cancel()
asyncio.create_task(self.p_drain_task(task))
asyncio.create_task(self.drain_task(task))
@typechecked
def handle_approval(self, request_id: str, decision: Dict):
backend/apps/agents/manager/builtin_mcp_servers.py
+1 -1
View File
@@ -78,7 +78,7 @@ def register_builtin_mcp_servers(
# Always-on meta-MCP server. Exposes MCPList / MCPSearch /
# MCPActivate so the model can discover and activate user MCPs at
# runtime. The activation gate (active_mcps filter in
# p_build_mcp_servers above) ensures the model cannot reach any
# build_mcp_servers above) ensures the model cannot reach any
# other MCP server's tools without going through this layer first.
mcp_meta_server_path = os.path.join(
agents_dir, "mcp_meta_server.py"
backend/apps/agents/manager/session/SessionLifecycleMixin.py
+8 -8
View File
@@ -31,11 +31,11 @@ logger = logging.getLogger(__name__)
class SessionLifecycleMixin:
@staticmethod
@typechecked
def p_build_search_text(session: AgentSession, max_len: int = 5000) -> str:
def build_search_text(session: AgentSession, max_len: int = 5000) -> str:
return build_search_text(session, max_len)
@typechecked
def p_sync_session_close(self, session: AgentSession, close_reason: str = "user"):
def sync_session_close(self, session: AgentSession, close_reason: str = "user"):
sync_session_close(session, close_reason)
@typechecked
@@ -72,10 +72,10 @@ class SessionLifecycleMixin:
if hasattr(session, '_cancel_event'):
session._cancel_event.set()
self.p_sync_session_close(session)
self.sync_session_close(session)
doc_data = session.model_dump(mode="json")
doc_data["search_text"] = self.p_build_search_text(session)
doc_data["search_text"] = self.build_search_text(session)
save_session(session_id, doc_data)
@@ -91,18 +91,18 @@ class SessionLifecycleMixin:
"dashboard_id": session.dashboard_id,
})
self.p_purge_session_memory(session_id)
self.purge_session_memory(session_id)
logger.info(f"Session {session_id} closed and persisted")
@typechecked
def p_purge_session_memory(self, session_id: str) -> None:
def purge_session_memory(self, session_id: str) -> None:
"""Drop a session from EVERY in-memory structure keyed by its id, so a
close or delete can't strand stale per-session state that lives until
the process dies. One chokepoint on purpose: a new per-session cache
wires its eviction in HERE and both removal paths get it for free."""
self.sessions.pop(session_id, None)
self.tasks.pop(session_id, None)
self.p_live_partial.pop(session_id, None)
self.live_partial.pop(session_id, None)
view_builder_render_retry_counts.pop(session_id, None)
view_builder_dirty_sessions.discard(session_id)
@@ -125,7 +125,7 @@ class SessionLifecycleMixin:
except asyncio.CancelledError:
pass
self.p_purge_session_memory(session_id)
self.purge_session_memory(session_id)
delete_session_file(session_id)
logger.info(f"Session {session_id} permanently deleted")
backend/apps/agents/manager/session/SessionPersistenceMixin.py
+3 -3
View File
@@ -1,7 +1,7 @@
"""Bulk session persistence across the WHOLE store, the startup/shutdown orchestration that
operates on every session at once (reconcile stale-running, flush-all on shutdown, restore-all
on boot). Split from SessionLifecycleMixin (which handles ONE session at a time) so each file is
one concern. self.sessions / self.p_sync_session_close resolve across the MRO as before."""
one concern. self.sessions / self.sync_session_close resolve across the MRO as before."""
import logging
@@ -50,9 +50,9 @@ class SessionPersistenceMixin:
# Tag this close as "shutdown" so the cloud can tell it apart
# from a user-initiated close. The desktop doesn't care; the
# tag rides along in the dump for whoever consumes it.
self.p_sync_session_close(session, close_reason="shutdown")
self.sync_session_close(session, close_reason="shutdown")
doc_data = session.model_dump(mode="json")
doc_data["search_text"] = self.p_build_search_text(session)
doc_data["search_text"] = self.build_search_text(session)
save_session(session_id, doc_data)
logger.info(f"Persisted session {session_id} on shutdown")
self.sessions.clear()
backend/apps/outputs/outputs.py
+2 -2
View File
@@ -639,8 +639,8 @@ async def vibe_code(body: VibeCodeRequest):
system=VIBE_CODE_SYSTEM_PROMPT,
messages=[{"role": "user", "content": user_message}],
)
from backend.apps.agents.core.aux_llm import _safe_resp_text
raw = _safe_resp_text(resp).strip()
from backend.apps.agents.core.aux_llm import safe_resp_text
raw = safe_resp_text(resp).strip()
if not raw:
return {
"message": "Aux model returned no content. Please try again.",
backend/apps/outputs/publish_scan.py
+2 -2
View File
@@ -113,7 +113,7 @@ async def _llm_findings(src: dict[str, str], settings) -> tuple[list[str], str]:
return [], "clean"
from backend.apps.agents.providers.registry import resolve_aux_model
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.core.aux_llm import _safe_resp_text
from backend.apps.agents.core.aux_llm import safe_resp_text
try:
model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
except Exception:
@@ -129,7 +129,7 @@ async def _llm_findings(src: dict[str, str], settings) -> tuple[list[str], str]:
except Exception:
logger.exception("publish LLM scan call failed; AST-only result stands")
return [], "clean"
text = _safe_resp_text(resp).strip()
text = safe_resp_text(resp).strip()
if text.startswith("```"):
text = text.split("\n", 1)[1] if "\n" in text else text[3:]
if text.endswith("```"):
backend/main.py
+1 -1
View File
@@ -868,7 +868,7 @@ async def session_compact(session_id: str):
session = agent_manager.sessions.get(session_id)
if not session:
return JSONResponse({"error": "session not found"}, status_code=404)
did_compact = agent_manager.p_maybe_compact(session, force=True)
did_compact = agent_manager.maybe_compact(session, force=True)
if did_compact:
session.needs_fresh_session = True
await _ws.send_to_session(session_id, "agent:context_status", {
backend/tests/formal/mcp_gate_proof.py
+1 -1
View File
@@ -1,7 +1,7 @@
"""Formal proof (Z3 / SMT) of the MCP dispatch-gate security invariant.
The product rule "MCP tools are reachable only after MCPActivate" is enforced at
dispatch in agent_manager.p_build_mcp_servers: for a gated session a server is
dispatch in agent_manager.build_mcp_servers: for a gated session a server is
forwarded to the model only if its sanitized name is in session.active_mcps.
tests/test_v2_invariants.py::test_mcp_gate_only_forwards_activated_servers
backend/tests/test_browser_agent_loop.py
+1 -1
View File
@@ -83,7 +83,7 @@ def _install(monkeypatch, primary, aux):
monkeypatch.setattr(cred_mod, "get_anthropic_client_for_model", _client_for, raising=True)
monkeypatch.setattr(BA, "load_builtin_permissions", lambda: {}, raising=True)
monkeypatch.setattr(am_mod.agent_manager, "p_sync_session_close", lambda *a, **k: None, raising=True)
monkeypatch.setattr(am_mod.agent_manager, "sync_session_close", lambda *a, **k: None, raising=True)
# fake WS: record browser commands, script results by action
sent = []
backend/tests/test_session_cleanup.py
+5 -5
View File
@@ -4,7 +4,7 @@ The orchestration core keeps several maps keyed by session id (the session
record, its asyncio task, the live partial-stream mirror, and two module-level
view-builder retry/dirty structures). Removal used to pop only `sessions` +
`tasks`, leaking the rest for the life of the process, an unbounded creep over
a long-running app. `p_purge_session_memory` is the single chokepoint both the
a long-running app. `purge_session_memory` is the single chokepoint both the
close and delete paths route through; this pins the invariant that after it
runs the id is gone from EVERY structure, while a sibling session is untouched.
@@ -18,15 +18,15 @@ def test_purge_session_memory_clears_every_structure():
mgr = am.AgentManager()
mgr.sessions = {"dead": object(), "alive": object()}
mgr.tasks = {"dead": object()}
mgr.p_live_partial = {"dead": {"text": "half a reply"}}
mgr.live_partial = {"dead": {"text": "half a reply"}}
vbs.view_builder_render_retry_counts["dead"] = 4
vbs.view_builder_dirty_sessions.add("dead")
mgr.p_purge_session_memory("dead")
mgr.purge_session_memory("dead")
assert "dead" not in mgr.sessions
assert "dead" not in mgr.tasks
assert "dead" not in mgr.p_live_partial
assert "dead" not in mgr.live_partial
assert "dead" not in vbs.view_builder_render_retry_counts
assert "dead" not in vbs.view_builder_dirty_sessions
# Only the target id is purged; an unrelated live session survives.
@@ -37,5 +37,5 @@ def test_purge_is_safe_on_an_untracked_id():
# Purging an id that was never tracked must be a quiet no-op, not a KeyError,
# so the delete/close paths can call it unconditionally.
mgr = am.AgentManager()
mgr.p_purge_session_memory("never-existed")
mgr.purge_session_memory("never-existed")
assert mgr.sessions == {}
backend/tests/test_streaming_harness.py
+10 -10
View File
@@ -1,4 +1,4 @@
"""Streaming harness: drive the real p_run_agent_loop with a MOCKED claude_agent_sdk.query
"""Streaming harness: drive the real run_agent_loop with a MOCKED claude_agent_sdk.query
that yields a controlled SDK message sequence, and assert the session state + emitted WS
events. This is the safety net for restructuring the streaming loop (it had no isolated
coverage), so it pins the observable contract: streamed text lands as an assistant message,
@@ -26,7 +26,7 @@ def _mock_query_yielding(*messages):
def _drive(monkeypatch, messages, prompt="hi"):
"""Run one p_run_agent_loop turn against a mocked SDK message stream; return (session, ws_events)."""
"""Run one run_agent_loop turn against a mocked SDK message stream; return (session, ws_events)."""
events = []
async def fake_send(session_id, event, data):
@@ -39,7 +39,7 @@ def _drive(monkeypatch, messages, prompt="hi"):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, prompt))
asyncio.run(mgr.run_agent_loop(session.id, prompt))
return session, events
@@ -82,7 +82,7 @@ def _capture_env(monkeypatch, settings, api_type, resolved_model, model_entry):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
return captured["options"].env
@@ -191,7 +191,7 @@ def test_loop_builds_direct_anthropic_key_env(monkeypatch):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
env = captured["options"].env
assert env == {"ANTHROPIC_API_KEY": "sk-ant-test123"} # direct key, no 9router proxy
@@ -223,7 +223,7 @@ def test_loop_with_session_cwd_runs_workspace_git_init(monkeypatch):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d", cwd="/tmp/openswarm-test-ws")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
assert called.get("cwd") == "/tmp/openswarm-test-ws" # the git-init path ran (no NameError)
assert session.status == "completed"
@@ -261,7 +261,7 @@ def test_full_streaming_turn_drives_the_complete_ws_contract(monkeypatch):
def test_loop_wires_all_four_hooks_to_a_live_hook_context(monkeypatch):
# Integration coverage the unit tests can't give: capture the ClaudeAgentOptions the real
# loop hands to query(), then invoke the WIRED hooks. This proves p_run_agent_loop builds a
# loop hands to query(), then invoke the WIRED hooks. This proves run_agent_loop builds a
# HookContext (all required fields, incl. the live `sessions` registry) and the four thin
# wrappers delegate to the extracted hook modules. The SDK never fires these under a mocked
# query, so without this the wiring (not just the functions) would be untested.
@@ -282,7 +282,7 @@ def test_loop_wires_all_four_hooks_to_a_live_hook_context(monkeypatch):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
options = captured["options"]
assert options is not None
@@ -380,7 +380,7 @@ def test_transient_capacity_error_is_retried_then_succeeds(monkeypatch):
from backend.apps.agents.core.models import AgentSession
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
assert state["n"] == 2 # retried exactly once
assert any(m.role == "assistant" and "Recovered" in str(m.content) for m in session.messages)
@@ -413,7 +413,7 @@ def test_thinking_pill_shows_per_turn_delta_not_cumulative(monkeypatch):
session = AgentSession(name="t", model="sonnet", dashboard_id="d")
session.tokens = {"input_fresh": 1000, "output": 500} # prior-turn accumulation
mgr.sessions[session.id] = session
asyncio.run(mgr.p_run_agent_loop(session.id, "hi"))
asyncio.run(mgr.run_agent_loop(session.id, "hi"))
assert pills, "expected a consolidated thinking pill"
assert pills[-1]["input_tokens"] == 150 # (1100-1000)+(550-500), not the cumulative 1650
backend/tests/test_v2_invariants.py
+38 -38
View File
@@ -83,7 +83,7 @@ async def test_gate_blocks_when_active_mcps_empty():
patch("backend.apps.agents.manager.RunSupportMixin.refresh_google_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
# allowed_tools includes mcp:Gmail, but active_mcps is empty
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"],
active_mcps=[],
)
@@ -102,7 +102,7 @@ async def test_gate_allows_only_activated_servers():
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=fake_tools), \
patch("backend.apps.agents.manager.RunSupportMixin.refresh_google_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"],
active_mcps=["gmail"], # sanitized name of "Gmail"
)
@@ -120,7 +120,7 @@ async def test_gate_unset_active_mcps_legacy_allows_all():
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=fake_tools), \
patch("backend.apps.agents.manager.RunSupportMixin.refresh_google_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail", "mcp:Slack"],
active_mcps=None, # legacy / unset
)
@@ -135,7 +135,7 @@ async def test_gate_disabled_tool_blocked_even_when_activated():
fake_tools = [_fake_tool("Gmail", enabled=False)]
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=fake_tools):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail"],
active_mcps=["gmail"],
)
@@ -149,7 +149,7 @@ async def test_gate_unauthed_tool_blocked():
fake_tools = [_fake_tool("Gmail", auth_status="disconnected")]
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=fake_tools):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail"],
active_mcps=["gmail"],
)
@@ -164,7 +164,7 @@ async def test_gate_allowed_tools_filter_intersects_active_mcps():
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=fake_tools), \
patch("backend.apps.agents.manager.RunSupportMixin.refresh_google_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail"], # mode-restricted
active_mcps=["gmail", "slack"], # both activated
)
@@ -197,7 +197,7 @@ async def test_gate_stress_random_activations():
patch("backend.apps.agents.manager.RunSupportMixin.refresh_airtable_token", new=AsyncMock(return_value=True)), \
patch("backend.apps.agents.manager.RunSupportMixin.refresh_hubspot_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=allowed,
active_mcps=active,
)
@@ -649,9 +649,9 @@ async def test_mcp_gate_only_forwards_activated_servers():
patch("backend.apps.agents.manager.RunSupportMixin.derive_mcp_config", side_effect=lambda t: {"command": "x"}):
allowed = ["__ALL__"]
# Boundary 1: empty activation list -> zero servers, always.
assert await mgr.p_build_mcp_servers(allowed, active_mcps=[]) == {}
assert await mgr.build_mcp_servers(allowed, active_mcps=[]) == {}
# Boundary 2: None (legacy) -> permission gate only, all forwarded.
assert set((await mgr.p_build_mcp_servers(allowed, active_mcps=None)).keys()) == set(names)
assert set((await mgr.build_mcp_servers(allowed, active_mcps=None)).keys()) == set(names)
# Property: forwarded set is ALWAYS a subset of the activated set, and
# equals exactly the activated-and-installed intersection.
rng = random.Random(1234)
@@ -660,7 +660,7 @@ async def test_mcp_gate_only_forwards_activated_servers():
# throw in a bogus name the gate must never invent a server for
if rng.random() < 0.3:
active = active + ["ghost-not-installed"]
forwarded = set((await mgr.p_build_mcp_servers(allowed, active_mcps=active)).keys())
forwarded = set((await mgr.build_mcp_servers(allowed, active_mcps=active)).keys())
assert forwarded <= set(active), f"leaked {forwarded - set(active)} for active={active}"
assert forwarded == (set(active) & set(names)), f"mismatch for active={active}"
@@ -967,10 +967,10 @@ async def test_concurrent_gate_calls_isolated():
patch("backend.apps.agents.manager.RunSupportMixin.refresh_google_token", new=AsyncMock(return_value=True)):
mgr = AgentManager()
results = await asyncio.gather(
mgr.p_build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["gmail"]),
mgr.p_build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["slack"]),
mgr.p_build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["notion"]),
mgr.p_build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=[]),
mgr.build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["gmail"]),
mgr.build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["slack"]),
mgr.build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=["notion"]),
mgr.build_mcp_servers(allowed_tools=["mcp:Gmail", "mcp:Slack", "mcp:Notion"], active_mcps=[]),
)
gmail_only, slack_only, notion_only, empty = results
assert set(gmail_only.keys()) == {"gmail"}
@@ -1051,7 +1051,7 @@ async def test_context_update_emitter_refreshes_session_tokens(monkeypatch):
s.framework_overhead_tokens = 42
s.active_mcps = ["github"]
await AgentManager().p_emit_context_update("x", s, input_tokens=250)
await AgentManager().emit_context_update("x", s, input_tokens=250)
assert s.tokens == {"input": 250, "output": 7}
assert sent == [(
@@ -1186,7 +1186,7 @@ async def test_gate_handles_missing_refresh_token_gracefully():
fake.auth_type = None # no oauth
with patch("backend.apps.agents.manager.RunSupportMixin.load_all_tools", return_value=[fake]):
mgr = AgentManager()
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:MyApiTool"],
active_mcps=["myapitool"],
)
@@ -1384,7 +1384,7 @@ def test_resolve_attachments_handles_missing_path_gracefully():
we emit a 'not found' refusal instead of crashing."""
from backend.apps.agents.agent_manager import AgentManager
mgr = AgentManager()
text, native, refusals = mgr.p_resolve_attachments(
text, native, refusals = mgr.resolve_attachments(
[{"path": "/var/folders/nonexistent/definitely-gone.pdf", "type": "file"}],
api_type="anthropic", model="opus-4-7",
)
@@ -1402,7 +1402,7 @@ def test_resolve_attachments_handles_directory_path_not_file():
tmpdir = tempfile.mkdtemp()
open(os.path.join(tmpdir, "a.txt"), "w").write("hello")
try:
text, native, refusals = mgr.p_resolve_attachments(
text, native, refusals = mgr.resolve_attachments(
[{"path": tmpdir, "type": "directory"}],
api_type="anthropic", model="opus-4-7",
)
@@ -1431,7 +1431,7 @@ def test_resolve_attachments_mixed_kinds_total_size_guard():
paths.append(fh.name)
with tempfile.NamedTemporaryFile(suffix=".md", mode="w", delete=False) as fh:
fh.write("# notes"); paths.append(fh.name)
text, native, refusals = mgr.p_resolve_attachments(
text, native, refusals = mgr.resolve_attachments(
[{"path": p, "type": "file"} for p in paths],
api_type="anthropic", model="opus-4-7",
)
@@ -1491,7 +1491,7 @@ def test_sniff_recognises_macos_paths_with_spaces():
try:
with open(path, "wb") as f:
f.write(b"%PDF-1.4\n")
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
assert native and native[0]["type"] == "document"
@@ -1552,7 +1552,7 @@ def test_sniff_handles_windows_style_backslash_path_string():
from backend.apps.agents.agent_manager import AgentManager
mgr = AgentManager()
# A path that doesn't exist (POSIX cannot interpret backslashes as separator)
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": r"C:\fake\path\nope.pdf", "type": "file"}],
api_type="anthropic", model="opus-4-7",
)
@@ -1591,7 +1591,7 @@ def test_resolve_attachments_classifies_renamed_binary_as_binary_not_pdf():
fh.write(b"PK\x03\x04fake zip masquerading as pdf")
path = fh.name
try:
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
assert not native
@@ -1664,7 +1664,7 @@ def test_anthropic_document_block_schema_matches_docs():
fh.write(b"%PDF-1.4\n%canonical schema test\n")
path = fh.name
try:
_t, native, _r = mgr.p_resolve_attachments(
_t, native, _r = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
block = native[0]
@@ -1909,7 +1909,7 @@ def test_resolve_attachments_openai_codex_refused_for_pdfs():
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="openai", model="gpt-5.3-codex",
)
assert not native
@@ -1928,7 +1928,7 @@ def test_resolve_attachments_openai_codex_still_refuses_pdf():
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="openai", model="gpt-5.3-codex",
)
assert not native
@@ -2029,7 +2029,7 @@ def test_resolve_attachments_anthropic_emits_native_document():
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
text, native, refusals = mgr.p_resolve_attachments(
text, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
assert native and native[0]["type"] == "document"
@@ -2051,7 +2051,7 @@ def test_resolve_attachments_openai_accepts_pdf_via_bypass_translator():
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
_text, native, refusals = mgr.p_resolve_attachments(
_text, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="openai", model="gpt-5.5",
)
assert native and native[0]["type"] == "document"
@@ -2143,7 +2143,7 @@ def test_resolve_attachments_gemini_emits_native_document_after_translator_fix()
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
_text, native, refusals = mgr.p_resolve_attachments(
_text, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="gemini", model="gemini-3.1-pro-api",
)
assert native and native[0]["type"] == "document"
@@ -2162,7 +2162,7 @@ def test_resolve_attachments_text_file_inlined_not_native():
fh.write("# hello\nworld")
path = fh.name
try:
text, native, refusals = mgr.p_resolve_attachments(
text, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="opus-4-7", model="opus-4-7",
)
assert not native
@@ -2182,7 +2182,7 @@ def test_resolve_attachments_pdf_refused_when_too_large():
fh.write(b"X" * (25 * 1024 * 1024))
path = fh.name
try:
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
assert not native
@@ -2208,7 +2208,7 @@ def test_resolve_attachments_refuses_when_total_exceeds_request_cap():
fh.write(b"%PDF-1.4\n")
fh.write(b"X" * (8 * 1024 * 1024))
paths.append(fh.name)
_t, native, refusals = mgr.p_resolve_attachments(
_t, native, refusals = mgr.resolve_attachments(
[{"path": p, "type": "file"} for p in paths],
api_type="anthropic", model="opus-4-7",
)
@@ -2234,7 +2234,7 @@ def test_resolve_attachments_anthropic_marks_last_document_ephemeral_for_cache()
with tempfile.NamedTemporaryFile(suffix=f"_{i}.pdf", delete=False) as fh:
fh.write(b"%PDF-1.4\n%test\n")
paths.append(fh.name)
_t, native, _r = mgr.p_resolve_attachments(
_t, native, _r = mgr.resolve_attachments(
[{"path": p, "type": "file"} for p in paths],
api_type="anthropic", model="opus-4-7",
)
@@ -2258,11 +2258,11 @@ def test_resolve_attachments_anthropic_does_mark_ephemeral_but_only_anthropic():
fh.write(b"%PDF-1.4\n%test\n")
path = fh.name
try:
_t, ant_native, _r = mgr.p_resolve_attachments(
_t, ant_native, _r = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="anthropic", model="opus-4-7",
)
assert ant_native and ant_native[0].get("cache_control") == {"type": "ephemeral"}
_t, or_native, _r = mgr.p_resolve_attachments(
_t, or_native, _r = mgr.resolve_attachments(
[{"path": path, "type": "file"}], api_type="openrouter", model="openrouter/openai/gpt-5",
)
assert or_native and "cache_control" not in or_native[0]
@@ -3033,7 +3033,7 @@ async def test_gate_100_sequential_calls_no_leak():
n = i % 10
active = [f"server{j}" for j in range(n)]
allowed = [f"mcp:Server{j}" for j in range(10)]
result = await mgr.p_build_mcp_servers(allowed_tools=allowed, active_mcps=active)
result = await mgr.build_mcp_servers(allowed_tools=allowed, active_mcps=active)
assert set(result.keys()) == set(active), \
f"iteration {i}: expected {set(active)}, got {set(result.keys())}"
@@ -3147,7 +3147,7 @@ async def test_e2e_session_lifecycle_with_mcp_activation():
s = AgentSession(id="e2e", name="End-to-end", model="sonnet", mode="agent")
# Step 1: fresh, gate blocks everything
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail", "mcp:Slack"],
active_mcps=s.active_mcps,
)
@@ -3161,7 +3161,7 @@ async def test_e2e_session_lifecycle_with_mcp_activation():
s.pending_continuation = True
# Step 3: continuation turn, gate passes gmail
result = await mgr.p_build_mcp_servers(
result = await mgr.build_mcp_servers(
allowed_tools=["mcp:Gmail", "mcp:Slack"],
active_mcps=s.active_mcps,
)
@@ -3192,7 +3192,7 @@ async def test_e2e_50_random_activation_sequences():
n = random.randint(0, len(sanitized))
active = random.sample(sanitized, n)
allowed = [f"mcp:{r}" for r in raw_names]
result = await mgr.p_build_mcp_servers(allowed, active)
result = await mgr.build_mcp_servers(allowed, active)
keys = set(result.keys())
assert keys == set(active), f"mismatch: active={active} keys={keys}"
backend/tests/test_ws_integration.py
+1 -1
View File
@@ -1,7 +1,7 @@
"""Real-server end-to-end WebSocket test: the closest in-repo proxy to a live desktop run. A WS
client connects to the actual FastAPI app, sends agent:send_message, and the agent loop runs and
streams its events BACK over the real WS transport. This exercises the whole live path the
desktop app uses, FastAPI app + WS route + ws_manager + send_message + p_run_agent_loop + the
desktop app uses, FastAPI app + WS route + ws_manager + send_message + run_agent_loop + the
extracted handlers + the broadcast, which the in-process harness (no server, no WS) can't cover.
The SDK and WS auth are mocked; everything else is the real running stack."""