fix(service): add safe nonblocking startup-health snapshots

src/server/startup-health-cache.ts

@@ -50,6 +50,23 @@ export interface StartupHealthCacheDeps {
   ) => Promise<StartupHealth | null>;
 }
 
+/**
+ * Return the last completed probe immediately and refresh it in the background.
+ *
+ * Settings are consumed by several dashboard controls. They must not block on a
+ * Windows service-manager probe; the dedicated /api/startup-health route owns
+ * the fresh, bounded diagnostic read.
+ */
+export function getStartupHealthSnapshot(
+  config: Pick<OcxConfig, "codexAutoStart">,
+  deps: StartupHealthCacheDeps = {},
+): StartupHealth {
+  const now = deps.now ?? Date.now;
+  if (cached && now() - cached.timestamp < CACHE_TTL_MS) return cached.value;
+  refreshInBackground(config, deps);
+  return cached ? markStartupHealthDiagnosticStale(cached.value) : conservativeFallback(config);
+}
+
 export function markStartupHealthDiagnosticStale(value: StartupHealth): StartupHealth {
   if (!value.localRoutingDependency) return { ...value, diagnosticStale: true };
   return {
@@ -134,15 +151,20 @@ function refreshInBackground(
 ): void {
   if (inflight) return;
   const startedGeneration = generation;
-  const probe = (deps.probe ?? runProbe)(config).then(value => {
-    if (startedGeneration === generation) {
-      cached = { timestamp: (deps.now ?? Date.now)(), value };
-    }
-    return value;
-  });
-  inflight = probe.finally(() => {
-    if (inflight === probe || startedGeneration === generation) inflight = null;
-  });
+  const probe: Promise<StartupHealth> = Promise.resolve()
+    .then(() => (deps.probe ?? runProbe)(config))
+    .then(value => {
+      if (startedGeneration === generation) {
+        cached = { timestamp: (deps.now ?? Date.now)(), value };
+      }
+      return value;
+    })
+    .catch(() => cached ? markStartupHealthDiagnosticStale(cached.value) : conservativeFallback(config))
+    .finally(() => {
+      // An invalidated probe must never clear the newer generation's flight.
+      if (inflight === probe) inflight = null;
+    });
+  inflight = probe;
 }
 
 /** Stale-while-revalidate: service-manager probes never hold open a model/UI request. */

src/vision/anthropic-describe.ts

@@ -10,6 +10,8 @@ import type { DescribeOutcome, VisionSettings } from "./describe";
 const ANTHROPIC_VISION_MAX_TOKENS = 1024;
 const ALLOWED_IMAGE_MIME = new Set(["image/png", "image/jpeg", "image/jpg", "image/webp", "image/gif"]);
 const MAX_IMAGE_BYTES = 20 * 1024 * 1024;
+/** Bound the sidecar SSE stream and its untrusted error body; the description is clamped downstream. */
+const MAX_SIDECAR_RESPONSE_BYTES = 64 * 1024;
 const DESCRIBE_INSTRUCTION =
   "You are a vision describer for a text-only model that cannot see the image. Describe the image " +
   "thoroughly and factually so that model can fully reason about it: transcribe any visible text " +
@@ -43,6 +45,34 @@ function buildImageBlock(imageUrl: string): { block?: AnthropicImageBlock; error
   return { error: "unsupported image URL scheme (expected data: or https:)" };
 }
 
+/** Read at most `MAX_SIDECAR_RESPONSE_BYTES` of an untrusted upstream body, then stop reading. */
+async function readBoundedText(res: Response): Promise<string> {
+  if (!res.body) return "";
+  const reader = res.body.getReader();
+  const decoder = new TextDecoder();
+  let out = "";
+  let seen = 0;
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      const remaining = MAX_SIDECAR_RESPONSE_BYTES - seen;
+      const accepted = value.byteLength <= remaining ? value : value.subarray(0, remaining);
+      seen += accepted.byteLength;
+      out += decoder.decode(accepted, { stream: true });
+      if (seen >= MAX_SIDECAR_RESPONSE_BYTES) {
+        try { void reader.cancel("vision sidecar error body byte limit reached").catch(() => undefined); }
+        catch { /* best-effort body teardown */ }
+        break;
+      }
+    }
+    out += decoder.decode();
+  } catch {
+    /* a failed error-body read must not mask the HTTP status we are about to report */
+  }
+  return out;
+}
+
 /** Fold Anthropic Messages text deltas into one description. Malformed frames are ignored. */
 export async function parseAnthropicVisionSSE(res: Response): Promise<DescribeOutcome> {
   if (!res.body) return { text: "", error: "anthropic vision sidecar returned no response body" };
@@ -52,6 +82,7 @@ export async function parseAnthropicVisionSSE(res: Response): Promise<DescribeOu
   const decoder = new TextDecoder();
   const reader = res.body.getReader();
   let buffer = "";
+  let responseBytes = 0;
 
   const processFrame = (rawFrame: string): void => {
     let dataLine = "";
@@ -76,12 +107,24 @@ export async function parseAnthropicVisionSSE(res: Response): Promise<DescribeOu
     for (;;) {
       const { done, value } = await reader.read();
       if (done) break;
-      buffer = (buffer + decoder.decode(value, { stream: true })).replace(/\r\n/g, "\n");
+      // Frames only fold on a `\n\n` separator, so an upstream that never emits one would grow
+      // `buffer` for the whole response. Accept a bounded prefix instead.
+      const remaining = MAX_SIDECAR_RESPONSE_BYTES - responseBytes;
+      const accepted = value.byteLength <= remaining ? value : value.subarray(0, remaining);
+      responseBytes += accepted.byteLength;
+      buffer = (buffer + decoder.decode(accepted, { stream: true })).replace(/\r\n/g, "\n");
       let separator: number;
       while ((separator = buffer.indexOf("\n\n")) !== -1) {
         processFrame(buffer.slice(0, separator));
         buffer = buffer.slice(separator + 2);
       }
+      if (responseBytes >= MAX_SIDECAR_RESPONSE_BYTES) {
+        // Keep the frames folded above, drop the unterminated tail, and do not wait on teardown.
+        try { void reader.cancel("vision sidecar response byte limit reached").catch(() => undefined); }
+        catch { /* best-effort body teardown */ }
+        buffer = "";
+        break;
+      }
     }
     buffer = (buffer + decoder.decode()).replace(/\r\n/g, "\n");
     if (buffer.trim()) processFrame(buffer);
@@ -164,7 +207,8 @@ export async function describeImageAnthropic(
       { abortSignal: linkedSignal.signal, label: "vision-sidecar-anthropic" },
     );
     if (!res.ok) {
-      const responseText = await res.text().catch(() => "");
+      // The body is untrusted and only feeds one auth-failure message, so read a bounded prefix.
+      const responseText = await readBoundedText(res);
       console.warn(`[vision] anthropic sidecar HTTP ${res.status} (${Date.now() - startedAt}ms)`);
       if (res.status === 401) {
         return { text: "", error: `anthropic vision sidecar auth failed: ${publicOAuthAuthenticationErrorMessage(new Error(responseText))}` };

tests/service/autostart-health.test.ts

@@ -3,7 +3,7 @@ import { deriveStartupHealth, formatStartupRoutingDetail, startupHealthSummary }
 import { unusedProxyWarningLines } from "../../src/cli/status";
 import { classifyCodexRouting, hasInjectedCodexRouting } from "../../src/codex/inject";
 import { handleManagementAPI } from "../../src/server/management-api";
-import { getCachedStartupHealth, invalidateStartupHealthCache, markStartupHealthDiagnosticStale } from "../../src/server/startup-health-cache";
+import { getCachedStartupHealth, getStartupHealthSnapshot, invalidateStartupHealthCache, markStartupHealthDiagnosticStale } from "../../src/server/startup-health-cache";
 import type { OcxConfig } from "../../src/types";
 
 const base = {
@@ -277,6 +277,89 @@ describe("Codex startup health", () => {
     await pendingProbe;
     invalidateStartupHealthCache();
   });
+
+  test("settings snapshot starts a probe without waiting for it", async () => {
+    invalidateStartupHealthCache();
+    let releaseProbe!: (value: ReturnType<typeof deriveStartupHealth>) => void;
+    const pendingProbe = new Promise<ReturnType<typeof deriveStartupHealth>>(resolve => {
+      releaseProbe = resolve;
+    });
+
+    const health = getStartupHealthSnapshot(
+      { codexAutoStart: true },
+      { probe: async () => pendingProbe },
+    );
+
+    expect(health.diagnosticStale).toBe(true);
+    releaseProbe(deriveStartupHealth({ ...base, routingKind: "native" }));
+    await pendingProbe;
+    invalidateStartupHealthCache();
+  });
+
+  test("snapshot preserves fresh protection and returns expired protection before a controlled probe settles", async () => {
+    invalidateStartupHealthCache();
+    let now = 1_000;
+    const config = { codexAutoStart: true };
+    const protectedHealth = deriveStartupHealth({ ...base, serviceInstalled: true, serviceViable: true, serviceEnabled: true, serviceRunning: true });
+    await getCachedStartupHealth(config, { now: () => now, probe: async () => protectedHealth, waitForProbe: probe => probe });
+    let calls = 0;
+    let release!: (value: typeof protectedHealth) => void;
+    const pending = new Promise<typeof protectedHealth>(resolve => { release = resolve; });
+    const deps = { now: () => now, probe: () => { calls += 1; return pending; }, waitForProbe: (probe: Promise<typeof protectedHealth>) => probe };
+    expect(getStartupHealthSnapshot(config, deps)).toBe(protectedHealth);
+    expect(calls).toBe(0);
+    now += 30_000;
+    const snapshot = getStartupHealthSnapshot(config, deps);
+    expect(snapshot).toMatchObject({ diagnosticStale: true, status: "at-risk", rebootSafe: false });
+    // Snapshot has returned while the manually controlled probe remains unresolved.
+    expect(getStartupHealthSnapshot(config, deps)).toEqual(snapshot);
+    const fresh = getCachedStartupHealth(config, deps);
+    const replacement = deriveStartupHealth({ ...base, routingKind: "custom-remote" });
+    release(replacement);
+    expect(await fresh).toBe(replacement);
+    expect(calls).toBe(1);
+    invalidateStartupHealthCache();
+  });
+
+  test.each(["reject", "throw"])("detached snapshot probe handles %s and permits a later retry", async (failure) => {
+    invalidateStartupHealthCache();
+    const config = { codexAutoStart: true };
+    const failed = getStartupHealthSnapshot(config, { probe: () => {
+      if (failure === "throw") throw new Error("controlled probe failure");
+      return Promise.reject(new Error("controlled probe failure"));
+    } });
+    expect(failed.diagnosticStale).toBe(true);
+    const settled = await getCachedStartupHealth(config, { waitForProbe: probe => probe });
+    expect(settled.diagnosticStale).toBe(true);
+    const replacement = deriveStartupHealth({ ...base, routingKind: "native" });
+    expect(await getCachedStartupHealth(config, { probe: async () => replacement, waitForProbe: probe => probe })).toBe(replacement);
+    invalidateStartupHealthCache();
+  });
+
+  test("invalidated probe cannot replace or clear a newer flight", async () => {
+    invalidateStartupHealthCache();
+    const config = { codexAutoStart: true };
+    type Health = ReturnType<typeof deriveStartupHealth>;
+    let oldRelease!: (value: Health) => void;
+    let newRelease!: (value: Health) => void;
+    const oldProbe = new Promise<Health>(resolve => { oldRelease = resolve; });
+    const newProbe = new Promise<Health>(resolve => { newRelease = resolve; });
+    getStartupHealthSnapshot(config, { probe: () => oldProbe });
+    const oldWait = getCachedStartupHealth(config, { waitForProbe: probe => probe });
+    invalidateStartupHealthCache();
+    getStartupHealthSnapshot(config, { probe: () => newProbe });
+    const newer = getCachedStartupHealth(config, { waitForProbe: probe => probe });
+    oldRelease(deriveStartupHealth(base));
+    await oldWait;
+    let spuriousCalls = 0;
+    getStartupHealthSnapshot(config, { probe: async () => { spuriousCalls += 1; return deriveStartupHealth(base); } });
+    const expected = deriveStartupHealth({ ...base, routingKind: "native" });
+    newRelease(expected);
+    expect(await newer).toBe(expected);
+    expect(getStartupHealthSnapshot(config)).toBe(expected);
+    expect(spuriousCalls).toBe(0);
+    invalidateStartupHealthCache();
+  });
 });
 import { ManagementRequest as Request } from "../helpers/management-auth";
 

tests/vision/vision-anthropic.test.ts

@@ -73,6 +73,34 @@ describe("Anthropic vision executor", () => {
     oauthAccessError = undefined;
   });
 
+  test.each([64 * 1024, 80 * 1024])("keeps only complete partial description frames at %i bytes without waiting for cancel", async (size) => {
+    const prefix = `data: ${JSON.stringify({ type: "content_block_delta", delta: { type: "text_delta", text: "partial 한글" } })}\n\n`;
+    const tail = `data: ${JSON.stringify({ type: "content_block_delta", delta: { type: "text_delta", text: "discard" } })}`;
+    const encoder = new TextEncoder();
+    const body = prefix + ":" + "x".repeat(64 * 1024 - encoder.encode(prefix + "\n\n" + tail).length - 1) + "\n\n" + tail;
+    let cancelled = false;
+    const out = await parseAnthropicVisionSSE(new Response(new ReadableStream<Uint8Array>({
+      start(controller) { controller.enqueue(encoder.encode(body + "z".repeat(size - 64 * 1024))); },
+      cancel() { cancelled = true; return new Promise<void>(() => {}); },
+    }, { highWaterMark: 0 })));
+    expect(cancelled).toBe(true);
+    expect(out).toEqual({ text: "partial 한글" });
+  });
+
+  test.each([401, 503])("bounds HTTP %i error bodies even when cancellation never settles", async (status) => {
+    let reads = 0;
+    let cancelled = false;
+    globalThis.fetch = (async () => new Response(new ReadableStream<Uint8Array>({
+      pull(controller) { reads += 1; controller.enqueue(new Uint8Array(4096).fill(120)); },
+      cancel() { cancelled = true; return new Promise<void>(() => {}); },
+    }, { highWaterMark: 0 }), { status })) as typeof fetch;
+    const out = await describeImageAnthropic(DATA_IMAGE, "high", "", "anthropic-vision-test", anthropicProvider, settings);
+    expect(reads).toBe(16);
+    expect(cancelled).toBe(true);
+    expect(out.error).toBe(status === 401
+      ? `anthropic vision sidecar auth failed: ${PUBLIC_OAUTH_ERROR}` : "anthropic vision sidecar HTTP 503");
+  });
+
   test("projects OAuth, upstream-auth, and transport failures onto safe replacement errors", async () => {
     oauthAccessError = new Error(`credential read failed at ${AUTH_ERROR_CANARY}`);
     const credentialFailure = await describeImageAnthropic(
@@ -225,6 +253,27 @@ describe("Anthropic vision executor", () => {
     expect(result).toEqual({ text: "first second" });
   });
 
+  test("an unterminated frame cannot buffer the stream without bound", async () => {
+    // A sidecar that never emits a frame separator: without a cap the parser accumulates the
+    // whole response in memory before it can fold anything.
+    let produced = 0;
+    let cancelled = false;
+    const chunk = new TextEncoder().encode(`data: {"filler":"${"x".repeat(64 * 1024)}"}`);
+    const body = new ReadableStream<Uint8Array>({
+      pull(c) {
+        if (produced > 8 * 1024 * 1024) { c.close(); return; }
+        produced += chunk.byteLength;
+        c.enqueue(chunk);
+      },
+      cancel() { cancelled = true; },
+    });
+    const out = await parseAnthropicVisionSSE(new Response(body, { status: 200 }));
+    expect(cancelled).toBe(true);
+    // The cap stops the read long before the producer would have finished on its own.
+    expect(produced).toBeLessThan(1024 * 1024);
+    expect(out.text).toBe("");
+  });
+
   test("malformed and terminal-error streams degrade to explicit errors", async () => {
     const malformed = await parseAnthropicVisionSSE(sseResponse(["{not-json", { type: "message_stop" }]));
     expect(malformed.text).toBe("");