diff --git a/packages/agent/docs/pico-v5-handoff.md b/packages/agent/docs/pico-v5-handoff.md index 1d4861c61..a537e664a 100644 --- a/packages/agent/docs/pico-v5-handoff.md +++ b/packages/agent/docs/pico-v5-handoff.md @@ -15,16 +15,16 @@ facades, membranes, document routing, view projection, events, or clone chains. ## 1. Records, cursors, and memory tables -Implement IDs, sequences, Session metadata with stable root ID, -`ConversationRecord`, `EntryRecord`, inputs, live/terminal `TaskRecord` values, -document metadata, storage writes, cursors, and detached `MemoryStorage` tables. +Implement IDs, sequences, reserved root conversation ID `1`, +`ConversationRecord`, `EntryRecord`, tagged inputs, live/terminal `TaskRecord` +values, document metadata, storage writes, backend-opaque JSON cursors, and +detached `MemoryStorage` tables. Reserve `Conversation` for the public conversation object, `Entry` for the typed entry definition, and `Task` for the typed executable definition returned by `defineTask()`. -Test atomic root-metadata/conversation creation, missing or dangling root -metadata rejection, mixed atomic commits, rollback, detached reads/writes, -cursor boundaries, +Test reserved root identity and immutable creation, mixed atomic commits, +rollback, detached reads/writes, cursor boundaries, fork-aware entry scans through deep ancestor caps, head lookup, entry-to-commit lookup, and full task replacement. diff --git a/packages/agent/docs/pico-v5.md b/packages/agent/docs/pico-v5.md index 91e630dc7..c52881b0b 100644 --- a/packages/agent/docs/pico-v5.md +++ b/packages/agent/docs/pico-v5.md @@ -7,7 +7,7 @@ Pico5 uses existing package types as follows: import type { Context, JsonValue } from "@earendil-works/chord"; import { applyImmutable, type Op } from "@earendil-works/chord/delta"; import type { - Message as ModelMessage, + Message, Models, ModelThinkingLevel, TextContent, @@ -66,6 +66,7 @@ these contracts. ```ts type Id = number; type Seq = number; +const ROOT_CONVERSATION_ID: Id = 1; type ConversationRecord = { readonly id: Id; @@ -78,8 +79,6 @@ type ConversationRecord = { readonly taskId: Id; }; }; - -type RequestMessage = ModelMessage; ``` The referenced pi-ai member is: @@ -105,14 +104,14 @@ order yields the effective prompt and tool set. type ContextEdit = { readonly target: Id; readonly action: "omit" | "replace"; - readonly messages?: readonly RequestMessage[]; + readonly messages?: readonly Message[]; }; type EntryRecord = { readonly id: Id; readonly conversationId: Id; readonly kind: string; - readonly model?: readonly RequestMessage[]; + readonly model?: readonly Message[]; readonly data?: JsonValue; readonly head?: Id; readonly edits?: readonly ContextEdit[]; @@ -123,16 +122,27 @@ type EntryDraft = Omit(kind: string): Entry; type ContextView = { readonly head: EntryRecord | undefined; readonly entries: readonly EntryRecord[]; - readonly messages: readonly RequestMessage[]; + readonly messages: readonly Message[]; }; type SettledInput = Input & { @@ -292,10 +302,10 @@ interface Conversation { context(context: Context): Promise; entries( query: Omit, - cursor: EntryCursor | undefined, + cursor: Cursor | undefined, limit: number, context: Context, - ): Promise>; + ): Promise>; fork( at: Id, spec: Omit, @@ -374,12 +384,11 @@ settles unknown or unmigratable live kinds as `orphaned` before resolving. IDs and names must be unique. No handler dispatches during open. Dynamic registration is available after open and does not resurrect a task already settled by that pass. -Storage contains one stable root conversation ID in Session metadata. Empty -storage creates the conversation and metadata together from `options.root`; -reopen reads that ID rather than inferring it from a conversation scan. -`options.root` never overwrites existing state. `root()` returns the stored -handle. A conversation with no configured model produces a durable `no_model` -generation failure. +The root conversation always has reserved ID `ROOT_CONVERSATION_ID` (`1`). Empty +storage creates that conversation from `options.root`; reopen looks it up by the +reserved ID. `options.root` never overwrites existing state. `root()` returns +that handle. A conversation with no configured model produces a durable +`no_model` generation failure. `resume()` is idempotent while running and only enables scheduling. It does not repeat open-time reconciliation. `suspend()` is terminal for that Harness @@ -1140,7 +1149,7 @@ conversation document containing tagged items: ```ts type InboxItem = - | { readonly id: Id; readonly mode: "steer" | "followUp"; readonly input: ModelMessage } + | { readonly id: Id; readonly mode: "steer" | "followUp"; readonly input: Message } | { readonly id: Id; readonly mode: "write"; readonly entry: EntryDraft }; ``` @@ -1769,11 +1778,7 @@ type Page = { readonly next?: C; }; -type Cursor = string & { readonly __kind?: K }; -type ConversationCursor = Cursor<"conversation">; -type EntryCursor = Cursor<"entry">; -type TaskCursor = Cursor<"task">; -type DocumentCursor = Cursor<"document">; +type Cursor = Readonly>; type EntryQuery = { readonly conversationId: Id; @@ -1810,12 +1815,7 @@ type StoredDocument = { ]; }; -type SessionMetadata = { - readonly rootConversationId: Id; -}; - type StorageWrite = - | { readonly type: "session.metadata"; readonly value: SessionMetadata } | { readonly type: "conversation"; readonly value: ConversationRecord } | { readonly type: "entry"; readonly value: EntryRecord } | { readonly type: "task"; readonly value: TaskRecord } @@ -1836,34 +1836,34 @@ type StorageWrite = | { readonly type: "document.retire"; readonly id: Id }; interface Storage { + /** The owning Session serializes calls on its mutation line. */ commit(writes: readonly StorageWrite[], context: Context): Promise; mintId(): Id; - metadata(context: Context): Promise; conversation(id: Id, context: Context): Promise; - scanConversations(cursor: ConversationCursor | undefined, limit: number, context: Context): Promise>; + scanConversations(cursor: Cursor | undefined, limit: number, context: Context): Promise>; entries(ids: readonly Id[], context: Context): Promise>; - scanEntries(query: EntryQuery, cursor: EntryCursor | undefined, limit: number, context: Context): Promise>; + scanEntries(query: EntryQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise>; entryCommit(id: Id, context: Context): Promise; task(id: Id, context: Context): Promise | undefined>; - scanTasks(query: TaskQuery, cursor: TaskCursor | undefined, limit: number, context: Context): Promise, TaskCursor>>; + scanTasks(query: TaskQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise, Cursor>>; input(id: Id, context: Context): Promise; inputByRequest(conversationId: Id, requestId: string, context: Context): Promise; document(id: Id, at: Seq | "current", context: Context): Promise; - scanDocuments(query: DocumentQuery, cursor: DocumentCursor | undefined, limit: number, context: Context): Promise>; + scanDocuments(query: DocumentQuery, cursor: Cursor | undefined, limit: number, context: Context): Promise>; close(context: Context): Promise; } ``` -`metadata()` is absent only for empty storage. The first root creation commits -its conversation and metadata together. Existing records with missing metadata, -or metadata naming a missing conversation, make open fail rather than guessing a -root. +Cursors are backend-owned JSON objects. Callers only round-trip them to the same +scan on the same storage; cross-storage or cross-query use is unsupported. The +Session owns the mutation line, so storage implementations do not add a second +caller-facing commit mutex. Each backend still makes one admitted batch atomic. `EntryQuery` supports conversation ancestry, a strict `before` entry cursor, kind filtering, and `withHead`. Document queries support definition, scope, diff --git a/packages/agent/src/harness/execution/tools.ts b/packages/agent/src/harness/execution/tools.ts index 1de0da118..f768781b1 100644 --- a/packages/agent/src/harness/execution/tools.ts +++ b/packages/agent/src/harness/execution/tools.ts @@ -181,10 +181,7 @@ export function finalizeToolCall( } /** Reconstruct the canonical tool result represented by a staged transcript message. */ -export function toolResultFromMessage( - message: ToolResultMessage, - terminate: boolean, -): AgentToolResult { +export function toolResultFromMessage(message: ToolResultMessage, terminate: boolean): AgentToolResult { return { content: message.content, details: message.details, @@ -200,7 +197,7 @@ export function createToolResultMessage(call: FinalizedToolCall): ToolResultMess toolCallId: call.toolCall.id, toolName: call.toolCall.name, content: call.result.content ?? [], - ...(call.result.details === undefined ? {} : { details: call.result.details }), + ...(call.result.details === undefined ? {} : { details: call.result.details as JsonValue }), ...(call.result.usage === undefined ? {} : { usage: call.result.usage }), isError: call.isError, timestamp: Date.now(), diff --git a/packages/agent/src/harness/runtime/drive/tool-placement.ts b/packages/agent/src/harness/runtime/drive/tool-placement.ts index b93d68112..5b0e6ee2a 100644 --- a/packages/agent/src/harness/runtime/drive/tool-placement.ts +++ b/packages/agent/src/harness/runtime/drive/tool-placement.ts @@ -50,15 +50,15 @@ export async function readToolBatchSource( type PlacementItem = { call: Extract; - message: ToolResultMessage; + message: ToolResultMessage; }; type PlacementRead = { items: PlacementItem[]; - turnResults?: ToolResultMessage[]; + turnResults?: ToolResultMessage[]; }; -function isToolResultMessage(value: unknown): value is ToolResultMessage { +function isToolResultMessage(value: unknown): value is ToolResultMessage { return typeof value === "object" && value !== null && "role" in value && value.role === "toolResult"; } @@ -107,7 +107,7 @@ async function readPlacement( items.push({ call, message: stored.value.payload }); } - let turnResults: ToolResultMessage[] | undefined; + let turnResults: ToolResultMessage[] | undefined; if (first === current.calls.length) { const placedIds = current.calls .filter((call) => call.status === "completed") diff --git a/packages/agent/src/harness/runtime/drive/tools.ts b/packages/agent/src/harness/runtime/drive/tools.ts index ccf16d3c9..1bff01053 100644 --- a/packages/agent/src/harness/runtime/drive/tools.ts +++ b/packages/agent/src/harness/runtime/drive/tools.ts @@ -36,7 +36,7 @@ import { } from "./tool-placement.ts"; type ToolCallTask = { completion: Promise }; -type ToolOutcome = { toolCall: AgentToolCall; message: ToolResultMessage; terminate: boolean }; +type ToolOutcome = { toolCall: AgentToolCall; message: ToolResultMessage; terminate: boolean }; type PreparedToolInvocation = | { kind: "ready"; cleared: ClearedToolCall } | { kind: "outcome"; outcome: ToolOutcome }; @@ -134,13 +134,13 @@ function syntheticMessage( toolCall: AgentToolCall, content: AgentToolResult["content"], options: { details?: unknown; usage?: AgentToolResult["usage"] } = {}, -): ToolResultMessage { +): ToolResultMessage { return { role: "toolResult", toolCallId: toolCall.id, toolName: toolCall.name, content, - ...(options.details === undefined ? {} : { details: options.details }), + ...(options.details === undefined ? {} : { details: options.details as JsonValue }), ...(options.usage === undefined ? {} : { usage: options.usage }), isError: true, timestamp: Date.now(), diff --git a/packages/agent/src/pico/index.ts b/packages/agent/src/pico/index.ts new file mode 100644 index 000000000..eb6b174d4 --- /dev/null +++ b/packages/agent/src/pico/index.ts @@ -0,0 +1,24 @@ +export { MemoryStorage } from "./memory-storage.ts"; +export type { + ContextEdit, + ConversationRecord, + Cursor, + DocumentIdentity, + DocumentMetadata, + EntryDraft, + EntryQuery, + EntryRecord, + Id, + Input, + LiveTask, + Page, + Seq, + Storage, + StorageWrite, + StoredError, + TaskOutcome, + TaskQuery, + TaskRecord, + TerminalTask, +} from "./types.ts"; +export { ROOT_CONVERSATION_ID } from "./types.ts"; diff --git a/packages/agent/src/pico/memory-storage.ts b/packages/agent/src/pico/memory-storage.ts new file mode 100644 index 000000000..5b531c35c --- /dev/null +++ b/packages/agent/src/pico/memory-storage.ts @@ -0,0 +1,249 @@ +import type { Context, JsonValue } from "@earendil-works/chord"; +import type { + ConversationRecord, + Cursor, + EntryQuery, + EntryRecord, + Id, + Input, + Page, + Seq, + Storage, + StorageWrite, + TaskQuery, + TaskRecord, +} from "./types.ts"; + +type StoredTask = TaskRecord; +type TableName = "conversation" | "entry" | "task" | "input"; + +type State = { + conversations: Map; + entries: Map; + entryCommits: Map; + tasks: Map; + inputs: Map; +}; + +/** Clone trusted JSON containers while sharing immutable primitives. */ +const clone = (value: T): T => { + if (value === null || typeof value !== "object") return value; + if (Array.isArray(value)) return value.map((item) => clone(item)) as T; + const source = value as Record; + const nullPrototype = Object.getPrototypeOf(value) === null; + const result = (nullPrototype ? Object.create(null) : {}) as Record; + for (const key of Object.keys(source)) { + const copied = clone(source[key]); + if (!nullPrototype && key in result) { + Object.defineProperty(result, key, { + value: copied, + writable: true, + enumerable: true, + configurable: true, + }); + } else { + result[key] = copied; + } + } + return result as T; +}; + +const cursorId = (cursor: Readonly> | undefined): Id | undefined => + cursor?.after as Id | undefined; + +const tableContaining = (state: State, id: Id): TableName | undefined => { + if (state.conversations.has(id)) return "conversation"; + if (state.entries.has(id)) return "entry"; + if (state.tasks.has(id)) return "task"; + if (state.inputs.has(id)) return "input"; + return undefined; +}; + +const page = (values: readonly T[], limit: number): Page => { + const items = values.slice(0, limit); + if (values.length <= limit) return { items: clone(items) }; + return { items: clone(items), next: { after: items.at(-1)!.id } }; +}; + +export class MemoryStorage implements Storage { + private readonly state: State = { + conversations: new Map(), + entries: new Map(), + entryCommits: new Map(), + tasks: new Map(), + inputs: new Map(), + }; + private nextId = 2; + private nextSeq = 1; + private closed = false; + + async commit(writes: readonly StorageWrite[], _context: Context): Promise { + this.assertOpen(); + const prepared = writes.map((write) => clone(write)); + this.checkImmutableIds(prepared); + const seq = this.nextSeq; + + for (const write of prepared) { + switch (write.type) { + case "conversation": + this.state.conversations.set(write.value.id, write.value); + break; + case "entry": + this.state.entries.set(write.value.id, write.value); + this.state.entryCommits.set(write.value.id, seq); + break; + case "task": + this.state.tasks.set(write.value.id, write.value); + break; + case "input": + this.state.inputs.set(write.value.id, write.value); + break; + } + this.nextId = Math.max(this.nextId, write.value.id + 1); + } + this.nextSeq++; + return seq; + } + + mintId(): Id { + this.assertOpen(); + if (!Number.isSafeInteger(this.nextId)) throw new Error("ID space is exhausted"); + return this.nextId++; + } + + async conversation(id: Id, _context: Context): Promise { + this.assertOpen(); + const value = this.state.conversations.get(id); + return value === undefined ? undefined : clone(value); + } + + async scanConversations( + cursor: Cursor | undefined, + limit: number, + _context: Context, + ): Promise> { + this.assertOpen(); + const after = cursorId(cursor); + const values = [...this.state.conversations.values()] + .filter((value) => after === undefined || value.id > after) + .sort((left, right) => left.id - right.id); + return page(values, limit); + } + + async entries(ids: readonly Id[], _context: Context): Promise> { + this.assertOpen(); + const result = new Map(); + for (const id of ids) { + const value = this.state.entries.get(id); + if (value !== undefined) result.set(id, clone(value)); + } + return result; + } + + async scanEntries( + query: EntryQuery, + cursor: Cursor | undefined, + limit: number, + _context: Context, + ): Promise> { + this.assertOpen(); + if (!this.state.conversations.has(query.conversationId)) { + throw new Error(`Unknown conversation: ${query.conversationId}`); + } + const beforeCursor = cursorId(cursor); + const visible: EntryRecord[] = []; + let currentId = query.conversationId; + let upperEntryId = Number.POSITIVE_INFINITY; + const visited = new Set(); + while (true) { + if (visited.has(currentId)) throw new Error("Conversation parent cycle"); + visited.add(currentId); + for (const entry of this.state.entries.values()) { + if (entry.conversationId !== currentId || entry.id > upperEntryId) continue; + if (query.before !== undefined && entry.id >= query.before) continue; + if (beforeCursor !== undefined && entry.id >= beforeCursor) continue; + if (query.kind !== undefined && entry.kind !== query.kind) continue; + if (query.withHead === true && entry.head === undefined) continue; + visible.push(entry); + } + const conversation = this.state.conversations.get(currentId)!; + if (conversation.parent === undefined) break; + upperEntryId = Math.min(upperEntryId, conversation.parent.at); + currentId = conversation.parent.conversationId; + } + visible.sort((left, right) => right.id - left.id); + return page(visible, limit); + } + + async entryCommit(id: Id, _context: Context): Promise { + this.assertOpen(); + return this.state.entryCommits.get(id); + } + + async task(id: Id, _context: Context): Promise { + this.assertOpen(); + const value = this.state.tasks.get(id); + return value === undefined ? undefined : clone(value); + } + + async scanTasks( + query: TaskQuery, + cursor: Cursor | undefined, + limit: number, + _context: Context, + ): Promise> { + this.assertOpen(); + const after = cursorId(cursor); + const values = [...this.state.tasks.values()] + .filter((value) => after === undefined || value.id > after) + .filter((value) => query.conversationId === undefined || value.conversationId === query.conversationId) + .filter((value) => query.kind === undefined || value.kind === query.kind) + .filter((value) => query.status === undefined || value.state.status === query.status) + .filter((value) => query.abortRequested === undefined || value.abortRequested === query.abortRequested) + .filter((value) => query.background === undefined || value.background === query.background) + .sort((left, right) => left.id - right.id); + return page(values, limit); + } + + async input(id: Id, _context: Context): Promise { + this.assertOpen(); + const value = this.state.inputs.get(id); + return value === undefined ? undefined : clone(value); + } + + async inputByRequest(conversationId: Id, requestId: string, _context: Context): Promise { + this.assertOpen(); + for (const value of this.state.inputs.values()) { + if (value.conversationId === conversationId && value.requestId === requestId) return clone(value); + } + return undefined; + } + + async close(_context: Context): Promise { + this.closed = true; + } + + private checkImmutableIds(writes: readonly StorageWrite[]): void { + const claimed = new Map(); + for (const write of writes) { + const table = write.type; + const id = write.value.id; + const existing = tableContaining(this.state, id); + const earlier = claimed.get(id); + if (table === "conversation" || table === "entry") { + if (existing !== undefined) throw new Error(`ID ${id} already belongs to ${existing}`); + if (earlier !== undefined) throw new Error(`ID ${id} is written more than once`); + } else { + if (existing !== undefined && existing !== table) { + throw new Error(`ID ${id} already belongs to ${existing}`); + } + if (earlier !== undefined && earlier !== table) throw new Error(`ID ${id} is written as two record types`); + } + claimed.set(id, table); + } + } + + private assertOpen(): void { + if (this.closed) throw new Error("MemoryStorage is closed"); + } +} diff --git a/packages/agent/src/pico/types.ts b/packages/agent/src/pico/types.ts new file mode 100644 index 000000000..6e336f9d8 --- /dev/null +++ b/packages/agent/src/pico/types.ts @@ -0,0 +1,221 @@ +import type { Context, JsonValue } from "@earendil-works/chord"; +import type { Message } from "@earendil-works/pi-ai"; + +/** Session-global durable record identifier. */ +export type Id = number; +/** Storage-assigned atomic commit sequence. */ +export type Seq = number; + +/** The root conversation always uses this reserved ID. */ +export const ROOT_CONVERSATION_ID: Id = 1; + +export type StoredError = { + readonly message: string; + readonly detail?: JsonValue; +}; + +export type ConversationRecord = { + readonly id: Id; + /** Transcript ancestry and the inclusive entry through which history is inherited. */ + readonly parent?: { + readonly conversationId: Id; + readonly at: Id; + }; + /** Durable task-ownership edge used for subtree abort and idle traversal. */ + readonly owner?: { + readonly conversationId: Id; + readonly taskId: Id; + }; +}; + +export type ContextEdit = { + readonly target: Id; + readonly action: "omit" | "replace"; + readonly messages?: readonly Message[]; +}; + +export type EntryRecord = { + readonly id: Id; + readonly conversationId: Id; + readonly kind: string; + /** Messages contributed to model context; absent for display or bookkeeping entries. */ + readonly model?: readonly Message[]; + readonly data?: JsonValue; + /** Entry at which the active context begins. */ + readonly head?: Id; + readonly edits?: readonly ContextEdit[]; + readonly byTaskId?: Id; +}; + +export type EntryDraft = Omit & { + readonly head?: Id | "self"; +}; + +type InputBase = { + readonly id: Id; + readonly conversationId: Id; + readonly requestId?: string; +}; + +/** Durable admission and settlement state for one host input. */ +export type Input = InputBase & + ( + | { + readonly status: "queued"; + readonly entry?: never; + readonly answer?: never; + readonly reason?: never; + readonly detail?: never; + } + | { + readonly status: "placed"; + /** User transcript entry created when this input was placed. */ + readonly entry: Id; + readonly answer?: never; + readonly reason?: never; + readonly detail?: never; + } + | { + readonly status: "done"; + /** User or passive-write transcript entry created when this input was placed. */ + readonly entry: Id; + /** Assistant answer entry; absent for a completed passive write. */ + readonly answer?: Id; + readonly reason?: never; + readonly detail?: never; + } + | { + readonly status: "unanswered"; + /** Present when the input was placed before becoming unanswered. */ + readonly entry?: Id; + readonly answer?: never; + readonly reason: string; + readonly detail?: JsonValue; + } + ); + +export type LiveTask = { + readonly status: "pending" | "running"; + readonly checkpoint: S; +}; + +export type TaskOutcome = + | { readonly status: "completed"; readonly result: R } + /** Expected task/domain failure explicitly committed by its implementation. */ + | { readonly status: "failed"; readonly error: StoredError; readonly result?: R } + | { readonly status: "aborted"; readonly reason?: string; readonly result?: R } + | { readonly status: "orphaned"; readonly reason: string } + /** Harness-detected contract failure, such as an uncaught throw or no durable progress. */ + | { readonly status: "faulted"; readonly error: StoredError }; + +export type TerminalTask = { + readonly status: "terminal"; + readonly outcome: TaskOutcome; +}; + +export type TaskRecord = { + readonly id: Id; + readonly conversationId: Id; + readonly kind: string; + readonly version: number; + readonly input: I; + readonly state: LiveTask | TerminalTask; + readonly after: readonly Id[]; + readonly background: boolean; + readonly abortRequested: boolean; + readonly memos?: Readonly>; +}; + +/** Persisted identity of one durable document incarnation. */ +export type DocumentIdentity = { + readonly id: Id; + readonly definitionId: string; + readonly definitionVersion: number; + readonly instanceId?: string; +} & ( + | { readonly scope: { readonly kind: "session" } } + | ({ readonly scope: { readonly kind: "conversation"; readonly conversationId: Id } } & ( + | { readonly history: "latest"; readonly fork: "current" | "initial" } + | { + readonly history: "rewindable"; + readonly fork: "asOf" | "current" | "initial"; + } + )) + | { readonly scope: { readonly kind: "task"; readonly taskId: Id } } +); + +/** Commit-sequence lifetime of one persisted document incarnation. */ +export type DocumentMetadata = DocumentIdentity & { + readonly createdAt: Seq; + readonly retiredAt?: Seq; +}; + +export type Page = { + readonly items: readonly T[]; + readonly next?: C; +}; + +/** Backend-owned, JSON-serializable continuation state. Callers must treat it as opaque. */ +export type Cursor = Readonly>; + +export type EntryQuery = { + readonly conversationId: Id; + /** Strict semantic cutoff: only entries with IDs less than this value match. */ + readonly before?: Id; + readonly kind?: string; + readonly withHead?: boolean; +}; + +export type TaskQuery = { + readonly conversationId?: Id; + readonly kind?: string; + readonly status?: "pending" | "running" | "terminal"; + readonly abortRequested?: boolean; + readonly background?: boolean; +}; + +/** Package 1 table writes. Document lifecycle writes are added in Package 2. */ +export type StorageWrite = + | { readonly type: "conversation"; readonly value: ConversationRecord } + | { readonly type: "entry"; readonly value: EntryRecord } + | { readonly type: "task"; readonly value: TaskRecord } + | { readonly type: "input"; readonly value: Input }; + +export interface Storage { + /** + * Persist one batch atomically and return its commit sequence. + * The owning Session serializes calls on its mutation line; Storage implementations + * do not provide a second caller-facing mutation mutex. + */ + commit(writes: readonly StorageWrite[], context: Context): Promise; + mintId(): Id; + + conversation(id: Id, context: Context): Promise; + scanConversations( + cursor: Cursor | undefined, + limit: number, + context: Context, + ): Promise>; + + entries(ids: readonly Id[], context: Context): Promise>; + scanEntries( + query: EntryQuery, + cursor: Cursor | undefined, + limit: number, + context: Context, + ): Promise>; + entryCommit(id: Id, context: Context): Promise; + + task(id: Id, context: Context): Promise | undefined>; + scanTasks( + query: TaskQuery, + cursor: Cursor | undefined, + limit: number, + context: Context, + ): Promise, Cursor>>; + + input(id: Id, context: Context): Promise; + inputByRequest(conversationId: Id, requestId: string, context: Context): Promise; + + close(context: Context): Promise; +} diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 0c4092d29..fb718b635 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -4,6 +4,7 @@ import type { AssistantMessageEvent, AssistantMessageEventStream, ImageContent, + JsonValue, Message, Model, SimpleStreamOptions, @@ -379,7 +380,7 @@ export interface AgentState { } /** Final or partial result produced by a tool. */ -export interface AgentToolResult { +export interface AgentToolResult { /** Text or image content returned to the model. */ content: (TextContent | ImageContent)[]; /** Arbitrary structured details for logs or UI rendering. */ diff --git a/packages/agent/test/harness/execution-tools.test.ts b/packages/agent/test/harness/execution-tools.test.ts index 1e951bac2..df9abb706 100644 --- a/packages/agent/test/harness/execution-tools.test.ts +++ b/packages/agent/test/harness/execution-tools.test.ts @@ -1,3 +1,4 @@ +import type { JsonObject } from "@earendil-works/pi-ai"; import { Type } from "typebox"; import { describe, expect, it, vi } from "vitest"; import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts"; @@ -14,7 +15,7 @@ import type { AgentToolCall, AgentToolResult } from "../../src/types.ts"; const parameters = Type.Object({ value: Type.String() }); -function call(arguments_: Record = { value: "input" }): AgentToolCall { +function call(arguments_: JsonObject = { value: "input" }): AgentToolCall { return { type: "toolCall", id: "call-1", name: "echo", arguments: arguments_ }; } @@ -231,6 +232,16 @@ describe("tool execution primitives", () => { expect(message.timestamp).toBeGreaterThanOrEqual(before); }); + it("preserves unusual JSON object keys in tool-result details", () => { + const cleared = clearPrepared(prepareToolCall(call(), [tool()])); + const details = JSON.parse('{"__proto__":{"preserved":true}}') as Record; + const finalized = finalizeToolCall(cleared, { result: { content: [], details }, isError: false }, undefined); + + const message = createToolResultMessage(finalized); + expect(Object.hasOwn(message.details as object, "__proto__")).toBe(true); + expect(message.details).toEqual(details); + }); + it("normalizes missing content from untyped tools", () => { const cleared = clearPrepared(prepareToolCall(call(), [tool()])); const finalized = finalizeToolCall( diff --git a/packages/agent/test/harness/pico3/helpers.ts b/packages/agent/test/harness/pico3/helpers.ts index 9c961b432..c562141e7 100644 --- a/packages/agent/test/harness/pico3/helpers.ts +++ b/packages/agent/test/harness/pico3/helpers.ts @@ -186,7 +186,7 @@ export function fake(o: FakeOptions): Models & { calls: number; requests: Reques type: "toolCall", id: `call_${n}_${index}`, name: tc.name, - arguments: tc.arguments as Record, + arguments: tc.arguments as ToolCall["arguments"], }; partial.content.push(call); yield { type: "toolcall_start", contentIndex: index, partial }; diff --git a/packages/agent/test/pico/memory-storage.test.ts b/packages/agent/test/pico/memory-storage.test.ts new file mode 100644 index 000000000..62be33a8e --- /dev/null +++ b/packages/agent/test/pico/memory-storage.test.ts @@ -0,0 +1,401 @@ +import type { JsonValue } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { describe, expect, it } from "vitest"; +import { MemoryStorage } from "../../src/pico/memory-storage.ts"; +import { + type EntryRecord, + type Id, + type Input, + ROOT_CONVERSATION_ID, + type StorageWrite, + type TaskRecord, +} from "../../src/pico/types.ts"; + +const context = BACKGROUND_CONTEXT; +type StoredTask = TaskRecord; + +async function createRoot(storage: MemoryStorage): Promise { + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + return ROOT_CONVERSATION_ID; +} + +function pendingTask(id: Id, conversationId: Id, phase = "ready"): StoredTask { + return { + id, + conversationId, + kind: "test.task", + version: 1, + input: { value: id }, + state: { status: "pending", checkpoint: { phase } }, + after: [], + background: false, + abortRequested: false, + }; +} + +function entry(id: Id, conversationId: Id, kind = "message", extra: Partial = {}): EntryRecord { + return { id, conversationId, kind, ...extra }; +} + +describe("Pico MemoryStorage", () => { + it("reserves ID 1 for the immutable root conversation", async () => { + const storage = new MemoryStorage(); + expect(storage.mintId()).toBe(2); + await expect(createRoot(storage)).resolves.toBe(ROOT_CONVERSATION_ID); + expect(await storage.conversation(ROOT_CONVERSATION_ID, context)).toEqual({ id: ROOT_CONVERSATION_ID }); + await expect( + storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context), + ).rejects.toThrow(`ID ${ROOT_CONVERSATION_ID} already belongs to conversation`); + }); + + it("commits mixed table writes atomically and rolls all of them back on failure", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const entryId = storage.mintId(); + const taskId = storage.mintId(); + const inputId = storage.mintId(); + const task = pendingTask(taskId, rootId); + const input: Input = { + id: inputId, + conversationId: rootId, + requestId: "request-1", + status: "placed", + entry: entryId, + }; + const initialSeq = await storage.commit( + [ + { type: "entry", value: entry(entryId, rootId, "user", { data: { text: "hello" } }) }, + { type: "task", value: task }, + { type: "input", value: input }, + ], + context, + ); + + expect(await storage.entries([entryId], context)).toEqual( + new Map([[entryId, entry(entryId, rootId, "user", { data: { text: "hello" } })]]), + ); + expect(await storage.task(taskId, context)).toEqual(task); + expect(await storage.input(inputId, context)).toEqual(input); + + const transientEntryId = storage.mintId(); + const runningTask: StoredTask = { + ...task, + state: { status: "running", checkpoint: { phase: "effect" } }, + }; + const doneInput: Input = { ...input, status: "done", answer: transientEntryId }; + await expect( + storage.commit( + [ + { type: "task", value: runningTask }, + { type: "input", value: doneInput }, + { type: "entry", value: entry(transientEntryId, rootId, "assistant") }, + { type: "conversation", value: { id: rootId } }, + ], + context, + ), + ).rejects.toThrow(`ID ${rootId} already belongs to conversation`); + + expect(await storage.task(taskId, context)).toEqual(task); + expect(await storage.input(inputId, context)).toEqual(input); + expect(await storage.entries([transientEntryId], context)).toEqual(new Map()); + expect(await storage.entryCommit(transientEntryId, context)).toBeUndefined(); + expect( + await storage.commit([{ type: "entry", value: entry(storage.mintId(), rootId, "after-rollback") }], context), + ).toBe(initialSeq + 1); + }); + + it("detaches retained writes and every returned record", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const entryId = storage.mintId(); + const taskId = storage.mintId(); + const inputId = storage.mintId(); + const entryData = { nested: [1, 2] }; + const checkpoint = { phase: "ready", nested: { count: 1 } }; + const detail = { codes: ["initial"] }; + const storedEntry = entry(entryId, rootId, "note", { data: entryData }); + const storedTask: StoredTask = { + ...pendingTask(taskId, rootId), + state: { status: "pending", checkpoint }, + }; + const storedInput: Input = { + id: inputId, + conversationId: rootId, + status: "unanswered", + reason: "failed", + detail, + }; + await storage.commit( + [ + { type: "entry", value: storedEntry }, + { type: "task", value: storedTask }, + { type: "input", value: storedInput }, + ], + context, + ); + + entryData.nested.push(3); + checkpoint.nested.count = 2; + detail.codes.push("mutated"); + expect((await storage.entries([entryId], context)).get(entryId)?.data).toEqual({ nested: [1, 2] }); + expect((await storage.task(taskId, context))?.state).toEqual({ + status: "pending", + checkpoint: { phase: "ready", nested: { count: 1 } }, + }); + expect((await storage.input(inputId, context))?.detail).toEqual({ codes: ["initial"] }); + + const readEntry = (await storage.entries([entryId], context)).get(entryId)!; + (readEntry.data as { nested: number[] }).nested.push(9); + const readTask = (await storage.task(taskId, context))!; + if (readTask.state.status !== "terminal") { + (readTask.state.checkpoint as { phase: string; nested: { count: number } }).nested.count = 9; + } + const readInput = (await storage.input(inputId, context))!; + (readInput.detail as { codes: string[] }).codes.push("read mutation"); + + expect((await storage.entries([entryId], context)).get(entryId)?.data).toEqual({ nested: [1, 2] }); + expect((await storage.task(taskId, context))?.state).toEqual({ + status: "pending", + checkpoint: { phase: "ready", nested: { count: 1 } }, + }); + expect((await storage.input(inputId, context))?.detail).toEqual({ codes: ["initial"] }); + }); + + it("paginates conversations by opaque cursor in ascending ID order", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const secondId = storage.mintId(); + const thirdId = storage.mintId(); + await storage.commit( + [ + { type: "conversation", value: { id: secondId } }, + { type: "conversation", value: { id: thirdId } }, + ], + context, + ); + + const first = await storage.scanConversations(undefined, 2, context); + expect(first.items.map(({ id }) => id)).toEqual([rootId, secondId]); + expect(first.next).toEqual({ after: secondId }); + const second = await storage.scanConversations(first.next, 2, context); + expect(second.items.map(({ id }) => id)).toEqual([thirdId]); + expect(second.next).toBeUndefined(); + }); + + it("scans deep fork history newest-first through every ancestor cap", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const rootFirst = storage.mintId(); + const rootForkPoint = storage.mintId(); + const rootExcludedSameCommit = storage.mintId(); + const rootEntriesSeq = await storage.commit( + [ + { type: "entry", value: entry(rootFirst, rootId) }, + { + type: "entry", + value: entry(rootForkPoint, rootId, "marker", { head: rootFirst }), + }, + { type: "entry", value: entry(rootExcludedSameCommit, rootId) }, + ], + context, + ); + const childId = storage.mintId(); + await storage.commit( + [ + { + type: "conversation", + value: { id: childId, parent: { conversationId: rootId, at: rootForkPoint } }, + }, + ], + context, + ); + const childForkPoint = storage.mintId(); + const childExcluded = storage.mintId(); + await storage.commit( + [ + { type: "entry", value: entry(childForkPoint, childId, "note") }, + { type: "entry", value: entry(childExcluded, childId) }, + ], + context, + ); + const rootExcludedLater = storage.mintId(); + await storage.commit([{ type: "entry", value: entry(rootExcludedLater, rootId) }], context); + const grandchildId = storage.mintId(); + await storage.commit( + [ + { + type: "conversation", + value: { id: grandchildId, parent: { conversationId: childId, at: childForkPoint } }, + }, + ], + context, + ); + const grandchildHead = storage.mintId(); + const grandchildTail = storage.mintId(); + const grandchildEntriesSeq = await storage.commit( + [ + { + type: "entry", + value: entry(grandchildHead, grandchildId, "marker", { head: grandchildHead }), + }, + { type: "entry", value: entry(grandchildTail, grandchildId) }, + ], + context, + ); + const childExcludedLater = storage.mintId(); + await storage.commit([{ type: "entry", value: entry(childExcludedLater, childId) }], context); + + const first = await storage.scanEntries({ conversationId: grandchildId }, undefined, 2, context); + expect(first.items.map(({ id }) => id)).toEqual([grandchildTail, grandchildHead]); + const second = await storage.scanEntries({ conversationId: grandchildId }, first.next, 2, context); + expect(second.items.map(({ id }) => id)).toEqual([childForkPoint, rootForkPoint]); + const third = await storage.scanEntries({ conversationId: grandchildId }, second.next, 2, context); + expect(third.items.map(({ id }) => id)).toEqual([rootFirst]); + expect(third.next).toBeUndefined(); + + expect( + ( + await storage.scanEntries({ conversationId: grandchildId, before: grandchildHead }, undefined, 10, context) + ).items.map(({ id }) => id), + ).toEqual([childForkPoint, rootForkPoint, rootFirst]); + expect( + ( + await storage.scanEntries({ conversationId: grandchildId, kind: "marker" }, undefined, 10, context) + ).items.map(({ id }) => id), + ).toEqual([grandchildHead, rootForkPoint]); + expect( + ( + await storage.scanEntries({ conversationId: grandchildId, withHead: true }, undefined, 10, context) + ).items.map(({ id }) => id), + ).toEqual([grandchildHead, rootForkPoint]); + + expect(await storage.entryCommit(rootFirst, context)).toBe(rootEntriesSeq); + expect(await storage.entryCommit(rootForkPoint, context)).toBe(rootEntriesSeq); + expect(await storage.entryCommit(grandchildHead, context)).toBe(grandchildEntriesSeq); + expect(await storage.entryCommit(grandchildTail, context)).toBe(grandchildEntriesSeq); + expect(await storage.entryCommit(999_999, context)).toBeUndefined(); + await expect(storage.scanEntries({ conversationId: 999_999 }, undefined, 10, context)).rejects.toThrow( + "Unknown conversation", + ); + }); + + it("replaces complete task records and pages filtered task scans", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const firstId = storage.mintId(); + const secondId = storage.mintId(); + const thirdId = storage.mintId(); + const first = { ...pendingTask(firstId, rootId), memos: { winner: "first" } } satisfies StoredTask; + const second = { ...pendingTask(secondId, rootId), background: true } satisfies StoredTask; + const third = { ...pendingTask(thirdId, rootId), abortRequested: true } satisfies StoredTask; + await storage.commit( + [ + { type: "task", value: first }, + { type: "task", value: second }, + { type: "task", value: third }, + ], + context, + ); + + const running: StoredTask = { + ...first, + state: { status: "running", checkpoint: { phase: "effect", attempt: 1 } }, + abortRequested: true, + }; + await storage.commit([{ type: "task", value: running }], context); + expect(await storage.task(firstId, context)).toEqual(running); + const terminal: StoredTask = { + id: firstId, + conversationId: rootId, + kind: first.kind, + version: first.version, + input: first.input, + state: { status: "terminal", outcome: { status: "completed", result: { entryId: 99 } } }, + after: [], + background: false, + abortRequested: true, + }; + await storage.commit([{ type: "task", value: terminal }], context); + expect(await storage.task(firstId, context)).toEqual(terminal); + + const pendingPage = await storage.scanTasks({ status: "pending" }, undefined, 1, context); + expect(pendingPage.items.map(({ id }) => id)).toEqual([secondId]); + expect(pendingPage.next).toBeDefined(); + expect( + (await storage.scanTasks({ status: "pending" }, pendingPage.next, 1, context)).items.map(({ id }) => id), + ).toEqual([thirdId]); + expect( + (await storage.scanTasks({ status: "terminal", abortRequested: true }, undefined, 10, context)).items, + ).toEqual([terminal]); + expect((await storage.scanTasks({ background: true }, undefined, 10, context)).items.map(({ id }) => id)).toEqual( + [secondId], + ); + }); + + it("indexes request IDs per conversation and replaces complete input records", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const secondConversationId = storage.mintId(); + await storage.commit([{ type: "conversation", value: { id: secondConversationId } }], context); + const firstId = storage.mintId(); + const secondId = storage.mintId(); + const otherConversationId = storage.mintId(); + const first: Input = { + id: firstId, + conversationId: rootId, + requestId: "same", + status: "queued", + }; + const second: Input = { + id: secondId, + conversationId: rootId, + requestId: "other", + status: "queued", + }; + const otherConversation: Input = { + id: otherConversationId, + conversationId: secondConversationId, + requestId: "same", + status: "queued", + }; + await storage.commit( + [ + { type: "input", value: first }, + { type: "input", value: second }, + { type: "input", value: otherConversation }, + ], + context, + ); + expect(await storage.inputByRequest(rootId, "same", context)).toEqual(first); + expect(await storage.inputByRequest(secondConversationId, "same", context)).toEqual(otherConversation); + + const placedSecond: Input = { ...second, status: "placed", entry: storage.mintId() }; + await storage.commit([{ type: "input", value: placedSecond }], context); + expect(await storage.input(secondId, context)).toEqual(placedSecond); + expect(await storage.inputByRequest(rootId, "other", context)).toEqual(placedSecond); + }); + + it("keeps one global record ID namespace and rejects exhausted ID minting", async () => { + const storage = new MemoryStorage(); + const rootId = await createRoot(storage); + const explicitEntryId = 100; + await storage.commit([{ type: "entry", value: entry(explicitEntryId, rootId) }], context); + expect(storage.mintId()).toBe(101); + await expect( + storage.commit([{ type: "task", value: pendingTask(explicitEntryId, rootId) }], context), + ).rejects.toThrow(`ID ${explicitEntryId} already belongs to entry`); + + await storage.commit([{ type: "entry", value: entry(Number.MAX_SAFE_INTEGER, rootId, "last-id") }], context); + expect(() => storage.mintId()).toThrow("ID space is exhausted"); + expect(() => storage.mintId()).toThrow("ID space is exhausted"); + }); + + it("rejects every operation after close", async () => { + const storage = new MemoryStorage(); + await createRoot(storage); + await storage.close(context); + await expect(storage.conversation(ROOT_CONVERSATION_ID, context)).rejects.toThrow("closed"); + await expect(storage.commit([] satisfies StorageWrite[], context)).rejects.toThrow("closed"); + expect(() => storage.mintId()).toThrow("closed"); + }); +}); diff --git a/packages/ai/src/api/bedrock-converse-stream.ts b/packages/ai/src/api/bedrock-converse-stream.ts index dcc747feb..ee66b4599 100644 --- a/packages/ai/src/api/bedrock-converse-stream.ts +++ b/packages/ai/src/api/bedrock-converse-stream.ts @@ -32,6 +32,8 @@ import type { AssistantMessage, CacheRetention, ImageContent, + JsonObject, + JsonValue, Model, ProviderEnv, ProviderResponse, @@ -437,7 +439,7 @@ function appendBedrockFailureDiagnostic( fallbackRequestId: string | undefined, ): void { const metadata = (error as SdkErrorMetadata)?.$metadata; - const details: Record = {}; + const details: JsonObject = {}; if (typeof metadata?.httpStatusCode === "number") details.status = metadata.httpStatusCode; @@ -916,7 +918,7 @@ function createRequiredTextBlock(text: string): ContentBlock.TextMember { return createNonBlankTextBlock(text) ?? { text: EMPTY_TEXT_PLACEHOLDER }; } -function sanitizeBedrockDocument(value: DocumentType): DocumentType { +function sanitizeBedrockDocument(value: JsonValue): DocumentType { if (Array.isArray(value)) { return value.map(sanitizeBedrockDocument); } diff --git a/packages/ai/src/api/mistral-conversations.ts b/packages/ai/src/api/mistral-conversations.ts index 29155f520..837d1ec11 100644 --- a/packages/ai/src/api/mistral-conversations.ts +++ b/packages/ai/src/api/mistral-conversations.ts @@ -721,7 +721,7 @@ async function consumeChatStream( ? toolCall.function.arguments : JSON.stringify(toolCall.function.arguments || {}); block.partialArgs = (block.partialArgs || "") + argsDelta; - block.arguments = parseStreamingJson>(block.partialArgs); + block.arguments = parseStreamingJson(block.partialArgs); stream.push({ type: "toolcall_delta", contentIndex: toolBlocksByKey.get(key)!, @@ -736,7 +736,7 @@ async function consumeChatStream( const block = output.content[index]; if (block.type !== "toolCall") continue; const toolBlock = block as ToolCall & { partialArgs?: string }; - toolBlock.arguments = parseStreamingJson>(toolBlock.partialArgs); + toolBlock.arguments = parseStreamingJson(toolBlock.partialArgs); // Finalize in-place and strip the scratch buffer so replay only // carries parsed arguments. delete toolBlock.partialArgs; diff --git a/packages/ai/src/api/openai-codex-responses.ts b/packages/ai/src/api/openai-codex-responses.ts index fb7461339..8c0e54466 100644 --- a/packages/ai/src/api/openai-codex-responses.ts +++ b/packages/ai/src/api/openai-codex-responses.ts @@ -357,7 +357,7 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons output, createAssistantMessageDiagnostic("provider_transport_failure", error, { configuredTransport: transport, - fallbackTransport: websocketStarted ? undefined : "sse", + ...(websocketStarted ? {} : { fallbackTransport: "sse" }), eventsEmitted: websocketStarted, phase: websocketStarted ? "after_message_stream_start" : "before_message_stream_start", requestBytes: new TextEncoder().encode(bodyJson).byteLength, diff --git a/packages/ai/src/api/pi-messages.ts b/packages/ai/src/api/pi-messages.ts index 46ceef412..4ed6ceadd 100644 --- a/packages/ai/src/api/pi-messages.ts +++ b/packages/ai/src/api/pi-messages.ts @@ -13,6 +13,8 @@ import type { AssistantMessage, AssistantMessageEvent, CacheRetention, + JsonObject, + JsonValue, Model, ProviderEnv, SimpleStreamOptions, @@ -95,9 +97,9 @@ type PiMessagesErrorBody = { export class PiMessagesResponseError extends Error { code?: string; - readonly diagnosticDetails: Record; + readonly diagnosticDetails: JsonObject; - constructor(message: string, code: string | undefined, diagnosticDetails: Record) { + constructor(message: string, code: string | undefined, diagnosticDetails: JsonObject) { super(message); this.name = "PiMessagesResponseError"; this.code = code; @@ -147,8 +149,8 @@ function createPiMessagesResponseError( url: url.toString(), status: response.status, statusText: response.statusText, - error: errorBody?.error, - body: errorBody ? undefined : truncateDiagnosticString(body), + ...(errorBody?.error === undefined ? {} : { error: errorBody.error as JsonValue }), + ...(errorBody ? {} : { body: truncateDiagnosticString(body) }), timestampMs: Date.now(), }); } diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 4dcfd08d5..3ca4ca17c 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -380,7 +380,7 @@ export interface ToolCall { type: "toolCall"; id: string; name: string; - arguments: Record; + arguments: JsonObject; thoughtSignature?: string; // Google-specific: opaque signature for reusing thought context /** OpenAI Responses namespace for calls to dynamically loaded or namespaced tools. */ namespace?: string; @@ -411,7 +411,53 @@ export interface Usage { export type StopReason = "pending" | "stop" | "length" | "toolUse" | "error" | "aborted" | "deferred"; -export type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue }; +export type JsonValue = null | boolean | number | string | readonly JsonValue[] | JsonObject; +export type JsonObject = { [key: string]: JsonValue }; + +type IsAny = 0 extends 1 & T ? true : false; +type IsExactlyJsonValue = [T] extends [JsonValue] ? ([JsonValue] extends [T] ? true : false) : false; +type IsJsonProperty = IsAny extends true + ? false + : unknown extends T + ? false + : [Exclude] extends [never] + ? true + : IsJsonCompatible>; +type InvalidJsonKeys = { + [TKey in keyof T]-?: TKey extends string | number ? (IsJsonProperty extends true ? never : TKey) : TKey; +}[keyof T]; +type IsJsonCompatible = IsAny extends true + ? false + : unknown extends T + ? false + : IsExactlyJsonValue extends true + ? true + : T extends null | boolean | number | string + ? true + : T extends undefined + ? false + : T extends readonly (infer TItem)[] + ? IsJsonCompatible + : T extends (...args: never[]) => unknown + ? false + : T extends object + ? [InvalidJsonKeys] extends [never] + ? true + : false + : false; + +/** The JSON representation of a typed in-memory value. Optional object properties remain optional. */ +export type JsonRepresentation = IsAny extends true + ? JsonValue + : unknown extends T + ? JsonValue + : [T] extends [JsonValue] + ? T + : T extends readonly unknown[] + ? { [TKey in keyof T]: JsonRepresentation> } + : T extends object + ? { [TKey in keyof T]: JsonRepresentation> } + : never; export interface DeferredHandle { provider: string; @@ -490,17 +536,19 @@ export interface AssistantMessage { timestamp: number; // Unix timestamp in milliseconds } -export interface ToolResultMessage { - role: "toolResult"; - toolCallId: string; - toolName: string; - content: (TextContent | ImageContent)[]; // Supports text and images - details?: TDetails; - /** Usage from the tool execution itself, if available. Not part of main LLM context accounting. */ - usage?: Usage; - isError: boolean; - timestamp: number; // Unix timestamp in milliseconds -} +export type ToolResultMessage = IsJsonCompatible extends true + ? { + role: "toolResult"; + toolCallId: string; + toolName: string; + content: (TextContent | ImageContent)[]; // Supports text and images + details?: JsonRepresentation; + /** Usage from the tool execution itself, if available. Not part of main LLM context accounting. */ + usage?: Usage; + isError: boolean; + timestamp: number; // Unix timestamp in milliseconds + } + : never; export type Message = SystemMessage | UserMessage | AssistantMessage | ToolResultMessage; diff --git a/packages/ai/src/utils/diagnostics.ts b/packages/ai/src/utils/diagnostics.ts index 89d135619..fdb1b4819 100644 --- a/packages/ai/src/utils/diagnostics.ts +++ b/packages/ai/src/utils/diagnostics.ts @@ -1,3 +1,5 @@ +import type { JsonObject } from "../types.ts"; + export interface DiagnosticErrorInfo { name?: string; message: string; @@ -9,7 +11,7 @@ export interface AssistantMessageDiagnostic { type: string; timestamp: number; error?: DiagnosticErrorInfo; - details?: Record; + details?: JsonObject; } export function formatThrownValue(value: unknown): string { @@ -32,7 +34,7 @@ export function extractDiagnosticError(error: unknown): DiagnosticErrorInfo { export function createAssistantMessageDiagnostic( type: string, error: unknown, - details?: Record, + details?: JsonObject, ): AssistantMessageDiagnostic { return { type, timestamp: Date.now(), error: extractDiagnosticError(error), details }; } diff --git a/packages/ai/test/bedrock-convert-messages.test.ts b/packages/ai/test/bedrock-convert-messages.test.ts index 4199cde65..e35169069 100644 --- a/packages/ai/test/bedrock-convert-messages.test.ts +++ b/packages/ai/test/bedrock-convert-messages.test.ts @@ -352,7 +352,7 @@ describe("bedrock convertMessages skips unknown content types", () => { }); it("removes empty property names only from replayed Bedrock input", async () => { - const toolArguments = { + const toolArguments: { path: string; edits: Array> } = { path: "/workspace/foobar/file.js", edits: [ { oldText: "first", newText: "updated first" }, diff --git a/packages/ai/test/constrained-sampling.test.ts b/packages/ai/test/constrained-sampling.test.ts index fc70ee1fe..b3ce0ad6a 100644 --- a/packages/ai/test/constrained-sampling.test.ts +++ b/packages/ai/test/constrained-sampling.test.ts @@ -227,7 +227,8 @@ describe("constrained tool sampling", () => { }, ], }); - for (const invalidArguments of [{}, { payload: 42 }]) { + const invalidArgumentsList: ToolCall["arguments"][] = [{}, { payload: 42 }]; + for (const invalidArguments of invalidArgumentsList) { replayedToolCall.arguments = invalidArguments; expect(() => convertResponsesMessages(makeModel(), context, new Set(["openai"]), { diff --git a/packages/ai/test/message-types.test.ts b/packages/ai/test/message-types.test.ts new file mode 100644 index 000000000..a3265d653 --- /dev/null +++ b/packages/ai/test/message-types.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, expectTypeOf, it } from "vitest"; +import type { Context, JsonValue, Message, ToolResultMessage } from "../src/types.ts"; + +const base = { + role: "toolResult" as const, + toolCallId: "call-1", + toolName: "read", + content: [], + isError: false, + timestamp: 1, +}; + +describe("message JSON types", () => { + it("preserves JSON-compatible detail types without requiring index signatures", () => { + interface Details { + path: string; + items: readonly string[]; + summary?: { + lines?: number; + }; + } + + const objectMessage: ToolResultMessage
= { + ...base, + details: { path: "a.ts", items: ["first"], summary: { lines: 3 } }, + }; + const arrayMessage: ToolResultMessage = { ...base, details: ["a", "b"] }; + const readonlyArrayMessage: ToolResultMessage = { + ...base, + details: ["a", { count: 1 }], + }; + const primitiveMessage: ToolResultMessage = { ...base, details: "diagnostic" }; + const nullMessage: ToolResultMessage = { ...base, details: null }; + const transcriptMessage: Message = objectMessage; + const readonlyTranscriptMessage: Message = readonlyArrayMessage; + const context: Context = { messages: [objectMessage, readonlyArrayMessage] }; + + expect(objectMessage.details?.summary?.lines).toBe(3); + expect(arrayMessage.details).toEqual(["a", "b"]); + expect(readonlyArrayMessage.details).toEqual(["a", { count: 1 }]); + expect(primitiveMessage.details).toBe("diagnostic"); + expect(nullMessage.details).toBeNull(); + expect(transcriptMessage).toBe(objectMessage); + expect(readonlyTranscriptMessage).toBe(readonlyArrayMessage); + expect(context.messages).toEqual([objectMessage, readonlyArrayMessage]); + expectTypeOf().toEqualTypeOf(); + }); + + it("rejects non-JSON detail types", () => { + type OptionalAny = { value?: any }; + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf void }>>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + }); +}); diff --git a/packages/ai/test/stream.test.ts b/packages/ai/test/stream.test.ts index 8450b52bd..6614b1164 100644 --- a/packages/ai/test/stream.test.ts +++ b/packages/ai/test/stream.test.ts @@ -305,6 +305,7 @@ async function multiTurn(model: Model, options?: StreamO expect(block.arguments).toBeTruthy(); const { a, b, operation } = block.arguments; + if (typeof a !== "number" || typeof b !== "number") throw new Error("Invalid math arguments"); let result: number; switch (operation) { case "add": diff --git a/packages/ai/test/validation.test.ts b/packages/ai/test/validation.test.ts index 1a212ac12..816f99a64 100644 --- a/packages/ai/test/validation.test.ts +++ b/packages/ai/test/validation.test.ts @@ -1,7 +1,7 @@ import { Type } from "typebox"; import { Compile } from "typebox/compile"; import { describe, expect, it } from "vitest"; -import type { Tool, ToolCall } from "../src/types.ts"; +import type { JsonValue, Tool, ToolCall } from "../src/types.ts"; import { validateToolArguments } from "../src/utils/validation.ts"; function createToolCallWithPlainSchema( @@ -27,7 +27,7 @@ function createToolCallWithPlainSchema( type: "toolCall", id: "tool-1", name: "echo", - arguments: { value }, + arguments: { value: value as JsonValue }, }; return { tool, toolCall }; diff --git a/packages/coding-agent/test/suite/agent-session-model-extension.test.ts b/packages/coding-agent/test/suite/agent-session-model-extension.test.ts index 39abd46be..a9f6a8d0e 100644 --- a/packages/coding-agent/test/suite/agent-session-model-extension.test.ts +++ b/packages/coding-agent/test/suite/agent-session-model-extension.test.ts @@ -3,6 +3,7 @@ import { fauxAssistantMessage, fauxToolCall, getCurrentSystemPrompt, + type JsonObject, type Model, type Usage, } from "@earendil-works/pi-ai"; @@ -345,7 +346,12 @@ describe("AgentSession model and extension characterization", () => { expect(getAssistantTexts(harness)).toContain("patched result"); const toolResult = harness.session.messages.find( - (message) => message.role === "toolResult" && message.details?.patched === true, + (message) => + message.role === "toolResult" && + typeof message.details === "object" && + message.details !== null && + !Array.isArray(message.details) && + (message.details as JsonObject).patched === true, ); expect(observedToolUsage).toEqual(toolUsage); expect(toolResult).toBeDefined(); diff --git a/packages/coding-agent/test/test-harness.ts b/packages/coding-agent/test/test-harness.ts index 0203e8aa3..0f83a9dec 100644 --- a/packages/coding-agent/test/test-harness.ts +++ b/packages/coding-agent/test/test-harness.ts @@ -17,6 +17,7 @@ import type { AssistantMessage, AssistantMessageEvent, AssistantMessageEventStream, + JsonObject, Model, SimpleStreamOptions, StopReason, @@ -68,7 +69,7 @@ export interface FauxResponse { /** Text content blocks. String shorthand becomes a single text block. */ text?: string; /** Tool calls to include in the response. */ - toolCalls?: Array<{ id?: string; name: string; args: Record }>; + toolCalls?: Array<{ id?: string; name: string; args: JsonObject }>; /** Thinking content. */ thinking?: string; /** Stop reason. Defaults to "stop", or "toolUse" if toolCalls are present, or "error" if error is set. */