feat(orchestrator): update a single egress secret in place (0081)

Add a single-secret counterpart to reprovision_from_secret's restore-all:
update ONE egress token for a running bottle without a relaunch, for
refreshing a short-lived host credential (e.g. the Codex access token)
whose launch-time snapshot has expired.

- registry_store.store_agent_secret: per-key upsert (delete+insert of the
  one row), the counterpart of store_agent_secrets' replace-all.
- OrchestratorCore.update_agent_secret: set the in-memory token AND upsert
  the re-encrypted row under the bottle's env_var_secret, leaving other
  tokens untouched.
- POST /bottles/<id>/secret (cli-only) + client.update_agent_secret.

Refs #510, #512

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

bot_bottle/orchestrator/api.py

@@ -50,6 +50,12 @@ class ReprovisionBody(_StrictModel):
     env_var_secret: StrictStr
 
 
+class SecretBody(_StrictModel):
+    name: StrictStr
+    value: StrictStr
+    env_var_secret: StrictStr
+
+
 class ReconcileBody(_StrictModel):
     live_source_ips: list[StrictStr]
     grace_seconds: float | None = None
@@ -259,6 +265,21 @@ def create_app(orch: OrchestratorCore, *, signing_key: str) -> FastAPI:
             return {"reprovisioned": True}
         raise HTTPException(404, "no stored secrets for this bottle")
 
+    @app.post("/bottles/{bottle_id}/secret")
+    def update_secret(bottle_id: str, body: SecretBody) -> dict[str, object]:
+        # Update ONE egress token for a running bottle in place — the
+        # single-secret form of reprovision, for refreshing a short-lived host
+        # credential (e.g. the Codex access token) without a relaunch. cli-only
+        # (not in _GATEWAY_ROUTES): a bottle must never set its own tokens.
+        if orch.update_agent_secret(
+            bottle_id,
+            _required(body.name, "name"),
+            _required(body.value, "value"),
+            _required(body.env_var_secret, "env_var_secret"),
+        ):
+            return {"updated": True}
+        raise HTTPException(404, "no such bottle")
+
     @app.delete("/bottles/{bottle_id}")
     def teardown(bottle_id: str) -> dict[str, object]:
         if orch.teardown_bottle(bottle_id):

bot_bottle/orchestrator/client.py

@@ -184,6 +184,27 @@ class OrchestratorClient:
             )
         return True
 
+    def update_agent_secret(
+        self, bottle_id: str, name: str, value: str, env_var_secret: str,
+    ) -> bool:
+        """Update ONE egress token for a running bottle in place
+        (`POST /bottles/<id>/secret`) — the single-secret form of
+        `reprovision_gateway`, for pushing a freshly-refreshed host credential
+        into a bottle without a relaunch. Returns True on success, False when the
+        orchestrator doesn't know the bottle (404)."""
+        status, _ = self._request(
+            "POST",
+            f"/bottles/{bottle_id}/secret",
+            {"name": name, "value": value, "env_var_secret": env_var_secret},
+        )
+        if status == 404:
+            return False
+        if not 200 <= status < 300:
+            raise OrchestratorClientError(
+                f"update_agent_secret {bottle_id}: HTTP {status}"
+            )
+        return True
+
     def teardown_bottle(self, bottle_id: str) -> bool:
         """Tear a bottle down (`DELETE /bottles/<id>`). False if the
         orchestrator didn't know it (404) — idempotent for cleanup paths."""

bot_bottle/orchestrator/service.py

@@ -379,6 +379,30 @@ class OrchestratorCore:
         self._tokens[bottle_id] = decrypted
         return True
 
+    def update_agent_secret(
+        self, bottle_id: str, name: str, value: str, env_var_secret: str,
+    ) -> bool:
+        """Update ONE egress token for a known bottle in place — the
+        single-secret form of ``reprovision_from_secret``.
+
+        Sets the in-memory token AND upserts the single re-encrypted row under
+        *env_var_secret* (the same key the rest of the rows are encrypted with, so
+        the whole set stays decryptable by a later ``reprovision_from_secret``),
+        leaving every other token untouched. Returns False if the bottle is
+        unknown.
+
+        Unlike ``reprovision_from_secret`` (which restores the values captured at
+        launch), this pushes a *caller-supplied* value — used to refresh a
+        short-lived host credential (e.g. the Codex access token) into a
+        still-running bottle without a relaunch."""
+        from .store.secret_store import encrypt_value
+        if self.registry.get(bottle_id) is None:
+            return False
+        self._tokens.setdefault(bottle_id, {})[name] = value
+        self.registry.store_agent_secret(
+            bottle_id, name, encrypt_value(env_var_secret, value))
+        return True
+
     # --- consolidated gateway ----------------------------------------------
 
     def gateway_status(self) -> dict[str, object]:

bot_bottle/orchestrator/store/registry_store.py

@@ -370,6 +370,30 @@ class RegistryStore(DbStore):
             )
         self._chmod()
 
+    def store_agent_secret(
+        self,
+        bottle_id: str,
+        key: str,
+        encrypted_value: str,
+        secret_type: str = "injected_env_var",
+    ) -> None:
+        """Upsert ONE encrypted secret row (env-var name → ciphertext) for
+        *bottle_id*, leaving the bottle's other secrets untouched — the per-key
+        counterpart of ``store_agent_secrets``' replace-all. Delete-then-insert
+        because the table carries no unique constraint to `ON CONFLICT` against."""
+        with self._connection() as conn:
+            conn.execute(
+                "DELETE FROM bottled_agent_secrets "
+                "WHERE bottled_agent_id = ? AND key = ? AND type = ?",
+                (bottle_id, key, secret_type),
+            )
+            conn.execute(
+                "INSERT INTO bottled_agent_secrets "
+                "(bottled_agent_id, key, value, type) VALUES (?, ?, ?, ?)",
+                (bottle_id, key, encrypted_value, secret_type),
+            )
+        self._chmod()
+
     def get_agent_secrets(
         self,
         bottle_id: str,

tests/unit/test_orchestrator_client.py

@@ -151,6 +151,31 @@ class TestReprovisionGateway(unittest.TestCase):
                 self.c.reprovision_gateway("b1", "key")
 
 
+class TestUpdateAgentSecret(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = OrchestratorClient("http://orch:8080")
+
+    def test_success_posts_single_secret(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(200, {"updated": True})) as opened:
+            self.assertTrue(self.c.update_agent_secret("b1", "A", "fresh", "key"))
+        request = opened.call_args.args[0]
+        self.assertEqual("POST", request.get_method())
+        self.assertTrue(request.full_url.endswith("/bottles/b1/secret"))
+        self.assertEqual(
+            {"name": "A", "value": "fresh", "env_var_secret": "key"},
+            json.loads(request.data),
+        )
+
+    def test_unknown_bottle_is_false(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(404)):
+            self.assertFalse(self.c.update_agent_secret("b1", "A", "v", "key"))
+
+    def test_other_status_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(400)):
+            with self.assertRaises(OrchestratorClientError):
+                self.c.update_agent_secret("b1", "A", "v", "key")
+
+
 class TestHealthAndPolicy(unittest.TestCase):
     def setUp(self) -> None:
         self.c = OrchestratorClient("http://orch:8080")

tests/unit/test_orchestrator_registry_store.py

@@ -199,6 +199,20 @@ class TestAgentSecrets(unittest.TestCase):
         got = self.store.get_agent_secrets("bottle-1")
         self.assertEqual({"K": "new", "K2": "v2"}, got)
 
+    def test_store_agent_secret_upserts_one_key_leaving_others(self) -> None:
+        self.store.store_agent_secrets("bottle-1", {"K": "v1", "K2": "v2"})
+        self.store.store_agent_secret("bottle-1", "K", "v1-new")  # update existing
+        self.store.store_agent_secret("bottle-1", "K3", "v3")     # insert new
+        self.assertEqual(
+            {"K": "v1-new", "K2": "v2", "K3": "v3"},
+            self.store.get_agent_secrets("bottle-1"),
+        )
+
+    def test_store_agent_secret_does_not_duplicate_rows(self) -> None:
+        self.store.store_agent_secret("bottle-1", "K", "v1")
+        self.store.store_agent_secret("bottle-1", "K", "v2")
+        self.assertEqual({"K": "v2"}, self.store.get_agent_secrets("bottle-1"))
+
     def test_delete_removes_secrets(self) -> None:
         self.store.store_agent_secrets("bottle-1", {"K": "v"})
         self.store.delete_agent_secrets("bottle-1")

tests/unit/test_orchestrator_server.py

@@ -125,6 +125,47 @@ class TestDispatch(unittest.TestCase):
             {"EGRESS_TOKEN_0": "upstream-secret"}, self.orch.tokens_for(bottle_id),
         )
 
+    def test_update_agent_secret_in_place(self) -> None:
+        key = base64.urlsafe_b64encode(b"unit-test-key").rstrip(b"=").decode()
+        status, payload = dispatch(
+            self.orch, "POST", "/bottles", _body({
+                "source_ip": "10.243.0.21",
+                "tokens": {"A": "old", "B": "keep"},
+                "env_var_secret": key,
+            }),
+        )
+        self.assertEqual(201, status)
+        bottle_id = payload["bottle_id"]
+        assert isinstance(bottle_id, str)
+        status, response = dispatch(
+            self.orch, "POST", f"/bottles/{bottle_id}/secret",
+            _body({"name": "A", "value": "fresh", "env_var_secret": key}),
+        )
+        self.assertEqual((200, {"updated": True}), (status, response))
+        self.assertEqual({"A": "fresh", "B": "keep"}, self.orch.tokens_for(bottle_id))
+
+    def test_update_agent_secret_validates_and_unknown_bottle(self) -> None:
+        status, _ = dispatch(self.orch, "POST", "/bottles/b1/secret", b"not-json")
+        self.assertEqual(400, status)
+        status, _ = dispatch(
+            self.orch, "POST", "/bottles/b1/secret", _body({"name": "A"}),
+        )
+        self.assertEqual(400, status)  # value + env_var_secret missing
+        status, _ = dispatch(
+            self.orch, "POST", "/bottles/ghost/secret",
+            _body({"name": "A", "value": "v", "env_var_secret": "k"}),
+        )
+        self.assertEqual(404, status)
+
+    def test_update_agent_secret_is_cli_only(self) -> None:
+        # A data-plane (`gateway`) caller must not set a bottle's tokens.
+        status, _ = dispatch(
+            self.orch, "POST", "/bottles/b1/secret",
+            _body({"name": "A", "value": "v", "env_var_secret": "k"}),
+            role=ROLE_GATEWAY,
+        )
+        self.assertEqual(403, status)
+
     def test_reprovision_validates_request_and_missing_rows(self) -> None:
         status, _ = dispatch(
             self.orch, "POST", "/bottles/b1/reprovision_gateway", b"not-json",

tests/unit/test_orchestrator_service.py

@@ -103,6 +103,25 @@ class TestOrchestrator(unittest.TestCase):
         self.assertTrue(self.orch.reprovision_from_secret(rec.bottle_id, key))
         self.assertEqual({"EGRESS_TOKEN_0": "secret"}, self.orch.tokens_for(rec.bottle_id))
 
+    def test_update_agent_secret_refreshes_one_token_in_place(self) -> None:
+        key = new_env_var_secret()
+        rec = self.orch.launch_bottle(
+            "10.243.0.20", tokens={"A": "old-a", "B": "keep-b"}, env_var_secret=key,
+        )
+        self.assertTrue(self.orch.update_agent_secret(rec.bottle_id, "A", "new-a", key))
+        # In memory: A refreshed, B left untouched.
+        self.assertEqual({"A": "new-a", "B": "keep-b"}, self.orch.tokens_for(rec.bottle_id))
+        # At rest: the whole set stays decryptable with the SAME env_var_secret,
+        # so a later reprovision restores the refreshed value (not the launch one).
+        self.orch._tokens.clear()
+        self.assertTrue(self.orch.reprovision_from_secret(rec.bottle_id, key))
+        self.assertEqual({"A": "new-a", "B": "keep-b"}, self.orch.tokens_for(rec.bottle_id))
+
+    def test_update_agent_secret_unknown_bottle_is_false(self) -> None:
+        self.assertFalse(
+            self.orch.update_agent_secret("ghost", "A", "v", new_env_var_secret())
+        )
+
     def test_reprovision_rejects_missing_rows_and_wrong_key(self) -> None:
         self.assertFalse(self.orch.reprovision_from_secret("missing", new_env_var_secret()))
         key = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE"