mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
feat(core): Trace AI Assistant sandbox operations (no-changelog) (#38170)
This commit is contained in:
parent
a58b24f2c4
commit
0380ffccfd
|
|
@ -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 |
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<KnowledgeBaseWorkspaceBundle> {
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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<MaterializedRuntimeSkills | undefined> {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<WorkflowSourceCompileResult> {
|
||||
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],
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Record<string, unknown>>(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<Record<string, unknown>>(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<number>();
|
||||
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<Record<string, string>>();
|
||||
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', () => {
|
||||
|
|
|
|||
|
|
@ -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<string, OtelApiSpan>;
|
||||
contexts: Map<string, OtelContext>;
|
||||
pendingOperations: Map<string, InstanceAiTraceRun>;
|
||||
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<T = unknown> {
|
|||
metadata?: Record<string, unknown>;
|
||||
inputs?: unknown;
|
||||
processOutputs?: (result: T) => unknown;
|
||||
processResult?: (
|
||||
result: T,
|
||||
) => InstanceAiTraceRunFinishOptions | Promise<InstanceAiTraceRunFinishOptions>;
|
||||
}
|
||||
|
||||
type NativeToolContext = ToolContext | InterruptibleToolContext;
|
||||
|
|
@ -904,7 +918,7 @@ export async function withCurrentTraceSpan<T>(
|
|||
fn: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const currentProductTrace = getCurrentProductTrace();
|
||||
if (!currentProductTrace) {
|
||||
if (!currentProductTrace || currentProductTrace.runtime.shutdown) {
|
||||
return await fn();
|
||||
}
|
||||
|
||||
|
|
@ -919,7 +933,7 @@ export async function withCurrentTraceSpan<T>(
|
|||
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<T>(
|
|||
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<InstanceAiTraceContext | undefined> {
|
||||
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<T>(
|
||||
threadId: string,
|
||||
operation: string,
|
||||
inputs: Record<string, unknown>,
|
||||
fn: () => Promise<T>,
|
||||
options?: SandboxLifecycleTraceOptions,
|
||||
): Promise<T> {
|
||||
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<InstanceAiTraceContext | undefined> {
|
||||
if (!isLangSmithTracingEnabled(!!options.proxyConfig)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
|
|
|
|||
81
packages/@n8n/instance-ai/src/tracing/sandbox-tracing.ts
Normal file
81
packages/@n8n/instance-ai/src/tracing/sandbox-tracing.ts
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
import { AsyncLocalStorage } from 'node:async_hooks';
|
||||
import { posix } from 'node:path';
|
||||
|
||||
import type { InstanceAiTraceRunFinishOptions } from '../types';
|
||||
|
||||
const ioTraceScope = new AsyncLocalStorage<boolean>();
|
||||
|
||||
interface SandboxTraceOptions<T> {
|
||||
inputs?: Record<string, unknown>;
|
||||
/** Batch spans summarize file transfers. I/O spans suppress transport details. */
|
||||
kind?: 'batch' | 'io';
|
||||
processResult?: (
|
||||
result: T,
|
||||
) => InstanceAiTraceRunFinishOptions | Promise<InstanceAiTraceRunFinishOptions>;
|
||||
}
|
||||
|
||||
export async function traceSandboxOperation<T>(
|
||||
operation: string,
|
||||
options: SandboxTraceOptions<T>,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
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<InstanceAiTraceRunFinishOptions> {
|
||||
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 } : {}),
|
||||
};
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<CommandResult>((_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
|
||||
|
|
|
|||
|
|
@ -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<void>
|
||||
>(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<void>>(
|
||||
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<
|
||||
|
|
|
|||
|
|
@ -22,6 +22,13 @@ import {
|
|||
type WriteOptions,
|
||||
} from '@n8n/agents';
|
||||
|
||||
import {
|
||||
traceSandboxOperation,
|
||||
sandboxFileBytes,
|
||||
sandboxTracePath,
|
||||
sandboxCommandTraceResult,
|
||||
} from '../tracing/sandbox-tracing';
|
||||
|
||||
export type RuntimeWorkspaceResolver = () => Promise<Workspace | undefined>;
|
||||
|
||||
export interface LazyRuntimeWorkspaceOptions {
|
||||
|
|
@ -219,35 +226,103 @@ class LazyRuntimeFilesystem extends BaseFilesystem {
|
|||
}
|
||||
|
||||
async readFile(path: string, options?: ReadOptions): Promise<string | Buffer> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<FileEntry[]> {
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<TBundle> {
|
||||
workspace: WorkspaceFileTarget;
|
||||
|
|
@ -30,59 +31,83 @@ export interface LoadPrebakedWorkspaceBundleOptions<TBundle> {
|
|||
export async function loadPrebakedWorkspaceBundle<TBundle extends { files: Map<string, string> }>(
|
||||
options: LoadPrebakedWorkspaceBundleOptions<TBundle>,
|
||||
): Promise<TBundle | undefined> {
|
||||
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<string, string>; manifestPath: string },
|
||||
>(options: MaterializeWorkspaceBundleOptions<TBundle>): Promise<TBundle> {
|
||||
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;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<T>(
|
|||
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<void> {
|
||||
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<string | null> {
|
||||
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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<T>(step: SandboxWorkspaceSetupStep, action: () => Promise<T>): Promise<T> {
|
||||
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<string, string>,
|
||||
): Promise<void> {
|
||||
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<SandboxWorkspace['filesystem']>;
|
||||
|
|
@ -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<boolean> {
|
||||
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<string, string>();
|
||||
const files = new Map<string, string>();
|
||||
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string | null> {
|
||||
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<void> {
|
||||
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<string, string>,
|
||||
options?: WorkspaceFileOptions,
|
||||
): Promise<void> {
|
||||
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);
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,12 +9,28 @@ vi.mock('@n8n/instance-ai', () => ({
|
|||
createSandbox: vi.fn(),
|
||||
createWorkspace: vi.fn(),
|
||||
setupSandboxWorkspace: vi.fn(),
|
||||
traceSandboxOperation: vi.fn(
|
||||
async <T>(_operation: string, _options: unknown, fn: () => Promise<T>) => await fn(),
|
||||
),
|
||||
withSandboxLifecycleTrace: vi.fn(
|
||||
async <T>(
|
||||
_threadId: string,
|
||||
_operation: string,
|
||||
_inputs: unknown,
|
||||
fn: () => Promise<T>,
|
||||
options?: { resolveConfig?: () => Promise<unknown> },
|
||||
) => {
|
||||
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<InstanceAiConfig>;
|
||||
runState?: Partial<InstanceAiSandboxRunState>;
|
||||
backgroundTasks?: Partial<InstanceAiSandboxBackgroundTasks>;
|
||||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -50,6 +50,13 @@ vi.mock('@n8n/instance-ai', async () => {
|
|||
),
|
||||
createLazyWorkspaceRuntimeSkillSource: vi.fn(({ source }) => source),
|
||||
setupSandboxWorkspace: vi.fn(),
|
||||
traceSandboxOperation: vi.fn(
|
||||
async <T>(_operation: string, _options: unknown, fn: () => Promise<T>) => await fn(),
|
||||
),
|
||||
withSandboxLifecycleTrace: vi.fn(
|
||||
async <T>(_threadId: string, _operation: string, _inputs: unknown, fn: () => Promise<T>) =>
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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 };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<User, 'id'>): 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<boolean> {
|
||||
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<void> {
|
||||
await this.routeTaskControl({ threadId, action: 'clear-thread' });
|
||||
async routeClearThreadState(threadId: string, userId?: string): Promise<void> {
|
||||
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<void> {
|
||||
async clearThreadState(threadId: string, userId?: string): Promise<void> {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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<Awaited<ReturnType<typeof createSandbox>>>;
|
||||
workspace: NonNullable<ReturnType<typeof createWorkspace>>;
|
||||
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<RuntimeSandboxEntry | undefined> {
|
||||
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<void> {
|
||||
async destroySandbox(
|
||||
threadId: string,
|
||||
reason = 'thread_cleanup',
|
||||
userId?: string,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
private async destroyUncachedSandbox(
|
||||
threadId: string,
|
||||
reason: string,
|
||||
userId?: string,
|
||||
): Promise<void> {
|
||||
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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user