diff --git a/packages/@n8n/instance-ai/docs/sandboxing.md b/packages/@n8n/instance-ai/docs/sandboxing.md index 454c5df8a73..7662a8bcc35 100644 --- a/packages/@n8n/instance-ai/docs/sandboxing.md +++ b/packages/@n8n/instance-ai/docs/sandboxing.md @@ -167,6 +167,47 @@ apply RBAC, project scope, session grants, and HITL confirmation rules. The workspace is for Instance AI build and runtime-skill work. It is not a general user workload platform. +## Tracing + +Sandbox operations use the active Instance AI trace. Each span inherits the +thread metadata. The sandbox does not retain a turn's trace handle. Skill +preparation runs inside the agent's lazy build and uses the same trace context. + +| Operation | Trace data | +|-----------|------------| +| Acquire and start | Provider, sandbox ID, cached reuse, shared acquisition, duration, and errors | +| Initialize workspace | Marker check, node catalog, base files, dependency installation, and optional SDK linking | +| Sync skills and knowledge base | Bundle hash, file counts, byte counts, reuse or upload, and manifest check result | +| Read, write, and edit files | File operations under the existing tool span, paths, byte counts, and errors | +| Execute commands and compile workflows | Command size, exit code, timeout or cancellation, compile result, and bounded diagnostics | +| Retry and fallback | File path, failed attempt, retry delay, and command fallback | +| Evict cache and destroy | Reason, provider, sandbox ID, and cleanup errors | + +Batch operations reject on the first failure. Their spans end at that point. +Transfers that already started can continue. Batch spans omit individual successful +file operations. Retry and fallback spans remain visible. Internal file spans +omit file contents. Command spans record byte counts instead of raw commands. Failed +commands include stdout and stderr after filtering, limited to 2,000 characters +each. The export filter also covers status messages and exception events. +Nonzero exit codes are command results and do not mark command spans as errors. +Timeouts, killed commands, and thrown execution errors still mark spans as errors. + +Cache eviction keeps the remote sandbox. Its trace records the time of eviction. +Uncached cleanup checks the provider before creating a trace. Disabled sandboxing +and uncached Daytona sandboxes require no cleanup trace. +Cleanup between turns creates an internal operation trace with the same +`thread_id`. These sandbox lifecycle traces use the normal LangSmith settings. +They do not require `N8N_INSTANCE_AI_TRACE_INTERNAL`. Proxy deployments resolve +fresh trace configuration for cleanup. Cleanup requests carry the owner ID so +other processes can trace cleanup after the thread row is deleted. Timers create +detached traces even when they inherit an earlier turn context. Trace setup and +finalization each have a one-second deadline. Trace failures do not change +sandbox results or prevent cleanup. Operations still open when a turn closes +end with a cancelled status; their underlying work can finish independently. + +Provider-side automatic stop and deletion are not reported. They require +provider notifications or polling. + ## Configuration | Variable | Default | Purpose | diff --git a/packages/@n8n/instance-ai/src/index.ts b/packages/@n8n/instance-ai/src/index.ts index 33000b623b8..c655d07d189 100644 --- a/packages/@n8n/instance-ai/src/index.ts +++ b/packages/@n8n/instance-ai/src/index.ts @@ -249,6 +249,11 @@ export const createInstanceAiTraceContext: typeof LangsmithTracingMod.createInst export const createInternalOperationTraceContext: typeof LangsmithTracingMod.createInternalOperationTraceContext = lazyFunction(() => loadLangsmithTracing().createInternalOperationTraceContext); +export { traceSandboxOperation } from './tracing/sandbox-tracing'; + +export const withSandboxLifecycleTrace: typeof LangsmithTracingMod.withSandboxLifecycleTrace = + lazyFunction(() => loadLangsmithTracing().withSandboxLifecycleTrace); + export const createTraceReplayOnlyContext: typeof LangsmithTracingMod.createTraceReplayOnlyContext = lazyFunction(() => loadLangsmithTracing().createTraceReplayOnlyContext); diff --git a/packages/@n8n/instance-ai/src/knowledge-base/materialize-knowledge-base.ts b/packages/@n8n/instance-ai/src/knowledge-base/materialize-knowledge-base.ts index 9a6cb39caa6..0a903ca3301 100644 --- a/packages/@n8n/instance-ai/src/knowledge-base/materialize-knowledge-base.ts +++ b/packages/@n8n/instance-ai/src/knowledge-base/materialize-knowledge-base.ts @@ -20,6 +20,7 @@ import { } from './build-templates-index'; export { KNOWLEDGE_BASE_TEMPLATES_DIR }; import { extractBuilderTemplatesArchive } from './extract-builder-templates-archive'; +import { traceSandboxOperation } from '../tracing/sandbox-tracing'; import { computeWorkspaceContentHash } from '../workspace/compute-workspace-content-hash'; import { loadPrebakedWorkspaceBundle, @@ -330,20 +331,30 @@ export async function loadPrebakedKnowledgeBaseBundle( export async function materializeKnowledgeBaseIntoWorkspace( options: MaterializeKnowledgeBaseOptions, ): Promise { - return await materializeWorkspaceBundle({ - workspace: options.workspace, - resourceLabel: KNOWLEDGE_BASE_FILE_LABEL, - logger: options.logger, - loadPrebaked: async () => await loadPrebakedKnowledgeBaseBundle(options), - buildBundle: async () => await buildKnowledgeBaseWorkspaceBundle(options), - materializedLogMessage: 'Materialized knowledge base into workspace', - materializedLogContext: (bundle) => ({ - root: options.root, - knowledgeBaseRoot: bundle.rootDir, - contentHash: bundle.contentHash, - fileCount: bundle.files.size, - }), - }); + return await traceSandboxOperation( + 'sync-knowledge-base', + { + processResult: (bundle) => ({ + outputs: { contentHash: bundle.contentHash, fileCount: bundle.files.size }, + }), + }, + async () => { + return await materializeWorkspaceBundle({ + workspace: options.workspace, + resourceLabel: KNOWLEDGE_BASE_FILE_LABEL, + logger: options.logger, + loadPrebaked: async () => await loadPrebakedKnowledgeBaseBundle(options), + buildBundle: async () => await buildKnowledgeBaseWorkspaceBundle(options), + materializedLogMessage: 'Materialized knowledge base into workspace', + materializedLogContext: (bundle) => ({ + root: options.root, + knowledgeBaseRoot: bundle.rootDir, + contentHash: bundle.contentHash, + fileCount: bundle.files.size, + }), + }); + }, + ); } export type { diff --git a/packages/@n8n/instance-ai/src/skills/materialize-runtime-skills.ts b/packages/@n8n/instance-ai/src/skills/materialize-runtime-skills.ts index 9b081be449d..c24db21a81d 100644 --- a/packages/@n8n/instance-ai/src/skills/materialize-runtime-skills.ts +++ b/packages/@n8n/instance-ai/src/skills/materialize-runtime-skills.ts @@ -15,6 +15,7 @@ import { getWorkspaceRoot } from '@n8n/agents/sandbox'; import { join as posixJoin, normalize as posixNormalize } from 'node:path/posix'; import type { Logger } from '../logger'; +import { traceSandboxOperation } from '../tracing/sandbox-tracing'; import { loadPrebakedWorkspaceBundle, materializeWorkspaceBundle, @@ -524,30 +525,41 @@ export async function materializeRuntimeSkillsIntoWorkspace({ root, logger, }: MaterializeRuntimeSkillsOptions): Promise { - if (source.registry.skills.length === 0) return undefined; - - return await materializeWorkspaceBundle({ - workspace, - resourceLabel: RUNTIME_SKILL_FILE_LABEL, - logger, - loadPrebaked: async () => - await loadPrebakedRuntimeSkillsBundle({ source, workspace, root, logger }), - buildBundle: async () => { - const bundle = await buildRuntimeSkillWorkspaceBundle({ source, root, logger }); - if (!bundle) { - throw new Error('Expected runtime skill bundle after registry validation'); - } - return bundle; + return await traceSandboxOperation( + 'sync-skills', + { + inputs: { skillsHash: source.registry.skillsHash, skillCount: source.registry.skills.length }, + processResult: (bundle) => ({ + outputs: { skillCount: bundle?.skills.length ?? 0, fileCount: bundle?.files.size ?? 0 }, + }), }, - materializedLogMessage: 'Materialized runtime skills into workspace', - materializedLogContext: (bundle) => ({ - root, - skillsRoot: bundle.rootDir, - registryPath: bundle.registryPath, - skillsHash: bundle.skillsHash, - count: bundle.skills.length, - }), - }); + async () => { + if (source.registry.skills.length === 0) return undefined; + + return await materializeWorkspaceBundle({ + workspace, + resourceLabel: RUNTIME_SKILL_FILE_LABEL, + logger, + loadPrebaked: async () => + await loadPrebakedRuntimeSkillsBundle({ source, workspace, root, logger }), + buildBundle: async () => { + const bundle = await buildRuntimeSkillWorkspaceBundle({ source, root, logger }); + if (!bundle) { + throw new Error('Expected runtime skill bundle after registry validation'); + } + return bundle; + }, + materializedLogMessage: 'Materialized runtime skills into workspace', + materializedLogContext: (bundle) => ({ + root, + skillsRoot: bundle.rootDir, + registryPath: bundle.registryPath, + skillsHash: bundle.skillsHash, + count: bundle.skills.length, + }), + }); + }, + ); } export function createLazyWorkspaceRuntimeSkillSource({ @@ -583,9 +595,7 @@ export function createLazyWorkspaceRuntimeSkillSource({ materializePromise ??= (async () => { const root = await getWorkspaceRoot(runtimeWorkspace); const options = { source, workspace: runtimeWorkspace, root, logger }; - const result = - (await loadPrebakedRuntimeSkillsBundle(options)) ?? - (await materializeRuntimeSkillsIntoWorkspace(options)); + const result = await materializeRuntimeSkillsIntoWorkspace(options); if (result) { materialized = result; workspaceSource.registry = result.source.registry; diff --git a/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts b/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts index 2043ae90ff9..bef71508cd1 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts @@ -15,6 +15,7 @@ import { detectSlackBlocksShape } from './detect-slack-blocks-shape'; import { detectUnparseableOpenAiSchema } from './detect-unparseable-openai-schema'; import { detectWrongKindLocatorValues } from './detect-wrong-kind-locator'; import { collectValidationIssues, type ValidationWarning } from './workflow-validation-warnings'; +import { traceSandboxOperation, sandboxFileBytes } from '../../tracing/sandbox-tracing'; import type { InstanceAiContext } from '../../types'; import { escapeSingleQuotes, runInSandbox } from '../../workspace/sandbox-fs'; import { joinWorkspacePath } from '../../workspace/workspace-paths'; @@ -412,30 +413,54 @@ export async function compileWorkflowSource( source: string, abortSignal?: AbortSignal, ): Promise { - let result: WorkflowSourceCompileResult; - if (isWorkflowJsonSourceFile(filePath)) { - result = parseWorkflowJsonSource(source); - } else if (isTypeScriptWorkflowSource(filePath)) { - result = await compileTypeScriptWorkflowSource(context, filePath, abortSignal); - } else { - result = { - success: false, - reason: 'workflow_source_unsupported_extension', - editable: true, - errors: [ - 'Workflow source file must be a TypeScript SDK file (.ts or .tsx) or WorkflowJSON file (.json).', - ], - summary: 'Workflow source file extension is unsupported.', - }; - } + return await traceSandboxOperation( + 'compile-workflow', + { + inputs: { path: filePath, bytes: sandboxFileBytes(source) }, + processResult: (result) => + result.success + ? { + outputs: { + success: true, + compiler: result.compiler, + warningCount: result.warnings.length, + }, + } + : { + outputs: { success: false, reason: result.reason, errors: result.errors }, + error: result.summary, + }, + }, + async () => { + let result: WorkflowSourceCompileResult; + if (isWorkflowJsonSourceFile(filePath)) { + result = parseWorkflowJsonSource(source); + } else if (isTypeScriptWorkflowSource(filePath)) { + result = await compileTypeScriptWorkflowSource(context, filePath, abortSignal); + } else { + result = { + success: false, + reason: 'workflow_source_unsupported_extension', + editable: true, + errors: [ + 'Workflow source file must be a TypeScript SDK file (.ts or .tsx) or WorkflowJSON file (.json).', + ], + summary: 'Workflow source file extension is unsupported.', + }; + } - if (!result.success) return result; + if (!result.success) return result; - const warnings = validateCompiledWorkflow(result.workflow, context, result.warnings); - const credentialWarnings = await collectCredentialResolutionWarnings(result.workflow, context); + const warnings = validateCompiledWorkflow(result.workflow, context, result.warnings); + const credentialWarnings = await collectCredentialResolutionWarnings( + result.workflow, + context, + ); - return { - ...result, - warnings: [...warnings, ...credentialWarnings], - }; + return { + ...result, + warnings: [...warnings, ...credentialWarnings], + }; + }, + ); } diff --git a/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts b/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts index b05bdbb3aa6..be9cb27b2e0 100644 --- a/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts +++ b/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts @@ -1,5 +1,6 @@ /* eslint-disable import-x/order */ import { createRuntimeSkillRegistry, type BuiltTool } from '@n8n/agents'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import type { Context, ContextManager } from '@opentelemetry/api'; import * as langsmithModule from 'langsmith'; import { jsonParse } from 'n8n-workflow'; @@ -20,7 +21,11 @@ import { shutdownProductTelemetryProviders, submitLangsmithUserFeedback, withCurrentTraceSpan, + withSandboxLifecycleTrace, } from '../langsmith-tracing'; +import { traceSandboxOperation, sandboxCommandTraceResult } from '../sandbox-tracing'; +import { runInSandbox } from '../../workspace/sandbox-fs'; +import { writeWorkspaceFileMap } from '../../workspace/workspace-files'; import { TraceWriter, type TraceToolCall, type TraceToolSuspend } from '../trace-replay'; vi.mock('@n8n/agents', async () => { @@ -2280,6 +2285,355 @@ describe('createInstanceAiTraceContext', () => { expect(tracing).toBeUndefined(); }); + describe('sandbox tracing', () => { + const createSandboxTrace = async (threadId = 'sandbox-thread') => { + const tracing = await createInstanceAiTraceContext({ + threadId, + runId: 'run', + messageId: 'message', + userId: 'user', + input: 'test', + }); + if (!tracing) throw new Error('Expected a trace'); + return tracing; + }; + + it('records command metadata under the current turn without command contents', async () => { + const first = await createSandboxTrace(); + await first.finishRun(first.rootRun); + const second = await createSandboxTrace(); + const command = 'node -e "writeFile(payload)"'; + const executeCommand = vi.fn( + async () => + await Promise.resolve({ + exitCode: 0, + stdout: 'private file contents', + stderr: '', + }), + ); + await second.withActiveSpan( + second.rootRun, + async () => await runInSandbox({ sandbox: { executeCommand } }, command), + ); + const span = agentsMock.getSpans().find((entry) => entry.name === 'sandbox: execute-command'); + expect(span).toMatchObject({ parentSpanId: second.rootRun.otelSpanId, ended: true }); + expect(span?.attributes['langsmith.metadata.thread_id']).toBe('sandbox-thread'); + expect(jsonParse>(String(span?.attributes['gen_ai.prompt']))).toEqual( + { + commandBytes: Buffer.byteLength(command), + }, + ); + expect(JSON.stringify(span)).not.toContain('private file contents'); + expect(JSON.stringify(span)).not.toContain(command); + }); + + it('records a nonzero command exit as a completed result', async () => { + const tracing = await createSandboxTrace(); + const commandResult = { exitCode: 1, stdout: '', stderr: 'No matching entry' }; + const executeCommand = vi.fn(async () => await Promise.resolve(commandResult)); + const result = await tracing.withActiveSpan( + tracing.rootRun, + async () => await runInSandbox({ sandbox: { executeCommand } }, 'grep pattern file'), + ); + expect(result).toEqual(commandResult); + const span = agentsMock.getSpans().find((entry) => entry.name === 'sandbox: execute-command'); + expect(span).toMatchObject({ ended: true, status: { code: 1 } }); + expect(span?.attributes['langsmith.metadata.final_status']).toBe('completed'); + expect( + jsonParse>(String(span?.attributes['gen_ai.completion'])), + ).toMatchObject({ exitCode: 1, stderr: 'No matching entry' }); + }); + + it('records a thrown command execution error and preserves it for the caller', async () => { + const tracing = await createSandboxTrace(); + const error = new Error('Command execution unavailable'); + const executeCommand = vi.fn(async () => await Promise.reject(error)); + await expect( + tracing.withActiveSpan( + tracing.rootRun, + async () => await runInSandbox({ sandbox: { executeCommand } }, 'command'), + ), + ).rejects.toBe(error); + const span = agentsMock.getSpans().find((entry) => entry.name === 'sandbox: execute-command'); + expect(span).toMatchObject({ ended: true, status: { code: 2 } }); + expect(span?.attributes['langsmith.metadata.final_status']).toBe('error'); + }); + + it.each([ + { status: { exitCode: 1 }, expectedError: undefined }, + { status: { exitCode: 0, timedOut: true }, expectedError: 'Command timed out' }, + { status: { exitCode: 0, killed: true }, expectedError: 'Command was killed' }, + ])('keeps bounded diagnostics for command result %j', async ({ status, expectedError }) => { + const key = `-----BEGIN PRIVATE KEY-----\n${'A'.repeat(2300)}\n-----END PRIVATE KEY-----`; + const result = await sandboxCommandTraceResult({ + ...status, + stdout: 'npm error EAI_AGAIN registry.npmjs.org', + stderr: key, + }); + expect(result.error).toBe(expectedError); + expect(result.outputs).toEqual( + expect.objectContaining({ + stdout: 'npm error EAI_AGAIN registry.npmjs.org', + }), + ); + expect(JSON.stringify(result.outputs)).not.toContain('A'.repeat(20)); + expect(JSON.stringify(result.outputs)).not.toContain('BEGIN PRIVATE KEY'); + const long = await sandboxCommandTraceResult({ + ...status, + stdout: 'x '.repeat(3000), + stderr: 'y '.repeat(3000), + }); + const diagnostics = long.outputs as { stdout: string; stderr: string }; + expect(diagnostics.stdout.length).toBeLessThanOrEqual(2000); + expect(diagnostics.stderr.length).toBeLessThanOrEqual(2000); + }); + + it.each(['privateKey', 'private_key', 'private-key'])( + 'filters %s fields in exported spans', + (key) => { + const value = 'opaque-test-material'; + const span = { + attributes: { [key]: value }, + events: [ + { + name: 'operation', + attributes: { [key]: value, details: { [key]: value, format: 'example' } }, + }, + ], + }; + const filtered = redactLangSmithTelemetrySpan(span); + expect(filtered).toMatchObject({ + attributes: { [key]: '[redacted]' }, + events: [ + { + name: 'operation', + attributes: { + [key]: '[redacted]', + details: { [key]: '[redacted]', format: 'example' }, + }, + }, + ], + }); + expect(JSON.stringify(filtered)).not.toContain(value); + }, + ); + + it('filters status and exception fields before export', () => { + const message = 'Could not open person@example.com.workflow.ts'; + const span = { + attributes: {}, + status: { code: 2, message }, + events: [ + { + name: 'exception', + time: [1, 2], + attributes: { + 'exception.message': message, + 'exception.stacktrace': `Error: ${message}\n at run()`, + }, + }, + ], + }; + const filtered = redactLangSmithTelemetrySpan(span); + expect(JSON.stringify(filtered)).not.toContain('person@example.com'); + expect(filtered).toMatchObject({ + status: { code: 2 }, + events: [{ name: 'exception', time: [1, 2] }], + }); + }); + + it('ends pending operations when their turn closes and preserves the eventual result', async () => { + const tracing = await createSandboxTrace(); + const started = createDeferredPromise(); + const release = createDeferredPromise(); + const operation = tracing.withActiveSpan( + tracing.rootRun, + async () => + await traceSandboxOperation('initialize-workspace', {}, async () => { + started.resolve(); + return await release.promise; + }), + ); + await started.promise; + await tracing.finishRun(tracing.rootRun); + const span = agentsMock + .getSpans() + .find((entry) => entry.name === 'sandbox: initialize-workspace'); + expect(span).toMatchObject({ ended: true, status: { code: 2 } }); + expect(span?.attributes['langsmith.metadata.final_status']).toBe('cancelled'); + release.resolve(42); + await expect(operation).resolves.toBe(42); + expect(span?.attributes['langsmith.metadata.final_status']).toBe('cancelled'); + }); + + it('returns batch failure before remaining writes finish', async () => { + const tracing = await createSandboxTrace(); + const slowWrite = createDeferredPromise(); + const writeFile = vi.fn(async (path: string) => { + if (path === 'first') throw new Error('first write failed'); + await slowWrite.promise; + }); + const batch = tracing.withActiveSpan( + tracing.rootRun, + async () => + await writeWorkspaceFileMap( + { filesystem: { writeFile } }, + new Map([ + ['first', 'one'], + ['second', 'two'], + ]), + ), + ); + let rejected = false; + const outcome = batch.catch((error: unknown) => { + rejected = true; + return error; + }); + try { + await vi.waitFor(() => expect(rejected).toBe(true)); + expect(writeFile).toHaveBeenCalledTimes(2); + expect(await outcome).toBeInstanceOf(Error); + const span = agentsMock.getSpans().find((entry) => entry.name === 'sandbox: write-files'); + expect(span).toMatchObject({ ended: true, status: { code: 2 } }); + } finally { + slowWrite.resolve(); + await outcome; + } + }); + + it('creates detached lifecycle roots despite a live ambient turn', async () => { + const tracing = await createSandboxTrace(); + await tracing.withActiveSpan( + tracing.rootRun, + async () => + await withSandboxLifecycleTrace('sandbox-thread', 'evict-cache', {}, async () => {}, { + detached: true, + }), + ); + const span = agentsMock + .getSpans() + .find((entry) => entry.name === 'internal: sandbox-evict-cache'); + expect(span?.parentSpanId).toBeUndefined(); + expect(span?.traceId).not.toBe(tracing.rootRun.otelTraceId); + expect(span?.attributes['langsmith.metadata.thread_id']).toBe('sandbox-thread'); + expect(span?.attributes['langsmith.metadata.final_status']).toBe('completed'); + }); + + it.each(['error', 'cancelled'])( + 'records lifecycle status %s and preserves the error', + async (status) => { + const error = new Error('cleanup failed'); + if (status === 'cancelled') error.name = 'AbortError'; + await expect( + withSandboxLifecycleTrace( + 'sandbox-thread', + 'destroy', + {}, + async () => await Promise.reject(error), + ), + ).rejects.toBe(error); + const span = agentsMock + .getSpans() + .find((entry) => entry.name === 'internal: sandbox-destroy'); + expect(span?.attributes['langsmith.metadata.final_status']).toBe(status); + expect(span).toMatchObject({ ended: true, status: { code: 2 } }); + }, + ); + + it('uses supplied owner configuration for proxy lifecycle traces', async () => { + delete process.env.LANGSMITH_API_KEY; + const getAuthHeaders = vi.fn( + async () => await Promise.resolve({ Authorization: 'Bearer proxy-test' }), + ); + await withSandboxLifecycleTrace( + 'deleted-thread', + 'destroy', + {}, + async () => { + expect(getAuthHeaders).toHaveBeenCalledTimes(1); + await Promise.resolve(); + }, + { + resolveConfig: async () => + await Promise.resolve({ + userId: 'owner', + proxyConfig: { apiUrl: 'https://proxy.example.com/langsmith', getAuthHeaders }, + }), + }, + ); + const span = agentsMock + .getSpans() + .find((entry) => entry.name === 'internal: sandbox-destroy'); + expect(span?.attributes['langsmith.metadata.thread_id']).toBe('deleted-thread'); + expect(span?.attributes['langsmith.metadata.user_id']).toBe('owner'); + expect(getAuthHeaders).toHaveBeenCalled(); + }); + + it('continues cleanup after the trace setup deadline and releases a late context', async () => { + vi.useFakeTimers(); + try { + const headers = createDeferredPromise>(); + const started = createDeferredPromise(); + const cleanup = vi.fn(async () => await Promise.resolve('done')); + const operation = withSandboxLifecycleTrace('sandbox-thread', 'destroy', {}, cleanup, { + resolveConfig: async () => + await Promise.resolve({ + userId: 'owner', + proxyConfig: { + apiUrl: 'https://proxy.example.com/langsmith', + getAuthHeaders: async () => { + started.resolve(); + return await headers.promise; + }, + }, + }), + }); + await started.promise; + await vi.advanceTimersByTimeAsync(1000); + await expect(operation).resolves.toBe('done'); + expect(cleanup).toHaveBeenCalledTimes(1); + headers.resolve({}); + await vi.waitFor(() => expect(agentsMock.getProvider().shutdown).toHaveBeenCalledTimes(1)); + expect(cleanup).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it('preserves results when trace creation or result processing fails', async () => { + const tracing = await createSandboxTrace(); + await expect( + tracing.withActiveSpan( + tracing.rootRun, + async () => + await traceSandboxOperation( + 'operation', + { + processResult: () => { + throw new Error('format failed'); + }, + }, + async () => await Promise.resolve(42), + ), + ), + ).resolves.toBe(42); + agentsMock.setBuildError(new Error('trace setup failed')); + const cleanup = vi.fn(async () => await Promise.resolve(43)); + await expect( + withSandboxLifecycleTrace('another-thread', 'destroy', {}, cleanup), + ).resolves.toBe(43); + expect(cleanup).toHaveBeenCalledTimes(1); + }); + + it('runs operations when tracing is disabled', async () => { + process.env.LANGSMITH_TRACING = 'false'; + const operation = vi.fn(async () => await Promise.resolve(42)); + await expect(traceSandboxOperation('write-file', {}, operation)).resolves.toBe(42); + await expect(withSandboxLifecycleTrace('thread', 'destroy', {}, operation)).resolves.toBe(42); + expect(operation).toHaveBeenCalledTimes(2); + expect(agentsMock.getSpans()).toEqual([]); + }); + }); }); describe('submitLangsmithUserFeedback', () => { diff --git a/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts b/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts index 9af080b57c1..c124d7689b6 100644 --- a/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts +++ b/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts @@ -1,6 +1,8 @@ import { LangSmithTelemetry, Telemetry, + isAbortError, + raceWithAbort, type AttributeValue, type BuiltTelemetry, type BuiltTool, @@ -19,6 +21,7 @@ import { import type { Context as OtelContext, Span as OtelApiSpan } from '@opentelemetry/api'; import { Client } from 'langsmith'; import { AsyncLocalStorage } from 'node:async_hooks'; +import { randomUUID } from 'node:crypto'; import { readFile } from 'node:fs/promises'; import { createRequire } from 'node:module'; import { dirname, join, parse } from 'node:path'; @@ -181,6 +184,7 @@ interface ProductOtelTraceRuntime { telemetry: BuiltTelemetry; spans: Map; contexts: Map; + pendingOperations: Map; shutdown: boolean; lifetime: ProductTelemetryLifetime; } @@ -487,6 +491,13 @@ async function releaseProductOtelRuntime( if (runtime.shutdown) return; runtime.shutdown = true; + for (const run of runtime.pendingOperations.values()) { + await finishProductSpanBestEffort(runtime, run, { + error: 'Operation did not finish before the trace closed', + metadata: { final_status: 'cancelled' }, + }); + } + runtime.pendingOperations.clear(); runtime.spans.clear(); runtime.contexts.clear(); otelTraceRuntimes.delete(traceId); @@ -723,6 +734,9 @@ interface CurrentTraceSpanOptions { metadata?: Record; inputs?: unknown; processOutputs?: (result: T) => unknown; + processResult?: ( + result: T, + ) => InstanceAiTraceRunFinishOptions | Promise; } type NativeToolContext = ToolContext | InterruptibleToolContext; @@ -904,7 +918,7 @@ export async function withCurrentTraceSpan( fn: () => Promise, ): Promise { const currentProductTrace = getCurrentProductTrace(); - if (!currentProductTrace) { + if (!currentProductTrace || currentProductTrace.runtime.shutdown) { return await fn(); } @@ -919,7 +933,7 @@ export async function withCurrentTraceSpan( canonicalName: options.canonicalName, runType: options.runType ?? 'chain', tags: options.tags, - metadata: options.metadata, + metadata: mergeMetadata(currentProductTrace.currentRun.metadata, options.metadata), inputs: options.inputs, parentRun: currentProductTrace.currentRun, ...(activeParentContext ? { parentContext: activeParentContext } : {}), @@ -928,19 +942,30 @@ export async function withCurrentTraceSpan( return await fn(); } + currentProductTrace.runtime.pendingOperations.set(spanRun.id, spanRun); try { const result = await withProductSpanContextBestEffort(currentProductTrace.runtime, spanRun, fn); + let finish: InstanceAiTraceRunFinishOptions = {}; + try { + finish = (await options.processResult?.(result)) ?? { + ...(options.processOutputs ? { outputs: options.processOutputs(result) } : {}), + }; + } catch { + // Trace serialization must not change the operation result. + } await finishProductSpanBestEffort(currentProductTrace.runtime, spanRun, { - ...(options.processOutputs ? { outputs: options.processOutputs(result) } : {}), - metadata: { final_status: 'completed' }, + ...finish, + metadata: { final_status: finish.error ? 'error' : 'completed', ...finish.metadata }, }); return result; } catch (error) { await finishProductSpanBestEffort(currentProductTrace.runtime, spanRun, { error: getErrorMessage(error), - metadata: { final_status: 'error' }, + metadata: { final_status: isAbortError(error) ? 'cancelled' : 'error' }, }); throw error; + } finally { + currentProductTrace.runtime.pendingOperations.delete(spanRun.id); } } @@ -1695,6 +1720,7 @@ async function createProductOtelRuntime( telemetry, spans: new Map(), contexts: new Map(), + pendingOperations: new Map(), shutdown: false, lifetime: new ProductTelemetryLifetime(telemetry), }; @@ -1870,7 +1896,117 @@ export async function continueInstanceAiTraceContext( export async function createInternalOperationTraceContext( options: CreateInternalOperationTraceContextOptions, ): Promise { - if (!isInternalOperationTracingEnabled() || !isLangSmithTracingEnabled(!!options.proxyConfig)) { + if (!isInternalOperationTracingEnabled()) return undefined; + return await createOperationTraceContext(options); +} + +const SANDBOX_TRACE_TIMEOUT_MS = 1_000; + +interface SandboxLifecycleTraceOptions { + detached?: boolean; + resolveConfig?: () => Promise<{ userId: string; proxyConfig?: ServiceProxyConfig }>; +} + +export async function withSandboxLifecycleTrace( + threadId: string, + operation: string, + inputs: Record, + fn: () => Promise, + options?: SandboxLifecycleTraceOptions, +): Promise { + const current = getCurrentProductTrace(); + if ( + !options?.detached && + current && + !current.runtime.shutdown && + current.currentRun.metadata?.thread_id === threadId + ) { + return await withCurrentTraceSpan( + { + name: `sandbox: ${operation}`, + canonicalName: `instance-ai.sandbox.${operation}`, + tags: ['sandbox'], + metadata: { sandbox_operation: operation }, + inputs, + }, + fn, + ); + } + + let tracing: InstanceAiTraceContext | undefined; + try { + if (isLangSmithTracingEnabled(true)) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), SANDBOX_TRACE_TIMEOUT_MS); + timer.unref(); + try { + tracing = await raceWithAbort(async () => { + const config = await options?.resolveConfig?.(); + if (controller.signal.aborted) return undefined; + const created = await createOperationTraceContext({ + threadId, + runId: randomUUID(), + userId: config?.userId ?? 'system', + proxyConfig: config?.proxyConfig, + operationName: `sandbox.${operation}`, + metadata: { sandbox_operation: operation }, + input: inputs, + }); + if (controller.signal.aborted && created) { + releaseTraceClient(created.rootRun.traceId); + return undefined; + } + return created; + }, controller.signal); + } finally { + clearTimeout(timer); + } + } + } catch { + // Trace setup must not prevent cleanup. + } + if (!tracing) { + return await productTraceStorage.exit(async () => await otelContext.with(ROOT_CONTEXT, fn)); + } + + try { + // Authentication was resolved within the setup deadline. + const runtime = otelTraceRuntimes.get(tracing.rootRun.traceId); + const result = runtime + ? await withProductSpanContextBestEffort(runtime, tracing.rootRun, fn) + : await fn(); + try { + await raceWithAbort( + async () => + await tracing.finishRun(tracing.rootRun, { metadata: { final_status: 'completed' } }), + AbortSignal.timeout(SANDBOX_TRACE_TIMEOUT_MS), + ); + } catch { + // Export failures must not change the cleanup result. + } + return result; + } catch (error) { + try { + await raceWithAbort( + async () => + await tracing.failRun(tracing.rootRun, error, { + final_status: isAbortError(error) ? 'cancelled' : 'error', + }), + AbortSignal.timeout(SANDBOX_TRACE_TIMEOUT_MS), + ); + } catch { + // Preserve the cleanup error. + } + throw error; + } finally { + releaseTraceClient(tracing.rootRun.traceId); + } +} + +async function createOperationTraceContext( + options: CreateInternalOperationTraceContextOptions, +): Promise { + if (!isLangSmithTracingEnabled(!!options.proxyConfig)) { return undefined; } diff --git a/packages/@n8n/instance-ai/src/tracing/sandbox-tracing.ts b/packages/@n8n/instance-ai/src/tracing/sandbox-tracing.ts new file mode 100644 index 00000000000..7f0d6256fbd --- /dev/null +++ b/packages/@n8n/instance-ai/src/tracing/sandbox-tracing.ts @@ -0,0 +1,81 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; +import { posix } from 'node:path'; + +import type { InstanceAiTraceRunFinishOptions } from '../types'; + +const ioTraceScope = new AsyncLocalStorage(); + +interface SandboxTraceOptions { + inputs?: Record; + /** Batch spans summarize file transfers. I/O spans suppress transport details. */ + kind?: 'batch' | 'io'; + processResult?: ( + result: T, + ) => InstanceAiTraceRunFinishOptions | Promise; +} + +export async function traceSandboxOperation( + operation: string, + options: SandboxTraceOptions, + fn: () => Promise, +): Promise { + if (options.kind === 'io' && ioTraceScope.getStore()) return await fn(); + + let tracing: typeof import('./langsmith-tracing'); + try { + tracing = await import('./langsmith-tracing.js'); + } catch { + return await fn(); + } + return await tracing.withCurrentTraceSpan( + { + name: `sandbox: ${operation}`, + canonicalName: `instance-ai.sandbox.${operation}`, + tags: ['sandbox'], + metadata: { sandbox_operation: operation }, + inputs: options.inputs, + processResult: options.processResult, + }, + async () => (options.kind ? await ioTraceScope.run(true, fn) : await fn()), + ); +} + +export function sandboxTracePath(path: string, root?: string): string { + return root && posix.isAbsolute(path) ? posix.relative(root, path) : path; +} + +export function sandboxFileBytes(content: string | Uint8Array): number { + return typeof content === 'string' ? Buffer.byteLength(content, 'utf8') : content.byteLength; +} + +export async function sandboxCommandTraceResult(result: { + exitCode: number; + stdout: string; + stderr: string; + timedOut?: boolean; + killed?: boolean; +}): Promise { + const error = result.timedOut + ? 'Command timed out' + : result.killed + ? 'Command was killed' + : undefined; + const includeDiagnostics = result.exitCode !== 0 || error !== undefined; + const { scrubTelemetryText } = await import('./trace-payloads.js'); + return { + outputs: { + exitCode: result.exitCode, + timedOut: result.timedOut ?? false, + killed: result.killed ?? false, + stdoutBytes: sandboxFileBytes(result.stdout), + stderrBytes: sandboxFileBytes(result.stderr), + ...(includeDiagnostics + ? { + stdout: scrubTelemetryText(result.stdout).slice(0, 2000), + stderr: scrubTelemetryText(result.stderr).slice(0, 2000), + } + : {}), + }, + ...(error ? { error } : {}), + }; +} diff --git a/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts b/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts index 29171762626..34b6aa16606 100644 --- a/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts +++ b/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts @@ -74,7 +74,7 @@ export const DEFAULT_TELEMETRY_REDACTION_OPTIONS: RedactionOptions = { }; /** Redact secrets + all PII from a free-text telemetry value before it egresses. */ -function scrubTelemetryText(value: string): string { +export function scrubTelemetryText(value: string): string { return redactText(value, DEFAULT_TELEMETRY_REDACTION_OPTIONS).text; } @@ -1030,6 +1030,19 @@ export function redactLangSmithTelemetrySpan(span: unknown): unknown { renameNativeToolSpanForLangSmith(span, attributes); moveNonLlmUsageAttributes(attributes); span.attributes = attributes; + if (isRecord(span.status) && typeof span.status.message === 'string') { + span.status = { + ...span.status, + message: truncateString(scrubTelemetryText(span.status.message)), + }; + } + if (Array.isArray(span.events)) { + span.events = span.events.map((event: unknown) => + isRecord(event) && isRecord(event.attributes) + ? { ...event, attributes: redactTelemetryJsonValue(event.attributes) } + : event, + ); + } return span; } diff --git a/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts b/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts index 488a5f74f04..2920fba7478 100644 --- a/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts +++ b/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts @@ -5,6 +5,7 @@ import { type WorkspaceFilesystem, type WorkspaceSandbox, } from '@n8n/agents'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { createLazyRuntimeWorkspace } from '../lazy-runtime-workspace'; @@ -98,6 +99,36 @@ describe('createLazyRuntimeWorkspace', () => { expect(ensureWorkspace).toHaveBeenCalledTimes(1); }); + it('reports cancellation when an active command returns a transport cancellation', async () => { + const { workspace, executeCommand } = createMockWorkspace(); + const started = createDeferredPromise(); + executeCommand.mockImplementation( + async (_command, _args, options) => + await new Promise((_resolve, reject) => { + options?.abortSignal?.addEventListener( + 'abort', + () => { + const error = new Error('canceled'); + error.name = 'CanceledError'; + reject(error); + }, + { once: true }, + ); + started.resolve(); + }), + ); + const lazyWorkspace = createLazyRuntimeWorkspace({ + ensureWorkspace: async () => await Promise.resolve(workspace), + }); + const controller = new AbortController(); + const operation = lazyWorkspace.sandbox!.executeCommand!('sleep 10', [], { + abortSignal: controller.signal, + }); + await started.promise; + controller.abort(); + await expect(operation).rejects.toMatchObject({ name: 'AbortError' }); + }); + it('retries workspace creation after the first lazy initialization fails', async () => { const { workspace } = createMockWorkspace(); const ensureWorkspace = vi diff --git a/packages/@n8n/instance-ai/src/workspace/__tests__/sandbox-setup.test.ts b/packages/@n8n/instance-ai/src/workspace/__tests__/sandbox-setup.test.ts index 03d5b76077a..b9cf4d609a7 100644 --- a/packages/@n8n/instance-ai/src/workspace/__tests__/sandbox-setup.test.ts +++ b/packages/@n8n/instance-ai/src/workspace/__tests__/sandbox-setup.test.ts @@ -62,6 +62,7 @@ vi.mock('../sandbox-fs', () => ({ escapeSingleQuotes: (value: string) => value.replace(/'/g, "'\\''"), })); +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { jsonParse } from 'n8n-workflow'; import type { Mock } from 'vitest'; @@ -420,6 +421,53 @@ describe('setupSandboxWorkspace', () => { expect(writtenPaths.some((p) => p.includes('/knowledge-base/templates/'))).toBe(true); }); + it.each(['mkdir', 'writeFile'])( + 'returns setup failure while another %s operation is pending', + async (operation) => { + const slowOperation = createDeferredPromise(); + const error = new Error('first operation failed'); + const runInSandbox: RunInSandboxMock = vi.fn(); + runInSandbox.mockRejectedValue(error); + const readFileViaSandbox: ReadFileViaSandboxMock = vi.fn(); + readFileViaSandbox.mockResolvedValue(null); + const setup = loadSetupSandboxWorkspaceWithFsMocks(runInSandbox, readFileViaSandbox); + const writeFile = vi.fn< + (...args: [string, string | Buffer, { recursive?: boolean }?]) => Promise + >(async (path) => { + if (operation !== 'writeFile') return; + if (path.endsWith('/package.json')) throw error; + if (path.endsWith('/tsconfig.json')) await slowOperation.promise; + }); + const mkdir = vi.fn<(...args: [string, { recursive?: boolean }?]) => Promise>( + async (path) => { + if (operation !== 'mkdir') return; + if (path.endsWith('/src')) throw error; + if (path.endsWith('/chunks')) await slowOperation.promise; + }, + ); + let rejected = false; + const outcome = setup( + createFilesystemWorkspace(writeFile, mkdir), + createSetupContext(), + ).catch((failure: unknown) => { + rejected = true; + return failure; + }); + try { + await vi.waitFor(() => expect(rejected).toBe(true)); + expect(await outcome).toMatchObject({ step: 'write-workspace-files' }); + if (operation === 'mkdir') { + expect(mkdir.mock.calls.some(([path]) => path.endsWith('/chunks'))).toBe(true); + } else { + expect(writeFile.mock.calls.some(([path]) => path.endsWith('/tsconfig.json'))).toBe(true); + } + } finally { + slowOperation.resolve(); + await outcome; + } + }, + ); + it('does not write the initialized marker when npm install fails', async () => { const runInSandbox: RunInSandboxMock = vi.fn< diff --git a/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts b/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts index 6a81464c2aa..c37e49410c4 100644 --- a/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts +++ b/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts @@ -22,6 +22,13 @@ import { type WriteOptions, } from '@n8n/agents'; +import { + traceSandboxOperation, + sandboxFileBytes, + sandboxTracePath, + sandboxCommandTraceResult, +} from '../tracing/sandbox-tracing'; + export type RuntimeWorkspaceResolver = () => Promise; export interface LazyRuntimeWorkspaceOptions { @@ -219,35 +226,103 @@ class LazyRuntimeFilesystem extends BaseFilesystem { } async readFile(path: string, options?: ReadOptions): Promise { - return await (await this.getFilesystem(options?.abortSignal)).readFile(path, options); + return await traceSandboxOperation( + 'read-file', + { + kind: 'io', + inputs: { path: sandboxTracePath(path, this.basePath) }, + processResult: (content) => ({ outputs: { bytes: sandboxFileBytes(content) } }), + }, + async () => await (await this.getFilesystem(options?.abortSignal)).readFile(path, options), + ); } async writeFile(path: string, content: FileContent, options?: WriteOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).writeFile(path, content, options); + await traceSandboxOperation( + 'write-file', + { + kind: 'io', + inputs: { path: sandboxTracePath(path, this.basePath), bytes: sandboxFileBytes(content) }, + }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).writeFile(path, content, options); + }, + ); } async appendFile(path: string, content: FileContent, options?: AppendOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).appendFile(path, content, options); + await traceSandboxOperation( + 'append-file', + { + kind: 'io', + inputs: { path: sandboxTracePath(path, this.basePath), bytes: sandboxFileBytes(content) }, + }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).appendFile(path, content, options); + }, + ); } async deleteFile(path: string, options?: RemoveOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).deleteFile(path, options); + await traceSandboxOperation( + 'delete-file', + { kind: 'io', inputs: { path: sandboxTracePath(path, this.basePath) } }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).deleteFile(path, options); + }, + ); } async copyFile(src: string, dest: string, options?: CopyOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).copyFile(src, dest, options); + await traceSandboxOperation( + 'copy-file', + { + kind: 'io', + inputs: { + src: sandboxTracePath(src, this.basePath), + dest: sandboxTracePath(dest, this.basePath), + }, + }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).copyFile(src, dest, options); + }, + ); } async moveFile(src: string, dest: string, options?: CopyOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).moveFile(src, dest, options); + await traceSandboxOperation( + 'move-file', + { + kind: 'io', + inputs: { + src: sandboxTracePath(src, this.basePath), + dest: sandboxTracePath(dest, this.basePath), + }, + }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).moveFile(src, dest, options); + }, + ); } async mkdir(path: string, options?: MkdirOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).mkdir(path, options); + await traceSandboxOperation( + 'mkdir', + { kind: 'io', inputs: { path: sandboxTracePath(path, this.basePath) } }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).mkdir(path, options); + }, + ); } async rmdir(path: string, options?: RemoveOptions): Promise { - await (await this.getFilesystem(options?.abortSignal)).rmdir(path, options); + await traceSandboxOperation( + 'rmdir', + { kind: 'io', inputs: { path: sandboxTracePath(path, this.basePath) } }, + async () => { + await (await this.getFilesystem(options?.abortSignal)).rmdir(path, options); + }, + ); } async readdir(path: string, options?: ListOptions): Promise { @@ -322,8 +397,27 @@ class LazyRuntimeSandbox extends BaseSandbox { throw new Error('Instance AI runtime sandbox does not support command execution.'); } + const executeCommand = sandbox.executeCommand.bind(sandbox); try { - return await sandbox.executeCommand(command, args, options); + return await traceSandboxOperation( + 'execute-command', + { + kind: 'io', + inputs: { + commandBytes: sandboxFileBytes(command), + argumentBytes: args.reduce((bytes, arg) => bytes + sandboxFileBytes(arg), 0), + argumentCount: args.length, + cwd: options?.cwd, + timeout: options?.timeout, + }, + processResult: sandboxCommandTraceResult, + }, + async () => + await raceWithAbort( + async () => await executeCommand(command, args, options), + options?.abortSignal, + ), + ); } finally { this.syncStatus(sandbox); } diff --git a/packages/@n8n/instance-ai/src/workspace/prebaked-workspace-bundle.ts b/packages/@n8n/instance-ai/src/workspace/prebaked-workspace-bundle.ts index cb283947a47..1450918d488 100644 --- a/packages/@n8n/instance-ai/src/workspace/prebaked-workspace-bundle.ts +++ b/packages/@n8n/instance-ai/src/workspace/prebaked-workspace-bundle.ts @@ -6,6 +6,7 @@ import { type WorkspaceFileTarget, } from './workspace-files'; import { parseVersionedWorkspaceManifest } from './workspace-manifest'; +import { traceSandboxOperation, sandboxFileBytes } from '../tracing/sandbox-tracing'; export interface LoadPrebakedWorkspaceBundleOptions { workspace: WorkspaceFileTarget; @@ -30,59 +31,83 @@ export interface LoadPrebakedWorkspaceBundleOptions { export async function loadPrebakedWorkspaceBundle }>( options: LoadPrebakedWorkspaceBundleOptions, ): Promise { - const manifestRaw = await readWorkspaceFile(options.workspace, options.manifestPath, { - logger: options.logger, - resourceLabel: options.resourceLabel, - }); - if (!manifestRaw) return undefined; + let decision = 'reused'; + let actualHash: string | undefined; + return await traceSandboxOperation( + 'check-bundle', + { + kind: 'batch', + inputs: { manifestPath: options.manifestPath, expectedHash: options.expectedHash }, + processResult: (bundle) => ({ + outputs: { decision, actualHash, fileCount: bundle?.files.size ?? 0 }, + }), + }, + async () => { + const manifestRaw = await readWorkspaceFile(options.workspace, options.manifestPath, { + logger: options.logger, + resourceLabel: options.resourceLabel, + }); + if (!manifestRaw) { + decision = 'missing'; + return undefined; + } - const manifest = parseVersionedWorkspaceManifest(manifestRaw, { - schemaVersion: options.schemaVersion, - hashField: options.hashField, - }); - if (!manifest) { - options.logger.debug(options.invalidManifestLogMessage, { - manifestPath: options.manifestPath, - }); - return undefined; - } + const manifest = parseVersionedWorkspaceManifest(manifestRaw, { + schemaVersion: options.schemaVersion, + hashField: options.hashField, + }); + if (!manifest) { + decision = 'invalid'; + options.logger.debug(options.invalidManifestLogMessage, { + manifestPath: options.manifestPath, + }); + return undefined; + } - if (manifest.hash !== options.expectedHash) { - options.logger.debug(options.staleManifestLogMessage, { - manifestPath: options.manifestPath, - [options.staleManifestLogKeys.expected]: options.expectedHash, - [options.staleManifestLogKeys.actual]: manifest.hash, - }); - return undefined; - } + actualHash = manifest.hash; + if (manifest.hash !== options.expectedHash) { + decision = 'stale'; + options.logger.debug(options.staleManifestLogMessage, { + manifestPath: options.manifestPath, + [options.staleManifestLogKeys.expected]: options.expectedHash, + [options.staleManifestLogKeys.actual]: manifest.hash, + }); + return undefined; + } - const bundle = await options.buildBundle(); - if (!bundle) return undefined; + const bundle = await options.buildBundle(); + if (!bundle) { + decision = 'empty'; + return undefined; + } - const payloadPaths = Array.from(bundle.files.keys()).filter( - (path) => path !== options.manifestPath, + const payloadPaths = Array.from(bundle.files.keys()).filter( + (path) => path !== options.manifestPath, + ); + const existenceChecks = await Promise.all( + payloadPaths.map(async (path) => ({ + path, + exists: + (await readWorkspaceFile(options.workspace, path, { + logger: options.logger, + resourceLabel: options.resourceLabel, + })) !== null, + })), + ); + const missingPath = existenceChecks.find((check) => !check.exists)?.path; + if (missingPath) { + decision = 'incomplete'; + options.logger.debug('Ignoring incomplete prebaked workspace bundle', { + manifestPath: options.manifestPath, + missingPath, + }); + return undefined; + } + + options.logger.debug(options.successLogMessage, options.successLogContext(bundle)); + return bundle; + }, ); - const existenceChecks = await Promise.all( - payloadPaths.map(async (path) => ({ - path, - exists: - (await readWorkspaceFile(options.workspace, path, { - logger: options.logger, - resourceLabel: options.resourceLabel, - })) !== null, - })), - ); - const missingPath = existenceChecks.find((check) => !check.exists)?.path; - if (missingPath) { - options.logger.debug('Ignoring incomplete prebaked workspace bundle', { - manifestPath: options.manifestPath, - missingPath, - }); - return undefined; - } - - options.logger.debug(options.successLogMessage, options.successLogContext(bundle)); - return bundle; } export interface MaterializeWorkspaceBundleOptions< @@ -101,26 +126,50 @@ export interface MaterializeWorkspaceBundleOptions< export async function materializeWorkspaceBundle< TBundle extends { files: Map; manifestPath: string }, >(options: MaterializeWorkspaceBundleOptions): Promise { - const prebaked = await options.loadPrebaked(); - if (prebaked) return prebaked; + let reused = false; + return await traceSandboxOperation( + 'materialize-bundle', + { + inputs: { resource: options.resourceLabel }, + processResult: (bundle) => ({ + outputs: { + reused, + fileCount: bundle.files.size, + bytesWritten: reused + ? 0 + : [...bundle.files.values()].reduce( + (sum, content) => sum + sandboxFileBytes(content), + 0, + ), + }, + }), + }, + async () => { + const prebaked = await options.loadPrebaked(); + if (prebaked) { + reused = true; + return prebaked; + } - const bundle = await options.buildBundle(); - const payloadFiles = new Map(bundle.files); - payloadFiles.delete(bundle.manifestPath); + const bundle = await options.buildBundle(); + const payloadFiles = new Map(bundle.files); + payloadFiles.delete(bundle.manifestPath); - await writeWorkspaceFileMap(options.workspace, payloadFiles, { - logger: options.logger, - resourceLabel: options.resourceLabel, - }); + await writeWorkspaceFileMap(options.workspace, payloadFiles, { + logger: options.logger, + resourceLabel: options.resourceLabel, + }); - const manifestContent = bundle.files.get(bundle.manifestPath); - if (manifestContent !== undefined) { - await writeWorkspaceFile(options.workspace, bundle.manifestPath, manifestContent, { - logger: options.logger, - resourceLabel: options.resourceLabel, - }); - } + const manifestContent = bundle.files.get(bundle.manifestPath); + if (manifestContent !== undefined) { + await writeWorkspaceFile(options.workspace, bundle.manifestPath, manifestContent, { + logger: options.logger, + resourceLabel: options.resourceLabel, + }); + } - options.logger.debug(options.materializedLogMessage, options.materializedLogContext(bundle)); - return bundle; + options.logger.debug(options.materializedLogMessage, options.materializedLogContext(bundle)); + return bundle; + }, + ); } diff --git a/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts b/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts index 95598e0ab52..752322621bc 100644 --- a/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts +++ b/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts @@ -19,6 +19,11 @@ import { import { formatErrorForLog } from '../error-formatting'; import type { Logger } from '../logger'; import { getTemplateTelemetrySession } from './template-telemetry'; +import { + traceSandboxOperation, + sandboxCommandTraceResult, + sandboxFileBytes, +} from '../tracing/sandbox-tracing'; export interface SandboxWorkspace extends SharedSandboxWorkspace { filesystem?: { @@ -96,9 +101,13 @@ export async function retryTransientSandboxIo( attempt, error: formatErrorForLog(error), }); - await sleepUntilAbort( - Math.min(baseMs * 2 ** (attempt - 1), SANDBOX_IO_RETRY_BACKOFF_CAP_MS), - options?.abortSignal, + const delayMs = Math.min(baseMs * 2 ** (attempt - 1), SANDBOX_IO_RETRY_BACKOFF_CAP_MS); + await traceSandboxOperation( + 'retry', + { + inputs: { path: filePath, attempt, delayMs, error: formatErrorForLog(error) }, + }, + async () => await sleepUntilAbort(delayMs, options?.abortSignal), ); } } @@ -117,7 +126,18 @@ export async function runInSandbox( command: string, cwdOrOptions?: string | RunInSandboxOptions, ): Promise<{ exitCode: number; stdout: string; stderr: string }> { - const result = await runInSharedSandbox(workspace, command, cwdOrOptions); + const result = await traceSandboxOperation( + 'execute-command', + { + kind: 'io', + inputs: { + commandBytes: sandboxFileBytes(command), + cwd: typeof cwdOrOptions === 'string' ? cwdOrOptions : cwdOrOptions?.cwd, + }, + processResult: sandboxCommandTraceResult, + }, + async () => await runInSharedSandbox(workspace, command, cwdOrOptions), + ); const session = getTemplateTelemetrySession(workspace); if (session) { @@ -143,48 +163,57 @@ export async function writeFileViaSandbox( content: string | Buffer, options?: SandboxIoRetryOptions, ): Promise { - await retryTransientSandboxIo( + return await traceSandboxOperation( + 'write-file', + { + kind: 'io', + inputs: { path: filePath, bytes: sandboxFileBytes(content), transport: 'command' }, + }, async () => { - const runWriteCommand = async (command: string) => { - const result = await runInSandbox(workspace, command, { - abortSignal: options?.abortSignal, - }); - if (result.exitCode !== 0) { - throw new Error(`Failed to write file ${filePath}: ${result.stderr}`); - } - }; + await retryTransientSandboxIo( + async () => { + const runWriteCommand = async (command: string) => { + const result = await runInSandbox(workspace, command, { + abortSignal: options?.abortSignal, + }); + if (result.exitCode !== 0) { + throw new Error(`Failed to write file ${filePath}: ${result.stderr}`); + } + }; - // Ensure parent directory exists - const dir = filePath.substring(0, filePath.lastIndexOf('/')); - if (dir) { - await runWriteCommand(`mkdir -p '${escapeSingleQuotes(dir)}'`); - } + // Ensure parent directory exists + const dir = filePath.substring(0, filePath.lastIndexOf('/')); + if (dir) { + await runWriteCommand(`mkdir -p '${escapeSingleQuotes(dir)}'`); + } - // Encode content as base64, transfer it in small chunks, then decode in the sandbox. - // Some providers run commands through spawn(), where a single huge argument can hit E2BIG. - const b64 = - typeof content === 'string' - ? Buffer.from(content, 'utf-8').toString('base64') - : content.toString('base64'); - const tempPath = `${filePath}.base64.tmp-${Date.now()}-${Math.random().toString(36).slice(2)}`; - const escapedTempPath = escapeSingleQuotes(tempPath); + // Encode content as base64, transfer it in small chunks, then decode in the sandbox. + // Some providers run commands through spawn(), where a single huge argument can hit E2BIG. + const b64 = + typeof content === 'string' + ? Buffer.from(content, 'utf-8').toString('base64') + : content.toString('base64'); + const tempPath = `${filePath}.base64.tmp-${Date.now()}-${Math.random().toString(36).slice(2)}`; + const escapedTempPath = escapeSingleQuotes(tempPath); - await runWriteCommand(`: > '${escapedTempPath}'`); + await runWriteCommand(`: > '${escapedTempPath}'`); - for (let offset = 0; offset < b64.length; offset += BASE64_WRITE_CHUNK_SIZE) { - const chunk = b64.slice(offset, offset + BASE64_WRITE_CHUNK_SIZE); - await runWriteCommand(`printf '%s' '${chunk}' >> '${escapedTempPath}'`); - } + for (let offset = 0; offset < b64.length; offset += BASE64_WRITE_CHUNK_SIZE) { + const chunk = b64.slice(offset, offset + BASE64_WRITE_CHUNK_SIZE); + await runWriteCommand(`printf '%s' '${chunk}' >> '${escapedTempPath}'`); + } - // Decode + cleanup in one shell expression; the exit reflects base64's - // status. Avoid the variable name `status` — it's a read-only builtin in - // zsh, which silently breaks the assignment and loses base64's exit code. - await runWriteCommand( - `base64 -d '${escapedTempPath}' > '${escapeSingleQuotes(filePath)}'; rc=$?; rm -f '${escapedTempPath}'; exit $rc`, + // Decode + cleanup in one shell expression; the exit reflects base64's + // status. Avoid the variable name `status` — it's a read-only builtin in + // zsh, which silently breaks the assignment and loses base64's exit code. + await runWriteCommand( + `base64 -d '${escapedTempPath}' > '${escapeSingleQuotes(filePath)}'; rc=$?; rm -f '${escapedTempPath}'; exit $rc`, + ); + }, + filePath, + options, ); }, - filePath, - options, ); } @@ -198,16 +227,28 @@ export async function readFileViaSandbox( filePath: string, options?: SandboxIoRetryOptions, ): Promise { - const result = await retryTransientSandboxIo( - async () => - await runInSandbox(workspace, `cat '${escapeSingleQuotes(filePath)}' 2>/dev/null`, { - abortSignal: options?.abortSignal, + return await traceSandboxOperation( + 'read-file', + { + kind: 'io', + inputs: { path: filePath, transport: 'command' }, + processResult: (result) => ({ + outputs: { found: result !== null, bytes: result === null ? 0 : sandboxFileBytes(result) }, }), - filePath, - options, + }, + async () => { + const result = await retryTransientSandboxIo( + async () => + await runInSandbox(workspace, `cat '${escapeSingleQuotes(filePath)}' 2>/dev/null`, { + abortSignal: options?.abortSignal, + }), + filePath, + options, + ); + if (result.exitCode !== 0) return null; + return result.stdout; + }, ); - if (result.exitCode !== 0) return null; - return result.stdout; } /** diff --git a/packages/@n8n/instance-ai/src/workspace/sandbox-setup.ts b/packages/@n8n/instance-ai/src/workspace/sandbox-setup.ts index 24d2c36bcc7..e55de513d81 100644 --- a/packages/@n8n/instance-ai/src/workspace/sandbox-setup.ts +++ b/packages/@n8n/instance-ai/src/workspace/sandbox-setup.ts @@ -48,6 +48,7 @@ import { } from './sandbox-fs'; import { joinWorkspacePath } from './workspace-paths'; import { materializeKnowledgeBaseIntoWorkspace } from '../knowledge-base/materialize-knowledge-base'; +import { traceSandboxOperation, sandboxFileBytes } from '../tracing/sandbox-tracing'; const hostRequire = createRequire(__filename); @@ -73,7 +74,7 @@ export class SandboxWorkspaceSetupError extends Error { async function setupStep(step: SandboxWorkspaceSetupStep, action: () => Promise): Promise { try { - return await action(); + return await traceSandboxOperation(step, {}, action); } catch (error) { throw new SandboxWorkspaceSetupError(step, error); } @@ -333,35 +334,52 @@ async function writeWorkspaceFiles( root: string, files: Map, ): Promise { - const filesystem = workspace.filesystem; - if (filesystem) { - // `writeFile` only creates parent dirs as a side-effect of writing a file. - await Promise.all( - ALWAYS_PRESENT_DIRS.map( - async (dir) => - await createWorkspaceDirectory(workspace, filesystem, joinWorkspacePath(root, dir)), - ), - ); - await Promise.all( - [...files].map( - async ([path, content]) => - await writeWorkspaceFile(workspace, filesystem, joinWorkspacePath(root, path), content), - ), - ); - return; - } + return await traceSandboxOperation( + 'write-files', + { + kind: 'batch', + inputs: { + fileCount: files.size, + bytes: [...files.values()].reduce((sum, content) => sum + sandboxFileBytes(content), 0), + }, + }, + async () => { + const filesystem = workspace.filesystem; + if (filesystem) { + // `writeFile` only creates parent dirs as a side-effect of writing a file. + await Promise.all( + ALWAYS_PRESENT_DIRS.map( + async (dir) => + await createWorkspaceDirectory(workspace, filesystem, joinWorkspacePath(root, dir)), + ), + ); + await Promise.all( + [...files].map( + async ([path, content]) => + await writeWorkspaceFile( + workspace, + filesystem, + joinWorkspacePath(root, path), + content, + ), + ), + ); + return; + } - const dirList = ALWAYS_PRESENT_DIRS.map( - (dir) => `'${escapeSingleQuotes(joinWorkspacePath(root, dir))}'`, - ).join(' '); - const result = await runInSandbox(workspace, `mkdir -p ${dirList}`); - if (result.exitCode !== 0) { - throw new Error(`Sandbox setup failed: ${result.stderr}`); - } + const dirList = ALWAYS_PRESENT_DIRS.map( + (dir) => `'${escapeSingleQuotes(joinWorkspacePath(root, dir))}'`, + ).join(' '); + const result = await runInSandbox(workspace, `mkdir -p ${dirList}`); + if (result.exitCode !== 0) { + throw new Error(`Sandbox setup failed: ${result.stderr}`); + } - for (const [path, content] of files) { - await writeFileViaSandbox(workspace, joinWorkspacePath(root, path), content); - } + for (const [path, content] of files) { + await writeFileViaSandbox(workspace, joinWorkspacePath(root, path), content); + } + }, + ); } type WorkspaceFilesystem = NonNullable; @@ -397,7 +415,11 @@ async function writeWorkspaceFile( await filesystem.writeFile(path, content, { recursive: true }); } catch (error) { try { - await writeFileViaSandbox(workspace, path, content); + await traceSandboxOperation( + 'file-command-fallback', + { inputs: { path } }, + async () => await writeFileViaSandbox(workspace, path, content), + ); } catch (fallbackError) { throw new Error( `Failed to write sandbox workspace file "${path}": ${getErrorMessage(error)}; command fallback failed: ${getErrorMessage(fallbackError)}`, @@ -451,87 +473,95 @@ export async function setupSandboxWorkspace( workspace: SandboxWorkspace, context: InstanceAiContext, ): Promise { - const root = await setupStep( - 'resolve-workspace-root', - async () => await getWorkspaceRoot(workspace), - ); - const markerFile = joinWorkspacePath(root, '.sandbox-initialized'); + return await traceSandboxOperation( + 'initialize-workspace', + { processResult: (initialized) => ({ outputs: { initialized, reused: !initialized } }) }, + async () => { + const root = await setupStep( + 'resolve-workspace-root', + async () => await getWorkspaceRoot(workspace), + ); + const markerFile = joinWorkspacePath(root, '.sandbox-initialized'); - // Check marker file for idempotency - const marker = await setupStep( - 'read-initialization-marker', - async () => await readWorkspaceFile(workspace, markerFile), - ); - if (marker !== null) { - await materializeKnowledgeBaseStep(workspace, root, context); - return false; - } + // Check marker file for idempotency + const marker = await setupStep( + 'read-initialization-marker', + async () => await readWorkspaceFile(workspace, markerFile), + ); + if (marker !== null) { + await materializeKnowledgeBaseStep(workspace, root, context); + return false; + } - // ── Collect all files ────────────────────────────────────────────────── + // ── Collect all files ────────────────────────────────────────────────── - const files = new Map(); + const files = new Map(); - files.set('package.json', PACKAGE_JSON); - files.set('tsconfig.json', TSCONFIG_JSON); - files.set('build.mjs', BUILD_MJS); + files.set('package.json', PACKAGE_JSON); + files.set('tsconfig.json', TSCONFIG_JSON); + files.set('build.mjs', BUILD_MJS); - // Node types catalog - const nodeTypes = await setupStep( - 'list-node-types', - async () => await context.nodeService.listSearchable(), - ); - const catalogLines = nodeTypes.map(formatNodeCatalogLine); - files.set('node-types/index.txt', catalogLines.join('\n')); + // Node types catalog + const nodeTypes = await setupStep( + 'list-node-types', + async () => await context.nodeService.listSearchable(), + ); + const catalogLines = nodeTypes.map(formatNodeCatalogLine); + files.set('node-types/index.txt', catalogLines.join('\n')); - // ── Write workspace files ────────────────────────────────────────────── + // ── Write workspace files ────────────────────────────────────────────── - await setupStep( - 'write-workspace-files', - async () => await writeWorkspaceFiles(workspace, root, files), - ); - await materializeKnowledgeBaseStep(workspace, root, context); + await setupStep( + 'write-workspace-files', + async () => await writeWorkspaceFiles(workspace, root, files), + ); + await materializeKnowledgeBaseStep(workspace, root, context); - // npm install (must run after package.json is in place) - await setupStep('install-dependencies', async () => { - // One deadline covers both attempts. The signal stops us waiting; it does not kill - // the remote command, which the sandbox collects at its own timeout. - const deadline = Date.now() + INSTALL_STEP_BUDGET_MS; - const install = async (flags: string) => - await runInSandbox(workspace, `npm install ${flags}`, { - cwd: root, - abortSignal: AbortSignal.timeout(Math.max(1, deadline - Date.now())), + // npm install (must run after package.json is in place) + await setupStep('install-dependencies', async () => { + // One deadline covers both attempts. The signal stops us waiting; it does not kill + // the remote command, which the sandbox collects at its own timeout. + const deadline = Date.now() + INSTALL_STEP_BUDGET_MS; + const install = async (flags: string) => + await runInSandbox(workspace, `npm install ${flags}`, { + cwd: root, + abortSignal: AbortSignal.timeout(Math.max(1, deadline - Date.now())), + }); + + let npmResult = await install(NPM_INSTALL_FLAGS); + if (npmResult.exitCode !== 0 && Date.now() < deadline) { + // The cached packument can be too old to resolve the pinned SDK version. That is + // the one failure the cache causes, and refreshing metadata is the only way out, + // so retry once with whatever budget is left. + context.logger.warn('Sandbox npm install failed against the cache; refreshing metadata', { + stderr: npmResult.stderr.slice(0, 500), + remainingMs: deadline - Date.now(), + }); + npmResult = await install(NPM_INSTALL_FLAGS_REFRESH_METADATA); + } + if (npmResult.exitCode !== 0) { + throw new Error(`Sandbox npm install failed: ${npmResult.stderr}`); + } }); - let npmResult = await install(NPM_INSTALL_FLAGS); - if (npmResult.exitCode !== 0 && Date.now() < deadline) { - // The cached packument can be too old to resolve the pinned SDK version. That is - // the one failure the cache causes, and refreshing metadata is the only way out, - // so retry once with whatever budget is left. - context.logger.warn('Sandbox npm install failed against the cache; refreshing metadata', { - stderr: npmResult.stderr.slice(0, 500), - remainingMs: deadline - Date.now(), - }); - npmResult = await install(NPM_INSTALL_FLAGS_REFRESH_METADATA); - } - if (npmResult.exitCode !== 0) { - throw new Error(`Sandbox npm install failed: ${npmResult.stderr}`); - } - }); + if (isLinkWorkspaceSdkEnabled()) { + await setupStep( + 'link-workspace-sdk', + async () => await linkWorkspaceSdkIfEnabled(workspace, root, context.logger), + ); + } - await setupStep( - 'link-workspace-sdk', - async () => await linkWorkspaceSdkIfEnabled(workspace, root, context.logger), + await setupStep( + 'write-initialization-marker', + async () => + await writeWorkspaceFiles( + workspace, + root, + new Map([['.sandbox-initialized', new Date().toISOString()]]), + ), + ); + + return true; + }, ); - - await setupStep( - 'write-initialization-marker', - async () => - await writeWorkspaceFiles( - workspace, - root, - new Map([['.sandbox-initialized', new Date().toISOString()]]), - ), - ); - - return true; } diff --git a/packages/@n8n/instance-ai/src/workspace/workspace-files.ts b/packages/@n8n/instance-ai/src/workspace/workspace-files.ts index 8f48310d4c5..0a193bc6c1e 100644 --- a/packages/@n8n/instance-ai/src/workspace/workspace-files.ts +++ b/packages/@n8n/instance-ai/src/workspace/workspace-files.ts @@ -9,6 +9,7 @@ import { writeFileViaSandbox, type SandboxWorkspace, } from './sandbox-fs'; +import { traceSandboxOperation, sandboxFileBytes } from '../tracing/sandbox-tracing'; import { isQuotaExhaustedError } from '../utils/quota-error'; export interface WorkspaceFileTarget { @@ -58,55 +59,67 @@ export async function readWorkspaceFile( filePath: string, options?: WorkspaceFileOptions, ): Promise { - const filesystem = workspace.filesystem; - const readFile = filesystem?.readFile; - if (filesystem && readFile) { - try { - return decodeWorkspaceFileContent( - await retryTransientSandboxIo( - // .call preserves the provider's `this` binding (e.g. LazyRuntimeFilesystem). - async () => - await readFile.call(filesystem, filePath, { - encoding: 'utf-8', - abortSignal: options?.abortSignal, - }), - filePath, - options, - ), - ); - } catch (error) { - // A provider failure is not a missing file — surface it instead of reporting null. - if (isTransientSandboxIoError(error)) { - throw new Error( - `Failed to read ${resourceLabel(options).toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, - { cause: error }, - ); + return await traceSandboxOperation( + 'read-file', + { + kind: 'io', + inputs: { path: filePath }, + processResult: (result) => ({ + outputs: { found: result !== null, bytes: result === null ? 0 : sandboxFileBytes(result) }, + }), + }, + async () => { + const filesystem = workspace.filesystem; + const readFile = filesystem?.readFile; + if (filesystem && readFile) { + try { + return decodeWorkspaceFileContent( + await retryTransientSandboxIo( + // .call preserves the provider's `this` binding (e.g. LazyRuntimeFilesystem). + async () => + await readFile.call(filesystem, filePath, { + encoding: 'utf-8', + abortSignal: options?.abortSignal, + }), + filePath, + options, + ), + ); + } catch (error) { + // A provider failure is not a missing file — surface it instead of reporting null. + if (isTransientSandboxIoError(error)) { + throw new Error( + `Failed to read ${resourceLabel(options).toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, + { cause: error }, + ); + } + options?.logger.debug(`${resourceLabel(options)} filesystem read missed`, { + path: filePath, + error: formatErrorForLog(error), + }); + return null; + } } - options?.logger.debug(`${resourceLabel(options)} filesystem read missed`, { - path: filePath, - error: formatErrorForLog(error), - }); - return null; - } - } - if (!workspace.sandbox) return null; + if (!workspace.sandbox) return null; - try { - return await readFileViaSandbox(workspace, filePath, options); - } catch (error) { - if (isTransientSandboxIoError(error)) { - throw new Error( - `Failed to read ${resourceLabel(options).toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, - { cause: error }, - ); - } - options?.logger.debug(`${resourceLabel(options)} command read missed`, { - path: filePath, - error: formatErrorForLog(error), - }); - return null; - } + try { + return await readFileViaSandbox(workspace, filePath, options); + } catch (error) { + if (isTransientSandboxIoError(error)) { + throw new Error( + `Failed to read ${resourceLabel(options).toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, + { cause: error }, + ); + } + options?.logger.debug(`${resourceLabel(options)} command read missed`, { + path: filePath, + error: formatErrorForLog(error), + }); + return null; + } + }, + ); } /** @@ -119,51 +132,61 @@ export async function writeWorkspaceFile( content: string, options?: WorkspaceFileOptions, ): Promise { - const label = resourceLabel(options); + return await traceSandboxOperation( + 'write-file', + { kind: 'io', inputs: { path: filePath, bytes: sandboxFileBytes(content) } }, + async () => { + const label = resourceLabel(options); + + const filesystem = workspace.filesystem; + if (filesystem) { + try { + await retryTransientSandboxIo( + async () => + await filesystem.writeFile(filePath, content, { + recursive: true, + abortSignal: options?.abortSignal, + }), + filePath, + options, + ); + return; + } catch (error) { + if (isAbortError(error)) throw error; + try { + await traceSandboxOperation( + 'file-command-fallback', + { inputs: { path: filePath } }, + async () => await writeFileViaSandbox(workspace, filePath, content, options), + ); + options?.logger.warn(`${label} filesystem write failed; used command fallback`, { + path: filePath, + error: formatErrorForLog(error), + }); + return; + } catch (fallbackError) { + if (isAbortError(fallbackError)) throw fallbackError; + // Preserve whichever path carries quota metadata so callers can + // classify the combined failure correctly. + throw new Error( + `Failed to write ${label.toLowerCase()} "${filePath}": ${formatErrorForLog(error)}; command fallback failed: ${formatErrorForLog(fallbackError)}`, + { cause: selectWriteFailureCause(error, fallbackError) }, + ); + } + } + } - const filesystem = workspace.filesystem; - if (filesystem) { - try { - await retryTransientSandboxIo( - async () => - await filesystem.writeFile(filePath, content, { - recursive: true, - abortSignal: options?.abortSignal, - }), - filePath, - options, - ); - return; - } catch (error) { - if (isAbortError(error)) throw error; try { await writeFileViaSandbox(workspace, filePath, content, options); - options?.logger.warn(`${label} filesystem write failed; used command fallback`, { - path: filePath, - error: formatErrorForLog(error), - }); - return; - } catch (fallbackError) { - if (isAbortError(fallbackError)) throw fallbackError; - // Preserve whichever path carries quota metadata so callers can - // classify the combined failure correctly. + } catch (error) { + if (isAbortError(error)) throw error; throw new Error( - `Failed to write ${label.toLowerCase()} "${filePath}": ${formatErrorForLog(error)}; command fallback failed: ${formatErrorForLog(fallbackError)}`, - { cause: selectWriteFailureCause(error, fallbackError) }, + `Failed to write ${label.toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, + { cause: error }, ); } - } - } - - try { - await writeFileViaSandbox(workspace, filePath, content, options); - } catch (error) { - if (isAbortError(error)) throw error; - throw new Error( - `Failed to write ${label.toLowerCase()} "${filePath}": ${formatErrorForLog(error)}`, - { cause: error }, - ); - } + }, + ); } export async function writeWorkspaceFileMap( @@ -171,9 +194,21 @@ export async function writeWorkspaceFileMap( files: Map, options?: WorkspaceFileOptions, ): Promise { - await Promise.all( - Array.from(files, async ([filePath, content]) => { - await writeWorkspaceFile(workspace, filePath, content, options); - }), + return await traceSandboxOperation( + 'write-files', + { + kind: 'batch', + inputs: { + fileCount: files.size, + bytes: [...files.values()].reduce((sum, content) => sum + sandboxFileBytes(content), 0), + }, + }, + async () => { + await Promise.all( + Array.from(files, async ([filePath, content]) => { + await writeWorkspaceFile(workspace, filePath, content, options); + }), + ); + }, ); } diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-sandbox.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-sandbox.service.test.ts index 302a3d0cf07..d2b1df266ec 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-sandbox.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-sandbox.service.test.ts @@ -9,12 +9,28 @@ vi.mock('@n8n/instance-ai', () => ({ createSandbox: vi.fn(), createWorkspace: vi.fn(), setupSandboxWorkspace: vi.fn(), + traceSandboxOperation: vi.fn( + async (_operation: string, _options: unknown, fn: () => Promise) => await fn(), + ), + withSandboxLifecycleTrace: vi.fn( + async ( + _threadId: string, + _operation: string, + _inputs: unknown, + fn: () => Promise, + options?: { resolveConfig?: () => Promise }, + ) => { + await options?.resolveConfig?.(); + return await fn(); + }, + ), })); import { createSandbox, createWorkspace, setupSandboxWorkspace, + withSandboxLifecycleTrace, type InstanceAiContext, type ManagedBackgroundTask, } from '@n8n/instance-ai'; @@ -31,6 +47,7 @@ import { const fakeUser = { id: 'user-1' } as User; type Overrides = { + resolveTracingConfig?: InstanceAiSandboxServiceOptions['resolveTracingConfig']; config?: Partial; runState?: Partial; backgroundTasks?: Partial; @@ -68,6 +85,7 @@ function createSandboxService(overrides: Overrides = {}) { backgroundTasks, settingsService, aiService, + resolveTracingConfig: overrides.resolveTracingConfig, }; const service = new InstanceAiSandboxService(options); return { service, logger, errorReporter, runState, backgroundTasks, settingsService, aiService }; @@ -835,6 +853,9 @@ describe('InstanceAiSandboxService', () => { expect(entry).toBeDefined(); vi.advanceTimersByTime(1000); + expect(vi.mocked(withSandboxLifecycleTrace).mock.calls[0]?.[4]).toMatchObject({ + detached: true, + }); // Eviction drops the cache entry but never destroys the remote workspace. expect(workspace.destroy).not.toHaveBeenCalled(); @@ -904,6 +925,36 @@ describe('InstanceAiSandboxService', () => { }); describe('destroySandbox', () => { + it.each([true, false])( + 'keeps the owner available for cleanup when cached=%s', + async (cached) => { + const resolveTracingConfig = vi.fn(async (_threadId: string, userId?: string) => { + if (!userId) throw new Error('Thread row is no longer available'); + return { userId }; + }); + const { service } = createSandboxService({ + config: { + sandboxEnabled: true, + sandboxProvider: 'n8n-sandbox', + n8nSandboxServiceUrl: 'http://sandbox.example', + }, + resolveTracingConfig, + }); + const destroy = vi.fn(async () => {}); + (createSandbox as Mock).mockResolvedValue({ id: 'sandbox-1', destroy }); + (createWorkspace as Mock).mockReturnValue({ init: vi.fn(async () => {}), destroy }); + if (cached) await service.getOrCreateWorkspaceEntry('thread-1', fakeUser); + await service.destroySandbox( + 'thread-1', + 'thread_cleanup', + cached ? undefined : fakeUser.id, + ); + expect(destroy).toHaveBeenCalledTimes(1); + expect(resolveTracingConfig).toHaveBeenCalledWith('thread-1', fakeUser.id); + service.stopSandboxExpiryTimers(); + }, + ); + it('destroys and removes the workspace for a thread', async () => { const { service } = createSandboxService({ config: { sandboxEnabled: true, sandboxProvider: 'daytona' }, @@ -983,16 +1034,35 @@ describe('InstanceAiSandboxService', () => { expect.any(Object), ); expect(sandbox.destroy).toHaveBeenCalledTimes(1); + expect(withSandboxLifecycleTrace).toHaveBeenCalledTimes(1); }); - it('does not attempt an uncached destroy for the daytona provider', async () => { - const { service } = createSandboxService({ - config: { sandboxEnabled: true, sandboxProvider: 'daytona' }, - }); - + it.each([ + { sandboxEnabled: false, sandboxProvider: 'n8n-sandbox' }, + { sandboxEnabled: false, sandboxProvider: 'daytona' }, + { sandboxEnabled: true, sandboxProvider: 'daytona' }, + ])('skips uncached cleanup and tracing for %j', async (config) => { + const resolveTracingConfig = vi.fn(async () => await Promise.resolve({ userId: 'owner' })); + const { service } = createSandboxService({ config, resolveTracingConfig }); await service.destroySandbox('thread-1'); - expect(createSandbox).not.toHaveBeenCalled(); + expect(withSandboxLifecycleTrace).not.toHaveBeenCalled(); + expect(resolveTracingConfig).not.toHaveBeenCalled(); + }); + + it('traces uncached cleanup when required configuration is missing', async () => { + const { service, logger } = createSandboxService({ + config: { sandboxEnabled: true, sandboxProvider: 'n8n-sandbox', n8nSandboxServiceUrl: '' }, + }); + await service.destroySandbox('thread-1'); + expect(withSandboxLifecycleTrace).toHaveBeenCalledTimes(1); + expect(createSandbox).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledWith('Failed to destroy sandbox', { + threadId: 'thread-1', + reason: 'thread_cleanup', + error: + 'N8N_SANDBOX_SERVICE_URL is required when Instance AI sandbox provider is n8n-sandbox.', + }); }); it('swallows uncached destroy errors and logs a warning', async () => { diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts index d97f8ec3f19..c98015c64ca 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts @@ -1486,7 +1486,7 @@ describe('InstanceAiController', () => { const result = await controller.deleteThread(req, res, THREAD_ID); expect(result).toEqual({ ok: true }); - expect(instanceAiService.routeClearThreadState).toHaveBeenCalledWith(THREAD_ID); + expect(instanceAiService.routeClearThreadState).toHaveBeenCalledWith(THREAD_ID, USER_ID); expect(memoryService.deleteThread).toHaveBeenCalledWith(THREAD_ID); }); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts index f41cb4cb9ee..b844f1ac653 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts @@ -50,6 +50,13 @@ vi.mock('@n8n/instance-ai', async () => { ), createLazyWorkspaceRuntimeSkillSource: vi.fn(({ source }) => source), setupSandboxWorkspace: vi.fn(), + traceSandboxOperation: vi.fn( + async (_operation: string, _options: unknown, fn: () => Promise) => await fn(), + ), + withSandboxLifecycleTrace: vi.fn( + async (_threadId: string, _operation: string, _inputs: unknown, fn: () => Promise) => + await fn(), + ), loadInstanceAiRuntimeSkillSource: vi.fn(() => ({ registry: { skillsHash: 'runtime-skills-hash', @@ -5298,12 +5305,12 @@ describe('InstanceAiService — cross-main task-control routing', () => { it('routeClearThreadState clears locally and fans out in multi-main', async () => { const service = buildTaskControlService(true); - await service.routeClearThreadState('thread-a'); + await service.routeClearThreadState('thread-a', 'owner-a'); - expect(service.clearThreadState).toHaveBeenCalledWith('thread-a'); + expect(service.clearThreadState).toHaveBeenCalledWith('thread-a', 'owner-a'); expect(service.publisher.publishCommand).toHaveBeenCalledWith({ command: 'relay-instance-ai-task-control', - payload: { threadId: 'thread-a', action: 'clear-thread' }, + payload: { threadId: 'thread-a', action: 'clear-thread', userId: 'owner-a' }, }); }); }); @@ -5335,22 +5342,29 @@ describe('InstanceAiService — cross-main task-control routing', () => { expect(service.sendCorrectionToTask).not.toHaveBeenCalled(); }); - it('routes cancel-task / cancel-thread / clear-thread to the local methods', async () => { - const service = buildTaskControlService(true); + it.each(['owner-a', undefined])( + 'routes task controls with owner %s to the local methods', + async (userId) => { + const service = buildTaskControlService(true); - await service.handleRelayTaskControl({ - threadId: 'thread-a', - taskId: 'task-1', - action: 'cancel-task', - }); - await service.handleRelayTaskControl({ threadId: 'thread-a', action: 'cancel-thread' }); - await service.handleRelayTaskControl({ threadId: 'thread-a', action: 'clear-thread' }); + await service.handleRelayTaskControl({ + threadId: 'thread-a', + taskId: 'task-1', + action: 'cancel-task', + }); + await service.handleRelayTaskControl({ threadId: 'thread-a', action: 'cancel-thread' }); + await service.handleRelayTaskControl({ + threadId: 'thread-a', + action: 'clear-thread', + userId, + }); - expect(service.cancelBackgroundTask).toHaveBeenCalledWith('thread-a', 'task-1'); - expect(service.cancelRun).toHaveBeenCalledWith('thread-a'); - expect(service.clearThreadState).toHaveBeenCalledWith('thread-a'); - expect(service.publisher.publishCommand).not.toHaveBeenCalled(); - }); + expect(service.cancelBackgroundTask).toHaveBeenCalledWith('thread-a', 'task-1'); + expect(service.cancelRun).toHaveBeenCalledWith('thread-a'); + expect(service.clearThreadState).toHaveBeenCalledWith('thread-a', userId); + expect(service.publisher.publishCommand).not.toHaveBeenCalled(); + }, + ); it('swallows and logs errors from a local action (no unhandled rejection on the sibling main)', async () => { const service = buildTaskControlService(true); diff --git a/packages/cli/src/modules/instance-ai/instance-ai.controller.ts b/packages/cli/src/modules/instance-ai/instance-ai.controller.ts index cb28945ab08..c502c18bbd4 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.controller.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.controller.ts @@ -853,7 +853,7 @@ export class InstanceAiController { ) { this.requireInstanceAiEnabled(); await this.assertThreadAccess(req.user.id, threadId); - await this.instanceAiService.routeClearThreadState(threadId); + await this.instanceAiService.routeClearThreadState(threadId, req.user.id); await this.memoryService.deleteThread(threadId); return { ok: true }; } diff --git a/packages/cli/src/modules/instance-ai/instance-ai.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.service.ts index ba8ae318da7..f3f41b1c15c 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -912,6 +912,12 @@ export class InstanceAiService { backgroundTasks: this.backgroundTasks, settingsService: this.settingsService, aiService: this.aiService, + resolveTracingConfig: async (threadId, userId) => { + const ownerId = userId ?? (await this.agentMemory.getThread(threadId))?.resourceId; + if (!ownerId) return { userId: 'system' }; + const { tracingProxyConfig } = await this.createProxyRunConfig({ id: ownerId }); + return { userId: ownerId, proxyConfig: tracingProxyConfig }; + }, }); this.terminalOutcome = new InstanceAiTerminalOutcomeService({ // The terminal guard and outcome-replay dedup must see the run's events @@ -958,7 +964,7 @@ export class InstanceAiService { this.liveness.start(); } - private async createProxyRunConfig(user: User): Promise<{ + private async createProxyRunConfig(user: Pick): Promise<{ searchProxyConfig?: ServiceProxyConfig; tracingProxyConfig?: ServiceProxyConfig; tokenManager?: ProxyTokenManager; @@ -1597,6 +1603,7 @@ export class InstanceAiService { taskId, action, correction, + userId, }: PubSubCommandMap['relay-instance-ai-task-control']): Promise { switch (action) { case 'correct': @@ -1616,7 +1623,7 @@ export class InstanceAiService { this.cancelRun(threadId); return false; case 'clear-thread': - await this.clearThreadState(threadId); + await this.clearThreadState(threadId, userId); return false; } } @@ -1655,8 +1662,12 @@ export class InstanceAiService { } } - async routeClearThreadState(threadId: string): Promise { - await this.routeTaskControl({ threadId, action: 'clear-thread' }); + async routeClearThreadState(threadId: string, userId?: string): Promise { + await this.routeTaskControl({ + threadId, + action: 'clear-thread', + ...(userId ? { userId } : {}), + }); } /** Apply a task-control action relayed from another main to this main's local @@ -1845,7 +1856,7 @@ export class InstanceAiService { * Remove all in-memory state associated with a thread. * Must be called when a thread is deleted so the maps don't leak. */ - async clearThreadState(threadId: string): Promise { + async clearThreadState(threadId: string, userId?: string): Promise { this.liveness.clearThreadState(threadId); // Clear run-state registry entries (active/suspended runs, confirmations, @@ -1886,7 +1897,7 @@ export class InstanceAiService { this.memoryTaskRegistry.clearThread(threadId); this.tracing.deleteTraceContextsForThread(threadId); await this.deleteAgentBuilderSessions(threadId); - await this.sandboxService.destroySandbox(threadId); + await this.sandboxService.destroySandbox(threadId, 'thread_cleanup', userId); await this.temporaryWorkflowService.reapForThreadCleanup(threadId); await this.suspendedThreads.dropPendingConfirmationsForThread(threadId); this.eventBus.clearThread(threadId); diff --git a/packages/cli/src/modules/instance-ai/sandbox/instance-ai-sandbox.service.ts b/packages/cli/src/modules/instance-ai/sandbox/instance-ai-sandbox.service.ts index e08938f5240..68c6d493b3c 100644 --- a/packages/cli/src/modules/instance-ai/sandbox/instance-ai-sandbox.service.ts +++ b/packages/cli/src/modules/instance-ai/sandbox/instance-ai-sandbox.service.ts @@ -7,6 +7,9 @@ import { createSandbox, createWorkspace, setupSandboxWorkspace, + traceSandboxOperation, + withSandboxLifecycleTrace, + type ServiceProxyConfig, type InstanceAiContext, type Logger, type ManagedBackgroundTask, @@ -29,6 +32,7 @@ const DEFAULT_SANDBOX_TTL_MS = 15 * 60 * 1000; /** Cached runtime sandbox + workspace pair for a single thread. */ export type RuntimeSandboxEntry = { + userId: string; sandbox: NonNullable>>; workspace: NonNullable>; configFingerprint: string; @@ -164,6 +168,10 @@ export type InstanceAiSandboxServiceOptions = { backgroundTasks: InstanceAiSandboxBackgroundTasks; settingsService: InstanceAiSandboxSettings; aiService: InstanceAiSandboxProxy; + resolveTracingConfig?: ( + threadId: string, + userId?: string, + ) => Promise<{ userId: string; proxyConfig?: ServiceProxyConfig }>; }; /** @@ -324,46 +332,74 @@ export class InstanceAiSandboxService { threadId: string, user: User, ): Promise { - const cacheGeneration = this.cacheGeneration; - const cacheState = await this.resolveSandboxCacheState(user); - if (cacheGeneration !== this.cacheGeneration) { - return await this.getOrCreateWorkspaceEntry(threadId, user); - } - const existing = this.sandboxes.get(threadId); - if (existing) { - if ( - existing.configFingerprint !== cacheState.fingerprint || - (this.isSandboxEntryExpired(existing) && !this.isSandboxInUse(threadId)) - ) { - this.evictSandboxEntry(threadId, existing); - } else { - this.touchSandboxEntry(threadId, existing); - return existing; - } - } + let outcome = 'acquired'; + return await traceSandboxOperation( + 'acquire', + { + inputs: { threadId }, + processResult: (entry) => ({ + outputs: { + outcome: entry ? outcome : 'disabled', + sandboxId: entry?.sandbox.id, + provider: entry?.sandbox.provider, + }, + }), + }, + async () => { + const cacheGeneration = this.cacheGeneration; + const cacheState = await traceSandboxOperation( + 'resolve-config', + { inputs: { threadId } }, + async () => await this.resolveSandboxCacheState(user), + ); + if (cacheGeneration !== this.cacheGeneration) { + return await this.getOrCreateWorkspaceEntry(threadId, user); + } + const existing = this.sandboxes.get(threadId); + if (existing) { + if ( + existing.configFingerprint !== cacheState.fingerprint || + (this.isSandboxEntryExpired(existing) && !this.isSandboxInUse(threadId)) + ) { + this.evictSandboxEntry( + threadId, + existing, + existing.configFingerprint !== cacheState.fingerprint ? 'config_changed' : 'idle', + ); + } else { + this.touchSandboxEntry(threadId, existing); + outcome = 'cached'; + return existing; + } + } - const pending = this.sandboxCreations.get(threadId); - if (pending?.fingerprint === cacheState.fingerprint) return await pending.promise; + const pending = this.sandboxCreations.get(threadId); + if (pending?.fingerprint === cacheState.fingerprint) { + outcome = 'shared_acquisition'; + return await pending.promise; + } - const creation = this.createWorkspaceEntry(threadId, user, cacheState); - const pendingCreation = { fingerprint: cacheState.fingerprint, promise: creation }; - this.sandboxCreations.set(threadId, pendingCreation); - try { - const entry = await creation; - if ( - entry && - cacheGeneration === this.cacheGeneration && - this.sandboxCreations.get(threadId) === pendingCreation - ) { - this.sandboxes.set(threadId, entry); - this.scheduleSandboxExpiry(threadId, entry); - } - return entry; - } finally { - if (this.sandboxCreations.get(threadId) === pendingCreation) { - this.sandboxCreations.delete(threadId); - } - } + const creation = this.createWorkspaceEntry(threadId, user, cacheState); + const pendingCreation = { fingerprint: cacheState.fingerprint, promise: creation }; + this.sandboxCreations.set(threadId, pendingCreation); + try { + const entry = await creation; + if ( + entry && + cacheGeneration === this.cacheGeneration && + this.sandboxCreations.get(threadId) === pendingCreation + ) { + this.sandboxes.set(threadId, entry); + this.scheduleSandboxExpiry(threadId, entry); + } + return entry; + } finally { + if (this.sandboxCreations.get(threadId) === pendingCreation) { + this.sandboxCreations.delete(threadId); + } + } + }, + ); } /** Get or create the shared runtime sandbox + workspace for a thread. */ @@ -413,10 +449,18 @@ export class InstanceAiSandboxService { const workspace = createWorkspace(sandbox); if (!sandbox || !workspace) return undefined; try { - await workspace.init(); + await traceSandboxOperation( + 'start', + { inputs: { threadId, sandboxId: sandbox.id, provider: sandbox.provider } }, + async () => await workspace.init(), + ); } catch (error) { try { - await workspace.destroy(); + await traceSandboxOperation( + 'cleanup-failed-start', + { inputs: { threadId, sandboxId: sandbox.id } }, + async () => await workspace.destroy(), + ); } catch { // Best-effort cleanup when the sandbox cannot start } @@ -433,6 +477,7 @@ export class InstanceAiSandboxService { } const entry: RuntimeSandboxEntry = { + userId: user.id, sandbox, workspace, configFingerprint: cacheState.fingerprint, @@ -465,7 +510,12 @@ export class InstanceAiSandboxService { } } - private evictSandboxEntry(threadId: string, entry: RuntimeSandboxEntry): void { + private evictSandboxEntry( + threadId: string, + entry: RuntimeSandboxEntry, + reason = 'settings_changed', + detached = false, + ): void { if (this.sandboxes.get(threadId) !== entry) return; this.sandboxes.delete(threadId); @@ -473,19 +523,44 @@ export class InstanceAiSandboxService { clearTimeout(entry.cleanupTimer); entry.cleanupTimer = undefined; } + void withSandboxLifecycleTrace( + threadId, + 'evict-cache', + { + reason, + sandboxId: entry.sandbox.id, + provider: entry.sandbox.provider, + occurredAt: new Date().toISOString(), + remoteDestroyed: false, + }, + async () => {}, + { resolveConfig: this.lifecycleTraceConfig(threadId, entry.userId), detached }, + ).catch(() => { + // Cache eviction does not depend on trace delivery. + }); } /** Destroy and remove the shared runtime workspace for a thread. */ - async destroySandbox(threadId: string, reason = 'thread_cleanup'): Promise { + async destroySandbox( + threadId: string, + reason = 'thread_cleanup', + userId?: string, + ): Promise { const entry = this.sandboxes.get(threadId); if (!entry?.sandbox) { - await this.destroyUncachedSandbox(threadId, reason); + await this.destroyUncachedSandbox(threadId, reason, userId); return; } - this.evictSandboxEntry(threadId, entry); + this.evictSandboxEntry(threadId, entry, reason); try { - await entry.workspace?.destroy(); + await withSandboxLifecycleTrace( + threadId, + 'destroy', + { reason, sandboxId: entry.sandbox.id, provider: entry.sandbox.provider }, + async () => await entry.workspace.destroy(), + { resolveConfig: this.lifecycleTraceConfig(threadId, entry.userId) }, + ); } catch (error) { this.logger.warn('Failed to destroy sandbox', { threadId, @@ -502,26 +577,50 @@ export class InstanceAiSandboxService { * that never existed is a cheap 404. Daytona is left to its own * auto-stop/auto-delete lifecycle. */ - private async destroyUncachedSandbox(threadId: string, reason: string): Promise { + private async destroyUncachedSandbox( + threadId: string, + reason: string, + userId?: string, + ): Promise { try { - const base = this.getSandboxConfigFromEnv(); - if (!base.enabled || base.provider !== 'n8n-sandbox') return; + if ( + !this.instanceAiConfig.sandboxEnabled || + normalizeSandboxProvider(this.instanceAiConfig.sandboxProvider) !== 'n8n-sandbox' + ) + return; - const settings = await this.options.settingsService.resolveN8nSandboxConfig(); - const config = withThreadScopedSandboxIdentity( - { - ...base, - serviceUrl: settings.serviceUrl ?? base.serviceUrl, - apiKey: settings.apiKey ?? base.apiKey, - }, + await withSandboxLifecycleTrace( threadId, + 'destroy', + { reason, cached: false }, + async () => { + const base = this.getSandboxConfigFromEnv(); + if (!base.enabled || base.provider !== 'n8n-sandbox') return; + + const settings = await this.options.settingsService.resolveN8nSandboxConfig(); + const config = withThreadScopedSandboxIdentity( + { + ...base, + serviceUrl: settings.serviceUrl ?? base.serviceUrl, + apiKey: settings.apiKey ?? base.apiKey, + }, + threadId, + ); + // Constructing the adapter makes no remote calls; destroy() issues the delete. + const sandbox = await createSandbox(config, { + logger: this.logger, + errorReporter: this.options.errorReporter, + }); + await traceSandboxOperation( + 'destroy-remote', + { + inputs: { sandboxId: sandbox?.id, provider: base.provider }, + }, + async () => await sandbox?.destroy?.(), + ); + }, + { resolveConfig: this.lifecycleTraceConfig(threadId, userId) }, ); - // Constructing the adapter makes no remote calls; destroy() issues the delete. - const sandbox = await createSandbox(config, { - logger: this.logger, - errorReporter: this.options.errorReporter, - }); - await sandbox?.destroy?.(); } catch (error) { this.logger.warn('Failed to destroy sandbox', { threadId, @@ -531,6 +630,11 @@ export class InstanceAiSandboxService { } } + private lifecycleTraceConfig(threadId: string, userId?: string) { + const resolve = this.options.resolveTracingConfig; + return resolve ? async () => await resolve(threadId, userId) : undefined; + } + private get sandboxTtlMs(): number { return this.instanceAiConfig?.builderSandboxTtlMs ?? DEFAULT_SANDBOX_TTL_MS; } @@ -573,7 +677,7 @@ export class InstanceAiSandboxService { this.touchSandboxEntry(threadId, entry); return; } - this.evictSandboxEntry(threadId, entry); + this.evictSandboxEntry(threadId, entry, 'idle', true); }, delay); entry.cleanupTimer.unref(); } diff --git a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts index 6f53819226a..92e267b658a 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts @@ -217,6 +217,7 @@ export type PubSubCommandMap = { */ 'relay-instance-ai-task-control': { threadId: string; + userId?: string; taskId?: string; action: 'correct' | 'cancel-task' | 'cancel-thread' | 'clear-thread'; correction?: string;