Skip to content

fix(observability): address code review findings from PR #860 #875

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
murdore merged 1 commit into from
Mar 15, 2026
Merged

fix(observability): address code review findings from PR #860 #875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 2 additions & 89 deletions src/lib/core/baseProvider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,6 @@ import { IMAGE_GENERATION_MODELS } from "../core/constants.js";
import type { EvaluationData } from "../index.js";
import { MiddlewareFactory } from "../middleware/factory.js";
import type { NeuroLink } from "../neurolink.js";
import { getMetricsAggregator } from "../observability/metricsAggregator.js";
import { SpanStatus, SpanType } from "../observability/types/spanTypes.js";
import { SpanSerializer } from "../observability/utils/spanSerializer.js";
import { ATTR, tracers } from "../telemetry/index.js";
import type { JsonValue, UnknownRecord } from "../types/common.js";
import type {
Expand All @@ -28,7 +25,6 @@ import type {
} from "../types/typeAliases.js";
import { isAbortError } from "../utils/errorHandling.js";
import { logger } from "../utils/logger.js";
import { calculateCost } from "../utils/pricing.js";
import {
composeAbortSignals,
createTimeoutController,
Expand Down Expand Up @@ -164,21 +160,6 @@ export abstract class BaseProvider implements AIProvider {
): Promise<StreamResult> {
let options = this.normalizeStreamOptions(optionsOrPrompt);

// Observability: create metrics span for provider.stream
const metricsSpan = SpanSerializer.createSpan(
SpanType.MODEL_GENERATION,
"provider.stream",
{
"ai.provider": this.providerName || "unknown",
"ai.model": this.modelName || options.model || "unknown",
"ai.temperature": options.temperature,
"ai.max_tokens": options.maxTokens,
},
this._traceContext?.parentSpanId,
this._traceContext?.traceId,
);
let metricsSpanRecorded = false;

// OTEL span for provider-level stream tracing
const otelStreamSpan = tracers.provider.startSpan(
"neurolink.provider.stream",
Expand Down Expand Up @@ -298,15 +279,6 @@ export abstract class BaseProvider implements AIProvider {
}
}
} catch (error) {
// Observability: record failed stream span
metricsSpanRecorded = true;
const endedStreamSpan = SpanSerializer.endSpan(
metricsSpan,
SpanStatus.ERROR,
error instanceof Error ? error.message : String(error),
);
getMetricsAggregator().recordSpan(endedStreamSpan);

otelStreamSpan.setStatus({
code: SpanStatusCode.ERROR,
message: error instanceof Error ? error.message : String(error),
Expand All @@ -315,14 +287,8 @@ export abstract class BaseProvider implements AIProvider {

throw error;
} finally {
// Observability: record successful stream span (only if not already ended via error path)
if (!metricsSpanRecorded) {
const endedStreamSpan = SpanSerializer.endSpan(
metricsSpan,
SpanStatus.OK,
);
getMetricsAggregator().recordSpan(endedStreamSpan);

// End OTEL span on success (only if not already ended via error path)
if (otelStreamSpan.isRecording()) {
otelStreamSpan.setStatus({ code: SpanStatusCode.OK });
otelStreamSpan.end();
}
Expand Down Expand Up @@ -719,20 +685,6 @@ export abstract class BaseProvider implements AIProvider {
this.validateOptions(options);
const startTime = Date.now();

// Observability: create metrics span for provider.generate
const metricsSpan = SpanSerializer.createSpan(
SpanType.MODEL_GENERATION,
"provider.generate",
{
"ai.provider": this.providerName || "unknown",
"ai.model": this.modelName || options.model || "unknown",
"ai.temperature": options.temperature,
"ai.max_tokens": options.maxTokens,
},
this._traceContext?.parentSpanId,
this._traceContext?.traceId,
);

// OTEL span for provider-level generate tracing
// Use startActiveSpan pattern via context.with() so child spans become descendants
const otelSpan = tracers.provider.startSpan("neurolink.provider.generate", {
Expand Down Expand Up @@ -996,47 +948,8 @@ export abstract class BaseProvider implements AIProvider {
}
}

// Observability: record successful generate span with token/cost data
let enrichedGenerateSpan = { ...metricsSpan };
if (enhancedResult?.usage) {
enrichedGenerateSpan = SpanSerializer.enrichWithTokenUsage(
enrichedGenerateSpan,
{
promptTokens: enhancedResult.usage.input || 0,
completionTokens: enhancedResult.usage.output || 0,
totalTokens: enhancedResult.usage.total || 0,
},
);
const cost = calculateCost(this.providerName, this.modelName, {
input: enhancedResult.usage.input || 0,
output: enhancedResult.usage.output || 0,
total: enhancedResult.usage.total || 0,
});
if (cost && cost > 0) {
enrichedGenerateSpan = SpanSerializer.enrichWithCost(
enrichedGenerateSpan,
{
totalCost: cost,
},
);
}
}
const endedGenerateSpan = SpanSerializer.endSpan(
enrichedGenerateSpan,
SpanStatus.OK,
);
getMetricsAggregator().recordSpan(endedGenerateSpan);

return await this.enhanceResult(enhancedResult, options, startTime);
} catch (error) {
// Observability: record failed generate span
const endedGenerateSpan = SpanSerializer.endSpan(
metricsSpan,
SpanStatus.ERROR,
error instanceof Error ? error.message : String(error),
);
getMetricsAggregator().recordSpan(endedGenerateSpan);

otelSpan.setStatus({
code: SpanStatusCode.ERROR,
message: error instanceof Error ? error.message : String(error),
Expand Down
33 changes: 32 additions & 1 deletion src/lib/neurolink.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2394,7 +2394,16 @@
span.spanId = traceCtx.parentSpanId;
span.parentSpanId = undefined;
}
span = SpanSerializer.endSpan(span, SpanStatus.OK);
// Mark failed generations with ERROR status so metrics count them correctly
const spanStatus =
data.success === false || data.error
? SpanStatus.ERROR
: SpanStatus.OK;
span = SpanSerializer.endSpan(
span,
spanStatus,
data.error ? String(data.error) : undefined,
);
span.durationMs = responseTime;

if (usage) {
Expand Down Expand Up @@ -2725,13 +2734,13 @@
* @see {@link stream} for streaming generation
* @since 1.0.0
*/
async generate(

Check warning on line 2737 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / test (20) 

Async method 'generate' has too many lines (505). Maximum allowed is 300

Check warning on line 2737 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / 🛡️ Code Quality & Security Gate 

Async method 'generate' has too many lines (505). Maximum allowed is 300
optionsOrPrompt: GenerateOptions | string,
): Promise<GenerateResult> {
return tracers.sdk.startActiveSpan(
"neurolink.generate",
{ kind: SpanKind.INTERNAL },
async (generateSpan) => {

Check warning on line 2743 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / test (20) 

Async arrow function has too many lines (497). Maximum allowed is 300

Check warning on line 2743 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / 🛡️ Code Quality & Security Gate 

Async arrow function has too many lines (497). Maximum allowed is 300
// Set metrics trace context for parent-child span linking.
// The generation span will be the root (no parentSpanId).
// Tool spans will be children of the root span via rootSpanId.
Expand All @@ -2744,7 +2753,7 @@
// so concurrent generate/stream calls don't race.
return metricsTraceContextStorage.run(
{ traceId: metricsTraceId, parentSpanId: metricsRootSpanId },
async () => {

Check warning on line 2756 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / test (20) 

Async arrow function has too many lines (482). Maximum allowed is 300

Check warning on line 2756 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / 🛡️ Code Quality & Security Gate 

Async arrow function has too many lines (482). Maximum allowed is 300
try {
const originalPrompt =
this._extractOriginalPrompt(optionsOrPrompt);
Expand Down Expand Up @@ -2836,7 +2845,7 @@
// Set session and user IDs from context for Langfuse spans and execute with proper async scoping
return await this.setLangfuseContextFromOptions(
options,
async () => {

Check warning on line 2848 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / test (20) 

Async arrow function has too many lines (357). Maximum allowed is 300

Check warning on line 2848 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / 🛡️ Code Quality & Security Gate 

Async arrow function has too many lines (357). Maximum allowed is 300
if (
this.conversationMemoryConfig?.conversationMemory
?.mem0Enabled &&
Expand Down Expand Up @@ -3199,6 +3208,28 @@
code: SpanStatusCode.ERROR,
message: error instanceof Error ? error.message : String(error),
});
// Emit generation:end on error so metrics listeners still record the failure.
// Note: variables declared inside try blocks are not accessible in error
// handlers, so we extract what we can from the original input.
const errProvider =
typeof optionsOrPrompt === "object"
? (optionsOrPrompt as GenerateOptions).provider || "unknown"
: "unknown";
const errModel =
typeof optionsOrPrompt === "object"
? (optionsOrPrompt as GenerateOptions).model || "unknown"
: "unknown";
try {
this.emitter.emit("generation:end", {
provider: errProvider,
model: errModel,
responseTime: 0,
error: error instanceof Error ? error.message : String(error),
success: false,
});
} catch (emitError: unknown) {
void emitError; // non-blocking — error event emission is best-effort
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
throw error;
} finally {
generateSpan.end();
Expand Down Expand Up @@ -3637,7 +3668,7 @@
* 4. Fall back to direct provider generation
* 5. Store conversation turn for future context
*/
private async generateTextInternal(

Check warning on line 3671 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / test (20) 

Async method 'generateTextInternal' has too many lines (426). Maximum allowed is 300

Check warning on line 3671 in src/lib/neurolink.ts

View workflow job for this annotation

GitHub Actions / 🛡️ Code Quality & Security Gate 

Async method 'generateTextInternal' has too many lines (426). Maximum allowed is 300
options: TextGenerationOptions,
): Promise<TextGenerationResult> {
return tracers.sdk.startActiveSpan(
Expand Down
7 changes: 6 additions & 1 deletion src/lib/observability/exporters/langfuseExporter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,9 @@ export class LangfuseExporter extends BaseExporter {
name: span.name,
userId: span.attributes["user.id"] as string | undefined,
sessionId: span.attributes["session.id"] as string | undefined,
metadata: span.attributes,
// Only pick safe, non-PII attributes for metadata — intentionally excludes
// input, output, error.stack, and other user content to match Braintrust exporter
metadata: filterSafeMetadata(span.attributes),
release: this.release,
tags: this.extractTags(span),
};
Expand Down Expand Up @@ -249,3 +251,6 @@ export class LangfuseExporter extends BaseExporter {
return tags;
}
}

// Safe metadata filtering imported from shared module to avoid duplication
import { filterSafeMetadata } from "../utils/safeMetadata.js";
8 changes: 6 additions & 2 deletions src/lib/observability/metricsAggregator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,8 +179,12 @@ export class MetricsAggregator {
recordSpan(span: SpanData): void {
// Enforce maximum spans limit
if (this.spans.length >= this.config.maxSpansRetained) {
this.spans.shift(); // Remove oldest span
// Note: We keep aggregated metrics, only raw spans are trimmed
const evicted = this.spans.shift(); // Remove oldest span
// Only trim latencyValues when the evicted span had a duration recorded
if (evicted?.durationMs !== undefined) {
this.latencyValues.shift();
}
Comment on lines 181 to +186

@coderabbitai coderabbitai Bot Mar 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Keep latency trimming tied to the evicted span, not only to max length.

Current logic can leave stale latency samples when the removed span had durationMs but latencyValues.length is still below maxSpansRetained (mixed spans with/without duration), which can skew latency stats.

💡 Suggested fix 
-    if (this.spans.length >= this.config.maxSpansRetained) {
-      this.spans.shift(); // Remove oldest span
-      // Trim latencyValues in sync to prevent unbounded memory growth
-      if (this.latencyValues.length >= this.config.maxSpansRetained) {
-        this.latencyValues.shift();
-      }
+    if (this.spans.length >= this.config.maxSpansRetained) {
+      const removedSpan = this.spans.shift(); // Remove oldest span
+      // Trim latencyValues in sync with the removed span
+      if (removedSpan?.durationMs !== undefined && this.latencyValues.length > 0) {
+        this.latencyValues.shift();
+      }
       // Note: We keep aggregated metrics, only raw spans and latency values are trimmed
     }
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@src/lib/observability/metricsAggregator.ts` around lines 181 - 186, When
evicting the oldest span in MetricsAggregator, remove the corresponding latency
sample for that specific evicted span instead of only trimming latencyValues by
length; change the block that currently calls this.spans.shift() to capture the
removed span (const evicted = this.spans.shift()), then if evicted.durationMs is
defined remove one matching entry from this.latencyValues (e.g., find the index
of evicted.durationMs and splice it out) so latencyValues stays aligned with
spans regardless of spans without durationMs; keep the maxSpansRetained guard
but base latency removal on the evicted span's durationMs rather than only on
this.latencyValues.length.

// Note: We keep aggregated metrics, only raw spans and latency values are trimmed
}

this.spans.push(span);
Expand Down
19 changes: 17 additions & 2 deletions src/lib/observability/spanProcessor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,8 @@ export class RedactionProcessor implements SpanProcessor {
"credentials",
"private_key",
"privateKey",
"stack",
"error.stack",
],
);
this.redactedValue = config?.redactedValue ?? "[REDACTED]";
Expand Down Expand Up @@ -308,11 +310,24 @@ export class BatchProcessor implements SpanProcessor {
}, this.flushIntervalMs);
}

// Note: flush() is intentionally synchronous. The onBatchReady callback is
// typed as `(spans: SpanData[]) => void` — callers must not pass async
// exporters. If async export is needed, the callback should handle its own
// error reporting (e.g. fire-and-forget with promise error handlers).
private flush(): void {
if (this.batch.length > 0 && this.onBatchReady) {
const spans = [...this.batch];
this.batch = [];
this.onBatchReady(spans);
try {
this.onBatchReady(spans);
this.batch = [];
} catch (flushError: unknown) {
// Keep spans for next flush attempt, but cap backlog growth
void flushError; // acknowledged — error is expected during exporter outages
const maxBacklog = this.batchSize * 20;
if (this.batch.length > maxBacklog) {
this.batch = this.batch.slice(this.batch.length - maxBacklog);
}
}
Comment on lines +320 to +330

@coderabbitai coderabbitai Bot Mar 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "1) Locate BatchProcessor constructions and onBatchReady assignments"
rg -nP --type=ts -C3 'new\s+BatchProcessor\s*\(|onBatchReady\s*:' src

echo
echo "2) Detect explicitly async onBatchReady callbacks"
rg -nP --type=ts -C2 'onBatchReady\s*:\s*async\b' src

echo
echo "Expected result:"
echo "- If any async onBatchReady exists, current flush() path can drop spans on rejection."
echo "- If none exist today, this is still a latent correctness bug for future callsites."

Repository: juspay/neurolink

Length of output: 127

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Check if the file exists and get its size
echo "=== Checking file existence and size ==="
if [ -f "src/lib/observability/spanProcessor.ts" ]; then
  wc -l src/lib/observability/spanProcessor.ts
  echo "File exists"
else
  echo "File not found at src/lib/observability/spanProcessor.ts"
  # Try to find the file
  echo "Searching for spanProcessor files..."
  fd -e ts spanProcessor
fi

Repository: juspay/neurolink

Length of output: 155

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read lines around 316-325 from the spanProcessor file
echo "=== Code at lines 316-325 ==="
sed -n '310,335p' src/lib/observability/spanProcessor.ts

echo
echo "=== Looking for flush method definition ==="
rg -n 'flush\s*\(' src/lib/observability/spanProcessor.ts | head -5

echo
echo "=== Looking for onBatchReady definition ==="
rg -n 'onBatchReady' src/lib/observability/spanProcessor.ts | head -10

Repository: juspay/neurolink

Length of output: 1235

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Search for BatchProcessor class definition and onBatchReady signature
echo "=== Search for class definition ==="
rg -n 'class\s+BatchProcessor' src/lib/observability/spanProcessor.ts

echo
echo "=== Get full context around flush method ==="
# Use ast-grep to find flush method with context
ast-grep --pattern $'private flush() {
  $$$
}'

echo
echo "=== Search for any async callbacks ==="
rg -n 'async\s*\(' src/lib/observability/spanProcessor.ts | head -10

Repository: juspay/neurolink

Length of output: 200

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Where is BatchProcessor instantiated? ==="
rg -n 'new\s+BatchProcessor' src --type=ts

echo
echo "=== All onBatchReady assignments/usages ==="
rg -n 'onBatchReady' src --type=ts -B2 -A2

echo
echo "=== Check if any async callbacks exist in the codebase ==="
# Look for patterns where async callbacks might be passed
rg -n 'async\s*\(\s*spans' src --type=ts
</parameter>

Repository: juspay/neurolink

Length of output: 105

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read the full config type and constructor
echo "=== Full BatchProcessor constructor and config type ==="
sed -n '272,295p' src/lib/observability/spanProcessor.ts

echo
echo "=== Check shutdown behavior ==="
sed -n '331,350p' src/lib/observability/spanProcessor.ts
</parameter>

Repository: juspay/neurolink

Length of output: 1515

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Search for BatchProcessor usage across entire codebase ==="
rg 'BatchProcessor' src --type=ts

echo
echo "=== Search for any async function passed as callback ==="
rg 'onBatchReady\s*[:=]' src --type=ts -A1

Repository: juspay/neurolink

Length of output: 447

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Full shutdown method ==="
sed -n '331,345p' src/lib/observability/spanProcessor.ts

echo
echo "=== Check SpanProcessor interface to understand expectations ==="
rg -n 'interface SpanProcessor' src --type=ts -A8

Repository: juspay/neurolink

Length of output: 1150

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== All flush() call sites ==="
sed -n '295,340p' src/lib/observability/spanProcessor.ts

echo
echo "=== Check if flush is called from timer/synchronous context ==="
rg -B5 'this.flush()' src/lib/observability/spanProcessor.ts

Repository: juspay/neurolink

Length of output: 2071

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Verify TypeScript behavior: can async be assigned to void-returning type? ==="
echo "Checking if there are any TypeScript compiler directives or strict settings..."
fd -e json tsconfig | head -3 | xargs cat 2>/dev/null | head -30

echo
echo "=== Check if BatchProcessor is exported for public use ==="
rg -n 'export.*BatchProcessor' src --type=ts
</parameter>

Repository: juspay/neurolink

Length of output: 1307

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Check the learning from previous PR about Promise handling ==="
echo "The learning suggests using .then()/.catch() for backward compatibility..."
echo ""
echo "=== Look for similar Promise handling patterns in the codebase ==="
rg 'onBatchReady\s*\(.*\)\.then|\.catch' src --type=ts

echo
echo "=== Check if there are examples of .then/.catch handling elsewhere ==="
rg '\.then\(|\.catch\(' src/lib/observability --type=ts -B2 -A2 | head -40

Repository: juspay/neurolink

Length of output: 5878

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Verify: Can an async function be assigned to a void-returning type in TypeScript strict mode? ==="
cat > /tmp/test.ts << 'EOF'
// Test TypeScript assignability with strict mode
type SyncCallback = (data: string[]) => void;
type AsyncCallback = (data: string[]) => Promise<void>;

const asyncFn: AsyncCallback = async (data) => { console.log(data); };
const syncFn: SyncCallback = asyncFn; // Does this error in strict mode?
EOF
echo "TypeScript strict mode allows assigning async functions to void-returning types"
echo "This is because Promise<void> is compatible with void in JavaScript execution"

echo
echo "=== Confirm the actual issue: unhandled promise rejection scenario ==="
echo "Current code path:"
echo "1. Line 317: this.onBatchReady(spans)  // If async, returns Promise but not awaited"
echo "2. Line 318: this.batch = []          // Clears batch immediately"
echo "3. If Promise rejects later: catch block doesn't execute, spans lost"

Repository: juspay/neurolink

Length of output: 629

flush() loses spans when onBatchReady is async.

Line 317 calls onBatchReady(spans) without awaiting. Although the type signature declares void, TypeScript allows assigning async functions to void-returning types. If an async callback is passed and rejects, the catch block (line 320) never runs, and line 318 has already cleared the batch, losing all spans.

This is critical during shutdown() (line 331), which is async but calls flush() synchronously at line 334. If the callback rejects, spans are lost without warning.

Fix: Use .then().catch() pattern to handle both sync and async callbacks while preserving backward compatibility:

Suggested fix (backward-compatible Promise handling) 
  private flush(): void {
    if (this.batch.length > 0 && this.onBatchReady) {
      const spans = [...this.batch];
      this.batch = [];
      
-     try {
-       this.onBatchReady(spans);
-     } catch {
-       // Keep spans for next flush attempt, but cap backlog growth
-       const maxBacklog = this.batchSize * 20;
-       if (this.batch.length > maxBacklog) {
-         this.batch = this.batch.slice(this.batch.length - maxBacklog);
-       }
+     const result = this.onBatchReady(spans);
+     if (result instanceof Promise) {
+       result.catch(() => {
+         // Keep spans for next flush attempt on rejection, but cap backlog growth
+         this.batch.unshift(...spans);
+         const maxBacklog = this.batchSize * 20;
+         if (this.batch.length > maxBacklog) {
+           this.batch = this.batch.slice(this.batch.length - maxBacklog);
+         }
+       });
+     }
     }
   }

This keeps flush() synchronous for callers while safely handling async callbacks.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@src/lib/observability/spanProcessor.ts` around lines 316 - 325, flush()
currently calls onBatchReady(spans) without handling promise rejections, which
can lose spans if onBatchReady is async and rejects; change flush() to wrap the
callback with Promise.resolve(this.onBatchReady(spans)).then(() => { clear
this.batch }) .catch(() => { keep this.batch but cap growth using this.batchSize
* 20 and slice to the newest entries }), ensuring synchronous callers of flush()
are not forced to await but both sync throws and async rejections are handled;
reference the methods/fields flush(), onBatchReady, this.batch and
this.batchSize (and consider shutdown() behavior) when applying the change.

}
}

Expand Down
31 changes: 31 additions & 0 deletions src/lib/observability/utils/safeMetadata.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/**
* Safe metadata filtering for observability exporters.
*
* Only these attribute keys are forwarded to third-party backends as trace
* metadata. User prompts (input), LLM responses (output), error stacks, and
* any other potentially sensitive data are excluded to prevent PII leaks.
*/

import type { SpanAttributes } from "../types/spanTypes.js";

// Only ai.* keys are forwarded as metadata. Stream metrics (chunk_count,
// content_length) should be accessed via span attributes directly, not via
// metadata sent to third-party backends.
export const SAFE_METADATA_KEYS = new Set([
"ai.provider",
"ai.model",
"ai.temperature",
"ai.max_tokens",
]);
Comment thread
coderabbitai[bot] marked this conversation as resolved.

export function filterSafeMetadata(
attributes: SpanAttributes,
): Record<string, unknown> {
const filtered: Record<string, unknown> = {};
for (const key of SAFE_METADATA_KEYS) {
if (attributes[key] !== undefined) {
filtered[key] = attributes[key];
}
}
return filtered;
}
7 changes: 6 additions & 1 deletion src/lib/observability/utils/spanSerializer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,9 @@ export class SpanSerializer {
name: span.name,
startTime: span.startTime,
endTime: span.endTime,
metadata: { ...span.attributes },
// Only pick safe, non-PII attributes for metadata — intentionally excludes
// input, output, error.stack, and other user content for PII safety
metadata: filterSafeMetadata(span.attributes),
level: span.status === SpanStatus.ERROR ? "ERROR" : "DEFAULT",
statusMessage: span.statusMessage,
input: span.attributes["input"],
Expand Down Expand Up @@ -389,3 +391,6 @@ export class SpanSerializer {
});
}
}

// Safe metadata filtering imported from shared module to avoid duplication
import { filterSafeMetadata } from "./safeMetadata.js";
8 changes: 2 additions & 6 deletions src/lib/providers/googleVertex.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1721,9 +1721,7 @@ export class GoogleVertexProvider extends BaseProvider {
},
{
role: "model",
parts: [
{ text: "Understood. I will follow these instructions." },
],
parts: [{ text: "OK" }],
},
...currentContents,
];
Expand Down Expand Up @@ -1999,9 +1997,7 @@ export class GoogleVertexProvider extends BaseProvider {
},
{
role: "model",
parts: [
{ text: "Understood. I will follow these instructions." },
],
parts: [{ text: "OK" }],
},
...currentContents,
];
Expand Down
7 changes: 5 additions & 2 deletions test/zod-schema-test-function.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,9 +394,12 @@ export async function testComplexZodSchemaMultiProvider(
const providerName = providerOverride || "vertex";
const modelName = modelOverride || undefined;

// Check if this is a Gemini MODEL - Gemini cannot use tools + JSON schema together
// Check if this is a Gemini MODEL - Gemini cannot use tools + JSON schema together.
// This is a documented Gemini limitation, not a bug in the SDK.
// Note: provider name alone is not sufficient - Vertex can run Claude too.
// When an explicit model is provided, we check its name. When no model is given,
// google-ai, googleAiStudio, and vertex all default to a Gemini model, so we
// treat them as Gemini. If someone passes e.g. --provider=vertex --model=claude-*,
// the modelName check will NOT match "gemini" and the test will correctly run.
const isGeminiModel =
modelName?.toLowerCase().includes("gemini") ||
(!modelName &&
Expand Down
Loading