docs(examples): add provider-owned batching example

examples/README.md

@@ -49,6 +49,7 @@ Focused demonstrations of specific AgentV capabilities. Each example includes it
 - [execution-metrics](features/execution-metrics/) - Metrics tracking (tokens, cost, latency)
 - [script-grader-with-llm-calls](features/script-grader-with-llm-calls/) - script graders with provider proxy for LLM calls
 - [batch-cli](features/batch-cli/) - Batch CLI evaluation
+- [provider-owned-batching](features/provider-owned-batching/) - CLI adapter-owned request batching behind normal per-case provider calls
 - [document-extraction](features/document-extraction/) - Document data extraction
 - [local-cli](features/local-cli/) - Local CLI providers
 - [compare](features/compare/) - Baseline comparison

examples/features/provider-owned-batching/README.md

@@ -0,0 +1,63 @@
+# Provider-Owned Batching
+
+This example shows the replacement pattern for throughput-oriented CLI tools:
+AgentV invokes the provider once per eval case, and the provider adapter owns
+any batching behind that normal provider boundary.
+
+The eval and provider catalog do not use `batch_requests`, `provider_batching`,
+or runner-owned grouping. `evals/suite.yaml` selects a normal provider label:
+
+```yaml
+providers:
+  - provider-owned-batch-cli
+evaluate_options:
+  max_concurrency: 3
+```
+
+`providers.yaml` then points at a single-case CLI adapter:
+
+```yaml
+- id: cli
+  label: provider-owned-batch-cli
+  command: bun run ./scripts/provider-owned-batch-adapter.ts {PROMPT_FILE} {OUTPUT_FILE} {EVAL_ID}
+  cwd: ..
+- id: mock
+  label: grader
+```
+
+The `grader` entry satisfies this repository's default grader label; the suite's
+checks are deterministic `contains` assertions.
+
+## How It Works
+
+1. AgentV starts ordinary per-case CLI provider invocations.
+2. Each adapter process appends its request to a provider-owned queue directory.
+3. The adapter uses a timeout trigger to flush all queued requests in one
+   synthetic batch.
+4. The batch writes one response file per queued request id.
+5. Each waiting adapter process returns its own response to AgentV, preserving
+   per-case response, error, and trace identity.
+
+The timeout flush is intentionally inside the adapter. A real provider could
+replace it with EOF on a long-lived process, an explicit flush command, or a
+future provider lifecycle drain hook without adding runner-level batch config.
+
+## Run
+
+From the repository root:
+
+```bash
+bun apps/cli/src/cli.ts validate examples/features/provider-owned-batching/evals/suite.yaml
+bun apps/cli/src/cli.ts eval examples/features/provider-owned-batching/evals/suite.yaml \
+  --providers examples/features/provider-owned-batching/providers.yaml
+```
+
+Focused smoke coverage for the provider-owned protocol:
+
+```bash
+bun examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts
+```
+
+The smoke test starts three adapter invocations concurrently and verifies that
+they are flushed as one provider-owned batch while still receiving three
+correlated responses.

examples/features/provider-owned-batching/evals/suite.yaml

@@ -0,0 +1,30 @@
+name: provider-owned-batching
+description: Demonstrates provider-owned batching behind ordinary per-case CLI provider invocation.
+providers:
+  - provider-owned-batch-cli
+tags:
+  - provider-owned-batching
+  - cli
+prompts:
+  - "{{ input }}"
+evaluate_options:
+  max_concurrency: 3
+tests:
+  - id: ticket-clear
+    assert:
+      - type: contains
+        value: decision=CLEAR
+    vars:
+      input: "screen support ticket: password reset succeeded"
+  - id: ticket-review
+    assert:
+      - type: contains
+        value: decision=REVIEW
+    vars:
+      input: "screen support ticket: billing dispute requires escalation"
+  - id: ticket-block
+    assert:
+      - type: contains
+        value: decision=BLOCK
+    vars:
+      input: "screen support ticket: fraud alert from blocked account"

examples/features/provider-owned-batching/providers.yaml

@@ -0,0 +1,7 @@
+- id: cli
+  label: provider-owned-batch-cli
+  command: bun run ./scripts/provider-owned-batch-adapter.ts {PROMPT_FILE} {OUTPUT_FILE} {EVAL_ID}
+  cwd: ..
+  timeout_seconds: 5
+- id: mock
+  label: grader

examples/features/provider-owned-batching/scripts/provider-owned-batch-adapter.ts

@@ -0,0 +1,163 @@
+#!/usr/bin/env bun
+import { mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises';
+import path from 'node:path';
+
+type QueuedRequest = {
+  readonly id: string;
+  readonly prompt: string;
+  readonly outputFile: string;
+  readonly queuedAt: string;
+};
+
+type BatchResponse = {
+  readonly id: string;
+  readonly text?: string;
+  readonly error?: string;
+  readonly duration_ms: number;
+};
+
+const [promptFile, outputFile, requestIdArg] = process.argv.slice(2);
+if (!promptFile || !outputFile) {
+  console.error('Usage: provider-owned-batch-adapter.ts <prompt-file> <output-file>');
+  process.exit(2);
+}
+
+const stateRoot = path.resolve(
+  process.env.PROVIDER_OWNED_BATCH_STATE_DIR ?? '.agentv/provider-owned-batching-state',
+);
+const queueDir = path.join(stateRoot, 'queue');
+const responseDir = path.join(stateRoot, 'responses');
+const batchLogPath = path.join(stateRoot, 'batches.jsonl');
+const lockDir = path.join(stateRoot, 'flush.lock');
+const flushDelayMs = Number(process.env.PROVIDER_OWNED_BATCH_FLUSH_DELAY_MS ?? 120);
+const waitTimeoutMs = Number(process.env.PROVIDER_OWNED_BATCH_WAIT_TIMEOUT_MS ?? 4_000);
+
+await mkdir(queueDir, { recursive: true });
+await mkdir(responseDir, { recursive: true });
+
+const prompt = await readFile(promptFile, 'utf8');
+const requestId = requestIdArg || process.env.AGENTV_EVAL_ID || stableRequestId(prompt, outputFile);
+const requestPath = path.join(queueDir, `${requestId}.${process.pid}.json`);
+const responsePath = path.join(responseDir, `${requestId}.json`);
+
+await writeJsonAtomic(requestPath, {
+  id: requestId,
+  prompt,
+  outputFile,
+  queuedAt: new Date().toISOString(),
+} satisfies QueuedRequest);
+
+await tryFlushBatch();
+
+const response = await waitForResponse(responsePath, waitTimeoutMs);
+await writeJsonAtomic(outputFile, response);
+
+async function tryFlushBatch(): Promise<void> {
+  try {
+    await mkdir(lockDir);
+  } catch {
+    return;
+  }
+
+  try {
+    await sleep(flushDelayMs);
+    const requestFiles = (await readdir(queueDir))
+      .filter((entry) => entry.endsWith('.json'))
+      .sort();
+    if (requestFiles.length === 0) {
+      return;
+    }
+
+    const requests: QueuedRequest[] = [];
+    for (const requestFile of requestFiles) {
+      const filePath = path.join(queueDir, requestFile);
+      try {
+        requests.push(JSON.parse(await readFile(filePath, 'utf8')) as QueuedRequest);
+      } catch {
+        // A concurrent writer has not finished its atomic rename yet. Leave it
+        // for the next provider-owned timeout flush.
+      }
+    }
+
+    const responses = runSyntheticBatch(requests);
+    const batchId = `batch-${Date.now()}-${process.pid}`;
+    await writeFile(
+      batchLogPath,
+      `${JSON.stringify({
+        batch_id: batchId,
+        trigger: 'timeout',
+        request_ids: requests.map((request) => request.id),
+        request_count: requests.length,
+      })}\n`,
+      { flag: 'a' },
+    );
+
+    for (const response of responses) {
+      await writeJsonAtomic(path.join(responseDir, `${response.id}.json`), response);
+    }
+    for (const requestFile of requestFiles) {
+      await rm(path.join(queueDir, requestFile), { force: true });
+    }
+  } finally {
+    await rm(lockDir, { recursive: true, force: true });
+  }
+}
+
+function runSyntheticBatch(requests: readonly QueuedRequest[]): readonly BatchResponse[] {
+  const startedAt = Date.now();
+  return requests.map((request) => {
+    const normalized = request.prompt.toLowerCase();
+    if (normalized.includes('fraud') || normalized.includes('blocked')) {
+      return buildResponse(request.id, 'BLOCK', startedAt);
+    }
+    if (normalized.includes('dispute') || normalized.includes('escalation')) {
+      return buildResponse(request.id, 'REVIEW', startedAt);
+    }
+    return buildResponse(request.id, 'CLEAR', startedAt);
+  });
+}
+
+function buildResponse(id: string, decision: string, startedAt: number): BatchResponse {
+  return {
+    id,
+    text: `decision=${decision}; request_id=${id}`,
+    duration_ms: Date.now() - startedAt,
+  };
+}
+
+async function waitForResponse(filePath: string, timeoutMs: number): Promise<BatchResponse> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    try {
+      return JSON.parse(await readFile(filePath, 'utf8')) as BatchResponse;
+    } catch {
+      await sleep(25);
+      await tryFlushBatch();
+    }
+  }
+  return {
+    id: requestId,
+    error: `Timed out waiting for provider-owned batch response for ${requestId}`,
+    duration_ms: timeoutMs,
+  };
+}
+
+async function writeJsonAtomic(filePath: string, value: unknown): Promise<void> {
+  await mkdir(path.dirname(filePath), { recursive: true });
+  const tempPath = `${filePath}.${process.pid}.tmp`;
+  await writeFile(tempPath, JSON.stringify(value, null, 2), 'utf8');
+  await rename(tempPath, filePath);
+}
+
+function stableRequestId(promptText: string, fallback: string): string {
+  const source = `${promptText}\n${fallback}`;
+  let hash = 0;
+  for (let index = 0; index < source.length; index += 1) {
+    hash = (hash * 31 + source.charCodeAt(index)) >>> 0;
+  }
+  return `request-${hash.toString(16)}`;
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}

examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts

@@ -0,0 +1,69 @@
+#!/usr/bin/env bun
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { spawn } from 'bun';
+
+const exampleRoot = path.resolve(import.meta.dir, '..');
+const adapter = path.join(exampleRoot, 'scripts/provider-owned-batch-adapter.ts');
+const tempRoot = await mkdtemp(path.join(os.tmpdir(), 'agentv-provider-owned-batching-'));
+const stateDir = path.join(tempRoot, 'state');
+
+const cases = [
+  ['ticket-clear', 'screen support ticket: password reset succeeded', 'decision=CLEAR'],
+  [
+    'ticket-review',
+    'screen support ticket: billing dispute requires escalation',
+    'decision=REVIEW',
+  ],
+  ['ticket-block', 'screen support ticket: fraud alert from blocked account', 'decision=BLOCK'],
+] as const;
+
+try {
+  await Promise.all(
+    cases.map(async ([id, prompt, expected]) => {
+      const promptFile = path.join(tempRoot, `${id}.prompt.txt`);
+      const outputFile = path.join(tempRoot, `${id}.output.json`);
+      await writeFile(promptFile, prompt, 'utf8');
+
+      const proc = spawn({
+        cmd: ['bun', 'run', adapter, promptFile, outputFile],
+        cwd: exampleRoot,
+        env: {
+          ...process.env,
+          AGENTV_EVAL_ID: id,
+          PROVIDER_OWNED_BATCH_STATE_DIR: stateDir,
+          PROVIDER_OWNED_BATCH_FLUSH_DELAY_MS: '150',
+        },
+        stdout: 'pipe',
+        stderr: 'pipe',
+      });
+
+      const exitCode = await proc.exited;
+      if (exitCode !== 0) {
+        const stderr = await new Response(proc.stderr).text();
+        throw new Error(`${id} exited ${exitCode}: ${stderr}`);
+      }
+
+      const output = JSON.parse(await readFile(outputFile, 'utf8')) as { text?: string };
+      if (!output.text?.includes(expected)) {
+        throw new Error(`${id} expected ${expected}, got ${JSON.stringify(output)}`);
+      }
+    }),
+  );
+
+  const batchLines = (await readFile(path.join(stateDir, 'batches.jsonl'), 'utf8'))
+    .trim()
+    .split(/\r?\n/)
+    .filter(Boolean);
+  const batches = batchLines.map((line) => JSON.parse(line) as { request_count?: number });
+  if (batches.length !== 1 || batches[0]?.request_count !== cases.length) {
+    throw new Error(
+      `Expected one batch with ${cases.length} requests, got ${batchLines.join('\\n')}`,
+    );
+  }
+
+  console.log(`provider-owned batching smoke passed: ${cases.length} requests flushed once`);
+} finally {
+  await rm(tempRoot, { recursive: true, force: true });
+}