diff --git a/docs/superpowers/plans/2026-08-29-native-research-workflow.md b/docs/superpowers/plans/2026-08-29-native-research-workflow.md new file mode 100644 index 0000000..01b0610 --- /dev/null +++ b/docs/superpowers/plans/2026-08-29-native-research-workflow.md @@ -0,0 +1,96 @@ +# Native Research Workflow Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace LangGraph/LangChain with native TypeScript orchestration and the OpenAI SDK without changing research behavior. + +**Architecture:** One native executor owns state and emits the existing `StreamEvent` union through a callback. Source runners execute through a bounded worker pool backed by `Promise.allSettled()`; `run()` uses a no-op emitter and `stream()` bridges the same executor to a small async queue. OpenAI structured responses use `responses.parse()` with the caller's Zod schema while existing Langfuse observations remain active around workflow steps. + +**Tech Stack:** TypeScript, OpenAI SDK, Zod, Vitest, Langfuse tracing, OpenTelemetry + +**Spec:** `docs/superpowers/specs/2026-08-29-native-research-workflow-design.md` + +## Global Constraints + +- Preserve `ResearchWorkflow.run()` and `ResearchWorkflow.stream()` public behavior. +- Preserve SSE payloads, abort propagation, retry/timeout rules, budgets, RRF/evidence, cache, storage, export, and UI behavior. +- Keep Langfuse manual observations, scores, masking, OTel startup, and flush. +- Remove `@langchain/langgraph`, `@langchain/core`, `@langchain/openai`, and `@langfuse/langchain` only. +- Do not add a replacement framework, event bus, agent loop, checkpoint store, or queue dependency. + +--- + +### Task 1: Native OpenAI structured adapter + +**Files:** +- Modify: `src/adapters/llm/types.ts` +- Modify: `src/adapters/llm/openai.ts` +- Modify: `src/adapters/llm/index.ts` +- Modify: `tests/helpers/mock-adapters.ts` +- Delete: `tests/unit/langchain-llm.test.ts` +- Create: `tests/unit/openai-llm.test.ts` + +**Interfaces:** +- Produces: `LLMAdapter.completeStructured(prompt, schema, options): Promise`. +- Produces: `LLMOptions.schemaName?: string` and `LLMInvocationContext` containing only `signal` and `budget`. + +- [ ] Write native-client tests proving Zod parsing, system/user input, abort forwarding, estimated-budget claim, actual usage recording, and null parsed-output rejection. +- [ ] Run `npm test -- tests/unit/openai-llm.test.ts` and confirm it fails because the adapter still expects LangChain. +- [ ] Replace the model factory with an injected minimal OpenAI client exposing `responses.parse()`; call `zodTextFormat(schema, schemaName)` and record `input_tokens`, `output_tokens`, and `total_tokens`. +- [ ] Remove unused free-text completion, model streaming, callback context, and usage-log storage from the interface, adapter, and mock. +- [ ] Run adapter, Profile, Analyst, and budget tests until green. + +### Task 2: Native concurrent workflow executor + +**Files:** +- Modify: `src/modules/workflow/state.ts` +- Modify: `src/modules/workflow/index.ts` +- Delete: `tests/unit/langgraph-runtime.test.ts` +- Create: `tests/unit/native-workflow-runtime.test.ts` +- Modify: `tests/integration/research-workflow.test.ts` + +**Interfaces:** +- Consumes: unchanged source runners and `ResearchBudget`. +- Produces: one `executeWorkflow(input, options, deps, runners, emit)` path used by both `run()` and `stream()`. + +- [ ] Add tests with controlled source promises proving overlap, `maxConcurrentSourceNodes`, early finding delivery, partial success after one rejection, abort propagation, exactly-once completion, and equivalent `run()`/`stream()` final state. +- [ ] Run the new workflow tests and confirm they fail against the graph implementation. +- [ ] Convert `ResearchWorkflowState` to a plain interface and delete `Annotation` state/reducers. +- [ ] Implement a minimal bounded mapper that submits active source jobs, collects results via `Promise.allSettled()`, and preserves the skipped LinkedIn result. +- [ ] Replace `dispatchCustomEvent()` with an injected async emitter; preserve the existing retry, timeout, query-budget, source-error, evidence, profile, diff, and analyst logic. +- [ ] Implement a local async event queue for `stream()` and a no-op emitter for `run()`; propagate executor errors once and close once. +- [ ] Run workflow unit/integration/e2e tests until green. + +### Task 3: Keep Langfuse without LangChain callbacks + +**Files:** +- Modify: `src/observability/langfuse.ts` +- Modify: `src/app/api/research/route.ts` +- Modify: `tests/unit/langfuse-observability.test.ts` +- Modify: `tests/unit/research-route-observability.test.ts` +- Modify: `tests/unit/research-cache-route.test.ts` + +**Interfaces:** +- Consumes: existing `traceResearch()`, `observeResearchStep()`, scores, masking, and flush functions. +- Produces: route calls workflow without a `callbacks` option. + +- [ ] Update tests to remove `createLangfuseCallback()` mocks/assertions while retaining observation, masking, score, failure, and flush coverage. +- [ ] Run observability and route tests and confirm they fail while callback plumbing remains. +- [ ] Delete the `CallbackHandler` import and `createLangfuseCallback()` function. +- [ ] Remove callback creation and forwarding from the research route; leave the manual root/source/Profile/Analyst observation flow unchanged. +- [ ] Run observability, cache-route, and research-route tests until green. + +### Task 4: Dependency cleanup and release verification + +**Files:** +- Modify: `package.json` +- Modify: `package-lock.json` +- Modify: any test names/descriptions that still claim LangGraph/LangChain behavior + +**Interfaces:** +- Produces: dependency tree with native `openai`, Zod, Langfuse, and OTel only. + +- [ ] Run `npm uninstall @langchain/langgraph @langchain/core @langchain/openai @langfuse/langchain`. +- [ ] Run `rg -n '@langchain|@langfuse/langchain|createLangfuseCallback|callbacks:' src tests package.json package-lock.json` and require no production matches. +- [ ] Run `npm test`, `npm run typecheck`, `npm run lint`, and `npm run build`. +- [ ] Run `git diff --check` and review the diff against every acceptance criterion in the spec. diff --git a/package-lock.json b/package-lock.json index 66889f1..e156003 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,11 +9,7 @@ "version": "0.0.2", "hasInstallScript": true, "dependencies": { - "@langchain/core": "1.2.9", - "@langchain/langgraph": "1.4.12", - "@langchain/openai": "1.5.10", "@langfuse/client": "^5.10.1", - "@langfuse/langchain": "5.10.1", "@langfuse/otel": "5.10.1", "@langfuse/tracing": "5.10.1", "@opentelemetry/sdk-node": "0.221.0", @@ -303,12 +299,6 @@ "node": ">=6.9.0" } }, - "node_modules/@cfworker/json-schema": { - "version": "4.1.1", - "resolved": "https://registry.npmjs.org/@cfworker/json-schema/-/json-schema-4.1.1.tgz", - "integrity": "sha512-gAmrUZSGtKc3AiBL71iNWxDsyUC5uMaKKGdvzYsBoTW/xi42JQHl7eKV2OYzCUqvc+D2RCcf7EXY2iCyFIk6og==", - "license": "MIT" - }, "node_modules/@ecies/ciphers": { "version": "0.2.6", "resolved": "https://registry.npmjs.org/@ecies/ciphers/-/ciphers-0.2.6.tgz", @@ -1207,145 +1197,6 @@ "url": "https://opencollective.com/js-sdsl" } }, - "node_modules/@langchain/core": { - "version": "1.2.9", - "resolved": "https://registry.npmjs.org/@langchain/core/-/core-1.2.9.tgz", - "integrity": "sha512-conzSEj9Zu1AyXJLXsSbgrtxtxinmI1yGqQ5CIJZSoV5rvv+yvQE/vgBnoySpBQ/bl3YPgj2FL/gbDjWykLSfg==", - "license": "MIT", - "dependencies": { - "@cfworker/json-schema": "^4.0.2", - "@standard-schema/spec": "^1.1.0", - "js-tiktoken": "^1.0.12", - "langsmith": ">=0.5.0 <1.0.0", - "mustache": "^4.2.0", - "p-queue": "^6.6.2", - "zod": "^3.25.76 || ^4" - }, - "engines": { - "node": ">=20" - } - }, - "node_modules/@langchain/langgraph": { - "version": "1.4.12", - "resolved": "https://registry.npmjs.org/@langchain/langgraph/-/langgraph-1.4.12.tgz", - "integrity": "sha512-63iH/igH5Fh5fHqmWp09YYWaDKKB9v4RCmYNJBrnQ224rFRbjebgyYW6o5RCczN5FZxIhQj+xT51rrNmG0zi5A==", - "license": "MIT", - "dependencies": { - "@langchain/langgraph-checkpoint": "^1.1.5", - "@langchain/langgraph-sdk": "~1.9.30", - "@langchain/protocol": "^0.0.18", - "@standard-schema/spec": "1.1.0" - }, - "engines": { - "node": ">=18" - }, - "peerDependencies": { - "@langchain/core": "^1.1.48", - "zod": "^3.25.32 || ^4.2.0" - } - }, - "node_modules/@langchain/langgraph-checkpoint": { - "version": "1.1.5", - "resolved": "https://registry.npmjs.org/@langchain/langgraph-checkpoint/-/langgraph-checkpoint-1.1.5.tgz", - "integrity": "sha512-BwDwl5VeTOh6CVuiIPgsUgfK51vTJDMSbFcSCUfjJWsl8/DPdK/mbv+ejxJstkSk/BlSPMP4JfXWcN6jD2ea2Q==", - "license": "MIT", - "engines": { - "node": ">=18" - }, - "peerDependencies": { - "@langchain/core": "^1.1.48" - } - }, - "node_modules/@langchain/langgraph-sdk": { - "version": "1.9.31", - "resolved": "https://registry.npmjs.org/@langchain/langgraph-sdk/-/langgraph-sdk-1.9.31.tgz", - "integrity": "sha512-y1sSdq39IPb6mOX43+JiSezVbUdA8EBEJ1gvn91GP0jrLG0EcSApeRDCjRouyDpPXZ51bQXEQhA8CiHM0mzcAw==", - "license": "MIT", - "dependencies": { - "@langchain/protocol": "^0.0.18", - "@types/json-schema": "^7.0.15", - "p-queue": "^9.0.1", - "p-retry": "^7.1.1" - }, - "peerDependencies": { - "@langchain/core": "^1.1.48", - "react": "^18 || ^19", - "react-dom": "^18 || ^19", - "svelte": "^4.0.0 || ^5.0.0", - "vue": "^3.0.0" - }, - "peerDependenciesMeta": { - "react": { - "optional": true - }, - "react-dom": { - "optional": true - }, - "svelte": { - "optional": true - }, - "vue": { - "optional": true - } - } - }, - "node_modules/@langchain/langgraph-sdk/node_modules/eventemitter3": { - "version": "5.0.4", - "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-5.0.4.tgz", - "integrity": "sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==", - "license": "MIT" - }, - "node_modules/@langchain/langgraph-sdk/node_modules/p-queue": { - "version": "9.3.3", - "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-9.3.3.tgz", - "integrity": "sha512-NXAOdnEe5FsZJfT4oK84lE1Y5cFFdWlRuOo5tww8DyNMxyRXwn39fIkUtNLKppcPC+UYU/bXujNCUGDv01y7CA==", - "license": "MIT", - "dependencies": { - "eventemitter3": "^5.0.4", - "p-timeout": "^7.0.0" - }, - "engines": { - "node": ">=20" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, - "node_modules/@langchain/langgraph-sdk/node_modules/p-timeout": { - "version": "7.0.1", - "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-7.0.1.tgz", - "integrity": "sha512-AxTM2wDGORHGEkPCt8yqxOTMgpfbEHqF51f/5fJCmwFC3C/zNcGT63SymH2ttOAaiIws2zVg4+izQCjrakcwHg==", - "license": "MIT", - "engines": { - "node": ">=20" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, - "node_modules/@langchain/openai": { - "version": "1.5.10", - "resolved": "https://registry.npmjs.org/@langchain/openai/-/openai-1.5.10.tgz", - "integrity": "sha512-4cxdgolkkXwnAiGEkNrue+ba7jUKjfBwleLCX5DrRVcRGrCc4w5EceblYZOIaHMY6+nhwMqIOtSzBWgcBCLfmw==", - "license": "MIT", - "dependencies": { - "js-tiktoken": "^1.0.12", - "openai": "^7.5.0", - "zod": "^3.25.76 || ^4" - }, - "engines": { - "node": ">=22" - }, - "peerDependencies": { - "@langchain/core": "^1.2.9" - } - }, - "node_modules/@langchain/protocol": { - "version": "0.0.18", - "resolved": "https://registry.npmjs.org/@langchain/protocol/-/protocol-0.0.18.tgz", - "integrity": "sha512-XW1egQtPfsGI41w2AMZNFZrUIwFSQHTjVMZs0OaTpCAvht/QLoaPN8FQcsysMVypOhupG28J29yOorrc70otBQ==", - "license": "MIT" - }, "node_modules/@langfuse/client": { "version": "5.10.1", "resolved": "https://registry.npmjs.org/@langfuse/client/-/client-5.10.1.tgz", @@ -1369,20 +1220,6 @@ "@opentelemetry/api": "^1.9.0" } }, - "node_modules/@langfuse/langchain": { - "version": "5.10.1", - "resolved": "https://registry.npmjs.org/@langfuse/langchain/-/langchain-5.10.1.tgz", - "integrity": "sha512-roKCdlyTmBVw1mT91yz3TUy+7xnvuBD1FaQqb6eR4H7/U8l40UGThP3c1wPKUOfIO57EGa9A/YwjGoc7YC2AIw==", - "license": "MIT", - "dependencies": { - "@langfuse/core": "^5.10.1", - "@langfuse/tracing": "^5.10.1" - }, - "peerDependencies": { - "@langchain/core": ">=0.3.8", - "@opentelemetry/api": "^1.9.0" - } - }, "node_modules/@langfuse/otel": { "version": "5.10.1", "resolved": "https://registry.npmjs.org/@langfuse/otel/-/otel-5.10.1.tgz", @@ -2758,6 +2595,7 @@ "version": "1.1.0", "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz", "integrity": "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==", + "dev": true, "license": "MIT" }, "node_modules/@supabase/auth-js": { @@ -3308,6 +3146,7 @@ "version": "7.0.15", "resolved": "https://registry.npmjs.org/@types/json-schema/-/json-schema-7.0.15.tgz", "integrity": "sha512-5+fP8P8MFNC+AyZCDxrB2pkZFPGzqQWUzpSeuuVLvm8VMcorNYavBqoFcxK8bQz4Qsbn4oUEEem4wDLfcysGHA==", + "dev": true, "license": "MIT" }, "node_modules/@types/json5": { @@ -6158,12 +5997,6 @@ "node": ">=0.10.0" } }, - "node_modules/eventemitter3": { - "version": "4.0.7", - "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz", - "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==", - "license": "MIT" - }, "node_modules/events": { "version": "3.3.0", "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz", @@ -7091,18 +6924,6 @@ "url": "https://github.com/sponsors/ljharb" } }, - "node_modules/is-network-error": { - "version": "1.3.2", - "resolved": "https://registry.npmjs.org/is-network-error/-/is-network-error-1.3.2.tgz", - "integrity": "sha512-PhBY86zaxNZUuWP6h13Vu5oFe0XY6/UlKzQnYFELzGVHygP3MxmvTfYSG7GN3aIab/iWudSMgjSnG9Dq+nHrgA==", - "license": "MIT", - "engines": { - "node": ">=16" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, "node_modules/is-number": { "version": "7.0.0", "resolved": "https://registry.npmjs.org/is-number/-/is-number-7.0.0.tgz", @@ -7342,15 +7163,6 @@ "url": "https://github.com/sponsors/panva" } }, - "node_modules/js-tiktoken": { - "version": "1.0.21", - "resolved": "https://registry.npmjs.org/js-tiktoken/-/js-tiktoken-1.0.21.tgz", - "integrity": "sha512-biOj/6M5qdgx5TKjDnFT1ymSpM5tbd3ylwDtrQvFQSu0Z7bBYko2dF+W/aUkXUPuk6IVpRxk/3Q2sHOzGlS36g==", - "license": "MIT", - "dependencies": { - "base64-js": "^1.5.1" - } - }, "node_modules/js-tokens": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-4.0.0.tgz", @@ -7453,39 +7265,6 @@ "json-buffer": "3.0.1" } }, - "node_modules/langsmith": { - "version": "0.9.0", - "resolved": "https://registry.npmjs.org/langsmith/-/langsmith-0.9.0.tgz", - "integrity": "sha512-tlg/aG7qezAKY6G3fgADSX7PkRj+JKoF3z7QNkCMsAOvwvuzhiwP9Amn1Z+zAIxuKoWuXQdIjtFN0LVmUC1oUQ==", - "license": "MIT", - "dependencies": { - "p-queue": "6.6.2" - }, - "peerDependencies": { - "@opentelemetry/api": "*", - "@opentelemetry/exporter-trace-otlp-proto": "*", - "@opentelemetry/sdk-trace-base": "*", - "openai": "*", - "ws": ">=7" - }, - "peerDependenciesMeta": { - "@opentelemetry/api": { - "optional": true - }, - "@opentelemetry/exporter-trace-otlp-proto": { - "optional": true - }, - "@opentelemetry/sdk-trace-base": { - "optional": true - }, - "openai": { - "optional": true - }, - "ws": { - "optional": true - } - } - }, "node_modules/language-subtag-registry": { "version": "0.3.23", "resolved": "https://registry.npmjs.org/language-subtag-registry/-/language-subtag-registry-0.3.23.tgz", @@ -8341,15 +8120,6 @@ "url": "https://github.com/sponsors/ljharb" } }, - "node_modules/p-finally": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/p-finally/-/p-finally-1.0.0.tgz", - "integrity": "sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow==", - "license": "MIT", - "engines": { - "node": ">=4" - } - }, "node_modules/p-limit": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-3.1.0.tgz", @@ -8382,49 +8152,6 @@ "url": "https://github.com/sponsors/sindresorhus" } }, - "node_modules/p-queue": { - "version": "6.6.2", - "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz", - "integrity": "sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==", - "license": "MIT", - "dependencies": { - "eventemitter3": "^4.0.4", - "p-timeout": "^3.2.0" - }, - "engines": { - "node": ">=8" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, - "node_modules/p-retry": { - "version": "7.1.1", - "resolved": "https://registry.npmjs.org/p-retry/-/p-retry-7.1.1.tgz", - "integrity": "sha512-J5ApzjyRkkf601HpEeykoiCvzHQjWxPAHhyjFcEUP2SWq0+35NKh8TLhpLw+Dkq5TZBFvUM6UigdE9hIVYTl5w==", - "license": "MIT", - "dependencies": { - "is-network-error": "^1.1.0" - }, - "engines": { - "node": ">=20" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, - "node_modules/p-timeout": { - "version": "3.2.0", - "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-3.2.0.tgz", - "integrity": "sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==", - "license": "MIT", - "dependencies": { - "p-finally": "^1.0.0" - }, - "engines": { - "node": ">=8" - } - }, "node_modules/pako": { "version": "0.2.9", "resolved": "https://registry.npmjs.org/pako/-/pako-0.2.9.tgz", diff --git a/package.json b/package.json index 2ea8938..51aeaed 100644 --- a/package.json +++ b/package.json @@ -14,11 +14,7 @@ "postinstall": "node -e \"const fs=require('fs'),p='node_modules/typescript/package.json';if(fs.existsSync(p)){const j=JSON.parse(fs.readFileSync(p,'utf8'));if(!j.bin||!j.bin.tsc){j.bin=j.bin||{};j.bin.tsc='./bin/tsc';fs.writeFileSync(p,JSON.stringify(j,null,2));const b='node_modules/typescript/bin/tsc';if(!fs.existsSync(b)){fs.writeFileSync(b,'#!/usr/bin/env node\\nrequire(\\'../../@typescript/native/bin/tsc\\');\\n',{mode:0o755});}}}\"" }, "dependencies": { - "@langchain/core": "1.2.9", - "@langchain/langgraph": "1.4.12", - "@langchain/openai": "1.5.10", "@langfuse/client": "^5.10.1", - "@langfuse/langchain": "5.10.1", "@langfuse/otel": "5.10.1", "@langfuse/tracing": "5.10.1", "@opentelemetry/sdk-node": "0.221.0", diff --git a/src/adapters/llm/openai.ts b/src/adapters/llm/openai.ts index c0e633a..4eefa55 100644 --- a/src/adapters/llm/openai.ts +++ b/src/adapters/llm/openai.ts @@ -1,166 +1,95 @@ -// ═══════════════════════════════════════════════════════ -// OpenAI LLM Adapter — LangChain Implementation -// Implements LLMAdapter using @langchain/openai and @langchain/core -// ═══════════════════════════════════════════════════════ - -import { ChatOpenAI } from "@langchain/openai"; -import { - AIMessage, - BaseMessage, - HumanMessage, - SystemMessage, -} from "@langchain/core/messages"; -import type { BaseChatModel } from "@langchain/core/language_models/chat_models"; -import type { Callbacks } from "@langchain/core/callbacks/manager"; -import { z } from "zod"; +import OpenAI from "openai"; +import { zodTextFormat } from "openai/helpers/zod"; +import type { z } from "zod"; import type { LLMAdapter, LLMOptions, LLMUsageLog } from "./types"; const DEFAULT_MODEL = "gpt-4o-mini"; +interface ParsedResponse { + output_parsed: unknown | null; + usage?: { + input_tokens: number; + output_tokens: number; + total_tokens: number; + } | null; +} + +interface OpenAIClientLike { + responses: { + parse( + body: unknown, + options?: { signal?: AbortSignal }, + ): Promise; + }; +} + export interface OpenAIAdapterOptions { - modelFactory?: (options?: LLMOptions) => BaseChatModel; + client?: OpenAIClientLike; } export class OpenAIAdapter implements LLMAdapter { - private apiKey: string; - private usageLogs: LLMUsageLog[] = []; - private modelFactory?: (options?: LLMOptions) => BaseChatModel; + private readonly client: OpenAIClientLike; constructor(apiKey: string, options?: OpenAIAdapterOptions) { - this.apiKey = apiKey; - this.modelFactory = options?.modelFactory; - } - - private getModel(options?: LLMOptions, defaultTemp = 0.3): BaseChatModel { - if (this.modelFactory) { - return this.modelFactory(options); - } - - return new ChatOpenAI({ - apiKey: this.apiKey, - modelName: options?.model ?? DEFAULT_MODEL, - temperature: options?.temperature ?? defaultTemp, - maxTokens: options?.maxTokens, - maxRetries: 2, - }); - } - - private buildMessages(prompt: string, options?: LLMOptions): BaseMessage[] { - const messages: BaseMessage[] = []; - if (options?.systemPrompt) { - messages.push(new SystemMessage(options.systemPrompt)); - } - messages.push(new HumanMessage(prompt)); - return messages; - } - - private estimateTokens(messages: BaseMessage[]): number { - let charCount = 0; - for (const msg of messages) { - charCount += typeof msg.content === "string" ? msg.content.length : 100; - } - return Math.max(10, Math.ceil(charCount / 4)); - } - - async complete(prompt: string, options?: LLMOptions): Promise { - const model = this.getModel(options, 0.3); - const messages = this.buildMessages(prompt, options); - - options?.context?.budget?.claimModelCall(this.estimateTokens(messages)); - - const response = await model.invoke(messages, { - signal: options?.context?.signal, - callbacks: options?.context?.callbacks as Callbacks, - }); - - const modelName = options?.model ?? DEFAULT_MODEL; - this.logUsage(response, modelName, options); - - return typeof response.content === "string" - ? response.content - : JSON.stringify(response.content); + this.client = options?.client ?? (new OpenAI({ apiKey }) as unknown as OpenAIClientLike); } async completeStructured( prompt: string, schema: z.ZodSchema, - options?: LLMOptions + options?: LLMOptions, ): Promise { - const model = this.getModel(options, 0.2); - const messages = this.buildMessages(prompt, options); - - options?.context?.budget?.claimModelCall(this.estimateTokens(messages)); - - const structuredModel = model.withStructuredOutput(schema, { - includeRaw: true, - }); - const result = (await structuredModel.invoke(messages, { - signal: options?.context?.signal, - callbacks: options?.context?.callbacks as Callbacks, - })) as { raw: BaseMessage; parsed: T | null }; - - const modelName = options?.model ?? DEFAULT_MODEL; - this.logUsage(result.raw, modelName, options); - if (result.parsed === null) { - throw new Error("Structured output parsing failed"); + const input = [ + ...(options?.systemPrompt + ? [{ role: "system" as const, content: options.systemPrompt }] + : []), + { role: "user" as const, content: prompt }, + ]; + const model = options?.model ?? DEFAULT_MODEL; + + options?.context?.budget?.claimModelCall(estimateTokens(input)); + + const response = await this.client.responses.parse( + { + model, + input, + temperature: options?.temperature, + max_output_tokens: options?.maxTokens, + text: { + format: zodTextFormat( + schema, + options?.schemaName ?? "structured_output", + ), + }, + }, + { signal: options?.context?.signal }, + ); + + if (response.usage) { + const usage: LLMUsageLog = { + model, + promptTokens: response.usage.input_tokens, + completionTokens: response.usage.output_tokens, + totalTokens: response.usage.total_tokens, + timestamp: new Date(), + }; + options?.context?.budget?.recordModelUsage(usage); } - return result.parsed; - } - - async *stream( - prompt: string, - options?: LLMOptions - ): AsyncGenerator { - const model = this.getModel(options, 0.3); - const messages = this.buildMessages(prompt, options); - - options?.context?.budget?.claimModelCall(this.estimateTokens(messages)); - - const stream = await model.stream(messages, { - signal: options?.context?.signal, - callbacks: options?.context?.callbacks as Callbacks, - }); - - for await (const chunk of stream) { - if (chunk.content) { - yield typeof chunk.content === "string" - ? chunk.content - : JSON.stringify(chunk.content); - } + if (response.output_parsed === null) { + throw new Error("Structured output parsing failed"); } - } - getUsageLogs(): LLMUsageLog[] { - return [...this.usageLogs]; + return response.output_parsed as T; } +} - private logUsage( - response: unknown, - model: string, - options?: LLMOptions - ): void { - if ( - response && - typeof response === "object" && - "usage_metadata" in response && - response.usage_metadata - ) { - const usage = (response as AIMessage).usage_metadata; - if (usage) { - const log: LLMUsageLog = { - model, - promptTokens: usage.input_tokens, - completionTokens: usage.output_tokens, - totalTokens: usage.total_tokens, - timestamp: new Date(), - }; - this.usageLogs.push(log); - options?.context?.budget?.recordModelUsage(log); - console.log( - `[LLM] ${model}: ${log.promptTokens}+${log.completionTokens}=${log.totalTokens} tokens` - ); - } - } - } +function estimateTokens( + input: ReadonlyArray<{ content: string }>, +): number { + const characterCount = input.reduce( + (total, message) => total + message.content.length, + 0, + ); + return Math.max(10, Math.ceil(characterCount / 4)); } diff --git a/src/adapters/llm/types.ts b/src/adapters/llm/types.ts index e175fae..76f9f98 100644 --- a/src/adapters/llm/types.ts +++ b/src/adapters/llm/types.ts @@ -11,7 +11,6 @@ export interface LLMBudget { export interface LLMInvocationContext { signal?: AbortSignal; - callbacks?: readonly unknown[]; budget?: LLMBudget; } @@ -21,6 +20,7 @@ export interface LLMOptions { maxTokens?: number; systemPrompt?: string; context?: LLMInvocationContext; + schemaName?: string; } export interface LLMUsageLog { @@ -32,16 +32,9 @@ export interface LLMUsageLog { } export interface LLMAdapter { - complete(prompt: string, options?: LLMOptions): Promise; completeStructured( prompt: string, schema: z.ZodSchema, options?: LLMOptions ): Promise; - stream( - prompt: string, - options?: LLMOptions - ): AsyncGenerator; - getUsageLogs?(): LLMUsageLog[]; } - diff --git a/src/app/api/research/route.ts b/src/app/api/research/route.ts index 110805b..05f4c07 100644 --- a/src/app/api/research/route.ts +++ b/src/app/api/research/route.ts @@ -39,7 +39,6 @@ import { createAnalystModule } from "@/modules/analyst"; import { createResearchWorkflow } from "@/modules/workflow"; import type { ResearchWorkflowState } from "@/modules/workflow/state"; import { - createLangfuseCallback, emitResearchScores, flushLangfuse, traceResearch, @@ -445,8 +444,6 @@ async function executeLiveWorkflow({ cacheMatchedBy: "none", cacheAction: existingProfile ? "refresh" : "auto", }; - const langfuseCallback = createLangfuseCallback(traceContext); - let finalState: ResearchWorkflowState | null = null; await traceResearch(traceContext, async (traceId) => { @@ -456,7 +453,6 @@ async function executeLiveWorkflow({ companyId, existingProfile, signal: controller.signal, - callbacks: langfuseCallback ? [langfuseCallback] : undefined, onComplete: async (state) => { finalState = state; updateResearchTraceOutcome(state); diff --git a/src/modules/research/index.ts b/src/modules/research/index.ts index 0d5250e..59e1c05 100644 --- a/src/modules/research/index.ts +++ b/src/modules/research/index.ts @@ -1,5 +1,5 @@ // ═══════════════════════════════════════════════════════ -// Research source runners used by the LangGraph workflow. +// Research source runners used by the native workflow. // ═══════════════════════════════════════════════════════ import type { diff --git a/src/modules/workflow/index.ts b/src/modules/workflow/index.ts index 552bcbb..e488982 100644 --- a/src/modules/workflow/index.ts +++ b/src/modules/workflow/index.ts @@ -1,12 +1,3 @@ -// ═══════════════════════════════════════════════════════ -// PartnerIQ Research Workflow (LangGraph StateGraph) -// Bounded parallel execution: 5 static source nodes fan-out, -// fan-in to deterministic evidence preparation, downstream profile/diff/analyst. -// ═══════════════════════════════════════════════════════ - -import { END, START, StateGraph } from "@langchain/langgraph"; -import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch"; -import type { Callbacks } from "@langchain/core/callbacks/manager"; import type { CompanyInput, CompanyProfile, @@ -28,30 +19,43 @@ import { ResearchQueryBudgetExceededError, type ResearchBudget, } from "@/modules/research/budget"; -import { createResearchSourceRunners, type ResearchSourceRunner } from "@/modules/research"; +import { + createResearchSourceRunners, + type ResearchSourceRunner, +} from "@/modules/research"; +import type { CrawlPolicy } from "@/modules/research/crawl-policy"; import { observeResearchStep, updateResearchObservationOutcome, } from "@/observability/langfuse"; -import { - ResearchWorkflowAnnotation, - type ResearchWorkflowState, -} from "./state"; - -const SSE_EVENT_NAME = "sse_event"; +import type { ResearchWorkflowState } from "./state"; + +const SOURCE_NAMES: SourceName[] = [ + "web_search", + "website", + "news", + "registry", + "linkedin", +]; +const SOURCE_EXECUTION_ORDER: SourceName[] = [ + "web_search", + "news", + "website", + "registry", + "linkedin", +]; + +type EventEmitter = (event: StreamEvent) => void | Promise; export interface ResearchWorkflowOptions { researchRunId: string; companyId?: string; existingProfile?: CompanyProfile | null; signal?: AbortSignal; - callbacks?: readonly unknown[]; sessionId?: string; onComplete?: (state: ResearchWorkflowState) => void | Promise; } -import type { CrawlPolicy } from "@/modules/research/crawl-policy"; - export interface ResearchWorkflowDeps { search: SearchAdapter; scraper: ScraperAdapter; @@ -65,15 +69,17 @@ export interface ResearchWorkflowDeps { export interface ResearchWorkflow { stream( input: CompanyInput, - options: ResearchWorkflowOptions + options: ResearchWorkflowOptions, ): AsyncGenerator; run( input: CompanyInput, - options: ResearchWorkflowOptions + options: ResearchWorkflowOptions, ): Promise; } -export function createResearchWorkflow(deps: ResearchWorkflowDeps): ResearchWorkflow { +export function createResearchWorkflow( + deps: ResearchWorkflowDeps, +): ResearchWorkflow { const runners = createResearchSourceRunners({ search: deps.search, scraper: deps.scraper, @@ -82,102 +88,218 @@ export function createResearchWorkflow(deps: ResearchWorkflowDeps): ResearchWork crawlPolicy: deps.crawlPolicy, }); - return { - async run(input, options) { - return await executeGraph(input, options, deps, runners); - }, + run: (input, options) => + executeWorkflow(input, options, deps, runners, () => undefined), async *stream(input, options) { - const activeSources: SourceName[] = ["web_search", "website", "news", "registry"]; - if (input.linkedinUrl) { - activeSources.push("linkedin"); - } - + const activeSources = SOURCE_NAMES.filter( + (source) => source !== "linkedin" || Boolean(input.linkedinUrl), + ); yield { event: "research:start", data: { sources: activeSources }, - } as StreamEvent; - - const app = compileResearchGraph(deps, runners, options); - - const eventStream = app.streamEvents( - createInitialState(input, options), - { - version: "v2", - signal: options.signal, - callbacks: options.callbacks as Callbacks, - maxConcurrency: deps.guards.maxConcurrentSourceNodes, - } - ); - - let fatalErrorEncountered: string | null = null; - let hasFindings = false; - let finalState: ResearchWorkflowState | null = null; - - for await (const event of eventStream) { - if (event.event === "on_chain_end") { - const output = (event.data as { output?: unknown }).output; - if (isResearchWorkflowState(output)) { - finalState = output; - } - } - if (event.event === "on_custom_event" && event.name === SSE_EVENT_NAME) { - const sse = event.data as StreamEvent; - if (sse.event === "research:finding") { - hasFindings = true; - } - if (sse.event === "error" && !sse.data.source) { - fatalErrorEncountered = sse.data.message; - } - yield sse; - } - } + }; - if (finalState) { - await options.onComplete?.(finalState); - } + const queue = new AsyncEventQueue(); + const execution = executeWorkflow( + input, + options, + deps, + runners, + (event) => queue.push(event), + ); + void execution.then( + () => queue.close(), + (error) => queue.fail(error), + ); - if (!hasFindings && !fatalErrorEncountered) { - yield { - event: "error", - data: { message: "Không tìm thấy thông tin nào về công ty này." }, - } as StreamEvent; + for await (const event of queue) { + yield event; } + await execution; }, }; } -async function executeGraph( +async function executeWorkflow( input: CompanyInput, options: ResearchWorkflowOptions, - deps: ResearchWorkflowDeps, - runners: Record -): Promise { - const app = compileResearchGraph(deps, runners, options); - - return (await app.invoke( - createInitialState(input, options), - { - signal: options.signal, - callbacks: options.callbacks as Callbacks, - maxConcurrency: deps.guards.maxConcurrentSourceNodes, - } - )) as ResearchWorkflowState; -} - -function compileResearchGraph( deps: ResearchWorkflowDeps, runners: Record, - options: ResearchWorkflowOptions, -) { + emit: EventEmitter, +): Promise { const budget = createResearchBudget({ maxLLMCalls: deps.guards.maxLLMCallsPerResearch, maxTokens: deps.guards.maxTokensPerResearch, maxQueries: deps.guards.maxQueriesPerResearch, maxConcurrentProviderCalls: deps.guards.maxConcurrentProviderCalls, }); - return buildGraph(deps, runners, budget, options).compile(); + const state = createInitialState(input, options); + const llmContext: LLMInvocationContext = { + signal: options.signal, + budget, + }; + + const sourceTasks = SOURCE_EXECUTION_ORDER.map((source) => async () => { + if (source === "linkedin" && !input.linkedinUrl) { + return skippedSource(source); + } + + return observeResearchStep(`source.${source}`, async () => { + const result = await executeSourceRunner( + source, + runners[source], + input, + budget, + deps.guards, + emit, + options.signal, + ); + if (result.status === "failed") { + updateResearchObservationOutcome("failed"); + } + return result; + }); + }); + + const settledSources = await settleWithConcurrency( + sourceTasks, + deps.guards.maxConcurrentSourceNodes, + ); + state.sourceResults = settledSources.map((result, index) => + result.status === "fulfilled" + ? result.value + : failedSource(SOURCE_EXECUTION_ORDER[index], result.reason), + ).sort( + (left, right) => SOURCE_NAMES.indexOf(left.source) - SOURCE_NAMES.indexOf(right.source), + ); + + throwIfAborted(options.signal); + + await observeResearchStep("evidence.prepare", async () => { + const prepared = prepareEvidence(state.sourceResults); + state.findings = prepared.findings; + state.outcome = prepared.outcome; + + if (state.findings.length === 0) { + const errorDetails = state.sourceResults + .filter((result) => result.error) + .map((result) => `${result.source}: ${result.error?.message}`); + const details = errorDetails.length > 0 + ? ` Chi tiết: ${errorDetails.join(" | ")}` + : ""; + state.fatalError = `Không tìm thấy thông tin nào về công ty này.${details}`; + state.outcome = "failed"; + updateResearchObservationOutcome("failed"); + await emit({ + event: "error", + data: { message: state.fatalError }, + }); + } + }); + + if (!state.fatalError) { + await buildProfile(state, options, deps, llmContext, emit); + } + if (!state.fatalError && state.profile) { + await buildDiff(state, deps, emit); + } + if (!state.fatalError && state.profile) { + await analyzeProfile(state, deps, llmContext, emit); + } + + throwIfAborted(options.signal); + await options.onComplete?.(state); + return state; +} + +async function buildProfile( + state: ResearchWorkflowState, + options: ResearchWorkflowOptions, + deps: ResearchWorkflowDeps, + llmContext: LLMInvocationContext, + emit: EventEmitter, +): Promise { + await observeResearchStep("profile.build", async () => { + await emit({ + event: "profile:building", + data: { message: "Đang tổng hợp hồ sơ công ty..." }, + }); + const targetCompanyId = + options.companyId || state.existingProfile?.id || options.researchRunId; + + try { + state.profile = await deps.profile.buildProfile( + state.findings, + state.input, + targetCompanyId, + state.existingProfile?.version, + llmContext, + ); + } catch (error) { + state.fatalError = error instanceof Error + ? error.message + : "Failed to build profile"; + state.outcome = "failed"; + updateResearchObservationOutcome("failed"); + await emit({ event: "error", data: { message: state.fatalError } }); + } + }); +} + +async function buildDiff( + state: ResearchWorkflowState, + deps: ResearchWorkflowDeps, + emit: EventEmitter, +): Promise { + await observeResearchStep("profile.diff", async () => { + if (!state.existingProfile || !state.profile) { + state.diff = null; + return; + } + + try { + state.diff = deps.profile.diffProfiles(state.profile, state.existingProfile); + } catch (error) { + state.fatalError = error instanceof Error + ? error.message + : "Failed to build profile diff"; + state.outcome = "failed"; + updateResearchObservationOutcome("failed"); + await emit({ event: "error", data: { message: state.fatalError } }); + } + }); +} + +async function analyzeProfile( + state: ResearchWorkflowState, + deps: ResearchWorkflowDeps, + llmContext: LLMInvocationContext, + emit: EventEmitter, +): Promise { + await observeResearchStep("analyst.analyze", async () => { + if (!state.profile) return; + + try { + state.report = await deps.analyst.analyze( + state.profile, + { previousProfile: state.existingProfile ?? undefined }, + llmContext, + ); + } catch (error) { + state.outcome = "partial"; + updateResearchObservationOutcome("partial"); + await emit({ + event: "error", + data: { + message: error instanceof Error + ? error.message + : "Không thể phân tích hồ sơ.", + }, + }); + } + }); } function createInitialState( @@ -198,255 +320,55 @@ function createInitialState( }; } -function buildGraph( - deps: ResearchWorkflowDeps, - runners: Record, - budget: ResearchBudget, - options: ResearchWorkflowOptions, -) { - const { signal } = options; - const llmContext: LLMInvocationContext = { - signal, - callbacks: options.callbacks, - budget, - }; - // Source Nodes - const createSourceNode = (source: SourceName) => { - return async (state: typeof ResearchWorkflowAnnotation.State) => - observeResearchStep(`source.${source}`, async () => { - if (source === "linkedin" && !state.input.linkedinUrl) { - return { - sourceResults: [ - { - source: "linkedin" as SourceName, - status: "skipped" as const, - findings: [], - attempts: 0, - durationMs: 0, - }, - ], - }; - } - - const runner = runners[source]; - const result = await executeSourceRunner( - source, - runner, - state.input, - budget, - deps.guards, - signal, - ); - if (result.status === "failed") { - updateResearchObservationOutcome("failed"); - } - return { - sourceResults: [result], - }; - }); - }; - - const tracedNode = ( - name: string, - node: (state: typeof ResearchWorkflowAnnotation.State) => Promise, - ) => - (state: typeof ResearchWorkflowAnnotation.State) => - observeResearchStep(name, async () => { - const result = await node(state); - if (result && typeof result === "object") { - const update = result as { fatalError?: unknown; outcome?: unknown }; - if (update.fatalError || update.outcome === "failed") { - updateResearchObservationOutcome("failed"); - } else if (update.outcome === "partial") { - updateResearchObservationOutcome("partial"); - } - } - return result; - }); - - return new StateGraph(ResearchWorkflowAnnotation) - .addNode("web_search", createSourceNode("web_search")) - .addNode("website", createSourceNode("website")) - .addNode("news", createSourceNode("news")) - .addNode("registry", createSourceNode("registry")) - .addNode("linkedin", createSourceNode("linkedin")) - .addNode("prepare_evidence", tracedNode("evidence.prepare", async (state) => { - const prepared = prepareEvidence(state.sourceResults); - if (prepared.findings.length === 0) { - const errorDetails = state.sourceResults - .filter((r) => r.error) - .map((r) => `${r.source}: ${r.error?.message}`); - const detailStr = - errorDetails.length > 0 ? ` Chi tiết: ${errorDetails.join(" | ")}` : ""; - const message = `Không tìm thấy thông tin nào về công ty này.${detailStr}`; - - await dispatchCustomEvent(SSE_EVENT_NAME, { - event: "error", - data: { message }, - } as StreamEvent); - return { - findings: [], - outcome: "failed" as const, - fatalError: message, - }; - } - - return { - findings: prepared.findings, - outcome: prepared.outcome, - }; - })) - .addNode("build_profile", tracedNode("profile.build", async (state) => { - if (state.fatalError || state.findings.length === 0) return {}; - - await dispatchCustomEvent(SSE_EVENT_NAME, { - event: "profile:building", - data: { message: "Đang tổng hợp hồ sơ công ty..." }, - } as StreamEvent); - - const targetCompanyId = - options.companyId || state.existingProfile?.id || options.researchRunId; - try { - const profile = await deps.profile.buildProfile( - state.findings, - state.input, - targetCompanyId, - state.existingProfile?.version, - llmContext, - ); - - return { profile }; - } catch (err) { - const message = err instanceof Error ? err.message : "Failed to build profile"; - await dispatchCustomEvent(SSE_EVENT_NAME, { - event: "error", - data: { message }, - } as StreamEvent); - return { fatalError: message, outcome: "failed" as const }; - } - })) - .addNode("build_diff", tracedNode("profile.diff", async (state) => { - if (state.fatalError || !state.profile) return {}; - - if (state.existingProfile) { - try { - const diff = deps.profile.diffProfiles(state.profile, state.existingProfile); - return { diff }; - } catch (err) { - const message = - err instanceof Error ? err.message : "Failed to build profile diff"; - await dispatchCustomEvent(SSE_EVENT_NAME, { - event: "error", - data: { message }, - } as StreamEvent); - return { fatalError: message, outcome: "failed" as const }; - } - } else { - return { diff: null }; - } - })) - .addNode("analyze", tracedNode("analyst.analyze", async (state) => { - if (state.fatalError || !state.profile) return {}; - - try { - const report = await deps.analyst.analyze( - state.profile, - { previousProfile: state.existingProfile ?? undefined }, - llmContext, - ); - - return { report }; - } catch (err) { - const message = err instanceof Error ? err.message : "Không thể phân tích hồ sơ."; - await dispatchCustomEvent(SSE_EVENT_NAME, { - event: "error", - data: { message }, - } as StreamEvent); - return { outcome: "partial" as const }; - } - })) - .addEdge(START, "web_search") - .addEdge(START, "website") - .addEdge(START, "news") - .addEdge(START, "registry") - .addEdge(START, "linkedin") - .addEdge("web_search", "prepare_evidence") - .addEdge("website", "prepare_evidence") - .addEdge("news", "prepare_evidence") - .addEdge("registry", "prepare_evidence") - .addEdge("linkedin", "prepare_evidence") - .addEdge("prepare_evidence", "build_profile") - .addEdge("build_profile", "build_diff") - .addEdge("build_diff", "analyze") - .addEdge("analyze", END); -} - async function executeSourceRunner( source: SourceName, runner: ResearchSourceRunner, input: CompanyInput, budget: ResearchBudget, guards: ResourceGuards, - signal?: AbortSignal + emit: EventEmitter, + signal?: AbortSignal, ): Promise { const startTime = Date.now(); + if (signal?.aborted) return failedSource(source, signal.reason, 1, 0); - if (signal?.aborted) { - return { - source, - status: "failed", - findings: [], - error: { - source, - type: "network_error", - message: "Execution aborted", - retryable: false, - }, - attempts: 1, - durationMs: 0, - }; - } - - await dispatchCustomEvent(SSE_EVENT_NAME, { + await emit({ event: "research:progress", data: { source, status: "started" }, - } as StreamEvent); + }); let attempts = 0; const maxRetries = guards.maxRetriesPerSource ?? 2; let lastError: SourceError | undefined; while (attempts <= maxRetries) { - attempts++; + attempts += 1; const timeoutSignal = AbortSignal.timeout(guards.sourceTimeoutMs); const attemptSignal = signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal; - try { - if (signal?.aborted) { - throw new Error("Execution aborted"); - } + try { + throwIfAborted(signal); const findings = await runWithAbortSignal( runner(input, { budget, signal: attemptSignal }), attemptSignal, ); for (const finding of findings) { - await dispatchCustomEvent(SSE_EVENT_NAME, { + await emit({ event: "research:finding", data: { source: finding.source, summary: finding.content.slice(0, 200), url: finding.url, }, - } as StreamEvent); + }); } - - await dispatchCustomEvent(SSE_EVENT_NAME, { + await emit({ event: "research:progress", data: { source, status: "done" }, - } as StreamEvent); + }); return { source, @@ -455,48 +377,38 @@ async function executeSourceRunner( attempts, durationMs: Date.now() - startTime, }; - } catch (err) { - if (err instanceof ResearchQueryBudgetExceededError) { - await dispatchCustomEvent(SSE_EVENT_NAME, { + } catch (error) { + if (error instanceof ResearchQueryBudgetExceededError) { + await emit({ event: "research:progress", data: { source, status: "done" }, - } as StreamEvent); - return { - source, - status: "skipped", - findings: [], - attempts, - durationMs: Date.now() - startTime, - }; + }); + return skippedSource(source, attempts, Date.now() - startTime); } - const message = err instanceof Error ? err.message : String(err); - const isTimeout = timeoutSignal.aborted || message.includes("timed out"); + + const message = error instanceof Error ? error.message : String(error); + const isTimeout = timeoutSignal.aborted && !signal?.aborted; const retryable = - isRetryableSourceError(err, isTimeout, signal) && + isRetryableSourceError(error, isTimeout, signal) && attempts <= maxRetries; - lastError = { source, type: isTimeout ? "timeout" : "network_error", message, retryable, }; - - if (!retryable || attempts > maxRetries) { - break; - } + if (!retryable || attempts > maxRetries) break; } } - await dispatchCustomEvent(SSE_EVENT_NAME, { + await emit({ event: "error", data: { message: lastError?.message ?? "Source execution failed", source }, - } as StreamEvent); - - await dispatchCustomEvent(SSE_EVENT_NAME, { + }); + await emit({ event: "research:progress", data: { source, status: "failed" }, - } as StreamEvent); + }); return { source, @@ -508,6 +420,66 @@ async function executeSourceRunner( }; } +export async function settleWithConcurrency( + tasks: ReadonlyArray<() => Promise>, + concurrency: number, +): Promise[]> { + if (tasks.length === 0) return []; + + const results = new Array>(tasks.length); + let nextIndex = 0; + const workerCount = Math.min(tasks.length, Math.max(1, concurrency)); + const workers = Array.from({ length: workerCount }, async () => { + while (nextIndex < tasks.length) { + const index = nextIndex; + nextIndex += 1; + try { + results[index] = { status: "fulfilled", value: await tasks[index]() }; + } catch (reason) { + results[index] = { status: "rejected", reason }; + } + } + }); + + await Promise.allSettled(workers); + return results; +} + +function skippedSource( + source: SourceName, + attempts = 0, + durationMs = 0, +): SourceExecutionResult { + return { + source, + status: "skipped", + findings: [], + attempts, + durationMs, + }; +} + +function failedSource( + source: SourceName, + error: unknown, + attempts = 1, + durationMs = 0, +): SourceExecutionResult { + return { + source, + status: "failed", + findings: [], + error: { + source, + type: "network_error", + message: error instanceof Error ? error.message : "Execution aborted", + retryable: false, + }, + attempts, + durationMs, + }; +} + function isRetryableSourceError( error: unknown, isTimeout: boolean, @@ -535,7 +507,9 @@ function runWithAbortSignal( signal: AbortSignal, ): Promise { return new Promise((resolve, reject) => { - const onAbort = () => reject(signal.reason); + const onAbort = () => reject( + signal.reason ?? new DOMException("Execution aborted", "AbortError"), + ); if (signal.aborted) { onAbort(); return; @@ -555,13 +529,54 @@ function runWithAbortSignal( }); } -function isResearchWorkflowState(value: unknown): value is ResearchWorkflowState { - if (!value || typeof value !== "object") return false; - const state = value as Partial; - return ( - typeof state.researchRunId === "string" && - Array.isArray(state.sourceResults) && - Array.isArray(state.findings) && - typeof state.outcome === "string" - ); +function throwIfAborted(signal?: AbortSignal): void { + if (!signal?.aborted) return; + throw signal.reason ?? new DOMException("Execution aborted", "AbortError"); +} + +type QueueItem = + | { type: "value"; value: T } + | { type: "done" } + | { type: "error"; error: unknown }; + +class AsyncEventQueue implements AsyncIterableIterator { + private readonly items: QueueItem[] = []; + private readonly waiters: Array<(item: QueueItem) => void> = []; + private closed = false; + + push(value: T): void { + if (this.closed) return; + this.enqueue({ type: "value", value }); + } + + close(): void { + if (this.closed) return; + this.closed = true; + this.enqueue({ type: "done" }); + } + + fail(error: unknown): void { + if (this.closed) return; + this.closed = true; + this.enqueue({ type: "error", error }); + } + + async next(): Promise> { + const item = this.items.shift() ?? await new Promise>( + (resolve) => this.waiters.push(resolve), + ); + if (item.type === "error") throw item.error; + if (item.type === "done") return { done: true, value: undefined }; + return { done: false, value: item.value }; + } + + [Symbol.asyncIterator](): AsyncIterableIterator { + return this; + } + + private enqueue(item: QueueItem): void { + const waiter = this.waiters.shift(); + if (waiter) waiter(item); + else this.items.push(item); + } } diff --git a/src/modules/workflow/state.ts b/src/modules/workflow/state.ts index 4812382..c32176a 100644 --- a/src/modules/workflow/state.ts +++ b/src/modules/workflow/state.ts @@ -1,8 +1,3 @@ -// ═══════════════════════════════════════════════════════ -// PartnerIQ Research Workflow State Schema (LangGraph Annotation) -// ═══════════════════════════════════════════════════════ - -import { Annotation } from "@langchain/langgraph"; import type { AnalysisReport, CompanyInput, @@ -25,40 +20,3 @@ export interface ResearchWorkflowState { outcome: ResearchOutcome; fatalError: string | null; } - -export const ResearchWorkflowAnnotation = Annotation.Root({ - researchRunId: Annotation(), - input: Annotation(), - sourceResults: Annotation({ - reducer: (prev, next) => (next ? prev.concat(next) : prev), - default: () => [], - }), - findings: Annotation({ - reducer: (_, next) => next ?? [], - default: () => [], - }), - existingProfile: Annotation({ - reducer: (_, next) => next, - default: () => null, - }), - profile: Annotation({ - reducer: (_, next) => next, - default: () => null, - }), - diff: Annotation({ - reducer: (_, next) => next, - default: () => null, - }), - report: Annotation({ - reducer: (_, next) => next, - default: () => null, - }), - outcome: Annotation({ - reducer: (_, next) => next ?? "running", - default: () => "running", - }), - fatalError: Annotation({ - reducer: (_, next) => next, - default: () => null, - }), -}); diff --git a/src/observability/langfuse.ts b/src/observability/langfuse.ts index 93cf34d..a34802e 100644 --- a/src/observability/langfuse.ts +++ b/src/observability/langfuse.ts @@ -1,9 +1,8 @@ // ═══════════════════════════════════════════════════════ // Langfuse Observability & Privacy Minimization -// Provides client-side masking, deterministic scoring, and OTel/LangChain callback +// Provides client-side masking, deterministic scoring, and OTel tracing // ═══════════════════════════════════════════════════════ -import { CallbackHandler } from "@langfuse/langchain"; import { LangfuseClient } from "@langfuse/client"; import { LangfuseSpanProcessor } from "@langfuse/otel"; import { @@ -369,44 +368,6 @@ export function initOpenTelemetry(): void { } } -export function createLangfuseCallback( - context: ResearchTraceContext -): CallbackHandler | null { - const isEnabled = isLangfuseEnabled(); - const publicKey = process.env.LANGFUSE_PUBLIC_KEY; - const secretKey = process.env.LANGFUSE_SECRET_KEY; - - if (!isEnabled || !publicKey || !secretKey) { - return null; - } - - try { - const isCacheHit = Boolean(context.cacheHit); - return new CallbackHandler({ - sessionId: context.sessionId, - version: APP_VERSION, - tags: [ - "workflow:research", - "surface:sse", - isCacheHit ? "cache:hit" : "cache:miss", - ], - traceMetadata: { - researchRunId: context.researchRunId, - companyId: context.companyId, - companyIdHash: hashCompanyIdentifier(context.companyId), - requestedSources: context.requestedSources, - appVersion: APP_VERSION, - cacheHit: isCacheHit, - cacheMatchedBy: context.cacheMatchedBy || "none", - cacheAction: context.cacheAction || "auto", - }, - }); - } catch (err) { - console.warn("[Langfuse] Failed to initialize CallbackHandler:", err); - return null; - } -} - export async function flushLangfuse(): Promise { if (_processor) { try { diff --git a/tests/helpers/mock-adapters.ts b/tests/helpers/mock-adapters.ts index 6615d64..64d4fd0 100644 --- a/tests/helpers/mock-adapters.ts +++ b/tests/helpers/mock-adapters.ts @@ -15,7 +15,7 @@ export class MockLLMAdapter implements LLMAdapter { this.responses.set(promptSubstring, response); } - async complete(prompt: string, options?: LLMOptions): Promise { + private responseFor(prompt: string, options?: LLMOptions): string { this.callLog.push({ prompt, options }); for (const [key, value] of this.responses) { if (prompt.includes(key)) return value; @@ -28,19 +28,9 @@ export class MockLLMAdapter implements LLMAdapter { schema: z.ZodSchema, options?: LLMOptions, ): Promise { - const raw = await this.complete(prompt, options); + const raw = this.responseFor(prompt, options); return schema.parse(JSON.parse(raw)); } - - async *stream( - prompt: string, - options?: LLMOptions, - ): AsyncGenerator { - const response = await this.complete(prompt, options); - for (const word of response.split(" ")) { - yield word + " "; - } - } } export class MockSearchAdapter implements SearchAdapter { diff --git a/tests/integration/research-workflow.test.ts b/tests/integration/research-workflow.test.ts index 0acb687..919fad1 100644 --- a/tests/integration/research-workflow.test.ts +++ b/tests/integration/research-workflow.test.ts @@ -12,7 +12,7 @@ import type { SearchOptions } from "@/adapters/search/types"; import type { ResourceGuards } from "@/config"; import type { CompanyInput, StreamEvent } from "@/lib/types"; -describe("ResearchWorkflow (LangGraph StateGraph)", () => { +describe("ResearchWorkflow (native executor)", () => { let llm: MockLLMAdapter; let search: MockSearchAdapter; let scraper: MockScraperAdapter; @@ -493,6 +493,75 @@ describe("ResearchWorkflow (LangGraph StateGraph)", () => { expect(progressEvents.some((p) => p.data.status === "failed")).toBe(true); }); + it("emits a finding before slower sibling sources finish", async () => { + guards.maxQueriesPerResearch = 2; + guards.maxScrapePagesPerResearch = 1; + search.setResults("FPT", [ + { title: "FPT", url: "https://fpt.com.vn", snippet: "FPT overview" }, + ]); + let slowSourceFinished = false; + scraper.extract = async (url: string) => { + await new Promise((resolve) => setTimeout(resolve, 40)); + slowSourceFinished = true; + return { + url, + title: "FPT", + text: "FPT company website content long enough for profile synthesis.", + }; + }; + let findingArrivedEarly = false; + + for await (const event of buildWorkflow().stream( + { name: "FPT", website: "https://fpt.com.vn" }, + { researchRunId: "early-finding" }, + )) { + if (event.event === "research:finding" && !slowSourceFinished) { + findingArrivedEarly = true; + } + } + + expect(findingArrivedEarly).toBe(true); + }); + + it("produces equivalent terminal state through run and stream", async () => { + guards.maxQueriesPerResearch = 2; + guards.maxScrapePagesPerResearch = 1; + search.setResults("FPT", [ + { title: "FPT", url: "https://fpt.com.vn", snippet: "FPT overview" }, + ]); + scraper.extract = async (url: string) => ({ + url, + title: "FPT", + text: "FPT company website content long enough for profile synthesis.", + }); + const workflow = buildWorkflow(); + const input = { name: "FPT", website: "https://fpt.com.vn" }; + const runState = await workflow.run(input, { researchRunId: "equivalent" }); + let streamState: typeof runState | undefined; + let completionCalls = 0; + + for await (const event of workflow.stream(input, { + researchRunId: "equivalent", + onComplete: (state) => { + completionCalls += 1; + streamState = state; + }, + })) { + void event; + } + + const projectState = (state: typeof runState) => ({ + outcome: state.outcome, + sources: state.sourceResults.map(({ source, status }) => ({ source, status })), + findingUrls: state.findings.map(({ url }) => url), + profileName: state.profile?.officialName, + hasAnalysis: Boolean(state.report), + }); + expect(completionCalls).toBe(1); + expect(streamState).toBeDefined(); + expect(projectState(streamState!)).toEqual(projectState(runState)); + }); + it("skips linkedin when no linkedinUrl is provided", async () => { const profileModule = createProfileModule({ llm }); const analystModule = createAnalystModule({ llm }); diff --git a/tests/unit/adapters.test.ts b/tests/unit/adapters.test.ts index b78c303..60fa067 100644 --- a/tests/unit/adapters.test.ts +++ b/tests/unit/adapters.test.ts @@ -16,18 +16,6 @@ describe("Adapters Unit Tests", () => { llm = new MockLLMAdapter(); }); - it("returns default mock response when no match", async () => { - const res = await llm.complete("Tell me about company X"); - expect(res).toBe('{"result": "mock response"}'); - expect(llm.callLog.length).toBe(1); - }); - - it("returns canned response on substring match", async () => { - llm.setResponse("FPT", JSON.stringify({ officialName: "FPT Corporation" })); - const res = await llm.complete("Analyze FPT now"); - expect(res).toContain("FPT Corporation"); - }); - it("supports completeStructured with zod schema", async () => { const schema = z.object({ name: z.string(), @@ -39,14 +27,6 @@ describe("Adapters Unit Tests", () => { expect(result.founded).toBe(1988); }); - it("supports streaming async generator", async () => { - llm.setResponse("hello", "Hello world from stream"); - const chunks: string[] = []; - for await (const chunk of llm.stream("hello")) { - chunks.push(chunk); - } - expect(chunks.join("")).toContain("Hello world from stream"); - }); }); describe("MockSearchAdapter", () => { diff --git a/tests/unit/langchain-llm.test.ts b/tests/unit/langchain-llm.test.ts deleted file mode 100644 index 48b6c90..0000000 --- a/tests/unit/langchain-llm.test.ts +++ /dev/null @@ -1,157 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { z } from "zod"; -import { BaseChatModel } from "@langchain/core/language_models/chat_models"; -import { AIMessage, BaseMessage } from "@langchain/core/messages"; -import { ChatGeneration, ChatResult } from "@langchain/core/outputs"; -import { RunnableLambda } from "@langchain/core/runnables"; -import type { ChatOpenAI } from "@langchain/openai"; -import { OpenAIAdapter } from "@/adapters/llm/openai"; -import type { LLMOptions } from "@/adapters/llm/types"; - -class FakeChatModel extends BaseChatModel { - lastSignal?: AbortSignal; - lastCallbacks?: unknown; - responses: AIMessage[]; - structuredParsedNull = false; - - constructor(responses: AIMessage[] = []) { - super({}); - this.responses = responses; - } - - _llmType(): string { - return "fake"; - } - - async _generate( - _messages: BaseMessage[], - options?: { signal?: AbortSignal; callbacks?: unknown } - ): Promise { - this.lastSignal = options?.signal; - this.lastCallbacks = options?.callbacks; - const next = this.responses.shift() ?? new AIMessage({ - content: "default response", - usage_metadata: { input_tokens: 5, output_tokens: 7, total_tokens: 12 }, - }); - return { - generations: [{ message: next, text: next.content as string } as ChatGeneration], - }; - } - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - override withStructuredOutput(schema: any, config?: { includeRaw?: boolean }): any { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - return RunnableLambda.from(async (_input: any, options?: any) => { - this.lastSignal = options?.signal; - this.lastCallbacks = options?.callbacks; - const msg = this.responses.shift(); - const text = msg ? (msg.content as string) : '{"name":"FPT"}'; - const parsed = this.structuredParsedNull - ? null - : (schema as z.ZodSchema).parse(JSON.parse(text)); - return config?.includeRaw ? { raw: msg, parsed } : parsed; - }); - } -} - -describe("LangChain-backed LLM Adapter", () => { - it("completes plain text and logs usage", async () => { - const fakeModel = new FakeChatModel([ - new AIMessage({ - content: "plain response", - usage_metadata: { input_tokens: 5, output_tokens: 7, total_tokens: 12 }, - }), - ]); - - const adapter = new OpenAIAdapter("test-key", { - modelFactory: () => fakeModel as unknown as ChatOpenAI, - }); - - const result = await adapter.complete("hello"); - expect(result).toBe("plain response"); - expect(adapter.getUsageLogs()[0].totalTokens).toBe(12); - }); - - it("completes structured output with caller Zod schema", async () => { - const schema = z.object({ name: z.string() }); - const fakeModel = new FakeChatModel([ - new AIMessage({ - content: JSON.stringify({ name: "FPT" }), - usage_metadata: { input_tokens: 10, output_tokens: 15, total_tokens: 25 }, - }), - ]); - - const adapter = new OpenAIAdapter("test-key", { - modelFactory: () => fakeModel as unknown as ChatOpenAI, - }); - let recordedTokens = 0; - - const result = await adapter.completeStructured("extract company", schema, { - context: { - budget: { - claimModelCall: () => undefined, - recordModelUsage: (usage) => { - recordedTokens += usage.totalTokens; - }, - }, - }, - }); - expect(result).toEqual({ name: "FPT" }); - expect(adapter.getUsageLogs()[0].totalTokens).toBe(25); - expect(recordedTokens).toBe(25); - }); - - it("logs raw usage before rejecting an unparsed structured response", async () => { - const fakeModel = new FakeChatModel([ - new AIMessage({ - content: "invalid structured response", - usage_metadata: { input_tokens: 10, output_tokens: 4, total_tokens: 14 }, - }), - ]); - fakeModel.structuredParsedNull = true; - const adapter = new OpenAIAdapter("test-key", { - modelFactory: () => fakeModel as unknown as ChatOpenAI, - }); - - await expect( - adapter.completeStructured("extract company", z.object({ name: z.string() })), - ).rejects.toThrow("Structured output parsing failed"); - expect(adapter.getUsageLogs()[0].totalTokens).toBe(14); - }); - - it("forwards signal, callbacks, and claims budget", async () => { - const fakeModel = new FakeChatModel(); - const adapter = new OpenAIAdapter("test-key", { - modelFactory: () => fakeModel as unknown as ChatOpenAI, - }); - - const controller = new AbortController(); - let claimed = 0; - const budget = { - claimModelCall: (tokens: number) => { - claimed += tokens; - }, - recordModelUsage: () => { - // no-op - }, - }; - - const callback = { - name: "test_handler", - handleLLMStart: () => {}, - }; - - const options: LLMOptions = { - context: { - signal: controller.signal, - callbacks: [callback], - budget, - }, - }; - - await adapter.complete("test prompt", options); - - expect(fakeModel.lastSignal).toBe(controller.signal); - expect(claimed).toBeGreaterThan(0); - }); -}); diff --git a/tests/unit/langfuse-observability.test.ts b/tests/unit/langfuse-observability.test.ts index dc51a82..53f2188 100644 --- a/tests/unit/langfuse-observability.test.ts +++ b/tests/unit/langfuse-observability.test.ts @@ -46,7 +46,6 @@ import { maskPartnerIqTelemetry, maskPartnerIqTelemetryData, calculateDeterministicScores, - createLangfuseCallback, emitResearchScores, flushLangfuse, initOpenTelemetry, @@ -219,20 +218,6 @@ describe("Langfuse Observability & Privacy Minimization", () => { expect(scores).toContainEqual({ name: "research_success", value: "partial" }); }); - it("returns no-op / null handler when LANGFUSE_ENABLED is false", () => { - const prev = process.env.LANGFUSE_ENABLED; - process.env.LANGFUSE_ENABLED = "false"; - - const handler = createLangfuseCallback({ - researchRunId: "run-1", - companyId: "fpt", - requestedSources: ["web_search"], - }); - - expect(handler).toBeNull(); - process.env.LANGFUSE_ENABLED = prev; - }); - it("creates one workflow observation under the research trace", async () => { vi.stubEnv("LANGFUSE_ENABLED", "true"); vi.stubEnv("LANGFUSE_PUBLIC_KEY", "pk-test"); diff --git a/tests/unit/langgraph-runtime.test.ts b/tests/unit/langgraph-runtime.test.ts deleted file mode 100644 index fd97bcf..0000000 --- a/tests/unit/langgraph-runtime.test.ts +++ /dev/null @@ -1,16 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { z } from "zod"; -import { END, START, StateGraph, StateSchema } from "@langchain/langgraph"; - -describe("LangGraph runtime", () => { - it("compiles and invokes Zod state", async () => { - const State = new StateSchema({ value: z.number() }); - const graph = new StateGraph(State) - .addNode("increment", ({ value }) => ({ value: value + 1 })) - .addEdge(START, "increment") - .addEdge("increment", END) - .compile(); - - await expect(graph.invoke({ value: 1 })).resolves.toMatchObject({ value: 2 }); - }); -}); diff --git a/tests/unit/native-workflow-runtime.test.ts b/tests/unit/native-workflow-runtime.test.ts new file mode 100644 index 0000000..f72f68e --- /dev/null +++ b/tests/unit/native-workflow-runtime.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { settleWithConcurrency } from "@/modules/workflow"; + +describe("native workflow runtime", () => { + it("limits concurrent tasks and settles every result after a rejection", async () => { + let active = 0; + let maxActive = 0; + const completed: number[] = []; + + const tasks = [0, 1, 2, 3].map((index) => async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => setTimeout(resolve, 10)); + active -= 1; + completed.push(index); + if (index === 1) throw new Error("source failed"); + return index; + }); + + const results = await settleWithConcurrency(tasks, 2); + + expect(maxActive).toBe(2); + expect(completed).toHaveLength(4); + expect(results.map((result) => result.status)).toEqual([ + "fulfilled", + "rejected", + "fulfilled", + "fulfilled", + ]); + }); +}); diff --git a/tests/unit/openai-llm.test.ts b/tests/unit/openai-llm.test.ts new file mode 100644 index 0000000..d856019 --- /dev/null +++ b/tests/unit/openai-llm.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, vi } from "vitest"; +import { z } from "zod"; +import { OpenAIAdapter } from "@/adapters/llm/openai"; + +describe("OpenAI structured LLM adapter", () => { + it("parses structured output and records the actual token usage", async () => { + const parse = vi.fn().mockResolvedValue({ + output_parsed: { name: "FPT" }, + usage: { input_tokens: 10, output_tokens: 15, total_tokens: 25 }, + }); + const adapter = new OpenAIAdapter("test-key", { + client: { responses: { parse } }, + }); + const recordModelUsage = vi.fn(); + const claimModelCall = vi.fn(); + const signal = new AbortController().signal; + + await expect( + adapter.completeStructured("extract company", z.object({ name: z.string() }), { + systemPrompt: "Return company data", + model: "gpt-test", + maxTokens: 200, + temperature: 0.1, + schemaName: "company_profile", + context: { + signal, + budget: { claimModelCall, recordModelUsage }, + }, + }), + ).resolves.toEqual({ name: "FPT" }); + + expect(parse).toHaveBeenCalledWith( + expect.objectContaining({ + model: "gpt-test", + input: [ + { role: "system", content: "Return company data" }, + { role: "user", content: "extract company" }, + ], + max_output_tokens: 200, + temperature: 0.1, + text: { format: expect.objectContaining({ name: "company_profile" }) }, + }), + { signal }, + ); + expect(claimModelCall).toHaveBeenCalledWith(expect.any(Number)); + expect(recordModelUsage).toHaveBeenCalledWith( + expect.objectContaining({ + model: "gpt-test", + promptTokens: 10, + completionTokens: 15, + totalTokens: 25, + }), + ); + }); + + it("records usage before rejecting a response without parsed output", async () => { + const recordModelUsage = vi.fn(); + const adapter = new OpenAIAdapter("test-key", { + client: { + responses: { + parse: vi.fn().mockResolvedValue({ + output_parsed: null, + usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 }, + }), + }, + }, + }); + + await expect( + adapter.completeStructured("extract", z.object({ name: z.string() }), { + context: { + budget: { + claimModelCall: vi.fn(), + recordModelUsage, + }, + }, + }), + ).rejects.toThrow("Structured output parsing failed"); + expect(recordModelUsage).toHaveBeenCalledWith( + expect.objectContaining({ totalTokens: 6 }), + ); + }); +}); diff --git a/tests/unit/research-cache-route.test.ts b/tests/unit/research-cache-route.test.ts index 8f8943c..0bd8e2f 100644 --- a/tests/unit/research-cache-route.test.ts +++ b/tests/unit/research-cache-route.test.ts @@ -73,7 +73,6 @@ vi.mock("@/modules/workflow", () => ({ })); vi.mock("@/observability/langfuse", () => ({ - createLangfuseCallback: () => null, emitResearchScores: vi.fn(async () => undefined), flushLangfuse: vi.fn(async () => undefined), traceResearch: async (_context: unknown, task: (traceId: string) => Promise) => diff --git a/tests/unit/research-route-observability.test.ts b/tests/unit/research-route-observability.test.ts index 0eebefd..ea4b048 100644 --- a/tests/unit/research-route-observability.test.ts +++ b/tests/unit/research-route-observability.test.ts @@ -33,7 +33,6 @@ vi.mock("@/modules/workflow", () => ({ }), })); vi.mock("@/observability/langfuse", () => ({ - createLangfuseCallback: () => null, emitResearchScores: observabilityMocks.emitResearchScores, flushLangfuse: observabilityMocks.flushLangfuse, traceResearch: async (