refactor(gateway): introduce the Gateway service ABC (docker impl)

Replace the docker backend's ad-hoc gateway wiring with the shared `Gateway`
service ABC (in the gateway package), the first of the two per-host service
classes that supersede the per-backend infra glue.

The contract is backend-neutral: `connect_to_orchestrator(url, gateway_token)`
binds the gateway to its control plane and brings it up, `address()` reports
the agent-facing proxy target, `ca_cert_pem()` vends the mitmproxy CA, and
`provisioning_transport()` hands back the exec/cp seam git-gate provisioning
stages per-bottle repos + deploy keys through. `GatewayProvisionError` +
`GatewayTransport` move to the ABC so every backend shares them.

Key trust-model change: the gateway no longer mints its own token. It never
holds the signing key (#469), so the orchestrator mints the role-scoped
`gateway` JWT and hands it in via `connect_to_orchestrator`; the gateway only
injects it. `DockerGateway.ensure_running` becomes `connect_to_orchestrator`
(stashing the URL + token as instance state), and the docker launch flow reads
`address()` / `provisioning_transport()` off the service rather than
re-deriving the container IP + transport itself.

macOS + firecracker gateways move under the ABC in following commits.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/unit/test_orchestrator_gateway.py

@@ -28,6 +28,7 @@ def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
 
 
 _ORCH_URL = "http://orchestrator:9000"
+_TOKEN = "gw-jwt-token"
 
 
 class TestDockerGateway(unittest.TestCase):
@@ -39,18 +40,27 @@ class TestDockerGateway(unittest.TestCase):
         self.addCleanup(use_bottle_root(Path(self._tmp.name)))
         # Resolver-only data plane (PRD 0070): running the gateway requires an
         # orchestrator URL, so the fixture supplies one.
-        self.sc = DockerGateway("bot-bottle-gateway:latest", orchestrator_url=_ORCH_URL)
+        self.sc = DockerGateway("bot-bottle-gateway:latest")
 
     def test_default_name(self) -> None:
         self.assertEqual(GATEWAY_NAME, self.sc.name)
 
-    def test_ensure_running_refuses_without_orchestrator_url(self) -> None:
+    def test_connect_refuses_without_orchestrator_url(self) -> None:
         # No policy source → the data-plane daemons would only crash-loop, so
         # the launch must fail closed with a clear error rather than start one.
         sc = DockerGateway("bot-bottle-gateway:latest")
         with patch(_RUN_DOCKER) as m:
             with self.assertRaises(GatewayError):
-                sc.ensure_running()
+                sc.connect_to_orchestrator("", _TOKEN)
+        m.assert_not_called()
+
+    def test_connect_refuses_without_gateway_token(self) -> None:
+        # The gateway never mints — it must be handed a pre-minted `gateway`
+        # token. Missing one is a misconfiguration, so fail closed before start.
+        sc = DockerGateway("bot-bottle-gateway:latest")
+        with patch(_RUN_DOCKER) as m:
+            with self.assertRaises(GatewayError):
+                sc.connect_to_orchestrator(_ORCH_URL, "")
         m.assert_not_called()
 
     def test_is_running_reads_docker_ps(self) -> None:
@@ -73,7 +83,7 @@ class TestDockerGateway(unittest.TestCase):
             return _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.ensure_running()
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
         self.assertEqual([], [c for c in calls if c[:2] == ["docker", "run"]])
         self.assertEqual([], [c for c in calls if c[:2] == ["docker", "rm"]])
 
@@ -93,7 +103,7 @@ class TestDockerGateway(unittest.TestCase):
             return _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.ensure_running()
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
         self.assertEqual(1, len([c for c in calls if c[:2] == ["docker", "run"]]))
         self.assertTrue(any(c[:2] == ["docker", "rm"] for c in calls))
 
@@ -105,7 +115,7 @@ class TestDockerGateway(unittest.TestCase):
             return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.ensure_running()
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
 
         runs = [c for c in calls if c[:2] == ["docker", "run"]]
         self.assertEqual(1, len(runs))
@@ -132,6 +142,45 @@ class TestDockerGateway(unittest.TestCase):
         # Data plane resolves policy against the orchestrator control plane.
         self.assertIn(f"BOT_BOTTLE_ORCHESTRATOR_URL={_ORCH_URL}", runs[0])
 
+    def test_connect_injects_the_pre_minted_gateway_token(self) -> None:
+        # The gateway presents the token the orchestrator handed it — it never
+        # mints (holds no signing key). The value rides the env (bare `--env
+        # NAME` on argv keeps it off `docker inspect`), not argv itself.
+        from bot_bottle.paths import ORCHESTRATOR_AUTH_JWT_ENV
+
+        run_env: dict[str, str] = {}
+
+        def fake(argv: list[str], **kw: object) -> Mock:
+            if argv[:2] == ["docker", "run"]:
+                env = kw.get("env")
+                if isinstance(env, dict):
+                    run_env.update(env)
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+        self.assertEqual(_TOKEN, run_env[ORCHESTRATOR_AUTH_JWT_ENV])
+
+    def test_address_reads_container_ip_on_the_agent_network(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout="172.18.0.2\n")) as m:
+            self.assertEqual("172.18.0.2", self.sc.address())
+        argv = m.call_args.args[0]
+        self.assertEqual(["docker", "inspect"], argv[:2])
+        self.assertIn(self.sc.name, argv)
+        self.assertTrue(any(self.sc.network in a for a in argv))
+
+    def test_address_raises_when_unassigned(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout="")):
+            with self.assertRaises(GatewayError):
+                self.sc.address()
+
+    def test_provisioning_transport_targets_the_gateway_container(self) -> None:
+        from bot_bottle.backend.docker.gateway_provision import DockerGatewayTransport
+
+        transport = self.sc.provisioning_transport()
+        assert isinstance(transport, DockerGatewayTransport)
+        self.assertEqual(self.sc.name, transport.gateway)
+
     def test_ensure_running_creates_network_when_missing(self) -> None:
         calls: list[list[str]] = []
 
@@ -142,7 +191,7 @@ class TestDockerGateway(unittest.TestCase):
             return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.ensure_running()
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
         creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
         self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
 
@@ -173,7 +222,7 @@ class TestDockerGateway(unittest.TestCase):
             return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.ensure_running()  # network inspect returns 0 → exists
+            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)  # network inspect returns 0 → exists
         self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
 
     def test_ensure_running_raises_on_docker_failure(self) -> None:
@@ -186,7 +235,7 @@ class TestDockerGateway(unittest.TestCase):
 
         with patch(_RUN_DOCKER, side_effect=fake):
             with self.assertRaises(GatewayError):
-                self.sc.ensure_running()
+                self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
 
     def test_stop_is_idempotent_on_missing(self) -> None:
         absent = _proc(returncode=1, stderr="Error: No such container: x")