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

[eric] agents: one-line every comment + collapse 3+ blank runs across manager/ surface (33 files, -550 lines, code byte-identical)

Browse Source
This commit is contained in:
ciregenz
2026-06-25 02:16:14 -07:00
parent 3960c2bd46
commit d5cd4db026
34 changed files with 150 additions and 700 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/manager/AgentLaunch.py
+5 -33
View File
@@ -37,10 +37,7 @@ class AgentLaunch(AgentManagerProtocol):
async def launch_agent(self, config: AgentConfig) -> AgentSession:
session_id = uuid4().hex
# Editing an existing App: when the user selected exactly one App card
# in App Builder mode, point the chat at that app's workspace so it
# edits in place. Without this the view-builder seed below fires (no
# target_directory) and registers a fresh empty "Untitled App" dupe.
# Editing an existing App: when the user selected exactly one App card in App Builder mode, point the chat at that app's workspace so it edits in place. Without this the view-builder seed below fires (no target_directory) and registers a fresh empty "Untitled App" dupe.
if (
config.mode == "view-builder"
and not config.target_directory
@@ -68,15 +65,7 @@ class AgentLaunch(AgentManagerProtocol):
os.makedirs(effective_cwd, exist_ok=True)
# Canvas-chat App Builder launch: when the user picks "App Builder"
# mode from the chat-input dropdown (no preexisting workspace, no
# target_directory passed in), the legacy code path only created an
# empty folder, so the agent could write files but the app never
# showed up in the Apps sidebar (no Output row, which is what the
# sidebar reads). Mirror the /workspace/seed endpoint's behavior
# here: seed the React template + register an Output row with
# workspace_id = session_id. Idempotent; safe if the session is
# ever re-launched with the same id.
# Canvas-chat App Builder launch: when the user picks "App Builder" mode from the chat-input dropdown (no preexisting workspace, no target_directory passed in), the legacy code path only created an empty folder, so the agent could write files but the app never showed up in the Apps sidebar (no Output row, which is what the sidebar reads). Mirror the /workspace/seed endpoint's behavior here: seed the React template + register an Output row with workspace_id = session_id. Idempotent; safe if the session is ever re-launched with the same id.
if config.mode == "view-builder" and not config.target_directory:
try:
from backend.apps.outputs.outputs import (
@@ -89,12 +78,7 @@ class AgentLaunch(AgentManagerProtocol):
session_id=session_id,
)
if output_id:
# Broadcast the new row so the Apps sidebar lights up
# immediately, even before the user clicks into it. The
# row name is still the placeholder ("Untitled App") at
# this point; the post-session meta-sync below fires a
# second upsert with the real name once the agent has
# written meta.json.
# Broadcast the new row so the Apps sidebar lights up immediately, even before the user clicks into it. The row name is still the placeholder ("Untitled App") at this point; the post-session meta-sync below fires a second upsert with the real name once the agent has written meta.json.
try:
new_output = load_output(output_id)
await ws_manager.broadcast_global("agent:output_upserted", {
@@ -108,13 +92,7 @@ class AgentLaunch(AgentManagerProtocol):
"still launch but the app may not appear in Apps sidebar"
)
# If the fallback chain landed on the user's home directory (no
# project dir, no default_folder set), re-route to a dedicated
# scratch workspace under ~/.openswarm/workspaces/<session_id>.
# This prevents us from writing .git/ (or anything else) into
# the user's $HOME and gives the CLI's Agent tool a clean repo
# to do worktree isolation inside. Users with a default_folder
# or target_directory set keep whatever they configured.
# If the fallback chain landed on the user's home directory (no project dir, no default_folder set), re-route to a dedicated scratch workspace under ~/.openswarm/workspaces/<session_id>. This prevents us from writing .git/ (or anything else) into the user's $HOME and gives the CLI's Agent tool a clean repo to do worktree isolation inside. Users with a default_folder or target_directory set keep whatever they configured.
home = os.path.expanduser("~")
if os.path.abspath(effective_cwd) == os.path.abspath(home):
effective_cwd = os.path.join(home, ".openswarm", "workspaces", session_id)
@@ -189,13 +167,7 @@ class AgentLaunch(AgentManagerProtocol):
timestamp=msg.timestamp,
branch_id=msg.branch_id,
parent_id=old_to_new_msg.get(msg.parent_id) if msg.parent_id else None,
# Sub-agents do NOT inherit parent's attached files. Each
# parent-message base64-expansion would re-fire in the
# sub-agent (cost explosion: a 25 MB PDF in parent +
# 5 InvokeAgent calls = 125 MB transmitted). The
# sub-agent receives the user's new message only; if it
# needs the file content, the parent message text from
# the prior turn already carries the model's summary.
# Sub-agents do NOT inherit parent's attached files. Each parent-message base64-expansion would re-fire in the sub-agent (cost explosion: a 25 MB PDF in parent + 5 InvokeAgent calls = 125 MB transmitted). The sub-agent receives the user's new message only; if it needs the file content, the parent message text from the prior turn already carries the model's summary.
context_paths=None,
attached_skills=msg.attached_skills,
forced_tools=msg.forced_tools,
backend/apps/agents/manager/AgentManagerProtocol.py
+1 -3
View File
@@ -28,9 +28,7 @@ class AgentManagerProtocol:
cancel_events: Dict[str, asyncio.Event]
if TYPE_CHECKING:
# Methods implemented on sibling mixins / AgentManager itself and called
# cross-mixin. Loose signatures on purpose: typeCheckingMode is off, so this
# only has to assert the names exist, not pin their call shapes.
# Methods implemented on sibling mixins / AgentManager itself and called cross-mixin. Loose signatures on purpose: typeCheckingMode is off, so this only has to assert the names exist, not pin their call shapes.
def run_agent_loop(self, *args: Any, **kwargs: Any) -> Any: ...
def generate_turn_label(self, *args: Any, **kwargs: Any) -> Any: ...
def commit_partial_now(self, *args: Any, **kwargs: Any) -> Any: ...
backend/apps/agents/manager/Messaging.py
+3 -20
View File
@@ -61,14 +61,7 @@ class Messaging(AgentManagerProtocol):
session_changed = False
if model and model != session.model:
# Cross-provider model switches force a session fork. The CLI's
# resume transcript stores Anthropic-format content blocks with
# Anthropic tool_use_ids; replaying them on a non-Anthropic
# provider via 9Router's claude→openai translator corrupts
# history silently (fixMissingToolResponses stubs missing tool
# responses with placeholder text). Forking starts a new CLI
# session so history is re-sent fresh in whichever format the
# new provider expects.
# Cross-provider model switches force a session fork. The CLI's resume transcript stores Anthropic-format content blocks with Anthropic tool_use_ids; replaying them on a non-Anthropic provider via 9Router's claude→openai translator corrupts history silently (fixMissingToolResponses stubs missing tool responses with placeholder text). Forking starts a new CLI session so history is re-sent fresh in whichever format the new provider expects.
from backend.apps.agents.providers.registry import get_api_type as get_api_type_for_model
if get_api_type_for_model(session.model) != get_api_type_for_model(model):
session.needs_fork = True
@@ -108,13 +101,7 @@ class Messaging(AgentManagerProtocol):
"message": user_msg.model_dump(mode="json"),
})
# Fire a background aux LLM call to generate a 3-6 word verb-phrase
# describing this turn ("Auditing the pull request", "Drafting your
# email"). The narrator pill swaps from its heuristic verb to this
# label as soon as it lands, usually ~500ms-1s into the turn,
# which is exactly when "Thinking…" starts feeling generic.
# Provider-agnostic via resolve_aux_model. Non-blocking; failure
# is silent and the heuristic stays.
# Fire a background aux LLM call to generate a 3-6 word verb-phrase describing this turn ("Auditing the pull request", "Drafting your email"). The narrator pill swaps from its heuristic verb to this label as soon as it lands, usually ~500ms-1s into the turn, which is exactly when "Thinking…" starts feeling generic. Provider-agnostic via resolve_aux_model. Non-blocking; failure is silent and the heuristic stays.
if not hidden and prompt:
try:
asyncio.create_task(
@@ -132,11 +119,7 @@ class Messaging(AgentManagerProtocol):
"session": session.model_dump(mode="json"),
})
# Browser fast path: a plainly browser-only first message skips the
# orchestrator LLM entirely (it was ~2/3 of the token bill on these
# tasks, spent deciding "delegate to a browser" and restating the
# outcome). Conservative gates + a cheap aux classifier; any miss or
# error falls through to the normal loop.
# Browser fast path: a plainly browser-only first message skips the orchestrator LLM entirely (it was ~2/3 of the token bill on these tasks, spent deciding "delegate to a browser" and restating the outcome). Conservative gates + a cheap aux classifier; any miss or error falls through to the normal loop.
fast_verdict = "no"
fast_brief = ""
if not hidden:
backend/apps/agents/manager/MockAgent.py
+1 -5
View File
@@ -99,11 +99,7 @@ class MockAgent(AgentManagerProtocol):
session.status = "completed"
session.closed_at = datetime.now()
# Mock branch (claude_agent_sdk missing): leave cost untouched so
# it stays at its 0.0 default. A fake nonzero value here would
# poison the cost shown in the session header during dev. The
# `_mock_run` flag is read by the close path so a mock session
# doesn't get reported to the cloud as a real one.
# Mock branch (claude_agent_sdk missing): leave cost untouched so it stays at its 0.0 default. A fake nonzero value here would poison the cost shown in the session header during dev. The `_mock_run` flag is read by the close path so a mock session doesn't get reported to the cloud as a real one.
setattr(session, "_mock_run", True)
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id,
backend/apps/agents/manager/RunSupport.py
+4 -9
View File
@@ -92,8 +92,7 @@ class RunSupport(AgentManagerProtocol):
if tool.auth_type == "oauth2" and tool.auth_status == "connected":
if tool.name.lower() in ("discord", "github"):
# Discord uses a shared bot token; GitHub OAuth-app tokens don't
# expire and carry no refresh_token. Nothing to refresh either way.
# Discord uses a shared bot token; GitHub OAuth-app tokens don't expire and carry no refresh_token. Nothing to refresh either way.
refreshed = True
elif tool.name.lower() == "airtable":
refreshed = await refresh_airtable_token(tool)
@@ -261,8 +260,7 @@ class RunSupport(AgentManagerProtocol):
session = self.sessions.get(session_id)
if not session:
return
# If a real run is in flight, the cache will be warmed by it;
# firing again is wasted tokens.
# If a real run is in flight, the cache will be warmed by it; firing again is wasted tokens.
existing = self.tasks.get(session_id)
if existing and not existing.done():
return
@@ -275,15 +273,12 @@ class RunSupport(AgentManagerProtocol):
from backend.apps.settings.credentials import get_anthropic_client
global_settings = load_settings()
# Free lane rotates pool accounts per call, so a warm ping primes a cache
# the next call won't hit, and worse it'd burn a metered run at idle (this
# fires on dashboard mount, not a user query). Skip it on the free trial.
# Free lane rotates pool accounts per call, so a warm ping primes a cache the next call won't hit, and worse it'd burn a metered run at idle (this fires on dashboard mount, not a user query). Skip it on the free trial.
if getattr(global_settings, "connection_mode", "own_key") == "free-trial":
return
client = get_anthropic_client(global_settings)
# Single ping with the same system + minimal user message.
# max_tokens=1 keeps it cheap; we don't care about the output.
# Single ping with the same system + minimal user message. max_tokens=1 keeps it cheap; we don't care about the output.
await client.messages.create(
model=entry.get("model_id", session.model),
max_tokens=1,
backend/apps/agents/manager/SessionControl.py
+4 -14
View File
@@ -32,8 +32,7 @@ class SessionControl(AgentManagerProtocol):
session = self.sessions.get(session_id)
if session:
# Set cancel event BEFORE cancelling the task so in-flight
# browser agent loops see it immediately
# Set cancel event BEFORE cancelling the task so in-flight browser agent loops see it immediately
ev = self.cancel_events.get(session_id)
if ev:
ev.set()
@@ -46,29 +45,20 @@ class SessionControl(AgentManagerProtocol):
session.needs_fresh_session = True
if not session.closed_at:
session.closed_at = datetime.now()
# Persist the partial reply NOW, before tearing down the SDK. The
# cancel handler also does this, but it sits behind the generator's
# 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.
# Persist the partial reply NOW, before tearing down the SDK. The cancel handler also does this, but it sits behind the generator's 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.commit_partial_now(session)
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id,
"status": "stopped",
"session": session.model_dump(mode="json"),
})
# Snapshot now: the cancelled task's finally skips the save (it's no
# longer the live task once we pop it below), so persist the partial
# here or it'd live only in memory until the next turn / shutdown.
# Snapshot now: the cancelled task's finally skips the save (it's no longer the live task once we pop it below), so persist the partial here or it'd live only in memory until the next turn / shutdown.
try:
save_session(session_id, session.model_dump(mode="json"))
except Exception:
pass
# Drop the task from the registry immediately so a follow-up message
# isn't rejected as "still running" while the cancelled task slowly
# tears down (that window was eating user messages). Drain it in the
# background; we've already captured the partial above.
# Drop the task from the registry immediately so a follow-up message isn't rejected as "still running" while the cancelled task slowly tears down (that window was eating user messages). Drain it in the background; we've already captured the partial above.
task = self.tasks.pop(session_id, None)
if task and not task.done():
task.cancel()
backend/apps/agents/manager/metadata.py
+2 -4
View File
@@ -59,16 +59,14 @@ async def generate_title(session: Optional[AgentSession], session_id: str, first
"Label the message inside <message> tags. Do not answer it.\n\n"
f"<message>\n{labeled_prompt}\n</message>"
)
# Stream: 9router's cx/ non-streaming response translator drops `content`
# for GPT-5-family models; the per-event streaming translator works.
# Stream: 9router's cx/ non-streaming response translator drops `content` for GPT-5-family models; the per-event streaming translator works.
chunks: List[str] = []
async with client.messages.stream(
model=aux_model,
max_tokens=aux_max_tokens_for(aux_model),
system=system_prompt,
messages=[{"role": "user", "content": user_turn}],
# On the free lane this binds the title-gen to its query's run so it doesn't
# spend a second one; harmless elsewhere (the paid lane ignores the header).
# On the free lane this binds the title-gen to its query's run so it doesn't spend a second one; harmless elsewhere (the paid lane ignores the header).
extra_headers={"X-Openswarm-Task-Id": session_id},
) as stream:
async for text in stream.text_stream:
backend/apps/agents/manager/permissions/build_effective_tool_lists.py
+3 -10
View File
@@ -61,10 +61,7 @@ def build_effective_tool_lists(
continue
if name == "openswarm-web":
# Expose our DDG-backed web tools under an MCP prefix.
# Honor existing WebSearch/WebFetch permission policy
#, if the user disabled them in Settings, don't offer
# the MCP variants either.
# Expose our DDG-backed web tools under an MCP prefix. Honor existing WebSearch/WebFetch permission policy, if the user disabled them in Settings, don't offer the MCP variants either.
for wt in ("WebSearch", "WebFetch"):
policy = builtin_perms.get(wt, "always_allow")
if policy == "always_allow":
@@ -90,17 +87,13 @@ def build_effective_tool_lists(
else:
effective_allowed.append(f"mcp__{name}__*")
# If the openswarm-web MCP was registered, the CLI's built-in
# WebSearch/WebFetch are guaranteed to fail (no Anthropic
# backend). Suppress them so the model picks our MCP variants
# and doesn't waste a turn on a broken tool.
# If the openswarm-web MCP was registered, the CLI's built-in WebSearch/WebFetch are guaranteed to fail (no Anthropic backend). Suppress them so the model picks our MCP variants and doesn't waste a turn on a broken tool.
if need_web_mcp:
effective_allowed = [t for t in effective_allowed if t not in ("WebSearch", "WebFetch")]
for wt_name in ("WebSearch", "WebFetch"):
if wt_name not in effective_disallowed:
effective_disallowed.append(wt_name)
# Claude's internal Cron* scheduler is denied in favour of the visible native
# one; withhold it from the SDK so the model doesn't even reach for it.
# Claude's internal Cron* scheduler is denied in favour of the visible native one; withhold it from the SDK so the model doesn't even reach for it.
for bt in path_gate.CLAUDE_INTERNAL_SCHEDULER_TOOLS:
if bt not in effective_disallowed:
effective_disallowed.append(bt)
backend/apps/agents/manager/permissions/gate_hooks.py
+3 -17
View File
@@ -67,13 +67,7 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona
tool_name = input_data.get("tool_name", "")
hook_event = input_data.get("hook_event_name", "PreToolUse")
# ToolSearch loop-breaker. Gated MCP servers are withheld from the
# SDK until MCPActivate, so the CLI's native ToolSearch can never
# find them; small models thrash (empty ToolSearch, retry) for
# minutes until the user pauses. Let the first couple through, then
# redirect to the gate. Any non-ToolSearch call is real progress, so
# the counter resets. Gated-server lookup is deferred behind the
# threshold so the common (non-looping) path stays free.
# ToolSearch loop-breaker. Gated MCP servers are withheld from the SDK until MCPActivate, so the CLI's native ToolSearch can never find them; small models thrash (empty ToolSearch, retry) for minutes until the user pauses. Let the first couple through, then redirect to the gate. Any non-ToolSearch call is real progress, so the counter resets. Gated-server lookup is deferred behind the threshold so the common (non-looping) path stays free.
if tool_name == "ToolSearch":
ctx.ts_loop_count += 1
if ctx.ts_loop_count >= TOOLSEARCH_LOOP_THRESHOLD:
@@ -81,11 +75,7 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona
reason = toolsearch_loop_redirect(ctx.ts_loop_count, gated)
if reason:
logger.info(f"[MCP-DEBUG] ToolSearch loop-breaker fired for {ctx.session_id} (n={ctx.ts_loop_count})")
# 2B-MCP: also surface a one-click connect offer to the USER for the vetted
# gated servers the agent keeps reaching for. Suggest-only: this just shows a
# card on the same channel the preflight uses; activation still requires
# MCPActivate + the dispatch gate, so it opens no side channel. Once per run,
# fail-open (an offer hiccup must never block the agent).
# 2B-MCP: also surface a one-click connect offer to the USER for the vetted gated servers the agent keeps reaching for. Suggest-only: this just shows a card on the same channel the preflight uses; activation still requires MCPActivate + the dispatch gate, so it opens no side channel. Once per run, fail-open (an offer hiccup must never block the agent).
if not ctx.mcp_offer_sent:
try:
from backend.apps.agents.core.mcp_preflight import offer_for_gated_server
@@ -110,11 +100,7 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona
else:
ctx.ts_loop_count = 0
# MCPSearch is the agent saying "I need an integration I don't have" (e.g. "no email
# connected"). Don't make the user read a wall of options: fire the same curated connect
# card the launch preflight uses, keyed to their original request. Non-blocking (the search
# proceeds) and once per run; covers the common path the ToolSearch-loop branch misses
# because a capable model does one MCPSearch instead of thrashing. Suggest-only as ever.
# MCPSearch is the agent saying "I need an integration I don't have" (e.g. "no email connected"). Don't make the user read a wall of options: fire the same curated connect card the launch preflight uses, keyed to their original request. Non-blocking (the search proceeds) and once per run; covers the common path the ToolSearch-loop branch misses because a capable model does one MCPSearch instead of thrashing. Suggest-only as ever.
if (tool_name.endswith("MCPSearch") or tool_name.endswith("MCPList")) and not ctx.mcp_offer_sent:
ctx.mcp_offer_sent = True
backend/apps/agents/manager/permissions/path_gate.py
+8 -24
View File
@@ -14,9 +14,7 @@ from typeguard import typechecked
from backend.apps.tools_lib.tools_lib import load_trusted_sensitive_paths
# Each entry: pattern -> (short label, plain-English risk). The label/risk is what the
# approval card shows, so it has to read clearly to a non-developer who has never heard
# of `~/.ssh/authorized_keys`.
# Each entry: pattern -> (short label, plain-English risk). The label/risk is what the approval card shows, so it has to read clearly to a non-developer who has never heard of `~/.ssh/authorized_keys`.
P_SENSITIVE_PATH_INFO: Dict[str, Tuple[str, str]] = {
"*/.ssh": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
@@ -45,9 +43,7 @@ P_SENSITIVE_PATH_PATTERNS: Tuple[str, ...] = tuple(P_SENSITIVE_PATH_INFO.keys())
P_PATH_GATED_TOOLS: Tuple[str, ...] = ("Write", "Edit", "NotebookEdit")
# OS-level scheduling across macOS/Linux/Windows. The agent must not install cron entries,
# launchd plists, Windows scheduled tasks, or PowerShell ScheduledTask cmdlets behind the
# user's back. Word-bounded so stray strings in echo etc. don't trip it.
# OS-level scheduling across macOS/Linux/Windows. The agent must not install cron entries, launchd plists, Windows scheduled tasks, or PowerShell ScheduledTask cmdlets behind the user's back. Word-bounded so stray strings in echo etc. don't trip it.
P_OS_SCHED_RE = re.compile(
r"\b("
r"crontab|launchctl|launchd|schtasks|systemd-run|"
@@ -58,10 +54,7 @@ P_OS_SCHED_RE = re.compile(
re.IGNORECASE,
)
# Catastrophic-path Bash gate. Bash is intentionally NOT in P_PATH_GATED_TOOLS (gating
# every `echo ... > /tmp/foo` would interrupt routine work), but a single redirected write
# to one of these can grant persistent attacker access or break the OS unrecoverably. The
# trust list is shared with Write/Edit so one "Always allow" covers both surfaces.
# Catastrophic-path Bash gate. Bash is intentionally NOT in P_PATH_GATED_TOOLS (gating every `echo ... > /tmp/foo` would interrupt routine work), but a single redirected write to one of these can grant persistent attacker access or break the OS unrecoverably. The trust list is shared with Write/Edit so one "Always allow" covers both surfaces.
P_BASH_CATASTROPHIC_INFO: Dict[str, Tuple[str, str]] = {
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
"/etc/sudoers": ("Sudo permissions (/etc/sudoers)", "Controls which commands can run with admin privileges."),
@@ -73,14 +66,12 @@ P_BASH_CATASTROPHIC_INFO: Dict[str, Tuple[str, str]] = {
}
P_BASH_CATASTROPHIC_PATTERNS: Tuple[str, ...] = tuple(P_BASH_CATASTROPHIC_INFO.keys())
# Pulls quoted strings AND bare path-like tokens out of a Bash command. Intentionally loose:
# a false positive just means an extra approval prompt, never a missed gate.
# Pulls quoted strings AND bare path-like tokens out of a Bash command. Intentionally loose: a false positive just means an extra approval prompt, never a missed gate.
P_BASH_PATH_TOKEN_RE = re.compile(
r"""(?P<quoted>"[^"]+"|'[^']+')|(?P<bare>[~/.][\w./~\-]*)"""
)
# Write operators we care about; presence alone isn't enough, a sensitive target in the
# same command is also required. Covers shell redirection and tools with a destination flag.
# Write operators we care about; presence alone isn't enough, a sensitive target in the same command is also required. Covers shell redirection and tools with a destination flag.
P_BASH_WRITE_OP_RE = re.compile(
r"(?:>>?|\btee\b|\bsed\s+-i\b|\bcp\b|\bmv\b|\bdd\b[^|]*\bof=|\binstall\b|\bchmod\b|\bchown\b|\brm\b|\btouch\b|\bmkdir\b|\bln\b)",
re.IGNORECASE,
@@ -98,9 +89,7 @@ def match_sensitive_pattern(file_path: str) -> Optional[str]:
norm = os.path.normpath(os.path.expanduser(file_path))
except Exception:
return None
# Forward-slash the path so patterns match on Windows too; os.path.normpath emits
# backslashes there and fnmatch treats '/' as literal. Without this the gate would
# silently no-op on Windows and a prompt-injected Write to ~/.ssh/... would pass.
# Forward-slash the path so patterns match on Windows too; os.path.normpath emits backslashes there and fnmatch treats '/' as literal. Without this the gate would silently no-op on Windows and a prompt-injected Write to ~/.ssh/... would pass.
if os.sep != '/':
norm = norm.replace(os.sep, '/')
trusted = set(load_trusted_sensitive_paths())
@@ -160,10 +149,7 @@ def extract_target_path(tool_name: str, tool_input: object) -> str:
return str(tool_input.get("file_path") or "")
# Native-scheduler MCP tools that commit or mutate a recurring schedule. Always-on
# MCP servers fall through to the always_allow default, so these would otherwise fire
# silently; force them through ApprovalBar. The Cron* tools are Claude's own internal
# scheduler, denied outright in favour of the visible/auditable native one.
# Native-scheduler MCP tools that commit or mutate a recurring schedule. Always-on MCP servers fall through to the always_allow default, so these would otherwise fire silently; force them through ApprovalBar. The Cron* tools are Claude's own internal scheduler, denied outright in favour of the visible/auditable native one.
p_SCHEDULE_GATED = {
"mcp__openswarm-schedule__ScheduleWorkflow",
"mcp__openswarm-schedule__UpdateScheduledWorkflow",
@@ -184,9 +170,7 @@ def maybe_override_policy(policy: str, tool_name: str, tool_input: object) -> Tu
return "ask", None
if tool_name in CLAUDE_INTERNAL_SCHEDULER_TOOLS:
return "deny", None
# Committing or mutating a native recurring schedule is the in-app twin of the
# crontab gate above: real, user-visible, hard-to-undo, so it goes through
# ApprovalBar every time regardless of the always_allow default.
# Committing or mutating a native recurring schedule is the in-app twin of the crontab gate above: real, user-visible, hard-to-undo, so it goes through ApprovalBar every time regardless of the always_allow default.
if tool_name in p_SCHEDULE_GATED:
return "ask", None
if tool_name == "Bash" and isinstance(tool_input, dict):
backend/apps/agents/manager/permissions/workflow_approval.py
+2 -5
View File
@@ -29,8 +29,7 @@ class WorkflowApprovalMemory(BaseModel):
step_usage: Dict[str, Dict[str, bool]] # per-step record: step_id -> {tool: approved}
remember: Optional[Callable[[str, str], None]] # persist a workflow-level decision to disk
ask_timeout: float
# The executor bumps this as it advances steps so the gate can record which
# tools each step touched. None on test runs that don't thread it.
# The executor bumps this as it advances steps so the gate can record which tools each step touched. None on test runs that don't thread it.
current_step_id: Optional[str] = None
@@ -78,9 +77,7 @@ def is_claude_schedule_skill(tool_name: str, tool_input: object) -> bool:
@typechecked
def note_tool_used(session_id: str, tool_name: str, approved: bool) -> None:
# Record which tools each step touched (in-memory; the executor/test path
# persists step_usage once at run end). Captures every tool the gate sees so
# a step's tool set is complete, not only the ones that prompted.
# Record which tools each step touched (in-memory; the executor/test path persists step_usage once at run end). Captures every tool the gate sees so a step's tool set is complete, not only the ones that prompted.
mem = p_approval_memory.get(session_id)
if mem is None or mem.current_step_id is None:
return
backend/apps/agents/manager/prompt/attachments.py
+8 -43
View File
@@ -102,37 +102,15 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
native: List[dict] = []
refusals: List[str] = []
# The Claude Agent SDK speaks only Anthropic content-block shape.
# 9router 0.3.60 translates `image` blocks to the per-provider
# native shape; we trust that (the existing `images` param has
# shipped on every provider since v1.0.29).
# `document` (PDF) blocks: native on Anthropic upstream. For
# Gemini, anthropic-proxy rewrites document→image (keeping
# media_type=application/pdf), and Gemini's inline_data accepts
# that mime type natively. For OpenRouter, anthropic-proxy
# detects document blocks + injects the file-parser plugin. For
# OpenAI we refuse PDFs (no 9router translator path for the
# type:file shape, and Codex OAuth can't hit /v1/files anyway).
# The Claude Agent SDK speaks only Anthropic content-block shape. 9router 0.3.60 translates `image` blocks to the per-provider native shape; we trust that (the existing `images` param has shipped on every provider since v1.0.29). `document` (PDF) blocks: native on Anthropic upstream. For Gemini, anthropic-proxy rewrites document→image (keeping media_type=application/pdf), and Gemini's inline_data accepts that mime type natively. For OpenRouter, anthropic-proxy detects document blocks + injects the file-parser plugin. For OpenAI we refuse PDFs (no 9router translator path for the type:file shape, and Codex OAuth can't hit /v1/files anyway).
api = (api_type or "anthropic").lower()
supports_image = api in ("anthropic", "gemini", "openai", "openrouter", "gemini-cli")
# PDFs flow per provider:
# - Anthropic: native document blocks pass through cleanly.
# - Gemini: anthropic_proxy rewrites document → image_url with
# data:application/pdf base64; 9router translates to Gemini
# inlineData natively.
# - OpenRouter: file-parser plugin injected in anthropic-proxy.
# - OpenAI direct (GPT-5.x non-codex): anthropic_proxy detects
# document block + bypasses 9router entirely, translating
# to OpenAI Chat Completions and streaming response back
# via anthropic_to_openai.py. Requires openai_api_key.
# - Codex (cx/): models don't support PDFs.
# PDFs flow per provider: - Anthropic: native document blocks pass through cleanly. - Gemini: anthropic_proxy rewrites document → image_url with data:application/pdf base64; 9router translates to Gemini inlineData natively. - OpenRouter: file-parser plugin injected in anthropic-proxy. - OpenAI direct (GPT-5.x non-codex): anthropic_proxy detects document block + bypasses 9router entirely, translating to OpenAI Chat Completions and streaming response back via anthropic_to_openai.py. Requires openai_api_key. - Codex (cx/): models don't support PDFs.
supports_pdf = api in ("anthropic", "gemini", "gemini-cli", "openrouter", "openai")
if api == "openai" and isinstance(model, str) and ("codex" in model.lower() or model.lower().startswith("cx/")):
supports_pdf = False
# Per-file inline caps (raw bytes, before base64). Going over
# means the request would 4xx, blow our 64MB SDK buffer, or
# exceed the API's per-request cap on its own.
# Per-file inline caps (raw bytes, before base64). Going over means the request would 4xx, blow our 64MB SDK buffer, or exceed the API's per-request cap on its own.
if api == "anthropic":
per_file_cap = 24 * 1024 * 1024
total_request_cap = 28 * 1024 * 1024 # under Anthropic's 32MB
@@ -149,17 +127,10 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
per_file_cap = 0
total_request_cap = 0
# Running total of base64-expanded bytes already committed to the
# request. Anything that would push us over total_request_cap gets
# refused with concrete recovery actions.
# Running total of base64-expanded bytes already committed to the request. Anything that would push us over total_request_cap gets refused with concrete recovery actions.
b64_total = 0
# Combined char budget across inline TEXT attachments. Per-file 512K read
# cap doesn't stop a user dropping 20 huge txt files in one turn and
# silently blowing the context window. Whole-file or refuse: partial files
# confuse the model and the user can't tell what's missing. Sized to
# roughly fit 1M-window models (~375K tokens at 4 chars/token) while
# leaving room for prior conversation, the prompt, and tool turns.
# Combined char budget across inline TEXT attachments. Per-file 512K read cap doesn't stop a user dropping 20 huge txt files in one turn and silently blowing the context window. Whole-file or refuse: partial files confuse the model and the user can't tell what's missing. Sized to roughly fit 1M-window models (~375K tokens at 4 chars/token) while leaving room for prior conversation, the prompt, and tool turns.
text_total_chars = 0
text_total_cap = 1_500_000
@@ -207,8 +178,7 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
if kind == "pdf":
if not supports_pdf:
if api == "openai":
# Falls here only for Codex variants (gpt-5.3-codex etc.),
# which don't accept PDFs even though their family does.
# Falls here only for Codex variants (gpt-5.3-codex etc.), which don't accept PDFs even though their family does.
refusals.append(
f"[Attached PDF {os.path.basename(path)} ({size // 1024} KB) cannot be read on Codex models. "
"Switch to a non-Codex GPT-5 (e.g. gpt-5.5), Claude, Gemini 3.x, or "
@@ -288,10 +258,7 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
except Exception as e:
sections.append(f"[Context: {path}, error reading: {e}]")
# Anthropic prompt caching: tag the last document block as ephemeral
# so a follow-up turn referencing the same PDF stays cache-warm.
# Per Anthropic docs, only the trailing cache_control marker matters
# for cache prefix scope; earlier markers are ignored.
# Anthropic prompt caching: tag the last document block as ephemeral so a follow-up turn referencing the same PDF stays cache-warm. Per Anthropic docs, only the trailing cache_control marker matters for cache prefix scope; earlier markers are ignored.
if api == "anthropic" and native:
for blk in reversed(native):
if blk.get("type") == "document":
@@ -302,9 +269,7 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
return context_text, native, refusals
# Legacy entry point retained for any external caller; routes to the
# new attachment resolver with anthropic-default routing (no native
# blocks emitted, so behavior is the safe text-only old path).
# Legacy entry point retained for any external caller; routes to the new attachment resolver with anthropic-default routing (no native blocks emitted, so behavior is the safe text-only old path).
@typechecked
def resolve_context_paths(context_paths: Optional[List]) -> str:
text, p_native, refusals = resolve_attachments(context_paths, api_type="anthropic", model="")
backend/apps/agents/manager/prompt/compose_turn_system_prompt.py
+5 -15
View File
@@ -29,11 +29,7 @@ def compose_turn_system_prompt(
selected_app_output_ids: Optional[List[str]],
selected_setting_ids: Optional[List[str]],
) -> Optional[str]:
# MCP servers and their tool inventories are intentionally NOT injected into the system
# prompt: the CLI's deferred-tool pool already exposes them by name via ToolSearch, and
# eagerly listing connected MCPs (account emails, full tool enumerations) here would defeat
# the deferral and leak every integration into every turn. The model discovers MCPs only
# when it actively calls ToolSearch; only the gated registry summary goes in.
# MCP servers and their tool inventories are intentionally NOT injected into the system prompt: the CLI's deferred-tool pool already exposes them by name via ToolSearch, and eagerly listing connected MCPs (account emails, full tool enumerations) here would defeat the deferral and leak every integration into every turn. The model discovers MCPs only when it actively calls ToolSearch; only the gated registry summary goes in.
browser_ctx = build_browser_context(session.dashboard_id, selected_browser_ids=selected_browser_ids)
mcp_registry_ctx = build_mcp_registry_summary(session.allowed_tools, session.active_mcps, get_all_tool_names)
composed_prompt = compose_system_prompt(
@@ -44,8 +40,7 @@ def compose_turn_system_prompt(
mcp_registry_ctx,
)
# Pin the agent's notion of "now" to the host wall clock + zone so it can answer
# day-of-week questions without hallucinating.
# Pin the agent's notion of "now" to the host wall clock + zone so it can answer day-of-week questions without hallucinating.
try:
from zoneinfo import ZoneInfo
# Best-effort IANA name for the host. Mirrors apps/service/client.py.
@@ -71,22 +66,17 @@ def compose_turn_system_prompt(
pass
if session.mode == "view-builder":
# Read the LIVE skill content rather than a frozen-at-import constant. The skill is
# registered at ~/.claude/skills/app_builder_skill.md; user edits in the Skills page
# land there and propagate to the agent's prompt next turn without a restart.
# Read the LIVE skill content rather than a frozen-at-import constant. The skill is registered at ~/.claude/skills/app_builder_skill.md; user edits in the Skills page land there and propagate to the agent's prompt next turn without a restart.
from backend.apps.outputs.view_builder_templates import load_app_builder_skill
skill_block = f"<app_builder_reference>\n{load_app_builder_skill()}\n</app_builder_reference>"
composed_prompt = f"{composed_prompt}\n\n{skill_block}" if composed_prompt else skill_block
# App cards the user picked via the dashboard element picker: give the agent each app's
# on-disk path + meta + SKILL.md pointer so it can edit them in place (the dashboard card's
# runtime live-reloads). Additive and independent of view-builder mode above.
# App cards the user picked via the dashboard element picker: give the agent each app's on-disk path + meta + SKILL.md pointer so it can edit them in place (the dashboard card's runtime live-reloads). Additive and independent of view-builder mode above.
app_ctx = build_selected_app_context(selected_app_output_ids)
if app_ctx:
composed_prompt = f"{composed_prompt}\n\n{app_ctx}" if composed_prompt else app_ctx
# The user can point the agent at specific Settings rows. Targeting aid only; the settings
# tools are always on regardless.
# The user can point the agent at specific Settings rows. Targeting aid only; the settings tools are always on regardless.
settings_ctx = build_selected_settings_context(selected_setting_ids)
if settings_ctx:
composed_prompt = f"{composed_prompt}\n\n{settings_ctx}" if composed_prompt else settings_ctx
backend/apps/agents/manager/prompt/prompt_context.py
+5 -19
View File
@@ -20,11 +20,7 @@ def resolve_mode(mode_id: str, get_all_tool_names: Callable[[], List[str]]) -> T
return get_all_tool_names(), None, None
# A run of this many ToolSearch calls with no other tool between them is the
# "looping on ToolSearch" wedge: the model hunts for a gated MCP server's tools,
# which ToolSearch can never see, gets empty results, and retries. Two free
# calls (a power user with many activated MCPs may legitimately ToolSearch to
# load a deferred tool); redirect on the third.
# A run of this many ToolSearch calls with no other tool between them is the "looping on ToolSearch" wedge: the model hunts for a gated MCP server's tools, which ToolSearch can never see, gets empty results, and retries. Two free calls (a power user with many activated MCPs may legitimately ToolSearch to load a deferred tool); redirect on the third.
TOOLSEARCH_LOOP_THRESHOLD = 3
@@ -244,8 +240,7 @@ def build_mcp_registry_summary(allowed_tools: List[str], active_mcps: List[str],
server_name = sanitize_server_name(tool.name)
desc = (getattr(tool, "description", None) or "").strip()
if not desc:
# Fall back to a generic blurb keyed on the tool name so the
# model still has *some* signal to MCPSearch against.
# Fall back to a generic blurb keyed on the tool name so the model still has *some* signal to MCPSearch against.
desc = f"{tool.name} integration"
line = f"- `{server_name}`, {desc}"
if server_name in active_set:
@@ -256,10 +251,7 @@ def build_mcp_registry_summary(allowed_tools: List[str], active_mcps: List[str],
if not active_lines and not available_lines:
return None
# Static preamble first (kept byte-identical across users so it caches),
# then the per-session server list. Worked-example uses generic
# placeholders so a Pro Anthropic prompt-cache hit isn't broken by
# one user's connector names differing from another's.
# Static preamble first (kept byte-identical across users so it caches), then the per-session server list. Worked-example uses generic placeholders so a Pro Anthropic prompt-cache hit isn't broken by one user's connector names differing from another's.
sections = ["<mcp_servers>"]
sections.append(
"MCP servers are gated: their tools are uncallable until the user "
@@ -314,12 +306,7 @@ def build_mcp_registry_summary(allowed_tools: List[str], active_mcps: List[str],
return "\n".join(sections)
# The agent runs on the claude_code preset (kept for its tool scaffolding, safety
# rules, and the exclude_dynamic_sections prompt-cache win, which a raw-string
# system prompt would all throw away). The preset opens with "You are Claude Code,
# Anthropic's official CLI", which leaks into chat. This block is APPENDED after the
# preset, so being later it overrides that identity. Edit AGENT_NAME / AGENT_BLURB
# to rebrand. Kept short so it costs ~80 cached tokens, not a wall.
# The agent runs on the claude_code preset (kept for its tool scaffolding, safety rules, and the exclude_dynamic_sections prompt-cache win, which a raw-string system prompt would all throw away). The preset opens with "You are Claude Code, Anthropic's official CLI", which leaks into chat. This block is APPENDED after the preset, so being later it overrides that identity. Edit AGENT_NAME / AGENT_BLURB to rebrand. Kept short so it costs ~80 cached tokens, not a wall.
AGENT_NAME = "OpenSwarm"
AGENT_IDENTITY = (
f"# Who you are\n"
@@ -340,8 +327,7 @@ AGENT_IDENTITY = (
@typechecked
def compose_system_prompt(default_prompt: Optional[str], mode_prompt: Optional[str], session_prompt: Optional[str], browser_ctx: Optional[str] = None, mcp_registry_ctx: Optional[str] = None) -> Optional[str]:
# Identity always leads so it overrides the preset's Claude Code persona, even
# when the user has no custom default/mode/session prompt of their own.
# Identity always leads so it overrides the preset's Claude Code persona, even when the user has no custom default/mode/session prompt of their own.
parts = [AGENT_IDENTITY] + [p for p in (default_prompt, mode_prompt, session_prompt, mcp_registry_ctx, browser_ctx) if p]
return "\n\n".join(parts)
backend/apps/agents/manager/prompt/tool_catalog.py
+1 -4
View File
@@ -20,10 +20,7 @@ FULL_TOOLS = [
"CronCreate", "CronList", "CronDelete",
"InvokeAgent",
"Agent",
# ToolSearch is the loader the CLI uses to expose deferred tool schemas
# on demand. Must be in the allowedTools whitelist or the model can't
# call it, which means none of the deferred extended tools become
# reachable even when the CLI advertises them in the system prompt.
# ToolSearch is the loader the CLI uses to expose deferred tool schemas on demand. Must be in the allowedTools whitelist or the model can't call it, which means none of the deferred extended tools become reachable even when the CLI advertises them in the system prompt.
"ToolSearch",
]
backend/apps/agents/manager/register_builtin_mcp_servers.py
+3 -15
View File
@@ -34,10 +34,7 @@ def register_builtin_mcp_servers(
agents_dir, "browser_agent_mcp_server.py"
)
backend_port = os.environ.get("OPENSWARM_PORT", "8324")
# Only the card the user actually picked in select-mode gets claimed for the
# task, so the sub drives that one instead of opening its own duplicate. Passing
# EVERY dashboard card here (the old behavior) made the sub force-grab a random,
# usually-parked card and never navigate it, which broke the bulk of browser tasks.
# Only the card the user actually picked in select-mode gets claimed for the task, so the sub drives that one instead of opening its own duplicate. Passing EVERY dashboard card here (the old behavior) made the sub force-grab a random, usually-parked card and never navigate it, which broke the bulk of browser tasks.
pre_selected_bids = [b for b in (selected_browser_ids or []) if b]
auth_tok = get_auth_token()
mcp_servers["openswarm-browser-agent"] = {
@@ -77,11 +74,7 @@ def register_builtin_mcp_servers(
"type": "stdio",
}
# 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
# build_mcp_servers above) ensures the model cannot reach any
# other MCP server's tools without going through this layer first.
# 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 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"
)
@@ -96,12 +89,7 @@ def register_builtin_mcp_servers(
"type": "stdio",
}
# Always-on settings-meta server: SettingsRead / SettingsWrite let the
# agent read and edit its own OpenSwarm Settings autonomously. The
# backend (/api/settings-meta) enforces the only two guardrails: it
# can't disconnect the credential powering this run, and reads come
# back with secrets redacted. No activation gate, Settings is the
# agent's own house, not a third-party MCP.
# Always-on settings-meta server: SettingsRead / SettingsWrite let the agent read and edit its own OpenSwarm Settings autonomously. The backend (/api/settings-meta) enforces the only two guardrails: it can't disconnect the credential powering this run, and reads come back with secrets redacted. No activation gate, Settings is the agent's own house, not a third-party MCP.
settings_meta_server_path = os.path.join(
agents_dir, "settings_meta_server.py"
)
backend/apps/agents/manager/run/RunOptions.py
+14 -62
View File
@@ -38,8 +38,7 @@ from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtoco
class RunOptions(AgentManagerProtocol):
# No return annotation: the returned tuple carries an SDK ClaudeAgentOptions, which can't be
# module-imported here (mock-mode would fail to import the manager); it's lazy-imported below.
# No return annotation: the returned tuple carries an SDK ClaudeAgentOptions, which can't be module-imported here (mock-mode would fail to import the manager); it's lazy-imported below.
@typechecked
async def build_agent_options(self, session: AgentSession, session_id: str, prompt: str,
prompt_content: Union[str, List], builtin_perms: Dict[str, str],
@@ -69,11 +68,7 @@ class RunOptions(AgentManagerProtocol):
return await post_tool_hook_mod.post_tool_hook(hook_ctx, input_data, tool_use_id, context)
_, mode_sys_prompt, _ = resolve_mode(session.mode, get_all_tool_names)
# 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 build_mcp_servers silently filters it out.
# Emit a context_status event so the model and UI both know.
# 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 build_mcp_servers silently filters it out. Emit a context_status event so the model and UI both know.
try:
p_enabled = {
sanitize_server_name(t.name)
@@ -105,10 +100,7 @@ class RunOptions(AgentManagerProtocol):
set_framework_overhead(session, composed_prompt)
# 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 build_mcp_servers docstring).
# 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 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(
@@ -116,9 +108,7 @@ class RunOptions(AgentManagerProtocol):
)
# Register the DDG-backed openswarm-web MCP only when the primary has no reliable
# native Anthropic web path (decided in tools/web.py); p_m feeds the registration log
# + provider branch just below, so it stays a loop local.
# Register the DDG-backed openswarm-web MCP only when the primary has no reliable native Anthropic web path (decided in tools/web.py); p_m feeds the registration log + provider branch just below, so it stays a loop local.
p_m = p_router_model_id if isinstance(p_router_model_id, str) else ""
need_web_mcp = should_register_web_mcp(
model=session.model,
@@ -146,21 +136,12 @@ class RunOptions(AgentManagerProtocol):
if effective_disallowed:
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 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.
# `p_router_model_id` and `p_api_type_for_session` were resolved 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
api_type = p_api_type_for_session
session.provider = api_type
# Capture the Claude CLI's stderr into a buffer so the retry
# classifier can see the real cause of a process crash (e.g.
# "No pool capacity available" from the OpenSwarm proxy, or the
# Anthropic SDK's 429/overloaded error body). Without this the
# SDK's ProcessError only stringifies to "Command failed with
# exit code 1 / Check stderr output for details", which masks
# transient capacity issues.
# Capture the Claude CLI's stderr into a buffer so the retry classifier can see the real cause of a process crash (e.g. "No pool capacity available" from the OpenSwarm proxy, or the Anthropic SDK's 429/overloaded error body). Without this the SDK's ProcessError only stringifies to "Command failed with exit code 1 / Check stderr output for details", which masks transient capacity issues.
p_stderr_buffer: List[str] = []
def p_stderr_cb(line: str) -> None:
@@ -174,12 +155,7 @@ class RunOptions(AgentManagerProtocol):
options_kwargs = {
"model": resolved_model,
# 64 MB ceiling on the SDK <-> CLI JSON-RPC channel. The
# default 5 MB blocked any base64'd PDF over ~3.5 MB; we
# now route PDFs/images as native content blocks, which
# base64-expand by ~33%. 64 MB clears the largest single
# Anthropic PDF (32 MB raw) with headroom for prompt +
# tool results sharing the same frame.
# 64 MB ceiling on the SDK <-> CLI JSON-RPC channel. The default 5 MB blocked any base64'd PDF over ~3.5 MB; we now route PDFs/images as native content blocks, which base64-expand by ~33%. 64 MB clears the largest single Anthropic PDF (32 MB raw) with headroom for prompt + tool results sharing the same frame.
"max_buffer_size": 64 * 1024 * 1024,
"permission_mode": "default",
"can_use_tool": can_use_tool,
@@ -193,8 +169,7 @@ class RunOptions(AgentManagerProtocol):
"disallowed_tools": effective_disallowed,
"include_partial_messages": True,
}
# cc/cx/gc/ag/gemini/openrouter prefixes force 9Router; route="api"
# bypasses to the provider's host directly; otherwise Pro proxy or key.
# cc/cx/gc/ag/gemini/openrouter prefixes force 9Router; route="api" bypasses to the provider's host directly; otherwise Pro proxy or key.
await configure_provider_env(
options_kwargs, session, resolved_model, api_type, global_settings, []
)
@@ -202,16 +177,12 @@ class RunOptions(AgentManagerProtocol):
options_kwargs["mcp_servers"] = mcp_servers
mcp_json_len = len(json.dumps({"mcpServers": mcp_servers}))
logger.info(f"[MCP-DEBUG] mcp_servers passed to SDK: {list(mcp_servers.keys())}, JSON length={mcp_json_len}")
# claude_code preset for BOTH system_prompt and tools so the CLI's
# deferred-tools scaffolding survives. Raw string would replace it.
# claude_code preset for BOTH system_prompt and tools so the CLI's deferred-tools scaffolding survives. Raw string would replace it.
options_kwargs["tools"] = {
"type": "preset",
"preset": "claude_code",
}
# exclude_dynamic_sections=True moves cwd/git/OS grounding out of
# the cached prefix and into the first user message, unlocks
# Anthropic prompt cache (~80% input-token cut, 13-31% faster TTFT).
# Trade-off: grounding freezes at turn 1.
# exclude_dynamic_sections=True moves cwd/git/OS grounding out of the cached prefix and into the first user message, unlocks Anthropic prompt cache (~80% input-token cut, 13-31% faster TTFT). Trade-off: grounding freezes at turn 1.
if composed_prompt:
options_kwargs["system_prompt"] = {
"type": "preset",
@@ -228,34 +199,19 @@ class RunOptions(AgentManagerProtocol):
if session.max_turns:
options_kwargs["max_turns"] = session.max_turns
# The claude_code preset auto-attaches the user's claude.ai-
# connected partner MCPs (`mcp__claude_ai_*`). Those bypass our
# MCPActivate gate, don't share OAuth state with the OpenSwarm
# Gmail/Calendar/Drive connectors the user actually configured
# here, and confuse the model into picking the partner shim
# instead of our vetted server. Hard-block them at the SDK
# layer so the model can't even attempt the call.
# The claude_code preset auto-attaches the user's claude.ai- connected partner MCPs (`mcp__claude_ai_*`). Those bypass our MCPActivate gate, don't share OAuth state with the OpenSwarm Gmail/Calendar/Drive connectors the user actually configured here, and confuse the model into picking the partner shim instead of our vetted server. Hard-block them at the SDK layer so the model can't even attempt the call.
options_kwargs["disallowed_tools"] = [
"mcp__claude_ai_*",
]
if session.cwd:
# Pre-existing sessions may have workspaces that predate
# the git-init block in launch_agent, leaving them
# without a valid HEAD. Ensure it here so subagent
# worktree-add always works.
# Pre-existing sessions may have workspaces that predate the git-init block in launch_agent, leaving them without a valid HEAD. Ensure it here so subagent worktree-add always works.
ensure_cwd_git_repo(session.cwd)
options_kwargs["cwd"] = session.cwd
inject_thinking_options(options_kwargs, session, prompt, resolved_model, api_type)
# Fresh-restart path: some session changes must not reuse the
# CLI's resume transcript. MCPActivate needs a new transport so
# tool schemas are reread; branch edits/switches need the model
# to see only get_branch_messages(session), not facts from the
# old branch's SDK transcript. Soft restart: drop resume +
# sdk_session_id, replay local history via the prompt, let the
# SDK build a clean session from the current app state.
# Fresh-restart path: some session changes must not reuse the CLI's resume transcript. MCPActivate needs a new transport so tool schemas are reread; branch edits/switches need the model to see only get_branch_messages(session), not facts from the old branch's SDK transcript. Soft restart: drop resume + sdk_session_id, replay local history via the prompt, let the SDK build a clean session from the current app state.
if session.needs_fresh_session:
if session.sdk_session_id:
logger.info(
@@ -283,11 +239,7 @@ class RunOptions(AgentManagerProtocol):
elif isinstance(prompt_content, list):
prompt_content.insert(0, {"type": "text", "text": history})
# Compaction trigger (Phase 2). Driven by live ctx_used ratio
# rather than turn count, fires when input_tokens/context_window
# crosses session.compact_threshold_pct (default 0.65). Cheap,
# programmatic summarization (no aux LLM call) so this adds
# zero latency on the user's turn.
# Compaction trigger (Phase 2). Driven by live ctx_used ratio rather than turn count, fires when input_tokens/context_window crosses session.compact_threshold_pct (default 0.65). Cheap, programmatic summarization (no aux LLM call) so this adds zero latency on the user's turn.
await pre_send_context_guard(self, session, session_id)
logger.info(f"[MCP-DEBUG] Creating ClaudeAgentOptions short={session.model} resolved={resolved_model} api_type={api_type}")
backend/apps/agents/manager/run/TurnRunner.py
+8 -38
View File
@@ -26,8 +26,7 @@ from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtoco
class TurnRunner(AgentManagerProtocol):
# `options` is the SDK ClaudeAgentOptions, lazy-imported below (so mock-mode can import the
# manager without the SDK present), so it's left unannotated; everything else is typed.
# `options` is the SDK ClaudeAgentOptions, lazy-imported below (so mock-mode can import the manager without the SDK present), so it's left unannotated; everything else is typed.
@typechecked
async def run_turn_with_retry(self, session: AgentSession, session_id: str,
prompt_content: Union[str, List], options,
@@ -44,11 +43,7 @@ class TurnRunner(AgentManagerProtocol):
}
async def p_run_streaming_turn():
# Per-turn thinking aggregation trackers (added for the
# "Thought for Ns · M tokens" persisted label). Without
# nonlocal, the int reassignments at AssistantMessage emission
# below shadow them as locals and the dict access at
# content_block_start crashes with UnboundLocalError.
# Per-turn thinking aggregation trackers (added for the "Thought for Ns · M tokens" persisted label). Without nonlocal, the int reassignments at AssistantMessage emission below shadow them as locals and the dict access at content_block_start crashes with UnboundLocalError.
async for message in query(
prompt=prompt_stream(),
options=options,
@@ -57,19 +52,12 @@ class TurnRunner(AgentManagerProtocol):
turn.current_turn_emitted = False
else:
turn.current_turn_emitted = True
# Stamp the turn's wall-clock start at the FIRST
# non-Result message we see, this is when the
# user actually started waiting. We use the same
# timestamp as the basis for "Thought for Ns"
# so the duration covers thinking + tool exec
# + assistant text generation.
# Stamp the turn's wall-clock start at the FIRST non-Result message we see, this is when the user actually started waiting. We use the same timestamp as the basis for "Thought for Ns" so the duration covers thinking + tool exec + assistant text generation.
if turn.started_ts is None:
turn.started_ts = time.time()
# Snapshot cumulative tokens at turn start;
# subtracted at emit time for per-turn deltas.
# Snapshot cumulative tokens at turn start; subtracted at emit time for per-turn deltas.
try:
# Baselines track the SAME fresh lane the pill reads,
# so the per-turn delta is fresh-minus-fresh.
# Baselines track the SAME fresh lane the pill reads, so the per-turn delta is fresh-minus-fresh.
if isinstance(session.tokens, dict):
turn.baseline_session_in = int(session.tokens.get("input_fresh", 0) or 0)
turn.baseline_session_out = int(session.tokens.get("output", 0) or 0)
@@ -88,15 +76,7 @@ class TurnRunner(AgentManagerProtocol):
turn.baseline_captured = True
except Exception:
pass
# Pre-emit thinking pill for routes whose
# translator strips reasoning content (cx/, gc/,
# ag/, gemini/). Without this, the pill emits
# at turn end and lands BELOW the assistant
# text in session.messages, visually wrong.
# Pre-emitting here gives the pill the same
# ordering as Anthropic's natural streaming
# path. Updates in place at turn end via the
# stable thinking.msg_id dedupe.
# Pre-emit thinking pill for routes whose translator strips reasoning content (cx/, gc/, ag/, gemini/). Without this, the pill emits at turn end and lands BELOW the assistant text in session.messages, visually wrong. Pre-emitting here gives the pill the same ordering as Anthropic's natural streaming path. Updates in place at turn end via the stable thinking.msg_id dedupe.
try:
p_route_strips_reasoning_pre = (
isinstance(resolved_model, str)
@@ -137,10 +117,7 @@ class TurnRunner(AgentManagerProtocol):
await p_run_streaming_turn()
break
except Exception as e:
# Make sure the consolidated-thinking ticker doesn't
# outlive the turn on error/retry. Without this, an
# exception mid-stream leaves a dangling task that
# keeps re-emitting against a stale msg id.
# Make sure the consolidated-thinking ticker doesn't outlive the turn on error/retry. Without this, an exception mid-stream leaves a dangling task that keeps re-emitting against a stale msg id.
if thinking.ticker_task is not None and not thinking.ticker_task.done():
thinking.ticker_task.cancel()
try:
@@ -159,14 +136,7 @@ class TurnRunner(AgentManagerProtocol):
f"mid_stream={mid_stream}); sleeping {wait}s before retry. "
f"exc={e!r} stderr_tail={stderr_snapshot[-400:]!r}"
)
# Finalize any in-flight stream messages so the UI
# doesn't leave them pinned as "still streaming" while
# we wait and restart. On resume the CLI re-runs the
# last turn from scratch (Anthropic doesn't persist
# in-progress responses), so the partial assistant
# text / tool call we emitted is now orphaned, cap
# it with stream_end and start the fresh turn under a
# new message id.
# Finalize any in-flight stream messages so the UI doesn't leave them pinned as "still streaming" while we wait and restart. On resume the CLI re-runs the last turn from scratch (Anthropic doesn't persist in-progress responses), so the partial assistant text / tool call we emitted is now orphaned, cap it with stream_end and start the fresh turn under a new message id.
if turn.stream_text_msg_id:
await ws_manager.send_to_session(session_id, "agent:stream_end", {
"session_id": session_id,
backend/apps/agents/manager/run/handle_run_error.py
+9 -39
View File
@@ -31,25 +31,15 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
logger.exception(f"Agent {session_id} error: {e}")
session.status = "error"
# Long-context-required 429 fork: surface a friendly overflow event
# so the frontend can render an actionable card ("Switch to Chat
# mode" / "Start a fresh chat") instead of a raw error blob. The
# user can't recover by waiting, this is a tier-gate, not a rate
# limit, so the UX matters.
# Long-context-required 429 fork: surface a friendly overflow event so the frontend can render an actionable card ("Switch to Chat mode" / "Start a fresh chat") instead of a raw error blob. The user can't recover by waiting, this is a tier-gate, not a rate limit, so the UX matters.
try:
p_stderr_tail = "\n".join(p_stderr_buffer[-50:])
except Exception:
p_stderr_tail = ""
# If we already streamed a substantive assistant response this
# turn, the user got their answer; the error fired on a
# subsequent step (title gen, follow-up tool turn, etc.).
# Don't blast a "context exceeded" card over a completed reply.
# If we already streamed a substantive assistant response this turn, the user got their answer; the error fired on a subsequent step (title gen, follow-up tool turn, etc.). Don't blast a "context exceeded" card over a completed reply.
p_streamed_substantive = bool(turn.stream_text_msg_id) and turn.current_turn_emitted
if p_streamed_substantive and is_long_context_error(e, extra_text=p_stderr_tail):
# Mark the session completed (not error), keep the assistant
# reply visible, and skip the overflow card. The next user
# turn will properly hit the pre-send guard if the chat is
# still over cap.
# Mark the session completed (not error), keep the assistant reply visible, and skip the overflow card. The next user turn will properly hit the pre-send guard if the chat is still over cap.
session.status = "completed"
if turn.stream_text_msg_id:
try:
@@ -105,11 +95,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
except Exception:
logger.debug("submit_diagnostic for context_overflow failed", exc_info=True)
elif is_transient_capacity_error(e, extra_text=p_stderr_tail):
# A genuine throttle (429/overload/capacity) that already burned
# the whole silent-backoff budget (the only way one reaches here).
# It's a limit, not a failure, so don't append a system-message
# card; emit a transient signal for the muted pill and mark the
# turn completed so it doesn't read as an error.
# A genuine throttle (429/overload/capacity) that already burned the whole silent-backoff budget (the only way one reaches here). It's a limit, not a failure, so don't append a system-message card; emit a transient signal for the muted pill and mark the turn completed so it doesn't read as an error.
session.status = "completed"
if turn.stream_text_msg_id:
try:
@@ -124,8 +110,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
"retry_after_s": parse_retry_after(e, p_stderr_tail),
})
elif is_free_trial_exhausted(e, extra_text=p_stderr_tail):
# Free runs spent. Flip back to own_key and show a friendly
# "connect a model" upsell instead of a raw 402.
# Free runs spent. Flip back to own_key and show a friendly "connect a model" upsell instead of a raw 402.
try:
from backend.apps.subscription.free_trial import clear_free_trial
await clear_free_trial(load_settings())
@@ -147,11 +132,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
"message": error_msg.model_dump(mode="json"),
})
elif is_out_of_tokens(e, extra_text=p_stderr_tail):
# The user's PROVIDER account is out of credits / over quota, distinct from
# OpenSwarm free-trial exhaustion above and from a 401 below ("credit balance
# too low", "insufficient_quota", "usage cap exceeded", OpenSwarm plan limit).
# Show a friendly card with the provider's reset hint when it gave one, instead
# of dropping to the raw-error blob in the else branch.
# The user's PROVIDER account is out of credits / over quota, distinct from OpenSwarm free-trial exhaustion above and from a 401 below ("credit balance too low", "insufficient_quota", "usage cap exceeded", OpenSwarm plan limit). Show a friendly card with the provider's reset hint when it gave one, instead of dropping to the raw-error blob in the else branch.
p_reset_hint = extract_reset_hint(f"{e!s}\n{p_stderr_tail}")
friendly_msg = (
"Your model provider reports you're out of credits or over your usage "
@@ -172,19 +153,10 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
"message": error_msg.model_dump(mode="json"),
})
elif is_auth_error(e, extra_text=p_stderr_tail):
# Three sub-cases the user can hit, with distinct fixes:
# 1. "No credentials for provider: claude", user picked a
# -cc route but doesn't have Claude Pro/Max connected
# via 9Router. Tell them to either connect Claude
# Pro/Max OR pick a non--cc model.
# 2. OpenSwarm Pro 401, bearer expired. Reconnect.
# 3. Anthropic API key 401, wrong key. Re-enter.
# Three sub-cases the user can hit, with distinct fixes: 1. "No credentials for provider: claude", user picked a -cc route but doesn't have Claude Pro/Max connected via 9Router. Tell them to either connect Claude Pro/Max OR pick a non--cc model. 2. OpenSwarm Pro 401, bearer expired. Reconnect. 3. Anthropic API key 401, wrong key. Re-enter.
p_model = (session.model or "").lower()
p_combined = f"{e!s}\n{p_stderr_tail}".lower()
# Codex/OpenAI subscription tokens rotate every ~2-3
# minutes, the user sees the rotation window as a 401
# with "reset after 1m 59s" or similar. Don't ask them to
# reconnect; just tell them to wait it out and retry.
# Codex/OpenAI subscription tokens rotate every ~2-3 minutes, the user sees the rotation window as a 401 with "reset after 1m 59s" or similar. Don't ask them to reconnect; just tell them to wait it out and retry.
if (
("codex/" in p_combined or "[codex/" in p_combined or p_model.startswith(("cx/", "gpt-")))
and ("authentication token is expired" in p_combined or "authentication token has expired" in p_combined or "401" in p_combined)
@@ -238,9 +210,7 @@ async def handle_run_error(e: Exception, session: AgentSession, session_id: str,
"message": error_msg.model_dump(mode="json"),
})
elif is_unknown_model_error(e, extra_text=p_stderr_tail):
# Upstream rejected the model code itself (e.g. Codex 1211 on a
# ChatGPT plan that lacks our GPT ids). Track it; the friendly
# "add an API key / pick another model" card is rendered frontend-side.
# Upstream rejected the model code itself (e.g. Codex 1211 on a ChatGPT plan that lacks our GPT ids). Track it; the friendly "add an API key / pick another model" card is rendered frontend-side.
try:
from backend.apps.service.client import submit_diagnostic
submit_diagnostic({
backend/apps/agents/manager/run/run_options_helpers.py
+7 -25
View File
@@ -13,8 +13,7 @@ from backend.apps.agents.manager.session.history_compaction import estimate_post
logger = logging.getLogger(__name__)
# `manager` is the AgentManager; it isn't annotated because typing it would import agent_manager
# back into a module agent_manager already imports (a cycle). Same reason self is never annotated.
# `manager` is the AgentManager; it isn't annotated because typing it would import agent_manager back into a module agent_manager already imports (a cycle). Same reason self is never annotated.
@typechecked
async def pre_send_context_guard(manager, session: AgentSession, session_id: str) -> None:
try:
@@ -34,24 +33,15 @@ async def pre_send_context_guard(manager, session: AgentSession, session_id: str
except Exception:
logger.exception("compaction failed; proceeding without it")
# Pre-send hard guard (Phase 2). After compaction, if the
# session is still over context_soft_cap_pct of the window,
# LRU-trim oldest active_mcps. Stops the 429 from ever
# firing on predictable overflow paths.
# Pre-send hard guard (Phase 2). After compaction, if the session is still over context_soft_cap_pct of the window, LRU-trim oldest active_mcps. Stops the 429 from ever firing on predictable overflow paths.
try:
# Use the most recent measurement (the prior turn's
# input_tokens) as the estimate. Conservative because the
# current turn's user prompt + any new history adds on top
#, but the first turn of a fresh session has tokens=0 so
# we only act once we've seen real numbers.
# Use the most recent measurement (the prior turn's input_tokens) as the estimate. Conservative because the current turn's user prompt + any new history adds on top, but the first turn of a fresh session has tokens=0 so we only act once we've seen real numbers.
p_est_tokens = session.tokens.get("input", 0)
p_hard_cap = int(session.context_window * session.context_soft_cap_pct)
if p_est_tokens >= p_hard_cap:
trimmed: List[str] = []
while p_est_tokens >= p_hard_cap and len(session.active_mcps) > 1:
# Keep at least one MCP active so the model can
# finish whatever it was doing; trim from oldest
# which is FIFO order in the list.
# Keep at least one MCP active so the model can finish whatever it was doing; trim from oldest which is FIFO order in the list.
trimmed.append(f"mcp:{session.active_mcps.pop(0)}")
p_est_tokens -= 8_000 # rough per-MCP schema cost
if trimmed:
@@ -61,11 +51,7 @@ async def pre_send_context_guard(manager, session: AgentSession, session_id: str
"trimmed": trimmed,
"estimate_after": p_est_tokens,
})
# Surface a visible system breadcrumb in the chat so
# the user (and the model on the next turn) know
# which MCPs got dropped. Without this, the model
# may keep trying to call a now-missing tool and
# the user has no idea why.
# Surface a visible system breadcrumb in the chat so the user (and the model on the next turn) know which MCPs got dropped. Without this, the model may keep trying to call a now-missing tool and the user has no idea why.
try:
p_names = ", ".join(t.replace("mcp:", "") for t in trimmed)
p_trim_msg = Message(
@@ -84,10 +70,7 @@ async def pre_send_context_guard(manager, session: AgentSession, session_id: str
})
except Exception:
logger.exception("failed to emit MCP-trimmed breadcrumb")
# Trimming changes mcp_servers / outputs context →
# rebuild options. The cheapest correct path is
# to flag for fork on next turn via needs_fork
# and let the existing fork path handle it.
# Trimming changes mcp_servers / outputs context → rebuild options. The cheapest correct path is to flag for fork on next turn via needs_fork and let the existing fork path handle it.
session.needs_fork = True
except Exception:
logger.exception("pre-send token guard failed; proceeding")
@@ -117,8 +100,7 @@ def register_web_mcp_server(mcp_servers: Dict, p_m: str) -> None:
import sys
import backend.apps.agents as p_agents_pkg
web_mcp_server_path = os.path.join(os.path.dirname(p_agents_pkg.__file__), "web_mcp_server.py")
# Tell the MCP which primary the session is using so it
# can route to that provider's native search tool.
# Tell the MCP which primary the session is using so it can route to that provider's native search tool.
if p_m.startswith(("gc/", "gemini/", "ag/")):
p_primary_hint = "gemini"
elif p_m.startswith("cx/"):
backend/apps/agents/manager/run_browser_fast_path.py
+5 -14
View File
@@ -37,11 +37,7 @@ async def run_browser_fast_path(
p_fp_path = verdict
logger.info(f"[browser-fast-path] direct dispatch for session {session_id} ({verdict})")
text = ""
# The fast-path skips the orchestrator, so the UI never gets the BrowserAgent
# tool-call that draws the "Browser Agent" bubble. Emit a synthetic tool_call/
# tool_result pair (same shape + mcp__ name the orchestrator uses) so the bubble
# shows here too. None until we actually dispatch a browser (a pure READ answer
# has no browser, so no bubble).
# The fast-path skips the orchestrator, so the UI never gets the BrowserAgent tool-call that draws the "Browser Agent" bubble. Emit a synthetic tool_call/ tool_result pair (same shape + mcp__ name the orchestrator uses) so the bubble shows here too. None until we actually dispatch a browser (a pure READ answer has no browser, so no bubble).
p_browser_tool = "mcp__openswarm-browser-agent__CreateBrowserAgent"
p_bubble_tid: Optional[str] = None
try:
@@ -80,8 +76,7 @@ async def run_browser_fast_path(
return (str(r.get("summary") or "")).strip()
if not text:
# show the "Browser Agent" bubble during the dispatch (it renders as
# running, then completes when we emit the matching result below)
# show the "Browser Agent" bubble during the dispatch (it renders as running, then completes when we emit the matching result below)
p_bubble_tid = uuid4().hex
p_tc = Message(role="tool_call", branch_id=session.active_branch_id,
content={"id": p_bubble_tid, "tool": p_browser_tool, "input": {"task": prompt}})
@@ -91,8 +86,7 @@ async def run_browser_fast_path(
first = await p_dispatch(browser_fast_path.compose_task(prompt, brief))
text = p_summary(first)
if browser_fast_path.dispatch_failed(first):
# Retry only transient failures; a dead dashboard fails the
# retry identically, so skip it and tell the user instead.
# Retry only transient failures; a dead dashboard fails the retry identically, so skip it and tell the user instead.
if not ws_manager.global_connections:
p_fp_path += "+no-dashboard"
text = browser_fast_path.NO_DASHBOARD_REPLY
@@ -100,9 +94,7 @@ async def run_browser_fast_path(
from backend.apps.agents.browser import browser_batch_replay
payload = browser_batch_replay.send_payload_from_log(first.get("action_log"), prompt)
if payload:
# The dead attempt had already typed into a composer, so a
# blind retry risks a double-send: a read-only probe's
# verdict gates the retry in code, not prose.
# The dead attempt had already typed into a composer, so a blind retry risks a double-send: a read-only probe's verdict gates the retry in code, not prose.
logger.info(f"[browser-fast-path] send-zone failure for {session_id}; payload probe before any retry")
probe_text = p_summary(await p_dispatch(browser_fast_path.send_probe_task(prompt, payload)))
pv = browser_fast_path.probe_verdict(probe_text)
@@ -131,8 +123,7 @@ async def run_browser_fast_path(
f"[browser-fast-path] session {session_id} done: path={p_fp_path} "
f"reply={len(text)}ch in {int((time.monotonic() - p_fp_t0) * 1000)}ms"
)
# Close the synthetic bubble (always, even if the dispatch threw) so it never
# hangs as "running"; the bubble pairs this result with its call positionally.
# Close the synthetic bubble (always, even if the dispatch threw) so it never hangs as "running"; the bubble pairs this result with its call positionally.
if p_bubble_tid:
p_tr = Message(role="tool_result", branch_id=session.active_branch_id,
content={"tool_use_id": p_bubble_tid, "tool": p_browser_tool, "text": "done"})
backend/apps/agents/manager/session/SessionLifecycle.py
+1 -7
View File
@@ -206,13 +206,7 @@ class SessionLifecycle(AgentManagerProtocol):
def get_all_sessions(self, dashboard_id: Optional[str] = None) -> List[AgentSession]:
if not dashboard_id:
return list(self.sessions.values())
# Memory first, then promote on-disk sessions for this dashboard, but
# ONLY ones the dashboard's layout still has a card for. A session keeps
# its dashboard_id when its card is deleted, so promoting by tag alone
# resurrected deleted chats on every reopen; the layout's cards are the
# real source of truth for what's on the board. Imported sessions ARE in
# the layout, so they still surface, and this bounds the disk read to
# once per session per run, like resume_session.
# Memory first, then promote on-disk sessions for this dashboard, but ONLY ones the dashboard's layout still has a card for. A session keeps its dashboard_id when its card is deleted, so promoting by tag alone resurrected deleted chats on every reopen; the layout's cards are the real source of truth for what's on the board. Imported sessions ARE in the layout, so they still surface, and this bounds the disk read to once per session per run, like resume_session.
result = [s for s in self.sessions.values() if s.dashboard_id == dashboard_id]
seen = {s.id for s in result}
card_ids = self.p_dashboard_card_ids(dashboard_id)
backend/apps/agents/manager/session/SessionPersistence.py
+2 -5
View File
@@ -32,8 +32,7 @@ class SessionPersistence(AgentManagerProtocol):
data["status"] = "stopped"
dirty = True
logger.info(f"Marked stale session {sid} as stopped")
# Mode migration: Chat was merged into Ask. Rewrite mode="chat"
# so old sessions keep loading after the chat.json file is gone.
# Mode migration: Chat was merged into Ask. Rewrite mode="chat" so old sessions keep loading after the chat.json file is gone.
if data.get("mode") == "chat":
data["mode"] = "ask"
dirty = True
@@ -50,9 +49,7 @@ class SessionPersistence(AgentManagerProtocol):
for req in list(session.pending_approvals):
ws_manager.resolve_approval(req.id, {"behavior": "deny", "message": "Server shutting down"})
session.pending_approvals = []
# 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.
# 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.sync_session_close(session, close_reason="shutdown")
doc_data = session.model_dump(mode="json")
doc_data["search_text"] = self.build_search_text(session)
backend/apps/agents/manager/session/apply_context_window.py
+1 -2
View File
@@ -21,8 +21,7 @@ def apply_context_window(session: AgentSession, settings: Optional[AppSettings]
try:
from backend.apps.agents.providers.registry import get_context_window
if settings is None:
# Falling back to load_settings() inside the guard lets get_context_window
# still find a model-default cap when the settings file itself is unreadable.
# Falling back to load_settings() inside the guard lets get_context_window still find a model-default cap when the settings file itself is unreadable.
try:
settings = load_settings()
except Exception:
backend/apps/agents/manager/session/history_compaction.py
+2 -5
View File
@@ -9,9 +9,7 @@ from backend.config.paths import SESSIONS_DIR
logger = logging.getLogger(__name__)
# One plain-English trust line, fenced by a tag. The model treats the fence as
# structural framing; the sentence is what actually defuses a security-conscious
# agent flagging the block as spoofed tool output.
# One plain-English trust line, fenced by a tag. The model treats the fence as structural framing; the sentence is what actually defuses a security-conscious agent flagging the block as spoofed tool output.
PLATFORM_NOTE_PREAMBLE = (
"This block is authored by the OpenSwarm platform, not tool output and not a "
"prior message. It is trusted context."
@@ -21,8 +19,7 @@ PLATFORM_NOTE_CLOSE = "</openswarm_platform_note>"
SESSION_RECAP_OPEN = "<openswarm_session_recap>"
SESSION_RECAP_CLOSE = "</openswarm_session_recap>"
# Per-turn caps so the re-grounded recap stays compact (summaries, not replays)
# and cannot reinflate the context window from one giant tool input/output.
# Per-turn caps so the re-grounded recap stays compact (summaries, not replays) and cannot reinflate the context window from one giant tool input/output.
RECAP_TOOL_INPUT_CAP = 200
RECAP_TOOL_RESULT_CAP = 500
backend/apps/agents/manager/session/session_store.py
+1 -2
View File
@@ -8,8 +8,7 @@ from backend.config.json_store import read_json_or_none, atomic_write_json
@typechecked
def sessions_dir() -> str:
# Resolve live so test patches on either the paths module or the
# agent_manager facade re-export land on the same directory.
# Resolve live so test patches on either the paths module or the agent_manager facade re-export land on the same directory.
from backend.apps.agents import agent_manager
return agent_manager.SESSIONS_DIR
backend/apps/agents/manager/session/workspace_git.py
+3 -10
View File
@@ -36,11 +36,7 @@ def ensure_cwd_git_repo(cwd: str, home: Optional[str] = None) -> None:
return
import subprocess as sp_git
# Case A: cwd is inside some git repo (possibly parent). Verify
# HEAD resolves. If the enclosing repo is broken (e.g. a stray
# `.git` in $HOME with no commits, which makes workspaces
# under ~/.openswarm/workspaces/ inherit a broken HEAD), we
# need to init a fresh repo AT cwd so it shadows the parent.
# Case A: cwd is inside some git repo (possibly parent). Verify HEAD resolves. If the enclosing repo is broken (e.g. a stray `.git` in $HOME with no commits, which makes workspaces under ~/.openswarm/workspaces/ inherit a broken HEAD), we need to init a fresh repo AT cwd so it shadows the parent.
inside = sp_git.run(
["git", "rev-parse", "--is-inside-work-tree"],
cwd=cwd,
@@ -66,12 +62,9 @@ def ensure_cwd_git_repo(cwd: str, home: Optional[str] = None) -> None:
stdout=sp_git.DEVNULL, stderr=sp_git.DEVNULL, timeout=10,
)
return
# .git is in a parent dir (broken home-dir repo, etc.).
# Init our own repo at cwd so it shadows the broken parent.
# Fall through to Case B.
# .git is in a parent dir (broken home-dir repo, etc.). Init our own repo at cwd so it shadows the broken parent. Fall through to Case B.
# Case B: cwd is not a git repo at all (or parent is broken):
# init + empty commit here.
# Case B: cwd is not a git repo at all (or parent is broken): init + empty commit here.
sp_git.run(
["git", "init", "-q", "-b", "main"],
cwd=cwd,
backend/apps/agents/manager/streaming/HookContext.py
+1 -2
View File
@@ -19,8 +19,7 @@ class HookContext(BaseModel):
prompt: str
builtin_perms: Dict[str, str]
policy_defaults: Dict[str, str]
# The manager's LIVE session registry (InstanceOf keeps the reference, so a sub-agent
# the post hook spawns is visible to the manager; a plain Dict field pydantic would copy).
# The manager's LIVE session registry (InstanceOf keeps the reference, so a sub-agent the post hook spawns is visible to the manager; a plain Dict field pydantic would copy).
sessions: InstanceOf[dict]
# tool_use_id -> wall-clock start (s); pre records it, post pops it for elapsed_ms.
tool_start_times: Dict[str, float] = {}
backend/apps/agents/manager/streaming/handle_assistant_message.py
+8 -58
View File
@@ -37,23 +37,14 @@ async def handle_assistant_message(
content_parts = []
new_thinking_parts = []
tool_uses = []
# Capture the latest Gemini thoughtSignature
# (and Anthropic's signature_delta if present)
# off any ThinkingBlock in this message. We
# store it on the turn's consolidated thinking
# message so it survives session.json
# serialization, and re-attach it on the next
# request so Google's continuity check passes.
# Capture the latest Gemini thoughtSignature (and Anthropic's signature_delta if present) off any ThinkingBlock in this message. We store it on the turn's consolidated thinking message so it survives session.json serialization, and re-attach it on the next request so Google's continuity check passes.
new_thought_signature: Optional[str] = None
for block in message.content:
if isinstance(block, ThinkingBlock):
thinking_text = getattr(block, "thinking", None) or getattr(block, "text", None) or ""
if thinking_text:
new_thinking_parts.append(thinking_text)
# Try multiple field-name variants, SDK
# versions and 9Router translations have
# used `signature`, `thoughtSignature`,
# and `thought_signature` over time.
# Try multiple field-name variants, SDK versions and 9Router translations have used `signature`, `thoughtSignature`, and `thought_signature` over time.
sig = (
getattr(block, "signature", None)
or getattr(block, "thoughtSignature", None)
@@ -70,43 +61,13 @@ async def handle_assistant_message(
"input": block.input,
})
# Accumulate this AssistantMessage's contributions
# into the turn-level thinking pill. We re-emit
# the SAME message id each time so the frontend
# dedupes (addMessage replaces by id) and the
# bubble updates live as more thought / tools
# arrive. This is what gives us "Thought for 18s
# · 412 tokens · 3 tools used" reflecting the
# whole turn rather than just one think-step.
#
# NOTE: tool count is incremented in the
# content_block_start (block_type=="tool_use")
# branch above, NOT here. That path fires for
# both Anthropic and 9Router-translated
# providers; counting again here would double.
# If a provider somehow doesn't surface
# content_block_start for tool blocks but DOES
# surface them in the AssistantMessage envelope
# (defensive case), the max() in the
# consolidated emit will still pick up the
# higher count.
# Accumulate this AssistantMessage's contributions into the turn-level thinking pill. We re-emit the SAME message id each time so the frontend dedupes (addMessage replaces by id) and the bubble updates live as more thought / tools arrive. This is what gives us "Thought for 18s · 412 tokens · 3 tools used" reflecting the whole turn rather than just one think-step. NOTE: tool count is incremented in the content_block_start (block_type=="tool_use") branch above, NOT here. That path fires for both Anthropic and 9Router-translated providers; counting again here would double. If a provider somehow doesn't surface content_block_start for tool blocks but DOES surface them in the AssistantMessage envelope (defensive case), the max() in the consolidated emit will still pick up the higher count.
if new_thinking_parts:
thinking.text_parts.extend(new_thinking_parts)
# Latch the most recent thoughtSignature, Gemini
# only validates against the LATEST one in the
# conversation history, so older signatures from
# earlier think-steps in the same turn are
# superseded by newer ones.
# Latch the most recent thoughtSignature, Gemini only validates against the LATEST one in the conversation history, so older signatures from earlier think-steps in the same turn are superseded by newer ones.
if new_thought_signature:
thinking.thought_signature = new_thought_signature
# Accumulate this message's total output tokens
# (SDK populates `usage.output_tokens` with the
# full output for the inference: thinking text +
# visible text + tool-call JSON args). Summing
# across the turn's AssistantMessages gives us
# "all output the model produced this turn,"
# which is what users intuit when they see a
# token count.
# Accumulate this message's total output tokens (SDK populates `usage.output_tokens` with the full output for the inference: thinking text + visible text + tool-call JSON args). Summing across the turn's AssistantMessages gives us "all output the model produced this turn," which is what users intuit when they see a token count.
try:
msg_usage = getattr(message, "usage", None) or {}
if isinstance(msg_usage, dict):
@@ -116,27 +77,16 @@ async def handle_assistant_message(
except Exception:
pass
# Re-emit the consolidated thinking message on
# every AssistantMessage (event-driven). The
# background ticker loop keeps it updating
# between events too, so the elapsed counter
# ticks even during tool execution / slow text
# generation gaps.
# Re-emit the consolidated thinking message on every AssistantMessage (event-driven). The background ticker loop keeps it updating between events too, so the elapsed counter ticks even during tool execution / slow text generation gaps.
if thinking.text_parts:
await thinking_mod.emit_consolidated_thinking(thinking, turn, session, session_id, sessions)
# Start the 1Hz ticker once we have a
# consolidated message in flight so the
# bubble keeps updating between SDK events.
# Start the 1Hz ticker once we have a consolidated message in flight so the bubble keeps updating between SDK events.
if thinking.ticker_task is None or thinking.ticker_task.done():
thinking.ticker_task = asyncio.create_task(thinking_mod.ticker_loop(thinking, turn, session, session_id, sessions))
if content_parts:
asst_text = "\n".join(content_parts)
# 9Router sometimes returns upstream 401s as
# the assistant reply (no SDK exception), so
# the catch-all auth handler never fires.
# Match the text pattern and surface a
# friendly system bubble instead.
# 9Router sometimes returns upstream 401s as the assistant reply (no SDK exception), so the catch-all auth handler never fires. Match the text pattern and surface a friendly system bubble instead.
lower_text = asst_text.lower()
looks_like_router_auth_error = (
("failed to authenticate" in lower_text and "401" in lower_text)
backend/apps/agents/manager/streaming/handle_result_message.py
+10 -65
View File
@@ -35,37 +35,18 @@ async def handle_result_message(
api_type: Optional[str],
global_settings: object,
) -> None:
# ResultMessage carries the AUTHORITATIVE per-turn
# output_tokens count. Some providers (notably
# OpenAI/Gemini through 9Router) only populate
# `usage.output_tokens` here, not on individual
# AssistantMessages. Fold this into the running
# turn aggregate BEFORE emitting the final
# consolidated thinking message, so the bubble's
# tokens segment reflects ground truth on those
# providers too.
# ResultMessage carries the AUTHORITATIVE per-turn output_tokens count. Some providers (notably OpenAI/Gemini through 9Router) only populate `usage.output_tokens` here, not on individual AssistantMessages. Fold this into the running turn aggregate BEFORE emitting the final consolidated thinking message, so the bubble's tokens segment reflects ground truth on those providers too.
try:
result_usage = getattr(message, "usage", None) or {}
if isinstance(result_usage, dict):
result_out = int(result_usage.get("output_tokens", 0) or 0)
# Take the max, if individual
# AssistantMessages already summed to a
# larger number we trust that; otherwise
# ResultMessage's count fills the gap.
# Take the max, if individual AssistantMessages already summed to a larger number we trust that; otherwise ResultMessage's count fills the gap.
if result_out > turn.output_tokens:
turn.output_tokens = result_out
except Exception:
pass
# Pre-populate session.tokens BEFORE emitting the
# final consolidated thinking pill. Order matters:
# emit_consolidated_thinking reads
# session.tokens["input"]/["output"] for the
# combined-total stamp on the pill. If we emit
# first, the pill freezes with input=0 because
# the ResultMessage hasn't been consumed yet
# (the writes below at line ~2918 wouldn't
# land until after the pill is already broadcast).
# Pre-populate session.tokens BEFORE emitting the final consolidated thinking pill. Order matters: emit_consolidated_thinking reads session.tokens["input"]/["output"] for the combined-total stamp on the pill. If we emit first, the pill freezes with input=0 because the ResultMessage hasn't been consumed yet (the writes below at line ~2918 wouldn't land until after the pill is already broadcast).
try:
pre_usage = getattr(message, "usage", None) or {}
if isinstance(pre_usage, dict):
@@ -76,28 +57,14 @@ async def handle_result_message(
pre_out = int(pre_usage.get("output_tokens", 0) or 0)
if pre_total_in > 0:
session.tokens["input"] = pre_total_in
# Pill reads the fresh lane: uncached input only,
# so re-read/cached context doesn't inflate it.
# Pill reads the fresh lane: uncached input only, so re-read/cached context doesn't inflate it.
session.tokens["input_fresh"] = pre_in
if pre_out > 0:
session.tokens["output"] = pre_out
except Exception:
pass
# Final consolidated emission with the full
# duration + authoritative tokens. The frontend
# bubble freezes on this final value.
# For routes whose translator strips reasoning
# content (cx/ for OpenAI, gc/ for Gemini),
# force-emit a pill even when no text or upstream
# token count was captured. Without this, GPT/
# Gemini turns show no thinking bubble at all
# because 9Router's translator doesn't carry
# reasoning_content across the Anthropic-shape
# round-trip. The frontend's ThinkingBubble
# detects empty content and renders a friendly
# "provider doesn't expose reasoning text"
# explanation instead of a blank panel.
# Final consolidated emission with the full duration + authoritative tokens. The frontend bubble freezes on this final value. For routes whose translator strips reasoning content (cx/ for OpenAI, gc/ for Gemini), force-emit a pill even when no text or upstream token count was captured. Without this, GPT/ Gemini turns show no thinking bubble at all because 9Router's translator doesn't carry reasoning_content across the Anthropic-shape round-trip. The frontend's ThinkingBubble detects empty content and renders a friendly "provider doesn't expose reasoning text" explanation instead of a blank panel.
route_strips_reasoning = (
isinstance(resolved_model, str)
and resolved_model.startswith(("cx/", "gc/", "ag/", "gemini/"))
@@ -135,8 +102,7 @@ async def handle_result_message(
thinking.block_starts = {}
session.sdk_session_id = getattr(message, "session_id", None)
# Pull usage first; SDK's total_cost_usd is wrong for OR
# (assumes Anthropic rates) and we recompute below.
# Pull usage first; SDK's total_cost_usd is wrong for OR (assumes Anthropic rates) and we recompute below.
usage = getattr(message, "usage", None) or {}
inp = out = cache_create = cache_read = total_input = 0
if isinstance(usage, dict):
@@ -158,13 +124,7 @@ async def handle_result_message(
elif resolved_model.startswith("openrouter/") and ":free" in resolved_model:
free_route = True
elif resolved_model.startswith("cp-"):
# User-configured custom OpenAI-compatible
# provider (Ollama Cloud, Together, Groq,
# local LMs, etc.). Pricing is unknowable
# without per-provider rate tables that
# would rot fast, zero out instead of
# showing the SDK's Anthropic-rate
# estimate, which is meaningless here.
# User-configured custom OpenAI-compatible provider (Ollama Cloud, Together, Groq, local LMs, etc.). Pricing is unknowable without per-provider rate tables that would rot fast, zero out instead of showing the SDK's Anthropic-rate estimate, which is meaningless here.
free_route = True
if api_type == "anthropic":
from backend.apps.settings.credentials import proxy_auth as proxy_auth
@@ -191,13 +151,7 @@ async def handle_result_message(
or resolved_model.startswith("cp-gemini/")
or resolved_model.startswith("cp-google/"))
):
# Direct OpenAI/Gemini API key lane. SDK's
# total_cost_usd is computed at Anthropic
# rates (Opus pricing), for GPT-5.4-Mini
# at $0.25/M input that's a 60x overcount
# ($30 instead of $0.04 per Mehmet-style
# 4-PDF turn). Use the published per-model
# rates instead.
# Direct OpenAI/Gemini API key lane. SDK's total_cost_usd is computed at Anthropic rates (Opus pricing), for GPT-5.4-Mini at $0.25/M input that's a 60x overcount ($30 instead of $0.04 per Mehmet-style 4-PDF turn). Use the published per-model rates instead.
from backend.apps.agents.providers.registry import get_direct_pricing
pricing = get_direct_pricing(resolved_model) or get_direct_pricing(session.model)
if pricing:
@@ -207,9 +161,7 @@ async def handle_result_message(
+ out * out_rate
) / 1_000_000
else:
# Unknown model in this family: zero out
# rather than ship an Anthropic-rate
# estimate that's wildly wrong.
# Unknown model in this family: zero out rather than ship an Anthropic-rate estimate that's wildly wrong.
cost = 0.0
session.cost_usd = cost
@@ -219,14 +171,7 @@ async def handle_result_message(
})
if isinstance(usage, dict):
# Per-turn context-usage broadcast. Drives the UI
# status pill and the auto-compact threshold. The
# denominator is the session's real model cap,
# populated from registry.get_context_window at
# session creation, restore, and model-switch
# (see apply_context_window). max(1, ...) is a
# belt-and-braces guard against zero/None drift
# from any future restore-from-disk corner case.
# Per-turn context-usage broadcast. Drives the UI status pill and the auto-compact threshold. The denominator is the session's real model cap, populated from registry.get_context_window at session creation, restore, and model-switch (see apply_context_window). max(1, ...) is a belt-and-braces guard against zero/None drift from any future restore-from-disk corner case.
ctx_window = max(1, getattr(session, "context_window", 0) or 200_000)
ctx_used_pct = round(total_input / ctx_window, 4) if total_input else 0.0
cache_read_pct = round(cache_read / total_input, 4) if total_input else 0.0
backend/apps/agents/manager/streaming/handle_stream_event.py
+6 -26
View File
@@ -34,10 +34,7 @@ async def handle_stream_event(
event_type = event.get("type")
if event_type == "content_block_start":
# Stamp the first stream event of the session
# so the session list can show "first response
# at HH:MM" on reload. Only the first turn
# sets this; later turns leave it untouched.
# Stamp the first stream event of the session so the session list can show "first response at HH:MM" on reload. Only the first turn sets this; later turns leave it untouched.
if session.first_response_at is None:
session.first_response_at = datetime.now()
@@ -56,19 +53,10 @@ async def handle_stream_event(
turn.stream_block_index_map[index] = turn.stream_text_msg_id
elif block_type == "thinking":
# Reasoning trace from thinking-capable models
# (GPT-5.3 Codex, Gemini 3 Pro/Flash, Claude
# with extended thinking). Rendered as a
# collapsible "thinking" message in the UI via
# the existing stream infrastructure, the
# frontend already handles role="thinking" for
# the DynamicIsland/agent card rendering.
# Reasoning trace from thinking-capable models (GPT-5.3 Codex, Gemini 3 Pro/Flash, Claude with extended thinking). Rendered as a collapsible "thinking" message in the UI via the existing stream infrastructure, the frontend already handles role="thinking" for the DynamicIsland/agent card rendering.
thinking_msg_id = uuid4().hex
turn.stream_block_index_map[index] = thinking_msg_id
# Server-stamp start so we can accumulate
# per-turn elapsed_ms across multiple
# thinking blocks (think → tool → think
# → answer turns sum correctly).
# Server-stamp start so we can accumulate per-turn elapsed_ms across multiple thinking blocks (think → tool → think → answer turns sum correctly).
thinking.block_starts[index] = time.time()
await ws_manager.send_to_session(session_id, "agent:stream_start", {
"session_id": session_id,
@@ -80,9 +68,7 @@ async def handle_stream_event(
tool_msg_id = uuid4().hex
turn.stream_tool_msg_ids_ordered.append(tool_msg_id)
turn.stream_block_index_map[index] = tool_msg_id
# Stream-level tool count for the thinking pill. OpenAI/Gemini-through-9Router
# AssistantMessage envelopes are sometimes incomplete, so this stream count guarantees
# "N tools used" renders cross-provider; the AssistantMessage path dedupes against it.
# Stream-level tool count for the thinking pill. OpenAI/Gemini-through-9Router AssistantMessage envelopes are sometimes incomplete, so this stream count guarantees "N tools used" renders cross-provider; the AssistantMessage path dedupes against it.
turn.tool_count += 1
await ws_manager.send_to_session(session_id, "agent:stream_start", {
"session_id": session_id,
@@ -112,8 +98,7 @@ async def handle_stream_event(
"delta": text_chunk,
})
elif msg_id and delta_type == "thinking_delta":
# Thinking content streams as thinking_delta
# with a "thinking" field (not "text")
# Thinking content streams as thinking_delta with a "thinking" field (not "text")
think_chunk = delta.get("thinking", "")
await ws_manager.send_to_session(session_id, "agent:stream_delta", {
"session_id": session_id,
@@ -132,12 +117,7 @@ async def handle_stream_event(
elif event_type == "content_block_stop":
index = event.get("index")
msg_id = turn.stream_block_index_map.get(index)
# If this was a thinking block, accumulate
# elapsed_ms server-side. We don't include
# per-block elapsed/tokens on the WS event
#, the pill stays in "Thinking…" until the
# AssistantMessage lands carrying the per-turn
# aggregate values.
# If this was a thinking block, accumulate elapsed_ms server-side. We don't include per-block elapsed/tokens on the WS event, the pill stays in "Thinking…" until the AssistantMessage lands carrying the per-turn aggregate values.
if index in thinking.block_starts:
thinking.total_ms += int(
(time.time() - thinking.block_starts.pop(index)) * 1000
backend/apps/agents/manager/streaming/post_tool_hook.py
+5 -29
View File
@@ -39,9 +39,7 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex
raw_response = input_data.get("tool_response", "")
# Accumulate per-tool latency on the session. Lets the cloud aggregate a
# tool-latency distribution into the existing daily.summary without firing
# per-tool events.
# Accumulate per-tool latency on the session. Lets the cloud aggregate a tool-latency distribution into the existing daily.summary without firing per-tool events.
hook_tool_name_early = input_data.get("tool_name", "")
if hook_tool_name_early and elapsed_ms is not None and elapsed_ms >= 0:
latencies = getattr(session, "tool_latencies", None)
@@ -152,8 +150,7 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex
usage = raw_response.get("usage", {})
if isinstance(usage, dict):
sub_tokens["input"] = usage.get("input_tokens", 0) + usage.get("cache_creation_input_tokens", 0) + usage.get("cache_read_input_tokens", 0)
# Pill-only lane: NEW (uncached) input, excludes the cached
# static prefix so the bubble shows what this turn added.
# Pill-only lane: NEW (uncached) input, excludes the cached static prefix so the bubble shows what this turn added.
sub_tokens["input_fresh"] = usage.get("input_tokens", 0)
sub_tokens["output"] = usage.get("output_tokens", 0)
if raw_response.get("total_cost_usd"):
@@ -163,21 +160,7 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex
sub_session_id = uuid4().hex
sub_name = agent_prompt[:50] if agent_prompt else "Sub-agent"
# Subagent context isolation invariant (Phase 3, Layer P):
# children DO NOT inherit the parent's active_mcps or
# compaction state. They start with the AgentSession
# defaults (empty lists). Reasoning:
# - Security: a parent that activated Gmail shouldn't
# leak Gmail tools to a subagent doing an unrelated
# task. The user only approved Gmail for the parent.
# - Token cost: subagents typically have a narrow task,
# they don't need the parent's full activated set.
# - Failure isolation: if the parent compacted history,
# the subagent shouldn't inherit a summary it can't
# re-expand.
# If a subagent ever needs a parent activation, the user
# must approve it explicitly via MCPActivate inside the
# subagent session, same gate as a fresh top-level chat.
# Subagent context isolation invariant (Phase 3, Layer P): children DO NOT inherit the parent's active_mcps or compaction state. They start with the AgentSession defaults (empty lists). Reasoning: - Security: a parent that activated Gmail shouldn't leak Gmail tools to a subagent doing an unrelated task. The user only approved Gmail for the parent. - Token cost: subagents typically have a narrow task, they don't need the parent's full activated set. - Failure isolation: if the parent compacted history, the subagent shouldn't inherit a summary it can't re-expand. If a subagent ever needs a parent activation, the user must approve it explicitly via MCPActivate inside the subagent session, same gate as a fresh top-level chat.
sub_session = AgentSession(
id=sub_session_id,
name=sub_name,
@@ -194,9 +177,7 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex
],
dashboard_id=session.dashboard_id,
parent_session_id=session_id,
# Explicit empty list (matches the model default) so
# the invariant is visible at the spawn site rather
# than relying on the field's default_factory.
# Explicit empty list (matches the model default) so the invariant is visible at the spawn site rather than relying on the field's default_factory.
active_mcps=[],
)
apply_context_window(sub_session)
@@ -209,12 +190,7 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex
result_payload["sub_session_id"] = sub_session_id
result_msg = Message(role="tool_result", content=result_payload, branch_id=session.active_branch_id)
# Spill oversized tool results to per-session disk storage.
# The replacement keeps the first 4KB inline so the model
# retains some signal; the rest lives on disk for the UI to
# surface in the compaction drawer. Crucially this happens
# at *write* time (before the next turn ships history to the
# SDK) so the bloat never re-enters context.
# Spill oversized tool results to per-session disk storage. The replacement keeps the first 4KB inline so the model retains some signal; the rest lives on disk for the UI to surface in the compaction drawer. Crucially this happens at *write* time (before the next turn ships history to the SDK) so the bloat never re-enters context.
try:
truncated_content, blob_path = truncate_large_tool_result(
result_msg.content, session.id, result_msg.id
backend/apps/agents/manager/streaming/state.py
+1 -2
View File
@@ -48,8 +48,7 @@ class TurnState(BaseModel):
output_tokens: int = 0
assistant_text_chars: int = 0
tool_input_chars: int = 0
# Cumulative-token snapshot taken at turn start; subtracted at emit time so the thinking
# pill shows THIS turn's new tokens, not the whole session's running total.
# Cumulative-token snapshot taken at turn start; subtracted at emit time so the thinking pill shows THIS turn's new tokens, not the whole session's running total.
baseline_session_in: int = 0
baseline_session_out: int = 0
baseline_children_in: int = 0
backend/apps/agents/manager/streaming/thinking.py
+8 -69
View File
@@ -38,13 +38,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
token count.
"""
upstream_reasoning_tokens: Optional[int] = None
# Probe 9Router for the upstream reasoning-token count
# whenever (a) there's no in-process text, OR (b) the
# caller flagged this as a force-emit for a route that
# strips reasoning. Case (b) is what makes the FINAL
# emit on GPT/Gemini show the real reasoning count
# (e.g. 196) instead of the heuristic chars/3.6 of the
# answer text (e.g. 13).
# Probe 9Router for the upstream reasoning-token count whenever (a) there's no in-process text, OR (b) the caller flagged this as a force-emit for a route that strips reasoning. Case (b) is what makes the FINAL emit on GPT/Gemini show the real reasoning count (e.g. 196) instead of the heuristic chars/3.6 of the answer text (e.g. 13).
if not thinking.text_parts or force_provider_unavailable:
try:
from backend.apps.nine_router import (
@@ -62,22 +56,10 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
and upstream_reasoning_tokens is None
and not force_provider_unavailable
):
# No text, no upstream signal, and caller didn't
# ask for the unavailable-pill, nothing to show.
# No text, no upstream signal, and caller didn't ask for the unavailable-pill, nothing to show.
return
joined_text = "\n".join(thinking.text_parts)
# Total turn output token estimate. Combines two sources:
# - SDK usage.output_tokens summed across completed
# AssistantMessages (authoritative for finished
# blocks).
# - chars/3.6 heuristic over the running streams of
# thinking + assistant-text + tool-input JSON
# (covers in-flight blocks the SDK hasn't billed
# yet, i.e. the answer the user is currently
# reading).
# Take the max so the number doesn't visually shrink as
# the SDK's authoritative count overtakes our running
# heuristic.
# Total turn output token estimate. Combines two sources: - SDK usage.output_tokens summed across completed AssistantMessages (authoritative for finished blocks). - chars/3.6 heuristic over the running streams of thinking + assistant-text + tool-input JSON (covers in-flight blocks the SDK hasn't billed yet, i.e. the answer the user is currently reading). Take the max so the number doesn't visually shrink as the SDK's authoritative count overtakes our running heuristic.
running_chars = (
len(joined_text)
+ turn.assistant_text_chars
@@ -85,11 +67,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
)
heuristic_tokens = max(1, round(running_chars / 3.6)) if running_chars else 0
turn_tokens: Optional[int] = None
# Priority order:
# 1. Upstream reasoning-token count from 9Router (the
# only honest signal for GPT/Gemini, captured above).
# 2. SDK-reported usage.output_tokens (Anthropic).
# 3. chars/3.6 heuristic over running streams (live UI).
# Priority order: 1. Upstream reasoning-token count from 9Router (the only honest signal for GPT/Gemini, captured above). 2. SDK-reported usage.output_tokens (Anthropic). 3. chars/3.6 heuristic over running streams (live UI).
if upstream_reasoning_tokens and upstream_reasoning_tokens > 0:
turn_tokens = upstream_reasoning_tokens
elif turn.output_tokens > 0 or heuristic_tokens > 0:
@@ -108,13 +86,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
pass
if turn.started_ts is not None:
turn.total_ms = int((time.time() - turn.started_ts) * 1000)
# Accumulate into session-level "agent active time" and
# the per-model breakdown so a session that spans
# multiple turns reports the total wall-clock time the
# agent was running. Per-model bucket uses the model
# active *now* (model can be switched mid-turn but the
# current value is the right attribution for the work
# just produced).
# Accumulate into session-level "agent active time" and the per-model breakdown so a session that spans multiple turns reports the total wall-clock time the agent was running. Per-model bucket uses the model active *now* (model can be switched mid-turn but the current value is the right attribution for the work just produced).
try:
session.agent_active_ms = int(getattr(session, "agent_active_ms", 0) or 0) + turn.total_ms
m = session.model or "unknown"
@@ -123,34 +95,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
pass
if thinking.msg_id is None:
thinking.msg_id = uuid4().hex
# Combined token total for the pill, input + output for
# the parent turn PLUS any work delegated to subagents
# (browser agents, invoke-agent forks) and tool MCP
# servers that produced their own usage on this turn.
# The user-visible answer to "how big is this turn" is
# the all-in sum, not just the primary's output. We sum
# every reachable source:
# - parent's input (session.tokens["input"],
# ResultMessage.usage at line ~2886)
# - parent's output (session.tokens["output"], same
# ResultMessage)
# - every direct sub-session whose parent_session_id
# points at this session (browser agents, sub-agent
# forks, invoke-agent calls book their own usage at
# subprocess return time, agent_manager.py:1365 +
# browser_agent.py:1000-1001)
# This mirrors how billing accumulates per-turn, caches,
# tool MCP servers that talk to LLMs (e.g. summarizers),
# and subagent reasoning all show up under the parent's
# "session.tokens" once their result lands.
# Read cumulative session totals + cumulative subagent
# totals at this moment, then subtract the turn-start
# baseline to get THIS TURN'S delta. Without subtracting,
# the second turn's pill would show turn-1 work added
# to turn-2 work, the third would show all three, etc.
# Pill uses the FRESH lane (uncached input only). session.tokens
# ["input"] stays full for the context-fullness bar + cost; the
# bubble shows the NEW tokens this turn, not the cached re-reads.
# Combined token total for the pill, input + output for the parent turn PLUS any work delegated to subagents (browser agents, invoke-agent forks) and tool MCP servers that produced their own usage on this turn. The user-visible answer to "how big is this turn" is the all-in sum, not just the primary's output. We sum every reachable source: - parent's input (session.tokens["input"], ResultMessage.usage at line ~2886) - parent's output (session.tokens["output"], same ResultMessage) - every direct sub-session whose parent_session_id points at this session (browser agents, sub-agent forks, invoke-agent calls book their own usage at subprocess return time, agent_manager.py:1365 + browser_agent.py:1000-1001) This mirrors how billing accumulates per-turn, caches, tool MCP servers that talk to LLMs (e.g. summarizers), and subagent reasoning all show up under the parent's "session.tokens" once their result lands. Read cumulative session totals + cumulative subagent totals at this moment, then subtract the turn-start baseline to get THIS TURN'S delta. Without subtracting, the second turn's pill would show turn-1 work added to turn-2 work, the third would show all three, etc. Pill uses the FRESH lane (uncached input only). session.tokens ["input"] stays full for the context-fullness bar + cost; the bubble shows the NEW tokens this turn, not the cached re-reads.
cum_in = 0
cum_out = 0
if isinstance(session.tokens, dict):
@@ -170,8 +115,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
except Exception:
pass
# Fall back to cumulative if the baseline wasn't captured
# (degenerate empty turn, better than showing zero).
# Fall back to cumulative if the baseline wasn't captured (degenerate empty turn, better than showing zero).
if turn.baseline_captured:
parent_in = max(0, cum_in - turn.baseline_session_in)
parent_out = max(0, cum_out - turn.baseline_session_out)
@@ -183,11 +127,7 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
children_in = cum_children_in
children_out = cum_children_out
# Fresh input + output = the NEW tokens this turn. The old
# framework-overhead subtraction is gone on purpose: it was an
# estimate to strip the cached static prefix out of the full
# input number, and the fresh lane already excludes that prefix
# exactly, so subtracting it again would double-discount to ~0.
# Fresh input + output = the NEW tokens this turn. The old framework-overhead subtraction is gone on purpose: it was an estimate to strip the cached static prefix out of the full input number, and the fresh lane already excludes that prefix exactly, so subtracting it again would double-discount to ~0.
turn_total_tokens: Optional[int] = (
parent_in + parent_out + children_in + children_out
)
@@ -221,7 +161,6 @@ async def emit_consolidated_thinking(thinking: ThinkingState, turn: TurnState, s
logger.exception("Failed to emit consolidated thinking message")
@typechecked
async def ticker_loop(thinking: ThinkingState, turn: TurnState, session: AgentSession, session_id: str, sessions: Dict[str, AgentSession]) -> None:
"""Re-emit the consolidated thinking message every 1s so