fix(cli): send messages added outside happy when resuming a session

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();

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<SyncedTranscript> | null = null;
     private claudeSessionProtocolState: ClaudeSessionProtocolState = {
         currentTurnId: null,
         uuidToProviderSubagent: new Map<string, string>(),
@@ -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<V3GetSessionMessagesResponse>(
@@ -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<SyncedTranscript>();
+    }
+
+    /**
+     * 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<SyncedTranscript> | null {
+        return this.syncedTranscriptFuture?.promise ?? null;
     }
 
     updateMetadata(handler: (metadata: Metadata) => Metadata) {

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

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 = {

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;

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<ReturnType<typeof collectSyncedTranscript>> | 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: [] });
+    });
+});