Merge pull request #1400 from Open-Source-Legal/claude/resolve-issue-1335-cRUxc

typing: annotate analyzer/shared/agents/badges/worker_uploads + shared protocols (#1335)

Author: JSv4

CHANGELOG.md

@@ -9,6 +9,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Mypy: type analyzer, shared, agents, badges, worker_uploads; introduce shared protocols** (Issue #1335): Brought the five smaller, interface-rich target packages over the ≥70% return-annotation bar called for by the issue and seeded `opencontractserver/types/protocols.py` with the four protocols requested in the scope:
+  - `VectorStoreProtocol` — minimum surface (`search` / `async_search`) implemented by `CoreAnnotationVectorStore` (`opencontractserver/llms/vector_stores/core_vector_stores.py`); imported and re-exported from that module so consumers can annotate against the protocol rather than the concrete dataclass.
+  - `PipelineComponentProtocol` — `title` / `description` / `author` / `dependencies` surface that the pipeline registry duck-types against; imported from `opencontractserver/pipeline/base/base_component.py` so any future parser/embedder/thumbnailer registered outside the inheritance hierarchy still type-checks against the same contract.
+  - `ToolProtocol` — `name` / `description` / `parameters` / `requires_approval` mirror of `CoreTool` (`opencontractserver/llms/tools/tool_factory.py`); framework adapters can accept any object satisfying it, no inheritance required.
+  - `PermissionedQueryManagerProtocol` — `visible_to_user(user) -> QuerySet` contract that `BaseVisibilityManager`, `PermissionManager`, `DocumentManager`, `AnnotationManager`, and `NoteManager` all satisfy (`opencontractserver/shared/Managers.py`); imported there so callers receiving "a permissioned manager" can type against the protocol instead of a concrete class.
+  - `StreamObserverProtocol` — duplicated `__call__(event)` shape from `opencontractserver/llms/types.StreamObserver` for callers in non-LLM modules (notifications, websockets) that need the contract without importing `llms.types`.
+  - **Coverage delta** (return-annotation coverage measured by AST walk, excluding `__init__.py`):
+    - `analyzer/`: 12.5% → **87.5%** (target ≥70%)
+    - `shared/`: 38.1% → **96.3%** (target ≥70%)
+    - `agents/`: 34.4% → **71.9%** (target ≥70%)
+    - `badges/`: 47.1% → **100%** (target ≥70%)
+    - `worker_uploads/`: 46.4% → **100%** (target ≥70%)
+  - **Files touched** (annotations only, zero behavior changes): `analyzer/{apps,checks,signals,startup,utils,admin,admin_views}.py`, `analyzer/management/commands/sync_doc_analyzers.py`, `agents/{apps,admin,memory}.py`, `badges/{apps,signals,models}.py`, `worker_uploads/{apps,auth,serializers,views,models,tasks}.py`, `shared/{utils,defaults,fields,mixins,Managers,QuerySets,decorators}.py`, plus the four protocol-consumer wirings above.
+  - **Bare-generic promotion**: every `dict` / `list` / `set` in newly-touched public signatures was widened to a parametrised form (`dict[str, Any]`, `list[Any]`, `dict[str, int]`, etc.). The `HasEmbeddingMixin` docstring example was tightened to match (`-> dict[str, Any]`).
+  - **`# type: ignore` audit**: every bare `# type: ignore` comment in the codebase now carries a specific error code. `opencontractserver/llms/tools/core_tools.py:413,416,418,420` (channels/asgiref import + `partial` kwarg-aware wrapper) tightened to `[import-not-found]` / `[call-arg]` with a one-line comment explaining why; `opencontractserver/utils/embeddings.py:171` to `[attr-defined]`; `opencontractserver/tests/test_pipeline_utils.py:395,423,431,438,446` cleaned up the duplicated `# type: ignore; type: ignore` markers and scoped them to `[import-not-found]`; `opencontractserver/tests/test_core_tool_factory.py:34,35,47` widened from bare to `[attr-defined]`. The total count went from 62 → 61, and the ratio of bare-to-scoped dropped to zero.
 - **Return-type annotations across `config/graphql/` resolvers and mutations** (Issue #1332, follow-up to #1331): The largest, least-typed subtree in the backend (459 function definitions, ~4.4% return-annotation coverage at baseline) is now at 100% return-annotation coverage. Touched files include every `*_mutations.py`, every `*_queries.py`, every `*_types.py`, plus `filters.py`, `base.py`, `base_types.py`, `security.py`, `optimized_file_resolvers.py`, `permissioning/permission_annotator/{middleware,mixins,utils}.py`, and the small utility modules. No behavioral changes — annotations only.
   - **`mutate(...)` on `graphene.Mutation` subclasses**: typed as forward references to the enclosing class (`-> "ClassName"`). Discovered and fixed the latent bug in `config/graphql/analysis_mutations.py:179` (`DeleteAnalysisMutation.mutate`) where the success path had no `return` statement; annotation is `-> "DeleteAnalysisMutation | None"` and an explicit `return None` was added to preserve the original implicit-None behavior on success.
   - **`resolve_*` methods**: typed as `-> Any` by default, refined where the GraphQL field type makes the runtime return obvious (e.g. `resolve_in_use -> bool`, `resolve_datacell_count -> int`).

opencontractserver/agents/admin.py

@@ -1,4 +1,6 @@
 from django.contrib import admin
+from django.db.models import QuerySet
+from django.http import HttpRequest
 from django.utils.html import format_html
 from guardian.admin import GuardedModelAdmin
 
@@ -146,7 +148,7 @@ class AgentActionResultAdmin(GuardedModelAdmin):
         ),
     )
 
-    def status_badge(self, obj):
+    def status_badge(self, obj: AgentActionResult) -> str:
         """Display status as a colored badge."""
         colors = {
             "pending": "#6c757d",  # gray
@@ -165,7 +167,7 @@ def status_badge(self, obj):
     status_badge.short_description = "Status"
     status_badge.admin_order_field = "status"
 
-    def corpus_action_link(self, obj):
+    def corpus_action_link(self, obj: AgentActionResult) -> str:
         """Link to the corpus action in admin."""
         if obj.corpus_action:
             name = (
@@ -182,7 +184,7 @@ def corpus_action_link(self, obj):
 
     corpus_action_link.short_description = "Corpus Action"
 
-    def document_link(self, obj):
+    def document_link(self, obj: AgentActionResult) -> str:
         """Link to the document in admin."""
         if obj.document:
             title = (
@@ -199,7 +201,7 @@ def document_link(self, obj):
 
     document_link.short_description = "Document"
 
-    def tools_count(self, obj):
+    def tools_count(self, obj: AgentActionResult) -> str:
         """Display count of tools executed."""
         if obj.tools_executed:
             count = len(obj.tools_executed)
@@ -212,7 +214,7 @@ def tools_count(self, obj):
 
     tools_count.short_description = "Tools"
 
-    def duration_display(self, obj):
+    def duration_display(self, obj: AgentActionResult) -> str:
         """Display execution duration."""
         duration = obj.duration_seconds
         if duration is not None:
@@ -228,7 +230,7 @@ def duration_display(self, obj):
 
     duration_display.short_description = "Duration"
 
-    def agent_response_display(self, obj):
+    def agent_response_display(self, obj: AgentActionResult) -> str:
         """Display agent response with formatting."""
         if obj.agent_response:
             # Truncate very long responses in admin
@@ -240,7 +242,7 @@ def agent_response_display(self, obj):
 
     agent_response_display.short_description = "Agent Response"
 
-    def tools_executed_display(self, obj):
+    def tools_executed_display(self, obj: AgentActionResult) -> str:
         """Display tools executed as formatted JSON."""
         if obj.tools_executed:
             import json
@@ -254,7 +256,7 @@ def tools_executed_display(self, obj):
 
     tools_executed_display.short_description = "Tools Executed"
 
-    def execution_metadata_display(self, obj):
+    def execution_metadata_display(self, obj: AgentActionResult) -> str:
         """Display execution metadata as formatted JSON."""
         if obj.execution_metadata:
             import json
@@ -265,7 +267,7 @@ def execution_metadata_display(self, obj):
 
     execution_metadata_display.short_description = "Execution Metadata"
 
-    def get_queryset(self, request):
+    def get_queryset(self, request: HttpRequest) -> QuerySet[AgentActionResult]:
         """Optimize queryset with select_related."""
         return (
             super()

opencontractserver/agents/apps.py

@@ -3,6 +3,6 @@
 
 
 class AgentsConfig(AppConfig):
-    default_auto_field = "django.db.models.BigAutoField"
-    name = "opencontractserver.agents"
+    default_auto_field: str = "django.db.models.BigAutoField"
+    name: str = "opencontractserver.agents"
     verbose_name = _("Agents")

opencontractserver/agents/memory.py

@@ -11,7 +11,7 @@
 import logging
 import re
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from channels.db import database_sync_to_async
 from django.core.files.base import ContentFile
@@ -74,7 +74,7 @@ def _build_empty_memory(corpus_id: int) -> str:
 # ---------------------------------------------------------------------------
 
 
-async def get_or_create_memory_document(corpus: Corpus, user) -> Document:
+async def get_or_create_memory_document(corpus: Corpus, user: Any) -> Document:
     """Get the existing memory document or create a new empty one.
 
     If the corpus already has a valid ``memory_document``, return it.
@@ -95,7 +95,7 @@ async def get_or_create_memory_document(corpus: Corpus, user) -> Document:
 
     from opencontractserver.corpuses.models import Corpus as CorpusModel
 
-    def _get_or_create_sync():
+    def _get_or_create_sync() -> Document:
         # Phase 1: Check under lock whether a memory document already exists.
         with transaction.atomic():
             locked = CorpusModel.objects.select_for_update().get(pk=corpus.pk)
@@ -164,7 +164,7 @@ async def read_memory_content(corpus: Corpus) -> str:
     if not corpus.memory_document_id:
         return ""
 
-    def _read():
+    def _read() -> str:
         doc = corpus.memory_document
         if doc is None or not doc.txt_extract_file:
             return ""
@@ -187,7 +187,9 @@ def _read():
     return await database_sync_to_async(_read)()
 
 
-async def update_memory_content(corpus: Corpus, new_content: str, user) -> Document:
+async def update_memory_content(
+    corpus: Corpus, new_content: str, user: Any
+) -> Document:
     """Update the memory document with new content.
 
     Creates the memory document if it doesn't exist.  Overwrites the
@@ -215,7 +217,7 @@ async def update_memory_content(corpus: Corpus, new_content: str, user) -> Docum
 
     doc = await get_or_create_memory_document(corpus, user)
 
-    def _update():
+    def _update() -> Document:
         from django.db import transaction
 
         with transaction.atomic():
@@ -431,7 +433,7 @@ def merge_curation_into_memory(
     current_content: str,
     collection_patterns: list[str],
     query_patterns: list[str],
-    refinements: list[dict],
+    refinements: list[dict[str, Any]],
 ) -> str:
     """Merge curation results into the existing memory document.
 

opencontractserver/analyzer/admin.py

@@ -1,5 +1,5 @@
 from django.contrib import admin
-from django.urls import path
+from django.urls import URLPattern, path
 from guardian.admin import GuardedModelAdmin
 
 from opencontractserver.analyzer.admin_views import AnalyzerSyncView
@@ -16,7 +16,7 @@ class AnalyzerAdmin(GuardedModelAdmin):
     list_display = ["id", "description", "task_name", "host_gremlin"]
     change_list_template = "admin/analyzer/analyzer_changelist.html"
 
-    def get_urls(self):
+    def get_urls(self) -> list[URLPattern]:
         urls = super().get_urls()
         custom_urls = [
             path(

opencontractserver/analyzer/admin_views.py

@@ -1,5 +1,8 @@
+from typing import Any
+
 from django.contrib import messages
 from django.contrib.admin.views.decorators import staff_member_required
+from django.http import HttpRequest, HttpResponse
 from django.shortcuts import redirect, render
 from django.urls import reverse
 from django.utils.decorators import method_decorator
@@ -17,11 +20,11 @@
 class AnalyzerSyncView(View):
     """Custom admin view for syncing doc analyzer tasks"""
 
-    template_name = "admin/analyzer/analyzer_sync.html"
+    template_name: str = "admin/analyzer/analyzer_sync.html"
 
-    def get_available_analyzers(self):
+    def get_available_analyzers(self) -> list[dict[str, Any]]:
         """Get info about all available doc analyzer tasks"""
-        analyzers = []
+        analyzers: list[dict[str, Any]] = []
 
         for task_name in celery_app.tasks.keys():
             analyzer_task = get_doc_analyzer_task_by_name(task_name)
@@ -48,8 +51,8 @@ def get_available_analyzers(self):
 
         return sorted(analyzers, key=lambda x: (x["exists"], x["task_name"]))
 
-    def get(self, request):
-        context = {
+    def get(self, request: HttpRequest) -> HttpResponse:
+        context: dict[str, Any] = {
             "title": "Sync Doc Analyzer Tasks",
             "analyzers": self.get_available_analyzers(),
             "opts": Analyzer._meta,
@@ -59,7 +62,7 @@ def get(self, request):
         }
         return render(request, self.template_name, context)
 
-    def post(self, request):
+    def post(self, request: HttpRequest) -> HttpResponse:
         if not request.user.has_perm("analyzer.add_analyzer"):
             messages.error(request, "You don't have permission to create analyzers.")
             return redirect(reverse("admin:analyzer_sync"))

opencontractserver/analyzer/apps.py

@@ -3,10 +3,10 @@
 
 
 class AnnotationsConfig(AppConfig):
-    default_auto_field = "django.db.models.BigAutoField"
-    name = "opencontractserver.analyzer"
+    default_auto_field: str = "django.db.models.BigAutoField"
+    name: str = "opencontractserver.analyzer"
 
-    def ready(self):
+    def ready(self) -> None:
         try:
             import opencontractserver.analyzer.signals  # noqa F401
             from opencontractserver.analyzer.models import GremlinEngine

opencontractserver/analyzer/checks.py

@@ -1,13 +1,15 @@
+from typing import Any
+
 from django.core.checks import Warning, register
 
 
 @register()
-def check_unsynced_analyzers(app_configs, **kwargs):
+def check_unsynced_analyzers(app_configs: Any, **kwargs: Any) -> list[Warning]:
     """
     Check if there are doc_analyzer_task decorated functions
     that haven't been synced to the database.
     """
-    warnings = []
+    warnings: list[Warning] = []
 
     try:
         from opencontractserver.analyzer.models import Analyzer

opencontractserver/analyzer/management/commands/sync_doc_analyzers.py

@@ -1,5 +1,7 @@
+from typing import Any
+
 from django.contrib.auth import get_user_model
-from django.core.management.base import BaseCommand
+from django.core.management.base import BaseCommand, CommandParser
 
 from opencontractserver.analyzer.models import Analyzer
 from opencontractserver.analyzer.utils import auto_create_doc_analyzers
@@ -8,14 +10,14 @@
 class Command(BaseCommand):
     help = "Synchronize doc_analyzer_task decorated functions with Analyzer database"
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument(
             "--dry-run",
             action="store_true",
             help="Show what would be created without making changes",
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args: Any, **options: Any) -> None:
         UserModel = get_user_model()
 
         if options["dry_run"]:

opencontractserver/analyzer/signals.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 from celery import chain
 from django.db import transaction
@@ -11,8 +12,9 @@
 logger = logging.getLogger(__name__)
 
 
-def install_gremlin_on_creation(sender, instance, created, **kwargs):
-
+def install_gremlin_on_creation(
+    sender: Any, instance: Any, created: bool, **kwargs: Any
+) -> None:
     # When we create a Gremlin object in DB, need to run async task to set it up
     if created:
         transaction.on_commit(
@@ -27,7 +29,7 @@ def install_gremlin_on_creation(sender, instance, created, **kwargs):
         )
 
 
-def handle_analysis_completion(sender, instance, **kwargs):
+def handle_analysis_completion(sender: Any, instance: Any, **kwargs: Any) -> None:
     """Handle analysis completion."""
     if hasattr(instance, "status") and instance.status == "COMPLETE":
         logger.info(f"Analysis {instance.id} completed")