feat(supervisor): wide-event module + kill switch

Author: nicktrn

apps/supervisor/src/env.ts

@@ -256,6 +256,11 @@ const Env = z
     // Debug
     DEBUG: BoolEnv.default(false),
     SEND_RUN_DEBUG_LOGS: BoolEnv.default(false),
+
+    // Wide-event observability - off by default. Emits one flat-keyed JSON
+    // line per natural unit of work (dequeue iteration, HTTP request, socket
+    // lifecycle). High-QPS hotpath, so the kill switch must be honoured.
+    TRIGGER_WIDE_EVENTS_ENABLED: BoolEnv.default(false),
   })
   .superRefine((data, ctx) => {
     if (data.COMPUTE_SNAPSHOTS_ENABLED && !data.TRIGGER_METADATA_URL) {

apps/supervisor/src/wideEvents/context.ts

@@ -0,0 +1,14 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import type { State } from "./state.js";
+
+/**
+ * AsyncLocalStorage threading per-operation `State` through the call stack.
+ * Wrappers enter a state via `wideEventStorage.run(state, () => fn())` and
+ * any code in the async call tree retrieves it via `fromContext()`.
+ */
+export const wideEventStorage = new AsyncLocalStorage<State>();
+
+/** Returns the State attached to the current async context, or null. */
+export function fromContext(): State | null {
+  return wideEventStorage.getStore() ?? null;
+}

apps/supervisor/src/wideEvents/emit.test.ts

@@ -0,0 +1,115 @@
+import { describe, it, expect } from "vitest";
+import { emit, EmitMessage } from "./emit.js";
+import { newState } from "./new.js";
+
+function captureEmit(state: Parameters<typeof emit>[0]): Record<string, unknown> {
+  const captured: string[] = [];
+  const origWrite = process.stdout.write;
+  process.stdout.write = ((chunk: unknown) => {
+    captured.push(String(chunk));
+    return true;
+  }) as typeof process.stdout.write;
+  try {
+    emit(state);
+  } finally {
+    process.stdout.write = origWrite;
+  }
+  expect(captured).toHaveLength(1);
+  const line = captured[0];
+  if (!line) throw new Error("no captured line");
+  return JSON.parse(line) as Record<string, unknown>;
+}
+
+describe("emit", () => {
+  it("emits a single line with the stable message + request_id", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 200;
+    s.ok = true;
+    s.durationMs = 5;
+    const out = captureEmit(s);
+    expect(out.msg).toBe(EmitMessage);
+    expect(out.request_id).toBe(s.requestId);
+    expect(out.service).toBe("supervisor");
+    expect(out.ok).toBe(true);
+    expect(out.status).toBe(200);
+    expect(out.duration_ms).toBe(5);
+  });
+
+  it("omits empty optional fields", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 200;
+    s.ok = true;
+    const out = captureEmit(s);
+    expect(out).not.toHaveProperty("trace_id");
+    expect(out).not.toHaveProperty("version");
+    expect(out).not.toHaveProperty("commit_sha");
+    expect(out).not.toHaveProperty("instance_id");
+    expect(out).not.toHaveProperty("error.code");
+  });
+
+  it("flattens meta keys as meta.<key>", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 200;
+    s.ok = true;
+    s.meta.run_id = "run_abc";
+    s.meta.deployment_id = "dep_xyz";
+    const out = captureEmit(s);
+    expect(out["meta.run_id"]).toBe("run_abc");
+    expect(out["meta.deployment_id"]).toBe("dep_xyz");
+    expect(out).not.toHaveProperty("meta");
+  });
+
+  it("flattens phases as phase.<name>.<field>", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 200;
+    s.ok = true;
+    s.phases.push({ name: "warm_start", durationMs: 12, ok: true, attempts: 1 });
+    s.phases.push({
+      name: "workload_create",
+      durationMs: 3,
+      ok: false,
+      attempts: 2,
+      errorCode: "Error",
+      errorMsg: "boom",
+      sub: { create_ms: 1 },
+    });
+    const out = captureEmit(s);
+    expect(out["phase.warm_start.duration_ms"]).toBe(12);
+    expect(out["phase.warm_start.ok"]).toBe(true);
+    expect(out["phase.warm_start.attempts"]).toBe(1);
+    expect(out["phase.workload_create.duration_ms"]).toBe(3);
+    expect(out["phase.workload_create.ok"]).toBe(false);
+    expect(out["phase.workload_create.attempts"]).toBe(2);
+    expect(out["phase.workload_create.error_code"]).toBe("Error");
+    expect(out["phase.workload_create.error_message"]).toBe("boom");
+    expect(out["phase.workload_create.create_ms"]).toBe(1);
+  });
+
+  it("includes error.code/message/kind when state.error is set", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 500;
+    s.error = { code: "InternalError", message: "kaboom", kind: "internal" };
+    const out = captureEmit(s);
+    expect(out["error.code"]).toBe("InternalError");
+    expect(out["error.message"]).toBe("kaboom");
+    expect(out["error.kind"]).toBe("internal");
+  });
+
+  it("truncates very long error messages", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.error = { code: "Big", message: "x".repeat(2000), kind: "internal" };
+    const out = captureEmit(s);
+    expect((out["error.message"] as string).length).toBe(512);
+  });
+
+  it("flattens extras at the top level", () => {
+    const s = newState({ service: "supervisor", env: {} });
+    s.statusCode = 200;
+    s.ok = true;
+    s.extras.route = "/health";
+    s.extras["dispatch.result"] = "hit";
+    const out = captureEmit(s);
+    expect(out.route).toBe("/health");
+    expect(out["dispatch.result"]).toBe("hit");
+  });
+});

apps/supervisor/src/wideEvents/emit.ts

@@ -0,0 +1,72 @@
+import type { State } from "./state.js";
+
+/**
+ * Stable slog message string for every wide event. Downstream filters (jq,
+ * Axiom queries, Vector pipelines) pin to this constant. The `service` field
+ * disambiguates which service emitted it.
+ */
+export const EmitMessage = "wide_event";
+
+const MAX_ERROR_MSG_BYTES = 512;
+
+/**
+ * Serializes a State as a single flat-keyed JSON line on stdout. Keys are
+ * flat (no nested objects) to keep jq filtering and Axiom indexing cheap.
+ * Empty optional fields are omitted.
+ */
+export function emit(state: State): void {
+  const out: Record<string, unknown> = {
+    msg: EmitMessage,
+    request_id: state.requestId,
+  };
+
+  if (state.traceId) out.trace_id = state.traceId;
+  appendIfSet(out, "service", state.service);
+  appendIfSet(out, "version", state.version);
+  appendIfSet(out, "commit_sha", state.commitSha);
+  appendIfSet(out, "region", state.region);
+  appendIfSet(out, "node_id", state.nodeId);
+  appendIfSet(out, "instance_id", state.instanceId);
+
+  out.ok = state.ok;
+  if (state.statusCode !== 0) out.status = state.statusCode;
+  out.duration_ms = state.durationMs;
+
+  if (state.error) {
+    appendIfSet(out, "error.code", state.error.code);
+    const msg =
+      state.error.message.length > MAX_ERROR_MSG_BYTES
+        ? state.error.message.slice(0, MAX_ERROR_MSG_BYTES)
+        : state.error.message;
+    appendIfSet(out, "error.message", msg);
+    appendIfSet(out, "error.kind", state.error.kind);
+  }
+
+  for (const [k, v] of Object.entries(state.meta)) {
+    out["meta." + k] = v;
+  }
+
+  for (const p of state.phases) {
+    const prefix = "phase." + p.name + ".";
+    out[prefix + "duration_ms"] = p.durationMs;
+    out[prefix + "ok"] = p.ok;
+    out[prefix + "attempts"] = p.attempts;
+    if (p.errorCode) out[prefix + "error_code"] = p.errorCode;
+    if (p.errorMsg) out[prefix + "error_message"] = p.errorMsg;
+    if (p.sub) {
+      for (const [sk, sv] of Object.entries(p.sub)) {
+        out[prefix + sk] = sv;
+      }
+    }
+  }
+
+  for (const [k, v] of Object.entries(state.extras)) {
+    out[k] = v;
+  }
+
+  process.stdout.write(JSON.stringify(out) + "\n");
+}
+
+function appendIfSet(out: Record<string, unknown>, key: string, value: string | undefined): void {
+  if (value) out[key] = value;
+}

apps/supervisor/src/wideEvents/index.ts

@@ -0,0 +1,28 @@
+/**
+ * Wide-event observability surface for the supervisor. One flat-keyed JSON
+ * line per natural unit of work (HTTP request, dequeue iteration, socket
+ * lifecycle event). Events join across services via `trace_id` (parsed from
+ * the inbound W3C `traceparent`) and `meta.run_id`.
+ *
+ * Off by default behind a kill switch - the dispatch hotpath runs at high
+ * QPS, so logging pressure must be cleanly removable.
+ */
+export { type Env, isValidRequestId, newState, type NewStateOptions } from "./new.js";
+export { emit, EmitMessage } from "./emit.js";
+export { parseTraceId } from "./traceparent.js";
+export { fromContext, wideEventStorage } from "./context.js";
+export {
+  type PhaseOpt,
+  recordPhase,
+  recordPhaseSince,
+  timePhase,
+} from "./record.js";
+export {
+  emitOneShot,
+  runWideEvent,
+  setExtra,
+  setMeta,
+  type WideEventLifecycleOptions,
+  type WideEventOptions,
+} from "./middleware.js";
+export type { ErrorInfo, PhaseRecord, State } from "./state.js";