refactor(remote): close the dispatch_agent_command seam

Migrate all 20 direct agent_registry.send_command call sites (api/servers,
terminal, fleet doctor/repair/fleet incl. the upgrade Thread target,
deployment_runner, survey, tunnel broker/publish, plugins SDK) onto the
shared dispatcher, and widen the ratchet from remote_*_service.py to a
tree-wide AST scan covering builtin-extensions/. The survey test's
transport stub gets the transport's real keyword signature.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/app/api/servers.py

@@ -19,6 +19,7 @@ from app.api._query import apply_query, QueryParseError
 from app.models import User
 from app.models.server import Server, ServerGroup, ServerMetrics, ServerCommand, AgentSession, AgentVersion, AgentRollout
 from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 from app.services.agent_fleet_service import fleet_service
 from app.services.discovery_service import discovery_service
 from app.services import connection_string as connection_string_codec
@@ -808,7 +809,7 @@ def ping_server(server_id):
         })
 
     # Send system:metrics command to get fresh data
-    result = agent_registry.send_command(
+    result = dispatch_agent_command(
         server_id=server_id,
         action='system:metrics',
         timeout=10.0
@@ -2316,7 +2317,7 @@ def trigger_agent_update(server_id):
         }), 503
 
     # Send update command to agent
-    result = agent_registry.send_command(
+    result = dispatch_agent_command(
         server_id=server_id,
         action='agent:update',
         params={
@@ -2688,7 +2689,7 @@ def remote_cloudflared_delete(server_id, tunnel_ref):
 @jwt_required()
 def remote_refresh_capabilities(server_id):
     user_id = get_jwt_identity()
-    result = agent_registry.send_command(
+    result = dispatch_agent_command(
         server_id=server_id, action='agent:recapabilities',
         params={}, user_id=user_id, timeout=20.0,
     )

backend/app/jobs/thread_ownership.py

@@ -62,7 +62,7 @@ THREAD_OWNERSHIP = {
         'lifecycle': LIFECYCLE_BOUNDED_DELIVERY,
         'rationale': 'Low-latency delivery attempt is owned and recoverable by its persisted command row.',
     },
-    'app/services/agent_fleet_service.py:upgrade_servers:agent_registry.send_command': {
+    'app/services/agent_fleet_service.py:upgrade_servers:dispatch_agent_command': {
         'owner': 'agent fleet upgrade',
         'lifecycle': LIFECYCLE_BOUNDED_FANOUT,
         'rationale': 'Parallel delivery fan-out creates persisted agent command state.',

backend/app/plugins_sdk/agents_sdk.py

@@ -103,8 +103,9 @@ class BoundAgents:
                 f"The agent on {server_id} does not support '{requires}'",
                 code='UNSUPPORTED')
 
-        result = registry.send_command(
-            server_id=server_id, action=action, params=params or {},
+        from app.services.remote_command_dispatcher import dispatch_agent_command
+        result = dispatch_agent_command(
+            server_id, action, params=params or {},
             timeout=float(timeout), user_id=user_id,
         ) or {}
 

backend/app/services/agent_fleet_service.py

@@ -15,7 +15,7 @@ from app import db
 from app.models.server import (
     Server, ServerGroup, AgentVersion, ServerCommand, AgentSession, AgentRollout
 )
-from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 
 
 class AgentFleetService:
@@ -119,7 +119,7 @@ class AgentFleetService:
                 continue
 
             threading.Thread(
-                target=agent_registry.send_command,
+                target=dispatch_agent_command,
                 args=(server_id, 'agent:update', params),
                 kwargs={'user_id': user_id, 'timeout': 60.0},
                 daemon=True
@@ -380,7 +380,7 @@ class AgentFleetService:
         app = create_app()
 
         with app.app_context():
-            result = agent_registry.send_command(
+            result = dispatch_agent_command(
                 server_id, action, params, user_id=user_id, timeout=60.0
             )
 

backend/app/services/deployment_runner.py

@@ -11,7 +11,7 @@ from typing import Any, Dict
 
 from app import db
 from app.models.deployment_job import DeploymentJob
-from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 from app.services.docker_service import DockerService
 from app.services.run_log_service import RunLogStream
 from app.services.telemetry_service import TelemetryService, generate_correlation_id
@@ -291,7 +291,7 @@ class DeploymentPlanRunner:
         return result
 
     def _send_agent_command(self, action: str, params: Dict[str, Any], timeout: float = 30.0) -> Any:
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id=self.job.target_server_id,
             action=action,
             params=params,

backend/app/services/fleet_doctor_service.py

@@ -34,6 +34,7 @@ import socket
 from datetime import datetime
 
 from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 from app.services.fleet_sweep import fleet_sweep
 
 logger = logging.getLogger(__name__)
@@ -148,7 +149,7 @@ class FleetDoctorService:
     def _probe_checks(cls, server_id, per_agent_timeout, caps):
         """One ``doctor:probe`` round trip → check rows, or ``None`` when the
         probe errors / returns an unusable payload (caller falls back to v1)."""
-        res = agent_registry.send_command(
+        res = dispatch_agent_command(
             server_id, 'doctor:probe', {'units': list(DOCTOR_UNITS)},
             timeout=per_agent_timeout)
         if not res or not res.get('success'):
@@ -195,7 +196,7 @@ class FleetDoctorService:
         restartable = bool(caps.get(RESTART_CAPABILITY))
         checks = []
         for unit in DOCTOR_UNITS:
-            res = agent_registry.send_command(
+            res = dispatch_agent_command(
                 server_id, 'systemd:status', {'unit': unit},
                 timeout=per_agent_timeout)
             if not res or not res.get('success'):
@@ -214,7 +215,7 @@ class FleetDoctorService:
 
         # Disk headroom from system:metrics (disk_percent = used %).
         disk_row = None
-        res = agent_registry.send_command(
+        res = dispatch_agent_command(
             server_id, 'system:metrics', {}, timeout=per_agent_timeout)
         if res and res.get('success') and isinstance(res.get('data'), dict):
             used = res['data'].get('disk_percent')

backend/app/services/fleet_repair_service.py

@@ -18,6 +18,7 @@ See ``docs/FLEET_CONTRACT.md`` (rule 6).
 import logging
 
 from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 
 logger = logging.getLogger(__name__)
 
@@ -92,7 +93,7 @@ class FleetRepairService:
                              f" (older agent — upgrade to enable remote repair)",
                     'code': 'UNSUPPORTED'}
 
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id,
             spec['action'],
             {spec['param_key']: target},

backend/app/services/survey_service.py

@@ -361,6 +361,7 @@ def run_survey(server_id: str, user_id: Optional[int] = None) -> Tuple[Optional[
     failed/timed-out command → an appropriate error — never a crash.
     """
     from app.services.agent_registry import agent_registry
+    from app.services.remote_command_dispatcher import dispatch_agent_command
 
     if not agent_registry.is_agent_connected(server_id):
         return None, {
@@ -388,7 +389,7 @@ def run_survey(server_id: str, user_id: Optional[int] = None) -> Tuple[Optional[
             'status': 500,
         }
 
-    result = agent_registry.send_command(
+    result = dispatch_agent_command(
         server_id,
         'survey:read',
         {'catalog': catalog},

backend/app/services/terminal_service.py

@@ -5,7 +5,7 @@ import logging
 from typing import Dict, Any, Optional
 from datetime import datetime
 
-from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 
 logger = logging.getLogger(__name__)
 
@@ -37,7 +37,7 @@ class TerminalService:
         session_id = f"term_{uuid.uuid4().hex[:12]}"
 
         # Send command to agent to create terminal
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id=server_id,
             action='terminal:create',
             params={
@@ -92,7 +92,7 @@ class TerminalService:
         if session['user_id'] != user_id:
             return {'success': False, 'error': 'Unauthorized'}
 
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id=session['server_id'],
             action='terminal:input',
             params={
@@ -127,7 +127,7 @@ class TerminalService:
         if session['user_id'] != user_id:
             return {'success': False, 'error': 'Unauthorized'}
 
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id=session['server_id'],
             action='terminal:resize',
             params={
@@ -164,7 +164,7 @@ class TerminalService:
         if session['user_id'] != user_id:
             return {'success': False, 'error': 'Unauthorized'}
 
-        result = agent_registry.send_command(
+        result = dispatch_agent_command(
             server_id=session['server_id'],
             action='terminal:close',
             params={

backend/tests/BASELINE_COUNT

@@ -1 +1 @@
-4116
+4430

backend/tests/test_remote_command_dispatcher.py

@@ -61,7 +61,7 @@ def test_dispatcher_supplies_stable_defaults(monkeypatch):
 
 
 def test_remote_feature_services_use_shared_dispatcher():
-    """New transport calls cannot bypass the single remote command seam."""
+    """Remote feature services must import the seam they are built on."""
     assert REMOTE_SERVICE_FILES, "expected remote feature services"
 
     for path in REMOTE_SERVICE_FILES:
@@ -72,13 +72,60 @@ def test_remote_feature_services_use_shared_dispatcher():
             and any(alias.name == "dispatch_agent_command" for alias in node.names)
             for node in tree.body
         )
-        direct_sends = [
-            node
-            for node in ast.walk(tree)
-            if isinstance(node, ast.Call)
-            and isinstance(node.func, ast.Attribute)
-            and node.func.attr == "send_command"
-        ]
-
         assert imports_dispatcher, f"{path.name} must import the shared dispatcher"
-        assert not direct_sends, f"{path.name} bypasses the shared dispatcher"
+
+
+BACKEND = Path(__file__).parents[1]
+REPO = BACKEND.parent
+
+# The transport itself and the one door allowed to call it.
+SEND_COMMAND_DOOR = {
+    BACKEND / "app" / "services" / "agent_registry.py",
+    BACKEND / "app" / "services" / "remote_command_dispatcher.py",
+}
+
+
+def _py_files(root):
+    for path in sorted(root.rglob("*.py")):
+        if "__pycache__" in path.parts or "node_modules" in path.parts:
+            continue
+        yield path
+
+
+def test_nothing_but_the_door_calls_send_command():
+    """The whole tree, not just remote_* services: every ``*.send_command(...)``
+    outside the dispatcher is a bypass of the shared remote command seam.
+
+    ``agent_registry.send_command`` has exactly one definition, so matching the
+    attribute name alone is precise — no receiver resolution needed. The scan
+    covers ``builtin-extensions/`` too, because the live ``app/plugins`` copies
+    of own-repo extensions are synced from there and are git-ignored.
+    """
+    offenders = []
+    for root in (BACKEND / "app", REPO / "builtin-extensions"):
+        for path in _py_files(root):
+            if path in SEND_COMMAND_DOOR:
+                continue
+            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+            for node in ast.walk(tree):
+                is_call = (
+                    isinstance(node, ast.Call)
+                    and isinstance(node.func, ast.Attribute)
+                    and node.func.attr == "send_command"
+                )
+                # A bare ``target=agent_registry.send_command`` reference (e.g.
+                # a Thread target) is the same bypass without the parentheses.
+                is_ref = (
+                    isinstance(node, ast.Attribute)
+                    and node.attr == "send_command"
+                    and isinstance(node.value, ast.Name)
+                    and node.value.id == "agent_registry"
+                )
+                if is_call or is_ref:
+                    rel = path.relative_to(REPO).as_posix()
+                    offenders.append(f"{rel}:{node.lineno}")
+
+    assert not offenders, (
+        "these call agent transport directly instead of "
+        "dispatch_agent_command():\n  " + "\n  ".join(sorted(set(offenders)))
+    )

backend/tests/test_survey_api.py

@@ -38,7 +38,7 @@ def _patch_agent(monkeypatch, *, connected=True, capable=True, result=None):
     monkeypatch.setattr(reg, 'get_capabilities', lambda sid: {'survey': capable})
     monkeypatch.setattr(
         reg, 'send_command',
-        lambda sid, action, params, user_id=None, timeout=None: (
+        lambda server_id, action, params=None, user_id=None, timeout=None: (
             result if result is not None else {'success': True, 'data': {}}))
 
 

builtin-extensions/serverkit-remote-access/backend/tunnel_broker_service.py

@@ -14,6 +14,7 @@ from app import db
 from app.models.server import Server
 from app.models.tunnel import Tunnel
 from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 from . import tunnel_netutil
 
 logger = logging.getLogger(__name__)
@@ -42,7 +43,7 @@ class TunnelBrokerService:
     def _cmd(server_id, action, params=None, user_id=None, timeout=20.0):
         """Send an agent command; raise TunnelBrokerError on failure.
         Returns the agent handler's ``data`` payload on success."""
-        res = agent_registry.send_command(
+        res = dispatch_agent_command(
             server_id=server_id, action=action, params=params or {},
             user_id=user_id, timeout=timeout,
         )
@@ -243,7 +244,7 @@ class TunnelBrokerService:
         # a disconnected agent (the row is the source of truth; #19 reconciles).
         for sid in (tunnel.edge_server_id, tunnel.private_server_id):
             try:
-                agent_registry.send_command(
+                dispatch_agent_command(
                     server_id=sid, action='wireguard:interface:down',
                     params={'interface': tunnel.interface_name}, user_id=user_id, timeout=10.0,
                 )
@@ -363,7 +364,7 @@ class TunnelBrokerService:
         port is already open, and the create response always carries the
         manual commands too. Returns a small outcome dict for the UI."""
         try:
-            res = agent_registry.send_command(
+            res = dispatch_agent_command(
                 server_id=edge_server_id, action='firewall:allow_port',
                 params={'port': port, 'protocol': 'udp'}, user_id=user_id, timeout=15.0,
             )

builtin-extensions/serverkit-remote-access/backend/tunnel_publish_service.py

@@ -15,7 +15,7 @@ from app import db
 from app.models.exposed_service import ExposedService
 from app.models.server import Server
 from app.models.tunnel import Tunnel
-from app.services.agent_registry import agent_registry
+from app.services.remote_command_dispatcher import dispatch_agent_command
 from app.services.dns_provider_service import DNSProviderService
 from app.services.environment_domain_service import EnvironmentDomainService
 from app.services.nginx_service import NginxService
@@ -93,7 +93,7 @@ class TunnelPublishService:
         try:
             # 1. Forwarder on the private peer (essential): WG IP:port → the
             #    real local service on 127.0.0.1:port.
-            fwd = agent_registry.send_command(
+            fwd = dispatch_agent_command(
                 server_id=tunnel.private_server_id, action='wireguard:forward',
                 params={'interface': tunnel.interface_name, 'listen_ip': tunnel.private_wg_ip,
                         'listen_port': port, 'target_host': '127.0.0.1', 'target_port': port},
@@ -163,7 +163,7 @@ class TunnelPublishService:
         # Stop the forwarder on the private peer (best-effort).
         if tunnel:
             try:
-                agent_registry.send_command(
+                dispatch_agent_command(
                     server_id=tunnel.private_server_id, action='wireguard:unforward',
                     params={'interface': tunnel.interface_name, 'listen_port': svc.port},
                     user_id=user_id, timeout=10.0)