chore: strengthen Python static analysis

.pre-commit-config.yaml

@@ -21,3 +21,12 @@ repos:
       - id: ruff-check
         args: [--exit-non-zero-on-fix]
       - id: ruff-format
+
+  - repo: local
+    hooks:
+      - id: ty-check
+        name: ty check
+        entry: uv run --locked ty check
+        language: system
+        pass_filenames: false
+        always_run: true

Makefile

@@ -16,8 +16,6 @@ check: ## Run code quality tools.
 	@uv lock --locked
 	@echo "🚀 Linting code: Running prek"
 	@uv run prek run -a
-	@echo "🚀 Static type checking: Running ty"
-	@uv run ty check
 
 .PHONY: test
 test: ## Test the code with pytest
@@ -47,7 +45,7 @@ harness-sync: ## Install the Bub replay harness environment.
 harness-check: ## Validate the Bub replay harness and committed scenarios.
 	@uv run ruff check e2e/bub
 	@uv run ruff format --check e2e/bub
-	@uv run ty check --project e2e/bub --python e2e/bub/.venv e2e/bub/src integrations/bub/src
+	@uv run ty check --project e2e/bub --python e2e/bub/.venv --python-version 3.12 e2e/bub/src integrations/bub/src
 	@uv run --project e2e/bub python -m pytest e2e/bub/tests
 	@uv run --project e2e/bub powercontext-e2e --help >/dev/null
 

e2e/bub/src/powercontext_e2e/harbor_agent.py

@@ -19,7 +19,7 @@
 import shlex
 from importlib.metadata import version
 from pathlib import Path
-from typing import Any
+from typing import Any, override
 
 from harbor.agents.installed import acp as harbor_acp
 from harbor.environments.base import BaseEnvironment
@@ -55,6 +55,7 @@ def __init__(self, **kwargs: Any) -> None:
             **kwargs,
         )
 
+    @override
     async def install(self, environment: BaseEnvironment) -> None:
         await self.exec_as_root(
             environment,

integrations/claude-code/plugins/powercontext/hooks/user_prompt_submit.py

@@ -24,10 +24,19 @@
 from hashlib import sha256
 from pathlib import Path
 from time import monotonic
-from typing import Any, Protocol, cast
+from typing import TYPE_CHECKING, Any, Protocol, TypeVar, cast
 from urllib.error import HTTPError
 from urllib.request import HTTPRedirectHandler, Request, build_opener
 
+if TYPE_CHECKING:
+    from typing_extensions import override
+else:
+    _MethodT = TypeVar("_MethodT")
+
+    def override(method: _MethodT, /) -> _MethodT:
+        return method
+
+
 _PLUGIN_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(_PLUGIN_ROOT))
 
@@ -62,6 +71,7 @@ def read(self, amount: int = -1) -> bytes: ...
 class _RejectRedirects(HTTPRedirectHandler):
     """Leave every 3xx response to urllib's default HTTP error handler."""
 
+    @override
     def redirect_request(
         self,
         req: Request,

integrations/codex/plugins/powercontext/hooks/recall.py

@@ -30,6 +30,8 @@
 from urllib.error import HTTPError
 from urllib.request import HTTPRedirectHandler, Request, build_opener
 
+from typing_extensions import override
+
 _PLUGIN_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(_PLUGIN_ROOT))
 
@@ -64,6 +66,7 @@ def read(self, amount: int = -1) -> bytes: ...
 class _RejectRedirects(HTTPRedirectHandler):
     """Leave every 3xx response to urllib's default HTTP error handler."""
 
+    @override
     def redirect_request(
         self,
         req: Request,

integrations/codex/plugins/powercontext/pyproject.toml

@@ -18,6 +18,7 @@ version = "0.2.0"
 requires-python = ">=3.11,<4.0"
 dependencies = [
     "pydantic-settings>=2.7,<3",
+    "typing-extensions>=4.12,<5",
 ]
 
 [tool.uv]

integrations/codex/plugins/powercontext/settings.py

@@ -23,6 +23,7 @@
 from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator
 from pydantic.fields import FieldInfo
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
+from typing_extensions import override
 
 _MCP_CONFIGURATION_PATH = Path(__file__).with_name(".mcp.json")
 _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"})
@@ -60,11 +61,13 @@ def validate_server_set(self) -> _McpConfiguration:
 class _McpEndpointSettingsSource(PydanticBaseSettingsSource):
     """Load the hook endpoint from the same file consumed by Codex MCP."""
 
+    @override
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         if field_name == "server_url":
             return _server_url_from_mcp_configuration(), field_name, False
         return None, field_name, False
 
+    @override
     def __call__(self) -> dict[str, Any]:
         return {"server_url": _server_url_from_mcp_configuration()}
 
@@ -114,6 +117,7 @@ def validate_authorization(cls, value: SecretStr | None) -> SecretStr | None:
         return value
 
     @classmethod
+    @override
     def settings_customise_sources(
         cls,
         settings_cls: type[BaseSettings],

integrations/hermes/plugins/powercontext/client.py

@@ -19,10 +19,19 @@
 import json
 from collections.abc import Callable
 from http.client import HTTPResponse
-from typing import Any
+from typing import TYPE_CHECKING, Any, TypeVar
 from urllib.error import HTTPError, URLError
 from urllib.request import HTTPRedirectHandler, Request, build_opener
 
+if TYPE_CHECKING:
+    from typing_extensions import override
+else:
+    _MethodT = TypeVar("_MethodT")
+
+    def override(method: _MethodT, /) -> _MethodT:
+        return method
+
+
 MAX_RESPONSE_BYTES = 1_048_576
 
 
@@ -43,6 +52,7 @@ class PowerContextTransportError(PowerContextError):
 
 
 class _NoRedirectHandler(HTTPRedirectHandler):
+    @override
     def redirect_request(self, req: Request, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> Request | None:
         return None
 

pyproject.toml

@@ -23,6 +23,7 @@ requires-python = ">=3.11,<4.0"
 dependencies = [
     "pydantic>=2.10,<3",
     "rfc8785>=0.1.4,<1",
+    "typing-extensions>=4.12,<5",
 ]
 classifiers = [
     "Intended Audience :: Developers",
@@ -120,6 +121,21 @@ extra-paths = [
 [tool.ty.src]
 exclude = ["e2e/bub", "integrations/bub", "evaluation"]
 
+[tool.ty.rules]
+division-by-zero = "error"
+missing-override-decorator = "error"
+missing-type-argument = "error"
+possibly-missing-attribute = "error"
+possibly-missing-import = "error"
+possibly-unresolved-reference = "error"
+unsupported-dynamic-base = "error"
+
+[[tool.ty.overrides]]
+include = ["tests/**"]
+
+[tool.ty.overrides.rules]
+missing-override-decorator = "ignore"
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 markers = [
@@ -133,6 +149,8 @@ fix = true
 
 [tool.ruff.lint]
 select = [
+    # flake8-async
+    "ASYNC",
     # flake8-2020
     "YTT",
     # flake8-bandit
@@ -145,6 +163,10 @@ select = [
     "C4",
     # flake8-debugger
     "T10",
+    # flake8-datetimez
+    "DTZ",
+    # flake8-logging
+    "LOG",
     # flake8-simplify
     "SIM",
     # isort
@@ -155,6 +177,8 @@ select = [
     "E", "W",
     # pyflakes
     "F",
+    # pylint errors
+    "PLE",
     # pygrep-hooks
     "PGH",
     # pyupgrade

src/powercontext/builtin/handoff_report/catalog_store.py

@@ -42,6 +42,7 @@
 )
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncConnection
+from typing_extensions import override
 
 from powercontext.builtin.handoff_report.errors import (
     HandoffReportCatalogArgumentError,
@@ -151,9 +152,11 @@ def __init__(self, items: tuple[CatalogItemT, ...], next_cursor: str | None) ->
         self.items = items
         self.next_cursor = next_cursor
 
+    @override
     def __repr__(self) -> str:
         return f"CatalogPage(items={self.items!r}, next_cursor={self.next_cursor!r})"
 
+    @override
     def __eq__(self, other: object) -> bool:
         return isinstance(other, CatalogPage) and self.items == other.items and self.next_cursor == other.next_cursor
 

src/powercontext/builtin/runtime/application.py

@@ -338,6 +338,7 @@ async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext:
                 service = context.artifacts.memory
                 current = await _head_or_none(service, context.artifacts.memory_artifact_id)
                 memory_hits = ()
+                search_mode: str | None = None
                 if current is not None:
                     result = await service.search(
                         request.query,
@@ -346,13 +347,14 @@ async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext:
                         mode="auto",
                     )
                     memory_hits = result.hits
+                    search_mode = result.mode
                 if span is not None:
                     attributes: dict[str, TraceAttribute] = {
                         _MEMORY_SEARCH_MEMORY_PRESENT: current is not None,
                         _MEMORY_SEARCH_RESULT_COUNT: len(memory_hits),
                     }
-                    if current is not None:
-                        attributes[_MEMORY_SEARCH_MODE] = result.mode
+                    if search_mode is not None:
+                        attributes[_MEMORY_SEARCH_MODE] = search_mode
                     span.set_attributes(attributes)
 
             experience_recall = self._runtime._experience_recall

src/powercontext/builtin/runtime/composition.py

@@ -22,6 +22,7 @@
 from typing import TYPE_CHECKING, TypeVar
 
 from pydantic import JsonValue
+from typing_extensions import override
 
 from powercontext.builtin.artifacts.experience import ExperienceCandidatePipeline, ExperienceGenerator
 from powercontext.builtin.artifacts.handoff import (
@@ -94,6 +95,7 @@ def __init__(self, issue: str) -> None:
 
 
 class _ContentEvidenceProjector(DefaultMemoryEvidenceProjector):
+    @override
     def project_source(self, source: Source, /) -> JsonValue:
         if isinstance(source, ContentSource):
             return {
@@ -106,6 +108,7 @@ def project_source(self, source: Source, /) -> JsonValue:
 
 
 class _ContentHandoffEvidenceProjector(DefaultHandoffEvidenceProjector):
+    @override
     def project_source(self, source: Source, /) -> JsonValue:
         if isinstance(source, ContentSource):
             return {

src/powercontext/client/client.py

@@ -357,8 +357,8 @@ async def _request_handoff_report_content(self, request: GetHandoffReportRequest
             mode="json",
             by_alias=True,
         )
+        span = ClientSpan.start(GET_HANDOFF_REPORT.operation_id)
         try:
-            span = ClientSpan.start(GET_HANDOFF_REPORT.operation_id)
             headers = {} if self._headers is None else dict(self._headers)
             span.inject(headers)
             response = await self._http_client.request(
@@ -582,8 +582,8 @@ async def _request(
             else:
                 json_payload = payload
 
+        span = ClientSpan.start(operation.operation_id)
         try:
-            span = ClientSpan.start(operation.operation_id)
             headers = {} if self._headers is None else dict(self._headers)
             span.inject(headers)
             response = await self._http_client.request(

src/powercontext/server/access.py

@@ -23,6 +23,7 @@
 
 from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
+from typing_extensions import override
 
 from powercontext._logging import log_safely
 from powercontext.server.context import current_request_id, is_internal_bridge
@@ -76,6 +77,7 @@ async def send_with_access_log(message: Message) -> None:
 class McpAccessLogMiddleware(Middleware):
     """Log logical MCP protocol requests instead of Streamable HTTP frames."""
 
+    @override
     async def on_request(
         self,
         context: MiddlewareContext[Any],

src/powercontext/server/logging.py

@@ -23,6 +23,7 @@
 from typing import Any
 
 from opentelemetry import trace
+from typing_extensions import override
 
 from powercontext.server.settings import ServerLoggingConfig
 
@@ -43,6 +44,7 @@
 
 
 class OperationalContextFilter(logging.Filter):
+    @override
     def filter(self, record: logging.LogRecord) -> bool:
         span_context = trace.get_current_span().get_span_context()
         if span_context.is_valid:
@@ -59,6 +61,7 @@ def filter(self, record: logging.LogRecord) -> bool:
 class JsonFormatter(logging.Formatter):
     """Render a stable operational record without serializing arbitrary extras."""
 
+    @override
     def format(self, record: logging.LogRecord) -> str:
         payload: dict[str, Any] = {
             "timestamp": datetime.fromtimestamp(record.created, UTC).isoformat(),
@@ -75,6 +78,7 @@ def format(self, record: logging.LogRecord) -> str:
 
 
 class _HumanContextFilter(OperationalContextFilter):
+    @override
     def filter(self, record: logging.LogRecord) -> bool:
         super().filter(record)
         request_id = getattr(record, "request_id", None)