fix(agent): close provisional event-envelope regressions #4515
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -114,6 +114,7 @@ const MANAGED_LOCAL_FAILURE_STAGES = [ | |
| "staging.losslessSnapshot", | ||
| "staging.measure", | ||
| "staging.sanitize", | ||
| "staging.preMeasure", | ||
| "staging.overflow", | ||
| "overflow.preMeasure", | ||
| "overflow.staged", | ||
|
|
@@ -868,7 +869,16 @@ function losslessDetachedClone<T>(value: T): T { | |
| } catch { | ||
| if (key === "transportFailure" && isManagedPlainRecord(descriptor.value)) { | ||
| const transport: Record<string, unknown> = {}; | ||
| for (const transportKey of [ | ||
| "kind", | ||
| "status", | ||
| "code", | ||
| "providerCode", | ||
| "openaiErrorCode", | ||
| "anthropicErrorType", | ||
| "retryAfterMs", | ||
| "headers", | ||
| ] as const) { | ||
| const transportDescriptor = Object.getOwnPropertyDescriptor(descriptor.value, transportKey); | ||
| if (!transportDescriptor || !("value" in transportDescriptor)) continue; | ||
| try { | ||
|
|
@@ -1147,11 +1157,12 @@ function warnManagedSnapshotFailure( | |
| * cancelled provider attempt is therefore unobservable to sessions and their | ||
| * side-effect consumers. Non-managed streams bypass this object entirely. | ||
| */ | ||
| type ManagedAttemptBatchItem = | ||
| | { type: "event"; event: AgentEvent } | ||
| | { type: "assistant_event"; message: AssistantMessage; event: AssistantMessageEvent }; | ||
|
|
||
| class ManagedAttemptTransaction { | ||
| #batch: ManagedAttemptBatchItem[] = []; | ||
| #stagedEventCount = 0; | ||
| #stagedBytes = 0; | ||
| /** Shape snapshot retained across discard() for bounded failure diagnostics. */ | ||
|
|
@@ -1171,6 +1182,10 @@ class ManagedAttemptTransaction { | |
|
|
||
| push(event: AgentEvent): void { | ||
| if (this.#committed) { | ||
| if (event.type === "message_end" || event.type === "turn_end") { | ||
| this.#batch.push({ type: "event", event }); | ||
| return; | ||
| } | ||
| this.stream.push(event); | ||
| return; | ||
| } | ||
|
|
@@ -1195,7 +1210,7 @@ class ManagedAttemptTransaction { | |
| } | ||
|
|
||
| flush(): void { | ||
| if (this.#discarded) return; | ||
| for (const item of this.#batch) { | ||
| if (item.type === "assistant_event") { | ||
| this.onAssistantMessageEvent?.(item.message, item.event); | ||
|
|
@@ -1209,6 +1224,47 @@ class ManagedAttemptTransaction { | |
| this.#committed = true; | ||
| } | ||
|
|
||
| flushNonTerminal(): void { | ||
| if (this.#discarded || this.#committed) return; | ||
| const retained: ManagedAttemptBatchItem[] = []; | ||
| for (const item of this.#batch) { | ||
| if (this.#isTerminalItem(item)) { | ||
| retained.push(item); | ||
| } else if (item.type === "assistant_event") { | ||
| this.onAssistantMessageEvent?.(item.message, item.event); | ||
| } else { | ||
| this.stream.push(item.event); | ||
| } | ||
| } | ||
| this.#batch = retained; | ||
| } | ||
|
|
||
| commitCallbacksAndUpdates(): void { | ||
| if (this.#discarded || this.#committed) return; | ||
| for (const item of this.#batch) { | ||
| if (item.type === "assistant_event") { | ||
| this.onAssistantMessageEvent?.(item.message, item.event); | ||
| } else if (item.event.type !== "message_end" && item.event.type !== "turn_end") { | ||
| this.stream.push(item.event); | ||
| } | ||
| } | ||
| this.#batch = this.#batch.filter( | ||
| item => item.type === "event" && (item.event.type === "message_end" || item.event.type === "turn_end"), | ||
| ); | ||
| this.#committed = true; | ||
| } | ||
|
|
||
| replacePendingAssistantMessage(message: AssistantMessage): void { | ||
| this.#batch = this.#batch.map(item => { | ||
| if (item.type === "assistant_event") { | ||
| return { ...item, message, event: this.#assistantEventSnapshot(item.event, message) }; | ||
| } | ||
| if (item.event.type === "message_end") return { ...item, event: { ...item.event, message } }; | ||
| if (item.event.type === "turn_end") return { ...item, event: { ...item.event, message } }; | ||
| return item; | ||
| }); | ||
| } | ||
|
|
||
| get committed(): boolean { | ||
| return this.#committed; | ||
| } | ||
|
|
@@ -1270,9 +1326,19 @@ class ManagedAttemptTransaction { | |
| #stage(event: AgentEvent): void { | ||
| if (this.snapshotMode === "lossless") { | ||
| const snapshot = this.#repairAssistantEvent(event); | ||
| let rawBytes: number | undefined; | ||
| try { | ||
| rawBytes = managedAttemptTextEncoder.encode(JSON.stringify(snapshot)).byteLength; | ||
| } catch { | ||
| rawBytes = undefined; | ||
| } | ||
| if (rawBytes !== undefined && this.#wouldOverflow(rawBytes)) { | ||
| this.discard(); | ||
| throw new ManagedAttemptSnapshotError("staging.preMeasure"); | ||
| } | ||
| let detached: AgentEvent; | ||
| try { | ||
| detached = this.#losslessAgentEventSnapshot(snapshot); | ||
| } catch { | ||
| this.discard(); | ||
| throw new ManagedAttemptSnapshotError("staging.losslessSnapshot"); | ||
|
|
@@ -1376,6 +1442,28 @@ class ManagedAttemptTransaction { | |
| return losslessDetachedClone(value); | ||
| } | ||
|
|
||
| #losslessAgentEventSnapshot(event: AgentEvent): AgentEvent { | ||
| switch (event.type) { | ||
| case "message_start": | ||
| case "message_end": | ||
| return { ...event, message: this.#losslessSnapshot(event.message) }; | ||
| case "message_update": { | ||
| const message = this.#losslessSnapshot(event.message); | ||
| if (message.role !== "assistant") return { ...event, message }; | ||
| const assistantMessageEvent = this.#assistantEventSnapshot(event.assistantMessageEvent, message); | ||
|
There was a problem hiding this comment.
When a streamed partial contains the live Useful? React with 👍 / 👎. |
||
| return { ...event, message, assistantMessageEvent }; | ||
| } | ||
| case "turn_end": | ||
| return { | ||
| ...event, | ||
| message: this.#losslessSnapshot(event.message), | ||
| toolResults: this.#losslessSnapshot(event.toolResults), | ||
| }; | ||
| default: | ||
| return this.#losslessSnapshot(event); | ||
| } | ||
| } | ||
|
|
||
| #assistantSnapshot(message: AssistantMessage): AssistantMessage { | ||
| return this.snapshotMode === "lossless" | ||
| ? this.#losslessSnapshot(message) | ||
|
|
@@ -1387,8 +1475,13 @@ class ManagedAttemptTransaction { | |
| const snapshot = this.#losslessSnapshot(event); | ||
| if (snapshot.type === "done") return { ...snapshot, message }; | ||
| if (snapshot.type === "error") return { ...snapshot, error: message }; | ||
| if (snapshot.type === "toolChoiceIncapability") return snapshot; | ||
| return { ...snapshot, partial: message }; | ||
| } | ||
|
|
||
| #isTerminalItem(item: ManagedAttemptBatchItem): boolean { | ||
| if (item.type === "assistant_event") return item.event.type === "done" || item.event.type === "error"; | ||
| return item.event.type === "message_end" || item.event.type === "turn_end"; | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -2043,6 +2136,18 @@ async function runLoopBody( | |
| } | ||
| await emitHarmonyAudit(config, err, "truncate_resume", harmonyRetryAttempt); | ||
| } else { | ||
| if (escapedToolTransaction?.committed) { | ||
| const contaminated = currentContext.messages.at(-1); | ||
| if (contaminated?.role !== "assistant") throw err; | ||
| const sanitized = escapedToolTransaction.acceptedAssistantSnapshot({ | ||
| ...contaminated, | ||
| content: [], | ||
| stopReason: "aborted", | ||
| providerPayload: undefined, | ||
| }); | ||
| escapedToolTransaction.replacePendingAssistantMessage(sanitized); | ||
| escapedToolTransaction.flush(); | ||
| } | ||
| if (harmonyRetryAttempt >= 2) { | ||
| await emitHarmonyAudit(config, err, "escalated", harmonyRetryAttempt); | ||
| throw new Error( | ||
|
|
@@ -2144,10 +2249,11 @@ async function runLoopBody( | |
| // back into the context the model samples from next. Drop the defective | ||
| // turn and re-request instead; the per-call rejection in | ||
| // `executeToolCalls` stays as the terminal answer once this budget is | ||
| // spent. Managed fallback reports the discarded attempt through the | ||
| // typed `escaped_arguments_discarded` outcome so the session policy | ||
| // owns a bounded same-model retry; the defect is never treated as | ||
| // provider evidence, so the fallback chain never advances on it. | ||
| if ( | ||
| message.stopReason !== "error" && | ||
| message.stopReason !== "aborted" && | ||
| escapedNonAsciiResampleAttempt < MAX_ESCAPED_NONASCII_RESAMPLES && | ||
|
|
@@ -2161,6 +2267,24 @@ async function runLoopBody( | |
| // still the tail: callbacks may append user/system history while the | ||
| // response settles, and none of that history belongs to this retry. | ||
| removeCommittedAssistantMessage(currentContext.messages, message); | ||
| // A managed invocation ends the run here and reports the discarded | ||
| // attempt to the session's fallback policy through the typed | ||
| // outcome below; the policy owns the same-model bounded retry and | ||
| // only falls back once it declines. The wire defect is not provider | ||
| // evidence, so the outcome deliberately carries no transport facts | ||
| // and the fallback chain never advances on it. | ||
| if (config.fallbackManaged) { | ||
| transaction?.discard(); | ||
| currentContext.messages.splice(contextMessageCount); | ||
| newMessages.splice(newMessageCount); | ||
| await config.onManagedAttemptOutcome?.({ | ||
| type: "escaped_arguments_discarded", | ||
| message, | ||
| scope: transaction?.scope, | ||
| }); | ||
| stream.end(newMessages); | ||
| return; | ||
| } | ||
| continue; | ||
| } | ||
| escapedNonAsciiResampleAttempt = 0; | ||
|
|
@@ -2220,23 +2344,27 @@ async function runLoopBody( | |
| } | ||
|
|
||
| // One provider invocation is committed before any tool can run. | ||
| if (escapedToolTransaction) { | ||
| const acceptedMessage = escapedToolTransaction.acceptedAssistantSnapshot(message); | ||
| const contextIndex = currentContext.messages.lastIndexOf(message); | ||
| if (contextIndex >= 0) currentContext.messages[contextIndex] = acceptedMessage; | ||
| const producedIndex = newMessages.lastIndexOf(message); | ||
| if (producedIndex >= 0) newMessages[producedIndex] = acceptedMessage; | ||
| message = acceptedMessage; | ||
| escapedToolTransaction.flushNonTerminal(); | ||
| // Tool-call updates are staged so an escaped turn can disappear | ||
| // atomically. Once accepted, drain every published update through the | ||
| // Agent/AgentSession consumers before dispatch: streaming edit guards | ||
| // can then abort the run before any tool execute() is entered. | ||
| if (message.stopReason !== "aborted" && message.stopReason !== "error") { | ||
| if (loopSignal.aborted) message.stopReason = "aborted"; | ||
|
There was a problem hiding this comment.
This commit changes shipped behavior in both AGENTS.md reference: AGENTS.md:L188-L188 Useful? React with 👍 / 👎. |
||
| if (stream.hasActiveConsumer) await stream.waitForConsumerDrain(new AbortController().signal); | ||
| if (loopSignal.aborted) message.stopReason = "aborted"; | ||
|
There was a problem hiding this comment.
When an event consumer aborts during Useful? React with 👍 / 👎. |
||
| } | ||
| escapedToolTransaction.replacePendingAssistantMessage(message); | ||
| escapedToolTransaction.flush(); | ||
|
Comment on lines
+2364
to
+2365
There was a problem hiding this comment.
When an accepted tool-call response streamed text first, the text path at Useful? React with 👍 / 👎. |
||
| } else { | ||
| transaction?.flush(); | ||
| } | ||
| if (config.fallbackManaged && message.stopReason !== "error" && message.stopReason !== "aborted") { | ||
| await config.onManagedAttemptAccepted?.(); | ||
|
|
@@ -2860,7 +2988,7 @@ async function streamAssistantResponse( | |
| // a later escaped tool call therefore falls through to the | ||
| // existing terminal per-call rejection instead. | ||
| if (event.type === "text_start" || event.type === "text_delta" || event.type === "text_end") { | ||
| provisionalToolTransaction?.commitCallbacksAndUpdates(); | ||
| } | ||
| } | ||
| break; | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When a response streams visible text and then triggers an unrecoverable text-surface Harmony leak, the text path marks this transaction committed, so its subsequent
message_endis retained here. Theabort_retrybranch at line 1993 then continues with a new transaction without flushing or discarding this batch, leaving subscribers with a publishedmessage_start/message_updatelifecycle that never receivesmessage_endand potentially leaving leaked text orphaned in the TUI. Settle the retained terminal—or explicitly reconcile the visible attempt—before taking that retry path.Useful? React with 👍 / 👎.