diff --git a/packages/happy-cli/src/api/apiSession.test.ts b/packages/happy-cli/src/api/apiSession.test.ts index 95ce5e39ab..baff95107b 100644 --- a/packages/happy-cli/src/api/apiSession.test.ts +++ b/packages/happy-cli/src/api/apiSession.test.ts @@ -865,6 +865,41 @@ describe('ApiSessionClient v3 messages API migration', () => { expect((client as any).lastReceivedSeq).toBe(1); }); + it('on reconnect, skips existing messages but records what they hold of the Claude conversation', async () => { + const client = new ApiSessionClient('fake-token', session); + const onUserMessage = vi.fn(); + client.onUserMessage(onUserMessage); + expect(client.syncedTranscript()).toBeNull(); + client.skipExistingMessages(); + + const history = [ + { role: 'user', content: { type: 'text', text: 'hello' } }, + { role: 'session', content: { role: 'agent', claudeUuid: 'a1', ev: { t: 'text', text: 'Hi!' } } }, + { role: 'user', content: { type: 'text', text: 'no reply yet' } }, + ]; + mockAxiosGet.mockResolvedValueOnce({ + data: { + messages: history.map((body, i) => ({ + id: `msg-${i + 1}`, + seq: i + 1, + content: { t: 'encrypted', c: encryptContent(session, body) }, + localId: null, + createdAt: 1000, + updatedAt: 1000, + })), + hasMore: false, + }, + }); + + await (client as any).fetchMessages(); + + expect(onUserMessage).not.toHaveBeenCalled(); + expect((client as any).lastReceivedSeq).toBe(3); + const synced = await client.syncedTranscript(); + expect([...synced!.claudeUuids]).toEqual(['a1']); + expect(synced!.trailingUserTexts).toEqual(['no reply yet']); + }); + it('fetchMessages uses incremental cursor and paginates while hasMore is true', async () => { const client = new ApiSessionClient('fake-token', session); const onUserMessage = vi.fn(); diff --git a/packages/happy-cli/src/api/apiSession.ts b/packages/happy-cli/src/api/apiSession.ts index ad8f75074f..a8a3b9ba1a 100644 --- a/packages/happy-cli/src/api/apiSession.ts +++ b/packages/happy-cli/src/api/apiSession.ts @@ -24,6 +24,8 @@ import { type ClaudeSessionProtocolState, } from '@/claude/utils/sessionProtocolMapper'; import { InvalidateSync } from '@/utils/sync'; +import { Future } from '@/utils/future'; +import { collectSyncedTranscript, type SyncedTranscript } from '@/claude/utils/transcriptResync'; import axios from 'axios'; /** @@ -219,6 +221,8 @@ export class ApiSessionClient extends EventEmitter { private closed = false; private ignoreArchiveSignal = false; private skipInitialMessages = false; + /** Set on reconnect: what the skipped history holds of the Claude conversation. */ + private syncedTranscriptFuture: Future | null = null; private claudeSessionProtocolState: ClaudeSessionProtocolState = { currentTurnId: null, uuidToProviderSubagent: new Map(), @@ -609,6 +613,7 @@ export class ApiSessionClient extends EventEmitter { logger.debug('[API] Reconnect mode: skipping existing messages, advancing lastReceivedSeq'); } + const skippedBodies: unknown[] = []; let afterSeq = this.lastReceivedSeq; while (true) { const response = await axios.get( @@ -631,14 +636,16 @@ export class ApiSessionClient extends EventEmitter { maxSeq = message.seq; } - if (skipRouting) continue; - if (message.content?.t !== 'encrypted') { continue; } try { const body = decrypt(this.encryptionKey, this.encryptionVariant, decodeBase64(message.content.c)); + if (skipRouting) { + skippedBodies.push(body); + continue; + } this.routeIncomingMessage(body); } catch (error) { logger.debug('[API] Failed to decrypt fetched message', { @@ -663,6 +670,9 @@ export class ApiSessionClient extends EventEmitter { break; } } + if (skipRouting) { + this.syncedTranscriptFuture?.resolve(collectSyncedTranscript(skippedBodies)); + } } private static readonly MAX_OUTBOX_BATCH_SIZE = 50; @@ -942,6 +952,15 @@ export class ApiSessionClient extends EventEmitter { skipExistingMessages() { this.skipInitialMessages = true; + this.syncedTranscriptFuture = new Future(); + } + + /** + * On reconnect, what the session history already holds of the Claude conversation, read while the + * existing messages are skipped; `null` when this client is not reattaching to a session. + */ + syncedTranscript(): Promise | null { + return this.syncedTranscriptFuture?.promise ?? null; } updateMetadata(handler: (metadata: Metadata) => Metadata) { diff --git a/packages/happy-cli/src/claude/runClaude.ts b/packages/happy-cli/src/claude/runClaude.ts index 65fe0d42a9..5eda05bcfc 100644 --- a/packages/happy-cli/src/claude/runClaude.ts +++ b/packages/happy-cli/src/claude/runClaude.ts @@ -39,6 +39,7 @@ import { getProjectPath } from './utils/path'; import { readFile } from 'node:fs/promises'; import { join } from 'node:path'; import { RawJSONLinesSchema, type RawJSONLines } from './types'; +import { backfillUnsyncedTranscript } from './utils/transcriptResync'; /** JavaScript runtime to use for spawning Claude Code */ export type JsRuntime = 'node' | 'bun' @@ -292,6 +293,20 @@ export async function runClaude(credentials: Credentials, options: StartOptions lifecycleState: 'running', archivedBy: undefined, })); + + // Reconnect backfill: the conversation may have continued outside Happy (plain + // `claude --resume`) while this session was not running. Send what the session history + // lacks before the scanners start, since they treat what is on disk as already sent. + const resumeFlagIndex = options.claudeArgs?.findIndex((arg) => arg === '--resume' || arg === '-r') ?? -1; + const resumedClaudeSessionId = resumeFlagIndex >= 0 ? options.claudeArgs?.[resumeFlagIndex + 1] : undefined; + if (resumedClaudeSessionId && !resumedClaudeSessionId.startsWith('-')) { + await backfillUnsyncedTranscript({ + synced: session.syncedTranscript(), + workingDirectory, + claudeSessionId: resumedClaudeSessionId, + send: (message) => session.sendClaudeSessionMessageFromLocalTranscript(message), + }); + } } // Fork backfill: when this Happy session was just spawned as a fork diff --git a/packages/happy-cli/src/claude/utils/sdkToLogConverter.test.ts b/packages/happy-cli/src/claude/utils/sdkToLogConverter.test.ts index 3097ee8102..d82e16f79a 100644 --- a/packages/happy-cli/src/claude/utils/sdkToLogConverter.test.ts +++ b/packages/happy-cli/src/claude/utils/sdkToLogConverter.test.ts @@ -19,6 +19,27 @@ describe('SDKToLogConverter', () => { converter = new SDKToLogConverter(context) }) + describe('uuid', () => { + it('keeps the SDK uuid, which is the entry uuid in the Claude JSONL', () => { + const sdkMessage = { + type: 'assistant', + uuid: '50ee533f-f7c0-4823-91ac-5ce43f9dec3e', + parent_tool_use_id: null, + message: { role: 'assistant', content: [{ type: 'text', text: 'Hi' }] }, + } as unknown as SDKAssistantMessage + expect(converter.convert(sdkMessage)?.uuid).toBe('50ee533f-f7c0-4823-91ac-5ce43f9dec3e') + }) + + it('generates one when the SDK message has none', () => { + const sdkMessage = { + type: 'assistant', + parent_tool_use_id: null, + message: { role: 'assistant', content: [{ type: 'text', text: 'Hi' }] }, + } as unknown as SDKAssistantMessage + expect(converter.convert(sdkMessage)?.uuid).toMatch(/^[0-9a-f-]{36}$/) + }) + }) + describe('User messages', () => { it('should convert SDK user message to log format', () => { const sdkMessage: SDKUserMessage = { diff --git a/packages/happy-cli/src/claude/utils/sdkToLogConverter.ts b/packages/happy-cli/src/claude/utils/sdkToLogConverter.ts index dfa185e0a7..8adda129b4 100644 --- a/packages/happy-cli/src/claude/utils/sdkToLogConverter.ts +++ b/packages/happy-cli/src/claude/utils/sdkToLogConverter.ts @@ -111,7 +111,10 @@ export class SDKToLogConverter { * Convert SDK message to log format */ convert(sdkMessage: SDKMessage): RawJSONLines | null { - const uuid = randomUUID() + // Keep the SDK's uuid: it is the entry's uuid in Claude's JSONL, which claudeUuid must match + // (rewind/fork read it from disk; resume matches the session history against the file). + const sdkUuid = (sdkMessage as { uuid?: unknown }).uuid + const uuid = typeof sdkUuid === 'string' && sdkUuid.length > 0 ? sdkUuid : randomUUID() const timestamp = new Date().toISOString() let parentUuid = this.lastUuid; let isSidechain = false; diff --git a/packages/happy-cli/src/claude/utils/transcriptResync.test.ts b/packages/happy-cli/src/claude/utils/transcriptResync.test.ts new file mode 100644 index 0000000000..f7c6573c64 --- /dev/null +++ b/packages/happy-cli/src/claude/utils/transcriptResync.test.ts @@ -0,0 +1,99 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { backfillUnsyncedTranscript, collectSyncedTranscript, unsyncedEntryIndexes } from './transcriptResync'; +import { getProjectPath } from './path'; + +const agent = (claudeUuid: string, text: string) => ({ role: 'session', content: { role: 'agent', claudeUuid, ev: { t: 'text', text } } }); +const appUser = (text: string) => ({ role: 'user', content: { type: 'text', text } }); +const user = (uuid: string, text: string) => ({ type: 'user', uuid, message: { content: text } }); +const assistant = (uuid: string) => ({ type: 'assistant', uuid, message: { content: [] } }); + +describe('collectSyncedTranscript', () => { + it('collects claudeUuids and the app-typed user texts after the last one', () => { + const synced = collectSyncedTranscript([appUser('hello'), agent('a1', 'Hi!'), appUser('again'), appUser('more')]); + expect([...synced.claudeUuids]).toEqual(['a1']); + expect(synced.trailingUserTexts).toEqual(['again', 'more']); + }); + + it('ignores bodies it does not understand', () => { + expect(collectSyncedTranscript([null, 'x', { role: 'session', content: { ev: { t: 'turn-start' } } }])).toEqual({ + claudeUuids: new Set(), + trailingUserTexts: [], + }); + }); +}); + +describe('unsyncedEntryIndexes', () => { + const synced = collectSyncedTranscript([appUser('hello'), agent('a1', 'Hi!')]); + + it('returns the entries added after the last one the history knows', () => { + const entries = [user('u1', 'hello'), assistant('a1'), user('u2', 'outside'), assistant('a2')]; + expect(unsyncedEntryIndexes(entries, synced)).toEqual([2, 3]); + }); + + it('returns nothing when nothing was added', () => { + expect(unsyncedEntryIndexes([user('u1', 'hello'), assistant('a1')], synced)).toEqual([]); + }); + + it('does not resend an app-typed message that never got a reply', () => { + const pending = collectSyncedTranscript([appUser('hello'), agent('a1', 'Hi!'), appUser('still there?')]); + const entries = [user('u1', 'hello'), assistant('a1'), user('u2', 'still there?'), user('u3', 'outside')]; + expect(unsyncedEntryIndexes(entries, pending)).toEqual([3]); + }); + + it('returns null when the history knows none of the entries', () => { + expect(unsyncedEntryIndexes([user('x1', 'other'), assistant('x2')], synced)).toBeNull(); + }); +}); + +describe('backfillUnsyncedTranscript', () => { + let dir: string; + let previousConfigDir: string | undefined; + const entry = (type: 'user' | 'assistant', uuid: string, text: string) => JSON.stringify(type === 'user' + ? { type, uuid, sessionId: 'c1', message: { role: 'user', content: text }, timestamp: new Date().toISOString() } + : { type, uuid, sessionId: 'c1', message: { role: 'assistant', content: [{ type: 'text', text }] }, timestamp: new Date().toISOString() }); + + beforeEach(async () => { + dir = join(tmpdir(), `backfill-${Date.now()}-${Math.random().toString(36).slice(2)}`); + previousConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = join(dir, 'config'); + await mkdir(getProjectPath(join(dir, 'work')), { recursive: true }); + await writeFile(join(getProjectPath(join(dir, 'work')), 'c1.jsonl'), [ + entry('user', 'u1', 'hello'), // sent to the app before happy stopped + entry('assistant', 'a1', 'Hi!'), // sent to the app before happy stopped + entry('user', 'u2', 'outside message'), // added later with plain `claude --resume c1` + entry('assistant', 'a2', 'reply to outside'), // added later with plain `claude --resume c1` + ].join('\n') + '\n'); + }); + + afterEach(async () => { + if (previousConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR; + else process.env.CLAUDE_CONFIG_DIR = previousConfigDir; + await rm(dir, { recursive: true, force: true }); + }); + + const run = async (synced: Promise> | null) => { + const sent: string[] = []; + const count = await backfillUnsyncedTranscript({ + synced, + workingDirectory: join(dir, 'work'), + claudeSessionId: 'c1', + send: async (message) => { sent.push((message as { uuid: string }).uuid); }, + timeoutMs: 200, + }); + return { count, sent }; + }; + + it('sends the messages added while happy was not running (slopus/happy#1861)', async () => { + const synced = collectSyncedTranscript([appUser('hello'), agent('a1', 'Hi!')]); + expect(await run(Promise.resolve(synced))).toEqual({ count: 2, sent: ['u2', 'a2'] }); + }); + + it('sends nothing when not reattaching, or when the history cannot be read', async () => { + expect(await run(null)).toEqual({ count: 0, sent: [] }); + expect(await run(Promise.reject(new Error('offline')))).toEqual({ count: 0, sent: [] }); + expect(await run(new Promise(() => {}))).toEqual({ count: 0, sent: [] }); + }); +}); diff --git a/packages/happy-cli/src/claude/utils/transcriptResync.ts b/packages/happy-cli/src/claude/utils/transcriptResync.ts new file mode 100644 index 0000000000..1865c138ae --- /dev/null +++ b/packages/happy-cli/src/claude/utils/transcriptResync.ts @@ -0,0 +1,127 @@ +/** + * Resuming a Happy session on an existing Claude conversation: which JSONL entries does the session + * history not have yet, and sending them (reconnect backfill). + * + * The conversation may have continued outside Happy (plain `claude --resume`) while Happy was not + * running. Those entries are in Claude's JSONL but were never sent. The session history records the + * Claude uuid of what was sent (`claudeUuid` on session-protocol envelopes), so everything after the + * last JSONL entry the history knows is new. User messages typed in the app carry no claudeUuid; the + * ones after the last known entry are matched by text so they are not sent twice. + */ + +import { readFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import { logger } from '@/ui/logger'; +import { RawJSONLinesSchema, type RawJSONLines } from '../types'; +import { getProjectPath } from './path'; + +/** What the session history already holds of the Claude conversation. */ +export type SyncedTranscript = { + claudeUuids: Set; + /** Texts of app-typed user messages after the last message with a claudeUuid, oldest first. */ + trailingUserTexts: string[]; +}; + +/** Builds a SyncedTranscript from decrypted session history bodies, in history (seq) order. */ +export function collectSyncedTranscript(bodies: unknown[]): SyncedTranscript { + const claudeUuids = new Set(); + let trailingUserTexts: string[] = []; + for (const body of bodies) { + const record = body as { role?: unknown; content?: { claudeUuid?: unknown; type?: unknown; text?: unknown } } | null; + if (record?.role === 'session' && typeof record.content?.claudeUuid === 'string') { + claudeUuids.add(record.content.claudeUuid); + trailingUserTexts = []; + } else if (record?.role === 'user' && record.content?.type === 'text' && typeof record.content.text === 'string') { + trailingUserTexts.push(record.content.text); + } + } + return { claudeUuids, trailingUserTexts }; +} + +type TranscriptEntry = { type: string; uuid?: string; message?: unknown }; + +/** + * Indexes of the entries to send: those after the last entry whose uuid the history knows, minus + * user entries already in the history as app-typed messages. `null` when the history knows none of + * the entries (e.g. a rewritten conversation): callers then treat the file as already sent. + */ +export function unsyncedEntryIndexes(entries: TranscriptEntry[], synced: SyncedTranscript): number[] | null { + let last = -1; + entries.forEach((entry, index) => { + if (entry.uuid && synced.claudeUuids.has(entry.uuid)) last = index; + }); + if (last === -1) return null; + + const pendingAppTexts = [...synced.trailingUserTexts]; + const indexes: number[] = []; + for (let index = last + 1; index < entries.length; index++) { + const entry = entries[index]; + const text = (entry.message as { content?: unknown } | undefined)?.content; + if (entry.type === 'user' && typeof text === 'string' && pendingAppTexts[0] === text) { + pendingAppTexts.shift(); + continue; + } + indexes.push(index); + } + return indexes; +} + +/** How long to wait for the session history before giving up on the backfill. */ +const SYNCED_TRANSCRIPT_TIMEOUT_MS = 10_000; + +/** + * Reconnect backfill: sends the entries of the Claude conversation that the session history lacks. + * Run once when reattaching to a Happy session, before the scanners start (they treat what is on disk + * as already sent). Returns the number of entries sent. + */ +export async function backfillUnsyncedTranscript(opts: { + synced: Promise | null; + workingDirectory: string; + claudeSessionId: string; + send: (message: RawJSONLines) => Promise; + timeoutMs?: number; +}): Promise { + if (!opts.synced) return 0; + const synced = await withTimeout(opts.synced, opts.timeoutMs ?? SYNCED_TRANSCRIPT_TIMEOUT_MS); + if (!synced) { + logger.debug('[RECONNECT BACKFILL] Session history not available; nothing sent'); + return 0; + } + const jsonlPath = join(getProjectPath(opts.workingDirectory), `${opts.claudeSessionId}.jsonl`); + let file: string; + try { + file = await readFile(jsonlPath, 'utf-8'); + } catch (error) { + logger.debug(`[RECONNECT BACKFILL] Failed to read ${jsonlPath}:`, error); + return 0; + } + const entries: RawJSONLines[] = []; + for (const line of file.split('\n')) { + if (line.trim().length === 0) continue; + let parsed: unknown; + try { parsed = JSON.parse(line); } catch { continue; } + const result = RawJSONLinesSchema.safeParse(parsed); + if (result.success && result.data.type !== 'summary') entries.push(result.data); + } + const indexes = unsyncedEntryIndexes(entries, synced); + if (!indexes) { + logger.debug(`[RECONNECT BACKFILL] The session history knows none of the ${entries.length} entries in ${jsonlPath}; nothing sent`); + return 0; + } + for (const index of indexes) { + await opts.send(entries[index]); + } + logger.debug(`[RECONNECT BACKFILL] Sent ${indexes.length} of ${entries.length} entries missing from the session history`); + return indexes.length; +} + +/** Resolves to the promise's value, or `null` if it rejects or does not settle within `ms`. */ +async function withTimeout(promise: Promise, ms: number): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { timer = setTimeout(() => resolve(null), ms); }); + try { + return await Promise.race([promise.catch(() => null), timeout]); + } finally { + clearTimeout(timer); + } +}