feat(types): Resolve strict mypy errors in conductor subsystem

This commit is contained in:
2026-03-04 01:13:42 -05:00
parent 6ebbf40d9d
commit c5ee50ff0b
4 changed files with 17 additions and 16 deletions

View File

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

View File

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

View File

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

View File

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