fix: pyright error in test_compose — annotate kwargs as dict[str, Any]

CI runs as root, so chmod 0o000 doesn't prevent reads. Both
test_skips_unreadable_file (TestFindRepoBottleFile) and
test_returns_false_when_file_unreadable_on_write now mock
Path.read_text to raise OSError directly.

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
     # Image startup policy. "fresh" preserves the normal build path;
     # "cached" reuses the current local image/artifact without rebuilding.
     image_policy: str = "fresh"
@@ -303,6 +306,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/backend/smolmachines/launch.py

@@ -93,7 +93,7 @@ def launch(
     try:
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
-        proxy_host = _proxy_host(plan, loopback_ip)
+        proxy_host = loopback_ip
         plan = _start_bundle(plan, network, proxy_host, stack)
 
         agent_from_path = _agent_from_path(plan)
@@ -291,14 +291,13 @@ def _launch_vm(
 ) -> None:
     """Create, patch, and start the smolvm VM; register teardown.
 
-    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias
+    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias.
-    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
+    This ensures the guest can only reach sidecar forwarders published
-    ensures the guest can only reach bundle ports published on that IP,
+    on that IP, not host localhost or another bottle's alias.
-    not the container IP directly. force_allowlist confirms the
+    force_allowlist confirms the allowlist persisted (patching smolvm
-    allowlist persisted (patching smolvm 0.8.0's silent-drop of
+    0.8.0's silent-drop of --allow-cidr when combined with --from) and
-    --allow-cidr when combined with --from) and fails closed if it
+    fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
-    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
+    makes --from and --smolfile mutually exclusive."""
-    and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
     _smolvm.machine_create(
         plan.machine_name,
@@ -345,12 +344,6 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
-def _proxy_host(plan: SmolmachinesBottlePlan, loopback_ip: str) -> str:
-    """Return the per-bottle host address used for TSI and forwarders."""
-    del plan
-    return loopback_ip
-
-
 def _label_for_port(port: int) -> str:
     if port == _EGRESS_PORT:
         return "egress"

bot_bottle/cli/start.py

@@ -224,6 +224,7 @@ def _start_headless(
         label=label,
         color=args.color or "",
         bottle_names=bottle_names,
+        headless=True,
         image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(

bot_bottle/git_gate_host_key.py

@@ -0,0 +1,268 @@
+"""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 subprocess
+import sys
+from pathlib import Path
+from typing import cast
+
+from .log import die, info
+from .manifest import Manifest
+from .yaml_subset import YamlSubsetError, parse_frontmatter, serialize_yaml_subset
+
+
+# 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 editing
+# ---------------------------------------------------------------------------
+
+
+def add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str:
+    """Return an updated copy of `file_text` with `host_key` set on the
+    named repo entry in the YAML frontmatter.
+
+    Parses the frontmatter into a dict, sets the key, and re-serializes.
+    Returns the original text unchanged when: the file has no frontmatter,
+    the git-gate.repos.<repo_name> entry is absent or already has a
+    host_key, or the frontmatter cannot be parsed."""
+    try:
+        fm, body = parse_frontmatter(file_text)
+    except YamlSubsetError:
+        return file_text
+
+    git_gate = fm.get("git-gate")
+    if not isinstance(git_gate, dict):
+        return file_text
+    repos = git_gate.get("repos")
+    if not isinstance(repos, dict):
+        return file_text
+    repo = repos.get(repo_name)
+    if not isinstance(repo, dict):
+        return file_text
+    if repo.get("host_key"):
+        return file_text
+
+    cast(dict[str, object], repo)["host_key"] = host_key
+    return f"---\n{serialize_yaml_subset(fm)}---\n{body}"
+
+
+def find_repo_bottle_file(bottles_dir: Path, repo_name: str) -> Path | None:
+    """Return the first `bottles_dir/*.md` that declares `repo_name` without
+    a `host_key`, without modifying anything.  Returns None if not found."""
+    if not bottles_dir.is_dir():
+        return None
+    for path in sorted(bottles_dir.glob("*.md")):
+        try:
+            text = path.read_text(encoding="utf-8")
+            fm, _ = 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) or repo.get("host_key"):
+            continue
+        return path
+    return None
+
+
+def find_and_update_bottle_file(
+    bottles_dir: Path, repo_name: str, host_key: str,
+) -> bool:
+    """Write `host_key` into the bottle file returned by `find_repo_bottle_file`.
+
+    Returns True on success, False when no suitable file is found or the
+    write fails."""
+    path = find_repo_bottle_file(bottles_dir, repo_name)
+    if path is None:
+        return False
+    try:
+        text = path.read_text(encoding="utf-8")
+    except (OSError, UnicodeDecodeError):
+        return False
+    updated = add_host_key_to_frontmatter(text, repo_name, host_key)
+    if updated == text:
+        return False
+    path.write_text(updated, encoding="utf-8")
+    info(f"wrote host_key for {repo_name!r} to {path}")
+    return True
+
+
+# ---------------------------------------------------------------------------
+# 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:
+            target_file = find_repo_bottle_file(bottles_dir, entry.Name)
+            if target_file is not None:
+                save = prompt_tty(
+                    f"Save host_key for {entry.Name!r} to {target_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}: could not write to {target_file}; "
+                            f"host_key kept in memory for this session only\n"
+                        )
+            else:
+                sys.stderr.write(
+                    f"git-gate: {label}: no bottle config file found for "
+                    f"{entry.Name!r}; 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",
+    "add_host_key_to_frontmatter",
+    "find_repo_bottle_file",
+    "find_and_update_bottle_file",
+    "prompt_tty",
+]

bot_bottle/yaml_subset.py

@@ -21,6 +21,11 @@ Public API:
         For a Markdown file with YAML frontmatter delimited by `---`
         lines. Returns (frontmatter_dict, body_text).
 
+    serialize_yaml_subset(data) -> str
+        Serialize a dict (as produced by parse_yaml_subset) back to
+        block-style YAML text.  The result ends with a newline and
+        can be parsed back by parse_yaml_subset.
+
 What we accept (block-style):
 
     key: value                # mapping entry, value is inline
@@ -576,3 +581,105 @@ def parse_frontmatter(text: str) -> tuple[dict[str, object], str]:
     fm = parse_yaml_subset(fm_text)
     body = text[body_start:]
     return fm, body
+
+
+# --- Serializer -------------------------------------------------------------
+
+
+def _needs_quoting(s: str) -> bool:
+    """True when the string must be single-quoted to survive a round-trip."""
+    if not s:
+        return True
+    if s in ("true", "false", "null", "~") or s in _RESERVED_BOOL_LIKE:
+        return True
+    if (
+        _INT_RX.match(s)
+        or _DATE_RX.match(s)
+        or _OCTAL_RX.match(s)
+        or _HEX_RX.match(s)
+        or _FLOAT_RX.match(s)
+    ):
+        return True
+    # Characters that have special meaning at the start of a YAML value
+    if s[0] in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">", "%", "@", "`"):
+        return True
+    return False
+
+
+def _yaml_scalar(v: object) -> str:
+    """Serialize a scalar Python value to its YAML text form."""
+    if v is None:
+        return "null"
+    if isinstance(v, bool):
+        return "true" if v else "false"
+    if isinstance(v, int):
+        return str(v)
+    s = str(v)
+    if _needs_quoting(s):
+        return "'" + s.replace("'", "''") + "'"
+    return s
+
+
+def _serialize_node(node: object, indent: int) -> list[str]:
+    """Return lines (without trailing newlines) for `node` at `indent`.
+
+    Called only for non-empty dicts and lists (the caller guards with
+    `isinstance(val, (dict, list)) and val`), plus scalars at the leaf."""
+    prefix = " " * indent
+    if isinstance(node, dict):
+        out: list[str] = []
+        for key, val in node.items():
+            if isinstance(val, (dict, list)) and val:
+                out.append(f"{prefix}{key}:")
+                out.extend(_serialize_node(val, indent + 2))
+            else:
+                scalar = (
+                    "{}" if isinstance(val, dict)
+                    else "[]" if isinstance(val, list)
+                    else _yaml_scalar(val)
+                )
+                out.append(f"{prefix}{key}: {scalar}")
+        return out
+    if isinstance(node, list):
+        out = []
+        for item in node:
+            if isinstance(item, dict) and item:
+                entries = list(item.items())
+                first_key, first_val = entries[0]
+                if isinstance(first_val, (dict, list)) and first_val:
+                    out.append(f"{prefix}- {first_key}:")
+                    out.extend(_serialize_node(first_val, indent + 4))
+                else:
+                    scalar = (
+                        "{}" if isinstance(first_val, dict)
+                        else "[]" if isinstance(first_val, list)
+                        else _yaml_scalar(first_val)
+                    )
+                    out.append(f"{prefix}- {first_key}: {scalar}")
+                cont = prefix + "  "
+                for key, val in entries[1:]:
+                    if isinstance(val, (dict, list)) and val:
+                        out.append(f"{cont}{key}:")
+                        out.extend(_serialize_node(val, indent + 4))
+                    else:
+                        scalar = (
+                            "{}" if isinstance(val, dict)
+                            else "[]" if isinstance(val, list)
+                            else _yaml_scalar(val)
+                        )
+                        out.append(f"{cont}{key}: {scalar}")
+            else:
+                out.append(f"{prefix}- {_yaml_scalar(item)}")
+        return out
+    return [_yaml_scalar(node)]  # pragma: no cover
+
+
+def serialize_yaml_subset(data: dict[str, object]) -> str:
+    """Serialize `data` (as produced by parse_yaml_subset) to YAML text.
+
+    Produces block-style output with 2-space indentation. The result ends
+    with a newline and can be parsed back by parse_yaml_subset. Keys are
+    emitted in iteration order (insertion order in Python 3.7+)."""
+    if not data:
+        return ""
+    return "\n".join(_serialize_node(data, 0)) + "\n"

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/unit/test_compose.py

@@ -267,7 +267,8 @@ 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, "supervise": True}]:
+        cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
+        for kwargs in cases:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])

tests/unit/test_git_gate_host_key.py

@@ -0,0 +1,698 @@
+"""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 (
+    add_host_key_to_frontmatter,
+    find_and_update_bottle_file,
+    find_repo_bottle_file,
+    prompt_tty,
+    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)
+
+
+# ---------------------------------------------------------------------------
+# add_host_key_to_frontmatter
+# ---------------------------------------------------------------------------
+
+
+class TestAddHostKeyToFrontmatter(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_inserts_host_key_at_repo_level(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        # Must be a sibling of `url` and `key`, not nested inside `key`
+        lines = result.splitlines()
+        hk_line = next(l for l in lines if "host_key" in l)
+        url_line = next(l for l in lines if "url:" in l)
+        self.assertEqual(
+            len(hk_line) - len(hk_line.lstrip()),
+            len(url_line) - len(url_line.lstrip()),
+        )
+
+    def test_preserves_body(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA")
+        self.assertIn("# Body text here", result)
+
+    def test_preserves_other_repos(self) -> None:
+        text = """\
+---
+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 = add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA")
+        self.assertIn("host_key: ssh-ed25519 AAAA", result)
+        self.assertIn("second:", result)
+
+    def test_no_change_when_host_key_already_present(self) -> None:
+        text = self._FILE.replace(
+            "        path: /dev/null\n",
+            "        path: /dev/null\n      host_key: ssh-ed25519 EXISTING\n",
+        )
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW")
+        self.assertNotIn("NEW", result)
+
+    def test_no_change_when_repo_not_found(self) -> None:
+        result = add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA")
+        self.assertEqual(self._FILE, result)
+
+    def test_no_change_without_frontmatter(self) -> None:
+        text = "No frontmatter here\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_no_closing_delimiter(self) -> None:
+        text = "---\ngit-gate:\n  repos:\n    myrepo:\n      url: x\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_git_gate_missing(self) -> None:
+        text = "---\nenv:\n  FOO: bar\n---\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_repos_missing(self) -> None:
+        text = "---\ngit-gate:\n  foo: bar\n---\n"
+        result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA")
+        self.assertEqual(text, result)
+
+    def test_no_change_when_repo_entry_not_a_dict(self) -> None:
+        text = "---\ngit-gate:\n  repos:\n    - item\n---\n"
+        result = add_host_key_to_frontmatter(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_returns_false_when_file_unreadable_on_write(self) -> None:
+        # find_repo_bottle_file succeeds, but the second read (for writing)
+        # raises OSError — e.g. file removed or permissions changed.
+        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.find_repo_bottle_file",
+                return_value=path,
+            ), patch.object(Path, "read_text", side_effect=OSError("Permission denied")):
+                result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA")
+            self.assertFalse(result)
+
+    def test_skips_when_update_produces_no_change(self) -> None:
+        # add_host_key_to_frontmatter returns the original text unchanged
+        # (e.g. because the entry is absent from the parsed structure).
+        # find_and_update_bottle_file 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.add_host_key_to_frontmatter",
+                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())
+
+
+# ---------------------------------------------------------------------------
+# find_repo_bottle_file
+# ---------------------------------------------------------------------------
+
+
+class TestFindRepoBottleFile(unittest.TestCase):
+    _CONTENT = """\
+---
+git-gate:
+  repos:
+    myrepo:
+      url: ssh://git@host/org/repo.git
+      key:
+        provider: static
+        path: /dev/null
+---
+"""
+
+    def _write(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_returns_path_when_repo_found(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            path = self._write(d, "dev", self._CONTENT)
+            result = find_repo_bottle_file(Path(d), "myrepo")
+            self.assertEqual(path, result)
+
+    def test_returns_none_when_no_matching_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            self._write(d, "dev", self._CONTENT)
+            self.assertIsNone(find_repo_bottle_file(Path(d), "other"))
+
+    def test_returns_none_when_host_key_already_set(self) -> None:
+        content = self._CONTENT.replace("        path: /dev/null\n",
+                                        "        path: /dev/null\n      host_key: X\n")
+        with tempfile.TemporaryDirectory() as d:
+            self._write(d, "dev", content)
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_returns_none_for_nonexistent_dir(self) -> None:
+        self.assertIsNone(find_repo_bottle_file(Path("/nonexistent"), "myrepo"))
+
+    def test_skips_unreadable_file(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "unreadable.md").write_text(self._CONTENT)
+            with patch.object(Path, "read_text", side_effect=OSError("Permission denied")):
+                self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_skips_file_with_non_dict_git_gate(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "dev.md").write_text("---\ngit-gate:\n  - item\n---\n")
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_skips_file_with_non_dict_repos(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "dev.md").write_text("---\ngit-gate:\n  repos:\n    - item\n---\n")
+            self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo"))
+
+    def test_save_prompt_includes_filename(self) -> None:
+        """preflight prompt must name the discovered bottle file."""
+        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",
+            )
+
+            mock_prompt = MagicMock(side_effect=["y", "n"])  # confirm key, skip save
+            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", mock_prompt,
+            ), patch("sys.stderr"):
+                preflight_host_keys(manifest, headless=False, home_md=Path(home))
+
+            save_call = next(
+                c for c in mock_prompt.call_args_list if "Save" in c.args[0]
+            )
+            self.assertIn(str(bottle_file), save_call.args[0])
+
+
+# ---------------------------------------------------------------------------
+# 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"],  # confirm key only; no save prompt when no file found
+            ), 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_interactive_warns_when_write_fails_after_prompt(self) -> None:
+        # User says yes to save, but find_and_update_bottle_file fails.
+        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",
+            )
+
+            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(
+                "bot_bottle.git_gate_host_key.find_and_update_bottle_file",
+                return_value=False,
+            ), 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()

tests/unit/test_smolmachines_provision.py

@@ -503,15 +503,6 @@ class TestProvisionGitUser(unittest.TestCase):
         self.assertIn("bot@example.com", calls[0][0])
 
 
-class TestProxyHost(unittest.TestCase):
-    """_proxy_host returns the per-bottle address used by the forwarder."""
-
-    def test_returns_loopback_alias(self):
-        plan = _plan()
-        result = _launch._proxy_host(plan, "127.0.0.16")
-        self.assertEqual("127.0.0.16", result)
-
-
 class TestLaunchResourceWiring(unittest.TestCase):
     def test_allocate_resources_uses_loopback_alias_and_bundle_name(self):
         plan = _plan()

tests/unit/test_yaml_subset.py

@@ -6,7 +6,7 @@ import textwrap
 import unittest
 
 from bot_bottle.yaml_subset import YamlSubsetError
-from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
+from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset, serialize_yaml_subset
 
 
 def _y(s: str):
@@ -457,5 +457,141 @@ class TestEdgeAndErrorBranches(unittest.TestCase):
         self.assertEqual(({}, ""), parse_frontmatter(""))
 
 
+# ---------------------------------------------------------------------------
+# serialize_yaml_subset
+# ---------------------------------------------------------------------------
+
+
+class TestSerializeYamlSubset(unittest.TestCase):
+    def _roundtrip(self, text: str) -> None:
+        """Parse then serialize then re-parse; result must equal original parse."""
+        parsed = parse_yaml_subset(text)
+        serialized = serialize_yaml_subset(parsed)
+        self.assertEqual(parsed, parse_yaml_subset(serialized))
+
+    # --- scalar values ---------------------------------------------------------
+
+    def test_string_bare(self) -> None:
+        self.assertEqual("key: value\n", serialize_yaml_subset({"key": "value"}))
+
+    def test_string_empty_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": ""})
+        self.assertIn("key: ''", out)
+        self._roundtrip("key: ''\n")
+
+    def test_string_true_like_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": "true"})
+        self.assertIn("'true'", out)
+        self.assertEqual({"key": "true"}, parse_yaml_subset(out))
+
+    def test_string_int_like_is_quoted(self) -> None:
+        out = serialize_yaml_subset({"key": "42"})
+        self.assertIn("'42'", out)
+        self.assertEqual({"key": "42"}, parse_yaml_subset(out))
+
+    def test_string_special_start_char_is_quoted(self) -> None:
+        for ch in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">"):
+            out = serialize_yaml_subset({"key": ch + "rest"})
+            self.assertIn("'", out, msg=f"expected quoting for {ch!r}")
+
+    def test_none_emits_null(self) -> None:
+        self.assertEqual("key: null\n", serialize_yaml_subset({"key": None}))
+
+    def test_bool_true(self) -> None:
+        self.assertEqual("key: true\n", serialize_yaml_subset({"key": True}))
+
+    def test_bool_false(self) -> None:
+        self.assertEqual("key: false\n", serialize_yaml_subset({"key": False}))
+
+    def test_int(self) -> None:
+        self.assertEqual("key: 42\n", serialize_yaml_subset({"key": 42}))
+
+    # --- nested dict -----------------------------------------------------------
+
+    def test_nested_dict(self) -> None:
+        data: dict[str, object] = {"outer": {"inner": "val"}}
+        out = serialize_yaml_subset(data)
+        self.assertIn("outer:", out)
+        self.assertIn("  inner: val", out)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_empty_dict_value_inline(self) -> None:
+        out = serialize_yaml_subset({"key": {}})
+        self.assertIn("key: {}", out)
+
+    # --- lists -----------------------------------------------------------------
+
+    def test_list_of_scalars(self) -> None:
+        data: dict[str, object] = {"items": ["a", "b", "c"]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_empty_list_value_inline(self) -> None:
+        out = serialize_yaml_subset({"key": []})
+        self.assertIn("key: []", out)
+
+    def test_list_of_mappings(self) -> None:
+        data: dict[str, object] = {"items": [{"name": "alpha", "val": 1}, {"name": "beta", "val": 2}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_mapping_with_nested_dict(self) -> None:
+        data: dict[str, object] = {"items": [{"name": "x", "sub": {"k": "v"}}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_first_key_is_nested_dict(self) -> None:
+        # Covers the branch where the FIRST key of a mapping list item
+        # has a nested dict/list value (not a scalar).
+        data: dict[str, object] = {"items": [{"cfg": {"a": 1, "b": 2}, "name": "x"}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    def test_list_item_mapping_continuation_keys(self) -> None:
+        data: dict[str, object] = {"items": [{"a": 1, "b": 2, "c": 3}]}
+        out = serialize_yaml_subset(data)
+        self.assertEqual(data, parse_yaml_subset(out))
+
+    # --- empty / trivial -------------------------------------------------------
+
+    def test_empty_dict(self) -> None:
+        self.assertEqual("", serialize_yaml_subset({}))
+
+    # --- realistic bottle frontmatter round-trip --------------------------------
+
+    def test_bottle_frontmatter_round_trip(self) -> None:
+        text = textwrap.dedent("""\
+            git-gate:
+              repos:
+                myrepo:
+                  url: ssh://git@host/org/repo.git
+                  key:
+                    provider: static
+                    path: /dev/null
+        """)
+        self._roundtrip(text)
+
+    def test_host_key_inserted_at_repo_level(self) -> None:
+        """The canonical use-case: host_key lands as a sibling of url/key."""
+        from typing import cast
+        text = textwrap.dedent("""\
+            git-gate:
+              repos:
+                myrepo:
+                  url: ssh://git@host/org/repo.git
+                  key:
+                    provider: static
+                    path: /dev/null
+        """)
+        data = parse_yaml_subset(text)
+        repos = cast(dict[str, object], cast(dict[str, object], data["git-gate"])["repos"])
+        cast(dict[str, object], repos["myrepo"])["host_key"] = "ssh-ed25519 AAAA"
+        out = serialize_yaml_subset(data)
+        parsed_back = parse_yaml_subset(out)
+        repo = cast(dict[str, object], cast(dict[str, object], cast(dict[str, object], parsed_back["git-gate"])["repos"])["myrepo"])
+        self.assertEqual("ssh-ed25519 AAAA", repo["host_key"])
+        self.assertIn("provider", cast(dict[str, object], repo["key"]))
+
+
 if __name__ == "__main__":
     unittest.main()