wehub-resource-sync
1260 lines
53 KiB
Python
1260 lines
53 KiB
Python
"""Tests for the pure RPC step→item mapper.
|
|
|
|
These exercise :func:`omnigent.antigravity_native_steps.map_step_to_events`
|
|
using the real recorded fixtures captured from live agy sessions (Task 1).
|
|
No I/O, no live agy: the mapper is driven with fixture dicts and event shapes
|
|
are asserted exactly.
|
|
|
|
Key assertions:
|
|
- PLANNER_RESPONSE with text → exactly one ``external_conversation_item``
|
|
``message`` (role assistant, ``output_text`` content). NO
|
|
``external_output_text_delta`` / ``output_text_delta`` event.
|
|
- USER_INPUT → ``[]`` (skipped — fixes user-dup).
|
|
- PLANNER_RESPONSE with tool_calls → ``function_call`` item(s) via allocator.
|
|
- RUN_COMMAND DONE → ``function_call_output`` carrying
|
|
``runCommand.combinedOutput.full``.
|
|
- RUN_COMMAND WAITING → ``function_call`` only (no output yet).
|
|
- ASK_QUESTION WAITING → ``function_call`` only (no output yet).
|
|
- ASK_QUESTION DONE → ``function_call_output`` carrying the formatted answer.
|
|
- CHECKPOINT / CONVERSATION_HISTORY → ``[]``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
from omnigent.antigravity_native_steps import (
|
|
OutboundEvent,
|
|
_execution_discriminator,
|
|
_ToolCallIdAllocator,
|
|
map_step_to_events,
|
|
output_reasoning_delta_event,
|
|
pending_interaction,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FIXTURES = Path(__file__).parent / "fixtures" / "antigravity" / "steps"
|
|
_CID = "test-conversation-id"
|
|
|
|
|
|
def _load(name: str) -> dict[str, Any]:
|
|
"""Load one step fixture by filename (without extension)."""
|
|
path = _FIXTURES / f"{name}.json"
|
|
return cast(dict[str, Any], json.loads(path.read_text()))
|
|
|
|
|
|
def _allocator() -> _ToolCallIdAllocator:
|
|
"""Fresh allocator for each test."""
|
|
return _ToolCallIdAllocator(conversation_id=_CID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: assert no delta event at all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _assert_no_delta(events: list[OutboundEvent]) -> None:
|
|
"""
|
|
Assert that none of the events are delta events.
|
|
|
|
The double-render fix requires that map_step_to_events emits NO
|
|
``external_output_text_delta`` events whatsoever — the old forwarder emitted
|
|
one delta per assistant text step; the new mapper drops it entirely.
|
|
"""
|
|
for event in events:
|
|
assert event.event_type != "external_output_text_delta", (
|
|
f"Unexpected delta event in output: {event}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# USER_INPUT → committed user message (#1155)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUserInputCommitted:
|
|
"""
|
|
USER_INPUT steps commit the user's turn as a ``message`` item.
|
|
|
|
The pure-RPC write path fires no "direct POST /events" to persist the user
|
|
turn, so the read path must mirror it (parity with claude/codex/cursor
|
|
native). Without it the web UI's optimistic bubble has no committed
|
|
counterpart and renders below the assistant reply (#1155).
|
|
"""
|
|
|
|
def test_user_input_commits_user_message(self) -> None:
|
|
"""USER_INPUT step → exactly one committed user ``message`` item."""
|
|
step = _load("user_input")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
ev = events[0]
|
|
assert ev.event_type == "external_conversation_item"
|
|
assert ev.data["item_type"] == "message"
|
|
assert ev.data["item_data"]["role"] == "user"
|
|
assert ev.data["item_data"]["content"] == [
|
|
{"type": "input_text", "text": "Say hello in one short sentence."}
|
|
]
|
|
# A user turn carries no response_id (not an assistant response).
|
|
assert "response_id" not in ev.data
|
|
|
|
def test_user_input_no_delta(self) -> None:
|
|
"""USER_INPUT commits a message but emits no streaming delta events."""
|
|
step = _load("user_input")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
_assert_no_delta(events)
|
|
|
|
def test_user_input_without_text_is_skipped(self) -> None:
|
|
"""A USER_INPUT step with no recoverable text emits nothing (no empty bubble)."""
|
|
step = {"type": "CORTEX_STEP_TYPE_USER_INPUT", "userInput": {}}
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PLANNER_RESPONSE (text only) → one message, NO delta
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPlannerResponseText:
|
|
"""PLANNER_RESPONSE with assistant text → exactly one message item, no delta."""
|
|
|
|
def test_returns_exactly_one_event(self) -> None:
|
|
"""
|
|
A text-only PLANNER_RESPONSE yields exactly one event.
|
|
|
|
No delta means no second event; the old forwarder emitted 2 (delta +
|
|
message); the new mapper emits 1.
|
|
"""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
|
|
def test_event_type_is_conversation_item(self) -> None:
|
|
"""The single event has type ``external_conversation_item``."""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].event_type == "external_conversation_item"
|
|
|
|
def test_item_type_is_message(self) -> None:
|
|
"""The event's ``item_type`` is ``"message"``."""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].data["item_type"] == "message"
|
|
|
|
def test_message_role_is_assistant(self) -> None:
|
|
"""The ``message`` item has role ``"assistant"``."""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["role"] == "assistant"
|
|
|
|
def test_message_content_is_output_text(self) -> None:
|
|
"""
|
|
Content list contains exactly one ``output_text`` block with the
|
|
fixture's ``plannerResponse.response`` text.
|
|
"""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
content = item_data["content"]
|
|
assert isinstance(content, list)
|
|
assert len(content) == 1
|
|
assert content[0]["type"] == "output_text"
|
|
expected_text = (
|
|
"Hello! I am Antigravity, your AI coding assistant, ready to help you with your tasks."
|
|
)
|
|
assert content[0]["text"] == expected_text
|
|
|
|
def test_no_delta_event(self) -> None:
|
|
"""
|
|
No ``external_output_text_delta`` event is emitted.
|
|
|
|
This is the primary double-render fix: the old forwarder emitted a delta
|
|
event before the message; the new mapper drops it entirely.
|
|
"""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
_assert_no_delta(events)
|
|
|
|
def test_step_index_from_fixture(self) -> None:
|
|
"""step_index on the event matches the fixture's sourceTrajectoryStepInfo.stepIndex."""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
# planner_response_text.json has stepIndex=2
|
|
assert events[0].step_index == 2
|
|
|
|
def test_response_id_stable(self) -> None:
|
|
"""response_id is deterministic: ``agy_<conversation_id>_<stepIndex>``."""
|
|
step = _load("planner_response_text")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].data["response_id"] == f"agy_{_CID}_2"
|
|
|
|
|
|
class TestPlannerResponseError:
|
|
"""An ERROR PLANNER_RESPONSE emits a visible error item (not a silent drop)."""
|
|
|
|
def test_error_planner_emits_one_error_message(self) -> None:
|
|
"""A terminal-ERROR planner yields one assistant ``message`` error marker.
|
|
|
|
Regression for #6: previously an ERROR planner returned ``[]`` (the branch
|
|
committed only at DONE), so a model/turn error looked identical to a normal
|
|
empty reply. It must surface a visible item.
|
|
"""
|
|
step = _load("planner_response_text")
|
|
step["status"] = "CORTEX_STEP_STATUS_ERROR"
|
|
step["plannerResponse"] = {} # no text/error detail -> generic marker
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
ev = events[0]
|
|
assert ev.event_type == "external_conversation_item"
|
|
assert ev.data["item_type"] == "message"
|
|
item = ev.data["item_data"]
|
|
assert isinstance(item, dict)
|
|
assert item["role"] == "assistant"
|
|
text = item["content"][0]["text"]
|
|
assert "status ERROR" in text
|
|
|
|
def test_error_planner_includes_error_detail_when_present(self) -> None:
|
|
"""Any ``plannerResponse.error`` text is folded into the marker."""
|
|
step = _load("planner_response_text")
|
|
step["status"] = "CORTEX_STEP_STATUS_ERROR"
|
|
step["plannerResponse"] = {"error": "model overloaded (503)"}
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert "model overloaded (503)" in events[0].data["item_data"]["content"][0]["text"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PLANNER_RESPONSE (tool_calls) → function_call events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPlannerResponseToolCallRunCommand:
|
|
"""PLANNER_RESPONSE with run_command tool call → function_call event(s)."""
|
|
|
|
def test_returns_one_function_call(self) -> None:
|
|
"""One tool call → one ``function_call`` event."""
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call"
|
|
|
|
def test_function_call_name(self) -> None:
|
|
"""The function_call name matches the fixture's toolCall name."""
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["name"] == "run_command"
|
|
|
|
def test_function_call_id_is_real_agy_id(self) -> None:
|
|
"""
|
|
call_id is the real agy-assigned id from plannerResponse.toolCalls[].id.
|
|
|
|
The fixture carries id="cbawg2v8"; the mapper must use that directly,
|
|
NOT synthesize a positional id from the allocator.
|
|
"""
|
|
alloc = _allocator()
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=alloc)
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
# Real agy id from the fixture
|
|
assert item_data["call_id"] == "cbawg2v8"
|
|
# Allocator must NOT have been advanced (real id was used instead)
|
|
assert alloc.invocation_count == 0
|
|
|
|
def test_function_call_arguments_strip_display_keys(self) -> None:
|
|
"""
|
|
``toolAction`` and ``toolSummary`` are stripped from the function
|
|
arguments; the real command args remain.
|
|
"""
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
args_text = item_data["arguments"]
|
|
assert isinstance(args_text, str)
|
|
args = json.loads(args_text)
|
|
assert "toolAction" not in args
|
|
assert "toolSummary" not in args
|
|
# Real args remain
|
|
assert "CommandLine" in args
|
|
|
|
def test_no_delta_event(self) -> None:
|
|
"""No delta event is emitted for a tool-call-only PLANNER_RESPONSE."""
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
_assert_no_delta(events)
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches fixture stepIndex=5."""
|
|
step = _load("planner_response_tool_call_run_command")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].step_index == 5
|
|
|
|
|
|
class TestPlannerResponseToolCallAskQuestion:
|
|
"""PLANNER_RESPONSE with ask_question tool call → function_call event."""
|
|
|
|
def test_returns_one_function_call(self) -> None:
|
|
"""One ask_question tool call → one ``function_call`` event."""
|
|
step = _load("planner_response_tool_call_ask_question")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call"
|
|
|
|
def test_function_call_name(self) -> None:
|
|
"""The function_call name is ``ask_question``."""
|
|
step = _load("planner_response_tool_call_ask_question")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["name"] == "ask_question"
|
|
|
|
def test_function_call_id_is_real_agy_id(self) -> None:
|
|
"""
|
|
call_id is the real agy-assigned id from plannerResponse.toolCalls[].id.
|
|
|
|
The fixture carries id="jfizoalt"; the allocator must NOT advance.
|
|
"""
|
|
alloc = _allocator()
|
|
step = _load("planner_response_tool_call_ask_question")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=alloc)
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["call_id"] == "jfizoalt"
|
|
assert alloc.invocation_count == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RUN_COMMAND DONE → function_call_output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunCommandDone:
|
|
"""RUN_COMMAND DONE step → ``function_call_output`` with combinedOutput."""
|
|
|
|
def test_returns_one_event(self) -> None:
|
|
"""One DONE run_command → one event."""
|
|
step = _load("run_command_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
|
|
def test_event_type_is_conversation_item(self) -> None:
|
|
"""event_type is ``external_conversation_item``."""
|
|
step = _load("run_command_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].event_type == "external_conversation_item"
|
|
|
|
def test_item_type_is_function_call_output(self) -> None:
|
|
"""item_type is ``function_call_output``."""
|
|
step = _load("run_command_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
|
|
def test_output_from_combined_output_full(self) -> None:
|
|
"""
|
|
The output text comes from ``runCommand.combinedOutput.full``.
|
|
|
|
The fixture has ``combinedOutput.full = '/Users/bryanli/...scratch\\n'``.
|
|
"""
|
|
step = _load("run_command_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["output"] == "/Users/bryanli/.gemini/antigravity-cli/scratch\n"
|
|
|
|
def test_call_id_is_real_agy_id(self) -> None:
|
|
"""
|
|
call_id is the real agy-assigned id from metadata.toolCall.id.
|
|
|
|
The fixture carries toolCall.id="cbawg2v8", matching the invocation
|
|
step's plannerResponse.toolCalls[0].id. The allocator must NOT be
|
|
consulted (no pending ids needed).
|
|
"""
|
|
step = _load("run_command_done")
|
|
alloc = _allocator()
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=alloc)
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
assert item_data["call_id"] == "cbawg2v8"
|
|
# Allocator was not used (no orphan id minted)
|
|
assert alloc.orphan_output_count == 0
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches fixture stepIndex=6."""
|
|
step = _load("run_command_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].step_index == 6
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RUN_COMMAND WAITING → function_call only (no output yet)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunCommandWaiting:
|
|
"""
|
|
RUN_COMMAND WAITING step → no function_call_output.
|
|
|
|
The command has been proposed but not yet approved/executed. The mapper must
|
|
NOT emit a ``function_call_output`` (no result exists). Task 5 extracts the
|
|
pending interaction for the bridge.
|
|
"""
|
|
|
|
def test_waiting_emits_no_output_event(self) -> None:
|
|
"""WAITING run_command → empty list (no function_call_output)."""
|
|
step = _load("run_command_waiting")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
def test_waiting_no_delta(self) -> None:
|
|
"""No delta event from a WAITING run_command."""
|
|
step = _load("run_command_waiting")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
_assert_no_delta(events)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RUN_COMMAND ERROR → error-marker output (close the dangling call)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunCommandError:
|
|
"""A terminal-ERROR tool step must still close its ``function_call``.
|
|
|
|
The invocation side emits a ``function_call`` for the tool unconditionally,
|
|
so an ERROR result (e.g. an ignored/timed-out interactive prompt that flips
|
|
WAITING→ERROR) must emit a paired ``function_call_output`` keyed on the same
|
|
id, or the web UI strands a perpetual in-progress tool card.
|
|
"""
|
|
|
|
def test_error_emits_one_output_event(self) -> None:
|
|
"""A failed (ERROR-status) run_command emits one function_call_output."""
|
|
step = _load("run_command_error")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
|
|
def test_error_output_keyed_on_real_id(self) -> None:
|
|
"""The error output pairs by the real agy id (matches the invocation)."""
|
|
step = _load("run_command_error")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = cast(dict[str, Any], events[0].data["item_data"])
|
|
# Fixture carries toolCall.id="cbawg2v8" — the same id the planner
|
|
# invocation emits, so the pair correlates.
|
|
assert item_data["call_id"] == "cbawg2v8"
|
|
|
|
def test_error_output_text_is_nonempty_marker(self) -> None:
|
|
"""The output is a non-empty error marker mentioning the ERROR status."""
|
|
step = _load("run_command_error")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = cast(dict[str, Any], events[0].data["item_data"])
|
|
output = item_data["output"]
|
|
assert isinstance(output, str) and output
|
|
assert "ERROR" in output
|
|
|
|
def test_error_step_index(self) -> None:
|
|
"""step_index matches fixture stepIndex=6."""
|
|
step = _load("run_command_error")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].step_index == 6
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool-result closure: empty output + unmapped result types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestToolResultClosure:
|
|
"""Every tool call is closed, even with empty or unmapped results.
|
|
|
|
Regression coverage for dangling ``function_call``s: a successful command
|
|
whose output proto3-omits empty ``combinedOutput.full``, and a result step
|
|
of a type the mapper has no extractor for (e.g. VIEW_FILE / CODE_ACTION),
|
|
must each still emit a ``function_call_output`` keyed on the real
|
|
``metadata.toolCall.id`` so the web UI's tool card resolves.
|
|
"""
|
|
|
|
def test_done_run_command_empty_output_still_closes(self) -> None:
|
|
"""DONE run_command with no combinedOutput → one event, empty output."""
|
|
step = _load("run_command_done")
|
|
# Proto3 omits empty scalars: drop combinedOutput to simulate a
|
|
# ``cd`` / ``mkdir`` / redirect that produced no captured output.
|
|
step["runCommand"].pop("combinedOutput", None)
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
item_data = cast(dict[str, Any], events[0].data["item_data"])
|
|
assert item_data["call_id"] == "cbawg2v8"
|
|
assert item_data["output"] == ""
|
|
|
|
def test_unmapped_tool_result_type_with_id_closes(self) -> None:
|
|
"""A result type with no extractor but a toolCall.id still closes the call."""
|
|
step = _load("run_command_done")
|
|
# Re-label as a result type the mapper has no extractor for; keep the
|
|
# real toolCall.id so the pair still correlates.
|
|
step["type"] = "CORTEX_STEP_TYPE_VIEW_FILE"
|
|
step.pop("runCommand", None)
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
item_data = cast(dict[str, Any], events[0].data["item_data"])
|
|
assert item_data["call_id"] == "cbawg2v8"
|
|
assert item_data["output"] == ""
|
|
|
|
def test_system_step_without_tool_id_is_skipped(self) -> None:
|
|
"""A non-tool step with no toolCall.id is NOT treated as a tool result."""
|
|
step = _load("checkpoint")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LIST_DIRECTORY DONE → function_call_output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListDirectoryDone:
|
|
"""LIST_DIRECTORY DONE step → ``function_call_output``."""
|
|
|
|
def test_returns_one_function_call_output(self) -> None:
|
|
"""DONE list_directory → one function_call_output event."""
|
|
step = _load("list_directory_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
|
|
def test_call_id_is_real_agy_id(self) -> None:
|
|
"""call_id is the real agy-assigned id from metadata.toolCall.id."""
|
|
step = _load("list_directory_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
# Fixture carries toolCall.id="h510vxi0"
|
|
assert item_data["call_id"] == "h510vxi0"
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches fixture stepIndex=10."""
|
|
step = _load("list_directory_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].step_index == 10
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ASK_QUESTION WAITING → no output (pending interaction)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAskQuestionWaiting:
|
|
"""
|
|
ASK_QUESTION WAITING → no function_call_output.
|
|
|
|
The question is awaiting user response. Key on ``status`` NOT on the
|
|
presence of ``requestedInteraction`` (which persists in the DONE fixture).
|
|
"""
|
|
|
|
def test_waiting_emits_no_event(self) -> None:
|
|
"""WAITING ask_question → empty list."""
|
|
step = _load("ask_question_waiting")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ASK_QUESTION DONE → function_call_output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAskQuestionDone:
|
|
"""ASK_QUESTION DONE step → function_call_output."""
|
|
|
|
def test_returns_function_call_output(self) -> None:
|
|
"""DONE ask_question → one function_call_output event."""
|
|
step = _load("ask_question_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].data["item_type"] == "function_call_output"
|
|
|
|
def test_call_id_is_real_agy_id(self) -> None:
|
|
"""call_id is the real agy-assigned id from metadata.toolCall.id."""
|
|
step = _load("ask_question_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
# Fixture carries toolCall.id="jfizoalt", matching the planner invocation
|
|
assert item_data["call_id"] == "jfizoalt"
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches fixture stepIndex=12."""
|
|
step = _load("ask_question_done")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events[0].step_index == 12
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CHECKPOINT and CONVERSATION_HISTORY → []
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSystemStepsSkipped:
|
|
"""CHECKPOINT and CONVERSATION_HISTORY system steps produce no events."""
|
|
|
|
def test_checkpoint_returns_empty(self) -> None:
|
|
"""CHECKPOINT step → ``[]``."""
|
|
step = _load("checkpoint")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
def test_conversation_history_returns_empty(self) -> None:
|
|
"""CONVERSATION_HISTORY step → ``[]``."""
|
|
step = _load("conversation_history")
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Slot-0 step index: absent stepIndex → treated as 0 (CDX-IMP1 + OPUS-MIN1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSlotZeroStepIndex:
|
|
"""
|
|
A PLANNER_RESPONSE step at slot 0 has no ``stepIndex`` in the proto
|
|
(proto omits zero-valued scalars). The mapper must treat it as index 0,
|
|
not silently drop the step.
|
|
"""
|
|
|
|
def test_absent_step_index_emits_event_at_zero(self) -> None:
|
|
"""Slot-0 PLANNER_RESPONSE (no stepIndex) → message event with step_index=0."""
|
|
import copy
|
|
|
|
base = _load("planner_response_text")
|
|
step = copy.deepcopy(base)
|
|
# Strip stepIndex to simulate proto-default omission
|
|
traj_info = step["metadata"]["sourceTrajectoryStepInfo"]
|
|
assert isinstance(traj_info, dict)
|
|
traj_info.pop("stepIndex", None)
|
|
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].step_index == 0
|
|
|
|
def test_string_encoded_step_index_accepted(self) -> None:
|
|
"""String-encoded stepIndex (e.g. ``'2'``) is accepted and parsed."""
|
|
import copy
|
|
|
|
base = _load("planner_response_text")
|
|
step = copy.deepcopy(base)
|
|
traj_info = step["metadata"]["sourceTrajectoryStepInfo"]
|
|
assert isinstance(traj_info, dict)
|
|
traj_info["stepIndex"] = "2" # String form
|
|
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
assert events[0].step_index == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# modifiedResponse precedence (OPUS-MIN2 / Task4-M1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestModifiedResponsePrecedence:
|
|
"""
|
|
``plannerResponse.modifiedResponse`` takes precedence over ``response``
|
|
when both are present and non-empty. Both fields appear in live fixtures
|
|
and are equal when no moderation has occurred. The preference is tested
|
|
with a synthetic step where they differ so the behavior is pinned.
|
|
"""
|
|
|
|
def test_modified_response_wins_when_different(self) -> None:
|
|
"""
|
|
When modifiedResponse differs from response, modifiedResponse is used.
|
|
|
|
The original fixture has both equal; this synthetic variant sets them
|
|
to different values to verify the precedence rule explicitly.
|
|
"""
|
|
import copy
|
|
|
|
base = _load("planner_response_text")
|
|
step = copy.deepcopy(base)
|
|
planner = step.get("plannerResponse")
|
|
assert isinstance(planner, dict)
|
|
planner["response"] = "Original text."
|
|
planner["modifiedResponse"] = "Post-moderation text."
|
|
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
content = item_data["content"]
|
|
assert isinstance(content, list)
|
|
# modifiedResponse wins
|
|
assert content[0]["text"] == "Post-moderation text."
|
|
|
|
def test_response_used_when_modified_absent(self) -> None:
|
|
"""When modifiedResponse is absent, response is used as fallback."""
|
|
import copy
|
|
|
|
base = _load("planner_response_text")
|
|
step = copy.deepcopy(base)
|
|
planner = step.get("plannerResponse")
|
|
assert isinstance(planner, dict)
|
|
planner.pop("modifiedResponse", None)
|
|
planner["response"] = "Fallback text."
|
|
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
content = item_data["content"]
|
|
assert isinstance(content, list)
|
|
assert content[0]["text"] == "Fallback text."
|
|
|
|
def test_response_used_when_modified_empty(self) -> None:
|
|
"""When modifiedResponse is empty string, response is used as fallback."""
|
|
import copy
|
|
|
|
base = _load("planner_response_text")
|
|
step = copy.deepcopy(base)
|
|
planner = step.get("plannerResponse")
|
|
assert isinstance(planner, dict)
|
|
planner["modifiedResponse"] = ""
|
|
planner["response"] = "Non-empty response."
|
|
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
assert len(events) == 1
|
|
item_data = events[0].data["item_data"]
|
|
assert isinstance(item_data, dict)
|
|
content = item_data["content"]
|
|
assert isinstance(content, list)
|
|
assert content[0]["text"] == "Non-empty response."
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Real-id pairing: planner → run_command sequence (OPUS-IMP1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRealIdPairing:
|
|
"""
|
|
Verify that real agy tool-call ids are used for invocation↔output pairing.
|
|
|
|
The RPC carries the same id on both the invocation
|
|
(``plannerResponse.toolCalls[].id``) and the result
|
|
(``metadata.toolCall.id``). The mapper uses those ids directly —
|
|
no FIFO position, no allocator — so pairing is order-independent.
|
|
"""
|
|
|
|
def test_planner_then_run_command_done_share_real_id(self) -> None:
|
|
"""
|
|
PLANNER_RESPONSE and RUN_COMMAND DONE → both carry the same real agy id.
|
|
|
|
The invocation call_id and the output call_id must equal the fixture's
|
|
agy-assigned id ("cbawg2v8"), not a positional allocator id.
|
|
"""
|
|
alloc = _allocator()
|
|
planner_step = _load("planner_response_tool_call_run_command")
|
|
planner_events = map_step_to_events(planner_step, conversation_id=_CID, allocator=alloc)
|
|
assert len(planner_events) == 1
|
|
planner_item_data = planner_events[0].data["item_data"]
|
|
assert isinstance(planner_item_data, dict)
|
|
invocation_call_id = planner_item_data["call_id"]
|
|
assert invocation_call_id == "cbawg2v8"
|
|
|
|
result_step = _load("run_command_done")
|
|
result_events = map_step_to_events(result_step, conversation_id=_CID, allocator=alloc)
|
|
assert len(result_events) == 1
|
|
result_item = result_events[0].data["item_data"]
|
|
assert isinstance(result_item, dict)
|
|
# Same real id — not a FIFO-synthesized orphan id
|
|
assert result_item["call_id"] == invocation_call_id
|
|
|
|
def test_two_results_out_of_order_pair_by_real_id(self) -> None:
|
|
"""
|
|
REGRESSION: two tool-result steps with DIFFERENT real ids delivered
|
|
out of order each pair to the correct invocation by real id.
|
|
|
|
FIFO would mis-pair: if result-B arrives before result-A, FIFO gives
|
|
result-B the call_id of invocation-A and result-A gets invocation-B's
|
|
id. Real-id pairing is immune to arrival order.
|
|
|
|
We simulate two consecutive PLANNER_RESPONSE steps each invoking a
|
|
different tool (run_command id="cbawg2v8", ask_question id="jfizoalt")
|
|
then deliver their result steps OUT OF ORDER (ask_question result first,
|
|
run_command result second).
|
|
"""
|
|
|
|
alloc = _allocator()
|
|
|
|
# Emit PLANNER_RESPONSE for run_command (id="cbawg2v8")
|
|
rc_planner = _load("planner_response_tool_call_run_command")
|
|
rc_planner_events = map_step_to_events(rc_planner, conversation_id=_CID, allocator=alloc)
|
|
assert len(rc_planner_events) == 1
|
|
rc_item = rc_planner_events[0].data["item_data"]
|
|
assert isinstance(rc_item, dict)
|
|
assert rc_item["call_id"] == "cbawg2v8"
|
|
|
|
# Emit PLANNER_RESPONSE for ask_question (id="jfizoalt")
|
|
aq_planner = _load("planner_response_tool_call_ask_question")
|
|
aq_planner_events = map_step_to_events(aq_planner, conversation_id=_CID, allocator=alloc)
|
|
assert len(aq_planner_events) == 1
|
|
aq_item = aq_planner_events[0].data["item_data"]
|
|
assert isinstance(aq_item, dict)
|
|
assert aq_item["call_id"] == "jfizoalt"
|
|
|
|
# Now deliver ask_question DONE result FIRST (out of order vs run_command)
|
|
aq_done = _load("ask_question_done")
|
|
aq_result_events = map_step_to_events(aq_done, conversation_id=_CID, allocator=alloc)
|
|
assert len(aq_result_events) == 1
|
|
aq_result_item = aq_result_events[0].data["item_data"]
|
|
assert isinstance(aq_result_item, dict)
|
|
# Must pair with ask_question id, NOT run_command id
|
|
assert aq_result_item["call_id"] == "jfizoalt"
|
|
|
|
# Then deliver run_command DONE result
|
|
rc_done = _load("run_command_done")
|
|
rc_result_events = map_step_to_events(rc_done, conversation_id=_CID, allocator=alloc)
|
|
assert len(rc_result_events) == 1
|
|
rc_result_item = rc_result_events[0].data["item_data"]
|
|
assert isinstance(rc_result_item, dict)
|
|
# Must pair with run_command id, NOT ask_question id
|
|
assert rc_result_item["call_id"] == "cbawg2v8"
|
|
|
|
# Allocator must not have been used at all (all ids were real)
|
|
assert alloc.invocation_count == 0
|
|
assert alloc.orphan_output_count == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task 5: pending_interaction extractor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPendingInteractionAskQuestionWaiting:
|
|
"""
|
|
ASK_QUESTION WAITING step → PendingInteraction with kind="ask_question".
|
|
|
|
The spec comes from requestedInteraction.askQuestion (NOT from the top-level
|
|
askQuestion field). trajectory_id and step_index come from
|
|
metadata.sourceTrajectoryStepInfo.
|
|
"""
|
|
|
|
def test_returns_not_none(self) -> None:
|
|
"""A WAITING ask_question step returns a PendingInteraction, not None."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
|
|
def test_kind_is_ask_question(self) -> None:
|
|
"""kind is 'ask_question'."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["kind"] == "ask_question"
|
|
|
|
def test_trajectory_id(self) -> None:
|
|
"""trajectory_id matches metadata.sourceTrajectoryStepInfo.trajectoryId."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["trajectory_id"] == "efb134b2-d69f-43de-bb54-c9ece346d8a3"
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches metadata.sourceTrajectoryStepInfo.stepIndex (12)."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["step_index"] == 12
|
|
|
|
def test_spec_is_ask_question_block(self) -> None:
|
|
"""spec equals the requestedInteraction.askQuestion block."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
spec = result["spec"]
|
|
assert isinstance(spec, dict)
|
|
assert "questions" in spec
|
|
|
|
def test_spec_questions_list(self) -> None:
|
|
"""spec.questions is a non-empty list."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
assert len(questions) == 1
|
|
|
|
def test_spec_question_text(self) -> None:
|
|
"""spec.questions[0].question contains the question text."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
q = questions[0]
|
|
assert isinstance(q, dict)
|
|
assert q["question"] == "What type of project or improvement would you like to focus on?"
|
|
|
|
def test_spec_options_list(self) -> None:
|
|
"""spec.questions[0].options is a list of 4 option dicts."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
first_q = questions[0]
|
|
assert isinstance(first_q, dict)
|
|
options = first_q.get("options")
|
|
assert isinstance(options, list)
|
|
assert len(options) == 4
|
|
|
|
def test_spec_option_id_and_text(self) -> None:
|
|
"""Each option has 'id' and 'text' keys."""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
first_q = questions[0]
|
|
assert isinstance(first_q, dict)
|
|
options = first_q.get("options")
|
|
assert isinstance(options, list)
|
|
first_option = options[0]
|
|
assert isinstance(first_option, dict)
|
|
assert first_option["id"] == "1"
|
|
assert "(Recommended)" in str(first_option["text"])
|
|
|
|
|
|
class TestPendingInteractionRunCommandWaiting:
|
|
"""
|
|
RUN_COMMAND WAITING step → PendingInteraction with kind="permission".
|
|
|
|
The spec comes from requestedInteraction.permission; resource.action and
|
|
resource.target are the authoritative fields.
|
|
"""
|
|
|
|
def test_returns_not_none(self) -> None:
|
|
"""A WAITING run_command step returns a PendingInteraction, not None."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
|
|
def test_kind_is_permission(self) -> None:
|
|
"""kind is 'permission'."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["kind"] == "permission"
|
|
|
|
def test_trajectory_id(self) -> None:
|
|
"""trajectory_id from metadata.sourceTrajectoryStepInfo.trajectoryId."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["trajectory_id"] == "efb134b2-d69f-43de-bb54-c9ece346d8a3"
|
|
|
|
def test_step_index(self) -> None:
|
|
"""step_index matches stepIndex=6."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["step_index"] == 6
|
|
|
|
def test_spec_is_permission_block(self) -> None:
|
|
"""spec equals the requestedInteraction.permission block."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
spec = result["spec"]
|
|
assert isinstance(spec, dict)
|
|
assert "resource" in spec
|
|
|
|
def test_spec_resource_action(self) -> None:
|
|
"""spec.resource.action is 'command'."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
resource = result["spec"]["resource"]
|
|
assert isinstance(resource, dict)
|
|
assert resource["action"] == "command"
|
|
|
|
def test_spec_resource_target(self) -> None:
|
|
"""spec.resource.target is 'pwd'."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
resource = result["spec"].get("resource")
|
|
assert isinstance(resource, dict)
|
|
assert resource["target"] == "pwd"
|
|
|
|
def test_spec_action_description(self) -> None:
|
|
"""spec.actionDescription is present when the fixture has it."""
|
|
step = _load("run_command_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
assert result["spec"].get("actionDescription") == "Running pwd command"
|
|
|
|
|
|
class TestPendingInteractionDoneReturnsNone:
|
|
"""
|
|
DONE steps with requestedInteraction MUST return None.
|
|
|
|
This is the adversarial trap (Task1-M2): both DONE fixtures contain
|
|
requestedInteraction, but status == CORTEX_STEP_STATUS_DONE. The
|
|
implementation must key on status, not on the presence of
|
|
requestedInteraction.
|
|
"""
|
|
|
|
def test_ask_question_done_returns_none(self) -> None:
|
|
"""
|
|
ask_question_done.json has status=DONE and still contains
|
|
requestedInteraction.askQuestion — must return None.
|
|
"""
|
|
step = _load("ask_question_done")
|
|
assert pending_interaction(step) is None
|
|
|
|
def test_run_command_done_returns_none(self) -> None:
|
|
"""
|
|
run_command_done.json has status=DONE and still contains
|
|
requestedInteraction.permission — must return None.
|
|
"""
|
|
step = _load("run_command_done")
|
|
assert pending_interaction(step) is None
|
|
|
|
|
|
class TestPendingInteractionIsMultiSelect:
|
|
"""
|
|
pending_interaction merges is_multi_select from metadata.toolCall.argumentsJson
|
|
into each spec.questions[i].
|
|
|
|
requestedInteraction.askQuestion does not carry is_multi_select; it lives
|
|
in argumentsJson. The fix builds a fresh spec dict with the flag injected.
|
|
"""
|
|
|
|
def test_fixture_is_multi_select_false(self) -> None:
|
|
"""
|
|
ask_question_waiting.json has is_multi_select=false in argumentsJson.
|
|
The merged spec must expose is_multi_select=False on questions[0].
|
|
"""
|
|
step = _load("ask_question_waiting")
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
q = questions[0]
|
|
assert isinstance(q, dict)
|
|
assert q.get("is_multi_select") is False
|
|
|
|
def test_synthetic_is_multi_select_true(self) -> None:
|
|
"""
|
|
Synthetic WAITING step with is_multi_select=true in argumentsJson.
|
|
The merged spec must expose is_multi_select=True on questions[0].
|
|
"""
|
|
import copy
|
|
|
|
base = _load("ask_question_waiting")
|
|
step = copy.deepcopy(base)
|
|
|
|
# Patch argumentsJson to set is_multi_select: true
|
|
metadata = step.get("metadata")
|
|
assert isinstance(metadata, dict)
|
|
tool_call = metadata.get("toolCall")
|
|
assert isinstance(tool_call, dict)
|
|
raw = tool_call.get("argumentsJson")
|
|
assert isinstance(raw, str)
|
|
args = json.loads(raw)
|
|
assert isinstance(args, dict)
|
|
aq_list = args.get("questions")
|
|
assert isinstance(aq_list, list)
|
|
first_q = aq_list[0]
|
|
assert isinstance(first_q, dict)
|
|
first_q["is_multi_select"] = True
|
|
tool_call["argumentsJson"] = json.dumps(args)
|
|
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
q = questions[0]
|
|
assert isinstance(q, dict)
|
|
assert q.get("is_multi_select") is True
|
|
|
|
def test_absent_arguments_json_defaults_false(self) -> None:
|
|
"""
|
|
When argumentsJson is absent, is_multi_select defaults to False without crashing.
|
|
"""
|
|
import copy
|
|
|
|
base = _load("ask_question_waiting")
|
|
step = copy.deepcopy(base)
|
|
|
|
# Remove argumentsJson from toolCall.
|
|
metadata = step.get("metadata")
|
|
assert isinstance(metadata, dict)
|
|
tool_call = metadata.get("toolCall")
|
|
assert isinstance(tool_call, dict)
|
|
tool_call.pop("argumentsJson", None)
|
|
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
q = questions[0]
|
|
assert isinstance(q, dict)
|
|
assert q.get("is_multi_select") is False
|
|
|
|
def test_malformed_arguments_json_defaults_false(self) -> None:
|
|
"""
|
|
When argumentsJson is not valid JSON, is_multi_select defaults to False without crashing.
|
|
"""
|
|
import copy
|
|
|
|
base = _load("ask_question_waiting")
|
|
step = copy.deepcopy(base)
|
|
|
|
metadata = step.get("metadata")
|
|
assert isinstance(metadata, dict)
|
|
tool_call = metadata.get("toolCall")
|
|
assert isinstance(tool_call, dict)
|
|
tool_call["argumentsJson"] = "{ not valid json !!!"
|
|
|
|
result = pending_interaction(step)
|
|
assert result is not None
|
|
questions = result["spec"].get("questions")
|
|
assert isinstance(questions, list)
|
|
q = questions[0]
|
|
assert isinstance(q, dict)
|
|
assert q.get("is_multi_select") is False
|
|
|
|
def test_input_step_not_mutated(self) -> None:
|
|
"""
|
|
pending_interaction must not mutate the input step dict.
|
|
requestedInteraction.askQuestion.questions[0] must remain unchanged.
|
|
"""
|
|
import copy
|
|
|
|
step = _load("ask_question_waiting")
|
|
original = copy.deepcopy(step)
|
|
|
|
pending_interaction(step)
|
|
|
|
# The original requestedInteraction.askQuestion block must be unchanged.
|
|
ri = step.get("requestedInteraction")
|
|
assert isinstance(ri, dict)
|
|
orig_ri = original.get("requestedInteraction")
|
|
assert isinstance(orig_ri, dict)
|
|
assert ri == orig_ri
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reasoning delta builder (output_reasoning_delta_event)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputReasoningDeltaEvent:
|
|
"""``output_reasoning_delta_event`` builds the transient reasoning delta."""
|
|
|
|
def test_shape_started_true(self) -> None:
|
|
"""
|
|
The first reasoning delta of a step carries ``started=True``.
|
|
|
|
``started`` tells the server to precede this delta with one
|
|
``response.reasoning.started`` (the SPA's new-reasoning-block marker).
|
|
"""
|
|
event = output_reasoning_delta_event(step_idx=2, delta="Let me think", started=True)
|
|
assert event.event_type == "external_output_reasoning_delta"
|
|
assert event.data == {"delta": "Let me think", "started": True}
|
|
assert event.step_index == 2
|
|
|
|
def test_shape_started_false(self) -> None:
|
|
"""A continuation reasoning delta carries ``started=False``."""
|
|
event = output_reasoning_delta_event(step_idx=5, delta=" some more", started=False)
|
|
assert event.data == {"delta": " some more", "started": False}
|
|
assert event.step_index == 5
|
|
|
|
def test_no_message_id(self) -> None:
|
|
"""
|
|
Reasoning deltas carry NO ``message_id``.
|
|
|
|
Unlike ``output_text_delta_event`` (whose ``message_id`` lets the SPA
|
|
coalesce text deltas into one block and reconcile against the committed
|
|
message), the SPA reasoning block is not keyed by a per-step id.
|
|
"""
|
|
event = output_reasoning_delta_event(step_idx=2, delta="x", started=True)
|
|
assert "message_id" not in event.data
|
|
|
|
|
|
class TestMapStepEmitsNoReasoning:
|
|
"""The pure mapper never emits a reasoning item — reasoning is delta-only."""
|
|
|
|
def test_done_planner_with_thinking_emits_no_reasoning_item(self) -> None:
|
|
"""
|
|
A DONE planner carrying ``thinking`` maps to message (+ tool calls) only.
|
|
|
|
Reasoning is surfaced solely as transient deltas by the streaming reader
|
|
(mirroring the in-process executor, which never commits a reasoning item).
|
|
The DONE-commit path must therefore not introduce a ``reasoning`` item or
|
|
any ``external_output_reasoning_delta``.
|
|
"""
|
|
step = _load("planner_response_text")
|
|
cast(dict[str, Any], step["plannerResponse"])["thinking"] = "internal chain of thought"
|
|
events = map_step_to_events(step, conversation_id=_CID, allocator=_allocator())
|
|
|
|
for event in events:
|
|
assert event.event_type != "external_output_reasoning_delta"
|
|
if event.event_type == "external_conversation_item":
|
|
assert event.data.get("item_type") != "reasoning"
|
|
# The committed assistant message is still produced (text unchanged).
|
|
item_types = [
|
|
e.data.get("item_type") for e in events if e.event_type == "external_conversation_item"
|
|
]
|
|
assert "message" in item_types
|
|
|
|
|
|
class TestExecutionDiscriminator:
|
|
"""``_execution_discriminator`` extracts the per-turn dedup discriminator."""
|
|
|
|
def test_prefers_execution_id(self) -> None:
|
|
"""``executionId`` is returned when present (preferred over createdAt)."""
|
|
step = {
|
|
"metadata": {
|
|
"executionId": "exec-123",
|
|
"createdAt": "2026-06-22T23:57:36.256051Z",
|
|
}
|
|
}
|
|
assert _execution_discriminator(step) == "exec-123"
|
|
|
|
def test_falls_back_to_created_at(self) -> None:
|
|
"""``createdAt`` is used when ``executionId`` is absent."""
|
|
step = {"metadata": {"createdAt": "2026-06-22T23:57:36.256051Z"}}
|
|
assert _execution_discriminator(step) == "2026-06-22T23:57:36.256051Z"
|
|
|
|
def test_skips_empty_execution_id(self) -> None:
|
|
"""An empty ``executionId`` is skipped in favour of ``createdAt``."""
|
|
step = {"metadata": {"executionId": "", "createdAt": "2026-06-22T23:57:36Z"}}
|
|
assert _execution_discriminator(step) == "2026-06-22T23:57:36Z"
|
|
|
|
def test_none_when_no_metadata(self) -> None:
|
|
"""A step with no ``metadata`` dict yields ``None``."""
|
|
assert _execution_discriminator({}) is None
|
|
assert _execution_discriminator({"metadata": "not-a-dict"}) is None
|
|
|
|
def test_none_when_no_discriminating_fields(self) -> None:
|
|
"""``metadata`` without ``executionId`` or ``createdAt`` yields ``None``."""
|
|
assert _execution_discriminator({"metadata": {"source": "user"}}) is None
|
|
|
|
def test_real_fixture_has_execution_id(self) -> None:
|
|
"""The recorded USER_INPUT fixture carries a usable ``executionId``."""
|
|
step = _load("user_input")
|
|
# Confirms the live wire shape this fix relies on (per-turn executionId).
|
|
assert _execution_discriminator(step) == "1df76a5f-0318-4c71-b31d-7e3b51a3d981"
|