From c5ee50ff0bcaccb144aa6f0d1f5359eca5555932 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Wed, 4 Mar 2026 01:13:42 -0500 Subject: [PATCH] feat(types): Resolve strict mypy errors in conductor subsystem --- .../plan.md | 10 +++++----- conductor_tech_lead.py | 12 ++++++------ dag_engine.py | 2 +- orchestrator_pm.py | 9 +++++---- 4 files changed, 17 insertions(+), 16 deletions(-) diff --git a/conductor/tracks/strict_static_analysis_and_typing_20260302/plan.md b/conductor/tracks/strict_static_analysis_and_typing_20260302/plan.md index 6016193..3ed09bb 100644 --- a/conductor/tracks/strict_static_analysis_and_typing_20260302/plan.md +++ b/conductor/tracks/strict_static_analysis_and_typing_20260302/plan.md @@ -15,11 +15,11 @@ - [x] WHAT: Add explicit type hints to all function arguments, return values, and complex dictionaries. Resolve `Any` bleeding. - [x] HOW: Surgical type annotations (`dict[str, Any]`, `list[str]`, etc.). - [x] SAFETY: Do not change runtime logic, only type signatures. -- [ ] Task: Resolve Conductor Subsystem Type Errors - - [ ] WHERE: `conductor_tech_lead.py`, `dag_engine.py`, `orchestrator_pm.py` - - [ ] WHAT: Enforce strict typing on track state, tickets, and DAG models. - - [ ] HOW: Standard python typing imports. - - [ ] SAFETY: Preserve JSON serialization compatibility. +- [x] Task: Resolve Conductor Subsystem Type Errors + - [x] WHERE: `conductor_tech_lead.py`, `dag_engine.py`, `orchestrator_pm.py` + - [x] WHAT: Enforce strict typing on track state, tickets, and DAG models. + - [x] HOW: Standard python typing imports. + - [x] SAFETY: Preserve JSON serialization compatibility. - [ ] Task: Conductor - User Manual Verification 'Phase 2: Core Library' (Protocol in workflow.md) ## Phase 3: GUI God-Object Typing Resolution diff --git a/conductor_tech_lead.py b/conductor_tech_lead.py index 1346e25..bddbee2 100644 --- a/conductor_tech_lead.py +++ b/conductor_tech_lead.py @@ -2,8 +2,9 @@ import json import ai_client import mma_prompts import re +from typing import Any -def generate_tickets(track_brief: str, module_skeletons: str) -> list[dict]: +def generate_tickets(track_brief: str, module_skeletons: str) -> list[dict[str, Any]]: """ Tier 2 (Tech Lead) call. Breaks down a Track Brief and module skeletons into discrete Tier 3 Tickets. @@ -18,7 +19,7 @@ def generate_tickets(track_brief: str, module_skeletons: str) -> list[dict]: ) # Set custom system prompt for this call old_system_prompt = ai_client._custom_system_prompt - ai_client.set_custom_system_prompt(system_prompt) + ai_client.set_custom_system_prompt(system_prompt or "") ai_client.current_tier = "Tier 2" try: # 3. Call Tier 2 Model @@ -38,20 +39,20 @@ def generate_tickets(track_brief: str, module_skeletons: str) -> list[dict]: match = re.search(r'\[\s*\{.*\}\s*\]', json_match, re.DOTALL) if match: json_match = match.group(0) - tickets = json.loads(json_match) + tickets: list[dict[str, Any]] = json.loads(json_match) return tickets except Exception as e: print(f"Error parsing Tier 2 response: {e}") return [] finally: # Restore old system prompt and clear tier tag - ai_client.set_custom_system_prompt(old_system_prompt) + ai_client.set_custom_system_prompt(old_system_prompt or "") ai_client.current_tier = None from dag_engine import TrackDAG from models import Ticket -def topological_sort(tickets: list[dict]) -> list[dict]: +def topological_sort(tickets: list[dict[str, Any]]) -> list[dict[str, Any]]: """ Sorts a list of tickets based on their 'depends_on' field. Raises ValueError if a circular dependency or missing internal dependency is detected. @@ -76,4 +77,3 @@ if __name__ == "__main__": test_skeletons = "class NewFeature: pass" tickets = generate_tickets(test_brief, test_skeletons) print(json.dumps(tickets, indent=2)) - diff --git a/dag_engine.py b/dag_engine.py index f61eb8b..d57fef5 100644 --- a/dag_engine.py +++ b/dag_engine.py @@ -96,7 +96,7 @@ class TrackDAG: visited = set() stack = [] - def visit(ticket_id: str): + def visit(ticket_id: str) -> None: """Internal recursive helper for topological sorting.""" if ticket_id in visited: return diff --git a/orchestrator_pm.py b/orchestrator_pm.py index c3e5d70..706ed6f 100644 --- a/orchestrator_pm.py +++ b/orchestrator_pm.py @@ -5,6 +5,7 @@ import mma_prompts import aggregate import summarize from pathlib import Path +from typing import Any, Optional CONDUCTOR_PATH: Path = Path("conductor") @@ -53,7 +54,7 @@ def get_track_history_summary() -> str: return "No previous tracks found." return "\n".join(summary_parts) -def generate_tracks(user_request: str, project_config: dict, file_items: list[dict], history_summary: str = None) -> list[dict]: +def generate_tracks(user_request: str, project_config: dict[str, Any], file_items: list[dict[str, Any]], history_summary: Optional[str] = None) -> list[dict[str, Any]]: """ Tier 1 (Strategic PM) call. Analyzes the project state and user request to generate a list of Tracks. @@ -72,7 +73,7 @@ def generate_tracks(user_request: str, project_config: dict, file_items: list[di user_message = "\n".join(user_message_parts) # Set custom system prompt for this call old_system_prompt = ai_client._custom_system_prompt - ai_client.set_custom_system_prompt(system_prompt) + ai_client.set_custom_system_prompt(system_prompt or "") try: # 3. Call Tier 1 Model (Strategic - Pro) # Note: We use gemini-1.5-pro or similar high-reasoning model for Tier 1 @@ -89,7 +90,7 @@ def generate_tracks(user_request: str, project_config: dict, file_items: list[di json_match = json_match.split("```json")[1].split("```")[0].strip() elif "```" in json_match: json_match = json_match.split("```")[1].split("```")[0].strip() - tracks = json.loads(json_match) + tracks: list[dict[str, Any]] = json.loads(json_match) # Ensure each track has a 'title' for the GUI for t in tracks: if "title" not in t: @@ -101,7 +102,7 @@ def generate_tracks(user_request: str, project_config: dict, file_items: list[di return [] finally: # Restore old system prompt - ai_client.set_custom_system_prompt(old_system_prompt) + ai_client.set_custom_system_prompt(old_system_prompt or "") if __name__ == "__main__": # Quick CLI test