Dato Agent: avoid Anthropic output-limit failures on large writes

dato-agent/.gitignore

@@ -2,3 +2,8 @@ dist
 node_modules
 .DS_Store
 coverage
+
+# Dato Agent live tests: secrets, spend ledger and eval output
+e2e/.env.local
+e2e/.spend.jsonl
+e2e/results/

dato-agent/e2e/README.md

@@ -0,0 +1,41 @@
+# Live tests and evals
+
+These tests run the real agent runtime against real provider APIs (Anthropic, OpenAI) and the real DatoCMS Remote MCP. Model output varies from run to run, so:
+
+- **Live tests** (`*.live.test.ts`) check that something still works. Each task is small and likely to succeed, and each test asserts on the project's end state, never on the script text.
+- **Evals** (`*.eval.ts`) compare request variants. They run each case several times and write a table of rates and costs to `e2e/results/`. They never fail on a model's choices, only on harness errors.
+
+Neither one runs in `npm test`, and neither one gates commits, pushes or releases.
+
+## Setup, once
+
+1. Put the provider test keys in the monorepo's root `.env`:
+   `ANTHROPIC_PLUGIN_TESTING_KEY=…` and `OPENAI_PLUGIN_TESTING_KEY=…`. Give each key a spend cap in the provider's console.
+2. Put the seed project in `e2e/.env.local` (git-ignored): `E2E_PROJECT_ID`, and `E2E_CMA_TOKEN` (a full-access API token of that project).
+3. Run `npm run e2e:seed` to create the schema the tests expect. It only adds what is missing.
+4. Run `npm run e2e:login`. In the browser window:
+   1. Sign in.
+   2. Pick **Only selected projects**, then the seed project.
+   3. Pick **Read and edit content**.
+   4. Click **Authorize**.
+
+   The script saves `E2E_MCP_ACCESS_TOKEN` to `e2e/.env.local`. When a test reports a 401 from the MCP, run it again.
+
+## Run
+
+```bash
+npm run test:live                               # every live test
+npm run eval -- e2e/output-limit.eval.ts        # one eval, every case, 3 repeats
+E2E_EVAL_CASES=64k-batch-nobudget E2E_EVAL_REPEATS=1 npm run eval -- e2e/output-limit.eval.ts
+```
+
+## How a run works
+
+1. `forkSeedEnvironment` forks the seed project's primary environment. Every run starts from the same schema and content, and the fork carries the seed's upgrade flags.
+2. `runTask` builds the runtime with `createAgentRuntime`, as the inspector does. It sends one message and approves every script the model proposes, until the turn ends.
+3. `installProviderTap` wraps `fetch` for provider calls. It records each response's usage and stop reason, and evals use it to rewrite requests (max_tokens, system prompt, beta features) without changing production code.
+4. The test reads the fork with the CMA client and asserts on what it finds. Then it deletes the fork.
+
+## Spend
+
+Every provider call goes into `e2e/.spend.jsonl` (git-ignored) with its estimated cost. A run refuses to start once a provider's total reaches `E2E_SPEND_CAP_USD` (default $10). Prices live in `e2e/lib/spend.ts`; add a model there before a test uses it.

dato-agent/e2e/lib/environments.ts

@@ -0,0 +1,34 @@
+import { buildClient, type Client } from '@datocms/cma-client-browser';
+
+export type ForkedEnvironment = {
+  client: Client;
+  environment: string;
+  destroy(): Promise<void>;
+};
+
+/**
+ * Forks the seed project's primary environment so a live run starts from the
+ * same schema and content every time. Forks inherit the source's opt-in
+ * upgrade flags and plugins.
+ */
+export async function forkSeedEnvironment(
+  cmaToken: string,
+  label: string,
+): Promise<ForkedEnvironment> {
+  const admin = buildClient({ apiToken: cmaToken });
+  const primary = (await admin.environments.list()).find(
+    (environment) => environment.meta.primary,
+  );
+  if (!primary) throw new Error('The seed project has no primary environment');
+  const environment = `e2e-${label}-${Date.now().toString(36)}`
+    .toLowerCase()
+    .replaceAll(/[^a-z0-9-]/g, '-');
+  await admin.environments.fork(primary.id, { id: environment });
+  return {
+    client: buildClient({ apiToken: cmaToken, environment }),
+    environment,
+    destroy: async () => {
+      await admin.environments.destroy(environment);
+    },
+  };
+}

dato-agent/e2e/lib/liveEnv.ts

@@ -0,0 +1,33 @@
+import { existsSync, readFileSync } from 'node:fs';
+import { parseEnv } from 'node:util';
+
+/** Secrets and IDs the live tests need. Every field is optional so suites can skip cleanly. */
+export type LiveEnv = {
+  projectId?: string;
+  cmaToken?: string;
+  mcpAccessToken?: string;
+  anthropicApiKey?: string;
+  openAiApiKey?: string;
+};
+
+const readEnvFile = (url: URL): Record<string, string | undefined> =>
+  existsSync(url) ? parseEnv(readFileSync(url, 'utf8')) : {};
+
+/**
+ * Reads `e2e/.env.local` (seed project and MCP token) and the monorepo's root
+ * `.env` (provider test keys). Real environment variables win over both files.
+ */
+export function loadLiveEnv(): LiveEnv {
+  const values = {
+    ...readEnvFile(new URL('../../../.env', import.meta.url)),
+    ...readEnvFile(new URL('../.env.local', import.meta.url)),
+    ...process.env,
+  };
+  return {
+    projectId: values.E2E_PROJECT_ID,
+    cmaToken: values.E2E_CMA_TOKEN,
+    mcpAccessToken: values.E2E_MCP_ACCESS_TOKEN,
+    anthropicApiKey: values.ANTHROPIC_PLUGIN_TESTING_KEY,
+    openAiApiKey: values.OPENAI_PLUGIN_TESTING_KEY,
+  };
+}

dato-agent/e2e/lib/providerTap.ts

@@ -0,0 +1,164 @@
+/**
+ * Wraps `globalThis.fetch` for provider calls made by the real SDKs inside the
+ * runtime. It records usage and stop reasons from each streamed response and can
+ * rewrite request bodies, so evals compare request variants without changing
+ * production code.
+ */
+
+export type ProviderCall = {
+  provider: 'anthropic' | 'openai';
+  model: string;
+  maxTokens?: number;
+  stopReason?: string;
+  inputTokens: number;
+  cacheReadTokens: number;
+  cacheWriteTokens: number;
+  outputTokens: number;
+};
+
+export type RequestRewrite = (
+  body: Record<string, unknown>,
+  headers: Headers,
+) => void;
+
+const providerOf = (url: string): ProviderCall['provider'] | undefined => {
+  if (url.startsWith('https://api.anthropic.com/v1/messages'))
+    return 'anthropic';
+  if (url.startsWith('https://api.openai.com/v1/responses')) return 'openai';
+  return undefined;
+};
+
+const numberAt = (value: unknown, key: string): number => {
+  const found = (value as Record<string, unknown> | undefined)?.[key];
+  return typeof found === 'number' ? found : 0;
+};
+
+/** Folds one parsed SSE `data:` payload into the call's totals. */
+function absorbEvent(call: ProviderCall, event: Record<string, unknown>): void {
+  if (call.provider === 'anthropic') {
+    const message = event.message as Record<string, unknown> | undefined;
+    const usage = (event.usage ?? message?.usage) as
+      | Record<string, unknown>
+      | undefined;
+    if (usage) {
+      call.inputTokens = Math.max(
+        call.inputTokens,
+        numberAt(usage, 'input_tokens'),
+      );
+      call.cacheReadTokens = Math.max(
+        call.cacheReadTokens,
+        numberAt(usage, 'cache_read_input_tokens'),
+      );
+      call.cacheWriteTokens = Math.max(
+        call.cacheWriteTokens,
+        numberAt(usage, 'cache_creation_input_tokens'),
+      );
+      call.outputTokens = Math.max(
+        call.outputTokens,
+        numberAt(usage, 'output_tokens'),
+      );
+    }
+    const delta = event.delta as Record<string, unknown> | undefined;
+    if (typeof delta?.stop_reason === 'string')
+      call.stopReason = delta.stop_reason;
+    return;
+  }
+  const response = event.response as Record<string, unknown> | undefined;
+  const usage = response?.usage as Record<string, unknown> | undefined;
+  if (!usage) return;
+  const cached = numberAt(usage.input_tokens_details, 'cached_tokens');
+  call.inputTokens = numberAt(usage, 'input_tokens') - cached;
+  call.cacheReadTokens = cached;
+  call.outputTokens = numberAt(usage, 'output_tokens');
+  const incomplete = response?.incomplete_details as
+    | Record<string, unknown>
+    | undefined;
+  call.stopReason =
+    typeof incomplete?.reason === 'string'
+      ? incomplete.reason
+      : String(response?.status ?? '');
+}
+
+/** Reads a copy of the SSE stream in the background; the SDK reads the other copy. */
+async function drain(stream: ReadableStream<Uint8Array>, call: ProviderCall) {
+  const decoder = new TextDecoder();
+  let buffer = '';
+  for await (const chunk of stream) {
+    buffer += decoder.decode(chunk, { stream: true });
+    const lines = buffer.split('\n');
+    buffer = lines.pop() ?? '';
+    for (const line of lines) {
+      if (!line.startsWith('data:')) continue;
+      try {
+        absorbEvent(call, JSON.parse(line.slice(5)));
+      } catch {
+        // Keep-alive or non-JSON frames.
+      }
+    }
+  }
+}
+
+/**
+ * Installs the tap and returns the calls it records plus a function that
+ * restores the original fetch.
+ */
+export function installProviderTap(rewrite?: RequestRewrite) {
+  const originalFetch = globalThis.fetch;
+  const calls: ProviderCall[] = [];
+  const pending: Promise<void>[] = [];
+
+  globalThis.fetch = async (input, init) => {
+    const request = new Request(input, init);
+    const provider = providerOf(request.url);
+    if (!provider || request.method !== 'POST')
+      return originalFetch(input, init);
+
+    const body = (await request.json()) as Record<string, unknown>;
+    const headers = new Headers(request.headers);
+    rewrite?.(body, headers);
+    const call: ProviderCall = {
+      provider,
+      model: String(body.model),
+      maxTokens:
+        typeof body.max_tokens === 'number'
+          ? body.max_tokens
+          : typeof body.max_output_tokens === 'number'
+            ? body.max_output_tokens
+            : undefined,
+      inputTokens: 0,
+      cacheReadTokens: 0,
+      cacheWriteTokens: 0,
+      outputTokens: 0,
+    };
+    calls.push(call);
+
+    const response = await originalFetch(request.url, {
+      method: 'POST',
+      headers,
+      body: JSON.stringify(body),
+      signal: request.signal,
+    }).catch((error: unknown) => {
+      // The runtime maps network failures to a generic message; keep the real cause.
+      call.stopReason = `fetch_error: ${error instanceof Error ? `${error.message} ${String((error as { cause?: unknown }).cause ?? '')}` : String(error)}`;
+      throw error;
+    });
+    if (!response.ok) call.stopReason = `http_${response.status}`;
+    if (!response.body) return response;
+    const [forSdk, forTap] = response.body.tee();
+    pending.push(drain(forTap, call).catch(() => undefined));
+    return new Response(forSdk, {
+      status: response.status,
+      statusText: response.statusText,
+      headers: response.headers,
+    });
+  };
+
+  return {
+    calls,
+    /** Waits for every tapped stream to finish, then puts the original fetch back. */
+    async restore() {
+      await Promise.all(pending);
+      globalThis.fetch = originalFetch;
+    },
+  };
+}

dato-agent/e2e/lib/runTask.ts

@@ -0,0 +1,96 @@
+import {
+  type AgentApprovalRequest,
+  type AgentRuntimeConfig,
+  type AgentTurnResult,
+  createAgentRuntime,
+} from '../../src/lib/agentRuntime';
+import {
+  installProviderTap,
+  type ProviderCall,
+  type RequestRewrite,
+} from './providerTap';
+import { assertUnderCap, recordCalls } from './spend';
+
+export type TaskRun = {
+  result: AgentTurnResult;
+  /** Every unsafe call the model asked to run, in order. All were approved. */
+  approvals: AgentApprovalRequest[];
+  calls: ProviderCall[];
+  usd: number;
+  text: string;
+};
+
+export type RunTaskOptions = {
+  label: string;
+  config: Omit<AgentRuntimeConfig, 'context' | 'navigation'> & {
+    provider: NonNullable<AgentRuntimeConfig['provider']>;
+  };
+  projectId: string;
+  environment: string;
+  message: string;
+  rewrite?: RequestRewrite;
+  /** Approval rounds before the run counts as stuck. */
+  maxApprovalRounds?: number;
+};
+
+const acknowledge = () => ({ presented: true });
+
+/**
+ * Runs one chat request the way the inspector does, approving every unsafe
+ * script the model proposes, until the turn ends. Usage goes to the spend ledger.
+ */
+export async function runTask(options: RunTaskOptions): Promise<TaskRun> {
+  assertUnderCap(
+    options.config.provider === 'anthropic' ? 'anthropic' : 'openai',
+  );
+  const tap = installProviderTap(options.rewrite);
+  const approvals: AgentApprovalRequest[] = [];
+  let text = '';
+  try {
+    // The SDKs bind fetch when the runtime builds them, so the tap goes in first.
+    const runtime = createAgentRuntime({
+      ...options.config,
+      mcpAccessLevel: 'unrestricted',
+      context: {
+        siteId: options.projectId,
+        environment: options.environment,
+        isEnvironmentPrimary: false,
+        siteName: 'claude-e2e-dato-agent',
+        surface: 'project',
+      },
+      navigation: {
+        openRecord: acknowledge,
+        showRecords: acknowledge,
+        presentRecords: acknowledge,
+        presentAssets: acknowledge,
+      },
+    });
+    let result = await runtime.runTurn({ message: options.message });
+    text += result.text;
+    for (
+      let round = 0;
+      result.status === 'approval_required' && result.responseId;
+      round++
+    ) {
+      if (round >= (options.maxApprovalRounds ?? 20)) break;
+      approvals.push(...result.approvals);
+      result = await runtime.submitApprovals({
+        responseId: result.responseId,
+        decisions: result.approvals.map((approval) => ({
+          approvalRequestId: approval.approvalRequestId,
+          approve: true,
+        })),
+      });
+      text += result.text;
+    }
+    await runtime.dispose?.();
+    return { result, approvals, calls: tap.calls, usd: 0, text };
+  } finally {
+    await tap.restore();
+  }
+}
+
+/** Records the run's provider calls in the ledger and fills in its cost. */
+export function settle(run: TaskRun, label: string): TaskRun {
+  return { ...run, usd: recordCalls(label, run.calls) };
+}