diff --git a/src/ai_client.py b/src/ai_client.py index 82dfe67..79f71e5 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -12,7 +12,6 @@ For Gemini: injects the initial context directly into system_instruction during chat creation to avoid massive history bloat. """ # ai_client.py -# ai_client.py import tomllib import asyncio import json @@ -53,7 +52,7 @@ events: EventEmitter = EventEmitter() def set_model_params(temp: float, max_tok: int, trunc_limit: int = 8000, top_p: float = 1.0) -> None: """ Sets global generation parameters like temperature and max tokens. - [C: src/app_controller.py:AppController._handle_request_event, src/app_controller.py:AppController.generate, tests/test_history_management.py:test_get_history_bleed_stats_basic] + [C: src/app_controller.py:AppController._handle_request_event, src/app_controller.py:AppController.generate] """ global _temperature, _max_tokens, _history_trunc_limit, _top_p _temperature = temp @@ -61,13 +60,6 @@ def set_model_params(temp: float, max_tok: int, trunc_limit: int = 8000, top_p: _history_trunc_limit = trunc_limit _top_p = top_p -def get_history_trunc_limit() -> int: - return _history_trunc_limit - -def set_history_trunc_limit(val: int) -> None: - global _history_trunc_limit - _history_trunc_limit = val - _gemini_client: Optional[genai.Client] = None _gemini_chat: Any = None _gemini_cache: Any = None @@ -438,7 +430,7 @@ def _classify_minimax_error(exc: Exception) -> ProviderError: def set_provider(provider: str, model: str) -> None: """ Updates the active LLM provider and model name. - [C: src/app_controller.py:AppController._handle_reset_session, src/app_controller.py:AppController._init_ai_and_hooks, src/app_controller.py:AppController.current_model, src/app_controller.py:AppController.current_provider, src/app_controller.py:AppController.do_fetch, src/multi_agent_conductor.py:run_worker_lifecycle, src/orchestrator_pm.py:generate_tracks, tests/conftest.py:reset_ai_client, tests/test_ai_cache_tracking.py:test_gemini_cache_tracking, tests/test_ai_client_cli.py:test_ai_client_send_gemini_cli, tests/test_api_events.py:test_send_emits_events_proper, tests/test_api_events.py:test_send_emits_tool_events, tests/test_deepseek_provider.py:test_deepseek_completion_logic, tests/test_deepseek_provider.py:test_deepseek_model_selection, tests/test_deepseek_provider.py:test_deepseek_payload_verification, tests/test_deepseek_provider.py:test_deepseek_reasoner_payload_verification, tests/test_deepseek_provider.py:test_deepseek_reasoning_logic, tests/test_deepseek_provider.py:test_deepseek_streaming, tests/test_deepseek_provider.py:test_deepseek_tool_calling, tests/test_gemini_cli_edge_cases.py:test_gemini_cli_loop_termination, tests/test_gemini_cli_integration.py:test_gemini_cli_full_integration, tests/test_gemini_cli_integration.py:test_gemini_cli_rejection_and_history, tests/test_gemini_cli_parity_regression.py:test_get_history_bleed_stats, tests/test_gemini_cli_parity_regression.py:test_send_invokes_adapter_send, tests/test_gui2_mcp.py:test_mcp_tool_call_is_dispatched, tests/test_history_management.py:test_get_history_bleed_stats_basic, tests/test_minimax_provider.py:test_minimax_default_model, tests/test_minimax_provider.py:test_minimax_history_bleed_stats, tests/test_minimax_provider.py:test_minimax_model_selection, tests/test_mma_agent_focus_phase1.py:test_append_comms_has_source_tier_key, tests/test_rag_integration.py:test_rag_integration, tests/test_tier4_interceptor.py:test_ai_client_passes_qa_callback, tests/test_tier4_interceptor.py:test_gemini_provider_passes_qa_callback_to_run_script, tests/test_token_usage.py:test_token_usage_tracking, tests/test_token_viz.py:test_get_history_bleed_stats_returns_all_keys_unknown_provider] + [C: src/app_controller.py:AppController._handle_reset_session, src/app_controller.py:AppController._init_ai_and_hooks, src/app_controller.py:AppController.current_model, src/app_controller.py:AppController.current_provider, src/app_controller.py:AppController.do_fetch, src/multi_agent_conductor.py:run_worker_lifecycle, src/orchestrator_pm.py:generate_tracks, tests/conftest.py:reset_ai_client, tests/test_ai_cache_tracking.py:test_gemini_cache_tracking, tests/test_ai_client_cli.py:test_ai_client_send_gemini_cli, tests/test_api_events.py:test_send_emits_events_proper, tests/test_api_events.py:test_send_emits_tool_events, tests/test_deepseek_provider.py:test_deepseek_completion_logic, tests/test_deepseek_provider.py:test_deepseek_model_selection, tests/test_deepseek_provider.py:test_deepseek_payload_verification, tests/test_deepseek_provider.py:test_deepseek_reasoner_payload_verification, tests/test_deepseek_provider.py:test_deepseek_reasoning_logic, tests/test_deepseek_provider.py:test_deepseek_streaming, tests/test_deepseek_provider.py:test_deepseek_tool_calling, tests/test_gemini_cli_edge_cases.py:test_gemini_cli_loop_termination, tests/test_gemini_cli_integration.py:test_gemini_cli_full_integration, tests/test_gemini_cli_integration.py:test_gemini_cli_rejection_and_history, tests/test_gemini_cli_parity_regression.py:test_send_invokes_adapter_send, tests/test_gui2_mcp.py:test_mcp_tool_call_is_dispatched, tests/test_minimax_provider.py:test_minimax_default_model, tests/test_minimax_provider.py:test_minimax_model_selection, tests/test_mma_agent_focus_phase1.py:test_append_comms_has_source_tier_key, tests/test_rag_integration.py:test_rag_integration, tests/test_tier4_interceptor.py:test_ai_client_passes_qa_callback, tests/test_tier4_interceptor.py:test_gemini_provider_passes_qa_callback_to_run_script, tests/test_token_usage.py:test_token_usage_tracking] """ global _provider, _model _provider = provider @@ -2444,165 +2436,12 @@ def _add_bleed_derived(d: dict[str, Any], sys_tok: int = 0, tool_tok: int = 0) - if os.environ.get("SLOP_TOOL_PRESET"): try: set_tool_preset(os.environ["SLOP_TOOL_PRESET"]) - except Exception as _e: + except Exception: pass -def get_history_bleed_stats(md_content: Optional[str] = None) -> dict[str, Any]: - """ - [C: tests/test_gemini_cli_parity_regression.py:test_get_history_bleed_stats, tests/test_history_management.py:test_get_history_bleed_stats_basic, tests/test_minimax_provider.py:test_minimax_history_bleed_stats, tests/test_token_viz.py:test_get_history_bleed_stats_returns_all_keys_unknown_provider] - """ - if _provider == "anthropic": - with _anthropic_history_lock: - history_snapshot = list(_anthropic_history) - sys_tok = max(1, int(len(md_content) / _CHARS_PER_TOKEN)) if md_content else 0 - current_tokens = _estimate_prompt_tokens([], history_snapshot) - if md_content: - current_tokens += max(1, int(len(md_content) / _CHARS_PER_TOKEN)) - limit_tokens = _ANTHROPIC_MAX_PROMPT_TOKENS - percentage = (current_tokens / limit_tokens) * 100 if limit_tokens > 0 else 0 - return _add_bleed_derived({ - "provider": "anthropic", - "limit": limit_tokens, - "current": current_tokens, - "percentage": percentage, - }, sys_tok=sys_tok, tool_tok=2500) - elif _provider == "gemini": - effective_limit = _history_trunc_limit if _history_trunc_limit > 0 else _GEMINI_MAX_INPUT_TOKENS - if _gemini_chat: - try: - _ensure_gemini_client() - if _gemini_client: - raw_history = list(_get_gemini_history_list(_gemini_chat)) - history: list[types.Content] = [] - for c in raw_history: - role = "model" if c.role in ["assistant", "model"] else "user" - history.append(types.Content(role=role, parts=c.parts)) - if md_content: - history.insert(0, types.Content(role="user", parts=[types.Part(text=md_content)])) - if not history: - return _add_bleed_derived({ - "provider": "gemini", - "limit": effective_limit, - "current": 0, - "percentage": 0, - }) - resp = _gemini_client.models.count_tokens( - model=_model, - contents=history - ) - current_tokens = cast(int, resp.total_tokens) - percentage = (current_tokens / effective_limit) * 100 if effective_limit > 0 else 0 - return _add_bleed_derived({ - "provider": "gemini", - "limit": effective_limit, - "current": current_tokens, - "percentage": percentage, - }, sys_tok=0, tool_tok=0) - except Exception: - pass - elif md_content: - try: - _ensure_gemini_client() - if _gemini_client: - resp = _gemini_client.models.count_tokens( - model=_model, - contents=[types.Content(role="user", parts=[types.Part(text=md_content)])] - ) - current_tokens = cast(int, resp.total_tokens) - percentage = (current_tokens / effective_limit) * 100 if effective_limit > 0 else 0 - return _add_bleed_derived({ - "provider": "gemini", - "limit": effective_limit, - "current": current_tokens, - "percentage": percentage, - }) - except Exception: - pass - return _add_bleed_derived({ - "provider": "gemini", - "limit": effective_limit, - "current": 0, - "percentage": 0, - }) - elif _provider == "gemini_cli": - effective_limit = _history_trunc_limit if _history_trunc_limit > 0 else _GEMINI_MAX_INPUT_TOKENS - limit_tokens = effective_limit - current_tokens = 0 - if _gemini_cli_adapter and _gemini_cli_adapter.last_usage: - u = _gemini_cli_adapter.last_usage - current_tokens = cast(int, u.get("input_tokens") or u.get("input", 0)) - percentage = (current_tokens / limit_tokens) * 100 if limit_tokens > 0 else 0 - return _add_bleed_derived({ - "provider": "gemini_cli", - "limit": limit_tokens, - "current": current_tokens, - "percentage": percentage, - }) - elif _provider == "deepseek": - limit_tokens = 64000 - current_tokens = 0 - with _deepseek_history_lock: - for msg in _deepseek_history: - content = msg.get("content", "") - if isinstance(content, str): - current_tokens += len(content) - elif isinstance(content, list): - for block in content: - if isinstance(block, dict): - text = block.get("text", "") - if isinstance(text, str): - current_tokens += len(text) - inp = block.get("input") - if isinstance(inp, dict): - import json as _json - current_tokens += len(_json.dumps(inp, ensure_ascii=False)) - if md_content: current_tokens += len(md_content) - current_tokens = max(1, int(current_tokens / _CHARS_PER_TOKEN)) - percentage = (current_tokens / limit_tokens) * 100 if limit_tokens > 0 else 0 - return _add_bleed_derived({ - "provider": "deepseek", - "limit": limit_tokens, - "current": current_tokens, - "percentage": percentage, - }) - elif _provider == "minimax": - limit_tokens = 204800 - current_tokens = 0 - with _minimax_history_lock: - for msg in _minimax_history: - content = msg.get("content", "") - if isinstance(content, str): - current_tokens += len(content) - elif isinstance(content, list): - for block in content: - if isinstance(block, dict): - text = block.get("text", "") - if isinstance(text, str): - current_tokens += len(text) - inp = block.get("input") - if isinstance(inp, dict): - import json as _json - current_tokens += len(_json.dumps(inp, ensure_ascii=False)) - if md_content: current_tokens += len(md_content) - current_tokens = max(1, int(current_tokens / _CHARS_PER_TOKEN)) - percentage = (current_tokens / limit_tokens) * 100 if limit_tokens > 0 else 0 - return _add_bleed_derived({ - "provider": "minimax", - "limit": limit_tokens, - "current": current_tokens, - "percentage": percentage, - }) - return _add_bleed_derived({ - "provider": _provider, - "limit": 0, - "current": 0, - "percentage": 0, - }) - def run_subagent_summarization(file_path: str, content: str, is_code: bool, outline: str) -> str: """ - Performs a stateless summarization request using a sub-agent prompt. - [C: src/summarize.py:summarise_file, tests/test_subagent_summarization.py:test_run_subagent_summarization_anthropic, tests/test_subagent_summarization.py:test_run_subagent_summarization_gemini] + Performs a stateless summarization request using a sub-agent prompt. [C: src/summarize.py:summarise_file, tests/test_subagent_summarization.py:test_run_subagent_summarization_anthropic, tests/test_subagent_summarization.py:test_run_subagent_summarization_gemini] """ prompt_tmpl = mma_prompts.TIER4_SUMMARIZE_CODE_PROMPT if is_code else mma_prompts.TIER4_SUMMARIZE_TEXT_PROMPT prompt = prompt_tmpl.format(file_path=file_path, outline=outline, content=content) diff --git a/tests/test_gemini_cli_parity_regression.py b/tests/test_gemini_cli_parity_regression.py index 4c27c78..dd37fcf 100644 --- a/tests/test_gemini_cli_parity_regression.py +++ b/tests/test_gemini_cli_parity_regression.py @@ -11,16 +11,3 @@ def test_send_invokes_adapter_send() -> None: ai_client.set_provider("gemini_cli", "gemini-2.0-flash") res = ai_client.send("context", "msg") assert res == "Hello from mock adapter" - -def test_get_history_bleed_stats() -> None: - import src.ai_client as ai_client - ai_client._gemini_cli_adapter = None - with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen: - mock_process = MagicMock() - mock_process.communicate.return_value = ('{"type": "message", "content": "txt"}', '') - mock_process.returncode = 0 - mock_popen.return_value = mock_process - ai_client.set_provider("gemini_cli", "gemini-2.0-flash") - ai_client.send("context", "msg") - stats = ai_client.get_history_bleed_stats() - assert stats["provider"] == "gemini_cli" diff --git a/tests/test_history_management.py b/tests/test_history_management.py index 8dd234a..1357648 100644 --- a/tests/test_history_management.py +++ b/tests/test_history_management.py @@ -112,22 +112,4 @@ def test_history_persistence_across_turns(tmp_path: Path) -> None: reloaded = project_manager.load_project(str(project_path)) active_disc = reloaded["discussion"]["active"] h2 = reloaded["discussion"]["discussions"][active_disc]["history"] - assert len(h2) >= 2 - -def test_get_history_bleed_stats_basic() -> None: - """Tests basic retrieval of history bleed statistics from the AI client.""" - ai_client.set_provider("gemini", "gemini-2.5-flash-lite") - # Before any message, it might be 0 or based on an empty context - stats = ai_client.get_history_bleed_stats() - assert "provider" in stats - assert stats["provider"] == "gemini" - assert "current" in stats - assert "limit" in stats, "Stats dictionary should contain 'limit'" - - # Test with a different limit - ai_client.set_model_params(0.0, 8192, 500) - stats = ai_client.get_history_bleed_stats() - assert "current" in stats, "Stats dictionary should contain 'current' token usage" - assert 'limit' in stats, "Stats dictionary should contain 'limit'" - assert stats['limit'] == 500 - assert isinstance(stats['current'], int) and stats['current'] >= 0 \ No newline at end of file + assert len(h2) >= 2 \ No newline at end of file diff --git a/tests/test_minimax_provider.py b/tests/test_minimax_provider.py index 5ff3e22..a4c771b 100644 --- a/tests/test_minimax_provider.py +++ b/tests/test_minimax_provider.py @@ -18,13 +18,6 @@ def test_minimax_list_models() -> None: assert "MiniMax-M2.1" in models assert "MiniMax-M2" in models -def test_minimax_history_bleed_stats() -> None: - ai_client.set_provider("minimax", "MiniMax-M2.5") - ai_client.reset_session() - stats = ai_client.get_history_bleed_stats(md_content="Test context") - assert stats["provider"] == "minimax" - assert stats["limit"] == 204800 - def test_minimax_in_providers_list() -> None: from src.models import PROVIDERS assert "minimax" in PROVIDERS diff --git a/tests/test_token_viz.py b/tests/test_token_viz.py index 841e110..3f0c88d 100644 --- a/tests/test_token_viz.py +++ b/tests/test_token_viz.py @@ -47,13 +47,6 @@ def test_add_bleed_derived_headroom_clamped_to_zero() -> None: result = ai_client._add_bleed_derived(d) assert result["headroom"] == 0 -def test_get_history_bleed_stats_returns_all_keys_unknown_provider() -> None: - """get_history_bleed_stats must return a valid dict even if provider is unknown.""" - ai_client.set_provider("unknown", "unknown") - stats = ai_client.get_history_bleed_stats() - for key in ["provider", "limit", "current", "percentage", "estimated_prompt_tokens", "headroom", "would_trim", "sys_tokens", "tool_tokens", "history_tokens"]: - assert key in stats - def test_app_token_stats_initialized_empty(app_instance: Any) -> None: """App._token_stats should start empty.""" assert app_instance.controller._token_stats == {}