chore: update quality badges

- Coverage: 83% - Core coverage: 95% [skip ci]
fix(git-gate): defer dynamic key provisioning
2026-07-01 16:51:08 +00:00 · 2026-07-01 12:45:46 -04:00 · 2026-07-01 11:53:26 -04:00
18 changed files with 126 additions and 1417 deletions
@@ -5,8 +5,8 @@
 # bot-bottle
 [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
-[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
-[![core coverage](https://img.shields.io/badge/core%20coverage-96%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
+[![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
 **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
@@ -37,7 +37,10 @@ from pathlib import Path
 from typing import Callable, Generator
 from ...egress import egress_resolve_token_values
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import info, warn
 from . import network as network_mod
 from . import util as docker_mod
@@ -118,6 +121,11 @@ def launch(
        git_gate_plan = plan.git_gate_plan
        if git_gate_plan.upstreams:
            git_gate_plan = provision_git_gate_dynamic_keys(
                plan.manifest.bottle,
                git_gate_plan,
                git_gate_state_dir(plan.slug),
            )
            git_gate_plan = dataclasses.replace(
                git_gate_plan,
                internal_network=internal_network,
@@ -28,7 +28,10 @@ from ...egress import (
    egress_resolve_token_values,
    egress_sidecar_env_entries,
 )
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import die, info, warn
 from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
@@ -98,6 +101,8 @@ def launch(
        egress_network = egress_network_name(plan.slug)
        _create_networks(internal_network, egress_network, stack)
        plan = _provision_git_gate_keys(plan)
        sidecar_name = sidecar_container_name(plan.slug)
        container_mod.force_remove_container(sidecar_name)
        _start_sidecar_bundle(plan, sidecar_name, internal_network, egress_network)
@@ -241,6 +246,19 @@ def _stamp_agent_urls(
    )
 def _provision_git_gate_keys(
    plan: MacosContainerBottlePlan,
 ) -> MacosContainerBottlePlan:
    if not plan.git_gate_plan.upstreams:
        return plan
    git_gate_plan = provision_git_gate_dynamic_keys(
        plan.manifest.bottle,
        plan.git_gate_plan,
        git_gate_state_dir(plan.slug),
    )
    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
 def _stage_git_gate(plan: MacosContainerBottlePlan, sidecar_name: str) -> None:
    gp = plan.git_gate_plan
    if not gp.upstreams:
@@ -41,7 +41,10 @@ from ..docker.git_gate import (
    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
    GIT_GATE_HOOK_IN_CONTAINER,
 )
-from ...git_gate import revoke_git_gate_provisioned_keys
+from ...git_gate import (
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
 )
 from ...log import info, warn
 from ...bottle_state import (
    egress_state_dir,
@@ -174,6 +177,7 @@ def _start_bundle(
 ) -> SmolmachinesBottlePlan:
    """Build the BundleLaunchSpec, resolve token env, start the
    sidecar bundle container, and register teardown."""
    plan = _provision_git_gate_keys(plan)
    bundle_spec = _bundle_launch_spec(plan, network, loopback_ip)
    token_env = _resolve_token_env(plan, dict(os.environ))
    _bundle.ensure_bundle_image(bundle_spec.image)
@@ -182,6 +186,19 @@ def _start_bundle(
    return plan
 def _provision_git_gate_keys(
    plan: SmolmachinesBottlePlan,
 ) -> SmolmachinesBottlePlan:
    if not plan.git_gate_plan.upstreams:
        return plan
    git_gate_plan = provision_git_gate_dynamic_keys(
        plan.manifest.bottle,
        plan.git_gate_plan,
        git_gate_state_dir(plan.slug),
    )
    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
 def _discover_urls(
    plan: SmolmachinesBottlePlan,
    loopback_ip: str,
@@ -27,34 +27,12 @@ from .start import _launch_bottle
 def cmd_resume(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(prog=f"{PROG} resume", add_help=True)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument(
        "--headless",
        action="store_true",
        help=(
            "non-interactive rehydrate: deliver --prompt to the agent and "
            "skip the y/N preflight. For orchestrators / the freeze-rehydrate "
            "loop."
        ),
    )
    parser.add_argument(
        "--prompt",
        default=None,
        help="follow-up prompt delivered to the agent (required with --headless)",
    )
    parser.add_argument(
        "identity",
        help="bottle identity from a prior `start` (see its session-end output)",
    )
    args = parser.parse_args(argv)
    if args.prompt and not args.headless:
        die("--prompt is only valid with --headless")
    if args.headless and not args.prompt:
        die(
            "--headless requires --prompt: "
            "./cli.py resume <identity> --headless --prompt 'Address the review'"
        )
    metadata = read_metadata(args.identity)
    if metadata is None:
        die(
@@ -78,6 +56,4 @@ def cmd_resume(argv: list[str]) -> int:
        spec,
        dry_run=args.dry_run,
        backend_name=backend_name,
        assume_yes=args.headless,
        headless_prompt_text=args.prompt or "",
    )
@@ -1,165 +0,0 @@
 """Forge abstraction (PRD forge-native-integration, chunk 3).
 The `Forge` abstract class is the provider-agnostic surface a forge
 sidecar dispatches to: read issues/comments, post comments, edit
 descriptions, and the membership / PR lookups the orchestrator needs.
 Each forge (Gitea first) implements it; the sidecar protocol and the
 agent prompt stay forge-agnostic.
 `signal_done` is deliberately *not* a `Forge` method — completion is a
 sidecar concept relayed to the orchestrator over a queue dir, not a
 forge API operation.
 `ScopedForge` enforces the PRD's **read-anywhere / write-scoped** model:
 reads pass through to any issue/PR for context; writes are rejected
 unless the target is the assigned issue or one of its PRs. This bounds
 the blast radius of a prompt-injected agent below repo-wide API-key
 permissions.
 """
 from __future__ import annotations
 import abc
 from collections.abc import Iterable
 from dataclasses import dataclass
@dataclass(frozen=True)
 class Issue:
    """A forge issue (not a PR — see `PullRequest`)."""
    number: int
    title: str
    body: str
    state: str  # "open" | "closed"
@dataclass(frozen=True)
 class PullRequest:
    """A forge pull request. Kept distinct from `Issue` even though some
    forges model PRs as issues on the wire: the domain objects carry
    different data (a PR has merge state) and are read through different
    methods (`read_pr` vs `read_issue`)."""
    number: int
    title: str
    body: str
    state: str  # "open" | "closed"
    merged: bool
@dataclass(frozen=True)
 class Comment:
    id: int
    user: str  # login of the comment author
    body: str
 class ForgeScopeError(PermissionError):
    """Raised by `ScopedForge` when a write targets an issue/PR outside
    the assigned scope."""
 class Forge(abc.ABC):
    """Provider-agnostic forge operations. Implementations wrap a
    per-provider HTTP client and translate to `Issue` / `Comment`."""
    @abc.abstractmethod
    def read_issue(self, number: int) -> Issue:
        """Read an issue body (read-anywhere)."""
    @abc.abstractmethod
    def read_pr(self, number: int) -> PullRequest:
        """Read a pull request, including its merge state (read-anywhere)."""
    @abc.abstractmethod
    def read_comments(self, number: int) -> list[Comment]:
        """Read a thread's comments (read-anywhere)."""
    @abc.abstractmethod
    def post_comment(self, number: int, body: str) -> None:
        """Post a comment to an issue or PR (write-scoped)."""
    @abc.abstractmethod
    def update_description(self, number: int, body: str) -> None:
        """Replace an issue or PR body (write-scoped)."""
    @abc.abstractmethod
    def is_org_member(self, org: str, username: str) -> bool:
        """Whether `username` is a member of `org`."""
    @abc.abstractmethod
    def get_pr_for_issue(self, number: int) -> int | None:
        """The PR number linked to an issue, or None when there is none."""
    @abc.abstractmethod
    def is_pr_open(self, number: int) -> bool:
        """Whether the given PR is still open."""
 class ScopedForge(Forge):
    """Read-anywhere / write-scoped wrapper around a concrete `Forge`.
    `post_comment` and `update_description` are rejected with
    `ForgeScopeError` unless the target number is the assigned issue or
    one of the assigned PRs. Every other method delegates unchanged, so
    reads, membership checks, and PR lookups work against any number for
    context.
    The writable set is fixed at construction. The sidecar reconstructs
    a `ScopedForge` when a PR is discovered (`get_pr_for_issue`) so the
    new PR becomes writable; this class does not mutate its own scope.
    """
    def __init__(
        self,
        inner: Forge,
        *,
        assigned_issue: int,
        assigned_prs: Iterable[int] = (),
    ) -> None:
        self._inner = inner
        self._assigned_issue = assigned_issue
        self._writable = {assigned_issue, *assigned_prs}
    @property
    def writable(self) -> frozenset[int]:
        return frozenset(self._writable)
    def _check_write(self, number: int) -> None:
        if number not in self._writable:
            allowed = ", ".join(str(n) for n in sorted(self._writable))
            raise ForgeScopeError(
                f"write to #{number} denied: out of assigned scope "
                f"(writable: {allowed})"
            )
    # --- read-anywhere: pass through --------------------------------------
    def read_issue(self, number: int) -> Issue:
        return self._inner.read_issue(number)
    def read_pr(self, number: int) -> PullRequest:
        return self._inner.read_pr(number)
    def read_comments(self, number: int) -> list[Comment]:
        return self._inner.read_comments(number)
    def is_org_member(self, org: str, username: str) -> bool:
        return self._inner.is_org_member(org, username)
    def get_pr_for_issue(self, number: int) -> int | None:
        return self._inner.get_pr_for_issue(number)
    def is_pr_open(self, number: int) -> bool:
        return self._inner.is_pr_open(number)
    # --- write-scoped: check then delegate --------------------------------
    def post_comment(self, number: int, body: str) -> None:
        self._check_write(number)
        self._inner.post_comment(number, body)
    def update_description(self, number: int, body: str) -> None:
        self._check_write(number)
        self._inner.update_description(number, body)
@@ -1,174 +0,0 @@
 """Gitea HTTP client + `GiteaForge` (PRD forge-native-integration, chunk 3).
 `GiteaClient` is the thin stdlib-only HTTP transport (mirrors
 `deploy_key_provisioner.py`: `urllib.request`, bounded timeouts,
 structured error bodies). `GiteaForge` adapts it to the provider-agnostic
 `Forge` surface.
 Unlike the option-2 design, the token is held here (the sidecar process
 owns it) and passed to the client directly — there is no agent-side
 cred-proxy route, because the agent never makes forge calls. The HTTP
 client is the one piece shared with `GiteaDeployKeyProvisioner`; the two
 are deliberately *not* unified behind a common abstract base (see the
 deferral note in the PRD).
 """
 from __future__ import annotations
 import json
 import urllib.error
 import urllib.request
 from typing import Any
 from ..forge.base import Comment, Forge, Issue, PullRequest
 # Bound every Gitea call: a hung instance must not stall the sidecar.
 _API_TIMEOUT_SECS = 30
 class GiteaClient:
    """Thin authenticated HTTP client for one repo's Gitea API.
    `api_url` is the API base *including* `/api/v1` (matching the
    `FORGE_GITEA_API` env var), e.g. `https://gitea.example.com/api/v1`.
    """
    def __init__(self, *, api_url: str, owner: str, repo: str, token: str) -> None:
        self._api_url = api_url.rstrip("/")
        self._owner = owner
        self._repo = repo
        self._token = token
    # --- low-level request -------------------------------------------------
    def _request(
        self, method: str, path: str, *, body: dict[str, Any] | None = None
    ) -> tuple[int, Any]:
        """Issue an authenticated request. Returns `(status, parsed_json)`;
        parsed_json is None when the response has no body. Raises
        `RuntimeError` on any non-2xx except where callers special-case
        the HTTPError themselves (membership 404)."""
        url = f"{self._api_url}{path}"
        data = json.dumps(body).encode() if body is not None else None
        headers = {"Authorization": f"token {self._token}"}
        if data is not None:
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS) as resp:
            raw = resp.read()
            parsed = json.loads(raw) if raw else None
            return resp.status, parsed
    def _repo_path(self, suffix: str) -> str:
        return f"/repos/{self._owner}/{self._repo}{suffix}"
    # --- operations --------------------------------------------------------
    def is_org_member(self, org: str, username: str) -> bool:
        """GET /orgs/{org}/members/{username}: 2xx → member, 404 → not.
        Other errors propagate so a misconfigured token fails loudly."""
        url = f"{self._api_url}/orgs/{org}/members/{username}"
        req = urllib.request.Request(
            url, headers={"Authorization": f"token {self._token}"}, method="GET"
        )
        try:
            with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS):
                return True
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                return False
            raise RuntimeError(
                f"org membership check failed for {org}/{username}: "
                f"HTTP {exc.code} — {_read_error_body(exc)}"
            ) from exc
    def get_issue(self, number: int) -> dict[str, Any]:
        _status, body = self._request("GET", self._repo_path(f"/issues/{number}"))
        return body or {}
    def get_comments(self, number: int) -> list[dict[str, Any]]:
        _status, body = self._request(
            "GET", self._repo_path(f"/issues/{number}/comments")
        )
        return body or []
    def post_comment(self, number: int, body: str) -> None:
        self._request(
            "POST",
            self._repo_path(f"/issues/{number}/comments"),
            body={"body": body},
        )
    def patch_issue_body(self, number: int, body: str) -> None:
        self._request(
            "PATCH", self._repo_path(f"/issues/{number}"), body={"body": body}
        )
    def get_pull(self, number: int) -> dict[str, Any]:
        _status, body = self._request("GET", self._repo_path(f"/pulls/{number}"))
        return body or {}
 class GiteaForge(Forge):
    """`Forge` over a `GiteaClient`."""
    def __init__(self, client: GiteaClient) -> None:
        self._client = client
    def read_issue(self, number: int) -> Issue:
        raw = self._client.get_issue(number)
        return Issue(
            number=int(raw.get("number", number)),
            title=str(raw.get("title", "")),
            body=str(raw.get("body", "") or ""),
            state=str(raw.get("state", "")),
        )
    def read_pr(self, number: int) -> PullRequest:
        raw = self._client.get_pull(number)
        return PullRequest(
            number=int(raw.get("number", number)),
            title=str(raw.get("title", "")),
            body=str(raw.get("body", "") or ""),
            state=str(raw.get("state", "")),
            merged=bool(raw.get("merged", False)),
        )
    def read_comments(self, number: int) -> list[Comment]:
        return [
            Comment(
                id=int(c.get("id", 0)),
                user=str((c.get("user") or {}).get("login", "")),
                body=str(c.get("body", "") or ""),
            )
            for c in self._client.get_comments(number)
        ]
    def post_comment(self, number: int, body: str) -> None:
        self._client.post_comment(number, body)
    def update_description(self, number: int, body: str) -> None:
        self._client.patch_issue_body(number, body)
    def is_org_member(self, org: str, username: str) -> bool:
        return self._client.is_org_member(org, username)
    def get_pr_for_issue(self, number: int) -> int | None:
        """Gitea models a PR as an issue with the same number, exposing a
        `pull_request` object on the issue. When the queried number is
        itself a PR, return it; otherwise None. (The orchestrator tracks
        the issue→PR mapping in forge state for the cross-number case.)"""
        raw = self._client.get_issue(number)
        if raw.get("pull_request"):
            return int(raw.get("number", number))
        return None
    def is_pr_open(self, number: int) -> bool:
        return self.read_pr(number).state == "open"
 def _read_error_body(exc: urllib.error.HTTPError) -> str:
    try:
        return exc.read().decode("utf-8", errors="replace")
    except Exception:  # pylint: disable=broad-exception-caught
        return ""
@@ -1,171 +0,0 @@
 """Forge state persistence (PRD forge-native-integration, chunk 2).
 The orchestrator tracks one record per forge-targeted issue so it can
 map an incoming webhook back to the bottle handling it, drive the
 freeze / rehydrate loop, and run the watchdog.
 State is stored in a local SQLite database in `~/.bot-bottle/`. Access
 goes through the thin `ForgeStateStore` CRUD interface so the backing
 store (location or engine) can be swapped without touching callers;
 `SqliteForgeStateStore` is the first implementation.
 """
 from __future__ import annotations
 import abc
 import json
 import sqlite3
 from dataclasses import dataclass, field
 from pathlib import Path
 from ...supervise import bot_bottle_root
 _DB_FILENAME = "bot-bottle.db"
 # Lifecycle: a bottle is launched (running), frozen on the done signal,
 # and destroyed when the PR closes.
 STATUS_RUNNING = "running"
 STATUS_FROZEN = "frozen"
 STATUS_DESTROYED = "destroyed"
@dataclass
 class ForgeState:
    """One forge-targeted issue's bottle lifecycle record."""
    owner: str
    repo: str
    issue_number: int
    slug: str
    agent_name: str
    bottle_names: list[str] = field(default_factory=list)
    backend_name: str = ""
    agent_git_user: str = ""
    pr_number: int | None = None
    status: str = STATUS_RUNNING
    last_checkin_at: str = ""
 class ForgeStateStore(abc.ABC):
    """Thin CRUD surface over forge state. Implementations back it with a
    concrete store; callers depend only on this interface so the storage
    location/engine is swappable."""
    @abc.abstractmethod
    def upsert(self, state: ForgeState) -> None:
        """Insert or replace the record keyed by (owner, repo, issue)."""
    @abc.abstractmethod
    def get(self, owner: str, repo: str, issue_number: int) -> ForgeState | None:
        """Fetch one record, or None when absent."""
    @abc.abstractmethod
    def delete(self, owner: str, repo: str, issue_number: int) -> None:
        """Remove a record. Missing is success (idempotent)."""
    @abc.abstractmethod
    def all(self) -> list[ForgeState]:
        """Every record, for the status table and the watchdog sweep."""
 def default_db_path() -> Path:
    return bot_bottle_root() / _DB_FILENAME
 class SqliteForgeStateStore(ForgeStateStore):
    """SQLite-backed `ForgeStateStore`. The database lives at
    `~/.bot-bottle/bot-bottle.db` by default; pass `db_path` to point at
    a different location (tests, alternate homes)."""
    def __init__(self, db_path: Path | None = None) -> None:
        self._db_path = db_path or default_db_path()
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        with self._connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS forge_state (
                    owner           TEXT    NOT NULL,
                    repo            TEXT    NOT NULL,
                    issue_number    INTEGER NOT NULL,
                    slug            TEXT    NOT NULL,
                    agent_name      TEXT    NOT NULL,
                    bottle_names    TEXT    NOT NULL,
                    backend_name    TEXT    NOT NULL,
                    agent_git_user  TEXT    NOT NULL,
                    pr_number       INTEGER,
                    status          TEXT    NOT NULL,
                    last_checkin_at TEXT    NOT NULL,
                    PRIMARY KEY (owner, repo, issue_number)
                )
                """
            )
    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self._db_path)
        conn.row_factory = sqlite3.Row
        return conn
    def upsert(self, state: ForgeState) -> None:
        with self._connect() as conn:
            conn.execute(
                """
                INSERT OR REPLACE INTO forge_state (
                    owner, repo, issue_number, slug, agent_name,
                    bottle_names, backend_name, agent_git_user,
                    pr_number, status, last_checkin_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    state.owner,
                    state.repo,
                    state.issue_number,
                    state.slug,
                    state.agent_name,
                    json.dumps(state.bottle_names),
                    state.backend_name,
                    state.agent_git_user,
                    state.pr_number,
                    state.status,
                    state.last_checkin_at,
                ),
            )
    def get(self, owner: str, repo: str, issue_number: int) -> ForgeState | None:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM forge_state "
                "WHERE owner = ? AND repo = ? AND issue_number = ?",
                (owner, repo, issue_number),
            ).fetchone()
        return _row_to_state(row) if row is not None else None
    def delete(self, owner: str, repo: str, issue_number: int) -> None:
        with self._connect() as conn:
            conn.execute(
                "DELETE FROM forge_state "
                "WHERE owner = ? AND repo = ? AND issue_number = ?",
                (owner, repo, issue_number),
            )
    def all(self) -> list[ForgeState]:
        with self._connect() as conn:
            rows = conn.execute(
                "SELECT * FROM forge_state ORDER BY owner, repo, issue_number"
            ).fetchall()
        return [_row_to_state(row) for row in rows]
 def _row_to_state(row: sqlite3.Row) -> ForgeState:
    return ForgeState(
        owner=row["owner"],
        repo=row["repo"],
        issue_number=row["issue_number"],
        slug=row["slug"],
        agent_name=row["agent_name"],
        bottle_names=json.loads(row["bottle_names"]),
        backend_name=row["backend_name"],
        agent_git_user=row["agent_git_user"],
        pr_number=row["pr_number"],
        status=row["status"],
        last_checkin_at=row["last_checkin_at"],
    )
@@ -30,7 +30,6 @@ backend-specific and lives on concrete subclasses (see
 from __future__ import annotations
 import dataclasses
 from abc import ABC
 from dataclasses import dataclass
 from pathlib import Path
@@ -53,6 +52,7 @@ from .git_gate_render import (
    _gitconfig_validate_value,
 )
 from .git_gate_provision import (
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
    _provision_dynamic_key,
    _resolve_identity_file,
@@ -93,20 +93,14 @@ class GitGate(ABC):
        entrypoint, pre-receive hook, and access-hook scripts (mode
        600) under `stage_dir`. Pure host-side, no docker subprocess.
-        For `gitea` key entries, also generates and registers
+        For `gitea` key entries, the returned upstream intentionally
-        a fresh deploy key via the forge API and writes the private key
+        has an empty identity file. Backend launch fills that in after
-        + key ID to `stage_dir`.
+        the operator confirms the preflight.
        Returned plan is incomplete: the launch step must fill
        `internal_network` / `egress_network` via `dataclasses.replace`
        before passing the plan to `.start`."""
-        upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
+        upstreams = git_gate_upstreams_for_bottle(bottle)
        for i, entry in enumerate(bottle.git):
            upstreams_list[i] = dataclasses.replace(
                upstreams_list[i],
                identity_file=_resolve_identity_file(entry, slug, stage_dir),
            )
        upstreams = tuple(upstreams_list)
        entrypoint = stage_dir / "git_gate_entrypoint.sh"
        entrypoint.write_text(git_gate_render_entrypoint(upstreams))
        entrypoint.chmod(0o600)
@@ -162,6 +156,7 @@ __all__ = [
    "git_gate_render_entrypoint",
    "git_gate_render_hook",
    "git_gate_render_access_hook",
    "provision_git_gate_dynamic_keys",
    "revoke_git_gate_provisioned_keys",
    "_gitconfig_validate_value",
    "_provision_dynamic_key",
@@ -9,10 +9,16 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.
 from __future__ import annotations
 import os
 import dataclasses
 from pathlib import Path
 from typing import TYPE_CHECKING
 from .log import info
 from .manifest import ManifestBottle, ManifestGitEntry
 from .git_gate_render import GitGateUpstream
 if TYPE_CHECKING:
    from .git_gate import GitGatePlan
 def _provision_dynamic_key(
    entry: ManifestGitEntry,
@@ -95,8 +101,45 @@ def _resolve_identity_file(entry: ManifestGitEntry, slug: str, stage_dir: Path)
    return entry.IdentityFile
 def provision_git_gate_dynamic_keys(
    bottle: ManifestBottle,
    plan: "GitGatePlan",
    stage_dir: Path,
 ) -> "GitGatePlan":
    """Provision dynamic git-gate keys and return an updated plan.
    This runs during backend launch, after the operator confirms the
    preflight. Plan preparation intentionally stays side-effect-light:
    dry-runs and aborted launches must not create remote deploy keys.
    """
    if not plan.upstreams:
        return plan
    upstreams_by_name: dict[str, GitGateUpstream] = {
        upstream.name: upstream for upstream in plan.upstreams
    }
    updated: list[GitGateUpstream] = []
    for entry in bottle.git:
        upstream = upstreams_by_name.get(entry.Name)
        if upstream is None:
            continue
        if entry.Key.provider == "gitea":
            identity_file = _provision_dynamic_key(entry, plan.slug, stage_dir)
            upstream = dataclasses.replace(upstream, identity_file=identity_file)
        updated.append(upstream)
    if len(updated) != len(plan.upstreams):
        updated_names = {u.name for u in updated}
        for upstream in plan.upstreams:
            if upstream.name not in updated_names:
                updated.append(upstream)
    return dataclasses.replace(plan, upstreams=tuple(updated))
 __all__ = [
    "revoke_git_gate_provisioned_keys",
    "provision_git_gate_dynamic_keys",
    "_provision_dynamic_key",
    "_resolve_identity_file",
 ]
@@ -16,11 +16,16 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import urlsplit
 from .git_gate import GIT_GATE_TIMEOUT_SECS
 DEFAULT_PORT = 9420
 # Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than
 # imported: this module ships as a flat top-level sibling in the sidecar
 # bundle image (see Dockerfile.sidecars), not as part of the bot_bottle
 # package, so `bot_bottle.git_gate` and its dependency chain aren't
 # available at runtime.
 GIT_GATE_TIMEOUT_SECS = 15
 # Bound memory use while still allowing ordinary git push packfiles.
 MAX_BODY_BYTES = 100 * 1024 * 1024
@@ -1,439 +0,0 @@
 # PRD prd-new: Forge native integration
 - **Status:** Draft
 - **Author:** claude
 - **Created:** 2026-06-29
 - **Issue:** #317
 ## Summary
 Add a webhook-driven orchestration layer that lets Gitea issues and PR comments
 drive bot-bottle sessions end-to-end with no operator in the loop for the happy
 path. An issue assigned to a member of the configured agent org and labelled
 with an agent name triggers a headless bottle launch; the bottle processes the
 issue, opens a PR, and interacts with the forge through a **forge sidecar** —
 the agent never touches the Gitea API or its credentials directly. The agent
 calls `signal_done(status, summary)` on the sidecar when a work unit is
 complete; the sidecar relays that to the orchestrator over a queue dir (the same
 pattern as the supervise sidecar), so completion is an unambiguous in-band
 signal rather than a comment the orchestrator has to parse. The orchestrator
 freezes the bottle. Subsequent PR comments rehydrate the frozen bottle. The
 bottle is destroyed when the PR closes.
 The forge sidecar is backed by a `Forge` abstract class with per-provider
 implementations (Gitea first), so the agent's prompts and the sidecar protocol
 stay forge-agnostic. The sidecar logs forge operations semantically ("read PR
 description", "posted comment", "signalled done"), giving richer provenance than
 post-hoc egress-byte parsing, and enforces a **read-anywhere / write-scoped**
 permission model: the agent may read for context but may only write to the
 issue and PRs it was assigned.
 Run provenance is exposed through a **provenance API** (the sidecar's structured
 operation log plus the run's metadata), not posted back into the forge. We do
 not surface a provenance footer in the PR — the audit record lives behind the
 API where it can be retained and queried, rather than as an editable comment.
 The separation of concerns across the two layers: bot-bottle owns the headless
 launch primitives, the forge sidecar + `Forge` abstraction, and forge state.
 `bot-bottle-orchestrator` (separate binary) owns the webhook listener, bottle
 lifecycle loop, and monitoring dashboard; it calls into bot-bottle via
 `./cli.py orchestrate`, a thin wrapper command. This PRD covers bot-bottle's
 side of that contract.
 ## Problem
 Today an operator must open the TUI, select an agent and bottle, confirm the
 preflight, and type prompts interactively. This blocks "issue → PR" automation
 and produces no durable audit record of what the agent did. The security model
 already provides the right isolation and egress controls, and `start --headless`
 (#315) already gives `bot-bottle-orchestrator` a non-interactive launch path.
 The missing pieces are a headless `resume` counterpart for rehydrating frozen
 bottles, a forge-interaction surface the agent uses to read context, post
 comments, and signal completion, and the provenance trail that makes the audit
 story legible to reviewers on every PR.
 That forge-interaction surface could be built two ways: (2) give the agent the
 Gitea API directly with cred-proxy injecting the token, or (3) put a forge
 sidecar between the agent and the forge. This PRD takes **option 3**. The
 deciding factors: a sidecar `signal_done` call is an unambiguous completion
 signal where comment-parsing is a correctness risk that surfaces in production;
 the sidecar produces a semantic audit trail rather than HTTP bytes, which is
 load-bearing for provenance (the stated product priority); and the sidecar can
 enforce scope tighter than repo-wide API-key permissions, reducing blast radius
 for a prompt-injected agent. The costs — a second sidecar process per forge run,
 a new failure mode if it crashes, and per-forge implementation cost — are
 accepted as the price of those properties.
 ## Goals / Success Criteria
 1. Headless launch already exists: `./cli.py start <agent> --headless --prompt`
   (#315) runs non-interactively with no TUI selectors or y/N preflight. This
   PRD builds on it rather than re-introducing it. The remaining gap is a
   matching headless `resume` path (`./cli.py resume --headless`), since
   rehydrating a frozen bottle for a new prompt is required by the freeze /
   rehydrate loop and `resume` has no non-interactive entry point today.
 2. An issue assigned to a member of the configured org (`FORGE_ORG`, default
   `bot-bottle`) and labelled `bot-bottle:<agent-name>` is the trigger
   convention. Org membership is verified via the Gitea API at event time.
 3. Forge-targeted bottles run a **forge sidecar** that exposes a small,
   forge-agnostic API (comment/issue/PR CRUD plus `signal_done`) over the same
   queue-dir + HTTP/JSON-RPC machinery as the supervise sidecar. The agent calls
   the sidecar; it never sees the forge token or forge-specific endpoints.
 4. The sidecar is backed by a `Forge` abstract class. Gitea is the first
   concrete implementation; adding a forge means a new subclass, not changes to
   the agent prompt or sidecar protocol. The sidecar enforces a read-anywhere /
   write-scoped model: writes are limited to the assigned issue and its PRs;
   reads are unrestricted for context.
 5. The agent calls `signal_done(status, summary)` on the sidecar when a work
   unit is complete; the sidecar relays it to the orchestrator over a queue dir.
   This is the done signal — no comment parsing. A watchdog timeout
   (configurable, default 30 min) causes the orchestrator to treat the run as
   done-without-self-report if the agent exits without signalling.
 6. Run provenance (agent name, bottle name(s), slug, timing, exit code,
   gitleaks result, egress summary, and the sidecar's semantic operation log)
   is available through a provenance API. It is **not** surfaced as a PR footer
   or any other forge comment.
 7. Forge state (issue → slug, status) is persisted in a local SQLite database
   under `~/.bot-bottle/` and survives orchestrator restarts.
 8. `./cli.py orchestrate status` lists active forge-managed bottles and their
   issue/PR URLs.
 9. Unit tests cover: label parsing, org-membership check path, forge state
   store CRUD (SQLite), headless launch arg construction, forge env var
   injection, sidecar request dispatch through the `Forge` abstraction,
   write-scope enforcement (reject writes outside the assigned issue/PRs), and
   `signal_done` queue relay.
 ## Non-goals
 - Webhook signature verification (HMAC-SHA256). Added as a follow-up.
 - The `bot-bottle-orchestrator` binary itself — this PRD covers bot-bottle's
  side of the interface only. The orchestrator is a separate project.
 - GitHub or GitLab support.
 - Multiple simultaneous forge bottles per issue.
 - Automatic retry on agent error exit.
 - Bottle destruction on issue close (PR close only; issue close is ambiguous).
 - Concurrent multi-issue handling (one blocking run per orchestrator process).
 - A monitoring dashboard (orchestrator-side concern).
 - Folding `DeployKeyProvisioner` into the `Forge` abstraction. Deploy-key
  provisioning runs at bottle-provision time on the host; the forge sidecar runs
  inside the bottle at agent time. The two have different lifecycles and actors,
  so coupling them into one class is deferred to a follow-up. This PRD only
  shares the Gitea HTTP client between them.
 ## Design
 ### Targeting convention
 An issue is forge-targeted when **both** hold:
 - At least one assignee is a member of the Gitea org named by `FORGE_ORG`
  (default `bot-bottle`). Checked via `GET /api/v1/orgs/{org}/members/{user}`.
 - At least one label has the prefix `bot-bottle:`. The suffix names the agent
  manifest, e.g. `bot-bottle:implementer` → agent `implementer`.
 `FORGE_ORG` is read at orchestrate-command startup. It is not embedded in
 manifests or state files; the orchestrator stamps its value into log output for
 auditability.
 An optional label `bot-bottle-bottle:<name>` overrides bottle selection. When
 absent the agent's default bottle is used.
 ### `./cli.py orchestrate` — the thin wrapper
 ```
 ./cli.py orchestrate start  --agent AGENT [--bottle BOTTLE ...] --prompt PROMPT
                            [--label LABEL] [--backend BACKEND]
 ./cli.py orchestrate resume --slug SLUG --prompt PROMPT [--backend BACKEND]
 ./cli.py orchestrate status
 ```
 `orchestrate start` is a thin shim over the already-shipped `start --headless`
 (#315): it forwards agent / bottle / label / prompt and adds the forge-specific
 wiring (`forge_env`, sidecar launch). It does not re-implement headless launch.
 The caller (`bot-bottle-orchestrator`) manages freeze, state, and the forge
 sidecar's done signal around it.
 `orchestrate resume` is the shim over the new `resume --headless` (below).
 `orchestrate status` prints the forge state table.
 ### Headless primitives — what exists vs. what's new
 Headless **start** already shipped in #315 and this PRD reuses it as-is:
 - `./cli.py start <agent> --headless --prompt TEXT` — no TUI selectors, no y/N
  preflight. Internally `_start_headless()` calls the shared `_launch_bottle()`
  with `assume_yes=True` and `headless_prompt_text=prompt`.
 - The prompt is delivered through `AgentProvider.headless_prompt(prompt)` —
  claude `-p`, codex positional, pi `-p`. The orchestrator does **not** hand-roll
  agent args; it relies on this provider abstraction. (An earlier draft proposed
  `start_headless` / `attach_agent_headless` helpers that constructed
  `--no-interactive`/`-p` directly — those are dropped as redundant with, and
  divergent from, what #315 merged.)
 Two additions are needed on top of #315:
 **1. A `forge_env` hook on the headless launch path.** The orchestrator needs to
 pass forge context + token through to the forge sidecar launched alongside the
 agent. This is a parameter threaded into `_launch_bottle` (the same core
 `start --headless` already uses), not a parallel launch function. The agent
 process itself does not receive the token.
 **2. `resume --headless`** — new in `bot_bottle/cli/resume.py`, mirroring the
 `--headless` flag on `start`:
 ```
 ./cli.py resume <slug> --headless --prompt TEXT
 ```
 It rehydrates a frozen bottle and runs one headless prompt via the same
 `assume_yes` + `headless_prompt` path, returning the agent's exit code. `resume`
 has no non-interactive entry point today, so this is genuinely new work rather
 than a rename of an existing helper.
 ### Forge sidecar
 Forge-targeted bottles run a forge sidecar alongside the agent, mirroring the
 supervise sidecar: a per-bottle process that exposes an HTTP/JSON-RPC endpoint
 over a Unix socket and relays events to the orchestrator through a queue dir.
 The agent calls the sidecar; the sidecar holds the forge token and makes the
 actual forge API calls. The agent never receives the credential and never sees a
 forge-specific endpoint — swapping Gitea for another forge does not change the
 agent prompt or the sidecar protocol.
 The sidecar is configured at launch from the forge context (owner, repo, issue,
 PR) and the token, supplied by the orchestrator — not baked into the agent
 manifest. Because the sidecar owns the token, forge traffic does not need a
 cred-proxy egress route on the agent; the agent's egress policy is unchanged by
 forge targeting.
 **Sidecar protocol** (forge-agnostic; each method maps to a `Forge` call):
 | Method | Scope | Purpose |
 |---|---|---|
 | `read_issue(number)` | read-anywhere | Read an issue body for context |
 | `read_pr(number)` | read-anywhere | Read a PR (incl. merge state) for context |
 | `read_comments(number)` | read-anywhere | Read a thread for context |
 | `post_comment(number, body)` | write-scoped | Post to the assigned issue/PR |
 | `update_description(number, body)` | write-scoped | Edit the assigned issue/PR body |
 | `signal_done(status, summary)` | — | Relay completion to the orchestrator |
 Issues and PRs are distinct domain objects (`Issue` vs `PullRequest`) read
 through distinct methods; a PR carries merge state an issue does not.
 **Scope enforcement** is read-anywhere / write-scoped: read methods accept any
 issue/PR number for context; write methods are rejected unless the target is the
 assigned issue or one of its PRs. This is tighter than Gitea's repo-wide API-key
 permissions and bounds the blast radius of a prompt-injected agent. Rejections
 are logged semantically (operation, target, reason) so the audit trail records
 attempted out-of-scope writes, not just allowed ones.
 **Semantic audit**: every sidecar call is logged as a structured operation
 ("read PR #318 description", "posted comment to #317", "signalled done:
 success") rather than as opaque HTTP bytes. This log feeds provenance directly,
 with no post-hoc egress-log parsing.
 ### `Forge` abstraction — `bot_bottle/contrib/forge/`
 The sidecar dispatches to a `Forge` abstract class. Each provider implements the
 operations behind the sidecar protocol:
 ```python
 class Forge(abc.ABC):
    @abc.abstractmethod
    def read_issue(self, number: int) -> Issue: ...
    @abc.abstractmethod
    def read_pr(self, number: int) -> PullRequest: ...
    @abc.abstractmethod
    def read_comments(self, number: int) -> list[Comment]: ...
    @abc.abstractmethod
    def post_comment(self, number: int, body: str) -> None: ...
    @abc.abstractmethod
    def update_description(self, number: int, body: str) -> None: ...
    @abc.abstractmethod
    def is_org_member(self, org: str, username: str) -> bool: ...
    @abc.abstractmethod
    def get_pr_for_issue(self, number: int) -> int | None: ...
    @abc.abstractmethod
    def is_pr_open(self, number: int) -> bool: ...
 ```
 `Issue` and `PullRequest` are separate frozen dataclasses — a PR adds `merged`.
 `ScopedForge` wraps a concrete `Forge` to enforce the read-anywhere /
 write-scoped model (`post_comment` / `update_description` raise `ForgeScopeError`
 outside the assigned issue and PRs).
 `GiteaForge` is the first and only concrete implementation in this PRD. It wraps
 the Gitea HTTP client (below). Adding GitHub or GitLab later is a new subclass;
 the sidecar, protocol, and agent prompt are untouched.
 > **Deferred:** `DeployKeyProvisioner` is *not* folded into `Forge` here.
 > Deploy-key provisioning runs on the host at provision time; the sidecar runs
 > in the bottle at agent time. They have different lifecycles and actors, so a
 > shared abstract base would couple two unrelated auth contexts. For now they
 > only share the Gitea HTTP client; a later PRD can revisit unification.
 ### Forge env vars
 The orchestrator passes forge context to the **sidecar** (not the agent) at
 launch. The agent does not need owner/repo/issue env vars to construct API
 calls, since it only names issue/PR numbers to the sidecar:
 | Var | Example | Purpose |
 |---|---|---|
 | `FORGE_GITEA_API` | `https://gitea.dideric.is/api/v1` | Base URL the sidecar calls |
 | `FORGE_OWNER` | `didericis` | Repo owner |
 | `FORGE_REPO` | `bot-bottle` | Repo name |
 | `FORGE_ISSUE_NUMBER` | `317` | Assigned issue (defines write scope) |
 | `FORGE_PR_NUMBER` | `318` | Assigned PR (empty until PR exists) |
 The agent's forge-specific prompt instructs it to call `signal_done` on the
 sidecar when a work unit is complete, and to use the sidecar for any
 comment/description writes. The instruction is forge-agnostic and is part of the
 forge prompt overlay, not the base agent manifest, so non-forge runs are
 unaffected.
 ### Done signal and watchdog
 The agent calls `signal_done(status, summary)` on the sidecar when it finishes a
 work unit. The sidecar writes the event to its queue dir; the orchestrator reads
 it and:
 1. Reads the forge state for `(owner, repo, issue_number)`.
 2. If `status == "running"`, treats the event as the done signal: freezes the
   bottle and sets `status = "frozen"`. Provenance is recorded via the
   provenance API — no comment is posted to the forge.
 Because completion is an explicit `signal_done` call, the orchestrator does not
 parse comment text to detect "done", and intermediate comments the agent posts
 mid-run cannot be mistaken for completion.
 **Watchdog**: the orchestrator tracks `last_checkin_at` in forge state, updated
 on each sidecar event. A background thread wakes every minute. If
 `now - last_checkin_at > FORGE_WATCHDOG_TIMEOUT` (default 30 min, configurable
 via env) and `status == "running"`, the orchestrator treats the run as
 done-without-self-report and freezes the bottle, flagging the run as incomplete
 in the provenance record.
 **Sidecar-death failure mode**: if the forge sidecar crashes mid-run the agent
 loses forge access while the bottle is otherwise healthy. The orchestrator
 detects a dead sidecar (socket/queue gone) the same way it detects a stalled
 agent and falls back to the watchdog path.
 ### Forge state — `bot_bottle/contrib/gitea/forge_state.py`
 State is stored in a local SQLite database at `~/.bot-bottle/bot-bottle.db`.
 Access goes through a thin CRUD interface, `ForgeStateStore`, so the storage
 location/engine can be swapped without touching callers. `SqliteForgeStateStore`
 is the first implementation.
 The `forge_state` table is keyed by `(owner, repo, issue_number)` and carries:
 `slug`, `agent_name`, `bottle_names` (JSON), `backend_name`, `agent_git_user`,
 `pr_number` (nullable), `status`, `last_checkin_at`.
 `status`: `"running"` | `"frozen"` | `"destroyed"`.
 Store interface:
 ```python
 class ForgeStateStore(abc.ABC):
    def upsert(self, state: ForgeState) -> None: ...
    def get(self, owner: str, repo: str, issue_number: int) -> ForgeState | None: ...
    def delete(self, owner: str, repo: str, issue_number: int) -> None: ...
    def all(self) -> list[ForgeState]: ...
 class SqliteForgeStateStore(ForgeStateStore):
    def __init__(self, db_path: Path | None = None) -> None: ...
 ```
 `upsert` uses `INSERT OR REPLACE` so a re-run for the same issue overwrites in
 place. The schema is created on first open.
 ### Provenance API
 Run provenance — agent, bottle(s), slug, timing, exit code, gitleaks result,
 egress summary, watchdog-fired flag, and the sidecar's semantic operation log —
 is exposed through a **provenance API**, not posted into the forge. There is no
 provenance footer or run-summary comment.
 The rationale (per the monetization positioning): a PR comment is mutable by any
 maintainer, unsigned, and per-PR, so it is worthless as an audit record and
 invites false trust. The authoritative record therefore lives behind the API,
 where it can be retained, queried, and (eventually) signed. Whether any
 projection of it ever appears in the forge is a separate, out-of-scope decision;
 this PR does not build one.
 The API surface itself (schema, transport, signing, retention) is **out of scope
 for this PRD** and belongs with the orchestrator / control-plane work. bot-bottle
 here only produces the raw material: the sidecar's semantic operation log and the
 run metadata the orchestrator collects.
 ### Gitea HTTP client — `bot_bottle/contrib/gitea/client.py`
 `GiteaForge` (and the existing `GiteaDeployKeyProvisioner`) share one thin HTTP
 client. Unlike the option-2 design, the token is held by the sidecar process and
 passed to the client directly — there is no agent-side cred-proxy route to
 inject it, because the agent never makes forge calls.
 ```python
 class GiteaClient:
    def __init__(self, *, api_url: str, owner: str, repo: str, token: str) -> None: ...
    def is_org_member(self, org: str, username: str) -> bool: ...
    def get_issue(self, number: int) -> dict: ...
    def get_comments(self, number: int) -> list[dict]: ...
    def post_comment(self, number: int, body: str) -> None: ...
    def patch_issue_body(self, number: int, body: str) -> None: ...
    def get_pull(self, number: int) -> dict: ...
 ```
 `GiteaForge` adapts this client to the `Forge` surface (mapping raw JSON to
 `Issue` / `PullRequest` / `Comment`). Sharing only the HTTP client (not an
 abstract base) is the deliberate boundary between the sidecar and the deploy-key
 provisioner — see the deferral note under the `Forge` abstraction.
 ### Implementation chunks
 1. **Headless additions on top of #315** — thread a `forge_env` parameter into
   the existing `_launch_bottle` core (the one `start --headless` already uses);
   add a `--headless` path to `cli/resume.py` reusing `assume_yes` +
   `headless_prompt`. No new `start_headless`/`attach_agent_headless` helpers.
   Tests: `forge_env` reaches the sidecar/`guest_env`; `resume --headless` skips
   the TUI and y/N preflight and returns the agent exit code.
 2. **Forge state** — `contrib/gitea/forge_state.py`: `ForgeState` dataclass,
   `ForgeStateStore` CRUD interface, `SqliteForgeStateStore`. Tests: round-trip,
   missing → None, `INSERT OR REPLACE` upsert, delete idempotent, `all()`
   ordering, persistence across store instances.
 3. **`Forge` abstraction + Gitea client** — `contrib/forge/base.py` (`Forge`
   ABC, `ScopedForge`, `Issue` / `PullRequest` / `Comment`) and
   `contrib/gitea/client.py` + `GiteaForge`: `is_org_member`, `read_issue`,
   `read_pr`, `read_comments`, `post_comment`, `update_description`,
   `get_pr_for_issue`, `is_pr_open`. Tests: mock `urllib.request.urlopen`,
   assert payloads and 404-as-false for membership; `ScopedForge` write-scope
   enforcement.
 4. **Forge sidecar** — sidecar process exposing the protocol over a Unix socket,
   queue-dir relay, write-scope enforcement, semantic op log, `signal_done`.
   Reuses the supervise sidecar bundle machinery. Tests: dispatch each method to
   the `Forge`, reject out-of-scope writes, `signal_done` writes a queue event,
   scope-rejection is logged.
 5. **`./cli.py orchestrate`** — `cli/orchestrate.py` with `start`, `resume`,
   `status` subcommands wired into `cli.py`; `start` launches the forge sidecar
   alongside the agent for forge-targeted runs. Tests: arg parsing, `start`
   delegates to `start --headless`, `resume` delegates to `resume --headless`.
 ## Provenance
 Run provenance is captured (sidecar semantic operation log + run metadata) and
 exposed through a provenance API. It is deliberately **not** surfaced in the
 forge — no footer, no run-summary comment. A mutable, unsigned PR comment is not
 an audit record; the authoritative record lives behind the API where it can be
 retained and signed. The `watchdog_fired` flag marks runs where the agent did
 not self-report completion so consumers of the API know the record may be
 incomplete.
 The provenance API's schema, transport, signing, and retention are out of scope
 for this PRD (control-plane work); bot-bottle here produces the raw material
 only.
@@ -1,75 +0,0 @@
 """Unit: `cli.py resume --headless` non-interactive rehydrate path.
 The freeze / rehydrate loop needs a non-interactive `resume`: deliver a
 follow-up prompt and skip the y/N preflight, reusing the same launch
 core (`assume_yes` + `headless_prompt_text`) as `start --headless`.
 """
 from __future__ import annotations
 import unittest
 from typing import Any
 from unittest.mock import MagicMock, patch
 import bot_bottle.cli.resume as resume_mod
 from bot_bottle.log import Die
 def _metadata():
    md = MagicMock()
    md.agent_name = "implementer"
    md.copy_cwd = False
    md.cwd = "/repo"
    md.identity = "implementer-abc12"
    md.bottle_names = ["claude"]
    md.backend = "docker"
    return md
 class ResumeHeadlessTest(unittest.TestCase):
    def setUp(self) -> None:
        self._launch = patch.object(
            resume_mod, "_launch_bottle", return_value=0
        ).start()
        patch.object(
            resume_mod, "read_metadata", return_value=_metadata()
        ).start()
        manifest = MagicMock()
        manifest.require_agent = MagicMock(return_value=None)
        patch.object(
            resume_mod.ManifestIndex, "resolve", return_value=manifest
        ).start()
        self.addCleanup(patch.stopall)
    def _launch_kwargs(self) -> dict[str, Any]:
        self._launch.assert_called_once()
        return dict(self._launch.call_args.kwargs)
    def test_headless_passes_assume_yes_and_prompt(self):
        rc = resume_mod.cmd_resume(
            ["implementer-abc12", "--headless", "--prompt", "Address the review"]
        )
        self.assertEqual(0, rc)
        kwargs = self._launch_kwargs()
        self.assertTrue(kwargs["assume_yes"])
        self.assertEqual("Address the review", kwargs["headless_prompt_text"])
    def test_interactive_resume_unchanged(self):
        resume_mod.cmd_resume(["implementer-abc12"])
        kwargs = self._launch_kwargs()
        self.assertFalse(kwargs["assume_yes"])
        self.assertEqual("", kwargs["headless_prompt_text"])
    def test_headless_without_prompt_errors(self):
        with self.assertRaises(Die):
            resume_mod.cmd_resume(["implementer-abc12", "--headless"])
        self._launch.assert_not_called()
    def test_prompt_without_headless_errors(self):
        with self.assertRaises(Die):
            resume_mod.cmd_resume(["implementer-abc12", "--prompt", "hi"])
        self._launch.assert_not_called()
 if __name__ == "__main__":
    unittest.main()
@@ -1,107 +0,0 @@
 """Unit: Forge abstraction + ScopedForge (PRD forge-native-integration)."""
 from __future__ import annotations
 import unittest
 from bot_bottle.contrib.forge.base import (
    Comment,
    Forge,
    ForgeScopeError,
    Issue,
    PullRequest,
    ScopedForge,
 )
 class _RecordingForge(Forge):
    """In-memory fake that records writes."""
    def __init__(self) -> None:
        self.comments: list[tuple[int, str]] = []
        self.descriptions: list[tuple[int, str]] = []
    def read_issue(self, number: int) -> Issue:
        return Issue(number=number, title="t", body="b", state="open")
    def read_pr(self, number: int) -> PullRequest:
        return PullRequest(
            number=number, title="pr", body="b", state="open", merged=False
        )
    def read_comments(self, number: int) -> list[Comment]:
        return [Comment(id=1, user="alice", body="hi")]
    def post_comment(self, number: int, body: str) -> None:
        self.comments.append((number, body))
    def update_description(self, number: int, body: str) -> None:
        self.descriptions.append((number, body))
    def is_org_member(self, org: str, username: str) -> bool:
        return username == "member"
    def get_pr_for_issue(self, number: int) -> int | None:
        return 99 if number == 17 else None
    def is_pr_open(self, number: int) -> bool:
        return True
 class TestScopedForgeReads(unittest.TestCase):
    def setUp(self) -> None:
        self.inner = _RecordingForge()
        self.scoped = ScopedForge(self.inner, assigned_issue=17, assigned_prs=[42])
    def test_reads_pass_through_to_any_number(self):
        # A number well outside the writable scope still reads fine.
        self.assertEqual(123, self.scoped.read_issue(123).number)
        self.assertEqual("alice", self.scoped.read_comments(500)[0].user)
    def test_read_pr_passes_through(self):
        pr = self.scoped.read_pr(999)
        self.assertIsInstance(pr, PullRequest)
        self.assertEqual(999, pr.number)
        self.assertFalse(pr.merged)
    def test_membership_and_pr_lookups_delegate(self):
        self.assertTrue(self.scoped.is_org_member("bot-bottle", "member"))
        self.assertFalse(self.scoped.is_org_member("bot-bottle", "stranger"))
        self.assertEqual(99, self.scoped.get_pr_for_issue(17))
        self.assertTrue(self.scoped.is_pr_open(8000))
 class TestScopedForgeWrites(unittest.TestCase):
    def setUp(self) -> None:
        self.inner = _RecordingForge()
        self.scoped = ScopedForge(self.inner, assigned_issue=17, assigned_prs=[42])
    def test_writable_set_is_issue_plus_prs(self):
        self.assertEqual(frozenset({17, 42}), self.scoped.writable)
    def test_write_to_assigned_issue_allowed(self):
        self.scoped.post_comment(17, "done")
        self.assertEqual([(17, "done")], self.inner.comments)
    def test_write_to_assigned_pr_allowed(self):
        self.scoped.update_description(42, "new body")
        self.assertEqual([(42, "new body")], self.inner.descriptions)
    def test_comment_outside_scope_rejected(self):
        with self.assertRaises(ForgeScopeError) as ctx:
            self.scoped.post_comment(500, "spam")
        self.assertIn("500", str(ctx.exception))
        self.assertEqual([], self.inner.comments)
    def test_description_outside_scope_rejected(self):
        with self.assertRaises(ForgeScopeError):
            self.scoped.update_description(500, "tamper")
        self.assertEqual([], self.inner.descriptions)
    def test_scope_error_is_permission_error(self):
        # Sidecars can catch the stdlib base type.
        self.assertIn(PermissionError, ForgeScopeError.__mro__)
 if __name__ == "__main__":
    unittest.main()
@@ -1,145 +0,0 @@
 """Unit: GiteaClient + GiteaForge (PRD forge-native-integration)."""
 from __future__ import annotations
 import json
 import unittest
 import urllib.error
 from io import BytesIO
 from unittest.mock import MagicMock, patch
 from bot_bottle.contrib.gitea.client import GiteaClient, GiteaForge
 def _client() -> GiteaClient:
    return GiteaClient(
        api_url="https://gitea.example.com/api/v1",
        owner="didericis",
        repo="bot-bottle",
        token="test-token",
    )
 def _resp(body: object, status: int = 200) -> MagicMock:
    resp = MagicMock()
    resp.read.return_value = json.dumps(body).encode() if body is not None else b""
    resp.status = status
    resp.__enter__ = lambda s: s  # type: ignore
    resp.__exit__ = MagicMock(return_value=False)
    return resp
 def _http_error(code: int, body: str = "") -> urllib.error.HTTPError:
    return urllib.error.HTTPError(
        url="http://x", code=code, msg="err", hdrs=None,  # type: ignore[arg-type]
        fp=BytesIO(body.encode()),
    )
 _URLOPEN = "bot_bottle.contrib.gitea.client.urllib.request.urlopen"
 class TestOrgMembership(unittest.TestCase):
    def test_member_returns_true_on_2xx(self):
        with patch(_URLOPEN, return_value=_resp(None, 204)) as m:
            self.assertTrue(_client().is_org_member("bot-bottle", "alice"))
        req = m.call_args.args[0]
        self.assertIn("/orgs/bot-bottle/members/alice", req.full_url)
    def test_nonmember_returns_false_on_404(self):
        with patch(_URLOPEN, side_effect=_http_error(404)):
            self.assertFalse(_client().is_org_member("bot-bottle", "stranger"))
    def test_other_http_error_raises(self):
        with patch(_URLOPEN, side_effect=_http_error(403, "forbidden")):
            with self.assertRaises(RuntimeError) as ctx:
                _client().is_org_member("bot-bottle", "alice")
        self.assertIn("403", str(ctx.exception))
 class TestForgeReads(unittest.TestCase):
    def test_read_issue_maps_fields(self):
        raw = {"number": 17, "title": "Bug", "body": "broken", "state": "open"}
        with patch(_URLOPEN, return_value=_resp(raw)) as m:
            issue = GiteaForge(_client()).read_issue(17)
        self.assertEqual((17, "Bug", "broken", "open"),
                         (issue.number, issue.title, issue.body, issue.state))
        self.assertIn("/repos/didericis/bot-bottle/issues/17",
                      m.call_args.args[0].full_url)
    def test_read_issue_tolerates_null_body(self):
        raw = {"number": 17, "title": "T", "body": None, "state": "open"}
        with patch(_URLOPEN, return_value=_resp(raw)):
            self.assertEqual("", GiteaForge(_client()).read_issue(17).body)
    def test_read_comments_maps_user_login(self):
        raw = [
            {"id": 1, "user": {"login": "alice"}, "body": "hi"},
            {"id": 2, "user": {"login": "bob"}, "body": "yo"},
        ]
        with patch(_URLOPEN, return_value=_resp(raw)):
            comments = GiteaForge(_client()).read_comments(17)
        self.assertEqual(["alice", "bob"], [c.user for c in comments])
        self.assertEqual([1, 2], [c.id for c in comments])
 class TestForgeWrites(unittest.TestCase):
    def test_post_comment_payload_and_url(self):
        with patch(_URLOPEN, return_value=_resp(None, 201)) as m:
            GiteaForge(_client()).post_comment(17, "done ✓")
        req = m.call_args.args[0]
        self.assertEqual("POST", req.method)
        self.assertIn("/repos/didericis/bot-bottle/issues/17/comments", req.full_url)
        self.assertEqual("done ✓", json.loads(req.data)["body"])
    def test_update_description_patches_issue(self):
        with patch(_URLOPEN, return_value=_resp(None, 200)) as m:
            GiteaForge(_client()).update_description(17, "edited")
        req = m.call_args.args[0]
        self.assertEqual("PATCH", req.method)
        self.assertTrue(req.full_url.endswith("/issues/17"))
        self.assertEqual("edited", json.loads(req.data)["body"])
    def test_auth_header_sent(self):
        with patch(_URLOPEN, return_value=_resp(None, 201)) as m:
            GiteaForge(_client()).post_comment(17, "x")
        self.assertEqual("token test-token",
                         m.call_args.args[0].headers["Authorization"])
 class TestPRHelpers(unittest.TestCase):
    def test_get_pr_for_issue_returns_number_when_issue_is_pr(self):
        raw = {"number": 18, "pull_request": {"merged": False}}
        with patch(_URLOPEN, return_value=_resp(raw)):
            self.assertEqual(18, GiteaForge(_client()).get_pr_for_issue(18))
    def test_get_pr_for_issue_none_for_plain_issue(self):
        raw = {"number": 17, "pull_request": None}
        with patch(_URLOPEN, return_value=_resp(raw)):
            self.assertIsNone(GiteaForge(_client()).get_pr_for_issue(17))
    def test_is_pr_open_true_when_state_open(self):
        with patch(_URLOPEN, return_value=_resp({"state": "open"})):
            self.assertTrue(GiteaForge(_client()).is_pr_open(18))
    def test_is_pr_open_false_when_closed(self):
        with patch(_URLOPEN, return_value=_resp({"state": "closed"})):
            self.assertFalse(GiteaForge(_client()).is_pr_open(18))
    def test_read_pr_maps_fields_including_merged(self):
        raw = {"number": 18, "title": "Fix", "body": "patch",
               "state": "closed", "merged": True}
        with patch(_URLOPEN, return_value=_resp(raw)) as m:
            pr = GiteaForge(_client()).read_pr(18)
        self.assertEqual((18, "Fix", "patch", "closed", True),
                         (pr.number, pr.title, pr.body, pr.state, pr.merged))
        self.assertIn("/repos/didericis/bot-bottle/pulls/18",
                      m.call_args.args[0].full_url)
    def test_read_pr_merged_defaults_false(self):
        with patch(_URLOPEN, return_value=_resp({"number": 18, "state": "open"})):
            self.assertFalse(GiteaForge(_client()).read_pr(18).merged)
 if __name__ == "__main__":
    unittest.main()
@@ -1,99 +0,0 @@
 """Unit: SQLite forge state store (PRD forge-native-integration)."""
 from __future__ import annotations
 import tempfile
 import unittest
 from dataclasses import replace
 from pathlib import Path
 from bot_bottle.contrib.gitea.forge_state import (
    STATUS_FROZEN,
    STATUS_RUNNING,
    ForgeState,
    SqliteForgeStateStore,
 )
 def _state(**over: object) -> ForgeState:
    base = ForgeState(
        owner="didericis",
        repo="bot-bottle",
        issue_number=17,
        slug="implementer-abc12",
        agent_name="implementer",
        bottle_names=["claude"],
        backend_name="docker",
        agent_git_user="didericis-claude",
        pr_number=42,
        status=STATUS_FROZEN,
        last_checkin_at="2026-06-29T12:04:12-04:00",
    )
    return replace(base, **over)
 class ForgeStateStoreTest(unittest.TestCase):
    def setUp(self) -> None:
        tmp = Path(self.enterContext(tempfile.TemporaryDirectory()))  # pylint: disable=consider-using-with
        self.store = SqliteForgeStateStore(tmp / "sub" / "bot-bottle.db")
    def test_round_trip(self):
        self.store.upsert(_state())
        self.assertEqual(_state(), self.store.get("didericis", "bot-bottle", 17))
    def test_missing_returns_none(self):
        self.assertIsNone(self.store.get("nobody", "nope", 1))
    def test_creates_db_parent_dirs(self):
        # setUp pointed at a non-existent 'sub/' dir; init must create it.
        self.assertIsNone(self.store.get("x", "y", 1))  # no raise
    def test_upsert_replaces(self):
        self.store.upsert(_state(status=STATUS_RUNNING))
        self.store.upsert(_state(status=STATUS_FROZEN))
        got = self.store.get("didericis", "bot-bottle", 17)
        assert got is not None
        self.assertEqual(STATUS_FROZEN, got.status)
        # Still one row, not two.
        self.assertEqual(1, len(self.store.all()))
    def test_delete_is_idempotent(self):
        self.store.upsert(_state())
        self.store.delete("didericis", "bot-bottle", 17)
        self.store.delete("didericis", "bot-bottle", 17)  # no raise
        self.assertIsNone(self.store.get("didericis", "bot-bottle", 17))
    def test_all_lists_across_repos_sorted(self):
        self.store.upsert(_state(issue_number=18, slug="other"))
        self.store.upsert(_state(issue_number=17))
        self.store.upsert(_state(owner="acme", repo="widget", issue_number=3))
        states = self.store.all()
        self.assertEqual(3, len(states))
        self.assertEqual(
            [("acme", 3), ("didericis", 17), ("didericis", 18)],
            [(s.owner, s.issue_number) for s in states],
        )
    def test_all_empty(self):
        self.assertEqual([], self.store.all())
    def test_bottle_names_list_preserved(self):
        self.store.upsert(_state(bottle_names=["claude", "dev"]))
        got = self.store.get("didericis", "bot-bottle", 17)
        assert got is not None
        self.assertEqual(["claude", "dev"], got.bottle_names)
    def test_pr_number_nullable(self):
        self.store.upsert(_state(pr_number=None))
        got = self.store.get("didericis", "bot-bottle", 17)
        assert got is not None
        self.assertIsNone(got.pr_number)
    def test_persists_across_store_instances(self):
        self.store.upsert(_state())
        reopened = SqliteForgeStateStore(self.store._db_path)  # pylint: disable=protected-access
        self.assertEqual(_state(), reopened.get("didericis", "bot-bottle", 17))
 if __name__ == "__main__":
    unittest.main()
@@ -14,6 +14,7 @@ from bot_bottle.git_gate import (
    git_gate_render_access_hook,
    git_gate_render_entrypoint,
    git_gate_render_hook,
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
    _resolve_identity_file,
    git_gate_upstreams_for_bottle,
@@ -371,6 +372,27 @@ class TestDynamicKeyProvisioning(unittest.TestCase):
            self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
        mock_provision.assert_called_once()
    def test_prepare_defers_gitea_key_provisioning(self):
        bottle = self._gitea_manifest().bottles["dev"]
        with patch("bot_bottle.git_gate_provision._provision_dynamic_key") as mock_provision:
            plan = _StubGate().prepare(bottle, "demo", self.stage)
        mock_provision.assert_not_called()
        self.assertEqual("", plan.upstreams[0].identity_file)
    def test_launch_time_helper_provisions_gitea_keys(self):
        bottle = self._gitea_manifest().bottles["dev"]
        plan = _StubGate().prepare(bottle, "demo", self.stage)
        with patch(
            "bot_bottle.git_gate_provision._provision_dynamic_key",
            return_value="/tmp/provisioned-key",
        ) as mock_provision:
            updated = provision_git_gate_dynamic_keys(bottle, plan, self.stage)
        mock_provision.assert_called_once_with(bottle.git[0], "demo", self.stage)
        self.assertEqual("/tmp/provisioned-key", updated.upstreams[0].identity_file)
    def test_revoke_skips_non_gitea_and_missing_id_file(self):
        revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage)
Author	SHA1	Message	Date
Quality Badge Bot	5970b785aa	chore: update quality badges - Coverage: 83% - Core coverage: 95% [skip ci]	2026-07-01 16:51:08 +00:00
didericis	2f5cf81cf5	fix(git-gate): defer dynamic key provisioning lint / lint (push) Successful in 1m59s Details test / unit (push) Successful in 49s Details test / integration (push) Successful in 23s Details test / coverage (push) Successful in 1m0s Details Update Quality Badges / update-badges (push) Successful in 53s Details	2026-07-01 12:45:46 -04:00
didericis	4a1e667306	fix(git-gate): inline GIT_GATE_TIMEOUT_SECS to fix git-http ImportError lint / lint (push) Successful in 1m56s Details test / unit (push) Successful in 48s Details test / integration (push) Successful in 20s Details test / coverage (push) Successful in 1m2s Details Update Quality Badges / update-badges (push) Successful in 52s Details git_http_backend.py is copied flat into the sidecar bundle image as a standalone script, not as part of the bot_bottle package, and git_gate.py/git_gate_render.py are never copied in. Its relative import of GIT_GATE_TIMEOUT_SECS crashed the git-http daemon (port 9420) on every startup, silently leaving the smart-HTTP git-gate transport down while the other sidecar daemons stayed up. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-07-01 11:53:26 -04:00