Skip to content

Commit eac1b96

Browse files
saltbosaltbo
committed
feat(chat): normalize session history for all runtimes
Move history normalization from browser to daemon. Each provider now exports a history reader that converts its native format to AgentEvent[]: - Claude: getClaudeHistory wraps getSessionMessages + mapSDKMessage - Codex: getCodexHistory reads ~/.codex/sessions/ JSONL by thread_id Wire format changes from { messages } to { events } — frontend receives pre-normalized RelayEvent[] and no longer parses provider-specific data. Gemini has no history API and returns empty for now.
1 parent be9115e commit eac1b96

12 files changed

Lines changed: 962 additions & 464 deletions

‎apps/web/src/hooks/useSessionRelay.ts‎

Lines changed: 4 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
import type { AgentEvent, ContentBlock } from "@agent-kanban/shared";
1+
import type { AgentEvent } from "@agent-kanban/shared";
22
import { useCallback, useEffect, useRef, useState } from "react";
33
import { getAuthToken } from "../lib/auth-client";
44

5-
export type { AgentEvent, ContentBlock };
5+
export type { AgentEvent };
66

77
export interface RelayEvent {
88
id: string;
@@ -15,69 +15,6 @@ interface UseSessionRelayOptions {
1515
enabled?: boolean;
1616
}
1717

18-
export function parseHistoryMessages(messages: any[]): RelayEvent[] {
19-
const events: RelayEvent[] = [];
20-
let counter = 0;
21-
for (const m of messages) {
22-
if (m.type === "assistant" && m.message?.content) {
23-
const blocks: ContentBlock[] = [];
24-
for (const block of m.message.content) {
25-
if (block.type === "thinking" && block.thinking) blocks.push({ type: "thinking", text: block.thinking });
26-
else if (block.type === "tool_use") blocks.push({ type: "tool_use", id: block.id, name: block.name, input: block.input });
27-
else if (block.type === "text" && block.text) blocks.push({ type: "text", text: block.text });
28-
}
29-
if (blocks.length > 0) {
30-
events.push({ id: m.uuid || `hist-${++counter}`, event: { type: "message", blocks }, timestamp: new Date().toISOString() });
31-
}
32-
} else if (m.type === "user" && m.message?.content != null) {
33-
// User messages from the Claude SDK come in two flavors:
34-
// - tool results (assistant attribution: blocks belong to a previous assistant turn)
35-
// - plain user input (human input: a real user message in the chat)
36-
// Plain text content can be either a string or an array of `{type:"text"}` blocks.
37-
const content = m.message.content;
38-
const toolBlocks: ContentBlock[] = [];
39-
const userTextParts: string[] = [];
40-
41-
if (typeof content === "string") {
42-
if (content.trim()) userTextParts.push(content);
43-
} else if (Array.isArray(content)) {
44-
for (const block of content) {
45-
if (block.type === "tool_result") {
46-
const output =
47-
typeof block.content === "string"
48-
? block.content
49-
: Array.isArray(block.content)
50-
? block.content
51-
.filter((c: any) => c.type === "text")
52-
.map((c: any) => c.text)
53-
.join("\n")
54-
: undefined;
55-
toolBlocks.push({ type: "tool_result", tool_use_id: block.tool_use_id, output, error: block.is_error });
56-
} else if (block.type === "text" && typeof block.text === "string" && block.text.trim()) {
57-
userTextParts.push(block.text);
58-
}
59-
}
60-
}
61-
62-
if (toolBlocks.length > 0) {
63-
events.push({
64-
id: m.uuid ? `${m.uuid}-tool` : `hist-${++counter}`,
65-
event: { type: "message", blocks: toolBlocks },
66-
timestamp: new Date().toISOString(),
67-
});
68-
}
69-
if (userTextParts.length > 0) {
70-
events.push({
71-
id: m.uuid ? `${m.uuid}-user` : `hist-${++counter}`,
72-
event: { type: "message.user", text: userTextParts.join("\n") },
73-
timestamp: new Date().toISOString(),
74-
});
75-
}
76-
}
77-
}
78-
return events;
79-
}
80-
8118
export type AgentStatus = "idle" | "working" | "done" | "rate_limited";
8219

8320
export function useSessionRelay({ sessionId, enabled = true }: UseSessionRelayOptions) {
@@ -137,9 +74,8 @@ export function useSessionRelay({ sessionId, enabled = true }: UseSessionRelayOp
13774
case "session:history": {
13875
historyLoaded.current = true;
13976
clearTimeout(historyRetryTimer.current);
140-
const messages = msg.messages as any[];
141-
if (Array.isArray(messages)) {
142-
const history = parseHistoryMessages(messages);
77+
const history = msg.events as RelayEvent[] | undefined;
78+
if (Array.isArray(history)) {
14379
setEvents((prev) => {
14480
const liveEvents = prev.filter((e) => e.id.startsWith("live-"));
14581
return [...history, ...liveEvents];

‎packages/cli/src/daemon/index.ts‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,16 @@
11
import { execSync } from "node:child_process";
22
import { mkdirSync, unlinkSync } from "node:fs";
33
import { arch, hostname, platform, release } from "node:os";
4-
import { getSessionMessages } from "@anthropic-ai/claude-agent-sdk";
54
import { MachineClient } from "../client/index.js";
65
import { getCredentials } from "../config.js";
76
import { generateDeviceId } from "../device.js";
87
import { createLogger } from "../logger.js";
98
import { PID_FILE, STATE_DIR } from "../paths.js";
9+
import { getClaudeHistory } from "../providers/claude.js";
10+
import { getCodexHistory } from "../providers/codex.js";
1011
import { getAvailableProviders } from "../providers/registry.js";
12+
import type { HistoryEvent } from "../providers/types.js";
13+
import { getSessionManager } from "../session/manager.js";
1114
import { migrateLegacySessions } from "../session/store.js";
1215
import { getVersion } from "../version.js";
1316
import { auditOrphanedTasks, cleanupLeaderSessions, cleanupStaleSessions } from "./cleanup.js";
@@ -20,6 +23,20 @@ import { UsageCollector } from "./usageCollector.js";
2023

2124
const logger = createLogger("daemon");
2225

26+
async function fetchSessionHistory(sessionId: string): Promise<HistoryEvent[]> {
27+
const session = getSessionManager().read(sessionId);
28+
if (!session) return [];
29+
30+
switch (session.runtime) {
31+
case "claude":
32+
return getClaudeHistory(sessionId);
33+
case "codex":
34+
return session.providerResumeToken ? getCodexHistory(session.providerResumeToken) : [];
35+
default:
36+
return [];
37+
}
38+
}
39+
2340
export interface DaemonOptions {
2441
maxConcurrent: number;
2542
pollInterval?: number;
@@ -88,8 +105,8 @@ export async function startDaemon(opts: DaemonOptions): Promise<void> {
88105
);
89106

90107
tunnel.onHistoryRequest((sessionId, requestId) => {
91-
getSessionMessages(sessionId)
92-
.then((messages) => tunnel.sendHistory(messages, requestId))
108+
fetchSessionHistory(sessionId)
109+
.then((events) => tunnel.sendHistory(events, requestId))
93110
.catch((e) => logger.warn(`History fetch failed for ${sessionId.slice(0, 8)}: ${e instanceof Error ? e.message : e}`));
94111
});
95112

‎packages/cli/src/daemon/tunnel.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -234,8 +234,8 @@ export class TunnelClient {
234234
this.send({ type: "agent:status", sessionId, status });
235235
}
236236

237-
sendHistory(messages: unknown[], requestId: string): void {
238-
this.send({ type: "session:history", messages, requestId });
237+
sendHistory(events: unknown[], requestId: string): void {
238+
this.send({ type: "session:history", events, requestId });
239239
}
240240

241241
onHumanMessage(handler: (sessionId: string, content: string) => void): void {

‎packages/cli/src/providers/claude.ts‎

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,9 @@ import { homedir, platform } from "node:os";
44
import { join } from "node:path";
55
import type { SubtaskStatus } from "@agent-kanban/shared";
66
import type { SDKAssistantMessage, SDKMessage, SDKPartialAssistantMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
7-
import { query } from "@anthropic-ai/claude-agent-sdk";
7+
import { getSessionMessages, query } from "@anthropic-ai/claude-agent-sdk";
88
import { createLogger } from "../logger.js";
9-
import type { AgentEvent, AgentHandle, AgentProvider, ContentBlock, ExecuteOpts, UsageInfo, UsageWindow } from "./types.js";
9+
import type { AgentEvent, AgentHandle, AgentProvider, ContentBlock, ExecuteOpts, HistoryEvent, UsageInfo, UsageWindow } from "./types.js";
1010
import { parseRetryAfterMs, UsageFetchError } from "./types.js";
1111

1212
const SUBTASK_STATUSES: readonly SubtaskStatus[] = ["completed", "failed", "stopped"] as const;
@@ -386,3 +386,22 @@ export const claudeProvider: AgentProvider = {
386386
return { windows, updated_at: new Date().toISOString() };
387387
},
388388
};
389+
390+
// ── History ──
391+
392+
/** Fetch Claude session history and normalize to AgentEvent stream. */
393+
export async function getClaudeHistory(sessionId: string): Promise<HistoryEvent[]> {
394+
const messages = await getSessionMessages(sessionId);
395+
const events: HistoryEvent[] = [];
396+
let counter = 0;
397+
for (const msg of messages) {
398+
// SessionMessage has { type, uuid, message, parent_tool_use_id }.
399+
// Reconstruct the shape mapSDKMessage expects.
400+
const sdkLike = { ...msg, message: msg.message } as unknown as SDKMessage;
401+
const event = mapSDKMessage(sdkLike);
402+
if (event) {
403+
events.push({ id: msg.uuid || `claude-hist-${++counter}`, event, timestamp: new Date().toISOString() });
404+
}
405+
}
406+
return events;
407+
}

‎packages/cli/src/providers/codex.ts‎

Lines changed: 98 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,13 @@
11
import { execSync } from "node:child_process";
2-
import { readFileSync } from "node:fs";
2+
import { readdirSync, readFileSync } from "node:fs";
33
import { homedir } from "node:os";
44
import { join } from "node:path";
55
import { Codex, type ThreadEvent } from "@openai/codex-sdk";
6-
import type { AgentEvent, AgentHandle, AgentProvider, ContentBlock, ExecuteOpts, UsageInfo, UsageWindow } from "./types.js";
6+
import type { AgentEvent, AgentHandle, AgentProvider, ContentBlock, ExecuteOpts, HistoryEvent, UsageInfo, UsageWindow } from "./types.js";
77
import { parseRetryAfterMs, UsageFetchError } from "./types.js";
88

99
const AUTH_PATH = join(homedir(), ".codex", "auth.json");
10+
const CODEX_SESSIONS_DIR = join(homedir(), ".codex", "sessions");
1011
const USAGE_API = "https://chatgpt.com/backend-api/wham/usage";
1112

1213
function readAccessToken(): string | null {
@@ -281,3 +282,98 @@ export const codexProvider: AgentProvider = {
281282
return { windows, updated_at: new Date().toISOString() };
282283
},
283284
};
285+
286+
// ── History from local JSONL ──
287+
288+
function findSessionFile(threadId: string): string | null {
289+
const suffix = `${threadId}.jsonl`;
290+
try {
291+
for (const year of readdirSync(CODEX_SESSIONS_DIR)) {
292+
const yearDir = join(CODEX_SESSIONS_DIR, year);
293+
for (const month of readdirSync(yearDir)) {
294+
const monthDir = join(yearDir, month);
295+
for (const day of readdirSync(monthDir)) {
296+
const dayDir = join(monthDir, day);
297+
for (const file of readdirSync(dayDir)) {
298+
if (file.endsWith(suffix)) return join(dayDir, file);
299+
}
300+
}
301+
}
302+
}
303+
} catch {
304+
/* dir missing */
305+
}
306+
return null;
307+
}
308+
309+
function mapResponseItem(payload: Record<string, any>): AgentEvent | null {
310+
switch (payload.type) {
311+
case "message": {
312+
if (payload.role === "assistant") {
313+
const texts = (payload.content ?? []).filter((c: any) => c.type === "output_text" && c.text).map((c: any) => c.text);
314+
if (texts.length > 0) {
315+
return { type: "message", blocks: [{ type: "text", text: texts.join("\n") }] };
316+
}
317+
}
318+
if (payload.role === "user") {
319+
const texts = (payload.content ?? []).filter((c: any) => c.type === "input_text" && c.text).map((c: any) => c.text);
320+
if (texts.length > 0) return { type: "message.user", text: texts.join("\n") };
321+
}
322+
return null;
323+
}
324+
case "function_call": {
325+
let input: Record<string, unknown> = {};
326+
if (payload.arguments) {
327+
try {
328+
input = JSON.parse(payload.arguments);
329+
} catch {
330+
input = { raw: payload.arguments };
331+
}
332+
}
333+
return {
334+
type: "message",
335+
blocks: [{ type: "tool_use", id: payload.call_id ?? `codex-hist-${Date.now()}`, name: payload.name ?? "tool", input }],
336+
};
337+
}
338+
case "function_call_output":
339+
return {
340+
type: "message",
341+
blocks: [{ type: "tool_result", tool_use_id: payload.call_id ?? "", output: payload.output }],
342+
};
343+
default:
344+
return null;
345+
}
346+
}
347+
348+
/** Read Codex session history from local JSONL files. */
349+
export function getCodexHistory(threadId: string): HistoryEvent[] {
350+
const file = findSessionFile(threadId);
351+
if (!file) return [];
352+
353+
const lines = readFileSync(file, "utf-8").split("\n");
354+
const events: HistoryEvent[] = [];
355+
let counter = 0;
356+
357+
for (const line of lines) {
358+
if (!line.trim()) continue;
359+
let row: any;
360+
try {
361+
row = JSON.parse(line);
362+
} catch {
363+
continue;
364+
}
365+
if (row.type !== "response_item") continue;
366+
// Skip developer/system messages
367+
if (row.payload?.role === "developer") continue;
368+
369+
const event = mapResponseItem(row.payload);
370+
if (event) {
371+
events.push({
372+
id: `codex-hist-${++counter}`,
373+
event,
374+
timestamp: row.timestamp ?? new Date().toISOString(),
375+
});
376+
}
377+
}
378+
return events;
379+
}

‎packages/cli/src/providers/types.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,13 @@ import type { AgentEvent, AgentRuntime, ContentBlock, UsageInfo, UsageWindow } f
22

33
export type { AgentEvent, AgentRuntime, ContentBlock, UsageInfo, UsageWindow };
44

5+
/** Normalized history entry returned by provider history readers. */
6+
export interface HistoryEvent {
7+
id: string;
8+
event: AgentEvent;
9+
timestamp: string;
10+
}
11+
512
export interface ExecuteOpts {
613
sessionId: string;
714
resumeToken?: string;

‎packages/cli/tests/tunnelClient.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ describe("TunnelClient — sendHistory()", () => {
274274
const messages = [{ role: "user", content: "hello" }];
275275
client.sendHistory(messages, "req-1");
276276
const msg = JSON.parse(lastCreatedWs!.sentMessages[0]);
277-
expect(msg.messages).toEqual(messages);
277+
expect(msg.events).toEqual(messages);
278278
});
279279
});
280280

0 commit comments

Comments
 (0)