From 36a686ee8dd73afe8242011db387040890e24093 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Tue, 6 Oct 2026 19:28:54 +0200 Subject: [PATCH] feat: record response, tool execution, and durable task times pi-ai's AssistantMessageEventStream sets AssistantMessage.durationMs with a monotonic clock for every API implementation. The agent loop and the durable tool task set ToolResultMessage.durationMs around execute(). Durable task records carry startedAt/endedAt stamped by the Session. The coding agent's bash renderer shows Took from the recorded duration, also after a reload. closes #10549 --- packages/agent/CHANGELOG.md | 4 + packages/agent/src/agent-loop.ts | 11 ++- packages/agent/src/types.ts | 12 ++- packages/agent/test/agent-loop.test.ts | 56 ++++++++++++++ packages/ai/CHANGELOG.md | 6 ++ packages/ai/src/api/lazy.ts | 6 +- packages/ai/src/types.ts | 14 +++- packages/ai/src/utils/event-stream.ts | 32 +++++++- packages/ai/test/event-stream.test.ts | 77 ++++++++++++++++++- packages/coding-agent/CHANGELOG.md | 2 + packages/coding-agent/docs/json.md | 2 +- .../coding-agent/src/core/agent-session.ts | 1 + .../src/core/export-html/tool-renderer.ts | 1 + .../coding-agent/src/core/extensions/types.ts | 7 ++ .../src/core/nested-tool-calls.ts | 2 + .../src/core/tools/renderers/bash.ts | 8 +- .../interactive/components/tool-execution.ts | 4 + .../src/modes/interactive/interactive-mode.ts | 2 +- .../test/codemode-renderer.test.ts | 1 + .../test/tool-execution-component.test.ts | 27 +++++++ packages/durable/CHANGELOG.md | 3 + packages/durable/docs/spec.md | 21 ++++- packages/durable/src/harness/harness.ts | 2 +- packages/durable/src/harness/tool.ts | 16 +++- packages/durable/src/session/session.ts | 9 ++- packages/durable/src/session/transaction.ts | 21 ++++- packages/durable/src/types.ts | 8 ++ .../test/harness-tasks-recovery.test.ts | 29 +++++++ packages/durable/test/harness-tasks.test.ts | 51 ++++++++++++ packages/durable/test/harness-tools.test.ts | 38 +++++++++ 30 files changed, 451 insertions(+), 22 deletions(-) diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 4f7aeb1ec..808ab5f50 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Added `durationMs` to tool result messages, `AgentToolCallOutcome`, and `tool_execution_end` events: how long `execute()` took, measured with a monotonic clock and excluding hooks. Calls that did not run have none ([#10549](https://github.com/earendil-works/pi/issues/10549)) + ## [1.0.4] - 2026-10-05 ## [1.0.3] - 2026-10-05 diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index b772985ca..f406aba56 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -675,6 +675,7 @@ type ImmediateToolCallOutcome = { type ExecutedToolCallOutcome = { result: AgentToolResult; isError: boolean; + durationMs: number; }; type FinalizedToolCallOutcome = AgentToolCallOutcome; @@ -824,6 +825,8 @@ async function executePreparedToolCall( ): Promise { const updateEvents: Promise[] = []; let acceptingUpdates = true; + const startedAt = performance.now(); + const elapsed = () => Math.round(performance.now() - startedAt); try { const result = await prepared.tool.execute( @@ -835,15 +838,18 @@ async function executePreparedToolCall( updateEvents.push(Promise.resolve(onUpdate(partialResult))); }, ); + const durationMs = elapsed(); acceptingUpdates = false; await Promise.all(updateEvents); - return { result, isError: result.isError === true }; + return { result, isError: result.isError === true, durationMs }; } catch (error) { + const durationMs = elapsed(); acceptingUpdates = false; await Promise.all(updateEvents); return { result: createErrorToolResult(error instanceof Error ? error.message : String(error)), isError: true, + durationMs, }; } finally { acceptingUpdates = false; @@ -899,6 +905,7 @@ async function finalizeExecutedToolCall( toolCall: prepared.toolCall, result, isError, + durationMs: executed.durationMs, }; } @@ -916,6 +923,7 @@ async function emitToolExecutionEnd(finalized: FinalizedToolCallOutcome, emit: A toolName: finalized.toolCall.name, result: finalized.result, isError: finalized.isError, + ...(finalized.durationMs === undefined ? {} : { durationMs: finalized.durationMs }), }); } @@ -930,6 +938,7 @@ function createToolResultMessage(finalized: FinalizedToolCallOutcome): ToolResul details: finalized.result.details, usage: finalized.result.usage, isError: finalized.isError, + ...(finalized.durationMs === undefined ? {} : { durationMs: finalized.durationMs }), timestamp: Date.now(), }; } diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 6e17c3c8c..dbfba74a0 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -450,6 +450,8 @@ export interface AgentToolCallOutcome { toolCall: AgentToolCall; result: AgentToolResult; isError: boolean; + /** Milliseconds `execute()` took, measured with a monotonic clock; absent when the tool did not run. */ + durationMs?: number; } /** @@ -526,4 +528,12 @@ export type AgentEvent = // Tool execution lifecycle | { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any } | { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any } - | { type: "tool_execution_end"; toolCallId: string; toolName: string; result: any; isError: boolean }; + | { + type: "tool_execution_end"; + toolCallId: string; + toolName: string; + result: any; + isError: boolean; + /** Milliseconds `execute()` took, measured with a monotonic clock; absent when the tool did not run. */ + durationMs?: number; + }; diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index c60c04a34..2ac2ada6a 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -4,6 +4,7 @@ import { EventStream, type Message, type Model, + type ToolResultMessage, type UserMessage, } from "@earendil-works/pi-ai"; import { Type } from "typebox"; @@ -412,6 +413,61 @@ describe("agentLoop with AgentMessage", () => { expect(toolResult?.role === "toolResult" ? toolResult.usage : undefined).toEqual(patchedToolUsage); }); + // #10549 + it("records how long execute() took on the tool result, excluding hooks", async () => { + const toolSchema = Type.Object({ value: Type.String() }); + const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + const tool: AgentTool = { + name: "echo", + label: "Echo", + description: "Echo tool", + parameters: toolSchema, + async execute(_toolCallId, params) { + await sleep(30); + return { content: [{ type: "text", text: params.value }], details: { value: params.value } }; + }, + }; + const config: AgentLoopConfig = { + model: createModel(), + convertToLlm: identityConverter, + beforeToolCall: async ({ toolCall }) => { + await sleep(100); + return toolCall.id === "blocked" ? { block: true, reason: "no" } : undefined; + }, + }; + let callIndex = 0; + const streamFn = () => { + const stream = new MockAssistantStream(); + queueMicrotask(() => { + const message = + callIndex === 0 + ? createAssistantMessage( + [ + { type: "toolCall", id: "ran", name: "echo", arguments: { value: "a" } }, + { type: "toolCall", id: "blocked", name: "echo", arguments: { value: "b" } }, + ], + "toolUse", + ) + : createAssistantMessage([{ type: "text", text: "done" }]); + stream.push({ type: "done", reason: callIndex === 0 ? "toolUse" : "stop", message }); + callIndex++; + }); + return stream; + }; + const stream = agentLoop([createUserMessage("go")], { messages: [], tools: [tool] }, config, undefined, streamFn); + for await (const _event of stream) { + // drain + } + const results = (await stream.result()).filter( + (message): message is ToolResultMessage => message.role === "toolResult", + ); + const [ran, blocked] = results; + expect(ran?.durationMs).toBeGreaterThanOrEqual(25); + expect(ran?.durationMs).toBeLessThan(100); + expect(blocked?.isError).toBe(true); + expect(blocked).not.toHaveProperty("durationMs"); + }); + it("should not execute tool calls from a length-truncated assistant message", async () => { const toolSchema = Type.Object({ value: Type.String() }); const executed: string[] = []; diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 9e44669bb..c109d5334 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -2,9 +2,15 @@ ## [Unreleased] +### Added + +- Added `durationMs` to `AssistantMessage`: `AssistantMessageEventStream` measures each response with a monotonic clock from the start of the request to its final message, for every API implementation, including direct calls. Deferred results fetched later stay untimed ([#10549](https://github.com/earendil-works/pi/issues/10549)) +- Added an optional `durationMs` to `ToolResultMessage` for the execution time of the tool ([#10549](https://github.com/earendil-works/pi/issues/10549)) + ### Fixed - Fixed `server_busy` and `servers are currently busy` provider errors ending the turn instead of being retried ([#10543](https://github.com/earendil-works/pi/issues/10543)) +- Fixed the error message of a failed lazy API setup, such as a module load or auth failure, using its failure time as `timestamp` instead of the request start - Reduced context-limit request failures by estimating input at 3.5 characters per token instead of 4 when calculating output limits ([#10497](https://github.com/earendil-works/pi/issues/10497)) ## [1.0.4] - 2026-10-05 diff --git a/packages/ai/src/api/lazy.ts b/packages/ai/src/api/lazy.ts index 55a564cdb..2f9cf0a35 100644 --- a/packages/ai/src/api/lazy.ts +++ b/packages/ai/src/api/lazy.ts @@ -1,7 +1,7 @@ import type { Api, AssistantMessage, AssistantMessageEvent, Model, ProviderStreams } from "../types.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; -function createSetupErrorMessage(model: Model, error: unknown): AssistantMessage { +function createSetupErrorMessage(model: Model, error: unknown, timestamp: number): AssistantMessage { return { role: "assistant", content: [], @@ -18,7 +18,7 @@ function createSetupErrorMessage(model: Model, error: unknown): AssistantMe }, stopReason: "error", errorMessage: error instanceof Error ? error.message : String(error), - timestamp: Date.now(), + timestamp, }; } @@ -52,7 +52,7 @@ export function lazyStream( setup() .then((inner) => forwardStream(outer, inner)) .catch((error) => { - const message = createSetupErrorMessage(model, error); + const message = createSetupErrorMessage(model, error, outer.startedAt); outer.push({ type: "error", reason: "error", error: message }); outer.end(message); }); diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 7469f3038..99a8fc4ef 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -568,7 +568,14 @@ export interface AssistantMessage { * Preserved for debugging and does not currently affect agent control flow. */ endTurn?: boolean; - timestamp: number; // Unix timestamp in milliseconds + /** Unix timestamp in milliseconds when the request started. */ + timestamp: number; + /** + * Milliseconds from `timestamp` until the response ended, measured with a monotonic clock. Set by + * `AssistantMessageEventStream` on the final message of a response it saw start; absent for legacy messages and + * for deferred results fetched later. + */ + durationMs?: number; } /** A tool call that another tool made while it ran, for example from a codemode script. */ @@ -605,7 +612,10 @@ export type ToolResultMessage = IsJsonCompatible /** Calls this tool made to other tools. Kept for the session record; not sent to the model. */ nestedCalls?: NestedToolCalls; isError: boolean; - timestamp: number; // Unix timestamp in milliseconds + /** Unix timestamp in milliseconds when the result was created. */ + timestamp: number; + /** Milliseconds the tool's execution took, measured with a monotonic clock. Absent for legacy results. */ + durationMs?: number; } : never; diff --git a/packages/ai/src/utils/event-stream.ts b/packages/ai/src/utils/event-stream.ts index c4f731d59..669deeaf3 100644 --- a/packages/ai/src/utils/event-stream.ts +++ b/packages/ai/src/utils/event-stream.ts @@ -26,7 +26,7 @@ class FifoQueue { export class EventStream implements AsyncIterable { private queue = new FifoQueue(); private waiting = new FifoQueue<(value: IteratorResult) => void>(); - private done = false; + protected done = false; private finalResultPromise: Promise; private resolveFinalResult!: (result: R) => void; private isComplete: (event: T) => boolean; @@ -88,7 +88,16 @@ export class EventStream implements AsyncIterable { } } +/** + * Event stream of one assistant response. It also times the response: the final message (`done` or `error` event, or + * the result passed to `end()`) gets `durationMs`, measured with a monotonic clock from the stream's creation, unless + * the message already has one or its `timestamp` predates the stream. A stream that forwards a response which started + * elsewhere, such as a deferred result fetched later, therefore leaves it untimed. + */ export class AssistantMessageEventStream extends EventStream { + readonly #startedAt = Date.now(); + readonly #startedAtMonotonic = performance.now(); + constructor() { super( (event) => event.type === "done" || event.type === "error", @@ -102,6 +111,27 @@ export class AssistantMessageEventStream extends EventStream { @@ -92,3 +93,77 @@ describe("EventStream", () => { expect(await secondEvent).toEqual({ value: undefined, done: true }); }); }); + +function message(timestamp: number, durationMs?: number): AssistantMessage { + return { + role: "assistant", + content: [], + api: "openai-responses", + provider: "openai", + model: "m", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp, + ...(durationMs === undefined ? {} : { durationMs }), + }; +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe("AssistantMessageEventStream timing", () => { + it("sets durationMs on the final done or error message of a response it saw start", async () => { + const done = new AssistantMessageEventStream(); + const answer = message(Date.now()); + await sleep(20); + done.push({ type: "done", reason: "stop", message: answer }); + expect(answer.durationMs).toBeGreaterThanOrEqual(15); + expect((await done.result()).durationMs).toBe(answer.durationMs); + + const failed = new AssistantMessageEventStream(); + const error = { ...message(Date.now()), stopReason: "error" as const }; + failed.push({ type: "error", reason: "error", error }); + expect(error.durationMs).toBeGreaterThanOrEqual(0); + + const ended = new AssistantMessageEventStream(); + const result = message(Date.now()); + ended.end(result); + expect(result.durationMs).toBeGreaterThanOrEqual(0); + }); + + it("keeps an existing duration, so a forwarding stream keeps the inner measurement", async () => { + const outer = new AssistantMessageEventStream(); + await sleep(20); + const inner = new AssistantMessageEventStream(); + const answer = message(Date.now()); + inner.push({ type: "done", reason: "stop", message: answer }); + const measured = answer.durationMs; + outer.push({ type: "done", reason: "stop", message: answer }); + expect(answer.durationMs).toBe(measured); + expect(measured).toBeLessThan(20); + + const preset = message(Date.now(), 1234); + new AssistantMessageEventStream().push({ type: "done", reason: "stop", message: preset }); + expect(preset.durationMs).toBe(1234); + }); + + it("leaves a message untimed when it started before the stream, such as a fetched deferred result", () => { + const fetched = message(Date.now() - 60_000); + new AssistantMessageEventStream().push({ type: "done", reason: "stop", message: fetched }); + expect(fetched.durationMs).toBeUndefined(); + }); + + it("does not time a message pushed after the stream completed", () => { + const stream = new AssistantMessageEventStream(); + stream.push({ type: "done", reason: "stop", message: message(Date.now()) }); + const late = message(Date.now()); + stream.push({ type: "done", reason: "stop", message: late }); + expect(late.durationMs).toBeUndefined(); + }); +}); diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 95dda73da..791e9d0a5 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -5,6 +5,7 @@ ### Added - Added `+name` and `-name` entries to `--tools`, which change the default tool selection instead of replacing it, for example `pi -t +codemode` +- Added `durationMs` to the tool render context and to `tool_execution_end` extension events: the recorded execution time of a final tool result ([#10549](https://github.com/earendil-works/pi/issues/10549)) - Added `outputPad` to the tool render context ([#10557](https://github.com/earendil-works/pi/pull/10557) by [@rwachtler](https://github.com/rwachtler)) ### Changed @@ -13,6 +14,7 @@ ### Fixed +- Fixed bash and PowerShell results losing `Took` after reloading a session, and the live `Took` including wall-clock steps; both now show the recorded execution time ([#10549](https://github.com/earendil-works/pi/issues/10549)) - Fixed managed installs keeping every old release; `pi update` now keeps only the new release and the one it updated from ([#10392](https://github.com/earendil-works/pi/issues/10392)) - Fixed standalone binaries loading `.env`, `.env.local`, and `.env.development` from the launch directory into Pi's environment ([#10473](https://github.com/earendil-works/pi/issues/10473)) - Fixed `!!` command headers losing their dim color once output arrives ([#10557](https://github.com/earendil-works/pi/pull/10557) by [@rwachtler](https://github.com/rwachtler)) diff --git a/packages/coding-agent/docs/json.md b/packages/coding-agent/docs/json.md index afa1f7b12..d2094a434 100644 --- a/packages/coding-agent/docs/json.md +++ b/packages/coding-agent/docs/json.md @@ -104,7 +104,7 @@ The top-level `usage` is the latest cumulative provider-reported usage for the a |---|---|---| | `tool_execution_start` | `toolCallId`, `toolName`, `args` | Tool execution started. | | `tool_execution_update` | `toolCallId`, `toolName`, `args`, `partialResult` | The tool reported a partial result. | -| `tool_execution_end` | `toolCallId`, `toolName`, `result`, `isError` | Tool execution finished. | +| `tool_execution_end` | `toolCallId`, `toolName`, `result`, `isError`, `durationMs` | Tool execution finished. `durationMs` is how long the tool's `execute()` took, measured with a monotonic clock; absent when the tool did not run. | Use `toolCallId` to correlate the lifecycle. `partialResult` is the latest partial result supplied by the tool. Whether it replaces or extends an earlier update depends on that tool's result contract. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 7388024e0..ebb775b8a 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -1354,6 +1354,7 @@ export class AgentSession { toolName: event.toolName, result: event.result, isError: event.isError, + ...(event.durationMs === undefined ? {} : { durationMs: event.durationMs }), }; await this._extensionRunner.emit(extensionEvent); } diff --git a/packages/coding-agent/src/core/export-html/tool-renderer.ts b/packages/coding-agent/src/core/export-html/tool-renderer.ts index 377de6415..1f8c1fd63 100644 --- a/packages/coding-agent/src/core/export-html/tool-renderer.ts +++ b/packages/coding-agent/src/core/export-html/tool-renderer.ts @@ -92,6 +92,7 @@ export function createToolHtmlRenderer(deps: ToolHtmlRendererDeps): ToolHtmlRend expanded, showImages: false, isError, + durationMs: undefined, outputPad: 1, }; }; diff --git a/packages/coding-agent/src/core/extensions/types.ts b/packages/coding-agent/src/core/extensions/types.ts index 8a4be6c27..278cc3e14 100644 --- a/packages/coding-agent/src/core/extensions/types.ts +++ b/packages/coding-agent/src/core/extensions/types.ts @@ -489,6 +489,11 @@ export interface ToolRenderContext { showImages: boolean; /** Whether the current result is an error. */ isError: boolean; + /** + * Milliseconds the tool's execution took, from the final result; `undefined` while it runs, when it did not run, or + * for results stored before durations were recorded. + */ + durationMs: number | undefined; /** Horizontal padding configured by the outputPad setting. Renderers with `renderShell: "self"` apply it themselves. */ outputPad: number; } @@ -1083,6 +1088,8 @@ export interface ToolExecutionEndEvent { toolName: string; result: any; isError: boolean; + /** Milliseconds `execute()` took, measured with a monotonic clock; absent when the tool did not run. */ + durationMs?: number; /** Set when another tool (for example a codemode script) made this call. */ parentToolCallId?: string; } diff --git a/packages/coding-agent/src/core/nested-tool-calls.ts b/packages/coding-agent/src/core/nested-tool-calls.ts index b014dc0da..ad5b79e63 100644 --- a/packages/coding-agent/src/core/nested-tool-calls.ts +++ b/packages/coding-agent/src/core/nested-tool-calls.ts @@ -124,6 +124,7 @@ export type NestedToolExecutionEvent = toolName: string; result: AgentToolResult; isError: boolean; + durationMs?: number; parentToolCallId: string; }; @@ -242,6 +243,7 @@ export class NestedToolCallRunner { toolName: name, result: outcome.result, isError: outcome.isError, + ...(outcome.durationMs === undefined ? {} : { durationMs: outcome.durationMs }), parentToolCallId: callerId, }); return outcome; diff --git a/packages/coding-agent/src/core/tools/renderers/bash.ts b/packages/coding-agent/src/core/tools/renderers/bash.ts index 1c0f7a5bc..1c47224b1 100644 --- a/packages/coding-agent/src/core/tools/renderers/bash.ts +++ b/packages/coding-agent/src/core/tools/renderers/bash.ts @@ -45,6 +45,7 @@ function rebuildBashResultRenderComponent( showImages: boolean, startedAt: number | undefined, endedAt: number | undefined, + durationMs: number | undefined, ): void { component.clear(); @@ -98,7 +99,11 @@ function rebuildBashResultRenderComponent( component.addChild(new Text(`\n${theme.fg("warning", `[${warnings.join(". ")}]`)}`, 0, 0)); } - if (startedAt !== undefined) { + // A final result's recorded duration wins: it is monotonic and survives reloads. The renderer's own clock is the + // fallback for live progress and for results stored without one. + if (!options.isPartial && durationMs !== undefined) { + component.addChild(new Text(`\n${theme.fg("muted", `Took ${formatDuration(durationMs)}`)}`, 0, 0)); + } else if (startedAt !== undefined) { const label = options.isPartial ? "Elapsed" : "Took"; const endTime = endedAt ?? Date.now(); component.addChild(new Text(`\n${theme.fg("muted", `${label} ${formatDuration(endTime - startedAt)}`)}`, 0, 0)); @@ -138,6 +143,7 @@ export function createShellRenderers(prompt: string): Pick; isError: boolean; details?: any; + durationMs?: number; }; private hideComponent = false; @@ -133,6 +134,7 @@ export class ToolExecutionComponent extends Container { expanded: this.expanded, showImages: this.showImages, isError: this.result?.isError ?? false, + durationMs: this.isPartial ? undefined : this.result?.durationMs, outputPad: this.outputPad, }; } @@ -187,6 +189,8 @@ export class ToolExecutionComponent extends Container { content: Array<{ type: string; text?: string; data?: string; mimeType?: string }>; details?: any; isError: boolean; + /** Execution time of a final result. */ + durationMs?: number; }, isPartial = false, ): void { diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 0928f918f..750af1075 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -3622,7 +3622,7 @@ export class InteractiveMode { if (event.isError) this.maybeShowInstallChangeWarning(); const component = this.pendingTools.get(event.toolCallId); if (component) { - component.updateResult({ ...event.result, isError: event.isError }); + component.updateResult({ ...event.result, isError: event.isError, durationMs: event.durationMs }); this.pendingTools.delete(event.toolCallId); this.ui.requestRender(); } diff --git a/packages/coding-agent/test/codemode-renderer.test.ts b/packages/coding-agent/test/codemode-renderer.test.ts index bd88c0c6f..dfda70145 100644 --- a/packages/coding-agent/test/codemode-renderer.test.ts +++ b/packages/coding-agent/test/codemode-renderer.test.ts @@ -26,6 +26,7 @@ function render( expanded, showImages: false, isError, + durationMs: undefined, outputPad: 1, } satisfies ToolRenderContext; const component = codemodeRenderers.renderResult?.( diff --git a/packages/coding-agent/test/tool-execution-component.test.ts b/packages/coding-agent/test/tool-execution-component.test.ts index 130d33a0f..f334f7d6b 100644 --- a/packages/coding-agent/test/tool-execution-component.test.ts +++ b/packages/coding-agent/test/tool-execution-component.test.ts @@ -262,6 +262,33 @@ describe("ToolExecutionComponent parity", () => { expect(completed).toContain(`Took ${formatted}`); }); + // #10549 + test("bash renderer shows a result's recorded duration, also for a result restored without a live start", () => { + const render = (live: boolean): string => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const component = new ToolExecutionComponent( + "bash", + "tool-bash-recorded", + { command: "sleep 4" }, + {}, + createBashToolDefinition(process.cwd(), { exposeSessionEnvironment: false }), + createFakeTui(), + process.cwd(), + ); + if (live) { + component.markExecutionStarted(); + component.updateResult({ content: [], isError: false }, true); + // The wall clock jumps; the recorded duration does not. + vi.advanceTimersByTime(3_600_000); + } + component.updateResult({ content: [], isError: false, durationMs: 4_200 }, false); + return stripAnsi(component.render(120).join("\n")); + }; + expect(render(true)).toContain("Took 4.2s"); + expect(render(false)).toContain("Took 4.2s"); + }); + test("does not duplicate built-in headers when passed the active built-in definition", () => { const component = new ToolExecutionComponent( "read", diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index e7c315955..c224b46cb 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -10,6 +10,9 @@ - `ToolExecutionApi.models` and `HookApi.models`: the Harness's `models`, so tools and hooks can resolve models and make requests with the same catalog, credentials, and request transforms as generation ([#10395](https://github.com/earendil-works/pi/issues/10395)). - `Conversation.context(context, { at })` returns the model context as of a visible earlier entry, the same view `fork(at)` would start with, without creating a conversation ([#10512](https://github.com/earendil-works/pi/issues/10512)). +- Task records carry `startedAt` and `endedAt`, wall-clock times the Session stamps at the first change to `running` and at the change to `terminal`, with `HarnessOptions.now` or the new `createSession(storage, { now })` option. `startedAt` survives waits and reopen. `inspect()` shows them on each live task's record. Records written by earlier versions have neither ([#10549](https://github.com/earendil-works/pi/issues/10549)). +- `pi.tool-result` messages carry `durationMs`: how long `execute()` took in that attempt, measured with a monotonic clock. Calls that did not execute, and interrupted or aborted calls, have none ([#10549](https://github.com/earendil-works/pi/issues/10549)). +- Assistant messages carry the `durationMs` pi-ai measures for each response ([#10549](https://github.com/earendil-works/pi/issues/10549)). ### Changed diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 8ce6fddf0..01247fec6 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -352,6 +352,7 @@ type HarnessOptions = { ) => ExecutionEnv | undefined | Promise; /** Runs in every commit that creates or forks a conversation, after the built-in creation hook (below). */ readonly conversationCreated?: (tx: Tx, conversation: ConversationRecord) => void | Promise; + /** Wall clock for sleeps, retries, message timestamps, and task lifecycle times. Default `Date.now`. */ readonly now?: () => number; readonly onReport?: (error: unknown) => void; }; @@ -1631,6 +1632,10 @@ type TaskRecord = { readonly owner?: TaskId; readonly background: boolean; readonly abortRequested: boolean; + /** Wall clock at the first change to `running`; kept through waits and recovery. */ + readonly startedAt?: number; + /** Wall clock at the change to `terminal`. */ + readonly endedAt?: number; } & ( | { readonly state: Extract, { status: "pending" | "running" | "waiting" }>; @@ -1887,6 +1892,17 @@ items 1 and 3 and its waiters follow in the final commit. The execution checkpoint and memos disappear from the terminal representation. Terminal records remain queryable for waits, waiters, inspection, and reopen. + +The Session stamps lifecycle times on task records with the clock it was opened +with (`HarnessOptions.now` for a Harness, `Date.now` by default): `startedAt` at +the first change to `running`, `endedAt` at the change to `terminal`. Once set, +each carries over from the replaced record, so `startedAt` survives waits, a +`completing` hold, and reopen, and the span between them includes those. A task +that never ran, such as one orphaned before its first reservation, has only +`endedAt`; records written by earlier versions have neither, and a task live +across the upgrade gets `startedAt` at its next run. `inspect()` shows +them on each live task's record. They are lifecycle times, not execution time: +a tool's execution time is its result's `durationMs` (section 7.3). There are no free-standing task dependencies: ordering comes from a task waiting on other tasks (section 5.5). An abort mark lets pending work reach its abort handler, or its `orphaned` settlement when its definition is unavailable. @@ -3596,9 +3612,12 @@ because the terminal record keeps it; the call is read from the assistant entry. diagnostics, and ends `failed` with `{ entryId }`. - The result commit bounds the content, appends the diagnostics block (section 7.3), appends one `pi.tool-result` entry with `model: [{ role: "toolResult", - toolCallId, toolName, content, details, isError, timestamp }]` and + toolCallId, toolName, content, details, isError, durationMs, timestamp }]` and `data: { diagnostics }`, marks the slot `done`, and completes with `{ entryId, control }`. An `isError` result still completes. + `durationMs` is how long `execute()` took in this attempt, measured with a + monotonic clock, including when it threw; results of calls that did not + execute, and `interrupted` or `aborted` results, have none. - A throw from `execute()`, or from `HarnessOptions.env` building its environment, becomes a `tool_error` result and the task ends `failed` with `{ entryId }`; once the invocation is signalled it propagates diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 0edb09296..fc598202a 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -173,7 +173,7 @@ class HarnessImpl extends SessionImpl implements #closed = false; constructor(storage: Storage, options: HarnessOptions, context: Context) { - super(storage); + super(storage, options.now); this.#storage = storage; this.#options = options; this.#report = options.onReport ?? (() => {}); diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 095793ce9..df2dac03a 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -250,10 +250,17 @@ async function run( let result: ToolExecutionResult; let ending = COMPLETED; + // Execution time of this attempt; a rerun after recovery measures only itself. + let durationMs: number | undefined; try { // Built for this call, so a rerun after recovery gets the conversation's environment at that time. const env = await runtime.env(context); - result = await tool.execute(args, { ...api, env }, context); + const startedAt = performance.now(); + try { + result = await tool.execute(args, { ...api, env }, context); + } finally { + durationMs = Math.round(performance.now() - startedAt); + } } catch (error) { if (runtime.signal.aborted) { ended = true; @@ -271,7 +278,7 @@ async function run( const pending = await progress.stop(); try { const settled = await finalResult(runtime, call, result, reported, context); - await settle(runtime, call, ending, () => settled, context); + await settle(runtime, call, ending, () => settled, context, durationMs); } catch (error) { for (const waiter of pending) waiter.reject(error); throw error; @@ -376,11 +383,12 @@ async function settle( ending: Ending, build: (slot: Readonly | undefined) => ToolExecutionResult, context: Context, + durationMs?: number, ): Promise { await runtime.commit(async (tx) => { const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId); const result = build(slot); - const entry = await appendToolResult(tx, runtime.conversationId, call, result, runtime.now()); + const entry = await appendToolResult(tx, runtime.conversationId, call, result, runtime.now(), durationMs); if (slot !== undefined) finishSlot(slot, entry.id); const entryId = entry.id; if (ending.status === "aborted") @@ -452,6 +460,7 @@ export async function appendToolResult( call: ToolCall, result: ToolExecutionResult, timestamp: number, + durationMs?: number, ): Promise> { const diagnostics = [...(result.diagnostics ?? [])]; const content: Content = [...(result.content ?? [])]; @@ -464,6 +473,7 @@ export async function appendToolResult( ...(result.details === undefined ? {} : { details: result.details }), ...(result.usage === undefined ? {} : { usage: result.usage }), isError: result.isError ?? false, + ...(durationMs === undefined ? {} : { durationMs }), timestamp, } as ToolResultMessage; if (result.usage !== undefined) await recordUsage(tx, conversationId, "tools", call.name, result.usage); diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index f1d8f1137..571d60858 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -45,9 +45,9 @@ import { } from "./observation.ts"; import { type LoadedDocument, Transaction, type TransactionHost, type TransactionScope } from "./transaction.ts"; -/** Open a Session kernel over one storage backend. */ -export function createSession(storage: Storage): Session { - return new SessionImpl(storage); +/** Open a Session kernel over one storage backend. `now` is the wall clock for task times; default `Date.now`. */ +export function createSession(storage: Storage, options?: { readonly now?: () => number }): Session { + return new SessionImpl(storage, options?.now); } /** @@ -66,10 +66,11 @@ export class SessionImpl implements Session { #closing: Promise | undefined; #poison: { readonly error: unknown } | undefined; - constructor(storage: Storage) { + constructor(storage: Storage, now: () => number = Date.now) { this.#storage = storage; this.#host = { storage, + now, cached: (id) => this.#documents.get(id), load: (definition, addressId, address, context) => this.#loadDocument(definition, addressId, address, context), install: (document) => { diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index f4186a279..4c0bfd3e6 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -98,6 +98,8 @@ export type LoadedDocument = { /** Session services used by a transaction while it holds the mutation line. */ export interface TransactionHost { readonly storage: Storage; + /** Wall clock for task lifecycle times. */ + now(): number; /** Return the cached current incarnation without loading. */ cached(addressId: string): LoadedDocument | undefined; /** Return the cached current incarnation, cold-loading and migrating it when necessary. */ @@ -467,7 +469,7 @@ export class Transaction implements Tx { } task.write = { kind: task.write?.kind === "create" ? "create" : "replace", - record: copyJson(value, TABLE_JSON_COPY_OPTIONS) as unknown as AnyTaskRecord, + record: copyJson(this.#stampTimes(value, candidate), TABLE_JSON_COPY_OPTIONS) as unknown as AnyTaskRecord, }; } @@ -943,6 +945,23 @@ export class Transaction implements Tx { } } + /** + * Lifecycle times: `startedAt` on the first change to `running`, `endedAt` on the change to `terminal`. Once set, + * they carry over from the replaced record; records written before they existed lack them. + */ + #stampTimes(value: AnyTaskRecord, candidate: AnyTaskRecord | undefined): AnyTaskRecord { + const status = value.state.status; + const now = (stamps: boolean) => (stamps ? this.#host.now() : undefined); + const startedAt = candidate?.startedAt ?? value.startedAt ?? now(status === "running"); + const endedAt = candidate?.endedAt ?? value.endedAt ?? now(status === "terminal"); + if (startedAt === value.startedAt && endedAt === value.endedAt) return value; + return { + ...value, + ...(startedAt === undefined ? {} : { startedAt }), + ...(endedAt === undefined ? {} : { endedAt }), + }; + } + #taskEntry(id: TaskId): TransactionTask { let task = this.#tasksById.get(id); if (task === undefined) { diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index fa6f88149..7b6447fa6 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -535,6 +535,14 @@ type TaskRecordBase = { readonly background: boolean; /** Durable abort mark checked before run-mode progress is committed. */ readonly abortRequested: boolean; + /** + * Wall-clock milliseconds of the first change to `running`, stamped by the Session. Kept through waits and recovery, + * so the span to `endedAt` includes them. Absent before the task first runs, and on records written by earlier + * versions. + */ + readonly startedAt?: number; + /** Wall-clock milliseconds of the change to `terminal`, stamped by the Session; absent while live. */ + readonly endedAt?: number; }; /** Complete replacement record for one durable task state machine. */ diff --git a/packages/durable/test/harness-tasks-recovery.test.ts b/packages/durable/test/harness-tasks-recovery.test.ts index 986103dec..1c2458773 100644 --- a/packages/durable/test/harness-tasks-recovery.test.ts +++ b/packages/durable/test/harness-tasks-recovery.test.ts @@ -144,6 +144,35 @@ describe("task recovery", () => { await third.harness.close(context); }); + // #10549 + it("keeps a task's startedAt across close and reopen and stamps endedAt when it settles", async () => { + const path = await sqlitePath(); + const service = new TransferService(); + const Transfer = transferTask(service, { first: true }); + let clock = 2_000; + const now = () => clock; + + const first = await openTasks(await openNodeSqliteStorage(path), [Transfer], { now }); + const root = await first.harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Transfer, { amount: 7 }, { ownership: { kind: "conversation" } }), + context, + ); + first.harness.resume(); + await eventually(() => service.calls === 1); + clock = 3_000; + await first.harness.close(context); + + clock = 9_000; + const second = await openTasks(await openNodeSqliteStorage(path), [Transfer], { now }); + const reopened = await second.harness.getTask(id, context); + expect([reopened?.state.status, reopened?.startedAt, reopened?.endedAt]).toEqual(["pending", 2_000, undefined]); + second.harness.resume(); + const receipt = await second.harness.waitForTask(id, context); + expect([receipt.startedAt, receipt.endedAt]).toEqual([2_000, 9_000]); + await second.harness.close(context); + }); + it("resumes abort work after close at every direct-task abort stage", async () => { const path = await sqlitePath(); const log: string[] = []; diff --git a/packages/durable/test/harness-tasks.test.ts b/packages/durable/test/harness-tasks.test.ts index 9bebc8fee..af4ac1d17 100644 --- a/packages/durable/test/harness-tasks.test.ts +++ b/packages/durable/test/harness-tasks.test.ts @@ -307,6 +307,57 @@ describe("task phases", () => { await harness.close(context); }); + // #10549 + it("stamps startedAt at the first run and endedAt when terminal, keeping startedAt through a wait", async () => { + let clock = 1_000; + const gate = deferred(); + let on: TaskId[] = []; + const Waiter = defineTask({ + name: "test.timed-waiter", + version: 1, + initial: () => ({ phase: "wait" }), + phases: { + wait: (_task, runtime, ctx) => { + const checkpoint = { phase: "resume" } as const; + return runtime.commit(() => ({ status: "waiting", checkpoint, on, policy: "allSettled" }), ctx); + }, + resume: (_task, runtime, ctx) => runtime.commit(() => completed(null), ctx), + }, + abort: async (_task, runtime, ctx) => runtime.commit(() => abortedWith("test"), ctx), + }); + const Held = gated("test.timed-held", gate.promise); + const { harness, root } = await openRoot([Waiter, Held], { now: () => clock }); + const held = await start(root, Held); + on = [held]; + const waiter = await root.commit( + (tx) => tx.createTask(Waiter, null, { ownership: { kind: "conversation" } }), + context, + ); + const times = async () => + (await harness.inspect(context)).tasks.map(({ record }) => [record.id, record.startedAt, record.endedAt]); + expect(await times()).toEqual([ + [held, undefined, undefined], + [waiter, undefined, undefined], + ]); + + clock = 2_000; + harness.resume(); + await eventually(async () => (await harness.getTask(waiter, context))?.state.status === "waiting"); + expect(await times()).toEqual([ + [held, 2_000, undefined], + [waiter, 2_000, undefined], + ]); + + // The waiter runs again at 5_000; its start stays the first run. + clock = 5_000; + gate.resolve(); + const settledWaiter = await harness.waitForTask(waiter, context); + const settledHeld = await harness.getTask(held, context); + expect([settledHeld?.startedAt, settledHeld?.endedAt]).toEqual([2_000, 5_000]); + expect([settledWaiter.startedAt, settledWaiter.endedAt]).toEqual([2_000, 5_000]); + await harness.close(context); + }); + it("keeps one registry snapshot per phase and refreshes it at the phase boundary", async () => { const seen: { phase: string; tools: string[]; same: boolean }[] = []; const phaseGate = deferred(); diff --git a/packages/durable/test/harness-tools.test.ts b/packages/durable/test/harness-tools.test.ts index 88cbdbd37..83ff0e015 100644 --- a/packages/durable/test/harness-tools.test.ts +++ b/packages/durable/test/harness-tools.test.ts @@ -428,6 +428,44 @@ describe("tool results", () => { await harness.close(context); }); + // #10549 + it("records how long execute() took, excluding hooks, and nothing for calls that did not run", async () => { + const setup = chatSetup(); + const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + addTool( + setup.registry, + tool("slow", async () => { + await sleep(30); + return { content: [] }; + }), + ); + addTool( + setup.registry, + tool("thrower", async () => { + await sleep(30); + throw new Error("boom"); + }), + ); + addHooks(setup.registry, ToolTask, { + beforeTool: async (call) => { + await sleep(100); + return call.id === "blocked" ? { block: "no" } : undefined; + }, + }); + const { harness, entries } = await run(setup, [ + calls(["slow", {}, "slow"], ["thrower", {}, "thrower"], ["slow", {}, "blocked"]), + DONE, + ]); + const byId = new Map(results(entries).map((result) => [result.toolCallId, result])); + for (const id of ["slow", "thrower"]) { + expect(byId.get(id)?.durationMs).toBeGreaterThanOrEqual(25); + expect(byId.get(id)?.durationMs).toBeLessThan(100); + } + expect(byId.get("thrower")?.isError).toBe(true); + expect(byId.get("blocked")).not.toHaveProperty("durationMs"); + await harness.close(context); + }); + it("validates arguments before and after beforeTool and applies blocks and replacements", async () => { const setup = chatSetup(); const seen: { text?: string }[] = [];