Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/api-reference/veryfront/agent.md
Original file line number Diff line number Diff line change
Expand Up @@ -1040,7 +1040,7 @@ Input delivered to a hosted agent-service detached execution callback.

| Name | Description | Source |
| ------------------------------------ | ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ |
| `AgentRuntime` | Implement agent runtime. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/index.ts#L840) |
| `AgentRuntime` | Implement agent runtime. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/index.ts#L844) |
| `AgentRuntimeMessageConversionError` | Error shape for agent runtime message conversion. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/runtime/message-adapter.ts#L138) |
| `AgentServiceAuthError` | Error shape for hosted service auth. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/auth.ts#L14) |
| `AppendConversationRunEventsError` | Error shape for append conversation run events. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable-append-errors.ts#L4) |
Expand Down
58 changes: 37 additions & 21 deletions src/agent/runtime/chat-stream-handler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,42 @@ export interface StreamingToolResult {
preliminary?: boolean;
}

/**
* Flush a tool call's buffered `tool-input-start` and input deltas to the
* client. The start event is withheld until the call's name is final, so a
* delegated remote tool is never announced under its unresolved namespace
* (#3123). Idempotent: `inputAnnounced` records that the buffer was drained,
* and it is also what gates the terminal `tool-output-error`, so a call that
* was never announced must be announced here before its failure can render.
*/
export function announceStreamedToolCallInput(
controller: ReadableStreamDefaultController,
encoder: TextEncoder,
toolCall: StreamingToolCall,
): void {
if (toolCall.inputAnnounced === true) {
return;
}

const dynamic = toolCall.dynamic ?? isDynamicTool(toolCall.name);
sendSSE(controller, encoder, {
type: "tool-input-start",
toolCallId: toolCall.id,
toolName: toolCall.name,
...(dynamic ? { dynamic: true } : {}),
});

for (const delta of toolCall.inputDeltas ?? []) {
sendSSE(controller, encoder, {
type: "tool-input-delta",
toolCallId: toolCall.id,
inputTextDelta: delta,
});
}

toolCall.inputAnnounced = true;
}

export interface StreamingReasoningPart {
id: string;
text: string;
Expand Down Expand Up @@ -772,27 +808,7 @@ export function processStreamInternal(
};

const announceToolInputStart = (toolCall: StreamingToolCall) => {
if (toolCall.inputAnnounced === true) {
return;
}

const dynamic = toolCall.dynamic ?? isDynamicTool(toolCall.name);
sendSSE(controller, encoder, {
type: "tool-input-start",
toolCallId: toolCall.id,
toolName: toolCall.name,
...(dynamic ? { dynamic: true } : {}),
});

for (const delta of toolCall.inputDeltas ?? []) {
sendSSE(controller, encoder, {
type: "tool-input-delta",
toolCallId: toolCall.id,
inputTextDelta: delta,
});
}

toolCall.inputAnnounced = true;
announceStreamedToolCallInput(controller, encoder, toolCall);
};

const ensureToolLifecycle = (part: {
Expand Down
53 changes: 49 additions & 4 deletions src/agent/runtime/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ import {
} from "./mcp-server-tool-sources.ts";
import { runWithRuntimeRemoteToolSources } from "./remote-tool-source-context.ts";
import {
announceStreamedToolCallInput,
createStreamState,
processStream,
type StreamingToolCall,
Expand Down Expand Up @@ -104,7 +105,10 @@ import {
prepareAgentRuntimeStep,
withIntegrationToolDiscoveryStatus,
} from "./agent-runtime-step.ts";
import { buildStreamedAssistantMessage } from "./streamed-assistant-message.ts";
import {
buildStreamedAssistantMessage,
isPersistedReasoningPart,
} from "./streamed-assistant-message.ts";
import {
type DeferredToolSummary,
flattenSystemInstructions,
Expand Down Expand Up @@ -2256,14 +2260,41 @@ export class AgentRuntime {

const streamedToolCalls = Array.from(state.toolCalls.values());
const finalToolResults = collectFinalStreamToolResults(state);
// Recovery replays the whole step, so it also re-emits this step's
// reasoning — duplicating it in the live stream and in history, with a
// signature that no longer matches the replayed content. Reasoning that
// was persisted is reasoning the client already saw, so fail closed.
// This is a stopgap: reasoning is default-on across the hosted catalog,
// which makes recovery inert on most hosted paths. See #3736 for the
// reconciliation protocol that would let it run again.
const hasExposedReasoning = state.reasoningParts.some(isPersistedReasoningPart);
const canRecoverInterruptedLocalToolBatch = !recoveredInterruptedLocalToolBatch &&
step + 1 < maxSteps;
step + 1 < maxSteps &&
!hasExposedReasoning;
const shouldContinue = shouldContinueAfterStreamStep(state, {
recoverInterruptedToolCalls: canRecoverInterruptedLocalToolBatch,
});
const shouldRecoverInterruptedLocalToolBatch = canRecoverInterruptedLocalToolBatch &&
shouldContinue &&
streamedToolCalls.some(isInterruptedClientToolCall);
// Exactly `shouldRecoverInterruptedLocalToolBatch` with the reasoning
// gate lifted: the batch this step would have replayed had it not
// already exposed reasoning. Re-asking is what separates "recovery was
// declined" from "this step merely carried reasoning";
// `shouldContinueAfterStreamStep` only reads state, so asking twice has
// no side effects, and the cheap conditions short-circuit ahead of it.
const declinedRecoveryForExposedReasoning = hasExposedReasoning &&
!recoveredInterruptedLocalToolBatch &&
step + 1 < maxSteps &&
streamedToolCalls.some(isInterruptedClientToolCall) &&
shouldContinueAfterStreamStep(state, { recoverInterruptedToolCalls: true });
if (declinedRecoveryForExposedReasoning) {
logger.warn("Declined interrupted local tool batch recovery after exposed reasoning", {
step,
toolName: streamedToolCalls.find(isInterruptedClientToolCall)?.name,
reasoningPartCount: state.reasoningParts.filter(isPersistedReasoningPart).length,
});
}
const assistantMessage = buildStreamedAssistantMessage({
...state,
accumulatedText: recoveryPresentationText,
Expand Down Expand Up @@ -2361,7 +2392,7 @@ export class AgentRuntime {

const recordIncompleteLocalToolError = async (
toolCall: StreamingToolCall,
options: { includeInResponse?: boolean } = {},
options: { includeInResponse?: boolean; announceInput?: boolean } = {},
): Promise<boolean> => {
if (
toolCall.providerExecuted === true ||
Expand All @@ -2370,6 +2401,18 @@ export class AgentRuntime {
) {
return false;
}
if (options.announceInput === true) {
// An interrupted call never reached `tool-input-end`, so its
// `tool-input-start` is still buffered and `inputAnnounced` is false
// — which would suppress the `tool-output-error` below. On the
// declined-recovery path that leaves the client with a reasoning
// block and then nothing at all. The run is terminating and the tool
// name is final, so flush the buffer and let the failure render as a
// tool card matching what the persisted message already records.
// Announcing is idempotent, so a call surfaced upstream is not
// reported twice.
announceStreamedToolCallInput(controller, encoder, toolCall);
Comment thread
kwakayama marked this conversation as resolved.
Resolved
}
const incompleteToolCall: ToolCall = {
id: toolCall.id,
name: toolCall.name,
Expand Down Expand Up @@ -2398,7 +2441,9 @@ export class AgentRuntime {
await persistToolResult(toolResult);
}
for (const toolCall of streamedToolCalls) {
await recordIncompleteLocalToolError(toolCall);
await recordIncompleteLocalToolError(toolCall, {
announceInput: declinedRecoveryForExposedReasoning,
});
}
sendSSE(controller, encoder, { type: "step-end" });
break;
Expand Down
Loading