diff options
| author | Adam Malczewski <[email protected]> | 2026-06-05 14:29:06 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-05 14:29:06 +0900 |
| commit | 4d94c530406567791dbe4ab06c838a83c2e26023 (patch) | |
| tree | 4e15e5ffee8bf4de68ddac55462aa7dc04e67ee7 /packages/kernel | |
| parent | 9ae09aad5d8d6232c55932af0d496b888166065f (diff) | |
| download | dispatch-4d94c530406567791dbe4ab06c838a83c2e26023.tar.gz dispatch-4d94c530406567791dbe4ab06c838a83c2e26023.zip | |
refactor(observability): pure-types contracts/logging + Span body channel; verbatim before/after -> LogRecord.body (273 tests)
contracts/logging.ts reduced to pure types; createLogger (+ helpers) moved to kernel/src/logging/ — @dispatch/kernel still exports it (host-bin/tool-read-file unaffected).
Span body channel (Option A): Logger.span / Span.child / Span.end accept an optional body string -> SpanOpenRecord.body / SpanCloseRecord.body. Large verbatim payloads now use body, not stringified attributes (store-fat-serve-thin; attributes stay thin/queryable for D9).
before: run-turn emits a 'prompt' span with the verbatim messages+tools in body (small scalars in attrs). after: provider.request span carries the verbatim request in body; attrs thin, auth self-redacted.
Verified: tsc -b clean, 273 tests, biome 0 warnings/0 infos. Live boot: prompt + provider.request bodies present and correlated (shared turnId); request.body no longer in attributes; auth-key leak count = 0.
Diffstat (limited to 'packages/kernel')
| -rw-r--r-- | packages/kernel/src/contracts/index.ts | 1 | ||||
| -rw-r--r-- | packages/kernel/src/contracts/logging.ts | 294 | ||||
| -rw-r--r-- | packages/kernel/src/host/host.test.ts | 113 | ||||
| -rw-r--r-- | packages/kernel/src/host/host.ts | 2 | ||||
| -rw-r--r-- | packages/kernel/src/index.ts | 1 | ||||
| -rw-r--r-- | packages/kernel/src/logging/index.ts | 1 | ||||
| -rw-r--r-- | packages/kernel/src/logging/logger.ts | 299 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.test.ts | 52 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.ts | 25 |
9 files changed, 490 insertions, 298 deletions
diff --git a/packages/kernel/src/contracts/index.ts b/packages/kernel/src/contracts/index.ts index e4eba87..a4c965a 100644 --- a/packages/kernel/src/contracts/index.ts +++ b/packages/kernel/src/contracts/index.ts @@ -81,7 +81,6 @@ export type { SpanOpenRecord, SpanStatus, } from "./logging.js"; -export { createLogger } from "./logging.js"; export type { FinishEvent, ProviderContract, diff --git a/packages/kernel/src/contracts/logging.ts b/packages/kernel/src/contracts/logging.ts index 8e1eef3..a8bab7c 100644 --- a/packages/kernel/src/contracts/logging.ts +++ b/packages/kernel/src/contracts/logging.ts @@ -1,9 +1,9 @@ /** * Logging contract — structured, correlated, span-capable Logger/Span ABI. * - * The kernel owns types + pure record-builders. NO I/O — the LogSink is - * injected by the host-bin. Logger/Span mint records and call sink.emit; - * sink errors are swallowed (D7 — the turn is sovereign). + * PURE TYPES ONLY — no implementations. The createLogger factory lives in + * ../logging/logger.js. The kernel owns types + pure record-builders. + * NO I/O — the LogSink is injected by the host-bin. * * Key properties: * - P3-safe: correlation flows via explicit child()/span() values, no ambient. @@ -11,6 +11,7 @@ * - D3: spans emitted incrementally (open at span(), close at end()). * - D6: extensionId auto-stamped by host, not caller-supplied. * - Flat scalar attributes (serializable D3, queryable D9). + * - Optional `body` field on spans for large verbatim payloads (store-fat-serve-thin). */ // --- Levels --- @@ -53,12 +54,16 @@ export interface Span { reason?: string, ) => void; /** Open a child span nested under this one. */ - readonly child: (name: string, attrs?: Attributes) => Span; + readonly child: (name: string, attrs?: Attributes, body?: string) => Span; /** * Close this span. Records duration + status. Optionally records an - * error and/or additional attributes. + * error, additional attributes, and/or a body payload. */ - readonly end: (outcome?: { readonly err?: unknown; readonly attrs?: Attributes }) => void; + readonly end: (outcome?: { + readonly err?: unknown; + readonly attrs?: Attributes; + readonly body?: string; + }) => void; } // --- Logger --- @@ -80,7 +85,7 @@ export interface Logger { */ readonly child: (ctx: Partial<LogContext> & { readonly attrs?: Attributes }) => Logger; /** Open a new span. Emits a `span-open` record immediately (D3). */ - readonly span: (name: string, attrs?: Attributes) => Span; + readonly span: (name: string, attrs?: Attributes, body?: string) => Span; } /** @@ -184,278 +189,3 @@ export interface LogDeps { readonly now: () => number; readonly newId: () => string; } - -// --- Pure record builder (no I/O) --- - -/** - * Internal state carried by a logger instance. Built by createLogger; - * never exposed outside this module. - */ -interface LoggerState { - readonly ctx: LogContext; - readonly attrs: Attributes | undefined; - readonly deps: LogDeps; - readonly sink: LogSink; -} - -function mergeAttributes( - base: Attributes | undefined, - extra: Attributes | undefined, -): Attributes | undefined { - if (base === undefined && extra === undefined) return undefined; - if (base === undefined) return extra; - if (extra === undefined) return base; - return { ...base, ...extra }; -} - -function isScalarAttr(value: unknown): value is string | number | boolean | null { - const t = typeof value; - return t === "string" || t === "number" || t === "boolean" || value === null; -} - -function emitLog(state: LoggerState, level: Level, msg: string, attrs?: Attributes): void { - const merged = mergeAttributes(state.attrs, attrs); - const base = { - kind: "log" as const, - level, - msg, - timestamp: state.deps.now(), - extensionId: state.ctx.extensionId, - }; - const record: LogLineRecord = - state.ctx.conversationId !== undefined || - state.ctx.turnId !== undefined || - state.ctx.spanId !== undefined || - state.ctx.parentSpanId !== undefined || - merged !== undefined - ? { - ...base, - ...(state.ctx.conversationId !== undefined - ? { conversationId: state.ctx.conversationId } - : {}), - ...(state.ctx.turnId !== undefined ? { turnId: state.ctx.turnId } : {}), - ...(state.ctx.spanId !== undefined ? { spanId: state.ctx.spanId } : {}), - ...(state.ctx.parentSpanId !== undefined ? { parentSpanId: state.ctx.parentSpanId } : {}), - ...(merged !== undefined ? { attributes: merged } : {}), - } - : base; - try { - state.sink.emit(record); - } catch { - // Swallow — D7: the turn is sovereign (never break the caller). - } -} - -function buildSpanOpen( - state: LoggerState, - name: string, - spanId: string, - attrs?: Attributes, -): SpanOpenRecord { - const base = { - kind: "span-open" as const, - spanId, - name, - timestamp: state.deps.now(), - extensionId: state.ctx.extensionId, - }; - const merged = mergeAttributes(state.attrs, attrs); - return { - ...base, - ...(state.ctx.conversationId !== undefined ? { conversationId: state.ctx.conversationId } : {}), - ...(state.ctx.turnId !== undefined ? { turnId: state.ctx.turnId } : {}), - ...(state.ctx.parentSpanId !== undefined ? { parentSpanId: state.ctx.parentSpanId } : {}), - ...(merged !== undefined ? { attributes: merged } : {}), - }; -} - -function buildSpanLink( - target: { readonly spanId: string; readonly turnId?: string }, - reason?: string, -): SpanLink { - return { - spanId: target.spanId, - ...(target.turnId !== undefined ? { turnId: target.turnId } : {}), - ...(reason !== undefined ? { reason } : {}), - }; -} - -/** - * Create a structured Logger. Pure factory — all I/O goes through the - * injected sink. `{ now, newId }` are injected for deterministic tests. - * - * @param ctx Initial correlation context (extensionId + optional ids). - * @param sink Fire-and-forget record sink. - * @param deps Clock + id generator. - * @param attrs Optional default attributes (from child()). - */ -export function createLogger( - ctx: LogContext, - sink: LogSink, - deps: LogDeps, - attrs?: Attributes, -): Logger { - const state: LoggerState = { ctx, attrs, deps, sink }; - - function makeSpan(name: string, spanAttrs?: Attributes, parentSpanId?: string): Span { - const spanId = deps.newId(); - const mergedParent = parentSpanId ?? state.ctx.spanId; - const spanCtx: LogContext = { - extensionId: ctx.extensionId, - ...(ctx.conversationId !== undefined ? { conversationId: ctx.conversationId } : {}), - ...(ctx.turnId !== undefined ? { turnId: ctx.turnId } : {}), - spanId, - ...(mergedParent !== undefined ? { parentSpanId: mergedParent } : {}), - }; - - const openRecord = buildSpanOpen(state, name, spanId, spanAttrs); - const spanAttrsMutable: Record<string, string | number | boolean | null> = - spanAttrs !== undefined ? { ...spanAttrs } : {}; - const links: SpanLink[] = []; - const openedAt = deps.now(); - - try { - sink.emit(openRecord); - } catch { - // Swallow — D7. - } - - const spanLogger = createLogger(spanCtx, sink, deps, state.attrs); - - const span: Span = { - id: spanId, - log: spanLogger, - setAttributes(newAttrs: Attributes): void { - for (const [key, value] of Object.entries(newAttrs)) { - spanAttrsMutable[key] = value; - } - }, - addLink(target, reason): void { - links.push(buildSpanLink(target, reason)); - }, - child(childName: string, childAttrs?: Attributes): Span { - return makeSpan(childName, childAttrs, spanId); - }, - end(outcome?): void { - const closedAt = deps.now(); - const err = outcome?.err; - let status: SpanStatus = "ok"; - if (err !== undefined && err !== null) { - status = "error"; - const errMsg = err instanceof Error ? err.message : String(err); - spanAttrsMutable["error.message"] = errMsg; - if (err instanceof Error && err.stack !== undefined) { - spanAttrsMutable["error.stack"] = err.stack; - } - } - if (outcome?.attrs !== undefined) { - for (const [key, value] of Object.entries(outcome.attrs)) { - spanAttrsMutable[key] = value; - } - } - - const hasAttrs = Object.keys(spanAttrsMutable).length > 0; - const hasLinks = links.length > 0; - const base = { - kind: "span-close" as const, - spanId, - name, - timestamp: closedAt, - durationMs: closedAt - openedAt, - status, - extensionId: ctx.extensionId, - }; - const closeRecord: SpanCloseRecord = { - ...base, - ...(ctx.conversationId !== undefined ? { conversationId: ctx.conversationId } : {}), - ...(ctx.turnId !== undefined ? { turnId: ctx.turnId } : {}), - ...(mergedParent !== undefined ? { parentSpanId: mergedParent } : {}), - ...(hasAttrs ? { attributes: { ...spanAttrsMutable } } : {}), - ...(hasLinks ? { links: [...links] } : {}), - }; - try { - sink.emit(closeRecord); - } catch { - // Swallow — D7. - } - }, - }; - - return span; - } - - const logger: Logger = { - debug(msg: string, attrs?: Attributes): void { - emitLog(state, "debug", msg, attrs); - }, - info(msg: string, attrs?: Attributes): void { - emitLog(state, "info", msg, attrs); - }, - warn(msg: string, attrs?: Attributes): void { - emitLog(state, "warn", msg, attrs); - }, - error(msg: string, attrs?: ErrorAttributes): void { - const err = attrs?.err; - if (err !== undefined && err !== null) { - // Extract scalar attributes (everything except err). - const scalarAttrs: Record<string, string | number | boolean | null> = {}; - if (attrs !== undefined) { - for (const [key, value] of Object.entries(attrs)) { - if (key !== "err" && isScalarAttr(value)) { - scalarAttrs[key] = value; - } - } - } - const merged = mergeAttributes( - state.attrs, - Object.keys(scalarAttrs).length > 0 ? scalarAttrs : undefined, - ); - const errMsg = err instanceof Error ? err.message : String(err); - const errorAttrs: Record<string, string | number | boolean | null> = { - ...(merged ?? {}), - "error.message": errMsg, - }; - if (err instanceof Error && err.stack !== undefined) { - errorAttrs["error.stack"] = err.stack; - } - emitLog(state, "error", msg, errorAttrs as Attributes); - } else { - // No err field — filter to scalar attributes only. - const scalarAttrs: Record<string, string | number | boolean | null> = {}; - if (attrs !== undefined) { - for (const [key, value] of Object.entries(attrs)) { - if (isScalarAttr(value)) { - scalarAttrs[key] = value; - } - } - } - emitLog( - state, - "error", - msg, - Object.keys(scalarAttrs).length > 0 ? (scalarAttrs as Attributes) : undefined, - ); - } - }, - child(childCtx: Partial<LogContext> & { readonly attrs?: Attributes }): Logger { - const convId = childCtx.conversationId ?? ctx.conversationId; - const tId = childCtx.turnId ?? ctx.turnId; - const sId = childCtx.spanId ?? ctx.spanId; - const pId = childCtx.parentSpanId ?? ctx.parentSpanId; - const newCtx: LogContext = { - extensionId: ctx.extensionId, - ...(convId !== undefined ? { conversationId: convId } : {}), - ...(tId !== undefined ? { turnId: tId } : {}), - ...(sId !== undefined ? { spanId: sId } : {}), - ...(pId !== undefined ? { parentSpanId: pId } : {}), - }; - const newAttrs = mergeAttributes(state.attrs, childCtx.attrs); - return createLogger(newCtx, sink, deps, newAttrs); - }, - span(name: string, attrs?: Attributes): Span { - return makeSpan(name, attrs); - }, - }; - - return logger; -} diff --git a/packages/kernel/src/host/host.test.ts b/packages/kernel/src/host/host.test.ts index 7688366..430447c 100644 --- a/packages/kernel/src/host/host.test.ts +++ b/packages/kernel/src/host/host.test.ts @@ -956,5 +956,118 @@ describe("createHost", () => { expect(spanCloses[0].attributes?.result).toBe("ok"); } }); + + it("span() with body emits body on span-open record", async () => { + let extLogger: Logger | undefined; + + const ext = createExtension("ext", { + activate: (host) => { + extLogger = host.logger; + }, + }); + + const host = createHost([ext], deps); + await host.activate(); + + const span = extLogger?.span("with-body", { key: "value" }, '{"payload":"hello"}'); + span?.end(); + + const spanOpens = logSink.records.filter((r) => r.kind === "span-open"); + expect(spanOpens).toHaveLength(1); + if (spanOpens[0]?.kind === "span-open") { + expect(spanOpens[0].body).toBe('{"payload":"hello"}'); + } + }); + + it("span() without body omits body field on span-open record", async () => { + let extLogger: Logger | undefined; + + const ext = createExtension("ext", { + activate: (host) => { + extLogger = host.logger; + }, + }); + + const host = createHost([ext], deps); + await host.activate(); + + const span = extLogger?.span("no-body"); + span?.end(); + + const spanOpens = logSink.records.filter((r) => r.kind === "span-open"); + expect(spanOpens).toHaveLength(1); + if (spanOpens[0]?.kind === "span-open") { + expect(spanOpens[0].body).toBeUndefined(); + } + }); + + it("child() with body emits body on child span-open record", async () => { + let extLogger: Logger | undefined; + + const ext = createExtension("ext", { + activate: (host) => { + extLogger = host.logger; + }, + }); + + const host = createHost([ext], deps); + await host.activate(); + + const span = extLogger?.span("parent"); + const child = span?.child("child-name", { k: "v" }, '{"child":"body"}'); + child?.end(); + span?.end(); + + const spanOpens = logSink.records.filter((r) => r.kind === "span-open"); + const childOpen = spanOpens.find((r) => r.kind === "span-open" && r.name === "child-name"); + expect(childOpen).toBeDefined(); + if (childOpen?.kind === "span-open") { + expect(childOpen.body).toBe('{"child":"body"}'); + } + }); + + it("end() with body emits body on span-close record", async () => { + let extLogger: Logger | undefined; + + const ext = createExtension("ext", { + activate: (host) => { + extLogger = host.logger; + }, + }); + + const host = createHost([ext], deps); + await host.activate(); + + const span = extLogger?.span("close-body"); + span?.end({ body: '{"result":"data"}' }); + + const spanCloses = logSink.records.filter((r) => r.kind === "span-close"); + expect(spanCloses).toHaveLength(1); + if (spanCloses[0]?.kind === "span-close") { + expect(spanCloses[0].body).toBe('{"result":"data"}'); + } + }); + + it("end() without body omits body field on span-close record", async () => { + let extLogger: Logger | undefined; + + const ext = createExtension("ext", { + activate: (host) => { + extLogger = host.logger; + }, + }); + + const host = createHost([ext], deps); + await host.activate(); + + const span = extLogger?.span("no-close-body"); + span?.end(); + + const spanCloses = logSink.records.filter((r) => r.kind === "span-close"); + expect(spanCloses).toHaveLength(1); + if (spanCloses[0]?.kind === "span-close") { + expect(spanCloses[0].body).toBeUndefined(); + } + }); }); }); diff --git a/packages/kernel/src/host/host.ts b/packages/kernel/src/host/host.ts index c7ec7a9..2331625 100644 --- a/packages/kernel/src/host/host.ts +++ b/packages/kernel/src/host/host.ts @@ -19,9 +19,9 @@ import type { ServiceHandle, } from "../contracts/hooks.js"; import type { LogDeps, Logger, LogSink } from "../contracts/logging.js"; -import { createLogger } from "../contracts/logging.js"; import type { ProviderContract } from "../contracts/provider.js"; import type { ToolContract } from "../contracts/tool.js"; +import { createLogger } from "../logging/logger.js"; import { resolveActivationOrder } from "./dag.js"; import { isApiVersionCompatible } from "./version.js"; diff --git a/packages/kernel/src/index.ts b/packages/kernel/src/index.ts index fc5d1ab..71a9d11 100644 --- a/packages/kernel/src/index.ts +++ b/packages/kernel/src/index.ts @@ -6,4 +6,5 @@ export * from "./bus/index.js"; export * from "./contracts/index.js"; export * from "./host/index.js"; +export * from "./logging/index.js"; export * from "./runtime/index.js"; diff --git a/packages/kernel/src/logging/index.ts b/packages/kernel/src/logging/index.ts new file mode 100644 index 0000000..71c723c --- /dev/null +++ b/packages/kernel/src/logging/index.ts @@ -0,0 +1 @@ +export { createLogger } from "./logger.js"; diff --git a/packages/kernel/src/logging/logger.ts b/packages/kernel/src/logging/logger.ts new file mode 100644 index 0000000..507ace6 --- /dev/null +++ b/packages/kernel/src/logging/logger.ts @@ -0,0 +1,299 @@ +/** + * Logger implementation — pure record-builder over an injected LogSink. + * + * All I/O goes through the sink. `{ now, newId }` are injected for + * deterministic tests (P2). No ambient state (P3). + */ + +import type { + Attributes, + ErrorAttributes, + Level, + LogContext, + LogDeps, + Logger, + LogLineRecord, + LogSink, + Span, + SpanCloseRecord, + SpanLink, + SpanOpenRecord, + SpanStatus, +} from "../contracts/logging.js"; + +interface LoggerState { + readonly ctx: LogContext; + readonly attrs: Attributes | undefined; + readonly deps: LogDeps; + readonly sink: LogSink; +} + +function mergeAttributes( + base: Attributes | undefined, + extra: Attributes | undefined, +): Attributes | undefined { + if (base === undefined && extra === undefined) return undefined; + if (base === undefined) return extra; + if (extra === undefined) return base; + return { ...base, ...extra }; +} + +function isScalarAttr(value: unknown): value is string | number | boolean | null { + const t = typeof value; + return t === "string" || t === "number" || t === "boolean" || value === null; +} + +function emitLog(state: LoggerState, level: Level, msg: string, attrs?: Attributes): void { + const merged = mergeAttributes(state.attrs, attrs); + const base = { + kind: "log" as const, + level, + msg, + timestamp: state.deps.now(), + extensionId: state.ctx.extensionId, + }; + const record: LogLineRecord = + state.ctx.conversationId !== undefined || + state.ctx.turnId !== undefined || + state.ctx.spanId !== undefined || + state.ctx.parentSpanId !== undefined || + merged !== undefined + ? { + ...base, + ...(state.ctx.conversationId !== undefined + ? { conversationId: state.ctx.conversationId } + : {}), + ...(state.ctx.turnId !== undefined ? { turnId: state.ctx.turnId } : {}), + ...(state.ctx.spanId !== undefined ? { spanId: state.ctx.spanId } : {}), + ...(state.ctx.parentSpanId !== undefined ? { parentSpanId: state.ctx.parentSpanId } : {}), + ...(merged !== undefined ? { attributes: merged } : {}), + } + : base; + try { + state.sink.emit(record); + } catch { + // Swallow — D7: the turn is sovereign (never break the caller). + } +} + +function buildSpanOpen( + state: LoggerState, + name: string, + spanId: string, + attrs?: Attributes, + body?: string, +): SpanOpenRecord { + const base = { + kind: "span-open" as const, + spanId, + name, + timestamp: state.deps.now(), + extensionId: state.ctx.extensionId, + }; + const merged = mergeAttributes(state.attrs, attrs); + return { + ...base, + ...(state.ctx.conversationId !== undefined ? { conversationId: state.ctx.conversationId } : {}), + ...(state.ctx.turnId !== undefined ? { turnId: state.ctx.turnId } : {}), + ...(state.ctx.parentSpanId !== undefined ? { parentSpanId: state.ctx.parentSpanId } : {}), + ...(merged !== undefined ? { attributes: merged } : {}), + ...(body !== undefined ? { body } : {}), + }; +} + +function buildSpanLink( + target: { readonly spanId: string; readonly turnId?: string }, + reason?: string, +): SpanLink { + return { + spanId: target.spanId, + ...(target.turnId !== undefined ? { turnId: target.turnId } : {}), + ...(reason !== undefined ? { reason } : {}), + }; +} + +/** + * Create a structured Logger. Pure factory — all I/O goes through the + * injected sink. `{ now, newId }` are injected for deterministic tests. + * + * @param ctx Initial correlation context (extensionId + optional ids). + * @param sink Fire-and-forget record sink. + * @param deps Clock + id generator. + * @param attrs Optional default attributes (from child()). + */ +export function createLogger( + ctx: LogContext, + sink: LogSink, + deps: LogDeps, + attrs?: Attributes, +): Logger { + const state: LoggerState = { ctx, attrs, deps, sink }; + + function makeSpan( + name: string, + spanAttrs?: Attributes, + parentSpanId?: string, + body?: string, + ): Span { + const spanId = deps.newId(); + const mergedParent = parentSpanId ?? state.ctx.spanId; + const spanCtx: LogContext = { + extensionId: ctx.extensionId, + ...(ctx.conversationId !== undefined ? { conversationId: ctx.conversationId } : {}), + ...(ctx.turnId !== undefined ? { turnId: ctx.turnId } : {}), + spanId, + ...(mergedParent !== undefined ? { parentSpanId: mergedParent } : {}), + }; + + const openRecord = buildSpanOpen(state, name, spanId, spanAttrs, body); + const spanAttrsMutable: Record<string, string | number | boolean | null> = + spanAttrs !== undefined ? { ...spanAttrs } : {}; + const links: SpanLink[] = []; + const openedAt = deps.now(); + + try { + sink.emit(openRecord); + } catch { + // Swallow — D7. + } + + const spanLogger = createLogger(spanCtx, sink, deps, state.attrs); + + const span: Span = { + id: spanId, + log: spanLogger, + setAttributes(newAttrs: Attributes): void { + for (const [key, value] of Object.entries(newAttrs)) { + spanAttrsMutable[key] = value; + } + }, + addLink(target, reason): void { + links.push(buildSpanLink(target, reason)); + }, + child(childName: string, childAttrs?: Attributes, childBody?: string): Span { + return makeSpan(childName, childAttrs, spanId, childBody); + }, + end(outcome?): void { + const closedAt = deps.now(); + const err = outcome?.err; + let status: SpanStatus = "ok"; + if (err !== undefined && err !== null) { + status = "error"; + const errMsg = err instanceof Error ? err.message : String(err); + spanAttrsMutable["error.message"] = errMsg; + if (err instanceof Error && err.stack !== undefined) { + spanAttrsMutable["error.stack"] = err.stack; + } + } + if (outcome?.attrs !== undefined) { + for (const [key, value] of Object.entries(outcome.attrs)) { + spanAttrsMutable[key] = value; + } + } + + const hasAttrs = Object.keys(spanAttrsMutable).length > 0; + const hasLinks = links.length > 0; + const base = { + kind: "span-close" as const, + spanId, + name, + timestamp: closedAt, + durationMs: closedAt - openedAt, + status, + extensionId: ctx.extensionId, + }; + const closeRecord: SpanCloseRecord = { + ...base, + ...(ctx.conversationId !== undefined ? { conversationId: ctx.conversationId } : {}), + ...(ctx.turnId !== undefined ? { turnId: ctx.turnId } : {}), + ...(mergedParent !== undefined ? { parentSpanId: mergedParent } : {}), + ...(hasAttrs ? { attributes: { ...spanAttrsMutable } } : {}), + ...(hasLinks ? { links: [...links] } : {}), + ...(outcome?.body !== undefined ? { body: outcome.body } : {}), + }; + try { + sink.emit(closeRecord); + } catch { + // Swallow — D7. + } + }, + }; + + return span; + } + + const logger: Logger = { + debug(msg: string, attrs?: Attributes): void { + emitLog(state, "debug", msg, attrs); + }, + info(msg: string, attrs?: Attributes): void { + emitLog(state, "info", msg, attrs); + }, + warn(msg: string, attrs?: Attributes): void { + emitLog(state, "warn", msg, attrs); + }, + error(msg: string, attrs?: ErrorAttributes): void { + const err = attrs?.err; + if (err !== undefined && err !== null) { + // Extract scalar attributes (everything except err). + const scalarAttrs: Record<string, string | number | boolean | null> = {}; + if (attrs !== undefined) { + for (const [key, value] of Object.entries(attrs)) { + if (key !== "err" && isScalarAttr(value)) { + scalarAttrs[key] = value; + } + } + } + const merged = mergeAttributes( + state.attrs, + Object.keys(scalarAttrs).length > 0 ? scalarAttrs : undefined, + ); + const errMsg = err instanceof Error ? err.message : String(err); + const errorAttrs: Record<string, string | number | boolean | null> = { + ...(merged ?? {}), + "error.message": errMsg, + }; + if (err instanceof Error && err.stack !== undefined) { + errorAttrs["error.stack"] = err.stack; + } + emitLog(state, "error", msg, errorAttrs as Attributes); + } else { + // No err field — filter to scalar attributes only. + const scalarAttrs: Record<string, string | number | boolean | null> = {}; + if (attrs !== undefined) { + for (const [key, value] of Object.entries(attrs)) { + if (isScalarAttr(value)) { + scalarAttrs[key] = value; + } + } + } + emitLog( + state, + "error", + msg, + Object.keys(scalarAttrs).length > 0 ? (scalarAttrs as Attributes) : undefined, + ); + } + }, + child(childCtx: Partial<LogContext> & { readonly attrs?: Attributes }): Logger { + const convId = childCtx.conversationId ?? ctx.conversationId; + const tId = childCtx.turnId ?? ctx.turnId; + const sId = childCtx.spanId ?? ctx.spanId; + const pId = childCtx.parentSpanId ?? ctx.parentSpanId; + const newCtx: LogContext = { + extensionId: ctx.extensionId, + ...(convId !== undefined ? { conversationId: convId } : {}), + ...(tId !== undefined ? { turnId: tId } : {}), + ...(sId !== undefined ? { spanId: sId } : {}), + ...(pId !== undefined ? { parentSpanId: pId } : {}), + }; + const newAttrs = mergeAttributes(state.attrs, childCtx.attrs); + return createLogger(newCtx, sink, deps, newAttrs); + }, + span(name: string, attrs?: Attributes, body?: string): Span { + return makeSpan(name, attrs, undefined, body); + }, + }; + + return logger; +} diff --git a/packages/kernel/src/runtime/run-turn.test.ts b/packages/kernel/src/runtime/run-turn.test.ts index 667476f..9c19027 100644 --- a/packages/kernel/src/runtime/run-turn.test.ts +++ b/packages/kernel/src/runtime/run-turn.test.ts @@ -2,9 +2,9 @@ import { describe, expect, it } from "vitest"; import type { ChatMessage } from "../contracts/conversation.js"; import type { AgentEvent } from "../contracts/events.js"; import type { LogDeps, Logger, LogRecord, LogSink } from "../contracts/logging.js"; -import { createLogger } from "../contracts/logging.js"; import type { ProviderContract, ProviderEvent } from "../contracts/provider.js"; import type { ToolContract, ToolExecuteContext, ToolResult } from "../contracts/tool.js"; +import { createLogger } from "../logging/logger.js"; import { runTurn } from "./run-turn.js"; function delay(ms: number): Promise<void> { @@ -1039,6 +1039,56 @@ describe("runTurn", () => { expect(stepCloses[0].attributes?.["error.message"]).toContain("provider exploded"); } }); + + it("emits a prompt span with verbatim body and small scalar attributes", async () => { + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + const provider = createFakeProvider([ + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const { logger, sink } = createTestLogger(); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + logger, + }); + + const promptOpens = sink.records.filter((r) => r.kind === "span-open" && r.name === "prompt"); + expect(promptOpens).toHaveLength(1); + + const promptOpen = promptOpens[0]; + if (promptOpen?.kind === "span-open") { + expect(promptOpen.body).toBeDefined(); + const parsed = JSON.parse(promptOpen.body as string); + expect(parsed.messages).toEqual([userMessage]); + expect(parsed.tools).toHaveLength(1); + expect(parsed.tools[0].name).toBe("echo"); + + expect(promptOpen.attributes?.messageCount).toBe(1); + expect(promptOpen.attributes?.toolCount).toBe(1); + } + + const promptCloses = sink.records.filter( + (r) => r.kind === "span-close" && r.name === "prompt", + ); + expect(promptCloses).toHaveLength(1); + + const logRecords = sink.records.filter( + (r) => + r.kind === "log" && r.kind === "log" && (r as { msg: string }).msg === "prompt:before", + ); + expect(logRecords).toHaveLength(0); + }); }); describe("provider logger threading", () => { diff --git a/packages/kernel/src/runtime/run-turn.ts b/packages/kernel/src/runtime/run-turn.ts index a78c31d..5e60641 100644 --- a/packages/kernel/src/runtime/run-turn.ts +++ b/packages/kernel/src/runtime/run-turn.ts @@ -167,22 +167,21 @@ async function executeStep(ctx: StepContext): Promise<StepResult> { let stepUsage = zeroUsage(); let finishReason = "stop"; - // Open a step span with the verbatim pre-mutation prompt in its body (BEFORE capture). + // Open a step span; capture the verbatim pre-mutation prompt via a + // "prompt" child span whose body holds the serialized messages+tools. let stepSpan: Span | undefined; try { stepSpan = ctx.logger.span("step"); - // Emit the verbatim pre-mutation prompt as a log record on the step span's logger. - // This is the "BEFORE" capture — the messages + tools as handed to provider.stream. - stepSpan.log.info("prompt:before", { - "prompt.messages": JSON.stringify(ctx.messages), - "prompt.tools": JSON.stringify( - ctx.tools.map((t) => ({ - name: t.name, - description: t.description, - parameters: t.parameters, - })), - ), - }); + const promptBody = JSON.stringify({ messages: ctx.messages, tools: ctx.tools }); + const promptSpan = stepSpan.child( + "prompt", + { + messageCount: ctx.messages.length, + toolCount: ctx.tools.length, + }, + promptBody, + ); + promptSpan.end(); } catch { // Swallow — D7. } |
