// @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 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",]);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.TaggedError<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.TaggedError<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`);} function shouldPersist(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 nativeEvent = typeof nested === "object" && nested !== null ? nested : event;    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;     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;  }} 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:      options.retentionCheckIntervalMs ?? DEFAULT_RETENTION_CHECK_INTERVAL_MS,    maxBufferedBytes: options.maxBufferedBytes ?? DEFAULT_MAX_BUFFERED_BYTES,    maxBufferedRecords: options.maxBufferedRecords ?? DEFAULT_MAX_BUFFERED_RECORDS,    attribution: options.attribution,  } satisfies ResolvedOptions;   const validations = [    ["maxBytes", resolved.maxBytes, 1],    ["maxFiles", resolved.maxFiles, 1],    ["batchWindowMs", resolved.batchWindowMs, 0],    ["maxTotalBytes", resolved.maxTotalBytes, 1],    ["maxAgeMs", resolved.maxAgeMs, 1],    ["retentionCheckIntervalMs", resolved.retentionCheckIntervalMs, 1],    ["maxBufferedBytes", resolved.maxBufferedBytes, 1],    ["maxBufferedRecords", resolved.maxBufferedRecords, 1],  ] as const;   for (const [option, value, minimum] of validations) {    const error = validateOption({ filePath, option, value, minimum });    if (error) return Effect.fail(error);  }  return Effect.succeed(resolved);} function drainPending(input: {  readonly directory: string;  readonly options: ResolvedOptions;  readonly state: StoreState;  readonly filePrefix: string;  readonly now: number;  readonly timerFired: boolean;  readonly close: boolean;}): readonly [DrainResult, StoreState] {  if (input.state.closed) {    return [{ attributions: [], failures: [] }, input.state];  }   const sinks = new Map(input.state.sinks);  const failures: Array<FileOperationFailure> = [];  const attributionByStream = new Map<    EventNdjsonStream,    { count: number; logicalWriteBytes: number }  >();  const recordsBySegment = new Map<string, Array<PendingRecord>>();   for (const record of input.state.pending) {    const records = recordsBySegment.get(record.threadSegment) ?? [];    records.push(record);    recordsBySegment.set(record.threadSegment, records);  }   for (const [threadSegment, records] of recordsBySegment) {    const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment);    let sink = sinks.get(threadSegment);    if (!sink) {      try {        sink = new RotatingFileSink({          filePath,          maxBytes: input.options.maxBytes,          maxFiles: input.options.maxFiles,          throwOnError: true,        });        sinks.set(threadSegment, sink);      } catch (cause) {        failures.push({ filePath, cause });        continue;      }    }     try {      writeBatchedMessages(sink, records, input.options.maxBytes, (writtenRecords) => {        for (const record of writtenRecords) {          const current = attributionByStream.get(record.stream) ?? {            count: 0,            logicalWriteBytes: 0,          };          attributionByStream.set(record.stream, {            count: current.count + 1,            logicalWriteBytes: current.logicalWriteBytes + record.bytes,          });        }      });    } catch (cause) {      sinks.delete(threadSegment);      failures.push({ filePath, cause });    }  }   const retentionDue =    input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs;  const retention = retentionDue    ? enforceRetention({        directory: input.directory,        maxTotalBytes: input.options.maxTotalBytes,        maxAgeMs: input.options.maxAgeMs,        activeFilePaths: new Set(          Array.from(sinks.keys(), (threadSegment) =>            providerLogPath(input.directory, input.filePrefix, threadSegment),          ),        ),        filePrefix: input.filePrefix,        now: input.now,      })    : { failures: [] };   return [    {      attributions: Array.from(attributionByStream, ([stream, value]) => ({        stream,        ...value,      })),      failures: [...failures, ...retention.failures],    },    {      pending: [],      pendingBytes: 0,      sinks,      flushScheduled: input.timerFired ? false : input.state.flushScheduled,      closed: input.close,      lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt,    },  ];} const serializeEvent = Effect.fnUntraced(function* (event: unknown) {  return yield* encodeUnknownJsonString(event).pipe(    Effect.catch((error) =>      logWarning("failed to serialize provider event log record", {

Lines 1–500 of 671. Load more