History

fix(server): flush bounded validation diagnostics before failure

Julius Marminge committed 2 months ago

Commit 2126d7a1 parent d54b58b6 files · +212 −29Browse files

Unified · Split

6 files
  1. apps/server/src/provider/AntigravityInstallation.test.ts +28 −4
  2. apps/server/src/provider/acp/AcpSessionRuntime.ts +26 −7
  3. apps/server/src/provider/acp/AntigravityAcpSupport.test.ts +69 −0
  4. apps/server/src/provider/acp/AntigravityAcpSupport.ts +7 −4
  5. apps/server/src/provider/antigravityAuthSupport.test.ts +34 −0
  6. apps/server/src/provider/antigravityAuthSupport.ts +48 −14
apps/server/src/provider/AntigravityInstallation.test.ts+28−4
@@ -438,6 +438,24 @@ it.layer(NodeServices.layer)("Antigravity installation", (it) => { · 437 unmodified lines
438438 stderr: [`discarded prefix ${"x".repeat(8_000)}\n`, "last startup diagnostic\n"],
439439 expected: "last startup diagnostic",
440440 },
441+ {
442+ name: "unterminated diagnostic",
443+ stderr: ["Failed to create an AF_INET6 socket."],
444+ expected: "Failed to create an AF_INET6 socket.",
445+ },
446+ {
447+ name: "oversized diagnostic",
448+ stderr: [`discarded prefix ${"x".repeat(65_536)} last startup diagnostic\n`],
449+ expected: "last startup diagnostic",
450+ },
451+ {
452+ name: "fragmented oversized authorization message",
453+ stderr: [
454+ `${ANTIGRAVITY_AUTH_BROWSER_MARKER}${"x".repeat(65_536)}`,
455+ "secret-authorization-tail\nlast startup diagnostic",
456+ ],
457+ expected: "last startup diagnostic",
458+ },
441459 {
442460 name: "no diagnostic",
443461 stderr: [],
@@ -453,6 +471,7 @@ it.layer(NodeServices.layer)("Antigravity installation", (it) => { · 9 unmodified lines
453471 }
454472 const helper = command.args[0] === "-e";
455473 const exited = yield* Deferred.make<ChildProcessSpawner.ExitCode>();
474+ const exitObserved = yield* Deferred.make<void>();
456475 const terminate = Deferred.succeed(exited, ChildProcessSpawner.ExitCode(1)).pipe(
457476 Effect.asVoid,
458477 );
@@ -461,22 +480,26 @@ it.layer(NodeServices.layer)("Antigravity installation", (it) => { · 2 unmodified lines
461480 pid: ChildProcessSpawner.ProcessId(helper ? 1 : 2),
462481 exitCode: helper
463482 ? Effect.succeed(ChildProcessSpawner.ExitCode(0))
464− : Deferred.await(exited),
483+ : Deferred.await(exited).pipe(
484+ Effect.tap(() => Deferred.succeed(exitObserved, undefined)),
485+ ),
465486 isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)),
466487 kill: () => terminate,
467488 unref: Effect.succeed(Effect.void),
468489 stdin: Sink.drain,
469490 stdout: helper
470491 ? Stream.empty
471− : Stream.fromEffect(Deferred.await(exited)).pipe(Stream.flatMap(() => Stream.empty)),
492+ : Stream.fromEffect(terminate).pipe(Stream.flatMap(() => Stream.empty)),
472493 stderr: helper
473494 ? Stream.make(
474495 encoder.encode(
475496 `${ANTIGRAVITY_AUTH_BROWSER_MARKER}${encodeJsonString(command.args.at(-1))}\n`,
476497 ),
477498 )
478− : Stream.fromIterable(testCase.stderr.map((line) => encoder.encode(line))).pipe(
479− Stream.ensuring(terminate),
499+ : Stream.fromEffect(Deferred.await(exitObserved)).pipe(
500+ Stream.flatMap(() =>
501+ Stream.fromIterable(testCase.stderr.map((line) => encoder.encode(line))),
502+ ),
480503 ),
481504 all: Stream.empty,
482505 getInputFd: () => Sink.drain,
@@ -494,6 +517,7 @@ it.layer(NodeServices.layer)("Antigravity installation", (it) => { · 11 unmodified lines
494517 expect(state.message).toContain(testCase.expected);
495518 expect(state.message?.length).toBeLessThan(4_200);
496519 expect(state.message).not.toContain("discarded prefix");
520+ expect(state.message).not.toContain("secret-authorization-tail");
497521 yield* Deferred.await(stagingReleased);
498522 yield* expectPreviousRelease(installation);
499523 }),
apps/server/src/provider/acp/AcpSessionRuntime.ts+26−7
@@ -104,6 +104,8 @@ export interface AcpSessionRuntimeOptions { · 103 unmodified lines
104104 ) => EffectAcpSchema.SessionNotification;
105105 /** Receives bounded stderr chunks. Redact secrets before logging. A failure closes the runtime. */
106106 readonly onStderr?: (text: string) => Effect.Effect<void, EffectAcpErrors.AcpError>;
107+ /** Flushes diagnostics when stderr ends. Failed initialization drains an exited child's stderr first. */
108+ readonly onStderrEnd?: Effect.Effect<void>;
107109 readonly requestLogger?: (event: AcpSessionRequestLogEvent) => Effect.Effect<void, never>;
108110 readonly protocolLogging?: {
109111 readonly logIncoming?: boolean;
@@ -449,13 +451,16 @@ export const make = ( · 339 unmodified lines
449451 ),
450452 );
451453
452− yield* child.stderr.pipe(
454+ const stderrFiber = yield* child.stderr.pipe(
453455 Stream.decodeText(),
454− Stream.runForEach((chunk) =>
455− (options.onStderr
456− ? options.onStderr(chunk.slice(-maxStderrChunkLength))
457− : Effect.void
458− ).pipe(
456+ Stream.runForEach(
457+ Effect.fn("AcpSessionRuntime.handleStderr")(
458+ function* (chunk: string) {
459+ if (!options.onStderr) return;
460+ for (let offset = 0; offset < chunk.length; offset += maxStderrChunkLength) {
461+ yield* options.onStderr(chunk.slice(offset, offset + maxStderrChunkLength));
462+ }
463+ },
459464 Effect.catch((error) =>
460465 Effect.gen(function* () {
461466 yield* Deferred.fail(stderrFailure, error);
@@ -465,6 +470,7 @@ export const make = ( · 3 unmodified lines
465470 ),
466471 ),
467472 ),
473+ Effect.ensuring(options.onStderrEnd ?? Effect.void),
468474 Effect.ignore,
469475 Effect.forkIn(runtimeScope),
470476 );
@@ -971,7 +977,20 @@ export const make = ( · 500 unmodified lines
971977 handleUnknownExtNotification: acp.handleUnknownExtNotification,
972978 handleExtRequest: acp.handleExtRequest,
973979 handleExtNotification: acp.handleExtNotification,
974− initialize: () => ensureConnected.pipe(Effect.andThen(sendInitialize)),
980+ initialize: () =>
981+ ensureConnected.pipe(
982+ Effect.andThen(sendInitialize),
983+ Effect.onError(() =>
984+ Effect.gen(function* () {
985+ if (!options.onStderrEnd) return;
986+ const running = yield* child.isRunning.pipe(Effect.orElseSucceed(() => true));
987+ if (running) return;
988+ // A descendant may inherit stderr, so a closed process cannot wait forever for EOF.
989+ yield* Fiber.await(stderrFiber).pipe(Effect.timeoutOption("1 second"));
990+ yield* options.onStderrEnd;
991+ }),
992+ ),
993+ ),
975994 start: () => start,
976995 getEvents: () => Stream.fromQueue(eventQueue),
977996 drainEvents,
apps/server/src/provider/acp/AntigravityAcpSupport.test.ts+69−0
@@ -8,8 +8,15 @@ import { · 7 unmodified lines
88 type RuntimeMode,
99 } from "@t3tools/contracts";
1010 import * as Effect from "effect/Effect";
11+import * as Deferred from "effect/Deferred";
12+import * as Exit from "effect/Exit";
13+import * as Fiber from "effect/Fiber";
1114 import * as FileSystem from "effect/FileSystem";
1215 import * as Path from "effect/Path";
16+import * as Sink from "effect/Sink";
17+import * as Stream from "effect/Stream";
18+import * as TestClock from "effect/testing/TestClock";
19+import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
1320 import * as EffectAcpErrors from "effect-acp/errors";
1421 import type * as EffectAcpSchema from "effect-acp/schema";
1522
@@ -19,8 +26,70 @@ import { · 3 unmodified lines
1926 antigravityPermissionMode,
2027 applyAntigravityAcpModelSelection,
2128 buildAntigravityPrompt,
29+ makeAntigravityAcpRuntime,
2230 } from "./AntigravityAcpSupport.ts";
2331
32+it.effect.each([false, true])(
33+ "drains exited validation stderr with inherited pipe: %s",
34+ (inherited) =>
35+ Effect.gen(function* () {
36+ const releaseStderr = yield* Deferred.make<void>();
37+ const draining = yield* Deferred.make<void>();
38+ const completed = yield* Deferred.make<void>();
39+ const diagnostics: string[] = [];
40+ const childProcessSpawner = ChildProcessSpawner.make(() =>
41+ Effect.succeed(
42+ ChildProcessSpawner.makeHandle({
43+ pid: ChildProcessSpawner.ProcessId(1),
44+ exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(1)),
45+ isRunning: Deferred.succeed(draining, undefined).pipe(Effect.as(false)),
46+ kill: () => Effect.void,
47+ unref: Effect.succeed(Effect.void),
48+ stdin: Sink.drain,
49+ stdout: Stream.empty,
50+ stderr: inherited
51+ ? Stream.make(new TextEncoder().encode("final startup diagnostic")).pipe(
52+ Stream.concat(
53+ Stream.fromEffect(Deferred.await(releaseStderr)).pipe(
54+ Stream.flatMap(() => Stream.empty),
55+ ),
56+ ),
57+ )
58+ : Stream.fromEffect(Deferred.await(releaseStderr)).pipe(
59+ Stream.map(() => new TextEncoder().encode("final startup diagnostic")),
60+ ),
61+ all: Stream.empty,
62+ getInputFd: () => Sink.drain,
63+ getOutputFd: () => Stream.empty,
64+ }),
65+ ),
66+ );
67+ const runtime = yield* makeAntigravityAcpRuntime({
68+ spawn: { command: "antigravity-fixture", args: [] },
69+ cwd: "/fixture",
70+ childProcessSpawner,
71+ clientInfo: { name: "test", version: "1" },
72+ onDiagnostic: (message) => Effect.sync(() => void diagnostics.push(message)),
73+ });
74+ const initialization = yield* runtime.initialize().pipe(
75+ Effect.exit,
76+ Effect.tap(() => Deferred.succeed(completed, undefined)),
77+ Effect.forkScoped,
78+ );
79+ expect(
80+ yield* Effect.raceFirst(
81+ Deferred.await(draining).pipe(Effect.as("draining")),
82+ Deferred.await(completed).pipe(Effect.as("completed")),
83+ ),
84+ ).toBe("draining");
85+ expect(yield* Deferred.isDone(completed)).toBe(false);
86+ if (inherited) yield* TestClock.adjust("1 second");
87+ else yield* Deferred.succeed(releaseStderr, undefined);
88+ expect(Exit.isFailure(yield* Fiber.join(initialization))).toBe(true);
89+ expect(diagnostics).toEqual(["final startup diagnostic"]);
90+ }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
91+);
92+
2493 const modelConfig = {
2594 id: "model",
2695 name: "Model",
apps/server/src/provider/acp/AntigravityAcpSupport.ts+7−4
@@ -31,6 +31,7 @@ export interface AntigravityAcpRuntimeInput extends Omit< · 30 unmodified lines
3131 | "cancelBehavior"
3232 | "clientCapabilities"
3333 | "onStderr"
34+ | "onStderrEnd"
3435 | "resumeMethod"
3536 | "transformSessionUpdate"
3637 | "transformStdout"
@@ -59,6 +60,10 @@ export const makeAntigravityAcpRuntime = Effect.fn("makeAntigravityAcpRuntime")( · 22 unmodified lines
5960 EffectAcpErrors.AcpError,
6061 Crypto.Crypto | Scope.Scope
6162 > {
63+ const handleStderr = makeAntigravityStderrHandler({
64+ ...(input.onAuthorizationUrl ? { onAuthorizationUrl: input.onAuthorizationUrl } : {}),
65+ ...(input.onDiagnostic ? { onDiagnostic: input.onDiagnostic } : {}),
66+ });
6267 const context = yield* Layer.build(
6368 AcpSessionRuntime.layer({
6469 ...input,
@@ -75,10 +80,8 @@ export const makeAntigravityAcpRuntime = Effect.fn("makeAntigravityAcpRuntime")( · 10 unmodified lines
7580 transformStdout: makeAntigravityStdoutTransform(
7681 input.onAuthorizationUrl ? { onAuthorizationUrl: input.onAuthorizationUrl } : {},
7782 ),
78− onStderr: makeAntigravityStderrHandler({
79− ...(input.onAuthorizationUrl ? { onAuthorizationUrl: input.onAuthorizationUrl } : {}),
80− ...(input.onDiagnostic ? { onDiagnostic: input.onDiagnostic } : {}),
81− }),
83+ onStderr: handleStderr,
84+ ...(input.onDiagnostic ? { onStderrEnd: handleStderr.flushDiagnostics } : {}),
8285 transformSessionUpdate: normalizeAntigravitySessionUpdate,
8386 }).pipe(
8487 Layer.provide(
apps/server/src/provider/antigravityAuthSupport.test.ts+34−0
@@ -391,6 +391,38 @@ describe("Antigravity stdout compatibility", () => { · 390 unmodified lines
391391 });
392392
393393 describe("Antigravity stderr compatibility", () => {
394+ it.effect.each([
395+ { name: "native", prefix: ANTIGRAVITY_AUTH_STDOUT_PREFIX },
396+ { name: "browser", prefix: ANTIGRAVITY_AUTH_BROWSER_MARKER },
397+ { name: "indented native", prefix: ` ${ANTIGRAVITY_AUTH_STDOUT_PREFIX}` },
398+ { name: "indented browser", prefix: ` ${ANTIGRAVITY_AUTH_BROWSER_MARKER}` },
399+ { name: "logged native", prefix: `INFO ${ANTIGRAVITY_AUTH_STDOUT_PREFIX}` },
400+ { name: "logged browser", prefix: `INFO ${ANTIGRAVITY_AUTH_BROWSER_MARKER}` },
401+ {
402+ name: "long log prefix",
403+ prefix: `${"logged prefix ".repeat(4_096)}${ANTIGRAVITY_AUTH_BROWSER_MARKER}`,
404+ },
405+ {
406+ name: "long indentation",
407+ prefix: `${" ".repeat(32_768)}${ANTIGRAVITY_AUTH_BROWSER_MARKER}`,
408+ },
409+ ])("discards oversized auth messages across fragment boundaries: $name", ({ prefix }) =>
410+ Effect.gen(function* () {
411+ const diagnostics: string[] = [];
412+ const handleStderr = makeAntigravityStderrHandler({
413+ onDiagnostic: (message) => Effect.sync(() => void diagnostics.push(message)),
414+ });
415+ yield* handleStderr(prefix.slice(0, 12));
416+ yield* handleStderr(`${prefix.slice(12)}${"x".repeat(32_768)}`);
417+ yield* handleStderr("secret-authorization-tail");
418+ yield* handleStderr.flushDiagnostics;
419+ yield* handleStderr("\nlast startup diagnostic");
420+ yield* handleStderr.flushDiagnostics;
421+ yield* handleStderr.flushDiagnostics;
422+ expect(diagnostics).toEqual(["last startup diagnostic"]);
423+ }),
424+ );
425+
394426 it.effect("keeps sign-in URLs out of collected native diagnostics", () =>
395427 Effect.gen(function* () {
396428 const diagnostics: string[] = [];
@@ -407,6 +439,8 @@ describe("Antigravity stderr compatibility", () => { · 10 unmodified lines
407439 yield* handleStderr(` ${ANTIGRAVITY_AUTH_STDOUT_PREFIX}${authorizationUrl}\n`);
408440 yield* handleStderr(` ${ANTIGRAVITY_AUTH_BROWSER_MARKER}${authorizationUrl}\n`);
409441 yield* handleStderr(`${ANTIGRAVITY_AUTH_BROWSER_MARKER}${authorizationUrl}\n`);
442+ yield* handleStderr(`INFO ${ANTIGRAVITY_AUTH_STDOUT_PREFIX}${authorizationUrl}\n`);
443+ yield* handleStderr(`INFO ${ANTIGRAVITY_AUTH_BROWSER_MARKER}${authorizationUrl}\n`);
410444 expect(diagnostics).toEqual(["Failed to create an AF_INET6 socket."]);
411445 }),
412446 );
apps/server/src/provider/antigravityAuthSupport.ts+48−14
@@ -470,23 +470,20 @@ export function makeAntigravityStderrHandler( · 469 unmodified lines
470470 } = {},
471471 ) {
472472 let pending = "";
473+ let oversizedLine: "diagnostic" | "discard" | undefined;
474+ const isAuthorizationLine = (message: string) =>
475+ message.includes(ANTIGRAVITY_AUTH_STDOUT_PREFIX) ||
476+ message.includes(ANTIGRAVITY_AUTH_BROWSER_MARKER);
473477 const handleLine = (line: string) => {
474478 const message = line.endsWith("\r") ? line.slice(0, -1) : line;
475− if (message.length > maxBrowserHelperLineLength) {
476− return Effect.void;
477− }
478479 const url = message.startsWith(ANTIGRAVITY_AUTH_STDOUT_PREFIX)
479480 ? Effect.succeed(message.slice(ANTIGRAVITY_AUTH_STDOUT_PREFIX.length))
480481 : message.startsWith(ANTIGRAVITY_AUTH_BROWSER_MARKER)
481482 ? decodeBrowserHelperUrl(message.slice(ANTIGRAVITY_AUTH_BROWSER_MARKER.length))
482483 : undefined;
483484 if (url === undefined) {
484− // Malformed or indented auth messages must not enter installation diagnostics.
485− const trimmed = message.trimStart();
486− if (
487− trimmed.startsWith(ANTIGRAVITY_AUTH_STDOUT_PREFIX) ||
488− trimmed.startsWith(ANTIGRAVITY_AUTH_BROWSER_MARKER)
489− ) {
485+ // Malformed or log-prefixed auth messages must not enter installation diagnostics.
486+ if (isAuthorizationLine(message)) {
490487 return Effect.void;
491488 }
492489 return input.onDiagnostic?.(message) ?? Effect.void;
@@ -503,10 +500,47 @@ export function makeAntigravityStderrHandler( · 10 unmodified lines
503500 );
504501 };
505502
506− return Effect.fn("antigravityAuthSupport.handleStderr")(function* (text: string) {
507− const lines = `${pending}${text}`.split("\n");
508− pending = lines.pop() ?? "";
509− if (pending.length > maxBrowserHelperLineLength) pending = "";
510− yield* Effect.forEach(lines, handleLine, { discard: true });
503+ const finishLine = () => {
504+ const message = pending;
505+ const kind = oversizedLine;
506+ pending = "";
507+ oversizedLine = undefined;
508+ if (kind === "discard") {
509+ return Effect.void;
510+ }
511+ if (kind === "diagnostic") return input.onDiagnostic?.(message) ?? Effect.void;
512+ return handleLine(message);
513+ };
514+
515+ const handleStderr = Effect.fn("antigravityAuthSupport.handleStderr")(function* (text: string) {
516+ let offset = 0;
517+ while (offset < text.length) {
518+ const newline = text.indexOf("\n", offset);
519+ const end = newline === -1 ? text.length : newline;
520+ if (oversizedLine !== "discard") {
521+ pending += text.slice(offset, end);
522+ if (pending.length > maxBrowserHelperLineLength) {
523+ oversizedLine =
524+ isAuthorizationLine(pending) || pending.trimStart().length === 0
525+ ? "discard"
526+ : "diagnostic";
527+ pending = oversizedLine === "discard" ? "" : pending.slice(-maxBrowserHelperLineLength);
528+ }
529+ }
530+ if (newline !== -1) yield* finishLine();
531+ offset = end + 1;
532+ }
533+ });
534+
535+ return Object.assign(handleStderr, {
536+ flushDiagnostics: Effect.suspend(() => {
537+ if (pending.length === 0 || oversizedLine === "discard" || isAuthorizationLine(pending)) {
538+ return Effect.void;
539+ }
540+ const message = pending.endsWith("\r") ? pending.slice(0, -1) : pending;
541+ pending = "";
542+ oversizedLine = undefined;
543+ return input.onDiagnostic?.(message) ?? Effect.void;
544+ }),
511545 });
512546 }