{"record":{"id":"e110089201fd558e","repo":"can1357/oh-my-pi","slug":"streaming-injection-is-unavailable","errorCode":null,"errorMessage":"Streaming injection is unavailable","messagePattern":"Streaming injection is unavailable","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"packages/coding-agent/src/session/yield-queue.ts","lineNumber":128,"sourceCode":"\t\t\t\treturn;\n\t\t\t}\n\t\t}\n\t}\n\n\tasync flush(mode: YieldFlushMode): Promise<void> {\n\t\tif (mode === \"idle\") {\n\t\t\tthis.#idleFlushPending = false;\n\t\t}\n\t\tconst idleMessages: BuiltMessage[] = [];\n\t\tfor (const [kind, dispatcher] of this.#dispatchers) {\n\t\t\tif (mode === \"idle\" && dispatcher.skipIdleFlush) continue;\n\t\t\tconst entries = this.#drain(kind);\n\t\t\tif (entries.length === 0) continue;\n\t\t\tconst built = this.#build(kind, dispatcher, entries);\n\t\t\tif (!built) continue;\n\t\t\tif (mode === \"streaming\") {\n\t\t\t\ttry {\n\t\t\t\t\tif (!this.#options.injectStreaming) throw new Error(\"Streaming injection is unavailable\");\n\t\t\t\t\tthis.#options.injectStreaming(built.message);\n\t\t\t\t\tthis.#resolveEntries(built.entries);\n\t\t\t\t} catch (error) {\n\t\t\t\t\tconst dispatchError = error instanceof Error ? error : new Error(String(error));\n\t\t\t\t\tthis.#rejectEntries(built.entries, dispatchError);\n\t\t\t\t\tlogger.warn(\"Yield queue streaming dispatch failed\", { kind, error: formatError(error) });\n\t\t\t\t}\n\t\t\t} else {\n\t\t\t\tidleMessages.push(built);\n\t\t\t}\n\t\t}\n\t\tif (mode === \"idle\" && idleMessages.length > 0) {\n\t\t\tfor (const item of idleMessages) this.#attachEntrySettlement(item);\n\t\t\ttry {\n\t\t\t\tawait this.#options.injectIdle(idleMessages.map(item => item.message));\n\t\t\t\tfor (const item of idleMessages) {\n\t\t\t\t\t(item.message as AgentMessage & { [ASIDE_MESSAGE_COMMIT]?: () => void })[ASIDE_MESSAGE_COMMIT]?.();\n\t\t\t\t}","sourceCodeStart":110,"sourceCodeEnd":146,"githubUrl":"https://github.com/can1357/oh-my-pi/blob/969062200754ea02cfac922e5ebb8c608c079e15/packages/coding-agent/src/session/yield-queue.ts#L110-L146","documentation":"YieldQueue.flush supports a 'streaming' dispatch mode that injects a queued message into an active stream via the injectStreaming option. If the mode is streaming but no injectStreaming callback was provided (it's optional), flush throws this sentinel error; it is caught immediately and converted into a rejection of the drained entries plus a warning log.","triggerScenarios":"Queue configured/invoked with mode='streaming' while options.injectStreaming is undefined — typically creating the queue without the streaming injector and then flushing in streaming mode mid-turn.","commonSituations":"Caller builds a YieldQueue for deferred (post-turn) injection but later calls flush with 'streaming' during an active turn; queue constructed before the streaming session exists so the injector couldn't be wired; refactors dropped the option.","solutions":["Provide injectStreaming when constructing/flushing the queue, or flush with a non-streaming mode (e.g. queue for after the turn).","Check mode selection: only use 'streaming' when an active streaming turn and injector exist.","Catch it downstream: entries are rejected with this error and logged, so handle rejected injection promises.","Re-create the queue with the injector once the streaming session is available."],"exampleFix":"// before\nnew YieldQueue({ /* no injectStreaming */ });\nqueue.flush(\"streaming\", dispatcher);\n// after\nnew YieldQueue({ injectStreaming: msg => session.inject(msg), /* ... */ });\nqueue.flush(\"streaming\", dispatcher);","handlingStrategy":"type-guard","validationCode":"if (mode === \"streaming\" && !options.injectStreaming) {\n  // defer injection until after the turn instead of flushing in streaming mode\n  mode = \"queued\";\n}\n","typeGuard":"function supportsStreaming(q: { options: { injectStreaming?: unknown } }): boolean {\n  return typeof q.options.injectStreaming === \"function\";\n}","tryCatchPattern":"try {\n  queue.flush(\"streaming\", dispatcher);\n} catch (err) {\n  // entries are already rejected inside flush; observe the warning and re-queue for post-turn delivery\n  queue.defer(builtEntries);\n}","preventionTips":["Always construct YieldQueue with injectStreaming if streaming-mode flush is possible.","Only flush with mode='streaming' during an active streaming turn.","Treat this error as 'injector not wired' and fall back to queued delivery.","Wire the injector as soon as the streaming session opens, before accepting queued messages."],"tags":["injection","streaming","api-misuse"],"backgroundTag":"missing-callback","analyzedSha":"969062200754ea02cfac922e5ebb8c608c079e15","analyzedAt":"2026-08-31T10:29:35.737Z","schemaVersion":2},"datasetVersion":"2026-08-31T14:17:45.589Z"}