ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

refactor(project_manager,aggregate,api_hook_client): replace weak type sites with aliases

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-06-21 12:39:17 -04:00
parent d0c0571bde
commit 833e99f2ec
3 changed files with 100 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aggregate.py
+20 -8
View File
@@ -33,6 +33,18 @@ from src.fuzzy_anchor import FuzzyAnchor
from src.file_cache import ASTParser
from src.paths import get_config_path
from src.performance_monitor import get_monitor
from src.type_aliases import (
CommsLog,
CommsLogCallback,
CommsLogEntry,
FileItem,
FileItems,
History,
HistoryMessage,
Metadata,
ToolCall,
ToolDefinition,
)
def find_next_increment(output_dir: Path, namespace: str) -> int:
@@ -143,7 +155,7 @@ def build_screenshots_section(base_dir: Path, screenshots: list[str]) -> str:
sections.append(f"### `{original}`\n\n![{path.name}]({path.as_posix()})")
return "\n\n---\n\n".join(sections)
def build_file_items(base_dir: Path, files: list[str | dict[str, Any]]) -> list[dict[str, Any]]:
def build_file_items(base_dir: Path, files: list[str | Metadata]) -> list[Metadata]:
"""
Return a list of dicts describing each file, for use by ai_client when it
wants to upload individual files rather than inline everything as markdown.
@@ -161,7 +173,7 @@ def build_file_items(base_dir: Path, files: list[str | dict[str, Any]]) -> list[
[C: src/app_controller.py:AppController._bg_task, src/orchestrator_pm.py:module, tests/test_aggregate_flags.py:test_auto_aggregate_skip, tests/test_aggregate_flags.py:test_force_full, tests/test_context_composition_phase6.py:test_view_mode_custom, tests/test_context_composition_phase6.py:test_view_mode_custom_empty_default_to_summary, tests/test_context_composition_phase6.py:test_view_mode_default_summary, tests/test_context_composition_phase6.py:test_view_mode_full, tests/test_context_composition_phase6.py:test_view_mode_none, tests/test_context_composition_phase6.py:test_view_mode_outline, tests/test_context_composition_phase6.py:test_view_mode_skeleton, tests/test_context_composition_phase6.py:test_view_mode_summary, tests/test_tiered_context.py:test_build_file_items_with_tiers, tests/test_tiered_context.py:test_build_files_section_with_dicts]
"""
with get_monitor().scope("build_file_items"):
items: list[dict[str, Any]] = []
items: list[Metadata] = []
parser = None
for entry_raw in files:
if isinstance(entry_raw, dict):
@@ -285,7 +297,7 @@ def build_file_items(base_dir: Path, files: list[str | dict[str, Any]]) -> list[
items.append({"path": path, "entry": entry, "content": content, "error": error, "mtime": mtime, "tier": tier, "auto_aggregate": auto_aggregate, "force_full": force_full, "view_mode": view_mode, "ast_signatures": ast_signatures, "ast_definitions": ast_definitions, "ast_mask": ast_mask, "custom_slices": custom_slices})
return items
def _build_files_section_from_items(file_items: list[dict[str, Any]]) -> str:
def _build_files_section_from_items(file_items: list[Metadata]) -> str:
"""
Build the files markdown section from pre-read file items (avoids double I/O).
[C: tests/test_aggregate_flags.py:test_auto_aggregate_skip, tests/test_context_composition_phase6.py:test_files_section_rendering, tests/test_tiered_context.py:test_build_files_section_with_dicts, tests/test_ui_summary_only_removal.py:test_aggregate_from_items_respects_auto_aggregate]
@@ -333,7 +345,7 @@ def build_beads_section(base_dir: Path) -> str:
for b in active: parts.append(f"- **{b.title}** ({b.id}): {b.description}")
return "\n\n".join(parts)
def build_markdown_from_items(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False, aggregation_strategy: str = "auto", execution_mode: str = "standard", base_dir: Path | None = None) -> str:
def build_markdown_from_items(file_items: list[Metadata], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False, aggregation_strategy: str = "auto", execution_mode: str = "standard", base_dir: Path | None = None) -> str:
"""Build markdown from pre-read file items instead of re-reading from disk."""
parts = []
# STATIC PREFIX: Files and Screenshots must go first to maximize Cache Hits
@@ -351,7 +363,7 @@ def build_markdown_from_items(file_items: list[dict[str, Any]], screenshot_base_
if history: parts.append("## Discussion History\n\n" + build_discussion_section(history))
return "\n\n---\n\n".join(parts)
def build_markdown_no_history(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], summary_only: bool = False, aggregation_strategy: str = "auto") -> str:
def build_markdown_no_history(file_items: list[Metadata], screenshot_base_dir: Path, screenshots: list[str], summary_only: bool = False, aggregation_strategy: str = "auto") -> str:
"""
Build markdown with only files + screenshots (no history). Used for stable caching.
[C: src/app_controller.py:AppController._do_generate, tests/test_history_management.py:test_aggregate_blacklist]
@@ -367,7 +379,7 @@ def build_discussion_text(history: list[str]) -> str:
return ""
return "## Discussion History\n\n" + build_discussion_section(history)
def build_tier3_context(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], focus_files: list[str]) -> str:
def build_tier3_context(file_items: list[Metadata], screenshot_base_dir: Path, screenshots: list[str], history: list[str], focus_files: list[str]) -> str:
"""
Tier 3 Context: Execution/Worker.
Full content for focus_files and files with tier=3, summaries/skeletons for others.
@@ -460,11 +472,11 @@ def build_tier3_context(file_items: list[dict[str, Any]], screenshot_base_dir: P
if history: parts.append("## Discussion History\n\n" + build_discussion_section(history))
return "\n\n---\n\n".join(parts)
def build_markdown(base_dir: Path, files: list[str | dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False, execution_mode: str = "standard") -> str:
def build_markdown(base_dir: Path, files: list[str | Metadata], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False, execution_mode: str = "standard") -> str:
file_items = build_file_items(base_dir, files)
return build_markdown_from_items(file_items, screenshot_base_dir, screenshots, history, summary_only=summary_only, aggregation_strategy='auto', execution_mode=execution_mode, base_dir=base_dir)
def run(config: dict[str, Any], aggregation_strategy: str = "auto") -> tuple[str, Path, list[dict[str, Any]]]:
def run(config: Metadata, aggregation_strategy: str = "auto") -> tuple[str, Path, list[Metadata]]:
"""
[C: simulation/sim_base.py:run_sim, src/ai_client.py:_send_anthropic, src/ai_client.py:_send_deepseek, src/ai_client.py:_send_gemini, src/ai_client.py:_send_gemini_cli, src/ai_client.py:_send_minimax, src/app_controller.py:AppController._cb_start_track, src/app_controller.py:AppController._do_generate, src/app_controller.py:AppController._process_event_queue, src/app_controller.py:AppController._start_track_logic, src/external_editor.py:_find_vscode_in_registry, src/gui_2.py:App._render_snapshot_tab, src/gui_2.py:App.run, src/gui_2.py:main, src/mcp_client.py:get_git_diff, src/project_manager.py:get_git_commit, src/rag_engine.py:RAGEngine._search_mcp, src/shell_runner.py:run_powershell, tests/conftest.py:kill_process_tree, tests/conftest.py:live_gui, tests/test_conductor_abort_event.py:test_conductor_abort_event_populated, tests/test_conductor_engine_v2.py:test_conductor_engine_dynamic_parsing_and_execution, tests/test_conductor_engine_v2.py:test_conductor_engine_run_executes_tickets_in_order, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_extended_sims.py:test_context_sim_live, tests/test_extended_sims.py:test_execution_sim_live, tests/test_extended_sims.py:test_tools_sim_live, tests/test_external_editor_gui.py:get_vscode_processes, tests/test_external_editor_gui.py:test_vscode_launches_with_diff_view, tests/test_gui_custom_window.py:test_app_window_is_borderless, tests/test_headless_simulation.py:module, tests/test_headless_verification.py:test_headless_verification_error_and_qa_interceptor, tests/test_headless_verification.py:test_headless_verification_full_run, tests/test_mock_gemini_cli.py:run_mock, tests/test_orchestration_logic.py:test_conductor_engine_run, tests/test_parallel_execution.py:test_conductor_engine_pool_integration, tests/test_sim_ai_settings.py:test_ai_settings_simulation_run, tests/test_sim_context.py:test_context_simulation_run, tests/test_sim_execution.py:test_execution_simulation_run, tests/test_sim_tools.py:test_tools_simulation_run]
"""