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

Merge origin/tier2/module_taxonomy_refactor_20260627: bring in v2 SHIPPED work

Browse Source 
Per post_module_taxonomy_de_cruft_20260627 Phase 0 prerequisite.
Master is at 6344b49f (pre-merge of v2 SHIPPED). This merge brings in
the 18 v2 SHIPPED commits that define the destination modules
(src.mma, src/project.py, src/project_files.py, src.tool_presets,
src.tool_bias, src.external_editor, src.personas,
src.workspace_manager, src.mcp_client) needed by the Phase 2
consumer migration in commit 8f11340b.

Conflicts resolved (all were import-block re-orderings between my
migration's update and v2 SHIPPED's update of the same files):
 - src/external_editor.py: took v2 SHIPPED version (class definitions
                                    + the no-alias import pattern)
 - src/personas.py: took v2 SHIPPED version
 - src/tool_bias.py: took v2 SHIPPED version
 - src/tool_presets.py: took v2 SHIPPED version
 - src/workspace_manager.py: took v2 SHIPPED version
 - src/ai_client.py: took v2 SHIPPED version (removes the 'as _FIC'
                              alias; uses 'from src.project_files import
                              FileItem' directly per the v2 SHIPPED style)
 - conductor/tracks/module_taxonomy_refactor_20260627/spec.md: took
                              HEAD version (my Phase 1 VC2 + VC10
                              corrections; the v2 SHIPPED version was
                              the pre-correction spec)
This commit is contained in:
ed
2026-06-26 13:51:05 -04:00
parent 6b0668f1a9 05647d94b5
commit 91a612887c
65 changed files with 4959 additions and 2463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+120 -20
View File
@@ -29,6 +29,7 @@ import sys
import threading
import time
import tomllib
from dataclasses import dataclass
# TODO(Ed): Eliminate These?
from collections import deque
@@ -44,18 +45,17 @@ from src import mma_prompts
from src import performance_monitor
from src import project_manager
from src import provider_state
from src.vendor_capabilities import VendorCapabilities, get_capabilities
# TODO(Ed): Eliminate these?
from src.events import EventEmitter
from src.gemini_cli_adapter import GeminiCliAdapter
from src.project_files import FileItem
from src.tool_bias import BiasProfile
from src.tool_presets import ToolPreset, Tool
from src.models import FileItem, ToolPreset, BiasProfile, Tool
from src.paths import get_credentials_path
from src.tool_bias import ToolBiasEngine
from src.tool_presets import ToolPresetManager
from src.tool_presets import ToolPresetManager
# VendorCapabilities, get_capabilities, list_models_for_vendor, register
# are defined in this file (see '#region: Vendor Capabilities'). Previously
# imported from src/vendor_capabilities.py (deleted in
# module_taxonomy_refactor_20260627 Phase 2.1).
PROVIDERS: List[str] = ["gemini", "anthropic", "gemini_cli", "deepseek", "minimax", "qwen", "grok", "llama"]
@@ -188,6 +188,110 @@ _project_context_marker: str = ""
#endregion: Provider Configuration
#region: Vendor Capabilities (moved from src/vendor_capabilities.py)
@dataclass(frozen=True)
class VendorCapabilities:
vendor: str
model: str
vision: bool = False
tool_calling: bool = True
caching: bool = False
streaming: bool = True
model_discovery: bool = True
context_window: int = 8192
cost_tracking: bool = True
cost_input_per_mtok: float = 0.0
cost_output_per_mtok: float = 0.0
notes: str = ''
local: bool = False
reasoning: bool = False
structured_output: bool = False
code_execution: bool = False
web_search: bool = False
x_search: bool = False
file_search: bool = False
mcp_support: bool = False
audio: bool = False
video: bool = False
grounding: bool = False
computer_use: bool = False
_VENDOR_REGISTRY: dict[tuple[str, str], "VendorCapabilities"] = {}
def register(cap: "VendorCapabilities") -> None:
_VENDOR_REGISTRY[(cap.vendor, cap.model)] = cap
def get_capabilities(vendor: str, model: str) -> "VendorCapabilities":
if (vendor, model) in _VENDOR_REGISTRY: return _VENDOR_REGISTRY[(vendor, model)]
if (vendor, '*') in _VENDOR_REGISTRY: return _VENDOR_REGISTRY[(vendor, '*')]
raise KeyError(f'No capabilities registered for vendor={vendor!r} model={model!r}')
def list_models_for_vendor(vendor: str) -> list[str]:
return sorted({m for v, m in _VENDOR_REGISTRY if v == vendor and m != '*'})
register(VendorCapabilities(vendor='minimax', model='*', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.7', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20, reasoning=True))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.5', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20, reasoning=True))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.1', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='grok', model='*', context_window=131072, cost_input_per_mtok=2.00, cost_output_per_mtok=10.00, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-2', context_window=131072, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-2-vision', vision=True, context_window=32768, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-beta', context_window=131072, cost_input_per_mtok=5.00, cost_output_per_mtok=15.00, web_search=True, x_search=True))
register(VendorCapabilities(vendor='llama', model='*', context_window=131072))
register(VendorCapabilities(vendor='llama', model='llama-3.1-8b-instant', context_window=131072, cost_input_per_mtok=0.05, cost_output_per_mtok=0.08))
register(VendorCapabilities(vendor='llama', model='llama-3.1-70b-versatile', context_window=131072, cost_input_per_mtok=0.59, cost_output_per_mtok=0.79))
register(VendorCapabilities(vendor='llama', model='llama-3.1-405b-reasoning', context_window=131072, cost_input_per_mtok=3.00, cost_output_per_mtok=3.00, reasoning=True))
register(VendorCapabilities(vendor='llama', model='llama-3.2-1b-preview', context_window=131072, cost_input_per_mtok=0.04, cost_output_per_mtok=0.04))
register(VendorCapabilities(vendor='llama', model='llama-3.2-3b-preview', context_window=131072, cost_input_per_mtok=0.06, cost_output_per_mtok=0.06))
register(VendorCapabilities(vendor='llama', model='llama-3.2-11b-vision-preview', vision=True, context_window=131072, cost_input_per_mtok=0.18, cost_output_per_mtok=0.18))
register(VendorCapabilities(vendor='llama', model='llama-3.2-90b-vision-preview', vision=True, context_window=131072, cost_input_per_mtok=0.90, cost_output_per_mtok=0.90))
register(VendorCapabilities(vendor='llama', model='llama-3.3-70b-specdec', context_window=131072, cost_input_per_mtok=0.59, cost_output_per_mtok=0.79))
register(VendorCapabilities(vendor='qwen', model='*', context_window=32768))
register(VendorCapabilities(vendor='qwen', model='qwen-turbo', context_window=1000000, cost_input_per_mtok=0.05, cost_output_per_mtok=0.10))
register(VendorCapabilities(vendor='qwen', model='qwen-plus', context_window=131072, cost_input_per_mtok=0.40, cost_output_per_mtok=1.20))
register(VendorCapabilities(vendor='qwen', model='qwen-max', context_window=32768, cost_input_per_mtok=2.00, cost_output_per_mtok=6.00))
register(VendorCapabilities(vendor='qwen', model='qwen-long', context_window=1000000, cost_input_per_mtok=0.07, cost_output_per_mtok=0.28, caching=True, notes='qwen-long supports custom chunked long-context caching'))
register(VendorCapabilities(vendor='qwen', model='qwen-vl-plus', vision=True, context_window=131072, cost_input_per_mtok=0.21, cost_output_per_mtok=0.63))
register(VendorCapabilities(vendor='qwen', model='qwen-vl-max', vision=True, context_window=32768, cost_input_per_mtok=0.50, cost_output_per_mtok=1.50))
register(VendorCapabilities(vendor='qwen', model='qwen-audio', context_window=32768, cost_input_per_mtok=0.10, cost_output_per_mtok=0.30, audio=True, notes='Audio input support added 2026-06-11 (v2 matrix)'))
register(VendorCapabilities(vendor='anthropic', model='*', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True, notes='Anthropic wildcard: Sonnet defaults. Per-model variations below.'))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-5-20250929', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-20250514', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-6', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-1-20250805', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-20250514', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-5-20251101', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-6', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-7', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-8', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-haiku-4-5-20251001', context_window=200000, cost_input_per_mtok=1.00, cost_output_per_mtok=5.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-fable-5', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='gemini', model='*', context_window=1000000, cost_input_per_mtok=1.25, cost_output_per_mtok=5.00, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True, notes='Gemini wildcard: 1M+ context window. Per-model variations below.'))
register(VendorCapabilities(vendor='gemini', model='gemini-3.1-pro-preview', context_window=1000000, cost_input_per_mtok=3.50, cost_output_per_mtok=10.50, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-3-flash-preview', context_window=1000000, cost_input_per_mtok=0.15, cost_output_per_mtok=0.60, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-2.5-flash', context_window=1000000, cost_input_per_mtok=0.15, cost_output_per_mtok=0.60, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-2.5-flash-lite', context_window=1000000, cost_input_per_mtok=0.075, cost_output_per_mtok=0.30, caching=True, vision=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='*', context_window=32768, cost_input_per_mtok=0.27, cost_output_per_mtok=1.10, reasoning=True, structured_output=True, notes='DeepSeek wildcard: V3 defaults. R1/reasoner variants below.'))
register(VendorCapabilities(vendor='deepseek', model='deepseek-v3', context_window=32768, cost_input_per_mtok=0.27, cost_output_per_mtok=1.10, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='deepseek-reasoner', context_window=32768, cost_input_per_mtok=0.55, cost_output_per_mtok=2.19, reasoning=True, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='deepseek-r1', context_window=32768, cost_input_per_mtok=0.55, cost_output_per_mtok=2.19, reasoning=True, structured_output=True))
#endregion: Vendor Capabilities
#region: Vendor State (moved from src/vendor_state.py)
@dataclass(frozen=True)
class VendorMetric:
key: str
label: str
value: str
state: str
tooltip: str
#endregion: Vendor State
#region: System Prompt Management
def set_custom_system_prompt(prompt: str) -> None:
@@ -412,7 +516,7 @@ def set_provider(provider: str, model: str, validate: bool = True) -> None:
When validate is True (default), the model is checked against the provider's
LIVE model list, which for gemini_cli/minimax means a blocking subprocess /
network call (and importing the provider SDK). Pass validate=False during
startup so the GUI's first frame is not blocked — AppController._fetch_models
startup so the GUI's first frame is not blocked ΓÇö AppController._fetch_models
corrects the model against the live list shortly after, off the main thread.
"""
global _provider, _model
@@ -726,7 +830,7 @@ def _parse_tool_args_result(tool_args_str: str) -> Result[Metadata]:
On JSON parse failure, returns Result(data={}, errors=[ErrorInfo(...)]).
The legacy caller accumulates errors into file_errors and falls back to
empty args (preserving original behavior). Per TIER1_REVIEW 2026-06-20:
empty-default is NOT a drain — the caller must observe the errors.
empty-default is NOT a drain ΓÇö the caller must observe the errors.
"""
try:
return Result(data=json.loads(tool_args_str))
@@ -1305,7 +1409,7 @@ def _list_anthropic_models_result() -> Result[list[str]]:
The previous version had:
except Exception as exc:
raise _classify_anthropic_error(exc) from exc
which raised an ErrorInfo as an Exception — a runtime bug. This
which raised an ErrorInfo as an Exception ΓÇö a runtime bug. This
migration follows the Phase 9 redo precedent: convert to Result[T].
"""
try:
@@ -2563,8 +2667,8 @@ def _send_grok(md_content: str, user_message: str, base_dir: str,
if file_items:
for fi in file_items:
if fi.get("is_image") and fi.get("base64_data"):
from src.project_files import FileItem as _FIC
fi_item = fi if isinstance(fi, _FIC) else _FIC.from_dict(fi)
from src.project_files import FileItem
fi_item = fi if isinstance(fi, FileItem) else FileItem.from_dict(fi)
user_content = f"[IMAGE: {fi_item.path or 'attachment'}]\n{user_content}"
if discussion_history and not history:
history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
@@ -2597,7 +2701,6 @@ def _send_grok(md_content: str, user_message: str, base_dir: str,
return Result(data="", errors=[_classify_openai_compatible_error(exc, source="ai_client.grok")])
def _list_grok_models() -> list[str]:
from src.vendor_capabilities import list_models_for_vendor
return list_models_for_vendor("grok")
def _send_minimax(md_content: str, user_message: str, base_dir: str,
@@ -2755,7 +2858,6 @@ def _extract_dashscope_tool_calls(resp: Any) -> list[Metadata]:
return out
def _list_qwen_models() -> list[str]:
from src.vendor_capabilities import list_models_for_vendor
return list_models_for_vendor("qwen")
def _send_qwen(md_content: str, user_message: str, base_dir: str,
@@ -2807,8 +2909,8 @@ def _send_qwen(md_content: str, user_message: str, base_dir: str,
if file_items:
for fi in file_items:
if fi.get("is_image") and fi.get("base64_data"):
from src.project_files import FileItem as _FIC
fi_item = fi if isinstance(fi, _FIC) else _FIC.from_dict(fi)
from src.project_files import FileItem
fi_item = fi if isinstance(fi, FileItem) else FileItem.from_dict(fi)
user_content = f"[IMAGE: {fi_item.path or 'attachment'}]\n{user_content}"
if discussion_history and not history:
history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
@@ -2900,8 +3002,8 @@ def _send_llama(md_content: str, user_message: str, base_dir: str,
if file_items:
for fi in file_items:
if fi.get("is_image") and fi.get("base64_data"):
from src.project_files import FileItem as _FIC
fi_item = fi if isinstance(fi, _FIC) else _FIC.from_dict(fi)
from src.project_files import FileItem
fi_item = fi if isinstance(fi, FileItem) else FileItem.from_dict(fi)
user_content = f"[IMAGE: {fi_item.path or 'attachment'}]\n{user_content}"
if discussion_history and not history:
history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
@@ -3017,13 +3119,11 @@ def _send_llama_native(md_content: str, user_message: str, base_dir: str,
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(exc), source="ai_client.llama_native", original=exc)])
def _list_llama_models() -> list[str]:
from src.vendor_capabilities import list_models_for_vendor
return list_models_for_vendor("llama")
def _get_llama_cost_tracking() -> bool:
if "localhost" in _llama_base_url or "127.0.0.1" in _llama_base_url:
return False
from src.vendor_capabilities import get_capabilities
try:
caps = get_capabilities("llama", _model)
return caps.cost_tracking
src/app_controller.py
+10 -11
View File
@@ -27,6 +27,7 @@ from src.module_loader import _require_warmed
from src import conductor_tech_lead
from src import events
from src import mcp_client
from src import mcp_tool_specs
from src import multi_agent_conductor
from src import orchestrator_pm
from src import paths
@@ -2054,7 +2055,7 @@ class AppController:
from src.personas import PersonaManager
self.persona_manager = PersonaManager(Path(self.active_project_path).parent if self.active_project_path else None)
from src.vendor_capabilities import get_capabilities
from src.ai_client import get_capabilities
try:
caps = get_capabilities(self.current_provider, self.current_model)
except KeyError:
@@ -2076,8 +2077,7 @@ class AppController:
self.ui_separate_tool_calls_panel = _uip.separate_tool_calls_panel
self.ui_auto_switch_layout = gui_cfg.get("auto_switch_layout", False)
self.ui_tier_layout_bindings = gui_cfg.get("tier_layout_bindings", {"Tier 1": "", "Tier 2": "", "Tier 3": "", "Tier 4": ""})
from src import bg_shader
bg_shader.get_bg().enabled = gui_cfg.get("bg_shader_enabled", False)
self.bg_shader_enabled = gui_cfg.get("bg_shader_enabled", False)
_default_windows = {
"Project Settings": True,
@@ -2107,7 +2107,7 @@ class AppController:
saved = self.config.get("gui", {}).get("show_windows", {})
self.show_windows = {k: saved.get(k, v) for k, v in _default_windows.items()}
agent_tools_cfg = self.project.get("agent", {}).get("tools", {})
self.ui_agent_tools = {t: agent_tools_cfg.get(t, True) for t in models.AGENT_TOOL_NAMES}
self.ui_agent_tools = {t: agent_tools_cfg.get(t, True) for t in mcp_tool_specs.tool_names()}
label = self.project.get("project", {}).get("name", "")
session_logger.reset_session(label=label)
# Trigger auto-start of MCP servers
@@ -2969,7 +2969,7 @@ class AppController:
proj["project"]["auto_scroll_tool_calls"] = self.ui_auto_scroll_tool_calls
proj.setdefault("gemini_cli", {})["binary_path"] = self.ui_gemini_cli_path
proj.setdefault("agent", {}).setdefault("tools", {})
for t_name in models.AGENT_TOOL_NAMES:
for t_name in mcp_tool_specs.tool_names():
proj["agent"]["tools"][t_name] = self.ui_agent_tools.get(t_name, True)
self._flush_disc_entries_to_project()
disc_sec = proj.setdefault("discussion", {})
@@ -3018,7 +3018,6 @@ class AppController:
self.config["rag"] = self.rag_config.to_dict()
self.config["projects"] = {"paths": self.project_paths, "active": self.active_project_path}
from src import bg_shader
# Update gui section while preserving other keys like bg_shader_enabled
gui_cfg = self.config.get("gui", {})
gui_cfg.update({
@@ -3033,7 +3032,7 @@ class AppController:
"separate_tier2": self.ui_separate_tier2,
"separate_tier3": self.ui_separate_tier3,
"separate_tier4": self.ui_separate_tier4,
"bg_shader_enabled": bg_shader.get_bg().enabled
"bg_shader_enabled": getattr(self, "bg_shader_enabled", False)
})
self.config["gui"] = gui_cfg
@@ -3270,7 +3269,7 @@ class AppController:
self.ui_auto_scroll_tool_calls = proj.get("project", {}).get("auto_scroll_tool_calls", True)
self.ui_word_wrap = proj.get("project", {}).get("word_wrap", True)
agent_tools_cfg = proj.get("agent", {}).get("tools", {})
self.ui_agent_tools = {t: agent_tools_cfg.get(t, True) for t in models.AGENT_TOOL_NAMES}
self.ui_agent_tools = {t: agent_tools_cfg.get(t, True) for t in mcp_tool_specs.tool_names()}
# MMA Tracks
self.tracks = project_manager.get_all_tracks(self.active_project_root)
# Restore MMA state
@@ -4283,7 +4282,7 @@ class AppController:
def _on_ai_stream(self, text: str) -> None:
"""Handles streaming text from the AI."""
self.event_queue.put("response", {"text": text, "status": "streaming...", "role": "AI"})
from src.vendor_capabilities import get_capabilities
from src.ai_client import get_capabilities
try:
caps = get_capabilities(self.current_provider, self.current_model)
except KeyError:
@@ -5161,7 +5160,7 @@ class AppController:
scripts/audit_no_models_config_io.py.
[C: src/app_controller.py:AppController.__init__]
"""
self.config = models._load_config_from_disk()
self.config = models.load_config_from_disk()
return self.config
def save_config(self) -> None:
@@ -5173,7 +5172,7 @@ class AppController:
scripts/audit_no_models_config_io.py.
[C: src/app_controller.py:AppController._cb_project_save, src/app_controller.py:AppController._do_generate]
"""
models._save_config_to_disk(self.config)
models.save_config_to_disk(self.config)
#endregion: --- Config I/O (single source of truth) ---
#endregion: MMA (Controller)
src/bg_shader.py
-75
View File
@@ -1,75 +0,0 @@
# src/bg_shader.py
import time
import math
from typing import Optional
from imgui_bundle import imgui, nanovg as nvg, hello_imgui
class BackgroundShader:
def __init__(self):
"""
[C: src/mcp_client.py:_DDGParser.__init__, src/mcp_client.py:_TextExtractor.__init__]
"""
self.enabled = False
self.start_time = time.time()
self.ctx: Optional[nvg.Context] = None
def render(self, width: float, height: float):
"""
[C: src/gui_2.py:App._gui_func, src/gui_2.py:App._render_discussion_entry_read_mode, src/gui_2.py:App._render_heavy_text, src/gui_2.py:App._render_markdown_test, src/gui_2.py:App._render_prior_session_view, src/gui_2.py:App._render_response_panel, src/gui_2.py:App._render_snapshot_tab, src/gui_2.py:App._render_text_viewer_window, src/markdown_helper.py:MarkdownRenderer._render_code_block, src/markdown_helper.py:MarkdownRenderer.render, src/markdown_helper.py:render, src/theme_2.py:render_post_fx, tests/test_theme_nerv_alert.py:test_alert_pulsing_render_active, tests/test_theme_nerv_alert.py:test_alert_pulsing_render_inactive, tests/test_theme_nerv_fx.py:TestThemeNervFx.test_alert_pulsing_render, tests/test_theme_nerv_fx.py:TestThemeNervFx.test_crt_filter_disabled, tests/test_theme_nerv_fx.py:TestThemeNervFx.test_crt_filter_render]
"""
if not self.enabled or width <= 0 or height <= 0:
return
# In imgui-bundle, hello_imgui handles the background.
# We can use the background_draw_list to draw primitives.
# Since we don't have raw GLSL easily in Python without PyOpenGL,
# we'll use a "faux-shader" approach with NanoVG or DrawList gradients.
t = time.time() - self.start_time
dl = imgui.get_background_draw_list()
# Base deep sea color
dl.add_rect_filled(imgui.ImVec2(0, 0), imgui.ImVec2(width, height), imgui.get_color_u32(imgui.ImVec4(0.01, 0.07, 0.20, 1.0)))
# Layer 1: Slow moving large blobs (FBM approximation)
for i in range(3):
phase = t * (0.1 + i * 0.05)
x = (math.sin(phase) * 0.5 + 0.5) * width
y = (math.cos(phase * 0.8) * 0.5 + 0.5) * height
radius = (0.4 + 0.2 * math.sin(t * 0.2)) * max(width, height)
col = imgui.ImVec4(0.02, 0.26, 0.55, 0.3)
dl.add_circle_filled(imgui.ImVec2(x, y), radius, imgui.get_color_u32(col), num_segments=32)
# Layer 2: Shimmering caustics (Animated Lines)
num_lines = 15
for i in range(num_lines):
offset = (t * 20.0 + i * (width / num_lines)) % width
alpha = 0.1 * (1.0 + math.sin(t + i))
col = imgui.get_color_u32(imgui.ImVec4(0.08, 0.60, 0.88, alpha))
p1 = imgui.ImVec2(offset, 0)
p2 = imgui.ImVec2(offset - 100, height)
dl.add_line(p1, p2, col, thickness=2.0)
# Vignette
center = imgui.ImVec2(width/2, height/2)
radius = max(width, height) * 0.8
# Draw multiple concentric circles for a soft vignette
for i in range(10):
r = radius + (i * 50)
alpha = (i / 10.0) * 0.5
dl.add_circle(center, r, imgui.get_color_u32(imgui.ImVec4(0, 0, 0, alpha)), num_segments=64, thickness=60.0)
_bg: Optional[BackgroundShader] = None
def get_bg():
"""
[C: src/gui_2.py:App._gui_func, src/gui_2.py:App._render_theme_panel]
"""
global _bg
if _bg is None:
_bg = BackgroundShader()
return _bg
src/command_palette.py
-208
View File
@@ -1,208 +0,0 @@
from __future__ import annotations
from imgui_bundle import imgui
from dataclasses import dataclass, field
from typing import Optional, Callable, List, Dict, Any
from src.result_types import ErrorInfo, ErrorKind, Result
@dataclass
class Command:
id: str
title: str
category: str
shortcut: Optional[str] = None
description: str = ""
enabled_when: Optional[str] = None
action: Optional[Callable] = None
@dataclass
class ScoredCommand:
command: Command
score: float
class CommandRegistry:
def __init__(self) -> None:
self._commands: Dict[str, Command] = {}
def register(self, command_or_callable: Any) -> Any:
if isinstance(command_or_callable, Command):
cmd = command_or_callable
else:
cmd = Command(
id=command_or_callable.__name__,
title=command_or_callable.__name__.replace("_", " ").title(),
category="uncategorized",
action=command_or_callable,
)
if cmd.id in self._commands:
raise ValueError(f"Command {cmd.id} already registered")
self._commands[cmd.id] = cmd
return command_or_callable
def all(self) -> List[Command]:
return list(self._commands.values())
def get(self, command_id: str) -> Command:
return self._commands.get(command_id) or Command(id="", title="", category="uncategorized", action=lambda: None)
def fuzzy_match(query: str, candidates: List[Command], top_n: int = 20) -> List[ScoredCommand]:
query_lower = query.lower()
scored: List[ScoredCommand] = []
for cmd in candidates:
title_lower = cmd.title.lower()
if not _is_subsequence(query_lower, title_lower):
continue
score = _compute_score(query_lower, title_lower)
scored.append(ScoredCommand(command=cmd, score=score))
scored.sort(key=lambda r: r.score, reverse=True)
return scored[:top_n]
def _is_subsequence(query: str, target: str) -> bool:
qi = 0
for ch in target:
if qi < len(query) and ch == query[qi]:
qi += 1
return qi == len(query)
def _compute_score(query: str, target: str) -> float:
score = 0.0
if target.startswith(query): score += 1.0
elif _starts_at_word_boundary(query, target): score += 0.5
if _is_contiguous(query, target): score += 0.3
gaps = _count_gaps(query, target)
score -= 0.1 * gaps
return score
def _starts_at_word_boundary(query: str, target: str) -> bool:
if not target.startswith(query):
return False
return len(query) == 0 or not query[0].isalnum() or len(target) == len(query) or not target[len(query)].isalnum()
def _is_contiguous(query: str, target: str) -> bool:
return query in target
def _count_gaps(query: str, target: str) -> int:
qi = 0
gaps = 0
last_match = -1
for ti, ch in enumerate(target):
if qi < len(query) and ch == query[qi]:
if last_match >= 0 and ti - last_match > 1: gaps += ti - last_match - 1
last_match = ti
qi += 1
return gaps
def _close_palette(app: Any) -> None:
"""Close the palette and reset all per-open state."""
app.show_command_palette = False
app._command_palette_query = ""
app._command_palette_selected = 0
app._command_palette_focused = False
app._command_palette_input_focused = False
def _execute(app: Any, command: Command) -> None:
"""Run a command and close the palette. Catches exceptions to keep the modal clean."""
if not command.action:
return
try:
command.action(app)
except (AttributeError, TypeError, ValueError, OSError) as e:
_cmd_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"Action {command.id} raised: {e}", source="command_palette._execute", original=e)])
print(f"[CommandPalette] Action {command.id} raised: {e}")
_close_palette(app)
def render_palette_modal(app: Any, commands: List[Command]) -> None:
"""Renders the interactive Command Palette modal. Exposes a text input query bar
and lists matching commands with fuzzy matching, supporting keyboard navigation (Up/Down/Enter/Esc).
SSDL: `[I:query_input] -> [B:results_list] => [B:execute_command]`
ASCII Layout Map:
+==================== Command Palette ===================+
| |query_text| |
| +-----------------------------------------------------+ |
| | > [category-a] Command Title A (selected) | |
| | [category-b] Command Title B | |
| +-----------------------------------------------------+ |
+========================================================+
"""
if not getattr(app, "show_command_palette", False):
return
viewport = imgui.get_main_viewport()
center = viewport.get_center()
imgui.set_next_window_pos((center.x - 300, center.y - 200), imgui.Cond_.always)
imgui.set_next_window_size((600, 400), imgui.Cond_.always)
if not hasattr(app, "_command_palette_query"): app._command_palette_query = ""
if not hasattr(app, "_command_palette_selected"): app._command_palette_selected = 0
if not hasattr(app, "_command_palette_focused"): app._command_palette_focused = False
# Set focus on the window + input field ONCE per open.
if not app._command_palette_focused:
imgui.set_next_window_focus()
app._command_palette_focused = True
# Escape closes the palette.
if imgui.is_key_pressed(imgui.Key.escape):
_close_palette(app)
return
expanded, opened = imgui.begin("Command Palette##manual_slop", True, imgui.WindowFlags_.no_collapse)
if not expanded or not opened:
app.show_command_palette = False
app._command_palette_focused = False
imgui.end()
return
# After the window is drawn, the input gets focus.
if not getattr(app, '_command_palette_input_focused', False):
imgui.set_keyboard_focus_here()
app._command_palette_input_focused = True
# Process Up/Down/Enter BEFORE input_text so we see the keys before the
# input field consumes them for cursor movement / text editing.
results = fuzzy_match(app._command_palette_query, commands, top_n=20)
if results: app._command_palette_selected = max(0, min(app._command_palette_selected, len(results) - 1))
else: app._command_palette_selected = 0
if imgui.is_key_pressed(imgui.Key.down_arrow):
if results:
app._command_palette_selected = min(app._command_palette_selected + 1, len(results) - 1)
if imgui.is_key_pressed(imgui.Key.up_arrow):
if results:
app._command_palette_selected = max(app._command_palette_selected - 1, 0)
if imgui.is_key_pressed(imgui.Key.enter) or imgui.is_key_pressed(imgui.Key.keypad_enter):
if results and 0 <= app._command_palette_selected < len(results):
_execute(app, results[app._command_palette_selected].command)
imgui.set_next_item_width(-1)
_, app._command_palette_query = imgui.input_text("##query", app._command_palette_query)
if imgui.begin_child("##results", (0, -1)):
for i, scored in enumerate(results):
is_selected = (i == app._command_palette_selected)
label = f"[{scored.command.category}] {scored.command.title}"
clicked, _ = imgui.selectable(label, is_selected)
if clicked:
app._command_palette_selected = i
_execute(app, scored.command)
if not results:
imgui.text_disabled("No matching commands.")
imgui.end_child()
imgui.end()
src/commands.py
+131 -28
View File
@@ -2,12 +2,12 @@ from __future__ import annotations
import webbrowser
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
from src import models
from src import theme_2
from src.module_loader import _require_warmed
from src.hot_reloader import HotReloader
from src.result_types import ErrorInfo, ErrorKind, Result
@@ -15,25 +15,138 @@ from src.result_types import ErrorInfo, ErrorKind, Result
if TYPE_CHECKING:
from src.gui_2 import App
# Lazy command registry (startup_speedup_20260606 Phase 5A)
# --------------------------------------------------------------------------
# The @registry.register decorator runs at module import time, but we want
# to defer the actual CommandRegistry creation (and the underlying
# src.command_palette import, ~244ms) until the palette is actually used.
# The proxy below makes @registry.register a no-op that just queues the
# function; the real CommandRegistry is built lazily on first access to
# any other registry attribute (.all, .get, etc.) by gui_2.py or tests.
# Command data classes + registry (moved from src/command_palette.py in
# module_taxonomy_refactor_20260627 Phase 1.3; the *rendering* function
# `render_palette_modal` lives in src/gui_2.py because it owns ImGui state)
# --------------------------------------------------------------------------
@dataclass
class Command:
id: str
title: str
category: str
shortcut: Optional[str] = None
description: str = ""
enabled_when: Optional[str] = None
action: Optional[Callable] = None
@dataclass
class ScoredCommand:
command: Command
score: float
class CommandRegistry:
def __init__(self) -> None:
self._commands: Dict[str, Command] = {}
def register(self, command_or_callable: Any) -> Any:
if isinstance(command_or_callable, Command):
cmd = command_or_callable
else:
cmd = Command(
id=command_or_callable.__name__,
title=command_or_callable.__name__.replace("_", " ").title(),
category="uncategorized",
action=command_or_callable,
)
if cmd.id in self._commands:
raise ValueError(f"Command {cmd.id} already registered")
self._commands[cmd.id] = cmd
return command_or_callable
def all(self) -> List[Command]:
return list(self._commands.values())
def get(self, command_id: str) -> Command:
return self._commands.get(command_id) or Command(id="", title="", category="uncategorized", action=lambda: None)
def fuzzy_match(query: str, candidates: List[Command], top_n: int = 20) -> List[ScoredCommand]:
query_lower = query.lower()
scored: List[ScoredCommand] = []
for cmd in candidates:
title_lower = cmd.title.lower()
if not _is_subsequence(query_lower, title_lower):
continue
score = _compute_score(query_lower, title_lower)
scored.append(ScoredCommand(command=cmd, score=score))
scored.sort(key=lambda r: r.score, reverse=True)
return scored[:top_n]
def _is_subsequence(query: str, target: str) -> bool:
qi = 0
for ch in target:
if qi < len(query) and ch == query[qi]:
qi += 1
return qi == len(query)
def _compute_score(query: str, target: str) -> float:
score = 0.0
if target.startswith(query): score += 1.0
elif _starts_at_word_boundary(query, target): score += 0.5
if _is_contiguous(query, target): score += 0.3
gaps = _count_gaps(query, target)
score -= 0.1 * gaps
return score
def _starts_at_word_boundary(query: str, target: str) -> bool:
if not target.startswith(query):
return False
return len(query) == 0 or not query[0].isalnum() or len(target) == len(query) or not target[len(query)].isalnum()
def _is_contiguous(query: str, target: str) -> bool:
return query in target
def _count_gaps(query: str, target: str) -> int:
qi = 0
gaps = 0
last_match = -1
for ti, ch in enumerate(target):
if qi < len(query) and ch == query[qi]:
if last_match >= 0 and ti - last_match > 1: gaps += ti - last_match - 1
last_match = ti
qi += 1
return gaps
def _close_palette(app: Any) -> None:
app.show_command_palette = False
app._command_palette_query = ""
app._command_palette_selected = 0
app._command_palette_focused = False
app._command_palette_input_focused = False
def _execute(app: Any, command: Command) -> None:
if not command.action:
return
try:
command.action(app)
except (AttributeError, TypeError, ValueError, OSError) as e:
_cmd_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"Action {command.id} raised: {e}", source="command_palette._execute", original=e)])
print(f"[CommandPalette] Action {command.id} raised: {e}")
_close_palette(app)
# --------------------------------------------------------------------------
# Eager registry (was _LazyCommandRegistry; the lazy pattern is no longer
# needed since src/commands.py is a thin data module, not the heavy
# command_palette.py that previously pulled in imgui at module load time)
# --------------------------------------------------------------------------
_PENDING_REGISTRATIONS: list[Callable] = []
_real_registry: Any = None
_real_registry: CommandRegistry | None = None
def _get_real_registry() -> CommandRegistry:
global _real_registry
if _real_registry is None:
_real_registry = CommandRegistry()
for func in _PENDING_REGISTRATIONS:
_real_registry.register(func)
return _real_registry
class _LazyCommandRegistry:
"""Proxy that defers CommandRegistry instantiation.
Behaves like a CommandRegistry from the caller's perspective:
- @registry.register decorates functions by queuing them
- .all, .get, etc. trigger real initialization on first access
class _EagerCommandRegistry:
"""Eager registry proxy. @registry.register queues until first .all/.get,
then materializes the real CommandRegistry and replays the queue.
"""
def register(self, command_or_callable: Any) -> Any:
@@ -44,17 +157,7 @@ class _LazyCommandRegistry:
return getattr(_get_real_registry(), name)
def _get_real_registry() -> Any:
global _real_registry
if _real_registry is None:
command_palette = _require_warmed("src.command_palette")
_real_registry = command_palette.CommandRegistry()
for func in _PENDING_REGISTRATIONS:
_real_registry.register(func)
return _real_registry
registry = _LazyCommandRegistry()
registry = _EagerCommandRegistry()
# --------------------------------------------------------------------------
src/diff_viewer.py
-173
View File
@@ -1,173 +0,0 @@
import difflib
import shutil
import os
from dataclasses import dataclass
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from src.result_types import ErrorInfo, ErrorKind, Result
@dataclass
class DiffHunk:
header: str
lines: List[str]
old_start: int
old_count: int
new_start: int
new_count: int
@dataclass
class DiffFile:
old_path: str
new_path: str
hunks: List[DiffHunk]
def parse_hunk_header(line: str) -> tuple[int, int, int, int]:
"""
[C: tests/test_diff_viewer.py:test_parse_hunk_header]
"""
if not line.startswith("@@"): return (-1, -1, -1, -1)
parts = line.split()
if len(parts) < 2: return (-1, -1, -1, -1)
old_part = parts[1][1:]
new_part = parts[2][1:]
old_parts = old_part.split(",")
new_parts = new_part.split(",")
old_start = int(old_parts[0])
old_count = int(old_parts[1]) if len(old_parts) > 1 else 1
new_start = int(new_parts[0])
new_count = int(new_parts[1]) if len(new_parts) > 1 else 1
return (old_start, old_count, new_start, new_count)
def parse_diff(diff_text: str) -> List[DiffFile]:
"""
[C: src/gui_2.py:App.request_patch_from_tier4, tests/test_diff_viewer.py:test_diff_line_classification, tests/test_diff_viewer.py:test_parse_diff_empty, tests/test_diff_viewer.py:test_parse_diff_none, tests/test_diff_viewer.py:test_parse_diff_with_context, tests/test_diff_viewer.py:test_parse_multiple_files, tests/test_diff_viewer.py:test_parse_simple_diff]
"""
if not diff_text or not diff_text.strip():
return []
files: List[DiffFile] = []
current_file: Optional[DiffFile] = None
current_hunk: Optional[DiffHunk] = None
for line in diff_text.split("\n"):
if line.startswith("--- "):
if current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
current_hunk = None
files.append(current_file)
path = line[4:]
if path.startswith("a/"):
path = path[2:]
current_file = DiffFile(old_path=path, new_path="", hunks=[])
elif line.startswith("+++ ") and current_file:
path = line[4:]
if path.startswith("b/"):
path = path[2:]
current_file.new_path = path
elif line.startswith("@@") and current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
hunk_info = parse_hunk_header(line)
if hunk_info:
old_start, old_count, new_start, new_count = hunk_info
current_hunk = DiffHunk(
header = line,
lines = [],
old_start = old_start,
old_count = old_count,
new_start = new_start,
new_count = new_count
)
else:
current_hunk = DiffHunk(
header = line,
lines = [],
old_start = 0,
old_count = 0,
new_start = 0,
new_count = 0
)
elif current_hunk is not None:
current_hunk.lines.append(line)
elif line and not line.startswith("diff ") and not line.startswith("index "):
pass
if current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
files.append(current_file)
return files
def get_line_color(line: str) -> str:
"""
[C: tests/test_diff_viewer.py:test_get_line_color]
"""
if line.startswith("+"): return "green"
elif line.startswith("-"): return "red"
elif line.startswith("@@"): return "cyan"
return ""
def apply_patch_to_file(patch_text: str, base_dir: str = ".") -> Tuple[bool, str]:
"""
[C: src/gui_2.py:App._apply_pending_patch, tests/test_diff_viewer.py:test_apply_patch_simple, tests/test_diff_viewer.py:test_apply_patch_with_context]
"""
diff_files = parse_diff(patch_text)
if not diff_files:
return False, "No valid diff found"
results = []
for df in diff_files:
file_path = Path(base_dir) / df.old_path
if not file_path.exists():
results.append(f"File not found: {file_path}")
continue
try:
with open(file_path, "r", encoding="utf-8") as f:
original_lines = f.read().splitlines(keepends=True)
new_lines = original_lines.copy()
offset = 0
for hunk in df.hunks:
hunk_old_start = hunk.old_start - 1
hunk_old_count = hunk.old_count
replace_start = hunk_old_start + offset
replace_count = hunk_old_count
hunk_new_content: List[str] = []
for line in hunk.lines:
if line.startswith("+") and not line.startswith("+++"):
hunk_new_content.append(line[1:] + "\n")
elif line.startswith(" ") or (line and not line.startswith(("-", "+", "@@"))):
hunk_new_content.append(line + "\n")
new_lines = new_lines[:replace_start] + hunk_new_content + new_lines[replace_start + replace_count:]
offset += len(hunk_new_content) - replace_count
with open(file_path, "w", encoding="utf-8", newline="") as f:
f.writelines(new_lines)
results.append(f"Patched: {file_path}")
except (OSError, ValueError, IndexError) as e:
_patch_err_result = Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"Error patching {file_path}: {e}", source="diff_viewer.apply_patch_to_file", original=e)])
return _patch_err_result.data, _patch_err_result.errors[0].message
return True, "\n".join(results)
src/external_editor.py
+58 -4
View File
@@ -6,10 +6,64 @@ import subprocess
import tempfile
# TODO(Ed): Eliminate these?
from pathlib import Path
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, List, Dict, Any
from src.result_types import ErrorInfo, ErrorKind, Result
from src.result_types import ErrorInfo, ErrorKind, Result
from src.type_aliases import Metadata
@dataclass
class TextEditorConfig:
name: str = ""
path: str = ""
diff_args: List[str] = field(default_factory=list)
def to_dict(self) -> Metadata:
return {
"name": self.name,
"path": self.path,
"diff_args": self.diff_args,
}
@classmethod
def from_dict(cls, data: Metadata) -> "TextEditorConfig":
return cls(
name = data["name"],
path = data["path"],
diff_args = data.get("diff_args", []),
)
@dataclass
class ExternalEditorConfig:
editors: Dict[str, TextEditorConfig] = field(default_factory=dict)
default_editor: Optional[str] = None
def get_default(self) -> TextEditorConfig:
if self.default_editor and self.default_editor in self.editors:
return self.editors[self.default_editor]
if self.editors:
return next(iter(self.editors.values()))
return EMPTY_TEXT_EDITOR_CONFIG
def to_dict(self) -> Metadata:
return {
"editors": {k: v.to_dict() for k, v in self.editors.items()},
"default_editor": self.default_editor,
}
@classmethod
def from_dict(cls, data: Metadata) -> "ExternalEditorConfig":
editors = {}
for name, ed_data in data.get("editors", {}).items():
if isinstance(ed_data, dict): editors[name] = TextEditorConfig.from_dict(ed_data)
elif isinstance(ed_data, str): editors[name] = TextEditorConfig(name=name, path=ed_data)
return cls(editors=editors, default_editor=data.get("default_editor"))
EMPTY_TEXT_EDITOR_CONFIG: TextEditorConfig = TextEditorConfig()
class ExternalEditorLauncher:
@@ -47,7 +101,7 @@ class ExternalEditorLauncher:
except FileNotFoundError as e:
return Result(data=None, errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"Editor binary not found: {cmd[0]}", source="external_editor.launch_diff_result", original=e)])
_cached_vscode_config: Optional[TextEditorConfig] = None
src/gui_2.py
+354 -21
View File
@@ -24,7 +24,8 @@ if _thirdparty not in sys.path:
from contextlib import ExitStack, nullcontext
from pathlib import Path
from typing import Optional, Any
from typing import Optional, Any, Callable, Dict, List
from dataclasses import dataclass, field
from imgui_bundle import imgui, hello_imgui, immapp, imgui_node_editor as ed, imgui_color_text_edit as ced
# Lazy proxies (startup_speedup_20260606 Phase 5D)
@@ -95,12 +96,11 @@ np = _LazyModule("numpy") # was: import numpy as np
filedialog = _LazyModule("tkinter", "filedialog") # was: from tkinter import filedialog
Tk = _LazyModule("tkinter", "Tk") # was: from tkinter import Tk
from src.diff_viewer import apply_patch_to_file
from src import ai_client
from src.ai_client import VendorMetric
from src import aggregate
from src import api_hooks
from src import app_controller
from src import bg_shader
from src import cost_tracker
from src import history
from src import imgui_scopes as imscope
@@ -114,7 +114,6 @@ from src import models
from src.models import GenerateRequest, ConfirmRequest
from src import mcp_client
from src import markdown_helper
from src import shaders
from src import synthesis_formatter
from src import theme_2 as theme
from src import thinking_parser
@@ -787,7 +786,7 @@ class App:
#TODO(Ed): Remove Exception based errors.
def _get_active_capabilities(self) -> "VendorCapabilities":
from src.vendor_capabilities import VendorCapabilities, get_capabilities
from src.ai_client import VendorCapabilities, get_capabilities
#TODO(Ed): Remove Exception based errors.
try:
caps = get_capabilities(self.current_provider, self.current_model)
@@ -1095,9 +1094,9 @@ class App:
pushed_prior_tint = False
# Render background shader
bg = bg_shader.get_bg()
ws = imgui.get_io().display_size
if bg.enabled: bg.render(ws.x, ws.y)
if getattr(self, 'bg_shader_enabled', False):
ws = imgui.get_io().display_size
get_bg().render(ws.x, ws.y)
theme.render_post_fx(ws.x, ws.y, self.ai_status, self.ui_crt_filter)
@@ -5630,8 +5629,7 @@ def render_vendor_state(app: App) -> None:
| Last Error | (none) | info |
+---------------------------------------------------------+
"""
from src.vendor_state import get_vendor_state
metrics = get_vendor_state(app)
metrics = _get_vendor_state_metrics(app)
if imgui.begin_table("vendor_state", 3, imgui.TableFlags_.row_bg | imgui.TableFlags_.borders):
imgui.table_setup_column("Metric", imgui.TableColumnFlags_.width_fixed, 180)
imgui.table_setup_column("Value", imgui.TableColumnFlags_.width_stretch)
@@ -6046,7 +6044,7 @@ def render_patch_modal(app: App) -> None:
if opened:
p_min = imgui.get_window_pos()
p_max = imgui.ImVec2(p_min.x + imgui.get_window_size().x, p_min.y + imgui.get_window_size().y)
shaders.draw_soft_shadow(imgui.get_background_draw_list(), p_min, p_max, imgui.ImVec4(0, 0, 0, 0.6), 25.0, 6.0)
draw_soft_shadow(imgui.get_background_draw_list(), p_min, p_max, imgui.ImVec4(0, 0, 0, 0.6), 25.0, 6.0)
imgui.text_colored(theme.get_color("status_warning"), "Tier 4 QA Generated a Patch")
imgui.separator()
@@ -6267,13 +6265,14 @@ def render_theme_panel(app: App) -> None:
ch_ct, ctrans = imgui.slider_float("##ctrans", theme.get_child_transparency(), 0.1, 1.0, "%.2f")
if ch_ct:
theme.set_child_transparency(ctrans)
bg = bg_shader.get_bg()
ch_bg, bg.enabled = imgui.checkbox("Animated Background Shader", bg.enabled)
if ch_bg:
bg_enabled = getattr(self, 'bg_shader_enabled', False)
ch_bg, new_bg = imgui.checkbox("Animated Background Shader", bg_enabled)
if ch_bg and new_bg != bg_enabled:
self.bg_shader_enabled = new_bg
gui_cfg = app.config.setdefault("gui", {})
gui_cfg["bg_shader_enabled"] = bg.enabled
app._flush_to_config()
app.save_config()
gui_cfg["bg_shader_enabled"] = new_bg
if hasattr(app, "_flush_to_config"): app._flush_to_config()
if hasattr(app, "save_config"): app.save_config()
ch_crt, app.ui_crt_filter = imgui.checkbox("CRT Filter", app.ui_crt_filter)
if ch_crt:
@@ -7015,11 +7014,9 @@ def render_track_proposal_modal(app: App) -> None:
if app._show_track_proposal_modal:
imgui.open_popup("Track Proposal")
if imgui.begin_popup_modal("Track Proposal", True, imgui.WindowFlags_.always_auto_resize)[0]:
from src import shaders #TODO(Ed): Review local import
p_min = imgui.get_window_pos()
p_max = imgui.ImVec2(p_min.x + imgui.get_window_size().x, p_min.y + imgui.get_window_size().y)
# Render soft shadow behind the modal
shaders.draw_soft_shadow(imgui.get_background_draw_list(), p_min, p_max, imgui.ImVec4(0, 0, 0, 0.6), 25.0, 6.0)
draw_soft_shadow(imgui.get_background_draw_list(), p_min, p_max, imgui.ImVec4(0, 0, 0, 0.6), 25.0, 6.0)
if app._show_track_proposal_modal:
imgui.text_colored(C_IN(), "Proposed Implementation Tracks")
@@ -8117,7 +8114,6 @@ def request_patch_from_tier4_result(app: "App", error: str, file_context: str) -
[C: src/gui_2.py:App.request_patch_from_tier4 (L1428 legacy wrapper)]
"""
from src.diff_viewer import parse_diff
try:
patch_text = ai_client.run_tier4_patch_generation(error, file_context)
if patch_text and "---" in patch_text and "+++" in patch_text:
@@ -8438,3 +8434,340 @@ def _capture_workspace_profile_ini_result(app: "App") -> Result[str]:
#endregion: Phase 8 Property Setter / State Result Helpers
#endregion: MMA
#region: Bg Shader (moved from src/bg_shader.py)
import time as _bg_time
import math as _bg_math
_bg: _Optional["BackgroundShader"] = None
class BackgroundShader:
def __init__(self):
self.start_time = _bg_time.time()
self.ctx: _Optional[_Any] = None
def render(self, width: float, height: float):
if width <= 0 or height <= 0:
return
t = _bg_time.time() - self.start_time
dl = imgui.get_background_draw_list()
dl.add_rect_filled(imgui.ImVec2(0, 0), imgui.ImVec2(width, height), imgui.get_color_u32(imgui.ImVec4(0.01, 0.07, 0.20, 1.0)))
for i in range(3):
phase = t * (0.1 + i * 0.05)
x = (_bg_math.sin(phase) * 0.5 + 0.5) * width
y = (_bg_math.cos(phase * 0.8) * 0.5 + 0.5) * height
radius = (0.4 + 0.2 * _bg_math.sin(t * 0.2)) * max(width, height)
col = imgui.ImVec4(0.02, 0.26, 0.55, 0.3)
dl.add_circle_filled(imgui.ImVec2(x, y), radius, imgui.get_color_u32(col), num_segments=32)
num_lines = 15
for i in range(num_lines):
offset = (t * 20.0 + i * (width / num_lines)) % width
alpha = 0.1 * (1.0 + _bg_math.sin(t + i))
col = imgui.get_color_u32(imgui.ImVec4(0.08, 0.60, 0.88, alpha))
p1 = imgui.ImVec2(offset, 0)
p2 = imgui.ImVec2(offset - 100, height)
dl.add_line(p1, p2, col, thickness=2.0)
center = imgui.ImVec2(width/2, height/2)
radius = max(width, height) * 0.8
for i in range(10):
r = radius + (i * 50)
alpha = (i / 10.0) * 0.5
dl.add_circle(center, r, imgui.get_color_u32(imgui.ImVec4(0, 0, 0, alpha)), num_segments=64, thickness=60.0)
def get_bg() -> BackgroundShader:
global _bg
if _bg is None:
_bg = BackgroundShader()
return _bg
#endregion: Bg Shader
#region: Shaders (moved from src/shaders.py)
def draw_soft_shadow(draw_list: imgui.ImDrawList, p_min: imgui.ImVec2, p_max: imgui.ImVec2, color: imgui.ImVec4, shadow_size: float = 10.0, rounding: float = 0.0) -> None:
r, g, b, a = color.x, color.y, color.z, color.w
steps = int(shadow_size)
if steps <= 0: return
alpha_step = a / steps
for i in range(steps):
current_alpha = a - (i * alpha_step)
current_alpha = current_alpha * (1.0 - (i / steps)**2)
if current_alpha <= 0.01:
continue
expand = float(i)
c_min = imgui.ImVec2(p_min.x - expand, p_min.y - expand)
c_max = imgui.ImVec2(p_max.x + expand, p_max.y + expand)
u32_color = imgui.get_color_u32(imgui.ImVec4(r, g, b, current_alpha))
draw_list.add_rect(
c_min,
c_max,
u32_color,
rounding + expand if rounding > 0 else 0.0,
flags=imgui.ImDrawFlags_.round_corners_all if rounding > 0 else imgui.ImDrawFlags_.none,
thickness=1.0
)
#endregion: Shaders
#region: Diff Viewer Operations (data classes live in src/patch_modal.py alongside PendingPatch; ops that gui_2 calls live here)
import difflib as _diff_difflib
import shutil as _diff_shutil
import os as _diff_os
from pathlib import Path as _diff_Path
from typing import List as _diff_List, Optional as _diff_Optional, Tuple as _diff_Tuple
from src.patch_modal import DiffHunk, DiffFile
def parse_hunk_header(line: str) -> tuple[int, int, int, int]:
if not line.startswith("@@"): return (-1, -1, -1, -1)
parts = line.split()
if len(parts) < 2: return (-1, -1, -1, -1)
old_part = parts[1][1:]
new_part = parts[2][1:]
old_parts = old_part.split(",")
new_parts = new_part.split(",")
old_start = int(old_parts[0])
old_count = int(old_parts[1]) if len(old_parts) > 1 else 1
new_start = int(new_parts[0])
new_count = int(new_parts[1]) if len(new_parts) > 1 else 1
return (old_start, old_count, new_start, new_count)
def parse_diff(diff_text: str) -> _diff_List[DiffFile]:
if not diff_text or not diff_text.strip():
return []
files: _diff_List[DiffFile] = []
current_file: _diff_Optional[DiffFile] = None
current_hunk: _diff_Optional[DiffHunk] = None
for line in diff_text.split("\n"):
if line.startswith("--- "):
if current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
current_hunk = None
files.append(current_file)
path = line[4:]
if path.startswith("a/"):
path = path[2:]
current_file = DiffFile(old_path=path, new_path="", hunks=[])
elif line.startswith("+++ ") and current_file:
path = line[4:]
if path.startswith("b/"):
path = path[2:]
current_file.new_path = path
elif line.startswith("@@") and current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
hunk_info = parse_hunk_header(line)
if hunk_info:
old_start, old_count, new_start, new_count = hunk_info
current_hunk = DiffHunk(
header = line,
lines = [],
old_start = old_start,
old_count = old_count,
new_start = new_start,
new_count = new_count
)
else:
current_hunk = DiffHunk(
header = line,
lines = [],
old_start = 0,
old_count = 0,
new_start = 0,
new_count = 0
)
elif current_hunk is not None:
current_hunk.lines.append(line)
elif line and not line.startswith("diff ") and not line.startswith("index "):
pass
if current_file:
if current_hunk:
current_file.hunks.append(current_hunk)
files.append(current_file)
return files
def get_line_color(line: str) -> str:
if line.startswith("+"): return "green"
elif line.startswith("-"): return "red"
elif line.startswith("@@"): return "cyan"
return ""
def apply_patch_to_file(patch_text: str, base_dir: str = ".") -> _diff_Tuple[bool, str]:
diff_files = parse_diff(patch_text)
if not diff_files:
return False, "No valid diff found"
results = []
for df in diff_files:
file_path = _diff_Path(base_dir) / df.old_path
if not file_path.exists():
results.append(f"File not found: {file_path}")
continue
try:
with open(file_path, "r", encoding="utf-8") as f:
original_lines = f.read().splitlines(keepends=True)
new_lines = original_lines.copy()
offset = 0
for hunk in df.hunks:
hunk_old_start = hunk.old_start - 1
hunk_old_count = hunk.old_count
replace_start = hunk_old_start + offset
replace_count = hunk_old_count
hunk_new_content: _diff_List[str] = []
for line in hunk.lines:
if line.startswith("+") and not line.startswith("+++"):
hunk_new_content.append(line[1:] + "\n")
elif line.startswith(" ") or (line and not line.startswith(("-", "+", "@@"))):
hunk_new_content.append(line + "\n")
new_lines = new_lines[:replace_start] + hunk_new_content + new_lines[replace_start + replace_count:]
offset += len(hunk_new_content) - replace_count
with open(file_path, "w", encoding="utf-8", newline="") as f:
f.writelines(new_lines)
results.append(f"Patched: {file_path}")
except (OSError, ValueError, IndexError) as e:
_patch_err_result = Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"Error patching {file_path}: {e}", source="diff_viewer.apply_patch_to_file", original=e)])
return _patch_err_result.data, _patch_err_result.errors[0].message
return True, "\n".join(results)
#endregion: Diff Viewer Operations
def draw_soft_shadow(draw_list: imgui.ImDrawList, p_min: imgui.ImVec2, p_max: imgui.ImVec2, color: imgui.ImVec4, shadow_size: float = 10.0, rounding: float = 0.0) -> None:
r, g, b, a = color.x, color.y, color.z, color.w
steps = int(shadow_size)
if steps <= 0: return
alpha_step = a / steps
for i in range(steps):
current_alpha = a - (i * alpha_step)
current_alpha = current_alpha * (1.0 - (i / steps)**2)
if current_alpha <= 0.01:
continue
expand = float(i)
c_min = imgui.ImVec2(p_min.x - expand, p_min.y - expand)
c_max = imgui.ImVec2(p_max.x + expand, p_max.y + expand)
u32_color = imgui.get_color_u32(imgui.ImVec4(r, g, b, current_alpha))
draw_list.add_rect(
c_min,
c_max,
u32_color,
rounding + expand if rounding > 0 else 0.0,
flags=imgui.ImDrawFlags_.round_corners_all if rounding > 0 else imgui.ImDrawFlags_.none,
thickness=1.0
)
#endregion: Shaders
#region: Command Palette Modal (rendering only; registry lives in src/commands.py)
from src.commands import Command as _CpCommand, fuzzy_match as _cp_fuzzy_match, _close_palette, _execute as _cp_execute
def render_palette_modal(app: Any, commands: List[Any]) -> None:
if not getattr(app, "show_command_palette", False):
return
viewport = imgui.get_main_viewport()
center = viewport.get_center()
imgui.set_next_window_pos((center.x - 300, center.y - 200), imgui.Cond_.always)
imgui.set_next_window_size((600, 400), imgui.Cond_.always)
if not hasattr(app, "_command_palette_query"): app._command_palette_query = ""
if not hasattr(app, "_command_palette_selected"): app._command_palette_selected = 0
if not hasattr(app, "_command_palette_focused"): app._command_palette_focused = False
if not app._command_palette_focused:
imgui.set_next_window_focus()
app._command_palette_focused = True
if imgui.is_key_pressed(imgui.Key.escape):
_close_palette(app)
return
expanded, opened = imgui.begin("Command Palette##manual_slop", True, imgui.WindowFlags_.no_collapse)
if not expanded or not opened:
app.show_command_palette = False
app._command_palette_focused = False
imgui.end()
return
if not getattr(app, '_command_palette_input_focused', False):
imgui.set_keyboard_focus_here()
app._command_palette_input_focused = True
results = _cp_fuzzy_match(app._command_palette_query, commands, top_n=20)
if results: app._command_palette_selected = max(0, min(app._command_palette_selected, len(results) - 1))
else: app._command_palette_selected = 0
if imgui.is_key_pressed(imgui.Key.down_arrow):
if results:
app._command_palette_selected = min(app._command_palette_selected + 1, len(results) - 1)
if imgui.is_key_pressed(imgui.Key.up_arrow):
if results:
app._command_palette_selected = max(app._command_palette_selected - 1, 0)
if imgui.is_key_pressed(imgui.Key.enter) or imgui.is_key_pressed(imgui.Key.keypad_enter):
if results and 0 <= app._command_palette_selected < len(results):
_cp_execute(app, results[app._command_palette_selected].command)
imgui.set_next_item_width(-1)
_, app._command_palette_query = imgui.input_text("##query", app._command_palette_query)
if imgui.begin_child("##results", (0, -1)):
for i, scored in enumerate(results):
is_selected = (i == app._command_palette_selected)
label = f"[{scored.command.category}] {scored.command.title}"
clicked, _ = imgui.selectable(label, is_selected)
if clicked:
app._command_palette_selected = i
_cp_execute(app, scored.command)
if not results:
imgui.text_disabled("No matching commands.")
imgui.end_child()
imgui.end()
#endregion: Command Palette Modal
#region: Vendor State Metrics (moved from src/vendor_state.py; VendorMetric dataclass lives in src/ai_client.py)
def _get_vendor_state_metrics(app: Any) -> list[Any]:
out: list[Any] = []
out.append(VendorMetric(
key = "provider_model",
label = "Provider / Model",
value = f"{app.current_provider} / {app.current_model}",
state = "info",
tooltip = "The vendor and model that will handle the next request."
))
ctrl = getattr(app, "controller", None)
tt = getattr(ctrl, "token_tracker", None) if ctrl else None
if tt and getattr(tt, "limit", 0):
pct = 100.0 * getattr(tt, "used", 0) / tt.limit
state = "warn" if pct > 75 else "ok"
out.append(VendorMetric(
key = "context_window",
label = "Context Window",
value = f"{tt.used:,} / {tt.limit:,} ({pct:.0f}%)",
state = state,
tooltip = "Used vs total context window for the current session."
))
else:
out.append(VendorMetric(
key = "context_window", label="Context Window", value="—", state="info",
tooltip = "No token tracker attached for the current provider."
))
if tt is not None:
hits = getattr(tt, "cache_hits", 0)
miss = getattr(tt, "cache_misses", 0)
total = hits + miss
rate = (100.0 * hits / total) if total else 0.0
out.append(VendorMetric(
key = "cache", label="Cache Hit Rate",
value = f"{rate:.0f}% ({hits:,}/{total:,})",
state = "ok" if rate > 50 else "info",
tooltip = "Server-side prompt cache hit rate for the current session."
))
else:
out.append(VendorMetric(
key = "cache", label="Cache Hit Rate", value="—", state="info",
tooltip = "No token tracker attached for the current provider."
))
quota = (getattr(ctrl, "vendor_quota", {}) or {}) if ctrl else {}
pct_left = quota.get("remaining_pct")
if pct_left is None:
out.append(VendorMetric(
key = "quota", label="Vendor Quota", value="—", state="info",
tooltip = "Vendor did not report quota for the current billing period."
))
else:
out.append(VendorMetric(
key = "quota", label="Vendor Quota",
value = f"{pct_left}% remaining",
state = "ok" if pct_left > 25 else "warn",
tooltip = "Approximate quota remaining for the current billing period."
))
err = getattr(ctrl, "last_error", None) if ctrl else None
out.append(VendorMetric(
key = "last_error", label="Last Error",
value = err.get("class", "none") if err else "none",
state = "error" if err else "ok",
tooltip = err.get("message", "No error since session start.") if err else "No error since session start."
))
return out
#endregion: Vendor State Metrics
src/mcp_client.py
+123 -6
View File
@@ -62,9 +62,10 @@ import subprocess
import urllib.request
import urllib.parse
from html.parser import HTMLParser
from pathlib import Path
from typing import Optional, Callable, Any, cast
from dataclasses import dataclass, field
from html.parser import HTMLParser
from pathlib import Path
from typing import Dict, List, Optional, Callable, Any, cast
from scripts import py_struct_tools
@@ -73,7 +74,123 @@ from src import models
from src import outline_tool
from src import summarize
from src import mcp_tool_specs
from src.result_types import ErrorInfo, ErrorKind, NilPath, Result
from src.result_types import ErrorInfo, ErrorKind, NilPath, Result
from src.type_aliases import Metadata
# --------------------------------------------------------------------------- MCP config dataclasses
# Moved from src/models.py in module_taxonomy_refactor_20260627 Phase 3i.
# These are the data layer of the MCP subsystem; they belong here.
@dataclass
class MCPServerConfig:
name: str
command: Optional[str] = None
args: List[str] = field(default_factory=list)
url: Optional[str] = None
auto_start: bool = False
def to_dict(self) -> Metadata:
res = {'auto_start': self.auto_start}
if self.command: res['command'] = self.command
if self.args: res['args'] = self.args
if self.url: res['url'] = self.url
return res
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "MCPServerConfig":
return cls(
name = name,
command = data.get('command'),
args = data.get('args', []),
url = data.get('url'),
auto_start = data.get('auto_start', False),
)
@dataclass
class MCPConfiguration:
mcpServers: Dict[str, MCPServerConfig] = field(default_factory=dict)
def to_dict(self) -> Metadata:
return {'mcpServers': {name: cfg.to_dict() for name, cfg in self.mcpServers.items()}}
@classmethod
def from_dict(cls, data: Metadata) -> "MCPConfiguration":
raw_servers = data.get('mcpServers', {})
parsed_servers = {name: MCPServerConfig.from_dict(name, cfg) for name, cfg in raw_servers.items()}
return cls(mcpServers=parsed_servers)
@dataclass
class VectorStoreConfig:
provider: str
url: Optional[str] = None
api_key: Optional[str] = None
collection_name: str = 'manual_slop'
mcp_server: Optional[str] = None
mcp_tool: Optional[str] = None
def to_dict(self) -> Metadata:
return {
"provider": self.provider,
"url": self.url,
"api_key": self.api_key,
"collection_name": self.collection_name,
"mcp_server": self.mcp_server,
"mcp_tool": self.mcp_tool,
}
@classmethod
def from_dict(cls, data: Metadata) -> "VectorStoreConfig":
return cls(
provider = data["provider"],
url = data.get("url"),
api_key = data.get("api_key"),
collection_name = data.get("collection_name", "manual_slop"),
mcp_server = data.get("mcp_server"),
mcp_tool = data.get("mcp_tool"),
)
@dataclass
class RAGConfig:
enabled: bool = False
vector_store: VectorStoreConfig = field(default_factory=lambda: VectorStoreConfig(provider='mock'))
embedding_provider: str = 'gemini'
chunk_size: int = 1000
chunk_overlap: int = 200
def to_dict(self) -> Metadata:
return {
"enabled": self.enabled,
"vector_store": self.vector_store.to_dict(),
"embedding_provider": self.embedding_provider,
"chunk_size": self.chunk_size,
"chunk_overlap": self.chunk_overlap,
}
@classmethod
def from_dict(cls, data: Metadata) -> "RAGConfig":
return cls(
enabled = data.get("enabled", False),
vector_store = VectorStoreConfig.from_dict(data.get("vector_store", {"provider": "mock"})),
embedding_provider = data.get("embedding_provider", "gemini"),
chunk_size = data.get("chunk_size", 1000),
chunk_overlap = data.get("chunk_overlap", 200),
)
def load_mcp_config(path: str) -> MCPConfiguration:
if not os.path.exists(path):
return MCPConfiguration()
with open(path, 'r', encoding='utf-8') as f:
try:
data = json.load(f)
return MCPConfiguration.from_dict(data)
except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
_mcp_err = Result(data=MCPConfiguration(), errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"failed to load MCP config: {e}", source="mcp_client.load_mcp_config", original=e)])
return _mcp_err.data
# ------------------------------------------------------------------ mutating tools sentinel
@@ -1621,7 +1738,7 @@ def get_ui_performance() -> str:
# ------------------------------------------------------------------ tool dispatch
class StdioMCPServer:
def __init__(self, config: models.MCPServerConfig):
def __init__(self, config: MCPServerConfig):
self.config = config
self.name = config.name
self.proc = None
@@ -1720,7 +1837,7 @@ class ExternalMCPManager:
"""Initialize the manager with an empty server registry."""
self.servers = {}
async def add_server(self, config: models.MCPServerConfig):
async def add_server(self, config: MCPServerConfig):
"""
Add and start a new MCP server from a configuration object.
[C: tests/test_external_mcp.py:test_external_mcp_real_process, tests/test_external_mcp.py:test_get_tool_schemas_includes_external]
src/mma.py
+227
View File
@@ -0,0 +1,227 @@
"""MMA (Multi-Model Architecture) core data structures.
Per module_taxonomy_refactor_20260627 Phase 3.1, the MMA Core (ThinkingSegment,
Ticket, Track, WorkerContext, TrackMetadata, TrackState) moved from
src/models.py to this module. The data domain is the ticket/track lifecycle
that drives the 4-Tier MMA execution.
The boundary wire schema `Metadata` (TypeAlias = dict[str, Any]) is
NOT defined here; it lives in src/type_aliases.py. This module's
TrackMetadata dataclass is the *typed* counterpart used for Track-level
metadata (id/name/status/timestamps).
"""
from __future__ import annotations
import datetime
from dataclasses import dataclass, field
from typing import List, Optional
from src.type_aliases import Metadata
@dataclass
class ThinkingSegment:
content: str
marker: str
def to_dict(self) -> Metadata:
return {"content": self.content, "marker": self.marker}
@classmethod
def from_dict(cls, data: Metadata) -> "ThinkingSegment":
return cls(content=data["content"], marker=data["marker"])
@dataclass
class Ticket:
id: str
description: str
target_symbols: List[str] = field(default_factory=list)
context_requirements: List[str] = field(default_factory=list)
depends_on: List[str] = field(default_factory=list)
status: str = "todo"
assigned_to: str = "unassigned"
priority: str = "medium"
target_file: Optional[str] = None
blocked_reason: Optional[str] = None
step_mode: bool = False
retry_count: int = 0
manual_block: bool = False
model_override: Optional[str] = None
persona_id: Optional[str] = None
def mark_blocked(self, reason: str) -> None:
self.status = "blocked"
self.blocked_reason = reason
def mark_manual_block(self, reason: str) -> None:
self.status = "blocked"
self.blocked_reason = f"[MANUAL] {reason}"
self.manual_block = True
def clear_manual_block(self) -> None:
if self.manual_block:
self.status = "todo"
self.blocked_reason = None
self.manual_block = False
def mark_complete(self) -> None:
self.status = "completed"
def to_dict(self) -> Metadata:
return {
"id": self.id,
"description": self.description,
"status": self.status,
"assigned_to": self.assigned_to,
"priority": self.priority,
"target_file": self.target_file,
"target_symbols": self.target_symbols,
"context_requirements": self.context_requirements,
"depends_on": self.depends_on,
"blocked_reason": self.blocked_reason,
"step_mode": self.step_mode,
"retry_count": self.retry_count,
"manual_block": self.manual_block,
"model_override": self.model_override,
"persona_id": self.persona_id,
}
@classmethod
def from_dict(cls, data: Metadata) -> "Ticket":
return cls(
id = data["id"],
description = data.get("description", ""),
status = data.get("status", "todo"),
assigned_to = data.get("assigned_to", "unassigned"),
priority = data.get("priority", "medium"),
target_file = data.get("target_file"),
target_symbols = data.get("target_symbols", []),
context_requirements = data.get("context_requirements", []),
depends_on = data.get("depends_on", []),
blocked_reason = data.get("blocked_reason"),
step_mode = data.get("step_mode", False),
retry_count = data.get("retry_count", 0),
manual_block = data.get("manual_block", False),
model_override = data.get("model_override"),
persona_id = data.get("persona_id"),
)
@dataclass
class Track:
id: str
description: str
tickets: List["Ticket"] = field(default_factory=list)
def to_dict(self) -> Metadata:
return {
"id": self.id,
"description": self.description,
"tickets": [t.to_dict() for t in self.tickets],
}
@classmethod
def from_dict(cls, data: Metadata) -> "Track":
return cls(
id = data["id"],
description = data.get("description", ""),
tickets = [Ticket.from_dict(t) for t in data.get("tickets", [])],
)
@dataclass
class WorkerContext:
ticket_id: str
model_name: str
messages: list[Metadata] = field(default_factory=list)
tool_preset: Optional[str] = None
persona_id: Optional[str] = None
@dataclass
class TrackMetadata:
id: str
name: str
status: Optional[str] = None
created_at: Optional[datetime.datetime] = None
updated_at: Optional[datetime.datetime] = None
def to_dict(self) -> Metadata:
return {
"id": self.id,
"name": self.name,
"status": self.status,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
@classmethod
def from_dict(cls, data: Metadata) -> "TrackMetadata":
created = data.get("created_at")
updated = data.get("updated_at")
if isinstance(created, str):
try:
created = datetime.datetime.fromisoformat(created)
except ValueError:
created = None
if isinstance(updated, str):
try:
updated = datetime.datetime.fromisoformat(updated)
except ValueError:
updated = None
return cls(
id = data["id"],
name = data.get("name", ""),
status = data.get("status"),
created_at = created,
updated_at = updated,
)
@dataclass
class TrackState:
metadata: Metadata = field(default_factory=dict)
discussion: List[Metadata] = field(default_factory=list)
tasks: List["Ticket"] = field(default_factory=list)
def to_dict(self) -> Metadata:
serialized_discussion = []
for item in self.discussion:
if isinstance(item, dict):
new_item = dict(item)
if "ts" in new_item and isinstance(new_item["ts"], datetime.datetime):
new_item["ts"] = new_item["ts"].isoformat()
serialized_discussion.append(new_item)
else:
serialized_discussion.append(item)
return {
"metadata": self.metadata.to_dict(),
"discussion": serialized_discussion,
"tasks": [t.to_dict() for t in self.tasks],
}
@classmethod
def from_dict(cls, data: Metadata) -> "TrackState":
discussion = data.get("discussion", [])
parsed_discussion = []
for item in discussion:
if isinstance(item, dict):
new_item = dict(item)
ts = new_item.get("ts")
if isinstance(ts, str):
try:
new_item["ts"] = datetime.datetime.fromisoformat(ts)
except ValueError:
pass
parsed_discussion.append(new_item)
else:
parsed_discussion.append(item)
return cls(
metadata = TrackMetadata.from_dict(data["metadata"]),
discussion = parsed_discussion,
tasks = [Ticket.from_dict(t) for t in data.get("tasks", [])],
)
EMPTY_TRACK_STATE: TrackState = TrackState()
src/models.py
+100 -1122
View File
File diff suppressed because it is too large Load Diff
src/patch_modal.py
+15
View File
@@ -2,6 +2,21 @@ from dataclasses import dataclass, field
from typing import Optional, Callable, List
@dataclass
class DiffHunk:
header: str
lines: List[str]
old_start: int
old_count: int
new_start: int
new_count: int
@dataclass
class DiffFile:
old_path: str
new_path: str
hunks: List[DiffHunk] = field(default_factory=list)
@dataclass
class PendingPatch:
patch_text: str = ""
src/personas.py
+96 -28
View File
@@ -1,10 +1,103 @@
"""Personas module: Persona dataclass + PersonaManager CRUD.
Per module_taxonomy_refactor_20260627 Phase 3.4, the Persona dataclass
moved from src/models.py into this module. PersonaManager (the ops layer
that loads/saves Persona instances to TOML) was already here.
"""
from __future__ import annotations
import tomllib
import tomli_w
from pathlib import Path
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Any, Optional
from src import paths
from src.type_aliases import Metadata
@dataclass
class Persona:
name: str
preferred_models: list[Metadata] = field(default_factory=list)
system_prompt: str = ''
tool_preset: Optional[str] = None
bias_profile: Optional[str] = None
context_preset: Optional[str] = None
aggregation_strategy: Optional[str] = None
@property
def provider(self) -> str:
if not self.preferred_models: return ""
return self.preferred_models[0].get("provider") or ""
@property
def model(self) -> str:
if not self.preferred_models: return ""
return self.preferred_models[0].get("model") or ""
@property
def temperature(self) -> float:
if not self.preferred_models: return 0.0
return float(self.preferred_models[0].get("temperature") or 0.0)
@property
def top_p(self) -> float:
if not self.preferred_models: return 1.0
return float(self.preferred_models[0].get("top_p") or 1.0)
@property
def max_output_tokens(self) -> int:
if not self.preferred_models: return 0
return int(self.preferred_models[0].get("max_output_tokens") or 0)
def to_dict(self) -> Metadata:
res = {"system_prompt": self.system_prompt}
if self.preferred_models:
processed = []
for m in self.preferred_models:
if isinstance(m, str):
processed.append({"model": m})
else:
processed.append(m)
res["preferred_models"] = processed
if self.tool_preset is not None: res["tool_preset"] = self.tool_preset
if self.bias_profile is not None: res["bias_profile"] = self.bias_profile
if self.context_preset is not None: res["context_preset"] = self.context_preset
if self.aggregation_strategy is not None: res["aggregation_strategy"] = self.aggregation_strategy
return res
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "Persona":
raw_models = data.get("preferred_models", [])
parsed_models = []
for m in raw_models:
if isinstance(m, str):
parsed_models.append({"model": m})
else:
parsed_models.append(m)
legacy = {}
for k in ["provider", "model", "temperature", "top_p", "max_output_tokens"]:
if data.get(k) is not None:
legacy[k] = data[k]
if legacy:
if not parsed_models:
parsed_models.append(legacy)
else:
for k, v in legacy.items():
if k not in parsed_models[0] or parsed_models[0][k] is None:
parsed_models[0][k] = v
return cls(
name = name,
preferred_models = parsed_models,
system_prompt = data.get("system_prompt", ""),
tool_preset = data.get("tool_preset"),
bias_profile = data.get("bias_profile"),
context_preset = data.get("context_preset"),
aggregation_strategy = data.get("aggregation_strategy"),
)
class PersonaManager:
"""Manages Persona profiles across global and project-specific files."""
@@ -13,9 +106,6 @@ class PersonaManager:
self.project_root = project_root
def _get_path(self, scope: str) -> Path:
"""
[C: src/tool_presets.py:ToolPresetManager.delete_bias_profile, src/tool_presets.py:ToolPresetManager.delete_preset, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.delete_profile, src/workspace_manager.py:WorkspaceManager.save_profile]
"""
if scope == "global":
return paths.get_global_personas_path()
elif scope == "project":
@@ -26,34 +116,23 @@ class PersonaManager:
raise ValueError("Invalid scope, must be 'global' or 'project'")
def load_all(self) -> Dict[str, Persona]:
"""
Merges global and project personas into a single dictionary.
[C: tests/test_persona_manager.py:test_delete_persona, tests/test_persona_manager.py:test_load_all_merged, tests/test_persona_manager.py:test_save_persona, tests/test_preset_manager.py:test_delete_preset, tests/test_preset_manager.py:test_load_all_merged, tests/test_preset_manager.py:test_save_preset_global, tests/test_preset_manager.py:test_save_preset_project, tests/test_presets.py:TestPresetManager.test_delete_preset, tests/test_presets.py:TestPresetManager.test_project_overwrites_global, tests/test_presets.py:TestPresetManager.test_save_and_load_global, tests/test_presets.py:TestPresetManager.test_save_and_load_project]
"""
personas = {}
global_path = paths.get_global_personas_path()
global_data = self._load_file(global_path)
for name, data in global_data.get("personas", {}).items():
personas[name] = Persona.from_dict(name, data)
if self.project_root:
project_path = paths.get_project_personas_path(self.project_root)
project_data = self._load_file(project_path)
for name, data in project_data.get("personas", {}).items():
personas[name] = Persona.from_dict(name, data)
return personas
def save_persona(self, persona: Persona, scope: str = "project") -> None:
"""
[C: tests/test_persona_manager.py:test_save_persona]
"""
path = self._get_path(scope)
data = self._load_file(path)
if "personas" not in data:
data["personas"] = {}
data["personas"][persona.name] = persona.to_dict()
self._save_file(path, data)
@@ -64,18 +143,13 @@ class PersonaManager:
project_data = self._load_file(project_path)
if name in project_data.get("personas", {}):
return "project"
global_path = paths.get_global_personas_path()
global_data = self._load_file(global_path)
if name in global_data.get("personas", {}):
return "global"
return "project"
def delete_persona(self, name: str, scope: str = "project") -> None:
"""
[C: tests/test_persona_manager.py:test_delete_persona]
"""
path = self._get_path(scope)
data = self._load_file(path)
if "personas" in data and name in data["personas"]:
@@ -83,9 +157,6 @@ class PersonaManager:
self._save_file(path, data)
def _load_file(self, path: Path) -> Dict[str, Any]:
"""
[C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.get_preset_scope, src/presets.py:PresetManager.load_all, src/presets.py:PresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.delete_profile, src/workspace_manager.py:WorkspaceManager.load_all_profiles, src/workspace_manager.py:WorkspaceManager.save_profile]
"""
if not path.exists():
return {}
try:
@@ -95,9 +166,6 @@ class PersonaManager:
return {}
def _save_file(self, path: Path, data: Dict[str, Any]) -> None:
"""
[C: src/presets.py:PresetManager.delete_preset, src/presets.py:PresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.delete_profile, src/workspace_manager.py:WorkspaceManager.save_profile]
"""
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "wb") as f:
tomli_w.dump(data, f)
tomli_w.dump(data, f)
src/project.py
+171
View File
@@ -0,0 +1,171 @@
"""Project configuration dataclasses and config I/O helpers.
Per module_taxonomy_refactor_20260627 Phase 3b, the project context
(ProjectContext + 5 sub-dataclasses) and config I/O helpers moved
from src/models.py to this module.
Per the 4-criteria decision rule:
- C1 (cross-system usage >= 3 systems): YES (project_manager, aggregate,
api_hooks, app_controller, gui_2, orchestrator_pm, tests)
- C2 (state machine / lifecycle): NO (just config; no state transitions)
- C3 (test file already exists): YES (test_project_context_20260627.py)
- C4 (substantial size): YES (ProjectContext + 5 sub + 3 helpers + 60+ lines)
Therefore: DEDICATED FILE = src/project.py
"""
from __future__ import annotations
import re
import tomllib
from dataclasses import dataclass, field
from typing import Any, List
from src.paths import get_config_path
from src.type_aliases import Metadata
# --------------------------------------------------------------------------- ProjectContext
@dataclass(frozen=True, slots=True)
class ProjectMeta:
name: str = ""
summary_only: bool = False
execution_mode: str = "standard"
@dataclass(frozen=True, slots=True)
class ProjectOutput:
namespace: str = "project"
output_dir: str = ""
@dataclass(frozen=True, slots=True)
class ProjectFiles:
base_dir: str = ""
paths: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True)
class ProjectScreenshots:
base_dir: str = "."
paths: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True)
class ProjectDiscussion:
roles: tuple[str, ...] = ()
history: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True)
class ProjectContext:
"""Typed return type for project_manager.flat_config(). Replaces the dict[str, Any] that flat_config() returned. Per conductor/tracks/cruft_elimination_20260627/SPEC_CORRECTION_phase_2.md."""
project: ProjectMeta = field(default_factory=ProjectMeta)
output: ProjectOutput = field(default_factory=ProjectOutput)
files: ProjectFiles = field(default_factory=ProjectFiles)
screenshots: ProjectScreenshots = field(default_factory=ProjectScreenshots)
context_presets: Metadata = field(default_factory=dict)
discussion: ProjectDiscussion = field(default_factory=ProjectDiscussion)
def to_dict(self) -> Metadata:
return {
"project": {
"name": self.project.name,
"summary_only": self.project.summary_only,
"execution_mode": self.project.execution_mode,
},
"output": {
"namespace": self.output.namespace,
"output_dir": self.output.output_dir,
},
"files": {
"base_dir": self.files.base_dir,
"paths": list(self.files.paths),
},
"screenshots": {
"base_dir": self.screenshots.base_dir,
"paths": list(self.screenshots.paths),
},
"context_presets": dict(self.context_presets),
"discussion": {
"roles": list(self.discussion.roles),
"history": list(self.discussion.history),
},
}
def __getitem__(self, key: str) -> Any:
return self.to_dict()[key]
def get(self, key: str, default: Any = None) -> Any:
return self.to_dict().get(key, default)
EMPTY_PROJECT_CONTEXT: ProjectContext = ProjectContext()
# --------------------------------------------------------------------------- Config IO helpers
def _clean_nones(data: Any) -> Any:
if isinstance(data, dict):
return {k: _clean_nones(v) for k, v in data.items() if v is not None}
elif isinstance(data, list):
return [_clean_nones(v) for v in data if v is not None]
return data
def load_config_from_disk() -> Metadata:
"""
Re-read the global config.toml from disk and return the parsed
dict. The single source of truth for the in-memory config is
the AppController's self.config attribute; this function is the
disk I/O primitive that the controller owns. Direct callers in
src/ are an architectural smell (bypassing the state owner) and
will be flagged by scripts/audit_no_models_config_io.py.
[C: src/app_controller.py:AppController.load_config, src/app_controller.py:AppController.__init__]
"""
with open(get_config_path(), "rb") as f:
return tomllib.load(f)
def save_config_to_disk(config: Metadata) -> None:
# tomli_w is loaded on-demand (sub-track 2 of startup_speedup_20260606).
# If it's already in sys.modules (e.g. warmed up or loaded by a prior
# call), the import is a fast lookup; otherwise it's a cold load paid
# only when the user actually saves config.
import tomli_w
config = _clean_nones(config)
with open(get_config_path(), "wb") as f:
tomli_w.dump(config, f)
# --------------------------------------------------------------------------- History utilities
def parse_history_entries(history_strings: list[str], roles: list[str]) -> list[Metadata]:
import re
from src import thinking_parser
entries = []
for raw in history_strings:
ts = ""
rest = raw
if rest.startswith("@"):
nl = rest.find("\n")
if nl != -1:
ts = rest[1:nl]
rest = rest[nl + 1:]
known = roles or ["User", "AI", "Vendor API", "System"]
role_pat = re.compile(r"^(" + "|".join(re.escape(r) for r in known) + r"):", re.IGNORECASE)
match = role_pat.match(rest)
role = match.group(1) if match else "User"
if match:
content = rest[match.end():].strip()
else:
content = rest
entry_obj = {"role": role, "content": content, "collapsed": True, "ts": ts}
if role == "AI" and ("<thinking>" in content or "<thought>" in content or "Thinking:" in content):
segments, parsed_content = thinking_parser.parse_thinking_trace(content)
if segments:
entry_obj["content"] = parsed_content
entry_obj["thinking_segments"] = [{"content": s.content, "marker": s.marker} for s in segments]
entries.append(entry_obj)
return entries
src/project_files.py
+188
View File
@@ -0,0 +1,188 @@
"""File-related project state dataclasses.
Per module_taxonomy_refactor_20260627 Phase 3c, the file + view + preset
dataclasses moved from src/models.py to this module.
Per the 4-criteria decision rule:
- C1 (cross-system usage >= 3 systems): YES (aggregate, app_controller,
gui_2, presets, context_presets, tests)
- C2 (state machine / lifecycle): NO (just data; no state transitions)
- C3 (test file already exists): YES (test_file_item_model.py,
test_view_presets.py, test_context_presets_*.py)
- C4 (substantial size): YES (FileItem has 10 fields + __post_init__ +
to_dict/from_dict; ContextPreset nests ContextFileEntry + list)
Therefore: DEDICATED FILE = src/project_files.py
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
from src.type_aliases import Metadata
@dataclass
class FileItem:
path: str
auto_aggregate: bool = True
force_full: bool = False
view_mode: str = 'full'
selected: bool = False
ast_signatures: bool = False
ast_definitions: bool = False
ast_mask: dict[str, str] = field(default_factory=dict)
custom_slices: list[dict] = field(default_factory=list)
injected_at: Optional[float] = None
def __post_init__(self):
if self.custom_slices:
normalized = []
for slc in self.custom_slices:
if isinstance(slc, dict):
new_slc = slc.copy()
if "tag" not in new_slc: new_slc["tag"] = None
if "comment" not in new_slc: new_slc["comment"] = None
normalized.append(new_slc)
else:
normalized.append(slc)
self.custom_slices = normalized
def to_dict(self) -> Metadata:
"""
[C: src/personas.py:PersonaManager.save_persona, src/presets.py:PresetManager.save_preset, src/project_manager.py:save_project, src/project_manager.py:save_track_state, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.save_profile, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_serialization_with_annotations, tests/test_event_serialization.py:test_user_request_event_serialization, tests/test_external_editor.py:TestExternalEditorConfig.test_to_dict, tests/test_external_editor.py:TestTextEditorConfig.test_to_dict, tests/test_file_item_model.py:test_file_item_to_dict, tests/test_gui_events_v2.py:test_user_request_event_payload, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_serialization, tests/test_persona_id.py:test_ticket_persona_id_serialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_serialization, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_thinking_gui.py:test_thinking_segment_model_compatibility, tests/test_ticket_queue.py:test_ticket_to_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_to_dict, tests/test_track_state_schema.py:test_track_state_to_dict_with_none, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return {
"path": self.path,
"auto_aggregate": self.auto_aggregate,
"force_full": self.force_full,
"view_mode": self.view_mode,
"ast_signatures": self.ast_signatures,
"ast_definitions": self.ast_definitions,
"ast_mask": self.ast_mask,
"custom_slices": self.custom_slices,
"injected_at": self.injected_at,
}
@classmethod
def from_dict(cls, data: Metadata) -> "FileItem":
"""
[C: src/personas.py:PersonaManager.load_all, src/presets.py:PresetManager.load_all, src/project_manager.py:load_project, src/project_manager.py:load_track_state, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets, src/workspace_manager.py:WorkspaceManager.load_all_profiles, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_from_dict_legacy, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_deserialization_with_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_dict_editors, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_string_editors, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_with_diff_args, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_without_diff_args, tests/test_file_item_model.py:test_file_item_from_dict, tests/test_file_item_model.py:test_file_item_from_dict_defaults, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_default_on_deserialize, tests/test_per_ticket_model.py:test_model_override_deserialization, tests/test_persona_id.py:test_ticket_persona_id_deserialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_deserialization, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_ticket_queue.py:test_ticket_from_dict_default_priority, tests/test_ticket_queue.py:test_ticket_from_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_from_dict, tests/test_track_state_schema.py:test_track_state_from_dict_empty_and_missing, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return cls(
path = data["path"],
auto_aggregate = data.get("auto_aggregate", True),
force_full = data.get("force_full", False),
view_mode = data.get("view_mode", 'full'),
ast_signatures = data.get("ast_signatures", False),
ast_definitions = data.get("ast_definitions", False),
ast_mask = data.get("ast_mask", {}),
custom_slices = data.get("custom_slices", []),
injected_at = data.get("injected_at"),
)
@dataclass
class Preset:
name: str
system_prompt: str
def to_dict(self) -> Metadata:
"""
[C: src/personas.py:PersonaManager.save_persona, src/presets.py:PresetManager.save_preset, src/project_manager.py:save_project, src/project_manager.py:save_track_state, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.save_profile, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_serialization_with_annotations, tests/test_event_serialization.py:test_user_request_event_serialization, tests/test_external_editor.py:TestExternalEditorConfig.test_to_dict, tests/test_external_editor.py:TestTextEditorConfig.test_to_dict, tests/test_file_item_model.py:test_file_item_to_dict, tests/test_gui_events_v2.py:test_user_request_event_payload, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_serialization, tests/test_persona_id.py:test_ticket_persona_id_serialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_serialization, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_thinking_gui.py:test_thinking_segment_model_compatibility, tests/test_ticket_queue.py:test_ticket_to_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_to_dict, tests/test_track_state_schema.py:test_track_state_to_dict_with_none, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return {"system_prompt": self.system_prompt}
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "Preset":
"""
[C: src/personas.py:PersonaManager.load_all, src/presets.py:PresetManager.load_all, src/project_manager.py:load_project, src/project_manager.py:load_track_state, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets, src/workspace_manager.py:WorkspaceManager.load_all_profiles, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_from_dict_legacy, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_deserialization_with_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_dict_editors, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_string_editors, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_with_diff_args, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_without_diff_args, tests/test_file_item_model.py:test_file_item_from_dict, tests/test_file_item_model.py:test_file_item_from_dict_defaults, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_default_on_deserialize, tests/test_per_ticket_model.py:test_model_override_deserialization, tests/test_persona_id.py:test_ticket_persona_id_deserialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_deserialization, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_ticket_queue.py:test_ticket_from_dict_default_priority, tests/test_ticket_queue.py:test_ticket_from_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_from_dict, tests/test_track_state_schema.py:test_track_state_from_dict_empty_and_missing, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return cls(name=name, system_prompt=data.get("system_prompt", ""))
@dataclass
class ContextFileEntry:
path: str
view_mode: str = "summary"
custom_slices: list = field(default_factory=list)
ast_mask: dict = field(default_factory=dict)
ast_signatures: bool = False
ast_definitions: bool = False
def to_dict(self) -> Metadata:
"""
[C: src/personas.py:PersonaManager.save_persona, src/presets.py:PresetManager.save_preset, src/project_manager.py:save_project, src/project_manager.py:save_track_state, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.save_profile, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_serialization_with_annotations, tests/test_event_serialization.py:test_user_request_event_serialization, tests/test_external_editor.py:TestExternalEditorConfig.test_to_dict, tests/test_external_editor.py:TestTextEditorConfig.test_to_dict, tests/test_file_item_model.py:test_file_item_to_dict, tests/test_gui_events_v2.py:test_user_request_event_payload, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_serialization, tests/test_persona_id.py:test_ticket_persona_id_serialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_serialization, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_thinking_gui.py:test_thinking_segment_model_compatibility, tests/test_ticket_queue.py:test_ticket_to_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_to_dict, tests/test_track_state_schema.py:test_track_state_to_dict_with_none, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return {"path": self.path, "view_mode": self.view_mode, "custom_slices": self.custom_slices, "ast_mask": self.ast_mask, "ast_signatures": self.ast_signatures, "ast_definitions": self.ast_definitions}
@classmethod
def from_dict(cls, data: Metadata) -> "ContextFileEntry":
"""
[C: src/personas.py:PersonaManager.load_all, src/presets.py:PresetManager.load_all, src/project_manager.py:load_project, src/project_manager.py:load_track_state, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets, src/workspace_manager.py:WorkspaceManager.load_all_profiles, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_from_dict_legacy, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_deserialization_with_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_dict_editors, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_string_editors, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_with_diff_args, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_without_diff_args, tests/test_file_item_model.py:test_file_item_from_dict, tests/test_file_item_model.py:test_file_item_from_dict_defaults, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_default_on_deserialize, tests/test_per_ticket_model.py:test_model_override_deserialization, tests/test_persona_id.py:test_ticket_persona_id_deserialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_deserialization, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_ticket_queue.py:test_ticket_from_dict_default_priority, tests/test_ticket_queue.py:test_ticket_from_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_from_dict, tests/test_track_state_schema.py:test_track_state_from_dict_empty_and_missing, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return cls(
path = data.get("path", ""),
view_mode = data.get("view_mode", "summary"),
custom_slices = data.get("custom_slices", []),
ast_mask = data.get("ast_mask", {}),
ast_signatures = data.get("ast_signatures", False),
ast_definitions = data.get("ast_definitions", False),
)
@dataclass
class NamedViewPreset:
name: str
view_mode: str
ast_mask: dict = field(default_factory=dict)
custom_slices: list = field(default_factory=list)
def to_dict(self) -> Metadata:
"""
[C: src/personas.py:PersonaManager.save_persona, src/presets.py:PresetManager.save_preset, src/project_manager.py:save_project, src/project_manager.py:save_track_state, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.save_profile, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_serialization_with_annotations, tests/test_event_serialization.py:test_user_request_event_serialization, tests/test_external_editor.py:TestExternalEditorConfig.test_to_dict, tests/test_external_editor.py:TestTextEditorConfig.test_to_dict, tests/test_file_item_model.py:test_file_item_to_dict, tests/test_gui_events_v2.py:test_user_request_event_payload, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_serialization, tests/test_persona_id.py:test_ticket_persona_id_serialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_serialization, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_thinking_gui.py:test_thinking_segment_model_compatibility, tests/test_ticket_queue.py:test_ticket_to_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_to_dict, tests/test_track_state_schema.py:test_track_state_to_dict_with_none, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return {"name": self.name, "view_mode": self.view_mode, "ast_mask": self.ast_mask, "custom_slices": self.custom_slices}
@classmethod
def from_dict(cls, data: Metadata) -> "NamedViewPreset":
"""
[C: src/personas.py:PersonaManager.load_all, src/presets.py:PresetManager.load_all, src/project_manager.py:load_project, src/project_manager.py:load_track_state, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets, src/workspace_manager.py:WorkspaceManager.load_all_profiles, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_from_dict_legacy, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_deserialization_with_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_dict_editors, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_string_editors, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_with_diff_args, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_without_diff_args, tests/test_file_item_model.py:test_file_item_from_dict, tests/test_file_item_model.py:test_file_item_from_dict_defaults, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_default_on_deserialize, tests/test_per_ticket_model.py:test_model_override_deserialization, tests/test_persona_id.py:test_ticket_persona_id_deserialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_deserialization, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_ticket_queue.py:test_ticket_from_dict_default_priority, tests/test_ticket_queue.py:test_ticket_from_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_from_dict, tests/test_track_state_schema.py:test_track_state_from_dict_empty_and_missing, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return cls(
name = data.get("name", ""),
view_mode = data.get("view_mode", "summary"),
ast_mask = data.get("ast_mask", {}),
custom_slices = data.get("custom_slices", []),
)
@dataclass
class ContextPreset:
name: str
files: list[ContextFileEntry] = field(default_factory=list)
screenshots: list[str] = field(default_factory=list)
description: str = ""
def to_dict(self) -> Metadata:
"""
[C: src/personas.py:PersonaManager.save_persona, src/presets.py:PresetManager.save_preset, src/project_manager.py:save_project, src/project_manager.py:save_track_state, src/tool_presets.py:ToolPresetManager.save_bias_profile, src/tool_presets.py:ToolPresetManager.save_preset, src/workspace_manager.py:WorkspaceManager.save_profile, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_serialization_with_annotations, tests/test_event_serialization.py:test_user_request_event_serialization, tests/test_external_editor.py:TestExternalEditorConfig.test_to_dict, tests/test_external_editor.py:TestTextEditorConfig.test_to_dict, tests/test_file_item_model.py:test_file_item_to_dict, tests/test_gui_events_v2.py:test_user_request_event_payload, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_serialization, tests/test_persona_id.py:test_ticket_persona_id_serialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_serialization, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_thinking_gui.py:test_thinking_segment_model_compatibility, tests/test_ticket_queue.py:test_ticket_to_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_to_dict, tests/test_track_state_schema.py:test_track_state_to_dict_with_none, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
return {
"files": [f.to_dict() for f in self.files],
"screenshots": self.screenshots,
"description": self.description,
}
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "ContextPreset":
"""
[C: src/personas.py:PersonaManager.load_all, src/presets.py:PresetManager.load_all, src/project_manager.py:load_project, src/project_manager.py:load_track_state, src/tool_presets.py:ToolPresetManager.load_all_bias_profiles, src/tool_presets.py:ToolPresetManager.load_all_presets, src/workspace_manager.py:WorkspaceManager.load_all_profiles, tests/test_bias_models.py:test_bias_profile_model, tests/test_bias_models.py:test_tool_model, tests/test_bias_models.py:test_tool_preset_extension, tests/test_context_presets_models.py:test_context_preset_from_dict_legacy, tests/test_context_presets_models.py:test_context_preset_serialization, tests/test_context_presets_models.py:test_file_view_preset_serialization, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_deserialization_with_annotations, tests/test_custom_slices_annotations.py:test_file_item_custom_slices_round_trip_annotations, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_dict_editors, tests/test_external_editor.py:TestExternalEditorConfig.test_from_dict_with_string_editors, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_with_diff_args, tests/test_external_editor.py:TestTextEditorConfig.test_from_dict_without_diff_args, tests/test_file_item_model.py:test_file_item_from_dict, tests/test_file_item_model.py:test_file_item_from_dict_defaults, tests/test_history_manager.py:TestHistoryManager.test_snapshot_roundtrip, tests/test_mcp_config.py:test_mcp_configuration_to_from_dict, tests/test_mcp_config.py:test_mcp_server_config_to_from_dict, tests/test_per_ticket_model.py:test_model_override_default_on_deserialize, tests/test_per_ticket_model.py:test_model_override_deserialization, tests/test_persona_id.py:test_ticket_persona_id_deserialization, tests/test_persona_models.py:test_persona_defaults, tests/test_persona_models.py:test_persona_deserialization, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations, tests/test_ticket_queue.py:test_ticket_from_dict_default_priority, tests/test_ticket_queue.py:test_ticket_from_dict_priority, tests/test_tiered_aggregation.py:test_persona_aggregation_strategy, tests/test_track_state_schema.py:test_track_state_from_dict, tests/test_track_state_schema.py:test_track_state_from_dict_empty_and_missing, tests/test_ui_summary_only_removal.py:test_file_item_serialization_with_flags]
"""
files_data = data.get("files", [])
return cls(
name = name,
files = [ContextFileEntry.from_dict(f) if isinstance(f, dict) else ContextFileEntry(path=str(f)) for f in files_data],
screenshots = data.get("screenshots", []),
description = data.get("description", ""),
)
src/shaders.py
-35
View File
@@ -1,35 +0,0 @@
from imgui_bundle import imgui
def draw_soft_shadow(draw_list: imgui.ImDrawList, p_min: imgui.ImVec2, p_max: imgui.ImVec2, color: imgui.ImVec4, shadow_size: float = 10.0, rounding: float = 0.0) -> None:
"""
Simulates a soft shadow effect by drawing multiple concentric rounded rectangles
with decreasing alpha values. This is a faux-shader effect using primitive batching.
"""
r, g, b, a = color.x, color.y, color.z, color.w
steps = int(shadow_size)
if steps <= 0: return
alpha_step = a / steps
for i in range(steps):
current_alpha = a - (i * alpha_step)
# Apply an easing function (e.g., cubic) for a smoother shadow falloff
current_alpha = current_alpha * (1.0 - (i / steps)**2)
if current_alpha <= 0.01:
continue
expand = float(i)
c_min = imgui.ImVec2(p_min.x - expand, p_min.y - expand)
c_max = imgui.ImVec2(p_max.x + expand, p_max.y + expand)
u32_color = imgui.get_color_u32(imgui.ImVec4(r, g, b, current_alpha))
draw_list.add_rect(
c_min,
c_max,
u32_color,
rounding + expand if rounding > 0 else 0.0,
flags=imgui.ImDrawFlags_.round_corners_all if rounding > 0 else imgui.ImDrawFlags_.none,
thickness=1.0
)
src/tool_bias.py
+27 -2
View File
@@ -1,7 +1,32 @@
from typing import List, Dict, Any, Optional
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from src.tool_presets import Tool, ToolPreset
from src.type_aliases import Metadata
from src.tool_presets import Tool, ToolPreset
@dataclass
class BiasProfile:
name: str
tool_weights: Dict[str, int] = field(default_factory=dict)
category_multipliers: Dict[str, float] = field(default_factory=dict)
def to_dict(self) -> Metadata:
return {
"name": self.name,
"tool_weights": self.tool_weights,
"category_multipliers": self.category_multipliers,
}
@classmethod
def from_dict(cls, data: Metadata) -> "BiasProfile":
return cls(
name = data["name"],
tool_weights = data.get("tool_weights", {}),
category_multipliers = data.get("category_multipliers", {}),
)
class ToolBiasEngine:
src/tool_presets.py
+54 -5
View File
@@ -1,11 +1,59 @@
from __future__ import annotations
import tomllib
import tomli_w
from pathlib import Path
from typing import Dict, List, Optional, Union, Any
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Union, Any
from src import paths
from src.tool_bias import BiasProfile
from src import paths
from src.type_aliases import Metadata
@dataclass
class Tool:
name: str
approval: str = 'auto'
weight: int = 3
parameter_bias: Dict[str, str] = field(default_factory=dict)
def to_dict(self) -> Metadata:
return {
"name": self.name,
"approval": self.approval,
"weight": self.weight,
"parameter_bias": self.parameter_bias,
}
@classmethod
def from_dict(cls, data: Metadata) -> "Tool":
return cls(
name=data["name"],
approval=data.get("approval", "auto"),
weight=data.get("weight", 3),
parameter_bias=data.get("parameter_bias", {}),
)
@dataclass
class ToolPreset:
name: str
categories: Dict[str, List[Union[Tool, Any]]] = field(default_factory=dict)
def to_dict(self) -> Metadata:
serialized_categories = {}
for cat, tools in self.categories.items():
serialized_categories[cat] = [t.to_dict() if isinstance(t, Tool) else t for t in tools]
return {"categories": serialized_categories}
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "ToolPreset":
raw_categories = data.get("categories", {})
parsed_categories = {}
for cat, tools in raw_categories.items():
parsed_categories[cat] = [Tool.from_dict(t) if isinstance(t, dict) else t for t in tools]
return cls(name=name, categories=parsed_categories)
class ToolPresetManager:
@@ -88,10 +136,11 @@ class ToolPresetManager:
del data["presets"][name]
self._write_raw(path, data)
def load_all_bias_profiles(self) -> Dict[str, BiasProfile]:
def load_all_bias_profiles(self) -> Dict[str, "BiasProfile"]:
"""
[C: tests/test_tool_preset_manager.py:test_bias_profiles_merged, tests/test_tool_preset_manager.py:test_delete_bias_profile, tests/test_tool_preset_manager.py:test_save_bias_profile]
"""
from src.tool_bias import BiasProfile
global_path = paths.get_global_tool_presets_path()
global_data = self._read_raw(global_path).get("bias_profiles", {})
src/vendor_capabilities.py
-91
View File
@@ -1,91 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class VendorCapabilities:
vendor: str
model: str
vision: bool = False
tool_calling: bool = True
caching: bool = False
streaming: bool = True
model_discovery: bool = True
context_window: int = 8192
cost_tracking: bool = True
cost_input_per_mtok: float = 0.0
cost_output_per_mtok: float = 0.0
notes: str = ''
# v2 fields (added 2026-06-11)
local: bool = False
reasoning: bool = False
structured_output: bool = False
code_execution: bool = False
web_search: bool = False
x_search: bool = False
file_search: bool = False
mcp_support: bool = False
audio: bool = False
video: bool = False
grounding: bool = False
computer_use: bool = False
_REGISTRY: dict[tuple[str, str], VendorCapabilities] = {}
def register(cap: VendorCapabilities) -> None:
_REGISTRY[(cap.vendor, cap.model)] = cap
def get_capabilities(vendor: str, model: str) -> VendorCapabilities:
if (vendor, model) in _REGISTRY: return _REGISTRY[(vendor, model)]
if (vendor, '*') in _REGISTRY: return _REGISTRY[(vendor, '*')]
raise KeyError(f'No capabilities registered for vendor={vendor!r} model={model!r}')
def list_models_for_vendor(vendor: str) -> list[str]:
return sorted({m for v, m in _REGISTRY if v == vendor and m != '*'})
register(VendorCapabilities(vendor='minimax', model='*', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.7', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20, reasoning=True))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.5', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20, reasoning=True))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2.1', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='minimax', model='MiniMax-M2', context_window=131072, cost_input_per_mtok=0.20, cost_output_per_mtok=0.20))
register(VendorCapabilities(vendor='grok', model='*', context_window=131072, cost_input_per_mtok=2.00, cost_output_per_mtok=10.00, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-2', context_window=131072, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-2-vision', vision=True, context_window=32768, web_search=True, x_search=True))
register(VendorCapabilities(vendor='grok', model='grok-beta', context_window=131072, cost_input_per_mtok=5.00, cost_output_per_mtok=15.00, web_search=True, x_search=True))
register(VendorCapabilities(vendor='llama', model='*', context_window=131072))
register(VendorCapabilities(vendor='llama', model='llama-3.1-8b-instant', context_window=131072, cost_input_per_mtok=0.05, cost_output_per_mtok=0.08))
register(VendorCapabilities(vendor='llama', model='llama-3.1-70b-versatile', context_window=131072, cost_input_per_mtok=0.59, cost_output_per_mtok=0.79))
register(VendorCapabilities(vendor='llama', model='llama-3.1-405b-reasoning', context_window=131072, cost_input_per_mtok=3.00, cost_output_per_mtok=3.00, reasoning=True))
register(VendorCapabilities(vendor='llama', model='llama-3.2-1b-preview', context_window=131072, cost_input_per_mtok=0.04, cost_output_per_mtok=0.04))
register(VendorCapabilities(vendor='llama', model='llama-3.2-3b-preview', context_window=131072, cost_input_per_mtok=0.06, cost_output_per_mtok=0.06))
register(VendorCapabilities(vendor='llama', model='llama-3.2-11b-vision-preview', vision=True, context_window=131072, cost_input_per_mtok=0.18, cost_output_per_mtok=0.18))
register(VendorCapabilities(vendor='llama', model='llama-3.2-90b-vision-preview', vision=True, context_window=131072, cost_input_per_mtok=0.90, cost_output_per_mtok=0.90))
register(VendorCapabilities(vendor='llama', model='llama-3.3-70b-specdec', context_window=131072, cost_input_per_mtok=0.59, cost_output_per_mtok=0.79))
register(VendorCapabilities(vendor='qwen', model='*', context_window=32768))
register(VendorCapabilities(vendor='qwen', model='qwen-turbo', context_window=1000000, cost_input_per_mtok=0.05, cost_output_per_mtok=0.10))
register(VendorCapabilities(vendor='qwen', model='qwen-plus', context_window=131072, cost_input_per_mtok=0.40, cost_output_per_mtok=1.20))
register(VendorCapabilities(vendor='qwen', model='qwen-max', context_window=32768, cost_input_per_mtok=2.00, cost_output_per_mtok=6.00))
register(VendorCapabilities(vendor='qwen', model='qwen-long', context_window=1000000, cost_input_per_mtok=0.07, cost_output_per_mtok=0.28, caching=True, notes='qwen-long supports custom chunked long-context caching'))
register(VendorCapabilities(vendor='qwen', model='qwen-vl-plus', vision=True, context_window=131072, cost_input_per_mtok=0.21, cost_output_per_mtok=0.63))
register(VendorCapabilities(vendor='qwen', model='qwen-vl-max', vision=True, context_window=32768, cost_input_per_mtok=0.50, cost_output_per_mtok=1.50))
register(VendorCapabilities(vendor='qwen', model='qwen-audio', context_window=32768, cost_input_per_mtok=0.10, cost_output_per_mtok=0.30, audio=True, notes='Audio input support added 2026-06-11 (v2 matrix)'))
register(VendorCapabilities(vendor='anthropic', model='*', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True, notes='Anthropic wildcard: Sonnet defaults. Per-model variations below.'))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-5-20250929', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-20250514', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-sonnet-4-6', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-1-20250805', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-20250514', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-5-20251101', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-6', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-7', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-opus-4-8', context_window=200000, cost_input_per_mtok=15.00, cost_output_per_mtok=75.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-haiku-4-5-20251001', context_window=200000, cost_input_per_mtok=1.00, cost_output_per_mtok=5.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='anthropic', model='claude-fable-5', context_window=200000, cost_input_per_mtok=3.00, cost_output_per_mtok=15.00, caching=True, structured_output=True, file_search=True, mcp_support=True, computer_use=True))
register(VendorCapabilities(vendor='gemini', model='*', context_window=1000000, cost_input_per_mtok=1.25, cost_output_per_mtok=5.00, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True, notes='Gemini wildcard: 1M+ context window. Per-model variations below.'))
register(VendorCapabilities(vendor='gemini', model='gemini-3.1-pro-preview', context_window=1000000, cost_input_per_mtok=3.50, cost_output_per_mtok=10.50, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-3-flash-preview', context_window=1000000, cost_input_per_mtok=0.15, cost_output_per_mtok=0.60, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-2.5-flash', context_window=1000000, cost_input_per_mtok=0.15, cost_output_per_mtok=0.60, caching=True, vision=True, video=True, audio=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='gemini', model='gemini-2.5-flash-lite', context_window=1000000, cost_input_per_mtok=0.075, cost_output_per_mtok=0.30, caching=True, vision=True, grounding=True, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='*', context_window=32768, cost_input_per_mtok=0.27, cost_output_per_mtok=1.10, reasoning=True, structured_output=True, notes='DeepSeek wildcard: V3 defaults. R1/reasoner variants below.'))
register(VendorCapabilities(vendor='deepseek', model='deepseek-v3', context_window=32768, cost_input_per_mtok=0.27, cost_output_per_mtok=1.10, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='deepseek-reasoner', context_window=32768, cost_input_per_mtok=0.55, cost_output_per_mtok=2.19, reasoning=True, structured_output=True))
register(VendorCapabilities(vendor='deepseek', model='deepseek-r1', context_window=32768, cost_input_per_mtok=0.55, cost_output_per_mtok=2.19, reasoning=True, structured_output=True))
src/vendor_state.py
-81
View File
@@ -1,81 +0,0 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class VendorMetric:
"""Atomic vendor-state metric.
[C: src/gui_2.py:render_vendor_state]
"""
key: str
label: str
value: str
state: str
tooltip: str
def get_vendor_state(app) -> list[VendorMetric]:
"""Aggregate per-vendor session state for the Operations Hub Vendor State tab.
[C: src/gui_2.py:render_vendor_state]
"""
out: list[VendorMetric] = []
out.append(VendorMetric(
key = "provider_model",
label = "Provider / Model",
value = f"{app.current_provider} / {app.current_model}",
state = "info",
tooltip = "The vendor and model that will handle the next request."
))
ctrl = getattr(app, "controller", None)
tt = getattr(ctrl, "token_tracker", None) if ctrl else None
if tt and getattr(tt, "limit", 0):
pct = 100.0 * getattr(tt, "used", 0) / tt.limit
state = "warn" if pct > 75 else "ok"
out.append(VendorMetric(
key = "context_window",
label = "Context Window",
value = f"{tt.used:,} / {tt.limit:,} ({pct:.0f}%)",
state = state,
tooltip = "Used vs total context window for the current session."
))
else:
out.append(VendorMetric(
key = "context_window", label="Context Window", value="—", state="info",
tooltip = "No token tracker attached for the current provider."
))
if tt is not None:
hits = getattr(tt, "cache_hits", 0)
miss = getattr(tt, "cache_misses", 0)
total = hits + miss
rate = (100.0 * hits / total) if total else 0.0
out.append(VendorMetric(
key = "cache", label="Cache Hit Rate",
value = f"{rate:.0f}% ({hits:,}/{total:,})",
state = "ok" if rate > 50 else "info",
tooltip = "Server-side prompt cache hit rate for the current session."
))
else:
out.append(VendorMetric(
key = "cache", label="Cache Hit Rate", value="—", state="info",
tooltip = "No token tracker attached for the current provider."
))
quota = (getattr(ctrl, "vendor_quota", {}) or {}) if ctrl else {}
pct_left = quota.get("remaining_pct")
if pct_left is None:
out.append(VendorMetric(
key = "quota", label="Vendor Quota", value="—", state="info",
tooltip = "Vendor did not report quota for the current billing period."
))
else:
out.append(VendorMetric(
key = "quota", label="Vendor Quota",
value = f"{pct_left}% remaining",
state = "ok" if pct_left > 25 else "warn",
tooltip = "Approximate quota remaining for the current billing period."
))
err = getattr(ctrl, "last_error", None) if ctrl else None
out.append(VendorMetric(
key = "last_error", label="Last Error",
value = err.get("class", "none") if err else "none",
state = "error" if err else "ok",
tooltip = err.get("message", "No error since session start.") if err else "No error since session start."
))
return out
src/workspace_manager.py
+30 -4
View File
@@ -1,10 +1,36 @@
import tomllib
import tomli_w
from pathlib import Path
from typing import Dict, Any, Optional, Union
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Any, Optional, Union
from src import paths
from src import paths
from src.type_aliases import Metadata
@dataclass
class WorkspaceProfile:
name: str
ini_content: str
show_windows: Dict[str, bool]
panel_states: Metadata
def to_dict(self) -> Metadata:
return {
"ini_content": self.ini_content,
"show_windows": self.show_windows,
"panel_states": self.panel_states,
}
@classmethod
def from_dict(cls, name: str, data: Metadata) -> "WorkspaceProfile":
return cls(
name = name,
ini_content = data.get("ini_content", ""),
show_windows = data.get("show_windows", {}),
panel_states = data.get("panel_states", {}),
)
class WorkspaceManager:
@@ -80,4 +106,4 @@ class WorkspaceManager:
def _save_file(self, path: Path, data: Dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "wb") as f:
tomli_w.dump(data, f)
tomli_w.dump(data, f)