Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 149 additions & 0 deletions packages/aft-bridge/src/__tests__/bridge-transport.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,155 @@ process.stdin.on("data", (chunk) => {
}
});

test("ignores stale child stdout end after replacement owns the buffer", async () => {
const script = writeExecutable(
"stale-stdout.js",
`#!/usr/bin/env node
process.stdin.setEncoding("utf8");
let buffer = "";
setInterval(() => {}, 1_000);
process.stdin.on("data", (chunk) => {
buffer += chunk;
let newline;
while ((newline = buffer.indexOf("\\n")) !== -1) {
const line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
const req = JSON.parse(line);
process.stdout.write(JSON.stringify({ id: req.id, success: true, warnings: [], pid: process.pid }) + "\\n");
}
});
`,
);
const bridge = new BinaryBridge(script, workDir, { timeoutMs: 5_000, maxRestarts: 0 });
const testBridge = bridge as unknown as {
process: ChildProcess | null;
stdoutBuffer: string;
};

try {
await bridge.send("ping");
const staleChild = testBridge.process;
if (!staleChild?.stdin) throw new Error("bridge child stdin is unavailable");

const closed = new Promise<void>((resolve) => staleChild.stdin?.once("close", resolve));
staleChild.stdin.destroy();
await closed;
await bridge.send("ping");

testBridge.stdoutBuffer = '{"id":"replacement"';
staleChild.stdout?.emit("end");
expect(testBridge.stdoutBuffer).toBe('{"id":"replacement"');
} finally {
await bridge.shutdown();
}
});

test("does not carry configure completion across child generations", async () => {
const script = writeExecutable(
"configure-generation.js",
`#!/usr/bin/env node
process.stdin.setEncoding("utf8");
let buffer = "";
let configured = false;
process.stdin.on("data", (chunk) => {
buffer += chunk;
let newline;
while ((newline = buffer.indexOf("\\n")) !== -1) {
const line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
const req = JSON.parse(line);
if (req.command === "configure") configured = true;
process.stdout.write(JSON.stringify({
id: req.id,
success: true,
warnings: [],
configured,
pid: process.pid,
}) + "\\n");
}
});
`,
);
const bridge = new BinaryBridge(script, workDir, { timeoutMs: 5_000, maxRestarts: 0 });
const testBridge = bridge as unknown as {
process: ChildProcess | null;
deliverConfigureWarnings: () => Promise<void>;
};
let releaseWarnings: (() => void) | undefined;
let warningsEntered: (() => void) | undefined;
const warningGate = new Promise<void>((resolve) => {
releaseWarnings = resolve;
});
const entered = new Promise<void>((resolve) => {
warningsEntered = resolve;
});
testBridge.deliverConfigureWarnings = async () => {
warningsEntered?.();
await warningGate;
};

try {
const first = bridge.send("ping");
await entered;
const staleChild = testBridge.process;
if (!staleChild?.stdin) throw new Error("bridge child stdin is unavailable");

const closed = new Promise<void>((resolve) => staleChild.stdin?.once("close", resolve));
staleChild.stdin.destroy();
await closed;
const second = bridge.send("ping");
releaseWarnings?.();
await Promise.allSettled([first, second]);

const third = await bridge.send("ping");
expect(third.configured).toBe(true);
} finally {
releaseWarnings?.();
await bridge.shutdown();
}
});

test("invalidates the child when stdin write throws synchronously", async () => {
const script = writeExecutable(
"sync-write-error.js",
`#!/usr/bin/env node
process.stdin.setEncoding("utf8");
let buffer = "";
process.stdin.on("data", (chunk) => {
buffer += chunk;
let newline;
while ((newline = buffer.indexOf("\\n")) !== -1) {
const line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
const req = JSON.parse(line);
process.stdout.write(JSON.stringify({ id: req.id, success: true, warnings: [], pid: process.pid }) + "\\n");
}
});
`,
);
const bridge = new BinaryBridge(script, workDir, { timeoutMs: 5_000, maxRestarts: 0 });
const testBridge = bridge as unknown as {
process: ChildProcess | null;
pending: Map<string, unknown>;
};

try {
const first = await bridge.send("ping");
const staleChild = testBridge.process;
if (!staleChild?.stdin) throw new Error("bridge child stdin is unavailable");
staleChild.stdin.write = (() => {
throw new Error("synchronous write failure");
}) as typeof staleChild.stdin.write;

await expect(bridge.send("mutate")).rejects.toThrow("synchronous write failure");
expect(testBridge.pending.size).toBe(0);
const second = await bridge.send("ping");
expect(second.pid).not.toBe(first.pid);
} finally {
await bridge.shutdown();
}
});

test("single timeout with child stdout activity keeps bridge warm and sibling alive", async () => {
const script = writeExecutable(
"alive-starvation.js",
Expand Down
17 changes: 17 additions & 0 deletions packages/aft-bridge/src/__tests__/error-contract.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,14 @@ import {
import {
BridgeTransportTimeoutError,
BridgeTransportUnavailableError,
BridgeTransportUnknownOutcomeError,
isBridgeTransportTimeout,
} from "../bridge.js";
import {
AftToolError,
adaptToolError,
BASH_TRANSPORT_DISPOSITION,
BRIDGE_TRANSPORT_UNKNOWN_OUTCOME_DISPOSITION,
isBashTransportDeadError,
SUBC_MODULE_RESTART_DISPOSITION,
} from "../error-contract.js";
Expand Down Expand Up @@ -111,6 +113,10 @@ describe("isBashTransportDeadError", () => {
}),
],
["route GOODBYE with unknown outcome", routeGoodbyeError()],
[
"standalone write with unknown outcome",
new BridgeTransportUnknownOutcomeError("write failed"),
],
["live bridge request timeout", new BridgeTransportTimeoutError("bash", 100, "bridge busy")],
["ordinary tool failure", new Error("command failed")],
];
Expand Down Expand Up @@ -175,6 +181,17 @@ describe("adaptToolError", () => {
expect(original.message).not.toContain(BASH_TRANSPORT_DISPOSITION);
});

test("an outcome-unknown standalone write never gets bash re-run guidance", () => {
const original = new BridgeTransportUnknownOutcomeError("stdin write failed");

adaptToolError("bash", original);

expect(original.message).toContain(BRIDGE_TRANSPORT_UNKNOWN_OUTCOME_DISPOSITION);
expect(original.message).toContain("UNKNOWN");
expect(original.message).toContain("never blind-retry a mutation");
expect(original.message).not.toContain(BASH_TRANSPORT_DISPOSITION);
});

test("disposition is appended once when the error passes through twice", () => {
const original = routeGoodbyeError();

Expand Down
67 changes: 50 additions & 17 deletions packages/aft-bridge/src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,14 @@ export class BridgeTransportUnavailableError extends Error {
}
}

/** The transport failed after a request may have been written to the child. */
export class BridgeTransportUnknownOutcomeError extends BridgeTransportUnavailableError {
constructor(message: string, options?: { cause?: unknown }) {
super(message, options);
this.name = "BridgeTransportUnknownOutcomeError";
}
}

/** Type guard for a transport-timeout rejection (bridge busy, retryable). */
export function isBridgeTransportTimeout(err: unknown): err is BridgeTransportTimeoutError {
return err instanceof Error && (err as { code?: unknown }).code === "transport_timeout";
Expand Down Expand Up @@ -588,7 +596,11 @@ export class BinaryBridge implements AftProjectTransport {
this.clearRestartResetTimer();
this.configured = false;
this.outstandingBackgroundTaskIds.clear();
this.rejectAllPending(error);
this.rejectAllPending(
error instanceof BridgeTransportUnknownOutcomeError
? error
: new BridgeTransportUnknownOutcomeError(error.message, { cause: error }),
);
}

hasPendingRequests(): boolean {
Expand Down Expand Up @@ -808,6 +820,14 @@ export class BinaryBridge implements AftProjectTransport {
if (!this.configured) {
if (command !== "configure" && command !== "version") {
if (!this._configurePromise) {
const configuringChild = this.process;
const requireConfiguringChild = (): void => {
if (this.process !== configuringChild) {
throw new BridgeTransportUnavailableError(
`${this.errorPrefix} Bridge process changed during configure; retry on replacement`,
);
}
};
// First caller — create the configure promise.
// All parallel callers await this same promise.
//
Expand All @@ -829,6 +849,7 @@ export class BinaryBridge implements AftProjectTransport {
},
implicitTransportOptions,
);
requireConfiguringChild();
if (configResult.success === false) {
throw new Error(
`${this.errorPrefix} Configure failed: ${configResult.message ?? "unknown error"}`,
Expand All @@ -838,7 +859,9 @@ export class BinaryBridge implements AftProjectTransport {
// and relayed through stderr → plugin log. No need to re-log here
// (doing so would just duplicate the same line in aft-plugin.log).
await this.deliverConfigureWarnings(configResult, params, options);
requireConfiguringChild();
await this.checkVersion(implicitTransportOptions);
requireConfiguringChild();
// Re-check liveness after version check — checkVersion() swallows
// errors as best-effort, so the bridge may have died without throwing.
if (!this.isAlive()) {
Expand Down Expand Up @@ -959,21 +982,27 @@ export class BinaryBridge implements AftProjectTransport {
}

requestSentAt = Date.now();
child.stdin.write(line, (err) => {
if (err) {
const error = new BridgeTransportUnavailableError(
`${this.errorPrefix} Failed to write to stdin: ${err.message}`,
{ cause: err },
);
const entry = this.pending.get(id);
if (entry) {
this.pending.delete(id);
clearTimeout(entry.timer);
entry.reject(error);
}
if (this.process === child) this.invalidateTransportProcess(error);
const handleWriteFailure = (cause: unknown): void => {
const writeError = cause instanceof Error ? cause : new Error(String(cause));
const error = new BridgeTransportUnknownOutcomeError(
`${this.errorPrefix} Failed to write to stdin: ${writeError.message}`,
{ cause: writeError },
);
const entry = this.pending.get(id);
if (entry) {
this.pending.delete(id);
clearTimeout(entry.timer);
entry.reject(error);
}
});
if (this.process === child) this.invalidateTransportProcess(error);
};
try {
child.stdin.write(line, (err) => {
if (err) handleWriteFailure(err);
});
} catch (err) {
handleWriteFailure(err);
}
});

if (
Expand Down Expand Up @@ -1305,19 +1334,23 @@ export class BinaryBridge implements AftProjectTransport {

const stdoutDecoder = new StringDecoder("utf8");
child.stdout?.on("data", (chunk: Buffer) => {
if (this.process !== currentChild) return;
this.onStdoutData(stdoutDecoder.write(chunk));
});
child.stdout?.on("end", () => {
if (this.process !== currentChild) return;
const remaining = stdoutDecoder.end();
if (remaining) this.onStdoutData(remaining);
this.flushStdoutBuffer();
});

const stderrDecoder = new StringDecoder("utf8");
child.stderr?.on("data", (chunk: Buffer) => {
if (this.process !== currentChild) return;
this.onStderrData(stderrDecoder.write(chunk));
});
child.stderr?.on("end", () => {
if (this.process !== currentChild) return;
const remaining = stderrDecoder.end();
if (remaining) this.onStderrData(remaining);
this.flushStderrBuffer();
Expand Down Expand Up @@ -1351,7 +1384,7 @@ export class BinaryBridge implements AftProjectTransport {
this.configured = false;
this.clearRestartResetTimer();
this.rejectAllPending(
new BridgeTransportUnavailableError(`${this.errorPrefix} Binary killed by ${signal}`),
new BridgeTransportUnknownOutcomeError(`${this.errorPrefix} Binary killed by ${signal}`),
);
return;
}
Expand Down Expand Up @@ -1622,7 +1655,7 @@ export class BinaryBridge implements AftProjectTransport {
}

this.rejectAllPending(
new BridgeTransportUnavailableError(
new BridgeTransportUnknownOutcomeError(
`${this.errorPrefix} Binary crashed (restarts: ${this._restartCount})${cause ? `: ${cause.message}` : ""} (see ${this.getLogFilePathVia()})`,
{ cause },
),
Expand Down
Loading
Loading