fix(cli): read stdin lazily so open pipes don't hang the CLI

src/commands/actor/push-data.ts

@@ -1,7 +1,7 @@
-import { cachedStdinInput } from '../../entrypoints/_shared.js';
 import { APIFY_STORAGE_TYPES, getApifyStorageClient, getDefaultStorageId } from '../../lib/actor.js';
 import { ApifyCommand } from '../../lib/command-framework/apify-command.js';
 import { Args } from '../../lib/command-framework/args.js';
+import { readStdin } from '../../lib/commands/read-stdin.js';
 import { error } from '../../lib/outputs.js';
 
 export class ActorPushDataCommand extends ApifyCommand<typeof ActorPushDataCommand> {
@@ -34,7 +34,7 @@ export class ActorPushDataCommand extends ApifyCommand<typeof ActorPushDataComma
 	async run() {
 		const { item: _item } = this.args;
 
-		const item = _item || cachedStdinInput;
+		const item = _item || (await readStdin());
 
 		if (!item) {
 			error({ message: 'No item was provided.' });

src/commands/datasets/push-items.ts

@@ -1,9 +1,9 @@
 import type { ApifyApiError } from 'apify-client';
 import chalk from 'chalk';
 
-import { cachedStdinInput } from '../../entrypoints/_shared.js';
 import { ApifyCommand } from '../../lib/command-framework/apify-command.js';
 import { Args } from '../../lib/command-framework/args.js';
+import { readStdin } from '../../lib/commands/read-stdin.js';
 import { tryToGetDataset } from '../../lib/commands/storages.js';
 import { error, success } from '../../lib/outputs.js';
 import { getLoggedClientOrThrow } from '../../lib/utils.js';
@@ -55,7 +55,7 @@ export class DatasetsPushDataCommand extends ApifyCommand<typeof DatasetsPushDat
 
 		let parsedData: Record<string, unknown> | Record<string, unknown>[];
 
-		const item = _item || cachedStdinInput;
+		const item = _item || (await readStdin());
 
 		if (!item) {
 			error({ message: 'No items were provided.' });

src/entrypoints/_shared.ts

@@ -11,16 +11,13 @@ import type { BuiltApifyCommand } from '../lib/command-framework/apify-command.j
 import { commandRegistry, internalRunCommand } from '../lib/command-framework/apify-command.js';
 import { CommandError } from '../lib/command-framework/CommandError.js';
 import { renderMainHelpMenu } from '../lib/command-framework/help.js';
-import { readStdin } from '../lib/commands/read-stdin.js';
 import { SUPPORTED_NODEJS_VERSION } from '../lib/consts.js';
 import { useCLIMetadata } from '../lib/hooks/useCLIMetadata.js';
 import { shouldSkipVersionCheck } from '../lib/hooks/useCLIVersionCheck.js';
 import { useCommandSuggestions } from '../lib/hooks/useCommandSuggestions.js';
 import { error } from '../lib/outputs.js';
 import { cliDebugPrint } from '../lib/utils/cliDebugPrint.js';
 
-export const cachedStdinInput = await readStdin();
-
 const cliMetadata = useCLIMetadata();
 
 export const USER_AGENT = `Apify CLI/${cliMetadata.version} (https://github.com/apify/apify-cli)`;

src/lib/command-framework/CommandError.ts

@@ -1,6 +1,6 @@
 import chalk from 'chalk';
 
-import { cachedStdinInput } from '../../entrypoints/_shared.js';
+import { peekStdin } from '../commands/read-stdin.js';
 import { useCLIMetadata } from '../hooks/useCLIMetadata.js';
 import type { BuiltApifyCommand } from './apify-command.js';
 import { selectiveRenderHelpForCommand } from './help.js';
@@ -218,7 +218,7 @@ export class CommandError extends Error {
 					'',
 					`- CLI version: \`${cliMetadata.fullVersionString}\``,
 					`- CLI debug logs (process.env.APIFY_CLI_DEBUG): ${process.env.APIFY_CLI_DEBUG ? 'Enabled' : 'Disabled'}`,
-					`- Stdin data? ${cachedStdinInput ? 'Yes' : 'No'}`,
+					`- Stdin data? ${peekStdin() ? 'Yes' : 'No'}`,
 				].join('\n');
 			}
 		}

src/lib/command-framework/apify-command.ts

@@ -9,7 +9,7 @@ import indentString from 'indent-string';
 import widestLine from 'widest-line';
 import wrapAnsi from 'wrap-ansi';
 
-import { cachedStdinInput } from '../../entrypoints/_shared.js';
+import { readStdin } from '../commands/read-stdin.js';
 import { detectAiAgent, detectCi, detectIsInteractive } from '../hooks/telemetry/detectEnvironment.js';
 import type { TrackEventMap } from '../hooks/telemetry/trackEvent.js';
 import { trackEvent } from '../hooks/telemetry/trackEvent.js';
@@ -360,7 +360,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 							this.args[camelCasedName] = String(rawArg);
 
 							if (rawArg === '-' && builderData.stdin) {
-								this.args[camelCasedName] = this._handleStdin(builderData.stdin);
+								this.args[camelCasedName] = await this._handleStdin(builderData.stdin);
 							}
 
 							if (builderData.catchAll) {
@@ -381,7 +381,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 			return;
 		}
 
-		this._parseFlags(rawFlags, rawTokens);
+		await this._parseFlags(rawFlags, rawTokens);
 
 		try {
 			await this.run();
@@ -450,7 +450,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 		return flagKey;
 	}
 
-	private _parseFlags(rawFlags: ParseResult['values'], rawTokens: ParseResult['tokens']) {
+	private async _parseFlags(rawFlags: ParseResult['values'], rawTokens: ParseResult['tokens']) {
 		if (!this.ctor.flags) {
 			return;
 		}
@@ -580,7 +580,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 
 							flagThatUsedStdin = baseFlagName;
 
-							this.flags[camelCasedName] = this._handleStdin(builderData.stdin);
+							this.flags[camelCasedName] = await this._handleStdin(builderData.stdin);
 						}
 
 						break;
@@ -705,12 +705,14 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 		});
 	}
 
-	private _handleStdin(mode: StdinMode) {
+	private async _handleStdin(mode: StdinMode) {
+		const input = await readStdin();
+
 		switch (mode) {
 			case StdinMode.Stringified:
-				return (cachedStdinInput?.toString('utf8') ?? '').trim();
+				return (input?.toString('utf8') ?? '').trim();
 			default:
-				return cachedStdinInput;
+				return input;
 		}
 	}
 

src/lib/commands/read-stdin.ts

@@ -2,10 +2,32 @@ import { once } from 'node:events';
 
 import { useStdin } from '../hooks/useStdin.js';
 
+let readPromise: Promise<Buffer | undefined> | undefined;
+let readResult: Buffer | undefined;
+
+/**
+ * Reads stdin to its end, at most once per process. Callers must only call this when the command
+ * actually wants stdin data — a pipe that stays open never ends, so this waits for as long as the
+ * writer keeps it open (#1206).
+ */
 export async function readStdin() {
-	const dataRef = await useStdin();
+	readPromise ??= _readStdin().then((data) => {
+		readResult = data;
+		return data;
+	});
 
-	const { hasData, waitDelay, stream } = dataRef;
+	return readPromise;
+}
+
+/**
+ * Stdin data read so far, without triggering a read. For diagnostics only.
+ */
+export function peekStdin() {
+	return readResult;
+}
+
+async function _readStdin() {
+	const { hasData, waitDelay, stream } = await useStdin();
 
 	if (!hasData) {
 		return;
@@ -45,10 +67,7 @@ export async function readStdin() {
 		}
 	} finally {
 		// Stop reading from stdin so its open handle can't keep the event loop (and
-		// the CLI) alive after the command finishes (#1206). This only helps when the
-		// await above settles ('end' or the no-data abort). A writer that sends data
-		// but never closes stdin still hangs up there; that needs the lazy stdin
-		// reading discussed in #1206.
+		// the CLI) alive after the command finishes (#1206).
 		stream.off('data', onData);
 		stream.pause();
 	}
@@ -57,9 +76,6 @@ export async function readStdin() {
 		clearTimeout(timeout);
 	}
 
-	// Mark further uses of useStdin / readStdin as having no more data since we've read it all
-	dataRef.hasData = false;
-
 	const concat = Buffer.concat(bufferChunks);
 
 	if (concat.length) {

src/lib/commands/resolve-input.ts

@@ -4,10 +4,10 @@ import process from 'node:process';
 
 import mime from 'mime';
 
-import { cachedStdinInput } from '../../entrypoints/_shared.js';
 import { CommandExitCodes } from '../consts.js';
 import { error } from '../outputs.js';
 import { getLocalInput } from '../utils.js';
+import { readStdin } from './read-stdin.js';
 
 interface InputOverrideOptions {
 	schemaHint?: string;
@@ -59,7 +59,7 @@ export async function getInputOverride(
 
 	if (!inputFlag && !inputFileFlag) {
 		// Try reading stdin
-		const stdin = cachedStdinInput;
+		const stdin = await readStdin();
 
 		if (stdin) {
 			try {

src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts

@@ -39,11 +39,13 @@ export function stdinCheckWrapper<Fn extends (...args: any[]) => any>(
 	}: StdinCheckWrapperOptions = {},
 ): (...args: NewFunctionArgs<Fn>) => Promise<Awaited<ReturnType<Fn>>> {
 	return async (input, ...rest) => {
-		const { isTTY, hasData } = await useStdin();
+		const { isTTY } = await useStdin();
 
 		const casted = input as StdinCheckWrapperInput<Awaited<ReturnType<Fn>>>;
 
-		if (isCI || (!isTTY && !hasData)) {
+		// Prompts need a terminal to read the answer from. Piped stdin is command input, not an
+		// answer source — before stdin became lazy (#1206) it was always drained by then anyway.
+		if (isCI || !isTTY) {
 			if (typeof casted.providedConfirmFromStdin === 'undefined') {
 				throw new Error(
 					casted.errorMessageForStdin ??

test/e2e/commands/stdin-held-open.test.ts

@@ -1,5 +1,6 @@
-import { mkdir, rm } from 'node:fs/promises';
+import { mkdir, open, rm } from 'node:fs/promises';
 import path from 'node:path';
+import process from 'node:process';
 import { fileURLToPath } from 'node:url';
 
 import { execa } from 'execa';
@@ -55,4 +56,36 @@ describe('[e2e] stdin held open (#1206)', () => {
 		expect(result.exitCode).toBe(1);
 		expect(result.stderr).toContain('Actor is of an unknown format');
 	});
+
+	// A named pipe is the harsher case: unlike the socket a spawned child gets, it has no wait
+	// deadline, so the old eager startup read blocked forever and the command never even ran.
+	// Opening the FIFO read-write keeps a writer attached, so it never reaches EOF, with no
+	// second process to manage. Windows has no mkfifo.
+	it.skipIf(process.platform === 'win32')('exits on its own when stdin is a named pipe with no writer', async () => {
+		const fifo = path.join(emptyDir, 'stdin.fifo');
+		await execa('mkfifo', [fifo]);
+
+		const handle = await open(fifo, 'r+');
+
+		try {
+			const result = await execa('node', [DistApify, 'run'], {
+				cwd: emptyDir,
+				reject: false,
+				timeout: EXIT_DEADLINE_MS,
+				stdin: handle.fd,
+				env: {
+					APIFY_CLI_DISABLE_TELEMETRY: '1',
+					APIFY_CLI_SKIP_UPDATE_CHECK: '1',
+					APIFY_DISABLE_KEYRING: '1',
+				},
+			});
+
+			expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false);
+			expect(result.exitCode).toBe(1);
+			expect(result.stderr).toContain('Actor is of an unknown format');
+		} finally {
+			await handle.close();
+			await rm(fifo, { force: true });
+		}
+	});
 });