Merge latest master into invariant service seam

packages/llm/llm-pi-ai/README.md

@@ -19,7 +19,7 @@ Configure credentials and deployment-specific transport settings per provider. O
         reasoning: high
       - provider: anthropic
         apiKey: !!js process.env.ANTHROPIC_API_KEY
-        maxRetries: 2
+        streamIdleTimeoutMs: 300000
       - provider: openrouter
         apiKey: !!js process.env.OPENROUTER_API_KEY
         headers:
@@ -30,7 +30,9 @@ Each provider name must exist in pi-ai's installed catalog and may appear only o
 
 The adapter exposes each configured provider's installed pi-ai models through `ctx.llm.listModels(provider)`. This is provider-neutral selector metadata derived from `getModels(provider)`; request-time resolution still performs the authoritative catalog lookup, so discovery does not create a second model registry.
 
-Supported profile fields are `provider`, `apiKey`, `baseURL`, `headers`, `reasoning`, `thinkingBudgets`, `cacheRetention`, `transport`, `timeoutMs`, `websocketConnectTimeoutMs`, `maxRetries`, and `maxRetryDelayMs`. They map to pi-ai's common stream options. Harness app attribution wins a conflicting configured header name.
+Supported profile fields are `provider`, `apiKey`, `baseURL`, `headers`, `reasoning`, `thinkingBudgets`, `cacheRetention`, `transport`, `timeoutMs`, `websocketConnectTimeoutMs`, and `streamIdleTimeoutMs`. The stream-idle interval is a positive finite Node timer delay, defaults to five minutes, and covers only an outstanding provider read, not consumer think time. Harness app attribution wins a conflicting configured header name.
+
+The adapter forces pi-ai's SDK `maxRetries` to zero so one `stream()` call makes one provider request. The removed profile fields `maxRetries` and `maxRetryDelayMs` fail load instead of silently multiplying or hiding the separately composed agent-level retry budget. Idle expiry aborts the SDK's stable request signal and surfaces `TIMEOUT`; an earlier caller abort remains `ABORTED`.
 
 ## Provider/model routing and replay
 
@@ -43,7 +45,7 @@ If a listener rewrites assembled assistant content, the loop drops replay state
 ## Vocabulary differences
 
 - pi-ai tool-call arguments are parsed objects; the harness stores raw JSON strings. The adapter parses input and re-stringifies output.
-- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted'}` chunks. Provider-specific error text and usage signals evaluated against the resolved model's context window normalize overflow to `CONTEXT_WINDOW_EXCEEDED`.
+- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted', failure}` chunks. Provider-specific error text distinguishes terminal `QUOTA` from transient `RATE_LIMIT`, while text and usage signals evaluated against the resolved model's context window normalize overflow to `CONTEXT_WINDOW_EXCEEDED`.
 - pi-ai folds reasoning tokens into output usage; there is no separate reasoning count to map.
 - `GenerateOptions.stop` is rejected with `UNSUPPORTED_OPTION` because pi-ai's common streaming surface cannot guarantee it across providers.
 
@@ -57,7 +59,7 @@ pi-ai installs several provider SDKs and lazy-loads the one selected by the cata
 
 ## Testing
 
-Unit tests use pi-ai catalog models redirected to local mock servers and cover provider/profile routing, native API selection, endpoint overrides, attribution, conversion, replay-state validation, and cross-provider/model replay within one adapter instance. Real-API coverage remains key-gated under `pnpm run test:e2e`.
+Unit tests use pi-ai catalog models redirected to local mock servers and cover provider/profile routing, one wire request per adapter call, idle-timeout response termination, caller abort, native API selection, endpoint overrides, attribution, conversion, replay-state validation, and cross-provider/model replay within one adapter instance. Real-API coverage remains key-gated under `pnpm run test:e2e`.
 
 ## Model Experience
 
@@ -95,3 +97,4 @@ Recorded response content appends to the next request and does not invalidate it
 - **`GenerateOptions.stop` is unsupported** — pi-ai's common stream options cannot guarantee stop-sequence behavior across providers, so the adapter rejects the field.
 - **In-history `system` messages use pi-ai's common context conversion** — provider-specific placement follows pi-ai rather than a harness-owned wire override.
 - **Provider HTTP status is unavailable** — pi-ai error events do not expose a stable HTTP status across providers; failures expose only stable harness error codes.
+- **Retry policy is not an adapter option** — SDK retries are disabled so durable agent steps and `llm/retry` events own every visible attempt; direct `ctx.llm.stream()` calls remain single-attempt.

packages/llm/llm-pi-ai/package.json

@@ -29,6 +29,7 @@
   "peerDependencies": {
     "@deepseek-ai/dsh-invariants": "^0.0.1",
     "@deepseek-ai/dsh-llm": "^0.0.1",
+    "@deepseek-ai/dsh-timeout": "^0.0.1",
     "cordis": "^4.0.0-rc.7"
   },
   "dependencies": {
@@ -39,6 +40,7 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
     "@deepseek-ai/dsh-llm-deepseek": "workspace:^",
+    "@deepseek-ai/dsh-timeout": "workspace:^",
     "cordis": "^4.0.0-rc.7"
   }
 }

packages/llm/llm-pi-ai/src/adapter.ts

@@ -16,7 +16,9 @@ import type {
 } from '@earendil-works/pi-ai'
 import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
 import type { GenerateOptions, LlmModelInfo, StreamChunk } from '@deepseek-ai/dsh-llm'
-import type { PiAiProviderProfile } from './config.ts'
+import { idleWatchdog, timeoutOf } from '@deepseek-ai/dsh-timeout'
+import { resolveProfiles } from './config.ts'
+import type { PiAiProviderProfile, ResolvedPiAiProviderProfile } from './config.ts'
 import { toPiContext } from './context.ts'
 import { toStreamChunks } from './stream.ts'
 
@@ -48,8 +50,8 @@ function profileOptions(profile: PiAiProviderProfile): SimpleStreamOptions {
     ...profile.transport === undefined ? {} : { transport: profile.transport },
     ...profile.timeoutMs === undefined ? {} : { timeoutMs: profile.timeoutMs },
     ...profile.websocketConnectTimeoutMs === undefined ? {} : { websocketConnectTimeoutMs: profile.websocketConnectTimeoutMs },
-    ...profile.maxRetries === undefined ? {} : { maxRetries: profile.maxRetries },
-    ...profile.maxRetryDelayMs === undefined ? {} : { maxRetryDelayMs: profile.maxRetryDelayMs },
+    // The agent recovery layer owns visible attempts; one adapter call is one SDK attempt.
+    maxRetries: 0,
   }
 }
 
@@ -68,11 +70,11 @@ function requestHeaders(headers: Readonly<Record<string, string>> | undefined):
  * request, so models need not be registered during the Cordis lifecycle.
  */
 export class PiAiAdapter extends LlmAdapter {
-  private readonly profiles: ReadonlyMap<string, PiAiProviderProfile>
+  private readonly profiles: ReadonlyMap<string, ResolvedPiAiProviderProfile>
 
   constructor(options: PiAiAdapterOptions) {
     super()
-    this.profiles = new Map(options.profiles.map(profile => [profile.provider, profile]))
+    this.profiles = new Map(resolveProfiles(options.profiles).map(profile => [profile.provider, profile]))
   }
 
   override listModels(provider: string): Promise<readonly LlmModelInfo[]> {
@@ -97,12 +99,12 @@ export class PiAiAdapter extends LlmAdapter {
     }
     const model = resolveModel(profile, options.model)
 
-    // Pi-ai has no iterator-return cancellation hook. Chain an internal signal
-    // and abort it when this generator exits so early consumers stop the HTTP stream.
-    const controller = new AbortController()
-    const onCallerAbort = (): void => { controller.abort(options.signal?.reason) }
-    if (options.signal?.aborted) controller.abort(options.signal.reason)
-    else options.signal?.addEventListener('abort', onCallerAbort, { once: true })
+    const consumer = new AbortController()
+    const upstream = options.signal === undefined
+      ? consumer.signal
+      : AbortSignal.any([options.signal, consumer.signal])
+    const streamIdleTimeoutMs = profile.streamIdleTimeoutMs
+    using watchdog = idleWatchdog(upstream, streamIdleTimeoutMs, 'LLM_STREAM_IDLE_TIMEOUT')
 
     try {
       const events = streamSimple(model, toPiContext(options), {
@@ -110,15 +112,44 @@ export class PiAiAdapter extends LlmAdapter {
         ...options.temperature === undefined ? {} : { temperature: options.temperature },
         ...options.maxTokens === undefined ? {} : { maxTokens: options.maxTokens },
         ...options.sessionId === undefined ? {} : { sessionId: String(options.sessionId) },
-        signal: controller.signal,
+        signal: watchdog.signal,
         // Profile headers are deployment-owned; attribution names are
         // Harness-owned and therefore win collisions.
         headers: requestHeaders(profile.headers),
       })
-      yield* toStreamChunks(events, model.contextWindow)
+      const iterator = toStreamChunks(events, model.contextWindow)[Symbol.asyncIterator]()
+      let exhausted = false
+      try {
+        while (true) {
+          const result = await watchdog.next(iterator)
+          const timeout = timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')
+          if (timeout !== undefined) throw timeout
+          if (result.done) {
+            exhausted = true
+            return
+          }
+          yield result.value
+        }
+      } finally {
+        if (!exhausted) {
+          consumer.abort('pi-ai stream consumer stopped')
+          try {
+            await iterator.return(undefined)
+          } catch (_abortedSdkTeardown) {
+            // The stable signal already owns SDK termination; return-time abort cannot add an outcome.
+          }
+        }
+      }
+    } catch (error: unknown) {
+      if (timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT') !== undefined) {
+        throw new LlmError(`pi-ai stream idle timeout after ${streamIdleTimeoutMs}ms`, 'TIMEOUT', { cause: error })
+      }
+      if (options.signal?.aborted) {
+        throw new LlmError('pi-ai request aborted by caller', 'ABORTED', { cause: error })
+      }
+      throw error
     } finally {
-      options.signal?.removeEventListener('abort', onCallerAbort)
-      controller.abort('consumer stopped streaming')
+      consumer.abort('pi-ai stream consumer stopped')
     }
   }
 }

packages/llm/llm-pi-ai/src/config.ts

@@ -7,6 +7,10 @@
 import { getProviders } from '@earendil-works/pi-ai'
 import type { CacheRetention, ThinkingBudgets, ThinkingLevel, Transport } from '@earendil-works/pi-ai'
 import z from 'schemastery'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+
+/** Default maximum idle interval while an adapter stream read is outstanding. */
+export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 300_000
 
 /** Configuration for one pi-ai provider route. */
 export interface PiAiProviderProfile {
@@ -30,10 +34,14 @@ export interface PiAiProviderProfile {
   timeoutMs?: number
   /** WebSocket connection timeout in milliseconds. */
   websocketConnectTimeoutMs?: number
-  /** Provider SDK retry count. */
-  maxRetries?: number
-  /** Maximum provider-requested retry delay in milliseconds. */
-  maxRetryDelayMs?: number
+  /** Maximum provider idle time while one stream read is outstanding. */
+  streamIdleTimeoutMs?: number
+}
+
+/** Validated profile with every adapter-owned default resolved. */
+export interface ResolvedPiAiProviderProfile extends PiAiProviderProfile {
+  /** Positive finite provider-idle interval after defaulting. */
+  streamIdleTimeoutMs: number
 }
 
 /** Plugin configuration: the non-empty provider profiles this instance owns. */
@@ -60,8 +68,7 @@ const profile = z.object({
   transport: z.union(['sse', 'websocket', 'websocket-cached', 'auto']),
   timeoutMs: z.natural(),
   websocketConnectTimeoutMs: z.natural(),
-  maxRetries: z.natural(),
-  maxRetryDelayMs: z.natural(),
+  streamIdleTimeoutMs: z.number().min(Number.MIN_VALUE).max(MAX_TIMER_DELAY_MS).default(DEFAULT_STREAM_IDLE_TIMEOUT_MS),
 })
 
 /** Runtime schema for {@link Config}. */
@@ -75,11 +82,18 @@ export const Config: z<Config> = z.object({
  * @param profiles - configured provider profiles.
  * @returns validated profiles in configuration order.
  */
-export function resolveProfiles(profiles: readonly PiAiProviderProfile[]): PiAiProviderProfile[] {
+export function resolveProfiles(profiles: readonly PiAiProviderProfile[]): ResolvedPiAiProviderProfile[] {
   if (profiles.length === 0) throw new Error('llm-pi-ai: providers must contain at least one profile')
   const supported = new Set<string>(getProviders())
   const seen = new Set<string>()
   return profiles.map((source) => {
+    const legacy = source as PiAiProviderProfile & {
+      maxRetries?: unknown
+      maxRetryDelayMs?: unknown
+    }
+    if ('maxRetries' in legacy || 'maxRetryDelayMs' in legacy) {
+      throw new Error('llm-pi-ai: maxRetries and maxRetryDelayMs were removed; compose agent recovery with dsh-llm-retry')
+    }
     if (source.provider.length === 0) throw new Error('llm-pi-ai: provider names must be non-empty')
     if (!supported.has(source.provider)) throw new Error(`llm-pi-ai: unknown pi-ai provider "${source.provider}"`)
     if (seen.has(source.provider)) throw new Error(`llm-pi-ai: duplicate provider profile "${source.provider}"`)
@@ -89,9 +103,18 @@ export function resolveProfiles(profiles: readonly PiAiProviderProfile[]): PiAiP
     if (source.baseURL !== undefined && source.baseURL.length === 0) {
       throw new Error(`llm-pi-ai: provider "${source.provider}" has an empty baseURL`)
     }
+    const streamIdleTimeoutMs = source.streamIdleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT_MS
+    if (!Number.isFinite(streamIdleTimeoutMs)
+      || streamIdleTimeoutMs <= 0
+      || streamIdleTimeoutMs > MAX_TIMER_DELAY_MS) {
+      throw new Error(
+        `llm-pi-ai: provider "${source.provider}" streamIdleTimeoutMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`,
+      )
+    }
     seen.add(source.provider)
     return {
       ...source,
+      streamIdleTimeoutMs,
       ...source.headers === undefined ? {} : { headers: { ...source.headers } },
       ...source.thinkingBudgets === undefined ? {} : { thinkingBudgets: { ...source.thinkingBudgets } },
     }

packages/llm/llm-pi-ai/src/stream.ts

@@ -8,7 +8,7 @@
  * @module dsh-llm-pi-ai/stream
  */
 
-import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, LlmError } from '@deepseek-ai/dsh-llm'
+import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, isQuotaExceededError, LlmError, QUOTA_EXCEEDED_CODE } from '@deepseek-ai/dsh-llm'
 import type { FinishReason, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
 import { isContextOverflow } from '@earendil-works/pi-ai'
 import type { AssistantMessage, AssistantMessageEvent, Usage as PiUsage } from '@earendil-works/pi-ai'
@@ -30,9 +30,15 @@ export function mapUsage(usage: PiUsage): TokenUsage {
 
 function classifyPiAiError(message: string): string {
   if (/\b(?:401|403)\b/.test(message)) return 'AUTH'
+  if (isQuotaExceededError(message)) return QUOTA_EXCEEDED_CODE
   if (/\b429\b|rate.?limit/i.test(message)) return 'RATE_LIMIT'
   if (/\b400\b|invalid.?request/i.test(message)) return 'INVALID_REQUEST'
   if (/\b5\d\d\b/.test(message)) return 'SERVER'
+  if (/\btime(?:d)?\s*out\b|timeout/i.test(message)) return 'TIMEOUT'
+  if (/\b(?:network|connection|socket|fetch)\b|\bECONN[A-Z]+\b/i.test(message)
+    || /\b(?:other side closed|HTTP2 request did not get a response|WebSocket closed unexpectedly)\b/i.test(message)) {
+    return 'TRANSPORT'
+  }
   return 'PI_AI_ERROR'
 }
 
@@ -52,8 +58,10 @@ export function mapStopReason(message: AssistantMessage, contextWindow?: number)
   if (piAiOverflow || harnessOverflow) {
     return {
       kind: 'error',
-      message: message.errorMessage ?? `pi-ai detected context overflow for model "${message.model}"`,
-      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      failure: {
+        message: message.errorMessage ?? `pi-ai detected context overflow for model "${message.model}"`,
+        code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      },
     }
   }
 
@@ -61,10 +69,13 @@ export function mapStopReason(message: AssistantMessage, contextWindow?: number)
     case 'stop': return { kind: 'stop' }
     case 'length': return { kind: 'max-tokens' }
     case 'toolUse': return { kind: 'tool-calls' }
-    case 'aborted': return { kind: 'aborted' }
+    case 'aborted': return {
+      kind: 'aborted',
+      failure: { message: message.errorMessage ?? 'pi-ai stream aborted', code: 'ABORTED' },
+    }
     case 'error': {
       const text = message.errorMessage ?? 'pi-ai stream error'
-      return { kind: 'error', message: text, code: classifyPiAiError(text) }
+      return { kind: 'error', failure: { message: text, code: classifyPiAiError(text) } }
     }
   }
 }

packages/llm/llm-pi-ai/tests/adapter.spec.ts

@@ -6,6 +6,7 @@ import LlmService, { CONTEXT_WINDOW_EXCEEDED_CODE, LlmError, userAgent } from '@
 import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
 import { PiAiAdapter } from '@deepseek-ai/dsh-llm-pi-ai'
 import { getModels } from '@earendil-works/pi-ai'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import { resolveProfiles } from '../src/config.ts'
 import { assemble } from './assemble.ts'
 
@@ -14,6 +15,8 @@ interface MockServer {
   paths: string[]
   requests: unknown[]
   headers: IncomingMessage['headers'][]
+  readonly closedResponses: number
+  responseClosed: Promise<void>
 }
 
 const servers: Server[] = []
@@ -23,11 +26,23 @@ afterEach(async () => {
   await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve))))
 })
 
-async function mockServer(script: { status?: number; events?: string[]; body?: string; delayMs?: number }[]): Promise<MockServer> {
+async function mockServer(script: {
+  status?: number
+  events?: string[]
+  body?: string
+  delayMs?: number
+  headers?: Record<string, string>
+}[]): Promise<MockServer> {
   const paths: string[] = []
   const requests: unknown[] = []
   const headers: IncomingMessage['headers'][] = []
+  let closedResponses = 0
+  const responseClosed = Promise.withResolvers<undefined>()
   const server = createServer((request: IncomingMessage, response: ServerResponse) => {
+    response.on('close', () => {
+      closedResponses += 1
+      responseClosed.resolve(undefined)
+    })
     let body = ''
     request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') })
     request.on('end', () => {
@@ -36,7 +51,7 @@ async function mockServer(script: { status?: number; events?: string[]; body?: s
       headers.push(request.headers)
       const behavior = script.shift() ?? { status: 500, body: 'script exhausted' }
       if (behavior.status !== undefined && behavior.status !== 200) {
-        response.writeHead(behavior.status, { 'content-type': 'application/json' })
+        response.writeHead(behavior.status, { 'content-type': 'application/json', ...behavior.headers })
         response.end(behavior.body ?? '{}')
         return
       }
@@ -56,7 +71,14 @@ async function mockServer(script: { status?: number; events?: string[]; body?: s
   await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
   const address = server.address()
   if (address === null || typeof address === 'string') throw new Error('no port')
-  return { url: `http://127.0.0.1:${address.port}`, paths, requests, headers }
+  return {
+    url: `http://127.0.0.1:${address.port}`,
+    paths,
+    requests,
+    headers,
+    responseClosed: responseClosed.promise,
+    get closedResponses() { return closedResponses },
+  }
 }
 
 const textEvents = [
@@ -107,8 +129,7 @@ describe('PiAiAdapter provider routing', () => {
       transport: 'sse',
       timeoutMs: 5000,
       websocketConnectTimeoutMs: 3000,
-      maxRetries: 0,
-      maxRetryDelayMs: 10,
+      streamIdleTimeoutMs: 10_000,
       thinkingBudgets: { high: 2048 },
     })
     await assemble(ctx, {
@@ -161,13 +182,35 @@ describe('PiAiAdapter provider routing', () => {
     const ctx = new Context()
     await ctx.plugin(LlmService)
     await ctx.plugin(LlmPiAi, {
-      providers: [{ provider: 'openai', apiKey: 'test-key', baseURL: `${server.url}/v1`, maxRetries: 0 }],
+      providers: [{ provider: 'openai', apiKey: 'test-key', baseURL: `${server.url}/v1` }],
     })
     const result = await assemble(ctx, { provider: 'openai', model: 'gpt-4.1', messages: [] })
     expect(result.finish.kind).toBe('error')
     expect(server.paths).toEqual(['/v1/responses'])
   })
 
+  it('forces one wire request for an SDK-retryable provider failure', async () => {
+    const server = await mockServer([
+      {
+        status: 429,
+        headers: { 'retry-after-ms': '1' },
+        body: JSON.stringify({ error: { message: 'retryable provider failure' } }),
+      },
+      { status: 500, body: JSON.stringify({ error: { message: 'hidden SDK retry' } }) },
+      { status: 500, body: JSON.stringify({ error: { message: 'second hidden SDK retry' } }) },
+    ])
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    await ctx.plugin(LlmPiAi, {
+      providers: [{ provider: 'openai', apiKey: 'test-key', baseURL: `${server.url}/v1` }],
+    })
+
+    const result = await assemble(ctx, { provider: 'openai', model: 'gpt-4.1', messages: [] })
+
+    expect(result.finish).toMatchObject({ kind: 'error' })
+    expect(server.paths).toEqual(['/v1/responses'])
+  })
+
   it('uses OpenAI Responses against an Azure project v1 path with its API key header', async () => {
     const server = await mockServer([{ status: 401, body: JSON.stringify({ error: { message: 'expected mock failure' } }) }])
     const ctx = new Context()
@@ -178,7 +221,6 @@ describe('PiAiAdapter provider routing', () => {
         apiKey: 'test-key',
         baseURL: `${server.url}/api/projects/openai/openai/v1`,
         headers: { 'api-key': 'test-key', Authorization: '' },
-        maxRetries: 0,
       }],
     })
     const result = await assemble(ctx, { provider: 'openai', model: 'gpt-5.5', messages: [] })
@@ -195,9 +237,10 @@ describe('PiAiAdapter provider routing', () => {
     [500, 'SERVER'],
   ] as const)('maps HTTP %s failures to %s', async (status, code) => {
     const server = await mockServer([{ status, body: JSON.stringify({ error: { message: `provider ${status}` } }) }])
-    const ctx = await harness(server.url, { maxRetries: 0 })
+    const ctx = await harness(server.url)
     const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
-    expect(result.finish).toMatchObject({ kind: 'error', code })
+    expect(result.finish).toMatchObject({ kind: 'error', failure: { code } })
+    expect(server.paths).toEqual(['/chat/completions'])
   })
 
   it('uses the resolved catalog context window for usage-based overflow detection', async () => {
@@ -218,10 +261,29 @@ describe('PiAiAdapter provider routing', () => {
 
     expect(result.finish).toEqual({
       kind: 'error',
-      message: `pi-ai detected context overflow for model "${model.id}"`,
-      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      failure: {
+        message: `pi-ai detected context overflow for model "${model.id}"`,
+        code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      },
     })
   })
+
+  it('stops the SDK request when the adapter idle watchdog expires', async () => {
+    const server = await mockServer([{ events: textEvents, delayMs: 200 }])
+    const ctx = await harness(server.url, { streamIdleTimeoutMs: 20 })
+
+    await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
+      .rejects.toMatchObject({ code: 'TIMEOUT' })
+    await Promise.race([
+      server.responseClosed,
+      new Promise<never>((_resolve, reject) => {
+        setTimeout(() => { reject(new Error('SDK request did not close after idle timeout')) }, 100)
+      }),
+    ])
+
+    expect(server.paths).toEqual(['/chat/completions'])
+    expect(server.closedResponses).toBe(1)
+  })
 })
 
 describe('provider profile lifecycle', () => {
@@ -280,16 +342,31 @@ describe('provider profile lifecycle', () => {
     expect(() => resolveProfiles([{ provider: 'openai', baseURL: '' }])).toThrow(/empty baseURL/)
   })
 
-  it('rejects negative or fractional stream tunables at schema validation', () => {
+  it.each(['maxRetries', 'maxRetryDelayMs'] as const)(
+    'rejects removed profile field %s instead of silently restoring hidden SDK retries',
+    async (field) => {
+      const legacy = { provider: 'openai', [field]: 2 }
+      expect(() => resolveProfiles([legacy as never])).toThrow(/removed.*agent recovery/i)
+      const ctx = new Context()
+      await ctx.plugin(LlmService)
+      await expect(ctx.plugin(LlmPiAi, { providers: [legacy as never] }))
+        .rejects.toThrow(/removed.*agent recovery/i)
+    },
+  )
+
+  it('rejects invalid stream tunables at plugin load', async () => {
     const invalid = [
       { timeoutMs: -1 },
       { websocketConnectTimeoutMs: -1 },
-      { maxRetries: -1 },
-      { maxRetries: 0.5 },
-      { maxRetryDelayMs: -1 },
+      { streamIdleTimeoutMs: 0 },
+      { streamIdleTimeoutMs: Number.NaN },
+      { streamIdleTimeoutMs: MAX_TIMER_DELAY_MS + 1 },
     ]
     for (const entry of invalid) {
-      expect(() => new LlmPiAi.Config({ providers: [{ provider: 'openai', ...entry }] })).toThrow()
+      const ctx = new Context()
+      await ctx.plugin(LlmService)
+      await expect(ctx.plugin(LlmPiAi, { providers: [{ provider: 'openai', ...entry }] }))
+        .rejects.toThrow()
     }
   })
 
@@ -301,11 +378,59 @@ describe('provider profile lifecycle', () => {
     })()).rejects.toMatchObject({ code: 'NO_ADAPTER' })
     expect(new LlmError('x', 'X')).toBeInstanceOf(Error)
   })
+
+  it('validates direct-constructor profiles at the embedding boundary', () => {
+    expect(() => new PiAiAdapter({
+      profiles: [{ provider: 'openai', streamIdleTimeoutMs: 0 }],
+    })).toThrow(/streamIdleTimeoutMs.*positive finite/)
+    expect(() => new PiAiAdapter({
+      profiles: [{ provider: 'openai', streamIdleTimeoutMs: MAX_TIMER_DELAY_MS + 1 }],
+    })).toThrow(/streamIdleTimeoutMs.*no greater/)
+  })
 })
 
 describe('abort wiring', () => {
+  it('preserves an unknown pre-dispatch adapter Error exactly', async () => {
+    const original = new Error('SDK context conversion exploded')
+    const message = Object.defineProperty({}, 'role', {
+      get() { throw original },
+    })
+    const adapter = new PiAiAdapter({ profiles: [{ provider: 'deepseek', apiKey: 'test-key' }] })
+    const drain = async (): Promise<void> => {
+      for await (const _chunk of adapter.stream({
+        provider: 'deepseek',
+        model: 'deepseek-v4-flash',
+        messages: [message as never],
+      })) { /* drain */ }
+    }
+
+    await expect(drain()).rejects.toBe(original)
+  })
+
+  it('lets a concurrent caller abort classify a pre-dispatch adapter failure', async () => {
+    const controller = new AbortController()
+    const original = new Error('conversion lost its caller')
+    const message = Object.defineProperty({}, 'role', {
+      get() {
+        controller.abort('caller cancelled during conversion')
+        throw original
+      },
+    })
+    const adapter = new PiAiAdapter({ profiles: [{ provider: 'deepseek', apiKey: 'test-key' }] })
+    const drain = async (): Promise<void> => {
+      for await (const _chunk of adapter.stream({
+        provider: 'deepseek',
+        model: 'deepseek-v4-flash',
+        messages: [message as never],
+        signal: controller.signal,
+      })) { /* drain */ }
+    }
+
+    await expect(drain()).rejects.toMatchObject({ code: 'ABORTED', cause: original })
+  })
+
   it('resolves catalog endpoints without an override before honoring pre-abort', async () => {
-    const adapter = new PiAiAdapter({ profiles: [{ provider: 'deepseek', apiKey: 'test-key', maxRetries: 0 }] })
+    const adapter = new PiAiAdapter({ profiles: [{ provider: 'deepseek', apiKey: 'test-key' }] })
     const controller = new AbortController()
     controller.abort('already stopped')
     const chunks = []

packages/llm/llm-pi-ai/tests/convert.spec.ts

@@ -485,20 +485,32 @@ describe('toStreamChunks', () => {
     )))
     expect(chunks).toEqual([
       { type: 'usage', usage: { inputTokens: 1, outputTokens: 0 } },
-      { type: 'finish', reason: { kind: 'error', message: 'boom', code: 'PI_AI_ERROR' } },
+      { type: 'finish', reason: { kind: 'error', failure: { message: 'boom', code: 'PI_AI_ERROR' } } },
     ])
   })
 
   it('maps aborted error events to aborted finish', async () => {
     const error = assistant({ stopReason: 'aborted' })
     const chunks = await collect(toStreamChunks(feed({ type: 'error', reason: 'aborted', error })))
-    expect(chunks.at(-1)).toEqual({ type: 'finish', reason: { kind: 'aborted' } })
+    expect(chunks.at(-1)).toEqual({
+      type: 'finish',
+      reason: { kind: 'aborted', failure: { message: 'pi-ai stream aborted', code: 'ABORTED' } },
+    })
   })
 
   it('rejects a stream that ends without done or error', async () => {
     await expect(collect(toStreamChunks(feed({ type: 'start', partial: assistant() }))))
       .rejects.toThrow(/without done\/error/)
   })
+
+  it('preserves an unknown SDK iterator Error exactly', async () => {
+    const original = Object.assign(new Error('SDK transport exploded'), { code: 'ECONNRESET' })
+    async function* failedSdkStream(): AsyncGenerator<AssistantMessageEvent> {
+      throw original
+    }
+
+    await expect(collect(toStreamChunks(failedSdkStream()))).rejects.toBe(original)
+  })
 })
 
 describe('mapStopReason / mapUsage', () => {
@@ -506,46 +518,65 @@ describe('mapStopReason / mapUsage', () => {
     ['stop', { kind: 'stop' }],
     ['length', { kind: 'max-tokens' }],
     ['toolUse', { kind: 'tool-calls' }],
-    ['aborted', { kind: 'aborted' }],
+    ['aborted', { kind: 'aborted', failure: { message: 'pi-ai stream aborted', code: 'ABORTED' } }],
   ] as const)('maps %s', (stopReason, expected) => {
     expect(mapStopReason(assistant({ stopReason }))).toEqual(expected)
   })
 
   it('defaults the error message when pi-ai omits it', () => {
     expect(mapStopReason(assistant({ stopReason: 'error' })))
-      .toEqual({ kind: 'error', message: 'pi-ai stream error', code: 'PI_AI_ERROR' })
+      .toEqual({ kind: 'error', failure: { message: 'pi-ai stream error', code: 'PI_AI_ERROR' } })
   })
 
   it('maps routable HTTP-ish error messages to stable codes', () => {
     expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 401: bad key' })))
-      .toMatchObject({ kind: 'error', code: 'AUTH' })
+      .toMatchObject({ kind: 'error', failure: { code: 'AUTH' } })
     expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 429: rate limit' })))
-      .toMatchObject({ kind: 'error', code: 'RATE_LIMIT' })
+      .toMatchObject({ kind: 'error', failure: { code: 'RATE_LIMIT' } })
+    expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 429: insufficient_quota' })))
+      .toMatchObject({ kind: 'error', failure: { code: 'QUOTA' } })
+    expect(mapStopReason(assistant({
+      stopReason: 'error',
+      errorMessage: 'OpenAI API error (429): You exceeded your current quota, please check your plan and billing details.',
+    }))).toMatchObject({ kind: 'error', failure: { code: 'QUOTA' } })
     expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 500: backend down' })))
-      .toMatchObject({ kind: 'error', code: 'SERVER' })
+      .toMatchObject({ kind: 'error', failure: { code: 'SERVER' } })
+    expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'provider timed out' })))
+      .toMatchObject({ kind: 'error', failure: { code: 'TIMEOUT' } })
+    expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'ECONNRESET socket closed' })))
+      .toMatchObject({ kind: 'error', failure: { code: 'TRANSPORT' } })
     expect(mapStopReason(assistant({
       stopReason: 'error',
       errorMessage: 'HTTP 400: input exceeds the model context window limit',
-    }))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
+    }))).toMatchObject({ kind: 'error', failure: { code: CONTEXT_WINDOW_EXCEEDED_CODE } })
     expect(mapStopReason(assistant({
       stopReason: 'error',
       errorMessage: 'HTTP 400: request too large for model context',
-    }))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
+    }))).toMatchObject({ kind: 'error', failure: { code: CONTEXT_WINDOW_EXCEEDED_CODE } })
     expect(mapStopReason(assistant({
       stopReason: 'error',
       errorMessage: 'HTTP 400: invalid input: temperature exceeds maximum allowed value',
-    }))).toMatchObject({ kind: 'error', code: 'INVALID_REQUEST' })
+    }))).toMatchObject({ kind: 'error', failure: { code: 'INVALID_REQUEST' } })
+  })
+
+  it.each([
+    'other side closed',
+    'HTTP2 request did not get a response',
+    'WebSocket closed unexpectedly',
+  ])('maps pi-ai transport wording %j', (errorMessage) => {
+    expect(mapStopReason(assistant({ stopReason: 'error', errorMessage })))
+      .toMatchObject({ kind: 'error', failure: { code: 'TRANSPORT' } })
   })
 
   it('uses pi-ai provider-specific overflow classification without losing rate-limit exclusions', () => {
     expect(mapStopReason(assistant({
       stopReason: 'error',
       errorMessage: 'prompt is too long: 213462 tokens > 200000 maximum',
-    }))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
+    }))).toMatchObject({ kind: 'error', failure: { code: CONTEXT_WINDOW_EXCEEDED_CODE } })
     expect(mapStopReason(assistant({
       stopReason: 'error',
       errorMessage: 'ThrottlingException: Too many tokens, rate limit reached',
-    }))).toMatchObject({ kind: 'error', code: 'RATE_LIMIT' })
+    }))).toMatchObject({ kind: 'error', failure: { code: 'RATE_LIMIT' } })
   })
 
   it('uses the resolved context window for silent and length-stop overflows', () => {
@@ -553,15 +584,17 @@ describe('mapStopReason / mapUsage', () => {
     expect(mapStopReason(silent)).toEqual({ kind: 'stop' })
     expect(mapStopReason(silent, 100)).toEqual({
       kind: 'error',
-      message: 'pi-ai detected context overflow for model "deepseek-v4-flash"',
-      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      failure: {
+        message: 'pi-ai detected context overflow for model "deepseek-v4-flash"',
+        code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      },
     })
 
     const truncated = assistant({ stopReason: 'length', usage: usage(80, 0, 19) })
     expect(mapStopReason(truncated)).toEqual({ kind: 'max-tokens' })
     expect(mapStopReason(truncated, 100)).toMatchObject({
       kind: 'error',
-      code: CONTEXT_WINDOW_EXCEEDED_CODE,
+      failure: { code: CONTEXT_WINDOW_EXCEEDED_CODE },
     })
   })
 

packages/llm/llm-pi-ai/tests/provider-apis.e2e.ts

@@ -70,7 +70,7 @@ function textOf(result: AssembledResult): string {
 
 function expectFinish(result: AssembledResult, expected: 'stop' | 'tool-calls'): void {
   if (result.finish.kind === 'error') {
-    throw new Error(`provider request failed (${result.finish.code ?? 'unknown'}): ${result.finish.message}`)
+    throw new Error(`provider request failed (${result.finish.failure.code}): ${result.finish.failure.message}`)
   }
   expect(result.finish.kind).toBe(expected)
 }

packages/llm/llm-pi-ai/tests/sdk-options.spec.ts

@@ -0,0 +1,35 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+const streamSimple = vi.hoisted(() => vi.fn())
+
+vi.mock('@earendil-works/pi-ai', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@earendil-works/pi-ai')>()
+  return { ...actual, streamSimple }
+})
+
+import { PiAiAdapter } from '../src/adapter.ts'
+
+afterEach(() => { streamSimple.mockReset() })
+
+describe('pi-ai SDK retry boundary', () => {
+  it('pins one SDK attempt even when the installed provider currently defaults to zero retries', async () => {
+    const failure = new Error('mock SDK boundary')
+    streamSimple.mockReturnValue({
+      async * [Symbol.asyncIterator](): AsyncGenerator<never> {
+        throw failure
+      },
+    })
+    const adapter = new PiAiAdapter({ profiles: [{ provider: 'openai', apiKey: 'test-key' }] })
+    const drain = async (): Promise<void> => {
+      for await (const _chunk of adapter.stream({
+        provider: 'openai',
+        model: 'gpt-4.1',
+        messages: [],
+      })) { /* drain */ }
+    }
+
+    await expect(drain()).rejects.toBe(failure)
+    expect(streamSimple).toHaveBeenCalledOnce()
+    expect(streamSimple.mock.calls[0]?.[2]).toMatchObject({ maxRetries: 0 })
+  })
+})

packages/llm/llm-pi-ai/tsconfig.json

@@ -22,6 +22,9 @@
     },
     {
       "path": "../../support/invariants"
+    },
+    {
+      "path": "../../util/timeout"
     }
   ]
 }