buzzert/Sybil-2
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agent): handle partial room timestamps without crashing

Browse Source
This commit is contained in:
buzzert
2026-10-25 10:16:37 -07:00
parent b743523fd5
commit a68caac378
7 changed files with 218 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/lib/agent-messages.ts
+55
View File
@@ -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<string, unknown> | undefined => value !== null && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown> : 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<Message, "createdAt" | "metadata">): 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;
}
web/src/lib/agent-optimistic-outbox.ts
+2 -1
View File
@@ -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 })); },
};
web/src/lib/agent-room.ts
+39 -25
View File
@@ -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<AgentRoomState, "sourceId" | "cursor" | "readCur
return incoming;
}
export function applyAgentPage(state: AgentRoomState, page: AgentMessagesPage): AgentRoomState {
page = parseAgentPage(page);
if (state.conversationId !== null && state.conversationId !== page.conversationId) throw new Error("Agent conversation changed; reload the page.");
// Validate before changing state. Failed delivery/decoding cannot advance replay position.
const cursor = page.cursor === null ? state.cursor : advanceAgentCursor(state.cursor, page.cursor);
@@ -132,7 +145,7 @@ export function applyAgentEvent(state: AgentRoomState, event: AgentRoomEvent): A
if (event.type === "read_state") return applyAgentReadState(state, event);
if (event.type !== "activity") return state;
if (state.conversationId !== null && state.conversationId !== event.conversationId) throw new Error("Agent activity belongs to a different conversation.");
return { ...state, conversationId: event.conversationId, activity: event.activity };
return { ...state, conversationId: event.conversationId, activity: parseAgentActivity(event.activity) };
}
export function agentUnreadCount(state: AgentRoomState): number {
return state.messages.filter(message => {
@@ -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<AgentDescriptor> { return this.request("", { signal }); }
messages(after: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
return this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal });
async descriptor(signal?: AbortSignal): Promise<AgentDescriptor> {
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<AgentMessagesPage> {
return this.request(`/messages?${new URLSearchParams({ ...(before === undefined ? {} : { before: normalizeAgentCursor(before) }), limit: "100" })}`, { signal });
async messages(after: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal }));
}
async history(before?: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ ...(before === undefined ? {} : { before: normalizeAgentCursor(before) }), limit: "100" })}`, { signal }));
}
readState(signal?: AbortSignal): Promise<AgentReadState> { return this.request("/read-state", { signal }); }
markRead(cursor: string, signal?: AbortSignal, sourceId?: string | null): Promise<AgentReadState> {
web/tests/agent-optimistic-outbox.test.mjs
+9
View File
@@ -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;
web/tests/agent-room.test.mjs
+86 -1
View File
@@ -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";