Compare commits

..

3 Commits

Author SHA1 Message Date
didericis 873d75f852 fix: resolve pyright errors in egress.py
Lint and Type Check / lint (push) Failing after 7m2s
test / unit (pull_request) Successful in 39s
test / integration (pull_request) Failing after 49s
- Add explicit type annotations to _route_to_yaml_fields return type and fields dict
- Add type: ignore for path_allowlist iteration which has object type

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-06-03 23:27:10 -04:00
didericis 1bd676de06 fix: resolve pyright errors in yaml_subset.py
- Add explicit type annotation for cur list in _split_flow
- Add unreachable return statement after die() in _split_key_value
- Add type cast for parse_yaml_subset return value

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-06-03 23:26:39 -04:00
didericis 0bf1532557 fix: resolve pyright type errors
- Fix launch.py provision callable signature to accept Bottle not str
- Rename _prompt_path to prompt_path to make it public (not protected)
- Fix PromptMode type handling in bottle.py files
- Update WorkspaceSpec protocol to use read-only properties for compatibility with frozen BottleSpec
- Fix pty_resize signal handler type annotation
- Update local_registry.py contextmanager return type to Generator (not Iterator)

These changes fix ~130 pyright errors related to type safety.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-06-03 23:25:41 -04:00
9 changed files with 27 additions and 18 deletions
+2 -2
View File
@@ -23,7 +23,7 @@ class DockerBottle(Bottle):
): ):
self.name = container self.name = container
self._teardown = teardown self._teardown = teardown
self._prompt_path = prompt_path_in_container self.prompt_path = prompt_path_in_container
self._agent_prompt_mode = agent_prompt_mode self._agent_prompt_mode = agent_prompt_mode
self.agent_command = agent_command self.agent_command = agent_command
self.agent_provider_template = ( self.agent_provider_template = (
@@ -36,7 +36,7 @@ class DockerBottle(Bottle):
) -> list[str]: ) -> list[str]:
full_argv = list(argv) full_argv = list(argv)
full_argv.extend( full_argv.extend(
prompt_args(self._agent_prompt_mode, self._prompt_path, argv=full_argv) prompt_args(self._agent_prompt_mode, self.prompt_path, argv=full_argv)
) )
cmd = ["docker", "exec"] cmd = ["docker", "exec"]
if tty: if tty:
+2 -2
View File
@@ -80,7 +80,7 @@ _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
def launch( def launch(
plan: DockerBottlePlan, plan: DockerBottlePlan,
*, *,
provision: Callable[[DockerBottlePlan, str], str | None], provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
) -> Generator[DockerBottle, None, None]: ) -> Generator[DockerBottle, None, None]:
"""Build, launch, and provision a Docker bottle via compose. """Build, launch, and provision a Docker bottle via compose.
Teardown on exit.""" Teardown on exit."""
@@ -218,7 +218,7 @@ def launch(
agent_command=plan.agent_command, agent_command=plan.agent_command,
agent_prompt_mode=plan.agent_prompt_mode, agent_prompt_mode=plan.agent_prompt_mode,
) )
bottle._prompt_path = provision(plan, bottle) bottle.prompt_path = provision(plan, bottle)
# Step 9: yield. exec_agent continues to use `docker exec -it` # Step 9: yield. exec_agent continues to use `docker exec -it`
# — the agent runs `sleep infinity` per the renderer's # — the agent runs `sleep infinity` per the renderer's
+2 -2
View File
@@ -72,7 +72,7 @@ class SmolmachinesBottle(Bottle):
# In-VM path to the agent's prompt file. None when the # In-VM path to the agent's prompt file. None when the
# agent declared no prompt (file still exists; we just # agent declared no prompt (file still exists; we just
# don't pass --append-system-prompt-file). # don't pass --append-system-prompt-file).
self._prompt_path = prompt_path self.prompt_path = prompt_path
# Env vars the agent process needs (HTTPS_PROXY, # Env vars the agent process needs (HTTPS_PROXY,
# CLAUDE_CODE_OAUTH_TOKEN, manifest-declared bottle env, …). # CLAUDE_CODE_OAUTH_TOKEN, manifest-declared bottle env, …).
# Forwarded on every `smolvm machine exec` via `-e K=V` # Forwarded on every `smolvm machine exec` via `-e K=V`
@@ -93,7 +93,7 @@ class SmolmachinesBottle(Bottle):
agent_tail = ["env", *_env_assignments_for("node", self._guest_env), agent_tail = ["env", *_env_assignments_for("node", self._guest_env),
self.agent_command] self.agent_command]
provider_prompt_args = prompt_args( provider_prompt_args = prompt_args(
self._agent_prompt_mode, self._prompt_path, argv=argv, self._agent_prompt_mode, self.prompt_path, argv=argv,
) )
if self._agent_prompt_mode == "read_prompt_file": if self._agent_prompt_mode == "read_prompt_file":
agent_tail += argv agent_tail += argv
+2 -2
View File
@@ -89,7 +89,7 @@ _SUPERVISE_PORT = SUPERVISE_PORT
def launch( def launch(
plan: SmolmachinesBottlePlan, plan: SmolmachinesBottlePlan,
*, *,
provision: Callable[[SmolmachinesBottlePlan, str], str | None], provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
) -> Generator[SmolmachinesBottle, None, None]: ) -> Generator[SmolmachinesBottle, None, None]:
"""Build + run the bottle and yield a handle; tear everything """Build + run the bottle and yield a handle; tear everything
down on exit. Errors during bringup unwind any partial state down on exit. Errors during bringup unwind any partial state
@@ -120,7 +120,7 @@ def launch(
agent_command=plan.agent_command, agent_command=plan.agent_command,
agent_prompt_mode=plan.agent_prompt_mode, agent_prompt_mode=plan.agent_prompt_mode,
) )
bottle._prompt_path = provision(plan, bottle) bottle.prompt_path = provision(plan, bottle)
yield bottle yield bottle
finally: finally:
@@ -42,7 +42,7 @@ import time
import uuid import uuid
from contextlib import contextmanager from contextlib import contextmanager
from dataclasses import dataclass from dataclasses import dataclass
from typing import Iterator from typing import Generator, Iterator
from ...log import die from ...log import die
@@ -98,7 +98,7 @@ class RegistryHandle:
@contextmanager @contextmanager
def ephemeral_registry() -> Iterator[RegistryHandle]: def ephemeral_registry() -> Generator[RegistryHandle, None, None]:
"""Bring up a per-session docker network + a `registry:2.8.3` """Bring up a per-session docker network + a `registry:2.8.3`
container on it (published on a random host port), yield a container on it (published on a random host port), yield a
`RegistryHandle`, force-remove both on exit. `RegistryHandle`, force-remove both on exit.
@@ -123,7 +123,7 @@ def main(argv: list[str]) -> int:
machine = argv[0] machine = argv[0]
inner = argv[2:] inner = argv[2:]
def sync(*_args) -> None: def sync(*_args: int) -> None:
size = _read_winsize() size = _read_winsize()
if size is None: if size is None:
return return
+3 -3
View File
@@ -216,14 +216,14 @@ def egress_token_env_map(
return out return out
def _route_to_yaml_fields(r: Route) -> dict: def _route_to_yaml_fields(r: Route) -> dict[str, object]:
"""Return the addon-visible fields for one route. """Return the addon-visible fields for one route.
Single authoritative mapping between EgressRoute (host-side) and Single authoritative mapping between EgressRoute (host-side) and
egress_addon_core.Route (sidecar-side). When a field is added to egress_addon_core.Route (sidecar-side). When a field is added to
the addon's Route that must appear in the YAML, add it here and the addon's Route that must appear in the YAML, add it here and
in egress_addon_core._parse_one together.""" in egress_addon_core._parse_one together."""
fields: dict = {"host": r.host} fields: dict[str, object] = {"host": r.host}
if r.auth_scheme and r.token_env: if r.auth_scheme and r.token_env:
fields["auth_scheme"] = r.auth_scheme fields["auth_scheme"] = r.auth_scheme
fields["token_env"] = r.token_env fields["token_env"] = r.token_env
@@ -252,7 +252,7 @@ def egress_render_routes(
lines.append(f' token_env: "{f["token_env"]}"') lines.append(f' token_env: "{f["token_env"]}"')
if "path_allowlist" in f: if "path_allowlist" in f:
lines.append(" path_allowlist:") lines.append(" path_allowlist:")
for p in f["path_allowlist"]: for p in f["path_allowlist"]: # type: ignore
lines.append(f' - "{p}"') lines.append(f' - "{p}"')
return "\n".join(lines) + "\n" return "\n".join(lines) + "\n"
+9 -2
View File
@@ -13,8 +13,15 @@ DEFAULT_WORKSPACE_MODE = "755"
class WorkspaceSpec(Protocol): class WorkspaceSpec(Protocol):
copy_cwd: bool @property
user_cwd: str def copy_cwd(self) -> bool:
"""Whether to copy the current working directory."""
...
@property
def user_cwd(self) -> str:
"""The user's current working directory."""
...
@dataclass(frozen=True) @dataclass(frozen=True)
+4 -2
View File
@@ -58,6 +58,7 @@ from __future__ import annotations
import re import re
from dataclasses import dataclass from dataclasses import dataclass
from typing import cast
class YamlSubsetError(ValueError): class YamlSubsetError(ValueError):
@@ -283,7 +284,7 @@ def _split_flow(body: str, lineno: int, kind: str) -> list[str]:
depth_c = 0 depth_c = 0
in_single = False in_single = False
in_double = False in_double = False
cur = [] cur: list[str] = []
for ch in body: for ch in body:
if ch == "'" and not in_double: if ch == "'" and not in_double:
in_single = not in_single in_single = not in_single
@@ -330,6 +331,7 @@ def _split_key_value(content: str, lineno: int) -> tuple[str, str]:
if i + 1 >= len(content) or content[i + 1] in (" ", "\t"): if i + 1 >= len(content) or content[i + 1] in (" ", "\t"):
return content[:i].strip(), content[i + 1:].lstrip() return content[:i].strip(), content[i + 1:].lstrip()
die(f"yaml-subset: line {lineno} missing `: ` separator: {content!r}") die(f"yaml-subset: line {lineno} missing `: ` separator: {content!r}")
return "", "" # unreachable, but needed for type checker
def _parse_block( def _parse_block(
@@ -536,7 +538,7 @@ def parse_yaml_subset(text: str) -> dict[str, object]:
) )
if not isinstance(value, dict): if not isinstance(value, dict):
die("yaml-subset: top-level value must be a mapping") die("yaml-subset: top-level value must be a mapping")
return value return cast(dict[str, object], value)
def parse_frontmatter(text: str) -> tuple[dict[str, object], str]: def parse_frontmatter(text: str) -> tuple[dict[str, object], str]: