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
This commit is contained in:
Mario Zechner 2026-10-06 19:28:54 +02:00
parent 636703a0a4
commit 36a686ee8d
30 changed files with 451 additions and 22 deletions

View file

@ -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

View file

@ -675,6 +675,7 @@ type ImmediateToolCallOutcome = {
type ExecutedToolCallOutcome = {
result: AgentToolResult<any>;
isError: boolean;
durationMs: number;
};
type FinalizedToolCallOutcome = AgentToolCallOutcome;
@ -824,6 +825,8 @@ async function executePreparedToolCall(
): Promise<ExecutedToolCallOutcome> {
const updateEvents: Promise<void>[] = [];
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(),
};
}

View file

@ -450,6 +450,8 @@ export interface AgentToolCallOutcome {
toolCall: AgentToolCall;
result: AgentToolResult<any>;
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;
};

View file

@ -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<typeof toolSchema, { value: string }> = {
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[] = [];

View file

@ -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

View file

@ -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<Api>, error: unknown): AssistantMessage {
function createSetupErrorMessage(model: Model<Api>, error: unknown, timestamp: number): AssistantMessage {
return {
role: "assistant",
content: [],
@ -18,7 +18,7 @@ function createSetupErrorMessage(model: Model<Api>, 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);
});

View file

@ -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<TDetails = JsonValue> = IsJsonCompatible<TDetails>
/** 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;

View file

@ -26,7 +26,7 @@ class FifoQueue<T> {
export class EventStream<T, R = T> implements AsyncIterable<T> {
private queue = new FifoQueue<T>();
private waiting = new FifoQueue<(value: IteratorResult<T>) => void>();
private done = false;
protected done = false;
private finalResultPromise: Promise<R>;
private resolveFinalResult!: (result: R) => void;
private isComplete: (event: T) => boolean;
@ -88,7 +88,16 @@ export class EventStream<T, R = T> implements AsyncIterable<T> {
}
}
/**
* 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<AssistantMessageEvent, AssistantMessage> {
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<AssistantMessageEve
},
);
}
/** Wall-clock start of this stream, for messages created before the response could start. */
get startedAt(): number {
return this.#startedAt;
}
override push(event: AssistantMessageEvent): void {
if (event.type === "done") this.#time(event.message);
else if (event.type === "error") this.#time(event.error);
super.push(event);
}
override end(result?: AssistantMessage): void {
if (result !== undefined) this.#time(result);
super.end(result);
}
#time(message: AssistantMessage): void {
if (this.done || message.durationMs !== undefined || message.timestamp < this.#startedAt) return;
message.durationMs = Math.max(0, Math.round(performance.now() - this.#startedAtMonotonic));
}
}
/** Factory function for AssistantMessageEventStream (for use in extensions) */

View file

@ -1,5 +1,6 @@
import { describe, expect, it } from "vitest";
import { EventStream } from "../src/utils/event-stream.ts";
import type { AssistantMessage } from "../src/types.ts";
import { AssistantMessageEventStream, EventStream } from "../src/utils/event-stream.ts";
// Regression tests for https://github.com/earendil-works/pi/issues/9055
describe("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();
});
});

View file

@ -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))

View file

@ -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.

View file

@ -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);
}

View file

@ -92,6 +92,7 @@ export function createToolHtmlRenderer(deps: ToolHtmlRendererDeps): ToolHtmlRend
expanded,
showImages: false,
isError,
durationMs: undefined,
outputPad: 1,
};
};

View file

@ -489,6 +489,11 @@ export interface ToolRenderContext<TState = any, TArgs = any> {
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;
}

View file

@ -124,6 +124,7 @@ export type NestedToolExecutionEvent =
toolName: string;
result: AgentToolResult<unknown>;
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;

View file

@ -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<ToolDefinition<any, a
context.showImages,
state.startedAt,
state.endedAt,
context.durationMs,
);
component.invalidate();
return component;

View file

@ -58,6 +58,7 @@ export class ToolExecutionComponent extends Container {
content: Array<{ type: string; text?: string; data?: string; mimeType?: string }>;
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 {

View file

@ -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();
}

View file

@ -26,6 +26,7 @@ function render(
expanded,
showImages: false,
isError,
durationMs: undefined,
outputPad: 1,
} satisfies ToolRenderContext;
const component = codemodeRenderers.renderResult?.(

View file

@ -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",

View file

@ -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

View file

@ -352,6 +352,7 @@ type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
) => ExecutionEnv | undefined | Promise<ExecutionEnv | undefined>;
/** 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<void>;
/** 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<I, S, R> = {
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<TaskState<S, R>, { 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

View file

@ -173,7 +173,7 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
#closed = false;
constructor(storage: Storage, options: HarnessOptions<Tool>, context: Context) {
super(storage);
super(storage, options.now);
this.#storage = storage;
this.#options = options;
this.#report = options.onReport ?? (() => {});

View file

@ -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<ToolSlot> | undefined) => ToolExecutionResult,
context: Context,
durationMs?: number,
): Promise<void> {
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<TypedEntry<{ diagnostics: ToolDiagnostic[] }>> {
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);

View file

@ -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<void> | 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) => {

View file

@ -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) {

View file

@ -535,6 +535,14 @@ type TaskRecordBase<I, R> = {
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. */

View file

@ -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[] = [];

View file

@ -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<null, { phase: "wait" } | { phase: "resume" }, null>({
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();

View file

@ -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 }[] = [];