fix(substrate): data-plane reliability, concurrency & scale hardening

src/analyst/analyst.test.ts

@@ -754,7 +754,15 @@ describe('ChatClient signal racing', () => {
     // documents the limit: races live in wrapLlmClient, mock passes
     // through. Either resolves (slow path) or rejects (when bound).
     // We just assert it eventually completes without hanging.
-    await Promise.race([p.catch(() => undefined), new Promise((r) => setTimeout(r, 200))])
-    expect(true).toBe(true)
+    const settled = await Promise.race([
+      p.then(
+        () => 'settled',
+        () => 'settled',
+      ),
+      new Promise<string>((r) => setTimeout(() => r('hung'), 200)),
+    ])
+    // The mock transport passes the signal through; the real contract here is
+    // only that the call SETTLES (resolves or rejects) and never hangs.
+    expect(settled).toBe('settled')
   })
 })

src/analyst/registry.budget.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it } from 'vitest'
+import { AnalystRegistry } from './registry'
+import type { Analyst, AnalystContext, AnalystFinding } from './types'
+
+function recorder(id: string, inputKind: Analyst['inputKind'], sink: { budget?: number }): Analyst {
+  return {
+    id,
+    description: id,
+    inputKind,
+    cost: { kind: 'deterministic' },
+    version: '1.0.0',
+    async analyze(_input: unknown, ctx: AnalystContext): Promise<AnalystFinding[]> {
+      sink.budget = ctx.budgetUsd
+      return []
+    },
+  }
+}
+
+describe('AnalystRegistry budget allocation', () => {
+  it('splits budget across analysts that actually run, not ones skipped for missing input', async () => {
+    const sink: { budget?: number } = {}
+    const reg = new AnalystRegistry()
+    reg.register(recorder('runs', 'trace-store', sink))
+    // artifact-dir input is NOT supplied → this analyst is skipped, so it must
+    // not dilute the budget of the one that runs.
+    reg.register(recorder('skipped', 'artifact-dir', {}))
+
+    await reg.run('t', { traceStore: {} as never }, { budget: { totalUsd: 10 } })
+
+    // runnableCount = 1 → full budget. The pre-fix code divided by 2 → 5.
+    expect(sink.budget).toBe(10)
+  })
+})

src/analyst/registry.ts

@@ -196,6 +196,13 @@ export class AnalystRegistry {
     let totalCost = 0
     let remainingUsd = budget?.totalUsd
 
+    // Budget is split only across analysts that actually run. Analysts skipped
+    // for missing input never spend, so counting them would under-budget the
+    // ones that do. routeInput is pure, so the pre-count is safe.
+    const runnableCount = selected.filter(
+      (a) => this.routeInput(a, inputs).kind !== 'missing',
+    ).length
+
     for (const analyst of selected) {
       const t0 = Date.now()
       const input = this.routeInput(analyst, inputs)
@@ -218,7 +225,7 @@ export class AnalystRegistry {
       const perBudget = allocateBudget(budget, {
         analyst,
         remainingUsd,
-        runningCount: selected.length,
+        runningCount: runnableCount,
       })
 
       const ctx: AnalystContext = {

src/client.test.ts

@@ -0,0 +1,124 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { ProductClient } from './client'
+
+function mkOk(body: unknown): Response {
+  return new Response(JSON.stringify(body), {
+    status: 200,
+    headers: { 'content-type': 'application/json' },
+  })
+}
+
+function mkErr(status: number, body: string): Response {
+  return new Response(body, { status })
+}
+
+const ROUTES = {
+  tasks: '/api/tasks',
+  events: '/api/events',
+  approvals: '/api/approvals',
+  vault: '/api/vault',
+  generations: '/api/generations',
+}
+
+function newClient() {
+  return new ProductClient({ baseUrl: 'https://app.test', routes: ROUTES })
+}
+
+describe('ProductClient — fail-loud on non-ok responses', () => {
+  const realFetch = globalThis.fetch
+  afterEach(() => {
+    globalThis.fetch = realFetch
+  })
+
+  it('getTasks throws on a 500 instead of masking it as zero results', async () => {
+    // OLD behavior: res.json() was called without checking res.ok, then the
+    // error body's missing `tasks` key was masked by `?? []`, so a 500 with an
+    // error JSON body returned [] — indistinguishable from a healthy empty set.
+    globalThis.fetch = vi.fn(async () =>
+      mkErr(500, JSON.stringify({ error: 'internal server error' })),
+    ) as unknown as typeof fetch
+
+    await expect(newClient().getTasks('ws-1')).rejects.toThrow(/HTTP 500/)
+  })
+
+  it('getApprovals throws on a 401 (auth) rather than returning [])', async () => {
+    globalThis.fetch = vi.fn(async () =>
+      mkErr(401, JSON.stringify({ error: 'unauthorized' })),
+    ) as unknown as typeof fetch
+
+    await expect(newClient().getApprovals('ws-1')).rejects.toThrow(/HTTP 401/)
+  })
+
+  it('getEvents fails loud when a 200 body omits the required array field', async () => {
+    // A wrong route or a contract drift can return 200 `{}`. The old `?? []`
+    // turned that into "zero events"; now it must surface as a defect.
+    globalThis.fetch = vi.fn(async () => mkOk({ unexpected: true })) as unknown as typeof fetch
+
+    await expect(newClient().getEvents('ws-1')).rejects.toThrow(/missing array field "events"/)
+  })
+
+  it('getVaultTree fails loud when a 200 body omits the tree array', async () => {
+    globalThis.fetch = vi.fn(async () => mkOk({})) as unknown as typeof fetch
+    await expect(newClient().getVaultTree('ws-1')).rejects.toThrow(/missing array field "tree"/)
+  })
+
+  it('returns the parsed array on a healthy 200', async () => {
+    globalThis.fetch = vi.fn(async () =>
+      mkOk({ tasks: [{ id: 't1', title: 'a', status: 'open', priority: 'high' }] }),
+    ) as unknown as typeof fetch
+
+    const tasks = await newClient().getTasks('ws-1')
+    expect(tasks).toHaveLength(1)
+    expect(tasks[0]!.id).toBe('t1')
+  })
+
+  it('a genuinely empty result set is still allowed (empty array, not masked error)', async () => {
+    globalThis.fetch = vi.fn(async () => mkOk({ tasks: [] })) as unknown as typeof fetch
+    const tasks = await newClient().getTasks('ws-1')
+    expect(tasks).toEqual([])
+  })
+
+  it('generic post throws with status + body on a 4xx', async () => {
+    globalThis.fetch = vi.fn(async () =>
+      mkErr(422, 'validation failed: name required'),
+    ) as unknown as typeof fetch
+
+    await expect(newClient().post('/api/x', { a: 1 })).rejects.toThrow(
+      /HTTP 422.*validation failed/,
+    )
+  })
+})
+
+describe('ProductClient — request timeout', () => {
+  const realFetch = globalThis.fetch
+  beforeEach(() => {
+    vi.useFakeTimers()
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+    globalThis.fetch = realFetch
+  })
+
+  it('aborts a hung request after the configured timeout', async () => {
+    // fetch resolves only if its signal aborts (mirrors how undici/fetch
+    // rejects an aborted request). A never-resolving server must not hang the
+    // harness forever.
+    globalThis.fetch = ((_url: string, init: RequestInit) =>
+      new Promise((_resolve, reject) => {
+        const signal = init.signal
+        if (signal) {
+          signal.addEventListener('abort', () => {
+            const err = new Error('The operation was aborted')
+            err.name = 'AbortError'
+            reject(err)
+          })
+        }
+      })) as unknown as typeof fetch
+
+    const client = new ProductClient({ baseUrl: 'https://app.test', routes: ROUTES, timeoutMs: 50 })
+    const p = client.get('/api/slow')
+    const assertion = expect(p).rejects.toThrow(/aborted/i)
+    await vi.advanceTimersByTimeAsync(60)
+    await assertion
+  })
+})

src/client.ts

@@ -5,14 +5,18 @@ import type { CheckResult, ProductClientConfig, RouteMap, TestResult } from './t
  *
  * Routes are config, not hardcoded. Each agent provides its own RouteMap.
  */
+const DEFAULT_REQUEST_TIMEOUT_MS = 30_000
+
 export class ProductClient {
   private baseUrl: string
   private routes: RouteMap
   private cookies: string = ''
+  private timeoutMs: number
 
   constructor(config: ProductClientConfig) {
     this.baseUrl = config.baseUrl.replace(/\/+$/, '')
     this.routes = config.routes
+    this.timeoutMs = config.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS
   }
 
   private route(name: keyof RouteMap): string {
@@ -21,6 +25,57 @@ export class ProductClient {
     return path
   }
 
+  /**
+   * Single HTTP boundary for every JSON request. Aborts after `timeoutMs`,
+   * checks `res.ok` BEFORE parsing, and throws `${method} ${path} failed:
+   * HTTP ${status} — ${body}` on a non-ok response so a 4xx/5xx error body
+   * can never be parsed as a success shape. Callers inspect the resolved
+   * value only after a successful return.
+   */
+  private async request(
+    method: string,
+    path: string,
+    body?: Record<string, unknown>,
+  ): Promise<Record<string, unknown>> {
+    const controller = new AbortController()
+    const timeout = setTimeout(() => controller.abort(), this.timeoutMs)
+    let res: Response
+    try {
+      res = await fetch(`${this.baseUrl}${path}`, {
+        method,
+        headers: {
+          ...(body ? { 'Content-Type': 'application/json' } : {}),
+          Origin: this.baseUrl,
+          Cookie: this.cookies,
+        },
+        ...(body ? { body: JSON.stringify(body) } : {}),
+        signal: controller.signal,
+      })
+    } finally {
+      clearTimeout(timeout)
+    }
+    if (!res.ok) {
+      const text = await res.text().catch(() => '')
+      throw new Error(`${method} ${path} failed: HTTP ${res.status} — ${text.slice(0, 500)}`)
+    }
+    return res.json() as Promise<Record<string, unknown>>
+  }
+
+  /**
+   * Read a required collection field. A missing/non-array field is a contract
+   * violation (wrong route, drift, partial body) — surface it loud instead of
+   * masking it as a healthy empty set. A genuinely empty `[]` passes through.
+   */
+  private requireArray<T>(res: Record<string, unknown>, field: string): T[] {
+    const value = res[field]
+    if (!Array.isArray(value)) {
+      throw new Error(
+        `Response missing array field "${field}" (got ${typeof value}: ${JSON.stringify(value)?.slice(0, 200)})`,
+      )
+    }
+    return value as T[]
+  }
+
   async signup(name: string, email: string, password: string): Promise<{ userId: string }> {
     const res = await this.post(this.route('signup'), { name, email, password })
     const user = res.user as Record<string, unknown> | undefined
@@ -121,19 +176,25 @@ export class ProductClient {
     workspaceId: string,
   ): Promise<{ id: string; title: string; status: string; priority: string }[]> {
     const res = await this.get(`${this.route('tasks')}?workspaceId=${workspaceId}`)
-    return (res.tasks ?? []) as { id: string; title: string; status: string; priority: string }[]
+    return this.requireArray<{ id: string; title: string; status: string; priority: string }>(
+      res,
+      'tasks',
+    )
   }
 
   async getEvents(workspaceId: string): Promise<{ id: string; title: string; type: string }[]> {
     const res = await this.get(`${this.route('events')}?workspaceId=${workspaceId}`)
-    return (res.events ?? []) as { id: string; title: string; type: string }[]
+    return this.requireArray<{ id: string; title: string; type: string }>(res, 'events')
   }
 
   async getApprovals(
     workspaceId: string,
   ): Promise<{ id: string; title: string; status: string; type: string }[]> {
     const res = await this.get(`${this.route('approvals')}?workspaceId=${workspaceId}`)
-    return (res.actions ?? []) as { id: string; title: string; status: string; type: string }[]
+    return this.requireArray<{ id: string; title: string; status: string; type: string }>(
+      res,
+      'actions',
+    )
   }
 
   async getVaultTree(workspaceId: string): Promise<string[]> {
@@ -146,7 +207,7 @@ export class ProductClient {
         if (node.children) extract(node.children)
       }
     }
-    extract((res.tree ?? []) as unknown[])
+    extract(this.requireArray<unknown>(res, 'tree'))
     return paths
   }
 
@@ -162,43 +223,22 @@ export class ProductClient {
     workspaceId: string,
   ): Promise<{ id: string; type: string; prompt: string }[]> {
     const res = await this.get(`${this.route('generations')}?workspaceId=${workspaceId}`)
-    return (res.generations ?? []) as { id: string; type: string; prompt: string }[]
+    return this.requireArray<{ id: string; type: string; prompt: string }>(res, 'generations')
   }
 
   /** Generic GET for custom routes */
   async get(path: string): Promise<Record<string, unknown>> {
-    const res = await fetch(`${this.baseUrl}${path}`, {
-      headers: { Cookie: this.cookies },
-    })
-    return res.json() as Promise<Record<string, unknown>>
+    return this.request('GET', path)
   }
 
   /** Generic POST for custom routes */
   async post(path: string, body: Record<string, unknown>): Promise<Record<string, unknown>> {
-    const res = await fetch(`${this.baseUrl}${path}`, {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        Origin: this.baseUrl,
-        Cookie: this.cookies,
-      },
-      body: JSON.stringify(body),
-    })
-    return res.json() as Promise<Record<string, unknown>>
+    return this.request('POST', path, body)
   }
 
   /** Generic PATCH for custom routes */
   async patch(path: string, body: Record<string, unknown>): Promise<Record<string, unknown>> {
-    const res = await fetch(`${this.baseUrl}${path}`, {
-      method: 'PATCH',
-      headers: {
-        'Content-Type': 'application/json',
-        Origin: this.baseUrl,
-        Cookie: this.cookies,
-      },
-      body: JSON.stringify(body),
-    })
-    return res.json() as Promise<Record<string, unknown>>
+    return this.request('PATCH', path, body)
   }
 }