fix: diagnose content review failures without blaming connectivity (#226)

Author: hetaoBackend

packages/local-runtime-v2/src/service/turn-system/agent-host/runner/local-agent-turn-runner.ts

@@ -1,3 +1,4 @@
+import { logger } from '../../../../infra/logging/index.js';
 import {
   PiTurnRunner,
   type LLMModelConfig,
@@ -633,6 +634,16 @@ async function settleNetworkStop(
   safetyWriter: LocalOutputSafetyEventWriter,
 ): Promise<void> {
   if (safetyWriter.terminalFailed) return;
+  logger.warn(
+    {
+      sessionId: input.sessionId,
+      turnId: input.turnId,
+      reviewOutcome: safetyWriter.reviewStopVariant === 'auth' ? 'auth_error' : 'unavailable',
+      outputSuppressed: true,
+      terminalStatus: 'completed',
+    },
+    '[content-safety] output review stopped turn',
+  );
   state.networkStopped = true;
   state.retractionVariant = safetyWriter.reviewStopVariant;
   state.approvedPartial = safetyWriter.getApprovedPartial();

packages/local-runtime/src/content-safety/api-v2.ts

@@ -16,6 +16,7 @@ import {
   managedBackendRoutingHeaders,
   type LocalRuntimeRoutingContext,
 } from '../runtime/routing-headers.js';
+import { createSafetyFailureReporter } from './diagnostics.js';
 import { resolveSafetyApiBase } from './api-base.js';
 
 /** biz-gateway uses V2 numeric scenes with its desktop HTTP envelope. */
@@ -50,26 +51,44 @@ export async function callLocalSafetyCheckV2(input: {
   const buildEnv = (input.buildEnv ?? getRuntimeBuildEnv)();
   const region = (input.region ?? getRuntimeRegion)();
   const token = input.authContext?.accessToken?.trim() || process.env.MAVIS_ACCESS_TOKEN?.trim();
-  const result = await postSafetyCheckV2({
-    url: `${resolveSafetyApiBase(region, buildEnv, input.testBaseURL)}/mavis/api/v2/content?require_auth=true`,
-    headers: {
-      'User-Agent': 'MiniMaxAgent',
-      ...(token ? { Authorization: `Bearer ${token}` } : {}),
-      ...managedBackendRoutingHeaders(input.routingContext, input.buildEnv ? buildEnv : undefined),
-    },
-    body: {
-      scene: input.request.scene,
-      ...(input.request.content_text !== undefined
-        ? { content_text: input.request.content_text }
-        : {}),
-      ...(input.request.sessionId !== undefined ? { sessionId: input.request.sessionId } : {}),
-      ...(input.request.image_url !== undefined ? { image_url: input.request.image_url } : {}),
-      ...(input.request.files !== undefined ? { files: input.request.files } : {}),
-    },
-    fetchImpl: input.fetchImpl,
-    signal: input.signal,
-  });
-  return parseLocalSafetyCheckV2Result(result);
+  const url = `${resolveSafetyApiBase(region, buildEnv, input.testBaseURL)}/mavis/api/v2/content?require_auth=true`;
+  const reportFailure = createSafetyFailureReporter(url, 'v2', input.request.scene);
+  try {
+    const result = await postSafetyCheckV2({
+      url,
+      headers: {
+        'User-Agent': 'MiniMaxAgent',
+        ...(token ? { Authorization: `Bearer ${token}` } : {}),
+        ...managedBackendRoutingHeaders(
+          input.routingContext,
+          input.buildEnv ? buildEnv : undefined,
+        ),
+      },
+      body: {
+        scene: input.request.scene,
+        ...(input.request.content_text !== undefined
+          ? { content_text: input.request.content_text }
+          : {}),
+        ...(input.request.sessionId !== undefined ? { sessionId: input.request.sessionId } : {}),
+        ...(input.request.image_url !== undefined ? { image_url: input.request.image_url } : {}),
+        ...(input.request.files !== undefined ? { files: input.request.files } : {}),
+      },
+      fetchImpl: input.fetchImpl,
+      signal: input.signal,
+    });
+    return parseLocalSafetyCheckV2Result(result);
+  } catch (error) {
+    reportFailure(
+      error instanceof SafetyCheckV2Error
+        ? {
+            failureKind: error.kind,
+            statusCode: error.statusCode,
+            transportKind: error.transportKind,
+          }
+        : { failureKind: 'internal' },
+    );
+    throw error;
+  }
 }
 
 function parseLocalSafetyCheckV2Result(body: Record<string, unknown>): LocalSafetyCheckV2Result {

packages/local-runtime/src/content-safety/api.ts

@@ -1,3 +1,8 @@
+import {
+  classifySafetyTransportError,
+  isSafetyCheckV2Record,
+} from "@mavis/shared/safety-check-v2";
+import { createSafetyFailureReporter } from "./diagnostics.js";
 import {
   getRuntimeBuildEnv,
   getRuntimeRegion,
@@ -8,7 +13,6 @@ import {
 import { callTurnSafetyApi } from "./turn-api.js";
 import { resolveSafetyApiBase } from "./api-base.js";
 
-import { logger } from "../common/logger.js";
 import type { LocalRuntimeAuthContext } from "../runtime/model-resolver.js";
 import {
   managedBackendRoutingHeaders,
@@ -126,12 +130,7 @@ export async function callSafetyApi(input: {
   const accessToken =
     input.authContext?.accessToken?.trim() ||
     process.env.MAVIS_ACCESS_TOKEN?.trim();
-  // TODO(shared-oauth): Remove this local-only diagnostic after the content-safety
-  // resource server accepts mcode-public Bearer tokens in the joint test environment.
-  // Never add the token, request body/content, or response body to these fields.
-  const logLocalUpstream =
-    resolvedBuildEnv === "dev" || resolvedBuildEnv === "test";
-  const startedAt = Date.now();
+  const reportFailure = createSafetyFailureReporter(url, "v1", input.scene);
   let response: Response;
   try {
     response = await input.fetchImpl(url, {
@@ -162,20 +161,10 @@ export async function callSafetyApi(input: {
       signal: AbortSignal.timeout(CONTENT_REVIEW_TIMEOUT_MS),
     });
   } catch (error) {
-    if (logLocalUpstream) {
-      logger.info(
-        {
-          url,
-          method: "POST",
-          status: null,
-          scene: input.scene,
-          hasBearer: Boolean(accessToken),
-          durationMs: Date.now() - startedAt,
-          errorType: error instanceof Error ? error.name : typeof error,
-        },
-        "[content-safety] upstream transport failure",
-      );
-    }
+    reportFailure({
+      failureKind: "transport",
+      transportKind: classifySafetyTransportError(error),
+    });
     // fetch threw (offline / DNS / connection refused / TLS) or the 10s timeout
     // fired: we never reached a verdict. No gateway confirmation → fail closed.
     return {
@@ -185,26 +174,21 @@ export async function callSafetyApi(input: {
     };
   }
 
-  if (logLocalUpstream) {
-    logger.info(
-      {
-        url,
-        method: "POST",
-        status: response.status,
-        scene: input.scene,
-        hasBearer: Boolean(accessToken),
-        durationMs: Date.now() - startedAt,
-        upstreamTraceId:
-          response.headers.get("trace-id") ??
-          response.headers.get("x-trace-id") ??
-          response.headers.get("x-request-id") ??
-          null,
-      },
-      "[content-safety] upstream response",
-    );
+  let text: string;
+  try {
+    text = await response.text();
+  } catch (error) {
+    reportFailure({
+      failureKind: "transport",
+      transportKind: classifySafetyTransportError(error),
+      statusCode: response.status,
+    });
+    return {
+      pass: false,
+      reason: "Service response could not be read",
+      errorKind: "local_error",
+    };
   }
-
-  const text = await response.text();
   if (!response.ok) {
     // Non-2xx from the gateway, split by class so an auth failure can never be a
     // bypass:
@@ -215,6 +199,10 @@ export async function callSafetyApi(input: {
     //          acceptable, so there is NO valid verdict on our side → `local_error`
     //          (fail closed). Treating 401/403 as `api_error` would let an
     //          unauthenticated user bypass every content gate.
+    reportFailure({
+      failureKind: response.status === 401 || response.status === 403 ? "auth" : "http",
+      statusCode: response.status,
+    });
     const error = readErrorText(text);
     return {
       pass: false,
@@ -240,6 +228,7 @@ export async function callSafetyApi(input: {
   try {
     data = JSON.parse(text) as typeof data;
   } catch {
+    reportFailure({ failureKind: "response", statusCode: response.status });
     // 2xx but the body is not JSON — we cannot read a verdict. Treat as no
     // gateway confirmation → fail closed on the input path.
     return {
@@ -249,8 +238,9 @@ export async function callSafetyApi(input: {
     };
   }
 
-  const pass = data.pass ?? data.safe;
+  const pass = isSafetyCheckV2Record(data) ? data.pass ?? data.safe : undefined;
   if (typeof pass !== "boolean") {
+    reportFailure({ failureKind: "response", statusCode: response.status });
     // 2xx JSON but no boolean pass/safe field — again no usable verdict → fail
     // closed on the input path.
     return {
@@ -262,6 +252,9 @@ export async function callSafetyApi(input: {
   const activeRegion = region();
   const upstreamCode =
     typeof data.errorCode === "number" ? data.errorCode : undefined;
+  if (!pass && upstreamCode === UPSTREAM_CODE.SERVICE_UNAVAILABLE) {
+    reportFailure({ failureKind: "upstream", statusCode: upstreamCode });
+  }
   return {
     pass,
     reason:

packages/local-runtime/src/content-safety/diagnostics.ts

@@ -0,0 +1,27 @@
+import { randomUUID } from 'node:crypto';
+import type { SafetyTransportKind } from '@mavis/shared/safety-check-v2';
+import { logger } from '../common/logger.js';
+
+/** One bounded record per failed check, in every build. Never log raw upstream data. */
+export function createSafetyFailureReporter(url: string, apiVersion: 'v1' | 'v2', scene: number) {
+  const startedAt = Date.now();
+  const reviewId = randomUUID();
+  const endpointHost = new URL(url).hostname;
+  return (failure: {
+    failureKind: 'transport' | 'auth' | 'http' | 'response' | 'upstream' | 'internal';
+    transportKind?: SafetyTransportKind;
+    statusCode?: number;
+  }): void => {
+    logger.warn(
+      {
+        reviewId,
+        endpointHost,
+        apiVersion,
+        scene,
+        durationMs: Date.now() - startedAt,
+        ...failure,
+      },
+      '[content-safety] review failed',
+    );
+  };
+}

packages/local-runtime/src/runtime/host.ts

@@ -1,3 +1,4 @@
+import { logger } from "../common/logger.js";
 import type { LocalEventWriter } from "../events/sink.js";
 import { PiTurnRunner } from "@mavis/agent-core/pi-turn-runner";
 import { RuntimeEventStatus, RuntimeEventType } from "@mavis/protocol";
@@ -437,6 +438,16 @@ export class LocalRuntimeHost {
         // writer, so keep it authoritative and do not append a synthetic
         // COMPLETED/network-notice outcome.
         if (safetyWriter.terminalFailed) break;
+        logger.warn(
+          {
+            sessionId: input.sessionId,
+            turnId: input.turnId,
+            reviewOutcome: "unavailable",
+            outputSuppressed: true,
+            terminalStatus: "completed",
+          },
+          "[content-safety] output review stopped turn",
+        );
         networkStopped = true;
         retractionVariant = "network";
         // Capture the approved prefix so the orchestration can rewrite pi history

packages/local-runtime/test/unit/content-safety-api-fail-policy.test.ts

@@ -276,46 +276,70 @@ describe('callSafetyApi fail-policy classification', () => {
     expect(headers.get('Token')).toBeNull();
   });
 
-  it('logs a sanitized upstream response in local test environments', async () => {
-    const log = vi.spyOn(logger, 'info').mockImplementation(() => {});
-    const fetchImpl = vi.fn(async () =>
-      new Response(JSON.stringify({ code: 401, message: 'auth failed' }), {
-        status: 401,
-        headers: {
-          'Content-Type': 'application/json',
-          'Trace-Id': 'trace-content-safety-test',
-        },
-      }),
-    ) as unknown as typeof fetch;
-
-    await callSafetyApi({
-      content: 'must-not-appear-in-log',
-      scene: SAFETY_SCENE.UserInput,
-      authContext: { accessToken: 'must-not-appear-token' },
-      fetchImpl,
-      region,
-      buildEnv,
-    });
-
-    expect(log).toHaveBeenCalledWith(
-      expect.objectContaining({
-        url: 'https://matrix-test.example.invalid/mavis/api/v1/content',
-        method: 'POST',
-        status: 401,
+  it.each([
+    ['auth', () => jsonResponse({ error: 'private-response' }, 401), 401],
+    ['http', () => jsonResponse({ error: 'private-response' }, 503), 503],
+    ['response', () => new Response('private-response'), 200],
+    ['response', () => jsonResponse(null), 200],
+    ['response', () => jsonResponse({ secret: 'private-response' }), 200],
+    [
+      'transport',
+      () => {
+        throw new TypeError('private-error https://secret.invalid/?token=private-token', {
+          cause: { code: 'ENOTFOUND' },
+        });
+      },
+      undefined,
+    ],
+    [
+      'transport',
+      () =>
+        new Response(
+          new ReadableStream({
+            start(controller) {
+              controller.error(new DOMException('private-error', 'TimeoutError'));
+            },
+          }),
+        ),
+      200,
+    ],
+  ] as const)(
+    'logs redacted %s failures in production',
+    async (failureKind, response, statusCode) => {
+      const log = vi.spyOn(logger, 'warn').mockImplementation(() => {});
+      const result = await callSafetyApi({
+        content: 'private-content',
         scene: SAFETY_SCENE.UserInput,
-        hasBearer: true,
-        durationMs: expect.any(Number),
-        upstreamTraceId: 'trace-content-safety-test',
-      }),
-      '[content-safety] upstream response',
-    );
-    const serialized = JSON.stringify(log.mock.calls);
-    expect(serialized).not.toContain('must-not-appear-token');
-    expect(serialized).not.toContain('must-not-appear-in-log');
-  });
+        authContext: { accessToken: 'private-token' },
+        attachments: [{ fileUrl: 'https://secret.invalid/private-file', fileName: 'private-name' }],
+        fetchImpl: async () => response(),
+        region,
+        buildEnv: () => 'prod',
+      });
+      expect(result.pass).toBe(false);
+      expect(reviewBlocks(result)).toBe(failureKind !== 'http');
+      expect(log).toHaveBeenCalledOnce();
+      expect(log).toHaveBeenCalledWith(
+        expect.objectContaining({
+          reviewId: expect.any(String),
+          endpointHost: 'agent.minimax.cn',
+          apiVersion: 'v1',
+          scene: SAFETY_SCENE.UserInput,
+          durationMs: expect.any(Number),
+          failureKind,
+          ...(statusCode === undefined ? { transportKind: 'dns' } : { statusCode }),
+        }),
+        '[content-safety] review failed',
+      );
+      const serialized = JSON.stringify(log.mock.calls);
+      expect(serialized).not.toContain('private-');
+      expect(serialized).not.toContain('secret.invalid');
+      expect(serialized).not.toContain('https://');
+    },
+  );
 
-  it('does not emit the temporary upstream diagnostic in production', async () => {
-    const log = vi.spyOn(logger, 'info').mockImplementation(() => {});
+  it('does not log successful reviews in production', async () => {
+    const log = vi.spyOn(logger, 'warn').mockImplementation(() => {});
 
     await callSafetyApi({
       content: 'hello',