authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 11:50:14-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 11:50:14-07:00
logd53237cbd14cbdc7fe6b1f1ec2f8aa224a3f7aff
tree0b8e0ff94344b48fd3c519d30a7504aa62245a94
parent0e718855b899b8535ee84e3721f4c5a0de0934cb
signature Signed by PGP key B5690EEEBB952194

[codex] sanitize ACP native event diagnostics (#3417)

Co-authored-by: codex <codex@users.noreply.github.com>

6 files changed, 273 insertions(+), 26 deletions(-)

apps/server/src/provider/Layers/EventNdjsonLogger.test.ts+36
......@@ -6,9 +6,13 @@ import * as NodePath from "node:path";
66import { ThreadId } from "@t3tools/contracts";
77import { assert, describe, it } from "@effect/vitest";
88import * as Effect from "effect/Effect";
9import * as Logger from "effect/Logger";
10import * as Schema from "effect/Schema";
911
1012import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
1113
14const encodeUnknownJson = Schema.encodeUnknownSync(Schema.UnknownFromJsonString);
15
1216function parseLogLine(line: string) {
1317 const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line);
1418 assert.notEqual(match, null);
......@@ -29,6 +33,38 @@ function parseLogLine(line: string) {
2933}
3034
3135describe("EventNdjsonLogger", () => {
36 it.effect("logs bounded diagnostics when an event cannot be serialized", () => {
37 const messages: Array<unknown> = [];
38 const logCapture = Logger.make<unknown, void>(({ message }) => {
39 if (Array.isArray(message)) {
40 messages.push(...message);
41 } else {
42 messages.push(message);
43 }
44 });
45 const secret = "secret-circular-event-value";
46
47 return Effect.gen(function* () {
48 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
49 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
50 const circular: Record<string, unknown> = { secret };
51 circular.self = circular;
52
53 try {
54 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native" });
55 assert.exists(logger);
56 if (!logger) return;
57 yield* logger.write(circular, ThreadId.make("thread-1"));
58
59 const serialized = encodeUnknownJson(messages);
60 assert.notInclude(serialized, secret);
61 assert.include(serialized, '"errorTag":"SchemaError"');
62 } finally {
63 NodeFS.rmSync(tempDir, { recursive: true, force: true });
64 }
65 }).pipe(Effect.provide(Logger.layer([logCapture], { mergeWithExisting: false })));
66 });
67
3268 it.effect("writes effect-style lines to thread-scoped files", () =>
3369 Effect.gen(function* () {
3470 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
apps/server/src/provider/Layers/EventNdjsonLogger.ts+9-8
......@@ -11,6 +11,7 @@ import * as NodePath from "node:path";
1111
1212import type { ThreadId } from "@t3tools/contracts";
1313import { RotatingFileSink } from "@t3tools/shared/logging";
14import { errorTag } from "@t3tools/shared/observability";
1415import * as Effect from "effect/Effect";
1516import * as Exit from "effect/Exit";
1617import * as Logger from "effect/Logger";
......@@ -31,8 +32,8 @@ export type EventNdjsonStream = "native" | "canonical" | "orchestration";
3132
3233export interface EventNdjsonLogger {
3334 readonly filePath: string;
34 write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void>;
35 close: () => Effect.Effect<void>;
35 write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void, never, never>;
36 close: () => Effect.Effect<void, never, never>;
3637}
3738
3839export interface EventNdjsonLoggerOptions {
......@@ -91,9 +92,9 @@ const toLogMessage = Effect.fn("toLogMessage")(function* (
9192): Effect.fn.Return<string | undefined> {
9293 return yield* encodeUnknownJsonString(event).pipe(
9394 Effect.catch((error) =>
94 logWarning("failed to serialize provider event log record", { error }).pipe(
95 Effect.as(undefined),
96 ),
95 logWarning("failed to serialize provider event log record", {
96 errorTag: errorTag(error),
97 }).pipe(Effect.as(undefined)),
9798 ),
9899 );
99100});
......@@ -124,7 +125,7 @@ const makeThreadWriter = Effect.fn("makeThreadWriter")(function* (input: {
124125 if (!sinkResult.ok) {
125126 yield* logWarning("failed to initialize provider thread log file", {
126127 filePath: input.filePath,
127 error: sinkResult.error,
128 errorTag: errorTag(sinkResult.error),
128129 });
129130 return undefined;
130131 }
......@@ -149,7 +150,7 @@ const makeThreadWriter = Effect.fn("makeThreadWriter")(function* (input: {
149150 if (!flushResult.ok) {
150151 yield* logWarning("provider event log batch flush failed", {
151152 filePath: input.filePath,
152 error: flushResult.error,
153 errorTag: errorTag(flushResult.error),
153154 });
154155 }
155156 }),
......@@ -187,7 +188,7 @@ export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function
187188 if (directoryReady !== true) {
188189 yield* logWarning("failed to create provider event log directory", {
189190 filePath,
190 error: directoryReady.error,
191 errorTag: errorTag(directoryReady.error),
191192 });
192193 return undefined;
193194 }
apps/server/src/provider/acp/AcpNativeLogging.test.ts created+137
......@@ -0,0 +1,137 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { ProviderDriverKind, ThreadId } from "@t3tools/contracts";
3import { assert, it } from "@effect/vitest";
4import * as Cause from "effect/Cause";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Logger from "effect/Logger";
8import * as Schema from "effect/Schema";
9import * as AcpErrors from "effect-acp/errors";
10
11import type { EventNdjsonLogger } from "../Layers/EventNdjsonLogger.ts";
12import { makeAcpNativeLoggerFactory } from "./AcpNativeLogging.ts";
13
14const nodeServicesIt = it.layer(NodeServices.layer);
15const encodeUnknownJson = Schema.encodeUnknownSync(Schema.UnknownFromJsonString);
16
17nodeServicesIt("ACP native logging", (it) => {
18 it.effect("records bounded request and protocol diagnostics without raw payloads", () =>
19 Effect.gen(function* () {
20 const records: Array<unknown> = [];
21 const nativeEventLogger: EventNdjsonLogger = {
22 filePath: "/tmp/provider-native.ndjson",
23 write: (event) => Effect.sync(() => void records.push(event)),
24 close: () => Effect.void,
25 };
26 const makeLogger = yield* makeAcpNativeLoggerFactory();
27 const logger = makeLogger({
28 nativeEventLogger,
29 provider: ProviderDriverKind.make("cursor"),
30 threadId: ThreadId.make("thread-1"),
31 });
32 const secret = "secret-token-value";
33 const requestLogger = logger.requestLogger;
34 const protocolLogger = logger.protocolLogging?.logger;
35 assert.exists(requestLogger);
36 assert.exists(protocolLogger);
37 if (!requestLogger || !protocolLogger) return;
38
39 yield* requestLogger({
40 method: "session/prompt",
41 payload: { prompt: secret, sessionId: secret },
42 status: "failed",
43 cause: Cause.fail(AcpErrors.AcpRequestError.internalError(secret, { token: secret })),
44 });
45 yield* protocolLogger({
46 direction: "incoming",
47 stage: "raw",
48 payload: `{"token":"${secret}"}`,
49 });
50 yield* protocolLogger({
51 direction: "outgoing",
52 stage: "decoded",
53 payload: {
54 _tag: "Request",
55 tag: "session/prompt",
56 payload: { prompt: secret },
57 },
58 });
59
60 const serialized = encodeUnknownJson(records);
61 assert.notInclude(serialized, secret);
62 assert.include(serialized, '"method":"session/prompt"');
63 assert.include(serialized, '"errorTag":"AcpRequestError"');
64 assert.include(serialized, '"reasonCount":1');
65 assert.include(serialized, '"valueType":"string"');
66 assert.include(serialized, '"messageTag":"Request"');
67 }),
68 );
69
70 it.effect("logs a structural tag when the native writer defects", () => {
71 const messages: Array<unknown> = [];
72 const logCapture = Logger.make<unknown, void>(({ message }) => {
73 if (Array.isArray(message)) {
74 messages.push(...message);
75 } else {
76 messages.push(message);
77 }
78 });
79 const secret = "secret-writer-failure";
80
81 return Effect.gen(function* () {
82 const makeLogger = yield* makeAcpNativeLoggerFactory();
83 const logger = makeLogger({
84 nativeEventLogger: {
85 filePath: "/tmp/provider-native.ndjson",
86 write: () => Effect.die(new Error(secret)),
87 close: () => Effect.void,
88 },
89 provider: ProviderDriverKind.make("cursor"),
90 threadId: ThreadId.make("thread-1"),
91 });
92 const requestLogger = logger.requestLogger;
93 assert.exists(requestLogger);
94 if (!requestLogger) return;
95
96 yield* requestLogger({
97 method: "session/prompt",
98 payload: {},
99 status: "started",
100 });
101
102 const serialized = encodeUnknownJson(messages);
103 assert.notInclude(serialized, secret);
104 assert.include(serialized, '"errorTag":"Die"');
105 assert.include(serialized, '"reasonCount":1');
106 }).pipe(Effect.provide(Logger.layer([logCapture], { mergeWithExisting: false })));
107 });
108
109 it.effect("preserves native writer interruption", () =>
110 Effect.gen(function* () {
111 const makeLogger = yield* makeAcpNativeLoggerFactory();
112 const logger = makeLogger({
113 nativeEventLogger: {
114 filePath: "/tmp/provider-native.ndjson",
115 write: () => Effect.interrupt,
116 close: () => Effect.void,
117 },
118 provider: ProviderDriverKind.make("cursor"),
119 threadId: ThreadId.make("thread-1"),
120 });
121 const requestLogger = logger.requestLogger;
122 assert.exists(requestLogger);
123 if (!requestLogger) return;
124
125 const exit = yield* requestLogger({
126 method: "session/prompt",
127 payload: {},
128 status: "started",
129 }).pipe(Effect.exit);
130
131 assert.isTrue(Exit.isFailure(exit));
132 if (Exit.isFailure(exit)) {
133 assert.isTrue(Cause.hasInterruptsOnly(exit.cause));
134 }
135 }),
136 );
137});
apps/server/src/provider/acp/AcpNativeLogging.ts+60-11
......@@ -1,4 +1,5 @@
11import type { ProviderDriverKind, ThreadId } from "@t3tools/contracts";
2import { causeErrorTag, errorTag } from "@t3tools/shared/observability";
23import * as Cause from "effect/Cause";
34import * as Crypto from "effect/Crypto";
45import * as DateTime from "effect/DateTime";
......@@ -8,13 +9,58 @@ import type * as EffectAcpProtocol from "effect-acp/protocol";
89import type { EventNdjsonLogger } from "../Layers/EventNdjsonLogger.ts";
910import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
1011
12function structuralMethod(value: string): string {
13 return value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) ? value : "unknown";
14}
15
16function summarizePayload(payload: unknown): Readonly<Record<string, unknown>> {
17 if (payload === null) return { valueType: "null" };
18 if (typeof payload === "string") {
19 return { valueType: "string", byteLength: new TextEncoder().encode(payload).byteLength };
20 }
21 if (payload instanceof Uint8Array) {
22 return { valueType: "bytes", byteLength: payload.byteLength };
23 }
24 if (Array.isArray(payload)) {
25 return { valueType: "array", itemCount: payload.length };
26 }
27 if (typeof payload !== "object") {
28 return { valueType: typeof payload };
29 }
30
31 try {
32 const record = payload as Record<string, unknown>;
33 return {
34 valueType: "object",
35 fieldCount: Object.keys(record).length,
36 ...(typeof record._tag === "string" ? { messageTag: errorTag(record) } : {}),
37 ...(typeof record.tag === "string" ? { method: structuralMethod(record.tag) } : {}),
38 };
39 } catch {
40 return { valueType: "object" };
41 }
42}
43
1144function formatRequestLogPayload(event: AcpSessionRuntime.AcpSessionRequestLogEvent) {
1245 return {
13 method: event.method,
46 method: structuralMethod(event.method),
1447 status: event.status,
15 request: event.payload,
16 ...(event.result !== undefined ? { result: event.result } : {}),
17 ...(event.cause !== undefined ? { cause: Cause.pretty(event.cause) } : {}),
48 request: summarizePayload(event.payload),
49 ...(event.result !== undefined ? { result: summarizePayload(event.result) } : {}),
50 ...(event.cause !== undefined
51 ? {
52 errorTag: causeErrorTag(event.cause),
53 reasonCount: event.cause.reasons.length,
54 }
55 : {}),
56 };
57}
58
59function formatProtocolLogPayload(event: EffectAcpProtocol.AcpProtocolLogEvent) {
60 return {
61 direction: event.direction,
62 stage: event.stage,
63 payload: summarizePayload(event.payload),
1864 };
1965}
2066
......@@ -47,12 +93,15 @@ export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory"
4793 input.threadId,
4894 );
4995 }).pipe(
50 Effect.catch((cause) =>
51 Effect.logWarning("Failed to write native ACP event log.", {
52 cause,
53 provider: input.provider,
54 threadId: input.threadId,
55 }),
96 Effect.catchCause((cause) =>
97 Cause.hasInterrupts(cause)
98 ? Effect.interrupt
99 : Effect.logWarning("Failed to write native ACP event log.", {
100 errorTag: causeErrorTag(cause),
101 reasonCount: cause.reasons.length,
102 provider: input.provider,
103 threadId: input.threadId,
104 }),
56105 ),
57106 );
58107
......@@ -70,7 +119,7 @@ export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory"
70119 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) =>
71120 writeNativeAcpLog({
72121 kind: "protocol",
73 payload: event,
122 payload: formatProtocolLogPayload(event),
74123 }),
75124 } satisfies NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"]>,
76125 }
packages/shared/src/observability.test.ts+9
......@@ -15,11 +15,20 @@ import * as Tracer from "effect/Tracer";
1515import {
1616 causeErrorTag,
1717 compactTraceAttributes,
18 errorTag,
1819 makeLocalFileTracer,
1920 makeTraceSink,
2021 type TraceRecord,
2122} from "./observability.ts";
2223
24describe("errorTag", () => {
25 it("reports structural tags without retaining arbitrary values", () => {
26 assert.equal(errorTag({ _tag: "AcpRequestError" }), "AcpRequestError");
27 assert.equal(errorTag(new TypeError("secret-token-value")), "TypeError");
28 assert.equal(errorTag({ _tag: "secret token value" }), "TaggedError");
29 });
30});
31
2332describe("causeErrorTag", () => {
2433 it("reports the tagged failure value instead of the Cause reason wrapper", () => {
2534 assert.equal(
packages/shared/src/observability.ts+22-7
......@@ -73,18 +73,33 @@ export interface OtlpTraceRecord extends BaseTraceRecord {
7373
7474export type TraceRecord = EffectTraceRecord | OtlpTraceRecord;
7575
76function taggedErrorName(error: unknown): string {
77 return typeof error === "object" && error !== null && "_tag" in error
78 ? String(error._tag)
79 : error instanceof Error
80 ? error.name
81 : typeof error;
76function isStructuralTag(value: unknown): value is string {
77 return (
78 typeof value === "string" &&
79 value.length > 0 &&
80 value.length <= 128 &&
81 /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value)
82 );
83}
84
85export function errorTag(error: unknown): string {
86 try {
87 if (typeof error === "object" && error !== null && "_tag" in error) {
88 return isStructuralTag(error._tag) ? error._tag : "TaggedError";
89 }
90 if (error instanceof Error) {
91 return isStructuralTag(error.name) ? error.name : "Error";
92 }
93 } catch {
94 return "UnknownError";
95 }
96 return typeof error;
8297}
8398
8499export function causeErrorTag(cause: Cause.Cause<unknown>): string {
85100 const failure = Cause.findErrorOption(cause);
86101 if (Option.isSome(failure)) {
87 return taggedErrorName(failure.value);
102 return errorTag(failure.value);
88103 }
89104 return cause.reasons[0]?._tag ?? "Empty";
90105}