mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(telemetry): emit the shutdown marker at the session's own disposal edge
Review finding (Codex P1), pinned red-first: the marker was tied to telemetry-plugin lifetime, but receivers key crash detection on its absence per session. A normally closed session in a long-running host retired silently (classified as a crash once stale), while a telemetry reload marked every still-live session as cleanly ended. The session/disposed handler now emits the marker at the session's own termination edge before retiring it; the dispose-time sweep only marks sessions still alive at application teardown (their own edge would fire unobserved). READMEs restate the marker semantics: telemetry stopped observing cleanly — a marker followed by more session events is a telemetry reload, not a session restart.
This commit is contained in:
@@ -23,7 +23,7 @@ Records carry the complete `event.data` as the seam's `telemetry/redact` waterfa
|
||||
|
||||
## Field mapping
|
||||
|
||||
Seam record → SDK log record: `time` → `timestamp`/`observedTimestamp`; `severity` → `severityNumber`/`severityText` (INFO 9 / WARN 13 / ERROR 17); `body` → the structured log body; `attributes` verbatim. Receivers dedupe on `(session.id, event.seq)`, alert on severity, and detect crashes by `shutdown`-record staleness (a session with activity, no `shutdown` ops record, gone stale ended uncleanly).
|
||||
Seam record → SDK log record: `time` → `timestamp`/`observedTimestamp`; `severity` → `severityNumber`/`severityText` (INFO 9 / WARN 13 / ERROR 17); `body` → the structured log body; `attributes` verbatim. Receivers dedupe on `(session.id, event.seq)`, alert on severity, and detect crashes by `shutdown`-record absence (a session with activity, no `shutdown` ops record, gone stale ended uncleanly). The marker means telemetry stopped observing the session cleanly — emitted at the session's own disposal, or at application teardown for sessions still running then; a marker followed by more of that session's events is a telemetry reload, not a session restart.
|
||||
|
||||
## Model Experience
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ The telemetry seam: the CAPTURE side of session-event reporting, behind a backen
|
||||
|
||||
## Capture points
|
||||
|
||||
The coordinator registers, all through the composing fiber's effects: `session/created` (adopt: record the header, read the log back through the projection — seed events from fork/resume never re-emit on the firehose), `session/event` (project, deep-copy, redact, hand off; zero I/O), `session/flush` (forward the optional `flush()` hint and return void — the loop's awaited parallel must never wait on telemetry), `session/disposed` (retire: release the adopted entry so a long-lived backend neither retains closed sessions nor stamps dispose-time markers for them), `agent/error` (the one live-bus relay; turn-enclosure structurally bars those errors from the log), a dispose effect (per still-adopted session emit its `shutdown` operational record, then await the backend's `shutdown()`; failures warn instead of throwing), and an adoption sweep of `ctx.sessions.list()` (a hot reload does not replay `session/created`).
|
||||
The coordinator registers, all through the composing fiber's effects: `session/created` (adopt: record the header, read the log back through the projection — seed events from fork/resume never re-emit on the firehose), `session/event` (project, deep-copy, redact, hand off; zero I/O), `session/flush` (forward the optional `flush()` hint and return void — the loop's awaited parallel must never wait on telemetry), `session/disposed` (emit the session's `shutdown` operational record at its own termination edge — where receivers key crash detection — then retire it, so a long-lived backend neither retains closed sessions nor re-marks them at unload), `agent/error` (the one live-bus relay; turn-enclosure structurally bars those errors from the log), a dispose effect (mark each session still alive at teardown, then await the backend's `shutdown()`; failures warn instead of throwing), and an adoption sweep of `ctx.sessions.list()` (a hot reload does not replay `session/created`).
|
||||
|
||||
## The redact waterfall
|
||||
|
||||
|
||||
@@ -35,19 +35,20 @@ const handoffCursor = new WeakMap<Session, number>()
|
||||
* 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` retires the session from the
|
||||
* adopted set — a long-lived backend must not retain closed sessions (and
|
||||
* their frozen event logs) or stamp dispose-time markers for sessions that
|
||||
* already ended. Disposal emits each still-adopted session's `shutdown`
|
||||
* operational record and then awaits the backend's `shutdown()`; a failure
|
||||
* there warns instead of throwing — best-effort reporting must not fail
|
||||
* application teardown.
|
||||
* `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.
|
||||
*/
|
||||
export class TelemetryCoordinator {
|
||||
/**
|
||||
* Sessions adopted by THIS fiber and still live, for dispose-time
|
||||
* `shutdown` records and double-adoption protection; `session/disposed`
|
||||
* retires entries.
|
||||
* Sessions adopted by THIS fiber and still live, for double-adoption
|
||||
* protection and the teardown sweep of unmarked sessions;
|
||||
* `session/disposed` marks and retires entries.
|
||||
*/
|
||||
private readonly adopted = new Set<Session>()
|
||||
/** Per session, the `turn:step` keys whose first chunk already shipped; rebuilt from the log on re-adoption. */
|
||||
@@ -64,10 +65,17 @@ export class TelemetryCoordinator {
|
||||
ctx.on('session/created', (session) => {
|
||||
this.adopt(session)
|
||||
})
|
||||
// Retirement is observe-only: the projection/cursor WeakMaps die with the
|
||||
// Session object; only the strong adopted set needs the explicit release.
|
||||
// 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.adopted.delete(session)
|
||||
this.contain(() => {
|
||||
if (!this.adopted.delete(session)) return
|
||||
this.handOff(shutdownRecord(session))
|
||||
})
|
||||
})
|
||||
ctx.on('session/event', (session, event) => {
|
||||
this.contain(() => {
|
||||
@@ -87,6 +95,10 @@ export class TelemetryCoordinator {
|
||||
})
|
||||
})
|
||||
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.
|
||||
for (const session of this.adopted) {
|
||||
this.contain(() => {
|
||||
this.handOff(shutdownRecord(session))
|
||||
@@ -214,7 +226,10 @@ export class TelemetryCoordinator {
|
||||
}
|
||||
}
|
||||
|
||||
/** Build the per-session clean-exit marker emitted at dispose, before the backend's `shutdown()`. */
|
||||
/**
|
||||
* Build the per-session clean-exit marker: emitted at the session's own
|
||||
* disposal edge, or at coordinator dispose for sessions still alive then.
|
||||
*/
|
||||
function shutdownRecord(session: Session): TelemetryRecord {
|
||||
return {
|
||||
channel: 'ops',
|
||||
|
||||
@@ -275,6 +275,26 @@ describe('TelemetryCoordinator lifecycle and containment', () => {
|
||||
expect(backend.flush).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('emits no marker for a session whose announcement was vetoed before adoption', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// A listener registered BEFORE the coordinator vetoes publication: the
|
||||
// store still emits the paired `session/disposed` for rollback, but the
|
||||
// coordinator never saw `session/created` — a marker for a session the
|
||||
// receiver saw no activity from would be noise, not signal.
|
||||
ctx.on('session/created', () => {
|
||||
throw new Error('vetoed by an earlier listener')
|
||||
})
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
expect(() => ctx.sessions.create(SessionId('vetoed'), { meta: {} })).toThrow('vetoed')
|
||||
expect(backend.records.filter(r => r.channel === 'ops')).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('emits each adopted session’s shutdown record before awaiting backend shutdown', async () => {
|
||||
const { ctx, backend, fiber } = await setup()
|
||||
liveSession(ctx, 's1')
|
||||
@@ -288,21 +308,25 @@ describe('TelemetryCoordinator lifecycle and containment', () => {
|
||||
expect(ops.every(r => !('event.seq' in r.attributes) && !('event.type' in r.attributes))).toBe(true)
|
||||
})
|
||||
|
||||
it('retires a disposed session: no retention, no stale shutdown marker at unload', async () => {
|
||||
it('emits the shutdown marker at the session’s own disposal edge, then retires it', async () => {
|
||||
const { ctx, backend, fiber } = await setup()
|
||||
liveSession(ctx, 'survivor')
|
||||
// A session owned by its own fiber: disposing the fiber detaches it from
|
||||
// the store and emits `session/disposed` — the authoritative retirement
|
||||
// signal a long-lived telemetry backend must honor, or every closed
|
||||
// session (and its full event log) stays strongly held for the backend's
|
||||
// lifetime and final unload emits shutdown markers for dead sessions.
|
||||
// the store and emits `session/disposed` — the authoritative termination
|
||||
// edge. The marker must ride THAT edge (receivers classify a session with
|
||||
// activity and no marker as crashed, so a normally closed session in a
|
||||
// long-running host must not look like a crash), and the session retires
|
||||
// from the adopted set so unload neither retains it nor re-marks it.
|
||||
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.sessions.create(SessionId('ephemeral'), { meta: {} })
|
||||
}, { inject: ['sessions'] }))
|
||||
await owner.dispose()
|
||||
const atEdge = backend.records.filter(r => r.channel === 'ops')
|
||||
expect(atEdge.map(r => r.attributes['session.id'])).toEqual(['ephemeral'])
|
||||
expect(atEdge[0]!.attributes['telemetry.op']).toBe('shutdown')
|
||||
await fiber.dispose()
|
||||
const ops = backend.records.filter(r => r.channel === 'ops')
|
||||
expect(ops.map(r => r.attributes['session.id'])).toEqual(['survivor'])
|
||||
expect(ops.map(r => r.attributes['session.id'])).toEqual(['ephemeral', 'survivor'])
|
||||
})
|
||||
|
||||
it('warns instead of throwing when backend shutdown fails', async () => {
|
||||
|
||||
Reference in New Issue
Block a user