feat(pty): forward user messages to local-mode Claude stdin

- Pipe stdin for local Claude processes so chat messages go directly
  to the running session instead of triggering a mode switch
- Add stdin message dedup filter in local launcher scanner callback
  (guards against user messages without a `message` payload)
- Restore lost messages to queue on claudeRemote launch failure
- Export extractRawUserTextContent for reuse in dedup logic

Author: junmo-kim

cli/src/api/apiSession.ts

@@ -55,7 +55,7 @@ const SYSTEM_INJECTION_PREFIXES = [
 // hub's scrollback ring). The tail always holds the latest full-screen redraw.
 const AGENT_TERMINAL_LOCAL_BUFFER_BYTES = 256 * 1024
 
-function extractRawUserTextContent(content: unknown): string | null {
+export function extractRawUserTextContent(content: unknown): string | null {
     if (typeof content === 'string') {
         return content
     }

cli/src/claude/claudeLocal.ts

@@ -19,6 +19,8 @@ export async function claudeLocal(opts: {
     claudeArgs?: string[]
     allowedTools?: string[]
     hookSettingsPath: string
+    /** Called when the child process's stdin is ready for writing. */
+    onStdinReady?: (write: (data: string) => void) => void
 }) {
 
     // Ensure project directory exists
@@ -95,7 +97,8 @@ export async function claudeLocal(opts: {
     const claudeCommand = getDefaultClaudeCodePath();
     logger.debug(`[ClaudeLocal] Using claude executable: ${claudeCommand}`);
 
-    // Spawn the process
+    // Spawn the process with pipe stdin so chat messages can be forwarded
+    // to the running Claude process instead of triggering a mode switch.
     try {
         await spawnWithTerminalGuard({
             command: claudeCommand,
@@ -108,7 +111,9 @@ export async function claudeLocal(opts: {
             installHint: 'Claude CLI',
             includeCause: true,
             logExit: true,
-            shell: false  // Use absolute path, no shell needed
+            shell: false,  // Use absolute path, no shell needed
+            stdio: ['pipe', 'inherit', 'inherit'],
+            onSpawned: opts.onStdinReady
         });
     } finally {
         cleanupMcpConfig?.();

cli/src/claude/claudeLocalLauncher.test.ts

@@ -54,7 +54,8 @@ function createSessionStub() {
             addSessionFoundCallback: () => {},
             removeSessionFoundCallback: () => {},
             consumeOneTimeFlags: () => {},
-            recordLocalLaunchFailure: () => {}
+            recordLocalLaunchFailure: () => {},
+            stdinMessageTexts: new Set<string>()
         },
         sentMessages
     }
@@ -133,4 +134,23 @@ describe('claudeLocalLauncher message filtering', () => {
 
         expect(sentMessages).toHaveLength(0)
     })
+
+    it('swallows exactly one stdin echo per forward, so a repeated identical message still surfaces', async () => {
+        const { session, sentMessages } = createSessionStub()
+        await claudeLocalLauncher(session as never)
+
+        // One "continue" was forwarded to claude via stdin (marked for dedup).
+        session.stdinMessageTexts.add('continue')
+        // Its JSONL echo is swallowed (already shown in chat via the web path)...
+        harness.scannerOnMessage!({ type: 'user', uuid: '1', message: { content: 'continue' } })
+        expect(sentMessages).toHaveLength(0)
+        // ...and the dedup entry is consumed on match, bounding the set.
+        expect(session.stdinMessageTexts.size).toBe(0)
+
+        // The user sends "continue" again; this echo has no pending forward to
+        // swallow it, so it must surface (regression: a value-keyed, never-cleared
+        // set would have silently dropped it).
+        harness.scannerOnMessage!({ type: 'user', uuid: '2', message: { content: 'continue' } })
+        expect(sentMessages).toHaveLength(1)
+    })
 })

cli/src/claude/claudeLocalLauncher.ts

@@ -3,6 +3,7 @@ import { Session } from "./session";
 import { createSessionScanner } from "./utils/sessionScanner";
 import { isClaudeChatVisibleMessage } from "./utils/chatVisibility";
 import { BaseLocalLauncher } from "@/modules/common/launcher/BaseLocalLauncher";
+import { extractRawUserTextContent } from "@/api/apiSession";
 
 export async function claudeLocalLauncher(session: Session): Promise<'switch' | 'exit'> {
 
@@ -25,6 +26,19 @@ export async function claudeLocalLauncher(session: Session): Promise<'switch' |
             if (!isClaudeChatVisibleMessage(message)) {
                 return
             }
+            // Skip the JSONL echo of a user message we already forwarded to the
+            // local process via stdin (it is already in the hub as a consumed
+            // message from the web chat). Without this the same user text would
+            // appear twice in the chat UI — once from the web path and once from
+            // the JSONL transcript. Swallow exactly ONE echo per forward by
+            // deleting on match: this both bounds the set and lets a later,
+            // identical message ("yes", "continue", ...) surface normally.
+            if (message.type === 'user') {
+                const text = extractRawUserTextContent(message.message?.content)
+                if (text && session.stdinMessageTexts.delete(text)) {
+                    return
+                }
+            }
             session.client.sendClaudeSessionMessage(message)
         }
     });
@@ -43,16 +57,28 @@ export async function claudeLocalLauncher(session: Session): Promise<'switch' |
         startedBy: session.startedBy,
         startingMode: session.startingMode,
         launch: async (abortSignal) => {
-            await claudeLocal({
-                path: session.path,
-                sessionId: session.sessionId,
-                abort: abortSignal,
-                claudeEnvVars: session.claudeEnvVars,
-                claudeArgs: session.claudeArgs,
-                mcpServers: session.mcpServers,
-                allowedTools: session.allowedTools,
-                hookSettingsPath: session.hookSettingsPath,
-            });
+            session.writeStdin = null;
+            try {
+                await claudeLocal({
+                    path: session.path,
+                    sessionId: session.sessionId,
+                    abort: abortSignal,
+                    claudeEnvVars: session.claudeEnvVars,
+                    claudeArgs: session.claudeArgs,
+                    mcpServers: session.mcpServers,
+                    allowedTools: session.allowedTools,
+                    hookSettingsPath: session.hookSettingsPath,
+                    onStdinReady: (write) => {
+                        session.writeStdin = (data: string) => write(data);
+                    }
+                });
+            } finally {
+                // The child has exited: drop the stdin writer so a message that
+                // races the mode-flip isn't routed to a destroyed pipe (and then
+                // acked as consumed but silently lost). onUserMessage falls back
+                // to the queue once this is null.
+                session.writeStdin = null;
+            }
         },
         onLaunchSuccess: () => {
             session.consumeOneTimeFlags();

cli/src/claude/claudeRemoteLauncher.ts

@@ -294,6 +294,9 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                 this.abortFuture = new Future<void>();
                 let modeHash: string | null = null;
                 let mode: EnhancedMode | null = null;
+                // Track the last message consumed from the queue so we can
+                // restore it if claudeRemote throws before processing.
+                let consumedMessage: { message: string; mode: EnhancedMode } | null = null;
                 try {
                     await claudeRemote({
                         sessionId: session.sessionId,
@@ -331,6 +334,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                 }
                                 modeHash = msg.hash;
                                 mode = msg.mode;
+                                consumedMessage = msg;
                                 permissionHandler.handleModeChange(mode.permissionMode);
                                 return {
                                     message: msg.message,
@@ -356,6 +360,8 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                             session.clearSessionId();
                         },
                         onReady: () => {
+                            // Message was successfully processed, stop tracking.
+                            consumedMessage = null;
                             logger.debug(
                                 `[claudeRemoteLauncher][async-debug] onReady callback ` +
                                 `(hasPending=${Boolean(pending)}, queueSize=${session.queue.size()})`
@@ -377,6 +383,23 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                     }
                 } catch (e) {
                     logger.debug('[remote]: launch error', e);
+                    // `consumedMessage` is only assigned inside the nextMessage
+                    // callback, which TS can't trace from this catch (it narrows
+                    // the var to null), so restore through a typed local.
+                    const restore = consumedMessage as { message: string; mode: EnhancedMode } | null;
+                    if (restore) {
+                        logger.debug('[remote]: restoring lost message to queue');
+                        // Restore via the public queue API so the waiter is
+                        // notified and the mode hash is recomputed — don't poke
+                        // the private backing array. NOTE: collectBatch already
+                        // dropped the consumed messages' localIds (only the
+                        // combined string survives), so the restored item is
+                        // localId-less; a later cancel-by-localId can't target it.
+                        // Full id restoration would require collectBatch to surface
+                        // the consumed ids.
+                        session.queue.unshift(restore.message, restore.mode);
+                        consumedMessage = null;
+                    }
                     if (!this.exitReason) {
                         const detail = e instanceof Error ? e.message : String(e);
                         session.client.sendSessionEvent({ type: 'message', message: `Process exited unexpectedly: ${detail}` });

cli/src/claude/runClaude.ts

@@ -404,6 +404,21 @@ export async function runClaude(options: StartOptions = {}): Promise<void> {
             return;
         }
 
+        // If in local mode with a live stdin pipe, forward the message
+        // directly to the running Claude process instead of pushing to
+        // the queue (which would trigger doSwitch and kill the process).
+        const sessionInstance = currentSessionRef.current;
+        if (sessionInstance?.writeStdin && sessionInstance.mode === 'local') {
+            logger.debug('[start] forwarding message to local process stdin');
+            sessionInstance.stdinMessageTexts.add(formattedText);
+            sessionInstance.writeStdin(formattedText + '\n');
+            if (localId) {
+                session.emitMessagesConsumed([localId]);
+            }
+            logger.debugLargeJson('User message forwarded to local stdin:', message)
+            return;
+        }
+
         // Push with resolved permission mode, model, system prompts, and tools
         const enhancedMode: EnhancedMode = {
             permissionMode: messagePermissionMode ?? 'default',

cli/src/claude/session.ts

@@ -21,6 +21,11 @@ export class Session extends AgentSessionBase<EnhancedMode> {
     readonly startedBy: 'runner' | 'terminal';
     readonly startingMode: 'local' | 'remote';
     localLaunchFailure: LocalLaunchFailure | null = null;
+    /** Function to write data to the local Claude process's stdin. */
+    writeStdin: ((data: string) => void) | null = null;
+    /** Texts of messages that were forwarded to local Claude via stdin.
+     *  Used by the sessionScanner to skip duplicate user messages. */
+    readonly stdinMessageTexts: Set<string> = new Set();
 
     constructor(opts: {
         api: ApiClient;

cli/src/utils/spawnWithAbort.ts

@@ -31,6 +31,8 @@ export type SpawnWithAbortOptions = {
     shell?: SpawnOptions['shell'];
     stdio?: StdioOptions;
     windowsHide?: SpawnOptions['windowsHide'];
+    /** Called after the child process is spawned with a function to write to stdin. */
+    onSpawned?: (writeStdin: (data: string) => void) => void;
 };
 
 export async function spawnWithAbort(options: SpawnWithAbortOptions): Promise<void> {
@@ -57,6 +59,14 @@ export async function spawnWithAbort(options: SpawnWithAbortOptions): Promise<vo
             windowsHide: options.windowsHide ?? process.platform === 'win32'
         });
 
+        if (options.onSpawned) {
+            options.onSpawned((data: string) => {
+                if (child.stdin && !child.stdin.destroyed) {
+                    child.stdin.write(data);
+                }
+            });
+        }
+
         let abortKillTimeout: NodeJS.Timeout | null = null;
 
         const abortHandler = () => {