diff --git a/packages/junior/src/api/conversations/events.ts b/packages/junior/src/api/conversations/events.ts index dfb55b653..3495f3654 100644 --- a/packages/junior/src/api/conversations/events.ts +++ b/packages/junior/src/api/conversations/events.ts @@ -12,6 +12,7 @@ export const conversationReportSourceEventTypes = [ "message_handled", "agent_step", "tool_execution_started", + "tool_execution_completed", "turn_started", "turn_routed", "turn_completed", @@ -181,6 +182,9 @@ export function conversationReportToolResultIds( return [ ...new Set( events.flatMap((event) => { + if (event.data.type === "tool_execution_completed") { + return [event.data.toolCallId]; + } if (event.data.type !== "agent_step") return []; const result = reportingToolResultMessageSchema.safeParse( event.data.message, @@ -346,6 +350,24 @@ export function projectConversationReportEventPage(args: { }, ], }; + } else if (event.data.type === "tool_execution_completed") { + const start = toolStarts.get(event.data.toolCallId); + data = { + type: "tool_calls", + calls: [ + { + toolCallId: event.data.toolCallId, + name: event.data.toolName, + status: event.data.outcome, + ...(start && start.seq < event.seq + ? { + startedAt: new Date(start.createdAtMs).toISOString(), + startedSeq: start.seq, + } + : {}), + }, + ], + }; } else if (event.data.type === "subagent_started") { subagentStarts.set(event.data.subagentInvocationId, { createdAtMs: event.createdAtMs, diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 1d5885999..05dbb9866 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -41,6 +41,7 @@ import { loadConnectedMcpProviders, loadTurnRoute, openConversationProjection, + recordToolExecutionCompleted, recordToolExecutionStarted, recordMcpProviderConnected, recordTurnRoute, @@ -410,17 +411,37 @@ async function executeAgentRunInPrivacyContext( surface, }); const runResume = resume; - const recordParentToolExecutionStart = async (event: { - args: unknown; - toolCallId: string; - toolName: string; - }) => { + let pendingToolActivityWrites = Promise.resolve(); + const recordParentToolExecution = async (event: + | { + args: unknown; + toolCallId: string; + toolName: string; + type: "tool_execution_start"; + } + | { + isError: boolean; + result: unknown; + toolCallId: string; + toolName: string; + type: "tool_execution_end"; + }, + ) => { try { - await recordToolExecutionStarted({ - conversationId, - toolCallId: event.toolCallId, - toolName: event.toolName, - }); + if (event.type === "tool_execution_start") { + await recordToolExecutionStarted({ + conversationId, + toolCallId: event.toolCallId, + toolName: event.toolName, + }); + } else { + await recordToolExecutionCompleted({ + conversationId, + isError: event.isError, + toolCallId: event.toolCallId, + toolName: event.toolName, + }); + } } catch (error) { // Host-only activity events are best-effort reporting writes; a // failed append must not abort the in-flight model turn. @@ -431,10 +452,17 @@ async function executeAgentRunInPrivacyContext( { "gen_ai.tool.name": event.toolName, }, - "Failed to record host-only tool execution start", + "Failed to record host-only tool execution", ); } }; + const enqueueParentToolExecution = ( + event: Parameters[0], + ): void => { + pendingToolActivityWrites = pendingToolActivityWrites.then(() => + recordParentToolExecution(event), + ); + }; const persistedConfigurationValues = policy.channelConfiguration ? await policy.channelConfiguration.resolveValues() : {}; @@ -998,16 +1026,25 @@ async function executeAgentRunInPrivacyContext( }); const unsubscribe = agent.subscribe((event) => { - if (event.type === "tool_execution_start") { - return recordParentToolExecutionStart(event); + if ( + event.type === "tool_execution_start" || + event.type === "tool_execution_end" + ) { + // Pi emits tool_execution_end before appending the corresponding + // toolResult message. Queue reporting writes without blocking that + // lifecycle so timeout recovery can snapshot the continuable result. + enqueueParentToolExecution(event); + return; } if (event.type === "turn_end" && event.toolResults.length > 0) { if (pendingHandoff) { - return; + return pendingToolActivityWrites; } - return runResume - .persistSafeBoundary([...agent!.state.messages]) - .then(() => undefined); + return pendingToolActivityWrites.then(() => + runResume + .persistSafeBoundary([...agent!.state.messages]) + .then(() => undefined), + ); } if (event.type === "message_end" && isAssistantMessage(event.message)) { if ( @@ -1336,6 +1373,7 @@ async function executeAgentRunInPrivacyContext( return authPauseOutcome; } } finally { + await pendingToolActivityWrites; unsubscribe(); } diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index a931afa09..00386e630 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -159,6 +159,15 @@ const toolExecutionStartedEventDataSchema = z }) .strict(); +const toolExecutionCompletedEventDataSchema = z + .object({ + type: z.literal("tool_execution_completed"), + toolCallId: z.string().min(1), + toolName: z.string().min(1), + outcome: z.enum(["completed", "error"]), + }) + .strict(); + const conversationMessageRoleSchema = z.union([ z.literal("user"), z.literal("assistant"), @@ -307,6 +316,7 @@ const appendableConversationEventDataSchema = z.union([ authorizationRequestedEventDataSchema, authorizationCompletedEventDataSchema, toolExecutionStartedEventDataSchema, + toolExecutionCompletedEventDataSchema, messageHandledEventDataSchema, messagesSummarizedEventDataSchema, turnStartedEventDataSchema, @@ -339,6 +349,7 @@ const knownConversationEventTypeSchema = z.enum([ "authorization_requested", "authorization_completed", "tool_execution_started", + "tool_execution_completed", "message_handled", "messages_summarized", "turn_started", diff --git a/packages/junior/src/chat/conversations/projection.ts b/packages/junior/src/chat/conversations/projection.ts index 8847e0382..4908a035b 100644 --- a/packages/junior/src/chat/conversations/projection.ts +++ b/packages/junior/src/chat/conversations/projection.ts @@ -458,3 +458,24 @@ export async function recordToolExecutionStarted(args: { }, ]); } + +/** Record a host-observed parent tool completion without persisting its output. */ +export async function recordToolExecutionCompleted(args: { + conversationId: string; + createdAtMs?: number; + isError: boolean; + toolCallId: string; + toolName: string; +}): Promise { + await getConversationEventStore().append(args.conversationId, [ + { + data: { + type: "tool_execution_completed", + toolCallId: args.toolCallId, + toolName: args.toolName, + outcome: args.isError ? "error" : "completed", + }, + createdAtMs: args.createdAtMs ?? Date.now(), + }, + ]); +} diff --git a/packages/junior/tests/integration/api/conversations/event-list.test.ts b/packages/junior/tests/integration/api/conversations/event-list.test.ts index e459cb7d1..50ebac55a 100644 --- a/packages/junior/tests/integration/api/conversations/event-list.test.ts +++ b/packages/junior/tests/integration/api/conversations/event-list.test.ts @@ -216,13 +216,10 @@ describe("conversation event list API", () => { }, { data: { - type: "agent_step", - message: { - role: "toolResult", - toolCallId: "search-before-page", - content: [{ type: "text", text: "two matches" }], - isError: false, - } as PiMessage, + type: "tool_execution_completed", + toolCallId: "search-before-page", + toolName: "search", + outcome: "completed", }, createdAtMs: 3, }, diff --git a/packages/junior/tests/integration/runtime/agent-run-provider-retry.test.ts b/packages/junior/tests/integration/runtime/agent-run-provider-retry.test.ts index f5513ead9..5c1813f54 100644 --- a/packages/junior/tests/integration/runtime/agent-run-provider-retry.test.ts +++ b/packages/junior/tests/integration/runtime/agent-run-provider-retry.test.ts @@ -80,6 +80,15 @@ vi.mock("@/chat/conversations/projection", async (importOriginal) => { } return actual.recordToolExecutionStarted(...args); }, + recordToolExecutionCompleted: async ( + ...args: Parameters + ) => { + sessionLogState.toolExecutionAppendCalls += 1; + if (sessionLogState.failToolExecutionAppend) { + throw new Error("store blip during host-only append"); + } + return actual.recordToolExecutionCompleted(...args); + }, }; }); @@ -223,6 +232,22 @@ vi.mock("@earendil-works/pi-agent-core", () => { this.recordRunFailure(error); return {}; } + try { + await Promise.all( + this.subscribers.map((subscriber) => + subscriber({ + type: "tool_execution_end", + toolCallId: "call_1", + toolName: "bash", + result: { content: [{ type: "text", text: "ok" }] }, + isError: false, + }), + ), + ); + } catch (error) { + this.recordRunFailure(error); + return {}; + } this.state.messages.push({ role: "toolResult", toolName: "bash", @@ -1094,7 +1119,7 @@ describe("executeAgentRun provider retry", () => { }), ); - expect(sessionLogState.toolExecutionAppendCalls).toBe(1); + expect(sessionLogState.toolExecutionAppendCalls).toBe(2); expect(reply.diagnostics.outcome).toBe("success"); expect(reply.text).toBe("Tool done."); });