diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index deb3beba9fb..6fd7b98f3ad 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -62,7 +62,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.insights_raw](public.insights_raw.md) | 5 | | BASE TABLE | | [public.installed_nodes](public.installed_nodes.md) | 4 | | BASE TABLE | | [public.installed_packages](public.installed_packages.md) | 6 | | BASE TABLE | -| [public.instance_ai_checkpoints](public.instance_ai_checkpoints.md) | 8 | | BASE TABLE | +| [public.instance_ai_checkpoints](public.instance_ai_checkpoints.md) | 9 | | BASE TABLE | | [public.instance_ai_events](public.instance_ai_events.md) | 7 | | BASE TABLE | | [public.instance_ai_iteration_logs](public.instance_ai_iteration_logs.md) | 6 | | BASE TABLE | | [public.instance_ai_mcp_registry_connections](public.instance_ai_mcp_registry_connections.md) | 7 | | BASE TABLE | @@ -819,6 +819,7 @@ erDiagram "public.instance_ai_checkpoints" { timestamp_3__with_time_zone createdAt timestamp_3__with_time_zone expiredAt + varchar_64_ hostRunId varchar_255_ key varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/postgres-schema/public.instance_ai_checkpoints.md b/docs/generated/postgres-schema/public.instance_ai_checkpoints.md index fccf53564d5..31a71035aad 100644 --- a/docs/generated/postgres-schema/public.instance_ai_checkpoints.md +++ b/docs/generated/postgres-schema/public.instance_ai_checkpoints.md @@ -6,6 +6,7 @@ | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | | expiredAt | timestamp(3) with time zone | | true | | | Soft-delete timestamp: null means live; non-null marks the row as a tombstone. | +| hostRunId | varchar(64) | | true | | | Host (Instance AI) run id; distinct from the agent-SDK runId parsed from the checkpoint key. | | key | varchar(255) | | false | [public.instance_ai_pending_confirmations](public.instance_ai_pending_confirmations.md) | | Opaque checkpoint key from the agent runtime. | | resourceId | varchar(255) | | true | | | Resource ID recorded by the agent runtime. | | runId | varchar(255) | | true | | | Run ID parsed from the checkpoint key when available. | @@ -45,6 +46,7 @@ erDiagram "public.instance_ai_checkpoints" { timestamp_3__with_time_zone createdAt timestamp_3__with_time_zone expiredAt + varchar_64_ hostRunId varchar_255_ key varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/postgres-schema/public.instance_ai_pending_confirmations.md b/docs/generated/postgres-schema/public.instance_ai_pending_confirmations.md index 36e06f0ff2b..c69162ce703 100644 --- a/docs/generated/postgres-schema/public.instance_ai_pending_confirmations.md +++ b/docs/generated/postgres-schema/public.instance_ai_pending_confirmations.md @@ -70,6 +70,7 @@ erDiagram "public.instance_ai_checkpoints" { timestamp_3__with_time_zone createdAt timestamp_3__with_time_zone expiredAt + varchar_64_ hostRunId varchar_255_ key varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/postgres-schema/public.instance_ai_threads.md b/docs/generated/postgres-schema/public.instance_ai_threads.md index 574683cf297..0aeab03baa0 100644 --- a/docs/generated/postgres-schema/public.instance_ai_threads.md +++ b/docs/generated/postgres-schema/public.instance_ai_threads.md @@ -70,6 +70,7 @@ erDiagram "public.instance_ai_checkpoints" { timestamp_3__with_time_zone createdAt timestamp_3__with_time_zone expiredAt + varchar_64_ hostRunId varchar_255_ key varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 75fca95ab7d..08a810673fe 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -62,7 +62,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [insights_raw](insights_raw.md) | 5 | | table | | [installed_nodes](installed_nodes.md) | 4 | | table | | [installed_packages](installed_packages.md) | 6 | | table | -| [instance_ai_checkpoints](instance_ai_checkpoints.md) | 8 | | table | +| [instance_ai_checkpoints](instance_ai_checkpoints.md) | 9 | | table | | [instance_ai_events](instance_ai_events.md) | 7 | | table | | [instance_ai_iteration_logs](instance_ai_iteration_logs.md) | 6 | | table | | [instance_ai_mcp_registry_connections](instance_ai_mcp_registry_connections.md) | 7 | | table | @@ -806,6 +806,7 @@ erDiagram "instance_ai_checkpoints" { datetime_3_ createdAt datetime_3_ expiredAt + VARCHAR_64_ hostRunId varchar_255_ key PK varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/sqlite-schema/instance_ai_checkpoints.md b/docs/generated/sqlite-schema/instance_ai_checkpoints.md index bb4bf5b4d74..60e9d063f83 100644 --- a/docs/generated/sqlite-schema/instance_ai_checkpoints.md +++ b/docs/generated/sqlite-schema/instance_ai_checkpoints.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "instance_ai_checkpoints" ("key" varchar(255) PRIMARY KEY NOT NULL, "runId" varchar(255), "threadId" varchar NOT NULL, "resourceId" varchar(255), "state" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "expiredAt" datetime(3), CONSTRAINT "instance_ai_checkpoints_state_tombstone_check" CHECK (("expiredAt" IS NOT NULL AND "state" IS NULL) OR "expiredAt" IS NULL), CONSTRAINT "FK_2b23f3f24a70bebb990203b011e" FOREIGN KEY ("threadId") REFERENCES "instance_ai_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) +CREATE TABLE "instance_ai_checkpoints" ("key" varchar(255) PRIMARY KEY NOT NULL, "runId" varchar(255), "threadId" varchar NOT NULL, "resourceId" varchar(255), "state" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "expiredAt" datetime(3), "hostRunId" VARCHAR(64), CONSTRAINT "instance_ai_checkpoints_state_tombstone_check" CHECK (("expiredAt" IS NOT NULL AND "state" IS NULL) OR "expiredAt" IS NULL), CONSTRAINT "FK_2b23f3f24a70bebb990203b011e" FOREIGN KEY ("threadId") REFERENCES "instance_ai_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) ``` @@ -17,6 +17,7 @@ CREATE TABLE "instance_ai_checkpoints" ("key" varchar(255) PRIMARY KEY NOT NULL, | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | | expiredAt | datetime(3) | | true | | | | +| hostRunId | VARCHAR(64) | | true | | | | | key | varchar(255) | | false | [instance_ai_pending_confirmations](instance_ai_pending_confirmations.md) | | | | resourceId | varchar(255) | | true | | | | | runId | varchar(255) | | true | | | | @@ -53,6 +54,7 @@ erDiagram "instance_ai_checkpoints" { datetime_3_ createdAt datetime_3_ expiredAt + VARCHAR_64_ hostRunId varchar_255_ key PK varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/sqlite-schema/instance_ai_pending_confirmations.md b/docs/generated/sqlite-schema/instance_ai_pending_confirmations.md index af0c868e2a2..3e9828d4a40 100644 --- a/docs/generated/sqlite-schema/instance_ai_pending_confirmations.md +++ b/docs/generated/sqlite-schema/instance_ai_pending_confirmations.md @@ -75,6 +75,7 @@ erDiagram "instance_ai_checkpoints" { datetime_3_ createdAt datetime_3_ expiredAt + VARCHAR_64_ hostRunId varchar_255_ key PK varchar_255_ resourceId varchar_255_ runId diff --git a/docs/generated/sqlite-schema/instance_ai_threads.md b/docs/generated/sqlite-schema/instance_ai_threads.md index 2fb410ef708..6bc2f82bcf2 100644 --- a/docs/generated/sqlite-schema/instance_ai_threads.md +++ b/docs/generated/sqlite-schema/instance_ai_threads.md @@ -76,6 +76,7 @@ erDiagram "instance_ai_checkpoints" { datetime_3_ createdAt datetime_3_ expiredAt + VARCHAR_64_ hostRunId varchar_255_ key PK varchar_255_ resourceId varchar_255_ runId diff --git a/packages/@n8n/agents/src/types/sdk/agent.ts b/packages/@n8n/agents/src/types/sdk/agent.ts index e13fe14ea22..76fbfcbb586 100644 --- a/packages/@n8n/agents/src/types/sdk/agent.ts +++ b/packages/@n8n/agents/src/types/sdk/agent.ts @@ -386,4 +386,11 @@ export interface SerializableAgentState { export type AgentPersistenceOptions = { threadId: string; resourceId: string; + /** + * The host application's own run id (distinct from the agent-SDK runId that + * keys checkpoints). Persisted with checkpoints so host-side recovery (e.g. + * Instance AI's interrupted-run sweep) can match a checkpoint to its run + * exactly instead of guessing by recency. + */ + hostRunId?: string; }; diff --git a/packages/@n8n/db/src/migrations/common/1784000000050-AddHostRunIdToInstanceAiCheckpoints.ts b/packages/@n8n/db/src/migrations/common/1784000000050-AddHostRunIdToInstanceAiCheckpoints.ts new file mode 100644 index 00000000000..1d8e383694d --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1784000000050-AddHostRunIdToInstanceAiCheckpoints.ts @@ -0,0 +1,31 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +/** + * Durable-log RFC (resilience phase): record the host (Instance AI) run id on + * checkpoints. The existing `runId` column holds the agent-SDK run id derived + * from the checkpoint key; the interrupted-run sweeper needs the host run id + * to match a crashed run's step checkpoint exactly instead of assuming the + * thread's newest running checkpoint belongs to it. + */ +export class AddHostRunIdToInstanceAiCheckpoints1784000000050 implements ReversibleMigration { + async up({ escape, runQuery, isPostgres }: MigrationContext) { + const tableName = escape.tableName('instance_ai_checkpoints'); + const column = escape.columnName('hostRunId'); + + await runQuery(`ALTER TABLE ${tableName} ADD COLUMN ${column} VARCHAR(64)`); + + // SQLite has no column comments. + if (isPostgres) { + await runQuery( + `COMMENT ON COLUMN ${tableName}.${column} IS 'Host (Instance AI) run id; distinct from the agent-SDK runId parsed from the checkpoint key.'`, + ); + } + } + + async down({ escape, runQuery }: MigrationContext) { + const tableName = escape.tableName('instance_ai_checkpoints'); + const column = escape.columnName('hostRunId'); + + await runQuery(`ALTER TABLE ${tableName} DROP COLUMN ${column}`); + } +} diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index 0f166723c29..ca60981fef7 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -223,6 +223,7 @@ import { CreateInstanceAiEventsTable1784000000046 } from '../common/178400000004 import { BackfillPreScopingOAuthGrantScopes1784000000047 } from '../common/1784000000047-BackfillPreScopingOAuthGrantScopes'; import { AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048 } from '../common/1784000000048-AddTriggerKindToWorkflowPublicationTriggerStatus'; import { AddScheduledTaskDispatchedAt1784000000049 } from '../common/1784000000049-AddScheduledTaskDispatchedAt'; +import { AddHostRunIdToInstanceAiCheckpoints1784000000050 } from '../common/1784000000050-AddHostRunIdToInstanceAiCheckpoints'; import type { Migration } from '../migration-types'; export const postgresMigrations: Migration[] = [ @@ -451,4 +452,5 @@ export const postgresMigrations: Migration[] = [ BackfillPreScopingOAuthGrantScopes1784000000047, AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048, AddScheduledTaskDispatchedAt1784000000049, + AddHostRunIdToInstanceAiCheckpoints1784000000050, ]; diff --git a/packages/@n8n/db/src/migrations/sqlite/index.ts b/packages/@n8n/db/src/migrations/sqlite/index.ts index b37916b4103..9c4de399983 100644 --- a/packages/@n8n/db/src/migrations/sqlite/index.ts +++ b/packages/@n8n/db/src/migrations/sqlite/index.ts @@ -215,6 +215,7 @@ import { CreateInstanceAiEventsTable1784000000046 } from '../common/178400000004 import { BackfillPreScopingOAuthGrantScopes1784000000047 } from '../common/1784000000047-BackfillPreScopingOAuthGrantScopes'; import { AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048 } from '../common/1784000000048-AddTriggerKindToWorkflowPublicationTriggerStatus'; import { AddScheduledTaskDispatchedAt1784000000049 } from '../common/1784000000049-AddScheduledTaskDispatchedAt'; +import { AddHostRunIdToInstanceAiCheckpoints1784000000050 } from '../common/1784000000050-AddHostRunIdToInstanceAiCheckpoints'; const sqliteMigrations: Migration[] = [ InitialMigration1588102412422, @@ -433,6 +434,7 @@ const sqliteMigrations: Migration[] = [ BackfillPreScopingOAuthGrantScopes1784000000047, AddTriggerKindToWorkflowPublicationTriggerStatus1784000000048, AddScheduledTaskDispatchedAt1784000000049, + AddHostRunIdToInstanceAiCheckpoints1784000000050, ]; export { sqliteMigrations }; diff --git a/packages/cli/src/events/maps/instance-ai.event-map.ts b/packages/cli/src/events/maps/instance-ai.event-map.ts index d0d4f1f68b7..3edcb9649d2 100644 --- a/packages/cli/src/events/maps/instance-ai.event-map.ts +++ b/packages/cli/src/events/maps/instance-ai.event-map.ts @@ -30,6 +30,11 @@ export type InstanceAiEventMap = { 'instance-ai-parser-fallback': { count: number; }; + /** The interrupted-run sweep resolved a crashed run. */ + 'instance-ai-run-swept': { + outcome: 'interrupted' | 'crash-resumed'; + toolInterruptedFacts: number; + }; 'instance-ai-run-finished': { /** 'suspended' is a non-terminal HITL segment: usage/tool counts only; the terminal event counts the run. */ status: 'completed' | 'cancelled' | 'error' | 'suspended'; diff --git a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts index 747942bf8b2..55e8450c6e2 100644 --- a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts @@ -130,6 +130,14 @@ export class PrometheusInstanceAiMetricsService implements PrometheusMetricsColl }); parserFallbacksTotal.inc(0); + const runsSweptTotal = new promClient.Counter({ + name: `${this.config.prefix}instance_ai_runs_swept_total`, + help: 'Crashed Instance AI runs resolved by the interrupted-run sweep.', + labelNames: ['outcome'], + }); + runsSweptTotal.inc({ outcome: 'interrupted' }, 0); + runsSweptTotal.inc({ outcome: 'crash-resumed' }, 0); + this.eventService.on('instance-ai-durable-log-drained', ({ rows, bytes }) => { durableLogRowsTotal.inc(rows); durableLogBytesTotal.inc(bytes); @@ -153,6 +161,9 @@ export class PrometheusInstanceAiMetricsService implements PrometheusMetricsColl this.eventService.on('instance-ai-parser-fallback', ({ count }) => { parserFallbacksTotal.inc(count); }); + this.eventService.on('instance-ai-run-swept', ({ outcome }) => { + runsSweptTotal.inc({ outcome }, 1); + }); this.eventService.on( 'instance-ai-run-finished', diff --git a/packages/cli/src/modules/instance-ai/entities/instance-ai-checkpoint.entity.ts b/packages/cli/src/modules/instance-ai/entities/instance-ai-checkpoint.entity.ts index 76001e08ef3..f755f80be58 100644 --- a/packages/cli/src/modules/instance-ai/entities/instance-ai-checkpoint.entity.ts +++ b/packages/cli/src/modules/instance-ai/entities/instance-ai-checkpoint.entity.ts @@ -13,6 +13,15 @@ export class InstanceAiCheckpoint extends WithTimestamps { @Column({ type: 'varchar', length: 255, nullable: true }) runId: string | null; + /** + * The Instance AI (host) run id, distinct from `runId` above (the agent-SDK + * id derived from the key). Lets the interrupted-run sweeper match a + * crashed run's step checkpoint exactly. Nullable: rows written before the + * column existed, and sub-agent checkpoints, don't carry it. + */ + @Column({ type: 'varchar', length: 64, nullable: true }) + hostRunId: string | null; + @ManyToOne(() => InstanceAiThread, { onDelete: 'CASCADE' }) @JoinColumn({ name: 'threadId' }) thread: InstanceAiThread; diff --git a/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts b/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts new file mode 100644 index 00000000000..42b375f23d7 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts @@ -0,0 +1,288 @@ +import type { InstanceAiEvent } from '@n8n/api-types'; +import type { Logger } from '@n8n/backend-common'; +import type { GlobalConfig } from '@n8n/config'; +import type { InstanceSettings } from 'n8n-core'; +import { mock } from 'vitest-mock-extended'; + +import type { EventService } from '@/events/event.service'; + +import { DurableLogMetrics } from '../durable-log-metrics'; +import { + InterruptedRunSweeper, + TOOL_INTERRUPTED_MESSAGE, + type InterruptedRunResumeHost, +} from '../interrupted-run-sweeper'; +import type { InstanceAiCheckpoint } from '../../entities/instance-ai-checkpoint.entity'; +import type { InstanceAiCheckpointRepository } from '../../repositories/instance-ai-checkpoint.repository'; +import type { InstanceAiEventLogRepository } from '../../repositories/instance-ai-event-log.repository'; + +const THREAD = 'thread-1'; +const RUN = 'run-1'; +const AGENT = `orchestrator:${RUN}`; + +function runStart(userId = 'user-1'): InstanceAiEvent { + return { + type: 'run-start', + runId: RUN, + agentId: AGENT, + userId, + payload: { messageId: 'm-1', messageGroupId: 'mg-1' }, + }; +} + +function toolCall(toolCallId: string, args: Record = {}): InstanceAiEvent { + return { + type: 'tool-call', + runId: RUN, + agentId: AGENT, + payload: { toolCallId, toolName: 'update-workflow', args }, + }; +} + +function toolResult(toolCallId: string): InstanceAiEvent { + return { type: 'tool-result', runId: RUN, agentId: AGENT, payload: { toolCallId, result: {} } }; +} + +function runningCheckpoint(overrides: Partial = {}): InstanceAiCheckpoint { + return { + key: 'agent:run-sdk-1', + runId: 'run-sdk-1', + hostRunId: RUN, + threadId: THREAD, + resourceId: 'user-1', + state: { + persistence: { threadId: THREAD, resourceId: 'user-1', hostRunId: RUN }, + status: 'running', + messageList: { messages: [] }, + pendingToolCalls: {}, + }, + expiredAt: null, + createdAt: new Date('2026-07-07T10:00:00Z'), + updatedAt: new Date('2026-07-07T10:00:00Z'), + ...overrides, + } as InstanceAiCheckpoint; +} + +interface Setup { + events?: InstanceAiEvent[]; + checkpoints?: InstanceAiCheckpoint[]; + isMultiMain?: boolean; + lastFactAt?: Date | null; + durableLog?: boolean; + host?: Partial; +} + +function buildSweeper(setup: Setup) { + const logger = mock(); + logger.scoped.mockReturnValue(logger); + + // A minimal in-memory log: publishes append durable facts (mirroring the + // event-bus graft), and the unfinished-runs query derives from the log + // contents — so a sweep's own run-finish makes the next sweep a no-op. + const log: InstanceAiEvent[] = [...(setup.events ?? [])]; + + const eventLogRepo = mock(); + // Mirrors the repository contract: DISTINCT (threadId, runId) pairs. + eventLogRepo.findUnfinishedRuns.mockImplementation(async () => [ + ...new Map( + log + .filter((e) => e.type === 'run-start') + .filter((s) => !log.some((f) => f.type === 'run-finish' && f.runId === s.runId)) + .map((s) => [s.runId, { threadId: THREAD, runId: s.runId }]), + ).values(), + ]); + eventLogRepo.getForRuns.mockImplementation(async (_threadId, runIds) => + log.filter((e) => runIds.includes(e.runId)), + ); + eventLogRepo.lastFactAt.mockResolvedValue(setup.lastFactAt ?? null); + + const checkpointRepo = mock(); + checkpointRepo.findActiveByThreadId.mockResolvedValue(setup.checkpoints ?? []); + + const published: InstanceAiEvent[] = []; + const eventBus = { + publish: (_threadId: string, event: InstanceAiEvent) => { + published.push(event); + log.push(event); + }, + }; + + const metrics = new DurableLogMetrics(mock()); + const host: InterruptedRunResumeHost = { + isRunLive: setup.host?.isRunLive ?? (() => false), + }; + + const sweeper = new InterruptedRunSweeper( + logger, + eventLogRepo, + checkpointRepo, + eventBus as never, + metrics, + { instanceAi: { durableLog: setup.durableLog ?? true } } as GlobalConfig, + { isMultiMain: setup.isMultiMain ?? false } as InstanceSettings, + ); + sweeper.setResumeHost(host); + return { sweeper, published, metrics, eventLogRepo }; +} + +describe('InterruptedRunSweeper', () => { + it('appends tool-interrupted facts and run-finish{interrupted} for a crashed run', async () => { + const { sweeper, published, metrics } = buildSweeper({ + events: [runStart(), toolCall('tc-done'), toolResult('tc-done'), toolCall('tc-inflight')], + }); + + await sweeper.sweep(); + + expect(published.map((e) => e.type)).toEqual(['tool-interrupted', 'run-finish']); + const interrupted = published[0]; + expect(interrupted.type === 'tool-interrupted' && interrupted.payload.toolCallId).toBe( + 'tc-inflight', + ); + expect(interrupted.type === 'tool-interrupted' && interrupted.payload.error).toBe( + TOOL_INTERRUPTED_MESSAGE, + ); + const finish = published[1]; + expect(finish.type === 'run-finish' && finish.payload.status).toBe('interrupted'); + expect(finish.type === 'run-finish' && finish.payload.reason).toBe('crash_interrupted'); + expect(metrics.sweep.runsMarkedInterrupted).toBe(1); + expect(metrics.sweep.toolInterruptedFacts).toBe(1); + }); + + it('does nothing when the durable log is disabled', async () => { + const { sweeper, published, eventLogRepo } = buildSweeper({ + events: [runStart()], + durableLog: false, + }); + + await sweeper.sweep(); + + expect(eventLogRepo.findUnfinishedRuns).not.toHaveBeenCalled(); + expect(published).toHaveLength(0); + }); + + it('is idempotent: a second sweep after the first is a no-op', async () => { + const { sweeper, published, metrics } = buildSweeper({ + events: [runStart(), toolCall('tc-inflight')], + }); + + await sweeper.sweep(); + expect(published.map((e) => e.type)).toEqual(['tool-interrupted', 'run-finish']); + + await sweeper.sweep(); + + expect(published).toHaveLength(2); + expect(metrics.sweep.runsMarkedInterrupted).toBe(1); + }); + + it('skips a run that is live in this process', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + host: { isRunLive: (_threadId, runId) => runId === RUN }, + }); + + await sweeper.sweep(); + + expect(published).toHaveLength(0); + }); + + it('sweeps an older crashed run while a different run is live on the thread', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + host: { isRunLive: (_threadId, runId) => runId === 'run-other' }, + }); + + await sweeper.sweep(); + + expect(published.at(-1)?.type).toBe('run-finish'); + }); + + it('skips HITL-suspended runs entirely (the confirmation orphan path owns them)', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + checkpoints: [ + runningCheckpoint({ + state: { + persistence: { threadId: THREAD, resourceId: 'user-1' }, + status: 'suspended', + messageList: { messages: [] }, + pendingToolCalls: { 'tc-p': { suspended: true } }, + } as never, + }), + ], + }); + + await sweeper.sweep(); + + expect(published).toHaveLength(0); + }); + + it('sweeps a crashed run even when a different run in the thread is HITL-suspended', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + checkpoints: [ + runningCheckpoint({ + key: 'agent:other', + hostRunId: 'run-other', + state: { + persistence: { threadId: THREAD, resourceId: 'user-1', hostRunId: 'run-other' }, + status: 'suspended', + messageList: { messages: [] }, + pendingToolCalls: { 'tc-p': { suspended: true } }, + } as never, + }), + ], + }); + + await sweeper.sweep(); + + expect(published.at(-1)?.type).toBe('run-finish'); + }); + + it('marks a run with a running step checkpoint interrupted (crash-resume is a later phase)', async () => { + const { sweeper, published, metrics } = buildSweeper({ + events: [runStart()], + checkpoints: [runningCheckpoint()], + }); + + await sweeper.sweep(); + + expect(published.at(-1)?.type).toBe('run-finish'); + expect(metrics.sweep.runsMarkedInterrupted).toBe(1); + }); + + it('multi-main: recent durable activity is a liveness heartbeat and skips the run', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + isMultiMain: true, + lastFactAt: new Date(), // written seconds ago: a sibling is driving it + }); + + await sweeper.sweep(); + + expect(published).toHaveLength(0); + }); + + it('multi-main: stale activity past the grace window is swept', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + isMultiMain: true, + lastFactAt: new Date(Date.now() - InterruptedRunSweeper.LIVENESS_GRACE_MS - 1000), + }); + + await sweeper.sweep(); + + expect(published.at(-1)?.type).toBe('run-finish'); + }); + + it('single-main ignores the grace window (immediate post-crash sweeps work)', async () => { + const { sweeper, published } = buildSweeper({ + events: [runStart()], + isMultiMain: false, + lastFactAt: new Date(), // fresh facts, but no siblings exist + }); + + await sweeper.sweep(); + + expect(published.at(-1)?.type).toBe('run-finish'); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts b/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts index b2405db9035..20cf3c0899f 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/durable-log-metrics.ts @@ -51,6 +51,14 @@ export class DurableLogMetrics { treesDerived: 0, }; + /** Interrupted-run sweep (phase 3). */ + sweep = { + runsExamined: 0, + runsMarkedInterrupted: 0, + runsCrashResumed: 0, + toolInterruptedFacts: 0, + }; + constructor(private readonly eventService: EventService) {} recordDrainBatch(rows: number, bytes: number): void { @@ -103,4 +111,18 @@ export class DurableLogMetrics { notifyParserFallbacks(count: number): void { if (count > 0) this.eventService.emit('instance-ai-parser-fallback', { count }); } + + recordSweepRunExamined(): void { + this.sweep.runsExamined++; + } + + recordSweepToolInterruptedFact(): void { + this.sweep.toolInterruptedFacts++; + } + + recordSweepOutcome(outcome: 'interrupted' | 'crash-resumed', toolInterruptedFacts: number): void { + if (outcome === 'interrupted') this.sweep.runsMarkedInterrupted++; + else this.sweep.runsCrashResumed++; + this.eventService.emit('instance-ai-run-swept', { outcome, toolInterruptedFacts }); + } } diff --git a/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts b/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts new file mode 100644 index 00000000000..89a42a2d0ad --- /dev/null +++ b/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts @@ -0,0 +1,192 @@ +import type { InstanceAiEvent } from '@n8n/api-types'; +import { Logger } from '@n8n/backend-common'; +import { GlobalConfig } from '@n8n/config'; +import { Service } from '@n8n/di'; +import { createSubAgentResourceIdPrefix, orchestratorAgentId } from '@n8n/instance-ai'; +import { InstanceSettings } from 'n8n-core'; + +import { DurableLogMetrics } from './durable-log-metrics'; +import { InProcessEventBus } from './in-process-event-bus'; +import { InstanceAiCheckpointRepository } from '../repositories/instance-ai-checkpoint.repository'; +import { InstanceAiEventLogRepository } from '../repositories/instance-ai-event-log.repository'; + +export const TOOL_INTERRUPTED_MESSAGE = + 'Interrupted by a process restart — effect unverified; verify before retrying.'; + +/** A tool call that was in flight (no terminal fact) when the process died. */ +interface InFlightToolCall { + toolCallId: string; + toolName: string; + agentId: string; + args: Record; +} + +/** + * The narrow slice of InstanceAiService the sweeper consults. Injected at init + * (module wiring) instead of DI to avoid a service-level dependency cycle. + */ +export interface InterruptedRunResumeHost { + /** Whether this specific run is live (active or suspended) in this process. */ + isRunLive(threadId: string, runId: string): boolean; +} + +/** + * Durable-log RFC (resilience phase): no zombie runs. + * + * Invariant: every run reaches a terminal state recorded in the log. A mid-run + * crash leaves a `run-start` with no `run-finish` — detection is a log query, + * resolution is appended facts: + * + * 1. In-flight tool calls (tool-call with no terminal fact) become durable + * `tool-interrupted` facts — effect unverified, never re-executed blindly. + * 2. One `run-finish { status: 'interrupted' }` is appended — the fold renders + * every in-flight item as terminated; no walk-and-mutate. (A later phase of + * the RFC re-drives runs with a `running` step checkpoint from that + * checkpoint instead; until then every swept run is marked interrupted.) + * + * Runs whose own checkpoint (matched by `hostRunId`) is HITL-`suspended` are + * skipped: the pending confirmation orphan path (SuspendedRunRestorer) owns + * their recovery. + * + * Multi-main safety, without a dedicated lease table: durable activity is the + * heartbeat. Step checkpoints upsert per step and log facts append per step, + * so a run whose newest fact or checkpoint write is younger than the grace + * window is treated as live on a sibling main and skipped. Single-main skips + * the grace window (no siblings; an unfinished run with no local live run is + * dead), which keeps immediate post-crash sweeps instant. Two mains booting + * together may still both mark one dead run — accepted: the duplicate + * terminal facts are benign (the fold is last-writer-wins per run), and a + * per-run claim would need the lease table this design deliberately avoids. + */ +@Service() +export class InterruptedRunSweeper { + /** Multi-main only: durable activity younger than this means "still driven". */ + static readonly LIVENESS_GRACE_MS = 2 * 60 * 1000; + + private resumeHost: InterruptedRunResumeHost | undefined; + + private readonly durableLogEnabled: boolean; + + constructor( + private readonly logger: Logger, + private readonly eventLogRepo: InstanceAiEventLogRepository, + private readonly checkpointRepo: InstanceAiCheckpointRepository, + private readonly eventBus: InProcessEventBus, + private readonly metrics: DurableLogMetrics, + globalConfig: GlobalConfig, + private readonly instanceSettings: InstanceSettings, + ) { + this.logger = this.logger.scoped('instance-ai'); + this.durableLogEnabled = globalConfig.instanceAi.durableLog; + } + + setResumeHost(host: InterruptedRunResumeHost): void { + this.resumeHost = host; + } + + /** Called from module init (startup). */ + async sweep(): Promise { + if (!this.durableLogEnabled) return; + + let unfinished; + try { + unfinished = await this.eventLogRepo.findUnfinishedRuns(); + } catch (error) { + this.logger.error('Interrupted-run sweep failed to query the event log', { error }); + return; + } + + for (const { threadId, runId } of unfinished) { + try { + await this.sweepRun(threadId, runId); + } catch (error) { + this.logger.error('Interrupted-run sweep failed for run', { threadId, runId, error }); + } + } + } + + private async sweepRun(threadId: string, runId: string): Promise { + this.metrics.recordSweepRunExamined(); + + // This specific run is live in this process (e.g. a sweep re-run raced an + // in-flight run) — not a zombie. A different run started on the same + // thread while the sweep was running must not shield it. + if (this.resumeHost?.isRunLive(threadId, runId)) return; + + const checkpoints = await this.checkpointRepo.findActiveByThreadId(threadId); + const subAgentPrefix = createSubAgentResourceIdPrefix(threadId); + // Exact hostRunId match only: every flag-on run post-dates the column + // (production never runs a hybrid pre/post-log state), and sub-agent or + // legacy rows carry null, so they never match another run's sweep. + const runCheckpoints = checkpoints.filter( + (row) => !row.resourceId?.startsWith(subAgentPrefix) && row.hostRunId === runId, + ); + + // A run suspended at HITL is recoverable through the pending-confirmation + // orphan path; marking it interrupted would destroy that recovery. Only + // this run's own checkpoint counts — another run suspended on the same + // thread must not shield a crashed run from being swept. + if (runCheckpoints.some((row) => row.state?.status === 'suspended')) return; + + // Multi-main: durable activity is the liveness heartbeat — a sibling + // main driving this run appends facts and upserts its checkpoint every + // step, so recent writes mean "not a zombie, skip this round". + if (this.instanceSettings.isMultiMain) { + const cutoff = Date.now() - InterruptedRunSweeper.LIVENESS_GRACE_MS; + const lastFact = await this.eventLogRepo.lastFactAt(threadId, runId); + const newestCheckpointAt = runCheckpoints + .map((row) => row.updatedAt?.getTime() ?? 0) + .reduce((a, b) => Math.max(a, b), 0); + const lastActivity = Math.max(lastFact?.getTime() ?? 0, newestCheckpointAt); + if (lastActivity > cutoff) return; + } + + const events = await this.eventLogRepo.getForRuns(threadId, [runId]); + const inFlight = collectInFlightToolCalls(events); + + for (const call of inFlight) { + this.eventBus.publish(threadId, { + type: 'tool-interrupted', + runId, + agentId: call.agentId, + payload: { toolCallId: call.toolCallId, error: TOOL_INTERRUPTED_MESSAGE }, + }); + this.metrics.recordSweepToolInterruptedFact(); + } + + this.eventBus.publish(threadId, { + type: 'run-finish', + runId, + agentId: orchestratorAgentId(runId), + payload: { status: 'interrupted', reason: 'crash_interrupted' }, + }); + this.metrics.recordSweepOutcome('interrupted', inFlight.length); + this.logger.info('Marked interrupted Instance AI run in the durable log', { + threadId, + runId, + inFlightToolCalls: inFlight.length, + }); + } +} + +/** tool-call facts with no matching terminal fact (result/error/interrupted). */ +export function collectInFlightToolCalls(events: InstanceAiEvent[]): InFlightToolCall[] { + const open = new Map(); + for (const event of events) { + if (event.type === 'tool-call') { + open.set(event.payload.toolCallId, { + toolCallId: event.payload.toolCallId, + toolName: event.payload.toolName, + agentId: event.agentId, + args: event.payload.args, + }); + } else if ( + event.type === 'tool-result' || + event.type === 'tool-error' || + event.type === 'tool-interrupted' + ) { + open.delete(event.payload.toolCallId); + } + } + return [...open.values()]; +} diff --git a/packages/cli/src/modules/instance-ai/instance-ai.module.ts b/packages/cli/src/modules/instance-ai/instance-ai.module.ts index 61155a76ae0..c28bbcd8c04 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.module.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.module.ts @@ -26,6 +26,20 @@ export class InstanceAiModule implements ModuleInterface { const { InstanceAiEventRelay } = await import('./instance-ai-event-relay.service.js'); Container.get(InstanceAiEventRelay); + // Durable-log flag (resilience phase): startup sweep resolves runs the + // previous process left mid-flight by converting their in-flight tool + // calls into tool-interrupted facts and appending run-finish{interrupted}. + const { GlobalConfig } = await import('@n8n/config'); + if (Container.get(GlobalConfig).instanceAi.durableLog) { + const { InterruptedRunSweeper } = await import('./event-bus/interrupted-run-sweeper.js'); + const { InstanceAiService } = await import('./instance-ai.service.js'); + const sweeper = Container.get(InterruptedRunSweeper); + sweeper.setResumeHost(Container.get(InstanceAiService)); + void sweeper.sweep().catch((error: unknown) => { + logger.error('Interrupted-run sweep failed on startup', { error }); + }); + } + if (process.env.E2E_TESTS === 'true' && process.env.NODE_ENV !== 'production') { await import('./instance-ai-test.controller.js'); } diff --git a/packages/cli/src/modules/instance-ai/instance-ai.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.service.ts index 69c861b367d..1e277c5e3fb 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -741,6 +741,18 @@ export class InstanceAiService { return this.runState.hasLiveRun(threadId); } + /** + * Whether this specific run is live (active or suspended) in this process. + * The interrupted-run sweeper uses this so a newer run on the same thread + * never shields an older crashed run from being swept. + */ + isRunLive(threadId: string, runId: string): boolean { + return ( + this.runState.getActiveRunId(threadId) === runId || + this.runState.getSuspendedRun(threadId)?.runId === runId + ); + } + getThreadStatus(threadId: string): InstanceAiThreadStatusResponse { const status = this.runState.getThreadStatus( threadId, @@ -819,6 +831,9 @@ export class InstanceAiService { persistence: { resourceId: user.id, threadId, + // Host run id, persisted with checkpoints so the interrupted-run + // sweep can match a crashed run's checkpoint exactly. + hostRunId: runId, }, providerOptions: { anthropic: { cacheControl: { type: 'ephemeral' } }, @@ -844,7 +859,7 @@ export class InstanceAiService { toolCallId, // Keep billing stopped/errored resumed runs (see stream-options builder). recoverUsageOnAbort: true, - persistence: { resourceId: user.id, threadId }, + persistence: { resourceId: user.id, threadId, hostRunId: runId }, // Must mirror buildOrchestratorAgentStreamOptions: without this request-level // cache directive, resumed (HITL) turns send no cache_control, so Anthropic // reprocesses the whole conversation uncached on every resume (~100K tokens). diff --git a/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts index 0cd73fc8601..43f5eec6db4 100644 --- a/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts +++ b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts @@ -6,6 +6,12 @@ import { jsonParse } from 'n8n-workflow'; import { InstanceAiEventLogEntry } from '../entities/instance-ai-event-log-entry.entity'; +/** A run that has a `run-start` fact but no terminal `run-finish` in the log. */ +export interface UnfinishedRun { + threadId: string; + runId: string; +} + @Service() export class InstanceAiEventLogRepository extends Repository { constructor(dataSource: DataSource) { @@ -90,4 +96,42 @@ export class InstanceAiEventLogRepository extends Repository(r.payload), })); } + + /** Timestamp of the run's most recent durable fact (sweep liveness proxy). */ + async lastFactAt(threadId: string, runId: string): Promise { + const row = await this.createQueryBuilder('e') + .select('MAX(e.createdAt)', 'max') + .where('e.threadId = :threadId', { threadId }) + .andWhere('e.runId = :runId', { runId }) + .getRawOne<{ max: string | Date | null }>(); + if (!row?.max) return null; + return row.max instanceof Date ? row.max : new Date(row.max); + } + + /** + * Interrupted-run sweep source: runs whose log has a `run-start` but no + * `run-finish`, as distinct (threadId, runId) pairs. Pure log query — + * liveness (is a main still driving it?) is the caller's concern. + */ + async findUnfinishedRuns(): Promise { + const rows = await this.createQueryBuilder('e') + .select('e.threadId', 'threadId') + .addSelect('e.runId', 'runId') + .distinct(true) + .where("e.type = 'run-start'") + .andWhere( + (qb) => + 'NOT EXISTS ' + + qb + .subQuery() + .select('1') + .from(InstanceAiEventLogEntry, 'f') + .where('f.threadId = e.threadId') + .andWhere('f.runId = e.runId') + .andWhere("f.type = 'run-finish'") + .getQuery(), + ) + .getRawMany(); + return rows; + } } diff --git a/packages/cli/src/modules/instance-ai/storage/__tests__/typeorm-agent-checkpoint-store.test.ts b/packages/cli/src/modules/instance-ai/storage/__tests__/typeorm-agent-checkpoint-store.test.ts index 52f225997f7..3f472d87a86 100644 --- a/packages/cli/src/modules/instance-ai/storage/__tests__/typeorm-agent-checkpoint-store.test.ts +++ b/packages/cli/src/modules/instance-ai/storage/__tests__/typeorm-agent-checkpoint-store.test.ts @@ -68,6 +68,7 @@ describe('TypeORMAgentCheckpointStore', () => { expect(checkpointRepo.create).toHaveBeenCalledWith({ key: 'checkpoint:run-1', runId: 'run-1', + hostRunId: null, threadId: 'thread-1', resourceId: 'user-1', state, diff --git a/packages/cli/src/modules/instance-ai/storage/typeorm-agent-checkpoint-store.ts b/packages/cli/src/modules/instance-ai/storage/typeorm-agent-checkpoint-store.ts index 9360200e19d..34c3a6de5a3 100644 --- a/packages/cli/src/modules/instance-ai/storage/typeorm-agent-checkpoint-store.ts +++ b/packages/cli/src/modules/instance-ai/storage/typeorm-agent-checkpoint-store.ts @@ -23,6 +23,7 @@ export class TypeORMAgentCheckpointStore implements CheckpointStore { const existing = await this.checkpointRepo.findOne({ where: { key } }); if (existing) { existing.runId = this.getRunId(key); + existing.hostRunId = state.persistence?.hostRunId ?? null; existing.threadId = threadId; existing.resourceId = state.persistence?.resourceId ?? null; existing.state = state; @@ -34,6 +35,7 @@ export class TypeORMAgentCheckpointStore implements CheckpointStore { const checkpoint = this.checkpointRepo.create({ key, runId: this.getRunId(key), + hostRunId: state.persistence?.hostRunId ?? null, threadId, resourceId: state.persistence?.resourceId ?? null, state,