|
|
@@ -1,12 +1,15 @@
|
|
|
/**
|
|
|
- * Capture coordinator: the seam's upstream half. Subscribes to the session
|
|
|
- * firehose plus the one live-bus relay (`agent/error`), applies the fixed
|
|
|
- * chunk projection, builds logical records, runs each through the
|
|
|
+ * Capture coordinator: the seam's upstream half. Live capture subscribes to
|
|
|
+ * the session firehose plus the one live-bus relay (`agent/error`). Both
|
|
|
+ * capture paths apply the fixed chunk projection, build logical records, and
|
|
|
+ * run each through the
|
|
|
* `telemetry/record` waterfall (deployment-mounted redaction rules;
|
|
|
- * pass-through when none), and hands the result to the backend — synchronously, with every
|
|
|
- * handler self-contained so a failing backend can never starve other
|
|
|
- * subscribers (cordis `emit` is stop-on-throw) or touch the agent loop.
|
|
|
- * Composed by a backend in its constructor.
|
|
|
+ * pass-through when none), then hands the result to the backend. Live capture
|
|
|
+ * follows the session firehose; on-demand capture replays the canonical log
|
|
|
+ * only when requested. Every synchronous handler is self-contained so a
|
|
|
+ * failing backend can never starve other subscribers (cordis `emit` is
|
|
|
+ * stop-on-throw) or touch the agent loop. Composed by a backend in its
|
|
|
+ * constructor.
|
|
|
*
|
|
|
* @module @deepseek-ai/dsh-session-telemetry/coordinator
|
|
|
*/
|
|
|
@@ -16,6 +19,16 @@ import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
|
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
|
import type { TelemetryBackend, TelemetryRecord, TelemetrySeverity } from './index.ts'
|
|
|
|
|
|
+/** Whether capture follows live events or reads the canonical log only when requested. */
|
|
|
+export type TelemetryCapture = 'live' | 'on-demand'
|
|
|
+
|
|
|
+/** One projected record ready for backend handoff. */
|
|
|
+interface ProjectedRecord {
|
|
|
+ readonly record: TelemetryRecord
|
|
|
+ /** Ledger cursor advanced only after the backend accepts this record. */
|
|
|
+ readonly seq?: number
|
|
|
+}
|
|
|
+
|
|
|
/**
|
|
|
* The handoff cursor: per session, the highest `seq` handed to a backend.
|
|
|
* Deliberately MODULE-scope ambient state — a narrow, documented exception
|
|
|
@@ -32,17 +45,17 @@ const handoffCursor = new WeakMap<Session, number>()
|
|
|
/**
|
|
|
* Install the telemetry capture side onto a context for one backend.
|
|
|
*
|
|
|
- * Registers the persistence-coordinator listener set plus the `agent/error`
|
|
|
- * relay, all through `ctx.effect()`/`ctx.on()` on the composing fiber, and
|
|
|
- * sweeps already-live sessions (a hot reload does not replay
|
|
|
- * `session/created`). A `session/disposed` emits the session's `shutdown`
|
|
|
- * operational record — the marker rides the session's own termination edge,
|
|
|
- * where receivers key crash detection — and retires it from the adopted set,
|
|
|
- * so a long-lived backend neither retains closed sessions (and their frozen
|
|
|
- * event logs) nor re-marks them at unload. Disposal marks the sessions still
|
|
|
- * alive at teardown (their own edge would fire unobserved) and then awaits
|
|
|
- * the backend's `shutdown()`; a failure there warns instead of throwing —
|
|
|
- * best-effort reporting must not fail application teardown.
|
|
|
+ * Live capture registers the persistence-coordinator listener set plus the
|
|
|
+ * `agent/error` relay, all through `ctx.effect()`/`ctx.on()` on the composing
|
|
|
+ * fiber, and sweeps already-live sessions (a hot reload does not replay
|
|
|
+ * `session/created`). A `session/disposed` captures the session's `shutdown`
|
|
|
+ * operational record at its own termination edge and retires it from the
|
|
|
+ * adopted set. On-demand capture registers none of those continuous listeners;
|
|
|
+ * {@link captureSession} reads the canonical log explicitly and never creates
|
|
|
+ * operational records. Disposal captures shutdown markers for live-adopted
|
|
|
+ * sessions, then awaits the backend's `shutdown()`; a failure there warns
|
|
|
+ * instead of throwing — best-effort reporting must not fail application
|
|
|
+ * teardown.
|
|
|
*/
|
|
|
export class TelemetryCoordinator {
|
|
|
/**
|
|
|
@@ -53,55 +66,55 @@ export class TelemetryCoordinator {
|
|
|
private readonly adopted = new Set<Session>()
|
|
|
/** Per session, the `turn:step` keys whose first chunk already shipped; rebuilt from the log on re-adoption. */
|
|
|
private readonly chunkSeen = new WeakMap<Session, Set<string>>()
|
|
|
-
|
|
|
/**
|
|
|
* @param ctx - the composing backend's context; listeners bind to its fiber.
|
|
|
* @param backend - the backend receiving records; owned elsewhere, never disposed here beyond `shutdown()` forwarding.
|
|
|
+ * @param capture - follow live events, or wait for explicit canonical-log capture.
|
|
|
*/
|
|
|
constructor(
|
|
|
private readonly ctx: Context,
|
|
|
private readonly backend: TelemetryBackend,
|
|
|
+ capture: TelemetryCapture = 'live',
|
|
|
) {
|
|
|
- ctx.on('session/created', (session) => {
|
|
|
- this.adopt(session)
|
|
|
- })
|
|
|
- // The session's own termination edge: emit the shutdown marker HERE —
|
|
|
- // receivers classify a session with activity and no marker as crashed,
|
|
|
- // so a normally closed session in a long-running host must get its
|
|
|
- // marker at disposal, not never. Then retire: the projection/cursor
|
|
|
- // WeakMaps die with the Session object; only the strong adopted set
|
|
|
- // needs the explicit release.
|
|
|
- ctx.on('session/disposed', (session) => {
|
|
|
- this.contain(() => {
|
|
|
- if (!this.adopted.delete(session)) return
|
|
|
- this.handOff(shutdownRecord(session))
|
|
|
+ if (capture === 'live') {
|
|
|
+ ctx.on('session/created', (session) => {
|
|
|
+ this.adopt(session)
|
|
|
})
|
|
|
- })
|
|
|
- ctx.on('session/event', (session, event) => {
|
|
|
- this.contain(() => {
|
|
|
- this.capture(session, event)
|
|
|
+ // Capture the shutdown marker at the session's own termination edge,
|
|
|
+ // then retire the only strong reference owned by this coordinator.
|
|
|
+ ctx.on('session/disposed', (session) => {
|
|
|
+ this.contain(() => {
|
|
|
+ if (!this.adopted.delete(session)) return
|
|
|
+ this.deliver(session, { record: this.redact(shutdownRecord(session)) })
|
|
|
+ })
|
|
|
})
|
|
|
- })
|
|
|
- // Parallel listeners are awaited by the loop at turn end; returning void
|
|
|
- // (not the SDK's flush promise) is the turn-latency contract.
|
|
|
- ctx.on('session/flush', (session) => {
|
|
|
- this.contain(() => {
|
|
|
- this.hintFlush(session)
|
|
|
+ ctx.on('session/event', (session, event) => {
|
|
|
+ this.contain(() => {
|
|
|
+ this.captureEvent(session, event)
|
|
|
+ })
|
|
|
})
|
|
|
- })
|
|
|
- ctx.on('agent/error', ({ agent, turn, step, error }) => {
|
|
|
- this.contain(() => {
|
|
|
- this.relayAgentError(agent, turn, step, error)
|
|
|
+ // Parallel listeners are awaited by the loop at turn end; returning void
|
|
|
+ // (not the SDK's flush promise) is the turn-latency contract.
|
|
|
+ ctx.on('session/flush', (session) => {
|
|
|
+ this.contain(() => {
|
|
|
+ this.hintFlush(session)
|
|
|
+ })
|
|
|
})
|
|
|
- })
|
|
|
+ ctx.on('agent/error', ({ agent, turn, step, error }) => {
|
|
|
+ this.contain(() => {
|
|
|
+ this.relayAgentError(agent, turn, step, error)
|
|
|
+ })
|
|
|
+ })
|
|
|
+ for (const session of ctx.sessions.list()) {
|
|
|
+ this.adopt(session)
|
|
|
+ }
|
|
|
+ }
|
|
|
ctx.effect(() => async () => {
|
|
|
- // Sessions still adopted here are alive through a whole-application
|
|
|
- // teardown (their own disposal edge will fire after telemetry is gone,
|
|
|
- // unobserved) — mark them now so the receiver sees a clean stop of
|
|
|
- // observation rather than a crash-shaped silence.
|
|
|
+ // Sessions still adopted here are alive through whole-application
|
|
|
+ // teardown, so capture the marker before the backend quiesces.
|
|
|
for (const session of this.adopted) {
|
|
|
this.contain(() => {
|
|
|
- this.handOff(shutdownRecord(session))
|
|
|
+ this.deliver(session, { record: this.redact(shutdownRecord(session)) })
|
|
|
})
|
|
|
}
|
|
|
try {
|
|
|
@@ -110,8 +123,28 @@ export class TelemetryCoordinator {
|
|
|
this.ctx.logger.warn(`telemetry: backend shutdown failed: ${String(error)}`)
|
|
|
}
|
|
|
}, 'telemetry capture')
|
|
|
- for (const session of ctx.sessions.list()) {
|
|
|
- this.adopt(session)
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Project and hand over the canonical session-log suffix after the handoff
|
|
|
+ * cursor, optionally stopping at an inclusive sequence boundary. Redaction
|
|
|
+ * runs during this call, so an on-demand caller retains no copied records
|
|
|
+ * before requesting capture and uses the policy mounted at that time.
|
|
|
+ * Backend and policy failures remain contained per event and do not starve
|
|
|
+ * later events in the same replay.
|
|
|
+ * @param session - session whose current canonical-log prefix may be handed over.
|
|
|
+ * @param throughSeq - optional last sequence included in this capture.
|
|
|
+ */
|
|
|
+ captureSession(session: Session, throughSeq?: number): void {
|
|
|
+ const cursor = handoffCursor.get(session) ?? session.firstLiveSeq - 1
|
|
|
+ // Containment is PER EVENT: one rejected record is withheld fail-closed
|
|
|
+ // while the rest of the historical replay proceeds.
|
|
|
+ for (const event of session.events) {
|
|
|
+ if (throughSeq !== undefined && event.seq > throughSeq) break
|
|
|
+ this.contain(() => {
|
|
|
+ if (event.seq <= cursor) this.track(session, event)
|
|
|
+ else this.captureEvent(session, event)
|
|
|
+ })
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -133,17 +166,7 @@ export class TelemetryCoordinator {
|
|
|
private adopt(session: Session): void {
|
|
|
if (this.adopted.has(session)) return
|
|
|
this.adopted.add(session)
|
|
|
- const cursor = handoffCursor.get(session) ?? session.firstLiveSeq - 1
|
|
|
- // Containment is PER EVENT, matching the firehose: one rejected record
|
|
|
- // is withheld fail-closed while the rest of the historical replay
|
|
|
- // proceeds — wrapping the whole loop would let a single failure silently
|
|
|
- // skip the remainder of the log on an already-adopted session.
|
|
|
- for (const event of session.events) {
|
|
|
- this.contain(() => {
|
|
|
- if (event.seq <= cursor) this.track(session, event)
|
|
|
- else this.capture(session, event)
|
|
|
- })
|
|
|
- }
|
|
|
+ this.captureSession(session)
|
|
|
}
|
|
|
|
|
|
/** Feed the chunk projection without handing off — the ≤cursor half of re-adoption. */
|
|
|
@@ -153,8 +176,8 @@ export class TelemetryCoordinator {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /** Project one event and hand it to the backend, advancing the cursor on handoff. */
|
|
|
- private capture(session: Session, event: SessionEvent): void {
|
|
|
+ /** Project, redact, and hand one event to the backend. */
|
|
|
+ private captureEvent(session: Session, event: SessionEvent): void {
|
|
|
if (event.type === 'assistant/chunk') {
|
|
|
const key = `${event.data.turn}:${event.data.step}`
|
|
|
const seen = this.seen(session)
|
|
|
@@ -165,27 +188,36 @@ export class TelemetryCoordinator {
|
|
|
if (seen.has(key)) return
|
|
|
seen.add(key)
|
|
|
}
|
|
|
- this.handOff({
|
|
|
- channel: 'ledger',
|
|
|
- time: event.time,
|
|
|
- severity: severityOf(event),
|
|
|
- attributes: identityOf(session, event),
|
|
|
- // The live event object is mutable and the backend serializes later;
|
|
|
- // append-time validation guarantees this clone cannot throw.
|
|
|
- body: structuredClone(event.data),
|
|
|
+ this.deliver(session, {
|
|
|
+ record: this.redact({
|
|
|
+ channel: 'ledger',
|
|
|
+ time: event.time,
|
|
|
+ severity: severityOf(event),
|
|
|
+ attributes: identityOf(session, event),
|
|
|
+ // The canonical event object is mutable and the backend serializes
|
|
|
+ // later; append-time validation guarantees this clone cannot throw.
|
|
|
+ body: structuredClone(event.data),
|
|
|
+ }),
|
|
|
+ seq: event.seq,
|
|
|
})
|
|
|
- handoffCursor.set(session, event.seq)
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Run the `telemetry/record` waterfall over one record and hand the result
|
|
|
- * to the backend. The innermost `next` passes the record through unchanged
|
|
|
- * — the seam ships no rules; exported data is as clean as the listeners a
|
|
|
- * deployment mounts. Callers run inside {@link contain}, so a throwing
|
|
|
- * rule withholds the record instead of reaching the loop (fail-closed).
|
|
|
+ * Run the `telemetry/record` waterfall at capture time. The innermost `next`
|
|
|
+ * passes the record through unchanged — the seam ships no rules; exported
|
|
|
+ * data is as clean as the listeners a deployment mounts. Callers run inside
|
|
|
+ * {@link contain}, so a throwing rule withholds the record instead of
|
|
|
+ * reaching the loop (fail-closed). On-demand capture invokes this waterfall
|
|
|
+ * while reading the canonical session log, not when the event was appended.
|
|
|
*/
|
|
|
- private handOff(record: TelemetryRecord): void {
|
|
|
- this.backend.emit(this.ctx.waterfall('telemetry/record', record, () => record))
|
|
|
+ private redact(record: TelemetryRecord): TelemetryRecord {
|
|
|
+ return this.ctx.waterfall('telemetry/record', record, () => record)
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Hand one redacted record to the backend, then advance its ledger cursor. */
|
|
|
+ private deliver(session: Session, pending: ProjectedRecord): void {
|
|
|
+ this.backend.emit(pending.record)
|
|
|
+ if (pending.seq !== undefined) handoffCursor.set(session, pending.seq)
|
|
|
}
|
|
|
|
|
|
/** Forward the turn-end boundary to the backend's optional flush hint. */
|
|
|
@@ -196,19 +228,21 @@ export class TelemetryCoordinator {
|
|
|
/** Relay one `agent/error` bus emission as an `agent-error` operational record. */
|
|
|
private relayAgentError(agent: Agent, turn: number, step: number, error: unknown): void {
|
|
|
const detail = errorDetail(error)
|
|
|
- this.handOff({
|
|
|
- channel: 'ops',
|
|
|
- time: Date.now(),
|
|
|
- severity: 'error',
|
|
|
- attributes: {
|
|
|
- 'telemetry.op': 'agent-error',
|
|
|
- 'session.id': String(agent.session.id),
|
|
|
- 'agent.id': agent.id,
|
|
|
- 'error.name': detail.name,
|
|
|
- turn,
|
|
|
- step,
|
|
|
- },
|
|
|
- body: detail,
|
|
|
+ this.deliver(agent.session, {
|
|
|
+ record: this.redact({
|
|
|
+ channel: 'ops',
|
|
|
+ time: Date.now(),
|
|
|
+ severity: 'error',
|
|
|
+ attributes: {
|
|
|
+ 'telemetry.op': 'agent-error',
|
|
|
+ 'session.id': String(agent.session.id),
|
|
|
+ 'agent.id': agent.id,
|
|
|
+ 'error.name': detail.name,
|
|
|
+ turn,
|
|
|
+ step,
|
|
|
+ },
|
|
|
+ body: detail,
|
|
|
+ }),
|
|
|
})
|
|
|
}
|
|
|
|