fix(git-gate): accept any key type from ssh-keyscan, prefer ed25519

Servers that don't offer ed25519 (ecdsa-only or rsa-only gitea
instances) would raise "ssh-keyscan returned no ed25519 key" because
`-t ed25519` produced no output.

Remove the `-t ed25519` restriction: ssh-keyscan now returns all
supported key types and the function picks the best available via
_KEY_TYPE_PREFERENCE (ed25519 > ecdsa > rsa > first).

Also adds four new fetch_host_key tests covering multi-type preference
logic and updates the stale "no ed25519 key" assertion.

README.md

@@ -5,7 +5,7 @@
 # bot-bottle
 
 [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
-[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
 [![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
 
 **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.

bot_bottle/backend/__init__.py

@@ -75,6 +75,9 @@ class BottleSpec:
     # Ordered bottle names selected at launch (issue #269). When non-empty
     # they are merged in order and replace the agent's `bottle:` field.
     bottle_names: tuple[str, ...] = ()
+    # True when launched via --headless (no TTY, no interactive prompts).
+    # The git-gate host-key preflight uses this to error rather than prompt.
+    headless: bool = False
 
 
 @dataclass(frozen=True)
@@ -300,6 +303,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
 
         self._preflight()
 
+        from ..git_gate_host_key import preflight_host_keys
+        manifest = preflight_host_keys(
+            manifest,
+            headless=spec.headless,
+            home_md=spec.manifest.home_md,
+        )
+
         manifest_bottle = manifest.bottle
         manifest_agent_provider = manifest_bottle.agent_provider
         agent_provider = get_provider(manifest_agent_provider.template)

bot_bottle/cli/start.py

@@ -209,6 +209,7 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
+        headless=True,
     )
     return _launch_bottle(
         spec,

bot_bottle/contrib/claude/Dockerfile

@@ -21,7 +21,7 @@ FROM node:22-slim
 # to it) works against egress's bumped TLS without the agent needing
 # local DNS.
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 \
+  && apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by claude-code itself

bot_bottle/git_gate_host_key.py

@@ -0,0 +1,298 @@
+"""Preflight host-key population for git-gate upstreams (issue #333).
+
+When a git-gate repo entry lacks a `host_key`, this module either:
+- headless: dies with a clear config error.
+- interactive: fetches the key via ssh-keyscan, prompts the operator to
+  confirm, and optionally persists it to the bottle config file on disk.
+
+Public entry point: `preflight_host_keys(manifest, headless=..., home_md=...)`.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+from .log import die, info
+from .manifest import Manifest
+
+
+# Preferred key types, most secure first.
+_KEY_TYPE_PREFERENCE = (
+    "ssh-ed25519",
+    "ecdsa-sha2-nistp256",
+    "ecdsa-sha2-nistp384",
+    "ecdsa-sha2-nistp521",
+    "ssh-rsa",
+)
+
+
+def fetch_host_key(host: str, port: str) -> str:
+    """Return an SSH public key for `host`:`port` via ssh-keyscan.
+
+    Returns the key in `<type> <base64-data>` format (the host prefix is
+    stripped so the result can be stored in `host_key` and later formatted
+    into a known_hosts line by `git_gate_known_hosts_line`).
+
+    Prefers ed25519 > ecdsa > rsa; falls back to the first key type
+    returned if none of the preferred types are present.
+
+    Raises `RuntimeError` on subprocess failure, timeout, or no result.
+    Uses only the Python stdlib (subprocess)."""
+    args = ["ssh-keyscan"]
+    if port and port != "22":
+        args += ["-p", port]
+    args.append(host)
+    try:
+        result = subprocess.run(
+            args, capture_output=True, text=True, timeout=15, check=False,
+        )
+    except OSError as e:
+        raise RuntimeError(
+            f"ssh-keyscan: could not launch for {host}:{port}: {e}"
+        ) from e
+    except subprocess.TimeoutExpired as e:
+        raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from e
+
+    # known_hosts format: "[host]:port type data"  or  "host type data"
+    # Strip the host/port prefix; collect "type -> type data" by type.
+    found: dict[str, str] = {}
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        parts = line.split(None, 2)
+        if len(parts) == 3 and parts[1] not in found:
+            found[parts[1]] = f"{parts[1]} {parts[2]}"
+
+    for preferred in _KEY_TYPE_PREFERENCE:
+        if preferred in found:
+            return found[preferred]
+    if found:
+        return next(iter(found.values()))
+
+    raise RuntimeError(
+        f"ssh-keyscan returned no host key for {host}:{port}."
+        + (f" stderr: {result.stderr.strip()!r}" if result.stderr.strip() else "")
+    )
+
+
+# ---------------------------------------------------------------------------
+# Frontmatter text-level editing
+# ---------------------------------------------------------------------------
+
+
+def _insert_host_key_in_frontmatter(fm_text: str, repo_name: str, host_key: str) -> str:
+    """Append `host_key: <key>` to the named repo entry in raw frontmatter text.
+
+    Locates the `<repo_name>:` block-mapping key, detects its child
+    indentation from the sibling keys already present, and inserts the
+    new line after the last child.  Returns the original text if the
+    repo entry is not found or already has a `host_key:` line."""
+    lines = fm_text.splitlines(keepends=True)
+
+    repo_pat = re.compile(r"^(\s+)" + re.escape(repo_name) + r"\s*:\s*(\r?\n)?$")
+    repo_line_idx: int | None = None
+    repo_indent: int | None = None
+    for i, line in enumerate(lines):
+        if repo_pat.match(line):
+            repo_indent = len(line) - len(line.lstrip())
+            repo_line_idx = i
+            break
+
+    if repo_line_idx is None or repo_indent is None:
+        return fm_text
+
+    child_indent: int | None = None
+    last_child_idx = repo_line_idx
+
+    for i in range(repo_line_idx + 1, len(lines)):
+        raw = lines[i].rstrip("\r\n")
+        if not raw.strip():
+            continue
+        indent = len(raw) - len(raw.lstrip())
+        if indent > repo_indent:
+            child_indent = indent
+            last_child_idx = i
+        else:
+            break
+
+    if child_indent is None:
+        child_indent = repo_indent + 2
+
+    block = "".join(lines[repo_line_idx + 1 : last_child_idx + 1])
+    if re.search(r"^\s*host_key\s*:", block, re.MULTILINE):
+        return fm_text
+
+    new_line = " " * child_indent + f"host_key: {host_key}\n"
+    lines.insert(last_child_idx + 1, new_line)
+    return "".join(lines)
+
+
+def _update_frontmatter_in_file(file_text: str, repo_name: str, host_key: str) -> str:
+    """Apply `_insert_host_key_in_frontmatter` to the frontmatter section of
+    a Markdown file, leaving the body unchanged.  Returns the original text
+    when no change is needed."""
+    lines = file_text.splitlines(keepends=True)
+    if not lines or lines[0].rstrip("\r\n") != "---":
+        return file_text
+
+    close_idx: int | None = None
+    for i in range(1, len(lines)):
+        if lines[i].rstrip("\r\n") == "---":
+            close_idx = i
+            break
+    if close_idx is None:
+        return file_text
+
+    fm_text = "".join(lines[1:close_idx])
+    body_text = "".join(lines[close_idx:])
+    updated_fm = _insert_host_key_in_frontmatter(fm_text, repo_name, host_key)
+    if updated_fm == fm_text:
+        return file_text
+    return "---\n" + updated_fm + body_text
+
+
+def _find_and_update_bottle_file(
+    bottles_dir: Path, repo_name: str, host_key: str,
+) -> bool:
+    """Scan `bottles_dir/*.md` for the first file that directly declares
+    `repo_name` without a `host_key`, update it, and return True.
+
+    Walks files in sorted order so the result is deterministic.  Does NOT
+    follow `extends:` chains — only files that explicitly declare the repo
+    entry are candidates.  Returns False when no such file is found."""
+    if not bottles_dir.is_dir():
+        return False
+
+    from .yaml_subset import YamlSubsetError, parse_frontmatter
+
+    for path in sorted(bottles_dir.glob("*.md")):
+        try:
+            text = path.read_text(encoding="utf-8")
+            fm, _body = parse_frontmatter(text)
+        except (OSError, UnicodeDecodeError, YamlSubsetError):
+            continue
+
+        git_gate = fm.get("git-gate")
+        if not isinstance(git_gate, dict):
+            continue
+        repos = git_gate.get("repos")
+        if not isinstance(repos, dict):
+            continue
+        repo = repos.get(repo_name)
+        if not isinstance(repo, dict):
+            continue
+        if repo.get("host_key"):
+            continue
+
+        updated = _update_frontmatter_in_file(text, repo_name, host_key)
+        if updated == text:
+            continue
+        path.write_text(updated, encoding="utf-8")
+        info(f"wrote host_key for {repo_name!r} to {path}")
+        return True
+
+    return False
+
+
+# ---------------------------------------------------------------------------
+# Interactive prompt helper
+# ---------------------------------------------------------------------------
+
+
+def _prompt_tty(message: str) -> str:
+    """Write `message` to stderr and read a line from /dev/tty (or stdin)."""
+    sys.stderr.write(message)
+    sys.stderr.flush()
+    try:
+        with open("/dev/tty", "r", encoding="utf-8") as tty:
+            return tty.readline().rstrip("\n")
+    except OSError:
+        return sys.stdin.readline().rstrip("\n")
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def preflight_host_keys(
+    manifest: Manifest,
+    *,
+    headless: bool,
+    home_md: Path | None,
+) -> Manifest:
+    """Ensure every git-gate repo entry has a `host_key` configured.
+
+    For entries whose `KnownHostKey` is empty:
+    - headless: calls `die()` with a clear message naming the repos.
+    - interactive: fetches the key via ssh-keyscan, shows it to the
+      operator, and requests confirmation.  If accepted, optionally
+      persists it to the bottle config file on disk; the key is always
+      applied in memory for this launch regardless of the persistence
+      choice.  Aborted confirmation calls `die()`.
+
+    Returns a (possibly updated) Manifest.  If all entries already have
+    host keys the original manifest is returned unchanged."""
+    bottle = manifest.bottle
+    missing = [e for e in bottle.git if not e.KnownHostKey]
+    if not missing:
+        return manifest
+
+    if headless:
+        names = ", ".join(repr(e.Name) for e in missing)
+        die(
+            f"git-gate: no host_key configured for repo(s) {names}. "
+            f"Add host_key to each bottle git-gate.repos entry, or run "
+            f"interactively once to have it fetched and saved automatically."
+        )
+
+    bottles_dir = (home_md / "bottles") if home_md is not None else None
+    updated_entries = list(bottle.git)
+
+    for entry in missing:
+        host = entry.UpstreamHost
+        port = entry.UpstreamPort
+        label = f"git-gate.repos[{entry.Name!r}]"
+
+        info(f"{label}: no host_key configured; fetching from {host}:{port}")
+        try:
+            key = fetch_host_key(host, port)
+        except RuntimeError as e:
+            die(f"git-gate: {label}: {e}")
+
+        sys.stderr.write(f"\ngit-gate: host key for {label}:\n  {key}\n\n")
+        confirm = _prompt_tty("Is this host key correct? [y/N] ")
+        if confirm.strip().lower() not in ("y", "yes"):
+            die(f"git-gate: {label}: host key not confirmed; aborting launch")
+
+        if bottles_dir is not None:
+            save = _prompt_tty(
+                f"Save host_key for {entry.Name!r} to the bottle config file? [y/N] "
+            )
+            if save.strip().lower() in ("y", "yes"):
+                ok = _find_and_update_bottle_file(bottles_dir, entry.Name, key)
+                if not ok:
+                    sys.stderr.write(
+                        f"git-gate: {label}: bottle file not found or not writable; "
+                        f"host_key kept in memory for this session only\n"
+                    )
+
+        idx = next(i for i, e in enumerate(updated_entries) if e.Name == entry.Name)
+        updated_entries[idx] = dataclasses.replace(entry, KnownHostKey=key)
+
+    updated_bottle = dataclasses.replace(bottle, git=tuple(updated_entries))
+    return dataclasses.replace(manifest, bottle=updated_bottle)
+
+
+__all__ = [
+    "fetch_host_key",
+    "preflight_host_keys",
+    "_insert_host_key_in_frontmatter",
+    "_find_and_update_bottle_file",
+    "_update_frontmatter_in_file",
+]

docs/prds/0068-smolmachines-linux.md

@@ -1,6 +1,6 @@
-# PRD prd-new: smolmachines backend on Linux
+# PRD 0068: smolmachines backend on Linux
 
-- **Status:** Draft
+- **Status:** Active
 - **Author:** Claude
 - **Created:** 2026-06-25
 - **Issue:** #283

requirements-dev.txt

@@ -3,5 +3,5 @@
 # These tools are used for code quality checks in CI/CD.
 
 pylint>=3.0.0
-pyright>=1.1.300
+pyright>=1.1.411
 coverage>=7.0.0

tests/integration/test_sandbox_escape.py

@@ -81,10 +81,10 @@ class TestSandboxEscape(unittest.TestCase):
         # those are missing rather than die-ing inside backend.prepare.
         backend_name = os.environ.get("BOT_BOTTLE_BACKEND", "docker")
         if backend_name == "smolmachines":
-            if sys.platform != "darwin":
+            if sys.platform not in ("darwin", "linux"):
                 raise unittest.SkipTest(
-                    "BOT_BOTTLE_BACKEND=smolmachines is macOS-only in "
-                    "v1 (libkrun TSI)"
+                    f"BOT_BOTTLE_BACKEND=smolmachines is not supported "
+                    f"on {sys.platform} (macOS and Linux only)"
                 )
             if shutil.which("smolvm") is None:
                 raise unittest.SkipTest(

tests/unit/test_git_gate_host_key.py

@@ -0,0 +1,581 @@
+"""Unit tests for git_gate_host_key (issue #333)."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.git_gate_host_key import (
+    _find_and_update_bottle_file,
+    _insert_host_key_in_frontmatter,
+    _prompt_tty,
+    _update_frontmatter_in_file,
+    fetch_host_key,
+    preflight_host_keys,
+)
+from bot_bottle.manifest import Manifest, ManifestIndex
+
+
+# ---------------------------------------------------------------------------
+# fetch_host_key
+# ---------------------------------------------------------------------------
+
+
+class TestFetchHostKey(unittest.TestCase):
+    def _run_result(self, stdout: str, returncode: int = 0) -> MagicMock:
+        r = MagicMock()
+        r.stdout = stdout
+        r.stderr = ""
+        r.returncode = returncode
+        return r
+
+    def test_returns_type_and_key(self) -> None:
+        stdout = "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key)
+
+    def test_non_default_port_passes_p_flag(self) -> None:
+        stdout = "[gitea.example.com]:30009 ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run:
+            fetch_host_key("gitea.example.com", "30009")
+        args = mock_run.call_args[0][0]
+        self.assertIn("-p", args)
+        self.assertIn("30009", args)
+
+    def test_default_port_omits_p_flag(self) -> None:
+        stdout = "github.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run:
+            fetch_host_key("github.com", "22")
+        args = mock_run.call_args[0][0]
+        self.assertNotIn("-p", args)
+
+    def test_skips_comment_lines(self) -> None:
+        stdout = "# comment\ngithub.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("github.com", "22")
+        self.assertEqual("ssh-ed25519 AAAA", key)
+
+    def test_skips_lines_without_three_parts(self) -> None:
+        # A line with only two whitespace-separated tokens is not a valid
+        # known_hosts entry; it should be skipped and the function should
+        # raise rather than return a partial result.
+        stdout = "malformed-line\ngithub.com ssh-ed25519 AAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("github.com", "22")
+        self.assertEqual("ssh-ed25519 AAAA", key)
+
+    def test_raises_on_empty_output(self) -> None:
+        with patch("subprocess.run", return_value=self._run_result("")):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("no host key", str(cm.exception))
+
+    def test_prefers_ed25519_over_ecdsa(self) -> None:
+        stdout = (
+            "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n"
+            "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n"
+        )
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key)
+
+    def test_falls_back_to_ecdsa_without_ed25519(self) -> None:
+        stdout = "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ecdsa-sha2-nistp256 BBBBB", key)
+
+    def test_falls_back_to_rsa_without_ed25519_or_ecdsa(self) -> None:
+        stdout = "gitea.example.com ssh-rsa AAAAB3NzaC1yc2EAAAA\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-rsa AAAAB3NzaC1yc2EAAAA", key)
+
+    def test_falls_back_to_first_for_unknown_type(self) -> None:
+        stdout = "gitea.example.com ssh-unknown CCCCC\n"
+        with patch("subprocess.run", return_value=self._run_result(stdout)):
+            key = fetch_host_key("gitea.example.com", "22")
+        self.assertEqual("ssh-unknown CCCCC", key)
+
+    def test_raises_includes_stderr_when_present(self) -> None:
+        r = self._run_result("")
+        r.stderr = "connection refused"
+        with patch("subprocess.run", return_value=r):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("connection refused", str(cm.exception))
+
+    def test_raises_on_os_error(self) -> None:
+        with patch("subprocess.run", side_effect=OSError("not found")):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("could not launch", str(cm.exception))
+
+    def test_raises_on_timeout(self) -> None:
+        import subprocess
+        with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("ssh-keyscan", 15)):
+            with self.assertRaises(RuntimeError) as cm:
+                fetch_host_key("example.com", "22")
+        self.assertIn("timed out", str(cm.exception))
+
+
+# ---------------------------------------------------------------------------
+# _prompt_tty
+# ---------------------------------------------------------------------------
+
+
+class TestPromptTty(unittest.TestCase):
+    def test_reads_from_tty_when_available(self) -> None:
+        import io
+        fake_tty = io.StringIO("yes\n")
+        with patch("builtins.open", return_value=fake_tty), patch("sys.stderr"):
+            result = _prompt_tty("question: ")
+        self.assertEqual("yes", result)
+
+    def test_falls_back_to_stdin_on_os_error(self) -> None:
+        import io
+        with patch("builtins.open", side_effect=OSError("no tty")), \
+                patch("sys.stdin", io.StringIO("fallback\n")), \
+                patch("sys.stderr"):
+            result = _prompt_tty("question: ")
+        self.assertEqual("fallback", result)
+
+
+# ---------------------------------------------------------------------------
+# _insert_host_key_in_frontmatter
+# ---------------------------------------------------------------------------
+
+
+class TestInsertHostKeyInFrontmatter(unittest.TestCase):
+    _FM = """\
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+"""
+
+    def test_inserts_host_key_after_last_child(self) -> None:
+        result = _insert_host_key_in_frontmatter(self._FM, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("      host_key: ssh-ed25519 AAAA", result)
+
+    def test_no_change_when_repo_not_found(self) -> None:
+        result = _insert_host_key_in_frontmatter(self._FM, "other", "ssh-ed25519 AAAA")
+        self.assertEqual(self._FM, result)
+
+    def test_no_change_when_host_key_already_present(self) -> None:
+        fm = self._FM + "      host_key: ssh-ed25519 EXISTING\n"
+        result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 NEW")
+        self.assertNotIn("NEW", result)
+        self.assertEqual(fm, result)
+
+    def test_preserves_other_repos(self) -> None:
+        fm = """\
+git-gate:
+  repos:
+    first:
+      url: ssh://git@host/org/first.git
+      key:
+        provider: static
+        path: /dev/null
+    second:
+      url: ssh://git@host/org/second.git
+      key:
+        provider: static
+        path: /dev/null
+"""
+        result = _insert_host_key_in_frontmatter(fm, "first", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        self.assertIn("second:", result)
+
+    def test_detects_child_indent_from_existing_keys(self) -> None:
+        # 4-space indent (non-standard but valid)
+        fm = """\
+git-gate:
+    repos:
+        myrepo:
+            url: ssh://git@host/org/repo.git
+            key:
+                provider: static
+                path: /dev/null
+"""
+        result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("            host_key: ssh-ed25519 AAAA", result)
+
+    def test_blank_lines_inside_block_do_not_end_scan(self) -> None:
+        fm = """\
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+
+      key:
+        provider: static
+        path: /dev/null
+"""
+        result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+
+    def test_repo_with_no_children_uses_default_indent(self) -> None:
+        # Edge case: repo entry has no child keys at all (empty block).
+        fm = "git-gate:\n  repos:\n    myrepo:\n"
+        result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA")
+        # Should fall back to repo_indent+2 = 4
+        self.assertIn("    host_key: ssh-ed25519 AAAA", result)
+
+
+# ---------------------------------------------------------------------------
+# _update_frontmatter_in_file
+# ---------------------------------------------------------------------------
+
+
+class TestUpdateFrontmatterInFile(unittest.TestCase):
+    _FILE = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+# Body text here
+"""
+
+    def test_updates_frontmatter_preserves_body(self) -> None:
+        result = _update_frontmatter_in_file(self._FILE, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        self.assertIn("# Body text here", result)
+
+    def test_no_change_without_frontmatter(self) -> None:
+        text = "No frontmatter here\n"
+        result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_repo_not_in_file(self) -> None:
+        result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA")
+        self.assertEqual(self._FILE, result)
+
+    def test_no_change_when_no_closing_delimiter(self) -> None:
+        text = "---\ngit-gate:\n  repos:\n    myrepo:\n      url: x\n"
+        result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+
+# ---------------------------------------------------------------------------
+# _find_and_update_bottle_file
+# ---------------------------------------------------------------------------
+
+
+class TestFindAndUpdateBottleFile(unittest.TestCase):
+    def _write_bottle(self, d: str, name: str, content: str) -> Path:
+        path = Path(d) / f"{name}.md"
+        path.write_text(content, encoding="utf-8")
+        return path
+
+    def test_updates_file_with_matching_repo(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+            path = self._write_bottle(d, "dev", content)
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertTrue(result)
+            updated = path.read_text()
+            self.assertIn("host_key: ssh-ed25519 AAAA", updated)
+
+    def test_returns_false_when_no_matching_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    other:
+      url: ssh://git@host/org/other.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+            self._write_bottle(d, "dev", content)
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_file_that_already_has_host_key(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+      host_key: ssh-ed25519 EXISTING
+---
+"""
+            path = self._write_bottle(d, "dev", content)
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 NEW")
+            self.assertFalse(result)
+            self.assertNotIn("NEW", path.read_text())
+
+    def test_returns_false_for_nonexistent_dir(self) -> None:
+        result = _find_and_update_bottle_file(
+            Path("/nonexistent/dir"), "myrepo", "ssh-ed25519 AAAA"
+        )
+        self.assertFalse(result)
+
+    def test_skips_file_without_git_gate_section(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write_bottle(d, "plain", "---\nenv:\n  FOO: bar\n---\n")
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_file_with_non_dict_repos(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write_bottle(d, "bad", "---\ngit-gate:\n  repos:\n    - item\n---\n")
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_invalid_utf8_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "broken.md"
+            path.write_bytes(b"\xff\xfe")  # invalid UTF-8
+            result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_unreadable_file(self) -> None:
+        import os
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "unreadable.md"
+            path.write_text("---\n---\n")
+            os.chmod(path, 0o000)
+            try:
+                result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+                self.assertFalse(result)
+            finally:
+                os.chmod(path, 0o644)
+
+    def test_skips_when_update_produces_no_change(self) -> None:
+        # _update_frontmatter_in_file returns the original text unchanged
+        # (e.g. because _insert_host_key_in_frontmatter can't locate the
+        # entry in the raw text).  Line 174 must `continue` rather than write.
+        content = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+        with tempfile.TemporaryDirectory() as d:
+            path = self._write_bottle(d, "dev", content)
+            with patch(
+                "bot_bottle.git_gate_host_key._update_frontmatter_in_file",
+                return_value=content,  # no change
+            ):
+                result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+            self.assertNotIn("host_key", path.read_text())
+
+
+# ---------------------------------------------------------------------------
+# preflight_host_keys
+# ---------------------------------------------------------------------------
+
+
+def _manifest_with_git(host_key: str = "") -> Manifest:
+    """Build a Manifest with one git entry; host_key defaults to empty."""
+    return ManifestIndex.from_json_obj({
+        "bottles": {
+            "dev": {
+                "git-gate": {
+                    "repos": {
+                        "myrepo": {
+                            "url": "ssh://git@gitea.example.com:30009/org/myrepo.git",
+                            "key": {"provider": "static", "path": "/dev/null"},
+                            **({"host_key": host_key} if host_key else {}),
+                        },
+                    },
+                },
+            },
+        },
+        "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+    }).load_for_agent("demo")
+
+
+class TestPreflightHostKeys(unittest.TestCase):
+    def test_no_op_when_all_keys_present(self) -> None:
+        manifest = _manifest_with_git("ssh-ed25519 AAAA")
+        result = preflight_host_keys(manifest, headless=False, home_md=None)
+        self.assertIs(result, manifest)
+
+    def test_headless_dies_when_key_missing(self) -> None:
+        manifest = _manifest_with_git()
+        with self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=True, home_md=None)
+
+    def test_interactive_fetches_and_confirms_key(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            return_value="ssh-ed25519 FETCHED",
+        ), patch(
+            "bot_bottle.git_gate_host_key._prompt_tty",
+            side_effect=["y", "n"],  # confirm key, don't persist
+        ), patch("sys.stderr"):
+            result = preflight_host_keys(manifest, headless=False, home_md=None)
+
+        self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+
+    def test_interactive_dies_when_key_not_confirmed(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            return_value="ssh-ed25519 FETCHED",
+        ), patch(
+            "bot_bottle.git_gate_host_key._prompt_tty",
+            return_value="n",
+        ), patch("sys.stderr"), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=False, home_md=None)
+
+    def test_interactive_persists_when_requested(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key._prompt_tty",
+                side_effect=["y", "y"],  # confirm key, yes persist
+            ), patch("sys.stderr"):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            written = bottle_file.read_text()
+            self.assertIn("host_key: ssh-ed25519 FETCHED", written)
+
+    def test_interactive_skips_persist_when_declined(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+            bottle_file = bottles_dir / "dev.md"
+            bottle_file.write_text(
+                "---\ngit-gate:\n  repos:\n    myrepo:\n"
+                "      url: ssh://git@gitea.example.com:30009/org/myrepo.git\n"
+                "      key:\n        provider: static\n        path: /dev/null\n---\n",
+                encoding="utf-8",
+            )
+
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key._prompt_tty",
+                side_effect=["y", "n"],  # confirm key, no persist
+            ), patch("sys.stderr"):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            written = bottle_file.read_text()
+            self.assertNotIn("host_key", written)
+
+    def test_interactive_dies_when_fetch_fails(self) -> None:
+        manifest = _manifest_with_git()
+
+        with patch(
+            "bot_bottle.git_gate_host_key.fetch_host_key",
+            side_effect=RuntimeError("connection refused"),
+        ), patch("sys.stderr"), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=False, home_md=None)
+
+    def test_interactive_warns_when_persist_file_not_found(self) -> None:
+        manifest = _manifest_with_git()
+
+        with tempfile.TemporaryDirectory() as home:
+            # bottles/ dir exists but does not contain the repo entry
+            bottles_dir = Path(home) / "bottles"
+            bottles_dir.mkdir()
+
+            import io
+            buf = io.StringIO()
+            with patch(
+                "bot_bottle.git_gate_host_key.fetch_host_key",
+                return_value="ssh-ed25519 FETCHED",
+            ), patch(
+                "bot_bottle.git_gate_host_key._prompt_tty",
+                side_effect=["y", "y"],  # confirm key, yes persist
+            ), patch("sys.stderr", buf):
+                result = preflight_host_keys(
+                    manifest, headless=False, home_md=Path(home),
+                )
+
+            self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey)
+            self.assertIn("kept in memory", buf.getvalue())
+
+    def test_headless_names_all_missing_repos_in_error(self) -> None:
+        manifest = ManifestIndex.from_json_obj({
+            "bottles": {
+                "dev": {
+                    "git-gate": {
+                        "repos": {
+                            "alpha": {
+                                "url": "ssh://git@host/org/alpha.git",
+                                "key": {"provider": "static", "path": "/dev/null"},
+                            },
+                            "beta": {
+                                "url": "ssh://git@host/org/beta.git",
+                                "key": {"provider": "static", "path": "/dev/null"},
+                            },
+                        },
+                    },
+                },
+            },
+            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+        }).load_for_agent("demo")
+
+        import io
+        buf = io.StringIO()
+        with patch("sys.stderr", buf), self.assertRaises(SystemExit):
+            preflight_host_keys(manifest, headless=True, home_md=None)
+
+
+if __name__ == "__main__":
+    unittest.main()