[eric] backend: headless mode drops the renderer-bound tools and denies approvals instantly

backend/apps/agents/manager/permissions/build_effective_tool_lists.py

@@ -18,6 +18,7 @@ from backend.apps.tools_lib.tools_lib import (
     load_all_tools as load_all_tools,
     sanitize_server_name as sanitize_server_name,
 )
+from backend.config.headless import apply_headless_denies
 
 # Mutation/exec tools a read-only session must never reach: Edit (rewrites files), Bash (rm/mv/overwrite),
 # NotebookEdit (rewrites notebooks). Write is intentionally NOT here, the audit needs its one report.
@@ -33,6 +34,8 @@ def build_effective_tool_lists(
     browser_delegation_tools: List[str],
     invoke_agent_tools: List[str],
 ) -> Tuple[List[str], List[str]]:
+    # Same shadow the server registration takes: headless, the renderer-bound built-ins go straight onto disallowed instead of being offered and failing when called.
+    builtin_perms = apply_headless_denies(builtin_perms)
     effective_allowed = [
         t for t in session.allowed_tools
         if t in FULL_TOOLS and builtin_perms.get(t, "always_allow") == "always_allow"

backend/apps/agents/manager/permissions/workflow_approval.py

@@ -16,6 +16,7 @@ from typeguard import typechecked
 from backend.apps.agents.manager.permissions.ApprovalDecision import ApprovalDecision
 from backend.apps.agents.manager.permissions.decision import request_user_approval
 from backend.apps.agents.manager.streaming.HookContext import HookContext
+from backend.config.headless import is_headless
 
 logger = logging.getLogger(__name__)
 
@@ -90,7 +91,8 @@ async def resolve_ask(
 ) -> ApprovalDecision:
     """Resolve an 'ask' policy. On a workflow run, reuse a remembered decision
     (this step first, then the workflow-level fallback) instead of prompting, and
-    persist any fresh non-sensitive answer so later fires don't re-ask. Shared by
+    persist any fresh non-sensitive answer so later fires don't re-ask. Headless,
+    anything still unresolved is denied on the spot instead of prompting. Shared by
     both gates so they can't disagree (and so the first one's answer is reused by
     the second within the same call)."""
     mem = p_approval_memory.get(ctx.session_id)
@@ -113,6 +115,12 @@ async def resolve_ask(
         if prior == "deny":
             note_tool_used(ctx.session_id, tool_name, False)
             return ApprovalDecision(behavior="deny", message="Denied by a remembered workflow permission")
+    # Headless there is no one to ask, so the request would just be broadcast into the void and come back denied ten minutes later; deny now, but only after the remembered decisions above got their say.
+    if is_headless():
+        return ApprovalDecision(
+            behavior="deny",
+            message="This run is headless, so nobody can approve a tool that asks. Use a tool that doesn't need approval, or report what you'd need permission for.",
+        )
     timeout = mem.ask_timeout if mem is not None else 600.0
     decision = await request_user_approval(
         ctx.session, ctx.session_id, tool_name, tool_input, ctx.builtin_perms,

backend/apps/agents/manager/register_builtin_mcp_servers.py

@@ -12,6 +12,7 @@ from typeguard import typechecked
 
 from backend.apps.agents.core.models import AgentSession
 from backend.auth import get_auth_token
+from backend.config.headless import apply_headless_denies
 
 
 @typechecked
@@ -24,6 +25,8 @@ def register_builtin_mcp_servers(
 ) -> Tuple[List[str], List[str]]:
     import backend.apps.agents as p_agents_pkg
     agents_dir = os.path.dirname(p_agents_pkg.__file__)
+    # Headless has no renderer for a webview or a UI component, so we shadow the map once here and let the existing deny short-circuits skip those servers; nothing below may read the un-shadowed one.
+    builtin_perms = apply_headless_denies(builtin_perms)
     browser_delegation_tools = ["CreateBrowserAgent", "BrowserAgent", "BrowserAgents", "AppAgent"]
     browser_all_denied = all(
         builtin_perms.get(t, "always_allow") == "deny"

backend/config/headless.py

@@ -0,0 +1,34 @@
+"""Headless mode: the backend running with no Electron renderer, no display, and no human
+(a Linux container). Single source of truth for the flag and for the tools that dead-end at a
+renderer, so they are dropped from the tool surface up front instead of hanging at call time."""
+
+import os
+from typing import Dict, FrozenSet
+
+from typeguard import typechecked
+
+# Each of these ends at the Electron renderer: browser/app delegation drives live webviews, ShowUI (the same gate AskUI rides) draws into the transcript, and AskUserQuestion waits on a person who isn't there.
+HEADLESS_DENIED_TOOLS: FrozenSet[str] = frozenset({
+    "CreateBrowserAgent",
+    "BrowserAgent",
+    "BrowserAgents",
+    "AppAgent",
+    "ShowUI",
+    "AskUserQuestion",
+})
+
+
+@typechecked
+def is_headless() -> bool:
+    """True when the backend was started with OPENSWARM_HEADLESS=1. Read per call rather than
+    frozen at import, so a launcher that sets it late still counts (and tests can flip it)."""
+    return os.environ.get("OPENSWARM_HEADLESS") == "1"
+
+
+@typechecked
+def apply_headless_denies(builtin_perms: Dict[str, str]) -> Dict[str, str]:
+    """The permission map with every renderer-bound tool forced to 'deny' when headless, and the
+    map itself untouched otherwise. Returns a copy so the mode never poisons the live snapshot."""
+    if not is_headless():
+        return builtin_perms
+    return {**builtin_perms, **{name: "deny" for name in HEADLESS_DENIED_TOOLS}}

backend/tests/test_headless_mode.py

@@ -0,0 +1,126 @@
+"""OPENSWARM_HEADLESS=1 gating: the tools that dead-end at an Electron renderer (browser/app
+delegation, ShowUI/AskUI, AskUserQuestion) must be gone from the effective tool surface, and an
+'ask' must deny on the spot instead of parking on the 600s approval timeout. Every case is paired
+with its headless-off twin, because a gate that can't be seen switching off proves nothing."""
+
+import pytest
+from unittest.mock import AsyncMock, patch
+
+from backend.apps.agents.core.models import AgentSession
+from backend.apps.agents.manager.permissions import workflow_approval
+from backend.apps.agents.manager.permissions.build_effective_tool_lists import build_effective_tool_lists
+from backend.apps.agents.manager.register_builtin_mcp_servers import register_builtin_mcp_servers
+from backend.apps.agents.manager.streaming.HookContext import HookContext
+from backend.config.headless import HEADLESS_DENIED_TOOLS
+
+BROWSER_DELEGATION = ("CreateBrowserAgent", "BrowserAgent", "BrowserAgents", "AppAgent")
+
+
+def p_session():
+    session = AgentSession(name="t", model="sonnet", dashboard_id="d")
+    session.allowed_tools = ["Read", "Bash", "AskUserQuestion"]
+    return session
+
+
+def p_run_the_real_pipeline():
+    """Registration then tool-list build, in the order the agent loop runs them."""
+    session = p_session()
+    mcp_servers = {}
+    browser_tools, invoke_tools = register_builtin_mcp_servers(
+        mcp_servers, session, {}, None, None)
+    allowed, disallowed = build_effective_tool_lists(
+        session, mcp_servers, {}, False, browser_tools, invoke_tools)
+    return mcp_servers, allowed, disallowed
+
+
+def p_ctx() -> HookContext:
+    session = p_session()
+    return HookContext(
+        session=session,
+        session_id=session.id,
+        prompt="hi",
+        builtin_perms={},
+        policy_defaults={},
+        sessions={},
+    )
+
+
+def test_the_denied_set_is_exactly_the_renderer_bound_tools():
+    assert HEADLESS_DENIED_TOOLS == frozenset(BROWSER_DELEGATION) | {"ShowUI", "AskUserQuestion"}
+
+
+def test_headless_drops_the_renderer_bound_servers_and_tools(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    mcp_servers, allowed, disallowed = p_run_the_real_pipeline()
+    assert "openswarm-browser-agent" not in mcp_servers
+    assert "openswarm-ui" not in mcp_servers
+    for tool in BROWSER_DELEGATION:
+        assert f"mcp__openswarm-browser-agent__{tool}" not in allowed
+    for ui_tool in ("ShowUI", "AskUI"):
+        assert f"mcp__openswarm-ui__{ui_tool}" not in allowed
+    assert "AskUserQuestion" not in allowed
+    assert "AskUserQuestion" in disallowed
+    # The rest of the surface is untouched; headless prunes the renderer, it doesn't lobotomise the agent.
+    assert "Read" in allowed and "Bash" in allowed
+    assert "openswarm-invoke-agent" in mcp_servers
+    assert "openswarm-apps" in mcp_servers
+
+
+def test_without_headless_every_one_of_them_is_offered(monkeypatch):
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    mcp_servers, allowed, _ = p_run_the_real_pipeline()
+    assert "openswarm-browser-agent" in mcp_servers
+    assert "openswarm-ui" in mcp_servers
+    for tool in BROWSER_DELEGATION:
+        assert f"mcp__openswarm-browser-agent__{tool}" in allowed
+    for ui_tool in ("ShowUI", "AskUI"):
+        assert f"mcp__openswarm-ui__{ui_tool}" in allowed
+
+
+def test_askuserquestion_survives_when_the_ui_server_is_absent(monkeypatch):
+    # With no openswarm-ui registered nothing else denies AskUserQuestion, so this isolates the headless gate.
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    allowed, disallowed = build_effective_tool_lists(p_session(), {}, {}, False, [], [])
+    assert "AskUserQuestion" in allowed and "AskUserQuestion" not in disallowed
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    allowed, disallowed = build_effective_tool_lists(p_session(), {}, {}, False, [], [])
+    assert "AskUserQuestion" not in allowed and "AskUserQuestion" in disallowed
+
+
+def test_only_the_exact_flag_value_turns_headless_on(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "0")
+    _, allowed, _ = p_run_the_real_pipeline()
+    assert "mcp__openswarm-ui__ShowUI" in allowed
+
+
+@pytest.mark.asyncio
+async def test_headless_denies_an_ask_without_ever_prompting(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    ask = AsyncMock()
+    with patch.object(workflow_approval, "request_user_approval", new=ask):
+        decision = await workflow_approval.resolve_ask(p_ctx(), "Bash", {"command": "ls"}, None)
+    assert decision.behavior == "deny"
+    assert not ask.called  # never broadcast into the void, so never a 600s park
+
+
+@pytest.mark.asyncio
+async def test_without_headless_an_ask_still_prompts(monkeypatch):
+    monkeypatch.delenv("OPENSWARM_HEADLESS", raising=False)
+    ask = AsyncMock(return_value=workflow_approval.ApprovalDecision(behavior="allow"))
+    with patch.object(workflow_approval, "request_user_approval", new=ask):
+        decision = await workflow_approval.resolve_ask(p_ctx(), "Bash", {"command": "ls"}, None)
+    assert decision.behavior == "allow"
+    assert ask.called
+
+
+@pytest.mark.asyncio
+async def test_headless_still_honors_a_remembered_allow(monkeypatch):
+    monkeypatch.setenv("OPENSWARM_HEADLESS", "1")
+    ctx = p_ctx()
+    workflow_approval.set_workflow_approval_memory(
+        ctx.session_id, decisions={"Bash": "allow"}, step_usage={}, remember=None, ask_timeout=5.0)
+    try:
+        decision = await workflow_approval.resolve_ask(ctx, "Bash", {"command": "ls"}, None)
+    finally:
+        workflow_approval.clear_workflow_approval_memory(ctx.session_id)
+    assert decision.behavior == "allow"