fix: make config store schema explicit

bot_bottle/backend/__init__.py

@@ -75,9 +75,6 @@ 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"
@@ -306,13 +303,6 @@ 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 = loopback_ip
+        proxy_host = _proxy_host(plan, loopback_ip)
         plan = _start_bundle(plan, network, proxy_host, stack)
 
         agent_from_path = _agent_from_path(plan)
@@ -291,13 +291,14 @@ def _launch_vm(
 ) -> None:
     """Create, patch, and start the smolvm VM; register teardown.
 
-    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias.
-    This ensures the guest can only reach sidecar forwarders published
-    on that IP, not host localhost or another bottle's alias.
-    force_allowlist confirms the allowlist persisted (patching smolvm
-    0.8.0's silent-drop of --allow-cidr when combined with --from) and
-    fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
-    makes --from and --smolfile mutually exclusive."""
+    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias
+    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
+    ensures the guest can only reach bundle ports published on that IP,
+    not the container IP directly. force_allowlist confirms the
+    allowlist persisted (patching smolvm 0.8.0's silent-drop of
+    --allow-cidr when combined with --from) and fails closed if it
+    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
+    and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
     _smolvm.machine_create(
         plan.machine_name,
@@ -344,6 +345,12 @@ 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,7 +224,6 @@ 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

@@ -1,268 +0,0 @@
-"""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,11 +21,6 @@ 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
@@ -581,105 +576,3 @@ 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.411
+pyright>=1.1.300
 coverage>=7.0.0

tests/unit/test_compose.py

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

tests/unit/test_git_gate_host_key.py

@@ -1,698 +0,0 @@
-"""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,6 +503,15 @@ 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, serialize_yaml_subset
+from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
 
 
 def _y(s: str):
@@ -457,141 +457,5 @@ 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()