From a68caac378a7d553de78571d9a28db0f0e095039 Mon Sep 17 00:00:00 2001 From: James Magahern Date: Sun, 25 Oct 2026 10:15:35 -0700 Subject: [PATCH] fix(agent): handle partial room timestamps without crashing --- docs/api/rest.md | 19 +++++ docs/api/streaming-chat.md | 8 ++ web/src/lib/agent-messages.ts | 55 ++++++++++++++ web/src/lib/agent-optimistic-outbox.ts | 3 +- web/src/lib/agent-room.ts | 64 +++++++++------- web/tests/agent-optimistic-outbox.test.mjs | 9 +++ web/tests/agent-room.test.mjs | 87 +++++++++++++++++++++- 7 files changed, 218 insertions(+), 27 deletions(-) create mode 100644 web/src/lib/agent-messages.ts diff --git a/docs/api/rest.md b/docs/api/rest.md index e9c0f81..3c71742 100644 --- a/docs/api/rest.md +++ b/docs/api/rest.md @@ -627,6 +627,25 @@ messages/DOM grow with explicit older loads and live updates. Offscreen web rows use CSS `content-visibility: auto`, which skips rendering work while retaining chronological DOM, selection, find and accessibility; it is not DOM virtualization. +Every room message, including protected tool rows and `activity.queuedMessages`, +has a string `id`, `role`, `content`, and ISO-8601 `createdAt`. Privacy filtering +retains this envelope and tool lifecycle metadata (`kind`, `toolCallId`, +`toolName`, `entryId`, `status`, and available `startedAt`/`completedAt`). Protected +tools use their public tool name as `content`; private arguments, previews, and +errors are withheld. Result-only pages may omit initiation metadata. + +Historical entries without a known time use `1970-01-01T00:00:00.000Z` as an +unknown-time sentinel. For compatibility with older partial responses, clients +may recover missing/invalid `createdAt` from `metadata.startedAt`, then +`metadata.completedAt`, or use that sentinel; never substitute the current time. +Older redacted tool rows containing only `id`, `role`, and `name` can retain their +identity with a tool-name label until a complete upsert arrives. The web client +uses the same normalization for REST, WebSocket, and queued activity, retains +known times across partial upserts, and orders by time, numeric `entryId` (when +present, before absent), then stable `id`. Unknown times display as “Earlier” +without a clock time. Invalid identities or non-tool message bodies reject the +page before advancing its replay cursor; timestamps never replace cursors. + ### `POST /v1/agent/messages` Body: `{ "content": "Check the oven", "requestId": "client-generated-uuid" }`. diff --git a/docs/api/streaming-chat.md b/docs/api/streaming-chat.md index 58e35d8..b432c54 100644 --- a/docs/api/streaming-chat.md +++ b/docs/api/streaming-chat.md @@ -326,6 +326,14 @@ with `metadata.kind: "tool_call"`, matching Sybil's tool renderer. REST pages an socket messages are upserts keyed by stable `id`. Save each returned cursor after applying its entire frame, even when it contains no visible messages. +Both message frames and queued activity use the required message envelope and +historical timestamp compatibility rules in [REST room history](./rest.md#get-v1agentmessages). +Protected tool redaction preserves `createdAt`, a safe string `content`, and +available lifecycle metadata; it does not remove the row's identity or replay +metadata. Partial tool results may omit `args`/`startedAt`; merge them without +losing a known creation time or downgrading a completed tool on replay. Validate +and normalize all rows before applying a frame or advancing its cursor. + Clients may send `{ "type": "ping" }` (receiving `{ "type": "pong" }`) or `{ "type": "subscribe", "after": "44" }` to change the replay cursor. Transport ping/pong heartbeats detect dead connections. Other writes belong on REST. diff --git a/web/src/lib/agent-messages.ts b/web/src/lib/agent-messages.ts new file mode 100644 index 0000000..f8b0b07 --- /dev/null +++ b/web/src/lib/agent-messages.ts @@ -0,0 +1,55 @@ +import type { Message } from "./api"; + +// The runtime also uses the Unix epoch for historical entries without a time. +// The transcript renders these as "Earlier", without inventing a send time. +export const UNKNOWN_AGENT_TIME = "1970-01-01T00:00:00.000Z"; +const object = (value: unknown): Record | undefined => value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record : undefined; +const timestamp = (value: unknown): string | undefined => { + if (typeof value !== "string" || !value.trim()) return undefined; + const time = Date.parse(value); + return Number.isFinite(time) && time !== 0 ? new Date(time).toISOString() : undefined; +}; + +export function agentMessageTime(message: Pick): string | undefined { + const info = object(message.metadata); + return timestamp(message.createdAt) ?? timestamp(info?.startedAt) ?? timestamp(info?.completedAt); +} + +/** Normalize historical/partial rows at both transports, before state or cursors change. */ +export function normalizeAgentMessage(value: unknown): Message { + const message = object(value); + if (!message || typeof message.id !== "string" || !message.id + || typeof message.role !== "string" || !["system", "user", "assistant", "tool"].includes(message.role)) throw new Error("Invalid agent message."); + const info = object(message.metadata); + // Older runtime REST redaction emitted protected tool rows with only id/role/name. + // Keep their identity for a later complete upsert; never discard a whole page. + const content = typeof message.content === "string" ? message.content + : message.role === "tool" && message.content == null ? (typeof message.name === "string" ? message.name : "Tool") : undefined; + if (content === undefined) throw new Error("Invalid agent message."); + const createdAt = agentMessageTime(message as Message) ?? UNKNOWN_AGENT_TIME; + const needsToolMetadata = message.role === "tool" && info?.kind !== "tool_call"; + if (message.createdAt === createdAt && message.content === content && !needsToolMetadata) return message as Message; + return { ...message, content, createdAt, + ...(needsToolMetadata ? { metadata: { ...info, kind: "tool_call" } } : {}), + } as Message; +} + +function cursor(message: Message): bigint | undefined { + const entryId = object(message.metadata)?.entryId; + if ((typeof entryId === "number" && Number.isSafeInteger(entryId) && entryId >= 0) + || (typeof entryId === "string" && /^\d+$/.test(entryId))) return BigInt(entryId); +} + +/** One total order for history, queue snapshots and optimistic rows, including ties. */ +export function compareAgentMessages(left: Message, right: Message): number { + const byTime = Date.parse(agentMessageTime(left) ?? UNKNOWN_AGENT_TIME) - Date.parse(agentMessageTime(right) ?? UNKNOWN_AGENT_TIME); + if (byTime) return byTime; + const a = cursor(left), b = cursor(right); + if (a !== b) { + if (a === undefined) return 1; + if (b === undefined) return -1; + return a < b ? -1 : 1; + } + return left.id < right.id ? -1 : left.id > right.id ? 1 : 0; +} diff --git a/web/src/lib/agent-optimistic-outbox.ts b/web/src/lib/agent-optimistic-outbox.ts index 2690e89..22499a2 100644 --- a/web/src/lib/agent-optimistic-outbox.ts +++ b/web/src/lib/agent-optimistic-outbox.ts @@ -1,5 +1,6 @@ import type { Message } from "./api"; import { agentAttachments, attachmentCaption, parseAgentAttachment, type AgentAttachment } from "./agent-attachments.ts"; +import { compareAgentMessages, normalizeAgentMessage } from "./agent-messages.ts"; export const AGENT_OUTBOX_STORAGE_KEY = "sybil.agent.optimistic-outbox.v1"; export type AgentDeliveryStatus = "sending" | "queued" | "complete" | "failed" | "unconfirmed"; @@ -165,7 +166,7 @@ export function createAgentOptimisticOutbox(options: { metadata: { kind: "message", status: row.status, channel: "sybil", requestId: row.requestId, ...(row.attachments?.length ? { agentAttachments: row.attachments } : {}), ...(row.error ? { error: row.error } : {}) }, }); } - return [...displayed.values()].sort((a, b) => a.createdAt.localeCompare(b.createdAt)); + return [...displayed.values()].map(normalizeAgentMessage).sort(compareAgentMessages); }, snapshot(): AgentOutboxRow[] { return [...rows.values()].map(row => ({ ...row })); }, }; diff --git a/web/src/lib/agent-room.ts b/web/src/lib/agent-room.ts index 0f59f8f..b2d4602 100644 --- a/web/src/lib/agent-room.ts +++ b/web/src/lib/agent-room.ts @@ -1,6 +1,7 @@ import type { Message } from "./api"; import { parseAgentAttachment, type AgentAttachment, type AgentAttachmentCapability } from "./agent-attachments.ts"; import { agentPushCursor, agentPushSourceId, agentPushUuid } from "./agent-push.ts"; +import { agentMessageTime, compareAgentMessages, normalizeAgentMessage, UNKNOWN_AGENT_TIME } from "./agent-messages.ts"; export type AgentActivity = { state: "idle" | "thinking" | "tool" | "waiting" | "error"; @@ -57,7 +58,9 @@ function mergeMessage(previous: Message, incoming: Message): Message { const prior = metadata(previous), next = metadata(incoming); // A delayed activity snapshot must not demote an already placed user message. if (prior?.status === "complete" && next?.status === "queued") return previous; - if (prior?.kind !== "tool_call" || next?.kind !== "tool_call") return { ...previous, ...incoming }; + const priorTime = agentMessageTime(previous), nextTime = agentMessageTime(incoming); + if (prior?.kind !== "tool_call" || next?.kind !== "tool_call") return { ...previous, ...incoming, + createdAt: nextTime ?? priorTime ?? UNKNOWN_AGENT_TIME }; const terminal = prior.status === "completed" || prior.status === "failed"; const staleStart = terminal && next.status === "initiated"; const merged = staleStart ? { ...defined(next), ...defined(prior) } : { ...defined(prior), ...defined(next) }; @@ -65,23 +68,32 @@ function mergeMessage(previous: Message, incoming: Message): Message { const oldArgs = object(prior.args), newArgs = object(next.args); if (oldArgs && (!newArgs || Object.keys(newArgs).length === 0)) merged.args = oldArgs; return { ...previous, ...incoming, ...(staleStart ? { content: previous.content, name: previous.name } : {}), - createdAt: previous.createdAt < incoming.createdAt ? previous.createdAt : incoming.createdAt, metadata: merged }; + createdAt: priorTime && nextTime ? (priorTime < nextTime ? priorTime : nextTime) : priorTime ?? nextTime ?? UNKNOWN_AGENT_TIME, metadata: merged }; } /** Replays and separate tool-start/result pages update stable rows instead of duplicating them. */ export function mergeAgentMessages(previous: readonly Message[], incoming: readonly Message[]): Message[] { - const messages = new Map(previous.map(message => [message.id, message])); - for (const message of incoming) { + const messages = new Map(previous.map(normalizeAgentMessage).map(message => [message.id, message])); + for (const message of incoming.map(normalizeAgentMessage)) { const prior = messages.get(message.id); messages.set(message.id, prior ? mergeMessage(prior, message) : message); } - return [...messages.values()].sort((a, b) => { - const byTime = a.createdAt.localeCompare(b.createdAt); - if (byTime) return byTime; - const left = messageCursor(a), right = messageCursor(b); - if (left !== null && right !== null && left !== right) return BigInt(left) < BigInt(right) ? -1 : 1; - return a.id.localeCompare(b.id); - }); + return [...messages.values()].sort(compareAgentMessages); +} + +function parseAgentPage(value: unknown): AgentMessagesPage { + const page = object(value); + if (!page || typeof page.conversationId !== "string" || !Array.isArray(page.messages) || typeof page.hasMore !== "boolean") throw new Error("Invalid agent messages frame."); + if (page.cursor !== null) normalizeAgentCursor(page.cursor); + if (page.beforeCursor != null) normalizeAgentCursor(page.beforeCursor); + return { ...page, beforeCursor: page.beforeCursor ?? null, messages: page.messages.map(normalizeAgentMessage) } as AgentMessagesPage; +} +function parseAgentActivity(value: unknown): AgentActivity { + const activity = object(value); + if (!activity || (activity.tools !== undefined && !Array.isArray(activity.tools)) + || (activity.queuedMessages !== undefined && !Array.isArray(activity.queuedMessages))) throw new Error("Invalid agent activity frame."); + return { ...activity, tools: activity.tools ?? [], + queuedMessages: ((activity.queuedMessages ?? []) as unknown[]).map(normalizeAgentMessage) } as AgentActivity; } export const createAgentRoomState = (): AgentRoomState => ({ conversationId: null, messages: [], cursor: "0", readCursor: "0", activity: null, @@ -99,6 +111,7 @@ function readSource(state: Pick { @@ -153,15 +166,12 @@ export function parseAgentRoomEvent(raw: string): AgentRoomEvent | null { || (event.updatedAt !== null && typeof event.updatedAt !== "string") || (event.sourceId !== undefined && event.sourceId !== null && !agentPushSourceId(event.sourceId)))) throw new Error("Invalid Agent read frame."); if (event.type === "messages") { - if (typeof event.conversationId !== "string" || !Array.isArray(event.messages) || typeof event.hasMore !== "boolean") throw new Error("Invalid agent messages frame."); - if (event.cursor !== null) normalizeAgentCursor(event.cursor); - for (const item of event.messages) { - const message = object(item); - if (!message || typeof message.id !== "string" || typeof message.content !== "string" || typeof message.createdAt !== "string" - || !["system", "user", "assistant", "tool"].includes(String(message.role))) throw new Error("Invalid agent message."); - } + return { ...parseAgentPage(event), type: "messages" }; + } + if (event.type === "activity") { + if (typeof event.conversationId !== "string") throw new Error("Invalid agent activity frame."); + return { ...event, activity: parseAgentActivity(event.activity) } as AgentRoomEvent; } - if (event.type === "activity" && (typeof event.conversationId !== "string" || !object(event.activity))) throw new Error("Invalid agent activity frame."); return event as AgentRoomEvent; } export function agentAuthFrame(token: string | null, cursor: string, presenceSessionId?: string): string { @@ -197,12 +207,16 @@ export class AgentRoomClient { } return value as T; } - descriptor(signal?: AbortSignal): Promise { return this.request("", { signal }); } - messages(after: string, signal?: AbortSignal): Promise { - return this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal }); + async descriptor(signal?: AbortSignal): Promise { + const value = object(await this.request("", { signal })); + if (!value || typeof value.id !== "string") throw new Error("Invalid agent descriptor."); + return { ...value, activity: parseAgentActivity(value.activity) } as AgentDescriptor; } - history(before?: string, signal?: AbortSignal): Promise { - return this.request(`/messages?${new URLSearchParams({ ...(before === undefined ? {} : { before: normalizeAgentCursor(before) }), limit: "100" })}`, { signal }); + async messages(after: string, signal?: AbortSignal): Promise { + return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal })); + } + async history(before?: string, signal?: AbortSignal): Promise { + return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ ...(before === undefined ? {} : { before: normalizeAgentCursor(before) }), limit: "100" })}`, { signal })); } readState(signal?: AbortSignal): Promise { return this.request("/read-state", { signal }); } markRead(cursor: string, signal?: AbortSignal, sourceId?: string | null): Promise { diff --git a/web/tests/agent-optimistic-outbox.test.mjs b/web/tests/agent-optimistic-outbox.test.mjs index 1b56dd1..a8d0334 100644 --- a/web/tests/agent-optimistic-outbox.test.mjs +++ b/web/tests/agent-optimistic-outbox.test.mjs @@ -7,6 +7,15 @@ const canonical = (id, content = "Hello", time = "2026-10-14T12:01:00.000Z") => metadata: { kind: "message", entryId: 123, status: "complete", channel: "sybil" }, }); const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; }; + +test("display sorts partial server rows and tied entry cursors consistently", () => { + const box = createAgentOptimisticOutbox({ submit: async () => assert.fail("no send"), + storage: { read: () => [], write: () => {} }, changed: () => {} }); + const rows = [canonical("z"), { id: "tool:1:partial", role: "tool", name: "owner_phone_context" }, + { ...canonical("a"), metadata: { entryId: 124 } }]; + assert.deepEqual(box.display(rows).map(row => row.id), ["tool:1:partial", "z", "a"]); + assert.deepEqual(box.display([...rows].reverse()), box.display(rows)); +}); function fixture(submit, legacy = {}) { let stored = null; let serial = 0; diff --git a/web/tests/agent-room.test.mjs b/web/tests/agent-room.test.mjs index 3f072e9..e624266 100644 --- a/web/tests/agent-room.test.mjs +++ b/web/tests/agent-room.test.mjs @@ -21,6 +21,91 @@ const sourceA = "4599d348-e053-4937-987d-1392746fb2ce", sourceB = "22222222-2222 const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; }; const sourceRoom = () => applyAgentPage(applyAgentReadState(createAgentRoomState(), { sourceId: sourceA, cursor: "90", updatedAt: null }), page("100", [message(100)])); +test("redacted historical tool rows survive REST loading, socket replay and outbox rendering", async () => { + const redacted = { id: "tool:1:protected", role: "tool", name: "owner_phone_context" }; + const input = page("3", [message(1), redacted, message(3)]); + const client = new AgentRoomClient({ fetch: async () => Response.json(input) }); + let state = createAgentRoomState(); + await syncAgentHistory(client, "0", update => { state = applyAgentPage(state, update); }); + const initial = state.messages; + assert.deepEqual((await client.history()).messages, (await client.messages("0")).messages, + "recent/older history uses the same boundary as forward replay"); + const row = initial.find(row => row.id === redacted.id); + assert.equal(row.createdAt, "1970-01-01T00:00:00.000Z", "unknown times are stable, never fabricated as now"); + assert.equal(row.content, "owner_phone_context"); + assert.equal(state.cursor, "3"); + const event = parseAgentRoomEvent(JSON.stringify({ type: "messages", ...input })); + state = applyAgentEvent(state, event); + assert.deepEqual(state.messages, initial, "replaying partial history is idempotent"); + const box = createAgentOptimisticOutbox({ submit: async () => assert.fail("no send"), + storage: { read: () => [], write: () => {} }, changed: () => {} }); + assert.deepEqual(box.display(agentDisplayMessages(state)), initial); +}); + +test("partial tool updates retain known time, terminal metadata and chronological ordering", () => { + const id = "tool:1:partial"; + const start = message(2, "tool", { id, metadata: { kind: "tool_call", entryId: 2, status: "initiated", startedAt: "2026-10-14T10:00:02.000Z" } }); + const done = message(3, "tool", { id, createdAt: undefined, metadata: { kind: "tool_call", entryId: 3, status: "completed", completedAt: "2026-10-14T10:00:03.000Z" } }); + const partial = { id, role: "tool", name: "owner_phone_context" }; + const rows = mergeAgentMessages([message(1), start], [done, partial, start, message(4)]); + assert.deepEqual(rows.map(row => row.id), ["entry:1", id, "entry:4"]); + assert.equal(rows[1].createdAt, start.createdAt); + assert.equal(rows[1].metadata.status, "completed"); + assert.equal(rows[1].metadata.completedAt, done.metadata.completedAt); + assert.equal(mergeAgentMessages([], [done])[0].createdAt, done.metadata.completedAt); + assert.equal(mergeAgentMessages([partial], [start])[0].createdAt, start.createdAt); +}); + +test("an unknown historical tool start never masks a known completion timestamp", () => { + const unknown = "1970-01-01T00:00:00.000Z", known = "2026-10-14T10:00:03.000Z"; + const start = message(2, "tool", { id: "tool:1:historical", createdAt: unknown, + metadata: { kind: "tool_call", entryId: 2, status: "initiated", startedAt: unknown } }); + const done = message(3, "tool", { id: start.id, createdAt: known, + metadata: { kind: "tool_call", entryId: 3, status: "completed", completedAt: known } }); + for (const rows of [[start, done], [done, start], [{ ...start, metadata: { ...start.metadata, ...done.metadata } }]]) { + const merged = mergeAgentMessages([], rows); + assert.equal(merged[0].createdAt, known); + assert.equal(merged[0].metadata.status, "completed"); + } +}); + +test("missing, null, invalid and tied timestamps have stable ordering independent of arrival order", () => { + const rows = [message(11, "assistant", { createdAt: null }), message(2, "assistant", { createdAt: undefined }), + message(3, "assistant", { createdAt: "not-a-date" }), message(4, "assistant", { createdAt: 0 }), + message(5, "assistant", { createdAt: null, metadata: { kind: "message" } })]; + const expected = ["entry:2", "entry:3", "entry:4", "entry:11", "entry:5"]; + assert.deepEqual(mergeAgentMessages([], rows).map(row => row.id), expected); + assert.deepEqual(mergeAgentMessages([], [...rows].reverse()).map(row => row.id), expected); + const tied = [message(11, "assistant", { createdAt: "2026-10-14T10:00:00Z" }), message(2, "assistant", { createdAt: "2026-10-14T11:00:00+01:00" })]; + assert.deepEqual(mergeAgentMessages([], tied).map(row => row.id), ["entry:2", "entry:11"]); +}); + +test("REST descriptors and socket activity normalize partial queued messages consistently", async () => { + const pending = message(1, "user", { id: "inbound:sybil:partial", createdAt: undefined, metadata: { kind: "message", status: "queued" } }); + const value = { id: "1", activity: activity([pending]) }; + const client = new AgentRoomClient({ fetch: async () => Response.json(value) }); + const descriptor = await client.descriptor(); + const fromRest = applyAgentEvent(createAgentRoomState(), { type: "activity", conversationId: descriptor.id, activity: descriptor.activity }); + const fromSocket = applyAgentEvent(createAgentRoomState(), parseAgentRoomEvent(JSON.stringify({ type: "activity", conversationId: "1", activity: value.activity }))); + assert.deepEqual(agentDisplayMessages(fromRest), agentDisplayMessages(fromSocket)); + assert.equal(agentDisplayMessages(fromRest)[0].createdAt, "1970-01-01T00:00:00.000Z"); + const malformed = { ...pending, content: undefined }; + assert.throws(() => applyAgentEvent(fromRest, { type: "activity", conversationId: "1", activity: activity([malformed]) }), /Invalid agent message/); + assert.equal(fromRest.cursor, "0"); +}); + +test("invalid message bodies fail at both transport boundaries before advancing replay", async () => { + for (const malformed of [null, { role: "assistant", content: "missing ID" }, message(2, "assistant", { content: null }), message(2, ["assistant"])]) { + const input = page("3", [message(1), malformed]); + const client = new AgentRoomClient({ fetch: async () => Response.json(input) }); + let applied = false; + await assert.rejects(syncAgentHistory(client, "0", () => { applied = true; }), /Invalid agent message/); + await assert.rejects(client.history(), /Invalid agent message/); + assert.equal(applied, false); + assert.throws(() => parseAgentRoomEvent(JSON.stringify({ type: "messages", ...input })), /Invalid agent message/); + } +}); + test("tool starts and results merge across replay without downgrading completed cards", () => { const start = message(2, "tool", { id: "tool:1:call1", metadata: { kind: "tool_call", entryId: 2, status: "initiated", args: { query: "cookie" }, startedAt: "2026-10-14T10:00:00Z" } }); const done = message(3, "tool", { id: start.id, metadata: { kind: "tool_call", entryId: 3, status: "completed", resultPreview: "found" } }); @@ -312,7 +397,7 @@ test("client uses Sybil auth and routes while websocket credentials remain in th let token = "first-session-token"; const client = new AgentRoomClient({ baseUrl: "https://sybil.example/api/", token: () => token, fetch: async (url, init) => { calls.push({ url, init }); return new Response(JSON.stringify(init.method === "POST" - ? { id: 1, status: "queued", messageId: "inbound:sybil:request-1" } : { cursor: "0", updatedAt: null }), { status: 200 }); + ? { id: 1, status: "queued", messageId: "inbound:sybil:request-1" } : url.includes("/messages?") ? page("0") : { cursor: "0", updatedAt: null }), { status: 200 }); } }); await client.messages("0"); token = "second-session-token";