diff --git a/apps/supervisor/src/index.ts b/apps/supervisor/src/index.ts index c0779eb72..3a1e6165f 100644 --- a/apps/supervisor/src/index.ts +++ b/apps/supervisor/src/index.ts @@ -259,6 +259,8 @@ class ManagedSupervisor { await runWideEvent( { ...this.wideEventOpts, + op: "dequeue", + kind: "inbound", traceparent, setup: (state) => { setMeta(state, "run_id", message.run.friendlyId); @@ -269,8 +271,6 @@ class ManagedSupervisor { setMeta(state, "deployment_id", message.deployment.friendlyId); } setMeta(state, "machine_preset", message.run.machine.name); - state.extras.op = "dequeue"; - state.extras.kind = "inbound"; state.extras.iteration = "dequeue"; state.extras.dequeue_response_ms = dequeueResponseMs; state.extras.polling_interval_ms = pollingIntervalMs; diff --git a/apps/supervisor/src/services/computeSnapshotService.ts b/apps/supervisor/src/services/computeSnapshotService.ts index a4684a104..35ac6acec 100644 --- a/apps/supervisor/src/services/computeSnapshotService.ts +++ b/apps/supervisor/src/services/computeSnapshotService.ts @@ -76,9 +76,9 @@ export class ComputeSnapshotService { this.timerWheel.submit(runFriendlyId, data); emitOneShot({ ...this.wideEventOpts, + op: "snapshot.schedule", + kind: "event", populate: (state) => { - state.extras.op = "snapshot.schedule"; - state.extras.kind = "event"; state.meta.run_id = runFriendlyId; state.meta.snapshot_id = data.snapshotFriendlyId; state.extras.runner_id = data.runnerId; @@ -98,9 +98,9 @@ export class ComputeSnapshotService { if (cancelled) { emitOneShot({ ...this.wideEventOpts, + op: "snapshot.canceled", + kind: "event", populate: (state) => { - state.extras.op = "snapshot.canceled"; - state.extras.kind = "event"; state.meta.run_id = runFriendlyId; }, }); @@ -121,7 +121,6 @@ export class ComputeSnapshotService { // become extras/meta on the same wide event - no nested emission. const state = fromContext(); if (state) { - state.extras.op = "snapshot.callback"; state.extras["snapshot.status"] = body.status; if (body.instance_id) state.extras["snapshot.instance_id"] = body.instance_id; if (body.duration_ms !== undefined) state.extras["snapshot.duration_ms"] = body.duration_ms; @@ -247,9 +246,9 @@ export class ComputeSnapshotService { await runWideEvent( { ...this.wideEventOpts, + op: "snapshot.dispatch", + kind: "scheduled", setup: (state) => { - state.extras.op = "snapshot.dispatch"; - state.extras.kind = "scheduled"; state.meta.run_id = snapshot.runFriendlyId; state.meta.snapshot_id = snapshot.snapshotFriendlyId; state.extras.runner_id = snapshot.runnerId; diff --git a/apps/supervisor/src/wideEvents/emit.ts b/apps/supervisor/src/wideEvents/emit.ts index a1644237d..f3f765cdd 100644 --- a/apps/supervisor/src/wideEvents/emit.ts +++ b/apps/supervisor/src/wideEvents/emit.ts @@ -27,6 +27,9 @@ export function emit(state: State): void { appendIfSet(out, "region", state.region); appendIfSet(out, "node_id", state.nodeId); + appendIfSet(out, "op", state.op); + appendIfSet(out, "kind", state.kind); + out.ok = state.ok; if (state.statusCode !== 0) out.status = state.statusCode; out.duration_ms = state.durationMs; diff --git a/apps/supervisor/src/wideEvents/middleware.test.ts b/apps/supervisor/src/wideEvents/middleware.test.ts index 18a6469d7..afb59f43d 100644 --- a/apps/supervisor/src/wideEvents/middleware.test.ts +++ b/apps/supervisor/src/wideEvents/middleware.test.ts @@ -20,7 +20,7 @@ describe("runWideEvent", () => { it("emits one event with ok=true when no statusCode is set", async () => { const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true, route: "/x", method: "POST" }, + { service: "supervisor", env: {}, enabled: true, op: "test", route: "/x", method: "POST" }, async () => undefined ); }); @@ -39,7 +39,7 @@ describe("runWideEvent", () => { it("derives ok from statusCode set via finalize", async () => { const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true }, + { service: "supervisor", env: {}, enabled: true, op: "test" }, async () => undefined, (state) => { state.statusCode = 200; @@ -56,7 +56,7 @@ describe("runWideEvent", () => { it("treats 4xx as ok=false", async () => { const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true }, + { service: "supervisor", env: {}, enabled: true, op: "test" }, async () => undefined, (state) => { state.statusCode = 400; @@ -73,7 +73,7 @@ describe("runWideEvent", () => { it("emits ok=false with error.kind=internal on throw", async () => { const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true }, + { service: "supervisor", env: {}, enabled: true, op: "test" }, async () => { throw new Error("boom"); } @@ -91,7 +91,7 @@ describe("runWideEvent", () => { it("threads state through AsyncLocalStorage", async () => { const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true }, + { service: "supervisor", env: {}, enabled: true, op: "test" }, async () => { setMeta(fromContext(), "run_id", "run_abc"); } @@ -108,7 +108,7 @@ describe("runWideEvent", () => { const tp = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: true, traceparent: tp }, + { service: "supervisor", env: {}, enabled: true, op: "test", traceparent: tp }, async () => undefined ); }); @@ -124,7 +124,7 @@ describe("runWideEvent", () => { { service: "supervisor", env: {}, - enabled: true, + enabled: true, op: "test", setup: (state) => { state.meta.run_id = "run_abc"; state.extras.iteration = "dequeue"; @@ -144,7 +144,7 @@ describe("runWideEvent", () => { let seenState: ReturnType = null; const lines = await captureStdout(async () => { await runWideEvent( - { service: "supervisor", env: {}, enabled: false }, + { service: "supervisor", env: {}, enabled: false, op: "test" }, async () => { seenState = fromContext(); } @@ -159,7 +159,7 @@ describe("runWideEvent", () => { await Promise.all( ["a", "b", "c"].map((tag) => runWideEvent( - { service: "supervisor", env: {}, enabled: true }, + { service: "supervisor", env: {}, enabled: true, op: "test" }, async () => { const s = fromContext(); if (!s) throw new Error("no state"); @@ -182,7 +182,7 @@ describe("emitOneShot", () => { emitOneShot({ service: "supervisor", env: {}, - enabled: true, + enabled: true, op: "test", populate: (s) => { s.meta.run_id = "run_abc"; s.extras.event = "run:start"; @@ -200,7 +200,7 @@ describe("emitOneShot", () => { it("emits nothing when disabled", async () => { const lines = await captureStdout(() => { - emitOneShot({ service: "supervisor", env: {}, enabled: false }); + emitOneShot({ service: "supervisor", env: {}, enabled: false, op: "test" }); }); expect(lines).toHaveLength(0); }); diff --git a/apps/supervisor/src/wideEvents/middleware.ts b/apps/supervisor/src/wideEvents/middleware.ts index a87a1885d..034c13641 100644 --- a/apps/supervisor/src/wideEvents/middleware.ts +++ b/apps/supervisor/src/wideEvents/middleware.ts @@ -18,6 +18,10 @@ export type WideEventOptions = { /** Per-invocation options layered on top of `WideEventOptions`. */ export type WideEventLifecycleOptions = WideEventOptions & { + /** Operation discriminator (`instance.create`, `dequeue`, ...). Required. */ + op: string; + /** Event shape: `inbound` | `outbound` | `event` | `scheduled`. Optional. */ + kind?: string; /** Route template (HTTP only) captured into `extras.route`. */ route?: string; /** HTTP method captured into `extras.method`. */ @@ -54,6 +58,8 @@ export async function runWideEvent( env: opts.env, inboundRequestId: opts.inboundRequestId, traceparent: opts.traceparent, + op: opts.op, + kind: opts.kind, }); if (opts.route) state.extras.route = opts.route; if (opts.method) state.extras.method = opts.method; @@ -94,6 +100,8 @@ export async function runWideEvent( */ export function emitOneShot( opts: WideEventOptions & { + op: string; + kind?: string; traceparent?: string; populate?: (state: State) => void; } @@ -103,6 +111,8 @@ export function emitOneShot( service: opts.service, env: opts.env, traceparent: opts.traceparent, + op: opts.op, + kind: opts.kind, }); if (opts.populate) opts.populate(state); state.ok = true; diff --git a/apps/supervisor/src/wideEvents/new.ts b/apps/supervisor/src/wideEvents/new.ts index 2b7c88b7c..79d3e1593 100644 --- a/apps/supervisor/src/wideEvents/new.ts +++ b/apps/supervisor/src/wideEvents/new.ts @@ -37,6 +37,10 @@ export type NewStateOptions = { inboundRequestId?: string; /** Optional inbound W3C traceparent (HTTP header, queue message field). */ traceparent?: string; + /** Operation discriminator. Dotted `noun.verb`. Defaults to empty (set later). */ + op?: string; + /** Event shape: `inbound` | `outbound` | `event` | `scheduled`. Defaults to empty. */ + kind?: string; }; /** @@ -62,6 +66,8 @@ export function newState(opts: NewStateOptions): State { commitSha: opts.env.commitSha, region: opts.env.region, nodeId: opts.env.nodeId, + op: opts.op ?? "", + kind: opts.kind ?? "", meta: {}, phases: [], ok: false, diff --git a/apps/supervisor/src/wideEvents/state.ts b/apps/supervisor/src/wideEvents/state.ts index 1929e5e98..a0e2997b1 100644 --- a/apps/supervisor/src/wideEvents/state.ts +++ b/apps/supervisor/src/wideEvents/state.ts @@ -22,6 +22,22 @@ export type State = { region?: string; nodeId?: string; + /** + * Operation discriminator. Dotted `noun.verb` (e.g. `instance.create`, + * `snapshot.dispatch`). Low cardinality - bounded set per service, not + * unbounded. Empty allowed during construction but expected to be set + * before emit. + */ + op: string; + + /** + * Event shape. `inbound` for received requests, `outbound` for outgoing + * calls, `event` for ambient occurrences with no meaningful duration, + * `scheduled` for timer-driven work. Empty allowed; omitted from emit + * when empty. + */ + kind: string; + // Caller-attached opaque metadata, flattened to `meta.` on emit. meta: Record; diff --git a/apps/supervisor/src/workloadServer/index.ts b/apps/supervisor/src/workloadServer/index.ts index e4bb4625a..f5c5ded14 100644 --- a/apps/supervisor/src/workloadServer/index.ts +++ b/apps/supervisor/src/workloadServer/index.ts @@ -196,15 +196,13 @@ export class WorkloadServer extends EventEmitter { { ...this.wideEventOpts, enabled, + op, + kind: "inbound", route, method, traceparent: this.headerValueFromRequest(ctx.req, "traceparent"), inboundRequestId: this.headerValueFromRequest(ctx.req, "x-request-id"), - setup: (state) => { - state.extras.op = op; - state.extras.kind = "inbound"; - this.attachRouteMeta(state, ctx.params); - }, + setup: (state) => this.attachRouteMeta(state, ctx.params), }, fn, (state) => { @@ -679,9 +677,9 @@ export class WorkloadServer extends EventEmitter { ) => { emitOneShot({ ...this.wideEventOpts, + op: event === "run_connected" ? "socket.run.connected" : "socket.run.disconnected", + kind: "event", populate: (state) => { - state.extras.op = event === "run_connected" ? "socket.run.connected" : "socket.run.disconnected"; - state.extras.kind = "event"; state.extras.event = event; setMeta(state, "run_id", friendlyId); if (socket.data.deploymentId) {