test: port PR 6229 acceptance coverage to main

apps/daemon/tests/cli-startup.test.ts

@@ -1,10 +1,13 @@
-import { chmod, mkdir, mkdtemp, rm } from 'node:fs/promises';
+import { chmod, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
+import { readFileSync } from 'node:fs';
 import http from 'node:http';
+import net from 'node:net';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { execFile, spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
 import { promisify } from 'node:util';
+import Database from 'better-sqlite3';
 import { describe, expect, it } from 'vitest';
 
 const execFileAsync = promisify(execFile);
@@ -90,6 +93,127 @@ describe('CLI startup boundaries', () => {
     }
   });
 
+  it('reconciles a durable running message after a real daemon process restart', { timeout: 60_000 }, async () => {
+    const root = await mkdtemp(join(tmpdir(), 'od-cli-daemon-restart-'));
+    const dataDir = join(root, 'data');
+    await mkdir(dataDir);
+    const port = await findFreePort();
+    const env = {
+      ...process.env,
+      OD_BIND_HOST: '127.0.0.1',
+      OD_DATA_DIR: dataDir,
+      OPEN_DESIGN_VELA_TELEMETRY: 'off',
+      OPEN_DESIGN_TELEMETRY_RELAY_URL: '',
+      LANGFUSE_PUBLIC_KEY: '',
+      LANGFUSE_SECRET_KEY: '',
+    };
+    const args = [
+      '--import', 'tsx', cliEntry,
+      'daemon', 'start', '--headless', '--port', String(port),
+    ];
+    const first = spawn(process.execPath, args, { cwd: daemonRoot, env });
+    const runId = 'run-real-process-restart';
+    const messageId = 'message-real-process-restart';
+    const conversationId = 'conversation-real-process-restart';
+    const projectId = 'project-real-process-restart';
+    const runDir = join(dataDir, 'runs', runId);
+    const statePath = join(runDir, 'state.json');
+
+    try {
+      await waitForStdoutLine(first, /\[od\] listening on (http:\/\/[^\s]+)/u);
+      const db = new Database(join(dataDir, 'app.sqlite'));
+      try {
+        const now = Date.now();
+        db.exec('PRAGMA foreign_keys = ON');
+        db.prepare(
+          `INSERT INTO projects (id, name, created_at, updated_at) VALUES (?, ?, ?, ?)`,
+        ).run(projectId, 'restart fixture', now, now);
+        db.prepare(
+          `INSERT INTO conversations (id, project_id, title, created_at, updated_at)
+           VALUES (?, ?, ?, ?, ?)`,
+        ).run(conversationId, projectId, 'restart fixture', now, now);
+        db.prepare(
+          `INSERT INTO messages
+             (id, conversation_id, role, content, run_id, run_status,
+              events_json, position, created_at, started_at)
+           VALUES (?, ?, 'assistant', '', ?, 'running', '[]', 0, ?, ?)`,
+        ).run(messageId, conversationId, runId, now, now);
+      } finally {
+        db.close();
+      }
+      await mkdir(runDir, { recursive: true });
+      await writeFile(statePath, `${JSON.stringify({
+        schemaVersion: 1,
+        id: runId,
+        projectId,
+        conversationId,
+        assistantMessageId: messageId,
+        agentId: 'claude',
+        status: 'running',
+        createdAt: Date.now() - 1_000,
+        updatedAt: Date.now(),
+        analyticsRecovery: {
+          context: {},
+          properties: { project_id: projectId, conversation_id: conversationId, run_id: runId },
+          insertId: 'restart-fixture-created',
+        },
+      })}\n`);
+
+      // SIGKILL models the process-loss case; graceful SIGTERM would run the
+      // normal shutdown path and would not exercise boot reconciliation.
+      first.kill('SIGKILL');
+      await waitForExit(first);
+
+      const second = spawn(process.execPath, args, { cwd: daemonRoot, env });
+      try {
+        const line = await waitForStdoutLine(second, /\[od\] listening on (http:\/\/[^\s]+)/u);
+        expect(line).toContain(`127.0.0.1:${port}`);
+        await waitFor(() => {
+          const state = JSON.parse(readFileSync(statePath, 'utf8')) as { status?: string };
+          const checkDb = new Database(join(dataDir, 'app.sqlite'), { readonly: true });
+          try {
+            const row = checkDb.prepare(`SELECT run_status AS status FROM messages WHERE id = ?`).get(messageId) as { status?: string } | undefined;
+            return state.status === 'failed' && row?.status === 'failed';
+          } finally {
+            checkDb.close();
+          }
+        });
+        const recoveredState = JSON.parse(await readFile(statePath, 'utf8')) as {
+          status: string;
+          errorCode?: string;
+          terminalRecoveryReason?: string;
+          analyticsRecovery?: { completedAt?: number };
+        };
+        expect(recoveredState).toMatchObject({
+          status: 'failed',
+          errorCode: 'DAEMON_RESTARTED',
+          terminalRecoveryReason: 'daemon_restart',
+          analyticsRecovery: { completedAt: expect.any(Number) },
+        });
+
+        const checkpoint = recoveredState.analyticsRecovery?.completedAt;
+        await terminateChild(second);
+        const third = spawn(process.execPath, args, { cwd: daemonRoot, env });
+        try {
+          await waitForStdoutLine(third, /\[od\] listening on (http:\/\/[^\s]+)/u);
+          await waitFor(() => {
+            const replayedState = JSON.parse(readFileSync(statePath, 'utf8')) as {
+              analyticsRecovery?: { completedAt?: number };
+            };
+            return replayedState.analyticsRecovery?.completedAt === checkpoint;
+          });
+        } finally {
+          await terminateChild(third);
+        }
+      } finally {
+        await terminateChild(second);
+      }
+    } finally {
+      await terminateChild(first);
+      await rm(root, { recursive: true, force: true });
+    }
+  });
+
   it('does not import daemon startup code for media client commands', async () => {
     const root = await mkdtemp(join(tmpdir(), 'od-cli-media-'));
     const dataDir = join(root, 'data');
@@ -445,6 +569,33 @@ function waitForStdoutLine(
   });
 }
 
+async function findFreePort(): Promise<number> {
+  const server = net.createServer();
+  await new Promise<void>((resolve, reject) => {
+    server.once('error', reject);
+    server.listen(0, '127.0.0.1', () => resolve());
+  });
+  const address = server.address();
+  const port = typeof address === 'object' && address ? address.port : 0;
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+  if (!port) throw new Error('failed to allocate a free TCP port');
+  return port;
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 10_000): Promise<void> {
+  const startedAt = Date.now();
+  while (Date.now() - startedAt < timeoutMs) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 50));
+  }
+  throw new Error('timed out waiting for daemon restart reconciliation');
+}
+
+async function waitForExit(child: ChildProcessWithoutNullStreams): Promise<void> {
+  if (child.exitCode !== null || child.signalCode !== null) return;
+  await new Promise<void>((resolve) => child.once('exit', () => resolve()));
+}
+
 async function terminateChild(child: ChildProcessWithoutNullStreams): Promise<void> {
   if (child.exitCode !== null || child.signalCode !== null) return;
   const exited = new Promise<void>((resolve) => {

apps/daemon/tests/langfuse-trace.test.ts

@@ -1006,6 +1006,44 @@ describe('buildTracePayload', () => {
     });
   });
 
+  it('keeps ACP usage on a failed run for post-run diagnostics', () => {
+    const batch = buildTracePayload(
+      makeCtx({
+        run: {
+          runId: 'run-failed-with-usage',
+          status: 'failed',
+          startedAt: 1_700_000_000_000,
+          endedAt: 1_700_000_004_500,
+          failure: {
+            failure_category: 'timeout',
+            failure_detail: 'timeout',
+            failure_stage: 'first_token_wait',
+            retryable: true,
+            user_action: 'retry',
+          },
+        },
+      }),
+    );
+    const trace = (batch[0] as any).body;
+    const generation = bodyOf(batch, 'generation-create', 'llm');
+
+    expect(trace.metadata.tokens).toMatchObject({
+      input: 1234,
+      output: 567,
+      total: 2051,
+      cacheReadInput: 200,
+      cacheCreationInput: 50,
+    });
+    expect(generation.usage).toMatchObject({
+      input: 1484,
+      output: 567,
+      total: 2051,
+      unit: 'TOKENS',
+    });
+    expect(trace.metadata.failure_category).toBe('timeout');
+    expect(trace.metadata.failure_detail).toBe('timeout');
+  });
+
   it('uses conversationId as sessionId when within length limit', () => {
     const batch = buildTracePayload(makeCtx());
     expect((batch[0] as any).body.sessionId).toBe(

apps/daemon/tests/proxy-routes.test.ts

@@ -394,6 +394,14 @@ describe('API proxy routes', () => {
       'https://token-plan-cn.xiaomimimo.com/anthropic',
       'https://token-plan-cn.xiaomimimo.com/anthropic/v1/messages',
     ],
+    [
+      'https://proxy.example.test/v1/',
+      'https://proxy.example.test/v1/messages',
+    ],
+    [
+      'https://proxy.example.test/custom/anthropic/v1/',
+      'https://proxy.example.test/custom/anthropic/v1/messages',
+    ],
   ])('routes Anthropic baseUrl %s to %s', async (input, expected) => {
     const fetchMock = vi.fn((req: FetchInput, init?: FetchInit) => {
       const url = String(req);

apps/daemon/tests/runtimes/run-terminal-reconciliation.test.ts

@@ -312,4 +312,63 @@ describe('durable run terminal reconciliation', () => {
     expect(JSON.parse(fs.readFileSync(path.join(runDir, 'state.json'), 'utf8')))
       .not.toHaveProperty('langfuseCompletedAt');
   });
+
+  it('retries an accepted telemetry delivery after a crash before checkpoint', async () => {
+    const runId = 'run-langfuse-crash-window';
+    const runDir = path.join(tmpDir, runId);
+    fs.mkdirSync(runDir, { recursive: true });
+    fs.writeFileSync(path.join(runDir, 'state.json'), JSON.stringify({
+      schemaVersion: 1,
+      id: runId,
+      projectId: 'p1',
+      conversationId: 'c1',
+      assistantMessageId: 'm1',
+      agentId: 'codex',
+      status: 'failed',
+      createdAt: 1_000,
+      updatedAt: 2_000,
+      errorCode: 'AGENT_EXIT_1',
+    }));
+    const calls: Array<Record<string, unknown>> = [];
+    let firstAttempt = true;
+    const reportLangfuse = vi.fn(async (args: Record<string, unknown>) => {
+      calls.push(args);
+      if (firstAttempt) {
+        firstAttempt = false;
+        // The upstream has accepted the request, but the daemon dies before
+        // reconcileDurableRunTerminals can persist langfuseCompletedAt.
+        throw new Error('simulated crash after telemetry acceptance');
+      }
+      return {
+        langfuse_expected: true,
+        langfuse_delivery_status: 'accepted' as const,
+      };
+    });
+    const options = {
+      analytics: { capture: vi.fn() },
+      appVersion: '0.15.1',
+      db,
+      reportLangfuse,
+      runsLogDir: tmpDir,
+    };
+
+    await expect(reconcileDurableRunTerminals(options)).rejects.toThrow(
+      'simulated crash after telemetry acceptance',
+    );
+    expect(JSON.parse(fs.readFileSync(path.join(runDir, 'state.json'), 'utf8')))
+      .not.toHaveProperty('langfuseCompletedAt');
+
+    await expect(reconcileDurableRunTerminals(options)).resolves.toMatchObject({
+      langfuseReplayed: 1,
+    });
+    expect(reportLangfuse).toHaveBeenCalledTimes(2);
+    expect(calls[1]).toMatchObject({
+      run: expect.objectContaining({ id: runId, status: 'failed' }),
+      persistedRunStatus: 'failed',
+      persistedEndedAt: 2_000,
+    });
+    expect(calls[1]?.run).toEqual(calls[0]?.run);
+    expect(JSON.parse(fs.readFileSync(path.join(runDir, 'state.json'), 'utf8')))
+      .toMatchObject({ langfuseCompletedAt: expect.any(Number) });
+  });
 });

e2e/lib/fake-agents.ts

@@ -156,6 +156,14 @@ async function emitRun(promptText) {
     emitServiceFailure(503);
     return;
   }
+  if (promptText.includes('Return a daemon model-not-found failure')) {
+    emitModelUnavailableFailure();
+    return;
+  }
+  if (promptText.includes('Return a daemon timeout failure')) {
+    emitTimeoutFailure();
+    return;
+  }
   if (promptText.includes('Return a daemon socket-drop failure')) {
     emitSocketDropFailure();
     return;
@@ -692,6 +700,24 @@ function emitServiceFailure(statusCode) {
   }
 }
 
+function emitModelUnavailableFailure() {
+  const message = 'The selected model is not available for this account: model not found.';
+  writeJson({ type: 'thread.started' });
+  writeJson({ type: 'turn.started' });
+  writeJson({ type: 'turn.failed', error: { message } });
+  process.exitCode = 0;
+  exitSoon(0);
+}
+
+function emitTimeoutFailure() {
+  const message = 'The upstream model request timed out while waiting for a response.';
+  writeJson({ type: 'thread.started' });
+  writeJson({ type: 'turn.started' });
+  writeJson({ type: 'turn.failed', error: { message } });
+  process.exitCode = 0;
+  exitSoon(0);
+}
+
 // Reproduces a connection that dropped mid-response. This shape is NOT guessed:
 // it was captured by pointing the real Claude Code CLI (2.1.168) at a fake
 // Anthropic endpoint that accepts the request, starts streaming, then destroys

e2e/tests/amr/turn.test.ts

@@ -30,7 +30,7 @@ import { describe, expect, test } from 'vitest';
 
 import { requestJson } from '@/vitest/http';
 import { listMessages } from '@/vitest/messages';
-import { startRun, waitForRunStatus } from '@/vitest/runs';
+import { readRunEvents, startRun, waitForRunStatus } from '@/vitest/runs';
 import { createSmokeSuite } from '@/vitest/suite';
 
 type ProjectResponse = {
@@ -288,6 +288,15 @@ describe('AMR chat-run end-to-end', () => {
       });
       expect(finalStatus.status).toBe('succeeded');
 
+      const runEvents = await readRunEvents(webUrl, run.runId);
+      expect(runEvents).toContain('"type":"usage"');
+      expect(runEvents).toContain('input_tokens');
+      expect(runEvents).toContain('output_tokens');
+      // This suite opts out of content telemetry. The ACP transport still
+      // persists the assistant transcript for the product, but the run event
+      // stream must not leak the user's raw prompt to telemetry consumers.
+      expect(runEvents).not.toContain(PROMPT);
+
       const messages = await listMessages(webUrl, projectId, conversationId);
       const assistantMessage = messages.find((m) => m.id === assistantMessageId);
       if (assistantMessage) {