diff --git a/.agents/skills/agent-core-dev/SKILL.md b/.agents/skills/agent-core-dev/SKILL.md new file mode 100644 index 000000000..8efbe0ab9 --- /dev/null +++ b/.agents/skills/agent-core-dev/SKILL.md @@ -0,0 +1,63 @@ +--- +name: agent-core-dev +description: Use when developing in packages/agent-core-v2 (the DI × Scope agent engine) — adding or modifying a domain Service, choosing a LifecycleScope, wiring DI dependencies, splitting a domain across scopes, gating behavior behind an experimental flag, raising coded errors, working on the permission system, writing DI/Scope tests, or porting business logic from agent-core (v1) to v2. Self-contained guide organized by development stage (orient → design → implement → test → verify) plus an align workflow for v1→v2 migration; each file carries the rules, examples, and red lines for its step. +--- + +# agent-core-dev + +> Develop `packages/agent-core-v2` by lifecycle stage. This skill is **self-contained**: every rule, recipe, and red line lives in the stage files below — it does not delegate to `packages/agent-core-v2/docs/`. + +`agent-core-v2` is the new agent engine built on the **DI × Scope** architecture (a port of `packages/agent-core`). Everything resolves through the container: a service declares an **identity**, its **dependencies**, and a **lifetime**; the container decides construction, singleton-per-scope, ordering, and disposal. The stage files restate the rules in imperative form so you can work without reading the source docs. + +## Lifecycle at a glance + +```text +Orient → Design → Implement → Test → Verify + │ │ │ │ │ + │ │ │ │ └─ lint:domain · typecheck · test · dep graph · red lines + │ │ │ └─ test.md + │ │ └─ implement.md (+ errors.md · flags.md · permission.md) + │ └─ design.md + └─ orient.md +``` + +Stages are ordered but not strictly linear: a test failure (stage 4) that reveals a wrong scope sends you back to design (stage 2); a `CyclicDependencyError` sends you to `design.md` §dependency-direction and `implement.md` §cycles. + +## Workflows + +End-to-end procedures that span the stages. Reach for these before reading the stage files individually. + +- [Align (port `agent-core` → `agent-core-v2`)](align.md): split a v1 class into semantic units, fix each unit's domain / scope / Service / dependencies, then migrate the logic and tests. Use when the task is "move feature X from v1 to v2" or "port `IXxxService` to v2". + +## Stages + +- [Stage 1 — Orient](orient.md): the DI black box (identity / dependencies / lifetime), the four `LifecycleScope` tiers and visibility, and the file-header comment convention. Read before touching business code. +- [Stage 2 — Design a service](design.md): pick a scope, split a domain across scopes, choose a calling style (direct call vs event vs hook), and direct dependencies. Decide *where things live and who knows whom* before coding. +- [Stage 3 — Implement](implement.md): the standard Service recipe and the DI building blocks — interface + identity, constructor injection, scoped registration, `Disposable`, eager vs delayed, `invokeFunction`, `createInstance`, child scopes, and the cycle-refactor playbook. + - Topic: [Service authoring](service-authoring.md) — file layout, naming, contract vs impl contents, interface style, constructor/field conventions, events, multi-Service domains, comment rules. + - Topic: [Errors](errors.md) — co-located `XxxError`, the central code registry, wire serialization, boundary translation. + - Topic: [Flags](flags.md) — `FLAG_DEFINITIONS`, `IFlagService.enabled(id)`, the `[experimental]` config section, resolution precedence. + - Topic: [Permission](permission.md) — composable chain-of-responsibility kernel, policy registry + composer, `modes`/`agentTypes` metadata, `resolveExecution`/`accesses`. + - Topic: [Telemetry](telemetry.md) — emitting events via `ITelemetryService`, context propagation, and appender destinations (`ConsoleAppender` / `CloudAppender`). +- [Stage 4 — Test](test.md): resolve the system under test by interface, pick `TestInstantiationService` vs `createScopedTestHost`, shared stubs, service groups, teardown. +- [Stage 5 — Verify & submit](verify.md): `lint:domain`, `typecheck`, `test`, updating the DI × Scope dependency map, and the pre-submit checklist. + +## How to use this skill + +Jump to the stage you are in and read that one file; each is self-contained and ends with its own red lines. Skim the global red lines below before submitting — they catch most mistakes across every stage. The repo's source of truth remains the code in `packages/agent-core-v2/src/`; this skill codifies the same rules so you do not have to re-derive them. + +## Global red lines + +Invariants that hold across every stage. Each is expanded in the stage file noted. + +1. No `new` on a class whose constructor carries `@IService` deps — inject with `@IX` or `accessor.get(IX)`. (implement.md) +2. `@IX` decorates constructor parameters only; parameter order depends on construction (static-first for `createInstance`, `@IX`-first for scoped services). (service-authoring.md) +3. Both interface and impl carry `_serviceBrand`; the `createDecorator` name is globally unique. (implement.md) +4. Parent scope never depends on child scope — short-lived may inject long-lived, never the reverse. (orient.md) +5. No cyclic dependencies — refactor (extract a third Service / use an event / re-scope); do not break the cycle with `Delayed`. (design.md, implement.md) +6. `ServicesAccessor` is valid only during `invokeFunction` — never stash it for async use. (implement.md) +7. Scope follows state identity — no `Map` at `Core` to fake per-session state. (design.md) +8. Foundational layers never know upstream ones; business code never depends on the edge layer (`gateway`/`rpc`). (design.md) +9. Throw coded errors; register codes centrally; branch on `code` across the wire, never `instanceof`. (errors.md) +10. Gate unreleased behavior behind a `FLAG_DEFINITIONS` flag; no ad-hoc env toggles. (flags.md) +11. Tests resolve the SUT by interface; shared stubs live under `test/`, never `src/`. (test.md) diff --git a/.agents/skills/agent-core-dev/telemetry.md b/.agents/skills/agent-core-dev/telemetry.md new file mode 100644 index 000000000..ba7ce5db7 --- /dev/null +++ b/.agents/skills/agent-core-dev/telemetry.md @@ -0,0 +1,92 @@ +# Topic — Telemetry + +Telemetry infrastructure for agent-core-v2: how business services emit events, how context propagates, and how events reach a destination through appenders. + +Telemetry is a **layer-1 root** domain (alongside `log`): pure `Core` scope, stateless, no business-domain dependencies. It is a thin facade — enrichment, batching, and transport belong to the appenders, not to this layer. + +## Where things live + +- `src/telemetry/telemetry.ts`: contract — `ITelemetryService` (facade), `ITelemetryAppender` (destination), `TelemetryProperties`, `nullTelemetryAppender`, and `TelemetryServiceOptions`. +- `src/telemetry/telemetryService.ts`: `TelemetryService` impl + `registerScopedService(LifecycleScope.Core, …)`. +- `src/telemetry/consoleAppender.ts`: `ConsoleAppender` — echoes events to a log function (dev / debug). +- `src/telemetry/cloudAppender.ts`: `CloudAppender` — batches + enriches + posts to the telemetry endpoint. +- `src/telemetry/cloudTransport.ts`: `CloudTransport` — HTTP transport behind `CloudAppender`. +- `src/telemetry/index.ts`: barrel. + +## Emitting events (business services) + +Inject `ITelemetryService` and call `track`: + +```ts +import { ITelemetryService } from '#/telemetry'; + +constructor(@ITelemetryService private readonly telemetry: ITelemetryService) {} + +this.telemetry.track('cron_fired', { task_id: taskId, latency_ms: 12 }); +``` + +`TelemetryService.track` merges the bound context into the properties and fans the event out to every registered appender. A single throwing appender is isolated via `onUnexpectedError` and never blocks the rest. + +### Context (sessionId / agentId / turnId) + +The service carries a bound context (`sessionId` / `agentId` / `turnId`) that is merged into every event. Bind it at construction or derive a scoped view: + +```ts +const child = telemetry.withContext({ agentId: 'main', turnId: 't1' }); +child.track('tool.call', { name: 'bash' }); // carries sessionId + agentId + turnId +``` + +`withContext(patch)` returns a new service sharing the same appenders; per-call properties override bound context on key collision. `setContext(patch)` mutates the bound context in place and propagates to appenders that implement `setContext`. + +## Appenders (destinations) + +An appender is the destination an event is fanned out to. It is **not a DI Service** — it is a plain object implementing `ITelemetryAppender`, held by `TelemetryService`. + +```ts +export interface ITelemetryAppender { + track(event: string, properties?: TelemetryProperties): void; + withContext?(patch: TelemetryContextPatch): ITelemetryAppender; + setContext?(patch: TelemetryContextPatch): void; + flush?(): Promise | void; + shutdown?(): Promise | void; +} +``` + +Built-in appenders: + +- `ConsoleAppender` — `[telemetry] ` to a log function (default `console.log`); options `prefix` / `pretty` / `log`. +- `CloudAppender` — batches events, enriches with common context (`app_name` / `version` / `platform` / …), and posts to `https://telemetry-logs.kimi.com/v1/event` through `CloudTransport` (Bearer auth, retry, on-disk fallback). Options: `homeDir` / `deviceId` / `sessionId?` / `appName` / `version` / `uiMode?` / `model?` / `getAccessToken?` / `endpoint?` / `flushThreshold?` / `flushIntervalMs?`. + +### Registering appenders (bootstrap) + +Appenders are added after the Core scope exists, by resolving the service and calling `addAppender`: + +```ts +const core = createCoreScope(); +const telemetry = core.accessor.get(ITelemetryService); + +telemetry.addAppender(new ConsoleAppender({ prefix: '[dev]' })); // dev echo +telemetry.addAppender(new CloudAppender({ // production + homeDir, deviceId, sessionId, + appName: 'kimi-code', version, uiMode: 'shell', model, + getAccessToken: () => auth.getCachedAccessToken(KIMI_CODE_PROVIDER_NAME), +})); +``` + +`addAppender` returns an `IDisposable` that removes the appender when disposed. `setAppender(appender)` resets to a single appender (mainly for tests). `removeAppender(appender)` drops one. + +> There is no production bootstrap wired yet — `TelemetryService` defaults to `[nullTelemetryAppender]`, so `track(...)` is a no-op until `addAppender` is called at startup. + +## Lifecycle + +- `setEnabled(false)` drops `track` (service-level switch); `setEnabled(true)` resumes. `flush` / `shutdown` are unaffected by the switch. +- `flush()` / `shutdown()` fan out to all appenders concurrently; a single rejecting appender is swallowed. Await `shutdown()` before process exit so buffered events (e.g. in `CloudAppender`) are sent. + +## Red lines (this topic) + +- Business services depend only on `ITelemetryService` — never import an appender class. +- Telemetry is layer-1 root: do not inject any business-domain service into it, and do not move it off `Core`. +- Appenders are plain `ITelemetryAppender` objects, not DI Services — register them with `addAppender`, never via `registerScopedService`. +- `track` is fire-and-forget and must not throw; appender `track` must be synchronous — buffer and send asynchronously via `flush` / `shutdown`. +- Await `telemetry.shutdown()` before process exit when a buffering appender is registered. +- Keep event names stable; properties must be JSON-serializable primitives (non-primitives are dropped by `CloudAppender`). diff --git a/packages/agent-core-v2/src/telemetry/cloudSink.ts b/packages/agent-core-v2/src/telemetry/cloudAppender.ts similarity index 89% rename from packages/agent-core-v2/src/telemetry/cloudSink.ts rename to packages/agent-core-v2/src/telemetry/cloudAppender.ts index b543e1d97..3a8d20b11 100644 --- a/packages/agent-core-v2/src/telemetry/cloudSink.ts +++ b/packages/agent-core-v2/src/telemetry/cloudAppender.ts @@ -1,14 +1,14 @@ /** - * `telemetry` domain (L1) — `CloudSink`, a `TelemetryClient` that batches - * events, enriches them with common context, and posts them to the telemetry - * endpoint through `CloudTransport`. Core-scoped; has no cross-domain + * `telemetry` domain (L1) — `CloudAppender`, an `ITelemetryAppender` that + * batches events, enriches them with common context, and posts them to the + * telemetry endpoint through `CloudTransport`. Core-scoped; has no cross-domain * collaborators and is independent of `@moonshot-ai/kimi-telemetry`. */ import { randomUUID } from 'node:crypto'; import { arch, platform, release } from 'node:os'; -import type { TelemetryClient, TelemetryContextPatch, TelemetryProperties } from './telemetry'; +import type { ITelemetryAppender, TelemetryContextPatch, TelemetryProperties } from './telemetry'; import { type CloudContext, type CloudPrimitive, @@ -18,7 +18,7 @@ import { isCloudPrimitive, } from './cloudTransport'; -export interface CloudSinkOptions { +export interface CloudAppenderOptions { readonly homeDir: string; readonly deviceId: string; readonly sessionId?: string; @@ -44,7 +44,7 @@ export interface CloudSinkOptions { const DEFAULT_FLUSH_THRESHOLD = 50; const DEFAULT_FLUSH_INTERVAL_MS = 30_000; -export class CloudSink implements TelemetryClient { +export class CloudAppender implements ITelemetryAppender { private readonly transport: CloudTransport; private readonly context: CloudContext; private readonly flushThreshold: number; @@ -54,7 +54,7 @@ export class CloudSink implements TelemetryClient { private buffer: EnrichedCloudEvent[] = []; private flushTimer: ReturnType | null = null; - constructor(options: CloudSinkOptions) { + constructor(options: CloudAppenderOptions) { this.deviceId = options.deviceId; this.sessionId = options.sessionId ?? null; this.flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD; @@ -142,7 +142,7 @@ function sanitizeProperties(input?: TelemetryProperties): CloudProperties { return out; } -function buildContext(options: CloudSinkOptions): CloudContext { +function buildContext(options: CloudAppenderOptions): CloudContext { const env = options.env ?? process.env; const context: CloudContext = { app_name: options.appName, diff --git a/packages/agent-core-v2/src/telemetry/cloudTransport.ts b/packages/agent-core-v2/src/telemetry/cloudTransport.ts index d497b17af..1ddbba6d1 100644 --- a/packages/agent-core-v2/src/telemetry/cloudTransport.ts +++ b/packages/agent-core-v2/src/telemetry/cloudTransport.ts @@ -1,6 +1,6 @@ /** * `telemetry` domain (L1) — `CloudTransport`, the HTTP transport behind - * `CloudSink`. Posts enriched events to the telemetry endpoint with Bearer + * `CloudAppender`. Posts enriched events to the telemetry endpoint with Bearer * auth, retry, and on-disk fallback for failed events. Core-scoped; has no * cross-domain collaborators and is independent of `@moonshot-ai/kimi-telemetry`. */ diff --git a/packages/agent-core-v2/src/telemetry/consoleSink.ts b/packages/agent-core-v2/src/telemetry/consoleAppender.ts similarity index 69% rename from packages/agent-core-v2/src/telemetry/consoleSink.ts rename to packages/agent-core-v2/src/telemetry/consoleAppender.ts index 1ec835290..cad044f98 100644 --- a/packages/agent-core-v2/src/telemetry/consoleSink.ts +++ b/packages/agent-core-v2/src/telemetry/consoleAppender.ts @@ -1,12 +1,12 @@ /** - * `telemetry` domain (L1) — `ConsoleSink`, a `TelemetryClient` that echoes - * events to a log function for development and debugging. Core-scoped; has no - * cross-domain collaborators. + * `telemetry` domain (L1) — `ConsoleAppender`, an `ITelemetryAppender` that + * echoes events to a log function for development and debugging. Core-scoped; + * has no cross-domain collaborators. */ -import type { TelemetryClient, TelemetryProperties } from './telemetry'; +import type { ITelemetryAppender, TelemetryProperties } from './telemetry'; -export interface ConsoleSinkOptions { +export interface ConsoleAppenderOptions { readonly prefix?: string; readonly pretty?: boolean; readonly log?: (message: string) => void; @@ -14,12 +14,12 @@ export interface ConsoleSinkOptions { const DEFAULT_PREFIX = '[telemetry]'; -export class ConsoleSink implements TelemetryClient { +export class ConsoleAppender implements ITelemetryAppender { private readonly prefix: string; private readonly pretty: boolean; private readonly log: (message: string) => void; - constructor(options: ConsoleSinkOptions = {}) { + constructor(options: ConsoleAppenderOptions = {}) { this.prefix = options.prefix ?? DEFAULT_PREFIX; this.pretty = options.pretty ?? false; this.log = options.log ?? defaultLog; diff --git a/packages/agent-core-v2/src/telemetry/index.ts b/packages/agent-core-v2/src/telemetry/index.ts index be9704798..88b5c0a49 100644 --- a/packages/agent-core-v2/src/telemetry/index.ts +++ b/packages/agent-core-v2/src/telemetry/index.ts @@ -1,11 +1,11 @@ /** * `telemetry` domain barrel — re-exports the `telemetry` contract, its scoped - * service (`telemetryService`), and the bundled sinks (`ConsoleSink`, - * `CloudSink`). Importing this barrel registers the `ITelemetryService` + * service (`telemetryService`), and the bundled appenders (`ConsoleAppender`, + * `CloudAppender`). Importing this barrel registers the `ITelemetryService` * binding into the scope registry. */ export * from './telemetry'; export * from './telemetryService'; -export * from './consoleSink'; -export * from './cloudSink'; +export * from './consoleAppender'; +export * from './cloudAppender'; diff --git a/packages/agent-core-v2/src/telemetry/telemetry.ts b/packages/agent-core-v2/src/telemetry/telemetry.ts index 7a673ac03..0653751db 100644 --- a/packages/agent-core-v2/src/telemetry/telemetry.ts +++ b/packages/agent-core-v2/src/telemetry/telemetry.ts @@ -1,12 +1,12 @@ /** - * `telemetry` domain (L1) — `ITelemetryService` contract and sink types. + * `telemetry` domain (L1) — `ITelemetryService` contract and appender types. * * Layer-1 root service: merges bound context into tracked events and fans - * them out to one or more `TelemetryClient` sinks. Core-scoped — stateless - * beyond its sink set and bound context; enrichment, batching, and transport - * are owned by the sinks, not by this layer. Defines the `TelemetryClient` - * sink contract, the `ITelemetryService` facade, the service options, and the - * no-op sink. + * them out to one or more `ITelemetryAppender` destinations. Core-scoped — + * stateless beyond its appender set and bound context; enrichment, batching, + * and transport are owned by the appenders, not by this layer. Defines the + * `ITelemetryAppender` contract, the `ITelemetryService` facade, the service + * options, and the null appender. */ import { createDecorator } from '#/_base/di/instantiation'; @@ -18,17 +18,17 @@ export type TelemetryProperties = Readonly | void; shutdown?(): Promise | void; } export interface TelemetryServiceOptions { - readonly client?: TelemetryClient; - readonly clients?: readonly TelemetryClient[]; + readonly appender?: ITelemetryAppender; + readonly appenders?: readonly ITelemetryAppender[]; readonly context?: TelemetryProperties; readonly sessionId?: string; readonly agentId?: string; @@ -40,17 +40,17 @@ export interface ITelemetryService { track(event: string, properties?: TelemetryProperties): void; withContext(patch: TelemetryContextPatch): ITelemetryService; setContext(patch: TelemetryContextPatch): void; - addSink(client: TelemetryClient): IDisposable; - removeSink(client: TelemetryClient): void; - setDelegate(client: TelemetryClient): void; + addAppender(appender: ITelemetryAppender): IDisposable; + removeAppender(appender: ITelemetryAppender): void; + setAppender(appender: ITelemetryAppender): void; setEnabled(enabled: boolean): void; flush(): Promise; shutdown(): Promise; } -export const noopTelemetryClient: TelemetryClient = { +export const nullTelemetryAppender: ITelemetryAppender = { track: () => {}, - withContext: () => noopTelemetryClient, + withContext: () => nullTelemetryAppender, setContext: () => {}, flush: () => {}, shutdown: () => {}, diff --git a/packages/agent-core-v2/src/telemetry/telemetryService.ts b/packages/agent-core-v2/src/telemetry/telemetryService.ts index 8d8609d26..d00ba0776 100644 --- a/packages/agent-core-v2/src/telemetry/telemetryService.ts +++ b/packages/agent-core-v2/src/telemetry/telemetryService.ts @@ -2,9 +2,9 @@ * `telemetry` domain (L1) — `ITelemetryService` implementation. * * Merges bound context into each tracked event and fans it out to the - * registered `TelemetryClient` sinks; owns the sink set, the enabled flag, - * and the bound context, but no enrichment or transport of its own. Bound at - * Core scope; has no cross-domain collaborators. + * registered `ITelemetryAppender` destinations; owns the appender set, the + * enabled flag, and the bound context, but no enrichment or transport of its + * own. Bound at Core scope; has no cross-domain collaborators. */ import { InstantiationType } from '#/_base/di/extensions'; @@ -14,8 +14,8 @@ import { onUnexpectedError } from '#/_base/errors/unexpectedError'; import { ITelemetryService, - noopTelemetryClient, - type TelemetryClient, + type ITelemetryAppender, + nullTelemetryAppender, type TelemetryContextPatch, type TelemetryProperties, type TelemetryServiceOptions, @@ -24,12 +24,12 @@ import { export class TelemetryService implements ITelemetryService { declare readonly _serviceBrand: undefined; - private sinks: TelemetryClient[]; + private appenders: ITelemetryAppender[]; private context: TelemetryProperties; private enabled = true; constructor(options: TelemetryServiceOptions = {}) { - this.sinks = resolveSinks(options); + this.appenders = resolveAppenders(options); this.context = { ...options.context, ...definedContext({ @@ -45,9 +45,9 @@ export class TelemetryService implements ITelemetryService { return; } const merged = { ...this.context, ...properties }; - for (const sink of this.sinks) { + for (const appender of this.appenders) { try { - sink.track(event, merged); + appender.track(event, merged); } catch (err) { onUnexpectedError(err); } @@ -56,7 +56,7 @@ export class TelemetryService implements ITelemetryService { withContext(patch: TelemetryContextPatch): ITelemetryService { const child = new TelemetryService({ - clients: this.sinks.map((sink) => sink.withContext?.(patch) ?? sink), + appenders: this.appenders.map((appender) => appender.withContext?.(patch) ?? appender), context: { ...this.context, ...patch }, }); child.enabled = this.enabled; @@ -65,22 +65,22 @@ export class TelemetryService implements ITelemetryService { setContext(patch: TelemetryContextPatch): void { this.context = { ...this.context, ...patch }; - for (const sink of this.sinks) { - sink.setContext?.(patch); + for (const appender of this.appenders) { + appender.setContext?.(patch); } } - addSink(client: TelemetryClient): IDisposable { - this.sinks.push(client); - return toDisposable(() => this.removeSink(client)); + addAppender(appender: ITelemetryAppender): IDisposable { + this.appenders.push(appender); + return toDisposable(() => this.removeAppender(appender)); } - removeSink(client: TelemetryClient): void { - this.sinks = this.sinks.filter((sink) => sink !== client); + removeAppender(appender: ITelemetryAppender): void { + this.appenders = this.appenders.filter((a) => a !== appender); } - setDelegate(client: TelemetryClient): void { - this.sinks = [client]; + setAppender(appender: ITelemetryAppender): void { + this.appenders = [appender]; } setEnabled(enabled: boolean): void { @@ -89,16 +89,16 @@ export class TelemetryService implements ITelemetryService { async flush(): Promise { await Promise.all( - this.sinks.map((sink) => - Promise.resolve(sink.flush?.()).catch(onUnexpectedError), + this.appenders.map((appender) => + Promise.resolve(appender.flush?.()).catch(onUnexpectedError), ), ); } async shutdown(): Promise { await Promise.all( - this.sinks.map((sink) => - Promise.resolve(sink.shutdown?.()).catch(onUnexpectedError), + this.appenders.map((appender) => + Promise.resolve(appender.shutdown?.()).catch(onUnexpectedError), ), ); } @@ -112,11 +112,11 @@ registerScopedService( 'telemetry', ); -function resolveSinks(options: TelemetryServiceOptions): TelemetryClient[] { - if (options.clients !== undefined) { - return options.clients.length > 0 ? [...options.clients] : [noopTelemetryClient]; +function resolveAppenders(options: TelemetryServiceOptions): ITelemetryAppender[] { + if (options.appenders !== undefined) { + return options.appenders.length > 0 ? [...options.appenders] : [nullTelemetryAppender]; } - return [options.client ?? noopTelemetryClient]; + return [options.appender ?? nullTelemetryAppender]; } function definedContext(input: TelemetryProperties): TelemetryProperties { diff --git a/packages/agent-core-v2/test/telemetry/cloudSink.test.ts b/packages/agent-core-v2/test/telemetry/cloudAppender.test.ts similarity index 83% rename from packages/agent-core-v2/test/telemetry/cloudSink.test.ts rename to packages/agent-core-v2/test/telemetry/cloudAppender.test.ts index dc89715af..749b83f96 100644 --- a/packages/agent-core-v2/test/telemetry/cloudSink.test.ts +++ b/packages/agent-core-v2/test/telemetry/cloudAppender.test.ts @@ -4,7 +4,7 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { CloudSink, type CloudSinkOptions } from '#/telemetry/cloudSink'; +import { CloudAppender, type CloudAppenderOptions } from '#/telemetry/cloudAppender'; interface CapturedRequest { readonly url: string; @@ -37,7 +37,7 @@ function statusResponse(status: number): Response { return new Response(null, { status }); } -function baseOptions(overrides: Partial = {}): CloudSinkOptions { +function baseOptions(overrides: Partial = {}): CloudAppenderOptions { return { homeDir: overrides.homeDir ?? '', deviceId: overrides.deviceId ?? 'dev', @@ -49,11 +49,11 @@ function baseOptions(overrides: Partial = {}): CloudSinkOption }; } -describe('CloudSink', () => { +describe('CloudAppender', () => { let homeDir: string; beforeEach(() => { - homeDir = mkdtempSync(join(tmpdir(), 'cloud-sink-')); + homeDir = mkdtempSync(join(tmpdir(), 'cloud-appender-')); }); afterEach(() => { @@ -62,7 +62,7 @@ describe('CloudSink', () => { it('sends a flattened, prefixed payload with user_id and context', async () => { const requests: CapturedRequest[] = []; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, deviceId: 'dev123', @@ -74,8 +74,8 @@ describe('CloudSink', () => { }), ); - sink.track('tool.call', { name: 'bash', count: 2 }); - await sink.flush(); + appender.track('tool.call', { name: 'bash', count: 2 }); + await appender.flush(); expect(requests).toHaveLength(1); expect(requests[0]?.url).toBe('https://telemetry-logs.kimi.com/v1/event'); @@ -94,7 +94,7 @@ describe('CloudSink', () => { it('sends Authorization header when a token is provided', async () => { const requests: CapturedRequest[] = []; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, getAccessToken: () => 'tok123', @@ -105,15 +105,15 @@ describe('CloudSink', () => { }), ); - sink.track('evt'); - await sink.flush(); + appender.track('evt'); + await appender.flush(); expect(requests[0]?.headers['Authorization']).toBe('Bearer tok123'); }); it('auto-flushes when the buffer reaches the threshold', async () => { let sends = 0; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, flushThreshold: 3, @@ -124,17 +124,17 @@ describe('CloudSink', () => { }), ); - sink.track('e1'); - sink.track('e2'); + appender.track('e1'); + appender.track('e2'); expect(sends).toBe(0); - sink.track('e3'); + appender.track('e3'); await new Promise((resolve) => setTimeout(resolve, 20)); expect(sends).toBe(1); }); it('shutdown flushes the remaining buffered events', async () => { let sends = 0; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, fetchImpl: makeFetch(() => { @@ -144,14 +144,14 @@ describe('CloudSink', () => { }), ); - sink.track('e1'); - await sink.shutdown(); + appender.track('e1'); + await appender.shutdown(); expect(sends).toBe(1); }); it('retries on 5xx and saves to disk after exhausting backoffs', async () => { let attempts = 0; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, fetchImpl: makeFetch(() => { @@ -161,8 +161,8 @@ describe('CloudSink', () => { }), ); - sink.track('evt'); - await sink.flush(); + appender.track('evt'); + await appender.flush(); expect(attempts).toBe(4); const files = readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')); @@ -171,7 +171,7 @@ describe('CloudSink', () => { it('retries a 401 once without the Authorization header', async () => { const seenAuths: (string | undefined)[] = []; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, getAccessToken: () => 'tok', @@ -185,29 +185,29 @@ describe('CloudSink', () => { }), ); - sink.track('evt'); - await sink.flush(); + appender.track('evt'); + await appender.flush(); expect(seenAuths).toEqual(['Bearer tok', undefined]); }); it('retryDiskEvents resends saved events and removes the file on success', async () => { let shouldFail = true; - const sink = new CloudSink( + const appender = new CloudAppender( baseOptions({ homeDir, fetchImpl: makeFetch(() => (shouldFail ? statusResponse(500) : okResponse())), }), ); - sink.track('evt'); - await sink.flush(); + appender.track('evt'); + await appender.flush(); expect( readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), ).toHaveLength(1); shouldFail = false; - await sink.retryDiskEvents(); + await appender.retryDiskEvents(); expect( readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), ).toHaveLength(0); diff --git a/packages/agent-core-v2/test/telemetry/consoleSink.test.ts b/packages/agent-core-v2/test/telemetry/consoleAppender.test.ts similarity index 54% rename from packages/agent-core-v2/test/telemetry/consoleSink.test.ts rename to packages/agent-core-v2/test/telemetry/consoleAppender.test.ts index 1f65a4600..fdd3aa6a8 100644 --- a/packages/agent-core-v2/test/telemetry/consoleSink.test.ts +++ b/packages/agent-core-v2/test/telemetry/consoleAppender.test.ts @@ -1,12 +1,12 @@ import { describe, expect, it } from 'vitest'; -import { ConsoleSink } from '#/telemetry/consoleSink'; +import { ConsoleAppender } from '#/telemetry/consoleAppender'; -describe('ConsoleSink', () => { +describe('ConsoleAppender', () => { it('logs event name and properties with the default prefix', () => { const lines: string[] = []; - const sink = new ConsoleSink({ log: (message) => lines.push(message) }); - sink.track('tool.call', { name: 'bash', count: 1 }); + const appender = new ConsoleAppender({ log: (message) => lines.push(message) }); + appender.track('tool.call', { name: 'bash', count: 1 }); expect(lines).toHaveLength(1); expect(lines[0]).toContain('[telemetry] tool.call'); expect(lines[0]).toContain('"name":"bash"'); @@ -15,22 +15,22 @@ describe('ConsoleSink', () => { it('uses a custom prefix', () => { const lines: string[] = []; - const sink = new ConsoleSink({ prefix: '[dbg]', log: (message) => lines.push(message) }); - sink.track('evt'); + const appender = new ConsoleAppender({ prefix: '[dbg]', log: (message) => lines.push(message) }); + appender.track('evt'); expect(lines[0]).toBe('[dbg] evt'); }); it('omits the payload when properties is undefined', () => { const lines: string[] = []; - const sink = new ConsoleSink({ log: (message) => lines.push(message) }); - sink.track('evt'); + const appender = new ConsoleAppender({ log: (message) => lines.push(message) }); + appender.track('evt'); expect(lines[0]).toBe('[telemetry] evt'); }); it('pretty-prints properties when requested', () => { const lines: string[] = []; - const sink = new ConsoleSink({ pretty: true, log: (message) => lines.push(message) }); - sink.track('evt', { a: 1 }); + const appender = new ConsoleAppender({ pretty: true, log: (message) => lines.push(message) }); + appender.track('evt', { a: 1 }); expect(lines[0]).toContain('\n'); }); }); diff --git a/packages/agent-core-v2/test/telemetry/telemetryService.test.ts b/packages/agent-core-v2/test/telemetry/telemetryService.test.ts index 9a5062e1b..d7271a7c3 100644 --- a/packages/agent-core-v2/test/telemetry/telemetryService.test.ts +++ b/packages/agent-core-v2/test/telemetry/telemetryService.test.ts @@ -8,13 +8,13 @@ import { setUnexpectedErrorHandler, } from '#/_base/errors/unexpectedError'; import { - type TelemetryClient, + type ITelemetryAppender, type TelemetryProperties, ITelemetryService, TelemetryService, } from '#/telemetry/index'; -class CapturingClient implements TelemetryClient { +class CapturingAppender implements ITelemetryAppender { readonly events: { event: string; properties?: TelemetryProperties }[] = []; flushCalls = 0; shutdownCalls = 0; @@ -36,23 +36,23 @@ describe('TelemetryService (unit)', () => { }); it('merges bound context into tracked properties', () => { - const client = new CapturingClient(); + const appender = new CapturingAppender(); const svc = new TelemetryService({ sessionId: 's1' }); - svc.setDelegate(client); + svc.setAppender(appender); svc.track('turn.start', { agentId: 'main' }); - expect(client.events[0]).toEqual({ + expect(appender.events[0]).toEqual({ event: 'turn.start', properties: { sessionId: 's1', agentId: 'main' }, }); }); - it('withContext merges context and shares the delegate', () => { - const client = new CapturingClient(); + it('withContext merges context and shares the appender', () => { + const appender = new CapturingAppender(); const root = new TelemetryService({ sessionId: 's1' }); - root.setDelegate(client); + root.setAppender(appender); const child = root.withContext({ agentId: 'main', turnId: 't1' }); child.track('tool.call', { name: 'bash' }); - expect(client.events[0]?.properties).toEqual({ + expect(appender.events[0]?.properties).toEqual({ sessionId: 's1', agentId: 'main', turnId: 't1', @@ -61,27 +61,27 @@ describe('TelemetryService (unit)', () => { }); it('per-call properties override bound context on key collision', () => { - const client = new CapturingClient(); + const appender = new CapturingAppender(); const svc = new TelemetryService({ sessionId: 's1' }); - svc.setDelegate(client); + svc.setAppender(appender); svc.track('evt', { sessionId: 'override' }); - expect(client.events[0]?.properties?.['sessionId']).toBe('override'); + expect(appender.events[0]?.properties?.['sessionId']).toBe('override'); }); - it('fans out to every sink passed via clients', () => { - const a = new CapturingClient(); - const b = new CapturingClient(); - const svc = new TelemetryService({ clients: [a, b] }); + it('fans out to every appender passed via appenders', () => { + const a = new CapturingAppender(); + const b = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [a, b] }); svc.track('evt', { x: 1 }); expect(a.events).toEqual([{ event: 'evt', properties: { x: 1 } }]); expect(b.events).toEqual([{ event: 'evt', properties: { x: 1 } }]); }); - it('addSink registers a sink and its disposable removes it', () => { - const a = new CapturingClient(); - const b = new CapturingClient(); - const svc = new TelemetryService({ client: a }); - const disposable = svc.addSink(b); + it('addAppender registers an appender and its disposable removes it', () => { + const a = new CapturingAppender(); + const b = new CapturingAppender(); + const svc = new TelemetryService({ appender: a }); + const disposable = svc.addAppender(b); svc.track('first'); expect(a.events).toHaveLength(1); expect(b.events).toHaveLength(1); @@ -91,57 +91,57 @@ describe('TelemetryService (unit)', () => { expect(b.events).toHaveLength(1); }); - it('removeSink stops delivery to that sink', () => { - const a = new CapturingClient(); - const b = new CapturingClient(); - const svc = new TelemetryService({ clients: [a, b] }); - svc.removeSink(a); + it('removeAppender stops delivery to that appender', () => { + const a = new CapturingAppender(); + const b = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [a, b] }); + svc.removeAppender(a); svc.track('evt'); expect(a.events).toHaveLength(0); expect(b.events).toHaveLength(1); }); it('setEnabled(false) drops track; setEnabled(true) resumes', () => { - const client = new CapturingClient(); - const svc = new TelemetryService({ client }); + const appender = new CapturingAppender(); + const svc = new TelemetryService({ appender }); svc.setEnabled(false); svc.track('dropped'); - expect(client.events).toHaveLength(0); + expect(appender.events).toHaveLength(0); svc.setEnabled(true); svc.track('sent'); - expect(client.events).toEqual([{ event: 'sent', properties: {} }]); + expect(appender.events).toEqual([{ event: 'sent', properties: {} }]); }); it('withContext child inherits enabled state at creation', () => { - const client = new CapturingClient(); - const root = new TelemetryService({ client }); + const appender = new CapturingAppender(); + const root = new TelemetryService({ appender }); root.setEnabled(false); const child = root.withContext({ turnId: 't1' }); child.track('dropped'); - expect(client.events).toHaveLength(0); + expect(appender.events).toHaveLength(0); }); - it('flush fans out to every sink', async () => { - const a = new CapturingClient(); - const b = new CapturingClient(); - const svc = new TelemetryService({ clients: [a, b] }); + it('flush fans out to every appender', async () => { + const a = new CapturingAppender(); + const b = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [a, b] }); await svc.flush(); expect(a.flushCalls).toBe(1); expect(b.flushCalls).toBe(1); }); - it('shutdown fans out to every sink', async () => { - const a = new CapturingClient(); - const b = new CapturingClient(); - const svc = new TelemetryService({ clients: [a, b] }); + it('shutdown fans out to every appender', async () => { + const a = new CapturingAppender(); + const b = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [a, b] }); await svc.shutdown(); expect(a.shutdownCalls).toBe(1); expect(b.shutdownCalls).toBe(1); }); - it('flush is a no-op for sinks without flush', async () => { - const minimal: TelemetryClient = { track() {} }; - const svc = new TelemetryService({ client: minimal }); + it('flush is a no-op for appenders without flush', async () => { + const minimal: ITelemetryAppender = { track() {} }; + const svc = new TelemetryService({ appender: minimal }); await expect(svc.flush()).resolves.toBeUndefined(); await expect(svc.shutdown()).resolves.toBeUndefined(); }); @@ -151,40 +151,40 @@ describe('TelemetryService (error isolation)', () => { beforeEach(() => setUnexpectedErrorHandler(() => {})); afterEach(() => resetUnexpectedErrorHandler()); - it('a throwing sink does not prevent delivery to other sinks', () => { - const bad: TelemetryClient = { + it('a throwing appender does not prevent delivery to other appenders', () => { + const bad: ITelemetryAppender = { track() { throw new Error('boom'); }, }; - const good = new CapturingClient(); - const svc = new TelemetryService({ clients: [bad, good] }); + const good = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [bad, good] }); expect(() => svc.track('evt')).not.toThrow(); expect(good.events).toEqual([{ event: 'evt', properties: {} }]); }); - it('flush tolerates a rejecting sink and still flushes the rest', async () => { - const bad: TelemetryClient = { + it('flush tolerates a rejecting appender and still flushes the rest', async () => { + const bad: ITelemetryAppender = { track() {}, async flush() { throw new Error('boom'); }, }; - const good = new CapturingClient(); - const svc = new TelemetryService({ clients: [bad, good] }); + const good = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [bad, good] }); await expect(svc.flush()).resolves.toBeUndefined(); expect(good.flushCalls).toBe(1); }); - it('shutdown tolerates a rejecting sink and still shuts down the rest', async () => { - const bad: TelemetryClient = { + it('shutdown tolerates a rejecting appender and still shuts down the rest', async () => { + const bad: ITelemetryAppender = { track() {}, async shutdown() { throw new Error('boom'); }, }; - const good = new CapturingClient(); - const svc = new TelemetryService({ clients: [bad, good] }); + const good = new CapturingAppender(); + const svc = new TelemetryService({ appenders: [bad, good] }); await expect(svc.shutdown()).resolves.toBeUndefined(); expect(good.shutdownCalls).toBe(1); });