Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions apps/web/src/lib/cloud-agent-next/message-ordering.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -56,4 +56,37 @@ describe('sortSessionMessagesForDisplay', () => {
it('handles an empty message list', () => {
expect(sortSessionMessagesForDisplay([])).toEqual([]);
});

it('orders timed rows turn by turn, matching the cloud-agent-next transcript', () => {
const row = (id: string, created: number, parentID?: string) => ({
info: {
id,
sessionID: 'ses_1',
role: parentID === undefined ? 'user' : 'assistant',
time: { created },
...(parentID === undefined ? {} : { parentID }),
},
parts: [],
});
// Ids disagree with time, and a later prompt falls between the steps of
// the turn that was running when it was sent.
const sorted = sortSessionMessagesForDisplay([
row('msg_d', 2000, 'msg_c'),
row('msg_c', 900),
row('msg_a', 1500),
row('msg_b', 1000, 'msg_c'),
]);

expect(sorted.map(m => m.info.id)).toEqual(['msg_c', 'msg_b', 'msg_d', 'msg_a']);
});

it('keeps every row when ids repeat', () => {
const input = [message('msg_2'), message('msg_1'), message('msg_2')];

expect(sortSessionMessagesForDisplay(input).map(m => m.info.id)).toEqual([
'msg_1',
'msg_2',
'msg_2',
]);
});
});
33 changes: 23 additions & 10 deletions apps/web/src/lib/cloud-agent-next/message-ordering.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import type { MessageInfo } from '@kilocode/cloud-agent-sdk';
import { orderMessageIds } from '@kilocode/cloud-agent-sdk/storage/helpers';

type IdentifiedSessionMessage = {
info: { id: string };
parts: Array<{ id: string }>;
Expand All @@ -8,24 +11,34 @@ type IdentifiedSessionMessage = {
*
* The session-ingest export endpoint streams messages and parts in ingest
* order (`ingested_at, id`), which can differ from conversation order when a
* session is re-ingested or history arrives out of order. The cloud-agent-next
* UI re-establishes display order by inserting each message/part into storage
* sorted by its time-ordered ID (`insertSorted` / `insertPartSorted` in
* `@kilocode/cloud-agent-sdk` storage). Message and part IDs are
* `msg_`/`part_` + big-endian hex timestamp + random suffix, so plain
* lexicographic ordering matches chronological order.
* session is re-ingested or history arrives out of order. Read-only views that
* render the raw export (the admin session trace viewer and the public
* shared-session page) apply the same turn order as the cloud-agent-next
* transcript (`orderMessageIds` in `@kilocode/cloud-agent-sdk`), so they match
* what users see. Parts keep their time-ordered id order, as in SDK storage.
*
* Read-only views that render the raw export (the admin session trace viewer
* and the public shared-session page) must apply the same ordering to match
* what users see.
* Messages are sorted by id first: that is the order `orderMessageIds` keeps
* when a row has no readable time.
*/
export function sortSessionMessagesForDisplay<T extends IdentifiedSessionMessage>(
messages: T[]
): T[] {
const byIdAscending = (a: { id: string }, b: { id: string }) =>
a.id < b.id ? -1 : a.id > b.id ? 1 : 0;

return messages
const sorted = messages
.map(message => ({ ...message, parts: [...message.parts].sort(byIdAscending) }))
.sort((a, b) => byIdAscending(a.info, b.info));
const byId = new Map(sorted.map(message => [message.info.id, message]));
if (byId.size !== sorted.length) return sorted;

// Export rows are unvalidated; `orderMessageIds` reads every field it uses
// defensively and keeps the id order for a row it cannot place.
const infos = new Map(
sorted.map(message => [message.info.id, message.info as unknown as MessageInfo])
);
return orderMessageIds(
sorted.map(message => message.info.id),
infos
).flatMap(id => byId.get(id) ?? []);
}
4 changes: 2 additions & 2 deletions apps/web/src/routers/admin-router.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2774,8 +2774,8 @@ export const adminRouter = createTRPCRouter({

try {
const snapshot = await fetchSessionSnapshot(input.session_id, session.kilo_user_id);
// The export streams messages/parts in ingest order; re-sort by the
// time-ordered IDs so the trace matches the cloud-agent-next UI.
// The export streams messages/parts in ingest order; re-sort in the
// transcript's turn order so the trace matches the cloud-agent-next UI.
return {
messages: sortSessionMessagesForDisplay(snapshot?.messages ?? []),
format: 'v2' as const,
Expand Down
2 changes: 2 additions & 0 deletions packages/app-shared/src/opencode.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@ export type UserMessage = {
variant?: string;
/** Kilo extension: marks a client-materialised row not yet confirmed by the server. */
synthetic?: boolean;
/** Kilo extension: server time a client-materialised row was queued; orders it until the server's own row replaces it. */
queuedAt?: number;
editorContext?: {
visibleFiles?: Array<string>;
openTabs?: Array<string>;
Expand Down
72 changes: 72 additions & 0 deletions packages/cloud-agent-sdk/src/chat-processor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -983,6 +983,76 @@ describe('createChatProcessor', () => {
});

describe('synthesizeQueuedUserMessage', () => {
it('records the server queue time on a new unconfirmed row', () => {
const storage = createMemoryStorage();
const processor = createChatProcessor(storage);
processor.synthesizeQueuedUserMessage({
messageId: 'msg-queued',
sessionId: 'ses-1',
content: 'hello',
createdAt: 100,
});
expect(storage.getMessageInfo('msg-queued')).toMatchObject({
time: { created: 100 },
queuedAt: 100,
synthetic: true,
});
expect(storage.getParts('msg-queued')).toMatchObject([{ text: 'hello', synthetic: true }]);
});

it('records the server queue time on an optimistic row without replacing its text or attachments', () => {
const storage = createMemoryStorage();
const processor = createChatProcessor(storage);
const optimistic = { ...makeUserMsg('msg-1'), synthetic: true };
storage.upsertMessage(optimistic);
const text = { ...makeTextPart('msg-1-text', 'msg-1', 'original text'), synthetic: true };
const file: Part = {
id: 'msg-1-file',
messageID: 'msg-1',
sessionID: 'ses-1',
type: 'file',
mime: 'image/png',
url: 'data:image/png;base64,AA==',
synthetic: true,
};
storage.upsertPart('msg-1', text);
storage.upsertPart('msg-1', file);
const parts = storage.getParts('msg-1');

processor.synthesizeQueuedUserMessage({
messageId: 'msg-1',
sessionId: 'ses-1',
content: undefined,
createdAt: 100,
});

expect(storage.getMessageInfo('msg-1')).toEqual({ ...optimistic, queuedAt: 100 });
expect(storage.getParts('msg-1')).toEqual(parts);
});

it('keeps the first queue time of a row on a repeated acknowledgement', () => {
const storage = createMemoryStorage();
const processor = createChatProcessor(storage);
const input = { messageId: 'msg-1', sessionId: 'ses-1', content: 'hello' };
processor.synthesizeQueuedUserMessage({ ...input, createdAt: 100 });
processor.synthesizeQueuedUserMessage({ ...input, createdAt: 500 });
expect(storage.getMessageInfo('msg-1')).toMatchObject({ queuedAt: 100, synthetic: true });
});

it('does not change a native row on queue acknowledgement', () => {
const storage = createMemoryStorage();
const processor = createChatProcessor(storage);
const native = makeUserMsg('msg-1');
storage.upsertMessage(native);
processor.synthesizeQueuedUserMessage({
messageId: 'msg-1',
sessionId: 'ses-1',
content: 'hello',
createdAt: 1,
});
expect(storage.getMessageInfo('msg-1')).toEqual(native);
});

it('inserts a synthetic user message with text part when storage is empty', () => {
const storage = createMemoryStorage();
const processor = createChatProcessor(storage);
Expand All @@ -997,6 +1067,7 @@ describe('createChatProcessor', () => {
const info = storage.getMessageInfo('msg-queued') as UserMessage;
expect(info?.role).toBe('user');
expect(info?.sessionID).toBe('ses-1');
expect(info?.synthetic).toBe(true);
const parts = storage.getParts('msg-queued');
expect(parts).toHaveLength(1);
const textPart = parts[0] satisfies Part as TextPart;
Expand All @@ -1019,6 +1090,7 @@ describe('createChatProcessor', () => {
messageId: 'msg-1',
sessionId: 'ses-1',
content: 'duplicate',
createdAt: 999,
});

expect(storage.getMessageIds()).toEqual(['msg-1']);
Expand Down
24 changes: 20 additions & 4 deletions packages/cloud-agent-sdk/src/chat-processor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,15 @@ type ChatProcessor = {
* Materialize a synthetic user message from a `cloud.message.queued` server
* event, unless an authoritative message with the same id already exists.
* The synthetic message is overwritten when the wrapper later emits the
* real `message.updated` payload for the same id.
* real `message.updated` payload for the same id. The server queue time is
* recorded as `queuedAt`, also on an existing optimistic row, so the row is
* ordered by server time until its native row arrives.
*/
synthesizeQueuedUserMessage(input: {
messageId: string;
sessionId: string;
content: string | undefined;
createdAt?: number | undefined;
}): void;
};

Expand Down Expand Up @@ -278,11 +281,23 @@ function createChatProcessor(
return true;
},

synthesizeQueuedUserMessage({ messageId, sessionId, content }) {
synthesizeQueuedUserMessage({ messageId, sessionId, content, createdAt }) {
const existing = sessionStorage.getMessageInfo(messageId);
if (existing) {
if (
createdAt !== undefined &&
existing.role === 'user' &&
existing.sessionID === sessionId &&
existing.synthetic === true &&
existing.queuedAt === undefined
) {
sessionStorage.upsertMessage({ ...existing, queuedAt: createdAt });
}
return;
}
// Empty or missing content can't form a renderable user message; wait for
// the authoritative `message.updated` payload from the wrapper instead.
if (!content) return;
if (sessionStorage.getMessageInfo(messageId)) return;

// Same unconfirmed marker as the optimistic send-time row: the
// authoritative `message.updated` for this id replaces the info (role and
Expand All @@ -293,10 +308,11 @@ function createChatProcessor(
id: messageId,
sessionID: sessionId,
role: 'user',
time: { created: Date.now() },
time: { created: createdAt ?? Date.now() },
agent: '',
model: { providerID: '', modelID: '' },
synthetic: true,
...(createdAt === undefined ? {} : { queuedAt: createdAt }),
};
sessionStorage.upsertMessage(syntheticMessage);

Expand Down
38 changes: 29 additions & 9 deletions packages/cloud-agent-sdk/src/normalizer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2358,30 +2358,50 @@ describe('normalize', () => {

describe('cloud.message.queued', () => {
it('normalizes with messageId, executionId, delivery', () => {
const result = normalize(
createRaw('cloud.message.queued', {
messageId: 'msg_1',
executionId: 'exe_1',
content: 'hi',
delivery: 'queued',
})
);
const raw = createRaw('cloud.message.queued', {
messageId: 'msg_1',
executionId: 'exe_1',
content: 'hi',
delivery: 'queued',
});
const result = normalize(raw);
expect(result).toEqual({
type: 'cloud.message.queued',
messageId: 'msg_1',
executionId: 'exe_1',
content: 'hi',
createdAt: Date.parse(raw.timestamp),
});
});

it('normalizes with only messageId', () => {
const result = normalize(createRaw('cloud.message.queued', { messageId: 'msg_2' }));
const raw = createRaw('cloud.message.queued', { messageId: 'msg_2' });
const result = normalize(raw);
expect(result).toEqual({
type: 'cloud.message.queued',
messageId: 'msg_2',
executionId: undefined,
content: undefined,
createdAt: Date.parse(raw.timestamp),
});
});

it('uses the persisted envelope time, not a timestamp supplied in content', () => {
const raw = createRaw('cloud.message.queued', { messageId: 'msg_1', createdAt: 999 });
expect(normalize({ ...raw, timestamp: '2026-09-01T00:00:00.000Z' })).toMatchObject({
createdAt: Date.parse('2026-09-01T00:00:00.000Z'),
});
});

it('keeps an unreadable timestamp unconfirmed without dropping the queued event', () => {
const raw = createRaw('cloud.message.queued', { messageId: 'msg_1', content: 'hi' });
const event = normalize({ ...raw, timestamp: 'unreadable' });
expect(event).toMatchObject({
type: 'cloud.message.queued',
messageId: 'msg_1',
content: 'hi',
});
expect(event).not.toHaveProperty('createdAt');
});

it('returns null when messageId is missing', () => {
Expand Down
5 changes: 5 additions & 0 deletions packages/cloud-agent-sdk/src/normalizer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,7 @@ export type ServiceEvent =
messageId: string;
executionId?: string | undefined;
content?: string | undefined;
createdAt?: number | undefined;
}
| {
type: 'cloud.message.canceled';
Expand Down Expand Up @@ -740,6 +741,10 @@ export function normalize(raw: CloudAgentEvent): NormalizedEvent | null {
if (raw.streamEventType === 'connected' && event?.type === 'connected') {
return { ...event, cloudSessionId: raw.sessionId };
}
if (event?.type === 'cloud.message.queued') {
const createdAt = Date.parse(raw.timestamp);
return Number.isFinite(createdAt) ? { ...event, createdAt } : event;
}
// Only a completion that carries a storable commit hash gets the envelope
// timestamp the commit record is anchored to. The schema already dropped a
// hash that could not be stored, so its presence is the whole test.
Expand Down
Loading
Loading