diff --git a/.github/integration-ci/integration_plan.py b/.github/integration-ci/integration_plan.py index 51e34dca0..19436af45 100644 --- a/.github/integration-ci/integration_plan.py +++ b/.github/integration-ci/integration_plan.py @@ -1,10 +1,10 @@ #!/usr/bin/env python3 -"""Plan the Codex/Pi, Go and Java integration checks, without changing Python CI. +"""Plan selected TypeScript, Go and Java integration checks, without changing Python CI. Uses the existing Python planner's NUL-safe, rename-aware git diff reader. The same files can land independently with each integration: absent target directories are ignored, but an existing target may not silently lose its manifest or tests. -CI changes run every present target; TS shared dependencies run both TS targets. +CI changes run every present target; TS shared dependencies run all TS targets. """ import argparse import json @@ -18,6 +18,7 @@ TARGETS = ( ('codex-sdk', 'typescript', 'typescript/packages/traceai_codex_sdk', 'package.json'), ('pi-coding-agent', 'typescript', 'typescript/packages/traceai_pi_coding_agent', 'package.json'), + ('instructor', 'typescript', 'typescript/packages/traceai_instructor', 'package.json'), ('openai', 'go', 'go/openai', 'go.mod'), ('a2a', 'go', 'go/a2a', 'go.mod'), ('eino', 'go', 'go/eino', 'go.mod'), diff --git a/.github/integration-ci/test_integration_plan.py b/.github/integration-ci/test_integration_plan.py index 91c6e5c3b..e1b42a6c6 100644 --- a/.github/integration-ci/test_integration_plan.py +++ b/.github/integration-ci/test_integration_plan.py @@ -42,13 +42,13 @@ def test_ci_edits_and_all_select_every_present_target(self): self.assertEqual(self.names([path]), expected) self.assertEqual(self.names([], all_targets=True), expected) - def test_shared_typescript_inputs_cover_both_consumers(self): + def test_shared_typescript_inputs_cover_all_consumers(self): for path in ['typescript/pnpm-lock.yaml', 'typescript/package.json', 'typescript/pnpm-workspace.yaml', 'typescript/.npmrc', 'typescript/tsconfig.base.json', 'typescript/packages/fi-core/src/index.ts', 'typescript/packages/fi-semantic-conventions/src/index.ts']: with self.subTest(path=path): - self.assertEqual(self.names([path]), {'codex-sdk', 'pi-coding-agent'}) + self.assertEqual(self.names([path]), {name for name, language, _, _ in TARGETS if language == 'typescript'}) def test_unrelated_paths_do_not_claim_coverage(self): self.assertEqual(self.names(['README.md', 'python/frameworks/openai/tracer.py', diff --git a/typescript/packages/traceai_instructor/README.md b/typescript/packages/traceai_instructor/README.md index 61557411b..cd05d3774 100644 --- a/typescript/packages/traceai_instructor/README.md +++ b/typescript/packages/traceai_instructor/README.md @@ -1,64 +1,96 @@ # @traceai/instructor -OpenTelemetry instrumentation for Instructor. Part of traceAI, Future AGI's OpenTelemetry instrumentation libraries. - -[Documentation](https://docs.futureagi.com/docs/integrations/traceai/instructor) · [Source](https://github.com/future-agi/traceAI/tree/dev/typescript/packages/traceai_instructor) · [Issues](https://github.com/future-agi/traceAI/issues) +Trace Instructor's TypeScript structured-output requests in Future AGI. Each extraction appears as a `CHAIN` span covering parsing and validation, with provider requests and retries beneath it when provider instrumentation is enabled. ## Install +Use Node.js 20 or later and Instructor 1.x with Zod 3. + ```sh -npm install @traceai/instructor @traceai/fi-core @opentelemetry/instrumentation @opentelemetry/api@^1.9.0 @opentelemetry/context-async-hooks@^2.0.1 -npm install @instructor-ai/instructor@"^1.0.0" +npm install @traceai/instructor @traceai/openai @traceai/fi-core @instructor-ai/instructor@^1.0.0 openai@^4.58.0 zod@^3.25.76 @opentelemetry/api@^1.9.0 @opentelemetry/instrumentation@^0.202.0 @opentelemetry/context-async-hooks@^2.0.1 ``` -The SDK ranges above match this instrumentor's module-version filters. Install -only the SDKs your application uses. For Future AGI export, set `FI_API_KEY` and -`FI_SECRET_KEY` in your environment; keep provider credentials outside source code. - -## Package entry points +Set `FI_API_KEY`, `FI_SECRET_KEY`, and `OPENAI_API_KEY` in your environment. Keep credentials outside source code. -This source revision fixes the package entry points. Both CommonJS `require()` -and native Node ESM `import` use the same CommonJS implementation; this avoids -broken, extensionless ESM build imports. The corrected packages still need a -patch release. A previously published `0.1.0` tarball is not changed by this README. +## Trace an extraction -## Quick start - -Load this setup **before** loading the SDK you want to instrument. This CommonJS -example registers tracing; it does not make a provider request. +Save this as `app.cjs`. Register instrumentation before loading Instructor or OpenAI. ```javascript const { context } = require("@opentelemetry/api"); const { AsyncLocalStorageContextManager } = require("@opentelemetry/context-async-hooks"); -const { register, ProjectType } = require("@traceai/fi-core"); -// Standalone process only: if your OTel SDK already installs a context manager, -// reuse it instead of registering another one. -context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()); const { registerInstrumentations } = require("@opentelemetry/instrumentation"); +const { register, ProjectType } = require("@traceai/fi-core"); const { InstructorInstrumentation } = require("@traceai/instructor"); -const tracerProvider = register({ - projectName: "instructor-app", +const { OpenAIInstrumentation } = require("@traceai/openai"); + +// In an existing OTel application, reuse its context manager and provider. +context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()); +const provider = register({ + projectName: "instructor-typescript", projectType: ProjectType.OBSERVE, }); +const traceConfig = { hideInputs: true, hideOutputs: true }; registerInstrumentations({ - tracerProvider, instrumentations: [new InstructorInstrumentation()], + tracerProvider: provider, + instrumentations: [ + new InstructorInstrumentation({ traceConfig }), + new OpenAIInstrumentation({ instrumentationConfig: { enabled: false }, traceConfig }), + ], }); + +const createInstructor = require("@instructor-ai/instructor").default; +const OpenAI = require("openai").default; +const { z } = require("zod"); +const client = createInstructor({ client: new OpenAI(), mode: "TOOLS" }); +const request = { + model: "gpt-4o-mini", + messages: [{ role: "user", content: "Extract the name and age: Ada is 36." }], + response_model: { + name: "Person", + schema: z.object({ name: z.string(), age: z.number() }), + }, +}; + +async function main() { + try { + const person = await client.chat.completions.create(request); + console.log(person.name, person.age); + } finally { + await provider.shutdown(); + } +} +main().catch((error) => { console.error(error); process.exitCode = 1; }); +``` + +Run `node app.cjs`, then open the `instructor-typescript` project in Future AGI. Look for `Instructor Extract: Person` with an OpenAI request beneath it. Keep provider instrumentation enabled to see the individual model calls made during validation retries. + +The example hides prompts, parsed results, and exception details on both instrumentors. To capture content, change the corresponding `traceConfig` options after reviewing your data policy. Do not put sensitive data in schema names: names identify the extraction span. + +## Stream structured output + +Replace the request inside `main()` with this block. Consume the stream or close it with `return()` so its extraction span finishes. A `break` in `for await` closes it automatically. + +```javascript +const stream = await client.chat.completions.create({ ...request, stream: true }); +for await (const partial of stream) { + console.log(partial); +} ``` -`register()` installs a tracer provider, not an async context manager. The -explicit context manager above keeps nested asynchronous calls in one trace. -Run your application's SDK calls after this setup. Call -`await tracerProvider.shutdown()` when the application finishes to flush pending -spans. For ESM, follow the documentation for your OpenTelemetry loader and SDK -instead of assuming import order alone enables automatic patching. +The span stays open through iteration, records the last parsed partial result, and reports iteration failures. Early closure sets `instructor.stream.cancelled = true` without claiming success. Instructor 1.0 opens the provider stream eagerly; newer versions can defer it until iteration. Instrumentation preserves that SDK behavior. Instructor does not support validation retries for streams. + +## Span data and lifecycle + +- Instructor is an extraction framework, not a model provider. The parent is a `CHAIN`, not a second `LLM` span. It does not claim a provider or duplicate provider token/cost attributes. +- Input capture contains the request messages, not the Zod schema, SDK client, or request-option headers. Output capture contains the parsed document without `_meta`. +- When Instructor supplies `_meta.usage`, numeric counts appear as `instructor.usage.prompt_tokens`, `instructor.usage.completion_tokens`, and `instructor.usage.total_tokens`. They describe supplied result metadata, not an aggregate of retries. Older SDK versions and some streaming paths do not supply usage; the instrumentor does not estimate it. +- Calls without `response_model` pass through without an Instructor span. The provider instrumentor can still trace them. +- `instrumentation.disable()` restores owned module factory patches. Existing clients stop starting extraction spans; in-flight operations still finish. After re-enabling instrumentation, create new Instructor clients. -## Compatibility and privacy +For manual setup, call `instrumentation.manuallyInstrument(instructorModule)` before constructing clients. The module object must be mutable. With native ESM, use your OpenTelemetry loader, or pass a mutable `{ default: createInstructor }` object and construct clients through that object's patched `default` property. Static import order alone does not enable automatic instrumentation. -This instruments the SDK hooks implemented by this package, not every API offered -by the provider. The install ranges above are intentionally limited to the module versions this -instrumentor patches; its peer declarations may be broader. -Traces can contain prompts, responses, and tool data; configure capture and -redaction before sending production traffic. +[Python guide](https://docs.futureagi.com/docs/integrations/traceai/instructor) · [Source](https://github.com/future-agi/traceAI/tree/dev/typescript/packages/traceai_instructor) · [Issues](https://github.com/future-agi/traceAI/issues) ## License diff --git a/typescript/packages/traceai_instructor/contract/readme.cjs b/typescript/packages/traceai_instructor/contract/readme.cjs new file mode 100644 index 000000000..23a2d78fb --- /dev/null +++ b/typescript/packages/traceai_instructor/contract/readme.cjs @@ -0,0 +1,70 @@ +// Execute the actual README application against loopback-only provider and OTLP endpoints. +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const http = require('node:http'); +const path = require('node:path'); +const vm = require('node:vm'); +const { createRequire } = require('node:module'); + +(async () => { + const packageDir = path.resolve(__dirname, '..'); + const requireFrom = createRequire(path.join(process.argv[2] || packageDir, 'package.json')); + const readme = fs.readFileSync(path.join(packageDir, 'README.md'), 'utf8'); + const blocks = [...readme.matchAll(/```javascript\n([\s\S]*?)```/g)].map(match => match[1]); + const streaming = process.argv[3] === 'stream'; + const code = streaming ? blocks[0].replace( + 'const person = await client.chat.completions.create(request);\n console.log(person.name, person.age);', blocks[1] + ) : blocks[0]; + assert(code && (!streaming || code.includes('for await'))); + const payloads = []; + let calls = 0; + const server = http.createServer(async (req, res) => { + const buffers = []; + for await (const chunk of req) buffers.push(chunk); + const body = JSON.parse(Buffer.concat(buffers).toString()); + if (req.url === '/tracer/v1/traces') { + payloads.push(body); + res.writeHead(200, { 'content-type': 'application/json' }).end('{}'); + } else if (req.url === '/v1/chat/completions') { + calls++; + const tool = { id: 'offline-tool', type: 'function', function: { name: 'Person', arguments: '{"name":"Ada","age":36}' } }; + if (body.stream) { + const chunk = { id: 'offline', object: 'chat.completion.chunk', created: 0, model: body.model, + choices: [{ index: 0, delta: { tool_calls: [{ index: 0, ...tool }] }, finish_reason: null }] }; + res.writeHead(200, { 'content-type': 'text/event-stream' }); + res.end(`data: ${JSON.stringify(chunk)}\n\ndata: [DONE]\n\n`); + } else { + res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({ + id: 'offline', object: 'chat.completion', created: 0, model: body.model, + choices: [{ index: 0, finish_reason: 'tool_calls', message: { role: 'assistant', content: null, tool_calls: [tool] } }], + usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 }, + })); + } + } else { res.writeHead(404).end(); } + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const endpoint = `http://127.0.0.1:${server.address().port}`; + Object.assign(process.env, { + FI_API_KEY: 'offline-fi-key', FI_SECRET_KEY: 'offline-fi-secret', OPENAI_API_KEY: 'offline-provider-key', + FI_BASE_URL: endpoint, OPENAI_BASE_URL: `${endpoint}/v1`, + }); + try { + await vm.runInNewContext(code, { require: requireFrom, process, console: { log() {}, error: console.error } }); + assert.equal(process.exitCode || 0, 0); + assert.equal(calls, 1); + const spans = payloads.flatMap(p => p.resourceSpans.flatMap(r => r.scopeSpans.flatMap(s => s.spans))); + assert.equal(spans.length, 2, JSON.stringify(spans)); + const parent = spans.find(s => s.name === 'Instructor Extract: Person'); + const child = spans.find(s => s !== parent); + assert(parent && child); + assert.equal(child.parentSpanId, parent.spanId); + assert.equal(child.traceId, parent.traceId); + assert.equal(parent.attributes.find(a => a.key === 'fi.span.kind').value.stringValue, 'CHAIN'); + assert.equal(child.attributes.find(a => a.key === 'fi.span.kind').value.stringValue, 'LLM'); + const wire = JSON.stringify(payloads); + for (const hidden of ['Ada', 'offline-fi-key', 'offline-fi-secret', 'offline-provider-key']) assert(!wire.includes(hidden), hidden); + console.log(`README ${streaming ? 'stream' : 'request'}: one extraction + one provider span; OTLP parenting and privacy passed`); + } finally { + await new Promise(resolve => server.close(resolve)); + } +})().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/typescript/packages/traceai_instructor/package.json b/typescript/packages/traceai_instructor/package.json index 44943483e..3cc671a33 100644 --- a/typescript/packages/traceai_instructor/package.json +++ b/typescript/packages/traceai_instructor/package.json @@ -33,20 +33,27 @@ "license": "Apache-2.0", "dependencies": { "@opentelemetry/api": "^1.9.0", + "@opentelemetry/core": "^2.0.1", "@opentelemetry/instrumentation": "^0.202.0", "@traceai/fi-core": "workspace:^", "@traceai/fi-semantic-conventions": "workspace:^" }, "devDependencies": { - "@opentelemetry/sdk-trace-base": "^1.30.0", - "@opentelemetry/sdk-trace-node": "^1.30.0", + "@instructor-ai/instructor": "1.7.0", + "@opentelemetry/otlp-transformer": "0.202.0", + "@opentelemetry/sdk-trace-base": "^2.0.1", + "@opentelemetry/sdk-trace-node": "^2.0.1", + "@traceai/openai": "0.1.14", "@types/jest": "^29.5.14", "@types/node": "^22.10.10", + "instructor-floor": "npm:@instructor-ai/instructor@1.0.0", "jest": "^29.7.0", + "openai": "4.58.0", "rimraf": "^6.0.1", "ts-jest": "^29.2.5", "tsc-alias": "^1.8.10", - "typescript": "^5.7.3" + "typescript": "^5.7.3", + "zod": "3.25.76" }, "peerDependencies": { "@instructor-ai/instructor": "^1.0.0" diff --git a/typescript/packages/traceai_instructor/src/__tests__/instrumentation.test.ts b/typescript/packages/traceai_instructor/src/__tests__/instrumentation.test.ts index d24ba29b6..472f1c752 100644 --- a/typescript/packages/traceai_instructor/src/__tests__/instrumentation.test.ts +++ b/typescript/packages/traceai_instructor/src/__tests__/instrumentation.test.ts @@ -1,270 +1,59 @@ -/** - * Tests for Instructor instrumentation. - */ - -import { InstructorInstrumentation, isPatched, _resetPatchedStateForTesting } from "../instrumentation"; +import { InstructorInstrumentation, isPatched } from "../instrumentation"; import { VERSION } from "../version"; -import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; -import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; - -describe("InstructorInstrumentation", () => { - const memoryExporter = new InMemorySpanExporter(); - const tracerProvider = new NodeTracerProvider({ - spanProcessors: [new SimpleSpanProcessor(memoryExporter)] - }); - tracerProvider.register(); - - const instrumentation = new InstructorInstrumentation(); - instrumentation.disable(); - instrumentation.setTracerProvider(tracerProvider); - - beforeEach(() => { - memoryExporter.reset(); - _resetPatchedStateForTesting(); - }); - - afterAll(() => { - instrumentation.disable(); - }); - - describe("VERSION", () => { - it("should export VERSION constant", () => { - expect(VERSION).toBe(require("../../package.json").version); - }); - }); - - describe("InstructorInstrumentation", () => { - it("should create instrumentation instance", () => { - const inst = new InstructorInstrumentation(); - expect(inst).toBeInstanceOf(InstructorInstrumentation); - }); - - it("should accept configuration options", () => { - const inst = new InstructorInstrumentation({ - traceConfig: { - hideInputs: true, - hideOutputs: true, - }, - }); - expect(inst).toBeInstanceOf(InstructorInstrumentation); - }); - - it("should have instrumentationName", () => { - const inst = new InstructorInstrumentation(); - expect(inst.instrumentationName).toBe("@traceai/fi-instrumentation-instructor"); - }); - - it("should have instrumentationVersion", () => { - const inst = new InstructorInstrumentation(); - expect(inst.instrumentationVersion).toBe(VERSION); - }); - }); - describe("isPatched", () => { - it("should return false when not patched", () => { - instrumentation.disable(); - expect(isPatched()).toBe(false); - }); - }); - - describe("manuallyInstrument", () => { - it("should handle null module", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - // Should not throw - inst.manuallyInstrument(null); - expect(true).toBe(true); - }); - - it("should mark module as patched after manual instrumentation", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const mockModule = { Instructor: {} }; - inst.manuallyInstrument(mockModule); - expect(mockModule).toHaveProperty("_fiPatched", true); - }); - }); - - describe("enable/disable", () => { - it("should enable instrumentation", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - // Should not throw - expect(true).toBe(true); - }); - - it("should disable instrumentation", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - inst.disable(); - // Should not throw - expect(true).toBe(true); - }); - }); +const instances: InstructorInstrumentation[] = []; +function make() { + const inst = new InstructorInstrumentation({ instrumentationConfig: { enabled: false } }); + instances.push(inst); + return inst; +} +afterEach(() => { for (const inst of instances.splice(0)) inst.disable(); }); + +it("retains the public instrumentor name and release-managed version", () => { + const inst = make(); + expect(VERSION).toMatch(/^\d+\.\d+\.\d+$/); + expect(inst.instrumentationName).toBe("@traceai/fi-instrumentation-instructor"); + expect(inst.instrumentationVersion).toBe(VERSION); }); -describe("Instructor Patching", () => { - const memoryExporter = new InMemorySpanExporter(); - const tracerProvider = new NodeTracerProvider({ - spanProcessors: [new SimpleSpanProcessor(memoryExporter)] - }); - tracerProvider.register(); - - beforeEach(() => { - memoryExporter.reset(); - _resetPatchedStateForTesting(); - }); - - it("should wrap Instructor.chat.completions.create when available", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const originalCreate = jest.fn().mockResolvedValue({ - name: "John", - age: 30, - }); - - const mockModule = { - Instructor: { - prototype: { - chat: { - completions: { - create: originalCreate, - }, - }, - }, - }, - }; - - inst.manuallyInstrument(mockModule); - expect((mockModule as any)._fiPatched).toBe(true); - }); - - it("should wrap default export when it is a function", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const mockCreate = jest.fn().mockResolvedValue({ extracted: "data" }); - const mockModule = { - default: jest.fn().mockReturnValue({ - chat: { - completions: { - create: mockCreate, - }, - }, - }), - }; - - inst.manuallyInstrument(mockModule); - expect((mockModule as any)._fiPatched).toBe(true); - }); - - it("should not double-patch module", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const mockModule = { - _fiPatched: true, - Instructor: { - prototype: { - chat: { - completions: { - create: jest.fn(), - }, - }, - }, - }, - }; - - const original = mockModule.Instructor.prototype.chat.completions.create; - inst.manuallyInstrument(mockModule); - - // Should not have changed - expect(mockModule.Instructor.prototype.chat.completions.create).toBe(original); - }); +it("ignores null and unsupported modules without claiming to patch them", () => { + const inst = make(); + inst.manuallyInstrument(null); + inst.manuallyInstrument({ Instructor: {} }); + expect(isPatched()).toBe(false); }); -describe("Structured Extraction Tracing", () => { - const memoryExporter = new InMemorySpanExporter(); - const tracerProvider = new NodeTracerProvider({ - spanProcessors: [new SimpleSpanProcessor(memoryExporter)] - }); - tracerProvider.register(); - - beforeEach(() => { - memoryExporter.reset(); - _resetPatchedStateForTesting(); - }); - - it("should handle response_model parameter", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const mockModule = { - Instructor: { - prototype: { - chat: { - completions: { - create: jest.fn().mockResolvedValue({ name: "Test" }), - }, - }, - }, - }, - }; - - inst.manuallyInstrument(mockModule); - expect((mockModule as any)._fiPatched).toBe(true); - }); - - it("should handle responseModel parameter (camelCase)", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); - - const mockModule = { - Instructor: { - prototype: { - chat: { - completions: { - create: jest.fn().mockResolvedValue({ result: "data" }), - }, - }, - }, - }, - }; - - inst.manuallyInstrument(mockModule); - expect((mockModule as any)._fiPatched).toBe(true); - }); +it("does not let a second instrumentor remove the first one's patch", () => { + const first = make(); const second = make(); + const original = () => ({ chat: { completions: { create: async () => ({}) } } }); + const module = { default: original }; + first.manuallyInstrument(module); const wrapped = module.default; + second.manuallyInstrument(module); second.disable(); + expect(module.default).toBe(wrapped); expect(isPatched()).toBe(true); + first.disable(); expect(module.default).toBe(original); expect(isPatched()).toBe(false); +}); - it("should capture model name in span", () => { - const inst = new InstructorInstrumentation(); - inst.setTracerProvider(tracerProvider); - inst.enable(); +it("preserves a later factory wrapper on disable", () => { + const inst = make(); const module = { default: () => ({}) }; + inst.manuallyInstrument(module); + const wrapped = module.default; + const later = () => wrapped(); module.default = later; + inst.disable(); expect(module.default).toBe(later); + expect(isPatched()).toBe(false); +}); - const mockModule = { - Instructor: { - prototype: { - chat: { - completions: { - create: jest.fn().mockResolvedValue({ extracted: "value" }), - }, - }, - }, - }, - }; +it("restores the legacy prototype route without retaining created clients", () => { + const inst = make(); const original = async () => ({}); + const module = { Instructor: { prototype: { chat: { completions: { create: original } } } } }; + inst.manuallyInstrument(module); + expect(module.Instructor.prototype.chat.completions.create).not.toBe(original); + inst.disable(); expect(module.Instructor.prototype.chat.completions.create).toBe(original); +}); - inst.manuallyInstrument(mockModule); - expect((mockModule as any)._fiPatched).toBe(true); - }); +it("can instrument a factory again after disabling", () => { + const inst = make(); const original = () => ({}); const module = { default: original }; + inst.manuallyInstrument(module); inst.disable(); + inst.enable(); inst.manuallyInstrument(module); + expect(module.default).not.toBe(original); expect(isPatched()).toBe(true); + inst.disable(); expect(module.default).toBe(original); }); diff --git a/typescript/packages/traceai_instructor/src/__tests__/lifecycle.test.ts b/typescript/packages/traceai_instructor/src/__tests__/lifecycle.test.ts new file mode 100644 index 000000000..480cf16d1 --- /dev/null +++ b/typescript/packages/traceai_instructor/src/__tests__/lifecycle.test.ts @@ -0,0 +1,282 @@ +import { context, trace, SpanStatusCode } from "@opentelemetry/api"; +import { suppressTracing } from "@opentelemetry/core"; +import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer"; +import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import { z } from "zod"; +import { InstructorInstrumentation, isPatched, _resetPatchedStateForTesting } from "../instrumentation"; +import { SemanticConventions as S, FISpanKind } from "@traceai/fi-semantic-conventions"; + +const exporter = new InMemorySpanExporter(); +const provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); +provider.register(); +const tracer = provider.getTracer("synthetic-provider"); +const usage = { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 }; +const request = () => ({ + model: "synthetic", messages: [{ role: "user", content: "private input" }], + response_model: { name: "Person", schema: z.object({ name: z.string() }) }, + max_retries: 0, +}); +const completion = (name: unknown = "Ada") => ({ + id: "offline", object: "chat.completion", model: "synthetic", created: 0, usage, + choices: [{ index: 0, finish_reason: "stop", message: { role: "assistant", content: JSON.stringify({ name }) } }], +}); +function chunks() { + return ['{"name":', '"Ada"}'].map((content, index) => ({ + id: "offline", object: "chat.completion.chunk", model: "synthetic", created: 0, + choices: [{ index: 0, delta: { content }, finish_reason: index === 1 ? "stop" : null }], + })); +} +const extractionSpans = () => exporter.getFinishedSpans().filter(s => s.name.startsWith("Instructor Extract:")); +const wire = () => Buffer.from(JsonTraceSerializer.serializeRequest(exporter.getFinishedSpans())!).toString(); +let inst: InstructorInstrumentation; +function instrument(module: any, traceConfig = {}) { + inst = new InstructorInstrumentation({ instrumentationConfig: { enabled: false }, traceConfig }); + inst.setTracerProvider(provider); + inst.enable(); + inst.manuallyInstrument(module); + return inst; +} +beforeEach(() => { _resetPatchedStateForTesting(); }); +afterEach(() => { inst?.disable(); exporter.reset(); jest.restoreAllMocks(); }); +afterAll(async () => { await provider.shutdown(); context.disable(); trace.disable(); }); + +describe.each([ + ["floor 1.0.0", "instructor-floor"], ["current 1.7.0", "@instructor-ai/instructor"], +])("real Instructor %s", (_version, packageName) => { + // Real Instructor + Zod + zod-stream; only the provider transport is synthetic. + function setup(create: (...args: any[]) => any, traceConfig = {}) { + const module = { ...require(packageName) }; + const factory = module.default; + instrument(module, traceConfig); + const client = module.default({ client: { baseURL: "https://api.openai.com", chat: { completions: { create } } }, mode: "JSON" }); + return { module, factory, client }; + } + + it("exports one extraction parent, parsed output and namespaced usage without duplicate LLM accounting", async () => { + const childIds: string[] = []; + const { client } = setup(async () => { + const child = tracer.startSpan("provider request", { attributes: { [S.FI_SPAN_KIND]: FISpanKind.LLM } }); + childIds.push(child.spanContext().spanId); child.end(); + return completion(); + }); + const result = await client.chat.completions.create(request()); + expect(result.name).toBe("Ada"); + const [parent] = extractionSpans(); + expect(parent.attributes[S.FI_SPAN_KIND]).toBe(FISpanKind.CHAIN); + expect(parent.attributes[S.LLM_PROVIDER]).toBeUndefined(); + expect(parent.attributes[S.LLM_TOKEN_COUNT_TOTAL]).toBeUndefined(); + // 1.0 strips metadata during Zod validation; 1.7 returns _meta.usage. + expect(parent.attributes["instructor.usage.total_tokens"]).toBe(packageName === "instructor-floor" ? undefined : 6); + expect(JSON.parse(parent.attributes[S.OUTPUT_VALUE] as string)).toEqual({ name: "Ada" }); + expect(JSON.parse(parent.attributes[S.INPUT_VALUE] as string)).toEqual(request().messages); + const child = exporter.getFinishedSpans().find(s => s.spanContext().spanId === childIds[0])!; + expect(child.parentSpanContext?.spanId).toBe(parent.spanContext().spanId); + const payload = JSON.parse(wire()); + expect(payload.resourceSpans.length).toBeGreaterThan(0); + expect(wire()).not.toContain('"llm.provider"'); + }); + + it("keeps a lazy stream open and runs provider work under its parent", async () => { + let parentId: string | undefined; + const create = jest.fn(async () => { + parentId = trace.getActiveSpan()?.spanContext().spanId; + return (async function* () { for (const chunk of chunks()) yield chunk; })(); + }); + const { client } = setup(create); + const stream = await client.chat.completions.create({ ...request(), stream: true }); + // 1.0 opens the provider stream eagerly; 1.7 defers it until next(). + expect(create).toHaveBeenCalledTimes(packageName === "instructor-floor" ? 1 : 0); + expect(extractionSpans()).toHaveLength(0); + const values = []; + for await (const item of stream) values.push(item); + expect(values.at(-1).name).toBe("Ada"); + const [span] = extractionSpans(); + expect(extractionSpans()).toHaveLength(1); + expect(parentId).toBe(span.spanContext().spanId); + expect(span.status.code).toBe(SpanStatusCode.OK); + expect(JSON.parse(span.attributes[S.OUTPUT_VALUE] as string)).toEqual({ name: "Ada" }); + expect(trace.getActiveSpan()).toBeUndefined(); + }); + + it("records lazy provider rejection instead of ending OK before execution", async () => { + jest.spyOn(console, "error").mockImplementation(() => {}); + const error = new Error("private stream failure"); + const { client } = setup(async () => { throw error; }); + if (packageName === "instructor-floor") { + await expect(client.chat.completions.create({ ...request(), stream: true })).rejects.toThrow(); + } else { + const stream = await client.chat.completions.create({ ...request(), stream: true }); + await expect(stream.next()).rejects.toThrow(); + } + expect(extractionSpans()).toHaveLength(1); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.ERROR); + expect(extractionSpans()[0].events.some(e => e.name === "exception")).toBe(true); + }); + + it("leaves passthrough provider calls unwrapped", async () => { + const value = completion(); + const { client } = setup(async () => value); + const { response_model, ...params } = request(); + expect(await client.chat.completions.create(params)).toBe(value); + expect(extractionSpans()).toHaveLength(0); + }); + + it("honors suppression without swallowing the result", async () => { + const { client } = setup(async () => completion()); + const result = await context.with(suppressTracing(context.active()), () => client.chat.completions.create(request())); + expect(result.name).toBe("Ada"); expect(extractionSpans()).toHaveLength(0); + }); + + it("restores the factory and makes existing clients inert on disable", async () => { + const { module, factory, client } = setup(async () => completion()); + inst.disable(); + expect(module.default).toBe(factory); + expect(isPatched()).toBe(false); + expect((await client.chat.completions.create(request())).name).toBe("Ada"); + expect(extractionSpans()).toHaveLength(0); + }); + + it("keeps validation retry provider attempts under one extraction", async () => { + jest.spyOn(console, "warn").mockImplementation(() => {}); + let attempts = 0; + const parents: Array = []; + const { client } = setup(async () => { + parents.push(trace.getActiveSpan()?.spanContext().spanId); + return completion(attempts++ === 0 ? 123 : "Ada"); + }); + expect((await client.chat.completions.create({ ...request(), max_retries: 1 })).name).toBe("Ada"); + expect(attempts).toBe(2); expect(extractionSpans()).toHaveLength(1); + expect(parents).toEqual([extractionSpans()[0].spanContext().spanId, extractionSpans()[0].spanContext().spanId]); + }); + + it("redacts input, output and exception details on the OTLP wire", async () => { + jest.spyOn(console, "error").mockImplementation(() => {}); + const create = jest.fn().mockResolvedValueOnce(completion("private output")).mockRejectedValueOnce(new Error("private error")); + const { client } = setup(create, { hideInputs: true, hideOutputs: true }); + await client.chat.completions.create(request()); + await expect(client.chat.completions.create(request())).rejects.toThrow("private error"); + const payload = wire(); + for (const secret of ["private input", "private output", "private error"]) expect(payload).not.toContain(secret); + expect(extractionSpans()[1].status.code).toBe(SpanStatusCode.ERROR); + }); +}); + +describe("iterator and patch contracts", () => { + function setup(create: (...args: any[]) => any) { + const module = { default: () => ({ chat: { completions: { create } } }) }; + instrument(module); + return { module, client: module.default() }; + } + it("closes a never-started iterator without executing it", async () => { + const execute = jest.fn(); + const { client } = setup(async () => (async function* () { execute(); yield { name: "Ada" }; })()); + const stream = await client.chat.completions.create(request()); + expect(await stream.return("cancelled")).toEqual({ done: true, value: "cancelled" }); + expect(execute).not.toHaveBeenCalled(); + expect(extractionSpans()).toHaveLength(1); + expect(extractionSpans()[0].attributes["instructor.stream.cancelled"]).toBe(true); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.UNSET); + }); + it("preserves return values and closes on break under extraction context", async () => { + let cleanupParent: string | undefined; + const { client } = setup(async () => (async function* () { + try { yield { name: "A" }; yield { name: "Ada" }; } + finally { cleanupParent = trace.getActiveSpan()?.spanContext().spanId; } + })()); + const stream = await client.chat.completions.create(request()); + for await (const value of stream) { expect(value.name).toBe("A"); break; } + const [span] = extractionSpans(); + expect(cleanupParent).toBe(span.spanContext().spanId); + expect(JSON.parse(span.attributes[S.OUTPUT_VALUE] as string)).toEqual({ name: "A" }); + await stream.return(); expect(extractionSpans()).toHaveLength(1); + }); + it("propagates iterator throw by identity and records the failure", async () => { + const { client } = setup(async () => (async function* () { yield { name: "A" }; })()); + const stream = await client.chat.completions.create(request()); + await stream.next(); const error = new Error("sentinel"); + await expect(stream.throw(error)).rejects.toBe(error); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.ERROR); + }); + it("retains only the last cumulative output and latest supplied usage", async () => { + const { client } = setup(async () => (async function* () { + yield { name: "A", _meta: { usage: { total_tokens: 2 } } }; + yield { name: "Ada", _meta: { usage } }; + })()); + for await (const _ of await client.chat.completions.create(request())) { /* consume */ } + const [span] = extractionSpans(); + expect(JSON.parse(span.attributes[S.OUTPUT_VALUE] as string)).toEqual({ name: "Ada" }); + expect(span.attributes["instructor.usage.total_tokens"]).toBe(6); + }); + it("patches multiple module copies without wrapping the same factory twice", async () => { + const { module, client } = setup(async () => ({ name: "Ada" })); + const factory = module.default; + inst.manuallyInstrument(module); expect(module.default).toBe(factory); + const other = { default: () => ({ chat: { completions: { create: async (_params: unknown) => ({ name: "Grace" }) } } }) }; + inst.manuallyInstrument(other); + await client.chat.completions.create(request()); + await other.default().chat.completions.create(request()); + expect(extractionSpans()).toHaveLength(2); + }); + it("records failure after a yielded value without changing the thrown object", async () => { + const error = new Error("iteration failed"); + const { client } = setup(async () => (async function* () { yield { name: "A" }; throw error; })()); + const stream = await client.chat.completions.create(request()); + await stream.next(); expect(extractionSpans()).toHaveLength(0); + await expect(stream.next()).rejects.toBe(error); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.ERROR); + await stream.next(); expect(extractionSpans()).toHaveLength(1); + }); + it("records cleanup errors rather than reporting cancellation as success", async () => { + const error = new Error("cleanup failed"); + const { client } = setup(async () => (async function* () { + try { yield { name: "A" }; } finally { throw error; } + })()); + const stream = await client.chat.completions.create(request()); await stream.next(); + await expect(stream.return()).rejects.toBe(error); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.ERROR); + }); + it("does not finish if a generator handles throw and yields another value", async () => { + const { client } = setup(async () => (async function* () { + try { yield { name: "A" }; } catch { yield { name: "Ada" }; } + })()); + const stream = await client.chat.completions.create(request()); await stream.next(); + expect((await stream.throw(new Error("recoverable"))).value.name).toBe("Ada"); + expect(extractionSpans()).toHaveLength(0); await stream.next(); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.OK); + }); + it("finishes an empty stream and ignores malformed or missing usage", async () => { + const { client } = setup(async () => (async function* () {})()); + const stream = await client.chat.completions.create(request()); + expect((await stream.next()).done).toBe(true); + expect(extractionSpans()[0].attributes[S.OUTPUT_VALUE]).toBeUndefined(); + expect(extractionSpans()[0].status.code).toBe(SpanStatusCode.OK); + }); + it("preserves result identity, request options and cyclic output", async () => { + const result: Record = { name: "Ada", _meta: { usage: { total_tokens: NaN } } }; + result.self = result; + const create = jest.fn(async () => result); const { client } = setup(create); + const params = request(); const options = { headers: { authorization: "synthetic-secret" } }; + expect(await client.chat.completions.create(params, options)).toBe(result); + expect(create).toHaveBeenCalledWith(params, options); + expect(wire()).not.toContain("synthetic-secret"); + expect(extractionSpans()[0].attributes["instructor.usage.total_tokens"]).toBeUndefined(); + }); + it("finishes an in-flight stream after disable without tracing new calls", async () => { + const { client } = setup(async () => (async function* () { yield { name: "Ada" }; })()); + const stream = await client.chat.completions.create(request()); inst.disable(); + for await (const _ of stream) { /* finish the active operation */ } + for await (const _ of await client.chat.completions.create(request())) { /* untraced */ } + expect(extractionSpans()).toHaveLength(1); + }); + it("isolates overlapping asynchronous requests", async () => { + const parents: string[] = []; + const { client } = setup(async () => { + await Promise.resolve(); parents.push(trace.getActiveSpan()!.spanContext().spanId); + return { name: "Ada" }; + }); + await Promise.all([client.chat.completions.create(request()), client.chat.completions.create(request())]); + expect(new Set(parents).size).toBe(2); + expect(parents.sort()).toEqual(extractionSpans().map(s => s.spanContext().spanId).sort()); + }); +}); diff --git a/typescript/packages/traceai_instructor/src/__tests__/readme.test.ts b/typescript/packages/traceai_instructor/src/__tests__/readme.test.ts new file mode 100644 index 000000000..2c6c52cff --- /dev/null +++ b/typescript/packages/traceai_instructor/src/__tests__/readme.test.ts @@ -0,0 +1,12 @@ +import { execFileSync } from "node:child_process"; +import path from "node:path"; + +it.each(["request", "stream"])("runs the documented %s with real SDKs and a loopback OTLP receiver", (mode) => { + const directory = path.resolve(__dirname, "../.."); + const output = execFileSync(process.execPath, [path.join(directory, "contract/readme.cjs"), directory, mode], { + encoding: "utf8", timeout: 20000, + // Do not inherit credentials, proxies, or tracing configuration from the test runner. + env: { PATH: process.env.PATH, NODE_ENV: "test" }, + }); + expect(output).toContain("OTLP parenting and privacy passed"); +}, 25000); diff --git a/typescript/packages/traceai_instructor/src/instrumentation.ts b/typescript/packages/traceai_instructor/src/instrumentation.ts index 723948922..044c40ced 100644 --- a/typescript/packages/traceai_instructor/src/instrumentation.ts +++ b/typescript/packages/traceai_instructor/src/instrumentation.ts @@ -1,27 +1,12 @@ -/** - * OpenTelemetry instrumentation for Instructor. - * - * Instructor is a library for structured output from LLMs using Pydantic models. - * This instrumentation captures structured extraction calls. - */ - +/** OpenTelemetry extraction/validation spans for Instructor's Zod-based SDK. */ import { InstrumentationBase, InstrumentationNodeModuleDefinition, } from "@opentelemetry/instrumentation"; -import { - Span, - SpanKind, - SpanStatusCode, - context, - trace, -} from "@opentelemetry/api"; +import { Context, Span, SpanKind, SpanStatusCode, context, trace } from "@opentelemetry/api"; +import { isTracingSuppressed } from "@opentelemetry/core"; import { FITracer, TraceConfigOptions } from "@traceai/fi-core"; -import { - SemanticConventions, - FISpanKind, - MimeType, -} from "@traceai/fi-semantic-conventions"; +import { SemanticConventions, FISpanKind, MimeType } from "@traceai/fi-semantic-conventions"; import { VERSION } from "./version"; export interface InstructorInstrumentationConfig { @@ -29,174 +14,223 @@ export interface InstructorInstrumentationConfig { traceConfig?: TraceConfigOptions; } -/** - * Instructor Instrumentation class. - * - * Provides automatic instrumentation for the @instructor-ai/instructor SDK. - */ +// The instrumentor does not import the SDK: doing so would load it before the require hook. +type Callable = (...args: any[]) => any; +type Completions = { create: Callable }; +type Client = { chat?: { completions?: Completions } }; +type InstructorModule = { + default?: Callable; + Instructor?: { prototype?: Client }; + _fiPatched?: boolean; +}; +type PatchState = { active: boolean; restore: Array<() => void> }; +let owners = new WeakMap(); +let activePatches = 0; + export class InstructorInstrumentation extends InstrumentationBase { - private fiTracer!: FITracer; - private _traceConfig?: TraceConfigOptions; + private readonly traceConfig?: TraceConfigOptions; + private readonly patches = new Map(); constructor(config: InstructorInstrumentationConfig = {}) { - super( - "@traceai/fi-instrumentation-instructor", - VERSION, - config.instrumentationConfig || {} - ); - this._traceConfig = config.traceConfig; - } - - override enable(): void { - super.enable(); - this.fiTracer = new FITracer({ - tracer: this.tracer, - traceConfig: this._traceConfig, + // InstrumentationBase can call enable() from its constructor. Initialize our state first. + super("@traceai/fi-instrumentation-instructor", VERSION, { + ...config.instrumentationConfig, enabled: false, }); + this.traceConfig = config.traceConfig; + this.setConfig({ ...config.instrumentationConfig, enabled: config.instrumentationConfig?.enabled !== false }); + if (this.getConfig().enabled) this.enable(); } protected init() { - const module = new InstrumentationNodeModuleDefinition( - "@instructor-ai/instructor", - ["^1.0.0"], - this.patch.bind(this), - this.unpatch.bind(this) + return new InstrumentationNodeModuleDefinition( + "@instructor-ai/instructor", ["^1.0.0"], this.patch.bind(this), this.unpatch.bind(this) ); - return module; } - /** - * Manually instrument the @instructor-ai/instructor module. - */ - manuallyInstrument(instructorModule: any): void { - if (instructorModule == null) return; - this.patch(instructorModule); + /** Patch a mutable module export object before constructing Instructor clients. */ + manuallyInstrument(instructorModule: InstructorModule | null | undefined): void { + if (instructorModule) this.patch(instructorModule); } - private patch(instructorModule: any & { _fiPatched?: boolean }): any { - if (instructorModule?._fiPatched || _isFIPatched) { - return instructorModule; - } + override disable(): void { + super.disable(); + // The OTel loader only unpatches modules it loaded, not manually supplied modules. + for (const module of this.patches.keys()) this.unpatch(module); + } + private patch(module: InstructorModule): InstructorModule { + if (owners.has(module) || module._fiPatched) return module; + if (typeof module.default !== "function" && !module.Instructor?.prototype?.chat?.completions) return module; + const state: PatchState = { active: true, restore: [] }; const instrumentation = this; - - // Wrap the Instructor class create method - if (instructorModule.Instructor?.prototype?.chat?.completions?.create) { - this._wrap( - instructorModule.Instructor.prototype.chat.completions, - "create", - (original: Function) => { - return function patchedCreate(this: any, ...args: any[]) { - return instrumentation.traceStructuredExtraction(original, this, args); - }; - } - ); - } - - // Wrap default export if it's a function - if (typeof instructorModule.default === "function") { - const originalDefault = instructorModule.default; - instructorModule.default = function patchedInstructor(...args: any[]) { - const client = originalDefault.apply(this, args); - if (client?.chat?.completions?.create) { - const originalCreate = client.chat.completions.create.bind(client.chat.completions); - client.chat.completions.create = function(...createArgs: any[]) { - return instrumentation.traceStructuredExtraction(originalCreate, this, createArgs); - }; + const clients = new WeakSet(); + const patchClient = (client: Client) => { + const completions = client?.chat?.completions; + if (!completions || typeof completions.create !== "function" || clients.has(completions)) return; + clients.add(completions); + const original = completions.create; + completions.create = function (...args: any[]) { + if (!state.active || isTracingSuppressed(context.active()) || !args[0]?.response_model) { + return original.apply(this, args); } + return instrumentation.traceStructuredExtraction(original, this, args); + }; + // Do not retain clients here. Their wrappers become inert when this module is disabled. + }; + if (typeof module.default === "function") { + const original = module.default; + const wrapped = function (this: unknown, ...args: any[]) { + const client = original.apply(this, args); + if (state.active) patchClient(client); return client; }; + module.default = wrapped; + state.restore.push(() => { if (module.default === wrapped) module.default = original; }); } - - instructorModule._fiPatched = true; - _isFIPatched = true; - return instructorModule; - } - - private unpatch(instructorModule: any & { _fiPatched?: boolean }): void { - if (instructorModule?.Instructor?.prototype?.chat?.completions?.create) { - this._unwrap(instructorModule.Instructor.prototype.chat.completions, "create"); - } - if (instructorModule) { - instructorModule._fiPatched = false; + const prototype = module.Instructor?.prototype; + if (prototype?.chat?.completions) { + const completions = prototype.chat.completions; + const original = completions.create; + patchClient(prototype); + const wrapped = completions.create; + state.restore.push(() => { if (completions.create === wrapped) completions.create = original; }); } - _isFIPatched = false; + module._fiPatched = true; + owners.set(module, state); + this.patches.set(module, state); + activePatches++; + return module; } - private async traceStructuredExtraction( - original: Function, - instance: any, - args: any[] - ): Promise { - const params = args[0] || {}; - const model = params.model || "unknown"; - const responseModel = params.response_model?.name || params.responseModel?.name || "unknown"; + private unpatch(module: InstructorModule): void { + const state = this.patches.get(module); + if (!state) return; + state.active = false; + for (const restore of state.restore) restore(); + module._fiPatched = false; + owners.delete(module); + this.patches.delete(module); + activePatches--; + } - const span = this.fiTracer.startSpan(`Instructor Extract: ${responseModel}`, { + private async traceStructuredExtraction(original: Callable, instance: unknown, args: any[]): Promise { + const params = args[0]; + const responseModel = typeof params.response_model?.name === "string" ? params.response_model.name : "unknown"; + // Resolve the tracer at call time, including providers assigned after enable(). + const tracer = new FITracer({ tracer: this.tracer, traceConfig: this.traceConfig }); + const span = tracer.startSpan(`Instructor Extract: ${responseModel}`, { kind: SpanKind.INTERNAL, attributes: { - [SemanticConventions.FI_SPAN_KIND]: FISpanKind.LLM, - [SemanticConventions.LLM_PROVIDER]: "instructor", - [SemanticConventions.LLM_MODEL_NAME]: model, - [SemanticConventions.INPUT_VALUE]: safeJsonStringify(params), + [SemanticConventions.FI_SPAN_KIND]: FISpanKind.CHAIN, + [SemanticConventions.INPUT_VALUE]: safeJsonStringify(params.messages), [SemanticConventions.INPUT_MIME_TYPE]: MimeType.JSON, - [SemanticConventions.RAW_INPUT]: safeJsonStringify(params), "instructor.response_model": responseModel, + "instructor.model": typeof params.model === "string" ? params.model : undefined, }, }); - const execContext = trace.setSpan(context.active(), span); - try { - const result = await context.with(execContext, () => { - return original.apply(instance, args); - }); - - span.setAttributes({ - [SemanticConventions.OUTPUT_VALUE]: safeJsonStringify(result), - [SemanticConventions.OUTPUT_MIME_TYPE]: MimeType.JSON, - [SemanticConventions.RAW_OUTPUT]: safeJsonStringify(result), - }); - + const result = await context.with(execContext, () => original.apply(instance, args)); + if (isAsyncIterable(result)) return traceIterator(result, span, execContext); + captureOutput(span, result); span.setStatus({ code: SpanStatusCode.OK }); span.end(); - return result; } catch (error) { - span.recordException(error as Error); - span.setStatus({ - code: SpanStatusCode.ERROR, - message: (error as Error).message, - }); - span.end(); + endWithError(span, error); throw error; } } } -// Global patched flag -let _isFIPatched = false; +/** Whether this package owns any active module patches. */ +export function isPatched(): boolean { return activePatches > 0; } -/** - * Check if the module has been patched. - */ -export function isPatched(): boolean { - return _isFIPatched; +/** Reset diagnostics for isolated tests; disable instrumentors first. */ +export function _resetPatchedStateForTesting(): void { + owners = new WeakMap(); + activePatches = 0; } -/** - * Reset the patched state (for testing only). - */ -export function _resetPatchedStateForTesting(): void { - _isFIPatched = false; +function isAsyncIterable(value: unknown): value is AsyncIterable { + return value != null && typeof (value as AsyncIterable)[Symbol.asyncIterator] === "function"; +} + +function traceIterator(source: AsyncIterable, span: Span, execContext: Context): AsyncIterableIterator { + const iterator = context.with(execContext, () => source[Symbol.asyncIterator]()); + let ended = false; + let cancelled = false; + let latest: unknown; + const finish = () => { + if (ended) return; + ended = true; + captureOutput(span, latest); + if (cancelled) span.setAttribute("instructor.stream.cancelled", true); + else span.setStatus({ code: SpanStatusCode.OK }); + span.end(); + latest = undefined; + }; + const advance = async (method: "next" | "return" | "throw", args: unknown[]): Promise> => { + try { + if (method === "return") cancelled = true; + const result = await context.with(execContext, () => { + const operation = iterator[method]; + if (operation) return operation.apply(iterator, args as []); + if (method === "throw") throw args[0]; + return { done: true, value: args[0] }; + }); + if (!ended) { + if (!result.done) { + latest = result.value; + captureUsage(span, latest); + } else finish(); + } + return result; + } catch (error) { + if (!ended) { ended = true; latest = undefined; endWithError(span, error); } + throw error; + } + }; + // Unlike an async-generator wrapper, return()/throw() before first next() also finalize the span. + return { + next: (...args: [] | [unknown]) => advance("next", args), + return: (...args: [unknown?]) => advance("return", args), + throw: (...args: [unknown?]) => advance("throw", args), + [Symbol.asyncIterator]() { return this; }, + }; } -// Helper functions +function captureOutput(span: Span, value: unknown): void { + if (value === undefined) return; + captureUsage(span, value); + // Instructor's metadata is not the extracted document (and can contain internal parser state). + const output = value !== null && typeof value === "object" && !Array.isArray(value) + ? Object.fromEntries(Object.entries(value).filter(([key]) => key !== "_meta")) : value; + const serialized = safeJsonStringify(output); + if (serialized !== undefined) span.setAttributes({ + [SemanticConventions.OUTPUT_VALUE]: serialized, + [SemanticConventions.OUTPUT_MIME_TYPE]: MimeType.JSON, + }); +} -function safeJsonStringify(obj: unknown): string { - try { - return JSON.stringify(obj); - } catch { - return String(obj); +function captureUsage(span: Span, value: unknown): void { + const usage = (value as { _meta?: { usage?: Record } } | null)?._meta?.usage; + // A CHAIN span is not another model request. Only provider instrumentation owns gen_ai usage/cost. + for (const key of ["prompt_tokens", "completion_tokens", "total_tokens"]) { + const count = usage?.[key]; + if (typeof count === "number" && Number.isFinite(count) && count >= 0) { + span.setAttribute(`instructor.usage.${key}`, count); + } } } + +function endWithError(span: Span, error: unknown): void { + const exception = error instanceof Error || typeof error === "string" ? error : { name: "Error" }; + span.recordException(exception); + span.setStatus({ code: SpanStatusCode.ERROR, message: error instanceof Error ? error.message : undefined }); + span.end(); +} + +function safeJsonStringify(value: unknown): string | undefined { + try { return JSON.stringify(value); } catch { return undefined; } +} diff --git a/typescript/pnpm-lock.yaml b/typescript/pnpm-lock.yaml index 4f07c068b..5d766e4e3 100644 --- a/typescript/pnpm-lock.yaml +++ b/typescript/pnpm-lock.yaml @@ -1147,12 +1147,12 @@ importers: packages/traceai_instructor: dependencies: - '@instructor-ai/instructor': - specifier: ^1.0.0 - version: 1.7.0(zod@4.3.6) '@opentelemetry/api': specifier: ^1.9.0 version: 1.9.0 + '@opentelemetry/core': + specifier: ^2.0.1 + version: 2.0.1(@opentelemetry/api@1.9.0) '@opentelemetry/instrumentation': specifier: ^0.202.0 version: 0.202.0(@opentelemetry/api@1.9.0) @@ -1163,21 +1163,36 @@ importers: specifier: workspace:^ version: link:../fi-semantic-conventions devDependencies: + '@instructor-ai/instructor': + specifier: 1.7.0 + version: 1.7.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76) + '@opentelemetry/otlp-transformer': + specifier: 0.202.0 + version: 0.202.0(@opentelemetry/api@1.9.0) '@opentelemetry/sdk-trace-base': - specifier: ^1.30.0 - version: 1.30.1(@opentelemetry/api@1.9.0) + specifier: ^2.0.1 + version: 2.0.1(@opentelemetry/api@1.9.0) '@opentelemetry/sdk-trace-node': - specifier: ^1.30.0 - version: 1.30.1(@opentelemetry/api@1.9.0) + specifier: ^2.0.1 + version: 2.0.1(@opentelemetry/api@1.9.0) + '@traceai/openai': + specifier: 0.1.14 + version: 0.1.14(openai@4.58.0(zod@3.25.76)) '@types/jest': specifier: ^29.5.14 version: 29.5.14 '@types/node': specifier: ^22.10.10 version: 22.15.33 + instructor-floor: + specifier: npm:@instructor-ai/instructor@1.0.0 + version: '@instructor-ai/instructor@1.0.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76)' jest: specifier: ^29.7.0 version: 29.7.0(@types/node@22.15.33)(ts-node@10.9.2(@types/node@22.15.33)(typescript@5.8.3)) + openai: + specifier: 4.58.0 + version: 4.58.0(zod@3.25.76) rimraf: specifier: ^6.0.1 version: 6.1.2 @@ -1190,6 +1205,9 @@ importers: typescript: specifier: ^5.7.3 version: 5.8.3 + zod: + specifier: 3.25.76 + version: 3.25.76 packages/traceai_lancedb: dependencies: @@ -3668,6 +3686,12 @@ packages: resolution: {integrity: sha512-bV0Tgo9K4hfPCek+aMAn81RppFKv2ySDQeMoSZuvTASywNTnVJCArCZE2FWqpvIatKu7VMRLWlR1EazvVhDyhQ==} engines: {node: '>=18.18'} + '@instructor-ai/instructor@1.0.0': + resolution: {integrity: sha512-FGh1KX97nJnkreAGVg135rI68wkuTedQ6nkaFgFXXer/CQc4Gxn6spI0fA/F4DYnGlRcPl+STddUn/sEoFyGsQ==} + peerDependencies: + openai: '>=4.28.0' + zod: '>=3.22.4' + '@instructor-ai/instructor@1.7.0': resolution: {integrity: sha512-7ifu2ZzFBMSOtkY2X5dd2RCxQvfO6Bx+Ag/9V+FkamLGGGciazlTKsEyQR6KSn7BXADLWLIDV296S+O+CifE/A==} peerDependencies: @@ -6004,6 +6028,11 @@ packages: '@traceai/fi-semantic-conventions@1.0.0': resolution: {integrity: sha512-P+idlTaWs5YBURzUcJ8dPut9Y32I8E2FxMJfQbU8xISoWOW2xX6w5+Ri7Xt0Ve/NJ6XvbgMqojN3aGH3ZJBaLw==} + '@traceai/openai@0.1.14': + resolution: {integrity: sha512-EMfHcOCZvNoIF0fbLqKVPv6+LdlZT8hS6mXI5Hcrt8iS2IH36WyOmyLe/b/cm/BrWeY4egKZCu/+ulNghcw3Iw==} + peerDependencies: + openai: ^4.0.0 || ^5.0.0 + '@tsconfig/node10@1.0.11': resolution: {integrity: sha512-DcRjDCujK/kCk/cUe8Xz8ZSpm8mS3mNNpta+jGCA6USEDfktlNvm1+IuZ9eTcDbNk41BHwpHHeW+N1lKCz4zOw==} @@ -9257,6 +9286,15 @@ packages: zod: optional: true + openai@4.58.0: + resolution: {integrity: sha512-iEjwfZ+3ZOyf/Rjfca0vrDA2y4Lt7W3auf2ZJGt6+puyvzGYwp22da4mrNcYJkcek7SngOqiQI3h/44XCSCO6A==} + hasBin: true + peerDependencies: + zod: ^3.23.8 + peerDependenciesMeta: + zod: + optional: true + openai@4.98.0: resolution: {integrity: sha512-TmDKur1WjxxMPQAtLG5sgBSCJmX7ynTsGmewKzoDwl1fRxtbLOsiR0FA/AOAAtYUmP6azal+MYQuOENfdU+7yg==} hasBin: true @@ -10750,6 +10788,12 @@ packages: zod-from-json-schema@0.5.2: resolution: {integrity: sha512-/dNaicfdhJTOuUd4RImbLUE2g5yrSzzDjI/S6C2vO2ecAGZzn9UcRVgtyLSnENSmAOBRiSpUdzDS6fDWX3Z35g==} + zod-stream@1.0.1: + resolution: {integrity: sha512-sjSzr/cyfwLeWAltDw3uWS4cAnvhtacD7TeMU0JoS/xm1NJKeIQL05qY2WYUfrSh9ftPf41F1+ysvz02ttWQwA==} + peerDependencies: + openai: '>=4.24.1' + zod: '>=3.22.4' + zod-stream@3.0.0: resolution: {integrity: sha512-BfW8KI+FCEMQOAIf8Qgc2q67iZ+LZ/zzk7alU45FWiZIrBSt0JMxNVGuc3o1yTpGFVk3zXiHt3TZvmUiad/9nw==} peerDependencies: @@ -10766,6 +10810,12 @@ packages: peerDependencies: zod: ^3.25 || ^4 + zod-validation-error@2.1.0: + resolution: {integrity: sha512-VJh93e2wb4c3tWtGgTa0OF/dTt/zoPCPzXq4V11ZjxmEAFaPi/Zss1xIZdEB5RD8GD00U0/iVXgqkF77RV7pdQ==} + engines: {node: '>=18.0.0'} + peerDependencies: + zod: ^3.18.0 + zod-validation-error@3.5.4: resolution: {integrity: sha512-+hEiRIiPobgyuFlEojnqjJnhFvg4r/i3cqgcm67eehZf/WBaK3g6cD02YU9mtdVxZjv8CzCA9n/Rhrs3yAAvAw==} engines: {node: '>=18.0.0'} @@ -12921,6 +12971,20 @@ snapshots: '@humanwhocodes/retry@0.4.3': {} + '@instructor-ai/instructor@1.0.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76)': + dependencies: + openai: 4.58.0(zod@3.25.76) + zod: 3.25.76 + zod-stream: 1.0.1(openai@4.58.0(zod@3.25.76))(zod@3.25.76) + zod-validation-error: 2.1.0(zod@3.25.76) + + '@instructor-ai/instructor@1.7.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76)': + dependencies: + openai: 4.58.0(zod@3.25.76) + zod: 3.25.76 + zod-stream: 3.0.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76) + zod-validation-error: 3.5.4(zod@3.25.76) + '@instructor-ai/instructor@1.7.0(zod@4.3.6)': dependencies: zod: 4.3.6 @@ -16269,12 +16333,12 @@ snapshots: '@grpc/grpc-js': 1.14.3 '@grpc/proto-loader': 0.7.15 '@opentelemetry/api': 1.9.1 - '@opentelemetry/context-async-hooks': 2.0.1(@opentelemetry/api@1.9.1) + '@opentelemetry/context-async-hooks': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/core': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/exporter-trace-otlp-grpc': 0.202.0(@opentelemetry/api@1.9.1) '@opentelemetry/exporter-trace-otlp-http': 0.202.0(@opentelemetry/api@1.9.1) '@opentelemetry/resources': 2.7.1(@opentelemetry/api@1.9.1) - '@opentelemetry/sdk-trace-base': 2.0.1(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace-base': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/sdk-trace-node': 2.7.1(@opentelemetry/api@1.9.1) '@opentelemetry/semantic-conventions': 1.41.1 '@protobuf-ts/grpc-transport': 2.11.1(@grpc/grpc-js@1.14.3) @@ -16288,6 +16352,16 @@ snapshots: '@opentelemetry/api': 1.9.1 uuid: 11.1.0 + '@traceai/openai@0.1.14(openai@4.58.0(zod@3.25.76))': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/instrumentation': 0.202.0(@opentelemetry/api@1.9.1) + '@traceai/fi-core': 1.1.0 + '@traceai/fi-semantic-conventions': 1.0.0 + openai: 4.58.0(zod@3.25.76) + transitivePeerDependencies: + - supports-color + '@tsconfig/node10@1.0.11': {} '@tsconfig/node12@1.0.11': {} @@ -20502,6 +20576,22 @@ snapshots: transitivePeerDependencies: - encoding + openai@4.58.0(zod@3.25.76): + dependencies: + '@types/node': 18.19.100 + '@types/node-fetch': 2.6.12 + '@types/qs': 6.14.0 + abort-controller: 3.0.0 + agentkeepalive: 4.6.0 + form-data-encoder: 1.7.2 + formdata-node: 4.4.1 + node-fetch: 2.7.0 + qs: 6.16.0 + optionalDependencies: + zod: 3.25.76 + transitivePeerDependencies: + - encoding + openai@4.98.0(ws@8.21.0)(zod@3.24.4): dependencies: '@types/node': 18.19.100 @@ -21208,6 +21298,11 @@ snapshots: sax@1.4.4: {} + schema-stream@3.2.2(zod@3.25.76): + dependencies: + ramda: 0.29.1 + zod: 3.25.76 + schema-stream@3.2.2(zod@4.3.6): dependencies: ramda: 0.29.1 @@ -22131,6 +22226,20 @@ snapshots: dependencies: zod: 4.3.6 + zod-stream@1.0.1(openai@4.58.0(zod@3.25.76))(zod@3.25.76): + dependencies: + openai: 4.58.0(zod@3.25.76) + schema-stream: 3.2.2(zod@3.25.76) + zod: 3.25.76 + zod-to-json-schema: 3.25.1(zod@3.25.76) + + zod-stream@3.0.0(openai@4.58.0(zod@3.25.76))(zod@3.25.76): + dependencies: + openai: 4.58.0(zod@3.25.76) + schema-stream: 3.2.2(zod@3.25.76) + zod: 3.25.76 + zod-to-json-schema: 3.25.1(zod@3.25.76) + zod-stream@3.0.0(zod@4.3.6): dependencies: schema-stream: 3.2.2(zod@4.3.6) @@ -22157,6 +22266,14 @@ snapshots: dependencies: zod: 4.6.5 + zod-validation-error@2.1.0(zod@3.25.76): + dependencies: + zod: 3.25.76 + + zod-validation-error@3.5.4(zod@3.25.76): + dependencies: + zod: 3.25.76 + zod-validation-error@3.5.4(zod@4.3.6): dependencies: zod: 4.3.6