Function runStdinStreamMode

apps/cli/src/commands/cli/stdin-stream.ts:342–977

({ host, jsonEmitter, setStreamRequestId }: StdinStreamModeOptions)

}

export async function runStdinStreamMode({ host, jsonEmitter, setStreamRequestId }: StdinStreamModeOptions) {
	let hasReceivedStdinCommand = false
	let shouldShutdown = false
	let activeTaskPromise: Promise<void> | null = null
	let fatalStreamError: Error | null = null
	let activeRequestId: string | undefined
	let activeTaskCommand: "start" | undefined
	let latestTaskId: string | undefined
	let cancelRequestedForActiveTask = false
	let awaitingPostCancelRecovery = false
	let hasSeenQueueState = false
	let lastQueueDepth = 0
	let lastQueueMessageIds: string[] = []
	const pendingQueuedMessageRequestIds: string[] = []
	const queueMessageRequestIdByMessageId = new Map<string, string>()

	const assignRequestIdsToNewQueueMessages = (queueMessageIds: string[]) => {
		for (const messageId of queueMessageIds) {
			if (queueMessageRequestIdByMessageId.has(messageId)) {
				continue
			}

			const requestId = pendingQueuedMessageRequestIds.shift()
			if (!requestId) {
				continue
			}

			queueMessageRequestIdByMessageId.set(messageId, requestId)
		}
	}

	const promoteRequestIdForDequeuedMessages = (queueMessageIds: string[]) => {
		if (lastQueueMessageIds.length === 0) {
			return
		}

		const remainingIds = new Set(queueMessageIds)

		for (const dequeuedMessageId of lastQueueMessageIds) {
			if (remainingIds.has(dequeuedMessageId)) {
				continue
			}

			const requestId = queueMessageRequestIdByMessageId.get(dequeuedMessageId)
			if (requestId) {
				setStreamRequestId(requestId)
			}
			queueMessageRequestIdByMessageId.delete(dequeuedMessageId)
		}
	}

	const waitForPreviousTaskToSettle = async () => {
		if (!activeTaskPromise) {
			return
		}

		try {
			await activeTaskPromise