// @effect-diagnostics nodeBuiltinImport:off/** * Best-effort provider event logging with one shared writer per thread. * * Native and canonical views share batching, rotation, and retention state so * they cannot race while appending to the same thread-scoped file. */import * as NodeFS from "node:fs";import * as NodePath from "node:path"; import type { ThreadId } from "@t3tools/contracts";import { RotatingFileSink } from "@t3tools/shared/logging";import { errorTag } from "@t3tools/shared/observability";import * as Clock from "effect/Clock";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Schema from "effect/Schema";import * as Scope from "effect/Scope";import * as SynchronizedRef from "effect/SynchronizedRef"; import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts";import type { ResourceAttribution } from "../../resourceTelemetry/ResourceAttribution.ts"; const MEBIBYTE = 1024 * 1024;const DAY_MS = 24 * 60 * 60 * 1_000;const DEFAULT_MAX_BYTES = 10 * MEBIBYTE;const DEFAULT_MAX_FILES = 10;const DEFAULT_BATCH_WINDOW_MS = 1_000;const DEFAULT_MAX_TOTAL_BYTES = 512 * MEBIBYTE;const DEFAULT_MAX_AGE_MS = 14 * DAY_MS;const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000;const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE;const DEFAULT_MAX_BUFFERED_RECORDS = 512;const MAX_RECORD_CHARACTERS = 64 * 1024;const MAX_RECORD_FIELDS = 1_024;const MAX_RECORD_DEPTH = 16;const GLOBAL_THREAD_SEGMENT = "_global";const LOG_SCOPE = "provider-observability";const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); const transientCanonicalEventTypes = new Set([  "content.delta",  "hook.progress",  "item.updated",  "task.progress",  "thread.realtime.audio.delta",  "tool.progress",  "turn.proposed.delta",]);const transientNativeMethods = new Set([  "item/agentMessage/delta",  "item/commandExecution/outputDelta",  "item/fileChange/outputDelta",  "item/plan/delta",  "item/reasoning/summaryTextDelta",  "item/reasoning/textDelta",  "thread/realtime/outputAudio/delta",  "thread/realtime/transcript/delta",  "turn/diff/updated",]);const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]); export type EventNdjsonStream = "native" | "canonical" | "orchestration"; export interface EventNdjsonLogger {  readonly filePath: string;  readonly write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void>;  readonly close: () => Effect.Effect<void>;} export interface EventNdjsonLogStore {  readonly filePath: string;  readonly logger: (stream: EventNdjsonStream) => EventNdjsonLogger;  readonly close: () => Effect.Effect<void>;} export interface EventNdjsonLogStoreOptions {  readonly maxBytes?: number;  readonly maxFiles?: number;  readonly batchWindowMs?: number;  readonly maxTotalBytes?: number;  readonly maxAgeMs?: number;  readonly retentionCheckIntervalMs?: number;  readonly maxBufferedBytes?: number;  readonly maxBufferedRecords?: number;  readonly attribution?: ResourceAttribution["Service"];} export interface EventNdjsonLoggerOptions extends EventNdjsonLogStoreOptions {  readonly stream: EventNdjsonStream;} export class EventNdjsonLogConfigurationError extends Schema.TaggedErrorClass<EventNdjsonLogConfigurationError>()(  "EventNdjsonLogConfigurationError",  {    filePath: Schema.String,    option: Schema.String,    value: Schema.Number,    minimum: Schema.Number,  },) {  override get message(): string {    return `Provider event log option '${this.option}' must be an integer >= ${this.minimum}; received ${this.value} for '${this.filePath}'`;  }} export class EventNdjsonLogDirectoryError extends Schema.TaggedErrorClass<EventNdjsonLogDirectoryError>()(  "EventNdjsonLogDirectoryError",  {    directory: Schema.String,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Failed to create provider event log directory '${this.directory}'`;  }} export type EventNdjsonLogStoreError =  | EventNdjsonLogConfigurationError  | EventNdjsonLogDirectoryError; interface ResolvedOptions {  readonly maxBytes: number;  readonly maxFiles: number;  readonly batchWindowMs: number;  readonly maxTotalBytes: number;  readonly maxAgeMs: number;  readonly retentionCheckIntervalMs: number;  readonly maxBufferedBytes: number;  readonly maxBufferedRecords: number;  readonly attribution: ResourceAttribution["Service"] | undefined;} export interface PendingRecord {  readonly stream: EventNdjsonStream;  readonly threadSegment: string;  readonly line: string;  readonly bytes: number;} interface StoreState {  readonly pending: Array<PendingRecord>;  readonly pendingBytes: number;  readonly sinks: ReadonlyMap<string, RotatingFileSink>;  readonly flushScheduled: boolean;  readonly closed: boolean;  readonly lastRetentionAt: number;} interface AttributionSummary {  readonly stream: EventNdjsonStream;  readonly count: number;  readonly logicalWriteBytes: number;} interface FileOperationFailure {  readonly filePath: string;  readonly cause: unknown;} interface RetentionResult {  readonly failures: ReadonlyArray<FileOperationFailure>;} interface DrainResult {  readonly attributions: ReadonlyArray<AttributionSummary>;  readonly failures: ReadonlyArray<FileOperationFailure>;} function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> {  return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE }));} function resolveThreadSegment(raw: string | null | undefined): string {  const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null;  return normalized ?? GLOBAL_THREAD_SEGMENT;} function resolveStreamLabel(stream: EventNdjsonStream): string {  return stream === "native" ? "NTIVE" : stream === "orchestration" ? "ORCH" : "CANON";} function providerLogPrefix(filePath: string): string {  const basename = NodePath.basename(filePath);  const extension = NodePath.extname(basename);  return `${extension.length > 0 ? basename.slice(0, -extension.length) : basename}.`;} function providerLogPath(directory: string, prefix: string, threadSegment: string): string {  return NodePath.join(directory, `${prefix}${threadSegment}.log`);} export function shouldPersistProviderEvent(stream: EventNdjsonStream, event: unknown): boolean {  if (stream === "orchestration" || typeof event !== "object" || event === null) {    return true;  }  try {    const type = Reflect.get(event, "type");    if (typeof type === "string" && transientCanonicalEventTypes.has(type)) {      return false;    }    if (stream !== "native") return true;     const nested = Reflect.get(event, "event");    const envelope = typeof nested === "object" && nested !== null ? nested : event;    // Decoded frames carry the same information as raw frames without another    // copy of every token delta. Decode failures have their own diagnostic frame.    if (Reflect.get(envelope, "stage") === "raw") return false;    const decodedPayload = Reflect.get(envelope, "payload");    const nativeEvent =      Reflect.get(envelope, "stage") === "decoded" &&      typeof decodedPayload === "object" &&      decodedPayload !== null        ? decodedPayload        : envelope;    const method = Reflect.get(nativeEvent, "method");    if (      typeof method === "string" &&      (transientNativeMethods.has(method) ||        method.startsWith("claude/stream_event/content_block_delta/"))    ) {      return false;    }     const nativeType = Reflect.get(nativeEvent, "type");    if (nativeType === "message.part.delta") return false;    if (nativeType === "stream_event") {      const streamEvent = Reflect.get(nativeEvent, "event");      if (        typeof streamEvent === "object" &&        streamEvent !== null &&        Reflect.get(streamEvent, "type") === "content_block_delta"      ) {        return false;      }    }     const payload = Reflect.get(nativeEvent, "payload");    if (typeof payload !== "object" || payload === null) return true;     if (method === "session/update") {      const update = Reflect.get(payload, "update");      if (typeof update !== "object" || update === null) return true;      const updateType = Reflect.get(update, "sessionUpdate");      return typeof updateType !== "string" || !transientAcpUpdates.has(updateType);    }     if (nativeType === "message.part.updated") {      const properties = Reflect.get(payload, "properties");      if (typeof properties !== "object" || properties === null) return true;      const part = Reflect.get(properties, "part");      if (typeof part !== "object" || part === null) return true;      const partType = Reflect.get(part, "type");      if (partType === "text" || partType === "reasoning") return false;      if (partType === "tool") {        // Running snapshots repeat growing output. Pending and terminal states stay in the log.        const state = Reflect.get(part, "state");        if (typeof state === "object" && state !== null) {          return Reflect.get(state, "status") !== "running";        }      }      return true;    }     return true;  } catch {    return true;  }} const summaryFields = [  "provider",  "protocol",  "kind",  "providerSessionId",  "direction",  "stage",  "type",  "subtype",  "method",  "id",  "threadId",  "turnId",  "requestId",  "session_id",  "status",  "is_error",  "api_error_status",  "terminal_reason",  "stop_reason",  "operation",  "code",  "willRetry",  "message",  "event",  "payload",  "params",  "result",  "thread",  "turn",  "error",  "turns",  "items",  "content",] as const; function summarizeProviderEvent(event: unknown): unknown {  let remainingFields = 128;  let remainingCharacters = 8 * 1024;  const summarize = (value: unknown, depth: number): unknown => {    if (typeof value === "string") {      if (value.length > Math.min(1_024, remainingCharacters)) {        return { omittedCharacters: value.length };      }      remainingCharacters -= value.length;      return value;    }    if (value === null || typeof value === "number" || typeof value === "boolean") return value;    if (typeof value !== "object") return undefined;    if (Array.isArray(value)) return { itemCount: value.length };    const summary: Record<string, unknown> = { truncated: true };    if (depth >= 6) return summary;    for (const key of summaryFields) {      if (remainingFields <= 0) break;      const nested = Reflect.get(value, key);      if (nested === undefined) continue;      remainingFields -= 1;      summary[key] = summarize(nested, depth + 1);    }    return summary;  };  try {    return summarize(event, 0);  } catch {    return { truncated: true };  }} /** Bounds traversal before adapters copy payloads or the logger encodes them. */export function boundProviderEventForLogging(event: unknown): unknown {  let remainingCharacters = MAX_RECORD_CHARACTERS;  let remainingFields = MAX_RECORD_FIELDS;  const ancestors = new WeakSet<object>();  const fits = (value: unknown, depth: number): boolean => {    if (typeof value === "string") {      remainingCharacters -= value.length;      return remainingCharacters >= 0;    }    if (typeof value !== "object" || value === null) return true;    if (depth > MAX_RECORD_DEPTH || ancestors.has(value)) return false;    if (Array.isArray(value) && value.length > remainingFields) return false;    ancestors.add(value);    for (const key in value) {      if (!Object.hasOwn(value, key)) continue;      remainingFields -= 1;      remainingCharacters -= key.length;      if (        remainingFields < 0 ||        remainingCharacters < 0 ||        !fits(Reflect.get(value, key), depth + 1)      )        return false;    }    ancestors.delete(value);    return true;  };  try {    if (fits(event, 0)) return event;  } catch {    // A failing accessor must not escape into provider processing.  }  return summarizeProviderEvent(event);} export function writeBatchedMessages(  sink: Pick<RotatingFileSink, "write">,  records: ReadonlyArray<PendingRecord>,  maxBytes: number,  onWritten: (records: ReadonlyArray<PendingRecord>) => void,): void {  let pendingRecords: Array<PendingRecord> = [];  let pendingBytes = 0;   const flush = () => {    if (pendingRecords.length === 0) return;    const writtenRecords = pendingRecords;    sink.write(writtenRecords.map((record) => record.line).join(""));    onWritten(writtenRecords);    pendingRecords = [];    pendingBytes = 0;  };   for (const record of records) {    if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) {      flush();    }    pendingRecords.push(record);    pendingBytes += record.bytes;    if (pendingBytes >= maxBytes) {      flush();    }  }  flush();} function isProviderLogFile(filePath: string, fileName: string, filePrefix: string): boolean {  if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false;  if (fileName.startsWith(filePrefix)) return true;   const descriptor = NodeFS.openSync(filePath, "r");  try {    const header = Buffer.alloc(256);    const bytesRead = NodeFS.readSync(descriptor, header, 0, header.byteLength, 0);    return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead));  } finally {    NodeFS.closeSync(descriptor);  }} function enforceRetention(input: {  readonly directory: string;  readonly maxTotalBytes: number;  readonly maxAgeMs: number;  readonly activeFilePaths: ReadonlySet<string>;  readonly filePrefix: string;  readonly now: number;}): RetentionResult {  const failures: Array<FileOperationFailure> = [];  const files: Array<{ filePath: string; mtimeMs: number; size: number }> = [];   let entries: ReadonlyArray<NodeFS.Dirent>;  try {    entries = NodeFS.readdirSync(input.directory, { withFileTypes: true });  } catch (cause) {    return { failures: [{ filePath: input.directory, cause }] };  }   for (const entry of entries) {    if (!entry.isFile()) continue;    const filePath = NodePath.join(input.directory, entry.name);    try {      if (!isProviderLogFile(filePath, entry.name, input.filePrefix)) continue;      const stat = NodeFS.statSync(filePath);      files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size });    } catch (cause) {      failures.push({ filePath, cause });    }  }   let totalBytes = files.reduce((total, file) => total + file.size, 0);  const remove = (file: (typeof files)[number]) => {    if (input.activeFilePaths.has(file.filePath)) return false;    try {      NodeFS.rmSync(file.filePath, { force: true });      totalBytes -= file.size;      return true;    } catch (cause) {      failures.push({ filePath: file.filePath, cause });      return false;    }  };   const retained = files.filter((file) => {    if (input.now - file.mtimeMs <= input.maxAgeMs) return true;    return !remove(file);  });   for (const file of retained.toSorted(    (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath),  )) {    if (totalBytes <= input.maxTotalBytes) break;    remove(file);  }   return { failures };} function validateOption(input: {  readonly filePath: string;  readonly option: string;  readonly value: number;  readonly minimum: number;}): EventNdjsonLogConfigurationError | undefined {  if (Number.isInteger(input.value) && input.value >= input.minimum) return undefined;  return new EventNdjsonLogConfigurationError(input);} function resolveOptions(  filePath: string,  options: EventNdjsonLogStoreOptions,): Effect.Effect<ResolvedOptions, EventNdjsonLogConfigurationError> {  const resolved = {    maxBytes: options.maxBytes ?? DEFAULT_MAX_BYTES,    maxFiles: options.maxFiles ?? DEFAULT_MAX_FILES,    batchWindowMs: options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS,    maxTotalBytes: options.maxTotalBytes ?? DEFAULT_MAX_TOTAL_BYTES,    maxAgeMs: options.maxAgeMs ?? DEFAULT_MAX_AGE_MS,    retentionCheckIntervalMs:

Lines 1–500 of 805. Load more