refactor(manifest): consolidate the manifest modules into a bot_bottle.manifest package
tracker-policy-pr / check-pr (pull_request) Failing after 11s
test / integration-docker (pull_request) Successful in 20s
lint / lint (push) Successful in 59s
test / unit (pull_request) Successful in 2m10s
test / integration-firecracker (pull_request) Successful in 3m29s
test / coverage (pull_request) Successful in 38s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 11s
test / integration-docker (pull_request) Successful in 20s
lint / lint (push) Successful in 59s
test / unit (pull_request) Successful in 2m10s
test / integration-firecracker (pull_request) Successful in 3m29s
test / coverage (pull_request) Successful in 38s
test / publish-infra (pull_request) Has been skipped
Move the nine root-level manifest modules into a bot_bottle/manifest/ package, dropping the redundant manifest_ prefix: the facade (manifest.py) becomes the package __init__ and keeps re-exporting Manifest / ManifestIndex and the piece types, while manifest_<x>.py become manifest/<x>.py (agent, bottle, egress, extends, git, loader, schema, util). Because the facade stays the package __init__, the ~13 callers that import `from bot_bottle.manifest import ...` are unchanged; only direct-piece imports move to `bot_bottle.manifest.<name>`. Inside the package, intra-manifest imports drop the prefix and the non-manifest siblings (log, yaml_subset, agent_provider) move to `..`. No behavior change; full unit suite green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,478 @@
|
||||
"""Manifest dataclasses (PRD 0011 layout).
|
||||
|
||||
Reads the per-file manifest tree:
|
||||
|
||||
$HOME/.bot-bottle/bottles/<name>.md — one bottle per file
|
||||
$HOME/.bot-bottle/agents/<name>.md — home-resident agents
|
||||
$CWD/.bot-bottle/agents/<name>.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: <bottle-name> # optional (PRD 0025)
|
||||
env: { <NAME>: <env-entry>, ... }
|
||||
git-gate: # optional (PRD 0047)
|
||||
user: { name: <str>, email: <str> } # optional
|
||||
repos: { <name>: <git-gate-entry>, ... } # optional
|
||||
egress: { routes: [ <egress-route>, ... ] }
|
||||
# route keys: host, matches, auth, role, dlp
|
||||
supervise: <bool> # optional (default true)
|
||||
nested_containers: <bool> # optional (default false)
|
||||
|
||||
Agent schema (frontmatter):
|
||||
bottle: <bottle-name> # required
|
||||
skills: [ <skill-name>, ... ] # optional
|
||||
git-gate:
|
||||
user: { name: <str>, email: <str> } # 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 ..log import warn
|
||||
from .util import ManifestError, as_json_object
|
||||
from .agent import ManifestAgent, ManifestAgentProvider
|
||||
from .bottle import ManifestBottle
|
||||
from .egress import (
|
||||
EGRESS_AUTH_SCHEMES,
|
||||
ManifestEgressConfig,
|
||||
ManifestEgressRoute,
|
||||
)
|
||||
from .extends import merge_bottles_runtime, resolve_bottles
|
||||
from .git import ManifestGitEntry, ManifestGitUser, ManifestKeyConfig
|
||||
from .loader import (
|
||||
check_stale_json,
|
||||
load_bottle_chain_from_dir,
|
||||
scan_agent_names,
|
||||
scan_bottle_names,
|
||||
)
|
||||
from .schema import validate_agent_frontmatter_keys
|
||||
from ..yaml_subset import YamlSubsetError, parse_frontmatter
|
||||
|
||||
# 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 _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)
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def _manifest_with_merged_git_user(
|
||||
agent: "ManifestAgent", raw_bottle: "ManifestBottle"
|
||||
) -> "Manifest":
|
||||
"""Build the single-value Manifest, overlaying the agent's git-gate.user
|
||||
onto the bottle (agent wins on non-empty, per-field). Shared by the eager
|
||||
and lazy load_for_agent paths."""
|
||||
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)
|
||||
|
||||
|
||||
def _resolve_effective_bottle_eager(
|
||||
agent_name: str,
|
||||
agent: "ManifestAgent",
|
||||
bottle_names: "tuple[str, ...]",
|
||||
bottles: "Mapping[str, ManifestBottle]",
|
||||
) -> "ManifestBottle":
|
||||
"""Return the effective ManifestBottle for the eager (from_json_obj) path.
|
||||
|
||||
When bottle_names is non-empty they are merged in order. When empty, falls
|
||||
back to agent.bottle. Raises ManifestError when neither is set."""
|
||||
if bottle_names:
|
||||
resolved: list[ManifestBottle] = []
|
||||
for bn in bottle_names:
|
||||
if bn not in bottles:
|
||||
available = ", ".join(sorted(bottles.keys())) or "(none)"
|
||||
raise ManifestError(
|
||||
f"bottle '{bn}' not defined. Available: {available}"
|
||||
)
|
||||
resolved.append(bottles[bn])
|
||||
return merge_bottles_runtime(resolved)
|
||||
|
||||
if not agent.bottle:
|
||||
raise ManifestError(
|
||||
f"agent '{agent_name}' has no 'bottle' field and no bottles were "
|
||||
f"selected at launch. Select at least one bottle or add "
|
||||
f"'bottle: <name>' to the agent manifest."
|
||||
)
|
||||
return bottles[agent.bottle]
|
||||
|
||||
|
||||
def _resolve_effective_bottle_lazy(
|
||||
agent_name: str,
|
||||
agent_bottle: str,
|
||||
bottle_names: "tuple[str, ...]",
|
||||
bottles_dir: "Path",
|
||||
) -> "ManifestBottle":
|
||||
"""Return the effective ManifestBottle for the lazy (from_md_dirs) path.
|
||||
|
||||
When bottle_names is non-empty they are resolved from disk and merged in
|
||||
order. When empty, falls back to agent_bottle. Raises ManifestError when
|
||||
neither is set."""
|
||||
if bottle_names:
|
||||
resolved = [load_bottle_chain_from_dir(bn, bottles_dir) for bn in bottle_names]
|
||||
return merge_bottles_runtime(resolved)
|
||||
|
||||
if not agent_bottle:
|
||||
raise ManifestError(
|
||||
f"agent '{agent_name}' has no 'bottle' field and no bottles were "
|
||||
f"selected at launch. Select at least one bottle or add "
|
||||
f"'bottle: <name>' to the agent manifest."
|
||||
)
|
||||
return load_bottle_chain_from_dir(agent_bottle, bottles_dir)
|
||||
|
||||
|
||||
@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/<name>.md — bottles (home-only)
|
||||
$HOME/.bot-bottle/agents/<name>.md — home agents
|
||||
$CWD/.bot-bottle/agents/<name>.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"
|
||||
|
||||
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)
|
||||
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}'")
|
||||
|
||||
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_bottle_names(self) -> list[str]:
|
||||
"""Sorted list of all discoverable bottle names.
|
||||
|
||||
In names-only mode (from resolve/from_md_dirs) this scans bottle
|
||||
filenames without reading their content. In eager mode (from
|
||||
from_json_obj) it returns the pre-parsed bottles' names."""
|
||||
if self.home_md is not None:
|
||||
return scan_bottle_names(self.home_md / "bottles")
|
||||
return sorted(self.bottles.keys())
|
||||
|
||||
@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:
|
||||
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,
|
||||
bottle_names: "tuple[str, ...] | None" = None,
|
||||
) -> "Manifest":
|
||||
"""Parse the named agent and its bottle; return a single-value Manifest.
|
||||
|
||||
`bottle_names` is an ordered list of bottles selected at launch time.
|
||||
When non-empty they are resolved and merged in order (index 0 = base;
|
||||
later entries override). When empty or None, falls back to the agent's
|
||||
own `bottle:` field. Raises ManifestError when neither is set.
|
||||
|
||||
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."""
|
||||
effective_bottle_names: tuple[str, ...] = bottle_names or ()
|
||||
if self.home_md is None:
|
||||
return self._load_for_agent_eager(agent_name, effective_bottle_names)
|
||||
return self._load_for_agent_lazy(agent_name, effective_bottle_names)
|
||||
|
||||
def _load_for_agent_eager(
|
||||
self, agent_name: str, bottle_names: tuple[str, ...]
|
||||
) -> "Manifest":
|
||||
"""Eager path (from_json_obj): data is 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 = _resolve_effective_bottle_eager(
|
||||
agent_name, agent, bottle_names, self.bottles
|
||||
)
|
||||
return _manifest_with_merged_git_user(agent, raw_bottle)
|
||||
|
||||
def _load_for_agent_lazy(
|
||||
self, agent_name: str, bottle_names: tuple[str, ...]
|
||||
) -> "Manifest":
|
||||
"""Lazy path (resolve/from_md_dirs): read and parse the agent file and
|
||||
its bottle chain from disk for the first time here."""
|
||||
assert self.home_md is not None # guaranteed by load_for_agent dispatch
|
||||
# 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())
|
||||
|
||||
# Determine the effective bottle name(s).
|
||||
agent_bottle = fm.get("bottle") or ""
|
||||
bottles_dir = self.home_md / "bottles"
|
||||
raw_bottle = _resolve_effective_bottle_lazy(
|
||||
agent_name, str(agent_bottle), bottle_names, bottles_dir
|
||||
)
|
||||
effective_bottle_name = (
|
||||
bottle_names[-1] if bottle_names else str(agent_bottle)
|
||||
)
|
||||
|
||||
# Build and validate the full ManifestAgent.
|
||||
agent_dict: dict[str, object] = {
|
||||
"skills": fm.get("skills", []),
|
||||
"prompt": body.strip(),
|
||||
}
|
||||
if agent_bottle:
|
||||
agent_dict["bottle"] = agent_bottle
|
||||
if "git-gate" in fm:
|
||||
agent_dict["git-gate"] = fm["git-gate"]
|
||||
# Pass the effective bottle name as the known-bottles set so agents
|
||||
# that have bottle: set are validated; agents without bottle: pass {}
|
||||
# since bottle_names were already resolved above.
|
||||
known = {effective_bottle_name} if effective_bottle_name else set()
|
||||
agent = ManifestAgent.from_dict(agent_name, agent_dict, known)
|
||||
|
||||
return _manifest_with_merged_git_user(agent, raw_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}"
|
||||
)
|
||||
@@ -0,0 +1,317 @@
|
||||
"""Agent configuration manifest dataclasses."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import cast
|
||||
|
||||
from ..agent_provider import PROVIDER_TEMPLATES
|
||||
from .util import ManifestError, as_json_object
|
||||
from .git import ManifestGitUser
|
||||
from .schema import AGENT_MODEL_KEYS, is_valid_entity_name
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestAgentProvider:
|
||||
"""Provider/template for the agent process inside a bottle.
|
||||
|
||||
`template` selects a built-in launch/runtime contract. `dockerfile`
|
||||
optionally points at a custom agent-image Dockerfile while leaving
|
||||
bot-bottle's gateway infrastructure intact.
|
||||
|
||||
`auth_token` names the host env var that holds the provider's OAuth
|
||||
token (Claude only). The provisioner injects a provider-owned egress
|
||||
route for api.anthropic.com that re-injects this token as the Bearer
|
||||
header, and sets a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent
|
||||
so the Claude Code CLI starts.
|
||||
|
||||
`forward_host_credentials` forwards the host provider auth token into
|
||||
the egress sidecar (Codex and Claude). For Codex this reads
|
||||
`~/.codex/auth.json`; for Claude it reads `~/.claude/.credentials.json`.
|
||||
"""
|
||||
|
||||
template: str = "claude"
|
||||
dockerfile: str = ""
|
||||
auth_token: str = ""
|
||||
forward_host_credentials: bool = False
|
||||
settings: dict[str, object] = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, bottle_name: str, raw: object) -> "ManifestAgentProvider":
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' agent_provider")
|
||||
for k in d:
|
||||
if k not in {
|
||||
"template",
|
||||
"dockerfile",
|
||||
"auth_token",
|
||||
"forward_host_credentials",
|
||||
"settings",
|
||||
}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider has unknown key {k!r}; "
|
||||
"allowed: template, dockerfile, auth_token, "
|
||||
"forward_host_credentials, settings"
|
||||
)
|
||||
template = d.get("template", "claude")
|
||||
if not isinstance(template, str) or not template:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.template must be a "
|
||||
f"non-empty string"
|
||||
)
|
||||
dockerfile = d.get("dockerfile", "")
|
||||
if not isinstance(dockerfile, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.dockerfile must be a "
|
||||
f"string (was {type(dockerfile).__name__})"
|
||||
)
|
||||
auth_token = d.get("auth_token", "")
|
||||
if not isinstance(auth_token, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.auth_token must be a "
|
||||
f"string (was {type(auth_token).__name__})"
|
||||
)
|
||||
if auth_token and template not in PROVIDER_TEMPLATES:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.auth_token is only "
|
||||
f"supported for built-in templates "
|
||||
f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
|
||||
)
|
||||
if auth_token and template != "claude":
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.auth_token is only "
|
||||
f"supported for template 'claude'"
|
||||
)
|
||||
forward_host_credentials = d.get("forward_host_credentials", False)
|
||||
if not isinstance(forward_host_credentials, bool):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
|
||||
f"must be a boolean (was {type(forward_host_credentials).__name__})"
|
||||
)
|
||||
if forward_host_credentials and template not in PROVIDER_TEMPLATES:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
|
||||
f"is only supported for built-in templates "
|
||||
f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
|
||||
)
|
||||
if forward_host_credentials and template not in {"codex", "claude"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
|
||||
"is only supported for templates 'codex' and 'claude'"
|
||||
)
|
||||
if forward_host_credentials and auth_token:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
|
||||
"and auth_token both set; use one or the other"
|
||||
)
|
||||
settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
|
||||
return cls(
|
||||
template=template,
|
||||
dockerfile=dockerfile,
|
||||
auth_token=auth_token,
|
||||
forward_host_credentials=forward_host_credentials,
|
||||
settings=settings,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestAgent:
|
||||
# Optional: when empty the operator selects bottles at launch time.
|
||||
bottle: str = ""
|
||||
skills: tuple[str, ...] = ()
|
||||
prompt: str = ""
|
||||
# Per-agent git identity (issue #94). Overlays the referenced
|
||||
# bottle's git-gate.user per-field at `Manifest.bottle_for`. Only
|
||||
# `user` is allowed at the agent level; `repos` stays bottle-only
|
||||
# because it carries credentials and host trust.
|
||||
git_user: ManifestGitUser = ManifestGitUser()
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, name: str, raw: object, bottle_names: set[str]) -> "ManifestAgent":
|
||||
d = as_json_object(raw, f"agent '{name}'")
|
||||
unknown = set(d.keys()) - AGENT_MODEL_KEYS
|
||||
if unknown:
|
||||
allowed = ", ".join(sorted(AGENT_MODEL_KEYS))
|
||||
raise ManifestError(
|
||||
f"agent '{name}' has unknown key(s) {sorted(unknown)}; "
|
||||
f"allowed keys are {allowed}."
|
||||
)
|
||||
|
||||
bottle_raw = d.get("bottle")
|
||||
bottle = ""
|
||||
if bottle_raw is not None:
|
||||
if not isinstance(bottle_raw, str) or not bottle_raw:
|
||||
raise ManifestError(
|
||||
f"agent '{name}' bottle must be a non-empty string when declared"
|
||||
)
|
||||
if bottle_raw not in bottle_names:
|
||||
available = ", ".join(sorted(bottle_names)) or "(none defined)"
|
||||
raise ManifestError(
|
||||
f"agent '{name}' references bottle '{bottle_raw}', which is not defined. "
|
||||
f"Available: {available}"
|
||||
)
|
||||
bottle = bottle_raw
|
||||
|
||||
skills: tuple[str, ...] = ()
|
||||
skills_raw = d.get("skills")
|
||||
if skills_raw is not None:
|
||||
if not isinstance(skills_raw, list):
|
||||
raise ManifestError(
|
||||
f"agent '{name}' skills must be an array "
|
||||
f"(was {type(skills_raw).__name__})"
|
||||
)
|
||||
collected: list[str] = []
|
||||
skills_list = cast(list[object], skills_raw)
|
||||
for i, skill in enumerate(skills_list):
|
||||
if not isinstance(skill, str):
|
||||
raise ManifestError(
|
||||
f"agent '{name}' skills[{i}] must be a string "
|
||||
f"(was {type(skill).__name__})"
|
||||
)
|
||||
# Skill names become host/guest path segments and are
|
||||
# interpolated into provisioning shell commands, so they
|
||||
# must fit the same kebab-case convention as bottle/agent
|
||||
# filenames — rejecting anything that could break out of a
|
||||
# path segment or inject shell metacharacters.
|
||||
if not is_valid_entity_name(skill):
|
||||
raise ManifestError(
|
||||
f"agent '{name}' skills[{i}] {skill!r} is not a valid "
|
||||
f"skill name; must match [a-z][a-z0-9-]*"
|
||||
)
|
||||
collected.append(skill)
|
||||
skills = tuple(collected)
|
||||
|
||||
prompt_raw = d.get("prompt")
|
||||
if prompt_raw is None:
|
||||
prompt = ""
|
||||
elif isinstance(prompt_raw, str):
|
||||
prompt = prompt_raw
|
||||
else:
|
||||
raise ManifestError(
|
||||
f"agent '{name}' prompt must be a string "
|
||||
f"(was {type(prompt_raw).__name__})"
|
||||
)
|
||||
|
||||
# git-gate: agents may declare only `git-gate.user` (name/email).
|
||||
# `git-gate.repos` is bottle-only — it carries credentials and host trust.
|
||||
git_user = ManifestGitUser()
|
||||
git_raw = d.get("git-gate")
|
||||
if git_raw is not None:
|
||||
gd = as_json_object(git_raw, f"agent '{name}' git-gate")
|
||||
for k in gd:
|
||||
if k != "user":
|
||||
raise ManifestError(
|
||||
f"agent '{name}' git-gate.{k} is not allowed at the "
|
||||
f"agent level; only git-gate.user (name/email) may be "
|
||||
f"set on an agent. git-gate.repos is bottle-only "
|
||||
f"(it carries credentials and host trust)."
|
||||
)
|
||||
if "user" in gd:
|
||||
git_user = ManifestGitUser.from_dict(name, gd["user"])
|
||||
|
||||
return cls(bottle=bottle, skills=skills, prompt=prompt, git_user=git_user)
|
||||
|
||||
|
||||
def _parse_provider_settings(
|
||||
bottle_name: str,
|
||||
template: str,
|
||||
raw: object,
|
||||
) -> dict[str, object]:
|
||||
if raw is None:
|
||||
return {}
|
||||
settings = as_json_object(raw, f"bottle '{bottle_name}' agent_provider.settings")
|
||||
|
||||
common_allowed = {"startup_args"}
|
||||
pi_allowed = {
|
||||
"provider",
|
||||
"base_url",
|
||||
"api",
|
||||
"api_key",
|
||||
"api_key_env",
|
||||
"models",
|
||||
"context_window",
|
||||
"max_tokens_field",
|
||||
"max_tokens",
|
||||
"supports_developer_role",
|
||||
"supports_reasoning_effort",
|
||||
}
|
||||
if template == "pi":
|
||||
allowed = common_allowed | pi_allowed
|
||||
elif template in ("claude", "codex"):
|
||||
allowed = common_allowed
|
||||
elif template not in PROVIDER_TEMPLATES:
|
||||
return dict(settings)
|
||||
else:
|
||||
allowed = common_allowed
|
||||
|
||||
for key in settings:
|
||||
if key not in allowed:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings has unknown "
|
||||
f"key {key!r}; allowed: {', '.join(sorted(allowed))}"
|
||||
)
|
||||
startup_args = settings.get("startup_args")
|
||||
if startup_args is not None:
|
||||
if not isinstance(startup_args, list):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.startup_args "
|
||||
f"must be an array of strings"
|
||||
)
|
||||
for i, arg in enumerate(startup_args):
|
||||
if not isinstance(arg, str) or not arg:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings."
|
||||
f"startup_args[{i}] must be a non-empty string"
|
||||
)
|
||||
if template != "pi":
|
||||
return dict(settings)
|
||||
|
||||
for key in ("provider", "base_url", "api", "api_key", "api_key_env"):
|
||||
value = settings.get(key)
|
||||
if value is not None and (not isinstance(value, str) or not value):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.{key} must "
|
||||
"be a non-empty string"
|
||||
)
|
||||
max_tokens_field = settings.get("max_tokens_field")
|
||||
if max_tokens_field is not None and max_tokens_field not in (
|
||||
"max_tokens", "max_completion_tokens",
|
||||
):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.max_tokens_field "
|
||||
"must be 'max_tokens' or 'max_completion_tokens'"
|
||||
)
|
||||
if settings.get("api_key") is not None and settings.get("api_key_env") is not None:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings may set either "
|
||||
"api_key or api_key_env, not both"
|
||||
)
|
||||
models = settings.get("models")
|
||||
if models is not None:
|
||||
if not isinstance(models, list) or not models:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.models must "
|
||||
"be a non-empty array of strings"
|
||||
)
|
||||
for i, model in enumerate(models):
|
||||
if not isinstance(model, str) or not model:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.models[{i}] "
|
||||
"must be a non-empty string"
|
||||
)
|
||||
for key in ("supports_developer_role", "supports_reasoning_effort"):
|
||||
value = settings.get(key)
|
||||
if value is not None and not isinstance(value, bool):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.{key} must "
|
||||
f"be a boolean (was {type(value).__name__})"
|
||||
)
|
||||
for key in ("context_window", "max_tokens"):
|
||||
value = settings.get(key)
|
||||
if value is not None and (
|
||||
not isinstance(value, int) or isinstance(value, bool) or value <= 0
|
||||
):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings.{key} must "
|
||||
f"be a positive integer (was {type(value).__name__})"
|
||||
)
|
||||
return dict(settings)
|
||||
@@ -0,0 +1,147 @@
|
||||
"""The `ManifestBottle` value type.
|
||||
|
||||
Split out of the package facade (`manifest/__init__.py`) so the `extends:`/
|
||||
loader resolvers can import it without a circular dependency: the facade
|
||||
imports those resolvers, while they only need this value type. Everything here
|
||||
depends on leaf modules (`util`, `agent`, `egress`, `git`, `schema`), so this
|
||||
module sits at the bottom of the manifest layer.
|
||||
|
||||
The facade re-exports `ManifestBottle`, so existing
|
||||
`from bot_bottle.manifest import ManifestBottle` callers are unaffected.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Mapping
|
||||
|
||||
from .util import ManifestError, as_json_object
|
||||
from .agent import ManifestAgentProvider
|
||||
from .egress import ManifestEgressConfig
|
||||
from .git import ManifestGitEntry, ManifestGitUser, parse_git_gate_config
|
||||
from .schema import BOTTLE_KEYS
|
||||
|
||||
__all__ = ["ManifestBottle"]
|
||||
|
||||
|
||||
def _empty_str_dict() -> dict[str, str]:
|
||||
return {}
|
||||
|
||||
|
||||
@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 daemon (PRD 0013). When true (the
|
||||
# default, issue #249), the launch step brings up a supervise
|
||||
# daemon that exposes egress MCP tools to the agent. Set
|
||||
# `supervise: false` to skip the gateway.
|
||||
supervise: bool = True
|
||||
# Guest-local container engine (issue #392). Not a host-daemon grant:
|
||||
# backends implement it inside the bottle or reject it. Gated because it
|
||||
# costs image weight, a resident service, and relaxed guest device modes
|
||||
# that the majority of bottles never need.
|
||||
nested_containers: bool = False
|
||||
# Source fields retained across extends/runtime composition. Boolean
|
||||
# defaults otherwise erase the distinction between "omitted" and an
|
||||
# explicitly declared value (especially False).
|
||||
declared_fields: frozenset[str] = frozenset()
|
||||
|
||||
@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 daemon (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 \"?<message>\" 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__})"
|
||||
)
|
||||
|
||||
nested_raw = d.get("nested_containers", False)
|
||||
if not isinstance(nested_raw, bool):
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' nested_containers must be a boolean "
|
||||
f"(was {type(nested_raw).__name__})"
|
||||
)
|
||||
|
||||
return cls(
|
||||
env=env, agent_provider=agent_provider, git=git,
|
||||
git_user=git_user, egress=egress, supervise=supervise_raw,
|
||||
nested_containers=nested_raw,
|
||||
declared_fields=frozenset(d),
|
||||
)
|
||||
@@ -0,0 +1,452 @@
|
||||
"""Egress routing manifest dataclasses and helpers (PRD 0017, PRD 0053)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import cast
|
||||
|
||||
from .util import ManifestError, as_json_object
|
||||
|
||||
EGRESS_AUTH_SCHEMES = ("Bearer", "token")
|
||||
|
||||
PATH_MATCH_TYPES = ("exact", "prefix", "regex")
|
||||
HEADER_MATCH_TYPES = ("exact", "regex")
|
||||
|
||||
VALID_METHODS = frozenset({
|
||||
"GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "TRACE",
|
||||
"CONNECT",
|
||||
})
|
||||
|
||||
OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets"})
|
||||
INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})
|
||||
|
||||
# What the proxy does on an outbound token match (PRD 0062).
|
||||
OUTBOUND_ON_MATCH_VALUES = ("block", "redact", "supervise")
|
||||
|
||||
|
||||
def validate_egress_routes(
|
||||
bottle_name: str,
|
||||
routes: tuple[ManifestEgressRoute, ...],
|
||||
) -> None:
|
||||
seen_hosts: dict[str, None] = {}
|
||||
for r in routes:
|
||||
key = r.Host.lower()
|
||||
if key in seen_hosts:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' egress.routes has duplicate host "
|
||||
f"{r.Host!r}; each host must be unique on the proxy."
|
||||
)
|
||||
seen_hosts[key] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestPathMatch:
|
||||
Type: str = "prefix"
|
||||
Value: str = ""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestHeaderMatch:
|
||||
Name: str = ""
|
||||
Value: str = ""
|
||||
Type: str = "exact"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestMatchEntry:
|
||||
Paths: tuple[ManifestPathMatch, ...] = ()
|
||||
Methods: tuple[str, ...] = ()
|
||||
Headers: tuple[ManifestHeaderMatch, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestEgressRoute:
|
||||
Host: str
|
||||
Matches: tuple[ManifestMatchEntry, ...] = ()
|
||||
AuthScheme: str = ""
|
||||
TokenRef: str = ""
|
||||
Role: tuple[str, ...] = ()
|
||||
GitFetch: bool = False
|
||||
OutboundDetectors: tuple[str, ...] | None = None
|
||||
InboundDetectors: tuple[str, ...] | None = None
|
||||
OutboundOnMatch: str = ""
|
||||
PreserveAuth: bool = False
|
||||
Inspect: bool = True
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "ManifestEgressRoute":
|
||||
label = f"bottle '{bottle_name}' egress.routes[{idx}]"
|
||||
d = as_json_object(raw, label)
|
||||
host = d.get("host")
|
||||
if not isinstance(host, str) or not host:
|
||||
raise ManifestError(f"{label} missing required string field 'host'")
|
||||
|
||||
inspect_raw = d.get("inspect", {})
|
||||
if inspect_raw is False:
|
||||
inspect = False
|
||||
inspect_d: dict[str, object] = {}
|
||||
else:
|
||||
inspect = True
|
||||
inspect_d = as_json_object(inspect_raw, f"{label} inspect")
|
||||
|
||||
# --- matches ---
|
||||
matches: tuple[ManifestMatchEntry, ...] = ()
|
||||
matches_raw = inspect_d.get("matches")
|
||||
if matches_raw is not None:
|
||||
if not isinstance(matches_raw, list):
|
||||
raise ManifestError(
|
||||
f"{label} matches must be an array "
|
||||
f"(was {type(matches_raw).__name__})"
|
||||
)
|
||||
matches_list = cast(list[object], matches_raw)
|
||||
entries: list[ManifestMatchEntry] = []
|
||||
for k, entry_raw in enumerate(matches_list):
|
||||
entries.append(
|
||||
_parse_match_entry(label, k, entry_raw)
|
||||
)
|
||||
matches = tuple(entries)
|
||||
|
||||
# --- auth ---
|
||||
auth_scheme = ""
|
||||
token_ref = ""
|
||||
if "auth" in inspect_d:
|
||||
auth_raw = inspect_d.get("auth")
|
||||
auth_d = as_json_object(auth_raw, f"{label} inspect.auth")
|
||||
if not auth_d:
|
||||
raise ManifestError(
|
||||
f"{label} auth is empty ({{}}); omit the 'auth' key "
|
||||
f"entirely if this route is unauthenticated. Otherwise "
|
||||
f"both 'scheme' and 'token_ref' are required."
|
||||
)
|
||||
auth_scheme_raw = auth_d.get("scheme")
|
||||
if not isinstance(auth_scheme_raw, str) or not auth_scheme_raw:
|
||||
raise ManifestError(
|
||||
f"{label} auth.scheme is required when 'auth' is set "
|
||||
f"(non-empty string)"
|
||||
)
|
||||
if auth_scheme_raw not in EGRESS_AUTH_SCHEMES:
|
||||
raise ManifestError(
|
||||
f"{label} auth.scheme {auth_scheme_raw!r} is not one of "
|
||||
f"{', '.join(EGRESS_AUTH_SCHEMES)}"
|
||||
)
|
||||
token_ref_raw = auth_d.get("token_ref")
|
||||
if not isinstance(token_ref_raw, str) or not token_ref_raw:
|
||||
raise ManifestError(
|
||||
f"{label} auth.token_ref is required when 'auth' is set "
|
||||
f"(name of the host env var holding the token value)"
|
||||
)
|
||||
for k in auth_d:
|
||||
if k not in ("scheme", "token_ref"):
|
||||
raise ManifestError(
|
||||
f"{label} auth has unknown key {k!r}; "
|
||||
f"only 'scheme' and 'token_ref' are accepted"
|
||||
)
|
||||
auth_scheme = auth_scheme_raw
|
||||
token_ref = token_ref_raw
|
||||
|
||||
# --- role (reserved) ---
|
||||
role_raw = d.get("role")
|
||||
roles: tuple[str, ...] = ()
|
||||
if role_raw is None:
|
||||
roles = ()
|
||||
elif isinstance(role_raw, str):
|
||||
roles = (role_raw,)
|
||||
elif isinstance(role_raw, list):
|
||||
role_list = cast(list[object], role_raw)
|
||||
collected_roles: list[str] = []
|
||||
for r in role_list:
|
||||
if not isinstance(r, str):
|
||||
msg = f"{label} role items must be strings (got {type(r).__name__})"
|
||||
raise ManifestError(msg)
|
||||
collected_roles.append(r)
|
||||
roles = tuple(collected_roles)
|
||||
else:
|
||||
raise ManifestError(
|
||||
f"{label} role must be a string or a list of strings "
|
||||
f"(was {type(role_raw).__name__})"
|
||||
)
|
||||
if roles:
|
||||
raise ManifestError(
|
||||
f"{label} role {roles[0]!r} is not accepted; "
|
||||
f"the 'role' field is reserved for future use"
|
||||
)
|
||||
|
||||
# --- DLP settings (inspection-only) ---
|
||||
outbound_detectors: tuple[str, ...] | None = None
|
||||
inbound_detectors: tuple[str, ...] | None = None
|
||||
outbound_on_match = ""
|
||||
if inspect:
|
||||
outbound_detectors, inbound_detectors, outbound_on_match = _parse_inspect_block(
|
||||
label, inspect_d,
|
||||
)
|
||||
|
||||
# --- git-over-HTTPS policy ---
|
||||
git_fetch = False
|
||||
if "git" in inspect_d:
|
||||
git_d = as_json_object(inspect_d.get("git"), f"{label} inspect.git")
|
||||
raw_fetch = git_d.get("fetch", False)
|
||||
if isinstance(raw_fetch, bool):
|
||||
git_fetch = raw_fetch
|
||||
else:
|
||||
raise ManifestError(
|
||||
f"{label} git.fetch must be a boolean "
|
||||
f"(was {type(raw_fetch).__name__})"
|
||||
)
|
||||
for k in git_d:
|
||||
if k != "fetch":
|
||||
raise ManifestError(
|
||||
f"{label} git has unknown key {k!r}; "
|
||||
f"only 'fetch' is accepted"
|
||||
)
|
||||
|
||||
# --- preserve_auth ---
|
||||
preserve_auth = False
|
||||
if "preserve_auth" in inspect_d:
|
||||
raw_preserve_auth = inspect_d.get("preserve_auth")
|
||||
if not isinstance(raw_preserve_auth, bool):
|
||||
raise ManifestError(
|
||||
f"{label} preserve_auth must be a boolean "
|
||||
f"(was {type(raw_preserve_auth).__name__})"
|
||||
)
|
||||
preserve_auth = raw_preserve_auth
|
||||
|
||||
for k in inspect_d:
|
||||
if k not in (
|
||||
"matches", "auth", "git", "preserve_auth",
|
||||
"outbound_detectors", "inbound_detectors", "outbound_on_match",
|
||||
):
|
||||
raise ManifestError(
|
||||
f"{label} inspect has unknown key {k!r}; accepted keys are "
|
||||
f"'matches', 'auth', 'git', 'preserve_auth', "
|
||||
f"'outbound_detectors', 'inbound_detectors', "
|
||||
f"'outbound_on_match'"
|
||||
)
|
||||
for k in d:
|
||||
if k not in ("host", "role", "inspect"):
|
||||
raise ManifestError(
|
||||
f"{label} has unknown key {k!r}; accepted keys are "
|
||||
f"'host', 'role', and 'inspect'"
|
||||
)
|
||||
|
||||
return cls(
|
||||
Host=host,
|
||||
Matches=matches,
|
||||
AuthScheme=auth_scheme,
|
||||
TokenRef=token_ref,
|
||||
Role=roles,
|
||||
GitFetch=git_fetch,
|
||||
OutboundDetectors=outbound_detectors,
|
||||
InboundDetectors=inbound_detectors,
|
||||
OutboundOnMatch=outbound_on_match,
|
||||
PreserveAuth=preserve_auth,
|
||||
Inspect=inspect,
|
||||
)
|
||||
|
||||
|
||||
def _parse_match_entry(
|
||||
route_label: str, k: int, raw: object,
|
||||
) -> ManifestMatchEntry:
|
||||
label = f"{route_label} matches[{k}]"
|
||||
d = as_json_object(raw, label)
|
||||
|
||||
paths: tuple[ManifestPathMatch, ...] = ()
|
||||
paths_raw = d.get("paths")
|
||||
if paths_raw is not None:
|
||||
if not isinstance(paths_raw, list):
|
||||
raise ManifestError(f"{label} paths must be an array")
|
||||
paths_list = cast(list[object], paths_raw)
|
||||
parsed_paths: list[ManifestPathMatch] = []
|
||||
for j, p_raw in enumerate(paths_list):
|
||||
parsed_paths.append(_parse_path_match(label, j, p_raw))
|
||||
paths = tuple(parsed_paths)
|
||||
|
||||
methods: tuple[str, ...] = ()
|
||||
methods_raw = d.get("methods")
|
||||
if methods_raw is not None:
|
||||
if not isinstance(methods_raw, list):
|
||||
raise ManifestError(f"{label} methods must be an array")
|
||||
methods_list = cast(list[object], methods_raw)
|
||||
normalised: list[str] = []
|
||||
for j, m in enumerate(methods_list):
|
||||
if not isinstance(m, str):
|
||||
raise ManifestError(
|
||||
f"{label} methods[{j}] must be a string"
|
||||
)
|
||||
upper = m.upper()
|
||||
if upper not in VALID_METHODS:
|
||||
raise ManifestError(
|
||||
f"{label} methods[{j}] {m!r} is not a valid HTTP method"
|
||||
)
|
||||
normalised.append(upper)
|
||||
methods = tuple(normalised)
|
||||
|
||||
headers: tuple[ManifestHeaderMatch, ...] = ()
|
||||
headers_raw = d.get("headers")
|
||||
if headers_raw is not None:
|
||||
if not isinstance(headers_raw, list):
|
||||
raise ManifestError(f"{label} headers must be an array")
|
||||
headers_list = cast(list[object], headers_raw)
|
||||
parsed_headers: list[ManifestHeaderMatch] = []
|
||||
for j, h_raw in enumerate(headers_list):
|
||||
parsed_headers.append(_parse_header_match(label, j, h_raw))
|
||||
headers = tuple(parsed_headers)
|
||||
|
||||
for key in d:
|
||||
if key not in ("paths", "methods", "headers"):
|
||||
raise ManifestError(f"{label} has unknown key {key!r}")
|
||||
|
||||
return ManifestMatchEntry(Paths=paths, Methods=methods, Headers=headers)
|
||||
|
||||
|
||||
def _parse_path_match(
|
||||
entry_label: str, j: int, raw: object,
|
||||
) -> ManifestPathMatch:
|
||||
label = f"{entry_label} paths[{j}]"
|
||||
d = as_json_object(raw, label)
|
||||
ptype = d.get("type", "prefix")
|
||||
if not isinstance(ptype, str) or ptype not in PATH_MATCH_TYPES:
|
||||
raise ManifestError(
|
||||
f"{label} type must be one of {', '.join(PATH_MATCH_TYPES)} "
|
||||
f"(got {ptype!r})"
|
||||
)
|
||||
value = d.get("value")
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ManifestError(f"{label} value must be a non-empty string")
|
||||
if ptype in ("exact", "prefix") and not value.startswith("/"):
|
||||
raise ManifestError(
|
||||
f"{label} value {value!r} must start with '/' for type {ptype!r}"
|
||||
)
|
||||
if ptype == "regex":
|
||||
try:
|
||||
re.compile(value)
|
||||
except re.error as e:
|
||||
raise ManifestError(
|
||||
f"{label} regex {value!r} failed to compile: {e}"
|
||||
) from e
|
||||
for k in d:
|
||||
if k not in ("type", "value"):
|
||||
raise ManifestError(f"{label} has unknown key {k!r}")
|
||||
return ManifestPathMatch(Type=ptype, Value=value)
|
||||
|
||||
|
||||
def _parse_header_match(
|
||||
entry_label: str, j: int, raw: object,
|
||||
) -> ManifestHeaderMatch:
|
||||
label = f"{entry_label} headers[{j}]"
|
||||
d = as_json_object(raw, label)
|
||||
name = d.get("name")
|
||||
if not isinstance(name, str) or not name:
|
||||
raise ManifestError(f"{label} name must be a non-empty string")
|
||||
value = d.get("value")
|
||||
if not isinstance(value, str):
|
||||
raise ManifestError(f"{label} value must be a string")
|
||||
htype = d.get("type", "exact")
|
||||
if not isinstance(htype, str) or htype not in HEADER_MATCH_TYPES:
|
||||
raise ManifestError(
|
||||
f"{label} type must be one of {', '.join(HEADER_MATCH_TYPES)} "
|
||||
f"(got {htype!r})"
|
||||
)
|
||||
if htype == "regex":
|
||||
try:
|
||||
re.compile(value)
|
||||
except re.error as e:
|
||||
raise ManifestError(
|
||||
f"{label} regex {value!r} failed to compile: {e}"
|
||||
) from e
|
||||
for k in d:
|
||||
if k not in ("name", "value", "type"):
|
||||
raise ManifestError(f"{label} has unknown key {k!r}")
|
||||
return ManifestHeaderMatch(Name=name, Value=value, Type=htype)
|
||||
|
||||
|
||||
def _parse_inspect_block(
|
||||
route_label: str,
|
||||
inspect: dict[str, object],
|
||||
) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
|
||||
"""Parse DLP settings from an inspected route."""
|
||||
label = f"{route_label} inspect"
|
||||
d = inspect
|
||||
|
||||
def _parse_field(
|
||||
field: str,
|
||||
valid_names: frozenset[str],
|
||||
) -> tuple[str, ...] | None:
|
||||
val = d.get(field)
|
||||
if val is None:
|
||||
return None
|
||||
if val is False:
|
||||
return ()
|
||||
if not isinstance(val, list):
|
||||
raise ManifestError(
|
||||
f"{label} {field} must be false, a list, or omitted"
|
||||
)
|
||||
items = cast(list[object], val)
|
||||
names: list[str] = []
|
||||
for j, item in enumerate(items):
|
||||
if not isinstance(item, str):
|
||||
raise ManifestError(
|
||||
f"{label} {field}[{j}] must be a string"
|
||||
)
|
||||
if item not in valid_names:
|
||||
raise ManifestError(
|
||||
f"{label} {field}[{j}] {item!r} is not a valid "
|
||||
f"detector; valid: {', '.join(sorted(valid_names))}"
|
||||
)
|
||||
names.append(item)
|
||||
return tuple(names)
|
||||
|
||||
outbound = _parse_field("outbound_detectors", OUTBOUND_DETECTOR_NAMES)
|
||||
inbound = _parse_field("inbound_detectors", INBOUND_DETECTOR_NAMES)
|
||||
|
||||
on_match = ""
|
||||
on_match_raw = d.get("outbound_on_match")
|
||||
if on_match_raw is not None:
|
||||
if not isinstance(on_match_raw, str) or on_match_raw not in OUTBOUND_ON_MATCH_VALUES:
|
||||
raise ManifestError(
|
||||
f"{label} outbound_on_match must be one of "
|
||||
f"{', '.join(OUTBOUND_ON_MATCH_VALUES)} (got {on_match_raw!r})"
|
||||
)
|
||||
on_match = on_match_raw
|
||||
|
||||
return outbound, inbound, on_match
|
||||
|
||||
|
||||
LOG_LEVELS = frozenset({0, 1, 2})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestEgressConfig:
|
||||
routes: tuple[ManifestEgressRoute, ...] = ()
|
||||
Log: int = 0
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, bottle_name: str, raw: object) -> "ManifestEgressConfig":
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' egress")
|
||||
routes_raw = d.get("routes")
|
||||
routes: tuple[ManifestEgressRoute, ...] = ()
|
||||
if routes_raw is not None:
|
||||
if not isinstance(routes_raw, list):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' egress.routes must be an array "
|
||||
f"(was {type(routes_raw).__name__})"
|
||||
)
|
||||
routes_list = cast(list[object], routes_raw)
|
||||
routes = tuple(
|
||||
ManifestEgressRoute.from_dict(bottle_name, i, entry)
|
||||
for i, entry in enumerate(routes_list)
|
||||
)
|
||||
validate_egress_routes(bottle_name, routes)
|
||||
log_raw = d.get("log", 0)
|
||||
if isinstance(log_raw, bool) or not isinstance(log_raw, int) \
|
||||
or log_raw not in LOG_LEVELS:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' egress.log must be 0, 1, or 2"
|
||||
)
|
||||
for k in d:
|
||||
if k not in ("routes", "log"):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' egress has unknown key {k!r}; "
|
||||
f"accepted keys are 'routes', 'log'"
|
||||
)
|
||||
return cls(routes=routes, Log=log_raw)
|
||||
@@ -0,0 +1,360 @@
|
||||
"""Internal bottle `extends:` resolution for manifests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .bottle import ManifestBottle
|
||||
from .egress import ManifestEgressConfig, validate_egress_routes
|
||||
from .git import ManifestGitUser, parse_git_gate_config
|
||||
from .util import ManifestError, as_json_object
|
||||
|
||||
|
||||
def _overlay_declared_bool(
|
||||
base: ManifestBottle,
|
||||
override: ManifestBottle,
|
||||
field: str,
|
||||
) -> bool:
|
||||
"""Overlay a defaulted boolean only when override declared it."""
|
||||
value = getattr(override if field in override.declared_fields else base, field)
|
||||
assert isinstance(value, bool)
|
||||
return value
|
||||
|
||||
|
||||
def merge_bottles_runtime(bottles: "list[ManifestBottle]") -> "ManifestBottle":
|
||||
"""Merge an ordered list of pre-resolved ManifestBottle objects.
|
||||
|
||||
Index 0 is the base; each subsequent entry is applied on top using
|
||||
the same field-merge rules as the file-based extends machinery:
|
||||
env: dict merge, later wins; git_user: per-field overlay, later
|
||||
wins on non-empty; git (repos): union by name, later wins; egress
|
||||
routes: concatenate; agent_provider, supervise, nested_containers:
|
||||
later replaces (presence-aware).
|
||||
|
||||
Defaulted booleans use presence-aware replacement: if the later bottle
|
||||
was loaded from a source that explicitly declared the key, its value
|
||||
wins (so an explicit `false` can override an earlier `true`). If the
|
||||
later bottle never mentioned the key, the earlier value is preserved.
|
||||
"""
|
||||
if not bottles:
|
||||
raise ValueError("merge_bottles_runtime requires at least one bottle")
|
||||
result = bottles[0]
|
||||
for override in bottles[1:]:
|
||||
result = _merge_two_bottles_runtime(result, override)
|
||||
return result
|
||||
|
||||
|
||||
def _merge_two_bottles_runtime(base: "ManifestBottle", override: "ManifestBottle") -> "ManifestBottle":
|
||||
merged_env = {**base.env, **override.env}
|
||||
|
||||
merged_git_user = ManifestGitUser(
|
||||
name=override.git_user.name or base.git_user.name,
|
||||
email=override.git_user.email or base.git_user.email,
|
||||
)
|
||||
|
||||
# git repos: union keyed by Name, override wins per-name.
|
||||
base_repos_by_name = {entry.Name: entry for entry in base.git}
|
||||
override_repos_by_name = {entry.Name: entry for entry in override.git}
|
||||
merged_repos_names = list(base_repos_by_name) + [
|
||||
n for n in override_repos_by_name if n not in base_repos_by_name
|
||||
]
|
||||
merged_git = tuple(
|
||||
override_repos_by_name[n] if n in override_repos_by_name else base_repos_by_name[n]
|
||||
for n in merged_repos_names
|
||||
)
|
||||
|
||||
merged_routes = base.egress.routes + override.egress.routes
|
||||
merged_egress = ManifestEgressConfig(routes=merged_routes, Log=override.egress.Log)
|
||||
|
||||
return ManifestBottle(
|
||||
env=merged_env,
|
||||
agent_provider=override.agent_provider,
|
||||
git=merged_git,
|
||||
git_user=merged_git_user,
|
||||
egress=merged_egress,
|
||||
supervise=_overlay_declared_bool(base, override, "supervise"),
|
||||
nested_containers=_overlay_declared_bool(
|
||||
base, override, "nested_containers"
|
||||
),
|
||||
declared_fields=base.declared_fields | override.declared_fields,
|
||||
)
|
||||
|
||||
|
||||
def resolve_bottles(raws: dict[str, dict[str, object]]) -> dict[str, ManifestBottle]:
|
||||
"""Apply `extends:` chains and return resolved ManifestBottle objects."""
|
||||
cache: dict[str, ManifestBottle] = {}
|
||||
# Per-bottle effective git-gate.repos, as raw dicts keyed by repo name.
|
||||
# Threaded alongside `cache` so a child can field-merge against its
|
||||
# parent's repos without reconstructing them from parsed entries.
|
||||
repos_cache: dict[str, dict[str, object]] = {}
|
||||
for name in raws:
|
||||
if name not in cache:
|
||||
_resolve_one_bottle(name, raws, cache, repos_cache, ())
|
||||
return cache
|
||||
|
||||
|
||||
def _resolve_one_bottle(
|
||||
name: str,
|
||||
raws: dict[str, dict[str, object]],
|
||||
cache: dict[str, ManifestBottle],
|
||||
repos_cache: dict[str, dict[str, object]],
|
||||
seen: tuple[str, ...],
|
||||
) -> ManifestBottle:
|
||||
if name in cache:
|
||||
return cache[name]
|
||||
if name in seen:
|
||||
chain = " -> ".join(seen + (name,))
|
||||
raise ManifestError(f"bottle '{name}' is in an extends cycle: {chain}")
|
||||
raw = raws[name]
|
||||
parent_name_raw = raw.get("extends")
|
||||
# Strip `extends:` before passing to ManifestBottle.from_dict so it
|
||||
# is not accidentally treated as a real ManifestBottle field by future
|
||||
# schema additions. It is only meaningful here.
|
||||
child_raw = {k: v for k, v in raw.items() if k != "extends"}
|
||||
|
||||
if parent_name_raw is None:
|
||||
bottle = ManifestBottle.from_dict(name, child_raw)
|
||||
cache[name] = bottle
|
||||
repos_cache[name] = _resolve_repos_raw({}, child_raw)
|
||||
return bottle
|
||||
|
||||
# Normalize to list, accepting both str and list[str].
|
||||
raw_list: list[object]
|
||||
if isinstance(parent_name_raw, str):
|
||||
raw_list = [parent_name_raw]
|
||||
elif isinstance(parent_name_raw, list):
|
||||
raw_list = parent_name_raw
|
||||
else:
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' extends must be a string or list of strings "
|
||||
f"(was {type(parent_name_raw).__name__})"
|
||||
)
|
||||
|
||||
# Validate each entry before resolving any of them.
|
||||
parent_names: list[str] = []
|
||||
for i, pname in enumerate(raw_list):
|
||||
if not isinstance(pname, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' extends[{i}] must be a string "
|
||||
f"(was {type(pname).__name__})"
|
||||
)
|
||||
parent_names.append(pname)
|
||||
if pname == name:
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' extends itself; remove the self-reference"
|
||||
)
|
||||
if pname not in raws:
|
||||
avail = ", ".join(sorted(raws.keys())) or "(none)"
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' extends '{pname}' which is not "
|
||||
f"defined. Available bottles: {avail}"
|
||||
)
|
||||
|
||||
combined_parent, combined_repos_raw = _fold_parents(
|
||||
parent_names, raws, cache, repos_cache, seen + (name,)
|
||||
)
|
||||
merged_repos_raw = _resolve_repos_raw(combined_repos_raw, child_raw)
|
||||
bottle = _merge_bottles(combined_parent, child_raw, merged_repos_raw, name)
|
||||
cache[name] = bottle
|
||||
repos_cache[name] = merged_repos_raw
|
||||
return bottle
|
||||
|
||||
|
||||
def _fold_parents(
|
||||
parent_names: list[str],
|
||||
raws: dict[str, dict[str, object]],
|
||||
cache: dict[str, ManifestBottle],
|
||||
repos_cache: dict[str, dict[str, object]],
|
||||
seen: tuple[str, ...],
|
||||
) -> tuple[ManifestBottle, dict[str, object]]:
|
||||
"""Resolve each parent and fold them left-to-right.
|
||||
|
||||
Later parents win over earlier ones on conflict. The `seen` tuple
|
||||
carries the current bottle's name so cycle detection works across
|
||||
every parent edge in the multi-parent graph."""
|
||||
first = parent_names[0]
|
||||
effective = _resolve_one_bottle(first, raws, cache, repos_cache, seen)
|
||||
effective_repos_raw = repos_cache[first]
|
||||
for pname in parent_names[1:]:
|
||||
later = _resolve_one_bottle(pname, raws, cache, repos_cache, seen)
|
||||
later_repos_raw = repos_cache[pname]
|
||||
effective, effective_repos_raw = _fold_two_bottles(
|
||||
effective, effective_repos_raw, later, later_repos_raw
|
||||
)
|
||||
return effective, effective_repos_raw
|
||||
|
||||
|
||||
def _fold_two_bottles(
|
||||
earlier: ManifestBottle,
|
||||
earlier_repos_raw: dict[str, object],
|
||||
later: ManifestBottle,
|
||||
later_repos_raw: dict[str, object],
|
||||
) -> tuple[ManifestBottle, dict[str, object]]:
|
||||
"""Combine two resolved parent bottles; later wins over earlier."""
|
||||
merged_env = {**earlier.env, **later.env}
|
||||
|
||||
merged_git_user = ManifestGitUser(
|
||||
name=later.git_user.name or earlier.git_user.name,
|
||||
email=later.git_user.email or earlier.git_user.email,
|
||||
)
|
||||
|
||||
# Repos: union by name; for same-name entries, later wins per-field.
|
||||
# Unlike _resolve_repos_raw, an empty later_repos_raw means "no repos
|
||||
# declared" — it does NOT clear the earlier parent's repos.
|
||||
names = list(earlier_repos_raw) + [
|
||||
n for n in later_repos_raw if n not in earlier_repos_raw
|
||||
]
|
||||
merged_repos_raw: dict[str, object] = {
|
||||
n: {
|
||||
**as_json_object(earlier_repos_raw.get(n, {}), "earlier parent repo"),
|
||||
**as_json_object(later_repos_raw.get(n, {}), "later parent repo"),
|
||||
}
|
||||
for n in names
|
||||
}
|
||||
if merged_repos_raw:
|
||||
merged_git, _ = parse_git_gate_config("_fold", {"repos": merged_repos_raw})
|
||||
else:
|
||||
merged_git = ()
|
||||
|
||||
# Egress: routes concatenate; scalar fields use last-wins.
|
||||
merged_egress = ManifestEgressConfig(
|
||||
routes=earlier.egress.routes + later.egress.routes,
|
||||
Log=later.egress.Log,
|
||||
)
|
||||
|
||||
return ManifestBottle(
|
||||
env=merged_env,
|
||||
agent_provider=later.agent_provider,
|
||||
git=merged_git,
|
||||
git_user=merged_git_user,
|
||||
egress=merged_egress,
|
||||
supervise=_overlay_declared_bool(earlier, later, "supervise"),
|
||||
nested_containers=_overlay_declared_bool(
|
||||
earlier, later, "nested_containers"
|
||||
),
|
||||
declared_fields=earlier.declared_fields | later.declared_fields,
|
||||
), merged_repos_raw
|
||||
|
||||
|
||||
def _merge_bottles(
|
||||
parent: ManifestBottle,
|
||||
child_raw: dict[str, object],
|
||||
merged_repos_raw: dict[str, object],
|
||||
name: str,
|
||||
) -> ManifestBottle:
|
||||
"""Apply PRD 0025 merge rules."""
|
||||
# git-gate.repos: when the child declares repos, inject the already
|
||||
# name-merged repo set (computed by _resolve_repos_raw) so the child
|
||||
# parses with the full inherited+overridden list (issue #237).
|
||||
if _child_declares_git_gate_repos(child_raw):
|
||||
git_raw = as_json_object(child_raw.get("git-gate", {}), "child git-gate")
|
||||
child_raw = {**child_raw, "git-gate": {**git_raw, "repos": merged_repos_raw}}
|
||||
|
||||
# Parse the child's declared fields into a ManifestBottle (with the
|
||||
# usual defaults for anything missing). Validation runs the same
|
||||
# way it would for a leaf bottle: typos / wrong types die here.
|
||||
child = ManifestBottle.from_dict(name, child_raw)
|
||||
|
||||
# env: dict merge, child wins on collision.
|
||||
merged_env = {**parent.env, **child.env}
|
||||
|
||||
# git-gate.user: per-field overlay. Each non-empty field on child
|
||||
# wins; empties fall through to parent. The default ManifestGitUser()
|
||||
# is two empty strings, so a child that omits git-gate.user
|
||||
# inherits the parent's user verbatim.
|
||||
merged_git_user = ManifestGitUser(
|
||||
name=child.git_user.name or parent.git_user.name,
|
||||
email=child.git_user.email or parent.git_user.email,
|
||||
)
|
||||
|
||||
# git-gate.repos: when declared, child.git already holds the merged
|
||||
# set (an explicit empty dict clears parent, leaving child.git empty).
|
||||
# When omitted, the parent's entries are inherited verbatim.
|
||||
if _child_declares_git_gate_repos(child_raw):
|
||||
merged_git = child.git
|
||||
else:
|
||||
merged_git = parent.git
|
||||
|
||||
# egress.routes: missing means inherit; otherwise parent and child
|
||||
# route lists concatenate. Other egress scalar fields remain
|
||||
# presence-driven overlays.
|
||||
merged_egress = (
|
||||
_merge_egress(parent.egress, child.egress, child_raw)
|
||||
if "egress" in child_raw
|
||||
else parent.egress
|
||||
)
|
||||
|
||||
# Presence-driven full-replace for the remaining scalar fields.
|
||||
merged_agent_provider = (
|
||||
child.agent_provider
|
||||
if "agent_provider" in child_raw
|
||||
else parent.agent_provider
|
||||
)
|
||||
merged_supervise = _overlay_declared_bool(parent, child, "supervise")
|
||||
merged_nested_containers = _overlay_declared_bool(
|
||||
parent, child, "nested_containers"
|
||||
)
|
||||
validate_egress_routes(name, merged_egress.routes)
|
||||
|
||||
return ManifestBottle(
|
||||
env=merged_env,
|
||||
agent_provider=merged_agent_provider,
|
||||
git=merged_git,
|
||||
git_user=merged_git_user,
|
||||
egress=merged_egress,
|
||||
supervise=merged_supervise,
|
||||
nested_containers=merged_nested_containers,
|
||||
declared_fields=parent.declared_fields | child.declared_fields,
|
||||
)
|
||||
|
||||
|
||||
def _resolve_repos_raw(
|
||||
parent_repos: dict[str, object],
|
||||
child_raw: dict[str, object],
|
||||
) -> dict[str, object]:
|
||||
"""Compute a bottle's effective git-gate.repos as raw dicts.
|
||||
|
||||
Repos are keyed by name. When the child omits git-gate.repos it
|
||||
inherits the parent's set verbatim; an explicit empty dict clears it.
|
||||
Otherwise parent and child unite by name, with same-name entries
|
||||
field-merged (parent fields are defaults, child fields win)."""
|
||||
if not _child_declares_git_gate_repos(child_raw):
|
||||
return parent_repos
|
||||
child_repos = _declared_repos_raw(child_raw)
|
||||
if not child_repos:
|
||||
return {}
|
||||
# Parent entries keep their order; child-only names are appended.
|
||||
names = list(parent_repos) + [n for n in child_repos if n not in parent_repos]
|
||||
return {
|
||||
name: {
|
||||
**as_json_object(parent_repos.get(name, {}), "parent git-gate repo"),
|
||||
**as_json_object(child_repos.get(name, {}), "child git-gate repo"),
|
||||
}
|
||||
for name in names
|
||||
}
|
||||
|
||||
|
||||
def _declared_repos_raw(child_raw: dict[str, object]) -> dict[str, object]:
|
||||
"""Return the child's explicitly declared git-gate.repos as raw dicts,
|
||||
or an empty dict when none are declared."""
|
||||
if not _child_declares_git_gate_repos(child_raw):
|
||||
return {}
|
||||
git_raw = as_json_object(child_raw.get("git-gate", {}), "child git-gate")
|
||||
return as_json_object(git_raw.get("repos", {}), "child git-gate.repos")
|
||||
|
||||
|
||||
def _child_declares_git_gate_repos(child_raw: dict[str, object]) -> bool:
|
||||
git_raw = child_raw.get("git-gate")
|
||||
if git_raw is None:
|
||||
return False
|
||||
git_obj = as_json_object(git_raw, "child git-gate")
|
||||
return "repos" in git_obj
|
||||
|
||||
|
||||
def _merge_egress(
|
||||
parent: ManifestEgressConfig,
|
||||
child: ManifestEgressConfig,
|
||||
child_raw: dict[str, object],
|
||||
) -> ManifestEgressConfig:
|
||||
child_egress_raw = as_json_object(child_raw.get("egress"), "child egress")
|
||||
routes = parent.routes + child.routes
|
||||
log = child.Log if "log" in child_egress_raw else parent.Log
|
||||
return ManifestEgressConfig(routes=routes, Log=log)
|
||||
@@ -0,0 +1,314 @@
|
||||
"""Git-related manifest dataclasses and helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .util import ManifestError, as_json_object
|
||||
|
||||
# Shell-safe characters for git-gate repo names. Names are embedded in
|
||||
# the generated entrypoint shell script (shlex.quote is the primary
|
||||
# defence; this regex is belt-and-suspenders and documents intent).
|
||||
_GIT_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
|
||||
|
||||
_KEY_PROVIDERS = {"static", "gitea"}
|
||||
|
||||
|
||||
def _opt_str(value: object, label: str) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if not isinstance(value, str):
|
||||
raise ManifestError(f"{label} must be a string (was {type(value).__name__})")
|
||||
return value
|
||||
|
||||
|
||||
def parse_git_upstream(url: str, label: str) -> tuple[str, str, str, str]:
|
||||
"""Parse `ssh://user@host[:port]/path` into (user, host, port, path).
|
||||
Dies if `url` doesn't match the ssh:// shape v1 supports. Default
|
||||
port is 22 (matches OpenSSH)."""
|
||||
if not url.startswith("ssh://"):
|
||||
raise ManifestError(f"{label} must be an ssh:// URL (was {url!r})")
|
||||
rest = url[len("ssh://"):]
|
||||
if "@" not in rest:
|
||||
raise ManifestError(
|
||||
f"{label} must include a user (e.g. ssh://git@host/path.git); "
|
||||
f"was {url!r}"
|
||||
)
|
||||
user, _, hostpart = rest.partition("@")
|
||||
if not user:
|
||||
raise ManifestError(f"{label} user is empty in {url!r}")
|
||||
if "/" not in hostpart:
|
||||
raise ManifestError(
|
||||
f"{label} must include a path (e.g. ssh://git@host/path.git); "
|
||||
f"was {url!r}"
|
||||
)
|
||||
hostport, _, path = hostpart.partition("/")
|
||||
if not path:
|
||||
raise ManifestError(f"{label} path is empty in {url!r}")
|
||||
if ":" in hostport:
|
||||
host, _, port = hostport.partition(":")
|
||||
if not port.isdigit():
|
||||
raise ManifestError(f"{label} port must be numeric in {url!r}")
|
||||
else:
|
||||
host = hostport
|
||||
port = "22"
|
||||
if not host:
|
||||
raise ManifestError(f"{label} host is empty in {url!r}")
|
||||
return (user, host, port, path)
|
||||
|
||||
|
||||
def validate_unique_git_names(bottle_name: str, git: tuple[ManifestGitEntry, ...]) -> None:
|
||||
seen: dict[str, None] = {}
|
||||
for g in git:
|
||||
if g.Name in seen:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.repos has duplicate name '{g.Name}'; "
|
||||
f"each entry maps to a distinct bare repo on the gate."
|
||||
)
|
||||
seen[g.Name] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestKeyConfig:
|
||||
"""Configuration for a repo's SSH key in git-gate.repos.
|
||||
|
||||
`provider` is either `"static"` (a pre-existing key on the host) or
|
||||
`"gitea"` (automatic deploy-key lifecycle via the Gitea API).
|
||||
|
||||
For `static`: `path` is the host-side absolute path to the SSH private key.
|
||||
|
||||
For `gitea`: `forge_token_env` is the name of a host-side env var
|
||||
carrying the Gitea API token; the value is read at provision time,
|
||||
never stored on the plan. `api_url` is the forge's HTTP API root; if
|
||||
empty, it is derived from the upstream URL's host at provision time."""
|
||||
|
||||
provider: str
|
||||
path: str = ""
|
||||
forge_token_env: str = ""
|
||||
api_url: str = ""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestGitEntry:
|
||||
"""One upstream the per-agent git-gate (PRD 0008) is allowed to
|
||||
talk to. `Upstream` is the real remote URL the agent would push to
|
||||
if there were no gate; the gate hosts a bare repo at /git/<Name>.git
|
||||
and `IdentityFile` is the SSH key the gate uses to push that repo
|
||||
upstream after gitleaks passes. The agent itself never holds the
|
||||
upstream credential.
|
||||
|
||||
The Upstream URL is parsed once at construction and the pieces are
|
||||
stashed in the `Upstream*` fields so the git-gate render step
|
||||
doesn't have to re-parse.
|
||||
|
||||
Manifest source: `git-gate.repos.<Name>` (PRD 0047/0048). A `key`
|
||||
block is required; `key.provider` is `"static"` or `"gitea"`. For
|
||||
`static`, `IdentityFile` is populated at parse time from `key.path`.
|
||||
For `gitea`, `IdentityFile` is populated at provision time."""
|
||||
|
||||
Name: str
|
||||
Upstream: str
|
||||
Key: ManifestKeyConfig = ManifestKeyConfig(provider="")
|
||||
IdentityFile: str = ""
|
||||
KnownHostKey: str = ""
|
||||
RemoteKey: str = ""
|
||||
UpstreamUser: str = ""
|
||||
UpstreamHost: str = ""
|
||||
UpstreamPort: str = ""
|
||||
UpstreamPath: str = ""
|
||||
|
||||
@classmethod
|
||||
def from_repos_entry(
|
||||
cls, bottle_name: str, repo_name: str, raw: object
|
||||
) -> "ManifestGitEntry":
|
||||
"""Parse one entry from `git-gate.repos.<repo_name>`.
|
||||
|
||||
YAML keys: `url` (required), `key` (required object with
|
||||
`provider`, and provider-specific fields), `host_key` (optional).
|
||||
The repo_name becomes `Name`."""
|
||||
if not repo_name:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.repos has an empty key"
|
||||
)
|
||||
if not _GIT_NAME_RE.match(repo_name):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.repos name {repo_name!r} is invalid; "
|
||||
f"allowed characters: A-Z a-z 0-9 . _ -"
|
||||
)
|
||||
label = f"git-gate.repos[{repo_name!r}]"
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' {label}")
|
||||
for k in d:
|
||||
if k not in {"url", "key", "host_key"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label} has unknown key {k!r}; "
|
||||
f"allowed: url, key, host_key"
|
||||
)
|
||||
upstream = d.get("url")
|
||||
if not isinstance(upstream, str) or not upstream:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label} missing required string field 'url'"
|
||||
)
|
||||
|
||||
if "key" not in d:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label} missing required 'key' block"
|
||||
)
|
||||
key_config = _parse_key_config(bottle_name, label, d["key"])
|
||||
|
||||
ident = key_config.path if key_config.provider == "static" else ""
|
||||
|
||||
khk = _opt_str(
|
||||
d.get("host_key"),
|
||||
f"bottle '{bottle_name}' {label} host_key",
|
||||
)
|
||||
user, host, port, path = parse_git_upstream(
|
||||
upstream, f"bottle '{bottle_name}' {label} url"
|
||||
)
|
||||
return cls(
|
||||
Name=repo_name,
|
||||
Upstream=upstream,
|
||||
Key=key_config,
|
||||
IdentityFile=ident,
|
||||
KnownHostKey=khk,
|
||||
RemoteKey=host,
|
||||
UpstreamUser=user,
|
||||
UpstreamHost=host,
|
||||
UpstreamPort=port,
|
||||
UpstreamPath=path,
|
||||
)
|
||||
|
||||
|
||||
def _parse_key_config(
|
||||
bottle_name: str, label: str, raw: object
|
||||
) -> ManifestKeyConfig:
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' {label}.key")
|
||||
provider = d.get("provider")
|
||||
if not isinstance(provider, str) or not provider:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key missing required "
|
||||
f"string field 'provider'"
|
||||
)
|
||||
if provider not in _KEY_PROVIDERS:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key provider {provider!r} is unknown; "
|
||||
f"allowed: {', '.join(sorted(_KEY_PROVIDERS))}"
|
||||
)
|
||||
|
||||
if provider == "gitea":
|
||||
for k in d:
|
||||
if k not in {"provider", "forge_token_env", "api_url"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key has unknown key {k!r} "
|
||||
f"for provider 'gitea'; allowed: provider, forge_token_env, api_url"
|
||||
)
|
||||
forge_token_env = d.get("forge_token_env")
|
||||
if not isinstance(forge_token_env, str) or not forge_token_env:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key missing required "
|
||||
f"string field 'forge_token_env' for provider 'gitea'"
|
||||
)
|
||||
api_url_raw = d.get("api_url", "")
|
||||
if not isinstance(api_url_raw, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key 'api_url' must be a string"
|
||||
)
|
||||
return ManifestKeyConfig(
|
||||
provider=provider,
|
||||
forge_token_env=forge_token_env,
|
||||
api_url=api_url_raw,
|
||||
)
|
||||
|
||||
# provider == "static"
|
||||
for k in d:
|
||||
if k not in {"provider", "path"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key has unknown key {k!r} "
|
||||
f"for provider 'static'; allowed: provider, path"
|
||||
)
|
||||
path = d.get("path")
|
||||
if not isinstance(path, str) or not path:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' {label}.key missing required "
|
||||
f"string field 'path' for provider 'static'"
|
||||
)
|
||||
return ManifestKeyConfig(provider=provider, path=path)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ManifestGitUser:
|
||||
"""Per-bottle `git config --global user.name` / `user.email`
|
||||
pair (issue #86). The agent's commits inside the bottle are
|
||||
attributed to this identity rather than the agent image's
|
||||
image-baked default (no user, or whatever the image dropped
|
||||
in). Either or both fields can be set independently.
|
||||
|
||||
`from_dict` is forgiving on shape (a single missing field is
|
||||
fine — we just skip that config line at provisioning) but
|
||||
strict on types (string-or-die)."""
|
||||
|
||||
name: str = ""
|
||||
email: str = ""
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, bottle_name: str, raw: object) -> "ManifestGitUser":
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' git-gate.user")
|
||||
for k in d:
|
||||
if k not in {"name", "email"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.user has unknown key {k!r}; "
|
||||
f"allowed: name, email"
|
||||
)
|
||||
name = d.get("name", "")
|
||||
email = d.get("email", "")
|
||||
if not isinstance(name, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.user.name must be a string "
|
||||
f"(was {type(name).__name__})"
|
||||
)
|
||||
if not isinstance(email, str):
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.user.email must be a string "
|
||||
f"(was {type(email).__name__})"
|
||||
)
|
||||
if not name and not email:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate.user is set but neither "
|
||||
f"name nor email is non-empty; remove the block or "
|
||||
f"fill at least one field."
|
||||
)
|
||||
return cls(name=name, email=email)
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
return not self.name and not self.email
|
||||
|
||||
|
||||
def parse_git_gate_config(
|
||||
bottle_name: str,
|
||||
raw: object,
|
||||
) -> tuple[tuple[ManifestGitEntry, ...], ManifestGitUser]:
|
||||
d = as_json_object(raw, f"bottle '{bottle_name}' git-gate")
|
||||
for k in d:
|
||||
if k not in {"user", "repos"}:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' git-gate has unknown key {k!r}; "
|
||||
f"allowed: user, repos"
|
||||
)
|
||||
|
||||
git_user = (
|
||||
ManifestGitUser.from_dict(bottle_name, d["user"])
|
||||
if "user" in d
|
||||
else ManifestGitUser()
|
||||
)
|
||||
|
||||
git: tuple[ManifestGitEntry, ...] = ()
|
||||
repos_raw = d.get("repos")
|
||||
if repos_raw is not None:
|
||||
repos = as_json_object(repos_raw, f"bottle '{bottle_name}' git-gate.repos")
|
||||
git = tuple(
|
||||
ManifestGitEntry.from_repos_entry(bottle_name, name, entry)
|
||||
for name, entry in repos.items()
|
||||
)
|
||||
validate_unique_git_names(bottle_name, git)
|
||||
|
||||
return git, git_user
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Internal per-file Markdown manifest loader."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from ..log import warn
|
||||
from .bottle import ManifestBottle
|
||||
from .extends import resolve_bottles
|
||||
from .schema import (
|
||||
entity_name_from_path,
|
||||
validate_bottle_frontmatter_keys,
|
||||
)
|
||||
from .util import ManifestError
|
||||
from ..yaml_subset import YamlSubsetError, parse_frontmatter
|
||||
|
||||
|
||||
def check_stale_json(dir_path: Path, md_dir: Path, label: str) -> None:
|
||||
"""Die if `<dir_path>/bot-bottle.json` exists but `md_dir` does
|
||||
not. The manifest format changed in PRD 0011 and we do not want
|
||||
to silently leave the JSON content unused."""
|
||||
legacy = dir_path / "bot-bottle.json"
|
||||
if legacy.is_file() and not md_dir.exists():
|
||||
raise ManifestError(
|
||||
f"found {legacy} but {md_dir} does not exist. The manifest "
|
||||
f"format changed in PRD 0011 — rewrite the JSON content "
|
||||
f"as per-file Markdown under {md_dir}/bottles/ and "
|
||||
f"{md_dir}/agents/. See README.md for the schema. "
|
||||
f"({label})"
|
||||
)
|
||||
|
||||
|
||||
def scan_bottle_names(bottles_dir: Path) -> list[str]:
|
||||
"""Scan `<bottles_dir>/*.md` for valid filenames and return sorted bottle names.
|
||||
|
||||
No file content is read. Invalid filenames are skipped with a warning."""
|
||||
result: list[str] = []
|
||||
if not bottles_dir.is_dir():
|
||||
return result
|
||||
for path in sorted(bottles_dir.glob("*.md")):
|
||||
name = entity_name_from_path(path)
|
||||
if name is None:
|
||||
warn(
|
||||
f"skipping {path}: filename must match "
|
||||
f"[a-z][a-z0-9-]*.md (got {path.name!r})"
|
||||
)
|
||||
continue
|
||||
result.append(name)
|
||||
return result
|
||||
|
||||
|
||||
def scan_agent_names(agents_dir: Path) -> dict[str, Path]:
|
||||
"""Scan `<agents_dir>/*.md` for valid filenames and return `{name: path}`.
|
||||
|
||||
No file content is read. Invalid filenames are skipped with a warning."""
|
||||
result: dict[str, Path] = {}
|
||||
if not agents_dir.is_dir():
|
||||
return result
|
||||
for path in sorted(agents_dir.glob("*.md")):
|
||||
name = entity_name_from_path(path)
|
||||
if name is None:
|
||||
warn(
|
||||
f"skipping {path}: filename must match "
|
||||
f"[a-z][a-z0-9-]*.md (got {path.name!r})"
|
||||
)
|
||||
continue
|
||||
result[name] = path
|
||||
return result
|
||||
|
||||
|
||||
def load_bottle_chain_from_dir(
|
||||
bottle_name: str, bottles_dir: Path
|
||||
) -> ManifestBottle:
|
||||
"""Load `bottle_name` and its full `extends:` chain from `bottles_dir`,
|
||||
returning the resolved ManifestBottle.
|
||||
|
||||
Only the files in the extends chain are read — unrelated bottle files
|
||||
are never touched. Raises ManifestError on parse or validation failure."""
|
||||
raws: dict[str, dict[str, object]] = {}
|
||||
to_load = [bottle_name]
|
||||
while to_load:
|
||||
name = to_load.pop()
|
||||
if name in raws:
|
||||
continue
|
||||
path = bottles_dir / f"{name}.md"
|
||||
if not path.is_file():
|
||||
avail = ", ".join(
|
||||
p.stem for p in sorted(bottles_dir.glob("*.md")) if p.is_file()
|
||||
) or "(none)"
|
||||
raise ManifestError(
|
||||
f"bottle '{name}' not found at {path}. "
|
||||
f"Available: {avail}"
|
||||
)
|
||||
try:
|
||||
fm, _body = parse_frontmatter(path.read_text())
|
||||
except OSError as e:
|
||||
raise ManifestError(f"could not read {path}: {e}") from e
|
||||
except YamlSubsetError as e:
|
||||
raise ManifestError(f"{path}: {e}") from e
|
||||
validate_bottle_frontmatter_keys(path, fm.keys())
|
||||
raws[name] = dict(fm)
|
||||
parent = fm.get("extends")
|
||||
if isinstance(parent, str):
|
||||
to_load.append(parent)
|
||||
elif isinstance(parent, list):
|
||||
to_load.extend(p for p in parent if isinstance(p, str))
|
||||
|
||||
return resolve_bottles(raws)[bottle_name]
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Internal manifest schema policy helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
# Filename-as-key uses kebab-case ASCII. The first character is a
|
||||
# letter so we don't conflict with hidden files / Markdown special
|
||||
# names (`.md`, `_template.md`, etc.). Filenames that fail this
|
||||
# pattern are skipped with a warning rather than crashing the load.
|
||||
_FILENAME_RX = re.compile(r"^[a-z][a-z0-9-]*$")
|
||||
|
||||
|
||||
# Frontmatter keys we accept on each entity. Anything not in these
|
||||
# sets dies with a "did you mean" pointer: typos should not silently
|
||||
# ghost into an empty config.
|
||||
BOTTLE_KEYS = frozenset(
|
||||
{
|
||||
"env", "extends", "agent_provider", "git-gate", "egress",
|
||||
"supervise", "nested_containers",
|
||||
}
|
||||
)
|
||||
AGENT_KEYS_REQUIRED: frozenset[str] = frozenset()
|
||||
AGENT_KEYS_OPTIONAL = frozenset({"bottle", "skills", "git-gate"})
|
||||
|
||||
# Claude Code subagent fields bot-bottle ignores at launch but does
|
||||
# not reject. This lets the same file double as
|
||||
# `~/.claude/agents/*.md` without modification.
|
||||
CLAUDE_CODE_AGENT_PASSTHROUGH_KEYS = frozenset({
|
||||
"name", "description", "model", "color", "memory",
|
||||
})
|
||||
AGENT_KEYS = (
|
||||
AGENT_KEYS_REQUIRED | AGENT_KEYS_OPTIONAL | CLAUDE_CODE_AGENT_PASSTHROUGH_KEYS
|
||||
)
|
||||
AGENT_MODEL_KEYS = AGENT_KEYS | frozenset({"prompt"})
|
||||
|
||||
|
||||
def is_valid_entity_name(name: str) -> bool:
|
||||
"""True if `name` fits the kebab-case `[a-z][a-z0-9-]*` convention
|
||||
shared by bottle/agent filenames and skill names. Names that satisfy
|
||||
this are also safe to interpolate into a host/guest path segment."""
|
||||
return bool(_FILENAME_RX.match(name))
|
||||
|
||||
|
||||
def entity_name_from_path(path: Path) -> str | None:
|
||||
"""Return the entity name implied by the filename, or None if the
|
||||
filename does not fit the [a-z][a-z0-9-]* convention."""
|
||||
if path.suffix != ".md":
|
||||
return None
|
||||
stem = path.stem
|
||||
if not is_valid_entity_name(stem):
|
||||
return None
|
||||
return stem
|
||||
|
||||
|
||||
def validate_bottle_frontmatter_keys(path: Path, keys: object) -> None:
|
||||
_validate_frontmatter_keys("bottle", path, keys, BOTTLE_KEYS)
|
||||
|
||||
|
||||
def validate_agent_frontmatter_keys(path: Path, keys: object) -> None:
|
||||
_validate_frontmatter_keys("agent", path, keys, AGENT_KEYS)
|
||||
|
||||
|
||||
def _validate_frontmatter_keys(
|
||||
kind: str,
|
||||
path: Path,
|
||||
keys: object,
|
||||
allowed_keys: frozenset[str],
|
||||
) -> None:
|
||||
from .util import ManifestError
|
||||
|
||||
key_set = set(keys) # type: ignore
|
||||
unknown = key_set - allowed_keys # type: ignore
|
||||
if unknown:
|
||||
allowed = ", ".join(sorted(allowed_keys))
|
||||
raise ManifestError(
|
||||
f"{kind} file {path}: unknown frontmatter key(s) "
|
||||
f"{sorted(unknown)}; allowed keys are {allowed}." # type: ignore
|
||||
)
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Shared manifest primitives used by all manifest sub-modules."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import cast
|
||||
|
||||
|
||||
class ManifestError(Exception):
|
||||
"""A manifest file (or the manifest tree) is invalid."""
|
||||
|
||||
|
||||
def as_json_object(value: object, label: str) -> dict[str, object]:
|
||||
"""Assert that `value` is a JSON object (str-keyed dict) and return
|
||||
a view typed as `dict[str, object]` so downstream `.get(...)` calls
|
||||
have a typed surface."""
|
||||
if not isinstance(value, dict):
|
||||
raise ManifestError(f"{label} must be a JSON object (was {type(value).__name__})")
|
||||
items = cast(dict[object, object], value)
|
||||
out: dict[str, object] = {}
|
||||
for k, v in items.items():
|
||||
if not isinstance(k, str):
|
||||
raise ManifestError(f"{label} keys must be strings (found {type(k).__name__})")
|
||||
out[k] = v
|
||||
return out
|
||||
Reference in New Issue
Block a user