refactor(ai): strip Goose harness (Phase 1) — native/simple host-side only

Remove the Goose agentic harness across the codebase per
docs/spec-native-harness.md §3. Goose made N sequential model calls inside the
sandbox (slow on CPU-only hardware) and forced an in-container→host Ollama
gateway that tripped the rootless-Podman slirp4netns loopback bug.

- bridge.py: delete _run_goose/_goose_argv/_goose_present + GOOSE_* consts and
  the present-cache; __init__ now takes harness="simple"/max_turns=5; granted
  !task runs _run_simple until the native loop lands (Phase 2).
- __main__.py: --harness {native,simple} (was {goose,simple}); drop
  --goose-max-turns, add --max-turns; default harness simple.
- app.rs: /ai start accepts native|simple (plain aliases simple) instead of a
  bare plain flag; refresh harness comments.
- sbx.rs: remove the in-container Ollama gateway (Docker host-gateway / Podman
  slirp4netns host-loopback) and the dk_bootstrap OLLAMA_HOST env — kills the
  slirp4netns loopback bug; drop Goose comments.
- bootstrap.sh: drop goose from the prereq probe.
- bootstrap-ai.sh: remove the entire Goose install block, --no-goose flag,
  GOOSE_INSTALLER_URL, host config writer, and goose_bin helper.
- sandbox-bootstrap.sh: remove the in-sandbox Goose binary install + config.
- spec-goose-harness.md: banner — harness portion superseded; Podman stays.

cargo check + py_compile clean. No Goose refs remain (headroom/ untouched).

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

cmd_chat/agent/bridge.py

@@ -85,13 +85,6 @@ ADVISORY_SYSTEM = (
 MAX_COMMANDS = 20
 MAX_BYTES = 8192
 
-# Goose harness limits. The container/VM is the blast radius (see
-# spec-goose-harness.md §4); these bound how much we relay + how long we wait.
-GOOSE_MAX_TURNS = 15        # cap Goose's agentic loop length
-GOOSE_MAX_OUTPUT = 16384    # max bytes of Goose output relayed to chat
-GOOSE_TIMEOUT = 300.0       # seconds before we kill a stuck Goose run
-GOOSE_FLUSH_SECS = 0.5      # throttle: at most ~2 chat updates/sec while streaming
-
 
 class AgentBridge(Client):
     def __init__(self, server: str, port: int, name: str, provider: Provider,
@@ -99,7 +92,7 @@ class AgentBridge(Client):
                  system_prompt: str | None = None, context_window: int = 12,
                  token_budget: int = 2000, embedder=None, rag_top_k: int = 4,
                  rag_min_score: float = 0.35, code_provider: Provider | None = None,
-                 harness: str = "goose", goose_max_turns: int = GOOSE_MAX_TURNS):
+                 harness: str = "simple", max_turns: int = 5):
         super().__init__(server, port, username=name, password=password,
                          insecure=insecure, no_tls=no_tls)
         self.name = name
@@ -128,17 +121,17 @@ class AgentBridge(Client):
         self.granted = False           # may we type into the shared PTY?
         self.can_sudo = False          # does our VM account have sudo?
         self._pending: list[str] | None = None  # destructive plan awaiting /confirm
-        # Default `!task` harness: "goose" (agentic loop run *inside* the sandbox)
-        # or "simple" (the legacy one-shot keystroke injector). Goose degrades to
-        # simple automatically when its binary isn't present in the sandbox.
-        self.harness = harness if harness in ("goose", "simple") else "goose"
-        self.goose_max_turns = goose_max_turns
+        # `!task` harness: "simple" (one-shot keystroke injector — the only one
+        # wired today) or "native" (bounded host-side Ollama tool-calling loop,
+        # lands in Phase 2 per docs/spec-native-harness.md; currently runs as
+        # simple). The default is "simple" until native is implemented.
+        self.harness = harness if harness in ("native", "simple") else "simple"
+        self.max_turns = max_turns  # cap for the native loop (Phase 2)
         # Where the shared sandbox lives, learned from the broker's `_sbx:status`
-        # frame so we can exec Goose into it. None until a sandbox is announced.
+        # frame so we can exec into it. None until a sandbox is announced.
         self.sbx_engine: str | None = None   # docker|podman|multipass|local
         self.sbx_name: str = ""              # container/instance handle ("" for local)
         self.sbx_backend: str | None = None  # cosmetic label from the broker
-        self._goose_present_cache: dict[tuple[str, str], bool] = {}
 
     @staticmethod
     def _est_tokens(text: str) -> int:
@@ -307,7 +300,7 @@ class AgentBridge(Client):
         (file transfer, sandbox data). The owner authorizes us via `/grant <name>`
         (or `/ai start <name> allow`), so we mirror the ACL here to know whether
         we're allowed to act; the status frame tells us which engine + container
-        the (co-located) broker is hosting so we can exec Goose into it."""
+        the (co-located) broker is hosting so we can exec commands into it."""
         try:
             frame = json.loads(text)
         except json.JSONDecodeError:
@@ -321,7 +314,6 @@ class AgentBridge(Client):
                 self.sbx_engine = None
                 self.sbx_name = ""
                 self.sbx_backend = None
-                self._goose_present_cache.clear()
             return
         if frame.get("_perm") != "acl":
             return
@@ -382,12 +374,10 @@ class AgentBridge(Client):
 
         - **Not granted** → advisory only: answer in chat, never touch the
           sandbox (`_advise`).
-        - **Granted** → act in the *spawned sandbox*. With the Goose harness
-          (default) run Goose's agentic loop INSIDE that sandbox — the container
-          /VM via the engine the broker advertised, or the host only for the
-          explicit `local` backend — and stream its output to chat. If Goose
-          isn't installed there, or the harness is `simple`, fall back to the
-          one-shot keystroke injector (`_run_simple`)."""
+        - **Granted** → act in the *spawned sandbox* via the one-shot keystroke
+          injector (`_run_simple`). The bounded host-side `native` tool-calling
+          loop (docs/spec-native-harness.md, Phase 2) will branch here on
+          `self.harness == "native"`; until then every granted task runs simple."""
         if not task:
             await self._send_chat(
                 ws, f"{asker}: tell me what to run, e.g. `/ai {self.name} !create a hello.py`.")
@@ -395,15 +385,6 @@ class AgentBridge(Client):
         if not self.granted:
             await self._advise(ws, task, asker)
             return
-        if self.harness == "goose" and self.sbx_engine is not None:
-            if await self._goose_present():
-                await self._run_goose(ws, task, asker)
-                return
-            await self._send_chat(
-                ws,
-                f"{asker}: Goose isn't installed in this sandbox — using the simple "
-                f"one-shot harness instead.",
-            )
         await self._run_simple(ws, task, asker)
 
     async def _advise(self, ws, task: str, asker: str) -> None:
@@ -430,139 +411,12 @@ class AgentBridge(Client):
             f"but here's how:\n{reply}",
         )
 
-    def _goose_argv(self, task: str) -> list[str] | None:
-        """argv that runs Goose *inside* the current sandbox, or None if we don't
-        know where it lives. `task` is a single argv element (never a shell
-        string), so untrusted room text can't inject shell metacharacters."""
-        eng, name = self.sbx_engine, self.sbx_name
-        goose = ["goose", "run", "-t", task, "--no-session", "-q",
-                 "--max-turns", str(self.goose_max_turns)]
-        if eng in ("docker", "podman"):
-            return [eng, "exec", "-i", name, *goose] if name else None
-        if eng == "multipass":
-            return ["multipass", "exec", name, "--", *goose] if name else None
-        if eng == "local":
-            return goose  # host shell — the explicit, warned exception
-        return None
-
-    async def _goose_present(self) -> bool:
-        """Is the Goose binary runnable inside the current sandbox? Cached per
-        (engine, name). A missing binary triggers the simple-injector fallback,
-        so Goose is the default but never load-bearing."""
-        eng, name = self.sbx_engine, self.sbx_name
-        if eng is None:
-            return False
-        key = (eng, name)
-        if key in self._goose_present_cache:
-            return self._goose_present_cache[key]
-        probe = "command -v goose >/dev/null 2>&1"
-        if eng in ("docker", "podman"):
-            argv = [eng, "exec", name, "sh", "-c", probe] if name else None
-        elif eng == "multipass":
-            argv = ["multipass", "exec", name, "--", "sh", "-c", probe] if name else None
-        elif eng == "local":
-            argv = ["sh", "-c", probe]
-        else:
-            argv = None
-        if argv is None:
-            return False
-        try:
-            proc = await asyncio.create_subprocess_exec(
-                *argv, stdout=asyncio.subprocess.DEVNULL,
-                stderr=asyncio.subprocess.DEVNULL)
-            present = await proc.wait() == 0
-        except (FileNotFoundError, OSError):
-            present = False  # the engine binary itself isn't on this host
-        self._goose_present_cache[key] = present
-        return present
-
-    async def _run_goose(self, ws, task: str, asker: str) -> None:
-        """Tier 1 (granted): run Goose's agentic loop INSIDE the spawned sandbox
-        and relay its output to chat — live preview while it runs, then a final
-        permanent line. Output is byte-capped and the run is time-bounded; the
-        container/VM is the blast radius. For the `local` backend (host shell) we
-        warn loudly first since there is no container isolation."""
-        argv = self._goose_argv(task)
-        if argv is None:
-            await self._send_chat(ws, f"{asker}: I can't locate the sandbox to run Goose in.")
-            return
-        if self.sbx_engine == "local":
-            await self._send_chat(
-                ws,
-                f"⚠ {self.name}: the sandbox is the LOCAL host shell (no container "
-                f"isolation) — Goose will run on this machine.",
-            )
-        await self._send_chat(ws, f"⛧ {self.name}: Goose working on — {task}")
-        try:
-            proc = await asyncio.create_subprocess_exec(
-                *argv,
-                stdin=asyncio.subprocess.DEVNULL,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.STDOUT,
-            )
-        except (FileNotFoundError, OSError) as e:
-            await self._send_chat(ws, f"{asker}: [goose launch failed: {e}]")
-            return
-
-        loop = asyncio.get_running_loop()
-        parts: list[str] = []
-        total = 0
-        truncated = False
-        timed_out = False
-        last_emit = 0.0
-        deadline = loop.time() + GOOSE_TIMEOUT
-        await self._send_typing(ws, True)
-        try:
-            while True:
-                remaining = deadline - loop.time()
-                if remaining <= 0:
-                    proc.kill()
-                    timed_out = True
-                    break
-                try:
-                    chunk = await asyncio.wait_for(proc.stdout.read(1024), timeout=remaining)
-                except asyncio.TimeoutError:
-                    proc.kill()
-                    timed_out = True
-                    break
-                if not chunk:
-                    break
-                text = chunk.decode(errors="replace")
-                room = GOOSE_MAX_OUTPUT - total
-                if room <= 0:
-                    truncated = True
-                    proc.kill()
-                    break
-                clip = text[:room]
-                parts.append(clip)
-                total += len(clip)
-                if len(clip) < len(text):
-                    truncated = True
-                    proc.kill()
-                    break
-                now = loop.time()
-                if now - last_emit >= GOOSE_FLUSH_SECS:
-                    await self._send_stream(ws, "".join(parts), False)
-                    last_emit = now
-        finally:
-            await self._send_stream(ws, "", True)  # clear the live preview
-            await self._send_typing(ws, False)
-        rc = await proc.wait()
-        suffix = ""
-        if timed_out:
-            suffix = "\n[goose timed out — killed]"
-        elif truncated:
-            suffix = f"\n[output capped at {GOOSE_MAX_OUTPUT} bytes]"
-        body = ("".join(parts).strip() or "(no output)") + suffix
-        self.transcript.append(Msg("assistant", "(goose) " + body[:1000]))
-        await self._send_chat(ws, f"⛧ {self.name} (goose) for {asker}:\n{body}")
-        self.success(f"goose run for {asker} exited rc={rc}")
-
     async def _run_simple(self, ws, task: str, asker: str) -> None:
-        """Legacy one-shot harness (granted): turn the request into shell commands
-        with the code provider and type them into the shared PTY via keystroke
-        frames. Guarded by the destructive-command check + blast-radius caps. Used
-        when harness=simple or as the Goose fallback."""
+        """One-shot harness (granted): turn the request into shell commands with
+        the code provider and type them into the shared PTY via keystroke frames.
+        Guarded by the destructive-command check + blast-radius caps. This is the
+        proven injector (commit 47019dd) and the default until the native
+        tool-calling loop lands (docs/spec-native-harness.md, Phase 2)."""
         await self._send_typing(ws, True)
         try:
             context = await self._model_messages(task)