fix(desktop, ink): don't wipe messages before final message (#65919)
* fix(desktop): preserve interim assistant text wiped at message.complete
When the agent emits interim text (commentary alongside tool calls, or the
attempted final answer before a verify-on-stop nudge), all UI surfaces
streamed it live but then wiped it at message.complete — keeping only the
final response. The user saw text appear during inference, then disappear.
This is the complete fix across all three layers: agent core, gateway
transport, and all UI surfaces (desktop + Ink TUI).
The verify-on-stop and pre_verify paths flagged the assistant's attempted
final answer as _verification_stop_synthetic, suppressing it from both
state.db and the UI. The user only saw the terse post-verification reply.
Now the assistant response is real content: it's persisted to state.db and
emitted as an interim message via _emit_interim_assistant_message(force_display=True)
before the verification loop runs. Only the synthetic nudge messages keep
the synthetic flags. The turn finalizer drops nudges from live history and
compares content (not just role) to avoid duplicating a published candidate.
Message sequence repair collapses verification candidates in the
consecutive-assistant merge.
Wire agent.interim_assistant_callback both at construction (_agent_cbs())
and per-turn (defense-in-depth), emitting a new message.interim event with
{text, already_streamed}. Gated on display.interim_assistant_messages
(default true). Cleared in the finally block so a stale closure can't
fire on a later turn.
Add message.interim to the GatewayEventName union (apps/shared) and a
typed payload to the TUI's GatewayEvent discriminated union.
The TUI already had the segment-anchoring machinery (flushStreamingSegment +
finalTail) but had no handler for message.interim. Added recordInterimMessage
+ interimBoundaryIndex to seal segments mid-turn, and updated
recordMessageComplete to only dedupe segments after the interim boundary.
Replaced the fragile sealed-set approach with a proper interimBoundaryPending
state flag on ClientSessionState. finalizeInterimAssistantMessage finalizes
the streaming bubble in place (or creates a standalone one), rotates the
stream ID so next deltas create a new bubble, and sets the flag. When the
final text equals an already-sealed interim, they stay as distinct messages.
Extracted mergeFinalAssistantText() as a pure function in chat-messages.ts,
used by both completeAssistantMessage and finalizeInterimAssistantMessage.
Split the bidirectional dedup predicate: reasoning is a restatement only when
the final FULLY covers it. A short final ("Done.") no longer swallows a
longer reasoning block that merely starts with it.
Honor display.interim_assistant_messages (default true) across all layers:
the tui_gateway gates the callback, the desktop wires it to a nanostores
atom via use-hermes-config. Updated hermes_cli/config.py and
cli-config.yaml.example comments to document the Desktop behavior.
_split_segment_tokens now accepts posix=False and _find_ad_hoc_match tries
both posix modes so ad-hoc verification scripts with Windows backslash
paths are matched correctly. (response_previewed forwarding from #53553
is not included — our emit-interim + persist approach makes it unnecessary
since the attempted answer is now surfaced before the verification loop.)
- tsc: clean (desktop + TUI + shared)
- vitest desktop: 73/73 pass (7 interim-sealing + 5 mergeFinalAssistantText + 4 config atom)
- vitest TUI: 83/83 pass (4 new message.interim tests)
- python: 390 tests pass (340 tui_gateway + 33 verification/finalizer + 6 config gating + 3 evidence + 8 continuation budget)
Co-authored-by: Liam Zhang <yingliang-zhang@users.noreply.github.com>
Co-authored-by: Lucas D'Alessandro <lucasfdale@users.noreply.github.com>
Co-authored-by: Eric Manganaro <superposition@users.noreply.github.com>
Co-authored-by: sweetcornna <sweetcornna@users.noreply.github.com>
Co-authored-by: DECK6 <DECK6@users.noreply.github.com>
Co-authored-by: matantsevs <matantsevs@users.noreply.github.com>
Co-authored-by: gitcommit90 <gitcommit90@users.noreply.github.com>
* fix: prefix-match interim streamed content to avoid benign duplicate bubbles
_interim_content_was_streamed used exact equality (streamed == visible_content),
so a final response that was the streamed text plus a trailing delta — or a
partial stream before the verify nudge fired — failed the match and left
_response_was_previewed false. The turn then showed two bubbles (interim +
identical final) instead of settling the interim in place.
Relax to a prefix check (visible_content.startswith(streamed)) in both the
core match and the desktop's settle-in-place gate. The TUI already used
prefix matching via finalTail. The reverse direction (streamed longer than
final) is intentionally not matched — that could suppress a needed resend
in the gateway path where already_streamed=True calls on_segment_break().
* test(desktop): add partial-stream-then-nudge dedup edge case
Third edge case for the interim-sealing dedup: model streams part of its
answer via message.delta, verify nudge fires, interim seals the streamed
prefix, then the final response is the same text plus a trailing delta.
Asserts one bubble (not two) containing the full final text.
Acceptance protocol #2 — covers all three dedup edges:
1. interim == final (existing)
2. interim = strict prefix of final (existing)
3. partial-stream-then-nudge (this commit)
---------
Co-authored-by: Liam Zhang <yingliang-zhang@users.noreply.github.com>
Co-authored-by: Lucas D'Alessandro <lucasfdale@users.noreply.github.com>
Co-authored-by: Eric Manganaro <superposition@users.noreply.github.com>
Co-authored-by: sweetcornna <sweetcornna@users.noreply.github.com>
Co-authored-by: DECK6 <DECK6@users.noreply.github.com>
Co-authored-by: matantsevs <matantsevs@users.noreply.github.com>
Co-authored-by: gitcommit90 <gitcommit90@users.noreply.github.com>
This commit is contained in:
ethernet
GitHub
parent
b520f507cc
commit
c363db81e0
|
|
@ -530,6 +530,12 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
|
|||
or m.get("finish_reason") == "incomplete"
|
||||
)
|
||||
|
||||
def _is_verification_candidate(m: Dict) -> bool:
|
||||
return m.get("finish_reason") in {
|
||||
"verification_required",
|
||||
"verify_hook_continue",
|
||||
}
|
||||
|
||||
collapsed: List[Dict] = []
|
||||
for msg in messages:
|
||||
if (
|
||||
|
|
@ -542,6 +548,16 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
|
|||
and not _is_codex_interim(collapsed[-1])
|
||||
):
|
||||
prev = collapsed[-1]
|
||||
# Verification candidate collapsing: when the earlier assistant
|
||||
# message is a provisional candidate (finish_reason =
|
||||
# verification_required / verify_hook_continue), the later
|
||||
# response supersedes it for model replay — replace rather than
|
||||
# union. Both remain durable in state.db; this only affects the
|
||||
# in-memory sequence sent to the model. (#65919 §7)
|
||||
if _is_verification_candidate(prev):
|
||||
collapsed[-1] = msg
|
||||
repairs += 1
|
||||
continue
|
||||
# Union tool_calls (preserve order, both may carry them).
|
||||
prev_calls = list(prev.get("tool_calls") or [])
|
||||
new_calls = list(msg.get("tool_calls") or [])
|
||||
|
|
|
|||
|
|
@ -690,6 +690,12 @@ def run_conversation(
|
|||
# user-facing result available; it must not be confused with error or
|
||||
# recovery text produced by unrelated exit paths.
|
||||
_pending_verification_response = None
|
||||
# Tracks whether the pending verification candidate was already streamed
|
||||
# to the user as interim content. The finalizer uses this to set
|
||||
# ``_response_was_previewed`` ONLY when the pending candidate is actually
|
||||
# reused as the final response — not merely because any interim was
|
||||
# streamed. (#65919 review: response-loss blocker)
|
||||
_pending_verification_response_previewed = False
|
||||
|
||||
# Per-turn tally of consecutive successful credential-pool token refreshes,
|
||||
# keyed by (provider, pool-entry-id). A persistent upstream 401 lets
|
||||
|
|
@ -5522,17 +5528,17 @@ def run_conversation(
|
|||
getattr(agent, "_verification_stop_nudges", 0) + 1
|
||||
)
|
||||
final_msg["finish_reason"] = "verification_required"
|
||||
final_msg["_verification_stop_synthetic"] = True
|
||||
# The assistant response is real content — persist it and
|
||||
# emit to the UI as an interim message so the user sees the
|
||||
# attempted final answer before the verification loop runs.
|
||||
# Only the nudge is flagged synthetic so it gets stripped
|
||||
# from the durable transcript (#65919 §7).
|
||||
agent._emit_interim_assistant_message(final_msg)
|
||||
messages.append(final_msg)
|
||||
# Keep the attempted final answer in model history so the
|
||||
# synthetic user nudge preserves role alternation, but do
|
||||
# not surface it to the user as an interim answer. The
|
||||
# whole point of this guard is to prevent premature
|
||||
# "done" claims before checks run. Both the attempted
|
||||
# answer and the nudge are flagged synthetic so neither
|
||||
# persists — otherwise the resumed transcript keeps a
|
||||
# premature "done" with the nudge stripped, producing an
|
||||
# assistant→assistant adjacency. (#55733)
|
||||
try:
|
||||
agent._flush_messages_to_session_db(messages, conversation_history)
|
||||
except Exception:
|
||||
logger.debug("verify-on-stop interim flush failed", exc_info=True)
|
||||
messages.append({
|
||||
"role": "user",
|
||||
"content": _verify_nudge,
|
||||
|
|
@ -5548,7 +5554,13 @@ def run_conversation(
|
|||
# continuation-budget exhaustion. ``final_response`` itself
|
||||
# must be cleared so the finalizer can distinguish this gate
|
||||
# from unrelated error/recovery exits. (#61631)
|
||||
# Track whether this candidate was already streamed so the
|
||||
# finalizer can mark the turn previewed only if the
|
||||
# candidate is actually reused as the final response.
|
||||
_pending_verification_response = final_response
|
||||
_pending_verification_response_previewed = (
|
||||
agent._interim_content_was_streamed(final_response or "")
|
||||
)
|
||||
final_response = None
|
||||
continue
|
||||
|
||||
|
|
@ -5587,12 +5599,17 @@ def run_conversation(
|
|||
if _verify_nudge2:
|
||||
agent._pre_verify_nudges = _attempt + 1
|
||||
final_msg["finish_reason"] = "verify_hook_continue"
|
||||
final_msg["_pre_verify_synthetic"] = True
|
||||
# Same alternation contract as verify-on-stop: keep the
|
||||
# attempted answer in history, follow it with a synthetic
|
||||
# user nudge, and don't surface the premature answer. Both
|
||||
# are flagged synthetic so neither persists. (#55733)
|
||||
# The assistant response is real content — persist it and
|
||||
# emit to the UI as an interim message so the user sees the
|
||||
# attempted final answer before the pre_verify loop runs.
|
||||
# Only the nudge is flagged synthetic so it gets stripped
|
||||
# from the durable transcript (#65919 §7).
|
||||
agent._emit_interim_assistant_message(final_msg)
|
||||
messages.append(final_msg)
|
||||
try:
|
||||
agent._flush_messages_to_session_db(messages, conversation_history)
|
||||
except Exception:
|
||||
logger.debug("pre_verify interim flush failed", exc_info=True)
|
||||
messages.append({
|
||||
"role": "user",
|
||||
"content": _verify_nudge2,
|
||||
|
|
@ -5602,6 +5619,9 @@ def run_conversation(
|
|||
logger.debug("pre_verify nudge issued (attempt %d)",
|
||||
agent._pre_verify_nudges)
|
||||
_pending_verification_response = final_response
|
||||
_pending_verification_response_previewed = (
|
||||
agent._interim_content_was_streamed(final_response or "")
|
||||
)
|
||||
final_response = None
|
||||
continue
|
||||
|
||||
|
|
@ -5649,6 +5669,9 @@ def run_conversation(
|
|||
# exhaustion path does not treat the narrated stop as
|
||||
# a completed answer.
|
||||
_pending_verification_response = final_response
|
||||
_pending_verification_response_previewed = (
|
||||
agent._interim_content_was_streamed(final_response or "")
|
||||
)
|
||||
final_response = None
|
||||
continue
|
||||
|
||||
|
|
@ -5773,6 +5796,7 @@ def run_conversation(
|
|||
_should_review_memory=_should_review_memory,
|
||||
_turn_exit_reason=_turn_exit_reason,
|
||||
_pending_verification_response=_pending_verification_response,
|
||||
_pending_verification_response_previewed=_pending_verification_response_previewed,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -42,6 +42,30 @@ def _is_pure_tool_call_tail(msg: dict) -> bool:
|
|||
return not flatten_message_text(msg.get("content")).strip()
|
||||
|
||||
|
||||
# Verification continuation scaffolding flags: verify-on-stop / pre_verify
|
||||
# inject a synthetic user nudge to keep the agent going one more turn.
|
||||
# These nudges must be stripped from returned/live history to avoid
|
||||
# role-alternation breaks and poisoning the resumed transcript. The
|
||||
# assistant response is real content and is not flagged. (#65919 §7)
|
||||
_VERIFICATION_CONTINUATION_FLAGS = (
|
||||
"_verification_stop_synthetic",
|
||||
"_pre_verify_synthetic",
|
||||
)
|
||||
|
||||
|
||||
def _drop_verification_continuation_scaffolding(messages) -> None:
|
||||
"""Remove verification-continuation nudge messages from *messages* in place.
|
||||
|
||||
Only the synthetic nudges carry these flags, so this strips just the
|
||||
nudges while preserving the real attempted-final-answer that was
|
||||
persisted to state.db.
|
||||
"""
|
||||
messages[:] = [
|
||||
m for m in messages
|
||||
if not (isinstance(m, dict) and any(m.get(f) for f in _VERIFICATION_CONTINUATION_FLAGS))
|
||||
]
|
||||
|
||||
|
||||
def finalize_turn(
|
||||
agent,
|
||||
*,
|
||||
|
|
@ -58,6 +82,7 @@ def finalize_turn(
|
|||
_should_review_memory,
|
||||
_turn_exit_reason,
|
||||
_pending_verification_response=None,
|
||||
_pending_verification_response_previewed=False,
|
||||
):
|
||||
"""Run the post-loop finalization and return the turn ``result`` dict.
|
||||
|
||||
|
|
@ -91,6 +116,11 @@ def finalize_turn(
|
|||
# fallible model call. The explicit pending value is the provenance
|
||||
# guard: unrelated error/recovery exits can never enter this branch.
|
||||
final_response = _pending_verification_response
|
||||
# Mark the turn as previewed only when the reused candidate was
|
||||
# actually streamed to the user as interim content. (#65919 review:
|
||||
# response-loss blocker)
|
||||
if _pending_verification_response_previewed:
|
||||
agent._response_was_previewed = True
|
||||
_turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})"
|
||||
iteration_limit_fallback = True
|
||||
preserved_verification_fallback = True
|
||||
|
|
@ -206,6 +236,12 @@ def finalize_turn(
|
|||
try:
|
||||
agent._drop_trailing_empty_response_scaffolding(messages)
|
||||
|
||||
# Drop verification-continuation nudges (synthetic user messages)
|
||||
# from the live history before the tail-assistant check — only the
|
||||
# nudges need stripping; the assistant candidate persists in
|
||||
# state.db. (#65919 §7)
|
||||
_drop_verification_continuation_scaffolding(messages)
|
||||
|
||||
# When the turn was interrupted and the last message is a tool
|
||||
# result, append a synthetic assistant message to close the
|
||||
# tool-call sequence. Without this, the session persists a
|
||||
|
|
@ -235,6 +271,10 @@ def finalize_turn(
|
|||
# single chokepoint every recovery ``break`` flows through, so the
|
||||
# invariant "delivered final_response ⇒ assistant row in transcript"
|
||||
# holds regardless of which path produced it. (#43849 / #44100)
|
||||
#
|
||||
# Compare content (not just role) so a verification candidate that
|
||||
# matches the final response is not duplicated at budget
|
||||
# exhaustion. (#65919 §7)
|
||||
if final_response and not interrupted:
|
||||
try:
|
||||
_tail = messages[-1] if messages else None
|
||||
|
|
@ -242,8 +282,10 @@ def finalize_turn(
|
|||
_tail = None
|
||||
_tail_role = _tail.get("role") if isinstance(_tail, dict) else None
|
||||
if _tail_role != "assistant":
|
||||
# Tail is not an assistant row — append the final response
|
||||
# so the durable turn closes with the answer (#43849/#44100).
|
||||
messages.append({"role": "assistant", "content": final_response})
|
||||
elif isinstance(_tail, dict) and _is_pure_tool_call_tail(_tail):
|
||||
elif isinstance(_tail, dict) and _tail.get("content") != final_response and _is_pure_tool_call_tail(_tail):
|
||||
# The tail IS an assistant row, but a *pure tool-call turn*:
|
||||
# tool_calls with no text of its own. The role check alone
|
||||
# leaves the #43849/#44100 invariant unmet — the user saw a
|
||||
|
|
@ -253,6 +295,11 @@ def finalize_turn(
|
|||
# instead of appending, so the durable turn ends with the answer
|
||||
# without disturbing the tool-call structure or creating an
|
||||
# assistant→assistant pair.
|
||||
#
|
||||
# The ``content != final_response`` guard prevents filling when
|
||||
# the tail already carries the final response text (verification
|
||||
# candidate collapse — the provisional answer was persisted and
|
||||
# reused as the terminal response, #65919 §7).
|
||||
_tail["content"] = final_response
|
||||
# The row may have already been flushed to SQLite by the
|
||||
# incremental tool-call persist (conversation_loop.py:4990),
|
||||
|
|
|
|||
|
|
@ -122,13 +122,13 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
|
|||
conn.commit()
|
||||
|
||||
|
||||
def _split_segment_tokens(command: str) -> list[list[str]]:
|
||||
def _split_segment_tokens(command: str, *, posix: bool = True) -> list[list[str]]:
|
||||
segments: list[list[str]] = []
|
||||
for segment in _SHELL_SPLIT_RE.split(command.strip()):
|
||||
if not segment:
|
||||
continue
|
||||
try:
|
||||
tokens = shlex.split(segment)
|
||||
tokens = shlex.split(segment, posix=posix)
|
||||
except ValueError:
|
||||
continue
|
||||
if tokens:
|
||||
|
|
@ -298,10 +298,13 @@ def _ad_hoc_script_args(tokens: list[str], root: str | Path | None) -> Optional[
|
|||
|
||||
|
||||
def _find_ad_hoc_match(command: str, root: str | Path | None) -> Optional[list[str]]:
|
||||
for tokens in _split_segment_tokens(command):
|
||||
trailing_args = _ad_hoc_script_args(tokens, root)
|
||||
if trailing_args is not None:
|
||||
return trailing_args
|
||||
# Try both posix=True (default) and posix=False (Windows backslash paths)
|
||||
# so ad-hoc verification scripts with backslash paths are matched on Windows.
|
||||
for posix in (True, False):
|
||||
for tokens in _split_segment_tokens(command, posix=posix):
|
||||
trailing_args = _ad_hoc_script_args(tokens, root)
|
||||
if trailing_args is not None:
|
||||
return trailing_args
|
||||
return None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ import { hasSessionInfoStatePatch, sessionInfoStatePatch, SUBAGENT_EVENT_TYPES,
|
|||
|
||||
const COMPACTION_RESUME_EVENT_TYPES = new Set([
|
||||
'message.delta',
|
||||
'message.interim',
|
||||
'thinking.delta',
|
||||
'reasoning.delta',
|
||||
'reasoning.available',
|
||||
|
|
@ -71,9 +72,10 @@ interface GatewayEventDeps {
|
|||
nativeSubagentSessionsRef: MutableRefObject<Set<string>>
|
||||
appendAssistantDelta: (sessionId: string, delta: string) => void
|
||||
appendReasoningDelta: (sessionId: string, delta: string, replace?: boolean) => void
|
||||
completeAssistantMessage: (sessionId: string, text: string) => void
|
||||
completeAssistantMessage: (sessionId: string, text: string, responsePreviewed?: boolean) => void
|
||||
failAssistantMessage: (sessionId: string, errorMessage: string) => void
|
||||
flushQueuedDeltas: (sessionId?: string) => void
|
||||
finalizeInterimAssistantMessage: (sessionId: string, text: string) => void
|
||||
queryClient: QueryClient
|
||||
refreshHermesConfig: () => Promise<void>
|
||||
sessionInterrupted: (sessionId: string) => boolean
|
||||
|
|
@ -103,6 +105,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
|
|||
completeAssistantMessage,
|
||||
failAssistantMessage,
|
||||
flushQueuedDeltas,
|
||||
finalizeInterimAssistantMessage,
|
||||
queryClient,
|
||||
refreshHermesConfig,
|
||||
sessionInterrupted,
|
||||
|
|
@ -386,6 +389,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
|
|||
awaitingResponse: true,
|
||||
sawAssistantPayload: false,
|
||||
interrupted: false,
|
||||
interimBoundaryPending: false,
|
||||
turnStartedAt: Date.now()
|
||||
}
|
||||
})
|
||||
|
|
@ -397,6 +401,18 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
|
|||
if (sessionId) {
|
||||
appendAssistantDelta(sessionId, coerceGatewayText(payload?.text))
|
||||
}
|
||||
} else if (event.type === 'message.interim') {
|
||||
// The agent emitted interim assistant commentary (text alongside tool
|
||||
// calls, or the attempted final answer before a verify-on-stop nudge).
|
||||
// Finalize it as its own sealed bubble so message.complete doesn't wipe
|
||||
// it — the text was already streamed via message.delta and is visible.
|
||||
if (sessionId) {
|
||||
flushQueuedDeltas(sessionId)
|
||||
const text = coerceGatewayText(payload?.text)
|
||||
if (text) {
|
||||
finalizeInterimAssistantMessage(sessionId, text)
|
||||
}
|
||||
}
|
||||
} else if (event.type === 'thinking.delta') {
|
||||
// thinking.delta carries the kawaii spinner status (face + verb from
|
||||
// KawaiiSpinner), not real reasoning. The bottom-of-thread loading
|
||||
|
|
@ -469,7 +485,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
|
|||
playCompletionSound()
|
||||
|
||||
const finalText = coerceGatewayText(payload?.text) || coerceGatewayText(payload?.rendered)
|
||||
completeAssistantMessage(sessionId, finalText)
|
||||
completeAssistantMessage(sessionId, finalText, payload?.response_previewed)
|
||||
|
||||
if (isActiveEvent) {
|
||||
setTurnStartedAt(null)
|
||||
|
|
@ -803,6 +819,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
|
|||
compactedTurnRef,
|
||||
completeAssistantMessage,
|
||||
failAssistantMessage,
|
||||
finalizeInterimAssistantMessage,
|
||||
flushQueuedDeltas,
|
||||
lastCwdInfoSessionRef,
|
||||
nativeSubagentSessionsRef,
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import {
|
|||
type ChatMessagePart,
|
||||
chatMessageText,
|
||||
type GatewayEventPayload,
|
||||
mergeFinalAssistantText,
|
||||
reasoningPart,
|
||||
renderMediaTags,
|
||||
upsertToolPart
|
||||
|
|
@ -370,8 +371,60 @@ export function useMessageStream({
|
|||
[flushQueuedDeltas, mutateStream, sessionInterrupted]
|
||||
)
|
||||
|
||||
const completeAssistantMessage = useCallback(
|
||||
const finalizeInterimAssistantMessage = useCallback(
|
||||
(sessionId: string, text: string) => {
|
||||
updateSessionState(sessionId, state => {
|
||||
if (state.interrupted) {
|
||||
return state
|
||||
}
|
||||
|
||||
const authoritativeText = renderMediaTags(text).trim()
|
||||
if (!authoritativeText) {
|
||||
return state
|
||||
}
|
||||
|
||||
const streamId = state.streamId
|
||||
|
||||
const replaceTextPart = (parts: ChatMessagePart[]) => {
|
||||
const visibleText = stripGeneratedImageEchoes(
|
||||
authoritativeText, generatedImageEchoSources(parts)
|
||||
).trim()
|
||||
return mergeFinalAssistantText(parts, visibleText)
|
||||
}
|
||||
|
||||
let nextMessages = state.messages
|
||||
if (streamId && nextMessages.some(m => m.id === streamId)) {
|
||||
// Finalize the existing streaming bubble in place
|
||||
nextMessages = nextMessages.map(m =>
|
||||
m.id === streamId
|
||||
? { ...m, parts: replaceTextPart(m.parts), pending: false }
|
||||
: m
|
||||
)
|
||||
} else {
|
||||
// No streaming bubble — create a standalone interim message
|
||||
nextMessages = [...nextMessages, {
|
||||
id: `assistant-interim-${Date.now()}`,
|
||||
role: 'assistant' as const,
|
||||
parts: [assistantTextPart(authoritativeText)],
|
||||
pending: false,
|
||||
branchGroupId: state.pendingBranchGroup ?? undefined
|
||||
}]
|
||||
}
|
||||
|
||||
return {
|
||||
...state,
|
||||
messages: nextMessages,
|
||||
streamId: null,
|
||||
interimBoundaryPending: true,
|
||||
sawAssistantPayload: state.sawAssistantPayload || Boolean(authoritativeText)
|
||||
}
|
||||
})
|
||||
},
|
||||
[updateSessionState]
|
||||
)
|
||||
|
||||
const completeAssistantMessage = useCallback(
|
||||
(sessionId: string, text: string, responsePreviewed?: boolean) => {
|
||||
let shouldHydrate = false
|
||||
|
||||
const completedState = updateSessionState(sessionId, state => {
|
||||
|
|
@ -394,27 +447,11 @@ export function useMessageStream({
|
|||
const streamId = state.streamId
|
||||
const finalText = renderMediaTags(text).trim()
|
||||
const completionError = completionErrorText(finalText)
|
||||
const normalize = (value: string) => value.replace(/\s+/g, ' ').trim()
|
||||
const interimBoundaryPending = state.interimBoundaryPending
|
||||
|
||||
const replaceTextPart = (parts: ChatMessagePart[]) => {
|
||||
const visibleFinalText = stripGeneratedImageEchoes(finalText, generatedImageEchoSources(parts)).trim()
|
||||
const dedupeReference = normalize(visibleFinalText)
|
||||
|
||||
const kept = parts.filter(part => {
|
||||
if (part.type === 'text') {
|
||||
return false
|
||||
}
|
||||
|
||||
if (part.type !== 'reasoning' || !dedupeReference) {
|
||||
return true
|
||||
}
|
||||
|
||||
const r = normalize(part.text)
|
||||
|
||||
return !(r && (dedupeReference.startsWith(r) || r.startsWith(dedupeReference)))
|
||||
})
|
||||
|
||||
return visibleFinalText ? [...kept, assistantTextPart(visibleFinalText)] : kept
|
||||
return mergeFinalAssistantText(parts, visibleFinalText)
|
||||
}
|
||||
|
||||
const completeMessage = (message: ChatMessage): ChatMessage =>
|
||||
|
|
@ -454,7 +491,27 @@ export function useMessageStream({
|
|||
const existing = prev[index]
|
||||
const existingText = chatMessageText(existing).trim()
|
||||
|
||||
if (existing.pending || (finalText && existingText === finalText)) {
|
||||
if (existing.pending || (!interimBoundaryPending && finalText && existingText === finalText)) {
|
||||
nextMessages = prev.map((message, messageIndex) =>
|
||||
messageIndex === index ? completeMessage(message) : message
|
||||
)
|
||||
} else if (
|
||||
interimBoundaryPending &&
|
||||
responsePreviewed &&
|
||||
finalText &&
|
||||
existingText &&
|
||||
finalText.startsWith(existingText)
|
||||
) {
|
||||
// The verification candidate was published provisionally as an
|
||||
// interim message and then reused as the terminal response
|
||||
// (continuation-budget fallback). Settle the interim in place
|
||||
// instead of creating a duplicate — the DB has one row, so the
|
||||
// live UI must agree. (#65919 review: duplicate-message blocker)
|
||||
//
|
||||
// Prefix match (not exact equality): the final response may be
|
||||
// the streamed text plus a trailing delta. mergeFinalAssistantText
|
||||
// (called via completeMessage) handles the actual merge — it
|
||||
// strips the old text parts and appends the full final text.
|
||||
nextMessages = prev.map((message, messageIndex) =>
|
||||
messageIndex === index ? completeMessage(message) : message
|
||||
)
|
||||
|
|
@ -480,6 +537,7 @@ export function useMessageStream({
|
|||
awaitingResponse: false,
|
||||
busy: false,
|
||||
needsInput: false,
|
||||
interimBoundaryPending: false,
|
||||
turnStartedAt: null
|
||||
}
|
||||
})
|
||||
|
|
@ -543,6 +601,7 @@ export function useMessageStream({
|
|||
awaitingResponse: false,
|
||||
busy: false,
|
||||
needsInput: false,
|
||||
interimBoundaryPending: false,
|
||||
turnStartedAt: null
|
||||
}
|
||||
})
|
||||
|
|
@ -560,6 +619,7 @@ export function useMessageStream({
|
|||
completeAssistantMessage,
|
||||
failAssistantMessage,
|
||||
flushQueuedDeltas,
|
||||
finalizeInterimAssistantMessage,
|
||||
queryClient,
|
||||
refreshHermesConfig,
|
||||
sessionInterrupted,
|
||||
|
|
@ -573,6 +633,7 @@ export function useMessageStream({
|
|||
appendReasoningDelta,
|
||||
completeAssistantMessage,
|
||||
handleGatewayEvent,
|
||||
finalizeInterimAssistantMessage,
|
||||
upsertToolCall
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,244 @@
|
|||
import { QueryClient } from '@tanstack/react-query'
|
||||
import { act, cleanup, render, waitFor } from '@testing-library/react'
|
||||
import { useEffect, useRef } from 'react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import type { ClientSessionState } from '@/app/types'
|
||||
import { createClientSessionState } from '@/lib/chat-runtime'
|
||||
import { chatMessageText } from '@/lib/chat-messages'
|
||||
import { $todosBySession, clearSessionTodos, setSessionTodos } from '@/store/todos'
|
||||
import type { RpcEvent } from '@/types/hermes'
|
||||
|
||||
import { useMessageStream } from './index'
|
||||
|
||||
const SID = 'session-1'
|
||||
|
||||
let handleEvent: ((event: RpcEvent) => void) | null = null
|
||||
let sessionStates: Map<string, ClientSessionState>
|
||||
let mockCompleteSound: ReturnType<typeof vi.fn>
|
||||
let mockHaptic: ReturnType<typeof vi.fn>
|
||||
|
||||
function Harness() {
|
||||
const activeSessionIdRef = useRef<string | null>(SID)
|
||||
const sessionStateByRuntimeIdRef = useRef(new Map<string, ClientSessionState>())
|
||||
const queryClientRef = useRef(new QueryClient())
|
||||
|
||||
const stream = useMessageStream({
|
||||
activeSessionIdRef,
|
||||
hydrateFromStoredSession: vi.fn(async () => undefined),
|
||||
queryClient: queryClientRef.current,
|
||||
refreshHermesConfig: vi.fn(async () => undefined),
|
||||
refreshSessions: vi.fn(async () => undefined),
|
||||
sessionStateByRuntimeIdRef,
|
||||
updateSessionState: (sessionId, updater) => {
|
||||
const current = sessionStateByRuntimeIdRef.current.get(sessionId) ?? createClientSessionState()
|
||||
const next = updater(current)
|
||||
sessionStateByRuntimeIdRef.current.set(sessionId, next)
|
||||
sessionStates.set(sessionId, next)
|
||||
|
||||
return next
|
||||
}
|
||||
})
|
||||
|
||||
useEffect(() => {
|
||||
handleEvent = stream.handleGatewayEvent
|
||||
}, [stream.handleGatewayEvent])
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
async function mountStream() {
|
||||
sessionStates = new Map()
|
||||
render(<Harness />)
|
||||
await waitFor(() => expect(handleEvent).not.toBeNull())
|
||||
}
|
||||
|
||||
const start = () => act(() => handleEvent!({ payload: {}, session_id: SID, type: 'message.start' }))
|
||||
const delta = (text: string) => act(() => handleEvent!({ payload: { text }, session_id: SID, type: 'message.delta' }))
|
||||
const interim = (text: string) =>
|
||||
act(() => handleEvent!({ payload: { text, already_streamed: true }, session_id: SID, type: 'message.interim' }))
|
||||
const complete = (text: string) =>
|
||||
act(() => handleEvent!({ payload: { text }, session_id: SID, type: 'message.complete' }))
|
||||
const completePreviewed = (text: string) =>
|
||||
act(() => handleEvent!({ payload: { text, response_previewed: true }, session_id: SID, type: 'message.complete' }))
|
||||
|
||||
function getState(): ClientSessionState {
|
||||
return sessionStates.get(SID) ?? createClientSessionState()
|
||||
}
|
||||
|
||||
function assistantText(): string {
|
||||
const state = getState()
|
||||
const last = [...state.messages].reverse().find(m => m.role === 'assistant' && !m.hidden)
|
||||
return last ? chatMessageText(last) : ''
|
||||
}
|
||||
|
||||
function assistantMessages(): string[] {
|
||||
const state = getState()
|
||||
return state.messages
|
||||
.filter(m => m.role === 'assistant' && !m.hidden)
|
||||
.map(m => chatMessageText(m))
|
||||
.filter(Boolean)
|
||||
}
|
||||
|
||||
describe('useMessageStream interim text sealing', () => {
|
||||
beforeEach(() => {
|
||||
handleEvent = null
|
||||
clearSessionTodos(SID)
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
cleanup()
|
||||
clearSessionTodos(SID)
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('preserves interim text that the final response does not include', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await delta('awaaaaa clean!! tsc zero errors')
|
||||
await interim('awaaaaa clean!! tsc zero errors')
|
||||
|
||||
await complete('All checks passed.')
|
||||
|
||||
const texts = assistantMessages()
|
||||
expect(texts).toContain('awaaaaa clean!! tsc zero errors')
|
||||
expect(texts).toContain('All checks passed.')
|
||||
})
|
||||
|
||||
it('dedupes interim text when the final response includes it', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await delta('Let me check the files.')
|
||||
await interim('Let me check the files.')
|
||||
|
||||
await complete('Let me check the files. Everything looks good.')
|
||||
|
||||
const texts = assistantMessages()
|
||||
expect(texts).not.toContain('Let me check the files.Let me check the files.')
|
||||
expect(texts.some(t => t.includes('Let me check the files. Everything looks good.'))).toBe(true)
|
||||
})
|
||||
|
||||
it('clears interimBoundaryPending at turn end so the next turn starts clean', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await delta('interim text')
|
||||
await interim('interim text')
|
||||
await complete('final text')
|
||||
|
||||
expect(getState().interimBoundaryPending).toBe(false)
|
||||
|
||||
await start()
|
||||
expect(getState().interimBoundaryPending).toBe(false)
|
||||
|
||||
await complete('new turn final')
|
||||
|
||||
const texts = assistantMessages()
|
||||
expect(texts[texts.length - 1]).toBe('new turn final')
|
||||
})
|
||||
|
||||
it('finalizes an interim segment without settling the turn', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await delta('streaming text')
|
||||
await interim('streaming text')
|
||||
|
||||
// Turn is still active — busy stays true
|
||||
expect(getState().busy).toBe(true)
|
||||
expect(getState().interimBoundaryPending).toBe(true)
|
||||
})
|
||||
|
||||
it('keeps an identical final completion distinct from an interim reply without response_previewed', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await interim('same reply')
|
||||
await complete('same reply')
|
||||
|
||||
// Without response_previewed, the interim and terminal replies are
|
||||
// distinct messages — the gateway didn't signal that the final reuses
|
||||
// the provisional candidate.
|
||||
const texts = assistantMessages()
|
||||
expect(texts.filter(t => t === 'same reply')).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('settles an identical final completion onto the interim when response_previewed', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await interim('same reply')
|
||||
await completePreviewed('same reply')
|
||||
|
||||
// With response_previewed, the final text is the same model response
|
||||
// that was published provisionally as an interim — settle onto the
|
||||
// existing interim instead of creating a duplicate. (#65919 review)
|
||||
const texts = assistantMessages()
|
||||
expect(texts.filter(t => t === 'same reply')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('settles a prefix-matched final onto the interim when response_previewed', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
// Interim text is a PREFIX of the final — the model streamed part of
|
||||
// its answer before the verify nudge fired, then the final includes
|
||||
// the same text plus a trailing delta.
|
||||
await interim('partial answer')
|
||||
await completePreviewed('partial answer with more detail')
|
||||
|
||||
// Prefix match: the final starts with the interim text, so settle
|
||||
// onto the interim instead of creating a duplicate bubble.
|
||||
const texts = assistantMessages()
|
||||
expect(texts.filter(t => t.includes('partial answer'))).toHaveLength(1)
|
||||
expect(texts[0]).toBe('partial answer with more detail')
|
||||
})
|
||||
|
||||
it('dedupes partial-stream-then-nudge: streamed prefix + interim + previewed final settles to one bubble', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
// The model streamed part of its answer via message.delta, then the
|
||||
// verify nudge fired. The interim seals the streamed text, then the
|
||||
// final response is the same text plus a trailing delta.
|
||||
await delta('partial streamed')
|
||||
await interim('partial streamed')
|
||||
await completePreviewed('partial streamed answer continued')
|
||||
|
||||
// One bubble, containing the full final text — not two.
|
||||
const texts = assistantMessages()
|
||||
expect(texts.filter(t => t.includes('partial streamed'))).toHaveLength(1)
|
||||
expect(texts[0]).toBe('partial streamed answer continued')
|
||||
})
|
||||
|
||||
it('ignores malformed message.interim payload', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
// No payload at all
|
||||
await act(() => handleEvent!({ type: 'message.interim' } as RpcEvent))
|
||||
// Empty text
|
||||
await act(() => handleEvent!({ payload: { text: '' }, session_id: SID, type: 'message.interim' } as RpcEvent))
|
||||
// Undefined text
|
||||
await act(() => handleEvent!({ payload: { text: undefined }, session_id: SID, type: 'message.interim' } as RpcEvent))
|
||||
|
||||
// Turn continues without finalizing or throwing
|
||||
expect(getState().busy).toBe(true)
|
||||
expect(getState().interimBoundaryPending).toBe(false)
|
||||
})
|
||||
|
||||
it('clears interimBoundaryPending on message.start', async () => {
|
||||
await mountStream()
|
||||
await start()
|
||||
|
||||
await delta('interim text')
|
||||
await interim('interim text')
|
||||
expect(getState().interimBoundaryPending).toBe(true)
|
||||
|
||||
// New turn starts
|
||||
await start()
|
||||
expect(getState().interimBoundaryPending).toBe(false)
|
||||
})
|
||||
})
|
||||
|
|
@ -820,6 +820,7 @@ describe('resumeSession failure recovery', () => {
|
|||
busy: false,
|
||||
cwd: '',
|
||||
fast: false,
|
||||
interimBoundaryPending: false,
|
||||
interrupted: false,
|
||||
messages: [],
|
||||
model: '',
|
||||
|
|
|
|||
|
|
@ -154,6 +154,8 @@ export interface ClientSessionState {
|
|||
sawAssistantPayload: boolean
|
||||
pendingBranchGroup: string | null
|
||||
interrupted: boolean
|
||||
/** True after message.interim finalized a bubble in the still-running turn. */
|
||||
interimBoundaryPending: boolean
|
||||
/** A blocking clarify prompt is waiting on the user for this session. Drives
|
||||
* the sidebar "needs input" indicator; cleared when the turn resumes/ends. */
|
||||
needsInput: boolean
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@ import {
|
|||
appendAssistantTextPart,
|
||||
appendReasoningPart,
|
||||
chatMessageText,
|
||||
mergeFinalAssistantText,
|
||||
preserveLocalAssistantErrors,
|
||||
reasoningPart,
|
||||
renderMediaTags,
|
||||
toChatMessages,
|
||||
upsertToolPart
|
||||
|
|
@ -785,3 +787,69 @@ describe('upsertToolPart', () => {
|
|||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('mergeFinalAssistantText', () => {
|
||||
it('removes all text parts and appends the final text', () => {
|
||||
const parts = [
|
||||
{ type: 'text' as const, text: 'streamed delta 1' },
|
||||
{ type: 'text' as const, text: 'streamed delta 2' },
|
||||
{ type: 'tool-call' as const, toolCallId: 'tc1', toolName: 'terminal', args: {} as never, argsText: '{}' }
|
||||
]
|
||||
|
||||
const result = mergeFinalAssistantText(parts, 'final answer')
|
||||
|
||||
expect(result.filter(p => p.type === 'text')).toHaveLength(1)
|
||||
expect(result.filter(p => p.type === 'text')[0]).toMatchObject({ text: 'final answer' })
|
||||
expect(result.some(p => p.type === 'tool-call')).toBe(true)
|
||||
})
|
||||
|
||||
it('drops reasoning that the final text fully covers (reasoning ⊆ final)', () => {
|
||||
const parts = [
|
||||
reasoningPart('Let me check the files.'),
|
||||
{ type: 'text' as const, text: 'streamed' }
|
||||
]
|
||||
|
||||
const result = mergeFinalAssistantText(parts, 'Let me check the files. Everything looks good.')
|
||||
|
||||
expect(result.filter(p => p.type === 'reasoning')).toHaveLength(0)
|
||||
expect(result.filter(p => p.type === 'text')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('keeps a longer reasoning block when the final text is only a short prefix', () => {
|
||||
// #61447: a short final ("Done.") must NOT swallow a longer reasoning block
|
||||
// that merely starts with it.
|
||||
const parts = [
|
||||
reasoningPart('Done. The root cause was a bare catch block swallowing Stripe errors. The fix adds proper error logging.'),
|
||||
{ type: 'text' as const, text: 'streamed' }
|
||||
]
|
||||
|
||||
const result = mergeFinalAssistantText(parts, 'Done.')
|
||||
|
||||
expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
|
||||
expect(result.filter(p => p.type === 'text')[0]).toMatchObject({ text: 'Done.' })
|
||||
})
|
||||
|
||||
it('keeps non-restating reasoning', () => {
|
||||
const parts = [
|
||||
reasoningPart('I analyzed the issue and found a race condition in the event loop.'),
|
||||
{ type: 'text' as const, text: 'streamed' }
|
||||
]
|
||||
|
||||
const result = mergeFinalAssistantText(parts, 'Fixed the race condition.')
|
||||
|
||||
expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
|
||||
expect(result.filter(p => p.type === 'text')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('handles empty final text', () => {
|
||||
const parts = [
|
||||
{ type: 'text' as const, text: 'streamed' },
|
||||
reasoningPart('some reasoning')
|
||||
]
|
||||
|
||||
const result = mergeFinalAssistantText(parts, '')
|
||||
|
||||
expect(result.filter(p => p.type === 'text')).toHaveLength(0)
|
||||
expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -87,6 +87,9 @@ export type GatewayEventPayload = {
|
|||
label?: string
|
||||
index?: number
|
||||
aggregator?: string
|
||||
// message.complete — signals the final text was already previewed via
|
||||
// interim_assistant_callback, so the UI can settle instead of duplicating.
|
||||
response_previewed?: boolean
|
||||
}
|
||||
|
||||
export function textPart(text: string): ChatMessagePart {
|
||||
|
|
@ -136,6 +139,49 @@ export function chatMessageText(message: ChatMessage): string {
|
|||
.join('')
|
||||
}
|
||||
|
||||
const normalizeWs = (value: string) => value.replace(/\s+/g, ' ').trim()
|
||||
|
||||
/**
|
||||
* Merge the final assistant text into a message's parts.
|
||||
*
|
||||
* - Removes all existing `text` parts (they were streamed deltas, now superseded
|
||||
* by the authoritative final response).
|
||||
* - Keeps `reasoning` parts, but drops one that the final text fully covers
|
||||
* (reasoning ⊆ final) — the final restates it. A short final ("Done.") must
|
||||
* NOT swallow a longer reasoning block that merely starts with it (#61447).
|
||||
* - Keeps all other part types (tool-call, image, etc.).
|
||||
* - Appends the final text as a new text part.
|
||||
*/
|
||||
export function mergeFinalAssistantText(
|
||||
parts: ChatMessagePart[],
|
||||
finalText: string
|
||||
): ChatMessagePart[] {
|
||||
const dedupeReference = normalizeWs(finalText)
|
||||
|
||||
const kept = parts.filter(part => {
|
||||
if (part.type === 'text') {
|
||||
// Sealed text parts were already finalized into their own bubbles —
|
||||
// this filter only runs on the LAST streaming bubble, so there are no
|
||||
// sealed parts here. All text parts are streamed deltas that get
|
||||
// replaced by the authoritative final text.
|
||||
return false
|
||||
}
|
||||
|
||||
if (part.type !== 'reasoning' || !dedupeReference) {
|
||||
return true
|
||||
}
|
||||
|
||||
// Reasoning is a restatement only when the final FULLY covers it.
|
||||
// The reverse direction is not considered — a short final must not
|
||||
// swallow a longer reasoning block (#61447).
|
||||
const r = normalizeWs(part.text)
|
||||
|
||||
return !(r && dedupeReference.startsWith(r))
|
||||
})
|
||||
|
||||
return finalText ? [...kept, assistantTextPart(finalText)] : kept
|
||||
}
|
||||
|
||||
const ATTACHED_CONTEXT_MARKER_RE = /(?:^|\n)--- Attached Context ---\s*\n/
|
||||
const CONTEXT_WARNINGS_MARKER_RE = /(?:^|\n)--- Context Warnings ---[\s\S]*$/
|
||||
const CONTEXT_REF_RE = /@(file|folder|url|image|tool|terminal):(?:"[^"\n]+"|'[^'\n]+'|`[^`\n]+`|\S+)/g
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ export function createClientSessionState(
|
|||
sawAssistantPayload: false,
|
||||
pendingBranchGroup: null,
|
||||
interrupted: false,
|
||||
interimBoundaryPending: false,
|
||||
needsInput: false,
|
||||
turnStartedAt: null,
|
||||
usage: null
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ describe('gateway event routing', () => {
|
|||
// output, and dropping them loses the live response until a refetch (#42178).
|
||||
expect(gatewayEventRequiresSessionId('message.delta')).toBe(false)
|
||||
expect(gatewayEventRequiresSessionId('message.complete')).toBe(false)
|
||||
expect(gatewayEventRequiresSessionId('message.interim')).toBe(false)
|
||||
expect(gatewayEventRequiresSessionId('reasoning.delta')).toBe(false)
|
||||
expect(gatewayEventRequiresSessionId('tool.start')).toBe(false)
|
||||
expect(gatewayEventRequiresSessionId('approval.request')).toBe(false)
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ const UNSCOPED_STREAM_EVENT_TYPES = new Set([
|
|||
'error',
|
||||
'message.complete',
|
||||
'message.delta',
|
||||
'message.interim',
|
||||
'message.start',
|
||||
'reasoning.available',
|
||||
'reasoning.delta',
|
||||
|
|
|
|||
|
|
@ -256,6 +256,7 @@ export interface HermesConfig {
|
|||
display?: {
|
||||
personality?: string
|
||||
skin?: string
|
||||
interim_assistant_messages?: boolean
|
||||
}
|
||||
terminal?: {
|
||||
cwd?: string
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ export type GatewayEventName =
|
|||
| 'session.info'
|
||||
| 'message.start'
|
||||
| 'message.delta'
|
||||
| 'message.interim'
|
||||
| 'message.complete'
|
||||
| 'thinking.delta'
|
||||
| 'reasoning.delta'
|
||||
|
|
|
|||
|
|
@ -1181,11 +1181,14 @@ display:
|
|||
# cleanup_progress: true
|
||||
cleanup_progress: false
|
||||
|
||||
# Gateway-only natural mid-turn assistant updates.
|
||||
# When true, completed assistant status messages are sent as separate chat
|
||||
# messages. This is independent of tool_progress and gateway streaming.
|
||||
# true: Send mid-turn assistant updates as separate messages (default)
|
||||
# false: Only send the final response
|
||||
# Natural mid-turn assistant updates.
|
||||
# On gateway platforms, when true, completed assistant status messages are
|
||||
# sent as separate chat messages. On the Desktop app, when true, mid-turn
|
||||
# assistant narration streamed between tool calls is kept in the transcript
|
||||
# instead of the bubble collapsing to only the final message on completion.
|
||||
# Independent of tool_progress and gateway streaming.
|
||||
# true: Keep/send mid-turn assistant updates (default)
|
||||
# false: Only keep/send the final response
|
||||
interim_assistant_messages: true
|
||||
|
||||
# Gateway-only long-running status heartbeats.
|
||||
|
|
|
|||
|
|
@ -189,6 +189,13 @@ class GatewayStreamConsumer:
|
|||
# subsequently failed.
|
||||
self._final_content_delivered = False
|
||||
self._delivered_commentary_texts: list[str] = []
|
||||
# Retains the finalized visible text of each streaming segment so
|
||||
# ``has_delivered_text`` can still match after ``_reset_segment_state``
|
||||
# clears ``_last_sent_text``. Without this, a segment break (triggered
|
||||
# by ``on_segment_break`` or ``on_commentary``) erases the only record
|
||||
# of what was delivered, and the gateway's final-send suppression
|
||||
# can't recognize an already-delivered response. (#65919 review)
|
||||
self._delivered_segment_texts: list[str] = []
|
||||
# Cache adapter lifecycle capability: only platforms that need an
|
||||
# explicit finalize call (e.g. DingTalk AI Cards) force us to make
|
||||
# a redundant final edit. Everyone else keeps the fast path.
|
||||
|
|
@ -320,7 +327,10 @@ class GatewayStreamConsumer:
|
|||
visible_prefix = self._visible_prefix().strip()
|
||||
if visible_prefix == target:
|
||||
return True
|
||||
return any(sent.strip() == target for sent in self._delivered_commentary_texts)
|
||||
return any(
|
||||
sent.strip() == target
|
||||
for sent in (*self._delivered_commentary_texts, *self._delivered_segment_texts)
|
||||
)
|
||||
|
||||
def on_segment_break(self) -> None:
|
||||
"""Finalize the current stream segment and start a fresh message."""
|
||||
|
|
@ -344,6 +354,13 @@ class GatewayStreamConsumer:
|
|||
def _reset_segment_state(self, *, preserve_no_edit: bool = False) -> None:
|
||||
if preserve_no_edit and self._message_id == "__no_edit__":
|
||||
return
|
||||
# Retain the finalized visible text of the current segment before
|
||||
# clearing ``_last_sent_text``, so ``has_delivered_text`` can still
|
||||
# match it after a segment break. (#65919 review)
|
||||
if self._last_sent_text:
|
||||
finalized = self._clean_for_display(self._last_sent_text).strip()
|
||||
if finalized:
|
||||
self._delivered_segment_texts.append(finalized)
|
||||
self._message_id = None
|
||||
self._message_created_ts = None
|
||||
self._accumulated = ""
|
||||
|
|
|
|||
|
|
@ -1948,7 +1948,7 @@ DEFAULT_CONFIG = {
|
|||
"first_lines": 2,
|
||||
"last_lines": 2,
|
||||
},
|
||||
"interim_assistant_messages": True, # Gateway: show natural mid-turn assistant status messages
|
||||
"interim_assistant_messages": True, # Gateway: send natural mid-turn assistant status messages. Desktop: keep mid-turn narration between tool calls instead of collapsing to the final message.
|
||||
# Codex Responses models narrate progress in a dedicated commentary
|
||||
# channel. When true (default), completed commentary messages are
|
||||
# delivered as visible mid-turn updates via the interim message path.
|
||||
|
|
|
|||
37
run_agent.py
37
run_agent.py
|
|
@ -229,11 +229,12 @@ _EPHEMERAL_SCAFFOLDING_FLAGS = (
|
|||
"_empty_recovery_synthetic",
|
||||
"_empty_terminal_sentinel",
|
||||
"_thinking_prefill",
|
||||
# verify-on-stop and pre_verify nudges append a synthetic assistant
|
||||
# "done" plus a synthetic user nudge to keep the agent going one more
|
||||
# turn before it can claim completion. Those messages exist only to
|
||||
# drive the verification loop; persisting them poisons the resumed
|
||||
# transcript and breaks prompt-prefix cache reuse on later turns. (#55733)
|
||||
# verify-on-stop and pre_verify nudges append a synthetic user nudge to
|
||||
# keep the agent going one more turn before it can claim completion.
|
||||
# The nudge exists only to drive the verification loop; persisting it
|
||||
# poisons the resumed transcript and breaks prompt-prefix cache reuse
|
||||
# on later turns. The assistant candidate is NOT synthetic — it is
|
||||
# persisted and emitted as an interim message (#65919).
|
||||
"_verification_stop_synthetic",
|
||||
"_pre_verify_synthetic",
|
||||
# kanban worker stop-guard: narrated exit without kanban_complete/block
|
||||
|
|
@ -4914,7 +4915,16 @@ class AIAgent:
|
|||
streamed = self._normalize_interim_visible_text(
|
||||
self._strip_think_blocks(getattr(self, "_current_streamed_assistant_text", "") or "")
|
||||
)
|
||||
return bool(streamed) and streamed == visible_content
|
||||
# Prefix match (not exact equality): the final response may be the
|
||||
# streamed text plus a trailing delta, or the stream may have been
|
||||
# partial when the verify nudge fired. In both cases the streamed
|
||||
# content is a prefix of the final — that's enough to mark it
|
||||
# previewed (fails safe to a benign duplicate, never loses text).
|
||||
# The reverse direction (streamed longer than final) is NOT matched:
|
||||
# that could suppress a needed resend in the gateway path where
|
||||
# already_streamed=True calls on_segment_break() instead of
|
||||
# on_commentary() (#65919 review).
|
||||
return bool(streamed) and visible_content.startswith(streamed)
|
||||
|
||||
def _extract_codex_interim_visible_parts(
|
||||
self,
|
||||
|
|
@ -5019,8 +5029,19 @@ class AIAgent:
|
|||
except Exception:
|
||||
logger.debug("interim_assistant_callback error", exc_info=True)
|
||||
|
||||
def _emit_interim_assistant_message(self, assistant_msg: Dict[str, Any]) -> None:
|
||||
"""Surface a real mid-turn assistant commentary message to the UI layer."""
|
||||
def _emit_interim_assistant_message(
|
||||
self, assistant_msg: Dict[str, Any]
|
||||
) -> None:
|
||||
"""Surface a real mid-turn assistant commentary message to the UI layer.
|
||||
|
||||
Does NOT set ``_response_was_previewed`` — that flag means "the final
|
||||
response was already shown to the user," but this helper is called for
|
||||
ordinary tool-call narration, intermediate acknowledgements, and
|
||||
verification candidates alike. Setting it here would cause the CLI to
|
||||
suppress a *different* final summary (e.g. from ``_handle_max_iterations``)
|
||||
when the only streamed text was unrelated mid-turn commentary. (#65919
|
||||
review: response-loss blocker)
|
||||
"""
|
||||
cb = getattr(self, "interim_assistant_callback", None)
|
||||
if cb is None or not isinstance(assistant_msg, dict):
|
||||
return
|
||||
|
|
|
|||
|
|
@ -296,3 +296,84 @@ def test_pending_response_records_kanban_timeout(monkeypatch):
|
|||
end_run=True,
|
||||
event_payload_extra={"budget_used": 60, "budget_max": 60},
|
||||
)
|
||||
|
||||
|
||||
def test_published_pending_candidate_is_not_duplicated_by_finalizer(monkeypatch):
|
||||
"""When budget exhaustion preserves a verification candidate that is
|
||||
already the tail assistant message, the finalizer must NOT append a
|
||||
duplicate. The content-comparison guard prevents this. (#65919 §7)
|
||||
"""
|
||||
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
|
||||
agent = _LimitAgent()
|
||||
report = "the composed report"
|
||||
|
||||
result = finalize_turn(
|
||||
agent,
|
||||
final_response=report,
|
||||
api_call_count=60,
|
||||
interrupted=False,
|
||||
failed=False,
|
||||
# The candidate is already in messages as the tail assistant.
|
||||
messages=[
|
||||
{"role": "user", "content": "task"},
|
||||
{"role": "assistant", "content": report},
|
||||
],
|
||||
conversation_history=[],
|
||||
effective_task_id="task",
|
||||
turn_id="turn",
|
||||
user_message="task",
|
||||
original_user_message="task",
|
||||
_should_review_memory=False,
|
||||
_turn_exit_reason="unknown",
|
||||
_pending_verification_response=report,
|
||||
)
|
||||
|
||||
# The tail assistant already matches final_response — no duplicate appended.
|
||||
roles = [m["role"] for m in result["messages"]]
|
||||
assert roles == ["user", "assistant"]
|
||||
# Persisted messages should also have no duplicate.
|
||||
assert agent.persisted_messages is not None
|
||||
persisted_roles = [m["role"] for m in agent.persisted_messages]
|
||||
assert persisted_roles == ["user", "assistant"]
|
||||
|
||||
|
||||
def test_terminal_verification_failure_is_persisted_as_one_correction(monkeypatch):
|
||||
"""When verification fails terminally (nudge present but budget exhausted),
|
||||
the finalizer drops the synthetic nudge and the assistant candidate
|
||||
persists as a single correction. No duplicate assistant appended. (#65919 §7)
|
||||
"""
|
||||
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
|
||||
agent = _LimitAgent()
|
||||
report = "terminal failure correction"
|
||||
|
||||
result = finalize_turn(
|
||||
agent,
|
||||
final_response=report,
|
||||
api_call_count=60,
|
||||
interrupted=False,
|
||||
failed=False,
|
||||
messages=[
|
||||
{"role": "user", "content": "task"},
|
||||
{"role": "assistant", "content": report},
|
||||
# Synthetic nudge — should be dropped by _drop_verification_continuation_scaffolding.
|
||||
{"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True},
|
||||
],
|
||||
conversation_history=[],
|
||||
effective_task_id="task",
|
||||
turn_id="turn",
|
||||
user_message="task",
|
||||
original_user_message="task",
|
||||
_should_review_memory=False,
|
||||
_turn_exit_reason="unknown",
|
||||
_pending_verification_response=report,
|
||||
)
|
||||
|
||||
# The nudge is dropped; the assistant candidate is the tail and matches
|
||||
# final_response, so no duplicate is appended.
|
||||
roles = [m["role"] for m in result["messages"]]
|
||||
assert roles == ["user", "assistant"]
|
||||
# The nudge is gone from persisted messages too.
|
||||
assert agent.persisted_messages is not None
|
||||
persisted_contents = [m.get("content") for m in agent.persisted_messages]
|
||||
assert "[System: run tests]" not in persisted_contents
|
||||
assert report in persisted_contents
|
||||
|
|
|
|||
|
|
@ -391,3 +391,36 @@ def test_recording_expires_old_edit_only_state(tmp_path, monkeypatch):
|
|||
status = verification_status(session_id="old-session", cwd=tmp_path)
|
||||
assert status["status"] == "unverified"
|
||||
assert status["changed_paths"] == []
|
||||
|
||||
|
||||
def test_windows_backslash_ad_hoc_script_path_is_matched(tmp_path, monkeypatch):
|
||||
"""Ad-hoc verification scripts with Windows backslash paths must be
|
||||
matched by ``_find_ad_hoc_match`` trying ``posix=False`` in addition to
|
||||
the default ``posix=True``. (#53553 / #65919)
|
||||
|
||||
On Linux, ``Path`` doesn't parse Windows backslash paths, so we mock
|
||||
``_is_temp_script_path`` to simulate the Windows environment where the
|
||||
path resolves correctly. The test verifies the posix=False splitting
|
||||
fallback — the actual fix from #53553.
|
||||
"""
|
||||
from agent.verification_evidence import _find_ad_hoc_match
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
(tmp_path / "package.json").write_text("{}", encoding="utf-8")
|
||||
|
||||
# On Windows, shlex.split(posix=True) eats backslashes as escape chars;
|
||||
# posix=False preserves them. Mock _is_temp_script_path so the test
|
||||
# focuses on the splitting fallback without needing a real Windows FS.
|
||||
def mock_is_temp_script(token, root):
|
||||
return "hermes-ad-hoc" in token and ".py" in token
|
||||
|
||||
monkeypatch.setattr(
|
||||
"agent.verification_evidence._is_temp_script_path",
|
||||
mock_is_temp_script,
|
||||
)
|
||||
|
||||
win_script = r"C:\Users\test\AppData\Local\Temp\hermes-ad-hoc-check.py"
|
||||
result = _find_ad_hoc_match(f"python {win_script}", tmp_path)
|
||||
assert result is not None, (
|
||||
"Windows backslash path should be matched via posix=False fallback"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,15 +1,15 @@
|
|||
"""Verification-loop synthetic scaffolding must never reach durable session state.
|
||||
|
||||
verify_on_stop / pre_verify append a synthetic assistant "done" plus a synthetic
|
||||
user nudge to keep the agent going one more turn before it can claim completion.
|
||||
These messages exist only to drive the loop; persisting them poisons the resumed
|
||||
transcript and breaks prompt-prefix cache reuse on later turns (#55733).
|
||||
verify_on_stop / pre_verify inject a synthetic user nudge to keep the agent
|
||||
going one more turn before it can claim completion. The assistant response is
|
||||
real content that persists and is emitted to the UI as an interim message.
|
||||
Only the nudge (the synthetic user message) is flagged, so only the nudge
|
||||
gets stripped from the durable transcript. This test file verifies:
|
||||
|
||||
Both persistence sinks (SQLite flush + JSON snapshot) route through the single
|
||||
``_is_ephemeral_scaffolding`` chokepoint, which is driven by
|
||||
``_EPHEMERAL_SCAFFOLDING_FLAGS``. These tests assert that the verification-loop
|
||||
flags are registered there and that both sinks drop the flagged messages while
|
||||
keeping the real conversation.
|
||||
- The verification-loop flags remain registered in
|
||||
``_EPHEMERAL_SCAFFOLDING_FLAGS`` (so nudges are stripped).
|
||||
- The DB flush drops only the nudge, keeping the assistant candidate.
|
||||
- The JSON log drops only the nudge, keeping the assistant candidate.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
|
@ -34,15 +34,16 @@ def test_verification_flags_registered_as_ephemeral(tmp_path, monkeypatch):
|
|||
assert "_verification_stop_synthetic" in ra._EPHEMERAL_SCAFFOLDING_FLAGS
|
||||
assert "_pre_verify_synthetic" in ra._EPHEMERAL_SCAFFOLDING_FLAGS
|
||||
|
||||
# The central classifier drives both persistence sinks.
|
||||
assert ra._is_ephemeral_scaffolding(
|
||||
{"role": "assistant", "content": "done", "_verification_stop_synthetic": True}
|
||||
)
|
||||
# The nudge messages ARE scaffolding (they carry the synthetic flag).
|
||||
assert ra._is_ephemeral_scaffolding(
|
||||
{"role": "user", "content": "[System: run tests]", "_pre_verify_synthetic": True}
|
||||
)
|
||||
# Real messages are not scaffolding.
|
||||
assert ra._is_ephemeral_scaffolding(
|
||||
{"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True}
|
||||
)
|
||||
# Real messages (including the assistant candidate) are not.
|
||||
assert not ra._is_ephemeral_scaffolding({"role": "user", "content": "hi"})
|
||||
assert not ra._is_ephemeral_scaffolding({"role": "assistant", "content": "premature done"})
|
||||
|
||||
|
||||
def _make_agent(ra, session_id, tmp_path):
|
||||
|
|
@ -64,14 +65,18 @@ def _make_agent(ra, session_id, tmp_path):
|
|||
return agent
|
||||
|
||||
|
||||
def test_db_flush_drops_verification_scaffolding(tmp_path, monkeypatch):
|
||||
def test_db_flush_drops_only_nudge_keeps_candidate(tmp_path, monkeypatch):
|
||||
"""The assistant candidate is NOT flagged synthetic, so it persists.
|
||||
Only the nudge (flagged synthetic) is dropped from the DB flush."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
ra = _fresh_run_agent(tmp_path)
|
||||
agent = _make_agent(ra, "sess_db", tmp_path)
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{"role": "assistant", "content": "premature done", "_verification_stop_synthetic": True},
|
||||
# Assistant candidate — NOT flagged synthetic, persists.
|
||||
{"role": "assistant", "content": "premature done"},
|
||||
# Nudge — flagged synthetic, gets dropped.
|
||||
{"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True},
|
||||
{"role": "assistant", "content": "verified and clean"},
|
||||
]
|
||||
|
|
@ -84,18 +89,24 @@ def test_db_flush_drops_verification_scaffolding(tmp_path, monkeypatch):
|
|||
]
|
||||
assert "hi" in persisted
|
||||
assert "verified and clean" in persisted
|
||||
assert "premature done" not in persisted
|
||||
# The assistant candidate persists — it is real content.
|
||||
assert "premature done" in persisted
|
||||
# Only the nudge is dropped.
|
||||
assert "[System: run tests]" not in persisted
|
||||
|
||||
|
||||
def test_json_log_drops_verification_scaffolding(tmp_path, monkeypatch):
|
||||
def test_json_log_drops_only_nudge_keeps_candidate(tmp_path, monkeypatch):
|
||||
"""The assistant candidate is NOT flagged synthetic, so it persists in the
|
||||
JSON log. Only the nudge (flagged synthetic) is dropped."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
ra = _fresh_run_agent(tmp_path)
|
||||
agent = _make_agent(ra, "sess_json", tmp_path)
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{"role": "assistant", "content": "premature done", "_pre_verify_synthetic": True},
|
||||
# Assistant candidate — NOT flagged synthetic, persists.
|
||||
{"role": "assistant", "content": "premature done"},
|
||||
# Nudge — flagged synthetic, gets dropped.
|
||||
{"role": "user", "content": "[System: run tests]", "_pre_verify_synthetic": True},
|
||||
{"role": "assistant", "content": "verified and clean"},
|
||||
]
|
||||
|
|
@ -106,5 +117,10 @@ def test_json_log_drops_verification_scaffolding(tmp_path, monkeypatch):
|
|||
assert log_file.exists()
|
||||
data = json.loads(log_file.read_text(encoding="utf-8"))
|
||||
contents = [m.get("content") for m in data["messages"]]
|
||||
assert contents == ["hi", "verified and clean"]
|
||||
# The assistant candidate persists — it is real content.
|
||||
assert "premature done" in contents
|
||||
assert "verified and clean" in contents
|
||||
assert "hi" in contents
|
||||
# Only the nudge is dropped.
|
||||
assert "[System: run tests]" not in contents
|
||||
assert all(not m.get("_pre_verify_synthetic") for m in data["messages"])
|
||||
|
|
|
|||
|
|
@ -2379,3 +2379,40 @@ class TestStripOrphanCloseTags:
|
|||
assert tag not in consumer._accumulated
|
||||
assert "trailing prose" in consumer._accumulated
|
||||
assert "more" in consumer._accumulated
|
||||
|
||||
|
||||
class TestHasDeliveredTextAfterSegmentBreak:
|
||||
"""has_delivered_text must find a delivered segment after a segment break,
|
||||
but must not claim text from a failed delivery. (#65919 review)"""
|
||||
|
||||
def test_finds_delivered_segment_after_segment_break(self):
|
||||
"""A successfully delivered segment must still be found by
|
||||
has_delivered_text after _reset_segment_state runs."""
|
||||
c = _make_consumer()
|
||||
# Simulate a successfully delivered segment
|
||||
c._last_sent_text = "Here is the first segment"
|
||||
c._reset_segment_state()
|
||||
# After the reset, has_delivered_text must still find it
|
||||
assert c.has_delivered_text("Here is the first segment") is True
|
||||
|
||||
def test_does_not_find_undelivered_text(self):
|
||||
"""Text that was never delivered must not be claimed."""
|
||||
c = _make_consumer()
|
||||
c._last_sent_text = "delivered text"
|
||||
c._reset_segment_state()
|
||||
assert c.has_delivered_text("never sent text") is False
|
||||
|
||||
def test_finds_commentary_text(self):
|
||||
"""has_delivered_text must find commentary text delivered via
|
||||
on_commentary."""
|
||||
c = _make_consumer()
|
||||
c._delivered_commentary_texts.append("interim commentary")
|
||||
assert c.has_delivered_text("interim commentary") is True
|
||||
|
||||
def test_does_not_match_empty(self):
|
||||
"""Empty/whitespace text must not match."""
|
||||
c = _make_consumer()
|
||||
c._last_sent_text = "some text"
|
||||
c._reset_segment_state()
|
||||
assert c.has_delivered_text("") is False
|
||||
assert c.has_delivered_text(" ") is False
|
||||
|
|
|
|||
|
|
@ -2371,6 +2371,37 @@ def test_interim_commentary_is_not_marked_already_streamed_when_stream_callback_
|
|||
}
|
||||
|
||||
|
||||
def test_interim_content_was_streamed_matches_prefix_not_exact(monkeypatch):
|
||||
"""_interim_content_was_streamed should return True when the streamed text
|
||||
is a PREFIX of the final content (trailing delta added after stream, or
|
||||
partial stream before verify nudge). Exact equality is too strict — it
|
||||
fails safe to a benign duplicate bubble instead of settling the interim.
|
||||
(#65919 review: prefix-based match like the TUI's finalTail dedup.)"""
|
||||
agent = _build_agent(monkeypatch)
|
||||
|
||||
# Exact match still works
|
||||
agent._current_streamed_assistant_text = "hello world"
|
||||
assert agent._interim_content_was_streamed("hello world") is True
|
||||
|
||||
# Streamed is a prefix of the final (trailing delta) — should match
|
||||
agent._current_streamed_assistant_text = "hello"
|
||||
assert agent._interim_content_was_streamed("hello world") is True
|
||||
|
||||
# Streamed is empty — should not match
|
||||
agent._current_streamed_assistant_text = ""
|
||||
assert agent._interim_content_was_streamed("hello world") is False
|
||||
|
||||
# Final is empty — should not match
|
||||
agent._current_streamed_assistant_text = "hello"
|
||||
assert agent._interim_content_was_streamed("") is False
|
||||
|
||||
# Streamed is LONGER than final (reverse direction) — should NOT match.
|
||||
# This is the unsafe direction: it could suppress a needed resend in the
|
||||
# gateway path where already_streamed=True calls on_segment_break().
|
||||
agent._current_streamed_assistant_text = "hello world extra"
|
||||
assert agent._interim_content_was_streamed("hello") is False
|
||||
|
||||
|
||||
def test_interim_commentary_preserves_assistant_content(monkeypatch):
|
||||
"""Interim commentary must not silently mutate assistant text containing
|
||||
literal <memory-context> markers — that's legitimate model output (docs,
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""End-to-end regression coverage for verification budget exhaustion (#61631)."""
|
||||
"""End-to-end regression coverage for verification budget exhaustion (#61631, #65919 §7)."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
|
@ -51,11 +51,12 @@ def _assert_pending_response_survives(agent, result):
|
|||
assert result["turn_exit_reason"] == "max_iterations_reached(1/1)"
|
||||
assert result["completed"] is False
|
||||
assert agent._handle_max_iterations.call_count == 0
|
||||
# The nudge is stripped by _drop_verification_continuation_scaffolding,
|
||||
# so the role sequence is [user, assistant] — the candidate is the
|
||||
# tail and matches final_response so it is not duplicated. (#65919 §7)
|
||||
assert [message["role"] for message in result["messages"]] == [
|
||||
"user",
|
||||
"assistant",
|
||||
"user",
|
||||
"assistant",
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -75,8 +76,8 @@ def test_verify_on_stop_preserves_composed_report_at_budget_limit(agent, monkeyp
|
|||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
_assert_pending_response_survives(agent, result)
|
||||
assert result["messages"][1]["_verification_stop_synthetic"] is True
|
||||
assert result["messages"][2]["_verification_stop_synthetic"] is True
|
||||
# The assistant response persists (it is real, unflagged content).
|
||||
assert not result["messages"][1].get("_verification_stop_synthetic")
|
||||
|
||||
|
||||
def test_pre_verify_preserves_composed_report_at_budget_limit(agent, monkeypatch):
|
||||
|
|
@ -100,8 +101,8 @@ def test_pre_verify_preserves_composed_report_at_budget_limit(agent, monkeypatch
|
|||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
_assert_pending_response_survives(agent, result)
|
||||
assert result["messages"][1]["_pre_verify_synthetic"] is True
|
||||
assert result["messages"][2]["_pre_verify_synthetic"] is True
|
||||
# The assistant response persists (it is real, unflagged content).
|
||||
assert not result["messages"][1].get("_pre_verify_synthetic")
|
||||
|
||||
|
||||
def test_intermediate_ack_uses_summary_instead_of_premature_text(agent, monkeypatch):
|
||||
|
|
@ -144,3 +145,177 @@ def test_later_verified_response_supersedes_pending_report(agent, monkeypatch):
|
|||
assert result["turn_exit_reason"] == "text_response(finish_reason=stop)"
|
||||
assert result["completed"] is True
|
||||
agent._handle_max_iterations.assert_not_called()
|
||||
|
||||
|
||||
def test_multiple_verification_retries_publish_each_candidate_once(agent, monkeypatch):
|
||||
"""Multiple verification retries should publish each candidate once, in order."""
|
||||
agent.max_iterations = 3
|
||||
agent.iteration_budget.max_total = 3
|
||||
answers = iter([
|
||||
_response("candidate one"),
|
||||
_response("candidate two"),
|
||||
_response("candidate three"),
|
||||
])
|
||||
agent._interruptible_api_call = lambda _kwargs: next(answers)
|
||||
agent._handle_max_iterations = MagicMock(return_value="replacement summary")
|
||||
monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
|
||||
|
||||
# Three nudges, then None (so the third candidate is the final response).
|
||||
nudge_side_effects = ["verify it", "verify it", None]
|
||||
|
||||
emitted = []
|
||||
agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"agent.verification_stop.build_verify_on_stop_nudge",
|
||||
side_effect=nudge_side_effects,
|
||||
),
|
||||
patch("hermes_cli.plugins.invoke_hook", return_value=[]),
|
||||
):
|
||||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
# Each candidate was emitted as an interim message, in order.
|
||||
assert emitted == ["candidate one", "candidate two"]
|
||||
# The final response is the last candidate.
|
||||
assert result["final_response"] == "candidate three"
|
||||
assert result["turn_exit_reason"] == "text_response(finish_reason=stop)"
|
||||
assert result["completed"] is True
|
||||
agent._handle_max_iterations.assert_not_called()
|
||||
|
||||
|
||||
def test_verification_false_finalizes_candidate_once(agent, monkeypatch):
|
||||
"""When verification returns false/exception, the candidate is finalized once."""
|
||||
agent._interruptible_api_call = lambda _kwargs: _response("the answer")
|
||||
agent._handle_max_iterations = MagicMock(return_value="replacement summary")
|
||||
monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
|
||||
|
||||
emitted = []
|
||||
agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
|
||||
|
||||
with (
|
||||
# build_verify_on_stop_nudge raises — simulates verification check failure
|
||||
patch(
|
||||
"agent.verification_stop.build_verify_on_stop_nudge",
|
||||
side_effect=RuntimeError("verify check crashed"),
|
||||
),
|
||||
patch("hermes_cli.plugins.invoke_hook", return_value=[]),
|
||||
):
|
||||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
# No interim emission because verification did not run (exception path
|
||||
# sets _verify_nudge = None, so the candidate becomes the final response
|
||||
# without an interim emission).
|
||||
assert result["final_response"] == "the answer"
|
||||
assert result["completed"] is True
|
||||
agent._handle_max_iterations.assert_not_called()
|
||||
|
||||
|
||||
def test_verify_on_stop_emits_interim_response_to_ui(agent, monkeypatch):
|
||||
"""The verify-on-stop path must emit the full response to the UI callback.
|
||||
|
||||
With no streaming set up in this test, _interim_content_was_streamed
|
||||
returns False, so already_streamed is False — the callback reports
|
||||
content the UI has not seen yet.
|
||||
"""
|
||||
agent._interruptible_api_call = lambda _kwargs: _response("composed report")
|
||||
agent._handle_max_iterations = MagicMock(return_value="replacement summary")
|
||||
monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
|
||||
|
||||
callback_calls = []
|
||||
|
||||
def capture_callback(text, *, already_streamed=None):
|
||||
callback_calls.append({"text": text, "already_streamed": already_streamed})
|
||||
|
||||
agent.interim_assistant_callback = capture_callback
|
||||
|
||||
with (
|
||||
patch("agent.verification_stop.build_verify_on_stop_nudge", return_value="verify it"),
|
||||
patch("hermes_cli.plugins.invoke_hook", return_value=[]),
|
||||
):
|
||||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
# The callback was called with the full response text and already_streamed=False
|
||||
assert len(callback_calls) == 1
|
||||
assert callback_calls[0]["text"] == "composed report"
|
||||
assert callback_calls[0]["already_streamed"] is False
|
||||
|
||||
# The candidate persists as the final response.
|
||||
assert result["final_response"] == "composed report"
|
||||
|
||||
|
||||
def test_streamed_interim_then_different_summary_not_marked_previewed(agent, monkeypatch):
|
||||
"""Ordinary interim narration followed by a different non-streamed summary.
|
||||
|
||||
The model streams "I'll inspect the files now" as an intermediate ack.
|
||||
_emit_interim_assistant_message is called for this ordinary narration,
|
||||
which must NOT set _response_was_previewed. Then _handle_max_iterations
|
||||
produces a different summary through the non-streaming Chat Completions
|
||||
path. The final result must NOT be marked as previewed — the interim was
|
||||
unrelated mid-turn commentary, not the final response — so the CLI renders
|
||||
the summary instead of suppressing it. (#65919 review: response-loss blocker)
|
||||
"""
|
||||
agent.valid_tool_names = ["web_search"]
|
||||
agent._intent_ack_continuation = True
|
||||
agent._looks_like_codex_intermediate_ack = MagicMock(return_value=True)
|
||||
agent._interruptible_api_call = lambda _kwargs: _response("I'll inspect the files now")
|
||||
agent._handle_max_iterations = MagicMock(return_value="Here is the summary of what I found.")
|
||||
monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "0")
|
||||
|
||||
emitted = []
|
||||
agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
|
||||
|
||||
with (
|
||||
patch("hermes_cli.plugins.has_hook", return_value=False),
|
||||
patch("hermes_cli.plugins.invoke_hook", return_value=[]),
|
||||
):
|
||||
result = agent.run_conversation("inspect /tmp/project")
|
||||
|
||||
# The final response is the different summary from _handle_max_iterations.
|
||||
assert result["final_response"] == "Here is the summary of what I found."
|
||||
# CRITICAL: response_previewed must be False — the interim narration was
|
||||
# NOT the final response, so the CLI must render the summary.
|
||||
assert result["response_previewed"] is False
|
||||
|
||||
|
||||
def test_streamed_verification_candidate_reused_marked_previewed(agent, monkeypatch):
|
||||
"""Verification candidate reused at budget exhaustion is marked previewed.
|
||||
|
||||
The model streams a verification candidate that is already streamed as
|
||||
interim content. The continuation budget is exhausted, so the finalizer
|
||||
reuses the pending verification candidate as the final response. The result
|
||||
must be marked as previewed so the CLI/desktop settle it once instead of
|
||||
duplicating. (#65919 review)
|
||||
"""
|
||||
agent._interruptible_api_call = lambda _kwargs: _response("composed report")
|
||||
agent._handle_max_iterations = MagicMock(return_value="replacement summary")
|
||||
monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
|
||||
|
||||
agent._turn_file_mutation_paths = {"changed.py"}
|
||||
|
||||
callback_calls = []
|
||||
|
||||
def capture_callback(text, *, already_streamed=None):
|
||||
callback_calls.append({"text": text, "already_streamed": already_streamed})
|
||||
|
||||
agent.interim_assistant_callback = capture_callback
|
||||
|
||||
# Simulate that the candidate text was already streamed. The streaming
|
||||
# buffer is cleared after the response is processed, so mock the check
|
||||
# directly — this is the condition the test validates: when the candidate
|
||||
# was streamed, the previewed flag propagates to the finalizer.
|
||||
with (
|
||||
patch.object(agent, "_interim_content_was_streamed", return_value=True),
|
||||
patch("agent.verification_stop.build_verify_on_stop_nudge", return_value="verify it"),
|
||||
patch("hermes_cli.plugins.invoke_hook", return_value=[]),
|
||||
):
|
||||
result = agent.run_conversation("edit changed.py")
|
||||
|
||||
# The candidate was already streamed, so the callback reports already_streamed=True.
|
||||
assert len(callback_calls) == 1
|
||||
assert callback_calls[0]["already_streamed"] is True
|
||||
# The candidate is reused as the final response.
|
||||
assert result["final_response"] == "composed report"
|
||||
# CRITICAL: response_previewed must be True — the reused candidate was
|
||||
# streamed as interim content, so the CLI/desktop settle it once.
|
||||
assert result["response_previewed"] is True
|
||||
|
|
|
|||
|
|
@ -0,0 +1,105 @@
|
|||
"""Tests for the interim_assistant_callback config gating in tui_gateway.
|
||||
|
||||
These tests exercise the real _agent_cbs() wiring rather than a local
|
||||
imitation, so a break in the production callback registration is caught.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
def test_load_interim_assistant_messages_defaults_true():
|
||||
from tui_gateway.server import _load_interim_assistant_messages
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={}):
|
||||
assert _load_interim_assistant_messages() is True
|
||||
|
||||
|
||||
def test_load_interim_assistant_messages_explicit_true():
|
||||
from tui_gateway.server import _load_interim_assistant_messages
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": True}}):
|
||||
assert _load_interim_assistant_messages() is True
|
||||
|
||||
|
||||
def test_load_interim_assistant_messages_explicit_false():
|
||||
from tui_gateway.server import _load_interim_assistant_messages
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": False}}):
|
||||
assert _load_interim_assistant_messages() is False
|
||||
|
||||
|
||||
def test_load_interim_assistant_messages_string_off():
|
||||
from tui_gateway.server import _load_interim_assistant_messages
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": "off"}}):
|
||||
assert _load_interim_assistant_messages() is False
|
||||
|
||||
|
||||
def test_agent_cbs_includes_interim_callback_when_enabled():
|
||||
"""_agent_cbs() includes interim_assistant_callback when the config is on.
|
||||
|
||||
Exercises the real _agent_cbs() wiring: the callback must be present in
|
||||
the returned dict and, when invoked, must emit a message.interim event
|
||||
with the text and already_streamed flag passed through.
|
||||
"""
|
||||
from tui_gateway.server import _agent_cbs
|
||||
|
||||
emitted: list[tuple] = []
|
||||
|
||||
def fake_emit(event_type, sid, payload=None):
|
||||
emitted.append((event_type, sid, payload))
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={}), \
|
||||
patch("tui_gateway.server._emit", side_effect=fake_emit):
|
||||
cbs = _agent_cbs("test-session")
|
||||
|
||||
assert "interim_assistant_callback" in cbs
|
||||
cb = cbs["interim_assistant_callback"]
|
||||
assert callable(cb)
|
||||
|
||||
# Invoke the real callback inside the patch context — the lambda
|
||||
# resolves _emit by name at call time, so it must be called while
|
||||
# the patch is active.
|
||||
cb("hello world", already_streamed=True)
|
||||
|
||||
assert len(emitted) == 1
|
||||
assert emitted[0][0] == "message.interim"
|
||||
assert emitted[0][1] == "test-session"
|
||||
assert emitted[0][2]["text"] == "hello world"
|
||||
assert emitted[0][2]["already_streamed"] is True
|
||||
|
||||
|
||||
def test_agent_cbs_omits_interim_callback_when_disabled():
|
||||
"""_agent_cbs() omits interim_assistant_callback when the config is off.
|
||||
|
||||
Exercises the real _agent_cbs() wiring: the callback must NOT be present
|
||||
in the returned dict when display.interim_assistant_messages is false.
|
||||
"""
|
||||
from tui_gateway.server import _agent_cbs
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": False}}):
|
||||
cbs = _agent_cbs("test-session")
|
||||
|
||||
assert "interim_assistant_callback" not in cbs
|
||||
|
||||
|
||||
def test_agent_cbs_interim_callback_passes_already_streamed_false():
|
||||
"""The real callback passes already_streamed=False by default."""
|
||||
from tui_gateway.server import _agent_cbs
|
||||
|
||||
emitted: list[tuple] = []
|
||||
|
||||
def fake_emit(event_type, sid, payload=None):
|
||||
emitted.append((event_type, sid, payload))
|
||||
|
||||
with patch("tui_gateway.server._load_cfg", return_value={}), \
|
||||
patch("tui_gateway.server._emit", side_effect=fake_emit):
|
||||
cbs = _agent_cbs("test-session")
|
||||
|
||||
cb = cbs["interim_assistant_callback"]
|
||||
cb("interim text")
|
||||
|
||||
assert emitted[0][2]["already_streamed"] is False
|
||||
assert emitted[0][2]["text"] == "interim text"
|
||||
|
|
@ -430,6 +430,20 @@ def _load_busy_input_mode() -> str:
|
|||
return raw if raw in {"queue", "steer", "interrupt"} else "interrupt"
|
||||
|
||||
|
||||
def _load_interim_assistant_messages() -> bool:
|
||||
"""Return whether interim assistant commentary should be surfaced to UIs.
|
||||
|
||||
Honors ``display.interim_assistant_messages`` (default true). When false,
|
||||
the tui_gateway does not install ``interim_assistant_callback``, so
|
||||
interim text from tool-call turns and verify-on-stop candidates is never
|
||||
emitted as ``message.interim`` — mirroring the messaging gateway's gating.
|
||||
"""
|
||||
display = _load_cfg().get("display")
|
||||
if not isinstance(display, dict):
|
||||
return True
|
||||
return is_truthy_value(display.get("interim_assistant_messages", True))
|
||||
|
||||
|
||||
def _notify_session_boundary(
|
||||
event_type: str, session_id: str | None, platform: str | None = None
|
||||
) -> None:
|
||||
|
|
@ -4307,7 +4321,7 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
|
|||
|
||||
|
||||
def _agent_cbs(sid: str) -> dict:
|
||||
return {
|
||||
callbacks = {
|
||||
"tool_start_callback": lambda tc_id, name, args: _on_tool_start(
|
||||
sid, tc_id, name, args
|
||||
),
|
||||
|
|
@ -4362,6 +4376,22 @@ def _agent_cbs(sid: str) -> dict:
|
|||
),
|
||||
}
|
||||
|
||||
# Interim assistant commentary (text alongside tool calls, or the attempted
|
||||
# final answer before a verify-on-stop nudge). Gated on
|
||||
# display.interim_assistant_messages (default true). Also set per-turn in
|
||||
# _run_prompt_submit as defense-in-depth — the per-turn set overwrites
|
||||
# this, and the finally block clears it so a stale closure can't fire.
|
||||
if _load_interim_assistant_messages():
|
||||
callbacks["interim_assistant_callback"] = (
|
||||
lambda text, *, already_streamed=False: _emit(
|
||||
"message.interim",
|
||||
sid,
|
||||
{"text": str(text), "already_streamed": bool(already_streamed)},
|
||||
)
|
||||
)
|
||||
|
||||
return callbacks
|
||||
|
||||
|
||||
def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
|
||||
"""Intentional workspace move from the project_* tools: re-anchor the live
|
||||
|
|
@ -10034,6 +10064,22 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
|
|||
payload["rendered"] = r
|
||||
_emit("message.delta", sid, payload)
|
||||
|
||||
# Surface interim assistant text (commentary emitted alongside
|
||||
# tool calls, or the attempted final answer before a verify-on-stop
|
||||
# nudge) so the desktop can seal it as its own segment instead of
|
||||
# losing it when message.complete replaces the streaming buffer.
|
||||
# Gated on display.interim_assistant_messages (default true).
|
||||
if _load_interim_assistant_messages():
|
||||
def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
|
||||
_emit("message.interim", sid, {
|
||||
"text": text,
|
||||
"already_streamed": already_streamed,
|
||||
})
|
||||
|
||||
agent.interim_assistant_callback = _interim_assistant_cb
|
||||
else:
|
||||
agent.interim_assistant_callback = None
|
||||
|
||||
run_kwargs = {
|
||||
"conversation_history": list(history),
|
||||
"stream_callback": _stream,
|
||||
|
|
@ -10155,6 +10201,8 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
|
|||
payload["reasoning"] = last_reasoning
|
||||
if status_note:
|
||||
payload["warning"] = status_note
|
||||
if result.get("response_previewed"):
|
||||
payload["response_previewed"] = True
|
||||
rendered = render_message(raw, cols)
|
||||
if rendered:
|
||||
payload["rendered"] = rendered
|
||||
|
|
@ -10338,6 +10386,9 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
|
|||
if home_token is not None:
|
||||
reset_hermes_home_override(home_token)
|
||||
_clear_session_context(session_tokens)
|
||||
# Clear the per-turn interim callback so a stale closure from
|
||||
# this turn can't fire during a later turn on the same agent.
|
||||
agent.interim_assistant_callback = None
|
||||
with session["history_lock"]:
|
||||
session["running"] = False
|
||||
session["last_active"] = time.time()
|
||||
|
|
|
|||
|
|
@ -1683,4 +1683,81 @@ describe('createGatewayEventHandler', () => {
|
|||
expect(openExternalUrlMock).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
describe('message.interim', () => {
|
||||
it('finalizes an interim segment without settling the turn', () => {
|
||||
const appended: Msg[] = []
|
||||
const onEvent = createGatewayEventHandler(buildCtx(appended))
|
||||
|
||||
onEvent({ payload: {}, type: 'message.start' } as any)
|
||||
onEvent({ payload: { text: 'streaming text' }, type: 'message.delta' } as any)
|
||||
onEvent({ payload: { already_streamed: true, text: 'streaming text' }, type: 'message.interim' } as any)
|
||||
|
||||
// Turn is still active — busy stays true, no completion messages appended
|
||||
expect(getUiState().busy).toBe(true)
|
||||
expect(appended).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('keeps identical interim and terminal replies as separate messages without response_previewed', () => {
|
||||
const appended: Msg[] = []
|
||||
const onEvent = createGatewayEventHandler(buildCtx(appended))
|
||||
|
||||
onEvent({ payload: {}, type: 'message.start' } as any)
|
||||
onEvent({ payload: { already_streamed: true, text: 'same reply' }, type: 'message.interim' } as any)
|
||||
onEvent({ payload: { text: 'same reply' }, type: 'message.complete' } as any)
|
||||
|
||||
const assistantMsgs = appended.filter(m => m.role === 'assistant' && m.text)
|
||||
expect(assistantMsgs).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('settles identical terminal reply onto interim when response_previewed', () => {
|
||||
const appended: Msg[] = []
|
||||
const onEvent = createGatewayEventHandler(buildCtx(appended))
|
||||
|
||||
onEvent({ payload: {}, type: 'message.start' } as any)
|
||||
onEvent({ payload: { already_streamed: true, text: 'same reply' }, type: 'message.interim' } as any)
|
||||
onEvent({ payload: { response_previewed: true, text: 'same reply' }, type: 'message.complete' } as any)
|
||||
|
||||
// With response_previewed, the terminal reply is the same model
|
||||
// response that was published provisionally — settle onto the
|
||||
// interim instead of duplicating. (#65919 review)
|
||||
const assistantMsgs = appended.filter(m => m.role === 'assistant' && m.text)
|
||||
expect(assistantMsgs).toHaveLength(1)
|
||||
expect(assistantMsgs[0]?.text).toBe('same reply')
|
||||
})
|
||||
|
||||
it('deduplicates flushed chunks within the terminal message after an interim boundary', () => {
|
||||
const appended: Msg[] = []
|
||||
const onEvent = createGatewayEventHandler(buildCtx(appended))
|
||||
|
||||
onEvent({ payload: {}, type: 'message.start' } as any)
|
||||
// Interim seals the first segment
|
||||
onEvent({ payload: { already_streamed: true, text: 'interim answer' }, type: 'message.interim' } as any)
|
||||
// Post-interim deltas that match the final text — these get deduped
|
||||
onEvent({ payload: { text: 'final answer' }, type: 'message.delta' } as any)
|
||||
onEvent({ payload: { text: 'final answer' }, type: 'message.complete' } as any)
|
||||
|
||||
const texts = appended.filter(m => m.role === 'assistant' && m.text).map(m => m.text)
|
||||
// interim + final, no duplication of the final
|
||||
expect(texts).toContain('interim answer')
|
||||
expect(texts.filter(t => t === 'final answer')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('ignores malformed message.interim payload', () => {
|
||||
const appended: Msg[] = []
|
||||
const onEvent = createGatewayEventHandler(buildCtx(appended))
|
||||
|
||||
onEvent({ payload: {}, type: 'message.start' } as any)
|
||||
// No payload at all
|
||||
onEvent({ type: 'message.interim' } as any)
|
||||
// Empty text
|
||||
onEvent({ payload: { text: '' }, type: 'message.interim' } as any)
|
||||
// Undefined text
|
||||
onEvent({ payload: { text: undefined }, type: 'message.interim' } as any)
|
||||
|
||||
// Turn continues without finalizing or throwing
|
||||
expect(getUiState().busy).toBe(true)
|
||||
expect(appended).toHaveLength(0)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -946,6 +946,14 @@ export function createGatewayEventHandler(ctx: GatewayEventHandlerContext): (ev:
|
|||
turnController.recordMessageDelta(ev.payload ?? {})
|
||||
|
||||
return
|
||||
case 'message.interim': {
|
||||
const text = ev.payload?.text
|
||||
if (typeof text === 'string' && text.trim()) {
|
||||
turnController.recordInterimMessage(text)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
case 'message.complete': {
|
||||
const { finalMessages, finalText, wasInterrupted } = turnController.recordMessageComplete(ev.payload ?? {})
|
||||
|
||||
|
|
|
|||
|
|
@ -126,6 +126,7 @@ class TurnController {
|
|||
private activeTools: ActiveTool[] = []
|
||||
private activeReasoningText = ''
|
||||
private reasoningSegmentIndex: null | number = null
|
||||
private interimBoundaryIndex: null | number = null
|
||||
private activityId = 0
|
||||
private reasoningStreamingTimer: Timer = null
|
||||
private reasoningTimer: Timer = null
|
||||
|
|
@ -554,7 +555,7 @@ class TurnController {
|
|||
this.flushPendingNotice()
|
||||
}
|
||||
|
||||
recordMessageComplete(payload: { rendered?: string; reasoning?: string; text?: string }) {
|
||||
recordMessageComplete(payload: { rendered?: string; reasoning?: string; response_previewed?: boolean; text?: string }) {
|
||||
this.closeReasoningSegment()
|
||||
|
||||
// Ink renders markdown via <Md>; the gateway's Rich-rendered ANSI
|
||||
|
|
@ -565,7 +566,15 @@ class TurnController {
|
|||
// only when the gateway elected not to send any (#16391).
|
||||
const rawText = (payload.text ?? payload.rendered ?? this.bufRef).trimStart()
|
||||
const split = splitReasoning(rawText)
|
||||
const finalText = finalTail(split.text, this.segmentMessages)
|
||||
// Only dedupe segments AFTER the interim boundary — interim-sealed
|
||||
// segments are preserved even if the final text includes them.
|
||||
// Exception: when response_previewed is true, the final text is the
|
||||
// same model response that was published provisionally as an interim
|
||||
// message. Dedupe against ALL segments (including sealed interims) so
|
||||
// the identical text doesn't render as a duplicate message. (#65919
|
||||
// review: duplicate-message blocker)
|
||||
const dedupeStart = payload.response_previewed ? 0 : (this.interimBoundaryIndex ?? 0)
|
||||
const finalText = finalTail(split.text, this.segmentMessages.slice(dedupeStart))
|
||||
const existingReasoning = this.reasoningText.trim() || String(payload.reasoning ?? '').trim()
|
||||
const savedReasoning = [existingReasoning, existingReasoning ? '' : split.reasoning].filter(Boolean).join('\n\n')
|
||||
const savedToolTokens = this.toolTokenAcc
|
||||
|
|
@ -672,6 +681,31 @@ class TurnController {
|
|||
}
|
||||
}
|
||||
|
||||
recordInterimMessage(text: string) {
|
||||
if (this.interrupted) {
|
||||
return
|
||||
}
|
||||
|
||||
const authoritativeText = text.trimStart()
|
||||
if (!authoritativeText) {
|
||||
return
|
||||
}
|
||||
|
||||
// If the streaming buffer hasn't caught up to the authoritative interim
|
||||
// text (e.g. the backend didn't stream every token), sync it so the
|
||||
// sealed segment matches what the user should see.
|
||||
if (this.bufRef.trimStart() !== authoritativeText) {
|
||||
this.bufRef = authoritativeText
|
||||
}
|
||||
|
||||
// Flush the current streaming buffer into a sealed segment — this is the
|
||||
// TUI equivalent of the desktop's finalizeInterimAssistantMessage. The
|
||||
// segment survives message.complete's finalTail dedupe because
|
||||
// interimBoundaryIndex marks it as interim-sealed.
|
||||
this.flushStreamingSegment()
|
||||
this.interimBoundaryIndex = this.segmentMessages.length
|
||||
}
|
||||
|
||||
recordReasoningAvailable(text: string, force = false) {
|
||||
if (this.interrupted || (!force && !getUiState().showReasoning)) {
|
||||
return
|
||||
|
|
@ -885,6 +919,7 @@ class TurnController {
|
|||
this.pendingSegmentTools = []
|
||||
this.protocolWarned = false
|
||||
this.reasoningSegmentIndex = null
|
||||
this.interimBoundaryIndex = null
|
||||
this.segmentMessages = []
|
||||
this.turnTools = []
|
||||
this.toolTokenAcc = 0
|
||||
|
|
@ -941,6 +976,7 @@ class TurnController {
|
|||
this.activeTools = []
|
||||
this.activeReasoningText = ''
|
||||
this.reasoningSegmentIndex = null
|
||||
this.interimBoundaryIndex = null
|
||||
this.turnTools = []
|
||||
this.toolTokenAcc = 0
|
||||
this.interrupted = false
|
||||
|
|
|
|||
|
|
@ -641,7 +641,12 @@ export type GatewayEvent =
|
|||
| { payload: SubagentEventPayload; session_id?: string; type: 'subagent.complete' }
|
||||
| { payload: { rendered?: string; text?: string }; session_id?: string; type: 'message.delta' }
|
||||
| {
|
||||
payload?: { reasoning?: string; rendered?: string; text?: string; usage?: Usage }
|
||||
payload: { already_streamed?: boolean; text: string }
|
||||
session_id?: string
|
||||
type: 'message.interim'
|
||||
}
|
||||
| {
|
||||
payload?: { reasoning?: string; rendered?: string; response_previewed?: boolean; text?: string; usage?: Usage }
|
||||
session_id?: string
|
||||
type: 'message.complete'
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue