@@ -437,6 +478,7 @@ export default function Tools() {
setModalState={setAddToolModalState}
getUserTools={getUserTools}
onToolAdded={handleToolAdded}
+ onDevicePaired={handleDevicePaired}
/>
} />
} />
} />
+ }
+ />
} />
} />
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index 31a2720c..ae507f82 100644
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -1,21 +1,33 @@
///
-import { defineConfig } from 'vite';
+import { defineConfig, loadEnv } from 'vite';
import react from '@vitejs/plugin-react';
import svgr from 'vite-plugin-svgr';
import path from 'path';
// https://vitejs.dev/config/
-export default defineConfig({
- plugins: [react(), svgr()],
- resolve: {
- alias: {
- '@': path.resolve(__dirname, './src'),
+export default defineConfig(({ mode }) => {
+ const env = loadEnv(mode, process.cwd(), '');
+ return {
+ plugins: [react(), svgr()],
+ resolve: {
+ alias: {
+ '@': path.resolve(__dirname, './src'),
+ },
},
- },
- test: {
- environment: 'happy-dom',
- globals: true,
- include: ['src/**/*.test.{ts,tsx}'],
- setupFiles: ['./vitest.setup.ts'],
- },
+ server: {
+ // Extra dev hosts (e.g. a tailscale name) come from VITE_ALLOWED_HOSTS in
+ // an untracked .env.local; machine-specific names stay out of the repo.
+ allowedHosts: env.VITE_ALLOWED_HOSTS
+ ? env.VITE_ALLOWED_HOSTS.split(',')
+ .map((h) => h.trim())
+ .filter(Boolean)
+ : [],
+ },
+ test: {
+ environment: 'happy-dom',
+ globals: true,
+ include: ['src/**/*.test.{ts,tsx}'],
+ setupFiles: ['./vitest.setup.ts'],
+ },
+ };
});
diff --git a/scripts/e2e/mock_llm_fixtures/01c733b9660eed626247ffb527486e12c2b0def48bc79fd4468533864c268441.json b/scripts/e2e/mock_llm_fixtures/01c733b9660eed626247ffb527486e12c2b0def48bc79fd4468533864c268441.json
new file mode 100644
index 00000000..19b90b15
--- /dev/null
+++ b/scripts/e2e/mock_llm_fixtures/01c733b9660eed626247ffb527486e12c2b0def48bc79fd4468533864c268441.json
@@ -0,0 +1,17 @@
+{
+ "_comment": "Tester fixture: 'please run whoami on my remote device' -> run_command tool call (whoami). Used to verify full mode auto-approves a safe non-denylisted command.",
+ "response": {
+ "content": "",
+ "tool_calls": [
+ {
+ "id": "call_remote_tester_whoami_1",
+ "type": "function",
+ "function": {
+ "name": "run_command",
+ "arguments": "{\"command\":\"whoami\",\"working_directory\":\"/tmp\",\"timeout_ms\":5000}"
+ }
+ }
+ ],
+ "finish_reason": "tool_calls"
+ }
+}
diff --git a/scripts/e2e/mock_llm_fixtures/b196e9532b6a979d959a570dcdec5f96cea35175091612545e24ff91ed65c8a3.json b/scripts/e2e/mock_llm_fixtures/b196e9532b6a979d959a570dcdec5f96cea35175091612545e24ff91ed65c8a3.json
new file mode 100644
index 00000000..20facd7b
--- /dev/null
+++ b/scripts/e2e/mock_llm_fixtures/b196e9532b6a979d959a570dcdec5f96cea35175091612545e24ff91ed65c8a3.json
@@ -0,0 +1,17 @@
+{
+ "_comment": "Tester fixture: 'please run mkdir /tmp/rd_v3_test_dir on my remote device' -> run_command tool call (mkdir /tmp/rd_v3_test_dir). Used to verify a write command in ask mode still pauses on the approval bar.",
+ "response": {
+ "content": "",
+ "tool_calls": [
+ {
+ "id": "call_remote_tester_mkdir_1",
+ "type": "function",
+ "function": {
+ "name": "run_command",
+ "arguments": "{\"command\":\"mkdir /tmp/rd_v3_test_dir\",\"working_directory\":\"/tmp\",\"timeout_ms\":5000}"
+ }
+ }
+ ],
+ "finish_reason": "tool_calls"
+ }
+}
diff --git a/scripts/e2e/mock_llm_fixtures/d42a7dccbd1f218bb095a28e2b4a9c878800e030223be5189ebe13274c65d21f.json b/scripts/e2e/mock_llm_fixtures/d42a7dccbd1f218bb095a28e2b4a9c878800e030223be5189ebe13274c65d21f.json
new file mode 100644
index 00000000..e52cc2f2
--- /dev/null
+++ b/scripts/e2e/mock_llm_fixtures/d42a7dccbd1f218bb095a28e2b4a9c878800e030223be5189ebe13274c65d21f.json
@@ -0,0 +1,17 @@
+{
+ "_comment": "Tester fixture: 'please run rm -rf / on my remote device' -> run_command tool call (rm -rf /). Used to verify full mode still pauses on the denylist-forced-prompt circuit-breaker.",
+ "response": {
+ "content": "",
+ "tool_calls": [
+ {
+ "id": "call_remote_tester_rm_1",
+ "type": "function",
+ "function": {
+ "name": "run_command",
+ "arguments": "{\"command\":\"rm -rf /\",\"working_directory\":\"/tmp\",\"timeout_ms\":5000}"
+ }
+ }
+ ],
+ "finish_reason": "tool_calls"
+ }
+}
diff --git a/scripts/e2e/mock_llm_fixtures/ed4ff698efb5a8c5f4d6f8ad1a68526da3813c82b0bb57ceaff298e959193d1f.json b/scripts/e2e/mock_llm_fixtures/ed4ff698efb5a8c5f4d6f8ad1a68526da3813c82b0bb57ceaff298e959193d1f.json
new file mode 100644
index 00000000..512f0eb1
--- /dev/null
+++ b/scripts/e2e/mock_llm_fixtures/ed4ff698efb5a8c5f4d6f8ad1a68526da3813c82b0bb57ceaff298e959193d1f.json
@@ -0,0 +1,17 @@
+{
+ "_comment": "Tester fixture: 'please run ls -la /tmp on my remote device' -> run_command tool call (ls -la /tmp).",
+ "response": {
+ "content": "",
+ "tool_calls": [
+ {
+ "id": "call_remote_tester_1",
+ "type": "function",
+ "function": {
+ "name": "run_command",
+ "arguments": "{\"command\":\"ls -la /tmp\",\"working_directory\":\"/tmp\",\"timeout_ms\":5000}"
+ }
+ }
+ ],
+ "finish_reason": "tool_calls"
+ }
+}
diff --git a/tests/agents/test_tool_executor.py b/tests/agents/test_tool_executor.py
index d37c5eae..80e5777e 100644
--- a/tests/agents/test_tool_executor.py
+++ b/tests/agents/test_tool_executor.py
@@ -457,6 +457,346 @@ class TestCheckPause:
assert result["action_name"] == "delete_all"
+@pytest.mark.unit
+class TestCheckPauseRemoteDevice:
+ """The gate must consult the live ``RemoteDeviceTool`` decision for
+ ``remote_device`` so ``approval_mode`` changes after pair time and
+ per-command heuristics (denylist, sticky) are respected.
+ """
+
+ def _make_call(self, name="run_command", call_id="c1", command="ls -la /tmp"):
+ import json
+
+ call = Mock()
+ call.name = name
+ call.id = call_id
+ call.arguments = json.dumps({"command": command})
+ call.thought_signature = None
+ return call
+
+ def _tools_dict(self, *, device_id="dev_abc"):
+ return {
+ "rd0": {
+ "id": "rd0",
+ "name": "remote_device",
+ "config": {"device_id": device_id},
+ "actions": [
+ {
+ "name": "run_command",
+ "description": "Execute on device",
+ "active": True,
+ "require_approval": False,
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "command": {
+ "type": "string",
+ "filled_by_llm": True,
+ "value": "",
+ },
+ },
+ "required": ["command"],
+ },
+ }
+ ],
+ }
+ }
+
+ def _patch_device(self, monkeypatch, device):
+ """Stub ``RemoteDeviceTool._load_device`` to return ``device``."""
+ from application.agents.tools import remote_device
+
+ monkeypatch.setattr(
+ remote_device.RemoteDeviceTool,
+ "_load_device",
+ lambda self: device,
+ )
+
+ def _patch_sticky(self, monkeypatch, patterns):
+ """Stub the sticky lookup to match any normalized pattern in ``patterns``."""
+ from application.agents.tools import remote_device
+
+ monkeypatch.setattr(
+ remote_device,
+ "normalize_command",
+ lambda cmd: cmd.split()[0] + " *" if cmd else "",
+ )
+ captured = {"patterns": set(patterns)}
+
+ class _StubRepo:
+ def __init__(self, conn):
+ pass
+
+ def has_pattern(self, device_id, user_id, pattern):
+ return pattern in captured["patterns"]
+
+ monkeypatch.setattr(
+ remote_device,
+ "DeviceAutoApprovePatternsRepository",
+ _StubRepo,
+ )
+
+ class _StubConn:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ return False
+
+ monkeypatch.setattr(remote_device, "db_readonly", lambda: _StubConn())
+
+ def test_full_non_deny_no_pause(self, monkeypatch):
+ """``full`` + a non-denylisted command auto-approves."""
+ executor = ToolExecutor(user="alice")
+ self._patch_device(
+ monkeypatch,
+ {
+ "id": "dev_abc",
+ "approval_mode": "full",
+ "status": "active",
+ },
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="whoami")
+ assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
+
+ def test_full_writing_command_no_pause(self, monkeypatch):
+ """``full`` auto-approves writes too (only the denylist stops it)."""
+ executor = ToolExecutor(user="alice")
+ self._patch_device(
+ monkeypatch,
+ {
+ "id": "dev_abc",
+ "approval_mode": "full",
+ "status": "active",
+ },
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="rm /tmp/foo")
+ assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
+
+ def test_full_denylist_forces_pause(self, monkeypatch):
+ """``full`` + a denylisted command still pauses (forced prompt)."""
+ executor = ToolExecutor(user="alice")
+ self._patch_device(
+ monkeypatch,
+ {
+ "id": "dev_abc",
+ "approval_mode": "full",
+ "status": "active",
+ },
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="rm -rf /")
+ result = executor.check_pause(tools_dict, call, "OpenAILLM")
+ assert result is not None
+ assert result["pause_type"] == "awaiting_approval"
+
+ def test_ask_mode_pauses(self, monkeypatch):
+ """``ask`` + any command pauses by default."""
+ executor = ToolExecutor(user="alice")
+ self._patch_device(
+ monkeypatch,
+ {
+ "id": "dev_abc",
+ "approval_mode": "ask",
+ "status": "active",
+ },
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="ls -la /tmp")
+ result = executor.check_pause(tools_dict, call, "OpenAILLM")
+ assert result is not None
+ assert result["pause_type"] == "awaiting_approval"
+
+ def test_ask_mode_sticky_match_no_pause(self, monkeypatch):
+ """``ask`` + a command matching a stored sticky pattern auto-approves."""
+ executor = ToolExecutor(user="alice")
+ self._patch_device(
+ monkeypatch,
+ {
+ "id": "dev_abc",
+ "approval_mode": "ask",
+ "status": "active",
+ },
+ )
+ # The stub normalizer turns ``ls -la /tmp`` into ``ls *``.
+ self._patch_sticky(monkeypatch, {"ls *"})
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="ls -la /tmp")
+ assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
+
+
+@pytest.mark.unit
+class TestCheckPauseRemoteDeviceHeadless:
+ """Headless gating for ``remote_device``.
+
+ A denylist-forced prompt must NOT be bypassed by the run's
+ ``tool_allowlist`` — otherwise a scheduled/headless run with the device
+ allowlisted would auto-execute even a denylisted command. Normal
+ (non-forced) approvals keep the allowlist bypass.
+ """
+
+ def _make_call(self, name="run_command", call_id="c1", command="whoami"):
+ import json
+
+ call = Mock()
+ call.name = name
+ call.id = call_id
+ call.arguments = json.dumps({"command": command})
+ call.thought_signature = None
+ return call
+
+ def _tools_dict(self, *, device_id="dev_abc"):
+ return {
+ "rd0": {
+ "id": "rd0",
+ "name": "remote_device",
+ "config": {"device_id": device_id},
+ "actions": [
+ {
+ "name": "run_command",
+ "description": "Execute on device",
+ "active": True,
+ "require_approval": False,
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "command": {
+ "type": "string",
+ "filled_by_llm": True,
+ "value": "",
+ },
+ },
+ "required": ["command"],
+ },
+ }
+ ],
+ }
+ }
+
+ def _patch_device(self, monkeypatch, device):
+ from application.agents.tools import remote_device
+
+ monkeypatch.setattr(
+ remote_device.RemoteDeviceTool,
+ "_load_device",
+ lambda self: device,
+ )
+
+ def _patch_sticky(self, monkeypatch, patterns):
+ from application.agents.tools import remote_device
+
+ monkeypatch.setattr(
+ remote_device,
+ "normalize_command",
+ lambda cmd: cmd.split()[0] + " *" if cmd else "",
+ )
+ captured = {"patterns": set(patterns)}
+
+ class _StubRepo:
+ def __init__(self, conn):
+ pass
+
+ def has_pattern(self, device_id, user_id, pattern):
+ return pattern in captured["patterns"]
+
+ monkeypatch.setattr(
+ remote_device, "DeviceAutoApprovePatternsRepository", _StubRepo
+ )
+
+ class _StubConn:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ return False
+
+ monkeypatch.setattr(remote_device, "db_readonly", lambda: _StubConn())
+
+ def test_headless_allowlisted_denylisted_command_denied(self, monkeypatch):
+ """Device allowlisted, but ``rm -rf /`` is a forced prompt -> denied."""
+ executor = ToolExecutor(
+ user="alice", headless=True, tool_allowlist={"rd0"}
+ )
+ self._patch_device(
+ monkeypatch,
+ {"id": "dev_abc", "approval_mode": "full", "status": "active"},
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="rm -rf /")
+ result = executor.check_pause(tools_dict, call, "OpenAILLM")
+ assert result is not None
+ assert result["pause_type"] == "headless_denied"
+ assert result["error_type"] == "tool_not_allowed"
+
+ def test_headless_allowlisted_normal_command_executes(self, monkeypatch):
+ """Device allowlisted + a normal command in ``full`` -> executes."""
+ executor = ToolExecutor(
+ user="alice", headless=True, tool_allowlist={"rd0"}
+ )
+ self._patch_device(
+ monkeypatch,
+ {"id": "dev_abc", "approval_mode": "full", "status": "active"},
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="whoami")
+ assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
+
+ def test_headless_not_allowlisted_denied(self, monkeypatch):
+ """Without the allowlist, an ask-mode command is denied headless."""
+ executor = ToolExecutor(user="alice", headless=True, tool_allowlist=set())
+ self._patch_device(
+ monkeypatch,
+ {"id": "dev_abc", "approval_mode": "ask", "status": "active"},
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="ls -la /tmp")
+ result = executor.check_pause(tools_dict, call, "OpenAILLM")
+ assert result is not None
+ assert result["pause_type"] == "headless_denied"
+
+ def test_headless_allowlisted_ask_mode_normal_executes(self, monkeypatch):
+ """ask-mode + allowlisted + non-denylisted -> bypass allowed."""
+ executor = ToolExecutor(
+ user="alice", headless=True, tool_allowlist={"rd0"}
+ )
+ self._patch_device(
+ monkeypatch,
+ {"id": "dev_abc", "approval_mode": "ask", "status": "active"},
+ )
+ self._patch_sticky(monkeypatch, set())
+
+ tools_dict = self._tools_dict()
+ executor.prepare_tools_for_llm(tools_dict)
+ call = self._make_call(command="ls -la /tmp")
+ # Not denylist-forced, so the allowlist bypass applies.
+ assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
+
+
@pytest.mark.unit
class TestToolExecutorExecute:
diff --git a/tests/core/test_openai_compatible.py b/tests/core/test_openai_compatible.py
index f8f52987..4001197a 100644
--- a/tests/core/test_openai_compatible.py
+++ b/tests/core/test_openai_compatible.py
@@ -74,7 +74,14 @@ def _write_together_yaml(directory: Path) -> Path:
@pytest.fixture(autouse=True)
-def _reset_registry():
+def _reset_registry(monkeypatch):
+ # Builtin openai_compatible catalogs (e.g. deepseek.yaml) activate when
+ # their api_key_env is present in os.environ, which the openai_compatible
+ # provider reads directly, independent of the patched settings. Clear it
+ # so a key leaked by another test (or present in the dev .env) doesn't add
+ # extra models to the exact-match assertions below. Mirrors the delenv
+ # guard in test_model_registry_yaml.py.
+ monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
ModelRegistry.reset()
yield
ModelRegistry.reset()
diff --git a/tests/devices/__init__.py b/tests/devices/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/devices/test_auto_approve_validation.py b/tests/devices/test_auto_approve_validation.py
new file mode 100644
index 00000000..bab1d240
--- /dev/null
+++ b/tests/devices/test_auto_approve_validation.py
@@ -0,0 +1,44 @@
+"""Input validation for the add-auto-approve-pattern route.
+
+These exercise ``add_auto_approve_pattern`` directly. The non-string and
+missing-command cases short-circuit with a 400 before any DB access, so no
+Postgres fixture is needed.
+"""
+
+from __future__ import annotations
+
+import pytest
+from flask import Flask
+
+from application.api.devices import routes as routes_module
+
+
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _call(app: Flask, body):
+ with app.test_request_context(
+ "/api/devices/dev_x/auto-approve",
+ method="POST",
+ json=body,
+ ):
+ from flask import request as flask_request
+
+ flask_request.decoded_token = {"sub": "user_abc"}
+ return routes_module.add_auto_approve_pattern("dev_x")
+
+
+@pytest.mark.parametrize("bad_command", [123, 12.5, True, ["ls"], {"c": 1}])
+def test_add_auto_approve_rejects_non_string_command(app, bad_command):
+ # A truthy-but-non-string command must 400 rather than crash in shlex.
+ response = _call(app, {"command": bad_command})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_command"
+
+
+def test_add_auto_approve_missing_command(app):
+ response = _call(app, {})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "missing_command"
diff --git a/tests/devices/test_broker_cleanup.py b/tests/devices/test_broker_cleanup.py
new file mode 100644
index 00000000..561cc744
--- /dev/null
+++ b/tests/devices/test_broker_cleanup.py
@@ -0,0 +1,77 @@
+"""Tests for ``DeviceBroker.cleanup_invocation`` pending removal."""
+
+from __future__ import annotations
+
+from queue import Empty
+
+from application.devices.broker import DeviceBroker
+
+
+def _dispatch_offline(broker: DeviceBroker):
+ # No session registered -> the invocation lands in _pending.
+ envelope = {"invocation_id": "inv_stale", "action": "run_command"}
+ return broker.dispatch_invocation("dev_x", "user_x", envelope)
+
+
+def test_cleanup_removes_pending_so_register_session_doesnt_replay():
+ # A timed-out invocation cleaned up while still pending must NOT be
+ # drained/executed by a later session registration.
+ broker = DeviceBroker()
+ inv = _dispatch_offline(broker)
+ assert broker._pending.get("dev_x") == [inv]
+
+ broker.cleanup_invocation(inv.invocation_id)
+ # Pending key gone entirely once the list empties.
+ assert "dev_x" not in broker._pending
+ assert broker.get_invocation(inv.invocation_id) is None
+
+ sess = broker.register_session("dev_x", "user_x")
+ # The stale invocation must not have been re-queued.
+ assert inv.invocation_id not in sess.invocations
+ try:
+ sess.invocation_queue.get_nowait()
+ raise AssertionError("stale invocation was dispatched after cleanup")
+ except Empty:
+ pass
+
+
+def test_cleanup_removes_from_active_session():
+ # A completed invocation must not linger in the active session's map,
+ # or a long-lived SSE session accumulates them unboundedly.
+ broker = DeviceBroker()
+ sess = broker.register_session("dev_live", "user_live")
+ inv = broker.dispatch_invocation(
+ "dev_live", "user_live",
+ {"invocation_id": "inv_live", "action": "run_command"},
+ )
+ # Dispatched into the active session.
+ assert "inv_live" in sess.invocations
+ assert broker.get_invocation("inv_live") is inv
+
+ # Drive it to completion via a control chunk, then clean up.
+ broker.submit_output_chunk(
+ "inv_live", {"stream": "control", "exit_code": 0, "duration_ms": 1}
+ )
+ broker.cleanup_invocation("inv_live")
+
+ assert "inv_live" not in sess.invocations
+ assert broker.get_invocation("inv_live") is None
+
+
+def test_cleanup_keeps_other_pending_invocations():
+ # Cleaning one invocation must leave a sibling pending invocation intact.
+ broker = DeviceBroker()
+ inv1 = broker.dispatch_invocation(
+ "dev_y", "user_y", {"invocation_id": "inv_1", "action": "run_command"}
+ )
+ inv2 = broker.dispatch_invocation(
+ "dev_y", "user_y", {"invocation_id": "inv_2", "action": "run_command"}
+ )
+ assert broker._pending.get("dev_y") == [inv1, inv2]
+
+ broker.cleanup_invocation(inv1.invocation_id)
+ assert broker._pending.get("dev_y") == [inv2]
+
+ sess = broker.register_session("dev_y", "user_y")
+ assert "inv_2" in sess.invocations
+ assert "inv_1" not in sess.invocations
diff --git a/tests/devices/test_broker_drain.py b/tests/devices/test_broker_drain.py
new file mode 100644
index 00000000..16dc784f
--- /dev/null
+++ b/tests/devices/test_broker_drain.py
@@ -0,0 +1,51 @@
+"""Tests for ``DeviceBroker.drain_output`` deadline handling."""
+
+from __future__ import annotations
+
+import time
+
+from application.devices.broker import DeviceBroker, INVOCATION_DONE
+
+
+def _dispatch(broker: DeviceBroker):
+ envelope = {"invocation_id": "inv_test", "action": "run_command"}
+ return broker.dispatch_invocation("dev_x", "user_x", envelope)
+
+
+def test_drain_output_returns_after_deadline_with_no_output():
+ # Device never connects / posts output and ``completed`` is never set:
+ # the generator must stop once the deadline passes instead of looping.
+ broker = DeviceBroker()
+ inv = _dispatch(broker)
+ deadline = time.time() + 0.2
+ start = time.time()
+ chunks = list(
+ broker.drain_output(inv.invocation_id, timeout=0.05, deadline=deadline)
+ )
+ elapsed = time.time() - start
+ assert chunks == []
+ assert not inv.completed.is_set()
+ assert elapsed < 2.0
+
+
+def test_drain_output_yields_then_stops_on_done():
+ # Sanity: a normal stream still drains and stops on INVOCATION_DONE,
+ # with a generous deadline that should not fire.
+ broker = DeviceBroker()
+ inv = _dispatch(broker)
+ inv.output_queue.put({"stream": "stdout", "chunk": "hello"})
+ inv.output_queue.put(INVOCATION_DONE)
+ deadline = time.time() + 5.0
+ chunks = list(
+ broker.drain_output(inv.invocation_id, timeout=0.05, deadline=deadline)
+ )
+ assert chunks == [{"stream": "stdout", "chunk": "hello"}]
+
+
+def test_drain_output_no_deadline_stops_when_completed():
+ # Without a deadline, the completed event still terminates the loop.
+ broker = DeviceBroker()
+ inv = _dispatch(broker)
+ inv.completed.set()
+ chunks = list(broker.drain_output(inv.invocation_id, timeout=0.05))
+ assert chunks == []
diff --git a/tests/devices/test_denylist.py b/tests/devices/test_denylist.py
new file mode 100644
index 00000000..17e9d390
--- /dev/null
+++ b/tests/devices/test_denylist.py
@@ -0,0 +1,212 @@
+"""Tests for application.devices.denylist."""
+
+from application.devices.denylist import check_denylist
+
+
+def test_rm_rf_slash():
+ assert check_denylist("rm -rf /") == "rm -rf /"
+
+
+def test_rm_rf_short_flag_order():
+ assert check_denylist("rm -fr /") == "rm -rf /"
+
+
+def test_rm_rf_slash_star():
+ assert check_denylist("rm -rf /*") == "rm -rf /"
+
+
+def test_rm_rf_long_flags():
+ assert check_denylist("rm --recursive --force /") == "rm -rf /"
+
+
+def test_rm_rf_no_preserve_root():
+ assert check_denylist("rm -rf --no-preserve-root /") == "rm -rf /"
+
+
+def test_rm_rf_no_preserve_root_before_flags():
+ assert check_denylist("rm --no-preserve-root -rf /") == "rm -rf /"
+
+
+def test_rm_rf_no_preserve_root_slash_star():
+ assert check_denylist("rm -rf --no-preserve-root /*") == "rm -rf /"
+
+
+def test_rm_rf_long_flags_with_no_preserve_root():
+ assert (
+ check_denylist("rm --recursive --no-preserve-root --force /")
+ == "rm -rf /"
+ )
+
+
+def test_rm_separated_flags_r_then_f():
+ # ``rm -r -f /`` (flags as separate tokens) must still trip.
+ assert check_denylist("rm -r -f /") == "rm -rf /"
+
+
+def test_rm_separated_flags_f_then_r():
+ assert check_denylist("rm -f -r /") == "rm -rf /"
+
+
+def test_rm_rfv_bundle_root():
+ assert check_denylist("rm -rfv /") == "rm -rf /"
+
+
+def test_rm_short_recursive_long_force():
+ assert check_denylist("rm -r --force /") == "rm -rf /"
+
+
+def test_rm_long_recursive_short_force_slash_star():
+ assert check_denylist("rm --recursive -f /*") == "rm -rf /"
+
+
+def test_rm_rf_tmp_subpath_safe():
+ assert check_denylist("rm -rf /tmp/foo") is None
+
+
+def test_rm_rf_relative_build_safe():
+ assert check_denylist("rm -rf ./build") is None
+
+
+def test_rm_rf_home_subpath_safe():
+ assert check_denylist("rm -rf /home/x") is None
+
+
+def test_rm_recursive_only_safe():
+ # Recursive without force isn't the catastrophic root-wipe form.
+ assert check_denylist("rm -r /tmp") is None
+
+
+def test_rm_force_only_slash_safe():
+ # Force without recursive (``rm -f /``) fails on a non-empty dir; not denied.
+ assert check_denylist("rm -f /") is None
+
+
+def test_rm_force_only_etc_safe():
+ # Force without recursive on a real path is not the root-wipe form.
+ assert check_denylist("rm -f /etc/hosts") is None
+
+
+def test_rm_separated_flags_subpath_safe():
+ # Separated recursive+force on a subpath stays safe.
+ assert check_denylist("rm -r -f /tmp/x") is None
+
+
+def test_rm_wrapped_timeout_separated_flags():
+ # ``timeout 5 rm -r -f /`` — wrapper + separated flags must still trip.
+ assert check_denylist("timeout 5 rm -r -f /") == "rm -rf /"
+
+
+def test_rm_wrapped_nice_separated_flags():
+ assert check_denylist("nice rm -f -r /") == "rm -rf /"
+
+
+def test_rm_wrapped_nohup_bundle():
+ assert check_denylist("nohup rm -rf /") == "rm -rf /"
+
+
+def test_rm_wrapped_timeout_duration_slash_star():
+ # ``timeout 30s rm -rf /*`` — duration token skipped, root-star target.
+ assert check_denylist("timeout 30s rm -rf /*") == "rm -rf /"
+
+
+def test_wrapped_safe_ls_not_denied():
+ # ``timeout 5 ls`` is harmless even though it's wrapped.
+ assert check_denylist("timeout 5 ls") is None
+
+
+def test_wrapped_rm_subpath_safe():
+ # ``nice rm -rf /tmp/foo`` is a subpath wipe; stays safe.
+ assert check_denylist("nice rm -rf /tmp/foo") is None
+
+
+def test_rm_rf_home_tilde():
+ assert check_denylist("rm -rf ~") == "rm -rf ~"
+
+
+def test_rm_rf_home_var():
+ assert check_denylist("rm -rf $HOME") == "rm -rf $HOME"
+
+
+def test_fork_bomb():
+ assert check_denylist(":(){:|:&};:") == "fork bomb"
+
+
+def test_fork_bomb_with_spaces():
+ assert check_denylist(":(){ :|:& };:") == "fork bomb"
+
+
+def test_dd_to_block_device():
+ assert (
+ check_denylist("dd if=/dev/zero of=/dev/sda")
+ == "dd to block device"
+ )
+
+
+def test_dd_with_random_to_nvme():
+ assert (
+ check_denylist("dd if=/dev/random of=/dev/nvme0n1")
+ == "dd to block device"
+ )
+
+
+def test_mkfs():
+ assert check_denylist("mkfs.ext4 /dev/sda1") == "mkfs"
+
+
+def test_mkfs_bare():
+ assert check_denylist("mkfs /dev/sda1") == "mkfs"
+
+
+def test_shutdown():
+ assert check_denylist("shutdown -h now") == "shutdown"
+
+
+def test_halt():
+ assert check_denylist("halt") == "halt"
+
+
+def test_poweroff():
+ assert check_denylist("poweroff") == "poweroff"
+
+
+def test_init_0():
+ assert check_denylist("init 0") == "init 0/6"
+
+
+def test_init_6():
+ assert check_denylist("init 6") == "init 0/6"
+
+
+def test_init_other_levels_safe():
+ # ``init 3`` shouldn't trigger.
+ assert check_denylist("init 3") is None
+
+
+def test_git_push_force_long():
+ assert check_denylist("git push --force origin main") == "git push --force"
+
+
+def test_git_push_force_with_lease_safe():
+ # --force-with-lease is allowed.
+ assert check_denylist("git push --force-with-lease origin main") is None
+
+
+def test_git_push_short_flag():
+ assert check_denylist("git push -f origin main") == "git push --force"
+
+
+def test_git_push_mirror():
+ assert check_denylist("git push --mirror origin") == "git push --force"
+
+
+def test_safe_command_no_match():
+ assert check_denylist("ls -la /tmp") is None
+
+
+def test_compound_with_dangerous_segment():
+ # ``echo safe && rm -rf /`` must still trip.
+ assert check_denylist("echo safe && rm -rf /") == "rm -rf /"
+
+
+def test_empty_command():
+ assert check_denylist("") is None
diff --git a/tests/devices/test_me_endpoint.py b/tests/devices/test_me_endpoint.py
new file mode 100644
index 00000000..bb42de86
--- /dev/null
+++ b/tests/devices/test_me_endpoint.py
@@ -0,0 +1,123 @@
+"""Tests for ``GET /api/devices/me``.
+
+Exercises the route handler directly through a Flask test-request context
+so we can mock ``DevicesRepository.find_by_token_hash`` without spinning
+up Postgres.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+import pytest
+from flask import Flask
+
+from application.api.devices import auth as auth_module
+from application.api.devices import session as session_module
+
+
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _device_row() -> dict:
+ return {
+ "id": "dev_abc",
+ "user_id": "user_xyz",
+ "name": "laptop",
+ "hostname": "laptop.local",
+ "os": "linux",
+ "arch": "amd64",
+ "status": "active",
+ "approval_mode": "full",
+ "description": "primary desktop",
+ "paired_at": datetime(2026, 5, 26, 14, 23, tzinfo=timezone.utc),
+ "last_seen_at": datetime(2026, 5, 26, 14, 30, tzinfo=timezone.utc),
+ "machine_pubkey_fingerprint": "fpfpfp",
+ "token_hash": "tokhash",
+ }
+
+
+def _call_me(app: Flask, token: str | None):
+ headers = {}
+ if token is not None:
+ headers["Authorization"] = f"Bearer {token}"
+ with app.test_request_context(
+ "/api/devices/me", method="GET", headers=headers,
+ ):
+ return session_module.me()
+
+
+def test_me_returns_device_record_on_valid_token(app):
+ """Valid token returns the device's own public fields, datetimes ISO."""
+ row = _device_row()
+
+ class _Repo:
+ def __init__(self, _conn):
+ self._row = row
+
+ def find_by_token_hash(self, _token_hash):
+ return self._row
+
+ def touch_last_seen(self, _device_id):
+ pass
+
+ class _Ctx:
+ def __enter__(self):
+ return None
+
+ def __exit__(self, *a):
+ return False
+
+ with patch.object(auth_module, "DevicesRepository", _Repo), \
+ patch.object(auth_module, "db_readonly", _Ctx), \
+ patch.object(auth_module, "db_session", _Ctx):
+ response = _call_me(app, "tok_good")
+ assert response.status_code == 200
+ payload = response.get_json()
+ assert payload["id"] == "dev_abc"
+ assert payload["name"] == "laptop"
+ assert payload["hostname"] == "laptop.local"
+ assert payload["os"] == "linux"
+ assert payload["status"] == "active"
+ assert payload["approval_mode"] == "full"
+ assert payload["description"] == "primary desktop"
+ assert payload["paired_at"] == "2026-05-26T14:23:00+00:00"
+ assert payload["last_seen_at"] == "2026-05-26T14:30:00+00:00"
+ # Internal fields never leak.
+ assert "token_hash" not in payload
+ assert "machine_pubkey_fingerprint" not in payload
+ assert "user_id" not in payload
+
+
+def test_me_rejects_missing_authorization(app):
+ """No Authorization header -> 401 missing_token, route handler not entered."""
+ response = _call_me(app, token=None)
+ assert response.status_code == 401
+ assert response.get_json()["error"] == "missing_token"
+
+
+def test_me_rejects_invalid_token(app):
+ """Unknown token -> 401 invalid_token."""
+
+ class _Repo:
+ def __init__(self, _conn):
+ pass
+
+ def find_by_token_hash(self, _token_hash):
+ return None
+
+ class _Ctx:
+ def __enter__(self):
+ return None
+
+ def __exit__(self, *a):
+ return False
+
+ with patch.object(auth_module, "DevicesRepository", _Repo), \
+ patch.object(auth_module, "db_readonly", _Ctx):
+ response = _call_me(app, "tok_bogus")
+ assert response.status_code == 401
+ assert response.get_json()["error"] == "invalid_token"
diff --git a/tests/devices/test_normalizer.py b/tests/devices/test_normalizer.py
new file mode 100644
index 00000000..f77a9b8f
--- /dev/null
+++ b/tests/devices/test_normalizer.py
@@ -0,0 +1,81 @@
+"""Tests for application.devices.normalizer."""
+
+from application.devices.normalizer import normalize_command, normalize_segment
+
+
+def test_normalize_git_checkout():
+ assert normalize_segment("git checkout main -- file.txt") == "git checkout *"
+
+
+def test_normalize_npm_install():
+ assert normalize_segment("npm install foo bar") == "npm install *"
+
+
+def test_normalize_ls_with_args():
+ assert normalize_segment("ls -la /tmp") == "ls *"
+
+
+def test_normalize_cat_path():
+ assert normalize_segment("cat /etc/passwd") == "cat *"
+
+
+def test_normalize_no_args():
+ assert normalize_segment("pwd") == "pwd"
+
+
+def test_normalize_two_word_no_args():
+ # ``git status`` has no additional args; pattern is just ``git status``.
+ assert normalize_segment("git status") == "git status"
+
+
+def test_normalize_empty():
+ assert normalize_segment("") == ""
+
+
+def test_normalize_command_compound_joins_all_segments():
+ # All segments normalized and joined; not just the first.
+ assert normalize_command("ls -la && rm -rf /tmp") == "ls * && rm *"
+
+
+def test_normalize_command_compound_distinguishes_tail():
+ # Sticky for ``ls /tmp && whoami`` must not match ``ls /tmp && rm /tmp/x``.
+ assert normalize_command("ls /tmp && whoami") == "ls * && whoami"
+ assert (
+ normalize_command("ls /tmp && rm /tmp/x")
+ != normalize_command("ls /tmp && whoami")
+ )
+
+
+def test_normalize_command_single_segment_unchanged():
+ # A non-compound command yields just its own pattern (no `` && ``).
+ assert normalize_command("ls -la /tmp") == "ls *"
+
+
+def test_normalize_command_skips_empty_segments():
+ # Trailing/empty connector segments are dropped, not joined as blanks.
+ assert normalize_command("ls -la && ") == "ls *"
+
+
+def test_normalize_command_none_when_empty():
+ assert normalize_command("") is None
+
+
+def test_normalize_command_none_when_only_connectors():
+ assert normalize_command("&& ;") is None
+
+
+def test_normalize_quoted_args():
+ assert normalize_segment('echo "hello world"') == "echo *"
+
+
+def test_normalize_docker_run():
+ assert normalize_segment("docker run -it nginx") == "docker run *"
+
+
+def test_normalize_kubectl_apply():
+ assert normalize_segment("kubectl apply -f deploy.yaml") == "kubectl apply *"
+
+
+def test_normalize_known_two_word_minimal():
+ # Just ``npm`` with no subcmd.
+ assert normalize_segment("npm") == "npm"
diff --git a/tests/devices/test_pairing_create.py b/tests/devices/test_pairing_create.py
new file mode 100644
index 00000000..97a81c81
--- /dev/null
+++ b/tests/devices/test_pairing_create.py
@@ -0,0 +1,283 @@
+"""Tests for the create-pairing endpoint's optional preset fields.
+
+Covers the friction-reducer A change: ``POST /api/devices/pairings`` now
+accepts ``name``, ``description``, and ``approval_mode`` in the body and
+stashes them in Redis so ``redeem_pairing`` can apply them to the new
+``devices`` row. These tests exercise the function directly (no Flask
+test client) because the broader pairing flow needs Redis + Postgres
+fixtures that aren't wired into this module yet.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Optional
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from flask import Flask
+
+from application.api.devices import pairing as pairing_module
+
+
+class _StubRedis:
+ """Minimal Redis stand-in capturing setex payloads.
+
+ ``eval`` re-implements the redeem-claim Lua: read the pairing JSON and,
+ only if ``status == 'pending'``, flip it to ``redeemed`` (returning 1),
+ else return 0. This lets the redeem tests exercise the atomic path.
+ """
+
+ def __init__(self) -> None:
+ self.setex_calls: list[tuple[str, int, str]] = []
+ self.store: dict[str, str] = {}
+
+ def setex(self, key: str, ttl: int, value: str) -> None:
+ self.setex_calls.append((key, ttl, value))
+ self.store[key] = value
+
+ def get(self, key: str) -> Optional[str]:
+ return self.store.get(key)
+
+ def delete(self, key: str) -> None:
+ self.store.pop(key, None)
+
+ def eval(self, _script: str, _numkeys: int, key: str, *_args) -> int:
+ raw = self.store.get(key)
+ if raw is None:
+ return 0
+ try:
+ state = json.loads(raw)
+ except Exception:
+ return 0
+ if state.get("status") != "pending":
+ return 0
+ state["status"] = "redeemed"
+ self.store[key] = json.dumps(state)
+ return 1
+
+
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _call_create(app: Flask, body: dict, redis: _StubRedis):
+ with app.test_request_context(
+ "/api/devices/pairings",
+ method="POST",
+ json=body,
+ ):
+ from flask import request as flask_request
+
+ flask_request.decoded_token = {"sub": "user_abc"}
+ with patch.object(pairing_module, "_redis", return_value=redis):
+ return pairing_module.create_pairing()
+
+
+def _stashed_state(redis: _StubRedis) -> dict:
+ assert redis.setex_calls, "no redis writes recorded"
+ # The first setex is the device_code -> state hash; the second is the
+ # user_code -> device_code index.
+ raw = redis.setex_calls[0][2]
+ return json.loads(raw)
+
+
+def test_create_pairing_stashes_name_description_and_mode(app):
+ redis = _StubRedis()
+ response = _call_create(
+ app,
+ {
+ "name": " laptop ",
+ "description": " desk machine ",
+ "approval_mode": "full",
+ },
+ redis,
+ )
+ payload = response.get_json()
+ assert response.status_code == 200
+ assert payload["user_code"], "expected a user_code in the response"
+ state = _stashed_state(redis)
+ assert state["requested_name"] == "laptop"
+ assert state["requested_description"] == "desk machine"
+ assert state["requested_approval_mode"] == "full"
+
+
+def test_create_pairing_defaults_when_body_omitted(app):
+ redis = _StubRedis()
+ response = _call_create(app, {}, redis)
+ assert response.status_code == 200
+ state = _stashed_state(redis)
+ assert state["requested_name"] is None
+ assert state["requested_description"] is None
+ assert state["requested_approval_mode"] is None
+
+
+@pytest.mark.parametrize("mode", ["bogus", "writes-only", "never"])
+def test_create_pairing_rejects_invalid_approval_mode(app, mode):
+ redis = _StubRedis()
+ response = _call_create(app, {"approval_mode": mode}, redis)
+ assert response.status_code == 400
+ body = response.get_json()
+ assert body["error"] == "invalid_approval_mode"
+ assert not redis.setex_calls
+
+
+@pytest.mark.parametrize("bad_name", [123, 12.5, True, ["x"], {"a": 1}])
+def test_create_pairing_rejects_non_string_name(app, bad_name):
+ # A non-string ``name`` must 400 instead of crashing on ``.strip()``.
+ redis = _StubRedis()
+ response = _call_create(app, {"name": bad_name}, redis)
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_name"
+ assert not redis.setex_calls
+
+
+def test_create_pairing_blank_strings_normalize_to_none(app):
+ redis = _StubRedis()
+ response = _call_create(
+ app,
+ {"name": " ", "description": ""},
+ redis,
+ )
+ assert response.status_code == 200
+ state = _stashed_state(redis)
+ assert state["requested_name"] is None
+ assert state["requested_description"] is None
+
+
+class _StubDevicesRepo:
+ def __init__(self) -> None:
+ self.create_args: dict = {}
+
+ def create(self, **kwargs) -> dict:
+ self.create_args = kwargs
+ return {"id": kwargs.get("device_id"), **kwargs}
+
+
+class _StubConn:
+ def __init__(self) -> None:
+ self.executed: list = []
+
+ def execute(self, *args, **kwargs):
+ self.executed.append((args, kwargs))
+ # Return an object with .fetchone()/.rowcount where needed; for
+ # the upsert helper we only need .fetchone() returning None.
+ return MagicMock(fetchone=lambda: None, rowcount=0)
+
+
+def test_redeem_applies_stashed_preferences(app, monkeypatch):
+ """End-to-end: stash via create, then redeem applies the values."""
+ redis = _StubRedis()
+ # Step 1 — create with preset name/description/mode
+ create_response = _call_create(
+ app,
+ {
+ "name": "edgebox",
+ "description": "raspberry pi",
+ "approval_mode": "full",
+ },
+ redis,
+ )
+ create_payload = create_response.get_json()
+ user_code = create_payload["user_code"].replace("-", "")
+ # Step 2 — simulate the CLI hitting redeem.
+ devices_repo = _StubDevicesRepo()
+ monkeypatch.setattr(
+ pairing_module, "DevicesRepository", lambda conn: devices_repo
+ )
+ monkeypatch.setattr(
+ pairing_module,
+ "_upsert_remote_device_user_tool",
+ lambda conn, **kwargs: None,
+ )
+
+ class _Sess:
+ def __enter__(self):
+ return _StubConn()
+
+ def __exit__(self, *args):
+ return False
+
+ monkeypatch.setattr(pairing_module, "db_session", _Sess)
+ monkeypatch.setattr(
+ pairing_module,
+ "fingerprint_pubkey",
+ lambda _pubkey: "fpfpfp",
+ )
+ monkeypatch.setattr(
+ pairing_module,
+ "hash_session_token",
+ lambda _tok: "tokhash",
+ )
+
+ with app.test_request_context(
+ "/api/devices/pairings/redeem",
+ method="POST",
+ json={
+ "user_code": user_code,
+ "hostname": "myhost",
+ "os": "linux",
+ "arch": "amd64",
+ "cli_version": "0.5.0",
+ "machine_pubkey": "Zm9v",
+ },
+ ):
+ with patch.object(pairing_module, "_redis", return_value=redis):
+ response = pairing_module.redeem_pairing()
+ assert response.status_code == 200, response.get_json()
+ args = devices_repo.create_args
+ assert args["approval_mode"] == "full"
+ assert args["description"] == "raspberry pi"
+ # Name prefix is the user-supplied value (collision suffix appended).
+ assert args["name"].startswith("edgebox")
+
+
+def test_redeem_falls_back_to_hostname_and_ask(app, monkeypatch):
+ redis = _StubRedis()
+ create_response = _call_create(app, {}, redis)
+ user_code = create_response.get_json()["user_code"].replace("-", "")
+ devices_repo = _StubDevicesRepo()
+ monkeypatch.setattr(
+ pairing_module, "DevicesRepository", lambda conn: devices_repo
+ )
+ monkeypatch.setattr(
+ pairing_module,
+ "_upsert_remote_device_user_tool",
+ lambda conn, **kwargs: None,
+ )
+
+ class _Sess:
+ def __enter__(self):
+ return _StubConn()
+
+ def __exit__(self, *args):
+ return False
+
+ monkeypatch.setattr(pairing_module, "db_session", _Sess)
+ monkeypatch.setattr(
+ pairing_module, "fingerprint_pubkey", lambda _pubkey: "fpfpfp",
+ )
+ monkeypatch.setattr(
+ pairing_module, "hash_session_token", lambda _tok: "tokhash",
+ )
+
+ with app.test_request_context(
+ "/api/devices/pairings/redeem",
+ method="POST",
+ json={
+ "user_code": user_code,
+ "hostname": "fallback-host",
+ "os": "linux",
+ "arch": "amd64",
+ "machine_pubkey": "Zm9v",
+ },
+ ):
+ with patch.object(pairing_module, "_redis", return_value=redis):
+ response = pairing_module.redeem_pairing()
+ assert response.status_code == 200
+ args = devices_repo.create_args
+ assert args["approval_mode"] == "ask"
+ assert args["description"] is None
+ assert args["name"].startswith("fallback-host")
diff --git a/tests/devices/test_pairing_redeem_atomic.py b/tests/devices/test_pairing_redeem_atomic.py
new file mode 100644
index 00000000..ed3d09c3
--- /dev/null
+++ b/tests/devices/test_pairing_redeem_atomic.py
@@ -0,0 +1,269 @@
+"""Concurrency test: two redeems of one user_code yield exactly one device.
+
+Exercises the real Lua claim against the local dev Redis (127.0.0.1:6379).
+Skips cleanly if Redis isn't reachable so CI without Redis still passes.
+The DB layer is stubbed; we only count how many device rows get created.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+from sqlalchemy.exc import IntegrityError
+
+from application.api.devices import pairing as pairing_module
+
+
+def _redis_or_skip():
+ redis = pytest.importorskip("redis")
+ try:
+ client = redis.Redis(host="127.0.0.1", port=6379, socket_connect_timeout=1)
+ client.ping()
+ except Exception: # noqa: BLE001 - any connection failure -> skip
+ pytest.skip("local Redis (127.0.0.1:6379) not available")
+ return client
+
+
+class _CountingDevicesRepo:
+ creates = 0
+
+ def __init__(self, conn):
+ pass
+
+ def create(self, **kwargs):
+ type(self).creates += 1
+ return {"id": kwargs.get("device_id"), **kwargs}
+
+
+class _StubConn:
+ def execute(self, *args, **kwargs):
+ return MagicMock(fetchone=lambda: None, rowcount=0)
+
+
+class _Sess:
+ def __enter__(self):
+ return _StubConn()
+
+ def __exit__(self, *args):
+ return False
+
+
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _seed_pairing(redis_client, user_id: str) -> str:
+ """Write a pending pairing + user-code index; return the raw user_code."""
+ device_code = f"dc_{uuid.uuid4().hex}"
+ user_code = "".join(
+ pairing_module.secrets.choice(pairing_module._USER_CODE_ALPHABET)
+ for _ in range(pairing_module._USER_CODE_LEN)
+ )
+ state = {
+ "device_code": device_code,
+ "user_code": pairing_module._format_user_code(user_code),
+ "user_code_raw": user_code,
+ "user_id": user_id,
+ "status": "pending",
+ "created_at": int(time.time()),
+ "requested_name": None,
+ "requested_description": None,
+ "requested_approval_mode": None,
+ }
+ redis_client.setex(
+ pairing_module._PAIRING_REDIS_PREFIX + device_code, 300, json.dumps(state)
+ )
+ redis_client.setex(
+ pairing_module._user_code_index_key(user_code), 300, device_code
+ )
+ return user_code
+
+
+def _redeem_once(app: Flask, redis_client, user_code: str):
+ with app.test_request_context(
+ "/api/devices/pairings/redeem",
+ method="POST",
+ json={
+ "user_code": user_code,
+ "hostname": "concurrent-host",
+ "os": "linux",
+ "arch": "amd64",
+ "machine_pubkey": "Zm9v",
+ },
+ ):
+ with patch.object(pairing_module, "_redis", return_value=redis_client):
+ return pairing_module.redeem_pairing()
+
+
+def test_two_redeems_of_one_code_create_exactly_one_device(app, monkeypatch):
+ redis_client = _redis_or_skip()
+ _CountingDevicesRepo.creates = 0
+ monkeypatch.setattr(pairing_module, "DevicesRepository", _CountingDevicesRepo)
+ monkeypatch.setattr(
+ pairing_module, "_upsert_remote_device_user_tool", lambda conn, **k: None
+ )
+ monkeypatch.setattr(pairing_module, "db_session", _Sess)
+ monkeypatch.setattr(pairing_module, "fingerprint_pubkey", lambda _p: "fpfpfp")
+ monkeypatch.setattr(pairing_module, "hash_session_token", lambda _t: "tokhash")
+
+ user_code = _seed_pairing(redis_client, f"user_{uuid.uuid4().hex}")
+ try:
+ first = _redeem_once(app, redis_client, user_code)
+ second = _redeem_once(app, redis_client, user_code)
+ finally:
+ # Best-effort cleanup of the seeded keys.
+ device_code = redis_client.get(
+ pairing_module._user_code_index_key(user_code)
+ )
+ if device_code:
+ if isinstance(device_code, (bytes, bytearray)):
+ device_code = device_code.decode()
+ redis_client.delete(pairing_module._PAIRING_REDIS_PREFIX + device_code)
+ redis_client.delete(pairing_module._user_code_index_key(user_code))
+
+ statuses = [first.status_code, second.status_code]
+ # Exactly one redeem wins (200); the other is rejected. Sequentially the
+ # winner deletes the user-code index so the loser sees 404; in a true
+ # race the loser loses the atomic claim and sees 409. Either way, no
+ # second device.
+ assert first.status_code == 200, first.get_json()
+ assert second.status_code in (404, 409), second.get_json()
+ assert statuses.count(200) == 1
+ assert _CountingDevicesRepo.creates == 1
+
+
+def _fake_integrity_error() -> IntegrityError:
+ return IntegrityError("INSERT ...", {}, Exception("duplicate key"))
+
+
+class _CollideOnceRepo:
+ """``create`` raises IntegrityError on its first call, then succeeds."""
+
+ calls = 0
+ names: list = []
+
+ def __init__(self, conn):
+ pass
+
+ def create(self, **kwargs):
+ type(self).calls += 1
+ type(self).names.append(kwargs.get("name"))
+ if type(self).calls == 1:
+ raise _fake_integrity_error()
+ return {"id": kwargs.get("device_id"), **kwargs}
+
+
+class _AlwaysCollideRepo:
+ """``create`` always raises IntegrityError (every name collides)."""
+
+ calls = 0
+
+ def __init__(self, conn):
+ pass
+
+ def create(self, **kwargs):
+ type(self).calls += 1
+ raise _fake_integrity_error()
+
+
+def test_redeem_retries_name_collision_then_succeeds(app, monkeypatch):
+ redis_client = _redis_or_skip()
+ _CollideOnceRepo.calls = 0
+ _CollideOnceRepo.names = []
+ monkeypatch.setattr(pairing_module, "DevicesRepository", _CollideOnceRepo)
+ monkeypatch.setattr(
+ pairing_module, "_upsert_remote_device_user_tool", lambda conn, **k: None
+ )
+ monkeypatch.setattr(pairing_module, "db_session", _Sess)
+ monkeypatch.setattr(pairing_module, "fingerprint_pubkey", lambda _p: "fpfpfp")
+ monkeypatch.setattr(pairing_module, "hash_session_token", lambda _t: "tokhash")
+
+ user_code = _seed_pairing(redis_client, f"user_{uuid.uuid4().hex}")
+ try:
+ resp = _redeem_once(app, redis_client, user_code)
+ finally:
+ device_code = redis_client.get(
+ pairing_module._user_code_index_key(user_code)
+ )
+ if device_code:
+ if isinstance(device_code, (bytes, bytearray)):
+ device_code = device_code.decode()
+ redis_client.delete(pairing_module._PAIRING_REDIS_PREFIX + device_code)
+ redis_client.delete(pairing_module._user_code_index_key(user_code))
+
+ # Redeem still succeeds; the second name attempt won.
+ assert resp.status_code == 200, resp.get_json()
+ assert _CollideOnceRepo.calls == 2
+ # The winning name differs from the first (colliding) candidate.
+ assert _CollideOnceRepo.names[0] != _CollideOnceRepo.names[1]
+ assert resp.get_json()["name"] == _CollideOnceRepo.names[1]
+
+
+def test_redeem_releases_claim_when_all_names_collide(app, monkeypatch):
+ redis_client = _redis_or_skip()
+ _AlwaysCollideRepo.calls = 0
+ monkeypatch.setattr(pairing_module, "DevicesRepository", _AlwaysCollideRepo)
+ monkeypatch.setattr(
+ pairing_module, "_upsert_remote_device_user_tool", lambda conn, **k: None
+ )
+ monkeypatch.setattr(pairing_module, "db_session", _Sess)
+ monkeypatch.setattr(pairing_module, "fingerprint_pubkey", lambda _p: "fpfpfp")
+ monkeypatch.setattr(pairing_module, "hash_session_token", lambda _t: "tokhash")
+
+ user_code = _seed_pairing(redis_client, f"user_{uuid.uuid4().hex}")
+ device_code = redis_client.get(
+ pairing_module._user_code_index_key(user_code)
+ )
+ if isinstance(device_code, (bytes, bytearray)):
+ device_code = device_code.decode()
+ try:
+ resp = _redeem_once(app, redis_client, user_code)
+
+ # Every attempt collided -> error returned.
+ assert resp.status_code == 500, resp.get_json()
+ assert _AlwaysCollideRepo.calls == pairing_module._DEVICE_NAME_RETRIES
+
+ # The claim was released: status reverted to pending and the
+ # user-code index restored, so the code can be redeemed again.
+ state = json.loads(
+ redis_client.get(pairing_module._PAIRING_REDIS_PREFIX + device_code)
+ )
+ assert state["status"] == "pending"
+ restored = redis_client.get(
+ pairing_module._user_code_index_key(user_code)
+ )
+ if isinstance(restored, (bytes, bytearray)):
+ restored = restored.decode()
+ assert restored == device_code
+ finally:
+ redis_client.delete(pairing_module._PAIRING_REDIS_PREFIX + device_code)
+ redis_client.delete(pairing_module._user_code_index_key(user_code))
+
+
+def test_claim_pairing_wins_exactly_once():
+ # Direct test of the atomic claim against real Redis: two claims on the
+ # same pending pairing key -> first True, second False. This is the 409
+ # race path (no index deletion involved).
+ redis_client = _redis_or_skip()
+ device_code = f"dc_{uuid.uuid4().hex}"
+ key = pairing_module._PAIRING_REDIS_PREFIX + device_code
+ redis_client.setex(key, 60, json.dumps({"status": "pending", "user_id": "u"}))
+ try:
+ first = pairing_module._claim_pairing(redis_client, device_code)
+ second = pairing_module._claim_pairing(redis_client, device_code)
+ assert first is True
+ assert second is False
+ # TTL preserved (claim must not drop expiry).
+ assert redis_client.ttl(key) > 0
+ # State flipped to redeemed, other fields intact.
+ stored = json.loads(redis_client.get(key))
+ assert stored["status"] == "redeemed"
+ assert stored["user_id"] == "u"
+ finally:
+ redis_client.delete(key)
diff --git a/tests/devices/test_remote_device_decision.py b/tests/devices/test_remote_device_decision.py
new file mode 100644
index 00000000..a5452a90
--- /dev/null
+++ b/tests/devices/test_remote_device_decision.py
@@ -0,0 +1,58 @@
+"""Tests for ``RemoteDeviceTool._decide_approval`` (two-mode model)."""
+
+from __future__ import annotations
+
+import pytest
+
+from application.agents.tools.remote_device import RemoteDeviceTool
+
+
+def _tool(monkeypatch, *, sticky=False):
+ tool = RemoteDeviceTool.__new__(RemoteDeviceTool)
+ tool.device_id = "dev_abc"
+ tool.user_id = "alice"
+ monkeypatch.setattr(tool, "_matches_sticky", lambda command: sticky)
+ return tool
+
+
+@pytest.mark.unit
+class TestDecideApproval:
+ def test_full_passthrough(self, monkeypatch):
+ tool = _tool(monkeypatch)
+ reason, mode = tool._decide_approval({"approval_mode": "full"}, "rm /tmp/x")
+ assert (reason, mode) == ("full_access_passthrough", "full")
+
+ def test_full_denylist_forces_prompt(self, monkeypatch):
+ tool = _tool(monkeypatch)
+ reason, mode = tool._decide_approval({"approval_mode": "full"}, "rm -rf /")
+ assert (reason, mode) == ("denylist_forced_prompt", "ask")
+
+ def test_ask_requires_approval(self, monkeypatch):
+ tool = _tool(monkeypatch, sticky=False)
+ reason, mode = tool._decide_approval({"approval_mode": "ask"}, "ls -la")
+ assert (reason, mode) == ("user_approval_required", "ask")
+
+ def test_ask_sticky_auto_approves(self, monkeypatch):
+ tool = _tool(monkeypatch, sticky=True)
+ reason, mode = tool._decide_approval({"approval_mode": "ask"}, "ls -la")
+ assert (reason, mode) == ("sticky_auto_approve", "full")
+
+ def test_ask_sticky_denylist_still_forces_prompt(self, monkeypatch):
+ # Regression: a "don't ask again" sticky pattern must not let a
+ # denylisted command (e.g. sticky ``rm *`` -> ``rm -rf /``, or sticky
+ # ``git push *`` -> ``git push --force --mirror``) auto-run. The
+ # denylist forces a prompt on every path, including the sticky one.
+ from application.devices.denylist import check_denylist
+
+ assert check_denylist("rm -rf /") is not None
+ assert check_denylist("git push --force --mirror") is not None
+
+ tool = _tool(monkeypatch, sticky=True)
+ for command in ("rm -rf /", "git push --force --mirror"):
+ reason, mode = tool._decide_approval({"approval_mode": "ask"}, command)
+ assert (reason, mode) == ("denylist_forced_prompt", "ask")
+
+ def test_missing_mode_defaults_to_ask(self, monkeypatch):
+ tool = _tool(monkeypatch, sticky=False)
+ reason, mode = tool._decide_approval({}, "ls -la")
+ assert (reason, mode) == ("user_approval_required", "ask")
diff --git a/tests/devices/test_session_ticket.py b/tests/devices/test_session_ticket.py
new file mode 100644
index 00000000..9a235af0
--- /dev/null
+++ b/tests/devices/test_session_ticket.py
@@ -0,0 +1,217 @@
+"""Tests for poll-issued session-ticket enforcement on the SSE upgrade.
+
+Two layers:
+
+* Broker unit tests for ``claim_ticket`` / ``validate_ticket`` (issue, match,
+ mismatch, expiry, absence).
+* Route tests proving the real CLI loop still works: ``/poll`` issues a
+ ticket and ``session_events`` accepts *that* ticket, while a mismatched
+ ticket is rejected with ``410`` before any stream opens.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from flask import Flask
+
+from application.api.devices import auth as auth_module
+from application.api.devices import session as session_module
+from application.devices.broker import DeviceBroker
+
+
+# ---------------------------------------------------------------------------
+# Broker unit tests
+# ---------------------------------------------------------------------------
+def _queue_work(broker: DeviceBroker, device_id: str = "dev_x") -> None:
+ # No active session -> the invocation parks in _pending, which is what
+ # makes claim_ticket hand out a ticket.
+ broker.dispatch_invocation(
+ device_id, "user_x", {"invocation_id": "inv_1", "action": "run_command"}
+ )
+
+
+def test_claim_ticket_none_when_no_work():
+ broker = DeviceBroker()
+ assert broker.claim_ticket("dev_x", 30) is None
+
+
+def test_validate_accepts_issued_ticket():
+ broker = DeviceBroker()
+ _queue_work(broker)
+ ticket = broker.claim_ticket("dev_x", 30)
+ assert ticket is not None
+ assert broker.validate_ticket("dev_x", ticket) is True
+
+
+def test_validate_rejects_wrong_ticket():
+ broker = DeviceBroker()
+ _queue_work(broker)
+ broker.claim_ticket("dev_x", 30)
+ assert broker.validate_ticket("dev_x", "st_not_the_one") is False
+
+
+def test_validate_rejects_when_no_ticket_issued():
+ broker = DeviceBroker()
+ # Never polled -> nothing issued.
+ assert broker.validate_ticket("dev_x", "st_anything") is False
+
+
+def test_validate_rejects_empty_session_id():
+ broker = DeviceBroker()
+ _queue_work(broker)
+ broker.claim_ticket("dev_x", 30)
+ assert broker.validate_ticket("dev_x", "") is False
+
+
+def test_validate_rejects_expired_ticket_and_evicts():
+ broker = DeviceBroker()
+ _queue_work(broker)
+ with patch("application.devices.broker.time.monotonic", return_value=1000.0):
+ ticket = broker.claim_ticket("dev_x", 30)
+ # 31s later the ticket is past its 30s TTL.
+ with patch("application.devices.broker.time.monotonic", return_value=1031.0):
+ assert broker.validate_ticket("dev_x", ticket) is False
+ # Evicted: a second check can't resurrect it.
+ assert broker.validate_ticket("dev_x", ticket) is False
+
+
+def test_register_session_consumes_issued_ticket_as_session_id():
+ # The issued ticket becomes the session_id, so the URL the CLI opens
+ # (= the ticket) matches the live session. This is the invariant the
+ # route enforcement relies on.
+ broker = DeviceBroker()
+ _queue_work(broker)
+ ticket = broker.claim_ticket("dev_x", 30)
+ sess = broker.register_session("dev_x", "user_x")
+ assert sess.session_id == ticket
+
+
+# ---------------------------------------------------------------------------
+# Route tests (poll -> SSE upgrade)
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _device_row() -> dict:
+ return {
+ "id": "dev_route",
+ "user_id": "user_route",
+ "name": "laptop",
+ "status": "active",
+ "approval_mode": "ask",
+ "machine_pubkey_fingerprint": "fp",
+ "token_hash": "tokhash",
+ }
+
+
+class _Repo:
+ def __init__(self, _conn):
+ pass
+
+ def find_by_token_hash(self, _token_hash):
+ return _device_row()
+
+ def touch_last_seen(self, _device_id):
+ pass
+
+
+class _Ctx:
+ def __enter__(self):
+ return None
+
+ def __exit__(self, *a):
+ return False
+
+
+def _patched_auth():
+ return [
+ patch.object(auth_module, "DevicesRepository", _Repo),
+ patch.object(auth_module, "db_readonly", _Ctx),
+ patch.object(auth_module, "db_session", _Ctx),
+ ]
+
+
+def _call(app: Flask, view, path: str, *args):
+ with app.test_request_context(
+ path, method="GET", headers={"Authorization": "Bearer tok_good"}
+ ):
+ ctxs = _patched_auth()
+ for c in ctxs:
+ c.start()
+ try:
+ return view(*args)
+ finally:
+ for c in ctxs:
+ c.stop()
+
+
+def test_poll_to_sse_upgrade_with_issued_ticket_works(app):
+ """The legitimate loop: /poll issues a ticket, session_events accepts it."""
+ broker = DeviceBroker()
+ # Queue work so /poll returns a ticket rather than 202.
+ broker.dispatch_invocation(
+ "dev_route", "user_route",
+ {"invocation_id": "inv_a", "action": "run_command"},
+ )
+ with patch.object(session_module, "get_broker", return_value=broker):
+ poll_resp = _call(app, session_module.poll, "/api/devices/poll")
+ assert poll_resp.status_code == 200
+ payload = poll_resp.get_json()
+ ticket = payload["session_ticket"]
+ assert payload["session_url"] == f"/api/devices/sessions/{ticket}/events"
+ assert payload["expires_in"] == 30
+
+ # CLI opens the exact session_url it was handed.
+ sse_resp = _call(
+ app,
+ session_module.session_events,
+ f"/api/devices/sessions/{ticket}/events",
+ ticket,
+ )
+ try:
+ assert sse_resp.status_code == 200
+ assert sse_resp.mimetype == "text/event-stream"
+ finally:
+ # Close the streamed (stream_with_context) generator so its
+ # deferred app-context teardown runs now instead of leaking into
+ # the next test as a "popped wrong app context" error.
+ sse_resp.close()
+
+
+def test_session_events_rejects_mismatched_ticket(app):
+ """A fabricated/mismatched session_id is 410 Gone, no stream opened."""
+ broker = DeviceBroker()
+ broker.dispatch_invocation(
+ "dev_route", "user_route",
+ {"invocation_id": "inv_b", "action": "run_command"},
+ )
+ with patch.object(session_module, "get_broker", return_value=broker):
+ # Poll issues the real ticket...
+ _call(app, session_module.poll, "/api/devices/poll")
+ # ...but the client opens a different one.
+ resp = _call(
+ app,
+ session_module.session_events,
+ "/api/devices/sessions/st_bogus/events",
+ "st_bogus",
+ )
+ assert resp.status_code == 410
+ assert resp.get_json()["error"] == "session_ticket_invalid"
+
+
+def test_session_events_rejects_when_never_polled(app):
+ """Opening the SSE stream without a prior poll is rejected (410)."""
+ broker = DeviceBroker()
+ with patch.object(session_module, "get_broker", return_value=broker):
+ resp = _call(
+ app,
+ session_module.session_events,
+ "/api/devices/sessions/st_anything/events",
+ "st_anything",
+ )
+ assert resp.status_code == 410
+ assert resp.get_json()["error"] == "session_ticket_invalid"
diff --git a/tests/devices/test_signature.py b/tests/devices/test_signature.py
new file mode 100644
index 00000000..9d41a49d
--- /dev/null
+++ b/tests/devices/test_signature.py
@@ -0,0 +1,42 @@
+"""Canonical signed-payload tests.
+
+The string here MUST match the DocsGPT-cli signer byte-for-byte (see
+``internal/host/identity.go`` ``CanonicalPayload`` and the Go test
+``TestCanonicalPayloadExactString``). If you change the format, change both
+repos together.
+"""
+
+from __future__ import annotations
+
+from application.api.devices.auth import _canonical_payload
+
+
+def test_canonical_payload_empty_body():
+ # sha256("") = e3b0c442...b855
+ got = _canonical_payload("GET", "/api/devices/poll", "1700000000", b"")
+ assert got == (
+ "GET /api/devices/poll 1700000000 "
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+ )
+
+
+def test_canonical_payload_none_body_equals_empty():
+ # The verifier passes request.get_data(); guard against a None slipping in.
+ assert _canonical_payload("GET", "/x", "1", None) == _canonical_payload(
+ "GET", "/x", "1", b""
+ )
+
+
+def test_canonical_payload_with_body_matches_cli():
+ # sha256("hello") = 2cf24dba...9824 — identical to the Go test constant.
+ got = _canonical_payload("POST", "/api/devices/x", "1700000000", b"hello")
+ assert got == (
+ "POST /api/devices/x 1700000000 "
+ "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
+ )
+
+
+def test_canonical_payload_body_changes_hash():
+ a = _canonical_payload("POST", "/x", "1", b'{"decision":"accept"}')
+ b = _canonical_payload("POST", "/x", "1", b'{"decision":"deny"}')
+ assert a != b
diff --git a/tests/devices/test_splitter.py b/tests/devices/test_splitter.py
new file mode 100644
index 00000000..74f8209f
--- /dev/null
+++ b/tests/devices/test_splitter.py
@@ -0,0 +1,75 @@
+"""Tests for application.devices.splitter."""
+
+from application.devices.splitter import head_token, head_tokens, split_command
+
+
+def test_single_segment():
+ assert split_command("ls -la") == ["ls -la"]
+
+
+def test_and_split():
+ assert split_command("ls && rm -rf /") == ["ls", "rm -rf /"]
+
+
+def test_or_split():
+ assert split_command("foo || bar") == ["foo", "bar"]
+
+
+def test_semicolon_split():
+ assert split_command("a; b; c") == ["a", "b", "c"]
+
+
+def test_pipe_split():
+ assert split_command("cat /etc/passwd | grep root") == [
+ "cat /etc/passwd",
+ "grep root",
+ ]
+
+
+def test_background_amp_split():
+ assert split_command("a & b") == ["a", "b"]
+
+
+def test_pipe_amp_split():
+ assert split_command("foo |& bar") == ["foo", "bar"]
+
+
+def test_newline_split():
+ assert split_command("a\nb\nc") == ["a", "b", "c"]
+
+
+def test_empty_command():
+ assert split_command("") == []
+
+
+def test_whitespace_only():
+ assert split_command(" ") == []
+
+
+def test_head_token_simple():
+ assert head_token("ls -la /tmp") == "ls"
+
+
+def test_head_token_wrapper_timeout():
+ assert head_token("timeout 30s ls -la") == "ls"
+
+
+def test_head_token_wrapper_nice():
+ # ``nice -n 10 ls`` should strip ``nice -n 10``.
+ assert head_token("nice -n 10 ls") == "ls"
+
+
+def test_head_token_wrapper_nohup():
+ assert head_token("nohup ls") == "ls"
+
+
+def test_head_token_empty():
+ assert head_token("") == ""
+
+
+def test_head_tokens_compound():
+ assert head_tokens("ls && rm -rf / | grep foo") == ["ls", "rm", "grep"]
+
+
+def test_compound_with_wrappers():
+ assert head_tokens("timeout 5 ls && nohup git push") == ["ls", "git"]
diff --git a/tests/devices/test_update_device.py b/tests/devices/test_update_device.py
new file mode 100644
index 00000000..95af78bb
--- /dev/null
+++ b/tests/devices/test_update_device.py
@@ -0,0 +1,109 @@
+"""Tests for ``PATCH /api/devices/`` body validation.
+
+The validation rejects bad ``name``/``description`` values before any DB
+access, so the 400 cases exercise the handler directly without Postgres.
+A valid PATCH is covered with the repository + session mocked.
+"""
+
+from __future__ import annotations
+
+
+import pytest
+from flask import Flask
+
+from application.api.devices import routes as routes_module
+
+
+@pytest.fixture
+def app():
+ return Flask(__name__)
+
+
+def _call_update(app: Flask, body, device_id: str = "dev_abc"):
+ with app.test_request_context(
+ f"/api/devices/{device_id}", method="PATCH", json=body,
+ ):
+ from flask import request as flask_request
+
+ flask_request.decoded_token = {"sub": "user_abc"}
+ return routes_module.update_device(device_id)
+
+
+def test_patch_rejects_null_name(app):
+ response = _call_update(app, {"name": None})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_name"
+
+
+def test_patch_rejects_blank_name(app):
+ response = _call_update(app, {"name": " "})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_name"
+
+
+def test_patch_rejects_non_string_name(app):
+ response = _call_update(app, {"name": 123})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_name"
+
+
+def test_patch_rejects_null_description(app):
+ response = _call_update(app, {"description": None})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_description"
+
+
+def test_patch_rejects_non_string_description(app):
+ response = _call_update(app, {"description": ["x"]})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_description"
+
+
+def test_patch_still_validates_approval_mode(app):
+ response = _call_update(app, {"approval_mode": "bogus"})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid_approval_mode"
+
+
+def test_patch_empty_body_is_no_fields(app):
+ response = _call_update(app, {})
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "no_fields"
+
+
+def test_patch_valid_name_trimmed_and_persisted(app, monkeypatch):
+ """A valid PATCH trims the name and writes the stripped value."""
+ captured: dict = {}
+
+ class _Repo:
+ def __init__(self, _conn):
+ pass
+
+ def update(self, _device_id, _user_id, fields):
+ captured.update(fields)
+ return True
+
+ def get(self, device_id, user_id=None):
+ return {"id": device_id, "name": fields_name(), "description": ""}
+
+ def fields_name():
+ return captured.get("name")
+
+ class _Sess:
+ def __enter__(self):
+ return None
+
+ def __exit__(self, *a):
+ return False
+
+ monkeypatch.setattr(routes_module, "DevicesRepository", _Repo)
+ monkeypatch.setattr(routes_module, "db_session", _Sess)
+ monkeypatch.setattr(
+ "application.api.devices.pairing._upsert_remote_device_user_tool",
+ lambda conn, **kwargs: None,
+ )
+
+ response = _call_update(app, {"name": " laptop ", "description": " box "})
+ assert response.status_code == 200
+ assert captured["name"] == "laptop"
+ assert captured["description"] == "box"
diff --git a/tests/storage/db/repositories/test_message_events_repository.py b/tests/storage/db/repositories/test_message_events_repository.py
index 0dcbef1f..aa04f123 100644
--- a/tests/storage/db/repositories/test_message_events_repository.py
+++ b/tests/storage/db/repositories/test_message_events_repository.py
@@ -134,6 +134,219 @@ class TestMessageEventsRepository:
"tool_calls": [],
}
+ def test_reconstruct_partial_includes_paused_tool_call(self, pg_conn):
+ """A paused (awaiting_approval) ``tool_call`` event must surface
+ in ``tool_calls`` so a conversation refresh re-renders the
+ approval bar."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "action_name": "run_command",
+ "arguments": {"command": "ls"},
+ "status": "awaiting_approval",
+ "device_id": "dev_abc",
+ },
+ },
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ tc = partial["tool_calls"][0]
+ assert tc["call_id"] == "c1"
+ assert tc["status"] == "awaiting_approval"
+ assert tc["device_id"] == "dev_abc"
+
+ def test_reconstruct_partial_tool_call_completed_replaces_pause(self, pg_conn):
+ """Later events for the same ``call_id`` overlay earlier ones,
+ so ``completed`` displaces ``awaiting_approval`` on refresh."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "awaiting_approval",
+ },
+ },
+ )
+ repo.record(
+ message_id, 1, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "completed",
+ "result": "ok",
+ },
+ },
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ assert partial["tool_calls"][0]["status"] == "completed"
+ assert partial["tool_calls"][0]["result"] == "ok"
+
+ def test_reconstruct_partial_multiple_tool_calls_preserve_order(self, pg_conn):
+ """First-seen order is preserved when multiple calls interleave."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {"type": "tool_call", "data": {"call_id": "c1", "status": "pending"}},
+ )
+ repo.record(
+ message_id, 1, "tool_call",
+ {"type": "tool_call", "data": {"call_id": "c2", "status": "pending"}},
+ )
+ repo.record(
+ message_id, 2, "tool_call",
+ {"type": "tool_call", "data": {"call_id": "c1", "status": "completed"}},
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert [tc["call_id"] for tc in partial["tool_calls"]] == ["c1", "c2"]
+ assert partial["tool_calls"][0]["status"] == "completed"
+ assert partial["tool_calls"][1]["status"] == "pending"
+
+ def test_reconstruct_partial_tool_call_overlays_bulk_tool_calls(self, pg_conn):
+ """Per-call ``tool_call`` events overlay an earlier bulk
+ ``tool_calls`` emit by ``call_id``."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_calls",
+ {
+ "type": "tool_calls",
+ "tool_calls": [
+ {"call_id": "c1", "status": "pending"},
+ {"call_id": "c2", "status": "pending"},
+ ],
+ },
+ )
+ repo.record(
+ message_id, 1, "tool_call",
+ {"type": "tool_call", "data": {"call_id": "c1", "status": "completed"}},
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 2
+ by_id = {tc["call_id"]: tc for tc in partial["tool_calls"]}
+ assert by_id["c1"]["status"] == "completed"
+ assert by_id["c2"]["status"] == "pending"
+
+ def test_reconstruct_partial_tool_call_skips_missing_call_id(self, pg_conn):
+ """Malformed per-call events without ``call_id`` are dropped, not crashing."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {"type": "tool_call", "data": {"status": "pending"}},
+ )
+ repo.record(
+ message_id, 1, "tool_call",
+ {"type": "tool_call", "data": {"call_id": "c1", "status": "completed"}},
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ assert partial["tool_calls"][0]["call_id"] == "c1"
+
+ def test_reconstruct_partial_paused_then_empty_bulk_preserves_overlay(
+ self, pg_conn
+ ):
+ """Realistic pause sequence: per-call ``awaiting_approval`` followed
+ by an empty bulk ``tool_calls: []`` from the classic-agent
+ end-of-turn yield. The empty bulk must not wipe the per-call entry.
+ """
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "awaiting_approval",
+ "device_id": "dev_abc",
+ },
+ },
+ )
+ repo.record(
+ message_id, 1, "tool_calls",
+ {"type": "tool_calls", "tool_calls": []},
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ assert partial["tool_calls"][0]["status"] == "awaiting_approval"
+ assert partial["tool_calls"][0]["device_id"] == "dev_abc"
+
+ def test_reconstruct_partial_paused_then_bulk_other_call_preserves_overlay(
+ self, pg_conn
+ ):
+ """A bulk emit for a different ``call_id`` must merge alongside the
+ per-call awaiting_approval entry rather than wipe it.
+ """
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "awaiting_approval",
+ },
+ },
+ )
+ repo.record(
+ message_id, 1, "tool_calls",
+ {
+ "type": "tool_calls",
+ "tool_calls": [
+ {"call_id": "c2", "status": "completed"},
+ ],
+ },
+ )
+ partial = repo.reconstruct_partial(message_id)
+ by_id = {tc["call_id"]: tc for tc in partial["tool_calls"]}
+ assert set(by_id) == {"c1", "c2"}
+ assert by_id["c1"]["status"] == "awaiting_approval"
+ assert by_id["c2"]["status"] == "completed"
+
+ def test_reconstruct_partial_completed_then_empty_bulk_preserves_overlay(
+ self, pg_conn
+ ):
+ """A completed per-call entry survives a subsequent empty bulk emit."""
+ message_id = _seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id, 0, "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "completed",
+ "result": "ok",
+ },
+ },
+ )
+ repo.record(
+ message_id, 1, "tool_calls",
+ {"type": "tool_calls", "tool_calls": []},
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ assert partial["tool_calls"][0]["status"] == "completed"
+ assert partial["tool_calls"][0]["result"] == "ok"
+
def test_reconstruct_partial_skips_malformed_payload(self, pg_conn):
message_id = _seed_message(pg_conn)
repo = MessageEventsRepository(pg_conn)
diff --git a/tests/test_continuation.py b/tests/test_continuation.py
index e48a1cc0..d7304fc3 100644
--- a/tests/test_continuation.py
+++ b/tests/test_continuation.py
@@ -409,6 +409,80 @@ class TestHandlerPauseSignaling:
]
assert len(pause_events) == 1
+ def test_pause_propagates_device_id_for_remote_device(self):
+ """``pause_info['device_id']`` (set in tool_executor for the
+ remote_device tool) must be copied into the emitted ``tool_call``
+ event so the approval UI can render the "don't ask again" button."""
+ handler = ConcreteHandler()
+ agent = self._make_agent()
+ agent.tool_executor.check_pause = Mock(return_value={
+ "call_id": "c1",
+ "name": "run_command_0",
+ "tool_name": "remote_device",
+ "tool_id": "0",
+ "action_name": "run_command",
+ "arguments": {"command": "ls"},
+ "pause_type": "awaiting_approval",
+ "device_id": "dev_abc",
+ "thought_signature": None,
+ })
+
+ call = ToolCall(id="c1", name="run_command_0", arguments='{"command": "ls"}')
+ gen = handler.handle_tool_calls(
+ agent, [call], {"0": {"name": "remote_device"}}, []
+ )
+
+ events = []
+ try:
+ while True:
+ events.append(next(gen))
+ except StopIteration:
+ pass
+
+ pause_events = [
+ e for e in events
+ if e.get("type") == "tool_call"
+ and e.get("data", {}).get("status") == "awaiting_approval"
+ ]
+ assert len(pause_events) == 1
+ assert pause_events[0]["data"].get("device_id") == "dev_abc"
+
+ def test_pause_omits_device_id_for_non_remote_tools(self):
+ """``device_id`` must NOT leak into pause events for tools that
+ don't ship one in ``pause_info``."""
+ handler = ConcreteHandler()
+ agent = self._make_agent()
+ agent.tool_executor.check_pause = Mock(return_value={
+ "call_id": "c1",
+ "name": "send_msg_0",
+ "tool_name": "telegram",
+ "tool_id": "0",
+ "action_name": "send_msg",
+ "arguments": {"text": "hello"},
+ "pause_type": "awaiting_approval",
+ "thought_signature": None,
+ })
+
+ call = ToolCall(id="c1", name="send_msg_0", arguments='{"text": "hello"}')
+ gen = handler.handle_tool_calls(
+ agent, [call], {"0": {"name": "telegram"}}, []
+ )
+
+ events = []
+ try:
+ while True:
+ events.append(next(gen))
+ except StopIteration:
+ pass
+
+ pause_events = [
+ e for e in events
+ if e.get("type") == "tool_call"
+ and e.get("data", {}).get("status") == "awaiting_approval"
+ ]
+ assert len(pause_events) == 1
+ assert "device_id" not in pause_events[0]["data"]
+
def test_mixed_execute_and_pause(self):
"""One tool executes, another needs approval."""
handler = ConcreteHandler()
@@ -989,3 +1063,115 @@ class TestContinuationServiceMarkResuming:
svc = cont_mod.ContinuationService()
# Not a UUID and no legacy row exists.
assert svc.mark_resuming("not-a-uuid", "alice") is False
+
+
+# ---------------------------------------------------------------------------
+# Refresh during pause: per-call ``tool_call`` events must reconstruct
+# into ``tool_calls`` so the approval bar re-renders.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestReconstructPartialToolCallReplay:
+ """The pause path in ``llm/handlers/base.py`` emits a per-call
+ ``tool_call`` event (status ``awaiting_approval``) — not a bulk
+ ``tool_calls`` snapshot. ``reconstruct_partial`` must overlay these
+ so a conversation refresh while paused still shows the approval bar.
+ """
+
+ @staticmethod
+ def _seed_message(conn):
+ from sqlalchemy import text
+
+ user_id = f"user-{uuid.uuid4().hex[:8]}"
+ conv_id = uuid.uuid4()
+ msg_id = uuid.uuid4()
+ conn.execute(
+ text("INSERT INTO users (user_id) VALUES (:u)"),
+ {"u": user_id},
+ )
+ conn.execute(
+ text(
+ "INSERT INTO conversations (id, user_id, name) "
+ "VALUES (:id, :u, 'test')"
+ ),
+ {"id": conv_id, "u": user_id},
+ )
+ conn.execute(
+ text(
+ "INSERT INTO conversation_messages (id, conversation_id, user_id, position) "
+ "VALUES (:id, :c, :u, 0)"
+ ),
+ {"id": msg_id, "c": conv_id, "u": user_id},
+ )
+ return str(msg_id)
+
+ def test_paused_tool_call_event_lands_in_tool_calls(self, pg_conn):
+ from application.storage.db.repositories.message_events import (
+ MessageEventsRepository,
+ )
+
+ message_id = self._seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ # Mirror the exact shape emitted at base.py:941-950 on pause.
+ repo.record(
+ message_id,
+ 0,
+ "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "call_remote_test_1",
+ "action_name": "run_command",
+ "arguments": {"command": "ls -la /tmp"},
+ "status": "awaiting_approval",
+ "device_id": "dev_abc",
+ },
+ },
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ tc = partial["tool_calls"][0]
+ assert tc["status"] == "awaiting_approval"
+ assert tc["device_id"] == "dev_abc"
+ assert tc["call_id"] == "call_remote_test_1"
+
+ def test_completed_event_replaces_paused_event(self, pg_conn):
+ from application.storage.db.repositories.message_events import (
+ MessageEventsRepository,
+ )
+
+ message_id = self._seed_message(pg_conn)
+ repo = MessageEventsRepository(pg_conn)
+ repo.record(
+ message_id,
+ 0,
+ "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "awaiting_approval",
+ },
+ },
+ )
+ repo.record(
+ message_id,
+ 1,
+ "tool_call",
+ {
+ "type": "tool_call",
+ "data": {
+ "tool_name": "remote_device",
+ "call_id": "c1",
+ "status": "completed",
+ "result": "/tmp listing here",
+ },
+ },
+ )
+ partial = repo.reconstruct_partial(message_id)
+ assert len(partial["tool_calls"]) == 1
+ assert partial["tool_calls"][0]["status"] == "completed"
+ assert partial["tool_calls"][0]["result"] == "/tmp listing here"