can1357/oh-my-pi · warning

Streaming injection is unavailable

Error message

Streaming injection is unavailable

What it means

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.

Source

Thrown at packages/coding-agent/src/session/yield-queue.ts:128

				return;
			}
		}
	}

	async flush(mode: YieldFlushMode): Promise<void> {
		if (mode === "idle") {
			this.#idleFlushPending = false;
		}
		const idleMessages: BuiltMessage[] = [];
		for (const [kind, dispatcher] of this.#dispatchers) {
			if (mode === "idle" && dispatcher.skipIdleFlush) continue;
			const entries = this.#drain(kind);
			if (entries.length === 0) continue;
			const built = this.#build(kind, dispatcher, entries);
			if (!built) continue;
			if (mode === "streaming") {
				try {
					if (!this.#options.injectStreaming) throw new Error("Streaming injection is unavailable");
					this.#options.injectStreaming(built.message);
					this.#resolveEntries(built.entries);
				} catch (error) {
					const dispatchError = error instanceof Error ? error : new Error(String(error));
					this.#rejectEntries(built.entries, dispatchError);
					logger.warn("Yield queue streaming dispatch failed", { kind, error: formatError(error) });
				}
			} else {
				idleMessages.push(built);
			}
		}
		if (mode === "idle" && idleMessages.length > 0) {
			for (const item of idleMessages) this.#attachEntrySettlement(item);
			try {
				await this.#options.injectIdle(idleMessages.map(item => item.message));
				for (const item of idleMessages) {
					(item.message as AgentMessage & { [ASIDE_MESSAGE_COMMIT]?: () => void })[ASIDE_MESSAGE_COMMIT]?.();
				}

View on GitHub (pinned to 9690622007)

Solutions

1. Provide injectStreaming when constructing/flushing the queue, or flush with a non-streaming mode (e.g. queue for after the turn).
2. Check mode selection: only use 'streaming' when an active streaming turn and injector exist.
3. Catch it downstream: entries are rejected with this error and logged, so handle rejected injection promises.
4. Re-create the queue with the injector once the streaming session is available.

Example fix

// before
new YieldQueue({ /* no injectStreaming */ });
queue.flush("streaming", dispatcher);
// after
new YieldQueue({ injectStreaming: msg => session.inject(msg), /* ... */ });
queue.flush("streaming", dispatcher);

if (mode === "streaming" && !options.injectStreaming) {
  // defer injection until after the turn instead of flushing in streaming mode
  mode = "queued";
}

function supportsStreaming(q: { options: { injectStreaming?: unknown } }): boolean {
  return typeof q.options.injectStreaming === "function";
}

try {
  queue.flush("streaming", dispatcher);
} catch (err) {
  // entries are already rejected inside flush; observe the warning and re-queue for post-turn delivery
  queue.defer(builtEntries);
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• No messages to continue from
• Cannot continue from message role: assistant
• Request was aborted
• No response body for V2 compaction streaming
• V2 compaction stream closed before response.completed

AI-assisted analysis of can1357/oh-my-pi@9690622007 (2026-08-31). Data as JSON: /api/errors/e110089201fd558e. Report an issue: GitHub.