fix: buffer stdio until exact session starts

docs/event-producer-consumer.md

@@ -15,7 +15,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
 | `agent/queued` | `emit` | [`packages/core/agent/src/types.ts:348`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - |
 | `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:473`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - |
 | `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:525`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`tool-skill`](../packages/skill/tool-skill) |
-| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:369`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`invariants`](../packages/support/invariants) |
+| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:369`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`invariants`](../packages/support/invariants), [`stdio-agent`](../packages/ui/stdio-agent) |
 | `agent/status` | `emit` | [`packages/core/agent/src/types.ts:333`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`invariants`](../packages/support/invariants), [`stdio-agent`](../packages/ui/stdio-agent) |
 | `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:540`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - |
 | `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:558`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) |

packages/ui/stdio-agent/README.md

@@ -32,7 +32,7 @@ The leaf `cordis.yml` supplies only the **swappable backends** — an LLM adapte
 | `welcome` | `ready.` | the stdin-chat banner |
 | `resumeSessionId` | — | resume a persisted session id instead of starting fresh (sourced from an env var in the leaf) |
 
-Fresh stdio sessions use the process launch directory as `session.header.cwd` and mint one combined `main-session-<uuid>` agent/session id, so durable restarts cannot collide. The app passes that exact opaque id to both its config-created agent and UI; an AgentLoop-only reload resumes materialized history under that id, while the UI's `main` text remains only a display label and never selects another registry root by prefix or insertion order. A resumed run binds both components to the exact `resumeSessionId` and keeps the cwd stored in the persisted session header.
+Fresh stdio sessions use the process launch directory as `session.header.cwd` and mint one combined `main-session-<uuid>` agent/session id, so durable restarts cannot collide. The app passes that exact opaque id to both its config-created agent and UI; an AgentLoop-only reload resumes materialized history under that id, while the UI's `main` text remains only a display label and never selects another registry root by prefix or insertion order. Readline buffers nonblank startup input for that identity until `agent/session-start`, so piped stdin cannot outrun asynchronous exact-id restoration or let EOF discard the queued prompt. A resumed run binds both components to the exact `resumeSessionId` and keeps the cwd stored in the persisted session header.
 
 ## The bin
 

packages/ui/stdio-agent/src/stdio-chat.ts

@@ -3,9 +3,10 @@
  * `steer()`, and renders the durable transcript to stdout. A UI is "just a
  * plugin" — it consumes the `session/event` feed (the assistant token stream,
  * turn/step boundaries, tool activity, todos) plus a few `agent/*` control
- * events (`agent/status`, `agent/created`/`agent/disposed`) and the `agents`
- * service. Dimmed chain-of-thought rendering plus robust piped-stdin EOF→idle
- * exit handling, configured via {@link Config}.
+ * events (`agent/status`, `agent/created`/`agent/disposed`,
+ * `agent/session-start`) and the `agents` service. Dimmed chain-of-thought
+ * rendering plus robust piped-stdin EOF→idle exit handling, configured via
+ * {@link Config}.
  *
  * An internal module of the stdio app, not a package of its own: the app's
  * front-door cluster always includes this UI, and nothing else composes it.
@@ -105,12 +106,6 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt
   const matchesConfiguredIdentity = (agent: Agent): boolean =>
     agent.id === config.sessionId && ctx.agents.roots().includes(agent)
   let target: Agent | undefined = ctx.agents.roots().find(agent => agent.id === config.sessionId)
-  ctx.on('agent/created', (agent) => {
-    if (matchesConfiguredIdentity(agent)) target = agent
-  })
-  ctx.on('agent/disposed', (agent) => {
-    if (target === agent) target = undefined
-  })
 
   // Transcript rendering off the durable `session/event` feed — the assistant
   // token stream, turn/step boundaries, tool activity, and todos all come from
@@ -158,7 +153,6 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt
   })
 
   ctx.effect(() => {
-    const reader = createInterface({ input, output, terminal: isTTYPair(input, output) })
     // Piped-input exit, once stdin reaches EOF:
     //  - If no line ever submitted work (empty stdin, blank-only lines), exit
     //    immediately — no turn will ever start, so there is nothing to wait
@@ -176,6 +170,36 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt
     let exitTimer: ReturnType<typeof setTimeout> | undefined
     let activeQuestion: PendingQuestion | undefined
     const questionQueue: PendingQuestion[] = []
+    const queuedInput: string[] = []
+    let targetReady = target !== undefined
+    let hadReadyTarget = targetReady
+
+    const submit = (agent: Agent, text: string): void => {
+      submittedWork = true
+      if (agent.status === 'running') {
+        agent.steer([{ type: 'text', text }])
+      } else {
+        agent.send([{ type: 'text', text }])
+      }
+    }
+
+    const disposeCreatedListener = ctx.on('agent/created', (agent) => {
+      if (!matchesConfiguredIdentity(agent)) return
+      target = agent
+      targetReady = false
+    })
+    const disposeSessionStartListener = ctx.on('agent/session-start', (agent) => {
+      if (agent !== target) return
+      targetReady = true
+      hadReadyTarget = true
+      for (const text of queuedInput.splice(0)) submit(agent, text)
+    })
+    const disposeDisposedListener = ctx.on('agent/disposed', (agent) => {
+      if (target !== agent) return
+      target = undefined
+      targetReady = false
+    })
+    const reader = createInterface({ input, output, terminal: isTTYPair(input, output) })
 
     const maybeExit = (): void => {
       if (disposed || !stdinClosed) return
@@ -351,16 +375,20 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt
       const text = line.trim()
       if (!text) return
       const agent = target
-      if (!agent) {
+      if (agent === undefined || !targetReady) {
+        // Initial exact-id restoration is asynchronous. Preserve input until
+        // session-start, the first supported point for queueing agent work.
+        // After a previously ready target disappears, a line in the HMR gap
+        // still fails loud unless its exact replacement is already publishing.
+        if (!hadReadyTarget || agent !== undefined) {
+          submittedWork = true
+          queuedInput.push(text)
+          return
+        }
         ctx.logger.error('ui-stdio: main agent is not running')
         return
       }
-      submittedWork = true
-      if (agent.status === 'running') {
-        agent.steer([{ type: 'text', text }])
-      } else {
-        agent.send([{ type: 'text', text }])
-      }
+      submit(agent, text)
     })
     reader.on('close', () => {
       // Fires for BOTH stdin EOF and plugin disposal (reader.close() below);
@@ -376,6 +404,9 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt
       disposePendingQuestions()
       disposeUserInteractionProvider()
       disposeStatusListener()
+      disposeCreatedListener()
+      disposeSessionStartListener()
+      disposeDisposedListener()
       reader.close()
     }
   }, 'ui-stdio')

packages/ui/stdio-agent/tests/stdio-chat.spec.ts

@@ -64,6 +64,13 @@ function makeAgent(id: string, status: AgentStatus = 'idle'): Agent & {
   } as never
 }
 
+/** Register a fake configured agent and cross the supported startup-work boundary. */
+function registerReady(ctx: Context, agent: Agent, source: 'startup' | 'resume' = 'startup'): () => void {
+  const dispose = ctx.agents.register(agent)
+  ctx.emit('agent/session-start', agent, source)
+  return dispose
+}
+
 /** A session stub whose `header.id` matches an agent's, for `session/event` emits. */
 function makeSession(id: string): Session {
   return { id, header: { id } } as Session
@@ -198,15 +205,21 @@ describe('createStdioChat rendering', () => {
     expect(out.text()).toContain('[main turn 5] ')
   })
 
-  it('accepts a lineage-bearing configured agent created after the UI installs', async () => {
+  it('buffers input for a lineage-bearing configured agent until its session starts', async () => {
     const { ctx, input } = await setup({ welcome: 'hi there', sessionId: 'resumed' })
+    input.feed('continue')
+    await new Promise(resolve => setImmediate(resolve))
+
     const unrelated = makeAgent('unrelated')
     ctx.agents.register(unrelated)
+    ctx.emit('agent/session-start', unrelated, 'startup')
     const resumed = makeAgent('resumed')
     ;(resumed.session.header as { parentSession?: string }).parentSession = 'persisted-parent'
     ctx.agents.register(resumed)
+    await new Promise(resolve => setImmediate(resolve))
+    expect(resumed.sent).toEqual([])
 
-    input.feed('continue')
+    ctx.emit('agent/session-start', resumed, 'resume')
     await new Promise(resolve => setImmediate(resolve))
 
     expect(unrelated.sent).toEqual([])
@@ -256,9 +269,11 @@ describe('createStdioChat rendering', () => {
     disposeOld()
     const replacement = makeAgent('main-session-fixed')
     ctx.agents.register(replacement)
-
     input.feed('after hmr')
     await new Promise(resolve => setImmediate(resolve))
+    expect(replacement.sent).toEqual([])
+    ctx.emit('agent/session-start', replacement, 'resume')
+    await new Promise(resolve => setImmediate(resolve))
 
     expect(prefixCollision.sent).toEqual([])
     expect(replacement.sent).toEqual([[{ type: 'text', text: 'after hmr' }]])
@@ -269,7 +284,7 @@ describe('createStdioChat rendering', () => {
     const unrelated = makeAgent('unrelated')
     ctx.agents.register(unrelated)
     const configured = makeAgent('main')
-    const disposeConfigured = ctx.agents.register(configured)
+    const disposeConfigured = registerReady(ctx, configured)
     const error = vi.spyOn(ctx.logger, 'error').mockImplementation(() => {})
 
     disposeConfigured()
@@ -693,7 +708,7 @@ describe('createStdioChat input', () => {
   it('sends a typed line to an idle agent', async () => {
     const { ctx, input } = await setup()
     const agent = makeAgent('main', 'idle')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('do a thing')
     await new Promise(r => setImmediate(r))
     expect(agent.sent).toEqual([[{ type: 'text', text: 'do a thing' }]])
@@ -703,7 +718,7 @@ describe('createStdioChat input', () => {
   it('steers a typed line into a running agent', async () => {
     const { ctx, input } = await setup()
     const agent = makeAgent('main', 'running')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('steer me')
     await new Promise(r => setImmediate(r))
     expect(agent.steered).toEqual([[{ type: 'text', text: 'steer me' }]])
@@ -719,18 +734,26 @@ describe('createStdioChat input', () => {
     expect(agent.sent).toEqual([])
   })
 
-  it('logs and drops a line when the target agent is not running', async () => {
+  it('buffers a line until the initial target session starts', async () => {
     const { ctx, input } = await setup()
     const spy = vi.spyOn(ctx.logger, 'error').mockImplementation(() => {})
     input.feed('nobody home')
     await new Promise(r => setImmediate(r))
-    expect(spy).toHaveBeenCalledWith('ui-stdio: main agent is not running')
+    expect(spy).not.toHaveBeenCalled()
+
+    const agent = makeAgent('main')
+    ctx.agents.register(agent)
+    await new Promise(r => setImmediate(r))
+    expect(agent.sent).toEqual([])
+    ctx.emit('agent/session-start', agent, 'startup')
+    await new Promise(r => setImmediate(r))
+    expect(agent.sent).toEqual([[{ type: 'text', text: 'nobody home' }]])
   })
 
   it('drives the exact app-configured resumed session', async () => {
     const { ctx, input } = await setup({ welcome: 'w', sessionId: 'worker' })
     const agent = makeAgent('worker')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent, 'resume')
     input.feed('hi')
     await new Promise(r => setImmediate(r))
     expect(agent.sent).toHaveLength(1)
@@ -749,7 +772,7 @@ describe('createStdioChat EOF exit', () => {
   it('waits for the agent to settle idle after running before exiting', async () => {
     const { ctx, input, exit } = await setup()
     const agent = makeAgent('main', 'idle')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('work')
     await new Promise(r => setImmediate(r))
     input.finish()
@@ -764,10 +787,31 @@ describe('createStdioChat EOF exit', () => {
     expect(exit).toHaveBeenCalledWith(0)
   })
 
+  it('keeps piped EOF pending until buffered startup input runs', async () => {
+    const { ctx, input, exit } = await setup()
+    input.feed('work')
+    input.finish()
+    await flushExit()
+    expect(exit).not.toHaveBeenCalled()
+
+    const agent = makeAgent('main', 'idle')
+    ctx.agents.register(agent)
+    await new Promise(r => setImmediate(r))
+    expect(agent.sent).toEqual([])
+    ctx.emit('agent/session-start', agent, 'startup')
+    await new Promise(r => setImmediate(r))
+    expect(agent.sent).toEqual([[{ type: 'text', text: 'work' }]])
+    ctx.emit('agent/status', agent, 'running')
+    ;(agent as { status: AgentStatus }).status = 'idle'
+    ctx.emit('agent/status', agent, 'idle')
+    await flushExit()
+    expect(exit).toHaveBeenCalledWith(0)
+  })
+
   it('schedules the exit only once when idle fires repeatedly', async () => {
     const { ctx, input, exit } = await setup()
     const agent = makeAgent('main', 'running')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('work')
     await new Promise(r => setImmediate(r))
     ctx.emit('agent/status', agent, 'running') // sawRunning = true
@@ -785,7 +829,7 @@ describe('createStdioChat EOF exit', () => {
   it('does not exit on an idle transition for a different agent', async () => {
     const { ctx, input, exit } = await setup()
     const agent = makeAgent('main', 'idle')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('work')
     await new Promise(r => setImmediate(r))
     input.finish()
@@ -799,7 +843,7 @@ describe('createStdioChat EOF exit', () => {
   it('does not exit while a turn is still running at EOF', async () => {
     const { ctx, input, exit } = await setup()
     const agent = makeAgent('main', 'idle')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('work')
     await new Promise(r => setImmediate(r))
     ctx.emit('agent/status', agent, 'running')
@@ -848,7 +892,7 @@ describe('createStdioChat disposal (HMR safety)', () => {
   it('removes the agent/status listener on dispose', async () => {
     const { ctx, fiber, input, exit } = await setup()
     const agent = makeAgent('main', 'idle')
-    ctx.agents.register(agent)
+    registerReady(ctx, agent)
     input.feed('work')
     await new Promise(r => setImmediate(r))
     await fiber.dispose()