fix(testkit): macOS compat + fix test 8c ordering (#24)

artifacts/api-server/src/lib/agent.ts

@@ -1,4 +1,6 @@
-import { anthropic } from "@workspace/integrations-anthropic-ai";
+import { makeLogger } from "./logger.js";
+
+const logger = makeLogger("agent");
 
 export interface EvalResult {
   accepted: boolean;
@@ -18,17 +20,79 @@ export interface AgentConfig {
   workModel?: string;
 }
 
+// ── Stub mode detection ───────────────────────────────────────────────────────
+// If Anthropic credentials are absent, all AI calls return canned responses so
+// the server starts and exercises the full payment/state-machine flow without
+// a real API key. This mirrors the LNbits stub pattern.
+const STUB_MODE =
+  !process.env["AI_INTEGRATIONS_ANTHROPIC_API_KEY"] ||
+  !process.env["AI_INTEGRATIONS_ANTHROPIC_BASE_URL"];
+
+if (STUB_MODE) {
+  logger.warn("no Anthropic key — running in STUB mode", { component: "agent", stub: true });
+}
+
+const STUB_EVAL: EvalResult = {
+  accepted: true,
+  reason: "Stub: request accepted for processing.",
+  inputTokens: 0,
+  outputTokens: 0,
+};
+
+const STUB_RESULT =
+  "Stub response: Timmy is running in stub mode (no Anthropic API key). " +
+  "Configure AI_INTEGRATIONS_ANTHROPIC_API_KEY to enable real AI responses.";
+
+// ── Lazy client ───────────────────────────────────────────────────────────────
+// Minimal local interface — avoids importing @anthropic-ai/sdk types directly.
+// Dynamic import avoids the module-level throw in the integrations client when
+// env vars are absent (the client.ts guard runs at module evaluation time).
+interface AnthropicLike {
+  messages: {
+    create(params: Record<string, unknown>): Promise<{
+      content: Array<{ type: string; text?: string }>;
+      usage: { input_tokens: number; output_tokens: number };
+    }>;
+    stream(params: Record<string, unknown>): AsyncIterable<{
+      type: string;
+      delta?: { type: string; text?: string };
+      usage?: { output_tokens: number };
+      message?: { usage: { input_tokens: number } };
+    }>;
+  };
+}
+
+let _anthropic: AnthropicLike | null = null;
+
+async function getClient(): Promise<AnthropicLike> {
+  if (_anthropic) return _anthropic;
+  // @ts-expect-error -- TS6305: integrations-anthropic-ai exports src directly; project-reference build not required at runtime
+  const mod = (await import("@workspace/integrations-anthropic-ai")) as { anthropic: AnthropicLike };
+  _anthropic = mod.anthropic;
+  return _anthropic;
+}
+
+// ── AgentService ─────────────────────────────────────────────────────────────
+
 export class AgentService {
   readonly evalModel: string;
   readonly workModel: string;
+  readonly stubMode: boolean = STUB_MODE;
 
   constructor(config?: AgentConfig) {
-    this.evalModel = config?.evalModel ?? process.env.EVAL_MODEL ?? "claude-haiku-4-5";
-    this.workModel = config?.workModel ?? process.env.WORK_MODEL ?? "claude-sonnet-4-6";
+    this.evalModel = config?.evalModel ?? process.env["EVAL_MODEL"] ?? "claude-haiku-4-5";
+    this.workModel = config?.workModel ?? process.env["WORK_MODEL"] ?? "claude-sonnet-4-6";
   }
 
   async evaluateRequest(requestText: string): Promise<EvalResult> {
-    const message = await anthropic.messages.create({
+    if (STUB_MODE) {
+      // Simulate a short eval delay so state-machine tests are realistic
+      await new Promise((r) => setTimeout(r, 300));
+      return { ...STUB_EVAL };
+    }
+
+    const client = await getClient();
+    const message = await client.messages.create({
       model: this.evalModel,
       max_tokens: 8192,
       system: `You are Timmy, an AI agent gatekeeper. Evaluate whether a request is acceptable to act on.
@@ -45,10 +109,10 @@ Respond ONLY with valid JSON: {"accepted": true, "reason": "..."} or {"accepted"
 
     let parsed: { accepted: boolean; reason: string };
     try {
-      const raw = block.text.replace(/^```(?:json)?\s*/i, "").replace(/\s*```$/, "").trim();
+      const raw = block.text!.replace(/^```(?:json)?\s*/i, "").replace(/\s*```$/, "").trim();
       parsed = JSON.parse(raw) as { accepted: boolean; reason: string };
     } catch {
-      throw new Error(`Failed to parse eval JSON: ${block.text}`);
+      throw new Error(`Failed to parse eval JSON: ${block.text!}`);
     }
 
     return {
@@ -60,7 +124,13 @@ Respond ONLY with valid JSON: {"accepted": true, "reason": "..."} or {"accepted"
   }
 
   async executeWork(requestText: string): Promise<WorkResult> {
-    const message = await anthropic.messages.create({
+    if (STUB_MODE) {
+      await new Promise((r) => setTimeout(r, 500));
+      return { result: STUB_RESULT, inputTokens: 0, outputTokens: 0 };
+    }
+
+    const client = await getClient();
+    const message = await client.messages.create({
       model: this.workModel,
       max_tokens: 8192,
       system: `You are Timmy, a capable AI agent. A user has paid for you to handle their request.
@@ -74,11 +144,61 @@ Fulfill it thoroughly and helpfully. Be concise yet complete.`,
     }
 
     return {
-      result: block.text,
+      result: block.text!,
       inputTokens: message.usage.input_tokens,
       outputTokens: message.usage.output_tokens,
     };
   }
+
+  /**
+   * Streaming variant of executeWork (#3). Calls onChunk for every text delta.
+   * In stub mode, emits the canned response word-by-word to exercise the SSE
+   * path end-to-end without a real Anthropic key.
+   */
+  async executeWorkStreaming(
+    requestText: string,
+    onChunk: (delta: string) => void,
+  ): Promise<WorkResult> {
+    if (STUB_MODE) {
+      const words = STUB_RESULT.split(" ");
+      for (const word of words) {
+        const delta = word + " ";
+        onChunk(delta);
+        await new Promise((r) => setTimeout(r, 40));
+      }
+      return { result: STUB_RESULT, inputTokens: 0, outputTokens: 0 };
+    }
+
+    const client = await getClient();
+    let fullText = "";
+    let inputTokens = 0;
+    let outputTokens = 0;
+
+    const stream = client.messages.stream({
+      model: this.workModel,
+      max_tokens: 8192,
+      system: `You are Timmy, a capable AI agent. A user has paid for you to handle their request.
+Fulfill it thoroughly and helpfully. Be concise yet complete.`,
+      messages: [{ role: "user", content: requestText }],
+    });
+
+    for await (const event of stream) {
+      if (
+        event.type === "content_block_delta" &&
+        event.delta?.type === "text_delta"
+      ) {
+        const delta = event.delta!.text ?? "";
+        fullText += delta;
+        onChunk(delta);
+      } else if (event.type === "message_delta" && event.usage) {
+        outputTokens = event.usage!.output_tokens;
+      } else if (event.type === "message_start" && event.message?.usage) {
+        inputTokens = event.message!.usage.input_tokens;
+      }
+    }
+
+    return { result: fullText, inputTokens, outputTokens };
+  }
 }
 
 export const agentService = new AgentService();

artifacts/api-server/src/lib/btc-oracle.ts

@@ -1,3 +1,7 @@
+import { makeLogger } from "./logger.js";
+
+const logger = makeLogger("btc-oracle");
+
 const COINGECKO_URL =
   "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd";
 
@@ -42,7 +46,10 @@ export async function getBtcPriceUsd(): Promise<number> {
     return price;
   } catch (err) {
     const fb = fallbackPrice();
-    console.warn(`[btc-oracle] Price fetch failed (using $${fb} fallback):`, err);
+    logger.warn("price fetch failed — using fallback", {
+      fallback_usd: fb,
+      error: err instanceof Error ? err.message : String(err),
+    });
     return fb;
   }
 }

artifacts/api-server/src/lib/event-bus.ts

@@ -0,0 +1,34 @@
+import { EventEmitter } from "events";
+
+export type JobEvent =
+  | { type: "job:state"; jobId: string; state: string }
+  | { type: "job:paid"; jobId: string; invoiceType: "eval" | "work" }
+  | { type: "job:completed"; jobId: string; result: string }
+  | { type: "job:failed"; jobId: string; reason: string };
+
+export type SessionEvent =
+  | { type: "session:state"; sessionId: string; state: string }
+  | { type: "session:paid"; sessionId: string; amountSats: number }
+  | { type: "session:balance"; sessionId: string; balanceSats: number };
+
+export type BusEvent = JobEvent | SessionEvent;
+
+class EventBus extends EventEmitter {
+  emit(event: "bus", data: BusEvent): boolean;
+  emit(event: string, ...args: unknown[]): boolean {
+    return super.emit(event, ...args);
+  }
+
+  on(event: "bus", listener: (data: BusEvent) => void): this;
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  on(event: string, listener: (...args: any[]) => void): this {
+    return super.on(event, listener);
+  }
+
+  publish(data: BusEvent): void {
+    this.emit("bus", data);
+  }
+}
+
+export const eventBus = new EventBus();
+eventBus.setMaxListeners(256);

artifacts/api-server/src/lib/histogram.ts

@@ -0,0 +1,45 @@
+const MAX_SAMPLES = 1_000;
+
+export interface BucketStats {
+  p50: number | null;
+  p95: number | null;
+  count: number;
+}
+
+export class LatencyHistogram {
+  private readonly buckets = new Map<string, number[]>();
+
+  record(route: string, durationMs: number): void {
+    let bucket = this.buckets.get(route);
+    if (!bucket) {
+      bucket = [];
+      this.buckets.set(route, bucket);
+    }
+    if (bucket.length >= MAX_SAMPLES) {
+      bucket.shift();
+    }
+    bucket.push(durationMs);
+  }
+
+  percentile(route: string, pct: number): number | null {
+    const bucket = this.buckets.get(route);
+    if (!bucket || bucket.length === 0) return null;
+    const sorted = [...bucket].sort((a, b) => a - b);
+    const idx = Math.floor((pct / 100) * sorted.length);
+    return sorted[Math.min(idx, sorted.length - 1)] ?? null;
+  }
+
+  snapshot(): Record<string, BucketStats> {
+    const out: Record<string, BucketStats> = {};
+    for (const [route, bucket] of this.buckets.entries()) {
+      out[route] = {
+        p50: this.percentile(route, 50),
+        p95: this.percentile(route, 95),
+        count: bucket.length,
+      };
+    }
+    return out;
+  }
+}
+
+export const latencyHistogram = new LatencyHistogram();

artifacts/api-server/src/lib/lnbits.ts

@@ -1,4 +1,7 @@
 import { randomBytes } from "crypto";
+import { makeLogger } from "./logger.js";
+
+const logger = makeLogger("lnbits");
 
 export interface LNbitsInvoice {
   paymentHash: string;
@@ -22,7 +25,7 @@ export class LNbitsService {
     this.apiKey = config?.apiKey ?? process.env.LNBITS_API_KEY ?? "";
     this.stubMode = !this.url || !this.apiKey;
     if (this.stubMode) {
-      console.warn("[LNbitsService] No LNBITS_URL/LNBITS_API_KEY — running in STUB mode. Invoices are simulated.");
+      logger.warn("no LNBITS_URL/LNBITS_API_KEY — running in STUB mode", { stub: true });
     }
   }
 
@@ -32,7 +35,7 @@ export class LNbitsService {
     if (this.stubMode) {
       const paymentHash = randomBytes(32).toString("hex");
       const paymentRequest = `lnbcrt${amountSats}u1stub_${paymentHash.slice(0, 16)}`;
-      console.log(`[stub] Created invoice: ${amountSats} sats — "${memo}" — hash=${paymentHash}`);
+      logger.info("stub invoice created", { amountSats, memo, paymentHash });
       return { paymentHash, paymentRequest };
     }
 
@@ -113,7 +116,7 @@ export class LNbitsService {
   async payInvoice(bolt11: string): Promise<string> {
     if (this.stubMode) {
       const paymentHash = randomBytes(32).toString("hex");
-      console.log(`[stub] Paid outgoing invoice — fake hash=${paymentHash}`);
+      logger.info("stub outgoing payment", { paymentHash, invoiceType: "outbound" });
       return paymentHash;
     }
 
@@ -140,7 +143,7 @@ export class LNbitsService {
       throw new Error("stubMarkPaid called on a real LNbitsService instance");
     }
     stubPaidInvoices.add(paymentHash);
-    console.log(`[stub] Marked invoice paid: hash=${paymentHash}`);
+    logger.info("stub invoice marked paid", { paymentHash, invoiceType: "inbound" });
   }
 
   // ── Private helpers ──────────────────────────────────────────────────────

artifacts/api-server/src/lib/logger.ts

@@ -0,0 +1,32 @@
+export type LogLevel = "debug" | "info" | "warn" | "error";
+
+export interface LogContext {
+  [key: string]: unknown;
+}
+
+function emit(level: LogLevel, component: string, message: string, ctx?: LogContext): void {
+  const line: Record<string, unknown> = {
+    timestamp: new Date().toISOString(),
+    level,
+    component,
+    message,
+    ...ctx,
+  };
+  const out = JSON.stringify(line);
+  if (level === "error" || level === "warn") {
+    console.error(out);
+  } else {
+    console.log(out);
+  }
+}
+
+export function makeLogger(component: string) {
+  return {
+    debug: (message: string, ctx?: LogContext) => emit("debug", component, message, ctx),
+    info:  (message: string, ctx?: LogContext) => emit("info",  component, message, ctx),
+    warn:  (message: string, ctx?: LogContext) => emit("warn",  component, message, ctx),
+    error: (message: string, ctx?: LogContext) => emit("error", component, message, ctx),
+  };
+}
+
+export const rootLogger = makeLogger("server");

artifacts/api-server/src/lib/metrics.ts

@@ -0,0 +1,118 @@
+import { db, jobs, invoices } from "@workspace/db";
+import { sql } from "drizzle-orm";
+import { latencyHistogram, type BucketStats } from "./histogram.js";
+
+export interface JobStateCounts {
+  awaiting_eval: number;
+  awaiting_work: number;
+  complete: number;
+  rejected: number;
+  failed: number;
+}
+
+export interface MetricsSnapshot {
+  uptime_s: number;
+  jobs: {
+    total: number;
+    by_state: JobStateCounts;
+  };
+  invoices: {
+    total: number;
+    paid: number;
+    conversion_rate: number | null;
+  };
+  earnings: {
+    total_sats: number;
+  };
+  latency: {
+    eval_phase: BucketStats | null;
+    work_phase: BucketStats | null;
+    routes: Record<string, BucketStats>;
+  };
+}
+
+const START_TIME = Date.now();
+
+export class MetricsService {
+  async snapshot(): Promise<MetricsSnapshot> {
+    const [jobsByState, invoiceCounts, earningsRow] = await Promise.all([
+      db
+        .select({
+          state: jobs.state,
+          count: sql<number>`cast(count(*) as int)`,
+        })
+        .from(jobs)
+        .groupBy(jobs.state),
+
+      db
+        .select({
+          total: sql<number>`cast(count(*) as int)`,
+          paid:  sql<number>`cast(sum(case when paid then 1 else 0 end) as int)`,
+        })
+        .from(invoices),
+
+      db
+        .select({
+          total_sats: sql<number>`cast(coalesce(sum(actual_amount_sats), 0) as int)`,
+        })
+        .from(jobs),
+    ]);
+
+    // Group raw DB states into operational state keys
+    const rawCounts: Record<string, number> = {};
+    let jobsTotal = 0;
+    for (const row of jobsByState) {
+      const n = Number(row.count);
+      rawCounts[row.state] = (rawCounts[row.state] ?? 0) + n;
+      jobsTotal += n;
+    }
+
+    const byState: JobStateCounts = {
+      awaiting_eval: (rawCounts["awaiting_eval_payment"] ?? 0) + (rawCounts["evaluating"] ?? 0),
+      awaiting_work: (rawCounts["awaiting_work_payment"] ?? 0) + (rawCounts["executing"] ?? 0),
+      complete:  rawCounts["complete"] ?? 0,
+      rejected:  rawCounts["rejected"] ?? 0,
+      failed:    rawCounts["failed"] ?? 0,
+    };
+
+    const invRow = invoiceCounts[0] ?? { total: 0, paid: 0 };
+    const invTotal = Number(invRow.total);
+    const invPaid  = Number(invRow.paid);
+    const conversionRate = invTotal > 0 ? invPaid / invTotal : null;
+
+    const totalSats = Number(earningsRow[0]?.total_sats ?? 0);
+
+    const allRoutes = latencyHistogram.snapshot();
+    const evalPhase = allRoutes["eval_phase"] ?? null;
+    const workPhase = allRoutes["work_phase"] ?? null;
+    const routeLatency: Record<string, BucketStats> = {};
+    for (const [key, stats] of Object.entries(allRoutes)) {
+      if (key !== "eval_phase" && key !== "work_phase") {
+        routeLatency[key] = stats;
+      }
+    }
+
+    return {
+      uptime_s: Math.floor((Date.now() - START_TIME) / 1000),
+      jobs: {
+        total: jobsTotal,
+        by_state: byState,
+      },
+      invoices: {
+        total: invTotal,
+        paid: invPaid,
+        conversion_rate: conversionRate,
+      },
+      earnings: {
+        total_sats: totalSats,
+      },
+      latency: {
+        eval_phase: evalPhase,
+        work_phase: workPhase,
+        routes: routeLatency,
+      },
+    };
+  }
+}
+
+export const metricsService = new MetricsService();

artifacts/api-server/src/lib/provisioner.ts

@@ -1,6 +1,9 @@
 import { generateKeyPairSync } from "crypto";
 import { db, bootstrapJobs } from "@workspace/db";
 import { eq } from "drizzle-orm";
+import { makeLogger } from "./logger.js";
+
+const logger = makeLogger("provisioner");
 
 const DO_API_BASE = "https://api.digitalocean.com/v2";
 const TS_API_BASE = "https://api.tailscale.com/api/v2";
@@ -458,9 +461,7 @@ export class ProvisionerService {
     this.tsTailnet = process.env.TAILSCALE_TAILNET ?? "";
     this.stubMode = !this.doToken;
     if (this.stubMode) {
-      console.warn(
-        "[ProvisionerService] No DO_API_TOKEN — running in STUB mode. Provisioning is simulated.",
-      );
+      logger.warn("no DO_API_TOKEN — running in STUB mode", { stub: true });
     }
   }
 
@@ -477,7 +478,7 @@ export class ProvisionerService {
       }
     } catch (err) {
       const message = err instanceof Error ? err.message : "Provisioning failed";
-      console.error(`[ProvisionerService] Error for job ${bootstrapJobId}:`, message);
+      logger.error("provisioning failed", { bootstrapJobId, error: message });
       await db
         .update(bootstrapJobs)
         .set({ state: "failed", errorMessage: message, updatedAt: new Date() })
@@ -486,7 +487,7 @@ export class ProvisionerService {
   }
 
   private async stubProvision(jobId: string): Promise<void> {
-    console.log(`[stub] Simulating provisioning for bootstrap job ${jobId}...`);
+    logger.info("stub provisioning started", { bootstrapJobId: jobId });
     const { privateKey } = generateSshKeypair();
     await new Promise((r) => setTimeout(r, 2000));
     const fakeDropletId = String(Math.floor(Math.random() * 900_000_000 + 100_000_000));
@@ -502,11 +503,11 @@ export class ProvisionerService {
         updatedAt: new Date(),
       })
       .where(eq(bootstrapJobs.id, jobId));
-    console.log(`[stub] Bootstrap job ${jobId} marked ready with fake credentials.`);
+    logger.info("stub provisioning complete", { bootstrapJobId: jobId });
   }
 
   private async realProvision(jobId: string): Promise<void> {
-    console.log(`[ProvisionerService] Provisioning real node for job ${jobId}...`);
+    logger.info("real provisioning started", { bootstrapJobId: jobId });
 
     // 1. SSH keypair (pure node:crypto)
     const { publicKey, privateKey } = generateSshKeypair();
@@ -525,7 +526,7 @@ export class ProvisionerService {
       try {
         tailscaleAuthKey = await getTailscaleAuthKey(this.tsApiKey, this.tsTailnet);
       } catch (err) {
-        console.warn("[ProvisionerService] Tailscale key failed — continuing without:", err);
+        logger.warn("Tailscale key failed — continuing without Tailscale", { error: String(err) });
       }
     }
 
@@ -534,7 +535,7 @@ export class ProvisionerService {
     if (this.doVolumeGb > 0) {
       const volName = `timmy-data-${jobId.slice(0, 8)}`;
       volumeId = await createVolume(volName, this.doVolumeGb, this.doRegion, this.doToken);
-      console.log(`[ProvisionerService] Volume created: id=${volumeId} (${this.doVolumeGb} GB)`);
+      logger.info("block volume created", { volumeId, sizeGb: this.doVolumeGb });
     }
 
     // 5. Create droplet
@@ -556,11 +557,11 @@ export class ProvisionerService {
       dropletPayload,
     );
     const dropletId = dropletData.droplet.id;
-    console.log(`[ProvisionerService] Droplet created: id=${dropletId}`);
+    logger.info("droplet created", { bootstrapJobId: jobId, dropletId });
 
     // 6. Poll for public IP (up to 2 min)
     const nodeIp = await pollDropletIp(dropletId, this.doToken, 120_000);
-    console.log(`[ProvisionerService] Node IP: ${nodeIp ?? "(not yet assigned)"}`);
+    logger.info("node ip assigned", { bootstrapJobId: jobId, nodeIp: nodeIp ?? "(not yet assigned)" });
 
     // 7. Tailscale hostname
     const tailscaleHostname =
@@ -589,7 +590,7 @@ export class ProvisionerService {
       })
       .where(eq(bootstrapJobs.id, jobId));
 
-    console.log(`[ProvisionerService] Bootstrap job ${jobId} ready.`);
+    logger.info("real provisioning complete", { bootstrapJobId: jobId });
   }
 }
 

artifacts/api-server/src/lib/rate-limiter.ts

@@ -0,0 +1,46 @@
+import { rateLimit, type Options } from "express-rate-limit";
+import { makeLogger } from "./logger.js";
+
+const logger = makeLogger("rate-limiter");
+
+function envInt(key: string, fallback: number): number {
+  const v = process.env[key];
+  const n = v ? parseInt(v, 10) : NaN;
+  return Number.isNaN(n) ? fallback : n;
+}
+
+function limiter(windowMs: number, max: number, overrideKey?: string) {
+  const resolvedMax = overrideKey ? envInt(overrideKey, max) : max;
+  return rateLimit({
+    windowMs,
+    max: resolvedMax,
+    standardHeaders: "draft-7",
+    legacyHeaders: false,
+    handler: (req, res) => {
+      const ip =
+        (req.headers["x-forwarded-for"] as string | undefined)?.split(",")[0]?.trim() ??
+        req.socket.remoteAddress ??
+        "unknown";
+      logger.warn("rate limit hit", {
+        route:    req.path,
+        method:   req.method,
+        ip,
+        retry_after_s: Math.ceil(windowMs / 1000),
+      });
+      res.status(429).json({
+        error: "rate_limited",
+        message: "Too many requests — please slow down.",
+        retryAfterSeconds: Math.ceil(windowMs / 1000),
+      });
+    },
+  } satisfies Partial<Options>);
+}
+
+// POST /api/jobs — 30 req/min per IP (configurable via RATE_LIMIT_JOBS)
+export const jobsLimiter = limiter(60_000, 30, "RATE_LIMIT_JOBS");
+
+// POST /api/sessions — 10 req/min per IP (configurable via RATE_LIMIT_SESSIONS)
+export const sessionsLimiter = limiter(60_000, 10, "RATE_LIMIT_SESSIONS");
+
+// POST /api/bootstrap — 3 req/hour per IP (configurable via RATE_LIMIT_BOOTSTRAP)
+export const bootstrapLimiter = limiter(60 * 60_000, 3, "RATE_LIMIT_BOOTSTRAP");

artifacts/api-server/src/lib/stream-registry.ts

@@ -0,0 +1,55 @@
+import { PassThrough } from "stream";
+
+interface StreamEntry {
+  stream: PassThrough;
+  createdAt: number;
+}
+
+class StreamRegistry {
+  private readonly streams = new Map<string, StreamEntry>();
+  private readonly TTL_MS = 5 * 60 * 1000;
+
+  register(jobId: string): PassThrough {
+    const existing = this.streams.get(jobId);
+    if (existing) {
+      existing.stream.destroy();
+    }
+    const stream = new PassThrough();
+    this.streams.set(jobId, { stream, createdAt: Date.now() });
+
+    stream.on("close", () => {
+      this.streams.delete(jobId);
+    });
+
+    this.evictExpired();
+    return stream;
+  }
+
+  get(jobId: string): PassThrough | null {
+    return this.streams.get(jobId)?.stream ?? null;
+  }
+
+  write(jobId: string, chunk: string): void {
+    this.streams.get(jobId)?.stream.write(chunk);
+  }
+
+  end(jobId: string): void {
+    const entry = this.streams.get(jobId);
+    if (entry) {
+      entry.stream.end();
+      this.streams.delete(jobId);
+    }
+  }
+
+  private evictExpired(): void {
+    const now = Date.now();
+    for (const [id, entry] of this.streams.entries()) {
+      if (now - entry.createdAt > this.TTL_MS) {
+        entry.stream.destroy();
+        this.streams.delete(id);
+      }
+    }
+  }
+}
+
+export const streamRegistry = new StreamRegistry();