Function runStreamCase

apps/cli/scripts/integration/lib/stream-harness.ts:66–152

(options: RunStreamCaseOptions)

}

export async function runStreamCase(options: RunStreamCaseOptions): Promise<void> {
	const cliRoot = process.env.ROO_CLI_ROOT ? path.resolve(process.env.ROO_CLI_ROOT) : defaultCliRoot
	const timeoutMs = options.timeoutMs ?? 120_000

	const child = execa(
		"pnpm",
		["dev", "--print", "--stdin-prompt-stream", "--provider", "openrouter", "--output-format", "stream-json"],
		{
			cwd: cliRoot,
			stdin: "pipe",
			stdout: "pipe",
			stderr: "pipe",
			reject: false,
			forceKillAfterDelay: 2_000,
		},
	)

	child.stderr?.on("data", (chunk) => {
		process.stderr.write(chunk)
	})

	let requestCounter = 0

	const context: StreamCaseContext = {
		cliRoot,
		timeoutMs,
		nextRequestId(prefix: string): string {
			requestCounter += 1
			return `${prefix}-${Date.now()}-${requestCounter}`
		},
		sendCommand(command: StreamCommand): void {
			if (child.stdin?.destroyed) {
				return
			}

			child.stdin.write(`${JSON.stringify(command)}\n`)
		},
	}

	let handlerError: Error | null = null
	let timedOut = false

	const timeout = setTimeout(() => {
		timedOut = true
		const message = options.onTimeoutMessage?.(context) ?? "timed out waiting for stream scenario completion"
		handlerError = new Error(message)
		child.kill("SIGTERM")
	}, timeoutMs)

	const rl = readline.createInterface({
		input: child.stdout!,
		crlfDelay: Infinity,
	})

	rl.on("line", (line) => {
		process.stdout.write(`${line}\n`)

		const event = parseEvent(line)