# Plan — Mid-turn injection engine parity (claude-cli ↔ openai)

## The contract (engine-blind, foundational)

A user sends a mid-turn message via HTTP `POST /agents/<agent>/chat` with the
active session's `sessionId`. Regardless of engine, the caller gets back the
**real assistant response to that message** (with content, tokens, durationMs)
once the agent has actually answered. The SSE / JSON response carries the same
shape and timing whether the agent is openai or claude-cli.

## Current divergence (verified)

### OpenAI mid-turn flow (works correctly)
1. `router.js:316–333` — when `runningSessions.has(sid)` and `message` provided,
   `runChat` routes to **`injectIntoActiveOpenaiSession`** (`router.js:148–266`).
2. That function:
   - Persists the user row + emits `chat.user_message`
   - Enqueues an `agent_messages` row with `from_agent=USER_SENTINEL` + a
     `correlation_id`
   - **Polls** the correlation row for up to 5 minutes (200 ms then 500 ms)
   - Returns the real `{content, tokenUsage, durationMs, …}` when the row's
     `response` column is populated.
3. The openai loop drains `USER_SENTINEL` rows at iteration start
   (`loop.js:334`), captures `correlationIds` into `pendingCorrelationIds`, runs
   the model, then calls `postCorrelatedResponse(cid, content, usage)`
   (`loop.js:491–494`).
4. SSE chat handler at `chat.js:121` sends `done` with the **real** result.

### claude-cli mid-turn flow (the bug)
1. `router.js:599` — `runClaudeChat` calls `pushToExistingSession(sid, message)`.
2. `pushToExistingSession` persists the user row, buffers content into
   `_pendingInjections` (in-memory), returns `true`.
3. `runClaudeChat` returns the **stub** at line 605:
   `{content: null, iterations: 0, durationMs: 0, tokenUsage: {0,0,0,0}, …}`.
4. SSE handler emits `done` with the stub → Studio sees `done` → spinner dies.
5. The actual response chunks land on the WS bus when `_pendingInjections`
   eventually flushes — but the SSE for THIS request already closed.

## What we want

claude-cli mid-turn injection must use the **same correlation-row + polling
contract** as openai. Same SSE `done` shape, same blocking behavior, same
`tokenUsage`. Studio's behavior should be identical.

## Architectural reality

The openai loop is consumer-managed: `runLoop` iterates inside the same
async function, drains at iteration start, posts correlated response per
iteration. Easy to plumb.

The claude-cli SDK manages its own internal loop. Our `runClaudeSession`
generator yields events to the consumer (`runClaudeChat`), which **breaks the
for-await on `chat.response`**. That break tears down the runtime via
`iterator.return()`. So if we drain mid-turn and push a new user message to
the SDK, the new turn's events are never observed by the consumer.

**Two-part fix:**

1. **Engine side** — make `runClaudeSession` aware of pending injections at
   the natural turn boundary (the SDK's `result` event). At that moment:
   - Drain `USER_SENTINEL` rows (filtered to `from_agent === USER_SENTINEL`
     only, NOT sub-agent reply rows — those are handled by wake to keep the
     async-inform streaming fix intact).
   - If any drained: push via `promptQueue.pushMessage` (real, not
     `pushSilent` — SDK ignores `shouldQuery: false`), track
     `pendingCorrelationIds`, **DO NOT yield `chat.response` yet** — continue
     the for-await; the SDK starts a new turn.
   - If none drained: yield `chat.response` as today, consumer breaks.
   - When the next turn's assistant content arrives:
     `postCorrelatedResponse(cid, content, usage)` for each pending
     correlation_id, then clear them.
   - On runtime teardown (cancel / error / final result with no pending):
     flush any remaining `pendingCorrelationIds` with audit text and clean DB
     rows (existing pattern at the outer finally).

2. **Router side** — replace the `pushToExistingSession` early-return with a
   call to a new `injectIntoActiveClaudeSession`. This function mirrors
   `injectIntoActiveOpenaiSession`:
   - Persist user row + emit `chat.user_message`.
   - Enqueue `agent_messages` row with `USER_SENTINEL` + `correlation_id`.
   - Poll the correlation row up to `INJECTION_TIMEOUT_MS` (5 min).
   - Race-recovery: if `runningSessions.has(sid)` becomes false mid-poll
     (active loop ended without draining our row), trust the post-teardown
     wake to pick it up — **don't recurse via continueFlag**, just keep
     polling. Wake will fire fresh `runChat` with the queued content;
     postCorrelatedResponse will be called by the engine when that runChat's
     turn finishes.
   - Return real `{content, tokenUsage, durationMs, …}` when the row's
     `response` column is populated.

## What stays the same

- **`pushToExistingSession`** — kept for the existing in-memory buffer
  semantics (used internally; not the HTTP-injection path anymore). May be
  dead code afterward, but leave the function intact for potential other
  callers.
- **`_pendingInjections`** buffer + `_inFlight` flag — unchanged.
- **Sub-agent async-inform path** — unchanged. Sub-agent reply rows go to
  `wake.js`, get drained there, flow through fresh runChat. The post-teardown
  wake hook stays.
- **Direct chat (no injection)** — unchanged.
- **OpenAI path** — completely untouched.

## Files to modify

| File | Change |
|---|---|
| `core/router.js` | Add `injectIntoActiveClaudeSession`. Replace the `pushToExistingSession` early-return at line 599-606 with a call to it. |
| `engines/claude-engine.js` | At the `result` handler before `chat.response` yield: drain USER_SENTINEL rows (filtered), pushMessage, track correlation_ids; only yield `chat.response` when no pending; on each subsequent assistant content yield, post correlated responses. Outer finally: flush any unflushed correlation_ids with audit text. |
| `core/queue.js` | Optional: add `drainUserSentinelOnly(agentName, sessionId)` helper or extend `drainNonFollowup` with a filter param. Cleaner. |

## Filter semantics (CRITICAL — two distinct drain paths)

`drainNonFollowup` currently drains ALL non-followup rows for an agent. We
need TWO disjoint drain filters now (criticizer #4 — race recovery):

- **Engine's mid-turn drain (NEW)** — `from_agent === USER_SENTINEL` ONLY.
  HTTP user injections that the active runtime should process before yielding
  chat.response.
- **Wake's drain (UPDATED)** — `from_agent !== USER_SENTINEL` ONLY.
  Sub-agent reply rows (auto-delivered from `postCorrelatedResponse`) and any
  other non-USER injection. **Wake must NOT touch USER_SENTINEL rows** — they
  belong to the active runtime; if the runtime ends without draining them,
  the injection caller's race-recovery path picks them up via
  `runChat({continueFlag: true, message: null})`, like openai does.

Add two helpers to `core/queue.js`:
```js
function drainUserSentinelOnly(agentName, sessionId) { ... }       // engine
function drainNonUserSentinel(agentName, sessionId) { ... }        // wake
```
Or extend `drainNonFollowup` with an `excludeUserSentinel` boolean param.

This keeps openai's loop (`loop.js:334`) calling the original
`drainNonFollowup` (drains everything — openai handles all rows uniformly).

## Engine drain at runtime startup (race-recovery support)

When `injectIntoActiveClaudeSession` recurses via `runChat({continueFlag:true,
message:null})` after the active runtime died, the new fresh `runClaudeSession`
must drain pending USER_SENTINEL rows and process them. Otherwise the runtime
sits idle waiting for input that already arrived.

Add a startup drain BEFORE the for-await loop in `runClaudeSession`:
```js
// After storeRuntime(...), BEFORE the for-await:
if (continueFlag && (!userMessage || userMessage === '')) {
  const { drainUserSentinelOnly } = require('../core/queue');
  const { messages, correlationIds } = drainUserSentinelOnly(agent.name, sessionId);
  if (messages.length > 0) {
    const seedContent = messages.map(m => m.content).join('\n\n');
    promptQueue.pushMessage(null, seedContent);
    db.addMessage({ sessionId, role: 'user', content: seedContent, modelKey });
    pendingCorrelationIds.push(...correlationIds);
  }
  // If no messages drained, continueFlag-with-no-message would push empty.
  // Skip the line ~351 initial pushMessage(userMessage) below in that case.
}
```

Need to track `continueFlag` is currently a `runChat` parameter — it's
threaded through to `runClaudeChat` (router.js:579) but not currently passed
to `runClaudeSession`. Need to thread it through one more level.

## Engine-side implementation sketch

In `runClaudeSession`'s `result` handler (currently around line 658-690):

```js
if (sdkMsg.type === 'result') {
  // ... existing usage accumulation ...

  if (mode !== 'chat') break;

  // ── NEW: drain USER_SENTINEL rows BEFORE yielding chat.response ──
  const { drainUserSentinelOnly } = require('../core/queue');
  const { messages: userInjections, correlationIds: injCids } =
      drainUserSentinelOnly(agent.name, sessionId);

  if (userInjections.length > 0) {
    // Push as REAL user messages (not pushSilent — SDK ignores that flag)
    for (const m of userInjections) {
      promptQueue.pushMessage(claudeSessionIdResolved, m.content);
    }
    pendingCorrelationIds.push(...injCids);
    // Don't yield chat.response — let the SDK process the new turn
    runtimeEntry._inFlight = false;  // we're between SDK-turns
    continue;  // back to for-await
  }

  // No injections — natural turn end
  yield { type: 'chat.response', content: lastContent, ... };

  // (rest of the existing flush logic for _pendingInjections, etc.)
  continue;
}
```

In the `assistant` handler (where `pendingCorrelationIds` is already
processed for sub-agent / async-inform), the existing code already handles
the `postCorrelatedResponse` flush at lines ~456-462. That same code
naturally handles our new HTTP-injection correlation IDs.

**TOKEN ACCOUNTING (per criticizer #7):** the existing call passes `null` for
usage. Update to pass per-turn tokens, mirroring `loop.js:485-493`:
```js
const correlatedUsage = {
  input:  perCallUsage?.input  || 0,
  output: perCallUsage?.output || 0,
  cache:  perCallUsage?.cache  || 0,
  cost:   perCallUsage?.cost   || 0,
};
postCorrelatedResponse(cid, textContent, correlatedUsage, { isFinalIteration });
```
Per-call tokens come from the assistant message's `sdkMsg.message.usage`
(already extracted at line ~477 as `tokenUsage`).

## Router-side implementation sketch

```js
// New helper in router.js — mirrors injectIntoActiveOpenaiSession
async function injectIntoActiveClaudeSession({ sid, agentName, message, agent, settings, cwd, runStartTime }) {
  // - persist user row + emit chat.user_message (mirror of openai)
  // - enqueue USER_SENTINEL row with correlation_id
  // - poll correlation row up to INJECTION_TIMEOUT_MS (5 min)
  // - RACE-RECOVERY (corrected per criticizer #4): if runningSessions.has(sid)
  //   becomes false mid-poll, the active runtime ended without draining our
  //   row. Since wake.js NO LONGER drains USER_SENTINEL rows (per the new
  //   filter), the row sits pending. We must recover by recursing:
  //     return runChat({ agentName, message: null, sessionId: sid,
  //                      continueFlag: true, cwd, settings });
  //   This matches openai's race-recovery exactly. continueFlag=true tells
  //   runChat to skip user-message-add (already persisted). The recursive
  //   runChat enters runClaudeChat, finds no active runtime, falls through
  //   to fresh runClaudeSession. The engine's drainUserSentinelOnly drains
  //   our row, processes it, postCorrelatedResponse fires, our caller's
  //   poll resolves with real content.
}
```

**continueFlag for claude-cli — verification needed:** runClaudeChat currently
takes `continueFlag` as a parameter. When called with continueFlag=true and
message=null, it should: skip user-message-add, run runClaudeSession with the
existing claude_session_id (resume), engine drains USER_SENTINEL rows on its
own at result-handler time. Verify this path during implementation.

Then in `runClaudeChat` (~line 599):
```js
// Replace the existing pushToExistingSession block
if (message && runningSessions.has(sid)) {
  return injectIntoActiveClaudeSession({
    sid, agentName, message, agent, settings, cwd,
    runStartTime: Date.now(),
  });
}
```

`pushToExistingSession` no longer called from runClaudeChat. It stays in the
engine module (dead code, but harmless).

## Edge cases to handle

1. **Cancel mid-injection** — user cancels session while `injectIntoActiveClaudeSession` is polling.
   - cancelRegistry signal aborts SDK → engine exits via outer finally.
   - Outer finally flushes pending correlation_ids with audit text
     ("(target session cancelled)") via existing pattern.
   - Poll sees `response` populated with audit text, returns it.
   - SSE emits `done` with that audit content. Cancelled flag must propagate.

2. **Concurrent injections** (rare but possible) — two HTTP requests during
   same active turn.
   - Each gets its own correlation_id. Both rows enqueued. Both polled.
   - **Decision (per criticizer #3):** mirror openai's behavior — JOIN the
     drained user contents into ONE turn (`messages.map(m => m.content).join('\n\n')`),
     pushMessage that single combined message, post the SAME response content
     to ALL pending cids when the turn finishes. Matches `loop.js:485-493`'s
     pattern of posting `content` to every `pendingCorrelationIds` entry.
   - The agent answers all questions in one combined response. Each polling
     caller receives the same response. This is acceptable UX and matches
     openai exactly.

3. **Race-recovery: runtime ends before engine drains the row** (REVISED per criticizer #4)
   - `runningSessions.has(sid)` becomes false inside the poll loop.
   - Wake.js NOW skips USER_SENTINEL rows (new filter), so the row stays
     pending in agent_messages.
   - Caller recurses: `await runChat({ agentName, message: null, sessionId: sid,
     continueFlag: true, cwd, settings })`. Same pattern as openai.
   - The recursive runChat → runClaudeChat with continueFlag=true; no message
     to add → enters runClaudeSession with `userMessage: ''`, engine sees no
     initial user message but runtime starts.
   - **Subtle issue:** runClaudeSession at line 350-352 unconditionally pushes
     userMessage and persists. If userMessage is empty, this would push an
     empty SDK message which historical comments say causes "No response
     requested." — bad. Need to handle continueFlag-with-empty-message:
     either skip the initial pushMessage, OR have engine drain
     USER_SENTINEL rows IMMEDIATELY at runtime startup (before the for-await
     even begins). The latter is cleaner.

4. **Buffered `_pendingInjections` behavior** — kept as-is. Some legacy
   callers might still rely on it (e.g., if any code pushes via direct
   `pushToExistingSession` outside of the chat HTTP path). Audit during
   implementation.

5. **Sub-agent async-inform path** — completely separate. Sub-agent's reply
   row has `from_agent=<sub-name>`, NOT `USER_SENTINEL`. Filter excludes it.
   It still goes via wake → fresh runChat. No change.

## Verification scenarios

Required to pass before declaring done:

1. **Direct chat** — 5/5 with full event stream. (regression check)
2. **Sub-agent async_inform** — 30/30 wake test. (regression check — must stay 100%)
3. **Mid-turn injection on openai (`assistant` agent)** — 5/5 with real
   response in SSE done. (regression check — openai untouched)
4. **Mid-turn injection on claude-cli (`cc-sonnet`)** — 5/5 with real response
   in SSE done. (the new behavior)
5. **Sleep-10s + mid-tool injection on cc-sonnet** — tool MUST complete
   normally (not killed), AND the injected message MUST get a real response.
   Specific assertions:
   - First request's tool: `success: true`
   - First request's content: contains expected output
   - Second request (mid-tool): blocks, returns real content (not stub),
     `iterations` ≥ 0 (not 0), real `tokenUsage`
6. **Cancel during mid-tool injection** — cancel propagates, both poll and
   active turn cleaned up. SSE `done` carries cancelled state.
7. **Concurrent two-injection** — both get their own real responses.
8. **Trace verification** — debug instrumentation on for first round of testing,
   verify the timeline matches the expected pattern.

## Criticizer focus areas (pre-impl)

- **Filter correctness** — does `drainUserSentinelOnly` correctly leave
  sub-agent reply rows in the queue? Auto-delivered rows have
  `from_agent=<sub-name>`, not `USER_SENTINEL`. ✓
- **chat.response yield timing** — by NOT yielding chat.response when
  injections are pending, are we breaking any other consumer assumption?
  Search for chat.response consumers.
- **Multi-turn within one runtime** — claude-cli runtime stays alive across
  multiple SDK turns when we don't break consumer. Does the SDK actually
  support this? Verify by reading SDK behavior comment at
  `engines/claude-engine.js:31-35`.
- **Token accounting** — current code accumulates `totalTokens` across
  iterations. Is the `tokenUsage` returned to the caller for the SECOND
  injected turn correct (just that turn's tokens, not cumulative)?
- **Response identification** — when assistant text arrives, multiple
  correlation_ids might be pending. Currently the openai pattern posts the
  SAME content to all. For claude-cli with multiple injections each as
  separate turns, each turn's content should go to ITS specific
  correlation_id. How to associate?
- **Race-recovery via wake** — if runtime dies mid-poll, post-teardown wake
  fires fresh runChat. Does wake call `postCorrelatedResponse` for the
  correlation_id? Looking at wake.js, it just runs runChat with the drained
  content as the message. The fresh runChat's runClaudeSession would handle
  postCorrelatedResponse via its existing assistant handler. ✓
- **DB persistence ordering** — `pushToExistingSession` does
  `db.addMessage` BEFORE buffering. Our new path enqueues to
  agent_messages (which the engine will later drain and push as a real
  user message). Does this also `db.addMessage` first? It must — Studio's
  history view expects the user row immediately. Look at
  `injectIntoActiveOpenaiSession:153` — yes, persists first.

## Criticizer focus areas (post-impl)

- Trace data: every BAD scenario (if any) MUST have a trace timeline
  showing where divergence happens.
- Token accounting matches openai's pattern.
- Cancel path: full cleanup in both poll and engine.
- Memory leak check on `pendingCorrelationIds` — flushed on all exit paths.

## Implementation checklist (post round-2 criticizer)

These specific corrections were called out and MUST be addressed:

### Mandatory code-level corrections

1. **Thread `continueFlag` to runClaudeSession** (criticizer #1).
   - `core/router.js:679` — add `continueFlag` to the `runClaudeSession({...})` opts.
   - `engines/claude-engine.js:238` — destructure `continueFlag` from opts.

2. **Startup drain BEFORE line 351's initial pushMessage** (criticizer #2).
   ```js
   // After storeRuntime, BEFORE the for-await:
   let skipInitialPush = false;
   if (continueFlag && (!userMessage || userMessage === '')) {
     const { drainUserSentinelOnly } = require('../core/queue');
     const { messages, correlationIds } = drainUserSentinelOnly(agent.name, sessionId);
     if (messages.length > 0) {
       const seedContent = messages.map(m => m.content).join('\n\n');
       promptQueue.pushMessage(null, seedContent);
       db.addMessage({ sessionId, role: 'user', content: seedContent, modelKey });
       pendingCorrelationIds.push(...correlationIds);
       skipInitialPush = true;
     }
   }
   if (!skipInitialPush) {
     promptQueue.pushMessage(null, userMessage);
     db.addMessage({ sessionId, role: 'user', content: userMessage, modelKey });
   }
   emitBusEvent({ type: 'chat.user_message', ... });  // unchanged
   ```

3. **postCorrelatedResponse with per-turn tokens** (criticizer #3 + #7).
   At `engines/claude-engine.js:464`:
   ```js
   const correlatedUsage = {
     input:  tokenUsage.input  || 0,    // tokenUsage extracted from sdkMsg.message.usage at line 477
     output: tokenUsage.output || 0,
     cache:  0,
     cost:   0,
   };
   for (const cid of pendingCorrelationIds) {
     postCorrelatedResponse(cid, textContent, correlatedUsage, { isFinalIteration });
   }
   ```

4. **Add queue.js helpers** (criticizer #4).
   ```js
   function drainUserSentinelOnly(agentName, sessionId = null) {
     // Same body as drainNonFollowup but with extra: if (msg.from_agent !== USER_SENTINEL) continue;
   }
   function drainNonUserSentinel(agentName, sessionId = null) {
     // Same body as drainNonFollowup but with extra: if (msg.from_agent === USER_SENTINEL) continue;
   }
   module.exports = { ..., drainUserSentinelOnly, drainNonUserSentinel };
   ```

5. **Update wake.js to use drainNonUserSentinel** (criticizer #4).
   At `core/wake.js:142-143`:
   ```js
   const { drainNonUserSentinel } = require('./queue');
   const { messages: drained } = drainNonUserSentinel(session.agent_name, targetSessionId);
   ```

6. **Engine result handler — drain ORDER + invariant** (concern A).
   ```js
   if (sdkMsg.type === 'result') {
     // ... usage accumulation (unchanged) ...

     if (mode !== 'chat') break;

     // Drain USER_SENTINEL rows that arrived during this turn
     const { drainUserSentinelOnly } = require('../core/queue');
     const { messages: userInjections, correlationIds: injCids } =
       drainUserSentinelOnly(agent.name, sessionId);

     if (userInjections.length > 0) {
       const seedContent = userInjections.map(m => m.content).join('\n\n');
       promptQueue.pushMessage(claudeSessionIdResolved, seedContent);
       pendingCorrelationIds.push(...injCids);
       runtimeEntry._inFlight = false;  // we're between SDK turns
       continue;  // back to for-await — DO NOT yield chat.response
     }

     // INVARIANT: chat.response is yielded ONLY when no injections are
     // pending. pendingCorrelationIds may still contain async_inform sub-agent
     // cids — those get flushed via postCorrelatedResponse in the assistant
     // handler when their content arrives. By the time we reach a result
     // event with no fresh USER_SENTINEL drain AND pendingCorrelationIds is
     // empty, the natural turn end has been reached for ALL injections.
     yield { type: 'chat.response', content: lastContent, ... };

     // Existing _pendingInjections flush (unchanged)
     if (runtimeEntry) {
       runtimeEntry._inFlight = false;
       if (runtimeEntry._pendingInjections.length > 0) { ... flush ... }
     }
     continue;
   }
   ```

7. **`injectIntoActiveClaudeSession` race-recovery** (criticizer #4 + #5).
   When `runningSessions.has(sid)` becomes false mid-poll:
   ```js
   if (!runningSessions.has(sid)) {
     // The active runtime ended. Wake.js no longer drains USER_SENTINEL
     // rows, so our row sits pending. Recurse to start a fresh runtime
     // that will drain it at startup.
     // Don't delete the row — the recursive runChat reads it via the
     // engine's startup drain (continueFlag=true path).
     return await runChat({
       agentName, message: null, sessionId: sid,
       continueFlag: true, cwd, settings,
     });
   }
   ```
   On final timeout (after 5 min): row stays pending in DB. Will be picked up
   by the next manual chat's wake fire. Acceptable — matches openai's
   timeout pattern.

## Workflow

1. Write this plan ✓
2. Pre-impl criticizer (focus on architectural concerns above)
3. Iterate plan if needed
4. Build trace instrumentation BEFORE implementing
5. Implement engine + router changes
6. Add tracing-on tests for first verification round
7. Post-impl criticizer on the actual code
8. Run all 8 verification scenarios with traces
9. Iterate if any scenario fails — diagnose from traces, fix, retest
10. Once all 8/8 pass: remove debug instrumentation
11. Final test pass on clean code (5/5 each scenario, 30/30 sub-agent)
12. Report metrics + final diff