diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index 4fba15f0..f186df8a 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -46,6 +46,6 @@ async def main() -> None: asyncio.run(main()) ``` -`events` is required. Each invocation calls `events(config, user_input, variables, history, session_id)` and the return value must be a fresh async iterable of that session only. The iterable must end: return it after `session.idle`, or close it yourself. `config().stream()` yields `chunk` then `done`. +`events` is required. Each invocation calls `events(config, user_input, variables, history, session_id)` and the return value must be a fresh async iterable of that session only. The iterable must end: return it after `session.idle`, or close it yourself. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. Pass `skip_judges=True`. This handler matches every agent-mode provider, including a judge, and it does not run the judge prompt. diff --git a/packages/opencode-agents/agents.md b/packages/opencode-agents/agents.md index 44d4597c..2410195e 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -26,7 +26,7 @@ Call `init_client()` before `config()`, the same way as the other handlers. The ## OTel -A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. Tool parts are not forwarded and do not open `execute_tool` spans. `config().stream()` is `chunk` then `done`. +A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The handler stream is the AG-UI text and tool subset. `config().stream()` is `chunk` then `done`. ## Call shape diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py index dcd4a8d9..a0aca9ce 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -2,7 +2,8 @@ The session is a caller-supplied async iterable of OpenCode bus events. This module does not start a process and does not construct an OpenCode SDK client. -The public stream is assistant text as ``chunk``, then one ``done``. +The handler yields the AG-UI text and tool subset, then one ``done``. +``config().stream()`` still yields ``chunk`` then ``done``. """ from __future__ import annotations @@ -23,6 +24,7 @@ set_output_content_attributes, text_message, ) +from launchdarkly_ai_server.agui import TurnEmitter from launchdarkly_ai_server.lifecycle import effective_capture_content from launchdarkly_ai_server.sdk_usage import report_usage from launchdarkly_ai_server.utils import number_or_zero @@ -56,19 +58,28 @@ class _Reducer: Text is emitted only for messages ``message.updated`` has marked assistant. A part snapshot is that part's full text so far: a later delta still appends, and a snapshot that repeats text already emitted adds nothing. A text delta - without ``partID`` is dropped, because the snapshot of that part cannot be - matched to it. Tool parts are not forwarded. + without ``partID`` is dropped. The first ``running`` tool state with complete + ``state.input`` opens the call. OpenCode runs the tool. ``state.raw`` is not + diffed into argument text. Events whose ``sessionID`` is not *session_id* are ignored. A missing ``sessionID`` is kept. The reducer does not adopt the first session it sees. A ``session.idle`` for this session ends the iterable. """ - def __init__(self, session_id: str) -> None: + def __init__(self, *, capture_content: bool, session_id: str) -> None: + self._emitter = TurnEmitter(capture_content=capture_content) self._session_id = session_id + self._started: set[str] = set() + self._text_message_id: str | None = None + self._text_open = False + self._held_tools: dict[str, list[Mapping[str, Any]]] = {} self._deltas: list[str] = [] self._roles: dict[str, str] = {} self._held: dict[str, list[tuple[str, str, str]]] = {} + self._parked: list[ + tuple[str, list[tuple[str, str, str]], list[Mapping[str, Any]]] + ] = [] self._part_text: dict[str, str] = {} self._usage_ids: set[str] = set() self._input = 0 @@ -122,7 +133,15 @@ def _accept_text( return [] if role != "assistant": return [] - return self._apply_text(part_id, kind, text) + if self._text_open and message_id != self._text_message_id: + self._held.setdefault(message_id, []).append((kind, part_id, text)) + return [] + events = self._begin_message(message_id) + events.extend(self._flush_held(message_id)) + events.extend(self._apply_text(part_id, kind, text)) + if self._text_open and self._text_message_id == message_id: + events.extend(self._release_tools(message_id)) + return events def _flush_held(self, message_id: str) -> list[dict[str, Any]]: held = self._held.pop(message_id, []) @@ -133,21 +152,100 @@ def _flush_held(self, message_id: str) -> list[dict[str, Any]]: def _apply_text(self, part_id: str, kind: str, text: str) -> list[dict[str, Any]]: if kind == "delta": - self._part_text[part_id] = self._part_text.get(part_id, "") + text + if part_id: + self._part_text[part_id] = self._part_text.get(part_id, "") + text return self._chunk(text) already = self._part_text.get(part_id, "") if text.startswith(already): suffix = text[len(already) :] - self._part_text[part_id] = text + if part_id: + self._part_text[part_id] = text elif not already: suffix = text - self._part_text[part_id] = text + if part_id: + self._part_text[part_id] = text else: return [] if not suffix: return [] return self._chunk(suffix) + def _begin_message(self, message_id: str) -> list[dict[str, Any]]: + if self._text_message_id == message_id: + return [] + events = self._end_open_text() + self._text_message_id = message_id + return events + + def _end_open_text(self) -> list[dict[str, Any]]: + events = self._emitter.end_text() + self._text_message_id = None + self._text_open = False + events.extend(self._flush_parked()) + return events + + def _flush_parked(self) -> list[dict[str, Any]]: + """Emit messages completed while other text was open, after that text ends.""" + parked = self._parked + self._parked = [] + events: list[dict[str, Any]] = [] + for message_id, held, tools in parked: + events.extend(self._emit_parked(message_id, held, tools)) + return events + + def _emit_parked( + self, + message_id: str, + held: list[tuple[str, str, str]], + tools: list[Mapping[str, Any]], + ) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + if held: + events.extend(self._begin_message(message_id)) + for kind, part_id, text in held: + events.extend(self._apply_text(part_id, kind, text)) + if self._text_open and self._text_message_id == message_id: + for part in tools: + events.extend(self._emit_tool(part)) + events.extend(self._emitter.end_text()) + self._text_message_id = None + self._text_open = False + return events + for part in tools: + events.extend(self._emit_tool(part)) + if self._text_message_id == message_id: + self._text_message_id = None + return events + + def _release_tools(self, message_id: str) -> list[dict[str, Any]]: + parts = self._held_tools.pop(message_id, []) + events: list[dict[str, Any]] = [] + for part in parts: + events.extend(self._emit_tool(part)) + return events + + def close(self) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + # End the open assistant message before leftover tools. tool_start + # copies that message into parentMessageId. Then apply every held + # assistant call, including one that completed after the text closed, + # before fail_open marks a started call as a failure. + events.extend(self._end_open_text()) + for message_id, held in list(self._held.items()): + if self._roles.get(message_id) != "assistant" or not held: + continue + tools = self._held_tools.pop(message_id, []) + events.extend(self._emit_parked(message_id, held, tools)) + self._held.pop(message_id, None) + for message_id, parts in list(self._held_tools.items()): + role = self._roles.get(message_id) + if role is None or role == "assistant": + for part in parts: + events.extend(self._emit_tool(part)) + self._held_tools.clear() + events.extend(self._emitter.fail_open()) + return events + def _event(self, raw: Mapping[str, Any]) -> list[dict[str, Any]]: kind = raw.get("type") properties = raw.get("properties") @@ -157,7 +255,11 @@ def _event(self, raw: Mapping[str, Any]) -> list[dict[str, Any]]: return self._delta(properties) if kind == "message.part.updated": part = properties.get("part") - if not isinstance(part, Mapping) or part.get("type") != "text": + if not isinstance(part, Mapping): + return [] + if part.get("type") == "tool": + return self._tool(part) + if part.get("type") != "text": return [] return self._text_snapshot(part) if kind == "message.updated": @@ -197,23 +299,105 @@ def _text_snapshot(self, part: Mapping[str, Any]) -> list[dict[str, Any]]: def _chunk(self, text: str) -> list[dict[str, Any]]: self._deltas.append(text) - return [{"type": "chunk", "text": text}] + events = self._emitter.text(text) + if events: + self._text_open = True + return events + + def _tool(self, part: Mapping[str, Any]) -> list[dict[str, Any]]: + message_id = part.get("messageID") + if not isinstance(message_id, str): + self._held_tools.setdefault("", []).append(part) + return [] + role = self._roles.get(message_id) + if role is None: + self._held_tools.setdefault(message_id, []).append(part) + return [] + if role != "assistant": + return [] + raw_id = part.get("callID") or part.get("id") or "" + call_id = raw_id if isinstance(raw_id, str) else "" + if call_id and call_id in self._started: + return self._emit_tool(part) + if self._text_open and self._text_message_id == message_id: + return self._emit_tool(part) + # Another message is streaming, or this message's text has not opened + # yet. Emitting now would parent the call to the wrong message. + if self._text_open or self._text_message_id == message_id: + self._held_tools.setdefault(message_id, []).append(part) + return [] + return self._emit_tool(part) + + def _emit_tool(self, part: Mapping[str, Any]) -> list[dict[str, Any]]: + state = part.get("state") + if not isinstance(state, Mapping): + return [] + status = state.get("status") + raw_id = part.get("callID") or part.get("id") or "" + call_id = raw_id if isinstance(raw_id, str) else "" + name = part.get("tool") if isinstance(part.get("tool"), str) else "" + if not call_id or status == "pending": + return [] + if status == "running": + if "input" not in state or not name: + return [] + events = self._emitter.tool_start(call_id, name) + if any(event.get("type") == "TOOL_CALL_START" for event in events): + self._started.add(call_id) + events.extend(self._emitter.tool_end(call_id, arguments=state.get("input"))) + return events + if status not in ("completed", "error"): + return [] + opened: list[dict[str, Any]] = [] + if name: + started = self._emitter.tool_start(call_id, name) + if any(event.get("type") == "TOOL_CALL_START" for event in started): + self._started.add(call_id) + opened.extend(started) + opened.extend(self._emitter.tool_end(call_id, arguments=state.get("input"))) + if status == "error": + opened.extend(self._emitter.tool_result(call_id, failed=True)) + else: + opened.extend(self._emitter.tool_result(call_id, state.get("output"))) + return opened def _message(self, info: Mapping[str, Any]) -> list[dict[str, Any]]: message_id = info.get("id") role = info.get("role") events: list[dict[str, Any]] = [] + completed_time = info.get("time") + finished = isinstance(completed_time, Mapping) and "completed" in completed_time if isinstance(message_id, str) and isinstance(role, str): self._roles[message_id] = role if role == "assistant": - events.extend(self._flush_held(message_id)) + # A reserved id is not an open AG-UI message. Only text that + # TurnEmitter has already yielded can take a parent or be split. + late_completion = ( + finished and self._text_open and message_id != self._text_message_id + ) + if late_completion: + # Keep the open message intact. The held text and tools + # still happened; emit them once that text ends. + held = self._held.pop(message_id, []) + tools = self._held_tools.pop(message_id, []) + if held or tools: + self._parked.append((message_id, held, tools)) + else: + events.extend(self._begin_message(message_id)) + events.extend(self._flush_held(message_id)) + if self._text_open and self._text_message_id == message_id: + events.extend(self._release_tools(message_id)) + elif finished: + events.extend(self._release_tools(message_id)) else: self._held.pop(message_id, None) - if role != "assistant" or not isinstance(message_id, str): - return events - completed = info.get("time") - if not isinstance(completed, Mapping) or "completed" not in completed: + self._held_tools.pop(message_id, None) + if role != "assistant" or not isinstance(message_id, str) or not finished: return events + if self._text_message_id == message_id and self._text_open: + events.extend(self._end_open_text()) + elif self._text_message_id == message_id: + self._text_message_id = None if message_id not in self._usage_ids: tokens = info.get("tokens") if isinstance(tokens, Mapping): @@ -323,7 +507,8 @@ def create_opencode_handler( the first one it sees. This function does not start OpenCode and does not construct an OpenCode SDK client. - The stream yields assistant text as ``chunk`` and one ``done``. Pass + The stream yields the AG-UI text and tool subset and one ``done``. + ``config().stream()`` still yields ``chunk`` then ``done``. Pass ``skip_judges=True`` to ``config()``: a judge would otherwise be routed here and *events* would run again on the judge prompt. *capture_content* follows the other handlers. @@ -346,8 +531,8 @@ def opencode_agents( *events* is required and is the same callable :func:`create_opencode_handler` takes. *variables*, *history*, *session_id*, and *capture_content* configure - that call. This does not construct an OpenCode SDK client. Pass - ``skip_judges=True`` so a judged config does not run *events* again. + that call. Pass ``skip_judges=True`` so a judged config does not run + *events* again. """ report_usage("opencode-agents.opencodeAgents", PACKAGE_NAME, __version__) variables = kwargs.pop("variables", None) @@ -391,7 +576,7 @@ async def _stream_impl( capture = effective_capture_content(content_override) span = start_root_span(config, variables or {}) ended: set[int] = set() - reducer = _Reducer(bound_session) + reducer = _Reducer(capture_content=capture, session_id=bound_session) try: instructions = ( config.get("instructions") if isinstance(config, dict) else None @@ -410,6 +595,8 @@ async def _stream_impl( ) ): yield event + for event in reducer.close(): + yield event output = "".join(reducer.deltas) if output: set_output_content_attributes( diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/spans.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/spans.py index b03e9b17..0b2aa2d2 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/spans.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/spans.py @@ -1,9 +1,9 @@ """Span construction for the OpenCode handler. -The public stream is ``chunk`` then ``done``. This module opens the -``invoke_agent`` root the other handlers open, plus the ``chat`` and -``execute_tool`` constructors the shared parity checks call. The reducer does -not emit tool events, so a run only starts the root. +``config().stream()`` is ``chunk`` then ``done``. The handler stream is the +AG-UI text and tool subset. This module opens the ``invoke_agent`` root, plus +the ``chat`` and ``execute_tool`` constructors the shared parity checks call. +A run starts the root. """ from __future__ import annotations diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 9ee80aaf..4e806c23 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -176,7 +176,7 @@ def test_capture_content_follows_configure(self) -> None: class TestText: - async def test_text_around_a_tool_is_chunks_then_done(self) -> None: + async def test_text_around_a_tool_keeps_text_and_the_call(self) -> None: events = await collect( [ assistant_open("m1"), @@ -192,16 +192,26 @@ async def test_text_around_a_tool_is_chunks_then_done(self) -> None: assistant_completed("m2"), ], ) - assert types_of(events) == ["chunk", "chunk", "done"] - assert [event["text"] for event in events if event["type"] == "chunk"] == [ + assert [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] == [ "Looking up", "Done.", ] + start = next(event for event in events if event["type"] == "TOOL_CALL_START") + assert start["toolCallId"] == "c1" + assert start["toolCallName"] == "bash" + assert start["parentMessageId"] + args = next(event for event in events if event["type"] == "TOOL_CALL_ARGS") + assert args["delta"] == '{"command":"ls"}' + result = next(event for event in events if event["type"] == "TOOL_CALL_RESULT") + assert result["content"] == "file.txt" assert events[-1]["output"] == "Looking upDone." - assert "file.txt" not in str(events) - assert "toolCallId" not in str(events) + assert types_of(events).count("TOOL_CALL_START") == 1 - async def test_a_failed_tool_does_not_leak_into_the_stream(self) -> None: + async def test_a_failed_tool_masks_the_provider_body(self) -> None: events = await collect( [ tool_part( @@ -212,10 +222,304 @@ async def test_a_failed_tool_does_not_leak_into_the_stream(self) -> None: ) ], ) - assert types_of(events) == ["done"] + result = next(event for event in events if event["type"] == "TOOL_CALL_RESULT") + assert result["content"] == "An error occurred." + assert result["metadata"] == {"launchdarkly": {"toolFailed": True}} + assert events[-1]["type"] == "done" assert events[-1]["output"] == "" assert "disk on fire" not in str(events) + async def test_a_tool_for_another_message_does_not_attach_to_open_text( + self, + ) -> None: + events = await collect( + [ + assistant_open("m2"), + assistant_open("m1"), + text_delta("m1", "p1", "A"), + tool_part( + "c2", "running", message_id="m2", raw_input={"command": "ls"} + ), + assistant_completed("m1"), + ] + ) + text = next(event for event in events if event["type"] == "TEXT_MESSAGE_START") + start = next(event for event in events if event["type"] == "TOOL_CALL_START") + ended = next(event for event in events if event["type"] == "TEXT_MESSAGE_END") + assert "parentMessageId" not in start + assert events.index(ended) < events.index(start) + assert ended["messageId"] == text["messageId"] + + async def test_a_tool_that_finishes_after_the_text_is_not_a_failure(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + tool_part("c1", "running", raw_input={"command": "ls"}), + assistant_completed("m1"), + tool_part("c1", "completed", output="file.txt"), + ] + ) + result = next(event for event in events if event["type"] == "TOOL_CALL_RESULT") + assert result["content"] == "file.txt" + assert "metadata" not in result + assert events[-1]["output"] == "Hi" + + async def test_a_tool_waits_until_assistant_text_is_open(self) -> None: + events = await collect( + [ + tool_part("c1", "running", raw_input={"command": "ls"}), + assistant_open("m1"), + text_delta("m1", "p1", "Looking"), + tool_part("c1", "completed", output="file.txt"), + assistant_completed("m1"), + ] + ) + text = next(event for event in events if event["type"] == "TEXT_MESSAGE_START") + start = next(event for event in events if event["type"] == "TOOL_CALL_START") + assert events.index(text) < events.index(start) + assert start["parentMessageId"] == text["messageId"] + ended = next(event for event in events if event["type"] == "TEXT_MESSAGE_END") + assert ended["messageId"] == text["messageId"] + + async def test_a_tool_before_the_role_waits_for_the_assistant_message(self) -> None: + events = await collect( + [ + tool_part("c1", "running", raw_input={"command": "ls"}), + text_delta("m1", "p1", "Looking"), + assistant_open("m1"), + tool_part("c1", "completed", output="file.txt"), + assistant_completed("m1"), + ] + ) + start = next(event for event in events if event["type"] == "TOOL_CALL_START") + text = next(event for event in events if event["type"] == "TEXT_MESSAGE_START") + assert start["parentMessageId"] == text["messageId"] + assert events.index(text) < events.index(start) + + async def test_a_late_completion_does_not_split_the_open_message(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "A"), + assistant_open("m2"), + text_delta("m2", "p2", "B1"), + text_delta("m3", "p3", "stale"), + tool_part( + "c3", + "completed", + message_id="m3", + raw_input={"command": "old"}, + output="old.txt", + ), + assistant_completed("m3"), + assistant_completed("m1"), + text_delta("m2", "p4", "B2"), + assistant_completed("m2"), + ] + ) + texts = [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert texts == ["A", "B1", "B2", "stale"] + message_ids = [ + event["messageId"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert message_ids[1] == message_ids[2] + assert message_ids[0] != message_ids[1] + assert message_ids[3] != message_ids[1] + open_end = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_END" + and event["messageId"] == message_ids[1] + ) + stale = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" and event["delta"] == "stale" + ) + assert events.index(open_end) < events.index(stale) + start = next( + event + for event in events + if event["type"] == "TOOL_CALL_START" and event["toolCallId"] == "c3" + ) + result = next( + event + for event in events + if event["type"] == "TOOL_CALL_RESULT" and event["toolCallId"] == "c3" + ) + assert result["content"] == "old.txt" + assert "metadata" not in result + assert start.get("parentMessageId") == message_ids[3] + assert events.index(open_end) < events.index(start) + assert events[-1]["output"] == "AB1B2stale" + + async def test_late_text_does_not_split_the_open_message(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "A"), + assistant_open("m2"), + text_delta("m2", "p2", "B1"), + bus( + "message.part.updated", + part={"id": "p1", "messageID": "m1", "type": "text", "text": "A"}, + ), + text_delta("m1", "p1", "X"), + text_delta("m2", "p2", "B2"), + assistant_completed("m2"), + ] + ) + b_ids = [ + event["messageId"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + and event["delta"] in ("B1", "B2") + ] + assert b_ids == [b_ids[0], b_ids[0]] + late = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" and event["delta"] == "X" + ) + assert late["messageId"] != b_ids[0] + assert events[-1]["output"] == "AB1B2X" + + async def test_a_later_snapshot_does_not_repeat_held_text(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "A"), + assistant_open("m2"), + text_delta("m2", "p2", "B1"), + text_delta("m1", "p1", "X"), + assistant_completed("m2"), + bus( + "message.part.updated", + part={"id": "p1", "messageID": "m1", "type": "text", "text": "AX"}, + ), + assistant_completed("m1"), + ] + ) + deltas = [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert deltas == ["A", "B1", "X"] + assert events[-1]["output"] == "AB1X" + + async def test_unfinished_assistant_text_is_emitted_at_session_end(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "A"), + assistant_open("m2"), + text_delta("m2", "p2", "B1"), + text_delta("m1", "p1", " later"), + ] + ) + b = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" and event["delta"] == "B1" + ) + later = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" and event["delta"] == " later" + ) + ended = next( + event + for event in events + if event["type"] == "TEXT_MESSAGE_END" + and event["messageId"] == b["messageId"] + ) + assert later["messageId"] != b["messageId"] + assert events.index(ended) < events.index(later) + assert events[-1]["output"] == "AB1 later" + + async def test_a_completed_message_is_emitted_when_nothing_is_open(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m2", "p2", "Kept"), + assistant_completed("m2"), + text_delta("m1", "p1", "After"), + assistant_completed("m1"), + ] + ) + contents = [ + event for event in events if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert [event["delta"] for event in contents] == ["Kept", "After"] + assert contents[0]["messageId"] != contents[1]["messageId"] + ends = [ + event["messageId"] + for event in events + if event["type"] == "TEXT_MESSAGE_END" + ] + assert ends == [contents[0]["messageId"], contents[1]["messageId"]] + assert events[-1]["output"] == "KeptAfter" + + async def test_capture_off_hides_tool_content(self) -> None: + built = create_opencode_handler( + capture_content=False, + session_id="ses", + events=source( + [ + assistant_open("m1"), + tool_part("c1", "running", raw_input={"command": "ls"}), + tool_part("c1", "completed", output="file.txt"), + assistant_completed("m1"), + ] + ), + ) + gen = await built.stream({}, "hi", None, None) + events = [event async for event in gen] + assert any(event["type"] == "TOOL_CALL_START" for event in events) + assert not any(event["type"] == "TOOL_CALL_ARGS" for event in events) + rendered = str(events) + assert "ls" not in rendered + assert "file.txt" not in rendered + + async def test_an_unlabeled_tool_does_not_inherit_an_open_message(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + tool_part( + "c1", "running", message_id="m2", raw_input={"command": "ls"} + ), + ] + ) + text = next(event for event in events if event["type"] == "TEXT_MESSAGE_START") + start = next(event for event in events if event["type"] == "TOOL_CALL_START") + assert "parentMessageId" not in start + ended = next(event for event in events if event["type"] == "TEXT_MESSAGE_END") + assert events.index(ended) < events.index(start) + assert ended["messageId"] == text["messageId"] + + async def test_a_user_message_tool_is_not_emitted(self) -> None: + events = await collect( + [ + tool_part( + "c1", "running", message_id="u1", raw_input={"command": "ls"} + ), + bus( + "message.updated", + info={"id": "u1", "role": "user", "time": {"created": 1}}, + ), + ] + ) + assert not any(event["type"].startswith("TOOL_CALL") for event in events) + async def test_pending_tool_state_emits_nothing(self) -> None: events = await collect( [tool_part("c1", "pending", raw_input={}, output=None)], @@ -256,8 +560,12 @@ async def test_a_text_snapshot_does_not_repeat_deltas(self) -> None: assistant_completed("m1"), ], ) - chunks = [event for event in events if event["type"] == "chunk"] - assert chunks == [{"type": "chunk", "text": "Hi"}] + deltas = [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert deltas == ["Hi"] assert events[-1]["output"] == "Hi" async def test_a_snapshot_then_later_deltas_append(self) -> None: @@ -281,8 +589,12 @@ async def test_a_snapshot_then_later_deltas_append(self) -> None: assistant_completed("m1"), ], ) - chunks = [event["text"] for event in events if event["type"] == "chunk"] - assert chunks == ["He", "llo"] + deltas = [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert deltas == ["He", "llo"] assert events[-1]["output"] == "Hello" async def test_user_text_is_not_the_assistant_reply(self) -> None: @@ -563,8 +875,12 @@ async def test_a_delta_without_part_id_is_not_repeated_by_its_snapshot( assistant_completed("m1"), ], ) - chunks = [event["text"] for event in events if event["type"] == "chunk"] - assert chunks == ["Hi"] + contents = [ + event["delta"] + for event in events + if event["type"] == "TEXT_MESSAGE_CONTENT" + ] + assert contents == ["Hi"] assert events[-1]["output"] == "Hi" async def test_skip_judges_runs_events_once(