Default the supervise flag to true

Issue #249: bottles should be supervised by default. Rather than
remove the flag (which would make supervision mandatory and is the
wrong plane for cost-control enforcement — see #251), keep the
opt-out and flip the default. Bottles that omit `supervise:` now get
the stuck-recovery sidecar; `supervise: false` still skips it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YcU7nerbg8cVj9R4EkpfLJ

bot_bottle/agent_provider.py

@@ -207,7 +207,8 @@ class AgentProvider(ABC):
     ) -> None:
         """Register the per-bottle supervise sidecar as an MCP server
         in the provider's in-guest config. Called by the backend after
-        the supervise sidecar is reachable."""
+        the supervise sidecar is reachable. No-op when
+        `plan.supervise_plan is None`."""
 
     def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None:
         """Install the egress MITM CA into the agent's trust store.

bot_bottle/backend/__init__.py

@@ -102,7 +102,7 @@ class BottlePlan(ABC):
         over a published host port)."""
         return "git"
     egress_plan: EgressPlan
-    supervise_plan: SupervisePlan
+    supervise_plan: SupervisePlan | None
     agent_provision: AgentProvisionPlan
 
     @property
@@ -332,7 +332,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         )
         agent_provision_plan = merge_provision_env_vars(agent_provision_plan)
         egress_plan = prepare_egress(manifest_bottle, slug, agent_provision_plan)
-        supervise_plan = prepare_supervise(slug)
+        supervise_plan = prepare_supervise(manifest_bottle, slug)
         git_gate_plan = prepare_git_gate(manifest_bottle, slug)
 
         return self._resolve_plan(
@@ -405,7 +405,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
                       agent_provision_plan: AgentProvisionPlan,
                       egress_plan: EgressPlan,
                       git_gate_plan: GitGatePlan,
-                      supervise_plan: SupervisePlan,
+                      supervise_plan: SupervisePlan | None,
                       stage_dir: Path) -> PlanT:
         """Backend-specific plan resolution: image/container names,
         env-file, prompt-file, proxy plan, runtime detection. Called by

bot_bottle/backend/docker/backend.py

@@ -70,7 +70,7 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
         agent_provision_plan: AgentProvisionPlan,
         egress_plan: EgressPlan,
         git_gate_plan: GitGatePlan,
-        supervise_plan: SupervisePlan,
+        supervise_plan: SupervisePlan | None,
         stage_dir: Path,
     ) -> DockerBottlePlan:
         return _resolve_plan.resolve_plan(
@@ -94,6 +94,8 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
         """Docker bottles reach the supervise sidecar via the
         compose-network alias `supervise:9100`. No per-bottle URL
         plumbing needed; the alias resolves inside the bridge."""
+        if plan.supervise_plan is None:
+            return ""
         return f"http://{SUPERVISE_HOSTNAME}:{SUPERVISE_PORT}/"
 
     def prepare_cleanup(self) -> DockerBottleCleanupPlan:

bot_bottle/backend/docker/compose.py

@@ -14,7 +14,7 @@ Conditional services follow the plan content:
   - agent + sidecars bundle: always.
   - git-gate:  iff plan.git_gate_plan.upstreams.
   - egress:    iff plan.egress_plan.routes.
-  - supervise: always (every bottle is supervised, issue #249).
+  - supervise: iff plan.supervise_plan is not None.
 """
 
 from __future__ import annotations
@@ -119,11 +119,13 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
     image, all daemons under a Python init supervisor.
 
     Daemon subset narrows via `BOT_BOTTLE_SIDECAR_DAEMONS` env.
-    egress and supervise are always present; git-gate is conditional.
+    egress is always present; git-gate / supervise are conditional.
     """
-    daemons: list[str] = ["egress", "supervise"]
+    daemons: list[str] = ["egress"]
     if plan.git_gate_plan.upstreams:
         daemons.append("git-gate")
+    if plan.supervise_plan is not None:
+        daemons.append("supervise")
 
     env: list[str] = [f"BOT_BOTTLE_SIDECAR_DAEMONS={','.join(daemons)}"]
     volumes: list[dict[str, Any]] = []
@@ -158,6 +160,7 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
 
     # --- supervise ----------------------------------------------------
     sp = plan.supervise_plan
+    if sp is not None:
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
             f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
@@ -170,9 +173,11 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
             "read_only": False,
         })
 
-    internal_aliases = [EGRESS_HOSTNAME, SUPERVISE_HOSTNAME]
+    internal_aliases = [EGRESS_HOSTNAME]
     if gp.upstreams:
         internal_aliases.append(GIT_GATE_HOSTNAME)
+    if sp is not None:
+        internal_aliases.append(SUPERVISE_HOSTNAME)
 
     service: dict[str, Any] = {
         "image": SIDECAR_BUNDLE_IMAGE,
@@ -226,10 +231,14 @@ def _agent_service(plan: DockerBottlePlan) -> dict[str, Any]:
     if plan.use_runsc:
         service["runtime"] = "runsc"
 
-    service["volumes"] = [_bind(
+    volumes: list[dict[str, Any]] = []
+    if plan.supervise_plan is not None:
+        volumes.append(_bind(
             plan.supervise_plan.current_config_dir,
             CURRENT_CONFIG_DIR_IN_AGENT,
-    )]
+        ))
+    if volumes:
+        service["volumes"] = volumes
 
     # The init supervisor inside the bundle owns intra-bundle
     # daemon ordering, so the agent only waits for the bundle
@@ -245,9 +254,12 @@ def _agent_proxy_url(plan: DockerBottlePlan) -> str:
 
 
 def _agent_no_proxy(plan: DockerBottlePlan) -> str:
-    """NO_PROXY for the agent: loopback plus the supervise hostname
-    (MCP long-poll must bypass the egress proxy)."""
-    hosts = ["localhost", "127.0.0.1", SUPERVISE_HOSTNAME]
+    """NO_PROXY for the agent: loopback always; supervise hostname
+    when the supervise sidecar is up (MCP long-poll must bypass
+    the egress proxy)."""
+    hosts = ["localhost", "127.0.0.1"]
+    if plan.supervise_plan is not None:
+        hosts.append(SUPERVISE_HOSTNAME)
     return ",".join(hosts)
 
 

bot_bottle/backend/docker/launch.py

@@ -130,8 +130,10 @@ def launch(
             mitmproxy_ca_host_path=egress_ca_host,
             mitmproxy_ca_cert_only_host_path=egress_ca_cert_only,
         )
+        supervise_plan = plan.supervise_plan
+        if supervise_plan is not None:
             supervise_plan = dataclasses.replace(
-            plan.supervise_plan,
+                supervise_plan,
                 internal_network=internal_network,
             )
         plan = dataclasses.replace(

bot_bottle/backend/docker/resolve_plan.py

@@ -37,7 +37,7 @@ def resolve_plan(
     resolved_env: ResolvedEnv,
     agent_provision_plan: AgentProvisionPlan,
     egress_plan: EgressPlan,
-    supervise_plan: SupervisePlan,
+    supervise_plan: SupervisePlan | None,
     git_gate_plan: GitGatePlan,
     stage_dir: Path,
 ) -> DockerBottlePlan:

bot_bottle/backend/macos_container/backend.py

@@ -52,7 +52,7 @@ class MacosContainerBottleBackend(
         agent_provision_plan: AgentProvisionPlan,
         egress_plan: EgressPlan,
         git_gate_plan: GitGatePlan,
-        supervise_plan: SupervisePlan,
+        supervise_plan: SupervisePlan | None,
         stage_dir: Path,
     ) -> MacosContainerBottlePlan:
         return _resolve_plan.resolve_plan(

bot_bottle/backend/macos_container/launch.py

@@ -222,6 +222,8 @@ def _stamp_agent_urls(
     sidecar_ip: str,
 ) -> MacosContainerBottlePlan:
     proxy_url = f"http://{sidecar_ip}:{EGRESS_PORT}"
+    supervise_url = ""
+    if plan.supervise_plan is not None:
         supervise_url = f"http://{sidecar_ip}:{SUPERVISE_PORT}/"
     git_gate_url = ""
     if plan.git_gate_plan.upstreams:
@@ -339,9 +341,11 @@ def _sidecar_dns() -> str:
 
 
 def _sidecar_daemons(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
-    daemons = ["egress", "supervise"]
+    daemons = ["egress"]
     if plan.git_gate_plan.upstreams:
         daemons += ["git-gate", "git-http"]
+    if plan.supervise_plan is not None:
+        daemons.append("supervise")
     return tuple(daemons)
 
 
@@ -351,6 +355,7 @@ def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
         env.extend(sorted(plan.egress_plan.token_env_map.keys()))
     if plan.git_gate_plan.upstreams:
         env.append(f"BOT_BOTTLE_GIT_GATE_READY_FILE={_GIT_GATE_READY_FILE}")
+    if plan.supervise_plan is not None:
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
             f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
@@ -378,6 +383,7 @@ def _sidecar_mounts(
         ))
 
     sp = plan.supervise_plan
+    if sp is not None:
         mounts.append((str(sp.queue_dir), QUEUE_DIR_IN_CONTAINER, False))
 
     return tuple(mounts)

bot_bottle/backend/macos_container/resolve_plan.py

@@ -30,7 +30,7 @@ def resolve_plan(
     resolved_env: ResolvedEnv,
     agent_provision_plan: AgentProvisionPlan,
     egress_plan: EgressPlan,
-    supervise_plan: SupervisePlan,
+    supervise_plan: SupervisePlan | None,
     git_gate_plan: GitGatePlan,
     stage_dir: Path,
 ) -> MacosContainerBottlePlan:

bot_bottle/backend/resolve_common.py

@@ -92,9 +92,11 @@ def prepare_egress(
     return Egress().prepare(bottle, slug, egress_dir, provision.egress_routes)
 
 
-def prepare_supervise(slug: str) -> SupervisePlan:
-    """Prepare the supervise sidecar state dir. Every bottle is
-    supervised (issue #249), so this always returns a plan."""
+def prepare_supervise(bottle: ManifestBottle, slug: str) -> SupervisePlan | None:
+    """Prepare the supervise sidecar state dir. Returns None when
+    bottle.supervise is falsy."""
+    if not bottle.supervise:
+        return None
     supervise_dir = supervise_state_dir(slug)
     supervise_dir.mkdir(parents=True, exist_ok=True)
     return Supervise().prepare(slug, supervise_dir)

bot_bottle/backend/smolmachines/backend.py

@@ -62,7 +62,7 @@ class SmolmachinesBottleBackend(
         agent_provision_plan: AgentProvisionPlan,
         egress_plan: EgressPlan,
         git_gate_plan: GitGatePlan,
-        supervise_plan: SupervisePlan,
+        supervise_plan: SupervisePlan | None,
         stage_dir: Path,
     ) -> SmolmachinesBottlePlan:
         return _resolve_plan.resolve_plan(

bot_bottle/backend/smolmachines/launch.py

@@ -206,6 +206,8 @@ def _discover_urls(
         )
         agent_git_gate_host = f"{loopback_ip}:{git_gate_host_port}"
 
+    agent_supervise_url = ""
+    if plan.supervise_plan is not None:
         supervise_host_port = _bundle.bundle_host_port(
             plan.slug, _SUPERVISE_PORT, host_ip=loopback_ip,
         )
@@ -297,14 +299,15 @@ def _bundle_launch_spec(
     """Build a BundleLaunchSpec from the resolved inner Plans.
 
     Daemons in the CSV:
-      - egress and supervise are always present.
+      - egress is always present.
       - git-gate + git-http are conditional on plan.git_gate_plan.upstreams.
+      - supervise is conditional on plan.supervise_plan.
 
     Env + volumes are the union of the sidecar daemons' needs, with
     daemon-private values only (HTTPS_PROXY is scoped to the
     egress process by egress_entrypoint.sh — see PRD 0024's bundle
     bind-address PR)."""
-    daemons: list[str] = ["egress", "supervise"]
+    daemons: list[str] = ["egress"]
     env: list[str] = []
     volumes: list[tuple[str, str, bool]] = []
 
@@ -344,6 +347,8 @@ def _bundle_launch_spec(
 
     # --- supervise --------------------------------------------
     sp = plan.supervise_plan
+    if sp is not None:
+        daemons.append("supervise")
         env += [
             f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
             f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
@@ -354,9 +359,11 @@ def _bundle_launch_spec(
     # Container ports the agent reaches from the smolvm guest —
     # published on host loopback so the guest can dial via TSI +
     # macOS networking. Egress is always the agent's HTTP/HTTPS proxy.
-    ports_to_publish: list[int] = [_EGRESS_PORT, _SUPERVISE_PORT]
+    ports_to_publish: list[int] = [_EGRESS_PORT]
     if gp.upstreams:
         ports_to_publish.append(_GIT_HTTP_PORT)
+    if sp is not None:
+        ports_to_publish.append(_SUPERVISE_PORT)
 
     return _bundle.BundleLaunchSpec(
         slug=plan.slug,

bot_bottle/backend/smolmachines/resolve_plan.py

@@ -52,7 +52,7 @@ def resolve_plan(
     resolved_env: ResolvedEnv,
     agent_provision_plan: AgentProvisionPlan,
     egress_plan: EgressPlan,
-    supervise_plan: SupervisePlan,
+    supervise_plan: SupervisePlan | None,
     git_gate_plan: GitGatePlan,
     stage_dir: Path,
 ) -> SmolmachinesBottlePlan:

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -68,8 +68,7 @@ class BundleLaunchSpec:
     image: str = SIDECAR_BUNDLE_IMAGE
     # Daemon subset CSV for BOT_BOTTLE_SIDECAR_DAEMONS. The
     # supervisor inside the bundle reads it to skip
-    # bottle-irrelevant daemons (e.g. git-gate when a bottle
-    # declares no upstreams).
+    # bottle-irrelevant daemons (e.g. supervise=False bottles).
     daemons_csv: str = "egress"
     # Plain "KEY=VALUE" strings + "KEY" bare names (the bare-name
     # form inherits the value from the docker-run subprocess env,

bot_bottle/contrib/claude/agent_provider.py

@@ -291,6 +291,8 @@ class ClaudeAgentProvider(AgentProvider):
 
         Failure is logged but not fatal — the bottle still works without
         the entry; the operator can register it manually."""
+        if plan.supervise_plan is None:
+            return
         info(f"registering supervise MCP server in agent claude config → {supervise_url}")
         r = bottle.exec(
             f"claude mcp add --scope user --transport http "

bot_bottle/contrib/codex/agent_provider.py

@@ -257,6 +257,8 @@ class CodexAgentProvider(AgentProvider):
 
         Mirrors the Claude provider's `claude mcp add` flow — failure
         is logged but not fatal."""
+        if plan.supervise_plan is None:
+            return
         info(f"registering supervise MCP server in agent codex config → {supervise_url}")
         r = bottle.exec(
             f"codex mcp add {_SUPERVISE_MCP_NAME} --url "

bot_bottle/manifest.py

@@ -19,6 +19,7 @@ Bottle schema (frontmatter):
     repos:      { <name>: <git-gate-entry>, ... }  # optional
   egress: { routes: [ <egress-route>, ... ] }
     # route keys: host, matches, auth, role, dlp
+  supervise:    <bool>                          # optional (default true)
 
 Agent schema (frontmatter):
   bottle:        <bottle-name>          # required
@@ -110,6 +111,13 @@ class ManifestBottle:
     # identity without any git-gate.repos upstreams, and vice versa.
     git_user: ManifestGitUser = field(default_factory=ManifestGitUser)
     egress: ManifestEgressConfig = field(default_factory=ManifestEgressConfig)
+    # Per-bottle stuck-recovery sidecar (PRD 0013). When true (the
+    # default, issue #249), the launch step brings up a supervise
+    # sidecar that exposes MCP tools to the agent (egress-block,
+    # capability-block) plus mounts the current-config dir read-only
+    # into the agent at /etc/bot-bottle/current-config. Set
+    # `supervise: false` to skip the sidecar and mount.
+    supervise: bool = True
 
     @classmethod
     def from_dict(cls, name: str, raw: object) -> "ManifestBottle":
@@ -144,13 +152,6 @@ class ManifestBottle:
                 f"removed. Move it under 'git-gate.user'."
             )
 
-        if "supervise" in d:
-            raise ManifestError(
-                f"bottle '{name}' has a 'supervise' field, which has been "
-                f"removed (issue #249). All bottles are now supervised; the "
-                f"flag was always-on in practice. Delete the field."
-            )
-
         unknown = set(d.keys()) - BOTTLE_KEYS
         if unknown:
             allowed = ", ".join(sorted(BOTTLE_KEYS))
@@ -189,9 +190,16 @@ class ManifestBottle:
             else ManifestEgressConfig()
         )
 
+        supervise_raw = d.get("supervise", True)
+        if not isinstance(supervise_raw, bool):
+            raise ManifestError(
+                f"bottle '{name}' supervise must be a boolean "
+                f"(was {type(supervise_raw).__name__})"
+            )
+
         return cls(
             env=env, agent_provider=agent_provider, git=git,
-            git_user=git_user, egress=egress,
+            git_user=git_user, egress=egress, supervise=supervise_raw,
         )
 
 

bot_bottle/manifest_extends.py

@@ -134,6 +134,9 @@ def _merge_bottles(
         if "agent_provider" in child_raw
         else parent.agent_provider
     )
+    merged_supervise = (
+        child.supervise if "supervise" in child_raw else parent.supervise
+    )
     validate_egress_routes(name, merged_egress.routes)
 
     return ManifestBottle(
@@ -142,6 +145,7 @@ def _merge_bottles(
         git=merged_git,
         git_user=merged_git_user,
         egress=merged_egress,
+        supervise=merged_supervise,
     )
 
 

bot_bottle/manifest_schema.py

@@ -16,7 +16,7 @@ _FILENAME_RX = re.compile(r"^[a-z][a-z0-9-]*$")
 # sets dies with a "did you mean" pointer: typos should not silently
 # ghost into an empty config.
 BOTTLE_KEYS = frozenset(
-    {"env", "extends", "agent_provider", "git-gate", "egress"}
+    {"env", "extends", "agent_provider", "git-gate", "egress", "supervise"}
 )
 AGENT_KEYS_REQUIRED = frozenset({"bottle"})
 AGENT_KEYS_OPTIONAL = frozenset({"skills", "git-gate"})

tests/integration/test_sidecar_bundle_compose.py

@@ -27,13 +27,14 @@ from tests._docker import skip_unless_docker
 
 
 def _manifest() -> ManifestIndex:
-    """Minimal bottle so the bundle exercises egress + supervise
-    (every bottle is supervised, issue #249). Git is off because a
-    meaningful git-gate test needs a real upstream and SSH keys —
-    out of scope for a bundle smoke."""
+    """Bottle with supervise on so the bundle exercises egress +
+    supervise. Git is off because a meaningful git-gate test needs
+    a real upstream and SSH keys — out of scope for a bundle smoke."""
     return ManifestIndex.from_json_obj({
         "bottles": {
-            "dev": {},
+            "dev": {
+                "supervise": True,
+            },
         },
         "agents": {
             "demo": {"skills": [], "prompt": "", "bottle": "dev"},

tests/unit/test_compose.py

@@ -40,11 +40,13 @@ STAGE = Path("/tmp/cb-stage")
 STATE = Path("/tmp/cb-state")
 
 
-def _manifest(*, with_git: bool, with_egress: bool) -> ManifestIndex:
+def _manifest(*, supervise: bool, with_git: bool, with_egress: bool) -> ManifestIndex:
     """Minimal manifest with the toggles the chunk-1 matrix needs.
     The renderer only reads from the plan, not the manifest, so this
     is just here to back BottleSpec."""
     bottle: dict[str, object] = {}
+    if supervise:
+        bottle["supervise"] = True
     if with_git:
         bottle["git-gate"] = {"repos": {
             "upstream": {
@@ -109,11 +111,10 @@ def _plan(
     *,
     with_git: bool = False,
     with_egress: bool = False,
+    supervise: bool = False,
 ) -> DockerBottlePlan:
     """Build a fully-resolved DockerBottlePlan. Toggles cover the
-    matrix the renderer's conditional-service logic branches on.
-    Every bottle is supervised (issue #249), so the supervise plan
-    is always present."""
+    matrix the renderer's conditional-service logic branches on."""
     upstreams: tuple[GitGateUpstream, ...] = ()
     if with_git:
         upstreams = (GitGateUpstream(
@@ -135,7 +136,7 @@ def _plan(
             roles=(),
         ),)
 
-    index = _manifest(with_git=with_git, with_egress=with_egress)
+    index = _manifest(supervise=supervise, with_git=with_git, with_egress=with_egress)
     spec = BottleSpec(
         manifest=index,
         agent_name="demo",
@@ -150,7 +151,7 @@ def _plan(
         forwarded_env={"CLAUDE_CODE_OAUTH_TOKEN": "x"},
         git_gate_plan=_git_gate_plan(upstreams),
         egress_plan=_egress_plan(routes),
-        supervise_plan=_supervise_plan(),
+        supervise_plan=_supervise_plan() if supervise else None,
         use_runsc=False,
         agent_provision=AgentProvisionPlan(
             template="claude",
@@ -219,8 +220,10 @@ class TestAgentAlwaysPresent(unittest.TestCase):
         proxy = [e for e in s["environment"] if e.startswith("HTTPS_PROXY=")][0]
         self.assertEqual("HTTPS_PROXY=http://egress:9099", proxy)
 
-    def test_agent_no_proxy_includes_supervise(self):
-        s = bottle_plan_to_compose(_plan())["services"]["agent"]
+    def test_agent_no_proxy_adds_supervise_when_enabled(self):
+        s = bottle_plan_to_compose(
+            _plan(supervise=True)
+        )["services"]["agent"]
         no_proxy = [e for e in s["environment"] if e.startswith("NO_PROXY=")][0]
         self.assertIn("supervise", no_proxy)
 
@@ -256,18 +259,22 @@ class TestAgentAlwaysPresent(unittest.TestCase):
     def test_agent_depends_only_on_sidecars(self):
         # Bundle shape: the init supervisor owns intra-bundle daemon
         # ordering, so the agent waits on the bundle container alone.
-        for kwargs in [{}, {"with_git": True, "with_egress": True}]:
+        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
 
-    def test_agent_current_config_always_mounted(self):
-        # Every bottle is supervised (issue #249), so the read-only
-        # current-config mount is always present in the agent.
-        agent = bottle_plan_to_compose(_plan())["services"]["agent"]
+    def test_agent_current_config_mount_only_with_supervise(self):
+        with_sv = bottle_plan_to_compose(_plan(supervise=True))["services"]["agent"]
         self.assertTrue(any(
             v["target"] == "/etc/bot-bottle/current-config"
-            for v in agent.get("volumes", [])
+            for v in with_sv.get("volumes", [])
+        ))
+        without_sv = bottle_plan_to_compose(_plan(supervise=False))["services"]["agent"]
+        # Either no volumes key at all, or no current-config target.
+        self.assertFalse(any(
+            v["target"] == "/etc/bot-bottle/current-config"
+            for v in without_sv.get("volumes", [])
         ))
 
 
@@ -285,7 +292,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertEqual({"sidecars", "agent"}, set(spec["services"].keys()))
 
     def test_emits_two_services_full_matrix(self):
-        spec = self._render(with_git=True, with_egress=True)
+        spec = self._render(with_git=True, with_egress=True, supervise=True)
         # Still two services — the bundle absorbs git-gate/egress/supervise.
         self.assertEqual({"sidecars", "agent"}, set(spec["services"].keys()))
 
@@ -308,16 +315,16 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertIn("egress", aliases)
 
     def test_internal_aliases_omit_inactive_sidecars(self):
-        # With no git-gate, that name is NOT aliased — keeps the alias
-        # list honest about what's actually listening inside the bundle.
-        # supervise is always present (issue #249).
+        # With no git-gate / supervise, those names are NOT aliased
+        # — keeps the alias list honest about what's actually
+        # listening inside the bundle.
         sc = self._render()["services"]["sidecars"]
         aliases = set(sc["networks"]["internal"]["aliases"])
         self.assertNotIn("git-gate", aliases)
-        self.assertIn("supervise", aliases)
+        self.assertNotIn("supervise", aliases)
 
     def test_internal_aliases_include_active_sidecars(self):
-        sc = self._render(with_git=True)["services"]["sidecars"]
+        sc = self._render(with_git=True, supervise=True)["services"]["sidecars"]
         aliases = set(sc["networks"]["internal"]["aliases"])
         self.assertIn("git-gate", aliases)
         self.assertIn("supervise", aliases)
@@ -329,11 +336,10 @@ class TestSidecarBundleShape(unittest.TestCase):
             for line in sc["environment"]
             if line.startswith("BOT_BOTTLE_SIDECAR_DAEMONS=")
         }
-        # egress + supervise are always present (issue #249).
-        self.assertEqual({"egress,supervise"}, daemons)
+        self.assertEqual({"egress"}, daemons)
 
     def test_daemons_csv_expands_with_optional_sidecars(self):
-        sc = self._render(with_git=True)["services"]["sidecars"]
+        sc = self._render(with_git=True, supervise=True)["services"]["sidecars"]
         for line in sc["environment"]:
             if line.startswith("BOT_BOTTLE_SIDECAR_DAEMONS="):
                 csv = line.split("=", 1)[1]
@@ -341,7 +347,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         else:
             self.fail("BOT_BOTTLE_SIDECAR_DAEMONS not in env")
         self.assertEqual(
-            ["egress", "supervise", "git-gate"],
+            ["egress", "git-gate", "supervise"],
             csv.split(","),
         )
 
@@ -370,7 +376,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertNotIn("EGRESS_TOKEN_0", env_strings)
 
     def test_supervise_env_present_when_active(self):
-        sc = self._render()["services"]["sidecars"]
+        sc = self._render(supervise=True)["services"]["sidecars"]
         env_strings = sc["environment"]
         self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings)
         self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
@@ -382,7 +388,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
 
     def test_volumes_union_full_matrix(self):
-        sc = self._render(with_git=True, with_egress=True)[
+        sc = self._render(with_git=True, with_egress=True, supervise=True)[
             "services"]["sidecars"]
         targets = {v["target"] for v in sc["volumes"]}
         self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
@@ -397,7 +403,7 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertNotIn("extra_hosts", sc)
 
     def test_agent_depends_on_bundle_only(self):
-        sc = self._render(with_git=True, with_egress=True)[
+        sc = self._render(with_git=True, with_egress=True, supervise=True)[
             "services"]["agent"]
         self.assertEqual(["sidecars"], sc["depends_on"])
 

tests/unit/test_contrib_claude_provider.py

@@ -50,8 +50,11 @@ def _plan(
     agent_prompt: str = "",
     skills: list[str] | None = None,
     agent_provision: AgentProvisionPlan | None = None,
+    supervise: bool = False,
 ) -> DockerBottlePlan:
     bottle_json: dict = {"agent_provider": {"template": "claude"}}  # type: ignore
+    if supervise:
+        bottle_json["supervise"] = True
     index = ManifestIndex.from_json_obj({
         "bottles": {"dev": bottle_json},
         "agents": {
@@ -67,6 +70,8 @@ def _plan(
         manifest=index, agent_name="demo",
         copy_cwd=False, user_cwd="/tmp/x",
     )
+    supervise_plan = None
+    if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
             queue_dir=Path("/tmp/queue"),
@@ -309,10 +314,17 @@ class TestClaudeUiProvision(unittest.TestCase):
 
 
 class TestClaudeSuperviseMcp(unittest.TestCase):
+    def test_noop_when_supervise_disabled(self):
+        bottle = _make_bottle()
+        ClaudeAgentProvider().provision_supervise_mcp(
+            _plan(supervise=False), bottle, _URL,
+        )
+        bottle.exec.assert_not_called()
+
     def test_runs_claude_mcp_add_as_node(self):
         bottle = _make_bottle()
         ClaudeAgentProvider().provision_supervise_mcp(
-            _plan(), bottle, _URL,
+            _plan(supervise=True), bottle, _URL,
         )
         bottle.exec.assert_called_once()
         script = bottle.exec.call_args.args[0]
@@ -328,7 +340,7 @@ class TestClaudeSuperviseMcp(unittest.TestCase):
             exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
         )
         ClaudeAgentProvider().provision_supervise_mcp(
-            _plan(), bottle, _URL,
+            _plan(supervise=True), bottle, _URL,
         )
 
 

tests/unit/test_contrib_codex_provider.py

@@ -50,8 +50,11 @@ def _plan(
     agent_prompt: str = "",
     skills: list[str] | None = None,
     agent_provision: AgentProvisionPlan | None = None,
+    supervise: bool = False,
 ) -> DockerBottlePlan:
     bottle_json: dict = {"agent_provider": {"template": "codex"}}  # type: ignore
+    if supervise:
+        bottle_json["supervise"] = True
     index = ManifestIndex.from_json_obj({
         "bottles": {"dev": bottle_json},
         "agents": {
@@ -67,6 +70,8 @@ def _plan(
         manifest=index, agent_name="demo",
         copy_cwd=False, user_cwd="/tmp/x",
     )
+    supervise_plan = None
+    if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
             queue_dir=Path("/tmp/queue"),
@@ -272,10 +277,17 @@ class TestCodexProvision(unittest.TestCase):
 
 
 class TestCodexSuperviseMcp(unittest.TestCase):
+    def test_noop_when_supervise_disabled(self):
+        bottle = _make_bottle()
+        CodexAgentProvider().provision_supervise_mcp(
+            _plan(supervise=False), bottle, _URL,
+        )
+        bottle.exec.assert_not_called()
+
     def test_runs_codex_mcp_add_as_node(self):
         bottle = _make_bottle()
         CodexAgentProvider().provision_supervise_mcp(
-            _plan(), bottle, _URL,
+            _plan(supervise=True), bottle, _URL,
         )
         bottle.exec.assert_called_once()
         script = bottle.exec.call_args.args[0]
@@ -290,7 +302,7 @@ class TestCodexSuperviseMcp(unittest.TestCase):
             exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
         )
         CodexAgentProvider().provision_supervise_mcp(
-            _plan(), bottle, _URL,
+            _plan(supervise=True), bottle, _URL,
         )
 
 

tests/unit/test_contrib_pi_provider.py

@@ -16,7 +16,6 @@ from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.contrib.pi.agent_provider import PiAgentProvider
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import ManifestIndex
 
 
@@ -78,11 +77,7 @@ def _plan(
             routes=(),
             token_env_map={},
         ),
-        supervise_plan=SupervisePlan(
-            slug="demo-abc12",
-            queue_dir=Path("/tmp/queue"),
-            current_config_dir=Path("/tmp/current-config"),
-        ),
+        supervise_plan=None,
         use_runsc=False,
         agent_provision=agent_provision or AgentProvisionPlan(
             template="pi", command="pi", prompt_mode="append_system_prompt",

tests/unit/test_docker_launch_committed_image.py

@@ -16,7 +16,6 @@ from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import ManifestIndex
 
 
@@ -56,11 +55,7 @@ def _plan(tmp: str) -> DockerBottlePlan:
             routes=(),
             token_env_map={},
         ),
-        supervise_plan=SupervisePlan(
-            slug=_SLUG,
-            queue_dir=stage / "supervise" / "queue",
-            current_config_dir=stage / "supervise" / "current-config",
-        ),
+        supervise_plan=None,
         agent_provision=AgentProvisionPlan(
             template="claude",
             command="claude",

tests/unit/test_docker_launch_teardown.py

@@ -21,7 +21,6 @@ from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import ManifestIndex
 
 _INDEX = ManifestIndex.from_json_obj({
@@ -57,11 +56,7 @@ def _plan(tmp: str) -> DockerBottlePlan:
             routes=(),
             token_env_map={},
         ),
-        supervise_plan=SupervisePlan(
-            slug="test-teardown-00001",
-            queue_dir=stage / "supervise" / "queue",
-            current_config_dir=stage / "supervise" / "current-config",
-        ),
+        supervise_plan=None,
         agent_provision=AgentProvisionPlan(
             template="claude",
             command="claude",

tests/unit/test_docker_provision_git_user.py

@@ -21,7 +21,6 @@ from bot_bottle.backend import Bottle, BottleSpec, ExecResult
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import ManifestIndex
 
 
@@ -80,11 +79,7 @@ def _plan(*, git_user: dict | None = None,  # type: ignore
             routes=(),
             token_env_map={},
         ),
-        supervise_plan=SupervisePlan(
-            slug="demo-abc12",
-            queue_dir=Path("/tmp/queue"),
-            current_config_dir=Path("/tmp/current-config"),
-        ),
+        supervise_plan=None,
         use_runsc=False,
         agent_provision=AgentProvisionPlan(
             template="claude",

tests/unit/test_macos_container_launch.py

@@ -15,7 +15,6 @@ from bot_bottle.backend.macos_container import launch
 from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import ManifestIndex
 
 _MANIFEST = ManifestIndex.from_json_obj({
@@ -28,6 +27,7 @@ def _plan(
     *,
     stage_dir: Path,
     git: bool = False,
+    supervise: bool = False,
     agent_git_gate_url: str = "",
     agent_supervise_url: str = "",
 ) -> MacosContainerBottlePlan:
@@ -67,8 +67,10 @@ def _plan(
         )
     else:
         git_gate_plan = SimpleNamespace(upstreams=())
-    # Every bottle is supervised (issue #249).
-    supervise_plan = SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
+    supervise_plan = (
+        SimpleNamespace(queue_dir=Path("/state/supervise/queue"))
+        if supervise else None
+    )
     agent_provision = SimpleNamespace(
         guest_env={"LITERAL": "value"},
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
@@ -99,7 +101,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
         self._tmp.cleanup()
 
     def test_sidecar_argv_uses_egress_network_first_and_explicit_dns(self):
-        plan = _plan(stage_dir=self.stage_dir)
+        plan = _plan(stage_dir=self.stage_dir, supervise=True)
         with patch.object(launch.os, "environ", {
             "BOT_BOTTLE_MACOS_CONTAINER_DNS": "9.9.9.9",
         }):
@@ -170,7 +172,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
     def test_git_gate_daemons_are_ready_gated(self):
         plan = _plan(stage_dir=self.stage_dir, git=True)
         self.assertEqual(
-            ("egress", "supervise", "git-gate", "git-http"),
+            ("egress", "git-gate", "git-http"),
             launch._sidecar_daemons(plan),
         )
         self.assertIn(
@@ -179,7 +181,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
         )
 
     def test_stamp_agent_urls_includes_git_http_when_git_gate_exists(self):
-        plan = _plan(stage_dir=self.stage_dir, git=True)
+        plan = _plan(stage_dir=self.stage_dir, git=True, supervise=True)
         with patch.object(launch.dataclasses, "replace") as replace:
             launch._stamp_agent_urls(plan, "192.168.128.2")
         replace.assert_called_once_with(
@@ -270,10 +272,7 @@ def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
         egress_plan=cast(EgressPlan, SimpleNamespace()),
-        supervise_plan=cast(
-            SupervisePlan,
-            SimpleNamespace(queue_dir=Path("/state/supervise/queue")),
-        ),
+        supervise_plan=None,
         agent_provision=AgentProvisionPlan(
             template="claude",
             command="claude",

tests/unit/test_manifest_agent_git_user.py

@@ -116,6 +116,7 @@ class TestAgentGitUserOverlay(unittest.TestCase):
         idx = ManifestIndex.from_json_obj({
             "bottles": {"dev": {
                 "env": {"FOO": "bar"},
+                "supervise": True,
                 "git-gate": {"user": {"name": "B"}},
             }},
             "agents": {"impl": {
@@ -126,6 +127,7 @@ class TestAgentGitUserOverlay(unittest.TestCase):
         b = idx.load_for_agent("impl").bottle
         self.assertEqual("a", b.git_user.name)
         self.assertEqual({"FOO": "bar"}, dict(b.env))
+        self.assertTrue(b.supervise)
 
 
 class TestAgentGitUserRejections(unittest.TestCase):

tests/unit/test_manifest_extends.py

@@ -42,26 +42,38 @@ class TestExtendsBasic(unittest.TestCase):
         # same way they did before the resolver landed.
         m = _build(dev={
             "env": {"FOO": "bar"},
+            "supervise": True,
         })
         b = m.bottles["dev"]
         self.assertEqual({"FOO": "bar"}, dict(b.env))
+        self.assertTrue(b.supervise)
 
     def test_child_inherits_parent_fields_unchanged(self):
         m = _build(
             base={
                 "env": {"BASE": "1"},
+                "supervise": True,
             },
             child={"extends": "base"},
         )
         c = m.bottles["child"]
         self.assertEqual({"BASE": "1"}, dict(c.env))
+        self.assertTrue(c.supervise)
+
+    def test_child_overrides_supervise_scalar(self):
+        m = _build(
+            base={"supervise": True},
+            off={"extends": "base", "supervise": False},
+        )
+        self.assertTrue(m.bottles["base"].supervise)
+        self.assertFalse(m.bottles["off"].supervise)
 
     def test_parent_resolved_once_for_multiple_children(self):
         # Two children sharing one parent: both inherit; the parent
         # is resolved once + cached. (Cache behavior is internal; we
         # observe correctness on both children.)
         m = _build(
-            base={"env": {"BASE": "1"}},
+            base={"env": {"BASE": "1"}, "supervise": True},
             a={"extends": "base", "env": {"A": "1"}},
             b={"extends": "base", "env": {"B": "1"}},
         )
@@ -354,6 +366,7 @@ class TestExtendsChain(unittest.TestCase):
         m = _build(
             grandparent={
                 "env": {"GP": "1"},
+                "supervise": True,
             },
             parent={
                 "extends": "grandparent",
@@ -368,6 +381,8 @@ class TestExtendsChain(unittest.TestCase):
             {"GP": "1", "P": "1", "C": "1"},
             dict(m.bottles["child"].env),
         )
+        # supervise threads through unchanged.
+        self.assertTrue(m.bottles["child"].supervise)
 
     def test_intermediate_can_override(self):
         m = _build(

tests/unit/test_plan_print_parity.py

@@ -19,7 +19,6 @@ from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.smolmachines.bottle_plan import SmolmachinesBottlePlan
 from bot_bottle.egress import EgressPlan, EgressRoute
 from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
-from bot_bottle.supervise import SupervisePlan
 from bot_bottle.manifest import Manifest, ManifestIndex
 
 
@@ -78,15 +77,6 @@ def _egress_plan(tmp: str) -> EgressPlan:
     )
 
 
-def _supervise_plan(tmp: str) -> SupervisePlan:
-    stage = Path(tmp)
-    return SupervisePlan(
-        slug="test-00001",
-        queue_dir=stage / "supervise" / "queue",
-        current_config_dir=stage / "supervise" / "current-config",
-    )
-
-
 def _agent_provision(tmp: str) -> AgentProvisionPlan:
     return AgentProvisionPlan(
         template="claude",
@@ -109,7 +99,7 @@ def _docker_plan(spec: BottleSpec, manifest: Manifest, tmp: str) -> DockerBottle
         stage_dir=stage,
         git_gate_plan=_git_gate_plan(tmp),
         egress_plan=_egress_plan(tmp),
-        supervise_plan=_supervise_plan(tmp),
+        supervise_plan=None,
         agent_provision=_agent_provision(tmp),
         slug="test-00001",
         forwarded_env={},
@@ -125,7 +115,7 @@ def _smolmachines_plan(spec: BottleSpec, manifest: Manifest, tmp: str) -> Smolma
         stage_dir=stage,
         git_gate_plan=_git_gate_plan(tmp),
         egress_plan=_egress_plan(tmp),
-        supervise_plan=_supervise_plan(tmp),
+        supervise_plan=None,
         agent_provision=_agent_provision(tmp),
         slug="test-00001",
         bundle_subnet="10.99.0.0/24",

tests/unit/test_smolmachines_provision.py

@@ -86,6 +86,7 @@ def _plan(
     stage_dir: Path | None = None,
     egress_routes: tuple[EgressRoute, ...] = (),
     egress_ca_path: Path = Path(),
+    supervise: bool = False,
     bundle_ip: str = "192.168.50.2",
     agent_git_gate_host: str = "127.0.0.1:55555",
     agent_supervise_url: str = "http://127.0.0.1:55556/",
@@ -107,6 +108,8 @@ def _plan(
         git_gate_json["user"] = git_user
     if git_gate_json:
         bottle_json["git-gate"] = git_gate_json
+    if supervise:
+        bottle_json["supervise"] = True
     index = ManifestIndex.from_json_obj({
         "bottles": {"dev": bottle_json},
         "agents": {
@@ -124,6 +127,8 @@ def _plan(
         copy_cwd=copy_cwd,
         user_cwd=user_cwd,
     )
+    supervise_plan = None
+    if supervise:
         supervise_plan = SupervisePlan(
             slug="demo-abc12",
             queue_dir=Path("/tmp/queue"),
@@ -400,7 +405,7 @@ class TestBundleLaunchSpec(unittest.TestCase):
         spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
 
         self.assertEqual(
-            "egress,supervise,git-gate,git-http",
+            "egress,git-gate,git-http",
             spec.daemons_csv,
         )
         self.assertIn(9420, spec.ports_to_publish)