From be245e334647334c8531765667c0f763fa2c83e1 Mon Sep 17 00:00:00 2001 From: Stan Tian Date: Fri, 4 Sep 2026 20:32:50 +0000 Subject: [PATCH] fix(chat): stop replaying a productive turn as an empty response A turn that streamed text and then called a tool arrived at the terminal chain with an empty final segment, because `assistant_text` is reset at every tool boundary and nothing recorded that visible output had already been flushed. The empty-response ladder read that as "the model returned nothing" and re-queued the ORIGINAL user message, which re-runs tool calls that already completed and re-derives an answer the user has already read. A productive turn (flushed visible text, a dispatched tool call, or thinking) can no longer take the verbatim-replay rung. It gets at most ONE continuation, worded so the model is not told its completed work produced no output, and that continuation consumes the remaining recovery budget. A genuinely activity-free empty turn keeps the existing bounded replay -> continue -> give-up ladder unchanged. The verdict also had no cause. Five physically different realities -- a provider that generated nothing, a terminal carrying no stop reason, no terminal event at all, a synthesized terminal, and a turn whose output was only tools or thinking -- collapsed onto one warning that printed none of the state the runner already held. Each now reports a closed cause and the rung it took, alongside booleans only: no prompt, no response text, no thinking, no tool arguments or results, no paths, no identities, no token counts and no costs. Two ACP-side silences are closed the same way: the pre-turn drain says how many leftover frames it destroyed (a count, never their contents), and a prompt stream that exhausts cleanly without a terminal completion now says so, which is the one state the dashboard cannot distinguish from a model that answered with nothing. --- docs/system-specs/modules/acp-client.md | 37 ++ docs/system-specs/modules/session.md | 48 ++- src/kiro_crew/acp/session_handle.py | 52 +++ src/kiro_crew/dashboard/chat_runner.py | 148 ++++++- src/kiro_crew/dashboard/chat_utils.py | 178 ++++++++ test/test_acp_runtime.py | 106 +++++ test/test_dashboard_chat.py | 486 +++++++++++++++++++++- test/test_subagent_delivery_ttl_anchor.py | 5 +- 8 files changed, 1044 insertions(+), 16 deletions(-) diff --git a/docs/system-specs/modules/acp-client.md b/docs/system-specs/modules/acp-client.md index 66d60f9fe50..9606c4bedd7 100644 --- a/docs/system-specs/modules/acp-client.md +++ b/docs/system-specs/modules/acp-client.md @@ -588,6 +588,43 @@ This leverages kiro-cli's `promptCapabilities.image: true` capability. The LLM r The summary carries **no message content** — only counts, types, and sizes — which is a hard requirement (issue #6022): the kiro-cli data dir is fenced precisely because it holds SSO tokens, so the diagnostics must never record block text, image bytes, or tool arguments. This lets an operator tell a stale/invalid model id apart from a structurally malformed payload the next time a turn is rejected as `Improperly formed request` (see the `_RE_MALFORMED_REQUEST` classifier), without ever exposing what the turn contained. The helper is defensive by contract: it never raises into the live prompt path (a malformed block list yields a partial/minimal summary), so a diagnostics failure can never break a turn. +### Turn-boundary loss diagnostics (content-free) + +Two places in `AcpSessionHandle.prompt` could destroy or omit a turn's evidence +silently. Both now report, and both report **only** a count or the bare fact — +never frame text, tool arguments, tool results, or frame SIZE, since a size leaks +response length. + +- **Pre-turn stale drain.** The drain empties the session queue of frames left by + an abandoned turn (see the cancel-unacked / stale / tool-stall / timeout paths, + which synthesize a terminal and return while the real kiro-cli turn keeps + emitting). Permission REQUESTS are answered rather than dropped; everything else + is discarded, which used to happen with no count and no log. It now counts the + discarded frames and emits **one** WARNING per turn carrying that count — one + line regardless of how many frames drained, so a burst cannot flood the log. + This matters downstream: a turn whose terminal was destroyed here reaches the + dashboard as an empty response with no attributable cause. The count is NOT + bridged into `chat_runner` — see the note below. +- **A prompt stream that ends without a terminal.** `_dispatch_events` + synthesizes an `EVENT_COMPLETE` on every exit path it knows about, so a + consumer that never receives one is looking at a path that has none. The + generator warns when it exhausts CLEANLY having yielded no terminal. + "Cleanly" is what makes the line spam-free: a consumer close + (`GeneratorExit`), a cancellation, and any raised error all skip it, and each + is already logged by whoever caused it. The terminal is marked as delivered + BEFORE its `yield`, so a consumer that closes the stream on the terminal is not + reported as having lost it. + +**Deliberately not bridged to the runner.** The drain count stays inside this +layer. Reaching `chat_runner` would mean a new field on the `AcpEvent` / +`LLMEvent` provider contract plus plumbing through `_dispatch` and the provider, +and `scripts/check_agent_sdk_boundary.py` baselines the dashboard modules at a +count that may not grow — a materially larger change than the fault it would +report. Unknown `sessionUpdate` discriminants in `_dispatch.py` are still ignored +silently for a related reason: `parse_session_update` is a pure function on a hot +path with no per-session state, so a bounded log there needs a dedupe set it does +not have, and an unbounded one would log per frame. + ## AcpRuntime & AcpSessionHandle (session multiplexing) diff --git a/docs/system-specs/modules/session.md b/docs/system-specs/modules/session.md index 99dc665e335..c28bde4e787 100644 --- a/docs/system-specs/modules/session.md +++ b/docs/system-specs/modules/session.md @@ -142,7 +142,8 @@ send time. cancellation is treated as a transient provider failure and recovered through a bounded three-rung ladder driven by `slot._empty_response_retries`: 1. **first empty** → the ORIGINAL message is silently re-queued at the - front of the slot queue (no visible card); + front of the slot queue (no visible card). Reached ONLY by a turn with no + activity — see the productive-turn exclusion below; 2. **second empty** (the same-message retry also produced nothing) → ONE synthetic continue nudge (`_EMPTY_AUTO_CONTINUE_MSG` — a DIFFERENT message, since re-sending the identical prompt tends to reproduce the @@ -153,6 +154,51 @@ send time. 3. **third empty** (the nudge also produced nothing) → terminal notice card asking the user to send a message; the counter resets so the next genuine user turn gets a fresh budget. + + **A PRODUCTIVE turn never reaches rung 1.** "Empty" at this branch means only + that the FINAL assistant segment is empty, which is not the same as "the turn + did nothing": `assistant_text` is reset at every tool boundary, so a turn that + streamed an answer and then called a tool arrives here with its answer already + flushed, persisted and on screen, and a tool-only turn arrives here having run + real side effects. Rung 1 re-queues the user's own message, so for either shape + it re-executes completed tool calls (a second `send_message`, a second write, a + second PR) and re-derives an answer the user has already read — observed in the + field as two consecutive billed `end_turn` turns, each with a preamble and + successful tool calls, both classified empty and the first verbatim-replayed. + `chat_utils.EmptyTurnActivity.productive` is the guard: a flushed visible + segment, a dispatched tool call, or thinking. A productive turn skips to rung 2, + which carries `_ACTIVITY_NO_REPLY_CONTINUE_MSG` instead — the same + `EMPTY_RESPONSE_RECOVERY_PREFIX` marker (so no new recovery card or locale pair + is needed) with a body that does NOT claim the turn produced nothing, because + that body is read by the model and would invite it to redo work whose side + effects already landed. Its notice card differs for the same reason. The ladder + bound is unchanged: a productive turn spends the same budget, it simply never + spends it on a replay. `_produced_visible_output` deliberately does NOT cover + this case — its narrow meaning (only the mid-turn resets that are not tool + boundaries: steer cut, compaction, clear, agent switch) is load-bearing for the + promise-only guard. + + **Turn-end diagnostics.** The branch emits ONE privacy-safe WARNING per empty + verdict, after the rung is chosen, naming a closed `cause` and `rung` plus + booleans: `provider_empty`, `tool_only`, `thinking_only`, `visible_partial`, + `no_terminal_event`, `synthetic_completion` or `other` + (`chat_utils.classify_empty_turn`, ranked most-specific first), and `replay` / + `continue` / `give_up`. `EmptyTurnActivity` carries whether a terminal + `EVENT_COMPLETE` arrived, whether the provider SYNTHESIZED it, the terminal stop + reason normalised onto a closed set (`chat_utils.normalize_stop_reason` — an + omitted reason answers `absent`, which is a distinct observation from a clean + `end_turn` and must not be laundered into one, and an unrecognised backend + string answers `other` rather than being echoed), whether text streamed, whether + a visible segment was flushed at a tool boundary, whether tools ran, whether + thinking ran, and whether the provider reported ANY billing dimension. Every + field is a bool or a closed constant by contract: no prompts, responses, + thinking, tool arguments or results, paths, identities, token counts or costs. + The predecessor logged only `Empty model response (attempt N)`, which could not + separate a provider that generated nothing from a turn whose answer a tool + boundary flushed away from a turn no terminal event ever closed — three faults + with three different owners, and one field incident hit all three in three + consecutive attempts. + Recovery rungs 1–2 skip persistence/consolidation/success-recording (the empty turn is never saved) and preserve all other retry budgets. Synthetic recovery messages (`_SYNTHETIC_RECOVERY_MSGS`: the post-transient CONTINUE diff --git a/src/kiro_crew/acp/session_handle.py b/src/kiro_crew/acp/session_handle.py index f9d3a895448..3606283bee7 100644 --- a/src/kiro_crew/acp/session_handle.py +++ b/src/kiro_crew/acp/session_handle.py @@ -906,6 +906,15 @@ async def _run_turn( # from an abandoned turn (or routed here for a backend child between # turns) gets the fail-closed reject; the live turn's requests are # handled by the dispatch loop as before. + # A DROPPED frame is invisible to every layer above: the abandoned turn's + # output vanishes here with nothing to show it existed, and a turn that + # loses its terminal this way reaches the dashboard as an empty response + # with no attributable cause. Count them and say how many, ONCE. Never + # what they were: a frame carries model text, tool arguments and tool + # results, and none of that belongs in a log — nor its size, which leaks + # response length. The count is bounded by the queue, and the log line is + # one per turn regardless of how many frames drained. + _stale_dropped = 0 while True: try: stale = self._queue.get_nowait() @@ -979,6 +988,19 @@ async def _run_turn( _stale_sid if _stale_sid != self._session_id else "" ), ) + else: + # Everything that is not a permission request is DISCARDED, which + # is correct (it belongs to a turn nobody is reading any more) but + # was silent. Count it. + _stale_dropped += 1 + + if _stale_dropped: + logger.warning( + "pre-turn drain discarded %d leftover frame(s) from a prior " + "abandoned turn on this session; those frames — possibly " + "including that turn's terminal — reached no consumer", + _stale_dropped, + ) self.last_prompt_stats = self.last_prompt_stats.carry_over() @@ -1019,6 +1041,16 @@ async def _run_turn( _mark(self._session_id, False) raise + # Did a terminal reach the consumer, and did this generator finish of its + # own accord? Together these answer a question no layer above can: the + # dashboard reads "no EVENT_COMPLETE" as an empty response and cannot tell + # whether the backend never closed the turn or the consumer simply walked + # away. Only a CLEAN exhaustion is reported, which is what makes the + # warning spam-free: a consumer close (GeneratorExit), a cancellation, and + # any raised error all leave `_exhausted_clean` False and are already + # logged by whoever caused them. + _yielded_terminal = False + _exhausted_clean = False try: # Surface any drain-time rejections (see the pre-turn drain above) # as crew-card activity before the turn's own events — the user @@ -1052,6 +1084,11 @@ async def _run_turn( # single choke point because `_dispatch_events` yields from 15 # places and every one of them funnels through this `async for`. self._parked_since = time.monotonic() + if event.kind == EVENT_COMPLETE: + # Set BEFORE the yield: a consumer that closes the stream ON + # the terminal still received it, and marking it after would + # report a lost terminal that was in fact delivered. + _yielded_terminal = True try: yield event finally: @@ -1064,11 +1101,26 @@ async def _run_turn( if self._parked_since is not None: self._parked_total += time.monotonic() - self._parked_since self._parked_since = None + # Reached only when the dispatch loop returned on its own — not on a + # close, a cancel, or an exception. + _exhausted_clean = True finally: if _mark is not None: _mark(self._session_id, False) if not self._turn_done.is_set(): self._turn_done.set() + if _exhausted_clean and not _yielded_terminal: + # The dispatch loop synthesizes a terminal on every path it knows + # about (timeout, stale, tool stall, cancel-unacked), so reaching + # here means one of its exits has none — and the consumer is left + # deciding what an unclosed turn means. Content-free by + # construction: this line carries no count, no text and no ids, + # because the only fact it has to report is that it happened. + logger.warning( + "prompt stream for this session ended without a terminal " + "completion event; the caller will see the turn as producing " + "nothing" + ) # ── Turn park state (readable from OUTSIDE the turn) ── diff --git a/src/kiro_crew/dashboard/chat_runner.py b/src/kiro_crew/dashboard/chat_runner.py index 1707dac3a61..25e19fc26d5 100644 --- a/src/kiro_crew/dashboard/chat_runner.py +++ b/src/kiro_crew/dashboard/chat_runner.py @@ -285,20 +285,27 @@ # classify them identically to the turn logic here). Re-exported under their # historical names so existing imports keep working. from kiro_crew.dashboard.chat_utils import ( # noqa: E402 + _ACTIVITY_NO_REPLY_CONTINUE_MSG, _COMPACTION_CONTINUE_MSG, _EMPTY_AUTO_CONTINUE_MSG, _POSTTOKEN_RECOVER_MSG, _PROMISE_ONLY_CONTINUE_MSG, _SYNTHETIC_RECOVERY_MSGS, CRON_NOTIFICATION_KIND, + EMPTY_RUNG_CONTINUE, + EMPTY_RUNG_GIVE_UP, + EMPTY_RUNG_REPLAY, SUBAGENT_COMPLETION_KIND, SYNTHETIC_RECOVERY_KIND, TRANSIENT_RETRY_KIND, + EmptyTurnActivity, RecoveryPayload, + classify_empty_turn, is_promise_only_terminal, is_synthetic_payload_item, is_synthetic_recovery_item, mint_options_token, + normalize_stop_reason, payload_for_replay, should_continue_after_compaction, should_notice_leaked_tool_call, @@ -5551,6 +5558,25 @@ def _steer_segment_cut() -> None: # reset the buffer WITHOUT a tool boundary — steer cut, compaction, clear, # agent switch) is load-bearing for the promise-only guard below. _turn_flushed_visible_text = False + # ── Content-free turn-end diagnostics (empty-response verdict) ── + # Booleans only, by contract. The empty-response branch below reaches its + # verdict from these, and a WARNING names the cause it derived; every field + # is safe to log because none of them can carry a prompt, a response, a tool + # argument, a path, an identity, a token count or a cost. What the incident + # they exist for needed was exactly this: whether a terminal event arrived at + # all, whether the provider or the backend closed the turn, and whether the + # turn had done work — none of which the single "Empty model response" + # warning could say. + # + # `_turn_flushed_visible_text` above and `_turn_tool_calls` / `_turn_thought` + # below already carry three of the observations, so only what nothing else + # records is added here. + _saw_terminal_event = False + # Retained past the EVENT_COMPLETE arm on purpose: the verdict is reached in + # the post-stream chain, which no longer has the event. + _terminal_synthetic = False + _saw_text_chunk = False + _turn_billed = False # Was this turn's prompt CONSUMED by the model? Reported to whoever armed the # turn (a queued sub-agent completion's retention clock -- see # ``_arm_queued_delivery_settlement``), because every handled-failure path @@ -7038,6 +7064,11 @@ def _queue_recovery( # reset of assistant_text above (planning turn only). if _orch_planning: _orch_plan_buf += safe_chunk + # Set BEFORE the `_turn_emitted` flip: the consumption report + # below must stay adjacent to that flip (pinned by + # test_subagent_delivery_ttl_anchor), so a diagnostic flag goes + # above it rather than between the two. + _saw_text_chunk = True _turn_emitted = True # tokens delivered — transient retry now unsafe await _report_consumed(irreversible=True) # Stream to the wire through the rolling buffer so a credential @@ -9103,6 +9134,15 @@ def _queue_recovery( # is asymmetric (a duplicate re-announce versus a pruned result). if event.stop_reason == STOP_REASON_END_TURN: await _report_consumed() + # Turn-end diagnostics. Read only from `event`, which nothing in + # this arm mutates, so the position is free — kept below the + # consumption gate because that gate's adjacency to the arm's start + # is pinned by test_subagent_delivery_ttl_anchor. + # `synthetic_completion` is readable ONLY here, and the + # empty-response verdict that needs it runs after the stream loop. + _saw_terminal_event = True + _terminal_synthetic = bool(event.synthetic_completion) + _turn_billed = usage_has_billing(event.usage) # Hang-attribution snapshot BEFORE the close-all safety net # below force-marks every card done: only children still # unfinished at the cut may count toward timeout attribution @@ -9868,12 +9908,40 @@ def _emit_error(msg: str, *, will_retry: bool = False) -> None: # the same message just re-hits the same gate. The `not _refusal_reasons` # guard lets it fall through to the refusal-recovery path below, which # hands the model the reason so it can adapt instead of looping. - logger.warning( - "Empty model response for slot %s (attempt %d)", - slot.key, - slot._empty_response_retries + 1, + # + # "Empty" here means only "the final assistant segment is empty", and + # that is NOT the same as "the turn did nothing". `assistant_text` is + # reset at every tool boundary, so a turn that answered and then called + # a tool arrives here with its answer already flushed, persisted and + # read — and `_produced_visible_output` does not cover that, by design + # (its narrow meaning is load-bearing for the promise-only guard). + # A tool-only turn arrives here too. The activity snapshot below is what + # separates those from a provider that genuinely returned nothing. + _empty_activity = EmptyTurnActivity( + saw_terminal=_saw_terminal_event, + terminal_synthetic=_terminal_synthetic, + stop_reason=normalize_stop_reason(_stop_reason), + saw_text=_saw_text_chunk, + flushed_visible=_turn_flushed_visible_text, + had_tools=_turn_tool_calls > 0, + had_thinking=_turn_thought, + billed=_turn_billed, ) - if _prompt_depth == 0 and slot._empty_response_retries < 1: + _empty_cause = classify_empty_turn(_empty_activity) + # Snapshot BEFORE the rungs, each of which may increment the counter. + # This is the ordinal of the turn just observed, which is what the + # predecessor warning reported. + _empty_attempt = slot._empty_response_retries + 1 + # THE load-bearing guard. A productive turn must never have its + # originating message replayed verbatim: rung 1 below re-queues + # `message` itself, which on such a turn re-runs tool calls that + # already completed (a second `send_message`, a second write, a second + # PR) and re-derives an answer the user has already read. A productive + # turn skips to the continuation rung, which tells the model the work + # above already happened. + _may_replay_verbatim = not _empty_activity.productive + if _prompt_depth == 0 and slot._empty_response_retries < 1 and _may_replay_verbatim: + _empty_rung = EMPTY_RUNG_REPLAY # Seamless self-heal: silently re-queue on the first empty # response. An ephemeral status indicator is not used here — it # is emitted at turn-teardown and the frontend drops it once the @@ -9909,20 +9977,48 @@ def _emit_error(msg: str, *, will_retry: bool = False) -> None: # session, with a transcript-visible notice so the recovery is # never invisible. Third empty falls through to the give-up # notice below — bounded, no loop. - slot._empty_response_retries += 1 - slot.append( - "notice", - "ℹ️ The model returned nothing twice — auto-continuing once.", - "msg msg-info", - ) + # A productive turn skipped the verbatim-replay rung entirely. + # Its ONE continuation consumes the remaining recovery budget: + # if that continuation is itself empty, the next turn must give + # up rather than enqueue a second continuation. Leaving the + # counter at 1 here would run `_EMPTY_AUTO_CONTINUE_MSG` next, + # contradicting both the notice ("continuing once") and the + # side-effect boundary this branch protects. + if _empty_activity.productive: + slot._empty_response_retries = 2 + else: + slot._empty_response_retries += 1 + _empty_rung = EMPTY_RUNG_CONTINUE + if _empty_activity.productive: + # Same rung, different words, because the words are read by + # the MODEL and by the user. "returned nothing twice" is + # false for a turn that streamed an answer or ran tools, and + # telling a model its completed work produced no output is an + # invitation to redo it — the side-effect duplication this + # path exists to avoid. + slot.append( + "notice", + "ℹ️ The turn ended without a closing reply — continuing " + "once from what already ran.", + "msg msg-info", + ) + _empty_continue_msg = _ACTIVITY_NO_REPLY_CONTINUE_MSG + else: + slot.append( + "notice", + "ℹ️ The model returned nothing twice — auto-continuing once.", + "msg msg-info", + ) + _empty_continue_msg = _EMPTY_AUTO_CONTINUE_MSG _queue_recovery( 0, - _EMPTY_AUTO_CONTINUE_MSG, + _empty_continue_msg, kind=SYNTHETIC_RECOVERY_KIND, payload=RecoveryPayload.CONTINUATION, ) _retrying_empty = True else: + _empty_rung = EMPTY_RUNG_GIVE_UP # Recoverable, usually-transient: the runner already silently # self-retried once (first empty = silent re-queue). Surface a # soft "notice" card (not a red "error" card) so a self-healing @@ -9936,6 +10032,34 @@ def _emit_error(msg: str, *, will_retry: bool = False) -> None: "again to continue." ) slot.append("notice", _empty_msg, "msg msg-info") + # ONE warning per empty verdict, emitted AFTER the rung is chosen so + # the log line carries the decision rather than only the symptom. The + # predecessor logged just "Empty model response (attempt N)", which + # could not distinguish a provider that generated nothing from a turn + # whose answer a tool boundary flushed away, from a turn no terminal + # event ever closed — three different faults with three different + # owners, and the field incident hit all three in three consecutive + # attempts. Every field here is a closed value or a bool by + # construction (see EmptyTurnActivity): no prompt, no response, no + # thinking, no tool arguments or results, no paths, no identities, no + # token counts and no costs. + logger.warning( + "Empty model response for slot %s (attempt %d) cause=%s rung=%s " + "stop_reason=%s terminal=%s synthetic=%s text=%s flushed_visible=%s " + "tools=%s thinking=%s billed=%s", + slot.key, + _empty_attempt, + _empty_cause, + _empty_rung, + _empty_activity.stop_reason, + _empty_activity.saw_terminal, + _empty_activity.terminal_synthetic, + _empty_activity.saw_text, + _empty_activity.flushed_visible, + _empty_activity.had_tools, + _empty_activity.had_thinking, + _empty_activity.billed, + ) # Fallback arm: a plan emitted BEFORE further tool calls was flushed out # of `assistant_text` (reset on each tool boundary), so the final-segment # detector above missed it and no [OPTION] gate would register — the diff --git a/src/kiro_crew/dashboard/chat_utils.py b/src/kiro_crew/dashboard/chat_utils.py index 90400431676..92a2db5404c 100644 --- a/src/kiro_crew/dashboard/chat_utils.py +++ b/src/kiro_crew/dashboard/chat_utils.py @@ -14,6 +14,7 @@ import re import time import uuid +from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from typing import TYPE_CHECKING, Any @@ -1574,6 +1575,23 @@ def _edit_queued_by_id(messages: list[dict], queue_id: str, content: str) -> boo "conversation above and respond now — do NOT restart from scratch and do " "NOT re-run steps or tools that already completed successfully." ) +_ACTIVITY_NO_REPLY_CONTINUE_MSG = ( + f"{EMPTY_RESPONSE_RECOVERY_PREFIX}\n" + "Your previous turn did work — it streamed text, called tools, or reasoned " + "— but ended without a closing reply, so the request looks unanswered. " + "Everything above already happened and its results are in the conversation: " + "answer now from what is there. Do NOT restart the request, and do NOT " + "re-run any tool or step that already completed." +) +#: Shares :data:`EMPTY_RESPONSE_RECOVERY_PREFIX` with +#: :data:`_EMPTY_AUTO_CONTINUE_MSG` rather than minting a marker of its own. The +#: marker is what ``RecoveryCard.tsx`` classifies a transcript row by, and both +#: bodies are the same event to a reader ("the turn ended without an answer, and +#: the runner continued it once"); a second marker would need a card row and a +#: locale pair in twelve catalogs to say nothing new. The BODIES must differ, +#: because this one is read by the MODEL: telling a turn that ran tools that it +#: "produced no output" invites it to redo work whose side effects already +#: landed, which is the failure this whole path exists to prevent. _PROMISE_ONLY_CONTINUE_MSG = ( f"{PROMISE_ONLY_RECOVERY_PREFIX}\n" "Your previous turn ended right after you said you would perform an action " @@ -1601,6 +1619,7 @@ def _edit_queued_by_id(messages: list[dict], queue_id: str, content: str) -> boo _BUSY_RECOVER_MSG, _POSTTOKEN_RECOVER_MSG, _EMPTY_AUTO_CONTINUE_MSG, + _ACTIVITY_NO_REPLY_CONTINUE_MSG, _PROMISE_ONLY_CONTINUE_MSG, _COMPACTION_CONTINUE_MSG, ) @@ -2042,6 +2061,165 @@ def should_notice_mixed_turn_leak( return has_leaked_tool_call(final_segment_text) +#: Normalised, CLOSED stop-reason vocabulary for the empty-turn diagnostic. The +#: raw wire value is never logged: a backend is free to invent a reason string, +#: and an unbounded value in a diagnostic is both a cardinality hazard and a +#: place model- or user-derived text could appear. +#: +#: The three literals are spelled here rather than imported from +#: ``kiro_crew.acp.types``, following the precedent in ``metrics/turns.py``: +#: ``scripts/check_agent_sdk_boundary.py`` baselines this file at ONE ACP edge +#: and a baselined file may not grow its count. Duplicating a wire constant is +#: only safe with a guard, so ``test_dashboard_chat.py`` pins each against the +#: ACP constant it mirrors — the test tree is outside the gate's scope, so the +#: pin can import what this module may not. +_STOP_END_TURN = "end_turn" +_STOP_CANCELLED_REASON = "cancelled" +_STOP_REFUSAL = "refusal" + +#: A terminal event arrived carrying NO stop reason at all. Deliberately its own +#: value rather than folded onto ``end_turn``: ``metrics.turns.turn_outcome`` +#: reads absence as a clean turn (correct for latency accounting, where the acp +#: path leaves it unset on every normal completion), but here the two are the +#: whole question — "the provider said the turn ended and produced nothing" is a +#: model-side event, while "the provider never said why it stopped" is a +#: transport-side one, and the incident that motivated these diagnostics could +#: not tell them apart. +STOP_REASON_ABSENT = "absent" +#: A terminal stop reason outside the closed set above. +STOP_REASON_OTHER = "other" + +#: Causes for a turn that reached the empty-response verdict. Closed set, +#: low-cardinality, content-free — safe for a log line and for a metric +#: attribute if one is ever added. +EMPTY_CAUSE_NO_TERMINAL = "no_terminal_event" +EMPTY_CAUSE_SYNTHETIC = "synthetic_completion" +EMPTY_CAUSE_VISIBLE_PARTIAL = "visible_partial" +EMPTY_CAUSE_TOOL_ONLY = "tool_only" +EMPTY_CAUSE_THINKING_ONLY = "thinking_only" +EMPTY_CAUSE_PROVIDER_EMPTY = "provider_empty" +EMPTY_CAUSE_OTHER = "other" + +#: Which rung of the empty-response ladder claimed the turn. +EMPTY_RUNG_REPLAY = "replay" +EMPTY_RUNG_CONTINUE = "continue" +EMPTY_RUNG_GIVE_UP = "give_up" + + +def normalize_stop_reason(stop_reason: str | None) -> str: + """Map a raw terminal stop reason onto the closed diagnostic vocabulary. + + ``None`` and ``""`` both answer :data:`STOP_REASON_ABSENT` — an omitted + reason, which is a distinct observation from a clean ``end_turn`` and must + not be laundered into one. Anything unrecognised answers + :data:`STOP_REASON_OTHER`, so no raw backend string is ever logged. + """ + raw = stop_reason or "" + if not raw: + return STOP_REASON_ABSENT + if raw in (_STOP_END_TURN, _STOP_CANCELLED_REASON, _STOP_REFUSAL): + return raw + if raw.startswith("error:"): + return "error" + return STOP_REASON_OTHER + + +@dataclass(frozen=True) +class EmptyTurnActivity: + """What a turn DID, in booleans only, as observed at the empty-response verdict. + + Every field is a bool or a value from a closed set. There are deliberately no + counts, no durations, no token or credit numbers, no paths, no ids, and no + text: this object exists to be written to a log line, and the incident it was + built for is one where the interesting facts (did a tool run? did the user + already read something?) are exactly the facts a privacy-safe diagnostic can + carry. A count would answer no question the bool does not, and token counts + and costs are billing data that has no business in a warning. + + ``billed`` is likewise a bool: whether the provider reported ANY billing + dimension for the turn (``llm_helpers.usage_has_billing``). It separates the + two shapes of "nothing came back" that look identical from the runner — a + turn the provider generated and charged for, versus one it never ran. + """ + + #: A terminal ``EVENT_COMPLETE`` arrived. False means the stream ended + #: without one, and every other field describes a turn nobody closed. + saw_terminal: bool = False + #: The terminal was SYNTHESIZED by the provider layer (watchdog, timeout, + #: cancel-unacked) rather than reported by the backend. Retained past the + #: event arm because the verdict below is reached long after it. + terminal_synthetic: bool = False + #: Normalised terminal stop reason — see :func:`normalize_stop_reason`. + stop_reason: str = STOP_REASON_ABSENT + #: At least one assistant text chunk streamed this turn. + saw_text: bool = False + #: A visible assistant segment was FLUSHED and persisted at a tool boundary, + #: so the user has already read text this turn even though the final segment + #: is empty. This is the incident's own shape. + flushed_visible: bool = False + #: At least one tool call was dispatched. + had_tools: bool = False + #: At least one thinking chunk arrived. + had_thinking: bool = False + #: The provider reported some billing dimension for the turn. + billed: bool = False + + @property + def productive(self) -> bool: + """True when the turn did work that can carry state or side effects. + + This is the load-bearing predicate: a turn that is productive must NEVER + have its originating message replayed verbatim, because the replay + re-executes tool calls that already completed and re-derives an answer + the user has already read. Text that merely STREAMED is not enough on its + own — an un-flushed partial segment is still in ``assistant_text`` and is + handled by the answer branch — so the three triggers are a flushed + visible segment, a dispatched tool call, and thinking, each of which + leaves the conversation in a state a replay would corrupt or duplicate. + """ + return self.flushed_visible or self.had_tools or self.had_thinking + + +def classify_empty_turn(activity: EmptyTurnActivity) -> str: + """Name the cause of an empty-response verdict, from the closed cause set. + + Ordered most-specific first, and the order is the point: the causes overlap + (a synthesized terminal usually also has tool activity), so a flat set of + predicates would report whichever the code happened to check first. The + ranking is by what an operator must act on. + + 1. :data:`EMPTY_CAUSE_NO_TERMINAL` — nobody closed the turn, so no other + field can be trusted to describe a complete picture. + 2. :data:`EMPTY_CAUSE_SYNTHETIC` — the provider layer closed it, so the + emptiness is ours, not the model's. + 3. :data:`EMPTY_CAUSE_VISIBLE_PARTIAL` — the user read an answer that a tool + boundary flushed away; the turn is not empty in any sense the user would + recognise. + 4. :data:`EMPTY_CAUSE_TOOL_ONLY` / :data:`EMPTY_CAUSE_THINKING_ONLY` — work + happened with nothing said. + 5. :data:`EMPTY_CAUSE_PROVIDER_EMPTY` — a clean ``end_turn`` with no + activity at all. The genuine provider-side empty, and the only cause for + which replaying the original message is the right recovery. + 6. :data:`EMPTY_CAUSE_OTHER` — a closed terminal with no activity and no + clean ``end_turn``, most importantly an OMITTED stop reason. Distinct + from ``provider_empty`` on purpose: the incident's third attempt looked + identical to a provider empty in the log and was not diagnosable. + """ + if not activity.saw_terminal: + return EMPTY_CAUSE_NO_TERMINAL + if activity.terminal_synthetic: + return EMPTY_CAUSE_SYNTHETIC + if activity.flushed_visible: + return EMPTY_CAUSE_VISIBLE_PARTIAL + if activity.had_tools: + return EMPTY_CAUSE_TOOL_ONLY + if activity.had_thinking: + return EMPTY_CAUSE_THINKING_ONLY + if activity.stop_reason == _STOP_END_TURN: + return EMPTY_CAUSE_PROVIDER_EMPTY + return EMPTY_CAUSE_OTHER + + def should_recover_promise_only( *, stop_reason: str, diff --git a/test/test_acp_runtime.py b/test/test_acp_runtime.py index febe26606db..775343e7797 100644 --- a/test/test_acp_runtime.py +++ b/test/test_acp_runtime.py @@ -7817,6 +7817,112 @@ async def test_handle_owned_rejections_are_sel_audited(): assert audited[0][2] == "stranded_request_pre_turn_drain" +@pytest.mark.asyncio +async def test_pre_turn_drain_counts_discarded_frames_without_logging_content(caplog): + """The pre-turn drain destroys leftover frames; it must SAY how many, and + nothing else. + + A dropped frame is invisible to every layer above: the abandoned turn's + output vanishes with nothing to show it existed, and a turn that loses its + terminal this way reaches the dashboard as an empty response with no + attributable cause. So the drain reports a COUNT — and only a count. Frame + text, tool arguments, tool results and even frame SIZE are all excluded: + a size leaks response length, and the kiro-cli data dir this material comes + from is fenced precisely because it holds credentials. + """ + import contextlib + import logging + + rt, _, _ = _make_runtime() + q = _register(rt, "sA") + handle = AcpSessionHandle("sA", q["sA"], rt) + rt.send_request = AsyncMock(return_value=9) + + # Three leftover NOTIFICATIONS from a prior abandoned turn. Not permission + # requests: those are answered rather than dropped, and must not be counted. + for _i, _secret in enumerate(("SECRETALPHA", "SECRETBETA", "SECRETGAMMA")): + q["sA"].put_nowait( + JsonRpcMessage.from_dict( + { + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "sA", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": _secret}, + }, + }, + } + ) + ) + + with caplog.at_level(logging.WARNING, logger="kiro_crew.acp.session_handle"): + gen = handle.prompt("hi", timeout=0.2) + with contextlib.suppress(StopAsyncIteration, asyncio.TimeoutError, Exception): + await asyncio.wait_for(gen.__anext__(), timeout=1.0) + await gen.aclose() + + _drain_lines = [ + rec.getMessage() for rec in caplog.records if "pre-turn drain discarded" in rec.getMessage() + ] + # ONE line per turn, not one per frame: a burst must not flood the log. + assert len(_drain_lines) == 1, f"expected one drain warning, got {_drain_lines}" + assert "3 leftover frame(s)" in _drain_lines[0] + for _secret in ("SECRETALPHA", "SECRETBETA", "SECRETGAMMA"): + assert _secret not in _drain_lines[0], f"{_secret!r} leaked into the drain warning" + + +@pytest.mark.asyncio +async def test_prompt_warns_when_the_stream_ends_without_a_terminal_event(caplog): + """A clean exhaustion with no EVENT_COMPLETE is reported; a consumer close is not. + + The dashboard reads a missing terminal as an empty response and cannot tell + whether the backend never closed the turn or the consumer walked away. Only + the first is a fault, so only the first warns — which is what keeps this line + out of the log on every ordinary cancel and every abandoned generator. + """ + import contextlib + import logging + + from kiro_crew.acp.types import EVENT_TEXT_CHUNK, AcpEvent + + _MARK = "ended without a terminal completion event" + + # 1. The dispatch loop returns of its own accord having yielded no terminal. + rt, _, _ = _make_runtime() + q = _register(rt, "sA") + handle = AcpSessionHandle("sA", q["sA"], rt) + rt.send_request = AsyncMock(return_value=9) + + async def _no_terminal(*a, **kw): + yield AcpEvent(kind=EVENT_TEXT_CHUNK, text="partial") + + handle._dispatch_events = _no_terminal # type: ignore[method-assign] + with caplog.at_level(logging.WARNING, logger="kiro_crew.acp.session_handle"): + async for _ev in handle.prompt("hi", timeout=0.2): + pass + assert any(_MARK in rec.getMessage() for rec in caplog.records) + + # 2. The SAME stream, abandoned by the consumer after one event. Identical + # absence of a terminal, but the consumer caused it — no warning. + caplog.clear() + rt2, _, _ = _make_runtime() + q2 = _register(rt2, "sB") + handle2 = AcpSessionHandle("sB", q2["sB"], rt2) + rt2.send_request = AsyncMock(return_value=9) + handle2._dispatch_events = _no_terminal # type: ignore[method-assign] + with caplog.at_level(logging.WARNING, logger="kiro_crew.acp.session_handle"): + gen = handle2.prompt("hi", timeout=0.2) + with contextlib.suppress(StopAsyncIteration): + await gen.__anext__() + await gen.aclose() + assert not any(_MARK in rec.getMessage() for rec in caplog.records), ( + "a consumer close was reported as a lost terminal — this is the log-spam " + "case the guard exists to exclude" + ) + + def test_missing_kind_is_not_a_resolved_shell_classification(): """A tool_call whose `kind` never arrived must NOT cache a shell classification: the miss-default False would otherwise read as a RESOLVED diff --git a/test/test_dashboard_chat.py b/test/test_dashboard_chat.py index 4e9c36ec5ac..b5419656ae1 100644 --- a/test/test_dashboard_chat.py +++ b/test/test_dashboard_chat.py @@ -4,6 +4,7 @@ import asyncio import json +import logging import os import re import threading @@ -16101,6 +16102,22 @@ async def _stream(msg): mock_client.stream = _stream mock_client.stream_command = _stream + @staticmethod + async def _cancel_background_tasks(state) -> None: + """Cancel AND await every task this isolated state spawned. + + `_run_chat` starts title/summary work that is irrelevant to these + recovery assertions. A bare `task.cancel()` leaves the coroutine pending + until the loop gets another tick, which surfaces as unawaited-coroutine + and destroyed-pending-task warnings under xdist. Await the cancellation + exactly; never sleep and never let one test's teardown spill into another. + """ + tasks = list(state._background_tasks) + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + @pytest.mark.asyncio async def test_first_empty_response_requeues_message(self, tmp_path: Path) -> None: """First empty response at depth 0 → message re-queued silently.""" @@ -16215,8 +16232,7 @@ def spy(self_slot, *a, **kw): with patch.object(_ChatSlot, "queue_insert", spy): await _run_chat(state, slot, "test message") - for _bg_task in list(state._background_tasks): - _bg_task.cancel() + await self._cancel_background_tasks(state) # The nudge (NOT the original message) is queued at the front. assert (0, _EMPTY_AUTO_CONTINUE_MSG) in calls @@ -16375,6 +16391,472 @@ async def _stream(msg): assert not any("returned nothing this turn" in m.get("content", "") for m in notice_msgs) +class TestProductiveTurnNeverReplaysVerbatim: + """A turn that DID work must never have the user's message replayed verbatim. + + ``assistant_text`` is reset at every tool boundary, so a turn that streamed a + preamble and then called a tool arrives at the terminal chain with an empty + final segment and takes the empty-response branch. Rung 1 of that ladder + re-queues the ORIGINAL message, which re-runs every tool call that already + completed — a second ``send_message``, a second write, a second PR — and + re-derives an answer the user has already read. + + The field incident these tests pin: two consecutive billed turns, each with an + assistant preamble and successful tool calls and each ending on a clean + ``end_turn``, were both classified empty; the first verbatim-replayed the + user's message. + + Reuses ``TestEmptyResponseRetry``'s harness rather than a second one, so a + change to how a slot is built cannot leave these tests driving a different + runner than the ladder tests beside them. + """ + + _make_state_and_slot = TestEmptyResponseRetry._make_state_and_slot + + @staticmethod + def _spy_queue(monkeypatch=None): + """Record ``(index, content)`` for every queue insert.""" + calls: list[tuple] = [] + orig = _ChatSlot.queue_insert + + def spy(self_slot, *a, **kw): + calls.append(a) + return orig(self_slot, *a, **kw) + + return calls, spy + + _cancel_background_tasks = staticmethod(TestEmptyResponseRetry._cancel_background_tasks) + + @staticmethod + def _text_then_tool_stream(client, executed): + """TEXT -> TOOL_CALL -> TOOL_RESULT -> COMPLETE(end_turn). + + ``executed`` counts dispatches so a replay of the original message would + be observable as a second execution rather than only as a queue entry. + """ + from kiro_crew.acp.types import STOP_REASON_END_TURN + from kiro_crew.providers.base import ( + EVENT_COMPLETE, + EVENT_TEXT_CHUNK, + EVENT_TOOL_CALL, + EVENT_TOOL_RESULT, + LLMEvent, + ) + + async def _stream(msg): + yield LLMEvent(kind=EVENT_TEXT_CHUNK, text="checking") + executed.append(msg) + yield LLMEvent( + kind=EVENT_TOOL_CALL, + tool_call_id="tc-1", + title="send_message", + ) + yield LLMEvent(kind=EVENT_TOOL_RESULT, tool_call_id="tc-1", text="sent") + yield LLMEvent(kind=EVENT_COMPLETE, stop_reason=STOP_REASON_END_TURN) + + client.stream = _stream + client.stream_command = _stream + + @staticmethod + def _tool_only_stream(client): + from kiro_crew.acp.types import STOP_REASON_END_TURN + from kiro_crew.providers.base import ( + EVENT_COMPLETE, + EVENT_TOOL_CALL, + EVENT_TOOL_RESULT, + LLMEvent, + ) + + async def _stream(msg): + yield LLMEvent(kind=EVENT_TOOL_CALL, tool_call_id="tc-1", title="send_message") + yield LLMEvent(kind=EVENT_TOOL_RESULT, tool_call_id="tc-1", text="sent") + yield LLMEvent(kind=EVENT_COMPLETE, stop_reason=STOP_REASON_END_TURN) + + client.stream = _stream + client.stream_command = _stream + + @staticmethod + def _thinking_only_stream(client): + from kiro_crew.acp.types import STOP_REASON_END_TURN + from kiro_crew.providers.base import EVENT_COMPLETE, EVENT_THINKING_CHUNK, LLMEvent + + async def _stream(msg): + yield LLMEvent(kind=EVENT_THINKING_CHUNK, text="hmm") + yield LLMEvent(kind=EVENT_COMPLETE, stop_reason=STOP_REASON_END_TURN) + + client.stream = _stream + client.stream_command = _stream + + @pytest.mark.asyncio + async def test_text_then_tool_turn_continues_instead_of_replaying(self, tmp_path: Path) -> None: + """The incident's own shape: preamble + a completed tool call, clean + end_turn. The original message must NOT be re-queued; exactly one + synthetic continuation may be, and the completed tool must not re-run.""" + from kiro_crew.dashboard.chat_utils import ( + _ACTIVITY_NO_REPLY_CONTINUE_MSG, + _EMPTY_AUTO_CONTINUE_MSG, + ) + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + executed: list[str] = [] + self._text_then_tool_stream(client, executed) + calls, spy = self._spy_queue() + + with patch.object(_ChatSlot, "queue_insert", spy): + await _run_chat(state, slot, "send the summary") + await self._cancel_background_tasks(state) + + # The verbatim replay is the defect. It must be absent. + assert (0, "send the summary") not in calls, ( + "the ORIGINAL user message was re-queued after a turn that already " + "ran a tool — the replay re-executes completed side effects" + ) + # Exactly one continuation, and it is the one whose body does not claim + # the turn produced nothing (which would invite the model to redo the + # completed call). + _continuations = [c for c in calls if c and c[1] == _ACTIVITY_NO_REPLY_CONTINUE_MSG] + assert len(_continuations) == 1, f"expected one continuation, got {calls}" + # The productive turn skipped rung 1 (verbatim replay), so this one + # continuation consumes the whole bounded ladder. If the continuation + # itself returns empty, it gives up rather than enqueueing a SECOND + # continuation whose wording would again invite rework. + assert slot._empty_response_retries == 2 + assert (0, _EMPTY_AUTO_CONTINUE_MSG) not in calls, ( + "the empty-response body tells the model its turn produced no " + "output, which is false for a turn that ran a tool" + ) + # The tool ran once. The queue was not drained in this test, so a second + # execution could only come from a replay dispatched inside this turn. + assert executed == ["send the summary"], f"tool dispatched more than once: {executed}" + + @pytest.mark.asyncio + async def test_tool_only_turn_does_not_replay_verbatim(self, tmp_path: Path) -> None: + """A turn with tool calls and no text at all still counts as productive.""" + from kiro_crew.dashboard.chat_utils import _ACTIVITY_NO_REPLY_CONTINUE_MSG + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + self._tool_only_stream(client) + calls, spy = self._spy_queue() + + with patch.object(_ChatSlot, "queue_insert", spy): + await _run_chat(state, slot, "post it") + await self._cancel_background_tasks(state) + + assert (0, "post it") not in calls, "tool-only turn verbatim-replayed the prompt" + assert (0, _ACTIVITY_NO_REPLY_CONTINUE_MSG) in calls + + @pytest.mark.asyncio + async def test_thinking_only_turn_does_not_replay_verbatim(self, tmp_path: Path) -> None: + """Thinking is work the replay would discard and re-charge for.""" + from kiro_crew.dashboard.chat_utils import _ACTIVITY_NO_REPLY_CONTINUE_MSG + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + self._thinking_only_stream(client) + calls, spy = self._spy_queue() + + with patch.object(_ChatSlot, "queue_insert", spy): + await _run_chat(state, slot, "think about it") + await self._cancel_background_tasks(state) + + assert (0, "think about it") not in calls, "thinking-only turn verbatim-replayed the prompt" + assert (0, _ACTIVITY_NO_REPLY_CONTINUE_MSG) in calls + + @pytest.mark.asyncio + async def test_activity_free_turn_keeps_the_verbatim_replay_rung(self, tmp_path: Path) -> None: + """The guard is scoped: a GENUINELY activity-free empty turn keeps rung 1. + + Without this, the fix could be 'never replay', which would silently + remove the self-heal that a real provider-side empty depends on. + """ + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + TestEmptyResponseRetry._make_empty_stream(self, client) + calls, spy = self._spy_queue() + + with ( + patch.object(_ChatSlot, "queue_insert", spy), + patch( + "kiro_crew.dashboard.chat_runner._start_next_queued_turn", + new=AsyncMock(return_value=False), + ), + ): + await _run_chat(state, slot, "test message") + await self._cancel_background_tasks(state) + + assert (0, "test message") in calls, "the activity-free replay rung was removed" + assert slot._empty_response_retries == 1 + + +class TestEmptyTurnDiagnostics: + """The empty-response WARNING must name a CLOSED cause, and carry no content. + + The incident it exists for had three attempts with three different causes and + one log line that could not tell them apart. These tests assert on the closed + vocabulary only — never on a prompt, a response, a tool argument, a path, an + identity, a token count or a cost, because asserting on those would pin + exactly the leak the diagnostics are designed to avoid. + """ + + _make_state_and_slot = TestEmptyResponseRetry._make_state_and_slot + _cancel_background_tasks = staticmethod(TestEmptyResponseRetry._cancel_background_tasks) + + @staticmethod + def _causes(caplog): + """The ``cause=`` values of every empty-response warning captured.""" + out = [] + for rec in caplog.records: + msg = rec.getMessage() + if "Empty model response" not in msg: + continue + for field in msg.split(): + if field.startswith("cause="): + out.append(field.split("=", 1)[1]) + return out + + @staticmethod + def _fields(caplog): + """``key=value`` pairs of the first empty-response warning captured.""" + for rec in caplog.records: + msg = rec.getMessage() + if "Empty model response" not in msg: + continue + return dict( + tuple(f.split("=", 1)) for f in msg.split() if "=" in f and not f.startswith("%") + ) + return {} + + def test_the_cause_and_rung_vocabularies_are_closed(self) -> None: + """Pure-unit: every cause the classifier can return, from its own inputs. + + Kept as a unit test over ``classify_empty_turn`` rather than seven + ``_run_chat`` drives: the ranking between overlapping causes is the part + that can regress silently, and a stream fixture cannot express + 'synthesized terminal AND tool activity' as cleanly as the snapshot can. + """ + from kiro_crew.dashboard.chat_utils import ( + EMPTY_CAUSE_NO_TERMINAL, + EMPTY_CAUSE_OTHER, + EMPTY_CAUSE_PROVIDER_EMPTY, + EMPTY_CAUSE_SYNTHETIC, + EMPTY_CAUSE_THINKING_ONLY, + EMPTY_CAUSE_TOOL_ONLY, + EMPTY_CAUSE_VISIBLE_PARTIAL, + STOP_REASON_ABSENT, + EmptyTurnActivity, + classify_empty_turn, + ) + + # No terminal outranks everything, including activity. + assert ( + classify_empty_turn(EmptyTurnActivity(saw_terminal=False, had_tools=True)) + == EMPTY_CAUSE_NO_TERMINAL + ) + # A synthesized terminal outranks activity: the emptiness is ours. + assert ( + classify_empty_turn( + EmptyTurnActivity(saw_terminal=True, terminal_synthetic=True, had_tools=True) + ) + == EMPTY_CAUSE_SYNTHETIC + ) + # A flushed visible segment outranks tools: the user read an answer. + assert ( + classify_empty_turn( + EmptyTurnActivity(saw_terminal=True, flushed_visible=True, had_tools=True) + ) + == EMPTY_CAUSE_VISIBLE_PARTIAL + ) + assert ( + classify_empty_turn(EmptyTurnActivity(saw_terminal=True, had_tools=True)) + == EMPTY_CAUSE_TOOL_ONLY + ) + assert ( + classify_empty_turn(EmptyTurnActivity(saw_terminal=True, had_thinking=True)) + == EMPTY_CAUSE_THINKING_ONLY + ) + # A clean end_turn with nothing at all is the only genuine provider empty. + assert ( + classify_empty_turn(EmptyTurnActivity(saw_terminal=True, stop_reason="end_turn")) + == EMPTY_CAUSE_PROVIDER_EMPTY + ) + # An OMITTED stop reason is NOT a provider empty — the distinction the + # incident's third attempt needed and did not have. + assert ( + classify_empty_turn( + EmptyTurnActivity(saw_terminal=True, stop_reason=STOP_REASON_ABSENT) + ) + == EMPTY_CAUSE_OTHER + ) + + def test_normalize_stop_reason_never_returns_a_raw_backend_string(self) -> None: + """Closed output set, and the ACP spellings it mirrors are pinned. + + ``chat_utils`` may not import ``kiro_crew.acp.types`` (the agent-SDK + boundary gate baselines it at one edge and a baselined file may not grow), + so it spells the three stop reasons as literals. The test tree is outside + that gate, so the pin lives here — a change to the backend's vocabulary + reddens here instead of silently reclassifying every turn as ``other``. + """ + from kiro_crew.acp.types import ( + STOP_REASON_CANCELLED, + STOP_REASON_END_TURN, + STOP_REASON_REFUSAL, + ) + from kiro_crew.dashboard.chat_utils import ( + _STOP_CANCELLED_REASON, + _STOP_END_TURN, + _STOP_REFUSAL, + STOP_REASON_ABSENT, + STOP_REASON_OTHER, + normalize_stop_reason, + ) + + assert _STOP_END_TURN == STOP_REASON_END_TURN + assert _STOP_CANCELLED_REASON == STOP_REASON_CANCELLED + assert _STOP_REFUSAL == STOP_REASON_REFUSAL + + assert normalize_stop_reason(None) == STOP_REASON_ABSENT + assert normalize_stop_reason("") == STOP_REASON_ABSENT + assert normalize_stop_reason(STOP_REASON_END_TURN) == STOP_REASON_END_TURN + assert normalize_stop_reason("error: tool stall") == "error" + # An invented backend reason is folded, never echoed. + _invented = "the model said stop because of /home/someone/secret.txt" + assert normalize_stop_reason(_invented) == STOP_REASON_OTHER + + @pytest.mark.asyncio + async def test_provider_empty_and_no_terminal_event_report_distinct_causes( + self, tmp_path: Path, caplog + ) -> None: + """Two turns that look identical to the old log line must not any more.""" + from kiro_crew.acp.types import STOP_REASON_END_TURN + from kiro_crew.dashboard.chat_utils import ( + EMPTY_CAUSE_NO_TERMINAL, + EMPTY_CAUSE_PROVIDER_EMPTY, + ) + from kiro_crew.providers.base import EVENT_COMPLETE, LLMEvent + + # A clean end_turn with nothing in it. + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + + async def _clean(msg): + yield LLMEvent(kind=EVENT_COMPLETE, stop_reason=STOP_REASON_END_TURN) + + client.stream = _clean + client.stream_command = _clean + with caplog.at_level(logging.WARNING, logger="kiro_crew.dashboard.chat_runner"): + await _run_chat(state, slot, "a", _prompt_depth=1) + await self._cancel_background_tasks(state) + assert self._causes(caplog) == [EMPTY_CAUSE_PROVIDER_EMPTY] + + # A stream that ends without any terminal event at all. + caplog.clear() + state2, slot2, client2, _run_chat2 = self._make_state_and_slot(tmp_path) + + async def _no_terminal(msg): + return + yield # pragma: no cover -- makes this an async generator + + client2.stream = _no_terminal + client2.stream_command = _no_terminal + with caplog.at_level(logging.WARNING, logger="kiro_crew.dashboard.chat_runner"): + await _run_chat2(state2, slot2, "a", _prompt_depth=1) + await self._cancel_background_tasks(state2) + assert self._causes(caplog) == [EMPTY_CAUSE_NO_TERMINAL] + + @pytest.mark.asyncio + async def test_omitted_stop_reason_is_not_reported_as_provider_empty( + self, tmp_path: Path, caplog + ) -> None: + """A terminal with no stopReason is its own observation.""" + from kiro_crew.dashboard.chat_utils import ( + EMPTY_CAUSE_OTHER, + STOP_REASON_ABSENT, + ) + from kiro_crew.providers.base import EVENT_COMPLETE, LLMEvent + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + + async def _stream(msg): + yield LLMEvent(kind=EVENT_COMPLETE) # no stop_reason + + client.stream = _stream + client.stream_command = _stream + with caplog.at_level(logging.WARNING, logger="kiro_crew.dashboard.chat_runner"): + await _run_chat(state, slot, "a", _prompt_depth=1) + await self._cancel_background_tasks(state) + + assert self._causes(caplog) == [EMPTY_CAUSE_OTHER] + assert self._fields(caplog).get("stop_reason") == STOP_REASON_ABSENT + + @pytest.mark.asyncio + async def test_productive_turn_reports_visible_partial_and_the_continue_rung( + self, tmp_path: Path, caplog + ) -> None: + """The incident's shape gets its own cause AND the rung it took. + + Asserting the rung is what makes the diagnostic answer 'what did the + runner DO', not only 'what did it see'. + """ + from kiro_crew.dashboard.chat_utils import ( + EMPTY_CAUSE_VISIBLE_PARTIAL, + EMPTY_RUNG_CONTINUE, + ) + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + TestProductiveTurnNeverReplaysVerbatim._text_then_tool_stream(client, []) + + with caplog.at_level(logging.WARNING, logger="kiro_crew.dashboard.chat_runner"): + await _run_chat(state, slot, "send the summary") + await self._cancel_background_tasks(state) + + assert self._causes(caplog) == [EMPTY_CAUSE_VISIBLE_PARTIAL] + _fields = self._fields(caplog) + assert _fields.get("rung") == EMPTY_RUNG_CONTINUE + assert _fields.get("flushed_visible") == "True" + assert _fields.get("tools") == "True" + + @pytest.mark.asyncio + async def test_the_warning_carries_no_content_and_no_billing_amounts( + self, tmp_path: Path, caplog + ) -> None: + """The privacy floor, asserted against the rendered line. + + Every diagnostic value is a bool or a closed constant, so the prompt text, + the streamed text, the tool title and the billed amounts must be absent + from the line even though the turn carried all of them. + """ + from kiro_crew.acp.types import STOP_REASON_END_TURN, TurnUsage + from kiro_crew.providers.base import ( + EVENT_COMPLETE, + EVENT_TEXT_CHUNK, + EVENT_TOOL_CALL, + LLMEvent, + ) + + state, slot, client, _run_chat = self._make_state_and_slot(tmp_path) + + async def _stream(msg): + yield LLMEvent(kind=EVENT_TEXT_CHUNK, text="SECRETPREAMBLE") + yield LLMEvent(kind=EVENT_TOOL_CALL, tool_call_id="tc-1", title="SECRETTOOL") + yield LLMEvent( + kind=EVENT_COMPLETE, + stop_reason=STOP_REASON_END_TURN, + usage=TurnUsage(credits=4.25, input_tokens=1234, output_tokens=99), + ) + + client.stream = _stream + client.stream_command = _stream + with caplog.at_level(logging.WARNING, logger="kiro_crew.dashboard.chat_runner"): + await _run_chat(state, slot, "SECRETPROMPT") + await self._cancel_background_tasks(state) + + _line = next( + rec.getMessage() for rec in caplog.records if "Empty model response" in rec.getMessage() + ) + for forbidden in ("SECRETPROMPT", "SECRETPREAMBLE", "SECRETTOOL", "4.25", "1234", "99"): + assert forbidden not in _line, f"{forbidden!r} leaked into {_line!r}" + # Billing presence IS reported — as a bool, which is the point. + assert "billed=True" in _line + + class TestExpandDollarSkills: """Runner-side ``_expand_dollar_skills``: redaction, chip, SEL audit, and empty/exception branches. The pure resolution logic is diff --git a/test/test_subagent_delivery_ttl_anchor.py b/test/test_subagent_delivery_ttl_anchor.py index 789cfeeb9ae..279788f0125 100644 --- a/test/test_subagent_delivery_ttl_anchor.py +++ b/test/test_subagent_delivery_ttl_anchor.py @@ -315,7 +315,10 @@ def test_run_chat_reports_consumption_through_the_callers_hook(self): # The retraction lives in the FIRST empty-response branch and happens # BEFORE the verbatim re-queue copies the callback. Reversing that order # drops the callback and strands the delivery after a successful replay. - first_empty_at = src.index("if _prompt_depth == 0 and slot._empty_response_retries < 1:") + # Anchored on the rung marker rather than the branch condition: that + # condition carries the productive-turn guard and is reformatted whenever + # it grows a term, while the marker names the rung this invariant is about. + first_empty_at = src.index("_empty_rung = EMPTY_RUNG_REPLAY") first_empty_end = src.index(" elif (", first_empty_at) first_empty = src[first_empty_at:first_empty_end] assert first_empty.count("await _report_consumed(False)") == 1