bot_bottle/backend/docker/launch.py

@@ -92,7 +92,7 @@ def launch(
     def teardown() -> None:
         try:
             stack.close()
-        except BaseException as exc:
+        except BaseException as exc:  # noqa: W0718 — teardown must not fail
             warn(
                 f"teardown failed for container {plan.container_name}"
                 f" (compose-down): {exc!r}"

bot_bottle/backend/docker/pipelock_apply.py

@@ -99,7 +99,7 @@ def fetch_current_yaml(slug: str) -> str:
                 f"could not fetch pipelock.yaml from {container}: "
                 f"{(r.stderr or '').strip() or 'container not running?'}"
             )
-        return Path(tmp_path).read_text()
+        return Path(tmp_path).read_text(encoding="utf-8")
     finally:
         try:
             Path(tmp_path).unlink()

bot_bottle/backend/docker/prepare.py

@@ -219,7 +219,7 @@ def resolve_plan(
             else Path(__file__).resolve().parent.parent.parent.parent / "Dockerfile.claude"
         )
         dockerfile_content = (
-            supervise_dockerfile_path.read_text()
+            supervise_dockerfile_path.read_text(encoding="utf-8")
             if supervise_dockerfile_path.is_file()
             else ""
         )

bot_bottle/backend/smolmachines/launch.py

@@ -139,7 +139,7 @@ def _teardown_smolmachines(
     teardown_exc: BaseException | None = None
     try:
         stack.close()
-    except BaseException as exc:
+    except BaseException as exc:  # noqa: W0718 — teardown must not fail
         teardown_exc = exc
         warn(f"smolmachines teardown failed: {exc!r}")
     bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)

bot_bottle/backend/smolmachines/local_registry.py

@@ -208,7 +208,6 @@ def _host_port(name: str) -> int:
         return int(port_str)
     except ValueError:
         die(f"unexpected `docker port` output: {line!r}")
-        return -1  # unreachable; die() never returns
 
 
 def _wait_ready(port: int) -> None:

bot_bottle/backend/smolmachines/loopback_alias.py

@@ -176,11 +176,11 @@ def force_allowlist(machine_name: str, allowed_cidrs: list[str]) -> None:
         con.close()
 
 
-def allocate(slug: str) -> str:
+def allocate(_slug: str) -> str:
     """Pick the lowest-numbered alias from the pool not already
     in use by a running smolmachines bundle. Bails when the pool
     is exhausted — the caller should report the limit to the
-    operator. `slug` is logged for traceability; not otherwise
+    operator. `_slug` is logged for traceability; not otherwise
     used (no on-disk reservation, allocation is purely
     docker-state-driven).
 
@@ -195,7 +195,7 @@ def allocate(slug: str) -> str:
     if not _is_macos():
         return "127.0.0.1"
     _ALLOC_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
-    with open(_ALLOC_LOCK_PATH, "w") as lf:
+    with open(_ALLOC_LOCK_PATH, "w", encoding="utf-8") as lf:
         fcntl.flock(lf, fcntl.LOCK_EX)
         return _allocate_locked()
 
@@ -211,7 +211,6 @@ def _allocate_locked() -> str:
         f"Stop a running bottle (`smolvm machine ls --json`) or "
         f"raise _POOL_END in loopback_alias.py."
     )
-    return ""  # unreachable; die() never returns
 
 
 def _alias_present(ip: str) -> bool:

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -223,7 +223,6 @@ def bundle_host_port(
         f"no port mapping on {host_ip} for {container} "
         f"{container_port}/tcp; got: {(result.stdout or '').strip()!r}"
     )
-    return -1  # unreachable; die() never returns
 
 
 def stop_bundle(slug: str) -> None:

bot_bottle/cli/_common.py

@@ -14,7 +14,7 @@ REPO_DIR = str(Path(__file__).resolve().parent.parent.parent)
 def read_tty_line() -> str:
     """Mirror `IFS= read -r REPLY </dev/tty`. Falls back to stdin."""
     try:
-        with open("/dev/tty", "r") as tty:
+        with open("/dev/tty", "r", encoding="utf-8") as tty:
             return tty.readline().rstrip("\n")
     except OSError:
         return sys.stdin.readline().rstrip("\n")

bot_bottle/cli/supervise.py

@@ -263,7 +263,7 @@ def edit_in_editor(content: str, *, suffix: str = ".tmp") -> str | None:
         path = f.name
     try:
         subprocess.run([editor, path], check=False)
-        with open(path) as f:
+        with open(path, encoding="utf-8") as f:
             edited = f.read()
         return edited if edited != content else None
     finally:
@@ -296,7 +296,7 @@ def cmd_supervise(argv: list[str]) -> int:
         else:
             error("supervise exited on a fatal error (no detail captured).")
         return e.code if isinstance(e.code, int) else 1
-    except Exception as e:
+    except Exception as e:  # noqa: W0718 — catch supervise crash for logging
         log_path = _write_crash_log(e)
         error(f"supervise crashed: {type(e).__name__}: {e}")
         error(f"full traceback written to {log_path}")
@@ -439,7 +439,7 @@ def _render(
     selected: int,
     status_line: str,
     *,
-    green_attr: int = 0,
+    green_attr: int = 0,  # noqa: F841 — unused, but required by interface
 ) -> None:
     stdscr.erase()
     h, w = stdscr.getmaxyx()

bot_bottle/cli/tui.py

@@ -89,7 +89,7 @@ def _run_picker(items: list[str], *, title: str, tty_fd: int) -> Optional[str]:
             curses.nocbreak()
             curses.echo()
             curses.endwin()
-    except Exception:
+    except Exception:  # noqa: W0718 — curses can raise many error types
         return None
     finally:
         sys.__stdin__ = orig_stdin    # type: ignore[assignment]

bot_bottle/contrib/gitea/deploy_key_provisioner.py

@@ -117,5 +117,5 @@ def _split_owner_repo(owner_repo: str) -> tuple[str, str]:
 def _read_error_body(exc: urllib.error.HTTPError) -> str:
     try:
         return exc.read().decode("utf-8", errors="replace")
-    except Exception:
+    except Exception:  # noqa: broad-exception-caught — safely fallback to empty error message
         return ""

bot_bottle/env.py

@@ -89,7 +89,7 @@ def _read_secret_silent(name: str, prompt_body: str) -> str:
     if not (sys.stdin.isatty() or sys.stderr.isatty()):
         # Fall back to /dev/tty so this still works when stdin is a pipe.
         try:
-            tty = open("/dev/tty", "r+")
+            tty = open("/dev/tty", "r+", encoding="utf-8")
         except OSError:
             die(
                 f"cannot prompt for secret '{name}': no tty available. "

bot_bottle/git_http_backend.py

@@ -157,7 +157,7 @@ class GitHttpHandler(BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(body)
 
-    def log_message(self, format: str, *args: object) -> None:  # type: ignore
+    def log_message(self, format: str, *args: object) -> None:  # type: ignore  # noqa: A002
         sys.stdout.write(format % args + "\n")
         sys.stdout.flush()
 

bot_bottle/manifest_egress.py

@@ -93,7 +93,7 @@ class PipelockRoutePolicy:
                 raise ManifestError(
                     f"{label}.ssrf_ip_allowlist[{j}] must be an IP address "
                     f"or CIDR (was {item!r}): {e}"
-                )
+                ) from e
             ssrf_ip_allowlist.append(item)
         return cls(
             TlsPassthrough=tls_passthrough_raw,

bot_bottle/manifest_loader.py

@@ -54,9 +54,9 @@ def load_bottles_from_dir(bottles_dir: Path) -> dict[str, Bottle]:
         try:
             fm, _body = parse_frontmatter(path.read_text())
         except OSError as e:
-            raise ManifestError(f"could not read {path}: {e}")
+            raise ManifestError(f"could not read {path}: {e}") from e
         except YamlSubsetError as e:
-            raise ManifestError(f"{path}: {e}")
+            raise ManifestError(f"{path}: {e}") from e
         validate_bottle_frontmatter_keys(path, fm.keys())
         raws[name] = fm
     return resolve_bottles(raws)
@@ -66,7 +66,7 @@ def load_agents_from_dir(
     agents_dir: Path,
     bottle_names: set[str],
     *,
-    source: str,
+    source: str,  # noqa: F841 — unused, but required by interface
 ) -> dict[str, Agent]:
     """Walk `<agents_dir>/*.md`, parse each as an agent, and return
     `{name: Agent}`. The Markdown body becomes the agent's prompt.
@@ -87,9 +87,9 @@ def load_agents_from_dir(
         try:
             fm, body = parse_frontmatter(path.read_text())
         except OSError as e:
-            raise ManifestError(f"could not read {path}: {e}")
+            raise ManifestError(f"could not read {path}: {e}") from e
         except YamlSubsetError as e:
-            raise ManifestError(f"{path}: {e}")
+            raise ManifestError(f"{path}: {e}") from e
         validate_agent_frontmatter_keys(path, fm.keys())
         # Build the dict Agent.from_dict expects. The body becomes
         # prompt; Claude Code passthrough fields stay in fm and get

bot_bottle/supervise.py

@@ -519,22 +519,22 @@ def _atomic_write(path: Path, content: str, *, mode: int) -> None:
 try:
     import fcntl as _fcntl
 
-    def _try_flock(fd: int) -> None:
+    def _try_flock(fd: int) -> None:  # type: ignore[reportRedeclaration]
         try:
             _fcntl.flock(fd, _fcntl.LOCK_EX)
         except OSError:
             pass
 
-    def _try_funlock(fd: int) -> None:
+    def _try_funlock(fd: int) -> None:  # type: ignore[reportRedeclaration]
         try:
             _fcntl.flock(fd, _fcntl.LOCK_UN)
         except OSError:
             pass
 except ImportError:  # pragma: no cover — Windows path
-    def _try_flock(fd: int) -> None:
+    def _try_flock(fd: int) -> None:  # noqa: F841 — Windows fallback
         return None
 
-    def _try_funlock(fd: int) -> None:
+    def _try_funlock(fd: int) -> None:  # noqa: F841 — Windows fallback
         return None
 
 

bot_bottle/supervise_server.py

@@ -590,7 +590,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
 
     server_version = f"{SERVER_NAME}/{SERVER_VERSION}"
 
-    def log_message(self, format: str, *args: typing.Any) -> None:
+    def log_message(self, format: str, *args: typing.Any) -> None:  # noqa: A002
         if os.environ.get("SUPERVISE_DEBUG"):
             super().log_message(format, *args)
 
@@ -630,7 +630,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
         except _RpcError as e:
             self._write_jsonrpc(jsonrpc_error(req.id, e.code, e.message))
             return
-        except Exception as e:  # pragma: no cover — defensive
+        except Exception as e:  # noqa: W0718 — catch-all for RPC dispatch errors
             sys.stderr.write(f"supervise: internal error: {e}\n")
             self._write_jsonrpc(jsonrpc_error(req.id, ERR_INTERNAL, "internal error"))
             return