shifathosn
/
rea
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(process): support explicit stdin closure

This commit is contained in:
morluto 2026-07-24 13:13:09 +00:00
parent 62ac73cdd7
commit 205bf1529b
8 changed files with 74 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/application/ProcessReactiveEffects.ts
View File

@ -14,6 +14,7 @@ import type { ProcessCaptureJournal } from "./ProcessCaptureJournal.js";
export interface ProcessReactiveTerminal {
write(data: string): void;
resize(columns: number, rows: number): void;
closeInput(): void;
kill(signal: "SIGINT" | "SIGTERM" | "SIGKILL"): void;
}
@ -56,6 +57,7 @@ const interactionData = (
: action.data;
if (action.type === "resize")
return `${String(action.columns)}x${String(action.rows)}`;
if (action.type === "close_stdin") return "";
return action.signal;
};
@ -64,6 +66,7 @@ const interactionType = (
): InteractionEvent["type"] => {
if (action.type === "send_input") return "input";
if (action.type === "resize") return "resize";
if (action.type === "close_stdin") return "stdin_close";
return "signal";
};
@ -77,7 +80,8 @@ const performTerminalAction = (
else if (action.type === "resize") {
terminal.resize(action.columns, action.rows);
renderer.resize(action.columns, action.rows, atMs);
} else terminal.kill(action.signal);
} else if (action.type === "close_stdin") terminal.closeInput();
else terminal.kill(action.signal);
};
const recordInteraction = (

22
src/application/ProcessReactiveHarness.ts
View File

@ -17,6 +17,7 @@ import type { CommandShimReplay } from "./CommandShimReplay.js";
import { ProcessReactiveCoordinator } from "./ProcessReactiveCoordinator.js";
import {
executeProcessReactiveEffects,
type ProcessReactiveTerminal,
unsupportedProcessReactiveFeatures,
} from "./ProcessReactiveEffects.js";
import { subscribeProcessReactiveObservations } from "./ProcessReactiveObservations.js";
@ -34,6 +35,25 @@ export interface ProcessReactiveHarness {
readonly unsubscribe: () => void;
}
const reactiveTerminalFor = (
terminal: IPty | undefined,
): ProcessReactiveTerminal | undefined => {
if (terminal === undefined) return undefined;
return {
write: (data) => terminal.write(data),
resize: (columns, rows) => terminal.resize(columns, rows),
closeInput: () => {
const end: unknown = Reflect.get(terminal, "end");
if (typeof end !== "function")
throw new ProcessCaptureError(
"PTY backend does not expose an stdin-close operation",
);
Reflect.apply(end, terminal, []);
},
kill: (signal) => terminal.kill(signal),
};
};
const controlKind = (
input: ProcessReactiveInput,
): ProcessReactiveControlRecord["kind"] | undefined =>
@ -81,7 +101,7 @@ export const startProcessReactiveHarness = (options: {
execute: (actions, signal) =>
executeProcessReactiveEffects(
{
terminal: options.terminal,
terminal: () => reactiveTerminalFor(options.terminal()),
renderer: options.renderer,
checkpoints: options.checkpoints,
interactions: options.capture.interactions,

14
src/domain/processCapture.ts
View File

@ -34,12 +34,12 @@ export interface RenderedTerminalFrame {
readonly serialized_state: string;
}
/** Scheduled input, resize, or signal with its observed dispatch outcome. */
/** Scheduled terminal interaction with its observed dispatch outcome. */
export interface InteractionEvent {
readonly sequence: number;
readonly scheduled_at_ms: number;
readonly dispatched_at_ms: number;
readonly type: "input" | "resize" | "signal";
readonly type: "input" | "resize" | "stdin_close" | "signal";
readonly data: string;
readonly outcome: "dispatched" | "target_exited" | "failed";
}
@ -307,7 +307,7 @@ export const processCaptureSchema: z.ZodType<ProcessCapture> = z.object({
sequence: z.number().int().nonnegative(),
scheduled_at_ms: z.number().int().nonnegative(),
dispatched_at_ms: z.number().int().nonnegative(),
type: z.enum(["input", "resize", "signal"]),
type: z.enum(["input", "resize", "stdin_close", "signal"]),
data: z.string(),
outcome: z.enum(["dispatched", "target_exited", "failed"]),
}),
@ -447,7 +447,13 @@ export const processCaptureSchema: z.ZodType<ProcessCapture> = z.object({
.max(PROCESS_REACTIVE_LIMITS.actionsPerTransition),
action_types: z
.array(
z.enum(["send_input", "resize", "send_signal", "checkpoint"]),
z.enum([
"send_input",
"resize",
"close_stdin",
"send_signal",
"checkpoint",
]),
)
.max(PROCESS_REACTIVE_LIMITS.actionsPerTransition),
}),

1
src/domain/processReactiveScenario.ts
View File

@ -190,6 +190,7 @@ const processReactiveActionSchema = z.discriminatedUnion("type", [
columns: z.number().int().min(1).max(1_000),
rows: z.number().int().min(1).max(1_000),
}),
z.strictObject({ type: z.literal("close_stdin") }),
z.strictObject({
type: z.literal("send_signal"),
target: signalTargetSchema,

24
src/domain/processReactiveTransition.ts
View File

@ -121,6 +121,23 @@ const applyProcessReactiveTransition = (input: {
};
};
const terminalControlObservationMatches = (
action: Exclude<
ProcessReactiveScenario["states"][number]["on"][number]["actions"][number],
{ readonly type: "checkpoint" | "send_input" }
>,
payload: Readonly<Record<string, unknown>>,
): boolean => {
if (action.type === "resize")
return (
payload["type"] === "resize" &&
payload["data"] === `${String(action.columns)}x${String(action.rows)}`
);
if (action.type === "close_stdin")
return payload["type"] === "stdin_close" && payload["data"] === "";
return payload["type"] === "signal" && payload["data"] === action.signal;
};
const effectObservationMatchesAction = (
action: ProcessReactiveScenario["states"][number]["on"][number]["actions"][number],
observation: ProcessObservation,
@ -160,12 +177,7 @@ const effectObservationMatchesAction = (
committedSensitiveInputs,
)
);
if (action.type === "resize")
return (
payload["type"] === "resize" &&
payload["data"] === `${String(action.columns)}x${String(action.rows)}`
);
return payload["type"] === "signal" && payload["data"] === action.signal;
return terminalControlObservationMatches(action, payload);
};
const expectedInputEvidence = (

8
tests/processCapture.test.ts
View File

@ -1558,6 +1558,7 @@ describe("process capture adapter", () => {
},
actions: [
{ type: "send_input", data: "accepted\n", sensitive: true },
{ type: "close_stdin" },
],
target: { kind: "goto", state: "finishing" },
},
@ -1620,6 +1621,13 @@ describe("process capture adapter", () => {
outcome: "dispatched",
}),
);
expect(result.value.interaction_events).toContainEqual(
expect.objectContaining({
type: "stdin_close",
data: "",
outcome: "dispatched",
}),
);
expect(result.value.frames.map(({ data }) => data).join("")).toContain(
"Done",
);

5
tests/processReactiveCoordinator.test.ts
View File

@ -114,6 +114,7 @@ const createHarness = (scenario: ProcessReactiveScenario) => {
write: (data: string) => terminalCalls.push(`write:${data}`),
resize: (columns: number, rows: number) =>
terminalCalls.push(`resize:${String(columns)}x${String(rows)}`),
closeInput: () => terminalCalls.push("close-input"),
kill: (signal: "SIGINT" | "SIGTERM" | "SIGKILL") =>
terminalCalls.push(`kill:${signal}`),
}),
@ -223,6 +224,7 @@ describe("process reactive coordinator", () => {
actions: [
{ type: "send_input", data: "secret", sensitive: true },
{ type: "resize", columns: 100, rows: 40 },
{ type: "close_stdin" },
{
type: "send_signal",
target: { kind: "root" },
@ -247,11 +249,13 @@ describe("process reactive coordinator", () => {
"write:secret",
"resize:100x40",
"render:100x40",
"close-input",
"kill:SIGINT",
]);
expect(harness.interactions.map(({ data }) => data)).toEqual([
"<redacted-input:6-bytes>",
"100x40",
"",
"SIGINT",
]);
expect(harness.checkpoints).toHaveLength(1);
@ -261,6 +265,7 @@ describe("process reactive coordinator", () => {
"obs.interaction_events.0",
"obs.interaction_events.1",
"obs.interaction_events.2",
"obs.interaction_events.3",
"obs.filesystem_checkpoints.0",
]);
await harness.coordinator.close();

8
tests/processReactiveRuntime.test.ts
View File

@ -112,7 +112,9 @@ const succeededEffect = (action: ProcessReactiveAction, order: number) => ({
? "input"
: action.type === "resize"
? "resize"
: "signal",
: action.type === "close_stdin"
? "stdin_close"
: "signal",
data:
action.type === "send_input"
? action.sensitive
@ -120,7 +122,9 @@ const succeededEffect = (action: ProcessReactiveAction, order: number) => ({
: action.data
: action.type === "resize"
? `${String(action.columns)}x${String(action.rows)}`
: action.signal,
: action.type === "close_stdin"
? ""
: action.signal,
outcome: "dispatched",
}),
});