diff --git a/AGENTS.md b/AGENTS.md index 4a80a4f..e928c3f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,4 +12,5 @@ - Use deterministic ordering, hashes, JSON results, and structured errors. - Fail closed on stale caches, invalid configuration, ambiguous IDs, and unauthorized families. - Keep dependencies small and pinned by compatible major version. -- Run formatting, static checks, focused tests, and the complete test suite before closing a gate. +- Run strict `pyright`, formatting, Ruff, compilation, focused tests, and the complete warning-strict + test suite before closing a gate. diff --git a/README.md b/README.md index c0a946e..7ec6d93 100644 --- a/README.md +++ b/README.md @@ -44,8 +44,14 @@ progressive hop-distance shading. ## Development +Install Pyright once with `npm install -g pyright`. DocForge configures it to use the repository +virtual environment and treats a clean strict run as a required development gate. + ```bash uv sync +pyright +uv run ruff check src tests +uv run ruff format --check src tests uv run python -m unittest discover -s tests -v uv run docforge --project-root tests/fixtures/alpha validate uv run docforge --project-root tests/fixtures/alpha render-status manual diff --git a/SLICE_HISTORY.md b/SLICE_HISTORY.md index 7a4086c..58d04c6 100644 --- a/SLICE_HISTORY.md +++ b/SLICE_HISTORY.md @@ -1,5 +1,20 @@ # Completed slices +## DFG-15 strict static typing gate + +### Changed + +- Made the existing strict Pyright configuration resolve DocForge's `.venv` automatically. +- Converted validated TOML, JSON, subprocess, socket, MCP, render, and visualization boundaries + from unknown dynamic values into explicit checked types. +- Kept runtime validation and fail-closed behavior at every untrusted input boundary. +- Made strict `pyright` an explicit repository development gate. + +### Verification + +- Pyright reports zero errors, warnings, or informational diagnostics across all source modules. +- Ruff lint and formatting, Python compilation, all 52 warning-strict tests, and diff checks pass. + ## DFG-14.1 detached viewer lifecycle correction ### Changed diff --git a/pyproject.toml b/pyproject.toml index 424983b..fdd7cca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,3 +30,5 @@ select = ["E", "F", "I", "UP", "B", "SIM"] pythonVersion = "3.12" typeCheckingMode = "strict" include = ["src"] +venvPath = "." +venv = ".venv" diff --git a/src/docforge/adapter_contract.py b/src/docforge/adapter_contract.py index c4bf179..23c1a59 100644 --- a/src/docforge/adapter_contract.py +++ b/src/docforge/adapter_contract.py @@ -355,7 +355,7 @@ class AdapterProject: def _resolved_directories( root: Path, paths: tuple[Path, ...], *, label: str ) -> tuple[Path, ...]: - resolved = [] + resolved: list[Path] = [] for path in paths: if path.is_symlink(): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") @@ -369,7 +369,7 @@ class AdapterProject: @staticmethod def _resolved_files(root: Path, paths: tuple[Path, ...], *, label: str) -> tuple[Path, ...]: - resolved = [] + resolved: list[Path] = [] for path in paths: if path.is_symlink(): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") diff --git a/src/docforge/changeset_contract.py b/src/docforge/changeset_contract.py index bb4f4d4..6fcc65a 100644 --- a/src/docforge/changeset_contract.py +++ b/src/docforge/changeset_contract.py @@ -6,7 +6,7 @@ import hashlib import json import re from pathlib import Path -from typing import Any +from typing import Any, cast from .errors import DocForgeError from .models import Node, ProjectService @@ -80,14 +80,16 @@ def edge_dict(edge: tuple[str, str, str], *, action: str) -> dict[str, str]: } -def validate_id(value: Any, field: str) -> None: +def validate_id(value: object, field: str) -> str: if not isinstance(value, str) or ID_PATTERN.fullmatch(value) is None: raise DocForgeError("invalid_id", f"{field} must be a stable ID", field=field) + return value -def validate_hash(value: Any, field: str) -> None: +def validate_hash(value: object, field: str) -> str: if not isinstance(value, str) or HASH_PATTERN.fullmatch(value) is None: raise DocForgeError("invalid_hash", f"{field} must be a lowercase SHA-256 hash") + return value def normalize_operation(operation: dict[str, Any], *, sequence: int) -> dict[str, Any]: @@ -109,21 +111,28 @@ def normalize_operation(operation: dict[str, Any], *, sequence: int) -> dict[str if not isinstance(relationships, list): raise DocForgeError("invalid_operation", "relationship_changes must be an array") normalized_relationships: list[dict[str, str]] = [] - for change in relationships: - if not isinstance(change, dict) or set(change) != RELATIONSHIP_KEYS: + for change_value in cast(list[object], relationships): + if not isinstance(change_value, dict): raise DocForgeError( "invalid_operation", "Relationship changes require exact structured fields" ) - if change["action"] not in {"add", "remove"}: + change = cast(dict[str, object], change_value) + if set(change) != set(RELATIONSHIP_KEYS): + raise DocForgeError( + "invalid_operation", "Relationship changes require exact structured fields" + ) + action = change["action"] + if not isinstance(action, str) or action not in {"add", "remove"}: raise DocForgeError("invalid_operation", "Relationship action is invalid") - for field in ("source_id", "relation", "target_id"): - validate_id(change[field], field) + source_id = validate_id(change["source_id"], "source_id") + relation = validate_id(change["relation"], "relation") + target_id = validate_id(change["target_id"], "target_id") normalized_relationships.append( { - "action": change["action"], - "source_id": change["source_id"], - "relation": change["relation"], - "target_id": change["target_id"], + "action": action, + "source_id": source_id, + "relation": relation, + "target_id": target_id, } ) keys = [ @@ -136,6 +145,7 @@ def normalize_operation(operation: dict[str, Any], *, sequence: int) -> dict[str if metadata is not None and not isinstance(metadata, dict): raise DocForgeError("invalid_operation", "metadata must be an object or null") if isinstance(metadata, dict): + metadata = cast(dict[str, object], metadata) unknown_metadata = sorted(set(metadata) - METADATA_KEYS) if unknown_metadata: raise DocForgeError( @@ -172,7 +182,12 @@ def normalize_operation(operation: dict[str, Any], *, sequence: int) -> dict[str def validate_document(project: ProjectService, document: Any, *, path: Path) -> dict[str, Any]: - if not isinstance(document, dict) or set(document) != CHANGESET_KEYS: + if not isinstance(document, dict): + raise DocForgeError( + "invalid_changeset", "Changeset has missing or unknown fields", path=path.name + ) + document = cast(dict[str, Any], document) + if set(document) != set(CHANGESET_KEYS): raise DocForgeError( "invalid_changeset", "Changeset has missing or unknown fields", path=path.name ) @@ -207,11 +222,17 @@ def validate_document(project: ProjectService, document: Any, *, path: Path) -> operations = document["operations"] if not isinstance(operations, list): raise DocForgeError("invalid_changeset", "operations must be an array") - if len(operations) > descriptor.limits.max_changeset_operations: + typed_operations = cast(list[object], operations) + if len(typed_operations) > descriptor.limits.max_changeset_operations: raise DocForgeError("changeset_operation_limit", "Changeset has too many operations") normalized: list[dict[str, Any]] = [] - for index, operation in enumerate(operations, start=1): - if not isinstance(operation, dict) or set(operation) != OPERATION_KEYS: + for index, operation_value in enumerate(typed_operations, start=1): + if not isinstance(operation_value, dict): + raise DocForgeError( + "invalid_changeset", "Stored operation has missing or unknown fields" + ) + operation = cast(dict[str, Any], operation_value) + if set(operation) != set(OPERATION_KEYS): raise DocForgeError( "invalid_changeset", "Stored operation has missing or unknown fields" ) diff --git a/src/docforge/changesets.py b/src/docforge/changesets.py index 0e658da..6bd8c78 100644 --- a/src/docforge/changesets.py +++ b/src/docforge/changesets.py @@ -6,7 +6,7 @@ import fcntl import json import os import tempfile -from collections.abc import Iterator +from collections.abc import Generator from contextlib import contextmanager from pathlib import Path from typing import Any @@ -517,7 +517,7 @@ class ChangesetStore: return root @contextmanager - def _lock(self) -> Iterator[None]: + def _lock(self) -> Generator[None]: root = self._root() lock_path = root / ".lock" try: diff --git a/src/docforge/config_validation.py b/src/docforge/config_validation.py index 8d5070a..e747630 100644 --- a/src/docforge/config_validation.py +++ b/src/docforge/config_validation.py @@ -3,8 +3,9 @@ from __future__ import annotations import re +from collections.abc import Mapping from pathlib import Path -from typing import Any +from typing import cast from .errors import DocForgeError @@ -13,7 +14,7 @@ AUTHORITIES = frozenset({"authoritative", "approved_plan", "derived", "proposal" _SECRET_PARTS = frozenset({".git", ".ssh", ".gnupg", "secrets", "credentials"}) -def require_string(document: dict[str, Any], key: str, source: Path) -> str: +def require_string(document: Mapping[str, object], key: str, source: Path) -> str: value = document.get(key) if not isinstance(value, str) or not value.strip(): raise DocForgeError("invalid_source", f"{source.name}: {key} must be a non-empty string") @@ -21,11 +22,16 @@ def require_string(document: dict[str, Any], key: str, source: Path) -> str: def string_list(value: object, *, key: str, source: Path) -> tuple[str, ...]: - if not isinstance(value, list) or any(not isinstance(item, str) or not item for item in value): + if not isinstance(value, list): raise DocForgeError("invalid_source", f"{source.name}: {key} must be a string list") - if len(value) != len(set(value)): + items: list[str] = [] + for item in cast(list[object], value): + if not isinstance(item, str) or not item: + raise DocForgeError("invalid_source", f"{source.name}: {key} must be a string list") + items.append(item) + if len(items) != len(set(items)): raise DocForgeError("invalid_source", f"{source.name}: {key} contains duplicates") - return tuple(value) + return tuple(items) def confined_path( diff --git a/src/docforge/context.py b/src/docforge/context.py index 78d7b9d..1e4cb7a 100644 --- a/src/docforge/context.py +++ b/src/docforge/context.py @@ -42,8 +42,7 @@ def compile_context( profile = _profile(snapshot, profile_id) selected_budget = profile.token_budget if budget is None else budget if ( - not isinstance(selected_budget, int) - or isinstance(selected_budget, bool) + isinstance(selected_budget, bool) or selected_budget < 1 or selected_budget > snapshot.descriptor.limits.max_context_tokens ): diff --git a/src/docforge/mcp_server.py b/src/docforge/mcp_server.py index 614651d..edd9b5e 100644 --- a/src/docforge/mcp_server.py +++ b/src/docforge/mcp_server.py @@ -6,7 +6,7 @@ import argparse import json from collections.abc import Callable from pathlib import Path -from typing import Any +from typing import Any, cast from mcp.server.fastmcp import FastMCP @@ -272,7 +272,7 @@ class DocForgeService: query=query, depth=depth, ) - snapshot = visualization["snapshot"] + snapshot = cast(dict[str, object], visualization["snapshot"]) return { "status": "ok", "project_id": snapshot["project_id"], @@ -396,6 +396,20 @@ def _create_bound_server(service: DocForgeService, *, read_only: bool) -> FastMC return service.visualize(node_id=node_id, query=query, depth=depth) + _registered_read_tools = ( + project_info, + get_contract, + get_node, + search, + filter_nodes, + backlinks, + dependencies, + impact, + get_context, + validate_project, + render_status, + visualize, + ) if read_only: return server @@ -531,6 +545,18 @@ def _create_bound_server(service: DocForgeService, *, read_only: bool) -> FastMC return service.invoke(lambda: service.rendering.preview(changeset_id, view_id)) + _registered_proposal_tools = ( + create_changeset, + list_changesets, + get_changeset, + propose_node_create, + propose_node_update, + propose_node_move, + propose_node_delete, + validate_changeset, + get_changeset_diff, + preview_changeset, + ) return server diff --git a/src/docforge/project.py b/src/docforge/project.py index d5192e9..f2ddcea 100644 --- a/src/docforge/project.py +++ b/src/docforge/project.py @@ -10,7 +10,7 @@ from collections import Counter from collections.abc import Mapping from dataclasses import replace from pathlib import Path -from typing import Any +from typing import Any, cast from .config_validation import ( AUTHORITIES, @@ -82,7 +82,7 @@ def _load_descriptor(root: Path) -> ProjectDescriptor: raise DocForgeError("missing_config", "Missing .docforge/project.toml") try: descriptor_bytes = descriptor_path.read_bytes() - document = tomllib.loads(descriptor_bytes.decode("utf-8")) + document = cast(dict[str, object], tomllib.loads(descriptor_bytes.decode("utf-8"))) except UnicodeDecodeError as error: raise DocForgeError("invalid_config", "Project descriptor is not UTF-8") from error except tomllib.TOMLDecodeError as error: @@ -119,6 +119,10 @@ def _load_descriptor(root: Path) -> ProjectDescriptor: raise DocForgeError( "invalid_config", "sources, derived, changesets, and graph tables are required" ) + sources = cast(dict[str, object], sources) + derived = cast(dict[str, object], derived) + changesets = cast(dict[str, object], changesets) + graph = cast(dict[str, object], graph) for table, allowed, name in ( (sources, _SOURCE_KEYS, "sources"), (derived, _DERIVED_KEYS, "derived"), @@ -192,9 +196,10 @@ def _load_descriptor(root: Path) -> ProjectDescriptor: raise DocForgeError("invalid_config", "changesets.writers must be an array of tables") proposal_writers: list[ProposalWriter] = [] writer_ids: set[str] = set() - for writer in writer_documents: - if not isinstance(writer, dict): + for writer_value in cast(list[object], writer_documents): + if not isinstance(writer_value, dict): raise DocForgeError("invalid_config", "Each changeset writer must be a table") + writer = cast(dict[str, object], writer_value) unknown_writer = sorted(set(writer) - _WRITER_KEYS) if unknown_writer: raise DocForgeError( @@ -241,6 +246,7 @@ def _load_descriptor(root: Path) -> ProjectDescriptor: limit_values = document.get("limits", {}) if not isinstance(limit_values, dict): raise DocForgeError("invalid_config", "limits must be a table") + limit_values = cast(dict[str, object], limit_values) defaults = Limits() unknown_limits = sorted(set(limit_values) - set(defaults.__dataclass_fields__)) if unknown_limits: @@ -269,9 +275,10 @@ def _load_descriptor(root: Path) -> ProjectDescriptor: raise DocForgeError("invalid_config", "profiles must be an array of tables") profiles: list[ContextProfile] = [] profile_ids: set[str] = set() - for profile in profile_documents: - if not isinstance(profile, dict): + for profile_value in cast(list[object], profile_documents): + if not isinstance(profile_value, dict): raise DocForgeError("invalid_config", "Each profile must be a table") + profile = cast(dict[str, object], profile_value) unknown_profile = sorted(set(profile) - _PROFILE_KEYS) if unknown_profile: raise DocForgeError( @@ -416,7 +423,7 @@ def _load_source_file( relative = path.relative_to(descriptor.root).as_posix() if path.suffix == ".md": record, content = _markdown_record(path, text) - node, edges = validated_node_from_record( + node, node_edges = validated_node_from_record( record, content=content, source=path, @@ -424,10 +431,10 @@ def _load_source_file( relations=descriptor.allowed_relations, hash_bytes=raw, ) - return (node,), edges + return (node,), node_edges if path.suffix == ".toml": try: - document = tomllib.loads(text) + document = cast(dict[str, object], tomllib.loads(text)) except tomllib.TOMLDecodeError as error: raise DocForgeError("invalid_source", f"{path.name}: invalid TOML: {error}") from error records = document.get("nodes") @@ -435,9 +442,10 @@ def _load_source_file( raise DocForgeError("invalid_source", f"{path.name}: TOML sources require [[nodes]]") nodes: list[Node] = [] edges: list[Edge] = [] - for index, record in enumerate(records): - if not isinstance(record, dict): + for index, record_value in enumerate(cast(list[object], records)): + if not isinstance(record_value, dict): raise DocForgeError("invalid_source", f"{path.name}: nodes must be tables") + record = cast(dict[str, Any], record_value) content = record.get("content") if not isinstance(content, str): raise DocForgeError( diff --git a/src/docforge/proposal_projection.py b/src/docforge/proposal_projection.py index a1c473c..382bb83 100644 --- a/src/docforge/proposal_projection.py +++ b/src/docforge/proposal_projection.py @@ -5,7 +5,7 @@ from __future__ import annotations import difflib from dataclasses import replace from pathlib import Path -from typing import Any +from typing import Any, cast from .changeset_contract import ( METADATA_KEYS, @@ -113,8 +113,13 @@ class ProposalProjector: raise DocForgeError("node_exists", "Create target already exists", node_id=node_id) if operation["expected_content_hash"] is not None: raise DocForgeError("invalid_operation", "Create expected_content_hash must be null") - metadata = operation["metadata"] - if not isinstance(metadata, dict) or not REQUIRED_CREATE_METADATA.issubset(metadata): + metadata_value = operation["metadata"] + if not isinstance(metadata_value, dict): + raise DocForgeError( + "invalid_operation", "Create requires complete node metadata", node_id=node_id + ) + metadata = cast(dict[str, Any], metadata_value) + if not REQUIRED_CREATE_METADATA.issubset(metadata): raise DocForgeError( "invalid_operation", "Create requires complete node metadata", node_id=node_id ) @@ -156,9 +161,10 @@ class ProposalProjector: ) -> None: if operation["target_source"] is not None: raise DocForgeError("invalid_operation", "Update cannot move a node") - metadata = operation["metadata"] - if metadata is not None and not isinstance(metadata, dict): + metadata_value = operation["metadata"] + if metadata_value is not None and not isinstance(metadata_value, dict): raise DocForgeError("invalid_operation", "Update metadata must be an object or null") + metadata = None if metadata_value is None else cast(dict[str, Any], metadata_value) if ( metadata is None and operation["content"] is None @@ -171,7 +177,7 @@ class ProposalProjector: "invalid_operation", "Update may change only relationships owned by its node" ) self._apply_relationship_changes(edges, changes) - merged = {**node_metadata(current), **(metadata or {})} + merged: dict[str, Any] = {**node_metadata(current), **(metadata or {})} content = current.content if operation["content"] is None else operation["content"] node = self._build_node( snapshot, diff --git a/src/docforge/render_config.py b/src/docforge/render_config.py index 84aa0df..b65f66a 100644 --- a/src/docforge/render_config.py +++ b/src/docforge/render_config.py @@ -3,6 +3,7 @@ from __future__ import annotations from pathlib import Path +from typing import cast from .config_validation import ID_PATTERN, confined_path, require_string, string_list from .errors import DocForgeError @@ -33,6 +34,7 @@ def load_render_config( return None if not isinstance(document, dict): raise DocForgeError("invalid_config", "render must be a table") + document = cast(dict[str, object], document) unknown = sorted(set(document) - _RENDER_KEYS) if unknown: raise DocForgeError("invalid_config", "render has unknown fields", fields=unknown) @@ -65,14 +67,16 @@ def load_render_config( view_documents = document.get("views") if not isinstance(view_documents, list) or not view_documents: raise DocForgeError("invalid_config", "render.views must contain at least one view") - if len(view_documents) > limits.max_render_views: + view_values = cast(list[object], view_documents) + if len(view_values) > limits.max_render_views: raise DocForgeError("invalid_config", "render.views exceeds the configured limit") views: list[RenderView] = [] view_ids: set[str] = set() output_paths: set[Path] = set() - for view_document in view_documents: - if not isinstance(view_document, dict): + for view_value in view_values: + if not isinstance(view_value, dict): raise DocForgeError("invalid_config", "Each render view must be a table") + view_document = cast(dict[str, object], view_value) unknown_view = sorted(set(view_document) - _VIEW_KEYS) if unknown_view: raise DocForgeError( diff --git a/src/docforge/rendering.py b/src/docforge/rendering.py index 8058bb8..4524d10 100644 --- a/src/docforge/rendering.py +++ b/src/docforge/rendering.py @@ -6,7 +6,7 @@ import fcntl import hashlib import os import tempfile -from collections.abc import Callable, Iterator +from collections.abc import Callable, Generator from contextlib import contextmanager from pathlib import Path @@ -279,7 +279,7 @@ class RenderService: } @contextmanager - def _lock(self) -> Iterator[None]: + def _lock(self) -> Generator[None]: root = self.project.descriptor.cache_root if root.resolve(strict=False) != root: raise DocForgeError("path_escape", "Render lock directory is not safe") diff --git a/src/docforge/visualization.py b/src/docforge/visualization.py index c059e10..cff7cf1 100644 --- a/src/docforge/visualization.py +++ b/src/docforge/visualization.py @@ -18,6 +18,7 @@ from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from time import monotonic +from typing import cast from .errors import DocForgeError from .index import APPLICATION_ID, INDEX_SCHEMA_VERSION, ProjectIndex, re_tokenize @@ -54,21 +55,35 @@ class VisualizationIndexSnapshot: self.max_query_chars = index.project.descriptor.limits.max_query_chars self.max_results = index.project.descriptor.limits.max_results self.max_depth = index.project.descriptor.limits.max_traversal_depth - self.identity = {key: checked[key] for key in self._IDENTITY_KEYS} + self.identity: dict[str, object] = {key: checked[key] for key in self._IDENTITY_KEYS} self._stat = self._safe_stat() @classmethod def from_spec(cls, spec: dict[str, object]) -> VisualizationIndexSnapshot: snapshot = cls.__new__(cls) - snapshot.path = Path(str(spec["path"])) - snapshot.title = str(spec["title"]) - snapshot.max_query_chars = int(spec["max_query_chars"]) - snapshot.max_results = int(spec["max_results"]) - snapshot.max_depth = int(spec["max_depth"]) + path = spec["path"] + title = spec["title"] + max_query_chars = spec["max_query_chars"] + max_results = spec["max_results"] + max_depth = spec["max_depth"] + if ( + not isinstance(path, str) + or not isinstance(title, str) + or type(max_query_chars) is not int + or type(max_results) is not int + or type(max_depth) is not int + ): + raise DocForgeError("invalid_index", "Visualization snapshot is invalid") + snapshot.path = Path(path) + snapshot.title = title + snapshot.max_query_chars = max_query_chars + snapshot.max_results = max_results + snapshot.max_depth = max_depth identity = spec["identity"] if not isinstance(identity, dict): raise DocForgeError("invalid_index", "Visualization identity is invalid") - snapshot.identity = {key: identity[key] for key in cls._IDENTITY_KEYS} + typed_identity = cast(dict[str, object], identity) + snapshot.identity = {key: typed_identity[key] for key in cls._IDENTITY_KEYS} snapshot._stat = snapshot._safe_stat() return snapshot @@ -131,7 +146,7 @@ class VisualizationIndexSnapshot: + " ORDER BY rank, nodes.node_id LIMIT ?", values, ).fetchall() - results = [] + results: list[dict[str, object]] = [] for row in rows: item = _node_dict(row, include_content=False) item.update({"rank": row["rank"], "snippet": row["snippet"]}) @@ -294,6 +309,9 @@ class VisualizationIndexSnapshot: **payload, } + def result(self, **payload: object) -> dict[str, object]: + return self._result(**payload) + class VisualizationRunner: """Start one token-protected loopback reader for one immutable project binding.""" @@ -387,7 +405,8 @@ class VisualizationRunner: def do_DELETE(self) -> None: # noqa: N802 self.do_POST() - def log_message(self, _format: str, *_args: object) -> None: + def log_message(self, format: str, *args: object) -> None: + del format, args return self._server = _VisualizationHttpServer(("127.0.0.1", 0), Handler) @@ -514,7 +533,7 @@ class VisualizationRunner: if parsed.path == f"{prefix}/api/overview": payload = reader.overview() elif parsed.path == f"{prefix}/api/heartbeat": - payload = reader._result( + payload = reader.result( viewer="alive", lease_seconds=self.lease_seconds, ) @@ -761,7 +780,7 @@ class DetachedVisualizationRunner: "visualization_unavailable", "The detached visualization worker rejected the snapshot", ) - return response["visualization"] + return cast(dict[str, object], response["visualization"]) def stop(self) -> None: process = self._process @@ -793,7 +812,7 @@ def _receive_worker_response(control: socket.socket) -> dict[str, object]: result = json.loads(payload) if not isinstance(result, dict): raise ValueError("Visualization worker returned an invalid response") - return result + return cast(dict[str, object], result) def _one(params: dict[str, list[str]], name: str) -> str: diff --git a/src/docforge/visualization_worker.py b/src/docforge/visualization_worker.py index 260d6a0..b2648dc 100644 --- a/src/docforge/visualization_worker.py +++ b/src/docforge/visualization_worker.py @@ -8,7 +8,7 @@ import os import socket import time from contextlib import suppress -from typing import Any +from typing import cast from .errors import DocForgeError from .visualization import VisualizationRunner @@ -20,7 +20,7 @@ def _parser() -> argparse.ArgumentParser: return parser -def _read_request(control: socket.socket) -> dict[str, Any]: +def _read_request(control: socket.socket) -> dict[str, object]: chunks: list[bytes] = [] size = 0 while True: @@ -37,7 +37,7 @@ def _read_request(control: socket.socket) -> dict[str, Any]: request = json.loads(payload) if not isinstance(request, dict): raise ValueError("Visualization launch request is invalid") - return request + return cast(dict[str, object], request) def _pid_exists(pid: int) -> bool: @@ -59,21 +59,45 @@ def main(argv: list[str] | None = None) -> int: try: request = _read_request(control) target = request["target"] - if not isinstance(target, dict): + snapshot = request["snapshot"] + token = request["token"] + initial_grace = request["initial_grace_seconds"] + lease = request["lease_seconds"] + monitor_interval = request["monitor_interval_seconds"] + owner = request["owner_pid"] + if not isinstance(target, dict) or not isinstance(snapshot, dict): raise ValueError("Visualization target is invalid") + target = cast(dict[str, object], target) + node_id = target.get("node_id") + query = target.get("query") + depth = target.get("depth") + if ( + (node_id is not None and not isinstance(node_id, str)) + or (query is not None and not isinstance(query, str)) + or type(depth) is not int + or not isinstance(token, str) + or not isinstance(initial_grace, int | float) + or isinstance(initial_grace, bool) + or not isinstance(lease, int | float) + or isinstance(lease, bool) + or not isinstance(monitor_interval, int | float) + or isinstance(monitor_interval, bool) + or type(owner) is not int + ): + raise ValueError("Visualization launch request is invalid") runner = VisualizationRunner( None, - snapshot_spec=request["snapshot"], - token=str(request["token"]), + snapshot_spec=cast(dict[str, object], snapshot), + token=token, register_atexit=False, - initial_grace_seconds=float(request["initial_grace_seconds"]), - lease_seconds=float(request["lease_seconds"]), - monitor_interval_seconds=float(request["monitor_interval_seconds"]), + initial_grace_seconds=float(initial_grace), + lease_seconds=float(lease), + monitor_interval_seconds=float(monitor_interval), ) visualization = runner.start( - node_id=target.get("node_id"), - query=target.get("query"), - depth=int(target["depth"]), + node_id=node_id, + query=query, + depth=depth, ) control.sendall( json.dumps( @@ -83,7 +107,7 @@ def main(argv: list[str] | None = None) -> int: ).encode("utf-8") + b"\n" ) - owner_pid = int(request["owner_pid"]) + owner_pid = owner except (DocForgeError, KeyError, OSError, TypeError, ValueError) as error: with suppress(OSError): control.sendall(