"""Manifest dataclasses (PRD 0011 layout). Reads the per-file manifest tree: $HOME/.bot-bottle/bottles/.md — one bottle per file $HOME/.bot-bottle/agents/.md — home-resident agents $CWD/.bot-bottle/agents/.md — cwd-supplied agents Each file is Markdown with YAML frontmatter. The frontmatter holds the structured config (see schema below); for agents the body is the system prompt, for bottles the body is human documentation (ignored by the parser). Bottle schema (frontmatter): extends: # optional (PRD 0025) env: { : , ... } git-gate: # optional (PRD 0047) user: { name: , email: } # optional repos: { : , ... } # optional egress: { routes: [ , ... ] } # route keys: host, matches, auth, role, dlp supervise: # optional (default true) Agent schema (frontmatter): bottle: # required skills: [ , ... ] # optional git-gate: user: { name: , email: } # optional; overlays bottle # Claude Code subagent passthrough fields — accepted, ignored: name, description, model, color, memory The agent file's Markdown body is the system prompt (stripped). Unknown top-level frontmatter keys raise ManifestError with a hint. Bottles can ONLY live under $HOME. A bottles/ dir under $CWD is a warn at load time and contributes nothing. The trust boundary is expressed as filesystem layout rather than resolver logic. Two types are exported: ManifestIndex — the multi-agent/bottle collection returned by resolve() and from_json_obj(). Used for agent selection (all_agent_names), validation (require_agent), and lazy loading (load_for_agent). This is the pre-preflight form. Manifest — a single-agent/bottle value type holding exactly one agent: ManifestAgent and one bottle: ManifestBottle (with the agent's git-gate.user already overlaid). Returned by load_for_agent(). This is the post-preflight form passed to backends. ManifestIndex.from_json_obj is preserved as a programmatic entry point (used by tests) that takes a dict with the same field names — useful for building manifests without on-disk files. """ from __future__ import annotations import os from dataclasses import dataclass, field, replace from pathlib import Path from typing import Mapping from .manifest_util import ManifestError, as_json_object from .manifest_agent import ManifestAgent, ManifestAgentProvider from .manifest_egress import ( EGRESS_AUTH_SCHEMES, ManifestEgressConfig, ManifestEgressRoute, ) from .manifest_git import ManifestGitEntry, ManifestGitUser, ManifestKeyConfig, parse_git_gate_config from .manifest_schema import BOTTLE_KEYS # Re-export everything that callers currently import from this module. __all__ = [ "ManifestError", "ManifestGitEntry", "ManifestGitUser", "ManifestKeyConfig", "ManifestAgentProvider", "EGRESS_AUTH_SCHEMES", "ManifestEgressRoute", "ManifestEgressConfig", "ManifestAgent", "ManifestBottle", "ManifestIndex", "Manifest", ] def _empty_str_dict() -> dict[str, str]: return {} def _section_dict(value: object, label: str) -> dict[str, object]: """Like as_json_object but treats absent/null as an empty section.""" if value is None: return {} return as_json_object(value, label) @dataclass(frozen=True) class ManifestBottle: env: Mapping[str, str] = field(default_factory=_empty_str_dict) agent_provider: ManifestAgentProvider = field(default_factory=ManifestAgentProvider) git: tuple[ManifestGitEntry, ...] = () # Per-bottle git identity (issue #86). Empty default — bottles # that don't set `git-gate.user:` in the manifest skip the # `git config --global` step entirely. A bottle can declare a user # identity without any git-gate.repos upstreams, and vice versa. git_user: ManifestGitUser = field(default_factory=ManifestGitUser) egress: ManifestEgressConfig = field(default_factory=ManifestEgressConfig) # Per-bottle stuck-recovery sidecar (PRD 0013). When true (the # default, issue #249), the launch step brings up a supervise # sidecar that exposes MCP tools to the agent (egress-block, # capability-block) plus mounts the current-config dir read-only # into the agent at /etc/bot-bottle/current-config. Set # `supervise: false` to skip the sidecar and mount. supervise: bool = True @classmethod def from_dict(cls, name: str, raw: object) -> "ManifestBottle": d = as_json_object(raw, f"bottle '{name}'") if "runtime" in d: raise ManifestError( f"bottle '{name}' has a 'runtime' field, which is no longer " f"supported. gVisor (runsc) is now auto-detected by the " f"backend; remove the 'runtime' field from the bottle " f"definition." ) if "ssh" in d: raise ManifestError( f"bottle '{name}' has an 'ssh' field, which has been removed " f"(PRD 0009). Declare upstreams under 'git-gate.repos' with " f"url + identity + host_key; the git-gate sidecar (PRD 0008) " f"holds the credential and gitleaks-scans pushes." ) if "git" in d: raise ManifestError( f"bottle '{name}' uses 'git' which has been replaced by " f"'git-gate' (PRD 0047). Move git.user → git-gate.user " f"and git.remotes → git-gate.repos (fields: url, identity, host_key)." ) if "git_user" in d: raise ManifestError( f"bottle '{name}' has a 'git_user' field, which has been " f"removed. Move it under 'git-gate.user'." ) unknown = set(d.keys()) - BOTTLE_KEYS if unknown: allowed = ", ".join(sorted(BOTTLE_KEYS)) raise ManifestError( f"bottle '{name}' has unknown key(s) {sorted(unknown)}; " f"allowed keys are {allowed}." ) env: dict[str, str] = {} env_raw = d.get("env") if env_raw is not None: env_dict = as_json_object(env_raw, f"bottle '{name}' env") for var, value in env_dict.items(): if not isinstance(value, str): raise ManifestError( f"env entry {var} in bottle '{name}' must be a JSON string " f"(was {type(value).__name__}). Use \"?\" for prompt-at-runtime." ) env[var] = value git: tuple[ManifestGitEntry, ...] = () git_user = ManifestGitUser() git_raw = d.get("git-gate") if git_raw is not None: git, git_user = parse_git_gate_config(name, git_raw) agent_provider = ( ManifestAgentProvider.from_dict(name, d["agent_provider"]) if "agent_provider" in d else ManifestAgentProvider() ) egress = ( ManifestEgressConfig.from_dict(name, d["egress"]) if "egress" in d else ManifestEgressConfig() ) supervise_raw = d.get("supervise", True) if not isinstance(supervise_raw, bool): raise ManifestError( f"bottle '{name}' supervise must be a boolean " f"(was {type(supervise_raw).__name__})" ) return cls( env=env, agent_provider=agent_provider, git=git, git_user=git_user, egress=egress, supervise=supervise_raw, ) def _merge_git_user( agent_user: ManifestGitUser, base_user: ManifestGitUser ) -> ManifestGitUser: """Merge the agent's git.user over the bottle's, agent-wins-on-non-empty.""" if agent_user.is_empty(): return base_user return ManifestGitUser( name=agent_user.name or base_user.name, email=agent_user.email or base_user.email, ) @dataclass(frozen=True) class Manifest: """Single-agent/bottle value type. Returned by ManifestIndex.load_for_agent(). `bottle` is the effective bottle with the agent's git-gate.user already overlaid per-field (agent wins on non-empty). Backends and provisioners use this directly — no agent_name lookup needed.""" agent: ManifestAgent bottle: ManifestBottle def git_identity_summary(self) -> str | None: """One-line effective git identity with per-field provenance, e.g. `name=claude (agent), email=eric@dideric.is (bottle)`. Returns None when neither agent nor bottle sets an identity.""" over = self.agent.git_user # agent's declared git_user (pre-merge) merged = self.bottle.git_user # effective git_user (post-merge) if merged.is_empty(): return None parts: list[str] = [] if merged.name: parts.append(f"name={merged.name} ({'agent' if over.name else 'bottle'})") if merged.email: parts.append(f"email={merged.email} ({'agent' if over.email else 'bottle'})") return ", ".join(parts) @dataclass(frozen=True) class ManifestIndex: """Multi-agent/bottle collection. The pre-preflight form. In lazy mode (from resolve()/from_md_dirs()) only filenames are scanned; no file content is read. In eager mode (from from_json_obj()) all agents and bottles are pre-parsed. Call load_for_agent() to get a single-value Manifest ready for backend use.""" bottles: Mapping[str, ManifestBottle] agents: Mapping[str, ManifestAgent] # Set by from_md_dirs; None in from_json_obj (test/programmatic) mode. # Stores the manifest root dirs so load_for_agent can locate files later. home_md: Path | None = field(default=None) cwd_md: Path | None = field(default=None) @classmethod def resolve(cls, cwd: str, *, missing_ok: bool = False) -> "ManifestIndex": """Walk the per-file manifest tree and build a ManifestIndex. Layout (PRD 0011): $HOME/.bot-bottle/bottles/.md — bottles (home-only) $HOME/.bot-bottle/agents/.md — home agents $CWD/.bot-bottle/agents/.md — cwd agents Cwd agents merge into the home agents on the same name (cwd wins). A bottles/ subdir under $CWD is logged as a warning and ignored — the filesystem layout IS the trust boundary. If `missing_ok` is true, a missing `$HOME/.bot-bottle/` returns an empty index instead of dying. This is for passive UI surfaces like the dashboard, which can still monitor already-running agents without launch config. If `bot-bottle.json` exists alongside a missing `.bot-bottle/` directory at either side, dies with a clear pointer at the README's manifest section — the manifest format changed in PRD 0011 and we don't silently fall back.""" home_dir = Path(os.environ["HOME"]) cwd_dir = Path(cwd) home_md = home_dir / ".bot-bottle" cwd_md = cwd_dir / ".bot-bottle" from .manifest_loader import check_stale_json check_stale_json(home_dir, home_md, "$HOME") if cwd_dir.resolve() != home_dir.resolve(): check_stale_json(cwd_dir, cwd_md, "$CWD") if not home_md.is_dir(): if missing_ok: return cls.from_json_obj({"bottles": {}, "agents": {}}) raise ManifestError( f"no manifest found: {home_md} does not exist. " f"See README.md for the per-file Markdown layout " f"(PRD 0011)." ) # When CWD == HOME (running from $HOME directly), pass the # same dir for both — _load_md_dirs will dedupe. cwd_md_arg = cwd_md if cwd_md.is_dir() and cwd_dir.resolve() != home_dir.resolve() else None return cls.from_md_dirs(home_md, cwd_md_arg) @classmethod def from_md_dirs( cls, home_dir: Path, cwd_dir: Path | None, ) -> "ManifestIndex": """Return a names-only ManifestIndex. No file content is read; only filenames are scanned for the agent selector. Full parsing happens later, per-agent, via `load_for_agent`. A `bottles/` subdir under `cwd_dir` is logged as a warning and ignored — the filesystem layout IS the trust boundary. Used by tests to build a ManifestIndex from fixture directories without touching `os.environ`.""" if cwd_dir is not None: stale_bottles = cwd_dir / "bottles" if stale_bottles.is_dir(): files = sorted(stale_bottles.glob("*.md")) if files: names = ", ".join(p.name for p in files) from .log import warn warn( f"ignoring bottle file(s) under " f"{stale_bottles}: {names}. Bottles can only " f"live under $HOME/.bot-bottle/bottles/ " f"(PRD 0011). Move them or delete." ) return cls(bottles={}, agents={}, home_md=home_dir, cwd_md=cwd_dir) @classmethod def from_json_obj(cls, obj: object) -> "ManifestIndex": """Validate and build a ManifestIndex from a raw JSON-like dict.""" d = as_json_object(obj, "manifest") raw_bottles_obj = _section_dict(d.get("bottles"), "manifest 'bottles'") raw_agents = _section_dict(d.get("agents"), "manifest 'agents'") # Coerce each bottle's raw to dict[str, object] so the # PRD 0025 resolver can apply extends-merge rules # consistently with the md-loader path. raw_bottles: dict[str, dict[str, object]] = {} for n, b in raw_bottles_obj.items(): raw_bottles[n] = as_json_object(b, f"bottle '{n}'") from .manifest_extends import resolve_bottles bottles = resolve_bottles(raw_bottles) bottle_names = set(bottles.keys()) agents: dict[str, ManifestAgent] = { n: ManifestAgent.from_dict(n, a, bottle_names) for n, a in raw_agents.items() } return cls(bottles=bottles, agents=agents) @property def all_agent_names(self) -> list[str]: """Sorted list of all discoverable agent names. In names-only mode (from resolve/from_md_dirs) this scans agent filenames without reading their content. In eager mode (from from_json_obj) it returns the pre-parsed agents' names.""" if self.home_md is not None: from .manifest_loader import scan_agent_names home_names = set(scan_agent_names(self.home_md / "agents").keys()) cwd_names: set[str] = set() if self.cwd_md is not None: cwd_names = set(scan_agent_names(self.cwd_md / "agents").keys()) return sorted(home_names | cwd_names) return sorted(self.agents.keys()) def load_for_agent(self, agent_name: str) -> "Manifest": """Parse the named agent and its bottle; return a single-value Manifest. In lazy mode (from resolve/from_md_dirs) the agent file and its bottle chain are read from disk for the first time here. In eager mode (from_json_obj) the data is already parsed; this just filters down to the requested agent and its bottle. The returned Manifest.bottle has the agent's git-gate.user already overlaid (agent wins on non-empty, per-field). Always raises ManifestError if the agent is unknown or invalid. Backends call this at preflight inside _validate.""" if self.home_md is None: # Eager manifest (from_json_obj): data already parsed; filter to # the one requested agent and its bottle so the returned Manifest # always holds exactly one agent and one bottle regardless of path. if agent_name not in self.agents: available = ", ".join(sorted(self.agents.keys())) or "(none)" raise ManifestError( f"agent '{agent_name}' not defined. Available: {available}" ) agent = self.agents[agent_name] raw_bottle = self.bottles[agent.bottle] merged = _merge_git_user(agent.git_user, raw_bottle.git_user) bottle = raw_bottle if merged == raw_bottle.git_user else replace(raw_bottle, git_user=merged) return Manifest(agent=agent, bottle=bottle) from .manifest_loader import load_bottle_chain_from_dir, scan_agent_names from .manifest_schema import validate_agent_frontmatter_keys from .yaml_subset import YamlSubsetError, parse_frontmatter # Locate the agent file; cwd wins over home on name collision. home_agents = scan_agent_names(self.home_md / "agents") cwd_agents: dict[str, Path] = {} if self.cwd_md is not None: cwd_agents = scan_agent_names(self.cwd_md / "agents") merged_agents = {**home_agents, **cwd_agents} if agent_name not in merged_agents: available = ", ".join(sorted(merged_agents.keys())) or "(none)" raise ManifestError( f"agent '{agent_name}' not defined. Available: {available}" ) agent_path = merged_agents[agent_name] try: fm, body = parse_frontmatter(agent_path.read_text()) except OSError as e: raise ManifestError(f"could not read {agent_path}: {e}") from e except YamlSubsetError as e: raise ManifestError(f"{agent_path}: {e}") from e validate_agent_frontmatter_keys(agent_path, fm.keys()) bottle_name = fm.get("bottle") if not isinstance(bottle_name, str) or not bottle_name: raise ManifestError( f"agent '{agent_name}' must declare a 'bottle' field " f"naming a defined bottle" ) # Load the bottle chain (may raise ManifestError). bottles_dir = self.home_md / "bottles" raw_bottle = load_bottle_chain_from_dir(bottle_name, bottles_dir) # Build and validate the full ManifestAgent. agent_dict: dict[str, object] = { "bottle": bottle_name, "skills": fm.get("skills", []), "prompt": body.strip(), } if "git-gate" in fm: agent_dict["git-gate"] = fm["git-gate"] agent = ManifestAgent.from_dict(agent_name, agent_dict, {bottle_name}) merged_user = _merge_git_user(agent.git_user, raw_bottle.git_user) bottle = raw_bottle if merged_user == raw_bottle.git_user else replace(raw_bottle, git_user=merged_user) return Manifest(agent=agent, bottle=bottle) def has_agent(self, name: str) -> bool: return name in self.agents def require_agent(self, name: str) -> None: """Check that `name` is a discoverable agent. In names-only mode this checks whether the .md file exists; in eager mode it checks the pre-parsed agents dict. Does NOT parse file content.""" if self.has_agent(name): return if self.home_md is not None: # Names-only mode: check file existence without parsing. home_path = self.home_md / "agents" / f"{name}.md" cwd_path = ( self.cwd_md / "agents" / f"{name}.md" if self.cwd_md else None ) if home_path.is_file() or (cwd_path and cwd_path.is_file()): return available = ", ".join(self.all_agent_names) or "(none)" raise ManifestError( f"agent '{name}' not defined. Available: {available}" )