chore(ci): add backend Ruff and Mypy checks

.github/workflows/ci.yml

@@ -129,6 +129,13 @@ jobs:
           python -m pip install -r apps/backend/requirements-dev.txt
           python -m pip install -e apps/backend --no-deps
 
+      - name: Static analysis
+        working-directory: apps/backend
+        run: |
+          ruff check app
+          ruff format --check app
+          mypy app
+
       - name: Test
         working-directory: apps/backend
         env:

apps/backend/README.md

@@ -49,6 +49,24 @@ npm run test:backend      # from the repository root
 
 Tests use per-test SQLite and the in-memory rate limiter by default. When `PARTHA_TEST_PG_URL` is set, the migration round trip runs against a fresh temporary PostgreSQL database and the PostgreSQL refresh-token concurrency test is enabled; without it, migrations fall back to SQLite and the concurrency test skips. Redis integration tests skip unless `PARTHA_TEST_REDIS_URL` is set. CI provides both services.
 
+## Static analysis
+
+The runtime lockfile deliberately excludes development tooling. Install the
+pinned development toolchain before running the backend static-analysis gates:
+
+```bash
+cd apps/backend
+python -m pip install -r requirements-dev.txt
+python -m pip install -e . --no-deps
+ruff check app
+ruff format --check app
+mypy app
+```
+
+Ruff targets the production package and Python 3.12, the project's minimum
+supported Python version. Run `ruff format app` to apply the repository's
+formatter, then review the resulting diff before committing.
+
 ## Migrations
 
 ```bash

apps/backend/app/ai/orchestrator.py

@@ -46,7 +46,9 @@ async def test_connection(self, request: AiProviderTestRequest) -> AiProviderTes
         provider = self.provider_factory.resolve(config)
         prompt = PromptBundle(system_prompt="Reply with the single word: ok", user_prompt="Connection test.")
         await provider.complete(config, prompt)
-        return AiProviderTestResponse(ok=True, message=f"{config.provider} connection succeeded.", checked_at=datetime.now(UTC))
+        return AiProviderTestResponse(
+            ok=True, message=f"{config.provider} connection succeeded.", checked_at=datetime.now(UTC)
+        )
 
     async def query(self, request: AiQueryRequest) -> AiQueryResponse:
         # Owner-scoped: another user's repository id resolves to None and gets

apps/backend/app/ai/prompt_builder.py

@@ -33,12 +33,14 @@ def render_repository_context(self, repository_context: RepositoryContext) -> st
             f"Frameworks: {', '.join(architecture.frameworks) if architecture.frameworks else 'Not detected'}",
             f"Entry points: {', '.join(architecture.entry_points) if architecture.entry_points else 'Not found'}",
             "Modules (roles are heuristic classifications):",
+            *[f"- {module.name} ({module.role}, {module.file_count} files)" for module in architecture.modules],
+            "Dependencies:",
             *[
-                f"- {module.name} ({module.role}, {module.file_count} files)"
-                for module in architecture.modules
+                self._render_dependency(
+                    dependency.name, dependency.version, dependency.declared_versions, dependency.has_version_conflict
+                )
+                for dependency in repository_context.dependencies
             ],
-            "Dependencies:",
-            *[self._render_dependency(dependency.name, dependency.version, dependency.declared_versions, dependency.has_version_conflict) for dependency in repository_context.dependencies],
             "Files:",
             *[f"- {file.path}" for file in repository_context.selected_files],
         ]

apps/backend/app/ai/providers/base.py

@@ -4,5 +4,4 @@
 
 
 class AiProvider(Protocol):
-    async def complete(self, config: AiProviderConfig, prompt: PromptBundle) -> AiProviderResponse:
-        ...
+    async def complete(self, config: AiProviderConfig, prompt: PromptBundle) -> AiProviderResponse: ...

apps/backend/app/ai/providers/config_store.py

@@ -145,7 +145,12 @@ def _resolve_key(
         """
         if config.api_key:
             return self.cipher.encrypt(config.api_key), config.api_key[-4:]
-        if config.provider != "ollama" and record is not None and record.provider == config.provider and record.encrypted_api_key:
+        if (
+            config.provider != "ollama"
+            and record is not None
+            and record.provider == config.provider
+            and record.encrypted_api_key
+        ):
             return record.encrypted_api_key, record.api_key_last4
         if config.provider == "ollama":
             return None, None

apps/backend/app/ai/repository_context.py

@@ -38,9 +38,7 @@ def build(self, record: RepositoryRecord, selected_file: str | None = None) -> R
         )
         dependencies: list[DependencyContext] = []
         for dependency in projection.dependencies[:20]:
-            declared_versions = tuple(
-                dict.fromkeys(declaration.version for declaration in dependency.declarations)
-            )
+            declared_versions = tuple(dict.fromkeys(declaration.version for declaration in dependency.declarations))
             has_conflict = len(declared_versions) > 1
             dependencies.append(
                 DependencyContext(

apps/backend/app/analysis/architecture.py

@@ -52,7 +52,6 @@
 }
 
 
-
 class ArchitectureAnalyzer:
     """Builds the Architecture read model exclusively from sealed ri.v1 snapshots.
 
@@ -83,7 +82,9 @@ def build_architecture(self, record: RepositoryRecord) -> ArchitectureResponse:
         remaining_node_ids = {node.id for node in nodes}
         for diagnostic in diagnostics:
             if diagnostic.node_ids is not None:
-                diagnostic.node_ids = [node_id for node_id in diagnostic.node_ids if node_id in remaining_node_ids] or None
+                diagnostic.node_ids = [
+                    node_id for node_id in diagnostic.node_ids if node_id in remaining_node_ids
+                ] or None
         self._set_relationship_states(modules, nodes, edges, unresolved_node_ids, covered_paths, facts is not None)
         layers = self._layers_for_nodes(nodes)
         arch_modules = [
@@ -509,9 +510,7 @@ def _architecture_endpoint_ids(
         if node is not None and not evidence:
             evidence = facts.node_evidence.get(node.id, [])
         exact = {
-            module_id
-            for item in evidence
-            for module_id in modules_by_file.get(self._normalize_path(item.path), [])
+            module_id for item in evidence for module_id in modules_by_file.get(self._normalize_path(item.path), [])
         }
         if exact:
             return exact
@@ -535,10 +534,7 @@ def _modules_for_evidence_scope(
             if not directory:
                 continue
             for module in module_by_id.values():
-                if any(
-                    self._path_is_within(self._normalize_path(file_path), directory)
-                    for file_path in module.files
-                ):
+                if any(self._path_is_within(self._normalize_path(file_path), directory) for file_path in module.files):
                     result.add(module.id)
         return result
 
@@ -558,8 +554,11 @@ def _set_relationship_states(
                 node.relationship_state = "connected"
             elif node.id in unresolved_node_ids:
                 node.relationship_state = "unresolved"
-            elif node.id in module_by_id and snapshot_available and module_by_id[node.id].files and all(
-                self._normalize_path(path) in covered_paths for path in module_by_id[node.id].files
+            elif (
+                node.id in module_by_id
+                and snapshot_available
+                and module_by_id[node.id].files
+                and all(self._normalize_path(path) in covered_paths for path in module_by_id[node.id].files)
             ):
                 node.relationship_state = "no-observed-relationships"
             else:
@@ -620,7 +619,12 @@ def _layers_for_nodes(self, nodes: list[ArchNode]) -> list[ArchLayer]:
         for node in nodes:
             layers.setdefault(node.layer, []).append(node.id)
         return [
-            ArchLayer(id=layer, name=layer.replace("-", " ").title(), order=LAYER_ORDER.get(layer, 99), nodes=node_ids)
+            ArchLayer(
+                id=layer,
+                name=layer.replace("-", " ").title(),
+                order=LAYER_ORDER.get(layer, 99),
+                nodes=node_ids,
+            )
             for layer, node_ids in sorted(layers.items(), key=lambda item: LAYER_ORDER.get(item[0], 99))
         ]
 
@@ -634,12 +638,45 @@ def _architecture_type(self, frameworks: list[str]) -> str:
     def _request_flow(self, modules: list[RepositoryModule]) -> list[RequestFlowStep]:
         module_roles = {module.role for module in modules}
         steps = [
-            RequestFlowStep(id="client", name="Client", type="frontend", description="Request enters the system.", details=["Browser or API client sends a request."]),
+            RequestFlowStep(
+                id="client",
+                name="Client",
+                type="frontend",
+                description="Request enters the system.",
+                details=["Browser or API client sends a request."],
+            ),
         ]
         if "route" in module_roles or "controller" in module_roles:
-            steps.append(RequestFlowStep(id="api", name="API Layer", type="controller", description="Route/controller handles input.", details=["Validate request", "Call service"]))
+            steps.append(
+                RequestFlowStep(
+                    id="api",
+                    name="API Layer",
+                    type="controller",
+                    description="Route/controller handles input.",
+                    details=["Validate request", "Call service"],
+                )
+            )
         if "service" in module_roles:
-            steps.append(RequestFlowStep(id="service", name="Service Layer", type="service", description="Business logic executes.", details=["Coordinate repository intelligence consumers", "Transform data"]))
+            steps.append(
+                RequestFlowStep(
+                    id="service",
+                    name="Service Layer",
+                    type="service",
+                    description="Business logic executes.",
+                    details=[
+                        "Coordinate repository intelligence consumers",
+                        "Transform data",
+                    ],
+                )
+            )
         if "repository" in module_roles:
-            steps.append(RequestFlowStep(id="repository", name="Repository Layer", type="repository", description="Persistence or source files are accessed.", details=["Read or write data"] ))
+            steps.append(
+                RequestFlowStep(
+                    id="repository",
+                    name="Repository Layer",
+                    type="repository",
+                    description="Persistence or source files are accessed.",
+                    details=["Read or write data"],
+                )
+            )
         return steps

apps/backend/app/analysis/authentication.py

@@ -41,9 +41,7 @@
 _GUARD_CLASSIFICATION = "auth_dependency"
 _SERVICE_CLASSIFICATION = "service"
 _MODEL_CLASSIFICATION = "model"
-_DIAGNOSTIC_CODES = frozenset(
-    {"RI-RES-UNRESOLVED", "RI-RES-AMBIGUOUS", "RI-EXT-UNSUPPORTED", "RI-SRC-MALFORMED"}
-)
+_DIAGNOSTIC_CODES = frozenset({"RI-RES-UNRESOLVED", "RI-RES-AMBIGUOUS", "RI-EXT-UNSUPPORTED", "RI-SRC-MALFORMED"})
 _MAX_DIAGNOSTICS = 50
 
 
@@ -274,11 +272,7 @@ def _reachable_service_model_edges(self, guard_node: RiNode) -> list[tuple[str,
                 used_nodes.add(walk_key)
                 walk_key = parent[walk_key][0]
 
-        return [
-            (edge.subject_key, edge.object_key, edge)
-            for edge in discovered_edges
-            if edge.object_key in used_nodes
-        ]
+        return [(edge.subject_key, edge.object_key, edge) for edge in discovered_edges if edge.object_key in used_nodes]
 
     def _role_of(self, stable_key: str) -> AuthRelationshipNodeKind:
         roles = self.classifications.get(stable_key, ())

apps/backend/app/analysis/manifest.py

@@ -154,8 +154,7 @@ def verify(
                 matches_stored_snapshot=False,
                 mismatched_fields=mismatched,
                 detail=(
-                    "The manifest is internally consistent but does not match the "
-                    "snapshot stored for this repository."
+                    "The manifest is internally consistent but does not match the snapshot stored for this repository."
                 ),
             )
         if named.snapshot_id != current.snapshot_id:
@@ -180,6 +179,4 @@ def verify(
     def _mismatched_fields(stored: RevisionManifest, submitted: RevisionManifest) -> list[str]:
         stored_fields = stored.model_dump(mode="json", by_alias=True)
         submitted_fields = submitted.model_dump(mode="json", by_alias=True)
-        return sorted(
-            key for key in stored_fields if stored_fields[key] != submitted_fields.get(key)
-        )
+        return sorted(key for key in stored_fields if stored_fields[key] != submitted_fields.get(key))

apps/backend/app/api/openapi.py

@@ -122,9 +122,7 @@ def remove_suppressed_automatic_validation_errors(document: dict[str, Any]) -> N
     """Remove FastAPI's synthetic 422 response from explicitly marked operations."""
     for path_item in document.get("paths", {}).values():
         for operation in path_item.values():
-            if not isinstance(operation, dict) or not operation.pop(
-                _SUPPRESS_AUTOMATIC_422_MARKER, False
-            ):
+            if not isinstance(operation, dict) or not operation.pop(_SUPPRESS_AUTOMATIC_422_MARKER, False):
                 continue
             operation.get("responses", {}).pop("422", None)
 

apps/backend/app/api/routes/ai.py

@@ -4,7 +4,14 @@
 
 from app.api.deps import get_ai_service, get_current_user
 from app.api.openapi import documented_responses
-from app.schemas.ai import AiProviderConfig, AiProviderPublicConfig, AiProviderTestRequest, AiProviderTestResponse, AiQueryRequest, AiQueryResponse
+from app.schemas.ai import (
+    AiProviderConfig,
+    AiProviderPublicConfig,
+    AiProviderTestRequest,
+    AiProviderTestResponse,
+    AiQueryRequest,
+    AiQueryResponse,
+)
 from app.services.ai_service import AiService
 
 # Every AI route requires auth; the repository and the provider config are both

apps/backend/app/api/routes/analysis.py

@@ -226,8 +226,7 @@ def get_authentication_explanation(
     dependencies=[Depends(require_repository_owner)],
     responses=documented_responses(
         200,
-        "Source text for one evidence citation, verified against the exact "
-        "snapshot revision it was cited from.",
+        "Source text for one evidence citation, verified against the exact snapshot revision it was cited from.",
         {
             "schemaVersion": "evidence-source.v1",
             "repositoryId": _REPOSITORY_ID,
@@ -240,7 +239,7 @@ def get_authentication_explanation(
             "endLine": 10,
             "status": "ready",
             "reason": None,
-            "content": "@app.get(\"/me\")\n",
+            "content": '@app.get("/me")\n',
             "truncated": False,
             "size": 16,
         },
@@ -390,9 +389,7 @@ def verify_revision_manifest(
                     ],
                 }
             ],
-            "edges": [
-                {"id": "edge_example", "source": "repo:root", "target": "dep:npm:react", "type": "depends-on"}
-            ],
+            "edges": [{"id": "edge_example", "source": "repo:root", "target": "dep:npm:react", "type": "depends-on"}],
             "totalDependencies": 1,
             "manifestCount": 1,
             "diagnostics": [],