diff --git a/scripts/tool_pilot_argon.py b/scripts/tool_pilot_argon.py index def1829e..a7d92f6c 100644 --- a/scripts/tool_pilot_argon.py +++ b/scripts/tool_pilot_argon.py @@ -1,17 +1,22 @@ -"""Internal pilot: Gemini 4 Argon with a PolicyEngine tool on the board's 100 US +"""Internal pilot: a board model with a PolicyEngine tool on the board's 100 US households. Not a board condition, and nothing here is published. +The model defaults to Gemini 4 Argon, the first one piloted; ``run --model`` +takes any board model whose provider has an entry in PROVIDER_KEYS. + The benchmark has one condition, ``no_tools`` (docs/benchmark_card.md, "Evaluation conditions"). This script stays outside the board pipeline: it writes only under --out-dir, never touches the run store, the snapshot, the -cache or the dashboard, and labels its rows ``gemini-4-argon+pe-tool`` so no -board export can take them for a board row. +cache or the dashboard, and labels its rows ``+pe-tool`` (for example +``gemini-4-argon+pe-tool``) so no board export can take them for a board row. Arms, on the same 100 households: - 1. No tools: Argon's frozen board predictions (release dashboard-data-20261014, - PR #221), read from the snapshot and never re-run. - 2. PolicyEngine tool: the board prompt plus TOOL_NOTICE, the board's - ``submit_outputs`` answer tool, and ``calculate_household``, which runs the + 1. No tools: the model's frozen board predictions (release + dashboard-data-20261014), read from the snapshot and never re-run. + 2. PolicyEngine tool: the board's prompt and request settings for the model, + plus a tool notice, the board's answer (``submit_outputs`` for a + tool-contract model, the JSON reply for a JSON-contract one), and + ``calculate_household``, which runs the reference engine on a situation the model writes as JSON. The reference engine is the installed policyengine-us (2.41.0) with the reference builder's convention reform, reference_audit/2026-09-28/fixes/latest_final.py @@ -26,7 +31,8 @@ Subcommands: oracle reference situations through the tool engine - run call the model (needs GEMINI_API_KEY); one JSON record per household + run call the model (needs its provider's key, PROVIDER_KEYS); one JSON + record per household score both arms through the board's analyze_no_tools on the scored outputs diagnose classify every arm-2 miss """ @@ -52,7 +58,7 @@ import threading import time from collections import Counter -from dataclasses import dataclass +from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Callable, Mapping @@ -72,6 +78,12 @@ MODEL_NAME = "gemini-4-argon" ARM1_LABEL = MODEL_NAME ARM2_LABEL = "gemini-4-argon+pe-tool" +# The API key each provider's models need, by litellm model-id prefix. +PROVIDER_KEYS = { + "gemini/": "GEMINI_API_KEY", + "gpt-": "OPENAI_API_KEY", + "claude-": "ANTHROPIC_API_KEY", +} YEAR = 2026 REFERENCE_ENGINE = "2.41.0" @@ -116,6 +128,16 @@ "once with your final answers, as instructed above." ) +# The notice for a JSON-contract model, whose API refuses a forced tool: the +# same tool, and the board's JSON reply in place of submit_outputs. +JSON_TOOL_NOTICE = ( + "\n\nYou also have a `calculate_household` function. It runs PolicyEngine US " + "(policyengine-us 2.41.0), an open-source rules engine for US federal and " + "state taxes and benefits, on a household situation you build from the facts " + "above. You may call it in up to 8 turns. Then give your final answers as the " + "single JSON object described above." +) + CALCULATE_DESCRIPTION = ( "Run PolicyEngine US (policyengine-us 2.41.0) on one household and return " "the requested variables. `situation` is a PolicyEngine household JSON " @@ -732,34 +754,6 @@ def load_board(snapshot: Path = SNAPSHOT): } -def build_initial_request(scenario, variables: list[str]) -> tuple[list[dict], dict]: - """The board's request for this household, plus the tool notice and tool.""" - from policybench.config import MODELS - from policybench.eval_no_tools import ( - _build_answer_tool, - _chat_completion_request_kwargs, - _completion_budget_ceiling, - ) - - model_id = MODELS[MODEL_NAME] - messages, kwargs = _chat_completion_request_kwargs(scenario, variables, model_id) - board_prompt = messages[0]["content"] - messages = [{"role": "user", "content": board_prompt + TOOL_NOTICE}] - answer_tool = _build_answer_tool(variables, country=scenario.country) - kwargs = {k: v for k, v in kwargs.items() if k not in {"messages", "caching"}} - kwargs["tools"] = [calculate_tool_schema(), answer_tool] - kwargs["tool_choice"] = "required" - kwargs["_board_prompt_sha256"] = hashlib.sha256(board_prompt.encode()).hexdigest() - kwargs["_budget_ceiling"] = _completion_budget_ceiling(model_id) - return messages, kwargs - - -def forced_answer_choice() -> dict: - from policybench.eval_no_tools import ANSWER_FUNCTION_NAME, _tool_choice_for - - return _tool_choice_for(ANSWER_FUNCTION_NAME) - - def _tool_calls(message) -> list: return list(getattr(message, "tool_calls", None) or []) @@ -812,6 +806,303 @@ def _has_thought_signature(message) -> bool: return bool(isinstance(fields, dict) and fields.get("thought_signatures")) +@dataclass(frozen=True) +class ModelProfile: + """How the board sends one model, and how its tool arm answers. + + ``api`` follows the board: the Responses API for GPT-5 and GPT-6 models + (``_uses_responses_api``), chat completions otherwise. ``contract`` is the + model card's answer contract: "tool" answers through the forced + ``submit_outputs`` function; "json" answers in the board's JSON reply, + for models whose API refuses a forced tool (the card says so for Claude + Opus 5.5), and offers ``calculate_household`` under tool_choice auto. + """ + + name: str + model_id: str + api: str + contract: str + + @property + def notice(self) -> str: + return TOOL_NOTICE if self.contract == "tool" else JSON_TOOL_NOTICE + + @property + def arm1_label(self) -> str: + return self.name + + @property + def arm2_label(self) -> str: + return f"{self.name}+pe-tool" + + +def model_profile(name: str = MODEL_NAME) -> ModelProfile: + from policybench.config import MODELS + from policybench.eval_no_tools import ( + _answer_contract_for_model, + _uses_responses_api, + ) + + model_id = MODELS[name] + return ModelProfile( + name=name, + model_id=model_id, + api="responses" if _uses_responses_api(model_id) else "chat", + contract=_answer_contract_for_model(model_id), + ) + + +def build_initial_request( + scenario, variables: list[str], profile: ModelProfile | None = None +) -> tuple[str, dict]: + """The board's prompt and request settings for this household, with the + tool notice appended and the calculate tool added.""" + from policybench.eval_no_tools import ( + _build_answer_tool, + _chat_completion_request_kwargs, + _completion_budget_ceiling, + _responses_request_kwargs, + ) + + profile = profile or model_profile() + if profile.api == "responses": + messages, kwargs = _responses_request_kwargs( + scenario, variables, profile.model_id + ) + kwargs = {k: v for k, v in kwargs.items() if k not in {"input"}} + else: + messages, kwargs = _chat_completion_request_kwargs( + scenario, variables, profile.model_id + ) + kwargs = {k: v for k, v in kwargs.items() if k not in {"messages", "caching"}} + board_prompt = messages[0]["content"] + kwargs.pop("tools", None) + kwargs.pop("tool_choice", None) + kwargs["_board_prompt_sha256"] = hashlib.sha256(board_prompt.encode()).hexdigest() + kwargs["_budget_ceiling"] = _completion_budget_ceiling(profile.model_id) + kwargs["_answer_tool"] = ( + _build_answer_tool(variables, country=scenario.country) + if profile.contract == "tool" + else None + ) + return board_prompt + profile.notice, kwargs + + +def forced_answer_choice(profile: ModelProfile | None = None) -> dict: + from policybench.eval_no_tools import ANSWER_FUNCTION_NAME, _tool_choice_for + + if profile is not None and profile.api == "responses": + return {"type": "function", "name": ANSWER_FUNCTION_NAME} + return _tool_choice_for(ANSWER_FUNCTION_NAME) + + +def _responses_tool(chat_tool: dict) -> dict: + """A chat-completions function tool in the Responses API's flat shape.""" + function = chat_tool["function"] + return { + "type": "function", + "name": function["name"], + "description": function["description"], + "parameters": function["parameters"], + } + + +@dataclass +class Call: + id: str + name: str | None + arguments: Any + + +@dataclass +class Turn: + response: Any + content: str | None + calls: list[Call] + finish_reason: str | None + reasoning_content: str | None = None + has_thought_signature: bool = False + + +CACHE_BREAKPOINT = {"type": "ephemeral"} + + +def _with_cache_breakpoint(message: dict) -> dict: + """The message with an Anthropic prompt-cache breakpoint at its end. + + Anthropic caches a prefix only at a marked block, and litellm carries the + mark from a text block or from a tool message. Gemini and OpenAI cache + repeated prefixes on their own. + """ + message = dict(message) + if message.get("role") == "tool": + message["cache_control"] = CACHE_BREAKPOINT + elif isinstance(message.get("content"), str): + message["content"] = [ + { + "type": "text", + "text": message["content"], + "cache_control": CACHE_BREAKPOINT, + } + ] + return message + + +class ChatConversation: + """A chat-completions conversation: one growing message list. + + With ``cache_breakpoint`` each request marks its last message, so a turn + reads the conversation so far from Anthropic's prompt cache. The stored + messages stay unmarked, which keeps one breakpoint per request. + """ + + budget_key = "max_completion_tokens" + + def __init__(self, prompt: str, cache_breakpoint: bool = False): + self.messages: list[dict] = [{"role": "user", "content": prompt}] + self.cache_breakpoint = cache_breakpoint + + def request_kwargs(self, base: dict, tools: list[dict], tool_choice) -> dict: + kwargs = dict(base) + messages = self.messages + if self.cache_breakpoint: + messages = [*messages[:-1], _with_cache_breakpoint(messages[-1])] + kwargs.update({"messages": messages, "tools": tools}) + if tool_choice is not None: + kwargs["tool_choice"] = tool_choice + if self.budget_key not in kwargs and "max_tokens" in kwargs: + self.budget_key = "max_tokens" + return kwargs + + def read(self, response, budget) -> Turn: + from policybench.eval_no_tools import _response_finish_reason + + choices = getattr(response, "choices", None) or [] + if not choices: + raise ProviderEmpty("response has no choices") + message = choices[0].message + return Turn( + response=response, + content=getattr(message, "content", None), + calls=[ + Call(_call_id(c), _call_name(c), _call_arguments(c)) + for c in _tool_calls(message) + ], + finish_reason=_response_finish_reason(response, responses_api=False), + reasoning_content=getattr(message, "reasoning_content", None), + has_thought_signature=_has_thought_signature(message), + ) + + def add_assistant(self, turn: Turn) -> None: + self.messages.append(_message_dict(turn.response.choices[0].message)) + + def add_tool_result(self, call: Call, result: dict) -> None: + self.messages.append( + { + "role": "tool", + "tool_call_id": call.id, + "name": call.name or "unknown", + "content": json.dumps(result), + } + ) + + def add_user(self, text: str) -> None: + self.messages.append({"role": "user", "content": text}) + + def transcript(self) -> list: + return self.messages + + +class ResponsesConversation: + """A Responses API conversation, chained by previous_response_id. + + The provider keeps each response, its reasoning included, so a turn sends + only what is new: the function call outputs and any user text. + """ + + budget_key = "max_output_tokens" + + def __init__(self, prompt: str): + self.prompt = prompt + self.previous_response_id: str | None = None + self.pending: list[dict] = [] + self.log: list[dict] = [] + + def request_kwargs(self, base: dict, tools: list[dict], tool_choice) -> dict: + kwargs = dict(base) + kwargs["tools"] = [_responses_tool(t) for t in tools] + if tool_choice is not None: + kwargs["tool_choice"] = tool_choice + if self.previous_response_id is None: + kwargs["input"] = self.prompt + else: + kwargs["input"] = list(self.pending) + kwargs["previous_response_id"] = self.previous_response_id + return kwargs + + def read(self, response, budget) -> Turn: + from policybench.eval_no_tools import ( + _get_response_item_attr, + _response_finish_reason, + _responses_content_and_tool_calls, + ) + + content, _ = _responses_content_and_tool_calls(response) + calls = [] + for item in getattr(response, "output", None) or []: + if _get_response_item_attr(item, "type") == "function_call": + calls.append( + Call( + _get_response_item_attr(item, "call_id") + or _get_response_item_attr(item, "id") + or "", + _get_response_item_attr(item, "name"), + _get_response_item_attr(item, "arguments"), + ) + ) + return Turn( + response=response, + content=content, + calls=calls, + finish_reason=_response_finish_reason( + response, responses_api=True, completion_budget_tokens=budget + ), + ) + + def add_assistant(self, turn: Turn) -> None: + self.log.append( + { + "input": self.prompt + if self.previous_response_id is None + else self.pending, + "response_id": getattr(turn.response, "id", None), + "content": turn.content, + "calls": [asdict(c) for c in turn.calls], + } + ) + self.previous_response_id = getattr(turn.response, "id", None) + self.pending = [] + + def add_tool_result(self, call: Call, result: dict) -> None: + self.pending.append( + { + "type": "function_call_output", + "call_id": call.id, + "output": json.dumps(result), + } + ) + + def add_user(self, text: str) -> None: + self.pending.append({"role": "user", "content": text}) + + def transcript(self) -> list: + return self.log + ([{"pending": self.pending}] if self.pending else []) + + +class ProviderEmpty(RuntimeError): + pass + + class Spend: def __init__(self, budget_usd: float, spent_usd: float = 0.0): self.budget_usd = budget_usd @@ -856,9 +1147,15 @@ def run_household( completion_fn: Callable[..., Any], spend: Spend, turn_cap: int = TOOL_TURN_CAP, + profile: ModelProfile | None = None, ) -> dict: - """One household's conversation: tool turns, then the board's answer tool.""" - from policybench.config import MODELS + """One household's conversation: tool turns, then the board's answer. + + A tool-contract model answers through ``submit_outputs`` (forced once the + cap is reached, as the board forces it). A JSON-contract model answers in + the board's JSON reply; its API refuses a forced tool, so the calculate + tool is offered under tool_choice auto and the cap is a user message. + """ from policybench.eval_no_tools import ( _apply_thinking_configuration, _build_answer_tool, @@ -866,80 +1163,171 @@ def run_household( _extract_usage_metadata, _merge_repair_cells, _missing_explanations, - _response_finish_reason, _serialize_response_payload, extract_explanations, extract_predictions, ) - model_id = MODELS[MODEL_NAME] - messages, base = build_initial_request(scenario, variables) + profile = profile or model_profile() + prompt, base = build_initial_request(scenario, variables, profile) board_prompt_sha256 = base.pop("_board_prompt_sha256") ceiling = base.pop("_budget_ceiling") - calculate_tool = base["tools"][0] + answer_tool = base.pop("_answer_tool") + calculate_tool = calculate_tool_schema() + conversation = ( + ResponsesConversation(prompt) + if profile.api == "responses" + else ChatConversation( + prompt, cache_breakpoint=profile.model_id.startswith("claude-") + ) + ) + json_contract = profile.contract == "json" record: dict[str, Any] = { "scenario_id": scenario.id, + "model": profile.name, + "api": profile.api, + "contract": profile.contract, "variables": variables, "board_prompt_sha256": board_prompt_sha256, - "tool_notice_sha256": hashlib.sha256(TOOL_NOTICE.encode()).hexdigest(), + "tool_notice_sha256": hashlib.sha256(profile.notice.encode()).hexdigest(), "turns": [], "status": "running", } predictions: dict[str, float | None] = {v: None for v in variables} explanations: dict[str, str | None] = {v: None for v in variables} - pending = list(variables) tool_turns = 0 repair_rounds = 0 - forced = False answer_variables = list(variables) household_started = time.perf_counter() + def tools_and_choice(kind: str): + if json_contract: + return [calculate_tool], "auto" + if kind == "repair": + return ( + [ + calculate_tool, + _build_answer_tool(answer_variables, country=scenario.country), + ], + forced_answer_choice(profile), + ) + if kind == "forced_answer": + return [calculate_tool, answer_tool], forced_answer_choice(profile) + return [calculate_tool, answer_tool], "required" + + def accept_answer(content, calls) -> list[str]: + """Merge an answer into the predictions; return the outputs still needed.""" + nonlocal predictions, explanations + got = extract_predictions(content, answer_variables, tool_calls=calls) + got_explanations = extract_explanations( + content, answer_variables, tool_calls=calls + ) + got, got_explanations = _enforce_explanation_value_contract( + got, got_explanations, answer_variables + ) + # The board's merge: a repair fills only cells still missing, and a + # valid earlier value or explanation always stands. + disagreements = record.setdefault("repair_disagreements", []) + predictions = _merge_repair_cells( + predictions, + {v: got.get(v) for v in answer_variables}, + field="prediction", + disagreements=disagreements, + ) + explanations = _merge_repair_cells( + explanations, + {v: got_explanations.get(v) for v in answer_variables}, + field="explanation", + disagreements=disagreements, + ) + pending = {v for v in variables if predictions[v] is None} + missing_explanation = set(_missing_explanations(explanations, variables)) + return [v for v in variables if v in pending | missing_explanation] + + def repair_text(still: list[str]) -> str: + kept = [ + f"{v} (keep your value {number}; its explanation must end with " + f"`value = {number}`)" + for v in still + if predictions[v] is not None + for number in [plain_number(predictions[v])] + ] + how = ( + "Reply with a JSON object holding only these outputs" + if json_contract + else "Call submit_outputs again with only these outputs" + ) + return ( + f"Your {'answer' if json_contract else 'submit_outputs call'} omitted " + f"or left unusable these outputs: {', '.join(still)}. {how}, each with " + "a numeric value and an explanation ending with `value = X`." + + ( + " Explanations only are missing for: " + "; ".join(kept) + "." + if kept + else "" + ) + ) + + def run_calculations(turn_record: dict, calls: list[Call]) -> bool: + """Run the turn's calculations; False if the pool had no worker.""" + for position, call in enumerate(calls): + if position >= MAX_CALCULATIONS_PER_TURN: + result, tool_latency = ( + { + "error": "not_run", + "message": ( + f"at most {MAX_CALCULATIONS_PER_TURN} calculations per turn" + ), + }, + 0.0, + ) + else: + result, tool_latency = tool_call(call.arguments) + turn_record["calls"].append( + { + "name": TOOL_NAME, + "arguments": call.arguments, + "result": result, + "tool_latency_seconds": tool_latency, + } + ) + if isinstance(result, dict) and result.get("error") == POOL_UNAVAILABLE: + return False + conversation.add_tool_result(call, result) + return True + try: + forced = False for turn_index in range(MAX_TOTAL_TURNS): if spend.exhausted(): record["status"] = "budget_stop" break if repair_rounds: kind = "repair" - tool_choice = forced_answer_choice() - tools = [ - calculate_tool, - _build_answer_tool(answer_variables, country=scenario.country), - ] - elif forced or tool_turns >= turn_cap: - kind = "forced_answer" - forced = True - tool_choice = forced_answer_choice() - tools = base["tools"] + elif not json_contract and (forced or tool_turns >= turn_cap): + kind, forced = "forced_answer", True else: kind = "open" - tool_choice = "required" - tools = base["tools"] - kwargs = { - k: v for k, v in base.items() if k not in {"tools", "tool_choice"} - } - kwargs.update( - { - "model": model_id, - "messages": messages, - "tools": tools, - "tool_choice": tool_choice, - } - ) - _apply_thinking_configuration(kwargs, model_id) - budget_key = ( - "max_completion_tokens" - if "max_completion_tokens" in kwargs - else "max_tokens" + tools, tool_choice = tools_and_choice(kind) + kwargs = conversation.request_kwargs( + {k: v for k, v in base.items() if k not in {"tools", "tool_choice"}} + | {"model": profile.model_id}, + tools, + tool_choice, ) + _apply_thinking_configuration(kwargs, profile.model_id) + budget_key = conversation.budget_key escalated_from = None try: response, started_at, completed_at, latency = _request_with_retries( completion_fn, kwargs ) - finish_reason = _response_finish_reason(response, responses_api=False) - if finish_reason == "length" and kwargs.get(budget_key, 0) < ceiling: - usage = _extract_usage_metadata(response, model_id, messages, "") + turn = conversation.read(response, kwargs.get(budget_key)) + if ( + turn.finish_reason == "length" + and kwargs.get(budget_key, 0) < ceiling + ): + usage = _extract_usage_metadata(response, profile.model_id, [], "") spend.add(usage.get("total_cost_usd")) record["turns"].append( { @@ -956,30 +1344,25 @@ def run_household( response, started_at, completed_at, latency = _request_with_retries( completion_fn, kwargs ) - finish_reason = _response_finish_reason( - response, responses_api=False - ) + turn = conversation.read(response, kwargs.get(budget_key)) except Exception as error: record["status"] = "provider_error" record["error"] = f"{type(error).__name__}: {error}"[:MAX_MESSAGE_CHARS] break - choices = getattr(response, "choices", None) or [] - if not choices: - record["status"] = "provider_error" - record["error"] = "response has no choices" - break - message = choices[0].message - content = getattr(message, "content", None) - calls = _tool_calls(message) raw = _serialize_response_payload( - content=content, tool_calls=calls, function_call=None + content=turn.content, + tool_calls=[ + {"id": c.id, "function": {"name": c.name, "arguments": c.arguments}} + for c in turn.calls + ], + function_call=None, ) usage = _extract_usage_metadata( - response, model_id, messages, raw or content or "" + response, profile.model_id, [], raw or turn.content or "" ) spend.add(usage.get("total_cost_usd")) - turn = { + turn_record = { "index": turn_index, "kind": kind, "tool_choice": tool_choice, @@ -988,191 +1371,129 @@ def run_household( "request_started_at": started_at, "request_completed_at": completed_at, "latency_seconds": latency, - "finish_reason": finish_reason, - "content": content, - "reasoning_content": getattr(message, "reasoning_content", None), - "has_thought_signature": _has_thought_signature(message), + "finish_reason": turn.finish_reason, + "content": turn.content, + "reasoning_content": turn.reasoning_content, + "has_thought_signature": turn.has_thought_signature, + "response_id": getattr(response, "id", None), "calls": [], **usage, } - record["turns"].append(turn) - messages.append(_message_dict(message)) + record["turns"].append(turn_record) + conversation.add_assistant(turn) - answer_calls = [c for c in calls if _call_name(c) == "submit_outputs"] - calculate_calls = [c for c in calls if _call_name(c) == TOOL_NAME] + answer_calls = [c for c in turn.calls if c.name == "submit_outputs"] + calculate_calls = [c for c in turn.calls if c.name == TOOL_NAME] other_calls = [ - c for c in calls if c not in answer_calls and c not in calculate_calls + c + for c in turn.calls + if c not in answer_calls and c not in calculate_calls ] + # A tool-contract answer: the submit_outputs call ends the turn. if answer_calls: - got = extract_predictions( - content, answer_variables, tool_calls=answer_calls - ) - got_explanations = extract_explanations( - content, answer_variables, tool_calls=answer_calls - ) - got, got_explanations = _enforce_explanation_value_contract( - got, got_explanations, answer_variables - ) - # The board's merge: a repair fills only cells still missing, and a - # valid earlier value or explanation always stands. - disagreements = record.setdefault("repair_disagreements", []) - predictions = _merge_repair_cells( - predictions, - {v: got.get(v) for v in answer_variables}, - field="prediction", - disagreements=disagreements, + still = accept_answer( + turn.content, + [ + {"function": {"name": c.name, "arguments": c.arguments}} + for c in answer_calls + ], ) - explanations = _merge_repair_cells( - explanations, - {v: got_explanations.get(v) for v in answer_variables}, - field="explanation", - disagreements=disagreements, - ) - turn["calls"].append( + turn_record["calls"].append( { "name": "submit_outputs", - "arguments": _call_arguments(answer_calls[0]), - "ignored_calls": len(calculate_calls) - + len(other_calls) - + len(answer_calls) - - 1, + "arguments": answer_calls[0].arguments, + "ignored_calls": len(turn.calls) - 1, } ) - pending = [v for v in variables if predictions[v] is None] - missing_explanation = _missing_explanations(explanations, variables) - still = [ - v for v in variables if v in set(pending) | set(missing_explanation) - ] - for call in calls: - messages.append( + for call in turn.calls: + conversation.add_tool_result( + call, + {"accepted": not still, "missing": still} + if call is answer_calls[0] + else {"error": "not run: submit_outputs ended the turn"}, + ) + elif calculate_calls and ( + tool_turns >= turn_cap or (not json_contract and kind != "open") + ): + # Past the cap, or in a turn that forced the answer, nothing + # runs. tool_choice auto cannot withhold the tool from a + # JSON-contract model; a forced turn should never carry a + # calculation, but a provider that sends one gets no result. + refusal = { + "error": "not_run", + "message": "no calculation turns are left; " + + ( + "reply with your final answers as the JSON object " + "described above" + if json_contract + else "call submit_outputs with your final answers" + ), + } + for call in turn.calls: + turn_record["calls"].append( { - "role": "tool", - "tool_call_id": _call_id(call), - "name": _call_name(call), - "content": json.dumps( - {"accepted": not still, "missing": still} - if call is answer_calls[0] - else {"error": "not run: submit_outputs ended the turn"} - ), + "name": call.name, + "arguments": call.arguments, + "result": refusal, + "tool_latency_seconds": 0.0, } ) - if not still: - record["status"] = "complete" - break - if repair_rounds >= MAX_REPAIR_ROUNDS: - record["status"] = "incomplete" - break - repair_rounds += 1 - answer_variables = still - kept = [ - f"{v} (keep your value {number}; its explanation must end " - f"with `value = {number}`)" - for v in still - if predictions[v] is not None - for number in [plain_number(predictions[v])] - ] - messages.append( - { - "role": "user", - "content": ( - "Your submit_outputs call omitted or left unusable these " - f"outputs: {', '.join(still)}. Call submit_outputs again " - "with only these outputs, each with a numeric value and an " - "explanation ending with `value = X`." - + ( - " Explanations only are missing for: " - + "; ".join(kept) - + "." - if kept - else "" - ) - ), - } - ) + conversation.add_tool_result(call, refusal) continue - - if calculate_calls: + elif calculate_calls: tool_turns += 1 - pool_down = False - for position, call in enumerate(calculate_calls): - if position >= MAX_CALCULATIONS_PER_TURN: - result, tool_latency = ( - { - "error": "not_run", - "message": ( - f"at most {MAX_CALCULATIONS_PER_TURN} " - "calculations per turn" - ), - }, - 0.0, - ) - else: - result, tool_latency = tool_call(_call_arguments(call)) - turn["calls"].append( - { - "name": TOOL_NAME, - "arguments": _call_arguments(call), - "result": result, - "tool_latency_seconds": tool_latency, - } - ) - if ( - isinstance(result, dict) - and result.get("error") == POOL_UNAVAILABLE - ): - pool_down = True - break - messages.append( - { - "role": "tool", - "tool_call_id": _call_id(call), - "name": TOOL_NAME, - "content": json.dumps(result), - } - ) - for call in other_calls: - messages.append( - { - "role": "tool", - "tool_call_id": _call_id(call), - "name": _call_name(call) or "unknown", - "content": json.dumps({"error": "unknown function"}), - } - ) - if pool_down: + if not run_calculations(turn_record, calculate_calls): # Infrastructure, not the model: a rerunnable status, so the # household is not scored on an answer the pool prevented. record["status"] = "harness_error" record["error"] = "the tool pool had no worker" break + for call in other_calls: + conversation.add_tool_result(call, {"error": "unknown function"}) if tool_turns >= turn_cap: - messages.append( - { - "role": "user", - "content": ( - f"You have used all {turn_cap} calculation turns. Call " - "submit_outputs now with your final answers." - ), - } + conversation.add_user( + f"You have used all {turn_cap} calculation turns. " + + ( + "Reply now with your final answers as the JSON object " + "described above." + if json_contract + else "Call submit_outputs now with your final answers." + ) ) continue - - for call in other_calls: - messages.append( - { - "role": "tool", - "tool_call_id": _call_id(call), - "name": _call_name(call) or "unknown", - "content": json.dumps({"error": "unknown function"}), - } + elif json_contract and turn.content: + # A JSON-contract answer: the reply itself. + still = accept_answer(turn.content, None) + answered = any(v is not None for v in predictions.values()) or any( + explanations.values() ) - messages.append( - { - "role": "user", - "content": "Call calculate_household or submit_outputs.", - } - ) + if not answered: + conversation.add_user( + "Reply with your final answers as the JSON object described " + "above, or call calculate_household." + ) + continue + else: + for call in other_calls: + conversation.add_tool_result(call, {"error": "unknown function"}) + conversation.add_user( + "Call calculate_household or give your final answers as the " + "JSON object described above." + if json_contract + else "Call calculate_household or submit_outputs." + ) + continue + + if not still: + record["status"] = "complete" + break + if repair_rounds >= MAX_REPAIR_ROUNDS: + record["status"] = "incomplete" + break + repair_rounds += 1 + answer_variables = still + conversation.add_user(repair_text(still)) else: record["status"] = "turn_limit" except Exception as error: # keep the turns, and their spend, so far @@ -1187,7 +1508,7 @@ def run_household( record["cap_hit"] = tool_turns >= turn_cap record["repair_rounds"] = repair_rounds record["wall_seconds"] = time.perf_counter() - household_started - record["messages"] = messages + record["messages"] = conversation.transcript() record["totals"] = household_totals(record) return record @@ -1234,6 +1555,9 @@ def household_totals(record: dict) -> dict: "completion_tokens": sum(_num(t.get("completion_tokens")) for t in turns), "reasoning_tokens": sum(_num(t.get("reasoning_tokens")) for t in turns), "cached_prompt_tokens": sum(_num(t.get("cached_prompt_tokens")) for t in turns), + "cache_write_prompt_tokens": sum( + _num(t.get("cache_write_prompt_tokens")) for t in turns + ), "cost_usd": sum(_num(t.get("total_cost_usd")) for t in turns), "model_seconds": sum(_num(t.get("latency_seconds")) for t in turns), "tool_seconds": sum(_num(c.get("tool_latency_seconds")) for c in calls), @@ -1413,12 +1737,31 @@ def resume_state(households_dir: Path) -> tuple[float, set[str], dict[str, dict] return spent, done, previous +def provider_key(model_id: str) -> str: + for prefix, key in PROVIDER_KEYS.items(): + if model_id.startswith(prefix): + return key + raise SystemExit(f"no provider key is known for {model_id}") + + +def records_model(records: Mapping[str, dict]) -> str: + """The one model a run directory's records belong to; records written + before the pilot took --model are Argon's.""" + models = {r.get("model", MODEL_NAME) for r in records.values()} + if len(models) > 1: + raise SystemExit(f"records from more than one model: {sorted(models)}") + return models.pop() if models else MODEL_NAME + + def command_run(args) -> None: _require_engine() - if not os.environ.get("GEMINI_API_KEY"): - raise SystemExit("GEMINI_API_KEY is not set") - from litellm import completion + profile = model_profile(args.model) + key = provider_key(profile.model_id) + if not os.environ.get(key): + raise SystemExit(f"{key} is not set") + from litellm import completion, responses + completion_fn = responses if profile.api == "responses" else completion board = load_board() out = Path(args.out_dir) households_dir = out / "households" @@ -1430,6 +1773,11 @@ def command_run(args) -> None: ids = ids[: args.limit] spent_before, done, previous = resume_state(households_dir) + if previous and records_model(previous) != profile.name: + raise SystemExit( + f"{households_dir} holds {records_model(previous)} records, " + f"not {profile.name}" + ) spend = Spend(args.budget_usd, spent_before) todo = [sid for sid in ids if sid not in done] manifest = { @@ -1440,8 +1788,11 @@ def command_run(args) -> None: "versions": _versions(), "fix_module": str(FIX_MODULE.relative_to(ROOT)), "fix_module_sha256": _sha256(FIX_MODULE), - "model": MODEL_NAME, - "tool_notice": TOOL_NOTICE, + "model": profile.name, + "model_id": profile.model_id, + "api": profile.api, + "answer_contract": profile.contract, + "tool_notice": profile.notice, "calculate_tool": calculate_tool_schema(), "turn_cap": TOOL_TURN_CAP, "max_calculations_per_turn": MAX_CALCULATIONS_PER_TURN, @@ -1476,12 +1827,14 @@ def one(sid: str) -> dict: board["scenarios"][sid], board["variables"][sid], pool.call, - completion, + completion_fn, spend, + profile=profile, ) except Exception as error: # one household must not end the run record = { "scenario_id": sid, + "model": profile.name, "variables": board["variables"][sid], "status": "harness_error", "error": f"{type(error).__name__}: {error}"[:MAX_MESSAGE_CHARS], @@ -1549,14 +1902,16 @@ def load_records(out_dir: Path) -> dict[str, dict]: return records -def arm2_predictions(records: Mapping[str, dict]) -> pd.DataFrame: +def arm2_predictions( + records: Mapping[str, dict], label: str = ARM2_LABEL +) -> pd.DataFrame: rows = [] for sid, record in records.items(): for variable in record["variables"]: value = record["predictions"].get(variable) rows.append( { - "model": ARM2_LABEL, + "model": label, "scenario_id": sid, "variable": variable, "prediction": value, @@ -1567,9 +1922,14 @@ def arm2_predictions(records: Mapping[str, dict]) -> pd.DataFrame: return pd.DataFrame(rows) -def arm1_predictions(snapshot: Path = SNAPSHOT) -> pd.DataFrame: +def arm1_predictions( + snapshot: Path = SNAPSHOT, label: str = ARM1_LABEL +) -> pd.DataFrame: frame = pd.read_csv(snapshot / "predictions.csv.gz", low_memory=False) - return frame[frame["model"] == ARM1_LABEL].copy() + rows = frame[frame["model"] == label].copy() + if rows.empty: + raise SystemExit(f"the snapshot has no frozen predictions for {label}") + return rows def command_score(args) -> None: @@ -1579,6 +1939,8 @@ def command_score(args) -> None: board = load_board() out = Path(args.out_dir) records = load_records(out) + profile = model_profile(records_model(records)) + arm1_label, arm2_label = profile.arm1_label, profile.arm2_label complete = sorted( sid for sid, r in records.items() if r.get("status") == "complete" ) @@ -1596,9 +1958,9 @@ def command_score(args) -> None: f"failure and score as misses until a resume reruns them: {leftover}" ) scored = board["scored"][board["scored"].scenario_id.isin(households)] - arm1 = arm1_predictions() + arm1 = arm1_predictions(label=arm1_label) arm1 = arm1[arm1.scenario_id.isin(households)] - arm2 = arm2_predictions({sid: records[sid] for sid in households}) + arm2 = arm2_predictions({sid: records[sid] for sid in households}, arm2_label) scenarios = board["scenarios_frame"][ board["scenarios_frame"].scenario_id.isin(households) ] @@ -1656,7 +2018,7 @@ def command_score(args) -> None: by_household = household_weighted_exact(scored, both, scenarios) means = by_household.groupby("model")["weighted_exact"].mean() published = summary.set_index("model")["weighted_exact"] - for model in (ARM1_LABEL, ARM2_LABEL): + for model in (arm1_label, arm2_label): if abs(means[model] - published[model]) > 1e-9: raise SystemExit( f"household scores for {model} average {means[model]}, not the " @@ -1666,7 +2028,7 @@ def command_score(args) -> None: index="scenario_id", columns="model", values="weighted_exact" ) wide.to_csv(out / "household_scores.csv") - bootstrap = paired_bootstrap(wide[ARM1_LABEL], wide[ARM2_LABEL]) + bootstrap = paired_bootstrap(wide[arm1_label], wide[arm2_label]) print(json.dumps(bootstrap, indent=2)) flips = ( @@ -1692,6 +2054,9 @@ def command_score(args) -> None: (out / "score.json").write_text( json.dumps( { + "model": profile.name, + "arm1_label": arm1_label, + "arm2_label": arm2_label, "households": len(households), "incomplete_households": sorted(set(records) - set(complete)), "rerunnable_households": leftover, @@ -2276,6 +2641,78 @@ def apply_reference_input(model: dict, reference: dict, diff: dict) -> dict | No return situation +def _amounts(side) -> list[float]: + if not isinstance(side, dict): + return [] + return [ + float(v) + for v in side.values() + if isinstance(v, (int, float)) and not isinstance(v, bool) + ] + + +# Reviewed re-entries: (the reference's input, another engine input a model +# put the same stated amount in). Each pair was read in a transcript; an +# equal amount anywhere else is only a candidate, since a number can match +# by chance (scenario_018's $50 of interest beside 50 weekly hours). +REENTRY_PAIRS = frozenset( + { + # Sol, scenario_018: "taxable interest income" entered as the + # engine's interest_income, the sum of taxable and tax-exempt interest. + ("taxable_interest_income", "interest_income"), + # Sol, scenario_088: "home mortgage interest" entered as + # mortgage_interest, the sum of deductible and non-deductible interest. + ("home_mortgage_interest", "mortgage_interest"), + } +) + + +def _entered_as(cause: dict, diffs: list[dict]) -> tuple[list[str], list[str]]: + """The model's other inputs, on the cause's unit and period, that hold + its amount: (reviewed re-entries, other numeric candidates).""" + wanted = { + period: float(value) + for period, value in (cause["reference"] or {}).items() + if isinstance(value, (int, float)) and not isinstance(value, bool) + } + confirmed, candidates = [], [] + for other in diffs: + if other is cause or other["entity"] != cause["entity"]: + continue + if other.get("model_unit") != cause.get("model_unit"): + continue + model_side = other["model"] if isinstance(other["model"], dict) else {} + if not any( + period in wanted + and isinstance(value, (int, float)) + and not isinstance(value, bool) + and abs(float(value) - wanted[period]) <= 1.0 + for period, value in model_side.items() + ): + continue + label = f"{other['entity']}.{other['variable']}" + if (cause["variable"], other["variable"]) in REENTRY_PAIRS: + confirmed.append(label) + else: + candidates.append(label) + return confirmed, candidates + + +def _entered_elsewhere(cause: dict, diffs: list[dict]) -> bool: + """Whether the model set this cause, or entered its amount elsewhere. + + A cause the model set itself (a non-empty model side) was entered + differently. One the model left unset was still entered if a reviewed + counterpart (``REENTRY_PAIRS``) holds the same amount, within $1, for the + same person or unit and period: Sol's $10,777 under ``mortgage_interest`` + for a reference ``home_mortgage_interest``. Any other equal number is a + candidate only. + """ + if cause["model"] not in ({}, None): + return True + return bool(_entered_as(cause, diffs)[0]) + + def attribute_miss( system, model: dict, reference: dict, scenario, output: str, target, diffs ) -> dict: @@ -2283,32 +2720,206 @@ def attribute_miss( Each difference is applied alone; the ones that bring the output within the board's tolerance of the reference are its single causes. When no - single input does, every applicable difference is applied together. + single input does, every applicable difference is applied together, and + if that fixes it, the necessary inputs are the ones whose omission from + the full set leaves the output wrong (leave-one-out). + + ``mapping_kind`` reads the causes (single, else necessary): + "entered_differently" when the model set one of them, or put a cause's + amount under a reviewed counterpart (``_entered_elsewhere``); "left_out" + otherwise. ``entered_as`` lists those counterparts, and + ``entered_as_candidates`` any other input holding the same amount. """ - causes = [] + causes, cause_diffs = [], [] for diff in diffs: situation = apply_reference_input(model, reference, diff) if situation is None: continue if hit(output, output_value(system, situation, scenario, output), target): causes.append(f"{diff['entity']}.{diff['variable']}") + cause_diffs.append(diff) joint = None + necessary, necessary_diffs = [], [] if not causes and diffs: - situation = model - for diff in diffs: - applied = apply_reference_input(situation, reference, diff) - situation = applied if applied is not None else situation - joint = hit(output, output_value(system, situation, scenario, output), target) - return {"single_causes": causes, "all_differences_fix_it": joint} + def apply_all(skip=None) -> dict: + situation = model + for diff in diffs: + if diff is skip: + continue + applied = apply_reference_input(situation, reference, diff) + situation = applied if applied is not None else situation + return situation + + joint = hit(output, output_value(system, apply_all(), scenario, output), target) + if joint: + for diff in diffs: + value = output_value(system, apply_all(skip=diff), scenario, output) + if not hit(output, value, target): + necessary.append(f"{diff['entity']}.{diff['variable']}") + necessary_diffs.append(diff) + basis = cause_diffs or necessary_diffs + kind = None + if basis: + kind = ( + "entered_differently" + if any(_entered_elsewhere(d, diffs) for d in basis) + else "left_out" + ) + # The inputs that hold an unset cause's amount: evidence for + # entered_differently, listed so a reader can tell a real re-entry + # (interest_income for taxable_interest_income) from a coincidence. + unset = [d for d in basis if d["model"] in ({}, None)] + found = [_entered_as(d, diffs) for d in unset] + return { + "single_causes": causes, + "all_differences_fix_it": joint, + "necessary_inputs": necessary, + "mapping_kind": kind, + "entered_as": sorted({n for confirmed, _ in found for n in confirmed}), + "entered_as_candidates": sorted({n for _, cand in found for n in cand}), + } -def main_call(record: dict, scenario, outputs: list[str]) -> dict | None: - """The successful call that computed the most of the household's outputs. - A model often ends with a narrow probe on a skeleton situation (one - person, no state); the situation its answers rest on is the one that - computed the outputs. Ties go to the later call. +def _entity_matches(call: dict, submitted, entities) -> list[str]: + """Results of one call within $1 of the submitted value, on the output's + entities only, as variable[member]@period.""" + if submitted is None: + return [] + matches = [] + for name, entry in call["result"]["results"].items(): + if not isinstance(entry, dict): + continue + if entities is not None and entry.get("entity") not in entities: + continue + for member, value in (entry.get("values") or {}).items(): + if isinstance(value, bool) or not isinstance(value, (int, float)): + continue + if abs(float(value) - float(submitted)) <= 1.0: + label = f"{name}[{member}]@{entry.get('period')}" + if label not in matches: + matches.append(label) + return matches + + +def resolve_unread_output( + verdict: dict, + record: dict, + scenario, + output: str, + submitted, + board_value_for: Callable[[Any], float | None], + output_entities=None, + binary: bool = False, +) -> dict: + """Settle a wrong_variable_or_period verdict with the board's own output. + + A model can read a variable other than the board's, such as a state's own + ``ca_income_tax_before_refundable_credits``, so the output's engine + variable is never computed. Numbers alone do not show which call an + answer came from, so a verdict changes only on agreeing evidence: + + - The holding calls are the successful ones with a result within $1 of + the submitted value on one of ``output_entities``, the entities of the + output's engine variables (an equal ``employment_income`` on a person + is not evidence for a tax-unit tax). + - ``board_value_for`` computes the board's output on the situation of + each holding call and of the main call, the one that computed the most + of the household's outputs. Only if every one of them equals the + submitted value does the miss become wrong_input_mapping, attributed + on the holding call that computed the most outputs. A lone probe that + happens to hold a summed answer cannot outvote the main situation. + - Otherwise the verdict stands, with the main call's situation and board + value as its evidence, every holding call's value listed, and a note + when some agree and some do not. + - A binary output is never reclassified: a 0 or 1 matches too much by + chance. + - Nor is a miss in a household where no call computed any of its + outputs: every call ties at zero, so no situation is the main one + (the tie rule would otherwise let the last probe vouch for itself). """ + if verdict.get("category") != "wrong_variable_or_period": + return verdict + variables = list(record["variables"]) + + def key(call): + return json.dumps(call.get("arguments"), sort_keys=True, default=str) + + calls = _successful_calls(record) + main, main_count = main_call_and_count(record, scenario, variables) + no_main = main is not None and main_count == 0 + holding = ( + [] + if binary + else [c for c in calls if _entity_matches(c, submitted, output_entities)] + ) + values: dict[str, float | None] = {} + for call in holding + ([main] if main is not None else []): + if key(call) not in values: + values[key(call)] = board_value_for(call.get("arguments")) + + def agrees(call) -> bool: + value = values[key(call)] + return value is not None and hit(output, submitted, value) + + if ( + holding + and not no_main + and main is not None + and all(agrees(c) for c in holding) + and agrees(main) + ): + call = main_call({"turns": [{"calls": holding}]}, scenario, variables) + read = ", ".join(_entity_matches(call, submitted, output_entities)) + return { + **verdict, + "category": "wrong_input_mapping", + "detail": f"read {read}, equal to the board's output on that situation", + "answering_call": "holds the submitted value", + "holding_calls": len(holding), + "holding_board_values": [values[key(c)] for c in holding], + "board_output_on_situation": values[key(call)], + "tool_value": values[key(call)], + "call_arguments": call.get("arguments"), + } + if main is None: + return verdict + settled = { + **verdict, + "answering_call": "main call", + "holding_calls": len(holding), + "holding_board_values": [values[key(c)] for c in holding], + "board_output_on_situation": values[key(main)], + "call_arguments": main.get("arguments"), + } + if no_main: + settled["detail"] = ( + "no call computed any of the household's outputs, so no situation " + "is the main one; not reclassified" + ) + elif values[key(main)] is None: + settled["detail"] = ( + "the main situation could not be computed (it no longer validates); " + "not reclassified" + ) + elif not holding and not binary and agrees(main): + settled["detail"] = ( + "the main situation's board output equals the answer, but no call's " + "results hold it (an answer the model built itself); not reclassified" + ) + elif holding and any(agrees(c) for c in holding + [main]): + settled["detail"] = ( + "ambiguous: the situations holding the submitted value, or the " + "main one, disagree on the board's output" + ) + return settled + + +def main_call_and_count( + record: dict, scenario, outputs: list[str] +) -> tuple[dict | None, int]: + """The successful call that computed the most of the household's outputs, + and how many it computed (-1 with no successful call).""" best, best_count = None, -1 for call in _successful_calls(record): count = sum( @@ -2317,7 +2928,17 @@ def main_call(record: dict, scenario, outputs: list[str]) -> dict | None: ) if count >= best_count: best, best_count = call, count - return best + return best, best_count + + +def main_call(record: dict, scenario, outputs: list[str]) -> dict | None: + """The successful call that computed the most of the household's outputs. + + A model often ends with a narrow probe on a skeleton situation (one + person, no state); the situation its answers rest on is the one that + computed the outputs. Ties go to the later call. + """ + return main_call_and_count(record, scenario, outputs)[0] def situation_fidelity(records, board, system, registry, defaults, out: Path) -> None: @@ -2387,6 +3008,8 @@ def situation_fidelity(records, board, system, registry, defaults, out: Path) -> def command_diagnose(args) -> None: + from policybench.spec import metric_type_for_output + board = load_board() out = Path(args.out_dir) records = load_records(out) @@ -2420,6 +3043,29 @@ def command_diagnose(args) -> None: verdict = classify_miss( record, scenario, row.variable, submitted, row.reference ) + if registry is not None: + + def board_value_for(arguments, scenario=scenario, output=row.variable): + request, _ = normalize_request(arguments, registry) + if request is None: + return None + return output_value(system, request["situation"], scenario, output) + + entities = { + registry[name].entity + for name in engine_variables_for_output(row.variable) + if name in registry + } + verdict = resolve_unread_output( + verdict, + record, + scenario, + row.variable, + submitted, + board_value_for, + output_entities=entities or None, + binary=metric_type_for_output(row.variable) != "amount", + ) diff = [] attribution = {} arguments = verdict.get("call_arguments") @@ -2459,6 +3105,15 @@ def command_diagnose(args) -> None: "submitted_matches": json.dumps(verdict.get("submitted_matches")), "single_causes": json.dumps(attribution.get("single_causes")), "all_differences_fix_it": attribution.get("all_differences_fix_it"), + "necessary_inputs": json.dumps(attribution.get("necessary_inputs")), + "mapping_kind": attribution.get("mapping_kind"), + "entered_as": json.dumps(attribution.get("entered_as")), + "entered_as_candidates": json.dumps( + attribution.get("entered_as_candidates") + ), + "holding_board_values": json.dumps(verdict.get("holding_board_values")), + "answering_call": verdict.get("answering_call"), + "board_output_on_situation": verdict.get("board_output_on_situation"), "input_diff": json.dumps(diff, default=str), } ) @@ -2499,6 +3154,11 @@ def main(argv: list[str] | None = None) -> None: p = sub.add_parser(name) p.add_argument("--out-dir", required=True) if name == "run": + p.add_argument( + "--model", + default=MODEL_NAME, + help="a board model name (policybench.config.MODELS)", + ) p.add_argument("--scenarios", nargs="*") p.add_argument("--limit", type=int) p.add_argument("--workers", type=int, default=4) diff --git a/tests/test_tool_pilot_argon.py b/tests/test_tool_pilot_argon.py index 9a9d87d1..45fd67a1 100644 --- a/tests/test_tool_pilot_argon.py +++ b/tests/test_tool_pilot_argon.py @@ -1,4 +1,4 @@ -"""The Argon tool pilot's validator, conversation loop, scoring and diagnosis. +"""The tool pilot's validator, conversation loop, scoring and diagnosis. Every model call is faked. The registry is a small stand-in for the engine's, so the validator tests need no PolicyEngine system; the tests that build the @@ -568,9 +568,10 @@ def test_the_prompt_is_the_board_prompt_plus_the_notice(): board_messages, _ = _chat_completion_request_kwargs( scenario, VARIABLES, "gemini/gemini-4-argon" ) - messages, kwargs = pilot.build_initial_request(scenario, VARIABLES) - assert messages[0]["content"] == board_messages[0]["content"] + pilot.TOOL_NOTICE - assert "caching" not in kwargs + prompt, kwargs = pilot.build_initial_request(scenario, VARIABLES) + assert prompt == board_messages[0]["content"] + pilot.TOOL_NOTICE + assert "caching" not in kwargs and "messages" not in kwargs + assert kwargs["_answer_tool"]["function"]["name"] == "submit_outputs" def test_the_cap_forces_the_answer_tool(): @@ -657,9 +658,11 @@ def test_hit_is_the_boards_exact_rule(): assert not pilot.hit("snap", float("nan"), 1.0) -def call_record(values: dict, period="2026", people=("head", "kid")): +def call_record( + values: dict, period="2026", people=("head", "kid"), entity="x", age=30 +): sit = situation() - sit["people"] = {p: {"age": 30} for p in people} + sit["people"] = {p: {"age": age} for p in people} return { "name": pilot.TOOL_NAME, "arguments": json.dumps( @@ -667,7 +670,7 @@ def call_record(values: dict, period="2026", people=("head", "kid")): ), "result": { "results": { - name: {"entity": "x", "period": period, "values": vals} + name: {"entity": entity, "period": period, "values": vals} for name, vals in values.items() } }, @@ -1301,3 +1304,942 @@ def test_plain_number_round_trips_through_the_boards_trailer_pattern(value, text assert pilot.plain_number(value) == text assert _extract_terminal_explanation_value(f"Reason. value = {text}") == value + + +# --------------------------------------------------------------------------- +# Other models: the Responses API (GPT-6 Sol) and the JSON contract (Opus 5.5). +# --------------------------------------------------------------------------- + +SOL = "gpt-6-sol" +OPUS = "claude-opus-5.5" + + +def test_profiles_follow_the_boards_api_and_answer_contract(): + argon, sol, opus = (pilot.model_profile(n) for n in (pilot.MODEL_NAME, SOL, OPUS)) + assert (argon.api, argon.contract) == ("chat", "tool") + assert (sol.api, sol.contract) == ("responses", "tool") + assert (opus.api, opus.contract) == ("chat", "json") + assert argon.notice == pilot.TOOL_NOTICE and sol.notice == pilot.TOOL_NOTICE + assert opus.notice == pilot.JSON_TOOL_NOTICE + assert (argon.arm1_label, argon.arm2_label) == (pilot.ARM1_LABEL, pilot.ARM2_LABEL) + assert sol.arm2_label == "gpt-6-sol+pe-tool" + # Both notices describe the same tool; only the way to answer differs. + shared = "You may call it in up to 8 turns. Then " + assert pilot.TOOL_NOTICE.split(shared)[0] == pilot.JSON_TOOL_NOTICE.split(shared)[0] + + +def test_a_responses_prompt_is_the_boards_input_plus_the_notice(): + from policybench.eval_no_tools import _responses_request_kwargs + + scenario = make_scenario() + profile = pilot.model_profile(SOL) + _, board = _responses_request_kwargs(scenario, VARIABLES, profile.model_id) + prompt, kwargs = pilot.build_initial_request(scenario, VARIABLES, profile) + assert prompt == board["input"] + pilot.TOOL_NOTICE + assert kwargs["max_output_tokens"] == board["max_output_tokens"] + assert "input" not in kwargs and "tools" not in kwargs + + +def test_a_json_contract_prompt_is_the_boards_plus_the_json_notice(): + from policybench.eval_no_tools import _chat_completion_request_kwargs + + scenario = make_scenario() + profile = pilot.model_profile(OPUS) + messages, board = _chat_completion_request_kwargs( + scenario, VARIABLES, profile.model_id + ) + prompt, kwargs = pilot.build_initial_request(scenario, VARIABLES, profile) + assert prompt == messages[0]["content"] + pilot.JSON_TOOL_NOTICE + assert kwargs["_answer_tool"] is None + assert kwargs["response_format"] == board["response_format"] + + +def function_call(name, arguments, call_id): + return SimpleNamespace( + type="function_call", + id=f"fc_{call_id}", + call_id=call_id, + name=name, + arguments=json.dumps(arguments), + status="completed", + ) + + +def responses_response(items, response_id, prompt=1000, completion=200, cached=0): + usage = SimpleNamespace( + input_tokens=prompt, + output_tokens=completion, + total_tokens=prompt + completion, + input_tokens_details=SimpleNamespace(cached_tokens=cached), + output_tokens_details=SimpleNamespace(reasoning_tokens=50), + ) + return SimpleNamespace( + id=response_id, + output=[SimpleNamespace(type="reasoning", id=f"rs_{response_id}"), *items], + output_text=None, + status="completed", + incomplete_details=None, + usage=usage, + model=SOL, + ) + + +class FakeResponses: + def __init__(self, responses): + self.responses = list(responses) + self.requests = [] + + def __call__(self, **kwargs): + self.requests.append(json.loads(json.dumps(kwargs, default=str))) + return self.responses.pop(0) + + +def sol_calculate(call_id): + return function_call( + pilot.TOOL_NAME, + {"situation": json.dumps(situation()), "variables": ["snap"]}, + call_id, + ) + + +def sol_submit(values, call_id="s1"): + outputs = { + k: {"value": v, "explanation": f"From the tool. value = {v}"} + for k, v in values.items() + } + return function_call("submit_outputs", {"outputs": outputs}, call_id) + + +def test_a_responses_conversation_chains_and_returns_each_call_output(): + model = FakeResponses( + [ + responses_response([sol_calculate("call_a")], "resp_1"), + responses_response( + [sol_submit({"snap": 2553, VARIABLES[1]: 0})], "resp_2", cached=800 + ), + ] + ) + spend = pilot.Spend(50.0) + record = pilot.run_household( + make_scenario(), + VARIABLES, + fake_tool, + model, + spend, + profile=pilot.model_profile(SOL), + ) + assert record["status"] == "complete" and record["model"] == SOL + assert record["predictions"] == {"snap": 2553.0, VARIABLES[1]: 0.0} + first, second = model.requests + assert first["model"] == SOL and first["tool_choice"] == "required" + assert first["input"].endswith(pilot.TOOL_NOTICE) + assert "previous_response_id" not in first + # The Responses API's flat tool shape. + assert [t["name"] for t in first["tools"]] == [pilot.TOOL_NAME, "submit_outputs"] + assert all(t["type"] == "function" for t in first["tools"]) + assert "max_output_tokens" in first and "max_completion_tokens" not in first + assert second["previous_response_id"] == "resp_1" + (output,) = second["input"] + assert output["type"] == "function_call_output" and output["call_id"] == "call_a" + assert json.loads(output["output"])["results"]["snap"]["values"] == {"spm": 2553.0} + # $2 / $10 per 1M; GPT cached input bills at a tenth of the input rate. + totals = record["totals"] + assert totals["cached_prompt_tokens"] == 800 + assert totals["cost_usd"] == pytest.approx( + (1000 * 2 + 200 * 10 + (200 * 2 + 800 * 0.2 + 200 * 10)) / 1e6 + ) + assert spend.spent_usd == pytest.approx(totals["cost_usd"]) + + +def test_a_responses_cap_forces_the_answer_function(): + cap = 2 + model = FakeResponses( + [responses_response([sol_calculate(f"c{i}")], f"r{i}") for i in range(cap)] + + [responses_response([sol_submit({"snap": 1, VARIABLES[1]: 2})], "rz")] + ) + record = pilot.run_household( + make_scenario(), + VARIABLES, + fake_tool, + model, + pilot.Spend(50.0), + turn_cap=cap, + profile=pilot.model_profile(SOL), + ) + assert record["status"] == "complete" and record["cap_hit"] + assert model.requests[cap]["tool_choice"] == { + "type": "function", + "name": "submit_outputs", + } + # The cap notice goes with the last call output, in the same request. + last = model.requests[cap]["input"] + assert [i.get("type") for i in last] == ["function_call_output", None] + assert "used all 2 calculation turns" in last[1]["content"] + + +def test_a_truncated_responses_turn_is_retried_at_the_ceiling(): + truncated = responses_response([], "r_cut", completion=16384) + truncated.status = "incomplete" + truncated.incomplete_details = SimpleNamespace(reason="max_output_tokens") + model = FakeResponses( + [truncated, responses_response([sol_submit({"snap": 1, VARIABLES[1]: 2})], "r")] + ) + record = pilot.run_household( + make_scenario(), + VARIABLES, + fake_tool, + model, + pilot.Spend(50.0), + profile=pilot.model_profile(SOL), + ) + assert record["status"] == "complete" + budgets = [r["max_output_tokens"] for r in model.requests] + assert budgets[0] < budgets[1] == 128_000 + # The retry resends the same turn: no response to chain from yet. + assert all("previous_response_id" not in r for r in model.requests) + assert record["totals"]["truncation_retries"] == 1 + + +def json_answer(values, fence=False): + text = json.dumps( + { + k: {"value": v, "explanation": f"From the tool. value = {v}"} + for k, v in values.items() + } + ) + return f"```json\n{text}\n```" if fence else text + + +def run_opus(model, **kwargs): + return pilot.run_household( + make_scenario(), + VARIABLES, + kwargs.pop("tool", fake_tool), + model, + kwargs.pop("spend", pilot.Spend(50.0)), + profile=pilot.model_profile(OPUS), + **kwargs, + ) + + +def test_a_json_contract_model_calculates_then_answers_in_json(): + model = FakeModel( + [ + response([calculate()]), + response([], content=json_answer({"snap": 2553, VARIABLES[1]: 0})), + ] + ) + spend = pilot.Spend(50.0) + record = run_opus(model, spend=spend) + assert record["status"] == "complete" and record["model"] == OPUS + assert record["predictions"] == {"snap": 2553.0, VARIABLES[1]: 0.0} + first, second = model.requests + # The API refuses a forced tool on this model: auto, and no answer tool. + assert [r["tool_choice"] for r in model.requests] == ["auto", "auto"] + assert [t["function"]["name"] for t in first["tools"]] == [pilot.TOOL_NAME] + assert first["model"] == "claude-opus-5-5" + prompt = first["messages"][0]["content"] + assert prompt[0]["text"].endswith(pilot.JSON_TOOL_NOTICE) + # One prompt-cache breakpoint per request, on its last message; the stored + # conversation stays unmarked. + assert prompt[0]["cache_control"] == {"type": "ephemeral"} + assert isinstance(second["messages"][0]["content"], str) + assert second["messages"][-1]["role"] == "tool" + assert second["messages"][-1]["cache_control"] == {"type": "ephemeral"} + assert not any("cache_control" in m for m in record["messages"]) + # $4 / $20 per 1M at 1,000 input and 200 output tokens a turn. + assert record["totals"]["cost_usd"] == pytest.approx(2 * (4000 + 4000) / 1e6) + + +def test_a_json_contract_cap_is_a_message_and_later_calls_do_not_run(): + ran = [] + + def counting_tool(args): + ran.append(args) + return fake_tool(args) + + cap = 2 + model = FakeModel( + [response([calculate(f"c{i}")]) for i in range(cap + 1)] + + [response([], content=json_answer({"snap": 1, VARIABLES[1]: 2}))] + ) + record = run_opus(model, tool=counting_tool, turn_cap=cap) + assert record["status"] == "complete" + assert len(ran) == cap and record["tool_turns"] == cap and record["cap_hit"] + cap_request = model.requests[cap] + assert ( + "used all 2 calculation turns" + in cap_request["messages"][-1]["content"][0]["text"] + ) + refused = model.requests[cap + 1]["messages"][-1] + assert refused["role"] == "tool" and "not_run" in refused["content"] + assert record["totals"]["tool_error_kinds"] == ["not_run"] + + +def test_a_json_contract_repair_asks_for_the_missing_output_in_json(): + model = FakeModel( + [ + response([], content=json_answer({"snap": 1}, fence=True)), + response([], content=json_answer({VARIABLES[1]: 2})), + ] + ) + record = run_opus(model) + assert record["status"] == "complete" and record["repair_rounds"] == 1 + assert record["predictions"] == {"snap": 1.0, VARIABLES[1]: 2.0} + repair = model.requests[1]["messages"][-1]["content"][0]["text"] + assert VARIABLES[1] in repair and "Reply with a JSON object" in repair + assert "submit_outputs" not in repair + + +def test_a_json_contract_reply_with_no_answer_is_nudged_not_repaired(): + model = FakeModel( + [ + response([], content="Let me work through this."), + response([], content=json_answer({"snap": 1, VARIABLES[1]: 2})), + ] + ) + record = run_opus(model) + assert record["status"] == "complete" and record["repair_rounds"] == 0 + nudge = model.requests[1]["messages"][-1]["content"][0]["text"] + assert "JSON object" in nudge and "calculate_household" in nudge + + +def test_records_name_one_model_and_old_records_are_argons(): + assert pilot.records_model({"a": {}, "b": {"model": pilot.MODEL_NAME}}) == ( + pilot.MODEL_NAME + ) + assert pilot.records_model({"a": {"model": SOL}}) == SOL + with pytest.raises(SystemExit): + pilot.records_model({"a": {"model": SOL}, "b": {"model": OPUS}}) + + +def test_arm_labels_follow_the_model(): + records = {"s": _record("s", ["snap"], {"snap": 1.0})} + assert set(pilot.arm2_predictions(records, "gpt-6-sol+pe-tool").model) == { + "gpt-6-sol+pe-tool" + } + frozen = pilot.arm1_predictions(label=SOL) + assert set(frozen.model) == {SOL} and frozen.scenario_id.nunique() == 100 + with pytest.raises(SystemExit): + pilot.arm1_predictions(label="no-such-model") + + +def test_every_pilot_model_has_a_provider_key(): + for name in (pilot.MODEL_NAME, SOL, OPUS): + assert pilot.provider_key(pilot.model_profile(name).model_id).endswith( + "_API_KEY" + ) + + +STEPS = st.lists( + st.one_of( + st.tuples(st.sampled_from(["calculate", "calculate_text"]), st.integers(1, 6)), + st.tuples( + st.sampled_from(["full", "partial", "prose", "empty", "truncated"]), + st.just(0), + ), + ), + max_size=16, +) + + +class ScriptedModel: + """Plays a script of turns in the model's own shape, then answers in full. + + ``calculate_text`` sends text beside its calls; ``truncated`` is a reply + cut off at the output budget, which the harness retries at the ceiling. + """ + + def __init__(self, profile, steps): + self.profile = profile + self.steps = list(steps) + self.requests = [] + self.responses = [] + + def __call__(self, **kwargs): + self.requests.append(json.loads(json.dumps(kwargs, default=str))) + kind, count = self.steps.pop(0) if self.steps else ("full", 0) + n = len(self.requests) + values = {"snap": 1, VARIABLES[1]: 2} + if kind == "partial": + values = {"snap": 1} + responses_api = self.profile.api == "responses" + if kind in {"calculate", "calculate_text", "truncated"}: + make = sol_calculate if responses_api else calculate + calls = [make(f"t{n}c{i}") for i in range(max(count, 1))] + content = "Checking with the tool." if kind == "calculate_text" else None + elif kind in {"full", "partial"} and self.profile.contract == "tool": + make = sol_submit if responses_api else submit + calls, content = [make(values, f"t{n}s")], None + elif kind in {"full", "partial"}: + calls, content = [], json_answer(values) + else: + calls, content = [], ("Thinking." if kind == "prose" else None) + if responses_api: + items = list(calls) + if content: + items.append( + SimpleNamespace( + type="message", + content=[SimpleNamespace(type="output_text", text=content)], + ) + ) + result = responses_response(items, f"resp_{n}") + if kind == "truncated": + result.status = "incomplete" + result.incomplete_details = SimpleNamespace(reason="max_output_tokens") + else: + result = response(calls, content=content) + if kind == "truncated": + result.choices[0].finish_reason = "length" + result.truncated = kind == "truncated" + self.responses.append(result) + return result + + +def _response_call_ids(profile, result) -> list[str]: + if profile.api == "responses": + return [i.call_id for i in result.output if i.type == "function_call"] + return [c.id for c in result.choices[0].message.tool_calls or []] + + +def _new_items(profile, request) -> list[tuple[str, str | None]]: + """What a request adds after the model's last turn, in order: + ("result", call id) for a function result, ("user", None) for user text.""" + if profile.api == "responses": + items = request["input"] + return [ + ("result", i["call_id"]) + if i.get("type") == "function_call_output" + else ("user", None) + for i in items + ] + messages = request["messages"] + last_assistant = max( + i for i, m in enumerate(messages) if m.get("role") == "assistant" + ) + return [ + ("result", m["tool_call_id"]) if m.get("role") == "tool" else ("user", None) + for m in messages[last_assistant + 1 :] + ] + + +@settings(max_examples=300, deadline=None, derandomize=True) +@given(st.sampled_from([pilot.MODEL_NAME, SOL, OPUS]), STEPS, st.integers(1, 4)) +def test_any_conversation_keeps_the_protocol_the_cap_and_the_spend(name, steps, cap): + profile = pilot.model_profile(name) + model = ScriptedModel(profile, steps) + ran = [] + + def counting_tool(args): + ran.append(args) + return fake_tool(args) + + spend = pilot.Spend(50.0) + record = pilot.run_household( + make_scenario(), + VARIABLES, + counting_tool, + model, + spend, + turn_cap=cap, + profile=profile, + ) + assert record["status"] in {"complete", "incomplete", "turn_limit"} + assert set(record["predictions"]) == set(VARIABLES) + # At most MAX_TOTAL_TURNS turns reach the conversation; a turn cut off + # below the ceiling costs one more request (intended), never more. + totals = record["totals"] + assert totals["model_turns"] <= pilot.MAX_TOTAL_TURNS + assert len(model.requests) == totals["model_turns"] + totals["truncation_retries"] + assert totals["truncation_retries"] <= totals["model_turns"] + # The cap: at most `cap` turns run calculations, each at most 4 of them. + assert record["tool_turns"] <= cap + assert len(ran) <= cap * pilot.MAX_CALCULATIONS_PER_TURN + # Spend: the shared counter saw exactly the household's turns. + assert spend.spent_usd == pytest.approx(record["totals"]["cost_usd"]) + # The protocol both APIs enforce: each function call of a response that + # reached the conversation gets exactly one result in the next request, + # before any user text, and a Responses request chains to that response. + # A response cut off below the budget ceiling never reaches it: the same + # request goes again at the ceiling. + from policybench.eval_no_tools import _completion_budget_ceiling + + ceiling = _completion_budget_ceiling(profile.model_id) + budget_key = ( + "max_output_tokens" if profile.api == "responses" else "max_completion_tokens" + ) + for i, (result, request) in enumerate(zip(model.responses, model.requests[1:])): + sent = model.requests[i] + if result.truncated and sent[budget_key] < ceiling: + assert request[budget_key] == ceiling + assert {k: v for k, v in request.items() if k != budget_key} == { + k: v for k, v in sent.items() if k != budget_key + } + continue + items = _new_items(profile, request) + results = [call_id for kind, call_id in items if kind == "result"] + assert sorted(results) == sorted(_response_call_ids(profile, result)) + kinds = [kind for kind, _ in items] + assert kinds == sorted(kinds), "user text came before a function result" + if profile.api == "responses": + assert request["previous_response_id"] == result.id + for request in model.requests: + names = [t.get("name") or t["function"]["name"] for t in request["tools"]] + if profile.contract == "json": + assert names == [pilot.TOOL_NAME] and request["tool_choice"] == "auto" + else: + assert names[0] == pilot.TOOL_NAME and names[1] == "submit_outputs" + if record["status"] == "complete": + assert all(v is not None for v in record["predictions"].values()) + + +# --------------------------------------------------------------------------- +# Settling unread outputs, and leave-one-out attribution. +# --------------------------------------------------------------------------- + +STATE_OUTPUT = "state_income_tax_before_refundable_credits" + + +def unread(calls, variables=(STATE_OUTPUT,)): + record = household(calls) + record["variables"] = list(variables) + return record + + +def test_a_state_variable_equal_to_the_boards_output_is_the_situations_miss(): + """Sol read ca_income_tax_before_refundable_credits; on its situation the + board's own output is the same number, so the inputs decide the miss.""" + call = call_record( + { + "ca_income_tax_before_refundable_credits": {"tu": 40483.32}, + "snap": {"spm": 0.0}, + }, + entity="tax_unit", + ) + record = unread([call], variables=(STATE_OUTPUT, "snap")) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 40483.32, 1) + assert verdict["category"] == "wrong_variable_or_period" + asked = [] + + def board_value_for(arguments): + asked.append(arguments) + return 40483.3 + + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 40483.32, + board_value_for, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_input_mapping" + assert settled["call_arguments"] == call["arguments"] == asked[0] + assert settled["tool_value"] == 40483.3 + assert "ca_income_tax_before_refundable_credits" in settled["detail"] + assert settled["answering_call"] == "holds the submitted value" + + +def test_a_variable_that_differs_from_the_boards_output_stays_a_wrong_variable(): + """Sol read state_income_tax (after refundable credits); its situation gives + the board's output exactly, so the situation was right and the read was not.""" + call = call_record({"state_income_tax": {"tu": 6363.34}}, entity="tax_unit") + record = unread([call]) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 6363.34, 1) + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 6363.34, + lambda a: 6828.34, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["board_output_on_situation"] == 6828.34 + + +def test_an_answer_no_call_returned_falls_back_to_the_main_call(): + """Opus summed two components itself; no single result holds its answer, + so the main call's situation is checked and the verdict stands.""" + parts = call_record( + { + "employee_social_security_tax": {"head": 3720.0}, + "employee_medicare_tax": {"head": 870.0}, + "snap": {"spm": 0.0}, + } + ) + record = unread([parts], variables=("payroll_tax", "snap")) + verdict = pilot.classify_miss(record, make_scenario(), "payroll_tax", 4590.0, 1) + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + "payroll_tax", + 4590.0, + lambda a: 4632.0, + output_entities={"tax_unit", "spm_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["answering_call"] == "main call" + assert settled["board_output_on_situation"] == 4632.0 + + +def test_other_verdicts_pass_through_unsettled(): + record = unread([call_record({STATE_OUTPUT: {"tu": 5.0}})]) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 5.0, 9.0) + assert verdict["category"] == "wrong_input_mapping" + + def never(arguments): + raise AssertionError("no engine call for a settled verdict") + + assert ( + pilot.resolve_unread_output( + verdict, record, make_scenario(), STATE_OUTPUT, 5.0, never + ) + is verdict + ) + + +def _diff(variable, model, reference, entity="person", unit="head"): + return { + "entity": entity, + "reference_unit": unit, + "model_unit": unit, + "variable": variable, + "model": model, + "reference": reference, + } + + +def test_leave_one_out_names_the_inputs_a_joint_fix_needs(monkeypatch): + """A toy output that needs two inputs together (as the 401(k) limit needs + the requested amounts and no post-limit override): no single input fixes + it, the full set does, and leave-one-out names the two that matter.""" + + def toy_value(system, situation, scenario, output): + head = situation["people"]["head"] + a = head.get("a", {}).get("2026", 0) + b = head.get("b", {}).get("2026", 0) + return 100.0 if (a, b) == (1, 1) else 0.0 + + monkeypatch.setattr(pilot, "output_value", toy_value) + model = {"people": {"head": {"c": {"2026": 5}}, "kid": {}}} + reference = {"people": {"head": {"a": {"2026": 1}, "b": {"2026": 1}}, "kid": {}}} + diffs = [ + _diff("a", {}, {"2026": 1}), + _diff("b", {}, {"2026": 1}), + _diff("c", {"2026": 5}, {}), + ] + result = pilot.attribute_miss(None, model, reference, None, "snap", 100.0, diffs) + assert result["single_causes"] == [] and result["all_differences_fix_it"] + assert result["necessary_inputs"] == ["person.a", "person.b"] + assert result["mapping_kind"] == "left_out" + + +def test_a_reviewed_re_entry_is_entered_differently_and_a_coincidence_is_not( + monkeypatch, +): + """Sol's scenario_018 on situation_diff's own diffs: the prompt's taxable + interest went into interest_income (a reviewed pair), and 50 weekly hours + happen to equal it. The pair makes it entered_differently; the hours are + only a candidate. Without the pair, the same miss is left_out.""" + + def toy_value(system, situation, scenario, output): + head = situation["people"]["head"] + return 0.0 if "taxable_interest_income" in head else 6.01 + 100 + + monkeypatch.setattr(pilot, "output_value", toy_value) + groups = {group: {} for group in pilot.GROUPS} + reference = { + "people": {"head": {"taxable_interest_income": {"2026": 50.0}}, "kid": {}}, + **groups, + } + model = { + "people": { + "head": { + "interest_income": {"2026": 50}, + "weekly_hours_worked": {"2026": 50}, + }, + "kid": {}, + }, + **groups, + } + diffs = pilot.situation_diff(model, reference, make_scenario()) + assert {d["variable"]: d["model"] for d in diffs}["taxable_interest_income"] == {} + result = pilot.attribute_miss(None, model, reference, None, "snap", 0.0, diffs) + assert result["single_causes"] == ["person.taxable_interest_income"] + assert result["mapping_kind"] == "entered_differently" + assert result["entered_as"] == ["person.interest_income"] + assert result["entered_as_candidates"] == ["person.weekly_hours_worked"] + + hours_only = { + "people": {"head": {"weekly_hours_worked": {"2026": 50}}, "kid": {}}, + **groups, + } + coincidence = pilot.attribute_miss( + None, + hours_only, + reference, + None, + "snap", + 0.0, + pilot.situation_diff(hours_only, reference, make_scenario()), + ) + assert coincidence["mapping_kind"] == "left_out" + assert coincidence["entered_as"] == [] + assert coincidence["entered_as_candidates"] == ["person.weekly_hours_worked"] + + +def test_an_amount_entered_under_another_input_is_entered_differently(monkeypatch): + """Sol's scenario_088: the reference's home_mortgage_interest is unset in + the model's situation, but the same $10,777 sits under mortgage_interest.""" + + def toy_value(system, situation, scenario, output): + # Outside the board's $1 tolerance either way. + return 0.0 if "home_mortgage_interest" in situation["people"]["head"] else 416.2 + + monkeypatch.setattr(pilot, "output_value", toy_value) + model = {"people": {"head": {"mortgage_interest": {"2026": 10777}}}} + reference = {"people": {"head": {"home_mortgage_interest": {"2026": 10777.0}}}} + diffs = [ + _diff("home_mortgage_interest", {}, {"2026": 10777.0}), + _diff("mortgage_interest", {"2026": 10777}, {}), + ] + result = pilot.attribute_miss(None, model, reference, None, "snap", 0.0, diffs) + assert result["single_causes"] == ["person.home_mortgage_interest"] + assert result["mapping_kind"] == "entered_differently" + assert result["entered_as"] == ["person.mortgage_interest"] + # The same cause with nothing like its amount elsewhere was left out, and a + # true flag elsewhere does not count as the amount. + unrelated = [ + _diff("home_mortgage_interest", {}, {"2026": 10777.0}), + _diff("is_disabled", {"2026": True}, {}), + _diff("rent", {"2026": 900}, {}), + ] + assert not pilot._entered_elsewhere(unrelated[0], unrelated) + assert pilot._entered_elsewhere(diffs[0], diffs) + other_person = [ + diffs[0], + {**diffs[1], "model_unit": "kid", "reference_unit": "kid"}, + ] + assert not pilot._entered_elsewhere(other_person[0], other_person) + + +def test_a_coincidental_probe_on_another_entity_is_not_evidence(): + """r2's case: the answer copies after-credit tax of $200 from the model's + situation, whose board output is $300; a later probe returns an + employment_income of $200 on a person. The probe does not count, so the + verdict stays a wrong variable on the real call.""" + real = call_record( + {"state_income_tax": {"tu": 200.0}, "snap": {"spm": 0.0}}, entity="tax_unit" + ) + probe = call_record({"employment_income": {"head": 200.0}}, entity="person", age=41) + record = unread([real, probe], variables=(STATE_OUTPUT, "snap")) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 200.0, 300) + board = {real["arguments"]: 300.0, probe["arguments"]: 200.0} + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 200.0, + board.get, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["call_arguments"] == real["arguments"] + assert settled["board_output_on_situation"] == 300.0 + + +def test_holding_calls_that_disagree_leave_the_verdict_marked_ambiguous(): + """Two tax-unit results both hold the answer, but only one situation's + board output equals it: no reclassification on evidence that splits.""" + first = call_record( + {"state_income_tax": {"tu": 200.0}, "snap": {"spm": 0.0}}, entity="tax_unit" + ) + second = call_record({"xx_credit": {"tu": 200.0}}, entity="tax_unit", age=41) + record = unread([first, second], variables=(STATE_OUTPUT, "snap")) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 200.0, 300) + board = {first["arguments"]: 300.0, second["arguments"]: 200.0} + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 200.0, + board.get, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["detail"].startswith("ambiguous") + assert settled["holding_calls"] == 2 + # The evidence reported is the main situation's, not the probe's (r3). + assert settled["call_arguments"] == first["arguments"] + assert settled["board_output_on_situation"] == 300.0 + assert settled["holding_board_values"] == [300.0, 200.0] + + +def test_agreeing_holding_calls_attribute_the_fullest_one(): + """Every holding call agrees; the situation attributed is the one that + computed the most of the household's outputs, not a later narrow probe.""" + full = call_record( + {"ca_income_tax": {"tu": 500.0}, "snap": {"spm": 0.0}}, + entity="tax_unit", + ) + probe = call_record({"ca_income_tax": {"tu": 500.0}}, entity="tax_unit", age=41) + record = unread([full, probe], variables=(STATE_OUTPUT, "snap")) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 500.0, 900) + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 500.0, + lambda a: 500.0, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_input_mapping" + assert settled["call_arguments"] == full["arguments"] + assert settled["holding_calls"] == 2 + + +def test_a_binary_output_is_never_reclassified(): + # The call computes SNAP, so it is the main one; r5's construction. + call = call_record( + {"is_wic_eligible": {"head": True, "kid": False}, "snap": {"spm": 0.0}}, + entity="person", + ) + record = unread([call], variables=("head_medicaid_eligible", "snap")) + verdict = pilot.classify_miss( + record, make_scenario(), "head_medicaid_eligible", 0.0, 1.0 + ) + assert verdict["category"] == "wrong_variable_or_period" + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + "head_medicaid_eligible", + 0.0, + lambda a: 0.0, + output_entities={"person"}, + binary=True, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["answering_call"] == "main call" + # Binary outputs skip the holding check, so no note may claim that no + # call holds the answer (r5): a 0 sits in many results. + assert "no call's results hold it" not in settled.get("detail", "") + + +def test_a_lone_probe_cannot_outvote_the_main_situation_for_a_summed_answer(): + """r3's case: Opus summed $3,720 and $870 into $4,590 on its main + situation, whose board payroll is $4,632. A later tax-unit probe, which + computes fewer outputs, returns $4,590 on a situation whose board payroll + is also $4,590. The probe is the only holding call, but the main + situation disagrees, so the verdict stands on the main call.""" + parts = call_record( + { + "employee_social_security_tax": {"tu": 3720.0}, + "employee_medicare_tax": {"tu": 870.0}, + "snap": {"spm": 0.0}, + "ssi": {"head": 0.0}, + }, + entity="tax_unit", + ) + probe = call_record({"xx_tax": {"tu": 4590.0}}, entity="tax_unit", age=41) + record = unread([parts, probe], variables=("payroll_tax", "snap", "ssi")) + verdict = pilot.classify_miss(record, make_scenario(), "payroll_tax", 4590.0, 4632) + assert verdict["category"] == "wrong_variable_or_period" + board = {parts["arguments"]: 4632.0, probe["arguments"]: 4590.0} + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + "payroll_tax", + 4590.0, + board.get, + output_entities={"tax_unit", "spm_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["call_arguments"] == parts["arguments"] + assert settled["board_output_on_situation"] == 4632.0 + assert settled["holding_board_values"] == [4590.0] + assert settled["detail"].startswith("ambiguous") + + +def test_no_main_situation_when_every_call_ties_at_zero(): + """r4's case: with payroll the only output and no call computing it, every + call ties at zero and the tie rule would make the last probe the main call, + vouching for itself. No situation is the main one, so nothing flips.""" + parts = call_record( + { + "employee_social_security_tax": {"tu": 3720.0}, + "employee_medicare_tax": {"tu": 870.0}, + }, + entity="tax_unit", + ) + probe = call_record({"xx_tax": {"tu": 4590.0}}, entity="tax_unit", age=41) + record = unread([parts, probe], variables=("payroll_tax",)) + verdict = pilot.classify_miss(record, make_scenario(), "payroll_tax", 4590.0, 4632) + assert verdict["category"] == "wrong_variable_or_period" + board = {parts["arguments"]: 4632.0, probe["arguments"]: 4590.0} + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + "payroll_tax", + 4590.0, + board.get, + output_entities={"tax_unit", "spm_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["detail"].startswith("no call computed any") + assert pilot.main_call_and_count(record, make_scenario(), ["payroll_tax"])[1] == 0 + + +def test_a_main_situation_equal_to_an_answer_no_call_holds_is_noted(): + """An answer the model built itself, on a main situation whose board output + equals it: strong evidence of a mapping miss, noted but not reclassified.""" + parts = call_record( + { + "employee_social_security_tax": {"tu": 3720.0}, + "employee_medicare_tax": {"tu": 870.0}, + "snap": {"spm": 0.0}, + }, + entity="tax_unit", + ) + record = unread([parts], variables=("payroll_tax", "snap")) + verdict = pilot.classify_miss(record, make_scenario(), "payroll_tax", 4590.0, 4632) + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + "payroll_tax", + 4590.0, + lambda a: 4590.0, + output_entities={"tax_unit", "spm_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert "no call's results hold it" in settled["detail"] + + +def test_an_uncomputable_main_situation_is_noted_not_called_a_disagreement(): + main = call_record( + {"ca_income_tax": {"tu": 500.0}, "snap": {"spm": 0.0}}, entity="tax_unit" + ) + record = unread([main], variables=(STATE_OUTPUT, "snap")) + verdict = pilot.classify_miss(record, make_scenario(), STATE_OUTPUT, 500.0, 900) + settled = pilot.resolve_unread_output( + verdict, + record, + make_scenario(), + STATE_OUTPUT, + 500.0, + lambda a: None, + output_entities={"tax_unit"}, + ) + assert settled["category"] == "wrong_variable_or_period" + assert settled["detail"].startswith("the main situation could not be computed")