Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 23 additions & 22 deletions src/entrypoints/serve_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ def _build_parser() -> argparse.ArgumentParser:
"--fallback-model", default=None, dest="fallback_model",
help="Model to switch to after repeated overloaded errors.",
)
parser.add_argument("--permission-mode", default="default", dest="permission_mode",
parser.add_argument("--permission-mode", default=None, dest="permission_mode",
help="default | acceptEdits | bypassPermissions | plan | auto")
parser.add_argument("--dangerously-skip-permissions", action="store_true",
dest="dangerously_skip_permissions",
Expand Down Expand Up @@ -160,28 +160,29 @@ def run_serve_subcommand(argv: list[str]) -> int:
bypass_requested=dangerously or allow_dangerously,
)

from src.permissions.modes import is_bypass_permissions_mode_disabled

disabled = is_bypass_permissions_mode_disabled()
if dangerously and not disabled:
args.permission_mode = "bypassPermissions"
elif dangerously and disabled:
logger.warning("Bypass permissions mode disabled by settings/policy; "
"ignoring --dangerously-skip-permissions")
if args.permission_mode == "bypassPermissions" and disabled:
logger.warning("Bypass permissions mode disabled by settings/policy; "
"ignoring --permission-mode bypassPermissions")
args.permission_mode = "default"

# Multi-session transport: bypass availability comes from FLAGS only,
# exactly like the agent-server's --http path. The desktop launcher (the
# single local operator) resolves settings at ITS boundary and forwards
# flags; folding host settings in here would unlock bypass for every
# client of this port. Lockdown still revokes an explicit request.
is_bypass_available = (dangerously or allow_dangerously) and not disabled

workspace = str(Path(args.workspace).resolve()) if args.workspace else str(Path.cwd())

# The desktop is an INTERACTIVE surface with a real user at the window, and
# this server is its own loopback, token-gated child — the same trust model
# as the TUI launcher spawning its agent-server. So it resolves permissions
# through the shared interactive resolver (src/cli.py + tui_launcher use it
# too), which means Full Access by default, exactly like `clawcodex`.
#
# Without this the desktop ran in "default" mode and asked approval for
# every Write/Bash; the resolver also carries the guards that make the
# implicit floor safe — an operator `disableBypassPermissionsMode`
# lockdown and root-outside-sandbox both drop it back to prompting, and a
# persisted `permissions.defaultMode` still wins.
from src.permissions.modes import resolve_interactive_permission_state

mode, is_bypass_available, bypass_selectable = resolve_interactive_permission_state(
permission_mode_cli=args.permission_mode,
dangerously_skip_permissions=dangerously,
allow_dangerously_skip_permissions=allow_dangerously,
cwd=workspace,
)
args.permission_mode = mode

if args.fallback_model and args.fallback_model == args.model:
print("serve: --fallback-model must differ from --model", file=sys.stderr)
return 2
Expand All @@ -197,7 +198,7 @@ def run_serve_subcommand(argv: list[str]) -> int:
fallback_model=args.fallback_model,
permission_mode=args.permission_mode,
is_bypass_available=is_bypass_available,
bypass_selectable=is_bypass_available,
bypass_selectable=bypass_selectable,
max_turns=args.max_turns,
)

Expand Down
61 changes: 57 additions & 4 deletions src/server/desktop_gateway_methods.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,38 @@
DESKTOP_CONTRACT = 1


# The agent's permission modes vs the desktop's approval vocabulary. The
# renderer's normalizeApprovalMode only knows manual|smart|off and silently
# coerces anything else to "manual", so sending the raw mode made a Full
# Access session render as "ask every time".
_APPROVAL_MODE_FOR = {
"bypassPermissions": "off", # Full Access — nothing to approve
"auto": "smart", # the classifier lane
"default": "manual",
"acceptEdits": "manual", # edits auto-approve, everything else asks
"plan": "manual",
}
_PERMISSION_MODE_FOR = {
"off": "bypassPermissions",
"smart": "auto",
"manual": "default",
}


def approval_mode_for(permission_mode: Any) -> str | None:
"""Agent permission mode → the desktop's manual|smart|off vocabulary."""
if not isinstance(permission_mode, str) or not permission_mode:
return None
return _APPROVAL_MODE_FOR.get(permission_mode, "manual")


def permission_mode_for(approval_mode: Any) -> str | None:
"""The desktop's approval mode → an agent permission mode."""
if not isinstance(approval_mode, str):
return None
return _PERMISSION_MODE_FOR.get(approval_mode.strip().lower())


def _init_session_info(init: dict[str, Any]) -> dict[str, Any]:
"""system/init frame → the ``session.info`` payload the renderer reads."""
payload: dict[str, Any] = {"running": False, "desktop_contract": DESKTOP_CONTRACT}
Expand All @@ -54,7 +86,7 @@ def _init_session_info(init: dict[str, Any]) -> dict[str, Any]:
payload["cwd"] = cwd
mode = init.get("permissionMode") or init.get("permission_mode")
if mode:
payload["approval_mode"] = mode
payload["approval_mode"] = approval_mode_for(mode)
model = init.get("model")
if model:
payload["model"] = model
Expand Down Expand Up @@ -145,7 +177,7 @@ async def publish_session_info(self, **extra: Any) -> None:
if settings.get("provider"):
payload["provider"] = str(settings["provider"])
if settings.get("permission_mode"):
payload["approval_mode"] = settings["permission_mode"]
payload["approval_mode"] = approval_mode_for(settings["permission_mode"])
effort = settings.get("reasoning_effort") or settings.get("effort")
if effort:
payload["reasoning_effort"] = str(effort)
Expand Down Expand Up @@ -212,7 +244,10 @@ async def _route(self, frame: dict[str, Any]) -> None:
# line too (permission mode may have flipped server-side).
mode = frame.get("permission_mode")
if mode:
await self._broadcast("session.info", {"approval_mode": mode, "running": False})
await self._broadcast(
"session.info",
{"approval_mode": approval_mode_for(mode), "running": False},
)
# …and republish the full line (model/provider/effort) — a turn can
# change them server-side (fallback model, plan-mode flip).
# Scheduled, never awaited: this control response routes through
Expand Down Expand Up @@ -333,6 +368,18 @@ async def config_set(self, key: str, value: Any, persist: bool = False) -> dict[
no control and succeed locally in the renderer, so an unknown key is a
silent ok here rather than an error.
"""
if key == "approvals.mode":
# Safety panel / `/approvals`: manual|smart|off → a permission mode.
mode = permission_mode_for(value)
if mode is None:
return {"ok": False, "error": f"unknown approvals mode {value!r}"}
reply = await self.control_query("set_permission_mode",
{"mode": mode, "persist": True})
res = reply if isinstance(reply, dict) else {}
applied = approval_mode_for(res.get("mode") or mode)
await self.publish_session_info()
return {"ok": res.get("ok") is not False, "value": applied,
"error": res.get("error")}
if key == "permission_mode":
reply = await self.control_query("set_permission_mode",
{"mode": value, "persist": persist})
Expand Down Expand Up @@ -705,7 +752,13 @@ async def config_get(self, params: dict[str, Any]) -> dict[str, Any]:
session = self._first_session(params)
if session is None:
return {}
return await session.control_query("get_settings", {}) or {}
settings = await session.control_query("get_settings", {}) or {}
# The Safety panel + /approvals read this single key and expect
# {value}; without it they always resolved to "manual" (the fallback)
# and misreported a Full Access session as asking every time.
if str(params.get("key") or "") == "approvals.mode":
return {"value": approval_mode_for(settings.get("permission_mode")) or "manual"}
return settings

async def config_set_rpc(self, params: dict[str, Any]) -> dict[str, Any]:
session = self._session(params)
Expand Down
84 changes: 80 additions & 4 deletions tests/server/test_desktop_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ def __init__(self) -> None:
self.shutdown_called = False
self.model = "fake"
self.provider = "fakeprov"
self.permission_mode = "bypassPermissions"

async def send_to_agent(self, frame: dict) -> None:
self.inbound.append(frame)
Expand All @@ -91,11 +92,14 @@ async def send_to_agent(self, frame: dict) -> None:
self.model = request.get("model") or self.model
self.provider = request.get("provider") or self.provider
reply = {"ok": True, "model": self.model}
elif subtype == "set_permission_mode":
self.permission_mode = request.get("mode") or self.permission_mode
reply = {"ok": True, "mode": self.permission_mode, "persisted": True}
elif subtype == "get_settings":
reply = {
"model": self.model,
"provider": self.provider,
"permission_mode": "default",
"permission_mode": self.permission_mode,
}
if reply is not None:
await self.queue.put(
Expand Down Expand Up @@ -145,7 +149,7 @@ async def messages_from_agent(self):
"type": "system",
"subtype": "init",
"cwd": "/tmp/w",
"permissionMode": "default",
"permissionMode": "bypassPermissions",
"model": "fake",
}
while True:
Expand Down Expand Up @@ -206,7 +210,8 @@ def test_create_submit_stream_complete(tmp_path: Path) -> None:
created = _drain_for_response(ws, 1, events)
session_id = created["result"]["session_id"]
assert session_id == "fake-1"
assert created["result"]["info"]["approval_mode"] == "default"
# Full Access maps to the desktop vocabulary: manual|smart|off.
assert created["result"]["info"]["approval_mode"] == "off"

_rpc(ws, 2, "prompt.submit", {"session_id": session_id, "text": "hi"})
_drain_for_response(ws, 2, events)
Expand Down Expand Up @@ -299,6 +304,77 @@ def reply_frames():
assert reply["response"]["updatedInput"] == {"command": "rm -rf /tmp/x"}


def test_serve_defaults_to_full_access_like_the_cli() -> None:
"""The desktop is an interactive surface, so serve resolves permissions
through the SAME resolver as `clawcodex` / the TUI launcher — Full Access
by default. Running in "default" mode made every Write/Bash raise an
approval the user never answered, and the tools timed out."""
from src.entrypoints.serve_cli import _build_parser
from src.permissions.modes import resolve_interactive_permission_state

# No --permission-mode → None, so the resolver's floor applies (pinning it
# to "default" here would defeat the implicit Full Access floor).
assert _build_parser().parse_args([]).permission_mode is None

mode, _available, selectable = resolve_interactive_permission_state(
permission_mode_cli=None,
dangerously_skip_permissions=False,
allow_dangerously_skip_permissions=False,
cwd=None,
)
assert mode == "bypassPermissions"
assert selectable is True


def test_approval_mode_vocabulary_mapping() -> None:
"""The renderer only knows manual|smart|off and coerces anything else to
"manual" — so a Full Access session rendered as "ask every time"."""
from src.server.desktop_gateway_methods import approval_mode_for, permission_mode_for

assert approval_mode_for("bypassPermissions") == "off"
assert approval_mode_for("auto") == "smart"
assert approval_mode_for("default") == "manual"
assert approval_mode_for("acceptEdits") == "manual"
assert approval_mode_for("plan") == "manual"
assert approval_mode_for(None) is None

assert permission_mode_for("off") == "bypassPermissions"
assert permission_mode_for("smart") == "auto"
assert permission_mode_for("manual") == "default"
assert permission_mode_for("nonsense") is None


def test_approvals_mode_get_and_set(tmp_path: Path) -> None:
"""The Safety panel round-trips a single `approvals.mode` key; without a
handler config.get returned the settings blob (no `value`) and the panel
always showed "manual"."""
state, agents = _fake_state(tmp_path)
with TestClient(build_app(state)) as client, _connect(client) as ws:
ws.receive_json()
events: list[dict] = []
_rpc(ws, 1, "session.create", {})
sid = _drain_for_response(ws, 1, events)["result"]["session_id"]

# Reads the live mode in the desktop's vocabulary.
_rpc(ws, 2, "config.get", {"session_id": sid, "key": "approvals.mode"})
assert _drain_for_response(ws, 2, events)["result"] == {"value": "off"}

# Writing maps back to a permission mode and persists it.
_rpc(ws, 3, "config.set", {"session_id": sid, "key": "approvals.mode",
"value": "manual"})
result = _drain_for_response(ws, 3, events)["result"]
assert result["ok"] is True and result["value"] == "manual"
assert agents[0].permission_mode == "default"

_rpc(ws, 4, "config.get", {"session_id": sid, "key": "approvals.mode"})
assert _drain_for_response(ws, 4, events)["result"] == {"value": "manual"}

# An unknown mode is refused rather than silently mapped.
_rpc(ws, 5, "config.set", {"session_id": sid, "key": "approvals.mode",
"value": "bogus"})
assert _drain_for_response(ws, 5, events)["result"]["ok"] is False


def test_init_session_info_carries_provider() -> None:
"""The picker only prefers the session's selection when BOTH model and
provider are set; without provider it fell back to the catalog while the
Expand All @@ -307,7 +383,7 @@ def test_init_session_info_carries_provider() -> None:

info = _init_session_info({
"cwd": "/w", "model": "m1", "provider": "p1",
"permissionMode": "default", "session_id": "s1",
"permissionMode": "bypassPermissions", "session_id": "s1",
})
assert info["model"] == "m1"
assert info["provider"] == "p1"
Expand Down
5 changes: 4 additions & 1 deletion tests/server/test_desktop_serve.py
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,10 @@ def test_serve_cli_parser_defaults() -> None:
args = _build_parser().parse_args([])
assert args.host == "127.0.0.1"
assert args.port == 0
assert args.permission_mode == "default"
# Unset, NOT "default": the interactive resolver supplies the mode (Full
# Access floor, honoring persisted defaultMode + lockdown). Pinning a
# value here would defeat it — see test_serve_defaults_to_full_access.
assert args.permission_mode is None


def test_serve_cli_accepts_desktop_spawn_shape() -> None:
Expand Down
Loading