fix(orchestrator): never inject notifications mid-turn (audit A5) (#589)

* feat(orchestrator): Joe hears loop run events

Phase 3: the goal-run event stream now feeds the orchestrator monitor —
blocked runs (pointing Joe at pending_prompts/respond_to_prompt),
awaiting_human escalations, and failures each produce one notification,
deduped per (runId, status) since blocked re-emits on prompt
re-broadcasts. Closes the gap where the supervisor only heard workflow
events while loops stalled invisibly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(orchestrator): never inject notifications mid-turn (audit A5)

All three delivery paths now gate on the orchestrator being idle:
sendOrchestratorNotification and the monitor's deliverToOrchestrator
route to the persistent outbox when Joe is mid-turn, and the outbox
flusher holds its digest under the same gate until the turn ends.
Closes the last mid-turn corruption path from the 2026-06 audit.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: alari76

server/orchestrator-monitor.ts

@@ -324,10 +324,11 @@ export class OrchestratorMonitor {
     if (!orchestratorId) return
 
     const session = this.sessions.get(orchestratorId)
-    if (!session?.claudeProcess?.isAlive()) {
-      // Orchestrator not running — hand the notification to the persistent
-      // outbox so it is replayed (as a digest) when the session comes back,
-      // instead of rotting in the in-memory buffer forever.
+    if (!session?.claudeProcess?.isAlive() || session.isProcessing) {
+      // Orchestrator not running, or mid-turn (injecting now would derail
+      // its active turn — audit item A5): hand the notification to the
+      // persistent outbox, whose flusher retries under the same idle gate,
+      // instead of letting it rot in the in-memory buffer.
       getOrchestratorOutbox().enqueue({
         label: notification.severity.toUpperCase(),
         title: notification.title,

server/orchestrator-notify.test.ts

@@ -20,10 +20,12 @@ import { sendOrchestratorNotification, type NotificationOutbox } from './orchest
 function makeSessions(opts: {
   exists: boolean
   alive: boolean
+  isProcessing?: boolean
 }) {
   const sentInputs: Array<{ id: string; data: string }> = []
   const session = opts.exists
     ? {
+        isProcessing: opts.isProcessing ?? false,
         claudeProcess: opts.alive
           ? { isAlive: vi.fn(() => true) }
           : { isAlive: vi.fn(() => false) },
@@ -64,6 +66,21 @@ describe('sendOrchestratorNotification', () => {
     )
   })
 
+  it('queues to the outbox instead of injecting mid-turn (A5)', () => {
+    const sessions = makeSessions({ exists: true, alive: true, isProcessing: true })
+    const outbox = makeOutbox()
+    const ok = sendOrchestratorNotification(sessions, {
+      parentSessionId: 'parent-id',
+      label: 'ACTION',
+      title: 'busy parent',
+      body: 'must not be interrupted',
+    }, outbox)
+
+    expect(ok).toBe(true)
+    expect(sessions.sendInput).not.toHaveBeenCalled()
+    expect(outbox.enqueue).toHaveBeenCalledWith({ label: 'ACTION', title: 'busy parent', body: 'must not be interrupted' })
+  })
+
   it('queues to the outbox and returns true when the parent session is missing', () => {
     const sessions = makeSessions({ exists: false, alive: false })
     const outbox = makeOutbox()

server/orchestrator-notify.ts

@@ -40,20 +40,25 @@ export interface NotificationOutbox {
  * Returns true when delivered immediately OR queued in the outbox for
  * replay (the outbox owns delivery from that point on). Returns false only
  * when queueing itself failed.
+ *
+ * Delivery is gated on the parent being idle: input sent while it is
+ * mid-turn lands inside the active turn and derails it (audit item A5).
+ * A busy parent gets the notification via the outbox's next flush tick,
+ * which applies the same idle gate.
  */
 export function sendOrchestratorNotification(
   sessions: SessionManager,
   args: OrchestratorNotifyArgs,
   outbox: NotificationOutbox = getOrchestratorOutbox(),
 ): boolean {
   const session = sessions.get(args.parentSessionId)
-  if (session?.claudeProcess?.isAlive()) {
+  if (session?.claudeProcess?.isAlive() && !session.isProcessing) {
     const message = `[Agent ${getAgentDisplayName()} Notification — ${args.label}]\n${args.title}\n${args.body}`
     sessions.sendInput(args.parentSessionId, message)
     return true
   }
 
-  // Parent unreachable — queue for replay when the orchestrator comes back.
+  // Parent unreachable or mid-turn — queue for replay by the outbox flusher.
   try {
     outbox.enqueue({ label: args.label, title: args.title, body: args.body })
     return true

server/orchestrator-outbox.test.ts

@@ -19,12 +19,13 @@ vi.mock('./orchestrator-manager.js', () => ({
 
 import { OrchestratorOutbox } from './orchestrator-outbox.js'
 
-function makeSessions(opts: { alive?: boolean; rateLimited?: boolean } = {}) {
-  const { alive = true, rateLimited = false } = opts
+function makeSessions(opts: { alive?: boolean; rateLimited?: boolean; isProcessing?: boolean } = {}) {
+  const { alive = true, rateLimited = false, isProcessing = false } = opts
   const sentInputs: Array<{ id: string; data: string }> = []
   return {
     isRateLimited: vi.fn(() => rateLimited),
     get: vi.fn(() => ({
+      isProcessing,
       claudeProcess: { isAlive: vi.fn(() => alive) },
     })),
     sendInput: vi.fn((id: string, data: string) => { sentInputs.push({ id, data }) }),
@@ -110,6 +111,18 @@ describe('OrchestratorOutbox', () => {
     expect(outbox.size()).toBe(1)
   })
 
+  it('flush holds the digest while the orchestrator is mid-turn (A5)', () => {
+    const outbox = new OrchestratorOutbox(filePath)
+    outbox.enqueue({ label: 'ACTION', title: 't', body: 'b' })
+    const sessions = makeSessions({ isProcessing: true })
+
+    expect(outbox.flush(sessions)).toBe(0)
+    expect(sessions.sendInput).not.toHaveBeenCalled()
+
+    // Next tick after the turn ends: delivered.
+    expect(outbox.flush(makeSessions())).toBe(1)
+  })
+
   it('flush delivers a single-item digest with the original label', () => {
     const outbox = new OrchestratorOutbox(filePath)
     outbox.enqueue({ label: 'ALERT', title: 'Workflow failed', body: 'Run x failed' })

server/orchestrator-outbox.ts

@@ -71,6 +71,10 @@ export class OrchestratorOutbox {
     if (!orchestratorId) return 0
     const session = sessions.get(orchestratorId)
     if (!session?.claudeProcess?.isAlive()) return 0
+    // Never inject mid-turn: input sent while the orchestrator is processing
+    // lands inside its active turn and derails it. Hold the digest for the
+    // next flush tick instead.
+    if (session.isProcessing) return 0
 
     const count = this.items.length
     const digest = this.buildDigest()