1import type { ProviderDriverKind, ThreadId } from "@t3tools/contracts";
2import { causeErrorTag, errorTag } from "@t3tools/shared/observability";
3import * as Cause from "effect/Cause";
4import * as Crypto from "effect/Crypto";
5import * as DateTime from "effect/DateTime";
6import * as Effect from "effect/Effect";
7import type * as EffectAcpProtocol from "effect-acp/protocol";
8
9import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
10import {
11 structuralProtocolMethod,
12 summarizeNativeProtocolPayload,
13} from "@t3tools/provider-core/server/nativeProtocolLogging";
14import type * as AcpSessionRuntime from "./AcpSessionRuntime.ts";
15
16const transientProtocolUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
17
18function structuralMethod(value: string): string {
19 return value.length <= 128 && /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value) ? value : "unknown";
20}
21
22function summarizePayload(payload: unknown): Readonly<Record<string, unknown>> {
23 if (payload === null) return { valueType: "null" };
24 if (typeof payload === "string") {
25 return { valueType: "string", byteLength: new TextEncoder().encode(payload).byteLength };
26 }
27 if (payload instanceof Uint8Array) {
28 return { valueType: "bytes", byteLength: payload.byteLength };
29 }
30 if (Array.isArray(payload)) {
31 return { valueType: "array", itemCount: payload.length };
32 }
33 if (typeof payload !== "object") {
34 return { valueType: typeof payload };
35 }
36
37 try {
38 const record = payload as Record<string, unknown>;
39 return {
40 valueType: "object",
41 fieldCount: Object.keys(record).length,
42 ...(typeof record._tag === "string" ? { messageTag: errorTag(record) } : {}),
43 ...(typeof record.tag === "string" ? { method: structuralMethod(record.tag) } : {}),
44 };
45 } catch {
46 return { valueType: "object" };
47 }
48}
49
50function formatRequestLogPayload(event: AcpSessionRuntime.AcpSessionRequestLogEvent) {
51 return {
52 method: structuralProtocolMethod(event.method),
53 status: event.status,
54 request: summarizeNativeProtocolPayload(event.payload),
55 ...(event.result !== undefined ? { result: summarizeNativeProtocolPayload(event.result) } : {}),
56 ...(event.cause !== undefined
57 ? {
58 errorTag: causeErrorTag(event.cause),
59 reasonCount: event.cause.reasons.length,
60 }
61 : {}),
62 };
63}
64
65function formatAcpProtocolLogPayload(event: EffectAcpProtocol.AcpProtocolLogEvent) {
66 return {
67 direction: event.direction,
68 stage: event.stage,
69 payload: summarizeNativeProtocolPayload(event.payload),
70 };
71}
72
73function isTransientProtocolMessage(message: unknown): boolean {
74 if (typeof message !== "object" || message === null) return false;
75 const method = Reflect.get(message, "tag") ?? Reflect.get(message, "method");
76 if (method !== "session/update") return false;
77
78 const payload = Reflect.get(message, "payload") ?? Reflect.get(message, "params");
79 if (typeof payload !== "object" || payload === null) return false;
80 const update = Reflect.get(payload, "update");
81 if (typeof update !== "object" || update === null) return false;
82 const updateType = Reflect.get(update, "sessionUpdate");
83 return typeof updateType === "string" && transientProtocolUpdates.has(updateType);
84}
85
86function rawChunkContainsOnlyTransientMessages(payload: string): boolean {
87 const lines = payload.split("\n");
88 const remainder = lines.pop() ?? "";
89 if (remainder.trim().length > 0) return false;
90
91 const messages: Array<unknown> = [];
92 for (const line of lines) {
93 if (line.trim().length === 0) continue;
94 try {
95 messages.push(JSON.parse(line));
96 } catch {
97 return false;
98 }
99 }
100 return messages.length > 0 && messages.every(isTransientProtocolMessage);
101}
102
103function filterTransientProtocolLog(
104 event: EffectAcpProtocol.AcpProtocolLogEvent,
105): EffectAcpProtocol.AcpProtocolLogEvent | undefined {
106 if (event.direction !== "incoming") return event;
107
108 if (event.stage === "raw" && typeof event.payload === "string") {
109 return rawChunkContainsOnlyTransientMessages(event.payload) ? undefined : event;
110 }
111
112 if (event.stage !== "decoded") return event;
113 if (!Array.isArray(event.payload)) {
114 return isTransientProtocolMessage(event.payload) ? undefined : event;
115 }
116
117 const payload = event.payload.filter((message) => !isTransientProtocolMessage(message));
118 return payload.length === 0 ? undefined : { ...event, payload };
119}
120
121export const makeAcpNativeLoggerFactory = Effect.fn("makeAcpNativeLoggerFactory")(function* () {
122 const crypto = yield* Crypto.Crypto;
123 return (input: {
124 readonly nativeEventLogger: ProviderEventLoggers.EventNdjsonLogger | undefined;
125 readonly provider: ProviderDriverKind;
126 readonly threadId: ThreadId;
127 readonly verboseProtocolLogging?: boolean;
128 }): Pick<AcpSessionRuntime.AcpSessionRuntimeOptions, "requestLogger" | "protocolLogging"> => {
129 const writeNativeAcpLog = (logInput: {
130 readonly kind: "request" | "protocol";
131 readonly payload: unknown;
132 }) =>
133 Effect.gen(function* () {
134 if (!input.nativeEventLogger) return;
135 const observedAt = DateTime.formatIso(yield* DateTime.now);
136 yield* input.nativeEventLogger.write(
137 {
138 observedAt,
139 event: {
140 id: yield* crypto.randomUUIDv4(),
141 kind: logInput.kind,
142 provider: input.provider,
143 createdAt: observedAt,
144 threadId: input.threadId,
145 payload: logInput.payload,
146 },
147 },
148 input.threadId,
149 );
150 }).pipe(
151 Effect.catchCause((cause) =>
152 Cause.hasInterrupts(cause)
153 ? Effect.interrupt
154 : Effect.logWarning("Failed to write native ACP event log.", {
155 errorTag: causeErrorTag(cause),
156 reasonCount: cause.reasons.length,
157 provider: input.provider,
158 threadId: input.threadId,
159 }),
160 ),
161 );
162
163 return {
164 requestLogger: (event: AcpSessionRuntime.AcpSessionRequestLogEvent) =>
165 writeNativeAcpLog({
166 kind: "request",
167 payload: formatRequestLogPayload(event),
168 }),
169 ...(input.nativeEventLogger && input.verboseProtocolLogging
170 ? {
171 protocolLogging: {
172 logIncoming: true,
173 logOutgoing: true,
174 logger: (event: EffectAcpProtocol.AcpProtocolLogEvent) => {
175 const filtered = filterTransientProtocolLog(event);
176 return filtered
177 ? writeNativeAcpLog({
178 kind: "protocol",
179 payload: formatAcpProtocolLogPayload(filtered),
180 })
181 : Effect.void;
182 },
183 } satisfies NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["protocolLogging"]>,
184 }
185 : {}),
186 };
187 };
188});