diff --git a/apps/cli/package.json b/apps/cli/package.json index f80a64ed64..37cb4138a0 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -124,7 +124,9 @@ "js-yaml": "^4.2.0" }, "devDependencies": { + "@deepseek-ai/dsh-llm-mock-server": "workspace:^", "@types/js-yaml": "^4.0.9", + "execa": "^10.0.0", "node-pty": "1.1.0" } } diff --git a/apps/cli/tests/telemetry-web.e2e.ts b/apps/cli/tests/telemetry-web.e2e.ts new file mode 100644 index 0000000000..ce58aaf2b2 --- /dev/null +++ b/apps/cli/tests/telemetry-web.e2e.ts @@ -0,0 +1,269 @@ +import { createServer, type Server } from 'node:http' +import { once } from 'node:events' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { createRequire } from 'node:module' +import { fileURLToPath } from 'node:url' +import { execa } from 'execa' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { startMockLlmServer, type MockLlmServer } from '@deepseek-ai/dsh-llm-mock-server' + +/** + * Keyless integration test for the web composition's telemetry row: boot the + * REAL `dsh web` tree (source launch) against an in-test OTLP/HTTP collector + * and a mock LLM server, drive one full turn over the /api carrier, then + * SIGINT — the shutdown drain must deliver the whole ledger plus the ops + * marker. Asserts what the collector actually received on the wire: OTLP + * JSON structure, resource identity, both instrumentation scopes, the + * session's event coverage in seq order, and the first-of-step chunk + * projection. Package-level capture/backend behavior is covered by + * session-telemetry-otel's own suites; this file pins the deployment wiring + * (cordis.yml row + env overrides) end to end. Skips when the frontend dist + * is not built (the web row fails loud without it). + */ + +const repoRoot = fileURLToPath(new URL('../../../', import.meta.url)) +const require = createRequire(new URL('../package.json', import.meta.url)) + +function frontendDistPresent(): boolean { + try { + require.resolve('@deepseek-ai/dsh-frontend/dist/index.html') + return true + } catch { + return false + } +} + +/** One decoded OTLP log record: flattened attributes plus the decoded body. */ +interface ReceivedRecord { + scope: string + severityText: string + timeUnixNano: string + attributes: Record + body: unknown +} + +/** Decode an OTLP JSON AnyValue into plain JS for readable assertions. */ +function decodeAnyValue(value: Record): unknown { + if ('stringValue' in value) return value['stringValue'] + if ('intValue' in value) return Number(value['intValue']) + if ('doubleValue' in value) return value['doubleValue'] + if ('boolValue' in value) return value['boolValue'] + if ('arrayValue' in value) { + return ((value['arrayValue'] as { values?: Record[] }).values ?? []).map(decodeAnyValue) + } + if ('kvlistValue' in value) { + const entries = (value['kvlistValue'] as { values?: { key: string; value: Record }[] }).values ?? [] + return Object.fromEntries(entries.map(entry => [entry.key, decodeAnyValue(entry.value)])) + } + return value +} + +/** In-test OTLP/HTTP logs collector: captures every POST /v1/logs payload. */ +class TestCollector { + readonly records: ReceivedRecord[] = [] + readonly badRequests: string[] = [] + private server: Server | undefined + url = '' + + async start(): Promise { + this.server = createServer((request, response) => { + const chunks: Buffer[] = [] + request.on('data', chunk => chunks.push(chunk as Buffer)) + request.on('end', () => { + const body = Buffer.concat(chunks).toString() + if (request.method !== 'POST' || request.url !== '/v1/logs' + || request.headers['content-type']?.includes('application/json') !== true) { + this.badRequests.push(`${request.method} ${request.url} ${request.headers['content-type']}`) + response.writeHead(400).end() + return + } + this.ingest(body) + response.writeHead(200, { 'content-type': 'application/json' }).end('{}') + }) + }) + this.server.listen(0, '127.0.0.1') + await once(this.server, 'listening') + const address = this.server.address() + if (address === null || typeof address === 'string') throw new Error('collector has no port') + this.url = `http://127.0.0.1:${address.port}/v1/logs` + } + + private ingest(body: string): void { + const payload = JSON.parse(body) as { + resourceLogs: { + resource: { attributes: { key: string; value: Record }[] } + scopeLogs: { + scope: { name: string } + logRecords: { + timeUnixNano?: string + severityText?: string + body?: Record + attributes?: { key: string; value: Record }[] + }[] + }[] + }[] + } + for (const resourceLog of payload.resourceLogs) { + const resource = Object.fromEntries( + resourceLog.resource.attributes.map(a => [a.key, decodeAnyValue(a.value)])) + expect(resource['service.name']).toBe('deepseek-harness') + expect(typeof resource['service.version']).toBe('string') + for (const scopeLog of resourceLog.scopeLogs) { + for (const record of scopeLog.logRecords) { + expect(record.timeUnixNano).toBeTypeOf('string') + expect(record.severityText).toBeTypeOf('string') + this.records.push({ + scope: scopeLog.scope.name, + severityText: record.severityText ?? '', + timeUnixNano: record.timeUnixNano ?? '', + attributes: Object.fromEntries((record.attributes ?? []).map(a => [a.key, decodeAnyValue(a.value)])), + body: record.body === undefined ? undefined : decodeAnyValue(record.body), + }) + } + } + } + } + + async stop(): Promise { + this.server?.close() + this.server?.closeAllConnections() + } +} + +/** Unary /api POST with the client-request envelope; unwraps the ok result. */ +async function rpc(base: string, method: string, payload: unknown): Promise { + const response = await fetch(`${base}/api/${method}`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ type: 'client-request', method, rpcId: `e2e-${method}-${Date.now()}`, payload }), + }) + const parsed = await response.json() as { result: { ok: boolean; value?: T; error?: unknown } } + if (!parsed.result.ok) throw new Error(`${method} failed: ${JSON.stringify(parsed.result.error)}`) + return parsed.result.value as T +} + +const PROMPT_TEXT = 'telemetry e2e probe: reply with one word' + +describe.skipIf(!frontendDistPresent())('web composition telemetry: OTLP collector receives the session ledger', () => { + const collector = new TestCollector() + let llm: MockLlmServer + /** Narrow structural view of the subprocess: execa's per-call generics do not unify under exactOptionalPropertyTypes. */ + let web: { + kill(signal: NodeJS.Signals): boolean + settled: Promise<{ exitCode?: number | undefined; stderr?: unknown }> + } | undefined + let webBase = '' + let dshHome = '' + + beforeAll(async () => { + await collector.start() + llm = await startMockLlmServer({ sequence: ['success'], repeatLast: true, successText: 'ok' }) + dshHome = mkdtempSync(join(tmpdir(), 'dsh-telemetry-e2e-')) + + const child = execa(process.execPath, ['--import', 'tsx/esm', 'apps/cli/src/bin.ts', 'web', '--port', '0'], { + cwd: repoRoot, + reject: false, + env: { + DSH_HOME: dshHome, + DSH_TELEMETRY_OTLP_URL: collector.url, + DSH_TELEMETRY_DISABLED: '', + DEEPSEEK_BASE_URL: llm.baseURL, + DEEPSEEK_API_KEY: 'mock-key', + }, + }) + web = { kill: signal => child.kill(signal), settled: child.then(result => result) } + // The URL line is the boot-settled signal; tsx source boot on a cold + // cache is slow, hence the generous window. + webBase = await new Promise((resolvePort, rejectPort) => { + const timer = setTimeout(() => { rejectPort(new Error('dsh web printed no URL within the boot window')) }, 150_000) + let seen = '' + child.stdout?.on('data', (chunk: Buffer) => { + seen += chunk.toString() + const match = /dsh web: (http:\/\/127\.0\.0\.1:\d+)/.exec(seen) + if (match !== null) { + clearTimeout(timer) + resolvePort(match[1] as string) + } + }) + void child.then((result) => { + clearTimeout(timer) + rejectPort(new Error(`dsh web exited before serving: ${String(result.stderr)}`)) + }) + }) + }, 180_000) + + afterAll(async () => { + // Idempotent: SIGKILL after the test's own SIGINT-exit is a no-op. + web?.kill('SIGKILL') + await web?.settled + await llm.close() + await collector.stop() + rmSync(dshHome, { recursive: true, force: true }) + }) + + it('streams the full ledger and drains the ops marker on SIGINT', async () => { + const { sessionId } = await rpc<{ sessionId: string }>(webBase, 'session.create', {}) + await rpc(webBase, 'session.prompt', { + sessionId, + mode: 'queue', + content: [{ type: 'text', text: PROMPT_TEXT }], + }) + + // Wait for the turn to finish via the RPC face (telemetry batches on its + // own 10s cadence, so the log — not the collector — is the completion signal). + const deadline = Date.now() + 60_000 + let sawTurnEnd = false + while (Date.now() < deadline && !sawTurnEnd) { + const history = await rpc<{ events: { event: { type: string } }[] }>( + webBase, 'session.history', { sessionId }) + sawTurnEnd = history.events.some(item => item.event.type === 'turn/end') + if (!sawTurnEnd) await new Promise(resolveDelay => setTimeout(resolveDelay, 500)) + } + expect(sawTurnEnd).toBe(true) + + // SIGINT → fiber dispose → coordinator emits shutdown markers → backend + // drain. Everything must reach the collector without waiting a batch tick. + web?.kill('SIGINT') + const result = await web?.settled + expect(result?.exitCode).toBe(130) + + expect(collector.badRequests).toEqual([]) + + const mine = collector.records.filter(record => record.attributes['session.id'] === sessionId) + const ledger = mine.filter(record => record.scope === '@deepseek-ai/dsh-session-telemetry-otel') + const ops = mine.filter(record => record.scope === '@deepseek-ai/dsh-session-telemetry-otel/ops') + + // Ledger coverage: the canonical turn shape arrived, each row carrying + // the identity attributes and an integer seq. + const types = ledger.map(record => record.attributes['event.type']) + for (const expected of ['turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end']) { + expect(types, expected).toContain(expected) + } + for (const record of ledger) { + expect(Number.isInteger(record.attributes['event.seq'])).toBe(true) + expect(record.severityText).toBeTruthy() + } + const seqs = ledger.map(record => record.attributes['event.seq'] as number) + expect([...seqs].sort((a, b) => a - b)).toEqual(seqs) + + // Body fidelity: the exported copy carries the event data (no redaction + // rule is mounted in this composition). + const userMessage = ledger.find(record => record.attributes['event.type'] === 'user/message') + expect(JSON.stringify(userMessage?.body)).toContain(PROMPT_TEXT) + + // Fixed chunk projection: at most the FIRST chunk of each (turn, step). + const chunkKeys = ledger + .filter(record => record.attributes['event.type'] === 'assistant/chunk') + .map((record) => { + const data = record.body as { turn: number; step: number } + return `${data.turn}:${data.step}` + }) + expect(new Set(chunkKeys).size).toBe(chunkKeys.length) + + // The drain proof: the session's clean-exit marker left the process + // before it died. + expect(ops.some(record => record.attributes['telemetry.op'] === 'shutdown')).toBe(true) + }, 120_000) +}) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 59f32ddf4b..c19bae4228 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -442,9 +442,15 @@ importers: specifier: ^4.2.0 version: 4.2.0 devDependencies: + '@deepseek-ai/dsh-llm-mock-server': + specifier: workspace:^ + version: link:../../packages/support/llm-mock-server '@types/js-yaml': specifier: ^4.0.9 version: 4.0.9 + execa: + specifier: ^10.0.0 + version: 10.0.0 node-pty: specifier: 1.1.0 version: 1.1.0(patch_hash=7a0c04f1f49d798a9ffe2f7f414c01064a44ca2489772d0c3e1235ab336755e6)