fix(cli): restore Pi session resume (#1206)

* fix(cli): restore Pi session resume

Use Pi's supported --session flag and keep the session initialized by the CLI instead of replacing it with a racing new_session RPC.

* test(cli): cover fresh Pi startup

cli/src/pi/runPi.test.ts

@@ -0,0 +1,112 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+type TransportOptions = { command: string; args: string[]; cwd: string };
+type LifecycleOptions = { stopKeepAlive: () => void };
+
+const harness = vi.hoisted(() => ({
+    transportOptions: null as TransportOptions | null,
+    sent: [] as unknown[],
+    session: {
+        keepAlive: vi.fn(),
+        onUserMessage: vi.fn(),
+        onCancelQueuedMessage: vi.fn(),
+        rpcHandlerManager: { registerHandler: vi.fn() },
+    },
+}));
+
+vi.mock('@/agent/sessionFactory', () => ({
+    bootstrapSession: vi.fn(async () => ({ api: {}, session: harness.session })),
+    bootstrapExistingSession: vi.fn(async () => ({ api: {}, session: harness.session })),
+}));
+
+vi.mock('@/agent/runnerLifecycle', () => ({
+    createRunnerLifecycle: vi.fn((options: LifecycleOptions) => {
+        return {
+            registerProcessHandlers: vi.fn(),
+            cleanupAndExit: vi.fn(async () => {
+                options.stopKeepAlive();
+            }),
+            markCrash: vi.fn(),
+            setExitCode: vi.fn(),
+            setArchiveReason: vi.fn(),
+            setSessionEndReason: vi.fn(),
+            hasExplicitSessionEndReason: vi.fn(() => true),
+        };
+    }),
+    createModeChangeHandler: vi.fn(() => vi.fn()),
+    setControlledByUser: vi.fn(),
+}));
+
+vi.mock('@/ui/logger', () => ({
+    logger: {
+        debug: vi.fn(),
+        getLogPath: vi.fn(() => '/tmp/hapi.log'),
+    },
+}));
+
+vi.mock('./piTransport', () => ({
+    PiTransport: class {
+        constructor(options: TransportOptions) {
+            harness.transportOptions = options;
+        }
+
+        onError(): void {}
+
+        onClose(): void {}
+
+        onEvent(): void {}
+
+        start(): void {}
+
+        send(command: unknown): void {
+            harness.sent.push(command);
+            if ((command as { type?: string }).type === 'get_commands') {
+                throw new Error('stop test transport');
+            }
+        }
+
+        kill(): void {}
+    },
+}));
+
+import { runPi } from './runPi';
+
+describe('runPi startup', () => {
+    beforeEach(() => {
+        harness.transportOptions = null;
+        harness.sent.length = 0;
+    });
+
+    it('lets Pi create a fresh session when no resume ID is provided', async () => {
+        await runPi({ workingDirectory: '/work' });
+
+        expect(harness.transportOptions).toEqual({
+            command: 'pi',
+            args: ['--mode', 'rpc'],
+            cwd: '/work',
+        });
+        expect(harness.sent).toEqual([
+            { type: 'get_state' },
+            { type: 'get_available_models' },
+            { type: 'get_commands' },
+        ]);
+    });
+
+    it('resumes with --session and keeps the session selected by Pi', async () => {
+        await runPi({
+            workingDirectory: '/work',
+            resumeSessionId: 'pi-session-123',
+        });
+
+        expect(harness.transportOptions).toEqual({
+            command: 'pi',
+            args: ['--mode', 'rpc', '--session', 'pi-session-123'],
+            cwd: '/work',
+        });
+        expect(harness.sent).toEqual([
+            { type: 'get_state' },
+            { type: 'get_available_models' },
+            { type: 'get_commands' },
+        ]);
+    });
+});

cli/src/pi/runPi.ts

@@ -74,7 +74,7 @@ export async function runPi(opts: {
 
     const transportArgs = ['--mode', 'rpc'];
     if (opts.resumeSessionId) {
-        transportArgs.push('--session-id', opts.resumeSessionId);
+        transportArgs.push('--session', opts.resumeSessionId);
     }
     const transport = new PiTransport({ command: 'pi', args: transportArgs, cwd: workingDirectory });
 
@@ -319,7 +319,7 @@ export async function runPi(opts: {
         const formattedText = formatMessageWithAttachments(message.content.text, message.content.attachments);
         // Gate the send behind Pi startup readiness. A prompt POSTed immediately
         // after spawn (supported handoff pattern — hapi-ping-peer, intake
-        // scripts) would otherwise reach Pi before new_session/get_state finish
+        // scripts) would otherwise reach Pi before its initial get_state finishes
         // and wedge the turn. runWhenReady delivers now if ready, else buffers
         // FIFO until the first get_state response drains it (issue #1143).
         // piIsStreaming is evaluated at delivery time so a message buffered
@@ -388,7 +388,8 @@ export async function runPi(opts: {
     readyFallback.unref?.();
     try {
         transport.start();
-        transport.send({ type: 'new_session' });
+        // Pi creates a fresh session (or resumes --session) before RPC mode
+        // starts. Sending new_session here races get_state and discards a resumed session.
         transport.send({ type: 'get_state' });
         transport.send({ type: 'get_available_models' });
         transport.send({ type: 'get_commands' });

cli/src/pi/session.test.ts

@@ -29,7 +29,7 @@ function createMockSession(): PiSession {
 // --- Ready gate + outbound buffer (Pi RPC ready-race, issue #1143) ---
 //
 // A prompt POSTed immediately after spawn used to be sent to Pi before
-// `new_session`/`get_state` finished, wedging the turn (agent_start then
+// Pi returned its initial `get_state`, wedging the turn (agent_start then
 // silence). runWhenReady buffers such sends until markReady() (fired when Pi's
 // get_state response lands), then drains them FIFO.
 

cli/src/pi/session.ts

@@ -47,7 +47,7 @@ export class PiSession {
     rpcResolver: PiRpcResolver | null = null;
 
     // Startup ready gate (issue #1143). Pi's socket goes `active` (spawn success)
-    // before `pi --mode rpc` finishes `new_session`/`get_state`, so a prompt sent
+    // before `pi --mode rpc` returns its initial `get_state`, so a prompt sent
     // in that window reaches Pi before its session is initialized and wedges
     // (agent_start, then silence). Outbound sends that assume a live Pi session
     // are queued via runWhenReady() and drained FIFO once markReady() fires (on