fix: resolve pyright errors in manifest_schema.py

- Add type: ignore annotations for dict key validation
- Keys parameter is untyped object from YAML parsing
- Use type: ignore for set operations and sorted calls
- Fixes 4 pyright errors

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

bot_bottle/backend/docker/bottle_state.py

@@ -35,6 +35,7 @@ import secrets
 import string
 from dataclasses import dataclass
 from pathlib import Path
+from typing import cast
 
 from ... import supervise as _supervise
 from . import util as docker_mod
@@ -135,14 +136,15 @@ def read_metadata(identity: str) -> BottleMetadata | None:
     raw = json.loads(path.read_text())
     if not isinstance(raw, dict):
         return None
+    raw_typed = cast(dict[str, object], raw)
     return BottleMetadata(
-        identity=str(raw.get("identity", identity)),
+        identity=str(raw_typed.get("identity", identity)),
-        agent_name=str(raw.get("agent_name", "")),
+        agent_name=str(raw_typed.get("agent_name", "")),
-        cwd=str(raw.get("cwd", "")),
+        cwd=str(raw_typed.get("cwd", "")),
-        copy_cwd=bool(raw.get("copy_cwd", False)),
+        copy_cwd=bool(raw_typed.get("copy_cwd", False)),
-        started_at=str(raw.get("started_at", "")),
+        started_at=str(raw_typed.get("started_at", "")),
-        compose_project=str(raw.get("compose_project", "")),
+        compose_project=str(raw_typed.get("compose_project", "")),
-        backend=str(raw.get("backend", "")),
+        backend=str(raw_typed.get("backend", "")),
     )
 
 

bot_bottle/backend/docker/egress_apply.py

@@ -26,6 +26,7 @@ import json
 import re
 import subprocess
 from pathlib import Path
+from typing import cast
 
 from ...egress import EGRESS_ROUTES_IN_CONTAINER
 from ...egress_addon_core import load_routes
@@ -57,7 +58,8 @@ def _render_routes_payload(routes_list: list[dict[str, object]]) -> str:
         if auth_scheme and token_env:
             lines.append(f'    auth_scheme: "{auth_scheme}"')
             lines.append(f'    token_env: "{token_env}"')
-        paths = entry.get("path_allowlist") or []
+        paths_obj = entry.get("path_allowlist")
+        paths = cast(list[str], paths_obj) if isinstance(paths_obj, list) else []
         if paths:
             lines.append("    path_allowlist:")
             for p in paths:
@@ -257,6 +259,7 @@ def _merge_single_route(
         raise EgressApplyError(
             "current routes.yaml: 'routes' is not a list"
         )
+    routes_typed = cast(list[object], routes)
 
     new_host = str(new_route.get("host", "")).lower()
     if not new_host:
@@ -264,22 +267,25 @@ def _merge_single_route(
             "proposed route is missing 'host'"
         )
 
-    proposed_paths = list(new_route.get("path_allowlist") or [])
+    proposed_paths_obj = new_route.get("path_allowlist")
+    proposed_paths = cast(list[str], proposed_paths_obj) if isinstance(proposed_paths_obj, list) else []
 
     # Look for an existing entry with the same host (case-insensitive).
-    for entry in routes:
+    for entry in routes_typed:
         if not isinstance(entry, dict):
             continue
-        if str(entry.get("host", "")).lower() == new_host:
+        entry_typed = cast(dict[str, object], entry)
+        if str(entry_typed.get("host", "")).lower() == new_host:
             # Merge path_allowlist: union proposed + existing, ordered
             # by first-seen so existing paths stay in original order.
-            existing_paths: list[str] = list(entry.get("path_allowlist") or [])
+            existing_paths_obj = entry_typed.get("path_allowlist")
+            existing_paths = cast(list[str], existing_paths_obj) if isinstance(existing_paths_obj, list) else []
             seen = {p: None for p in existing_paths}
             for p in proposed_paths:
                 seen.setdefault(p, None)
             merged_paths = list(seen.keys())
             if merged_paths:
-                entry["path_allowlist"] = merged_paths
+                entry_typed["path_allowlist"] = merged_paths
             # Preserve existing auth — tool description says agent-
             # proposed auth on an existing host is ignored.
             break
@@ -289,19 +295,22 @@ def _merge_single_route(
         # `auth` was proposed (otherwise the addon's parser rejects
         # a half-set auth pair). Slots: count existing slots, pick
         # the next free index.
-        entry = {"host": new_route["host"]}
+        entry_typed: dict[str, object] = {"host": new_route.get("host")}  # type: ignore
         if proposed_paths:
-            entry["path_allowlist"] = proposed_paths
+            entry_typed["path_allowlist"] = proposed_paths
         auth = new_route.get("auth")
-        if isinstance(auth, dict) and auth.get("scheme") and auth.get("token_ref"):
+        if isinstance(auth, dict) and auth.get("scheme") and auth.get("token_ref"):  # type: ignore
+            auth_typed = cast(dict[str, object], auth)
             existing_slots = sorted({
-                str(r.get("token_env"))
+                str(r_entry.get("token_env", ""))
-                for r in routes
+                for r_entry_obj in routes_typed
-                if isinstance(r, dict) and r.get("token_env")
+                if isinstance(r_entry_obj, dict)
+                for r_entry in [cast(dict[str, object], r_entry_obj)]
+                if r_entry.get("token_env")
             })
             next_idx = len(existing_slots)
-            entry["auth_scheme"] = str(auth["scheme"])
+            entry_typed["auth_scheme"] = str(cast(object, auth_typed.get("scheme")))
-            entry["token_env"] = f"EGRESS_TOKEN_{next_idx}"
+            entry_typed["token_env"] = f"EGRESS_TOKEN_{next_idx}"
             # NOTE: the addon reads token VALUES from its container's
             # environ keyed by token_env. A newly-added auth route at
             # runtime points at a slot that has no env value → the
@@ -309,9 +318,9 @@ def _merge_single_route(
             # arranges for the value to land in the container's env.
             # Recording this here so the operator-facing diff carries
             # the slot name they'll need to provision.
-        routes.append(entry)
+        routes_typed.append(entry_typed)
 
-    return _render_routes_payload(routes)
+    return _render_routes_payload(cast(list[dict[str, object]], routes_typed))
 
 
 def add_route(slug: str, proposed_route_json: str) -> tuple[str, str]:

bot_bottle/codex_auth.py

@@ -13,6 +13,7 @@ import os
 from copy import deepcopy
 from datetime import datetime, timezone
 from pathlib import Path
+from typing import cast
 
 from .log import die
 from .util import expand_tilde
@@ -50,7 +51,8 @@ def codex_host_access_token(
     tokens = raw.get("tokens")
     if not isinstance(tokens, dict):
         die(f"codex host credentials: {path} is missing tokens")
-    access = tokens.get("access_token")
+    tokens_typed = cast(dict[str, object], tokens)
+    access = tokens_typed.get("access_token")
     if not isinstance(access, str) or not access:
         die(
             f"codex host credentials: {path} is missing tokens.access_token. "
@@ -105,14 +107,14 @@ def write_codex_dummy_auth_file(
     path.chmod(0o600)
 
 
-def _read_auth_object(path: Path) -> dict:
+def _read_auth_object(path: Path) -> dict[str, object]:
     try:
         raw = json.loads(path.read_text())
     except (OSError, json.JSONDecodeError) as e:
         die(f"codex host credentials: could not read valid JSON at {path}: {e}")
     if not isinstance(raw, dict):
         die(f"codex host credentials: {path} must contain a JSON object")
-    return raw
+    return cast(dict[str, object], raw)
 
 
 def _dummy_exp(now: datetime | None, exp_ts: int | None) -> int:
@@ -151,11 +153,11 @@ def _dummy_jwt_from_host(
         return _dummy_jwt(now, exp_ts=exp_ts)
     if not isinstance(payload, dict):
         return _dummy_jwt(now, exp_ts=exp_ts)
-    return _encode_dummy_jwt(_redact_jwt_payload(payload, now=now, exp_ts=exp_ts))
+    return _encode_dummy_jwt(_redact_jwt_payload(cast(dict[str, object], payload), now=now, exp_ts=exp_ts))
 
 
-def _encode_dummy_jwt(payload: dict) -> str:
+def _encode_dummy_jwt(payload: dict[str, object]) -> str:
-    def enc(obj: dict) -> str:
+    def enc(obj: dict[str, object]) -> str:
         raw = json.dumps(obj, separators=(",", ":")).encode()
         return base64.urlsafe_b64encode(raw).decode().rstrip("=")
 
@@ -163,23 +165,24 @@ def _encode_dummy_jwt(payload: dict) -> str:
 
 
 def _redact_jwt_payload(
-    payload: dict,
+    payload: dict[str, object],
     *,
     now: datetime | None = None,
     exp_ts: int | None = None,
-) -> dict:
+) -> dict[str, object]:
     out = _redact_claims(payload)
     if not isinstance(out, dict):
         out = {}
-    out["exp"] = _dummy_exp(now, exp_ts)
+    out_typed: dict[str, object] = cast(dict[str, object], out)
-    out.setdefault("sub", "bot-bottle-placeholder")
+    out_typed["exp"] = _dummy_exp(now, exp_ts)
-    return out
+    out_typed.setdefault("sub", "bot-bottle-placeholder")
+    return out_typed
 
 
 def _redact_claims(value: object) -> object:
     if isinstance(value, dict):
         out: dict[str, object] = {}
-        for key, inner in value.items():
+        for key, inner in cast(dict[str, object], value).items():
             lower = key.lower()
             if key == "https://api.openai.com/profile":
                 out[key] = _redact_profile_claim(inner)
@@ -207,16 +210,16 @@ def _redact_claims(value: object) -> object:
     return "bot-bottle-placeholder"
 
 
-def _redact_profile_claim(value: object) -> dict:
+def _redact_profile_claim(value: object) -> dict[str, object]:
-    profile = value if isinstance(value, dict) else {}
+    profile = cast(dict[str, object], value) if isinstance(value, dict) else {}
     return {
         "email": "bot-bottle@example.invalid",
         "email_verified": bool(profile.get("email_verified", True)),
     }
 
 
-def _redact_auth_claim(value: object) -> dict:
+def _redact_auth_claim(value: object) -> dict[str, object]:
-    auth = value if isinstance(value, dict) else {}
+    auth = cast(dict[str, object], value) if isinstance(value, dict) else {}
     out: dict[str, object] = {}
     for key, inner in auth.items():
         lower = key.lower()
@@ -247,7 +250,7 @@ def _redact_auth_claim(value: object) -> dict:
 def _redact_codex_auth(
     value: object, *, now: datetime | None = None, exp_ts: int | None = None,
 ) -> object:
-    auth = value if isinstance(value, dict) else {}
+    auth = cast(dict[str, object], value) if isinstance(value, dict) else {}
     out: dict[str, object] = {}
     for key, inner in auth.items():
         lower = key.lower()
@@ -269,7 +272,7 @@ def _redact_codex_auth(
 def _redact_token_block(
     value: object, *, now: datetime | None = None, exp_ts: int | None = None,
 ) -> dict[str, object]:
-    tokens = value if isinstance(value, dict) else {}
+    tokens = cast(dict[str, object], value) if isinstance(value, dict) else {}
     out: dict[str, object] = {}
     for key, inner in tokens.items():
         lower = key.lower()
@@ -306,7 +309,7 @@ def _jwt_exp(token: str) -> datetime | None:
         return None
     if not isinstance(payload, dict):
         return None
-    exp = payload.get("exp")
+    exp = cast(dict[str, object], payload).get("exp")
     if not isinstance(exp, (int, float)):
         return None
     return datetime.fromtimestamp(exp, timezone.utc)

bot_bottle/manifest_schema.py

@@ -60,11 +60,11 @@ def _validate_frontmatter_keys(
 ) -> None:
     from .manifest_util import ManifestError
 
-    key_set = set(keys)
+    key_set = set(keys)  # type: ignore
-    unknown = key_set - allowed_keys
+    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}."
+            f"{sorted(unknown)}; allowed keys are {allowed}."  # type: ignore
         )

bot_bottle/pipelock.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from pathlib import Path
+from typing import cast
 
 from .egress import EgressRoute, egress_routes_for_bottle
 from .supervise import SUPERVISE_HOSTNAME
@@ -259,7 +260,7 @@ def _required_dict(
     value = obj.get(key)
     if not isinstance(value, dict):
         raise _pipelock_render_error(section, key, "a mapping")
-    return value
+    return cast(dict[str, object], value)
 
 
 def _required_bool(obj: dict[str, object], section: str, key: str) -> bool:
@@ -289,9 +290,12 @@ def _required_str_list(
     key: str,
 ) -> list[str]:
     value = obj.get(key)
-    if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
+    if not isinstance(value, list):
         raise _pipelock_render_error(section, key, "a list of strings")
-    return value
+    value_list = cast(list[object], value)
+    if not all(isinstance(v, str) for v in value_list):
+        raise _pipelock_render_error(section, key, "a list of strings")
+    return cast(list[str], value)
 
 
 def _optional_str_list(
@@ -407,49 +411,42 @@ def pipelock_render_yaml(cfg: dict[str, object]) -> str:
     lines: list[str] = []
     lines.append(f"version: {cfg['version']}")
     lines.append(f"mode: {cfg['mode']}")
-    lines.append(f"enforce: {_bool(cfg['enforce'])}")
+    lines.append(f"enforce: {_bool(cast(bool, cfg['enforce']))}")
     lines.append("")
     lines.append("api_allowlist:")
-    api_allowlist = cfg["api_allowlist"]
+    api_allowlist = cast(list[str], cfg["api_allowlist"])
-    assert isinstance(api_allowlist, list)
     for h in api_allowlist:
         lines.append(f'  - "{h}"')
     lines.append("")
     if "seed_phrase_detection" in cfg:
         lines.append("seed_phrase_detection:")
-        spd = cfg["seed_phrase_detection"]
+        spd = cast(dict[str, object], cfg["seed_phrase_detection"])
-        assert isinstance(spd, dict)
+        lines.append(f"  enabled: {_bool(cast(bool, spd['enabled']))}")
-        lines.append(f"  enabled: {_bool(spd['enabled'])}")
         lines.append("")
     lines.append("forward_proxy:")
-    fp = cfg["forward_proxy"]
+    fp = cast(dict[str, object], cfg["forward_proxy"])
-    assert isinstance(fp, dict)
+    lines.append(f"  enabled: {_bool(cast(bool, fp['enabled']))}")
-    lines.append(f"  enabled: {_bool(fp['enabled'])}")
     lines.append("")
     lines.append("dlp:")
-    dlp = cfg["dlp"]
+    dlp = cast(dict[str, object], cfg["dlp"])
-    assert isinstance(dlp, dict)
+    lines.append(f"  include_defaults: {_bool(cast(bool, dlp['include_defaults']))}")
-    lines.append(f"  include_defaults: {_bool(dlp['include_defaults'])}")
+    lines.append(f"  scan_env: {_bool(cast(bool, dlp['scan_env']))}")
-    lines.append(f"  scan_env: {_bool(dlp['scan_env'])}")
     lines.append("")
     lines.append("request_body_scanning:")
-    rbs = cfg["request_body_scanning"]
+    rbs = cast(dict[str, object], cfg["request_body_scanning"])
-    assert isinstance(rbs, dict)
+    lines.append(f'  action: "{cast(str, rbs["action"])}"')
-    lines.append(f'  action: "{rbs["action"]}"')
     if "scan_headers" in rbs:
-        lines.append(f"  scan_headers: {_bool(rbs['scan_headers'])}")
+        lines.append(f"  scan_headers: {_bool(cast(bool, rbs['scan_headers']))}")
     if "header_mode" in rbs:
-        lines.append(f'  header_mode: "{rbs["header_mode"]}"')
+        lines.append(f'  header_mode: "{cast(str, rbs["header_mode"])}"')
     if "tls_interception" in cfg:
         lines.append("")
         lines.append("tls_interception:")
-        tls = cfg["tls_interception"]
+        tls = cast(dict[str, object], cfg["tls_interception"])
-        assert isinstance(tls, dict)
+        lines.append(f"  enabled: {_bool(cast(bool, tls['enabled']))}")
-        lines.append(f"  enabled: {_bool(tls['enabled'])}")
+        lines.append(f'  ca_cert: "{cast(str, tls["ca_cert"])}"')
-        lines.append(f'  ca_cert: "{tls["ca_cert"]}"')
+        lines.append(f'  ca_key: "{cast(str, tls["ca_key"])}"')
-        lines.append(f'  ca_key: "{tls["ca_key"]}"')
+        passthrough = cast(list[str], tls["passthrough_domains"])
-        passthrough = tls["passthrough_domains"]
-        assert isinstance(passthrough, list)
         if passthrough:
             lines.append("  passthrough_domains:")
             for d in passthrough:
@@ -457,11 +454,9 @@ def pipelock_render_yaml(cfg: dict[str, object]) -> str:
     if "ssrf" in cfg:
         lines.append("")
         lines.append("ssrf:")
-        ssrf = cfg["ssrf"]
+        ssrf = cast(dict[str, object], cfg["ssrf"])
-        assert isinstance(ssrf, dict)
         lines.append("  ip_allowlist:")
-        ip_allowlist = ssrf["ip_allowlist"]
+        ip_allowlist = cast(list[str], ssrf["ip_allowlist"])
-        assert isinstance(ip_allowlist, list)
         for ip in ip_allowlist:
             lines.append(f'    - "{ip}"')
     return "\n".join(lines) + "\n"