refactor: make CI lint, format, type checks, and runtime tests pass

.claude/skills/porting-to-canyonos/prepare.py

@@ -195,7 +195,9 @@ def _refresh_app(
     if conflicts:
         conflicts = sorted(set(conflicts))
         shown = "\n  ".join(conflicts[:20])
-        suffix = "" if len(conflicts) <= 20 else f"\n  ... and {len(conflicts) - 20} more"
+        suffix = (
+            "" if len(conflicts) <= 20 else f"\n  ... and {len(conflicts) - 20} more"
+        )
         raise ValueError(
             "refresh found files changed in both the source and .car/app:\n  "
             f"{shown}{suffix}\nResolve them in .car/app, then update the source "

.github/workflows/ci.yml

@@ -24,10 +24,10 @@ jobs:
           enable-cache: true
 
       - name: "Run Ruff: Lint"
-        run: uvx ruff check .
+        run: uvx ruff@0.16.7 check .
 
       - name: "Run Ruff: Format Check"
-        run: uvx ruff format --check .
+        run: uvx ruff@0.16.7 format --check .
 
       - name: "Install dependencies for Ty"
         run: |
@@ -38,5 +38,5 @@ jobs:
         run: uvx ty check
 
 # If you want to run this locally, install act and run "act pull_request"
-# For fixing Ruff lint errors: uvx ruff check --fix .
-# For fixing Ruff formatting errors: uvx ruff format .
+# For fixing Ruff lint errors: uvx ruff@0.16.7 check --fix .
+# For fixing Ruff formatting errors: uvx ruff@0.16.7 format .

canyonos_core/OTLP_Exporter/convert.py

@@ -4,7 +4,12 @@
 conversion.
 """
 
-from opentelemetry.sdk.trace import EXCEPTION_MESSAGE, EXCEPTION_TYPE, Event, ReadableSpan
+from opentelemetry.sdk.trace import (
+    EXCEPTION_MESSAGE,
+    EXCEPTION_TYPE,
+    Event,
+    ReadableSpan,
+)
 from opentelemetry.trace import SpanContext, SpanKind, TraceFlags
 from opentelemetry.trace.status import Status, StatusCode
 
@@ -41,7 +46,10 @@ def waiting_row_to_span(row):
     )
     parent = (
         SpanContext(
-            trace_id=trace_id, span_id=parent_span_id, is_remote=False, trace_flags=_SAMPLED
+            trace_id=trace_id,
+            span_id=parent_span_id,
+            is_remote=False,
+            trace_flags=_SAMPLED,
         )
         if parent_span_id
         else None

canyonos_core/OTLP_Exporter/db.py

@@ -5,7 +5,7 @@
 import os
 import sqlite3
 
-from canyonos_core.controller.utils import pricing 
+from canyonos_core.controller.utils import pricing
 # Will need to eventually delete dependency on this and move to OTLP
 # It is currently stored here for backcompat with the old telemetry collecting
 
@@ -32,6 +32,7 @@ def _log_cost_failure(kind, exc):
         exc_info=True,
     )
 
+
 # Demo-only multipliers for scaling displayed costs, DELETE FOR MORE ACCURATE METRICS
 _TOKEN_COST_MULTIPLIER = 10000
 _SERVER_COST_MULTIPLIER = 100000
@@ -68,6 +69,7 @@ def _log_cost_failure(kind, exc):
     sent BOOLEAN DEFAULT 0
 """
 
+
 def init_db(db_path=DB_PATH):
     """Create the waiting table if it doesn't already exist."""
     conn = sqlite3.connect(db_path)
@@ -81,12 +83,33 @@ def init_db(db_path=DB_PATH):
 # `sent` is deliberately excluded here so re-upserting a waiting row (e.g. GC
 # re-writing it from Redis) never resets it back to unsent.
 _COLUMNS = [
-    "future_id", "parent_id", "session_id", "project_id", "agent_id", "model",
-    "cpu", "gpu", "started_at", "finished_at", "execution_time_ms", "queue_time_ms",
-    "input_token_count", "output_token_count", "token_count", "errors",
-    "failed", "server_cost", "token_cost", "total_cost",
-    "cached_tokens", "cache_hit_ratio", "error_name", "error_message",
-    "name", "input", "output",
+    "future_id",
+    "parent_id",
+    "session_id",
+    "project_id",
+    "agent_id",
+    "model",
+    "cpu",
+    "gpu",
+    "started_at",
+    "finished_at",
+    "execution_time_ms",
+    "queue_time_ms",
+    "input_token_count",
+    "output_token_count",
+    "token_count",
+    "errors",
+    "failed",
+    "server_cost",
+    "token_cost",
+    "total_cost",
+    "cached_tokens",
+    "cache_hit_ratio",
+    "error_name",
+    "error_message",
+    "name",
+    "input",
+    "output",
 ]
 
 _WAITING_UPSERT = """
@@ -126,7 +149,10 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH
             if not fid or not session_id:
                 missing = ", ".join(
                     field
-                    for field, present in (("future_id", fid), ("request_id", session_id))
+                    for field, present in (
+                        ("future_id", fid),
+                        ("request_id", session_id),
+                    )
                     if not present
                 )
                 logger.warning(
@@ -217,7 +243,9 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH
                     "token_cost": token_cost,
                     "total_cost": server_cost + token_cost,
                     "cached_tokens": cached_tokens,
-                    "cache_hit_ratio": cached_tokens / token_count if token_count else 0.0,
+                    "cache_hit_ratio": cached_tokens / token_count
+                    if token_count
+                    else 0.0,
                     "error_name": raw.get("error_name"),
                     "error_message": raw.get("error") or raw.get("error_message"),
                     "name": name or agent_id or "unknown_agent",

canyonos_core/OTLP_Exporter/otel_exporter.py

@@ -8,6 +8,7 @@
 import sqlite3
 import sys
 import time
+from typing import Literal, TypedDict, cast
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from canyonos_core.controller.utils.redis_client import RedisClient
@@ -51,11 +52,24 @@
 _grpc_partial_success_warned = False
 # Destination name -> whether its last export succeeded, so recovery is reported.
 _destination_healthy = {}
-DESTINATIONS_KEY = "otel:destinations"  # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY
+DESTINATIONS_KEY = (
+    "otel:destinations"  # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY
+)
 _redis = None
 
+DestinationProtocol = Literal["grpc", "http", "http/protobuf"]
+
+
+class Destination(TypedDict):
+    name: str
+    protocol: DestinationProtocol
+    endpoint: str
+    headers: dict[str, str] | None
+    insecure: bool | None
+    timeout: float | None
 
-def _validate_destination(destination, index):
+
+def _validate_destination(destination, index) -> Destination:
     if not isinstance(destination, dict):
         raise ValueError(f"destination {index} must be an object")
 
@@ -79,9 +93,7 @@ def _validate_destination(destination, index):
         if not isinstance(headers, dict):
             raise ValueError(f"destination {name!r} headers must be an object")
         if any(
-            not isinstance(key, str)
-            or not key.strip()
-            or not isinstance(value, str)
+            not isinstance(key, str) or not key.strip() or not isinstance(value, str)
             for key, value in headers.items()
         ):
             raise ValueError(
@@ -102,11 +114,11 @@ def _validate_destination(destination, index):
 
     return {
         "name": name.strip(),
-        "protocol": protocol,
+        "protocol": cast(DestinationProtocol, protocol),
         "endpoint": endpoint.strip(),
         "headers": headers,
         "insecure": insecure,
-        "timeout": timeout,
+        "timeout": float(timeout) if timeout is not None else None,
     }
 
 
@@ -168,14 +180,7 @@ def __call__(self, response, *args, **kwargs):
 
 def _build_exporter(destination):
     """Construct one OTLP exporter, and its partial-success recorder when supported."""
-    kwargs = {
-        "endpoint": destination["endpoint"],
-    }
-    if destination["headers"] is not None: kwargs["headers"] = destination["headers"]  # fmt: skip
-    if destination["timeout"] is not None: kwargs["timeout"] = destination["timeout"]  # fmt: skip
-
     if destination["protocol"] == "grpc":
-        if destination["insecure"] is not None: kwargs["insecure"] = destination["insecure"]  # fmt: skip
         global _grpc_partial_success_warned
         if not _grpc_partial_success_warned:
             _grpc_partial_success_warned = True
@@ -184,7 +189,15 @@ def _build_exporter(destination):
                 "discards the response body, so spans this receiver rejects "
                 "individually will still be marked sent."
             )
-        return GrpcOTLPSpanExporter(**kwargs), None
+        return (
+            GrpcOTLPSpanExporter(
+                endpoint=destination["endpoint"],
+                headers=destination["headers"],
+                timeout=destination["timeout"],
+                insecure=destination["insecure"],
+            ),
+            None,
+        )
 
     if destination["insecure"] is not None:
         logger.warning(
@@ -195,8 +208,15 @@ def _build_exporter(destination):
     recorder = _PartialSuccessRecorder(destination["name"])
     session = requests.Session()
     session.hooks["response"].append(recorder)
-    kwargs["session"] = session
-    return HttpOTLPSpanExporter(**kwargs), recorder
+    return (
+        HttpOTLPSpanExporter(
+            endpoint=destination["endpoint"],
+            headers=destination["headers"],
+            timeout=destination["timeout"],
+            session=session,
+        ),
+        recorder,
+    )
 
 
 def _probe_destination(destination_name, exporter):
@@ -213,7 +233,9 @@ def _probe_destination(destination_name, exporter):
         reachable = False
         detail = f": {e}"
     if reachable:
-        logger.info("OTel destination %s answered a connectivity check.", destination_name)
+        logger.info(
+            "OTel destination %s answered a connectivity check.", destination_name
+        )
     else:
         logger.warning(
             "OTel destination %s did not answer a connectivity check, so nothing "
@@ -227,7 +249,9 @@ def _build_exporters(raw):
     """Build one OTLP exporter per configured destination."""
     destinations = _configured_destinations(raw)
     if destinations is None:
-        raise RuntimeError(f"{DESTINATIONS_KEY} is not set; otel.destinations is required")
+        raise RuntimeError(
+            f"{DESTINATIONS_KEY} is not set; otel.destinations is required"
+        )
 
     exporters = []
     recorders = {}
@@ -279,6 +303,9 @@ def _reload_destinations_if_changed():
     # Invalid Redis values are logged and ignored -- keep the previous exporters
     # running rather than tearing down a working config over a bad update.
     global _exporters, _last_destinations_raw
+    if _redis is None:
+        logger.error("Cannot reload OTel destinations before Redis is initialized.")
+        return
     try:
         raw = _redis.get(DESTINATIONS_KEY)
     except Exception as e:
@@ -396,9 +423,7 @@ def _waiting_row_count():
         finally:
             conn.close()
     except Exception as e:
-        logger.error(
-            "Failed to count rows in %s: %s", db.DB_PATH, e, exc_info=True
-        )
+        logger.error("Failed to count rows in %s: %s", db.DB_PATH, e, exc_info=True)
         return None
 
 
@@ -548,7 +573,10 @@ def main():
         _last_destinations_raw = _redis.get(DESTINATIONS_KEY)
     except Exception as e:
         logger.error(
-            "Fatal: cannot reach Redis to read %s: %s", DESTINATIONS_KEY, e, exc_info=True
+            "Fatal: cannot reach Redis to read %s: %s",
+            DESTINATIONS_KEY,
+            e,
+            exc_info=True,
         )
         raise
     try:
@@ -561,7 +589,9 @@ def main():
             exc_info=True,
         )
         raise
-    logger.info("OTel exporter process started with %d destination(s).", len(_exporters))
+    logger.info(
+        "OTel exporter process started with %d destination(s).", len(_exporters)
+    )
     try:
         last_poll = 0
         while _running:

canyonos_core/cli.py

@@ -55,13 +55,19 @@ def _load_config(config_path):
 
 
 def _artifact_prefix(root):
-    return ARTIFACT_DIR_NAME if os.path.isdir(os.path.join(root, ARTIFACT_DIR_NAME)) else ""
+    return (
+        ARTIFACT_DIR_NAME
+        if os.path.isdir(os.path.join(root, ARTIFACT_DIR_NAME))
+        else ""
+    )
 
 
 def _normalize_requirements(agent_cfg):
     """Return an agent's `requirements` list, or [] if absent/null/malformed."""
     requirements = agent_cfg.get("requirements") or []
-    if not isinstance(requirements, list) or not all(isinstance(r, str) for r in requirements):
+    if not isinstance(requirements, list) or not all(
+        isinstance(r, str) for r in requirements
+    ):
         # The requirements list is bad, assuming file has no requirements and logging error
         logger.warning(
             "Agent '%s': `requirements` must be a list of strings, got %r; ignoring.",
@@ -114,8 +120,8 @@ def _write_bake_file(bake_targets, bake_file_path, platform):
                 "tags": [target["image_name"]],
                 "platforms": [platform],
                 "output": ["type=docker"],
-              # type=docker could be changed to tarring it up, which would be
-              # faster but skipped because that change would alter canyonos deploy
+                # type=docker could be changed to tarring it up, which would be
+                # faster but skipped because that change would alter canyonos deploy
             }
             for target in bake_targets
         }
@@ -234,7 +240,9 @@ def _run_build(config_path):
     project_dir = os.path.abspath(os.getcwd())
     prefix = _artifact_prefix(project_dir)
     artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir
-    source_root = os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir
+    source_root = (
+        os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir
+    )
     package_dir = _get_package_dir()
 
     # -------------------------------------------------------------- #

canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py

@@ -198,7 +198,9 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         raise
 
 
-def _bootstrap_instance(host, spec, replica_index, cfg, redis_host, redis_port, agent_id):
+def _bootstrap_instance(
+    host, spec, replica_index, cfg, redis_host, redis_port, agent_id
+):
     """Run the agent container over SSH."""
     ssh_user = cfg["ssh_user"]
 
@@ -264,7 +266,9 @@ def _bootstrap_instance(host, spec, replica_index, cfg, redis_host, redis_port,
     )
     logger.info(
         "docker save|load returncode=%s stdout=%s stderr=%s",
-        result.returncode, result.stdout, result.stderr,
+        result.returncode,
+        result.stdout,
+        result.stderr,
     )
     if result.returncode != 0:
         raise RuntimeError(f"Failed to transfer image to {host}: {result.stderr}")