diff --git a/packages/agent-core-v2/src/log/formatter.ts b/packages/agent-core-v2/src/log/formatter.ts new file mode 100644 index 000000000..cb47ede03 --- /dev/null +++ b/packages/agent-core-v2/src/log/formatter.ts @@ -0,0 +1,205 @@ +/** + * `log` domain (L1) — logfmt entry formatter. + * + * Renders a `LogEntry` as a single logfmt line (`ISO LEVEL msg k=v ...`), + * redacts secret-shaped keys and raw secret patterns, truncates oversized + * fields, optionally colorizes the level with ANSI, and indents error stacks. + * Pure — no I/O, no DI. + */ + +import type { LogContext, LogEntry, LogEntryError } from './log'; + +export const MSG_MAX_CHARS = 200; +export const CTX_VALUE_MAX_CHARS = 2048; +export const STACK_MAX_BYTES = 2048; +export const ENTRY_MAX_BYTES = 4096; +export const REDACT_MAX_DEPTH = 10; + +const REDACTED_KEYS: ReadonlySet = new Set([ + 'authorization', + 'apikey', + 'token', + 'refreshtoken', + 'accesstoken', + 'idtoken', + 'password', + 'secret', + 'clientsecret', + 'apisecret', + 'cookie', + 'setcookie', + 'bearer', +]); + +const SAFE_KEY_RE = /^[\w.-]+$/; +const ELLIPSIS = '…'; +const TRUNCATED_TAIL = ` …truncated`; +const REDACTED = '[REDACTED]'; +const RAW_SECRET_PATTERNS: readonly RegExp[] = [ + /\b(authorization\s*[:=]\s*bearer\s+)[^\s"'`]+/gi, + /\b((?:api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|token|password|secret)\s*[:=]\s*)[^\s"'`]+/gi, + /\b(cookie\s*[:=]\s*)[^\r\n]+/gi, +]; + +const LEVEL_LABEL: Record, string> = { + error: 'ERROR', + warn: 'WARN ', + info: 'INFO ', + debug: 'DEBUG', +}; + +const ANSI_LEVEL: Record, string> = { + error: '', + warn: '', + info: '', + debug: '', +}; +const ANSI_RESET = ''; + +function normalizeKey(key: string): string { + return key.toLowerCase().replaceAll(/[_\-.]/g, ''); +} + +export function redactCtx(ctx: LogContext): LogContext { + const seen = new WeakSet(); + const walk = (value: unknown, depth: number): unknown => { + if (depth > REDACT_MAX_DEPTH) return '[REDACTED:depth]'; + if (value === null || typeof value !== 'object') return value; + if (seen.has(value)) return '[REDACTED:cycle]'; + seen.add(value); + if (Array.isArray(value)) { + return value.map((item) => walk(item, depth + 1)); + } + const out: Record = {}; + for (const [key, raw] of Object.entries(value as Record)) { + out[key] = REDACTED_KEYS.has(normalizeKey(key)) ? REDACTED : walk(raw, depth + 1); + } + return out; + }; + return walk(ctx, 0) as LogContext; +} + +export interface FormatOptions { + readonly ansi?: boolean; + readonly omitContextKeys?: readonly string[]; +} + +export interface FormattedEntry { + readonly text: string; + readonly dropped: boolean; +} + +function truncate(value: string, max: number): string { + return value.length <= max ? value : value.slice(0, max - 1) + ELLIPSIS; +} + +function serializeValue(raw: unknown): string { + if (typeof raw === 'string') return redactString(raw); + if (raw === undefined) return 'undefined'; + if (raw === null) return 'null'; + if ( + typeof raw === 'number' || + typeof raw === 'boolean' || + typeof raw === 'bigint' || + typeof raw === 'symbol' + ) { + return String(raw); + } + try { + const json = JSON.stringify(raw); + if (json !== undefined) return json; + } catch { + } + if (typeof raw === 'function') return raw.name === '' ? '[Function]' : `[Function: ${raw.name}]`; + return Object.prototype.toString.call(raw); +} + +function redactString(value: string): string { + let out = value; + for (const pattern of RAW_SECRET_PATTERNS) { + out = out.replace(pattern, `$1${REDACTED}`); + } + return out; +} + +function quote(value: string): string { + return `"${value.replaceAll('\\', '\\\\').replaceAll('"', '\\"').replaceAll('\n', '\\n')}"`; +} + +function formatPair(key: string, raw: unknown): string { + const limited = truncate(serializeValue(raw), CTX_VALUE_MAX_CHARS); + const renderedKey = SAFE_KEY_RE.test(key) ? key : quote(key); + const renderedVal = /[\s="\\]/.test(limited) || limited.length === 0 ? quote(limited) : limited; + return `${renderedKey}=${renderedVal}`; +} + +function clipBytes(text: string, maxBytes: number): string { + if (Buffer.byteLength(text, 'utf-8') <= maxBytes) return text; + let lo = 0; + let hi = text.length; + while (lo < hi) { + const mid = (lo + hi + 1) >> 1; + if ( + Buffer.byteLength(text.slice(0, mid), 'utf-8') <= + maxBytes - Buffer.byteLength(TRUNCATED_TAIL, 'utf-8') + ) { + lo = mid; + } else { + hi = mid - 1; + } + } + return text.slice(0, lo) + TRUNCATED_TAIL; +} + +function clipStack(stack: string): string { + if (Buffer.byteLength(stack, 'utf-8') <= STACK_MAX_BYTES) return stack; + return clipBytes(stack, STACK_MAX_BYTES); +} + +function indentStack(stack: string): string { + return stack + .split('\n') + .map((line, i) => (i === 0 ? ` ${line}` : ` ${line.trimStart()}`)) + .join('\n'); +} + +export function formatEntry(entry: LogEntry, options: FormatOptions = {}): FormattedEntry { + const ctx = entry.ctx ? redactCtx(entry.ctx) : undefined; + const omitContextKeys = new Set(options.omitContextKeys ?? []); + const msg = truncate(entry.msg, MSG_MAX_CHARS); + const pairs: string[] = []; + if (ctx) { + for (const [k, v] of Object.entries(ctx)) { + if (omitContextKeys.has(k)) continue; + if (v !== undefined) pairs.push(formatPair(k, v)); + } + } + + const time = new Date(entry.t).toISOString(); + const label = LEVEL_LABEL[entry.level]; + const rendered = pairs.length === 0 + ? `${time} ${label} ${msg}` + : `${time} ${label} ${msg} ${pairs.join(' ')}`; + + let head = Buffer.byteLength(rendered, 'utf-8') > ENTRY_MAX_BYTES + ? clipBytes(rendered, ENTRY_MAX_BYTES) + : rendered; + + if (options.ansi === true) { + head = `${ANSI_LEVEL[entry.level]}${head}${ANSI_RESET}`; + } + + if (entry.error?.stack) { + head = `${head}\n${indentStack(clipStack(redactString(entry.error.stack)))}`; + } else if (entry.error?.message) { + head = `${head}\n Error: ${redactString(entry.error.message)}`; + } + + return { text: head, dropped: false }; +} + +export function extractError(value: Error): LogEntryError { + return typeof value.stack === 'string' + ? { message: value.message, stack: value.stack } + : { message: value.message }; +} diff --git a/packages/agent-core-v2/src/log/index.ts b/packages/agent-core-v2/src/log/index.ts index 93b3b5000..a9e5b8ac1 100644 --- a/packages/agent-core-v2/src/log/index.ts +++ b/packages/agent-core-v2/src/log/index.ts @@ -1,8 +1,12 @@ /** - * `log` domain barrel — re-exports the `log` contract and its scoped service - * (`logService`). Importing this barrel registers the `ILogService` and - * `ILogSink` bindings into the scope registry. + * `log` domain barrel — re-exports the logging contract and its scoped + * services. Importing this barrel registers the `ILogService` / `ILogWriterService` + * (Core) and `ISessionLogService` (Session) bindings into the scope registry. */ export * from './log'; +export * from './logConfig'; export * from './logService'; +export * from './sessionLogService'; +export * from './logWriter'; +export * from './formatter'; diff --git a/packages/agent-core-v2/src/log/log.ts b/packages/agent-core-v2/src/log/log.ts index 8c8359ff1..a3d8b3c26 100644 --- a/packages/agent-core-v2/src/log/log.ts +++ b/packages/agent-core-v2/src/log/log.ts @@ -3,11 +3,13 @@ * * Defines the public contract of logging: the `LogEntry` / `LogLevel` model, * the `ILogger` / `ILogService` used by other domains to emit leveled entries, - * and the `ILogSink` they are written to. Core-scoped — one shared instance - * for the process. + * the `ILogWriterService` they are written to, and the per-session `ISessionLogService` + * that owns a session-scoped sink. `ILogService` is Core-scoped; + * `ISessionLogService` is Session-scoped. */ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { ScopeSeed } from '#/_base/di/scope'; export type LogLevel = 'off' | 'error' | 'warn' | 'info' | 'debug'; @@ -28,12 +30,14 @@ export interface LogEntry { readonly error?: LogEntryError; } -export interface ILogSink { +export interface ILogWriterService { write(entry: LogEntry): void; + flush?(): Promise; + close?(): Promise; } -export const ILogSink: ServiceIdentifier = - createDecorator('logSink'); +export const ILogWriterService: ServiceIdentifier = + createDecorator('logWriterService'); export interface ILogger { error(message: string, payload?: LogPayload): void; @@ -47,6 +51,7 @@ export interface ILogService extends ILogger { readonly _serviceBrand: undefined; readonly level: LogLevel; setLevel(level: LogLevel): void; + flush(): Promise; } export const ILogService: ServiceIdentifier = @@ -64,3 +69,29 @@ export function levelEnabled(level: LogLevel, configured: LogLevel): boolean { if (level === 'off' || configured === 'off') return false; return LEVEL_ORDER[level] <= LEVEL_ORDER[configured]; } + +export interface ISessionLogService extends ILogger { + readonly _serviceBrand: undefined; + flush(): Promise; + close(): Promise; +} + +export const ISessionLogService: ServiceIdentifier = + createDecorator('sessionLogService'); + +export interface ISessionLogOptions { + readonly sessionId: string; + readonly sessionDir: string; +} + +export const ISessionLogOptions: ServiceIdentifier = + createDecorator('sessionLogOptions'); + +export function sessionLogSeed(sessionId: string, sessionDir: string): ScopeSeed { + return [ + [ + ISessionLogOptions as ServiceIdentifier, + { sessionId, sessionDir } satisfies ISessionLogOptions, + ], + ]; +} diff --git a/packages/agent-core-v2/src/log/logConfig.ts b/packages/agent-core-v2/src/log/logConfig.ts new file mode 100644 index 000000000..030b6f455 --- /dev/null +++ b/packages/agent-core-v2/src/log/logConfig.ts @@ -0,0 +1,78 @@ +/** + * `log` domain (L1) — runtime logging configuration. + * + * Builds the `LoggingConfig` from `KIMI_LOG_*` environment variables plus + * defaults, resolves the global and per-session log paths, and exposes the + * `ILogOptions` seed used to inject the resolved config into a Core scope. + */ + +import { join } from 'pathe'; + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { ScopeSeed } from '#/_base/di/scope'; + +import type { LogLevel } from './log'; + +export const DEFAULT_LOG_LEVEL: LogLevel = 'info'; +export const DEFAULT_GLOBAL_MAX_BYTES = 6 * 1024 * 1024; +export const DEFAULT_GLOBAL_FILES = 5; +export const DEFAULT_SESSION_MAX_BYTES = 5 * 1024 * 1024; +export const DEFAULT_SESSION_FILES = 3; + +export interface LoggingConfig { + readonly level: LogLevel; + readonly globalLogPath: string; + readonly globalMaxBytes: number; + readonly globalFiles: number; + readonly sessionMaxBytes: number; + readonly sessionFiles: number; +} + +export interface ILogOptions extends LoggingConfig {} + +export const ILogOptions: ServiceIdentifier = + createDecorator('logOptions'); + +export interface ResolveLoggingInput { + readonly homeDir: string; + readonly env?: NodeJS.ProcessEnv; +} + +export function resolveGlobalLogPath(homeDir: string): string { + return join(homeDir, 'logs', 'kimi-code.log'); +} + +export function resolveSessionLogPath(sessionDir: string): string { + return join(sessionDir, 'logs', 'kimi-code.log'); +} + +export function resolveLoggingConfig(input: ResolveLoggingInput): LoggingConfig { + const env = input.env ?? process.env; + return { + level: parseLevel(env['KIMI_LOG_LEVEL']) ?? DEFAULT_LOG_LEVEL, + globalLogPath: resolveGlobalLogPath(input.homeDir), + globalMaxBytes: parsePositiveInt(env['KIMI_LOG_GLOBAL_MAX_BYTES']) ?? DEFAULT_GLOBAL_MAX_BYTES, + globalFiles: parsePositiveInt(env['KIMI_LOG_GLOBAL_FILES']) ?? DEFAULT_GLOBAL_FILES, + sessionMaxBytes: + parsePositiveInt(env['KIMI_LOG_SESSION_MAX_BYTES']) ?? DEFAULT_SESSION_MAX_BYTES, + sessionFiles: parsePositiveInt(env['KIMI_LOG_SESSION_FILES']) ?? DEFAULT_SESSION_FILES, + }; +} + +export function logSeed(config: LoggingConfig): ScopeSeed { + return [[ILogOptions as ServiceIdentifier, config satisfies ILogOptions]]; +} + +function parseLevel(value: string | undefined): LogLevel | undefined { + if (value === undefined) return undefined; + const v = value.toLowerCase().trim(); + if (v === 'off' || v === 'error' || v === 'warn' || v === 'info' || v === 'debug') return v; + return undefined; +} + +function parsePositiveInt(value: string | undefined): number | undefined { + if (value === undefined) return undefined; + const n = Number.parseInt(value, 10); + if (!Number.isFinite(n) || n <= 0) return undefined; + return n; +} diff --git a/packages/agent-core-v2/src/log/logService.ts b/packages/agent-core-v2/src/log/logService.ts index a19adb34a..82f591bcb 100644 --- a/packages/agent-core-v2/src/log/logService.ts +++ b/packages/agent-core-v2/src/log/logService.ts @@ -1,8 +1,8 @@ /** - * `log` domain (L1) — `ILogService` implementation and built-in sinks. + * `log` domain (L1) — `ILogService` implementation and built-in writers. * * Filters entries by the configured `LogLevel` and writes them to the bound - * `ILogSink`; provides the console and in-memory `ILogSink` implementations. + * `ILogWriterService`; provides the console and in-memory `ILogWriterService` implementations. * Bound at Core scope. */ @@ -17,7 +17,7 @@ import { type LogLevel, type LogPayload, ILogService, - ILogSink, + ILogWriterService, levelEnabled, } from './log'; @@ -47,14 +47,14 @@ function extractContext(payload: LogPayload): LogContext | undefined { return undefined; } -export class MemoryLogSink implements ILogSink { +export class MemoryLogWriterService implements ILogWriterService { readonly entries: LogEntry[] = []; write(entry: LogEntry): void { this.entries.push(entry); } } -export class ConsoleLogSink implements ILogSink { +export class ConsoleLogWriterService implements ILogWriterService { write(entry: LogEntry): void { const line = entry.ctx !== undefined ? `${entry.msg} ${JSON.stringify(entry.ctx)}` : entry.msg; switch (entry.level) { @@ -82,7 +82,7 @@ export class LogService implements ILogService { private _level: LogLevel; constructor( - @ILogSink private readonly sink: ILogSink, + @ILogWriterService protected readonly writer: ILogWriterService, private readonly bound: LogContext = {}, level: LogLevel = 'info', ) { @@ -97,6 +97,10 @@ export class LogService implements ILogService { this._level = level; } + flush(): Promise { + return this.writer.flush?.() ?? Promise.resolve(); + } + error(message: string, payload?: LogPayload): void { this.emit('error', message, payload); } @@ -111,7 +115,7 @@ export class LogService implements ILogService { } child(ctx: LogContext): ILogger { - return new LogService(this.sink, { ...this.bound, ...ctx }, this._level); + return new LogService(this.writer, { ...this.bound, ...ctx }, this._level); } private emit( @@ -133,14 +137,14 @@ export class LogService implements ILogService { ...(ctx !== undefined ? { ctx } : {}), ...(error !== undefined ? { error } : {}), }; - this.sink.write(entry); + this.writer.write(entry); } } registerScopedService( LifecycleScope.Core, - ILogSink, - ConsoleLogSink, + ILogWriterService, + ConsoleLogWriterService, InstantiationType.Eager, 'log', ); diff --git a/packages/agent-core-v2/src/log/logWriter.ts b/packages/agent-core-v2/src/log/logWriter.ts new file mode 100644 index 000000000..d1d2db818 --- /dev/null +++ b/packages/agent-core-v2/src/log/logWriter.ts @@ -0,0 +1,319 @@ +/** + * `log` domain (L1) — rotating file writer. + * + * Writes formatted `LogEntry` lines to a size-rotated file, exposing `flush` + * and a synchronous `flushSync` for process-exit paths; `FileLogWriterService` adapts + * it to the `ILogWriterService` contract. Uses `node:fs` rather than `kaos` because + * rotation needs atomic rename and synchronous append. + */ + +import { appendFileSync, mkdirSync } from 'node:fs'; +import { mkdir, open, rename, stat, unlink } from 'node:fs/promises'; +import { dirname } from 'pathe'; + +import { InstantiationType } from '#/_base/di/extensions'; +import { Disposable } from '#/_base/di/lifecycle'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; + +import { formatEntry, type FormatOptions } from './formatter'; +import { ILogWriterService, type LogEntry, ISessionLogOptions } from './log'; +import { ILogOptions, resolveSessionLogPath } from './logConfig'; + +export const PENDING_MAX = 1000; +const STDERR_NOTICE_INTERVAL_MS = 30_000; + +class AsyncSerialQueue { + private tail: Promise = Promise.resolve(); + + run(task: () => Promise): Promise { + const next = this.tail.then(task, task); + this.tail = next.catch(() => {}); + return next; + } +} + +export interface RotatingWriter { + enqueue(line: string): void; + flush(): Promise; + close(): Promise; + flushSync(): void; +} + +export interface RotatingFileWriterOptions { + readonly path: string; + readonly maxBytes: number; + readonly files: number; +} + +async function syncDir(dirPath: string): Promise { + if (process.platform === 'win32') return; + const dirFh = await open(dirPath, 'r'); + try { + await dirFh.sync(); + } finally { + await dirFh.close(); + } +} + +export class RotatingFileWriter implements RotatingWriter { + private readonly queue = new AsyncSerialQueue(); + private pending: string[] = []; + private dropped = 0; + private closed = false; + private lastStderrNotice = 0; + private currentBytes = -1; + private directorySynced = false; + + constructor(private readonly options: RotatingFileWriterOptions) {} + + enqueue(line: string): void { + if (this.closed) return; + if (this.pending.length >= PENDING_MAX) { + this.pending.shift(); + this.dropped++; + } + this.pending.push(line); + this.scheduleDrain(); + } + + async flush(): Promise { + return this.queue.run(() => this.drain()); + } + + async close(): Promise { + if (this.closed) return; + this.closed = true; + try { + await this.flush(); + } catch { + } + } + + flushSync(): void { + if (this.closed || this.pending.length === 0) return; + try { + mkdirSync(dirname(this.options.path), { recursive: true }); + const body = this.pending.join('') + this.takeDroppedNotice(); + this.pending = []; + appendFileSync(this.options.path, body); + } catch (error) { + this.noteFailure(error); + } + } + + private scheduleDrain(): void { + if (this.closed) return; + queueMicrotask(() => { + if (this.closed || this.pending.length === 0) return; + this.queue.run(() => this.drain()).catch(() => {}); + }); + } + + private async drain(): Promise { + if (this.pending.length === 0) return true; + const droppedLine = this.takeDroppedNotice(); + const lines = droppedLine === '' ? [...this.pending] : [...this.pending, droppedLine]; + this.pending = []; + try { + await mkdir(dirname(this.options.path), { recursive: true }); + if (this.currentBytes < 0) { + this.currentBytes = await this.statSize(this.options.path); + } + await this.appendLines(lines); + + if (!this.directorySynced) { + await syncDir(dirname(this.options.path)); + this.directorySynced = true; + } + + return true; + } catch (error) { + this.noteFailure(error); + this.restorePending(lines); + return false; + } + } + + private restorePending(lines: readonly string[]): void { + const restored = [...lines, ...this.pending]; + const overflow = restored.length - PENDING_MAX; + if (overflow <= 0) { + this.pending = restored; + return; + } + this.dropped += overflow; + this.pending = restored.slice(overflow); + } + + private async appendLines(lines: readonly string[]): Promise { + let chunk = ''; + let chunkBytes = 0; + for (const line of lines) { + const lineBytes = Buffer.byteLength(line, 'utf-8'); + if ( + chunkBytes > 0 && + (chunkBytes + lineBytes > this.options.maxBytes || + this.currentBytes + chunkBytes + lineBytes > this.options.maxBytes) + ) { + await this.appendChunk(chunk); + chunk = ''; + chunkBytes = 0; + } + + if ( + chunkBytes === 0 && + this.currentBytes > 0 && + this.currentBytes + lineBytes > this.options.maxBytes + ) { + await this.rotate(); + } + + chunk += line; + chunkBytes += lineBytes; + } + if (chunkBytes > 0) { + await this.appendChunk(chunk); + } + } + + private async appendChunk(chunk: string): Promise { + const fh = await open(this.options.path, 'a'); + try { + await fh.appendFile(chunk, 'utf-8'); + await fh.sync(); + } finally { + await fh.close(); + } + this.currentBytes += Buffer.byteLength(chunk, 'utf-8'); + if (this.currentBytes >= this.options.maxBytes) { + await this.rotate(); + } + } + + private async rotate(): Promise { + const { path, files } = this.options; + for (let i = files - 2; i >= 1; i--) { + const from = `${path}.${i}`; + const to = `${path}.${i + 1}`; + try { + await rename(from, to); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + } + try { + await rename(path, `${path}.1`); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + try { + await unlink(`${path}.${files}`); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + this.currentBytes = 0; + this.directorySynced = false; + } + + private async statSize(p: string): Promise { + try { + const s = await stat(p); + return s.size; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return 0; + throw error; + } + } + + private takeDroppedNotice(): string { + if (this.dropped === 0) return ''; + const line = `... dropped ${this.dropped} entries ...\n`; + this.dropped = 0; + return line; + } + + private noteFailure(error: unknown): void { + const now = Date.now(); + if (now - this.lastStderrNotice < STDERR_NOTICE_INTERVAL_MS) return; + this.lastStderrNotice = now; + const code = (error as NodeJS.ErrnoException)?.code ?? 'UNKNOWN'; + try { + process.stderr.write(`[logger] write failed: ${code}\n`); + } catch {} + } +} + +export interface FileLogWriterServiceOptions extends RotatingFileWriterOptions { + readonly format?: FormatOptions; +} + +export class FileLogWriterService implements ILogWriterService { + private readonly sink: RotatingFileWriter; + private readonly format: FormatOptions; + + constructor(options: FileLogWriterServiceOptions) { + this.sink = new RotatingFileWriter(options); + this.format = options.format ?? {}; + } + + write(entry: LogEntry): void { + const { text, dropped } = formatEntry(entry, this.format); + if (dropped) return; + this.sink.enqueue(text + '\n'); + } + + async flush(): Promise { + await this.sink.flush(); + } + + close(): Promise { + return this.sink.close(); + } + + flushSync(): void { + this.sink.flushSync(); + } +} + +export class SessionFileLogWriterService extends Disposable implements ILogWriterService { + private readonly inner: FileLogWriterService; + + constructor( + @ILogOptions options: ILogOptions, + @ISessionLogOptions session: ISessionLogOptions, + ) { + super(); + this.inner = new FileLogWriterService({ + path: resolveSessionLogPath(session.sessionDir), + maxBytes: options.sessionMaxBytes, + files: options.sessionFiles, + format: { omitContextKeys: ['sessionId'] }, + }); + } + + write(entry: LogEntry): void { + this.inner.write(entry); + } + + flush(): Promise { + return this.inner.flush(); + } + + close(): Promise { + return this.inner.close(); + } + + override dispose(): void { + this.inner.flushSync(); + void this.inner.close(); + super.dispose(); + } +} + +registerScopedService( + LifecycleScope.Session, + ILogWriterService, + SessionFileLogWriterService, + InstantiationType.Delayed, + 'log', +); diff --git a/packages/agent-core-v2/src/log/sessionLogService.ts b/packages/agent-core-v2/src/log/sessionLogService.ts new file mode 100644 index 000000000..d291a6657 --- /dev/null +++ b/packages/agent-core-v2/src/log/sessionLogService.ts @@ -0,0 +1,39 @@ +/** + * `log` domain (L1) — `ISessionLogService` implementation. + * + * Per-session logger: binds `sessionId` to every entry and writes to the + * Session-scoped `ILogWriterService` (a rotating file writer owned by the Session scope). + * Bound at Session scope (Delayed) so sessions that never emit logs allocate + * nothing. + */ + +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; + +import { ILogWriterService, ISessionLogOptions, ISessionLogService } from './log'; +import { ILogOptions } from './logConfig'; +import { LogService } from './logService'; + +export class SessionLogService extends LogService implements ISessionLogService { + declare readonly _serviceBrand: undefined; + + constructor( + @ILogWriterService writer: ILogWriterService, + @ILogOptions options: ILogOptions, + @ISessionLogOptions session: ISessionLogOptions, + ) { + super(writer, { sessionId: session.sessionId }, options.level); + } + + close(): Promise { + return this.writer.close?.() ?? Promise.resolve(); + } +} + +registerScopedService( + LifecycleScope.Session, + ISessionLogService, + SessionLogService, + InstantiationType.Delayed, + 'log', +); diff --git a/packages/agent-core-v2/test/log/formatter.test.ts b/packages/agent-core-v2/test/log/formatter.test.ts new file mode 100644 index 000000000..0bc019275 --- /dev/null +++ b/packages/agent-core-v2/test/log/formatter.test.ts @@ -0,0 +1,233 @@ +import { describe, expect, it } from 'vitest'; + +import { + CTX_VALUE_MAX_CHARS, + ENTRY_MAX_BYTES, + MSG_MAX_CHARS, + STACK_MAX_BYTES, + extractError, + formatEntry, + redactCtx, +} from '#/log/formatter'; +import type { LogEntry } from '#/log/log'; + +const FIXED_TIME = Date.UTC(2026, 4, 19, 10, 12, 30, 123); + +function baseEntry(overrides: Partial = {}): LogEntry { + return { + t: FIXED_TIME, + level: 'info', + msg: 'diagnostic event', + ...overrides, + }; +} + +describe('formatter — logfmt rendering', () => { + it('renders timestamp, level, msg without ctx', () => { + const { text } = formatEntry(baseEntry()); + expect(text).toBe('2026-05-19T10:12:30.123Z INFO diagnostic event'); + }); + + it('renders ctx as k=v pairs', () => { + const { text } = formatEntry( + baseEntry({ ctx: { sessionId: 'ses_abc', workDir: '/repo' } }), + ); + expect(text).toContain('sessionId=ses_abc'); + expect(text).toContain('workDir=/repo'); + }); + + it('omits selected ctx keys', () => { + const { text } = formatEntry( + baseEntry({ ctx: { sessionId: 'ses_abc', workDir: '/repo' } }), + { omitContextKeys: ['sessionId'] }, + ); + expect(text).not.toContain('sessionId=ses_abc'); + expect(text).toContain('workDir=/repo'); + }); + + it('quotes ctx values that contain spaces or special chars', () => { + const { text } = formatEntry(baseEntry({ ctx: { path: '/Users/foo bar/x' } })); + expect(text).toContain('path="/Users/foo bar/x"'); + }); + + it('renders all level labels at fixed width', () => { + for (const level of ['error', 'warn', 'info', 'debug'] as const) { + const { text } = formatEntry(baseEntry({ level })); + const label = + level === 'error' ? 'ERROR' : level === 'warn' ? 'WARN ' : level === 'info' ? 'INFO ' : 'DEBUG'; + expect(text).toContain(` ${label} `); + } + }); + + it('does not include ANSI when ansi=false', () => { + const { text } = formatEntry(baseEntry({ level: 'error' }), { ansi: false }); + expect(text).not.toMatch(/\[/); + }); + + it('includes ANSI when ansi=true', () => { + const { text } = formatEntry(baseEntry({ level: 'error' }), { ansi: true }); + expect(text).toMatch(/\[31m/); + expect(text).toMatch(/\[0m/); + }); +}); + +describe('formatter — error extraction', () => { + it('attaches stack as indented multi-line block', () => { + const err = new Error('boom'); + err.stack = 'Error: boom\n at fn (file.ts:1:1)'; + const ext = extractError(err); + const { text } = formatEntry( + baseEntry({ level: 'error', msg: 'failure', error: { message: ext.message, stack: ext.stack } }), + ); + expect(text).toMatch(/\n Error: boom\n {4}at fn/); + }); + + it('falls back to message-only line when no stack', () => { + const { text } = formatEntry(baseEntry({ level: 'error', error: { message: 'no stack' } })); + expect(text).toMatch(/\n Error: no stack$/); + }); + + it('redacts secrets in error stack and message lines', () => { + const { text: stackText } = formatEntry( + baseEntry({ + level: 'error', + error: { + message: 'failed', + stack: + 'Error: request failed token=abc123\nAuthorization: Bearer secret-token\ncookie: sid=secret-cookie', + }, + }), + ); + expect(stackText).toContain('token=[REDACTED]'); + expect(stackText).toContain('Authorization: Bearer [REDACTED]'); + expect(stackText).toContain('cookie: [REDACTED]'); + expect(stackText).not.toContain('abc123'); + expect(stackText).not.toContain('secret-token'); + expect(stackText).not.toContain('secret-cookie'); + + const { text: messageText } = formatEntry( + baseEntry({ level: 'error', error: { message: 'failed access_token=abc123' } }), + ); + expect(messageText).toContain('access_token=[REDACTED]'); + expect(messageText).not.toContain('abc123'); + }); + + it('clips stack to STACK_MAX_BYTES with truncation marker', () => { + const longStack = 'Error: x\n' + ' at frame()\n'.repeat(1000); + const { text } = formatEntry(baseEntry({ error: { message: 'x', stack: longStack } })); + expect(text).toContain('…truncated'); + expect(Buffer.byteLength(text, 'utf-8')).toBeLessThan(STACK_MAX_BYTES + 4096); + }); +}); + +describe('formatter — limits', () => { + it('truncates msg over MSG_MAX_CHARS with ellipsis', () => { + const longMsg = 'x'.repeat(MSG_MAX_CHARS + 50); + const { text } = formatEntry(baseEntry({ msg: longMsg })); + expect(text).toContain('…'); + }); + + it('truncates a single ctx value over CTX_VALUE_MAX_CHARS', () => { + const big = 'y'.repeat(CTX_VALUE_MAX_CHARS + 50); + const { text } = formatEntry(baseEntry({ ctx: { huge: big } })); + expect(text).toMatch(/huge="?y{300,}…/); + }); + + it('byte-slices the rendered head when entry exceeds ENTRY_MAX_BYTES', () => { + const ctx: Record = {}; + for (let i = 0; i < 1000; i++) ctx[`k${i}`] = 'v'.repeat(50); + const { text } = formatEntry(baseEntry({ ctx, msg: 'x'.repeat(MSG_MAX_CHARS) })); + const head = text.split('\n')[0] ?? ''; + expect(Buffer.byteLength(head, 'utf-8')).toBeLessThanOrEqual(ENTRY_MAX_BYTES); + expect(text).toContain('…truncated'); + }); +}); + +describe('formatter — auto-redact', () => { + it('redacts top-level sensitive keys', () => { + const out = redactCtx({ + token: 'abc', + apiKey: 'def', + cookie: 'ghi', + password: 'jkl', + user: 'x', + }); + expect(out['token']).toBe('[REDACTED]'); + expect(out['apiKey']).toBe('[REDACTED]'); + expect(out['cookie']).toBe('[REDACTED]'); + expect(out['password']).toBe('[REDACTED]'); + expect(out['user']).toBe('x'); + }); + + it('redacts case- and separator-normalized keys', () => { + const out = redactCtx({ + API_KEY: '1', + access_token: '2', + 'Refresh-Token': '3', + Authorization: '4', + client_secret: '5', + api_secret: '6', + }); + expect(out['API_KEY']).toBe('[REDACTED]'); + expect(out['access_token']).toBe('[REDACTED]'); + expect(out['Refresh-Token']).toBe('[REDACTED]'); + expect(out['Authorization']).toBe('[REDACTED]'); + expect(out['client_secret']).toBe('[REDACTED]'); + expect(out['api_secret']).toBe('[REDACTED]'); + }); + + it('redacts common secret assignments inside raw string values', () => { + const { text } = formatEntry( + baseEntry({ + ctx: { + stderrTail: 'Authorization: Bearer abc123\napi_key=def456\ncookie: session=ghi789', + }, + }), + ); + expect(text).toContain('Authorization: Bearer [REDACTED]'); + expect(text).toContain('api_key=[REDACTED]'); + expect(text).toContain('cookie: [REDACTED]'); + expect(text).not.toContain('abc123'); + expect(text).not.toContain('def456'); + expect(text).not.toContain('ghi789'); + }); + + it('recurses into nested objects', () => { + const out = redactCtx({ headers: { Authorization: 'Bearer xxx', 'X-Trace': '1' } }); + const headers = out['headers'] as Record; + expect(headers['Authorization']).toBe('[REDACTED]'); + expect(headers['X-Trace']).toBe('1'); + }); + + it('recurses into arrays of objects', () => { + const out = redactCtx({ tokens: [{ token: 'a' }, { token: 'b' }] }); + const tokens = out['tokens'] as Array>; + expect(tokens[0]?.['token']).toBe('[REDACTED]'); + expect(tokens[1]?.['token']).toBe('[REDACTED]'); + }); + + it('collapses cycles to [REDACTED:cycle]', () => { + const a: Record = { name: 'a' }; + a['self'] = a; + const out = redactCtx({ a }); + const wrap = out['a'] as Record; + expect(wrap['self']).toBe('[REDACTED:cycle]'); + }); + + it('collapses deep nesting to [REDACTED:depth]', () => { + let leaf: Record = { n: 'leaf' }; + for (let i = 0; i < 20; i++) leaf = { down: leaf }; + const out = redactCtx({ chain: leaf }); + const json = JSON.stringify(out); + expect(json).toContain('[REDACTED:depth]'); + }); +}); + +describe('extractError', () => { + it('captures message and stack', () => { + const e = new Error('boom'); + const result = extractError(e); + expect(result.message).toBe('boom'); + expect(result.stack).toMatch(/Error: boom/); + }); +}); diff --git a/packages/agent-core-v2/test/log/logConfig.test.ts b/packages/agent-core-v2/test/log/logConfig.test.ts new file mode 100644 index 000000000..f63d0b250 --- /dev/null +++ b/packages/agent-core-v2/test/log/logConfig.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, it } from 'vitest'; + +import { + DEFAULT_GLOBAL_FILES, + DEFAULT_GLOBAL_MAX_BYTES, + DEFAULT_LOG_LEVEL, + DEFAULT_SESSION_FILES, + DEFAULT_SESSION_MAX_BYTES, + ILogOptions, + logSeed, + resolveGlobalLogPath, + resolveLoggingConfig, + resolveSessionLogPath, +} from '#/log/logConfig'; +import { createScopedTestHost } from '#/_base/di/test'; + +describe('resolveLoggingConfig', () => { + it('uses defaults when env is empty', () => { + const cfg = resolveLoggingConfig({ homeDir: '/home/kimi', env: {} }); + expect(cfg.level).toBe(DEFAULT_LOG_LEVEL); + expect(cfg.globalLogPath).toBe('/home/kimi/logs/kimi-code.log'); + expect(cfg.globalMaxBytes).toBe(DEFAULT_GLOBAL_MAX_BYTES); + expect(cfg.globalFiles).toBe(DEFAULT_GLOBAL_FILES); + expect(cfg.sessionMaxBytes).toBe(DEFAULT_SESSION_MAX_BYTES); + expect(cfg.sessionFiles).toBe(DEFAULT_SESSION_FILES); + }); + + it('reads level and sizes from env', () => { + const cfg = resolveLoggingConfig({ + homeDir: '/h', + env: { + KIMI_LOG_LEVEL: 'debug', + KIMI_LOG_GLOBAL_MAX_BYTES: '1024', + KIMI_LOG_GLOBAL_FILES: '7', + KIMI_LOG_SESSION_MAX_BYTES: '2048', + KIMI_LOG_SESSION_FILES: '4', + }, + }); + expect(cfg.level).toBe('debug'); + expect(cfg.globalMaxBytes).toBe(1024); + expect(cfg.globalFiles).toBe(7); + expect(cfg.sessionMaxBytes).toBe(2048); + expect(cfg.sessionFiles).toBe(4); + }); + + it('ignores invalid level and non-positive sizes', () => { + const cfg = resolveLoggingConfig({ + homeDir: '/h', + env: { + KIMI_LOG_LEVEL: 'verbose', + KIMI_LOG_GLOBAL_MAX_BYTES: '-5', + KIMI_LOG_GLOBAL_FILES: 'abc', + }, + }); + expect(cfg.level).toBe(DEFAULT_LOG_LEVEL); + expect(cfg.globalMaxBytes).toBe(DEFAULT_GLOBAL_MAX_BYTES); + expect(cfg.globalFiles).toBe(DEFAULT_GLOBAL_FILES); + }); + + it('falls back to process.env when env is omitted', () => { + const cfg = resolveLoggingConfig({ homeDir: '/h' }); + expect(cfg.globalLogPath).toBe('/h/logs/kimi-code.log'); + }); +}); + +describe('path resolution', () => { + it('resolves the global log path under homeDir/logs', () => { + expect(resolveGlobalLogPath('/home/kimi')).toBe('/home/kimi/logs/kimi-code.log'); + }); + + it('resolves the session log path under sessionDir/logs', () => { + expect(resolveSessionLogPath('/sessions/s1')).toBe('/sessions/s1/logs/kimi-code.log'); + }); +}); + +describe('logSeed', () => { + it('seeds ILogOptions into a Core scope', () => { + const cfg = resolveLoggingConfig({ homeDir: '/h', env: { KIMI_LOG_LEVEL: 'warn' } }); + const host = createScopedTestHost(logSeed(cfg)); + const opts = host.core.accessor.get(ILogOptions); + expect(opts.level).toBe('warn'); + expect(opts.globalLogPath).toBe('/h/logs/kimi-code.log'); + host.dispose(); + }); +}); diff --git a/packages/agent-core-v2/test/log/logService.test.ts b/packages/agent-core-v2/test/log/logService.test.ts index d0ae6a90e..79860ca04 100644 --- a/packages/agent-core-v2/test/log/logService.test.ts +++ b/packages/agent-core-v2/test/log/logService.test.ts @@ -1,22 +1,42 @@ -import { beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { InstantiationType } from '#/_base/di/extensions'; -import { LifecycleScope, _clearScopedRegistryForTests } from '#/_base/di/scope'; -import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { DisposableStore } from '#/_base/di/lifecycle'; import { - ConsoleLogSink, + LifecycleScope, + _clearScopedRegistryForTests, + registerScopedService, +} from '#/_base/di/scope'; +import { createScopedTestHost, createServices, stubPair } from '#/_base/di/test'; +import type { TestInstantiationService } from '#/_base/di/test'; +import { + ConsoleLogWriterService, ILogService, - ILogSink, + ILogWriterService, LogService, - MemoryLogSink, + MemoryLogWriterService, levelEnabled, } from '#/log/index'; -import { registerScopedService } from '#/_base/di/scope'; -describe('LogService (unit)', () => { +describe('LogService', () => { + let disposables: DisposableStore; + let ix: TestInstantiationService; + let sink: MemoryLogWriterService; + + beforeEach(() => { + disposables = new DisposableStore(); + sink = new MemoryLogWriterService(); + ix = createServices(disposables, { + additionalServices: (reg) => { + reg.defineInstance(ILogWriterService, sink); + reg.define(ILogService, LogService); + }, + }); + }); + afterEach(() => disposables.dispose()); + it('emits entries to the sink at/above the configured level', () => { - const sink = new MemoryLogSink(); - const log = new LogService(sink, {}, 'info'); + const log = ix.get(ILogService); log.debug('hidden'); log.info('hello'); log.warn('careful'); @@ -25,8 +45,7 @@ describe('LogService (unit)', () => { }); it('extracts Error payload onto entry.error', () => { - const sink = new MemoryLogSink(); - const log = new LogService(sink, {}, 'info'); + const log = ix.get(ILogService); const err = new Error('boom'); log.error('failed', err); expect(sink.entries[0]?.error?.message).toBe('boom'); @@ -34,15 +53,15 @@ describe('LogService (unit)', () => { }); it('merges object payload into ctx', () => { - const sink = new MemoryLogSink(); - const log = new LogService(sink, {}, 'debug'); + const log = ix.get(ILogService); + log.setLevel('debug'); log.info('with ctx', { requestId: 'r1', count: 2 }); expect(sink.entries[0]?.ctx).toEqual({ requestId: 'r1', count: 2 }); }); it('child merges bound context and bound wins over payload', () => { - const sink = new MemoryLogSink(); - const parent = new LogService(sink, {}, 'debug'); + const parent = ix.get(ILogService); + parent.setLevel('debug'); const child = parent.child({ sessionId: 's1', agentId: 'main' }); child.info('evt', { sessionId: 'override', extra: 'x' }); expect(sink.entries[0]?.ctx).toEqual({ @@ -53,21 +72,37 @@ describe('LogService (unit)', () => { }); it('child chains accumulate context', () => { - const sink = new MemoryLogSink(); - const root = new LogService(sink, {}, 'debug'); + const root = ix.get(ILogService); + root.setLevel('debug'); const leaf = root.child({ a: 1 }).child({ b: 2 }); leaf.info('evt'); expect(sink.entries[0]?.ctx).toEqual({ a: 1, b: 2 }); }); it('setLevel changes filtering at runtime', () => { - const sink = new MemoryLogSink(); - const log = new LogService(sink, {}, 'error'); + const log = ix.get(ILogService); + log.setLevel('error'); log.info('hidden'); log.setLevel('info'); log.info('shown'); expect(sink.entries.map((e) => e.msg)).toEqual(['shown']); }); + + it('flush delegates to the sink when present', async () => { + let flushed = false; + (sink as MemoryLogWriterService & { flush?: () => Promise }).flush = () => { + flushed = true; + return Promise.resolve(); + }; + const log = ix.get(ILogService); + await log.flush(); + expect(flushed).toBe(true); + }); + + it('flush resolves when the sink has no flush', async () => { + const log = ix.get(ILogService); + await expect(log.flush()).resolves.toBeUndefined(); + }); }); describe('levelEnabled', () => { @@ -84,8 +119,8 @@ describe('ILogService (scoped)', () => { _clearScopedRegistryForTests(); registerScopedService( LifecycleScope.Core, - ILogSink, - ConsoleLogSink, + ILogWriterService, + ConsoleLogWriterService, InstantiationType.Eager, 'log', ); @@ -99,8 +134,8 @@ describe('ILogService (scoped)', () => { }); it('resolves ILogService from the Core scope with its sink injected', () => { - const sink = new MemoryLogSink(); - const host = createScopedTestHost([stubPair(ILogSink, sink)]); + const sink = new MemoryLogWriterService(); + const host = createScopedTestHost([stubPair(ILogWriterService, sink)]); const log = host.core.accessor.get(ILogService); log.info('scoped-hello'); expect(sink.entries.map((e) => e.msg)).toEqual(['scoped-hello']); @@ -108,8 +143,8 @@ describe('ILogService (scoped)', () => { }); it('a scoped child logger bound to sessionId is resolvable downstream', () => { - const sink = new MemoryLogSink(); - const host = createScopedTestHost([stubPair(ILogSink, sink)]); + const sink = new MemoryLogWriterService(); + const host = createScopedTestHost([stubPair(ILogWriterService, sink)]); const root = host.core.accessor.get(ILogService); const sessionLog = root.child({ sessionId: 's1' }); sessionLog.warn('bound'); diff --git a/packages/agent-core-v2/test/log/logWriter.test.ts b/packages/agent-core-v2/test/log/logWriter.test.ts new file mode 100644 index 000000000..6a07f9af4 --- /dev/null +++ b/packages/agent-core-v2/test/log/logWriter.test.ts @@ -0,0 +1,206 @@ +import { mkdtemp, readFile, readdir, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'pathe'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { FileLogWriterService, PENDING_MAX, RotatingFileWriter } from '#/log/logWriter'; +import type { LogEntry } from '#/log/log'; + +let workDir: string; + +beforeEach(async () => { + workDir = await mkdtemp(join(tmpdir(), 'logger-sinks-')); +}); + +afterEach(async () => { + await rm(workDir, { recursive: true, force: true }); +}); + +async function listLogs(dir: string): Promise { + return (await readdir(dir)).toSorted(); +} + +describe('RotatingFileWriter', () => { + it('writes single line to active file', async () => { + const sink = new RotatingFileWriter({ + path: join(workDir, 'app.log'), + maxBytes: 1024, + files: 3, + }); + sink.enqueue('hello\n'); + await sink.flush(); + const text = await readFile(join(workDir, 'app.log'), 'utf-8'); + expect(text).toBe('hello\n'); + }); + + it('rotates when active file exceeds maxBytes', async () => { + const path = join(workDir, 'app.log'); + const sink = new RotatingFileWriter({ path, maxBytes: 64, files: 3 }); + for (let i = 0; i < 20; i++) { + sink.enqueue(`line${i} ${'x'.repeat(20)}\n`); + await sink.flush(); + } + const files = await listLogs(workDir); + expect(files).toContain('app.log'); + expect(files).toContain('app.log.1'); + }); + + it('evicts oldest archive after files=N rolls', async () => { + const path = join(workDir, 'app.log'); + const sink = new RotatingFileWriter({ path, maxBytes: 32, files: 2 }); + for (let i = 0; i < 50; i++) { + sink.enqueue(`${i.toString().padStart(3, '0')} ${'x'.repeat(30)}\n`); + await sink.flush(); + } + // Final write to ensure active file exists post-rotation + sink.enqueue('final\n'); + await sink.flush(); + const files = await listLogs(workDir); + expect(files).toEqual(expect.arrayContaining(['app.log'])); + // files = 2 → active + at most 1 archive; no app.log.2 or higher + expect(files.some((f) => /^app\.log\.[2-9]$/.test(f))).toBe(false); + }); + + it('rotates a large pending batch instead of writing it as one oversized file', async () => { + const path = join(workDir, 'app.log'); + const maxBytes = 128; + const sink = new RotatingFileWriter({ path, maxBytes, files: 3 }); + for (let i = 0; i < 24; i++) { + sink.enqueue(`line${i.toString().padStart(2, '0')} ${'x'.repeat(24)}\n`); + } + + await sink.flush(); + + const files = await listLogs(workDir); + expect(files).toContain('app.log.1'); + for (const file of files) { + expect((await stat(join(workDir, file))).size).toBeLessThanOrEqual(maxBytes); + } + }); + + it('drops oldest when pending overflows', async () => { + const path = join(workDir, 'app.log'); + const sink = new RotatingFileWriter({ path, maxBytes: 1_000_000, files: 2 }); + const over = PENDING_MAX + 500; + for (let i = 0; i < over; i++) { + sink.enqueue(`line${i}\n`); + } + await sink.flush(); + const text = await readFile(path, 'utf-8'); + expect(text).toMatch(/\.\.\. dropped \d+ entries \.\.\./); + // First lines (oldest) should be gone + expect(text).not.toContain('line0\n'); + // Latest should be present + expect(text).toContain(`line${over - 1}\n`); + }); + + it('does not throw when fs write fails; emits stderr notice', async () => { + const stderrSpy = vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + // Force failure by passing an invalid path char on POSIX + const badWriter = new RotatingFileWriter({ + path: '\0/invalid/path', + maxBytes: 1024, + files: 2, + }); + badWriter.enqueue('x\n'); + expect(await badWriter.flush()).toBe(false); + expect( + stderrSpy.mock.calls.some((c) => String(c[0]).includes('[logger] write failed')), + ).toBe(true); + stderrSpy.mockRestore(); + }); + + it('keeps restored pending bounded after repeated write failures', async () => { + const stderrSpy = vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + const badWriter = new RotatingFileWriter({ + path: '\0/invalid/path', + maxBytes: 1024, + files: 2, + }); + try { + for (let round = 0; round < 3; round++) { + for (let i = 0; i < PENDING_MAX + 25; i++) { + badWriter.enqueue(`round${round}-line${i}\n`); + } + expect(await badWriter.flush()).toBe(false); + } + const pending = (badWriter as unknown as { pending: readonly string[] }).pending; + expect(pending.length).toBeLessThanOrEqual(PENDING_MAX); + } finally { + stderrSpy.mockRestore(); + } + }); + + it('returns true when flush writes successfully', async () => { + const path = join(workDir, 'app.log'); + const sink = new RotatingFileWriter({ path, maxBytes: 1024, files: 2 }); + sink.enqueue('ok\n'); + expect(await sink.flush()).toBe(true); + }); + + it('serializes concurrent writes without interleaving lines', async () => { + const path = join(workDir, 'app.log'); + const sink = new RotatingFileWriter({ path, maxBytes: 10_000_000, files: 2 }); + const N = 500; + for (let i = 0; i < N; i++) { + sink.enqueue(`line${i.toString().padStart(4, '0')}\n`); + } + await sink.flush(); + const text = await readFile(path, 'utf-8'); + const lines = text.split('\n').filter((l) => l.length > 0); + expect(lines.length).toBe(N); + for (const line of lines) { + expect(line).toMatch(/^line\d{4}$/); + } + }); +}); + +describe('FileLogWriterService (ILogWriterService)', () => { + function entry(overrides: Partial = {}): LogEntry { + return { + t: Date.UTC(2026, 4, 19, 10, 12, 30, 123), + level: 'info', + msg: 'hello', + ...overrides, + }; + } + + it('formats entries as logfmt lines', async () => { + const path = join(workDir, 'app.log'); + const sink = new FileLogWriterService({ path, maxBytes: 1_000_000, files: 2 }); + sink.write(entry({ ctx: { requestId: 'r1' } })); + await sink.flush(); + const text = await readFile(path, 'utf-8'); + expect(text).toContain('INFO hello'); + expect(text).toContain('requestId=r1'); + await sink.close(); + }); + + it('redacts secret ctx values before writing', async () => { + const path = join(workDir, 'app.log'); + const sink = new FileLogWriterService({ path, maxBytes: 1_000_000, files: 2 }); + sink.write(entry({ ctx: { token: 'super-secret' } })); + await sink.flush(); + const text = await readFile(path, 'utf-8'); + expect(text).toContain('token=[REDACTED]'); + expect(text).not.toContain('super-secret'); + await sink.close(); + }); + + it('omits configured context keys', async () => { + const path = join(workDir, 'app.log'); + const sink = new FileLogWriterService({ + path, + maxBytes: 1_000_000, + files: 2, + format: { omitContextKeys: ['sessionId'] }, + }); + sink.write(entry({ ctx: { sessionId: 's1', requestId: 'r1' } })); + await sink.flush(); + const text = await readFile(path, 'utf-8'); + expect(text).not.toContain('sessionId'); + expect(text).toContain('requestId=r1'); + await sink.close(); + }); +}); diff --git a/packages/agent-core-v2/test/log/sessionLogService.test.ts b/packages/agent-core-v2/test/log/sessionLogService.test.ts new file mode 100644 index 000000000..a39378fd9 --- /dev/null +++ b/packages/agent-core-v2/test/log/sessionLogService.test.ts @@ -0,0 +1,128 @@ +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'pathe'; + +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { InstantiationType } from '#/_base/di/extensions'; +import { + LifecycleScope, + _clearScopedRegistryForTests, + registerScopedService, +} from '#/_base/di/scope'; +import { createScopedTestHost } from '#/_base/di/test'; +import { + ILogWriterService, + ISessionLogService, + resolveLoggingConfig, + resolveSessionLogPath, + sessionLogSeed, +} from '#/log/index'; +import { SessionFileLogWriterService } from '#/log/logWriter'; +import { logSeed } from '#/log/logConfig'; +import { SessionLogService } from '#/log/sessionLogService'; + +let homeDir: string; +let sessionDir: string; + +beforeEach(async () => { + _clearScopedRegistryForTests(); + registerScopedService( + LifecycleScope.Session, + ILogWriterService, + SessionFileLogWriterService, + InstantiationType.Delayed, + 'log', + ); + registerScopedService( + LifecycleScope.Session, + ISessionLogService, + SessionLogService, + InstantiationType.Delayed, + 'log', + ); + homeDir = await mkdtemp(join(tmpdir(), 'session-log-')); + sessionDir = join(homeDir, 'sessions', 's1'); +}); + +afterEach(async () => { + await rm(homeDir, { recursive: true, force: true }); +}); + +function buildHost() { + const cfg = resolveLoggingConfig({ + homeDir, + env: { KIMI_LOG_LEVEL: 'debug', KIMI_LOG_SESSION_MAX_BYTES: '1024', KIMI_LOG_SESSION_FILES: '2' }, + }); + return createScopedTestHost(logSeed(cfg)); +} + +async function readSessionLog(): Promise { + try { + return await readFile(resolveSessionLogPath(sessionDir), 'utf-8'); + } catch { + return ''; + } +} + +describe('SessionLogService', () => { + it('writes entries to the per-session log file', async () => { + const host = buildHost(); + const session = host.child(LifecycleScope.Session, 's1', sessionLogSeed('s1', sessionDir)); + const log = session.accessor.get(ISessionLogService); + log.info('session event', { requestId: 'r1' }); + await log.flush(); + const text = await readSessionLog(); + expect(text).toContain('session event'); + expect(text).toContain('requestId=r1'); + host.dispose(); + }); + + it('omits sessionId from per-session lines', async () => { + const host = buildHost(); + const session = host.child(LifecycleScope.Session, 's1', sessionLogSeed('s1', sessionDir)); + const log = session.accessor.get(ISessionLogService); + log.info('evt'); + await log.flush(); + const text = await readSessionLog(); + expect(text).not.toContain('sessionId'); + host.dispose(); + }); + + it('child logger accumulates context and writes to the same file', async () => { + const host = buildHost(); + const session = host.child(LifecycleScope.Session, 's1', sessionLogSeed('s1', sessionDir)); + const log = session.accessor.get(ISessionLogService); + log.child({ agentId: 'main' }).warn('child event'); + await log.flush(); + const text = await readSessionLog(); + expect(text).toContain('child event'); + expect(text).toContain('agentId=main'); + host.dispose(); + }); + + it('close flushes and a subsequent write is dropped', async () => { + const host = buildHost(); + const session = host.child(LifecycleScope.Session, 's1', sessionLogSeed('s1', sessionDir)); + const log = session.accessor.get(ISessionLogService); + log.info('before-close'); + await log.close(); + log.info('after-close'); + const text = await readSessionLog(); + expect(text).toContain('before-close'); + expect(text).not.toContain('after-close'); + host.dispose(); + }); + + it('dispose flushes pending entries synchronously', () => { + const host = buildHost(); + const session = host.child(LifecycleScope.Session, 's1', sessionLogSeed('s1', sessionDir)); + const log = session.accessor.get(ISessionLogService); + log.info('on-dispose'); + host.dispose(); + // dispose() is synchronous and uses flushSync; read after the call returns. + return readSessionLog().then((text) => { + expect(text).toContain('on-dispose'); + }); + }); +}); diff --git a/packages/agent-core-v2/test/log/stubs.ts b/packages/agent-core-v2/test/log/stubs.ts index 90ca1487a..235da252a 100644 --- a/packages/agent-core-v2/test/log/stubs.ts +++ b/packages/agent-core-v2/test/log/stubs.ts @@ -28,6 +28,7 @@ export function stubLog(): ILogService { _serviceBrand: undefined, level: 'info', setLevel: () => {}, + flush: () => Promise.resolve(), }; }