fix(types): make manifest.py clean under pyright strict
test / run tests/run_tests.py (pull_request) Successful in 14s

- log.die() typed NoReturn so pyright knows it terminates control flow
  (was returning the unreachable Die instance type).
- manifest.py: raw inputs typed object (not Any) and narrowed via a new
  _as_json_object helper that validates str keys and returns
  dict[str, object]. Eliminates the Unknown cascade through .get()
  calls under strict.
- _from_dict classmethods renamed to from_dict so cross-class
  construction (Bottle.from_dict from Manifest.from_json_obj, etc.)
  doesn't trip reportPrivateUsage.
- _SUPPORTED_RUNTIMES typed tuple[Runtime, ...] so the membership
  check narrows runtime_raw to Literal["runc", "runsc"] and the
  # type: ignore[assignment] is no longer needed.
- Bottle.env uses a typed _empty_str_dict factory; bare dict resolves
  to dict[Unknown, Unknown] under strict.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-10 21:34:03 -04:00
parent 1f36d53f7b
commit e9a3de49af
2 changed files with 97 additions and 72 deletions
+2 -1
View File
@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import sys import sys
from typing import NoReturn
def info(msg: str) -> None: def info(msg: str) -> None:
@@ -18,6 +19,6 @@ class Die(SystemExit):
fatal exit from an unrelated SystemExit.""" fatal exit from an unrelated SystemExit."""
def die(msg: str) -> "Die": def die(msg: str) -> NoReturn:
print(f"claude-bottle: error: {msg}", file=sys.stderr) print(f"claude-bottle: error: {msg}", file=sys.stderr)
raise Die(1) raise Die(1)
+95 -71
View File
@@ -33,13 +33,17 @@ import json
import os import os
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import Any, Literal, Mapping from typing import Literal, Mapping, cast
from .log import die from .log import die
_SUPPORTED_RUNTIMES: tuple[str, ...] = ("runc", "runsc")
Runtime = Literal["runc", "runsc"] Runtime = Literal["runc", "runsc"]
_SUPPORTED_RUNTIMES: tuple[Runtime, ...] = ("runc", "runsc")
def _empty_str_dict() -> dict[str, str]:
return {}
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -52,26 +56,22 @@ class SshEntry:
KnownHostKey: str = "" KnownHostKey: str = ""
@classmethod @classmethod
def _from_dict(cls, bottle_name: str, idx: int, raw: Any) -> "SshEntry": def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "SshEntry":
if not isinstance(raw, dict): d = _as_json_object(raw, f"bottle '{bottle_name}' ssh[{idx}]")
die( host = d.get("Host")
f"bottle '{bottle_name}' ssh[{idx}] must be a JSON object "
f"(was {_json_type(raw)})"
)
host = raw.get("Host")
if not isinstance(host, str) or not host: if not isinstance(host, str) or not host:
die(f"bottle '{bottle_name}' ssh[{idx}] missing required string field 'Host'") die(f"bottle '{bottle_name}' ssh[{idx}] missing required string field 'Host'")
ident = raw.get("IdentityFile") ident = d.get("IdentityFile")
if not isinstance(ident, str) or not ident: if not isinstance(ident, str) or not ident:
die( die(
f"bottle '{bottle_name}' ssh '{host}' missing required string field " f"bottle '{bottle_name}' ssh '{host}' missing required string field "
f"'IdentityFile'" f"'IdentityFile'"
) )
hostname = _opt_str(raw.get("Hostname"), f"bottle '{bottle_name}' ssh '{host}' Hostname") hostname = _opt_str(d.get("Hostname"), f"bottle '{bottle_name}' ssh '{host}' Hostname")
user = _opt_str(raw.get("User"), f"bottle '{bottle_name}' ssh '{host}' User") user = _opt_str(d.get("User"), f"bottle '{bottle_name}' ssh '{host}' User")
port = _opt_port(raw.get("Port"), f"bottle '{bottle_name}' ssh '{host}' Port") port = _opt_port(d.get("Port"), f"bottle '{bottle_name}' ssh '{host}' Port")
khk = _opt_str( khk = _opt_str(
raw.get("KnownHostKey"), d.get("KnownHostKey"),
f"bottle '{bottle_name}' ssh '{host}' KnownHostKey", f"bottle '{bottle_name}' ssh '{host}' KnownHostKey",
) )
return cls( return cls(
@@ -89,13 +89,9 @@ class BottleEgress:
allowlist: tuple[str, ...] = () allowlist: tuple[str, ...] = ()
@classmethod @classmethod
def _from_dict(cls, bottle_name: str, raw: Any) -> "BottleEgress": def from_dict(cls, bottle_name: str, raw: object) -> "BottleEgress":
if not isinstance(raw, dict): d = _as_json_object(raw, f"bottle '{bottle_name}' egress")
die( allow = d.get("allowlist")
f"bottle '{bottle_name}' egress must be a JSON object "
f"(was {_json_type(raw)})"
)
allow = raw.get("allowlist")
if allow is None: if allow is None:
return cls() return cls()
if not isinstance(allow, list): if not isinstance(allow, list):
@@ -103,33 +99,34 @@ class BottleEgress:
f"bottle '{bottle_name}' egress.allowlist must be an array " f"bottle '{bottle_name}' egress.allowlist must be an array "
f"(was {_json_type(allow)})" f"(was {_json_type(allow)})"
) )
for i, host in enumerate(allow): items: list[str] = []
allow_list = cast(list[object], allow)
for i, host in enumerate(allow_list):
if not isinstance(host, str): if not isinstance(host, str):
die( die(
f"bottle '{bottle_name}' egress.allowlist[{i}] must be a string " f"bottle '{bottle_name}' egress.allowlist[{i}] must be a string "
f"(was {_json_type(host)})" f"(was {_json_type(host)})"
) )
return cls(allowlist=tuple(allow)) items.append(host)
return cls(allowlist=tuple(items))
@dataclass(frozen=True) @dataclass(frozen=True)
class Bottle: class Bottle:
env: Mapping[str, str] = field(default_factory=dict) env: Mapping[str, str] = field(default_factory=_empty_str_dict)
ssh: tuple[SshEntry, ...] = () ssh: tuple[SshEntry, ...] = ()
egress: BottleEgress = field(default_factory=BottleEgress) egress: BottleEgress = field(default_factory=BottleEgress)
runtime: Runtime = "runc" runtime: Runtime = "runc"
@classmethod @classmethod
def _from_dict(cls, name: str, raw: Any) -> "Bottle": def from_dict(cls, name: str, raw: object) -> "Bottle":
if not isinstance(raw, dict): d = _as_json_object(raw, f"bottle '{name}'")
die(f"bottle '{name}' must be a JSON object (was {_json_type(raw)})")
env_raw = raw.get("env")
env: dict[str, str] = {} env: dict[str, str] = {}
env_raw = d.get("env")
if env_raw is not None: if env_raw is not None:
if not isinstance(env_raw, dict): env_dict = _as_json_object(env_raw, f"bottle '{name}' env")
die(f"bottle '{name}' env must be a JSON object (was {_json_type(env_raw)})") for var, value in env_dict.items():
for var, value in env_raw.items():
if not isinstance(value, str): if not isinstance(value, str):
die( die(
f"env entry {var} in bottle '{name}' must be a JSON string " f"env entry {var} in bottle '{name}' must be a JSON string "
@@ -137,25 +134,28 @@ class Bottle:
) )
env[var] = value env[var] = value
ssh_raw = raw.get("ssh")
ssh: tuple[SshEntry, ...] = () ssh: tuple[SshEntry, ...] = ()
ssh_raw = d.get("ssh")
if ssh_raw is not None: if ssh_raw is not None:
if not isinstance(ssh_raw, list): if not isinstance(ssh_raw, list):
die(f"bottle '{name}' ssh must be an array (was {_json_type(ssh_raw)})") die(f"bottle '{name}' ssh must be an array (was {_json_type(ssh_raw)})")
ssh_list = cast(list[object], ssh_raw)
ssh = tuple( ssh = tuple(
SshEntry._from_dict(name, i, entry) for i, entry in enumerate(ssh_raw) SshEntry.from_dict(name, i, entry)
for i, entry in enumerate(ssh_list)
) )
egress_raw = raw.get("egress") egress_raw = d.get("egress")
egress = ( egress = (
BottleEgress._from_dict(name, egress_raw) BottleEgress.from_dict(name, egress_raw)
if egress_raw is not None if egress_raw is not None
else BottleEgress() else BottleEgress()
) )
runtime_raw = raw.get("runtime") runtime_raw = d.get("runtime")
runtime: Runtime
if runtime_raw is None: if runtime_raw is None:
runtime: Runtime = "runc" runtime = "runc"
else: else:
if not isinstance(runtime_raw, str): if not isinstance(runtime_raw, str):
die(f"bottle '{name}' runtime must be a string (was {_json_type(runtime_raw)})") die(f"bottle '{name}' runtime must be a string (was {_json_type(runtime_raw)})")
@@ -164,7 +164,7 @@ class Bottle:
f"bottle '{name}' runtime '{runtime_raw}' is not supported. " f"bottle '{name}' runtime '{runtime_raw}' is not supported. "
f"Use one of: {', '.join(_SUPPORTED_RUNTIMES)}." f"Use one of: {', '.join(_SUPPORTED_RUNTIMES)}."
) )
runtime = runtime_raw # type: ignore[assignment] runtime = runtime_raw
return cls(env=env, ssh=ssh, egress=egress, runtime=runtime) return cls(env=env, ssh=ssh, egress=egress, runtime=runtime)
@@ -176,11 +176,10 @@ class Agent:
prompt: str = "" prompt: str = ""
@classmethod @classmethod
def _from_dict(cls, name: str, raw: Any, bottle_names: set[str]) -> "Agent": def from_dict(cls, name: str, raw: object, bottle_names: set[str]) -> "Agent":
if not isinstance(raw, dict): d = _as_json_object(raw, f"agent '{name}'")
die(f"agent '{name}' must be a JSON object (was {_json_type(raw)})")
bottle = raw.get("bottle") bottle = d.get("bottle")
if not isinstance(bottle, str) or not bottle: if not isinstance(bottle, str) or not bottle:
die(f"agent '{name}' must declare a 'bottle' field naming a defined bottle") die(f"agent '{name}' must declare a 'bottle' field naming a defined bottle")
if bottle not in bottle_names: if bottle not in bottle_names:
@@ -190,20 +189,23 @@ class Agent:
f"Available: {available}" f"Available: {available}"
) )
skills_raw = raw.get("skills")
skills: tuple[str, ...] = () skills: tuple[str, ...] = ()
skills_raw = d.get("skills")
if skills_raw is not None: if skills_raw is not None:
if not isinstance(skills_raw, list): if not isinstance(skills_raw, list):
die(f"agent '{name}' skills must be an array (was {_json_type(skills_raw)})") die(f"agent '{name}' skills must be an array (was {_json_type(skills_raw)})")
for i, skill in enumerate(skills_raw): collected: list[str] = []
skills_list = cast(list[object], skills_raw)
for i, skill in enumerate(skills_list):
if not isinstance(skill, str): if not isinstance(skill, str):
die( die(
f"agent '{name}' skills[{i}] must be a string " f"agent '{name}' skills[{i}] must be a string "
f"(was {_json_type(skill)})" f"(was {_json_type(skill)})"
) )
skills = tuple(skills_raw) collected.append(skill)
skills = tuple(collected)
prompt_raw = raw.get("prompt") prompt_raw = d.get("prompt")
if prompt_raw is None: if prompt_raw is None:
prompt = "" prompt = ""
elif isinstance(prompt_raw, str): elif isinstance(prompt_raw, str):
@@ -235,30 +237,32 @@ class Manifest:
if cwd_doc is None and home_doc is None: if cwd_doc is None and home_doc is None:
die(f"no claude-bottle.json found in {cwd} or {os.environ['HOME']}") die(f"no claude-bottle.json found in {cwd} or {os.environ['HOME']}")
h = home_doc or {} h: dict[str, object] = home_doc if home_doc is not None else {}
c = cwd_doc or {} c: dict[str, object] = cwd_doc if cwd_doc is not None else {}
merged: dict[str, Any] = { h_bottles = _section_dict(h.get("bottles"), "bottles")
"bottles": {**(h.get("bottles") or {}), **(c.get("bottles") or {})}, c_bottles = _section_dict(c.get("bottles"), "bottles")
"agents": {**(h.get("agents") or {}), **(c.get("agents") or {})}, h_agents = _section_dict(h.get("agents"), "agents")
c_agents = _section_dict(c.get("agents"), "agents")
merged: dict[str, object] = {
"bottles": {**h_bottles, **c_bottles},
"agents": {**h_agents, **c_agents},
} }
return cls.from_json_obj(merged) return cls.from_json_obj(merged)
@classmethod @classmethod
def from_json_obj(cls, obj: Any) -> "Manifest": def from_json_obj(cls, obj: object) -> "Manifest":
"""Validate and build a Manifest from a raw JSON-like dict.""" """Validate and build a Manifest from a raw JSON-like dict."""
if not isinstance(obj, dict): d = _as_json_object(obj, "manifest")
die(f"manifest must be a JSON object (was {_json_type(obj)})") raw_bottles = _section_dict(d.get("bottles"), "manifest 'bottles'")
raw_agents = _section_dict(d.get("agents"), "manifest 'agents'")
raw_bottles = obj.get("bottles") or {} bottles: dict[str, Bottle] = {
raw_agents = obj.get("agents") or {} n: Bottle.from_dict(n, b) for n, b in raw_bottles.items()
if not isinstance(raw_bottles, dict): }
die(f"manifest 'bottles' must be a JSON object (was {_json_type(raw_bottles)})")
if not isinstance(raw_agents, dict):
die(f"manifest 'agents' must be a JSON object (was {_json_type(raw_agents)})")
bottles = {n: Bottle._from_dict(n, b) for n, b in raw_bottles.items()}
bottle_names = set(bottles.keys()) bottle_names = set(bottles.keys())
agents = {n: Agent._from_dict(n, a, bottle_names) for n, a in raw_agents.items()} agents: dict[str, Agent] = {
n: Agent.from_dict(n, a, bottle_names) for n, a in raw_agents.items()
}
return cls(bottles=bottles, agents=agents) return cls(bottles=bottles, agents=agents)
def has_agent(self, name: str) -> bool: def has_agent(self, name: str) -> bool:
@@ -293,18 +297,38 @@ class Manifest:
return self.bottles[self.agents[agent_name].bottle] return self.bottles[self.agents[agent_name].bottle]
def _load_json_or_die(path: Path) -> dict[str, Any]: 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):
die(f"{label} must be a JSON object (was {_json_type(value)})")
items = cast(dict[object, object], value)
out: dict[str, object] = {}
for k, v in items.items():
if not isinstance(k, str):
die(f"{label} keys must be strings (found {_json_type(k)})")
out[k] = v
return out
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 _load_json_or_die(path: Path) -> dict[str, object]:
try: try:
with path.open() as f: with path.open() as f:
doc = json.load(f) doc: object = json.load(f)
except json.JSONDecodeError: except json.JSONDecodeError:
die(f"claude-bottle.json at {path} is not valid JSON") die(f"claude-bottle.json at {path} is not valid JSON")
if not isinstance(doc, dict): return _as_json_object(doc, f"claude-bottle.json at {path}")
die(f"claude-bottle.json at {path} must be a JSON object")
return doc
def _opt_str(value: Any, label: str) -> str: def _opt_str(value: object, label: str) -> str:
if value is None: if value is None:
return "" return ""
if not isinstance(value, str): if not isinstance(value, str):
@@ -312,7 +336,7 @@ def _opt_str(value: Any, label: str) -> str:
return value return value
def _opt_port(value: Any, label: str) -> str: def _opt_port(value: object, label: str) -> str:
"""Port accepts string or int (JSON-friendly) and is normalized to str.""" """Port accepts string or int (JSON-friendly) and is normalized to str."""
if value is None: if value is None:
return "" return ""
@@ -325,7 +349,7 @@ def _opt_port(value: Any, label: str) -> str:
die(f"{label} must be a string or number (was {_json_type(value)})") die(f"{label} must be a string or number (was {_json_type(value)})")
def _json_type(value: Any) -> str: def _json_type(value: object) -> str:
"""Mirror jq's type names for parity with the original bash error messages.""" """Mirror jq's type names for parity with the original bash error messages."""
if value is None: if value is None:
return "null" return "null"