administrator/DocForge2
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Enforce strict Pyright gate

Browse source
This commit is contained in:
Andraxion 2026-07-24 22:26:01 -04:00
parent 90898cfd63
commit 2841042b9c
16 changed files with 214 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

3
AGENTS.md
View file

@ -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.

6
README.md
View file

@ -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

15
SLICE_HISTORY.md
View file

@ -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

2
pyproject.toml
View file

@ -30,3 +30,5 @@ select = ["E", "F", "I", "UP", "B", "SIM"]
pythonVersion = "3.12"
typeCheckingMode = "strict"
include = ["src"]
venvPath = "."
venv = ".venv"

4
src/docforge/adapter_contract.py
View file

@ -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")

53
src/docforge/changeset_contract.py
View file

@ -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"
)

4
src/docforge/changesets.py
View file

@ -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:

16
src/docforge/config_validation.py
View file

@ -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(

3
src/docforge/context.py
View file

@ -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
):

30
src/docforge/mcp_server.py
View file

@ -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

30
src/docforge/project.py
View file

@ -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(

18
src/docforge/proposal_projection.py
View file

@ -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,

10
src/docforge/render_config.py
View file

@ -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(

4
src/docforge/rendering.py
View file

@ -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")

43
src/docforge/visualization.py
View file

@ -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:

50
src/docforge/visualization_worker.py
View file

@ -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(