refactor(typing): Replace Any with precise types across codebase

conftest.py

@@ -24,7 +24,7 @@
 @pytest.fixture(autouse=True)
 def add_doctest_fixtures(
     request: pytest.FixtureRequest,
-    doctest_namespace: dict[str, t.Any],
+    doctest_namespace: dict[str, object],
 ) -> None:
     """Harness pytest fixtures to doctests namespace."""
     from _pytest.doctest import DoctestItem

src/vcspull/_internal/private_path.py

@@ -34,7 +34,11 @@ class PrivatePath(PrivatePathBase):
     '~/notes.txt'
     """
 
-    def __new__(cls, *args: t.Any, **kwargs: t.Any) -> PrivatePath:
+    def __new__(
+        cls,
+        *args: str | os.PathLike[str],
+        **kwargs: object,
+    ) -> PrivatePath:
         return super().__new__(cls, *args, **kwargs)
 
     @classmethod

src/vcspull/cli/__init__.py

@@ -28,6 +28,20 @@
 log = logging.getLogger(__name__)
 
 
+SubparserTuple: t.TypeAlias = tuple[
+    argparse.ArgumentParser,  # sync
+    argparse.ArgumentParser,  # list
+    argparse.ArgumentParser,  # status
+    argparse.ArgumentParser,  # search
+    argparse.ArgumentParser,  # add
+    argparse.ArgumentParser,  # discover
+    argparse.ArgumentParser,  # fmt
+    argparse.ArgumentParser,  # migrate
+    argparse.ArgumentParser,  # import
+    argparse.ArgumentParser,  # worktree
+]
+
+
 def build_description(
     intro: str,
     example_blocks: t.Sequence[tuple[str | None, t.Sequence[str]]],
@@ -323,7 +337,7 @@ def build_description(
 @t.overload
 def create_parser(
     return_subparsers: t.Literal[True],
-) -> tuple[argparse.ArgumentParser, t.Any]: ...
+) -> tuple[argparse.ArgumentParser, SubparserTuple]: ...
 
 
 @t.overload
@@ -332,7 +346,7 @@ def create_parser(return_subparsers: t.Literal[False]) -> argparse.ArgumentParse
 
 def create_parser(
     return_subparsers: bool = False,
-) -> argparse.ArgumentParser | tuple[argparse.ArgumentParser, t.Any]:
+) -> argparse.ArgumentParser | tuple[argparse.ArgumentParser, SubparserTuple]:
     """Create CLI argument parser for vcspull."""
     parser = argparse.ArgumentParser(
         prog="vcspull",

src/vcspull/cli/_output.py

@@ -8,6 +8,72 @@
 from dataclasses import dataclass, field
 from enum import Enum
 
+JsonPrimitive: t.TypeAlias = str | int | float | bool | None
+JsonValue: t.TypeAlias = JsonPrimitive | dict[str, "JsonValue"] | list["JsonValue"]
+JsonObject: t.TypeAlias = t.Mapping[str, JsonValue]
+OutputPayload: t.TypeAlias = "JsonObject | PlanEntryPayload | PlanSummaryPayload"
+
+
+class _PlanEntryPayloadRequired(t.TypedDict):
+    """Required keys for a plan-entry JSON payload."""
+
+    format_version: str
+    type: str
+    name: str
+    path: str
+    workspace_root: str
+    action: str
+
+
+class PlanEntryPayload(_PlanEntryPayloadRequired, total=False):
+    """Typed JSON payload for a plan entry (optional keys omitted when unset)."""
+
+    detail: str
+    url: str
+    branch: str
+    remote_branch: str
+    current_rev: str
+    target_rev: str
+    ahead: int
+    behind: int
+    dirty: bool
+    error: str
+    diagnostics: list[str]
+
+
+class _PlanSummaryPayloadRequired(t.TypedDict):
+    """Required keys for a plan-summary JSON payload."""
+
+    format_version: str
+    type: str
+    clone: int
+    update: int
+    unchanged: int
+    blocked: int
+    errors: int
+    total: int
+
+
+class PlanSummaryPayload(_PlanSummaryPayloadRequired, total=False):
+    """Typed JSON payload for a plan summary (optional keys omitted when unset)."""
+
+    duration_ms: int
+
+
+class PlanWorkspacePayload(t.TypedDict):
+    """Typed JSON payload for a workspace grouping."""
+
+    path: str
+    operations: list[PlanEntryPayload]
+
+
+class PlanResultPayload(t.TypedDict):
+    """Typed JSON payload for a plan result."""
+
+    format_version: str
+    workspaces: list[PlanWorkspacePayload]
+    summary: PlanSummaryPayload
+
 
 class OutputMode(Enum):
     """Output format modes."""
@@ -47,7 +113,7 @@ class PlanEntry:
     error: str | None = None
     diagnostics: list[str] = field(default_factory=list)
 
-    def to_payload(self) -> dict[str, t.Any]:
+    def to_payload(self) -> PlanEntryPayload:
         """Convert the plan entry into a serialisable payload.
 
         Examples
@@ -67,7 +133,7 @@ def to_payload(self) -> dict[str, t.Any]:
         >>> payload["format_version"]
         '1'
         """
-        payload: dict[str, t.Any] = {
+        payload: PlanEntryPayload = {
             "format_version": "1",
             "type": "operation",
             "name": self.name,
@@ -122,7 +188,7 @@ def total(self) -> int:
         """
         return self.clone + self.update + self.unchanged + self.blocked + self.errors
 
-    def to_payload(self) -> dict[str, t.Any]:
+    def to_payload(self) -> PlanSummaryPayload:
         """Convert the summary to a serialisable payload.
 
         Examples
@@ -134,7 +200,7 @@ def to_payload(self) -> dict[str, t.Any]:
         >>> payload["total"]
         3
         """
-        payload: dict[str, t.Any] = {
+        payload: PlanSummaryPayload = {
             "format_version": "1",
             "type": "summary",
             "clone": self.clone,
@@ -167,6 +233,29 @@ class PlanResult:
     entries: list[PlanEntry]
     summary: PlanSummary
 
+    def to_workspace_mapping(self) -> dict[str, list[PlanEntry]]:
+        """Group plan entries by workspace root."""
+        grouped: dict[str, list[PlanEntry]] = {}
+        for entry in self.entries:
+            grouped.setdefault(entry.workspace_root, []).append(entry)
+        return grouped
+
+    def to_json_object(self) -> PlanResultPayload:
+        """Return the JSON structure for ``--json`` output."""
+        workspaces: list[PlanWorkspacePayload] = []
+        for workspace_root, entries in self.to_workspace_mapping().items():
+            workspaces.append(
+                {
+                    "path": workspace_root,
+                    "operations": [entry.to_payload() for entry in entries],
+                },
+            )
+        return {
+            "format_version": "1",
+            "workspaces": workspaces,
+            "summary": self.summary.to_payload(),
+        }
+
 
 class OutputFormatter:
     """Manages output formatting for different modes (human, JSON, NDJSON)."""
@@ -186,17 +275,18 @@ def __init__(self, mode: OutputMode = OutputMode.HUMAN) -> None:
         <OutputMode.JSON: 'json'>
         """
         self.mode = mode
-        self._json_buffer: list[dict[str, t.Any]] = []
+        self._json_buffer: list[OutputPayload] = []
 
-    def emit(self, data: dict[str, t.Any] | PlanEntry | PlanSummary) -> None:
+    def emit(self, data: OutputPayload | PlanEntry | PlanSummary) -> None:
         """Emit a data event.
 
         Parameters
         ----------
-        data : dict | PlanEntry | PlanSummary
+        data : OutputPayload | PlanEntry | PlanSummary
             Event data to emit. PlanEntry and PlanSummary instances are serialised
             automatically.
         """
+        payload: OutputPayload
         if isinstance(data, (PlanEntry, PlanSummary)):
             payload = data.to_payload()
         else:

src/vcspull/cli/add.py

@@ -37,6 +37,13 @@
 log = logging.getLogger(__name__)
 
 
+class _OrderedItem(t.TypedDict):
+    """Ordered config entry preserving label/section pairs."""
+
+    label: str
+    section: object
+
+
 class AddAction(enum.Enum):
     """Action resolved for a single repo during ``vcspull add``."""
 
@@ -251,37 +258,40 @@ def _normalize_detected_url(remote: str | None) -> tuple[str, str]:
 
 
 def _build_ordered_items(
-    top_level_items: list[tuple[str, t.Any]] | None,
-    raw_config: dict[str, t.Any],
-) -> list[dict[str, t.Any]]:
+    top_level_items: list[tuple[str, object]] | None,
+    raw_config: dict[str, object],
+) -> list[_OrderedItem]:
     """Return deep-copied top-level items preserving original ordering."""
-    source: list[tuple[str, t.Any]] = top_level_items or list(raw_config.items())
+    source: list[tuple[str, object]] = top_level_items or list(raw_config.items())
 
-    ordered: list[dict[str, t.Any]] = []
+    ordered: list[_OrderedItem] = []
     for label, section in source:
         ordered.append({"label": label, "section": copy.deepcopy(section)})
     return ordered
 
 
 def _aggregate_from_ordered_items(
-    items: list[dict[str, t.Any]],
-) -> dict[str, t.Any]:
+    items: list[_OrderedItem],
+) -> dict[str, object]:
     """Collapse ordered top-level items into a mapping grouped by label."""
-    aggregated: dict[str, t.Any] = {}
+    aggregated: dict[str, object] = {}
     for entry in items:
         label = entry["label"]
         section = entry["section"]
         if isinstance(section, dict):
-            workspace_section = aggregated.setdefault(label, {})
+            workspace_section = t.cast(
+                "dict[str, object]",
+                aggregated.setdefault(label, {}),
+            )
             for repo_name, repo_config in section.items():
-                workspace_section[repo_name] = copy.deepcopy(repo_config)
+                workspace_section[str(repo_name)] = copy.deepcopy(repo_config)
         else:
             aggregated[label] = copy.deepcopy(section)
     return aggregated
 
 
 def _collapse_ordered_items_to_dict(
-    ordered_items: list[dict[str, t.Any]],
+    ordered_items: list[_OrderedItem],
 ) -> dict[str, t.Any]:
     """Collapse ordered items into a flat dict for JSON serialization.
 
@@ -326,10 +336,10 @@ def _collapse_ordered_items_to_dict(
 
 
 def _collect_duplicate_sections(
-    items: list[dict[str, t.Any]],
-) -> dict[str, list[t.Any]]:
+    items: list[_OrderedItem],
+) -> dict[str, list[object]]:
     """Return mapping of labels to their repeated sections (>= 2 occurrences)."""
-    occurrences: dict[str, list[t.Any]] = {}
+    occurrences: dict[str, list[object]] = {}
     for entry in items:
         label = entry["label"]
         occurrences.setdefault(label, []).append(copy.deepcopy(entry["section"]))
@@ -341,7 +351,7 @@ def _collect_duplicate_sections(
 
 def _save_ordered_items(
     config_file_path: pathlib.Path,
-    ordered_items: list[dict[str, t.Any]],
+    ordered_items: list[_OrderedItem],
 ) -> None:
     """Persist ordered items in the format matching the config file extension.
 
@@ -610,9 +620,9 @@ def add_repo(
             config_file_path = home_configs[0]
 
     # Load existing config
-    raw_config: dict[str, t.Any]
-    duplicate_root_occurrences: dict[str, list[t.Any]]
-    top_level_items: list[tuple[str, t.Any]]
+    raw_config: dict[str, object]
+    duplicate_root_occurrences: dict[str, list[object]]
+    top_level_items: list[tuple[str, object]]
     display_config_path = str(PrivatePath(config_file_path))
 
     if config_file_path.exists() and config_file_path.is_file():
@@ -666,7 +676,7 @@ def add_repo(
     new_repo_entry = build_repo_entry(url, rev=rev, shallow=shallow, depth=depth)
 
     def _ensure_workspace_label_for_merge(
-        config_data: dict[str, t.Any],
+        config_data: dict[str, object],
     ) -> tuple[str, bool]:
         workspace_map: dict[pathlib.Path, str] = {}
         for label, section in config_data.items():
@@ -700,7 +710,7 @@ def _ensure_workspace_label_for_merge(
         return workspace_label, relabelled
 
     def _prepare_no_merge_items(
-        items: list[dict[str, t.Any]],
+        items: list[_OrderedItem],
     ) -> tuple[str, int, bool]:
         matching_indexes: list[int] = []
         for idx, entry in enumerate(items):

src/vcspull/cli/discover.py

@@ -416,8 +416,8 @@ def discover_repos(
     config_scope = _classify_config_scope(config_file_path, cwd=cwd, home=home)
     allow_relative_workspace = config_scope == "project"
 
-    raw_config: dict[str, t.Any]
-    duplicate_root_occurrences: dict[str, list[t.Any]]
+    raw_config: dict[str, object]
+    duplicate_root_occurrences: dict[str, list[object]]
     if config_file_path.exists() and config_file_path.is_file():
         try:
             (
@@ -844,18 +844,21 @@ def discover_repos(
             )
             workspace_map[workspace_path] = workspace_label
 
-        if workspace_label not in raw_config:
-            raw_config[workspace_label] = {}
-        elif not isinstance(raw_config[workspace_label], dict):
+        section = raw_config.get(workspace_label)
+        if section is None:
+            section = {}
+            raw_config[workspace_label] = section
+        elif not isinstance(section, dict):
             log.warning(
                 "Workspace root '%s' in config is not a dictionary. Skipping repo %s.",
                 workspace_label,
                 repo_name,
             )
             continue
 
-        if repo_name not in raw_config[workspace_label]:
-            raw_config[workspace_label][repo_name] = build_repo_entry(
+        workspace_section = t.cast("dict[str, object]", raw_config[workspace_label])
+        if repo_name not in workspace_section:
+            workspace_section[repo_name] = build_repo_entry(
                 repo_url,
                 rev=rev,
                 shallow=repo_shallow,