Repository navigation
Conversation
- Introduced OpenTelemetry instrumentation in a new file `instrumentation.ts` to enable tracing for the service. - Configured NodeSDK with OTLPTraceExporter, BatchSpanProcessor, and W3CTraceContextPropagator for enhanced tracing capabilities. - Created detailed session documentation for `CONTINUITY_ses_45e3.md`, outlining goals, constraints, progress, and next steps for implementing lifecycle hooks and OpenTelemetry tracing in `src/mastra/tools/e2b.ts`. - Documented progress and analysis in `CONTINUITY_ses_45e6.md`, focusing on standardizing OpenTelemetry patterns across tools and ensuring proper lifecycle hook implementation. - Conducted a comprehensive analysis of the AgentStack project in `CONTINUITY_ses_45fc.md`, detailing the project's structure, technology stack, and current state for onboarding new team members.
Learn moreAll Green is an AI agent that automatically: ✅ Addresses code review comments ✅ Fixes failing CI checks ✅ Resolves merge conflicts |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
|
🤖 Hi @ssdeanx, I've received your request, and I'm working on it now! You can track my progress in the logs for more details. |
|
Caution Review failedThe pull request is closed. Summary by CodeRabbitRelease Notes
✏️ Tip: You can customize this high-level summary in your review settings. WalkthroughAdds comprehensive OpenTelemetry instrumentation across the platform: new Node.js SDK initialization, observability configuration with bridge and redaction, logging enhanced with trace context, tool execute signatures updated to accept context, and database vector search parameters tuned from HNSW to IVFFlat indexing. Changes
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant Tool as E2B Tool
participant Tracer as OpenTelemetry Tracer
participant Span as Active Span
participant Logger as Logger (w/ Trace)
participant Sandbox as E2B Sandbox
App->>Tool: execute(inputData, context)
Note over Tool: RequestContext available via context param
Tool->>Tracer: Start span for tool operation
Tracer-->>Span: Active span created
Tool->>Span: Add attributes (tool.id, input size)
Tool->>Logger: Log tool start + trace context
Logger->>Logger: Attach span context + baggage
Tool->>Sandbox: Execute tool operation
Sandbox-->>Tool: Result/output
Tool->>Span: Record span event
Tool->>Logger: Log tool completion + metrics
Logger->>Logger: Append to workflow.log with trace
Tool-->>App: Return result with trace metadata
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
✨ Finishing touches
🧪 Generate unit tests (beta)
📜 Recent review detailsConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (10)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.
Summary of ChangesHello @ssdeanx, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the project's observability and maintainability by integrating OpenTelemetry for distributed tracing and metrics across key components, particularly within the E2B tools. It standardizes how tools report their activity and progress, ensuring a consistent and detailed view of operations. Additionally, it refines the overall observability setup and provides crucial documentation for project understanding and future development. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
|
🤖 I'm sorry @ssdeanx, but I was unable to process your request. Please see the logs for more details. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces comprehensive OpenTelemetry (OTEL) instrumentation and logging enhancements across the codebase, particularly for E2B sandbox tools. A new instrumentation.ts file sets up the OTEL NodeSDK, while logger.ts is updated with OTEL helpers to enrich logs with trace context and baggage, although the main log formatter for this is currently commented out. The Mastra observability configuration in index.ts has been refactored to use a DefaultExporter with sensitive data filtering and an OtelBridge, replacing a previous OtelExporter setup. The package.json reflects a version bump and numerous dependency updates, including new OTEL-related packages. In pg-storage.ts, PostgreSQL connection limits and vector index configurations were adjusted, shifting from HNSW to IVFFlat for some indexes. Most significantly, the e2b.ts file underwent a major refactoring to integrate OTEL tracing and metrics into all its tools, adding spans, attributes, events, and error handling, and standardizing tool lifecycle hooks. However, the readFile tool was accidentally removed, the runCode tool's implementation was fundamentally changed from sandbox.runCode to sandbox.commands.run leading to breaking changes in output and supported options, and the contextManager in instrumentation.ts is set to undefined, which disables asynchronous context propagation for tracing. Additionally, there is commented-out code in the observability configuration and a potential loss of baggage metadata in the mergeAndSetBaggage helper.
| }) | ||
| ), | ||
| ], | ||
| contextManager: undefined, |
There was a problem hiding this comment.
Setting contextManager to undefined disables automatic asynchronous context propagation. This means that trace.getActiveSpan() and otelContext.active() will not work correctly across async/await calls, which will break the tracing instrumentation implemented in logger.ts and e2b.ts. To fix this, you should remove this line entirely to let the SDK use the default AsyncLocalStorageContextManager.
| import { createTool } from '@mastra/core/tools'; | ||
| import z from 'zod'; | ||
| import { log } from '../config/logger'; | ||
| import { FilesystemEventType, FileType, Sandbox } from '@e2b/code-interpreter' |
There was a problem hiding this comment.
The readFile tool, which existed in the previous version of this file, appears to have been accidentally removed during this refactoring. All other tools were preserved and instrumented, but readFile is now missing, which constitutes a breaking change.
| // log(object) { | ||
| // return attachTraceAndBaggage(object) | ||
| // }, |
There was a problem hiding this comment.
The log formatter, which is responsible for calling attachTraceAndBaggage to enrich all logs with trace context, is commented out. This means that no logs will be automatically correlated with traces, which undermines a key benefit of this instrumentation. Please uncomment this block to enable automatic log enrichment.
| // log(object) { | |
| // return attachTraceAndBaggage(object) | |
| // }, | |
| log(object) { | |
| return attachTraceAndBaggage(object) | |
| }, |
| export const runCode = createTool({ | ||
| id: 'runCode', | ||
| description: 'Execute code in the e2b sandbox', | ||
| inputSchema: z.object({ | ||
| sandboxId: z | ||
| .string() | ||
| .describe('The sandboxId for the sandbox to run code in'), | ||
| code: z.string().describe('The code to execute'), | ||
| runCodeOpts: z | ||
| .object({ | ||
| language: z | ||
| .enum(['python', 'javascript', 'typescript']) | ||
| .default('python') | ||
| .describe('The programming language of the code'), | ||
| }) | ||
| .optional() | ||
| .describe('Options for code execution'), | ||
| }), | ||
| outputSchema: z.object({ | ||
| execution: z.string().describe('JSON string containing execution results'), | ||
| }), | ||
| execute: async (inputData, context) => { | ||
| const span = tracer.startSpan('runCode') | ||
| const startTime = Date.now() | ||
|
|
||
| try { | ||
| logToolStart('runCode', inputData, context) | ||
|
|
||
| const sandbox = await Sandbox.connect(inputData.sandboxId) | ||
|
|
||
| // Execute code using the appropriate interpreter based on language | ||
| const language = inputData.runCodeOpts?.language ?? 'python' | ||
| let command: string | ||
|
|
||
| if (language === 'python') { | ||
| command = `python -c "${inputData.code.replace(/"/g, '\\"')}"` | ||
| } else if (language === 'javascript') { | ||
| command = `node -e "${inputData.code.replace(/"/g, '\\"')}"` | ||
| } else if (language === 'typescript') { | ||
| // For TypeScript, we need to compile and run | ||
| command = `npx tsx -e "${inputData.code.replace(/"/g, '\\"')}"` | ||
| } else { | ||
| throw new Error(`Unsupported language: ${language}`) | ||
| } | ||
|
|
||
| const result = await sandbox.commands.run(command) | ||
|
|
||
| span.setStatus({ code: SpanStatusCode.OK }) | ||
| span.setAttribute('sandbox.id', inputData.sandboxId) | ||
| span.setAttribute('code.language', language) | ||
| span.setAttribute('code.length', inputData.code.length) | ||
|
|
||
| // Format result to match expected structure with logs.stdout and logs.stderr as arrays | ||
| const executionResult = { | ||
| logs: { | ||
| stdout: result.stdout ? result.stdout.split('\n') : [], | ||
| stderr: result.stderr ? result.stderr.split('\n') : [], | ||
| }, | ||
| exitCode: result.exitCode, | ||
| success: result.exitCode === 0, | ||
| } | ||
|
|
||
| const output = { | ||
| execution: JSON.stringify(executionResult), | ||
| } | ||
|
|
||
| logToolComplete('runCode', output, Date.now() - startTime, context) | ||
| return output | ||
| } catch (e) { | ||
| span.setStatus({ | ||
| code: SpanStatusCode.ERROR, | ||
| message: e instanceof Error ? e.message : String(e), | ||
| }) | ||
| logToolError('runCode', e, context) | ||
| const errObj = | ||
| e instanceof Error | ||
| ? { message: e.message, stack: e.stack } | ||
| : { message: String(e) } | ||
| throw new Error(JSON.stringify(errObj)) | ||
| } finally { | ||
| span.end() | ||
| } | ||
| }, | ||
| onInputStart: ({ toolCallId, messages, abortSignal }) => { | ||
| log.info('Run code tool input streaming started', { | ||
| toolCallId, | ||
| messageCount: messages.length, | ||
| abortSignal: abortSignal?.aborted, | ||
| hook: 'onInputStart', | ||
| }) | ||
| }, | ||
| onInputAvailable: ({ input, toolCallId, messages, abortSignal }) => { | ||
| log.info('Run code received complete input', { | ||
| toolCallId, | ||
| messageCount: messages.length, | ||
| sandboxId: input.sandboxId, | ||
| codeLength: input.code.length, | ||
| language: input.runCodeOpts?.language ?? 'python', | ||
| abortSignal: abortSignal?.aborted, | ||
| hook: 'onInputAvailable', | ||
| }) | ||
| }, | ||
| onOutput: ({ output, toolCallId, toolName, abortSignal }) => { | ||
| log.info('Run code completed', { | ||
| toolCallId, | ||
| toolName, | ||
| executionLength: output.execution.length, | ||
| abortSignal: abortSignal?.aborted, | ||
| hook: 'onOutput', | ||
| }) | ||
| }, | ||
| }) |
There was a problem hiding this comment.
The implementation of the runCode tool has been fundamentally changed from using sandbox.runCode to sandbox.commands.run. This introduces several breaking changes:
- Different Output: The output format is now a simple object with
stdout/stderrarrays, which differs from the richer object returned by the originalrunCodemethod. - Unsupported Options: The
runCodeOptsno longer supportsenvs,timeoutMS, orrequestTimeoutMs. - Language Aliases: The supported languages have changed (e.g.,
jsis no longer an alias forjavascript).
This change is likely to break integrations that rely on this tool. It's recommended to revert to using sandbox.runCode and wrap it with the new tracing and logging logic to maintain compatibility.
| import { W3CTraceContextPropagator } from "@opentelemetry/core"; | ||
|
|
||
| const sdk = new NodeSDK({ | ||
| serviceName: "my-service", |
There was a problem hiding this comment.
The serviceName is hardcoded. It's a best practice to make this configurable via an environment variable to easily identify the service in different observability environments (e.g., development, staging, production).
| serviceName: "my-service", | |
| serviceName: process.env.OTEL_SERVICE_NAME ?? "agentstack", |
| if (!span) { | ||
| return { traceId: undefined, spanId: undefined } | ||
| } | ||
| const sc = (span as any).spanContext ? (span as any).spanContext() : undefined |
There was a problem hiding this comment.
The cast (span as any) is unnecessary and bypasses type safety. The spanContext() method is a standard part of the Span interface from @opentelemetry/api. You can call it directly on the span object.
| const sc = (span as any).spanContext ? (span as any).spanContext() : undefined | |
| const sc = span.spanContext(); |
| const BaggageToObject = (b: unknown) => { | ||
| if (b === null || b === undefined) { | ||
| return {} | ||
| } | ||
| const obj: Record<string, string> = {} | ||
| for (const [k, v] of (b as any).getAllEntries()) { | ||
| obj[k] = String(v.value) | ||
| } | ||
| return obj | ||
| } |
There was a problem hiding this comment.
The BaggageToObject function takes b as unknown and then casts it to any. This can be made more type-safe by typing b as Baggage | undefined | null and avoiding the any cast, as getAllEntries() is a standard method on the Baggage interface.
| const BaggageToObject = (b: unknown) => { | |
| if (b === null || b === undefined) { | |
| return {} | |
| } | |
| const obj: Record<string, string> = {} | |
| for (const [k, v] of (b as any).getAllEntries()) { | |
| obj[k] = String(v.value) | |
| } | |
| return obj | |
| } | |
| const BaggageToObject = (b: Baggage | undefined | null) => { | |
| if (!b) { | |
| return {} | |
| } | |
| const obj: Record<string, string> = {} | |
| for (const [k, v] of b.getAllEntries()) { | |
| obj[k] = String(v.value) | |
| } | |
| return obj | |
| } |
| let data: { | ||
| workflowId: string | ||
| input: Record<string, unknown> | ||
| timestamp: string | ||
| _trace?: Record<string, unknown> | ||
| _Baggage?: Record<string, string> | ||
| } = { | ||
| workflowId, | ||
| input, | ||
| timestamp: new Date().toISOString(), | ||
| } | ||
| data = attachTraceAndBaggage(data) as typeof data |
There was a problem hiding this comment.
There is significant code duplication across the logWorkflowStart, logWorkflowEnd, logStepStart, and other logging functions. Each one re-defines a data type and manually calls attachTraceAndBaggage. If you enable the Pino log formatter (as suggested in another comment), it will handle this automatically for all logs, and you can remove the manual enrichment from each of these functions, greatly simplifying the code.
| // strategy: 'realtime', | ||
| // maxBatchSize: 500, | ||
| // maxBufferSize: 1000, | ||
| // maxBatchWaitMs: 6000, | ||
| maxRetries: 5, | ||
| } | ||
| ), // Studio access0 | ||
| ], | ||
| includeInternalSpans: true, | ||
|
|
||
| bridge: new OtelBridge( | ||
| { | ||
| name: "mastra-otel-bridge", | ||
| // otelTracer: { | ||
| // startSpan: (name: string, options: SpanOptions | undefined) => { | ||
| // const tracer = trace.getTracer("mastra-otel-bridge-tracer"); | ||
| // return tracer.startSpan(name, options); | ||
| // }, | ||
| // }, | ||
| logger: log, | ||
| logLevel: 'debug', | ||
| logLevel: 'info', | ||
| // batchsize: 500, | ||
| timeout: 50000, | ||
| serializationOptions: { | ||
| maxStringLength: 1024, // Maximum length for string values (default: 1024) | ||
| maxDepth: 6, // Maximum depth for nested objects (default: 6) | ||
| maxArrayLength: 50, // Maximum number of items in arrays (default: 50) | ||
| maxObjectKeys: 50, // Maximum number of keys in objects (default: 50) | ||
| }, | ||
| resourceAttributes: { | ||
| 'deployment.environment': 'dev', | ||
| }, | ||
| }), | ||
| ], | ||
| } | ||
| ), | ||
| }, | ||
| // exporters: [ | ||
| // new DefaultExporter(), // Studio access | ||
|
|
||
| // new OtelExporter({ | ||
| // provider: { | ||
| // traceloop: { | ||
| // apiKey: process.env.TRACELOOP_API_KEY, | ||
| // destinationId: 'my-destination', | ||
| // }, | ||
| // }, | ||
| // batchSize: 500, | ||
| // timeout: 50000, | ||
| // logger: log, | ||
| // logLevel: 'debug', | ||
| // resourceAttributes: { | ||
| // 'deployment.environment': 'dev', | ||
| // }, | ||
| // }), | ||
| // ], |
There was a problem hiding this comment.
There is a large amount of commented-out code in the observability configuration. This makes the file difficult to read and maintain. Please remove this dead code before merging.
| for (const k of Object.keys(entries)) { | ||
| merged[k] = { value: entries[k] } | ||
| } |
There was a problem hiding this comment.
In the mergeAndSetBaggage helper, when adding or updating baggage entries, you are creating a new entry with only a value. If an entry with the same key already existed and had associated metadata, that metadata will be lost. This could lead to a loss of context if other services rely on that metadata.
There was a problem hiding this comment.
Pull request overview
This pull request introduces OpenTelemetry instrumentation and enhances lifecycle hooks for E2B sandbox tools. The changes include comprehensive tracing, metrics collection, and improved observability across the application.
Key Changes
- Added OpenTelemetry auto-instrumentation with OTLP exporter for distributed tracing
- Refactored 13 E2B tools with enhanced OpenTelemetry spans, attributes, events, and metrics
- Integrated trace context and baggage into the logging system for correlated observability
- Updated observability configuration with sampling strategies and sensitive data filtering
Reviewed changes
Copilot reviewed 10 out of 11 changed files in this pull request and generated 33 comments.
Show a summary per file
| File | Description |
|---|---|
| instrumentation.ts | New OpenTelemetry NodeSDK configuration with auto-instrumentation |
| src/mastra/tools/e2b.ts | Major refactor (823→1697 lines): Added comprehensive OTEL tracing, metrics, lifecycle hooks, and helper functions |
| src/mastra/config/logger.ts | Enhanced with trace context and baggage propagation |
| src/mastra/index.ts | Updated observability config with DefaultExporter, sampling, and OtelBridge |
| src/mastra/config/pg-storage.ts | Adjusted connection pool size, IVFFlat configuration, and title generation |
| src/mastra/tools/document-chunking.tool.ts | Minor: Added type keyword to RequestContext import |
| package.json | Version bump and dependency updates |
| thoughts/ledgers/*.md | Session documentation for development context |
| const handle = await sandbox.files.watchDir( | ||
| inputData.path, | ||
| async (event) => { | ||
| events.push({ | ||
| type: event.type, | ||
| name: event.name, | ||
| timestamp: new Date().toISOString(), | ||
| }) | ||
| }, | ||
| { | ||
| recursive: inputData.recursive, | ||
| } | ||
| ) | ||
|
|
||
| // Watch for the specified duration | ||
| await new Promise((resolve) => | ||
| setTimeout(resolve, inputData.watchDuration) | ||
| ) | ||
|
|
||
| // Stop watching | ||
| await handle.stop() | ||
|
|
||
| span.setStatus({ code: SpanStatusCode.OK }) | ||
| span.setAttribute('sandbox.id', inputData.sandboxId) | ||
| span.setAttribute('watch.path', inputData.path) | ||
| span.setAttribute('watch.recursive', inputData.recursive) | ||
| span.setAttribute('watch.duration', inputData.watchDuration) | ||
| span.setAttribute('events.count', events.length) | ||
|
|
||
| const result = { | ||
| watchStarted: true, | ||
| path: inputData.path, | ||
| events, | ||
| } | ||
|
|
||
| logToolComplete( | ||
| 'watchDirectory', | ||
| result, | ||
| Date.now() - startTime, | ||
| context | ||
| ) | ||
| return result | ||
| } catch (e) { | ||
| span.setStatus({ | ||
| code: SpanStatusCode.ERROR, | ||
| message: e instanceof Error ? e.message : String(e), | ||
| }) | ||
| logToolError('watchDirectory', e, context) | ||
| const errObj = | ||
| e instanceof Error | ||
| ? { message: e.message, stack: e.stack } | ||
| : { message: String(e) } | ||
| throw new Error(JSON.stringify(errObj)) | ||
| } finally { | ||
| span.end() | ||
| } |
There was a problem hiding this comment.
Potential memory leak in watchDirectory tool. The tool creates a file watcher at line 1323 that captures events in a closure, but if an error occurs during the watch setup or execution, the watcher handle may not be properly cleaned up. The handle.stop() call is only in the success path (line 1343), not in the catch block. If an exception occurs after the watcher is created but before it's stopped, the watcher will continue running and leaking resources.
| try { | ||
| const info = await sandbox.files.getInfo(inputData.path) | ||
|
|
||
| span.setStatus({ code: SpanStatusCode.OK }) | ||
| span.setAttribute('sandbox.id', inputData.sandboxId) | ||
| span.setAttribute('file.path', inputData.path) | ||
| span.setAttribute('file.exists', true) | ||
| if (info.type !== null && info.type !== undefined) { | ||
| span.setAttribute('file.type', info.type) | ||
| } | ||
|
|
||
| const result = { | ||
| exists: true, | ||
| path: inputData.path, | ||
| type: info.type, | ||
| } | ||
|
|
||
| logToolComplete( | ||
| 'checkFileExists', | ||
| result, | ||
| Date.now() - startTime, | ||
| context | ||
| ) | ||
| return result | ||
| } catch { | ||
| // If getInfo fails, the file doesn't exist | ||
| span.setStatus({ code: SpanStatusCode.OK }) | ||
| span.setAttribute('sandbox.id', inputData.sandboxId) | ||
| span.setAttribute('file.path', inputData.path) | ||
| span.setAttribute('file.exists', false) | ||
|
|
||
| const result = { | ||
| exists: false, | ||
| path: inputData.path, | ||
| } | ||
|
|
||
| logToolComplete( | ||
| 'checkFileExists', | ||
| result, | ||
| Date.now() - startTime, | ||
| context | ||
| ) | ||
| return result | ||
| } |
There was a problem hiding this comment.
Inconsistent error handling pattern in checkFileExists. The inner try-catch block at line 1027 catches errors from sandbox.files.getInfo() silently without any error recording or logging (line 1051), treating file-not-found as a normal case. However, if this inner catch triggers due to a different error (e.g., network failure, permission issues), it will be incorrectly interpreted as "file doesn't exist". The implementation should distinguish between "file not found" errors and other types of errors.
| 'tool.input_size': JSON.stringify(input).length, | ||
| 'user.id': toolContext?.requestContext?.userId, | ||
| }) | ||
|
|
||
| // Merge into baggage for downstream services and traces | ||
| mergeAndSetBaggage({ 'tool.name': toolName, 'tool.start_time': Date.now().toString() }) | ||
|
|
||
| // Also set attributes on the span so it's available in OTEL exporters | ||
| try { | ||
| span.setAttribute('tool.name', toolName) | ||
| span.setAttribute('tool.input_size', JSON.stringify(input).length) | ||
| } catch (err) { | ||
| // ignore if span does not support setAttribute | ||
| } | ||
| } | ||
|
|
||
| // Include trace info and baggage in the structured log | ||
| const traceInfo = getActiveSpanContextInfo() | ||
| const bag = propagation.getBaggage(otelContext.active()) | ||
| logToolExecution(toolName, { originalInput: input, _trace: traceInfo, _Baggage: BaggageToObject(bag), _userId: toolContext?.requestContext?.userId }) | ||
| } | ||
|
|
||
| const logToolComplete = ( | ||
| toolName: string, | ||
| output: any, | ||
| duration: number, | ||
| toolContext?: any | ||
| ) => { | ||
| const span = trace.getActiveSpan() | ||
| if (span) { | ||
| // Add span event for tool completion | ||
| span.addEvent('tool.complete', { | ||
| 'tool.name': toolName, | ||
| 'tool.duration_ms': duration, | ||
| 'tool.output_size': JSON.stringify(output).length, | ||
| 'user.id': toolContext?.requestContext?.userId, | ||
| }) | ||
|
|
||
| // Update baggage | ||
| mergeAndSetBaggage({ 'tool.duration': duration.toString(), 'tool.completed_at': Date.now().toString() }) | ||
|
|
||
| // Set attributes on span | ||
| try { | ||
| span.setAttribute('tool.duration_ms', duration) | ||
| span.setAttribute('tool.output_size', JSON.stringify(output).length) | ||
| } catch (err) { | ||
| // ignore | ||
| } | ||
| } | ||
|
|
||
| // Record metrics | ||
| toolExecutionCounter.add(1, { | ||
| 'tool.name': toolName, | ||
| 'tool.status': 'success', | ||
| 'user.id': toolContext?.requestContext?.userId, | ||
| }) | ||
| toolExecutionDuration.record(duration, { | ||
| 'tool.name': toolName, | ||
| 'tool.status': 'success', | ||
| 'user.id': toolContext?.requestContext?.userId, | ||
| }) | ||
|
|
||
| const traceInfo = getActiveSpanContextInfo() | ||
| const bag = propagation.getBaggage(otelContext.active()) | ||
|
|
||
| logToolExecution( | ||
| toolName, | ||
| { _trace: traceInfo, _Baggage: BaggageToObject(bag), _userId: toolContext?.requestContext?.userId }, | ||
| output | ||
| ) | ||
| } | ||
|
|
||
| const logToolError = (toolName: string, error: any, context?: any) => { | ||
| const span = trace.getActiveSpan() | ||
| if (span) { | ||
| // Add span event for tool error | ||
| span.addEvent('tool.error', { | ||
| 'tool.name': toolName, | ||
| 'error.message': error instanceof Error ? error.message : String(error), | ||
| 'error.stack': error instanceof Error ? error.stack : undefined, | ||
| 'user.id': context?.requestContext?.userId | ||
| }) | ||
|
|
||
| // Record error in span | ||
| span.recordException(error instanceof Error ? error : new Error(String(error))) | ||
| } | ||
|
|
||
| // Record error metrics | ||
| toolExecutionCounter.add(1, { | ||
| 'tool.name': toolName, | ||
| 'tool.status': 'error', | ||
| 'error.type': error instanceof Error ? error.constructor.name : 'Unknown' | ||
| }) | ||
|
|
||
| logError(toolName, error, { | ||
| toolName, | ||
| userId: context?.requestContext?.userId | ||
| }) | ||
| } | ||
|
|
||
| export const createSandbox = createTool({ | ||
| id: 'createSandbox', | ||
| description: 'Create an e2b sandbox', | ||
| inputSchema: z.object({ | ||
| metadata: z.record(z.string(), z.string()).optional().describe('Custom metadata for the sandbox'), | ||
| envs: z.record(z.string(), z.string()).optional().describe(` | ||
| id: 'createSandbox', | ||
| description: 'Create an e2b sandbox', | ||
| inputSchema: z.object({ | ||
| metadata: z | ||
| .record(z.string(), z.string()) | ||
| .optional() | ||
| .describe('Custom metadata for the sandbox'), | ||
| envs: z.record(z.string(), z.string()).optional().describe(` | ||
| Custom environment variables for the sandbox. | ||
| Used when executing commands and code in the sandbox. | ||
| Can be overridden with the \`envs\` argument when executing commands or code. | ||
| `), | ||
| timeoutMS: z.number().optional().describe(` | ||
| timeoutMS: z.number().optional().describe(` | ||
| Timeout for the sandbox in **milliseconds**. | ||
| Maximum time a sandbox can be kept alive is 24 hours (86_400_000 milliseconds) for Pro users and 1 hour (3_600_000 milliseconds) for Hobby users. | ||
| @default 300_000 // 5 minutes | ||
| `), | ||
| }), | ||
| outputSchema: z.object({ | ||
| sandboxId: z.string(), | ||
| }), | ||
| onInputStart: ({ toolCallId, messages, abortSignal }) => { | ||
| log.info('Create sandbox tool input streaming started', { | ||
| toolCallId, | ||
| hook: 'onInputStart', | ||
| }) | ||
| }, | ||
| onInputAvailable: ({ input, toolCallId, messages, abortSignal }) => { | ||
| log.info('Create sandbox received complete input', { | ||
| toolCallId, | ||
| hasMetadata: !!input.metadata, | ||
| hasEnvs: !!input.envs, | ||
| hook: 'onInputAvailable', | ||
| }) | ||
| }, | ||
| onOutput: ({ output, toolCallId, toolName, abortSignal }) => { | ||
| log.info('Create sandbox completed', { | ||
| toolCallId, | ||
| toolName, | ||
| sandboxId: output.sandboxId, | ||
| hook: 'onOutput', | ||
| }) | ||
| }, | ||
| execute: async (sandboxOptions) => { | ||
| try { | ||
| const sandbox = await Sandbox.create(sandboxOptions); | ||
|
|
||
| return { | ||
| sandboxId: sandbox.sandboxId, | ||
| }; | ||
| } catch (e) { | ||
| const errObj = e instanceof Error ? { message: e.message, stack: e.stack } : { message: String(e) }; | ||
| throw new Error(JSON.stringify(errObj)); | ||
| } | ||
| }, | ||
| }); | ||
| }), | ||
| outputSchema: z.object({ | ||
| sandboxId: z.string(), | ||
| }), | ||
| execute: async (inputData, context) => { | ||
| const span = tracer.startSpan('createSandbox') | ||
| const startTime = Date.now() | ||
| const requestContext = context?.requestContext as E2BRequestContext | ||
| const userId = requestContext?.userId | ||
| if (userId && userId.length > 0) { | ||
| span.setAttribute('user.id', userId) | ||
| } | ||
|
|
||
| export const runCode = createTool({ | ||
| id: 'runCode', | ||
| description: 'Run code in an e2b sandbox', | ||
| inputSchema: z.object({ | ||
| sandboxId: z.string().describe('The sandboxId for the sandbox to run the code'), | ||
| code: z.string().describe('The code to run in the sandbox'), | ||
| runCodeOpts: z | ||
| .object({ | ||
| language: z | ||
| .enum(['ts', 'js', 'python']) | ||
| .default('python') | ||
| .describe('language used for code execution. If not provided, default python context is used'), | ||
| envs: z.record(z.string(), z.string()).optional().describe('Custom environment variables for code execution.'), | ||
| timeoutMS: z.number().optional().describe(` | ||
| Timeout for the code execution in **milliseconds**. | ||
| @default 60_000 // 60 seconds | ||
| `), | ||
| requestTimeoutMs: z.number().optional().describe(` | ||
| Timeout for the request in **milliseconds**. | ||
| @default 30_000 // 30 seconds | ||
| `), | ||
| }) | ||
| .optional() | ||
| .describe('Run code options'), | ||
| }), | ||
| outputSchema: z.object({ | ||
| execution: z.string().describe('Serialized representation of the execution results'), | ||
| }), | ||
| onInputStart: ({ toolCallId, messages, abortSignal }) => { | ||
| log.info('Run code tool input streaming started', { | ||
| toolCallId, | ||
| hook: 'onInputStart', | ||
| }) | ||
| }, | ||
| onInputAvailable: ({ input, toolCallId, messages, abortSignal }) => { | ||
| log.info('Run code received complete input', { | ||
| toolCallId, | ||
| sandboxId: input.sandboxId, | ||
| codeLength: input.code.length, | ||
| language: input.runCodeOpts?.language, | ||
| hook: 'onInputAvailable', | ||
| }) | ||
| }, | ||
| onOutput: ({ output, toolCallId, toolName, abortSignal }) => { | ||
| log.info('Run code completed', { | ||
| toolCallId, | ||
| toolName, | ||
| executionLength: output.execution.length, | ||
| hook: 'onOutput', | ||
| }) | ||
| }, | ||
| execute: async (inputData) => { | ||
| try { | ||
| const sandbox = await Sandbox.connect(inputData.sandboxId); | ||
| // Add custom resource attributes | ||
| span.setAttribute('service.name', 'e2b-tools') | ||
| span.setAttribute('service.version', '1.0.0') | ||
| span.setAttribute('sandbox.operation', 'create') | ||
|
|
||
| const execution = await sandbox.runCode(inputData.code, inputData.runCodeOpts); | ||
| // Set baggage for cross-service context propagation | ||
| const currentBaggage = propagation.getBaggage(otelContext.active()) | ||
| if (currentBaggage) { | ||
| const keys = Array.from(currentBaggage.getAllEntries()).map(([key]) => key) | ||
| span.setAttribute('baggage.keys', keys.join(',')) | ||
| } | ||
|
|
||
| return { | ||
| execution: JSON.stringify(execution.toJSON()), | ||
| }; | ||
| } catch (e) { | ||
| const errObj = e instanceof Error ? { message: e.message, stack: e.stack } : { message: String(e) }; | ||
| throw new Error(JSON.stringify(errObj)); | ||
| } | ||
| }, | ||
| }); | ||
|
|
||
| export const readFile = createTool({ | ||
| id: 'readFile', | ||
| description: 'Read a file from the e2b sandbox', | ||
| inputSchema: z.object({ | ||
| sandboxId: z.string().describe('The sandboxId for the sandbox to read the file from'), | ||
| path: z.string().describe('The path to the file to read'), | ||
| }), | ||
| outputSchema: z.object({ | ||
| content: z.string().describe('The content of the file'), | ||
| path: z.string().describe('The path of the file that was read'), | ||
| }), | ||
| onInputStart: ({ toolCallId, messages, abortSignal }) => { | ||
| log.info('Read file tool input streaming started', { | ||
| toolCallId, | ||
| hook: 'onInputStart', | ||
| }) | ||
| }, | ||
| onInputAvailable: ({ input, toolCallId, messages, abortSignal }) => { | ||
| log.info('Read file received complete input', { | ||
| toolCallId, | ||
| sandboxId: input.sandboxId, | ||
| path: input.path, | ||
| hook: 'onInputAvailable', | ||
| }) | ||
| }, | ||
| onOutput: ({ output, toolCallId, toolName, abortSignal }) => { | ||
| log.info('Read file completed', { | ||
| toolCallId, | ||
| toolName, | ||
| contentLength: output.content.length, | ||
| path: output.path, | ||
| hook: 'onOutput', | ||
| }) | ||
| }, | ||
| execute: async (inputData) => { | ||
| try { | ||
| const sandbox = await Sandbox.connect(inputData.sandboxId); | ||
| const fileContent = await sandbox.files.read(inputData.path); | ||
|
|
||
| return { | ||
| content: fileContent, | ||
| path: inputData.path, | ||
| }; | ||
| } catch (e) { | ||
| const errObj = e instanceof Error ? { message: e.message, stack: e.stack } : { message: String(e) }; | ||
| throw new Error(JSON.stringify(errObj)); | ||
| } | ||
| }, | ||
| }); | ||
| // Add span event for input validation | ||
| span.addEvent('input.validation', { | ||
| 'input.has_metadata': !!inputData.metadata, | ||
| 'input.has_envs': !!inputData.envs, | ||
| 'input.timeout_ms': inputData.timeoutMS ?? 300000 | ||
| }) | ||
|
|
||
| try { | ||
| // Add start event and input size | ||
| const inputSize = JSON.stringify(inputData).length | ||
| span.addEvent('tool.start', { | ||
| ts: Date.now(), | ||
| input_size: inputSize, | ||
| }) | ||
| span.setAttribute('tool.input_size', inputSize) | ||
|
|
||
| await context?.writer?.custom({ | ||
| type: 'data-tool-progress', | ||
| data: { | ||
| status: 'in-progress', | ||
| message: `Creating sandbox...`, | ||
| stage: 'createSandbox', | ||
| }, | ||
| id: 'createSandbox', | ||
| }) | ||
|
|
||
| logToolStart('createSandbox', inputData, context) | ||
|
|
||
| const sandbox = await Sandbox.create(inputData) | ||
|
|
||
| // Record successful sandbox creation metrics | ||
| const creationDuration = Date.now() - startTime | ||
| sandboxOperationsCounter.add(1, { | ||
| 'operation.type': 'create', | ||
| 'operation.status': 'success', | ||
| 'sandbox.id': sandbox.sandboxId, | ||
| }) | ||
| sandboxOperationDuration.record(creationDuration, { | ||
| 'operation.type': 'create', | ||
| 'operation.status': 'success', | ||
| 'sandbox.id': sandbox.sandboxId, | ||
| }) | ||
|
|
||
| span.setStatus({ code: SpanStatusCode.OK }) | ||
| span.setAttribute('sandbox.id', sandbox.sandboxId) | ||
| span.setAttribute('sandbox.created', true) | ||
|
|
||
| // Add span event for successful creation | ||
| span.addEvent('sandbox.created', { | ||
| 'sandbox.id': sandbox.sandboxId, | ||
| 'creation.time_ms': Date.now() - startTime | ||
| }) | ||
|
|
||
| const result = { | ||
| sandboxId: sandbox.sandboxId, | ||
| } | ||
|
|
||
| // Output size attribute & complete event | ||
| const outputSize = JSON.stringify(result).length |
There was a problem hiding this comment.
Performance concern: Synchronous JSON.stringify calls on potentially large objects. Throughout the e2b.ts file, JSON.stringify() is called synchronously on input/output data for calculating sizes (e.g., lines 95, 129, 247, 252, 296). For large inputs or outputs, this can block the event loop. Consider using streaming or chunked approaches for size calculation, or at minimum add size limits to prevent performance degradation.
| // Message management | ||
| lastMessages: parseInt(process.env.MEMORY_LAST_MESSAGES ?? '500'), | ||
| generateTitle: process.env.THREAD_GENERATE_TITLE !== 'true', | ||
| generateTitle: true, |
There was a problem hiding this comment.
The generateTitle configuration change on line 87 in pg-storage.ts appears to have the logic inverted. The original code was generateTitle: process.env.THREAD_GENERATE_TITLE !== 'true' (meaning it defaults to false and only generates when env var is NOT 'true'), but the new code is generateTitle: true (always generate). This behavioral change should be intentional and documented, as it may impact performance by always generating titles for threads.
| generateTitle: true, | |
| generateTitle: process.env.THREAD_GENERATE_TITLE !== 'false', |
| export const e2bToolSet = { | ||
| createSandbox, | ||
| writeFile, | ||
| writeFiles, | ||
| listFiles, | ||
| deleteFile, | ||
| createDirectory, | ||
| getFileInfo, | ||
| checkFileExists, | ||
| getFileSize, | ||
| watchDirectory, | ||
| runCommand, | ||
| runCode, | ||
| } as const |
There was a problem hiding this comment.
The runCode tool is missing the readFile tool from the original implementation. Comparing the original code (which had readFile exported) with the new code, the readFile tool has been completely removed from the exports and tool definitions. This is a breaking change that removes existing functionality users may depend on.
|
|
||
| const logToolComplete = ( | ||
| toolName: string, | ||
| output: any, |
There was a problem hiding this comment.
Unexpected any. Specify a different type.
| output: any, | |
| output: unknown, |
| toolName: string, | ||
| output: any, | ||
| duration: number, | ||
| toolContext?: any |
There was a problem hiding this comment.
Unexpected any. Specify a different type.
| } catch (err) { | ||
| // ignore |
There was a problem hiding this comment.
'err' is defined but never used.
| } catch (err) { | |
| // ignore | |
| } catch (error) { | |
| logError('Failed to set span attributes in logToolComplete', { error }) |
| ) | ||
| } | ||
|
|
||
| const logToolError = (toolName: string, error: any, context?: any) => { |
There was a problem hiding this comment.
Unexpected any. Specify a different type.
| ) | ||
| } | ||
|
|
||
| const logToolError = (toolName: string, error: any, context?: any) => { |
There was a problem hiding this comment.
Unexpected any. Specify a different type.
| const logToolError = (toolName: string, error: any, context?: any) => { | |
| const logToolError = (toolName: string, error: unknown, context?: { requestContext?: E2BRequestContext }) => { |
instrumentation.tsto enable tracing for the service.CONTINUITY_ses_45e3.md, outlining goals, constraints, progress, and next steps for implementing lifecycle hooks and OpenTelemetry tracing insrc/mastra/tools/e2b.ts.CONTINUITY_ses_45e6.md, focusing on standardizing OpenTelemetry patterns across tools and ensuring proper lifecycle hook implementation.CONTINUITY_ses_45fc.md, detailing the project's structure, technology stack, and current state for onboarding new team members.