diff --git a/README.md b/README.md index 36b3d869..ef11eb2a 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,37 @@ active workbench's exact source-backed API. Native exposes only the complete tool families belonging to the human-selected VibeCAD ribbon. A provider can never select or switch a workbench, ribbon, or authoring mode for itself. + +### Anthropic and Gemini conversation budgets + +Long tool loops use a 512 KiB serialized request budget. Above 75% of that budget, +older bulky successful observations can become explicit references to data that +must be read again when needed. The latest two tool batches, original user +instructions, tool arguments/signatures, exact source/API reads, failures, +pending jobs and critical state are retained. A current state snapshot accompanies +reduced history. If the remaining request cannot fit, VibeCAD stops before +generation and explains how to continue; completed CAD work remains available. + +Integrators can pass these options in the provider run context (these are Python +integration settings, not GUI preferences): + +```python +context["_vibecad_provider_options"] = { + "history_budget_bytes": 512 * 1024, # 0 disables the byte limit + # "context_window_tokens": 200_000, # optional, set for the selected model + # "output_reserve_tokens": 8192, # Gemini estimate; not an API output cap +} +``` + +The optional context-window check reserves Anthropic's requested maximum output +or Gemini's configured reserve, then estimates input from serialized JSON bytes +divided by four. Tools, system text and encoded images count toward the byte +budget. This estimate is not a tokenizer or a guarantee for image-token charges. +A configured context-window check remains enabled even with a zero byte limit. +Large exact reads may require a narrower read or a larger budget. History +management adds no summarization API call. Diagnostic events separate estimated +input from provider-reported usage when available. + ### Save a Key in the OS Keyring 1. Select the provider first. Keys are stored separately for OpenAI, Anthropic, and Gemini. diff --git a/src/Mod/VibeCAD/VibeCADProvider.py b/src/Mod/VibeCAD/VibeCADProvider.py index 36947fca..985ad71c 100644 --- a/src/Mod/VibeCAD/VibeCADProvider.py +++ b/src/Mod/VibeCAD/VibeCADProvider.py @@ -1970,6 +1970,7 @@ def run( try: provider_context = dict(context) provider_context["_vibecad_provider_options"] = { + **dict(context.get("_vibecad_provider_options") or {}), "web_search_enabled": self.web_search_enabled, "compaction_model": self.compaction_model, } @@ -5460,6 +5461,220 @@ def _gemini_forced_tool_completion( return _json_safe(arguments) + +DEFAULT_PROVIDER_HISTORY_BYTES = 512 * 1024 +_PROVIDER_HISTORY_PROTECTED_KEYS = { + "source", "code", "api", "api_text", "input_schema", "schema", + "operation", "job", "background_jobs", "background_job", + "error", "errors", "failure_code", "failure_stage", "cancelled", + "next_action", "next_actions", "human_steering", "verification", + "vibecad_state_after", "native_state", "modeling_surface", + "document", "object", "object_name", "created", "updated", "deleted", + "changed", "transaction", "expected_outputs", "affected_outputs", +} + + +class _ProviderHistoryBudgetExceeded(RuntimeError): + def __init__(self, accounting: dict[str, Any]) -> None: + super().__init__( + "I stopped before sending a request that exceeds the conversation budget. " + "Completed CAD work is retained. Continue with a narrower request or a " + "scoped source/API read; the provider history budget can also be raised. " + "Exact reads, failures, pending jobs and signed tool calls were not truncated." + ) + self.accounting = accounting + + +def _provider_history_has_protected_value(value: Any) -> bool: + if isinstance(value, dict): + for key, item in value.items(): + name = str(key).lower() + if (name in _PROVIDER_HISTORY_PROTECTED_KEYS + or "revision" in name or name.endswith(("_id", "_sha256")) + or (name == "ok" and item is False) + or (name == "status" and item in ( + "pending", "queued", "running", "failed", "error", "cancelled" + ))): + return True + if _provider_history_has_protected_value(item): + return True + elif isinstance(value, list): + return any(_provider_history_has_protected_value(item) for item in value) + return False + + +def _provider_history_reference(content: Any, tool_name: str, call_id: str) -> str | None: + # Exact source/API responses remain whole, even after newer tool batches. + if not tool_name or not isinstance(content, str) or any( + name in tool_name.lower() for name in ("source", "api") + ): + return None + try: + result = json.loads(content) + except (TypeError, ValueError): + return None + if not isinstance(result, dict) or result.get("ok") is not True: + return None + if "vibecad_history_reference" in result: + return None + # Never reduce an unresolved/error result, including nested operation status. + def unresolved(value: Any) -> bool: + if isinstance(value, dict): + if (value.get("ok") is False or value.get("error") or value.get("errors") + or value.get("failure_code") or value.get("cancelled") + or any(key in value for key in ("source", "code", "api_text", "input_schema"))): + return True + if value.get("status") in ("pending", "queued", "running", "failed", "error", "cancelled"): + return True + return any(unresolved(item) for item in value.values()) + return isinstance(value, list) and any(unresolved(item) for item in value) + if unresolved(result): + return None + summary = dict(result) + omitted = [] + for key, value in result.items(): + if _provider_history_has_protected_value({key: value}): + continue + if _provider_json_bytes(value) <= 1024: + continue + summary[key] = { + "_vibecad_value_omitted": True, + "reason": "older_tool_history", + "json_bytes": _provider_json_bytes(value), + } + omitted.append(key) + if not omitted: + return None + summary["vibecad_history_reference"] = { + "tool_call_id": call_id, + "tool": tool_name, + "original_sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), + "original_utf8_bytes": len(content.encode("utf-8")), + "omitted_fields": omitted, + "recovery": ( + "This is an older observation, not exact data. Use the available read " + "tools to inspect the current fact before relying on omitted fields. " + "The original call arguments are retained; do not replay a mutation." + ), + } + encoded = json.dumps(summary, ensure_ascii=True, separators=(",", ":")) + return encoded if len(encoded.encode("utf-8")) < len(content.encode("utf-8")) else None + + +def _provider_budget_history( + request: dict[str, Any], context: dict[str, Any], *, + provider: str, state: dict[str, Any], output_reserve_tokens: int, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Bound serialized input; token figures are estimates, never billed usage. + + Optional context-window accounting uses ceil(serialized JSON bytes / 4). + This includes encoded images but is not a provider tokenizer or vision-token + guarantee. Exact reads/critical state take priority over automatic reduction. + """ + options = state.setdefault("options", { + name: _provider_option_value(context, name) + for name in ("history_budget_bytes", "context_window_tokens") + }) + configured = options["history_budget_bytes"] + limit = DEFAULT_PROVIDER_HISTORY_BYTES if configured is None else max(0, int(configured)) + window = options["context_window_tokens"] + reserve = max(0, int(output_reserve_tokens)) + effective_limit = limit + if window is not None: + context_bytes = max(0, (int(window) - reserve) * 4) + effective_limit = min(limit, context_bytes) if limit else context_bytes + enabled = bool(limit) or window is not None + before = _provider_json_bytes(request) + messages = list(request["messages"]) + old_snapshot = state.get("snapshot") + messages = [message for message in messages if message is not old_snapshot] + updated = dict(request, messages=messages) + reduced = 0 + + def refresh_snapshot() -> None: + nonlocal messages + visible = _model_visible_context(context) + visible = {key: value for key, value in visible.items() + if key not in {"view_screenshot", "reference_images"}} + if isinstance(context.get("native_state"), dict): + visible["native_state"] = _json_safe(context["native_state"]) + snapshot = {"role": "user", "content": json.dumps({ + "vibecad_history_live_state": visible, + "instruction": "Continue the original user request. Read current exact data for omitted historical fields.", + }, ensure_ascii=True, separators=(",", ":"))} + state["snapshot"] = snapshot + messages.append(snapshot) + + if state.get("active"): + refresh_snapshot() + target = effective_limit * 3 // 4 + current_size = _provider_json_bytes(updated) + if enabled and current_size > target: + # Retain the latest two assistant/tool batches in full. All assistant + # content, call IDs/arguments and thought signatures are always retained. + assistant_indices = [i for i, m in enumerate(messages) if m.get("role") == "assistant"] + cutoff = assistant_indices[-2] if len(assistant_indices) >= 2 else 0 + calls: dict[str, str] = {} + for message in messages: + for call in message.get("tool_calls", []): + calls[str(call["id"])] = str(call["function"]["name"]) + content = message.get("content") + if isinstance(content, list): + for block in content: + if block.get("type") == "tool_use": + calls[str(block["id"])] = str(block["name"]) + for index in range(cutoff): + if current_size <= target: + break + message = messages[index] + if provider == "gemini" and message.get("role") == "tool": + call_id = str(message.get("tool_call_id", "")) + reference = _provider_history_reference( + message.get("content"), calls.get(call_id, ""), call_id + ) + if reference is not None: + messages[index] = dict(message, content=reference) + reduced += 1 + elif provider == "anthropic" and message.get("role") == "user": + content = message.get("content") + if not isinstance(content, list): + continue + blocks = list(content) + for block_index, block in enumerate(content): + if block.get("type") != "tool_result" or block.get("is_error"): + continue + call_id = str(block.get("tool_use_id", "")) + reference = _provider_history_reference( + block.get("content"), calls.get(call_id, ""), call_id + ) + if reference is not None: + blocks[block_index] = dict(block, content=reference) + reduced += 1 + messages[index] = dict(message, content=blocks) + current_size += _provider_json_bytes(messages[index]) - _provider_json_bytes(message) + if reduced and not state.get("active"): + state["active"] = True + refresh_snapshot() + + after = _provider_json_bytes(updated) + accounting = { + "event": "provider_history_budget", + "provider": provider, + "before_json_bytes": before, + "request_json_bytes": after, + "history_limit_bytes": effective_limit if enabled else None, + "compacted_results": reduced, + "estimator": "ceil(serialized_json_bytes/4); images included as encoded bytes", + "estimated_input_tokens": (after + 3) // 4, + "output_reserve_tokens": reserve, + "estimated_total_tokens": (after + 3) // 4 + reserve, + "context_window_tokens": window, + } + if enabled and after > effective_limit: + raise _ProviderHistoryBudgetExceeded(accounting) + return updated, accounting + + def _gemini_child_main( conn, prompt: str, @@ -5540,6 +5755,9 @@ def build_tool_surface( client_kwargs["timeout"] = timeout_seconds client = openai.OpenAI(**client_kwargs) + history_state: dict[str, Any] = {} + reserve_option = _provider_option_value(context, "output_reserve_tokens") + output_reserve = 8192 if reserve_option is None else max(0, int(reserve_option)) turn = 1 while max_turns is None or max_turns <= 0 or turn <= max_turns: sdk_request: dict[str, Any] = { @@ -5551,6 +5769,12 @@ def build_tool_surface( sdk_request["tools"] = tool_definitions if reasoning_effort: sdk_request["reasoning_effort"] = reasoning_effort + sdk_request, history_accounting = _provider_budget_history( + sdk_request, live_context, provider="gemini", state=history_state, + output_reserve_tokens=output_reserve, + ) + messages = sdk_request["messages"] + _send_child_progress(conn, dict(history_accounting, turn=turn)) _capture_outbound_request( live_context, provider="gemini", @@ -5579,9 +5803,16 @@ def build_tool_surface( turn=turn, ) chunk_count = 0 + token_usage: dict[str, Any] = {} finish_reason = "" for chunk in stream: chunk_count += 1 + usage = getattr(chunk, "usage", None) + if usage is not None: + for name in ("prompt_tokens", "completion_tokens", "total_tokens"): + value = usage.get(name) if isinstance(usage, dict) else getattr(usage, name, None) + if type(value) is int and value >= 0: + token_usage[name] = value choices = getattr(chunk, "choices", None) or [] if not choices: continue @@ -5652,6 +5883,7 @@ def build_tool_surface( conn, { "event": "gemini_stream_completed", + **({"token_usage": token_usage} if token_usage else {}), "turn": turn, "chunk_count": chunk_count, "finish_reason": finish_reason, @@ -5773,6 +6005,9 @@ def build_tool_surface( "error": "Google Gemini provider turn limit reached.", } ) + except _ProviderHistoryBudgetExceeded as exc: + conn.send({"type": "done", "final_output": str(exc), + "raw": {"stalled": True, "reason": "input_budget", **exc.accounting}}) except BaseException as exc: _send_child_error(conn, "Google Gemini provider", exc) finally: @@ -5900,7 +6135,10 @@ def build_tool_surface( "effort": _anthropic_adaptive_effort(reasoning_effort) } + history_state: dict[str, Any] = {} + def _stream_response(turn: int, attempt: int) -> Any: + nonlocal messages # The SDK rejects non-streaming requests that could exceed ten # minutes (large max_tokens plus thinking budgets), so always # stream and accumulate the final message. @@ -5922,6 +6160,12 @@ def _stream_response(turn: int, attempt: int) -> Any: ] if thinking is not None: sdk_request["output_config"] = {"effort": "low"} + sdk_request, history_accounting = _provider_budget_history( + sdk_request, live_context, provider="anthropic", state=history_state, + output_reserve_tokens=sdk_request["max_tokens"], + ) + messages = sdk_request["messages"] + _send_child_progress(conn, dict(history_accounting, turn=turn, attempt=attempt)) _capture_outbound_request( live_context, provider="anthropic", @@ -6381,6 +6625,9 @@ def _stream_response_with_retries(turn: int) -> Any: "raw": {"stalled": True, "reason": "turn_limit"}, } ) + except _ProviderHistoryBudgetExceeded as exc: + conn.send({"type": "done", "final_output": str(exc), + "raw": {"stalled": True, "reason": "input_budget", **exc.accounting}}) except BaseException as exc: _send_child_error(conn, "Anthropic provider", exc) finally: diff --git a/src/Mod/VibeCAD/vibecad_tests/test_provider_history_budget.py b/src/Mod/VibeCAD/vibecad_tests/test_provider_history_budget.py new file mode 100644 index 00000000..850ab135 --- /dev/null +++ b/src/Mod/VibeCAD/vibecad_tests/test_provider_history_budget.py @@ -0,0 +1,229 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later + +"""Cumulative request budgeting without changing signed tool conversations.""" + +import json +import sys +from types import SimpleNamespace + +import pytest +import VibeCADProvider as provider + +from .test_provider_subprocess import _CollectingConnection, _SequenceAnthropicMessages + + +@pytest.mark.parametrize("engine", ["gemini", "anthropic"]) +@pytest.mark.parametrize("budget", [100000, 0]) +def test_productive_history_is_bounded_before_request(monkeypatch, engine, budget): + requests = [] + context = { + "modeling_surface": {"engine": "native"}, + "native_state": {"revision": 1}, + "provider_tool_schemas": [{ + "name": "state.read", "description": "Read", + "parameters": {"type": "object", "properties": {"target": {"type": "string"}}}, + }], + "_vibecad_provider_options": {"history_budget_bytes": budget}, + } + class Connection(_CollectingConnection): + def recv(self): + state = json.loads(json.dumps(context)) + state["native_state"]["revision"] = len(requests) + 1 + state.pop("_vibecad_provider_options", None) + return {"type": "tool_result", "context": state, + "result": {"ok": True, "revision": len(requests), "payload": "x" * 20000}} + def response(): + n = len(requests) + if n > 12: + return SimpleNamespace(content=[SimpleNamespace(type="text", text="Done")], stop_reason="end_turn") + return SimpleNamespace(content=[ + SimpleNamespace(type="thinking", thinking="reason", signature="signed"), + SimpleNamespace(type="tool_use", id=f"call-{n}", name="state_read", input={"target": str(n)}) + ], stop_reason="tool_use") + class Messages: + def stream(self, **kwargs): + requests.append(json.loads(json.dumps(kwargs))) + return _SequenceAnthropicMessages([response()]).stream(**kwargs) + def create(self, **kwargs): + requests.append(json.loads(json.dumps(kwargs))) + n = len(requests) + call = SimpleNamespace(index=0, id=f"call-{n}", type="function", + function=SimpleNamespace(name="state_read", arguments=json.dumps({"target": str(n)})), + extra_content={"google": {"thought_signature": f"signed-{n}"}}) + return iter([SimpleNamespace(choices=[SimpleNamespace( + delta=SimpleNamespace(content="Done" if n > 12 else None, + tool_calls=[] if n > 12 else [call]), + finish_reason="stop" if n > 12 else "tool_calls")])]) + messages = Messages() + monkeypatch.setitem(sys.modules, "anthropic", SimpleNamespace( + Anthropic=lambda **kw: SimpleNamespace(messages=messages), + BadRequestError=type("BadRequestError", (Exception,), {}))) + monkeypatch.setitem(sys.modules, "openai", SimpleNamespace( + OpenAI=lambda **kw: SimpleNamespace(chat=SimpleNamespace(completions=messages), close=lambda: None))) + monkeypatch.setattr(provider, "_validate_provider_wire_surface", lambda c: None) + conn = Connection() + child = provider._gemini_child_main if engine == "gemini" else provider._anthropic_child_main + child(conn, "Keep the width exactly 25 mm.", context, "mock", "key", None, 1.0, 14, False) + assert conn.messages[-1]["type"] == "done" + assert conn.messages[-1]["final_output"] == "Done" + assert len(requests) == 13 + sizes = [provider._provider_json_bytes(r) for r in requests] + if budget: + assert max(sizes) <= budget + assert "vibecad_history_reference" in json.dumps(requests[-1]) + assert "Keep the width exactly 25 mm." in json.dumps(requests[-1]) + snapshot = json.loads(requests[-1]["messages"][-1]["content"]) + assert snapshot["vibecad_history_live_state"]["native_state"]["revision"] == 13 + else: + assert sizes[-1] > 240000 + last = requests[-1]["messages"] + if engine == "gemini": + calls = [c for m in last for c in m.get("tool_calls", [])] + results = [m["tool_call_id"] for m in last if m["role"] == "tool"] + assert [c["id"] for c in calls] == results + assert all(c["extra_content"]["google"]["thought_signature"] == f"signed-{i}" + for i, c in enumerate(calls, 1)) + else: + blocks = [b for m in last if isinstance(m["content"], list) for b in m["content"]] + assert [b["id"] for b in blocks if b["type"] == "tool_use"] == [ + b["tool_use_id"] for b in blocks if b["type"] == "tool_result"] + assert all(b["signature"] == "signed" for b in blocks if b["type"] == "thinking") + print(engine, budget, "requests", len(requests), "cumulative_bytes", sum(sizes), "largest_bytes", max(sizes)) + +@pytest.mark.parametrize("engine", ["gemini", "anthropic"]) +@pytest.mark.parametrize("kind", ["failure", "pending", "source", "api", "nested_revision"]) +def test_history_budget_preserves_critical_results(monkeypatch, engine, kind): + protected = {"ok": True, "payload": "protected" * 2500} + name = "state_read" + if kind == "failure": protected["ok"] = False + if kind == "pending": protected["job"] = {"id": "job-1", "status": "running"} + if kind == "source": protected["source"] = "exact code" * 2500 + if kind == "api": name = "vibescript_read_api" + if kind == "nested_revision": + protected = {"ok": True, "payload": {"revision": 42, "data": "exact" * 4000}} + payloads = [protected] + [{"ok": True, "data": "x" * 20000} for _ in range(4)] + messages = [{"role": "user", "content": "Preserve every user constraint."}] + for i, payload in enumerate(payloads): + call_id = str(i) + if engine == "gemini": + messages.extend([ + {"role": "assistant", "tool_calls": [{"id": call_id, "function": {"name": name, "arguments": "{}"}, + "extra_content": {"google": {"thought_signature": "signed"}}}]}, + {"role": "tool", "tool_call_id": call_id, "content": json.dumps(payload)}, + ]) + else: + messages.extend([ + {"role": "assistant", "content": [{"type": "thinking", "thinking": "reason", "signature": "signed"}, + {"type": "tool_use", "id": call_id, "name": name, "input": {}}]}, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": call_id, "content": json.dumps(payload)}]}, + ]) + name = "state_read" + request = {"messages": messages, "tools": [{"name": "state_read"}]} + original = json.loads(json.dumps(request)) + updated, accounting = provider._provider_budget_history(request, { + "_vibecad_provider_options": {"history_budget_bytes": 130000}, + "native_state": {"revision": 99, "background_job": {"id": "live", "status": "running"}}, + }, provider=engine, state={}, output_reserve_tokens=8192) + assert accounting["compacted_results"] > 0 + assert updated["messages"][2] == original["messages"][2] + assert request == original # No rewriting earlier captured requests. + assert updated["messages"][0] == original["messages"][0] + assert updated["tools"] == original["tools"] + snapshot = json.loads(updated["messages"][-1]["content"]) + assert snapshot["vibecad_history_live_state"]["native_state"]["revision"] == 99 + + +@pytest.mark.parametrize("part", ["messages", "system", "tools", "image", "source"]) +def test_history_budget_counts_irreducible_payloads(part): + request = {"messages": [{"role": "user", "content": "keep me"}]} + if part == "messages": + request["messages"][0]["content"] = "prompt" * 1000 + elif part == "image": + request["messages"][0]["content"] = [{"type": "image", "source": {"type": "base64", "data": "a" * 6000}}] + elif part == "source": + request["messages"].extend([ + {"role": "assistant", "tool_calls": [{"id": "read", "function": {"name": "read_source", "arguments": "{}"}}]}, + {"role": "tool", "tool_call_id": "read", "content": json.dumps({"ok": True, "source": "exact" * 1500})}, + ]) + else: + request[part] = [{"text": "required" * 1000}] + original = json.loads(json.dumps(request)) + with pytest.raises(provider._ProviderHistoryBudgetExceeded) as error: + provider._provider_budget_history(request, { + "_vibecad_provider_options": {"history_budget_bytes": 1024} + }, provider="gemini", state={}, output_reserve_tokens=8192) + assert error.value.accounting["request_json_bytes"] > 1024 + assert request == original + + +def test_history_context_window_reserves_output(): + request = {"messages": [{"role": "user", "content": "x" * 5000}]} + with pytest.raises(provider._ProviderHistoryBudgetExceeded) as error: + provider._provider_budget_history(request, { + "_vibecad_provider_options": {"history_budget_bytes": 100000, "context_window_tokens": 9000} + }, provider="anthropic", state={}, output_reserve_tokens=8192) + accounting = error.value.accounting + assert accounting["history_limit_bytes"] == (9000 - 8192) * 4 + assert accounting["estimated_total_tokens"] == accounting["estimated_input_tokens"] + 8192 + assert "measured" not in accounting["estimator"] + + +@pytest.mark.parametrize("engine", ["gemini", "anthropic"]) +def test_oversized_input_stops_before_generation(monkeypatch, engine): + requests = [] + def create(**kwargs): + requests.append(kwargs) + raise AssertionError("oversized request reached the model") + monkeypatch.setitem(sys.modules, "openai", SimpleNamespace( + OpenAI=lambda **kw: SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create)), close=lambda: None))) + monkeypatch.setitem(sys.modules, "anthropic", SimpleNamespace( + Anthropic=lambda **kw: SimpleNamespace(messages=SimpleNamespace(stream=create)), + BadRequestError=type("BadRequestError", (Exception,), {}))) + monkeypatch.setattr(provider, "_validate_provider_wire_surface", lambda c: None) + conn = _CollectingConnection() + child = provider._gemini_child_main if engine == "gemini" else provider._anthropic_child_main + child(conn, "required user instruction" * 1000, { + "provider_tool_schemas": [], + "_vibecad_provider_options": {"history_budget_bytes": 1024}, + }, "mock", "key", None, 1.0, 2, False) + assert requests == [] + assert conn.messages[-1]["type"] == "done" + assert conn.messages[-1]["raw"]["reason"] == "input_budget" + assert conn.closed + + +def test_anthropic_parent_preserves_history_configuration(monkeypatch): + captured = [] + def run(**kwargs): + captured.append(kwargs["context"]["_vibecad_provider_options"]) + return provider.ProviderResult(final_output="done", raw=None) + monkeypatch.setattr(provider, "_run_provider_subprocess", run) + provider.AnthropicProvider().run("hello", { + "_vibecad_provider_options": {"history_budget_bytes": 100000, "context_window_tokens": 200000} + }) + assert captured[0]["history_budget_bytes"] == 100000 + assert captured[0]["context_window_tokens"] == 200000 + +def test_gemini_available_usage_is_separate_from_budget_estimates(monkeypatch): + requests = [] + def create(**kwargs): + requests.append(kwargs) + return iter([ + SimpleNamespace(choices=[SimpleNamespace( + delta=SimpleNamespace(content="Done", tool_calls=[]), finish_reason="stop")]), + SimpleNamespace(choices=[], usage={"prompt_tokens": 321, "completion_tokens": 12, "total_tokens": 333}), + ]) + monkeypatch.setitem(sys.modules, "openai", SimpleNamespace( + OpenAI=lambda **kw: SimpleNamespace( + chat=SimpleNamespace(completions=SimpleNamespace(create=create)), close=lambda: None))) + monkeypatch.setattr(provider, "_validate_provider_wire_surface", lambda c: None) + conn = _CollectingConnection() + provider._gemini_child_main(conn, "hello", {"provider_tool_schemas": []}, + "mock", "key", None, 1.0, 2, False) + assert len(requests) == 1 + events = [m["event"] for m in conn.messages if m.get("type") == "progress"] + completed = next(e for e in events if e["event"] == "gemini_stream_completed") + assert completed["token_usage"] == {"prompt_tokens": 321, "completion_tokens": 12, "total_tokens": 333} + budget = next(e for e in events if e["event"] == "provider_history_budget") + assert "estimated_input_tokens" in budget + assert "token_usage" not in budget