opensource/n8n
3
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-09-17 17:42:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): Make the Instance AI durable event log the only storage path (no-changelog) (#36729)

Browse Source 
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Raúl Gómez Morales <raul00gm@gmail.com>
This commit is contained in:
Jaakko Husso
2026-08-28 12:00:45 +00:00
committed by GitHub
co-authored by Claude Opus 5 Raúl Gómez Morales
parent fa43bc8892
commit 3e6edc11b7
41 changed files with 474 additions and 1629 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/@n8n/api-types/src/schemas/instance-ai.schema.ts
+2 -2
View File
@@ -185,8 +185,8 @@ export const instanceAiEventTypeSchema = z.enum([
export type InstanceAiEventType = z.infer<typeof instanceAiEventTypeSchema>;
/**
* Live-only event types under the durable log (`N8N_INSTANCE_AI_DURABLE_LOG`):
* never persisted, their SSE frames carry no `id:` line, and the browser's
* Live-only event types: never persisted, their SSE frames carry no `id:` line,
* and the browser's
* replay cursor never points at them. Deltas are transport, not state: a
* completed segment replays as a coalesced block fact instead. One list,
* shared by the writer (what to persist) and the frontend (which frames to
packages/@n8n/config/src/configs/instance-ai.config.ts
-27
View File
@@ -173,37 +173,10 @@ export class InstanceAiConfig {
@Env('N8N_INSTANCE_AI_CONFIRMATION_TIMEOUT')
confirmationTimeout: number = 24 * Time.hours.toMilliseconds;
/** Scan and redact secrets/PII from agent output before it reaches the user. */
@Env('N8N_INSTANCE_AI_OUTPUT_REDACTION_ENABLED')
outputRedactionEnabled: boolean = true;
/** Redact credential/secret patterns from agent output. Applies only when output redaction is enabled. */
@Env('N8N_INSTANCE_AI_OUTPUT_REDACTION_SECRETS')
outputRedactionSecrets: boolean = true;
/** Comma-separated PII categories to redact from agent output. Available: email, phone, credit-card, ssn-us, iban, crypto-wallet, ip, mac, url. Empty = no PII scanning. */
@Env('N8N_INSTANCE_AI_OUTPUT_REDACTION_PII')
outputRedactionPii: string = 'credit-card';
/** Replacement text substituted for each redacted match in agent output. */
@Env('N8N_INSTANCE_AI_OUTPUT_REDACTION_PLACEHOLDER')
outputRedactionPlaceholder: string = '[REDACTED]';
/** Capture orchestrator LLM steps and workflow code snapshots for the dev debug panel. */
@Env('N8N_INSTANCE_AI_RUN_DEBUG_ENABLED')
runDebugEnabled: boolean = false;
/**
* Persist Instance AI events to a durable DB log (`instance_ai_events`)
* and serve SSE replay + history from it. Default on since Gate A of the
* durable-log rollout (pre-existing runs are backfilled by migration);
* `false` restores the legacy in-memory bus + stored-snapshot history as
* an off switch until the legacy paths sunset at Gate B. See RFC:
* instance-ai durable event log.
*/
@Env('N8N_INSTANCE_AI_DURABLE_LOG')
durableLog: boolean = true;
/** Enable extended thinking / reasoning for the orchestrator agent. */
@Env('N8N_INSTANCE_AI_THINKING_ENABLED')
thinkingEnabled: boolean = true;
packages/@n8n/config/test/config.test.ts
-5
View File
@@ -363,13 +363,8 @@ describe('GlobalConfig', () => {
snapshotRetention: 86_400_000,
checkpointGcRetention: 604_800_000,
confirmationTimeout: 86_400_000,
outputRedactionEnabled: true,
outputRedactionSecrets: true,
outputRedactionPii: 'credit-card',
outputRedactionPlaceholder: '[REDACTED]',
runDebugEnabled: false,
thinkingEnabled: true,
durableLog: true,
mcpConnectionsEnabled: false,
canvasNodeContextEnabled: false,
activationCapped: false,
packages/@n8n/instance-ai/docs/architecture.md
+10 -13
View File
@@ -49,7 +49,8 @@ graph TB
subgraph EventSystem ["Event System"]
OrcAgent -->|publishes| EventBus
EvalSetupAgent -->|publishes| EventBus
EventBus --> ThreadStorage[Thread Event Storage]
EventBus --> DurableLog[Durable Event Log]
DurableLog --> EventsTable[(instance_ai_events)]
end
subgraph Filesystem ["Filesystem Access"]
@@ -68,9 +69,8 @@ graph TB
subgraph Storage ["Storage"]
Memory --> PostgreSQL[PostgreSQL<br/>main n8n database]
Memory --> SQLite[SQLite<br/>main n8n database]
ThreadStorage -->|durable log on| PostgreSQL
ThreadStorage -->|durable log on| SQLite
ThreadStorage -->|durable log off| InMemory[Per-thread memory buffer]
EventsTable --> PostgreSQL
EventsTable --> SQLite
end
subgraph Sandbox ["Sandbox (Optional)"]
@@ -217,9 +217,9 @@ The n8n integration layer.
- **Adapter** — bridges n8n services to agent interfaces, enforces RBAC permissions
- **Memory service** — thread lifecycle, message persistence, expiration
- **Settings service** — admin settings (model, MCP, sandbox), user preferences
- **Event bus** — in-process EventEmitter (single instance) or Redis Pub/Sub
(queue mode). The durable log is the default replay store. With the durable
log disabled, replay uses a 500-event or 2 MB in-memory buffer per thread.
- **Event bus** — live fan-out only: in-process EventEmitter (single instance)
plus a Redis Pub/Sub relay to sibling mains (queue mode). Persistence and
replay belong to the durable event log, not the bus
- **Filesystem** — `LocalGateway` (remote daemon via SSE protocol).
See `docs/filesystem-access.md`
- **Persistence** — 13 TypeORM entity/repository pairs for threads, messages,
@@ -282,12 +282,9 @@ The event bus decouples agent execution from event delivery:
- All events carry `runId` (correlates to triggering message) and `agentId`
- Durable SSE facts use monotonically increasing per-thread `id` values for replay
- SSE supports both `Last-Event-ID` header and `?lastEventId` query parameter
- Event storage depends on `N8N_INSTANCE_AI_DURABLE_LOG`: on (the default),
coalesced step-level facts are appended to the `instance_ai_events` table
(the durable replay source, ids survive restarts) while token deltas remain
live-only and are not retained; off (the rollback switch until Gate B), events
live only in a bounded in-memory buffer (500 events / 2 MB per thread,
FIFO-evicted, ids reset on restart)
- Coalesced step-level facts are appended to the `instance_ai_events` table —
the only replay source, so ids survive restarts — while token deltas remain
live-only and are not retained
- No need to pipe sub-agent streams through orchestrator tool execution
- One active run per thread (additional `POST /chat` is rejected while active)
- Cancellation via `POST /instance-ai/chat/:threadId/cancel` (idempotent)
packages/@n8n/instance-ai/docs/configuration.md
+13 -34
View File
@@ -147,23 +147,13 @@ These environment variables are read directly by `BuilderTemplatesService`.
| `N8N_INSTANCE_AI_CONFIRMATION_TIMEOUT` | number | `86400000` | Timeout in ms for HITL confirmation requests. 0 = no timeout. |
| `N8N_INSTANCE_AI_CHECKPOINT_GC_RETENTION` | number | `604800000` | Retention period in ms for expired checkpoint tombstones before hard deletion. `0` keeps tombstones. |
### Output Filtering
### Output filtering
The stream output redactor runs only when the durable event log is disabled.
The durable event log is enabled by default, so these settings do not redact
the default durable stream path. On the legacy path, the scan covers assistant
text, reasoning, tool-call arguments, tool results, tool errors, and confirmation
text. It records category counts when it redacts content.
| Variable | Type | Default | Description |
|----------|------|---------|-------------|
| `N8N_INSTANCE_AI_OUTPUT_REDACTION_ENABLED` | boolean | `true` | Master switch. When `false`, output passes through untouched. |
| `N8N_INSTANCE_AI_OUTPUT_REDACTION_SECRETS` | boolean | `true` | Redact credential/secret patterns (API keys, tokens, auth headers, `key=value` pairs). |
| `N8N_INSTANCE_AI_OUTPUT_REDACTION_PII` | string | `credit-card` | Comma-separated PII categories. Available values are `email`, `phone`, `credit-card`, `ssn-us`, `iban`, `crypto-wallet`, `ip`, `mac`, and `url`. Empty disables PII scanning. Unrecognized values are ignored. |
| `N8N_INSTANCE_AI_OUTPUT_REDACTION_PLACEHOLDER` | string | `[REDACTED]` | Replacement text substituted for each redacted match. |
Secret detection matches known token shapes and explicit secret fields. PII
detection applies the pattern registered for each selected category.
Instance AI does not scan or redact agent output on the streaming path.
Agent output is stored raw, consistent with workflow execution data, and
redaction applies at egress boundaries instead — the LangSmith telemetry
redactor is a separate layer and is unaffected. There are no
`N8N_INSTANCE_AI_OUTPUT_REDACTION_*` settings.
## Provider connections
@@ -229,19 +219,13 @@ The event bus transport is selected automatically:
- **Single instance**: In-process `EventEmitter` — zero infrastructure
- **Queue mode**: Redis Pub/Sub — uses n8n's existing Redis connection
Event persistence is controlled by `N8N_INSTANCE_AI_DURABLE_LOG` (default
`true` since Gate A of the durable-log rollout; pre-existing runs are
backfilled by migration). On, coalesced step-level facts (completed
text/reasoning blocks, tool calls and results, run lifecycle) are appended to
the `instance_ai_events` table and replay reads the database; token deltas
are never persisted. Rows cascade-delete with their thread
(`N8N_INSTANCE_AI_THREAD_TTL_DAYS`). Setting it to `false` is the rollback
switch until the legacy paths sunset at Gate B: events then live only in a
bounded in-memory buffer per thread (500 events / 2 MB, FIFO-evicted; ids
reset on restart, so replay does not survive a restart). That bound is
per-thread only: a buffer is released when its thread is deleted or expires,
so a main's memory scales with the number of threads it has served until the
process restarts. The main logs a warning at boot when the switch is set.
Events are persisted to the durable event log, which is the only storage
path — there is no setting to turn it off. Coalesced step-level facts
(completed text/reasoning blocks, tool calls and results, run lifecycle) are
appended to the `instance_ai_events` table and replay reads the database;
token deltas are never persisted. Rows cascade-delete with their thread
(`N8N_INSTANCE_AI_THREAD_TTL_DAYS`). Nothing is retained in the process, so
cursors stay valid across restarts and across mains sharing one database.
Runtime behavior:
- One active run per thread. Additional `POST /instance-ai/chat/:threadId`
@@ -295,11 +279,6 @@ N8N_INSTANCE_AI_VERTEX_PROJECT_ID=my-gcp-project
N8N_INSTANCE_AI_VERTEX_LOCATION=global
N8N_INSTANCE_AI_VERTEX_SERVICE_ACCOUNT_JSON='{"type":"service_account",...}'
# Legacy non-durable stream filtering — email with a custom placeholder
N8N_INSTANCE_AI_DURABLE_LOG=false
N8N_INSTANCE_AI_OUTPUT_REDACTION_PII=email
N8N_INSTANCE_AI_OUTPUT_REDACTION_PLACEHOLDER=‹redacted›
# Observational memory tuning
N8N_INSTANCE_AI_OBSERVER_MESSAGE_TOKENS=30000
N8N_INSTANCE_AI_REFLECTOR_OBSERVATION_TOKENS=40000
packages/@n8n/instance-ai/docs/streaming-protocol.md
+16 -14
View File
@@ -421,14 +421,20 @@ graph LR
S2[Sub-Agent B] -->|publish| Bus
end
Bus --> Store[Replay Storage]
Bus -->|enqueue| Log[Durable Event Log]
Log -->|"drained (seq assigned)"| Bus
Log --> DB[(instance_ai_events)]
Bus --> SSE[SSE Endpoint]
Bus -->|relay| Siblings[Sibling mains]
SSE --> FE[Frontend]
```
All events are published to a per-thread channel on the event bus and delivered
to connected SSE clients. The durable log persists replayable facts. Ephemeral
transport events remain live-only.
All events are published to a per-thread channel on the event bus, which
enqueues them into the durable event log. The log assigns each durable fact a
per-thread `seq`, persists it, and hands the event back to the bus for
delivery to connected SSE clients and — in multi-main — to sibling mains.
Ephemeral transport events remain live-only, and the bus itself retains
nothing.
### Implementations
@@ -437,13 +443,10 @@ transport events remain live-only.
| Single instance | In-process `EventEmitter` | Zero infrastructure |
| Queue mode | Redis Pub/Sub | n8n already uses Redis |
Replay storage depends on `N8N_INSTANCE_AI_DURABLE_LOG`. On (the default),
the durable event log (`instance_ai_events`) is the replay source: coalesced
step-level facts are appended with a per-thread `seq` assigned by the
writer's drain, so cursors stay valid across restarts and across mains
sharing one database. Off (the rollback switch until Gate B), replay serves
from a bounded in-memory buffer per thread (500 events / 2 MB, FIFO-evicted;
ids reset on restart).
The durable event log (`instance_ai_events`) is the only replay source:
coalesced step-level facts are appended with a per-thread `seq` assigned by
the writer's drain, so cursors stay valid across restarts and across mains
sharing one database.
### Reconnection & Replay (Canonical Rule)
@@ -471,9 +474,8 @@ connection may occasionally deliver a lower id after a higher one. The
frontend therefore tracks its reconnect cursor as the max id seen and drops
already-seen ids on replay overlap.
With the durable log enabled (`N8N_INSTANCE_AI_DURABLE_LOG`), ids are
database-assigned sequence numbers and only DURABLE facts carry an `id:`
line. Ephemeral frames (`text-delta`, `reasoning-delta`, `status`,
Ids are database-assigned sequence numbers, and only DURABLE facts carry an
`id:` line. Ephemeral frames (`text-delta`, `reasoning-delta`, `status`,
`filesystem-request`) are live-only: their SSE frames have no `id:` line, so
the browser's replay cursor never points at them (the same mechanism as the
`run-sync` control frames). On replay, the deltas a client missed are covered
packages/@n8n/instance-ai/evaluations/harness/in-memory-event-bus.ts
-19
View File
@@ -28,25 +28,6 @@ export function createInMemoryEventBus(): InstanceAiEventBus {
);
};
},
getEventsAfter(threadId, afterId) {
return (storeByThread.get(threadId) ?? []).filter(
(event) => event.id !== undefined && event.id > afterId,
);
},
getEventsForRun(threadId, runId) {
return (storeByThread.get(threadId) ?? [])
.map((event) => event.event)
.filter((event) => 'runId' in event && event.runId === runId);
},
getEventsForRuns(threadId, runIds) {
const runIdSet = new Set(runIds);
return (storeByThread.get(threadId) ?? [])
.map((event) => event.event)
.filter((event) => 'runId' in event && runIdSet.has(event.runId));
},
async getNextEventId(threadId) {
return await Promise.resolve((storeByThread.get(threadId) ?? []).length + 1);
},
};
}
packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts
+5 -29
View File
@@ -13,7 +13,11 @@ export interface StoredEvent {
type Unsubscribe = () => void;
/** Domain-level interface -- no transport details leak through. */
/**
* Domain-level interface -- no transport details leak through. Publish and
* subscribe only: events are persisted to `instance_ai_events`, and every read
* (replay, run-scoped, cursor seeding) goes through the durable event log.
*/
export interface InstanceAiEventBus {
/**
* Publish an event to a thread channel.
@@ -26,32 +30,4 @@ export interface InstanceAiEventBus {
* Returns an unsubscribe function.
*/
subscribe(threadId: string, handler: (storedEvent: StoredEvent) => void): Unsubscribe;
/**
* Retrieve all persisted events for a thread with id > afterId.
* Used for replay on reconnect.
* Returns events in id order (ascending).
*/
getEventsAfter(threadId: string, afterId: number): StoredEvent[];
/**
* Retrieve all persisted events for a thread that belong to a specific run.
* More efficient than getEventsAfter(threadId, 0) + filter when only one
* run's events are needed (e.g. building agent tree snapshots).
*/
getEventsForRun(threadId: string, runId: string): InstanceAiEvent[];
/**
* Retrieve all persisted events for a thread that belong to any of the
* specified runs. Used for rebuilding merged assistant turns that span
* multiple auto-follow-up runs.
*/
getEventsForRuns(threadId: string, runIds: string[]): InstanceAiEvent[];
/**
* Get the next event ID that will be assigned for a thread.
* Used to seed the frontend's SSE replay cursor after message hydration.
* Async because multi-main implementations read a shared sequence.
*/
getNextEventId(threadId: string): Promise<number>;
}
packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts
-1
View File
@@ -8,7 +8,6 @@ function createEventBus() {
publish: vi.fn(),
subscribe: vi.fn(),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
};
packages/@n8n/instance-ai/src/runtime/__tests__/stream-runner.test.ts
-1
View File
@@ -25,7 +25,6 @@ function createEventBus() {
publish: vi.fn(),
subscribe: vi.fn(),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
};
packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts
+5 -1
View File
@@ -41,7 +41,11 @@ export interface ResumableStreamContext {
onActivity?: () => void;
/** Stop consuming after the current chunk has been mapped and published. */
stopSignal?: () => OrchestratorRunStopSignal | undefined;
/** Output-redaction policy: omit for the safe default, or `false` to disable. */
/**
* Redaction policy. `false` disables scanning; OMITTING IT ENABLES the
* default policy, which on the durable-log path would persist redacted text
* — Instance AI passes `false` everywhere (raw-at-rest, INS-837).
*/
outputRedaction?: RedactionOptions | false;
}
packages/@n8n/instance-ai/src/runtime/stream-runner.ts
+5 -1
View File
@@ -31,7 +31,11 @@ export interface StreamRunOptions {
logger: Logger;
onActivity?: () => void;
stopSignal?: () => OrchestratorRunStopSignal | undefined;
/** Output-redaction policy: omit for the safe default, or `false` to disable. */
/**
* Redaction policy. `false` disables scanning; OMITTING IT ENABLES the
* default policy, which on the durable-log path would persist redacted text
* — Instance AI passes `false` everywhere (raw-at-rest, INS-837).
*/
outputRedaction?: RedactionOptions | false;
}
packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts
+2 -2
View File
@@ -62,8 +62,8 @@ function formatWorkSummaryCounts(workSummary?: WorkSummary): string {
}
function hasText(event: InstanceAiEvent): boolean {
// The durable log stores coalesced text-block facts, never deltas, so
// flag-on guard reads must recognize both shapes of streamed text.
// The durable log stores coalesced text-block facts, never deltas, so guard
// reads must recognize both shapes of streamed text.
return (
(event.type === 'text-delta' || event.type === 'text-block') &&
event.payload.text.trim().length > 0
packages/@n8n/instance-ai/src/stream/__tests__/consume-with-hitl.test.ts
-4
View File
@@ -16,10 +16,6 @@ function createEventBus(): InstanceAiEventBus {
return {
publish: vi.fn(),
subscribe: vi.fn().mockReturnValue(() => {}),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
};
}
packages/@n8n/instance-ai/src/stream/consume-with-hitl.ts
+10 -2
View File
@@ -35,7 +35,11 @@ export interface ConsumeWithHitlOptions {
resumeOptions?: Record<string, unknown>;
/** Native agent persistence owner for suspended sub-agent state. */
persistence?: { threadId: string; resourceId: string };
/** Output-redaction policy: omit for the safe default, or `false` to disable. */
/**
* Redaction policy. `false` disables scanning; OMITTING IT ENABLES the
* default policy, which on the durable-log path would persist redacted text
* — Instance AI passes `false` everywhere (raw-at-rest, INS-837).
*/
outputRedaction?: RedactionOptions | false;
}
@@ -127,7 +131,11 @@ export interface ConsumeStreamCascadingOptions {
logger: Logger;
threadId: string;
abortSignal: AbortSignal;
/** Output-redaction policy: omit for the safe default, or `false` to disable. */
/**
* Redaction policy. `false` disables scanning; OMITTING IT ENABLES the
* default policy, which on the durable-log path would persist redacted text
* — Instance AI passes `false` everywhere (raw-at-rest, INS-837).
*/
outputRedaction?: RedactionOptions | false;
}
packages/@n8n/instance-ai/src/stream/output-redaction.ts
+8 -4
View File
@@ -15,8 +15,10 @@ import type { Logger } from '../logger';
* secret patterns plus credit-card numbers. Other PII categories (`email`,
* `ssn-us`) are implemented but off by default until we decide which to enable.
*
* Instance AI always redacts matches. The SDK's `GuardrailStrategy` also defines
* `block` and `warn`, but those are not implemented here.
* Instance AI's own streams pass `false` (raw-at-rest, INS-837), so this policy
* applies only to callers that opt in. When it does run it always redacts
* matches; the SDK's `GuardrailStrategy` also defines `block` and `warn`, but
* those are not implemented here.
*/
export const DEFAULT_OUTPUT_REDACTION_OPTIONS: RedactionOptions = {
secrets: true,
@@ -29,8 +31,10 @@ interface OutputRedactorContext {
runId: string;
agentId: string;
/**
* Redaction policy: omit for the safe default, pass options to customise, or
* `false` to disable scanning entirely (events pass through untouched).
* Redaction policy: omit for the default policy, pass options to customise,
* or `false` to disable scanning entirely (events pass through untouched).
* NOTE: omission means ENABLED — callers on a persistence path must pass
* `false` explicitly or the stored text is redacted.
*/
options?: RedactionOptions | false;
}
packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.cascade-restart.test.ts
-4
View File
@@ -314,10 +314,6 @@ function createEventBusStub(): InstanceAiEventBus {
return {
publish: () => {},
subscribe: () => () => {},
getEventsAfter: () => [],
getEventsForRun: () => [],
getEventsForRuns: () => [],
getNextEventId: async () => await Promise.resolve(1),
};
}
packages/@n8n/instance-ai/src/tools/orchestration/__tests__/complete-checkpoint.tool.test.ts
-4
View File
@@ -36,10 +36,6 @@ function makeContext(
eventBus: {
publish: vi.fn(),
subscribe: vi.fn(),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
},
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
domainTools: createToolRegistry(),
packages/@n8n/instance-ai/src/tools/orchestration/__tests__/plan.tool.test.ts
-4
View File
@@ -25,10 +25,6 @@ function createMockContext(overrides: Partial<OrchestrationContext> = {}): Orche
eventBus: {
publish: vi.fn(),
subscribe: vi.fn(),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
},
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
domainTools: createToolRegistry(),
packages/@n8n/instance-ai/src/tools/orchestration/__tests__/report-verification-verdict.tool.test.ts
-4
View File
@@ -31,10 +31,6 @@ function createMockContext(overrides: Partial<OrchestrationContext> = {}): Orche
eventBus: {
publish: vi.fn(),
subscribe: vi.fn(),
getEventsAfter: vi.fn(),
getNextEventId: vi.fn(),
getEventsForRun: vi.fn().mockReturnValue([]),
getEventsForRuns: vi.fn().mockReturnValue([]),
},
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
domainTools: createToolRegistry(),
packages/@n8n/instance-ai/src/types.ts
+5 -1
View File
@@ -1697,7 +1697,11 @@ export interface OrchestrationContext {
checkpointStore?: CheckpointStore;
eventBus: InstanceAiEventBus;
logger: Logger;
/** Output-redaction policy for sub-agent streams: omit for the safe default, or `false` to disable. */
/**
* Redaction policy. `false` disables scanning; OMITTING IT ENABLES the
* default policy, which on the durable-log path would persist redacted text
* — Instance AI passes `false` everywhere (raw-at-rest, INS-837).
*/
outputRedaction?: RedactionOptions | false;
trackTelemetry?: (eventName: string, properties: Record<string, GenericValue>) => void;
/**
packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts
+1 -2
View File
@@ -74,8 +74,7 @@ export class PrometheusInstanceAiMetricsService implements PrometheusMetricsColl
},
});
// Durable event log (RFC: instance-ai durable event log). All series are
// flat when N8N_INSTANCE_AI_DURABLE_LOG is off.
// Durable event log (RFC: instance-ai durable event log).
const durableLogRowsTotal = new promClient.Counter({
name: `${this.config.prefix}instance_ai_durable_log_rows_total`,
help: 'Durable Instance AI event rows appended (structural facts + coalesced blocks).',
packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts
+23 -33
View File
@@ -83,13 +83,10 @@ function installLogDouble(rows: LogRow[] = []): void {
}
const mockDurableLogMetrics = { recordFoldRead: vi.fn(), notifyParserFallbacks: vi.fn() };
function createService(
options: { threadTtlDays?: number; durableLog?: boolean } = {},
): InstanceAiMemoryService {
function createService(options: { threadTtlDays?: number } = {}): InstanceAiMemoryService {
const mockConfig = {
instanceAi: {
threadTtlDays: options.threadTtlDays ?? 0,
durableLog: options.durableLog ?? false,
},
database: {
type: 'postgresdb',
@@ -413,8 +410,8 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
it('derives the tree from the log even when the stored snapshot is degenerate', async () => {
// The stored snapshot was built over an evicted buffer: an empty
// cancelled tree with none of the run's work (the INS-595 bug family).
// Snapshot rows keep being written flag-on (the rollback path until Gate
// B), but they are never read once the thread has log rows.
// Snapshot rows keep being written, but they are never read once the
// thread has log rows.
mockDbSnapshotStorage.getForWindow.mockResolvedValue([
{
tree: makeTree({ status: 'cancelled', textContent: '', timeline: [], toolCalls: [] }),
@@ -445,7 +442,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
const assistant = result.messages[1];
@@ -485,7 +482,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
const assistant = result.messages[1];
@@ -558,7 +555,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1', {
excludeRunIds: ['run_b'],
});
@@ -609,7 +606,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1', {
excludeRunIds: ['run_b'],
});
@@ -647,7 +644,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1', {
excludeRunIds: ['run_b'],
excludeMessageGroupIds: ['mg-1'],
@@ -703,7 +700,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
// Only run_done's entry is derived, exactly as the driving main would.
@@ -735,7 +732,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(mockDbSnapshotStorage.getForWindow).not.toHaveBeenCalled();
@@ -804,7 +801,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(result.messages).toHaveLength(2);
@@ -919,7 +916,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
// Four messages, no trailing orphan card.
@@ -984,7 +981,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
const timeline = result.messages[1].agentTree?.timeline ?? [];
@@ -998,7 +995,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
{ tree, runId: 'run_abc', createdAt: at, updatedAt: at },
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(result.messages[1].agentTree).toStrictEqual(tree);
@@ -1012,7 +1009,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
]);
mockEventLogRepository.getRunStarts.mockRejectedValue(new Error('db down'));
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(result.messages[1].agentTree).toStrictEqual(tree);
@@ -1046,20 +1043,13 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
),
]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(result.messages[1].agentTree).toStrictEqual(tree);
expect(mockDurableLogMetrics.recordFoldRead).not.toHaveBeenCalled();
});
it('never reads the log when the flag is off', async () => {
const service = createService();
await service.getRichMessages('user-1', 'thread-1');
expect(mockEventLogRepository.getRunStarts).not.toHaveBeenCalled();
});
describe('hydration is scoped to the requested page', () => {
const oldAt = new Date('2025-06-01T00:00:00.000Z');
@@ -1112,7 +1102,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
// parser surfaced it as a message of its own.
setLogRows([...runRows('run_old', oldAt), ...runRows('run_recent', at)]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(mockEventLogRepository.getForThreadRuns).toHaveBeenCalledWith('thread-1', [
@@ -1132,7 +1122,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
const laterAt = new Date('2026-01-01T00:00:09.000Z');
setLogRows([...runRows('run_recent', at), ...runRows('run_live', laterAt)]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(mockEventLogRepository.findRunIdsInWindow).toHaveBeenCalledWith('thread-1', {
@@ -1157,7 +1147,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
});
setLogRows([...runRows('run_recent', at)]);
const service = createService({ durableLog: true });
const service = createService();
await service.getRichMessages('user-1', 'thread-1', { page: 1 });
expect(mockListMessages).toHaveBeenCalledWith(
@@ -1200,7 +1190,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
mockListMessages.mockResolvedValue({ messages: [] });
setLogRows([...runRows('run_recent', at)]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1', { page: 3 });
expect(mockEventLogRepository.findRunIdsInWindow).not.toHaveBeenCalled();
@@ -1219,7 +1209,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
});
setLogRows([...runRows('run_parent', at, 'mg-1'), ...runRows('run_bg', afterPage, 'mg-1')]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1', { page: 1 });
// `run_bg` is outside the page bounds; it rides in on its group.
@@ -1238,7 +1228,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
mockListMessages.mockResolvedValue({ messages: [] });
setLogRows([...runRows('run_live', at)]);
const service = createService({ durableLog: true });
const service = createService();
const result = await service.getRichMessages('user-1', 'thread-1');
expect(mockEventLogRepository.findRunIdsInWindow).toHaveBeenCalledWith('thread-1', {});
@@ -1251,7 +1241,7 @@ describe('InstanceAiMemoryService.getRichMessages — durable-log fold-on-read',
mockListMessages.mockResolvedValue({ messages: [userMessage, assistantMessage] });
setLogRows([...runRows('run_recent', at)]);
const service = createService({ durableLog: true });
const service = createService();
await service.getRichMessages('user-1', 'thread-1');
expect(mockEventLogRepository.findRunIdsInWindow).toHaveBeenCalledWith('thread-1', {
packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts
+4 -6
View File
@@ -218,7 +218,6 @@ function createService(snapshotTree?: InstanceAiAgentNode): {
};
const options = {
durableLog: false,
eventBus: deps.eventBus,
dbSnapshotStorage: deps.dbSnapshotStorage,
agentMemory: {},
@@ -275,7 +274,7 @@ describe('InstanceAiTerminalOutcomeService — terminal outcome replay', () => {
expect(deps.dbSnapshotStorage.updateLast).toHaveBeenCalledTimes(1);
expect(deps.eventBus.publish).toHaveBeenCalledWith('thread-a', {
type: 'text-delta',
type: 'text-block',
runId: outcome.runId,
agentId: 'orchestrator-run-1',
responseId: `background-outcome:${outcome.id}`,
@@ -342,7 +341,7 @@ describe('InstanceAiTerminalOutcomeService — terminal outcome replay', () => {
await service.replayUndeliveredTerminalOutcomes('thread-a', { delivery: 'event' });
expect(deps.eventBus.publish).toHaveBeenCalledWith('thread-a', {
type: 'text-delta',
type: 'text-block',
runId: outcome.runId,
agentId: 'orchestrator-run-1',
responseId: `background-outcome:${outcome.id}`,
@@ -389,7 +388,7 @@ describe('InstanceAiTerminalOutcomeService — background outcome recording', ()
expect(deps.eventBus.publish).toHaveBeenCalledWith(
'thread-a',
expect.objectContaining({
type: 'text-delta',
type: 'text-block',
payload: { text: 'The background workflow-builder task finished.' },
}),
);
@@ -416,12 +415,11 @@ describe('InstanceAiTerminalOutcomeService — background outcome recording', ()
});
describe('InstanceAiTerminalOutcomeService — durable-log outcome lines', () => {
it('publishes the outcome line as a text-block when the durable log is on', async () => {
it('publishes the outcome line as a persisted text-block, not a trailing delta', async () => {
// A trailing delta would race the coalescer's idle flush on an immediate
// page reload; a text-block is persisted before it is emitted live.
const { deps } = createService(makeAgentTree());
const service = new InstanceAiTerminalOutcomeService({
durableLog: true,
eventBus: deps.eventBus,
dbSnapshotStorage: deps.dbSnapshotStorage,
agentMemory: {},
packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts
+27 -16
View File
@@ -121,7 +121,7 @@ describe('InstanceAiController', () => {
const publisher = mock<Publisher>();
const urlService = mock<UrlService>();
const globalConfig = mock<GlobalConfig>({
instanceAi: { gatewayApiKey: 'static-key', durableLog: false },
instanceAi: { gatewayApiKey: 'static-key' },
editorBaseUrl: 'http://localhost:5678',
port: 5678,
});
@@ -164,6 +164,11 @@ describe('InstanceAiController', () => {
beforeEach(() => {
vi.clearAllMocks();
// SSE replay reads the durable log; default to an empty thread so tests
// only stub what they exercise.
eventLog.getEventsAfter.mockResolvedValue([]);
eventLog.getEventsForRuns.mockResolvedValue([]);
eventLog.getOpenSegments.mockReturnValue([]);
settingsService.isInstanceAiEnabled.mockReturnValue(true);
});
@@ -409,7 +414,7 @@ describe('InstanceAiController', () => {
it('should bootstrap run-sync from the richer persisted snapshot when live events are incomplete', async () => {
memoryService.checkThreadOwnership.mockResolvedValue('owned');
eventBus.getEventsAfter.mockReturnValue([]);
eventLog.getEventsAfter.mockResolvedValue([]);
instanceAiService.getThreadStatus.mockReturnValue({
hasActiveRun: true,
isSuspended: false,
@@ -417,7 +422,7 @@ describe('InstanceAiController', () => {
} as never);
instanceAiService.getMessageGroupId.mockReturnValue('mg-1');
instanceAiService.getRunIdsForMessageGroup.mockReturnValue(['run-1']);
eventBus.getEventsForRuns.mockReturnValue([
eventLog.getEventsForRuns.mockResolvedValue([
{
type: 'run-start',
runId: 'run-1',
@@ -488,12 +493,14 @@ describe('InstanceAiController', () => {
} as never);
instanceAiService.getMessageGroupId.mockReturnValue('mg-1');
instanceAiService.getRunIdsForMessageGroup.mockReturnValue(['run-1']);
eventBus.getEventsForRuns.mockReturnValue([]);
eventBus.getEventsAfter.mockReturnValue([]);
eventLog.getEventsForRuns.mockResolvedValue([]);
eventLog.getEventsAfter.mockResolvedValue([]);
let subscribeHandler: ((stored: { id: number; event: unknown }) => void) | undefined;
eventBus.subscribe.mockImplementation((_threadId, handler) => {
subscribeHandler = handler as typeof subscribeHandler;
// The bootstrap also registers a buffering subscription; the live
// delivery handler is the first one.
subscribeHandler ??= handler as (stored: { id: number; event: unknown }) => void;
return vi.fn();
});
@@ -506,7 +513,7 @@ describe('InstanceAiController', () => {
};
memoryService.getLatestRunSnapshot.mockImplementation(async () => {
subscribeHandler!(midAwaitEvent);
eventBus.getEventsAfter.mockReturnValue([midAwaitEvent] as never);
eventLog.getEventsAfter.mockResolvedValue([midAwaitEvent] as never);
return undefined;
});
@@ -546,8 +553,8 @@ describe('InstanceAiController', () => {
} as never);
instanceAiService.getMessageGroupId.mockReturnValue('mg-1');
instanceAiService.getRunIdsForMessageGroup.mockReturnValue(['run-1']);
eventBus.getEventsForRuns.mockReturnValue([]);
eventBus.getEventsAfter.mockReturnValue([
eventLog.getEventsForRuns.mockResolvedValue([]);
eventLog.getEventsAfter.mockResolvedValue([
{ id: 1, event: { type: 'text-delta', runId: 'run-1', agentId: 'a1', payload: {} } },
] as never);
@@ -597,10 +604,12 @@ describe('InstanceAiController', () => {
// Capture the subscribe handler
let subscribeHandler: ((stored: { id: number; event: unknown }) => void) | undefined;
eventBus.subscribe.mockImplementation((_threadId, handler) => {
subscribeHandler = handler as typeof subscribeHandler;
// The bootstrap also registers a buffering subscription; the live
// delivery handler is the first one.
subscribeHandler ??= handler as (stored: { id: number; event: unknown }) => void;
return vi.fn();
});
eventBus.getEventsAfter.mockReturnValue([]);
eventLog.getEventsAfter.mockResolvedValue([]);
instanceAiService.getThreadStatus.mockReturnValue({
hasActiveRun: false,
isSuspended: false,
@@ -644,10 +653,12 @@ describe('InstanceAiController', () => {
let subscribeHandler: ((stored: { id: number; event: unknown }) => void) | undefined;
eventBus.subscribe.mockImplementation((_threadId, handler) => {
subscribeHandler = handler as typeof subscribeHandler;
// The bootstrap also registers a buffering subscription; the live
// delivery handler is the first one.
subscribeHandler ??= handler as (stored: { id: number; event: unknown }) => void;
return vi.fn();
});
eventBus.getEventsAfter.mockReturnValue([]);
eventLog.getEventsAfter.mockResolvedValue([]);
instanceAiService.getThreadStatus.mockReturnValue({
hasActiveRun: false,
isSuspended: false,
@@ -1515,7 +1526,7 @@ describe('InstanceAiController', () => {
it('should return rich messages with nextEventId', async () => {
const richResult = mock<Omit<InstanceAiRichMessagesResponse, 'nextEventId'>>();
memoryService.getRichMessages.mockResolvedValue(richResult);
eventBus.getNextEventId.mockResolvedValue(42);
eventLog.getNextEventId.mockResolvedValue(42);
const query = mock<InstanceAiThreadMessagesQuery>({
limit: 50,
page: 0,
@@ -1985,7 +1996,7 @@ describe('InstanceAiController', () => {
});
});
describe('InstanceAiController — durable-log SSE replay (flag on)', () => {
describe('InstanceAiController — durable-log SSE replay', () => {
const instanceAiService = mock<InstanceAiService>();
const memoryService = mock<InstanceAiMemoryService>();
const settingsService = mock<InstanceAiSettingsService>();
@@ -1993,7 +2004,7 @@ describe('InstanceAiController — durable-log SSE replay (flag on)', () => {
const eventLog = mock<DurableEventLog>();
const durableLogMetrics = mock<DurableLogMetrics>();
const globalConfig = mock<GlobalConfig>({
instanceAi: { gatewayApiKey: 'static-key', durableLog: true },
instanceAi: { gatewayApiKey: 'static-key' },
editorBaseUrl: 'http://localhost:5678',
port: 5678,
});
packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts
+10 -20
View File
@@ -673,12 +673,7 @@ type TerminalGuardOrderServiceInternals = {
suspendedThreads: { dropPendingConfirmationsForThread: Mock; persistPendingConfirmation: Mock };
logger: { warn: Mock; error: Mock };
instanceAiErrorReporter: ReturnType<typeof createInstanceAiErrorReporterMock>;
instanceAiConfig: {
outputRedactionEnabled: boolean;
outputRedactionSecrets: boolean;
outputRedactionPii: string;
outputRedactionPlaceholder: string;
};
instanceAiConfig: {};
tracing: {
finalizeRunTracing: Mock;
finalizeDetachedTraceRun: Mock;
@@ -780,7 +775,6 @@ type SnapshotServiceInternals = {
getEventsForRuns: Mock;
};
eventLog: { flush: Mock; getEventsForRuns: Mock };
instanceAiConfig: { durableLog: boolean };
tracing: { getTraceContext: Mock };
logger: { warn: Mock };
};
@@ -815,12 +809,7 @@ function createTerminalGuardOrderService(): TerminalGuardOrderServiceInternals {
};
service.logger = { warn: vi.fn(), error: vi.fn() };
service.instanceAiErrorReporter = createInstanceAiErrorReporterMock();
service.instanceAiConfig = {
outputRedactionEnabled: true,
outputRedactionSecrets: true,
outputRedactionPii: 'credit-card',
outputRedactionPlaceholder: '[REDACTED]',
};
service.instanceAiConfig = {};
service.tracing = {
finalizeRunTracing: vi.fn(async () => {}),
finalizeDetachedTraceRun: vi.fn(async () => {}),
@@ -845,7 +834,6 @@ function createTerminalGuardOrderService(): TerminalGuardOrderServiceInternals {
service.preserveHitlOnShutdown = new Set();
service.terminalOutcome = new InstanceAiTerminalOutcomeService({
durableLog: false,
eventBus: service.eventBus,
dbSnapshotStorage: {},
agentMemory: {},
@@ -885,7 +873,6 @@ function createSnapshotService(): SnapshotServiceInternals {
getEventsForRuns: vi.fn(() => []),
};
service.eventLog = { flush: vi.fn(async () => {}), getEventsForRuns: vi.fn(async () => []) };
service.instanceAiConfig = { durableLog: false };
service.tracing = { getTraceContext: vi.fn(() => undefined) };
service.logger = { warn: vi.fn() };
return service;
@@ -936,6 +923,7 @@ describe('InstanceAiService — runtime workspace setup', () => {
abortSignal: AbortSignal,
) => Promise<{
orchestrationContext: {
outputRedaction?: unknown;
workspace?: unknown;
runtimeSkills?: {
registry: { skillsHash: string; skills: Array<{ id: string }> };
@@ -1078,6 +1066,10 @@ describe('InstanceAiService — runtime workspace setup', () => {
new AbortController().signal,
);
// OutputRedactor treats an OMITTED policy as ENABLED (`options !== false`),
// so this must stay an explicit false or every stream is scanned and the
// durable log stores redacted text instead of raw (INS-837).
expect(environment.orchestrationContext.outputRedaction).toBe(false);
expect(createLazyRuntimeWorkspace).toHaveBeenCalledTimes(2);
expect(createLazyRuntimeWorkspace).toHaveBeenNthCalledWith(
2,
@@ -2995,7 +2987,7 @@ describe('InstanceAiService — agent tree snapshots', () => {
save: vi.fn(async () => {}),
updateLast: vi.fn(async () => {}),
};
service.eventBus.getEventsForRuns.mockReturnValue([terminalEvent]);
service.eventLog.getEventsForRuns.mockResolvedValue([terminalEvent]);
await service.saveAgentTreeSnapshot(
'thread-a',
@@ -3010,7 +3002,7 @@ describe('InstanceAiService — agent tree snapshots', () => {
messageGroupId: 'group-old',
runId: 'run-background',
});
expect(service.eventBus.getEventsForRuns).toHaveBeenCalledWith('thread-a', [
expect(service.eventLog.getEventsForRuns).toHaveBeenCalledWith('thread-a', [
'run-original',
'run-background',
]);
@@ -3059,9 +3051,8 @@ describe('InstanceAiService — agent tree snapshots', () => {
);
});
it('reads snapshot input from the durable log instead of the bus when the flag is on', async () => {
it('reads snapshot input from the durable log', async () => {
const service = createSnapshotService();
service.instanceAiConfig.durableLog = true;
const logEvent: InstanceAiEvent = {
type: 'text-delta',
runId: 'run-1',
@@ -3078,7 +3069,6 @@ describe('InstanceAiService — agent tree snapshots', () => {
await service.saveAgentTreeSnapshot('thread-a', 'run-1', snapshotStorage);
expect(service.eventLog.getEventsForRuns).toHaveBeenCalledWith('thread-a', ['run-1']);
expect(service.eventBus.getEventsForRun).not.toHaveBeenCalled();
// Read-own-writes barrier: the drain settles before the snapshot input is
// read, so a just-published terminal fact can't be missing from the tree.
expect(service.eventLog.flush).toHaveBeenCalledWith('thread-a');
packages/cli/src/modules/instance-ai/__tests__/output-redaction-config.test.ts
-65
View File
@@ -1,65 +0,0 @@
import type { InstanceAiConfig } from '@n8n/config';
import { resolveOutputRedaction } from '../output-redaction-config';
function config(overrides: Partial<InstanceAiConfig> = {}): InstanceAiConfig {
return {
outputRedactionEnabled: true,
outputRedactionSecrets: true,
outputRedactionPii: 'email,credit-card,ssn-us',
outputRedactionPlaceholder: '[REDACTED]',
...overrides,
} as InstanceAiConfig;
}
describe('resolveOutputRedaction', () => {
it('returns false when disabled', () => {
expect(resolveOutputRedaction(config({ outputRedactionEnabled: false }))).toBe(false);
});
it('returns false when the durable log is on, even with redaction enabled (raw-at-rest)', () => {
expect(resolveOutputRedaction(config({ durableLog: true }))).toBe(false);
});
it('maps secrets, PII categories, and placeholder from config', () => {
expect(resolveOutputRedaction(config())).toEqual({
secrets: true,
detect: ['email', 'credit-card', 'ssn-us'],
placeholder: '[REDACTED]',
});
});
it('trims and drops unknown PII categories', () => {
expect(resolveOutputRedaction(config({ outputRedactionPii: 'email, bogus , ssn-us' }))).toEqual(
{
secrets: true,
detect: ['email', 'ssn-us'],
placeholder: '[REDACTED]',
},
);
});
it('drops unsupported categories (e.g. address, which has no detector)', () => {
expect(
resolveOutputRedaction(config({ outputRedactionPii: 'email,phone,address,ssn-us' })),
).toMatchObject({ detect: ['email', 'phone', 'ssn-us'] });
});
it('honors the secrets toggle and an empty PII list', () => {
expect(
resolveOutputRedaction(config({ outputRedactionSecrets: false, outputRedactionPii: '' })),
).toEqual({ secrets: false, detect: [], placeholder: '[REDACTED]' });
});
it('uses a custom placeholder', () => {
expect(resolveOutputRedaction(config({ outputRedactionPlaceholder: '***' }))).toMatchObject({
placeholder: '***',
});
});
it('omits the placeholder when configured blank (engine default applies)', () => {
expect(resolveOutputRedaction(config({ outputRedactionPlaceholder: '' }))).not.toHaveProperty(
'placeholder',
);
});
});
packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts
+106 -570
View File
@@ -1,28 +1,24 @@
import type { Logger } from '@n8n/backend-common';
import type { InstanceAiEvent } from '@n8n/api-types';
import type { GlobalConfig } from '@n8n/config';
import { mock } from 'vitest-mock-extended';
import type { Logger } from '@n8n/backend-common';
import type { InstanceSettings } from 'n8n-core';
import { mock } from 'vitest-mock-extended';
import type { Publisher } from '@/scaling/pubsub/publisher.service';
import type { DurableEventLog } from '../durable-event-log';
import { InProcessEventBus } from '../in-process-event-bus';
function makeEvent(type: string, runId: string): InstanceAiEvent {
type EmitFn = (drained: { id?: number; event: InstanceAiEvent; live: boolean }) => void;
function makeEvent(text: string, runId: string): InstanceAiEvent {
return {
type: 'text-delta',
runId,
agentId: 'agent-001',
payload: { text: `${type}-${runId}` },
payload: { text },
};
}
/** Flush the per-thread drain: each batch awaits one (mock) Redis round trip. */
async function flushDrain() {
await new Promise((resolve) => setImmediate(resolve));
}
describe('InProcessEventBus', () => {
let bus: InProcessEventBus;
let publisher: ReturnType<typeof mock<Publisher>>;
@@ -30,499 +26,97 @@ describe('InProcessEventBus', () => {
let logger: ReturnType<typeof mock<Logger>>;
let instanceSettings: { isMultiMain: boolean };
/** Shared fake Redis sequence — one Map plays the role of the Redis server,
* so two bus instances built in one test behave like two mains. */
let seqByKey: Map<string, number>;
let redisFailure: Error | null;
let incrbyCalls: Array<{ key: string; count: number }>;
let deletedKeys: string[];
const redisClient = {
multi: (): unknown => {
let incrArgs: { key: string; count: number } | null = null;
const chain = {
incrby(key: string, count: number) {
incrArgs = { key, count };
return chain;
},
expire() {
return chain;
},
async exec() {
if (redisFailure) throw redisFailure;
const { key, count } = incrArgs!;
incrbyCalls.push({ key, count });
const value = (seqByKey.get(key) ?? 0) + count;
seqByKey.set(key, value);
return [
[null, value],
[null, 1],
];
},
};
return chain;
},
async get(key: string) {
if (redisFailure) throw redisFailure;
const value = seqByKey.get(key);
return value === undefined ? null : String(value);
},
async del(key: string) {
deletedKeys.push(key);
seqByKey.delete(key);
return 1;
},
};
function buildBus({ durableLog = false } = {}) {
function buildBus() {
logger = mock<Logger>();
logger.scoped.mockReturnValue(logger);
publisher = mock<Publisher>();
publisher.publishCommand.mockResolvedValue(undefined);
publisher.getClient.mockReturnValue(redisClient as never);
// Flag off: the durable log is never touched, so a bare mock suffices.
eventLog = mock<DurableEventLog>();
const globalConfig = mock<GlobalConfig>({
redis: { prefix: 'n8n' },
instanceAi: { durableLog },
});
return new InProcessEventBus(
logger,
instanceSettings as InstanceSettings,
publisher,
eventLog,
globalConfig,
);
return new InProcessEventBus(logger, instanceSettings as InstanceSettings, publisher, eventLog);
}
/** Route a publish through the mocked drain and hand back its emit callback. */
function publishAndCaptureEmit(threadId: string, event: InstanceAiEvent): EmitFn {
bus.publish(threadId, event);
const call = eventLog.publish.mock.calls.at(-1)!;
expect(call[0]).toBe(threadId);
// publish() stamps `ts` onto a copy before handing it to the log
expect(call[1]).toEqual({ ...event, ts: expect.any(Number) });
return call[2] as EmitFn;
}
beforeEach(() => {
instanceSettings = { isMultiMain: false };
seqByKey = new Map();
redisFailure = null;
incrbyCalls = [];
deletedKeys = [];
bus = buildBus();
});
afterEach(() => {
bus.clear();
});
describe('publish (single-main)', () => {
it('should assign monotonically increasing IDs per thread in the same tick', () => {
describe('publish', () => {
it('enqueues into the durable log rather than storing anything itself', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
bus.publish('thread-1', makeEvent('c', 'run_1'));
const events = bus.getEventsAfter('thread-1', 0);
expect(events).toHaveLength(3);
expect(events[0].id).toBe(1);
expect(events[1].id).toBe(2);
expect(events[2].id).toBe(3);
expect(eventLog.publish).toHaveBeenCalledTimes(1);
});
it('should use independent ID sequences per thread', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
bus.publish('thread-2', makeEvent('c', 'run_2'));
it('stamps ts once and leaves an already-stamped event alone', () => {
bus.publish('thread-1', { ...makeEvent('a', 'run_1'), ts: 42 });
const events1 = bus.getEventsAfter('thread-1', 0);
const events2 = bus.getEventsAfter('thread-2', 0);
expect(events1).toHaveLength(2);
expect(events1[0].id).toBe(1);
expect(events1[1].id).toBe(2);
expect(events2).toHaveLength(1);
expect(events2[0].id).toBe(1);
});
it('should not touch Redis', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
expect(seqByKey.size).toBe(0);
expect(eventLog.publish.mock.calls[0][1]).toEqual(expect.objectContaining({ ts: 42 }));
});
});
describe('publish (multi-main, shared sequence)', () => {
beforeEach(() => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
describe('drained events', () => {
it('emits a durable fact to subscribers with its DB seq', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
publishAndCaptureEmit('thread-1', event)({ id: 7, event, live: true });
expect(received).toEqual([{ id: 7, event }]);
});
it('assigns ids from the shared Redis sequence in publish order', async () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
bus.publish('thread-1', makeEvent('c', 'run_1'));
await flushDrain();
it('emits ephemeral events without an id, so the replay cursor skips them', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
const events = bus.getEventsAfter('thread-1', 0);
expect(events.map((e) => e.id)).toEqual([1, 2, 3]);
expect(events.map((e) => e.event.payload)).toEqual([
{ text: 'a-run_1' },
{ text: 'b-run_1' },
{ text: 'c-run_1' },
]);
publishAndCaptureEmit('thread-1', event)({ event, live: true });
expect(received).toEqual([{ event }]);
expect(received[0]).not.toHaveProperty('id');
});
it('sequences events queued during a Redis round trip as one INCRBY batch', async () => {
bus.publish('thread-1', makeEvent('a', 'run_1')); // drains alone
bus.publish('thread-1', makeEvent('b', 'run_1')); // queued during the round trip
bus.publish('thread-1', makeEvent('c', 'run_1')); // queued during the round trip
await flushDrain();
it('drops a coalesced block entirely — subscribers already saw its deltas', () => {
const received: unknown[] = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
expect(incrbyCalls.map((c) => c.count)).toEqual([1, 2]);
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([1, 2, 3]);
});
publishAndCaptureEmit('thread-1', event)({ id: 3, event, live: false });
it('continues the sequence started by another main', async () => {
const otherMain = buildBus();
otherMain.publish('thread-1', makeEvent('a', 'run_1'));
otherMain.publish('thread-1', makeEvent('b', 'run_1'));
await flushDrain();
bus.publish('thread-1', makeEvent('c', 'run_1'));
await flushDrain();
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([3]);
});
it('falls back to local ids above the high-water mark when Redis fails', async () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
await flushDrain();
redisFailure = new Error('connection lost');
bus.publish('thread-1', makeEvent('c', 'run_1'));
await flushDrain();
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([1, 2, 3]);
});
it('keeps fallback ids above ids observed from relayed events', async () => {
redisFailure = new Error('connection lost');
// A sibling produced up to id 7 — observed via relay without a subscriber.
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 7, event: makeEvent('x', 'run_1') },
});
bus.publish('thread-1', makeEvent('a', 'run_1'));
await flushDrain();
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([8]);
expect(received).toHaveLength(0);
expect(publisher.publishCommand).not.toHaveBeenCalled();
});
});
describe('subscribe', () => {
it('should receive events published after subscription', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
expect(received).toHaveLength(2);
expect(received[0].id).toBe(1);
expect(received[1].id).toBe(2);
});
it('should not receive events from other threads', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
bus.publish('thread-2', makeEvent('a', 'run_2'));
expect(received).toHaveLength(0);
});
it('should stop delivery after unsubscribe', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
it('delivers only the subscribed thread, and stops after unsubscribe', () => {
const received: unknown[] = [];
const unsubscribe = bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
bus.publish('thread-1', makeEvent('a', 'run_1'));
publishAndCaptureEmit('thread-2', event)({ id: 1, event, live: true });
expect(received).toHaveLength(0);
const emit = publishAndCaptureEmit('thread-1', event);
emit({ id: 1, event, live: true });
expect(received).toHaveLength(1);
unsubscribe();
bus.publish('thread-1', makeEvent('b', 'run_1'));
emit({ id: 2, event, live: true });
expect(received).toHaveLength(1);
});
});
describe('getEventsAfter', () => {
it('should return all events when afterId is 0', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
bus.publish('thread-1', makeEvent('c', 'run_1'));
const events = bus.getEventsAfter('thread-1', 0);
expect(events).toHaveLength(3);
});
it('should skip events with id <= afterId', () => {
for (let i = 0; i < 7; i++) {
bus.publish('thread-1', makeEvent(`e${i}`, 'run_1'));
}
const events = bus.getEventsAfter('thread-1', 5);
expect(events).toHaveLength(2);
expect(events[0].id).toBe(6);
expect(events[1].id).toBe(7);
});
it('should return empty array for unknown thread', () => {
const events = bus.getEventsAfter('nonexistent', 0);
expect(events).toEqual([]);
});
it('should return empty array when all events are before cursor', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
const events = bus.getEventsAfter('thread-1', 10);
expect(events).toEqual([]);
});
});
describe('getNextEventId', () => {
it('should return 1 for a new thread', async () => {
await expect(bus.getNextEventId('thread-1')).resolves.toBe(1);
});
it('should return the next sequential ID after publishing', async () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_1'));
await expect(bus.getNextEventId('thread-1')).resolves.toBe(3);
});
it('reads the shared sequence in multi-main, so any main returns the same cursor', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
const otherMain = buildBus();
otherMain.publish('thread-1', makeEvent('a', 'run_1'));
otherMain.publish('thread-1', makeEvent('b', 'run_1'));
await flushDrain();
// This main never buffered the thread, but agrees on the next id.
await expect(bus.getNextEventId('thread-1')).resolves.toBe(3);
});
it('falls back to the local high-water mark when Redis fails', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
bus.publish('thread-1', makeEvent('a', 'run_1'));
await flushDrain();
redisFailure = new Error('connection lost');
await expect(bus.getNextEventId('thread-1')).resolves.toBe(2);
});
});
describe('getEventsForRun', () => {
it('should return only events matching the given runId', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.publish('thread-1', makeEvent('b', 'run_2'));
bus.publish('thread-1', makeEvent('c', 'run_1'));
bus.publish('thread-1', makeEvent('d', 'run_2'));
const run1Events = bus.getEventsForRun('thread-1', 'run_1');
expect(run1Events).toHaveLength(2);
expect(run1Events.every((e) => e.runId === 'run_1')).toBe(true);
const run2Events = bus.getEventsForRun('thread-1', 'run_2');
expect(run2Events).toHaveLength(2);
expect(run2Events.every((e) => e.runId === 'run_2')).toBe(true);
});
it('should return empty array for unknown thread', () => {
expect(bus.getEventsForRun('nonexistent', 'run_1')).toEqual([]);
});
it('should return empty array when no events match the runId', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
expect(bus.getEventsForRun('thread-1', 'run_99')).toEqual([]);
});
it('should return unwrapped InstanceAiEvent objects (not StoredEvent)', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
const events = bus.getEventsForRun('thread-1', 'run_1');
expect(events[0]).not.toHaveProperty('id'); // No StoredEvent wrapper
expect(events[0]).toHaveProperty('type');
expect(events[0]).toHaveProperty('runId');
expect(events[0]).toHaveProperty('agentId');
});
it('includes events still awaiting a sequence number (same-main read-your-writes)', () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
bus.publish('thread-1', makeEvent('a', 'run_1'));
// No drain flush: the event has no id yet, but same-main callers
// (terminal outcomes, tracing, snapshots) must still see it.
expect(bus.getEventsForRun('thread-1', 'run_1')).toHaveLength(1);
expect(bus.getEventsAfter('thread-1', 0)).toHaveLength(0);
});
});
describe('clear', () => {
it('should remove all stored events and listeners', async () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
bus.publish('thread-1', makeEvent('a', 'run_1'));
expect(received).toHaveLength(1);
bus.clear();
// Events cleared
expect(bus.getEventsAfter('thread-1', 0)).toEqual([]);
await expect(bus.getNextEventId('thread-1')).resolves.toBe(1);
// Listener removed — new publish should not reach old handler
bus.publish('thread-1', makeEvent('b', 'run_1'));
expect(received).toHaveLength(1);
});
});
describe('clearThread', () => {
it('deletes the shared sequence key in multi-main', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
bus.publish('thread-1', makeEvent('a', 'run_1'));
await flushDrain();
expect(seqByKey.size).toBe(1);
bus.clearThread('thread-1');
await flushDrain();
expect(deletedKeys).toEqual(['n8n:instance-ai:event-seq:thread-1']);
expect(bus.getEventsAfter('thread-1', 0)).toEqual([]);
});
it('does not touch Redis in single-main', async () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
bus.clearThread('thread-1');
await flushDrain();
expect(deletedKeys).toEqual([]);
});
});
describe('cross-main relay', () => {
it('does not relay when single-main', () => {
bus.publish('thread-1', makeEvent('a', 'run_1'));
expect(publisher.publishCommand).not.toHaveBeenCalled();
});
it('relays each event with its producer-assigned id when multi-main', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
const event = makeEvent('a', 'run_1');
bus.publish('thread-1', event);
await flushDrain();
expect(publisher.publishCommand).toHaveBeenCalledWith({
command: 'relay-instance-ai-event',
payload: {
threadId: 'thread-1',
// publish() stamps the publish time onto the event
storedEvent: { id: 1, event: { ...event, ts: expect.any(Number) } },
},
});
});
it('still delivers locally even when relaying', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
const received: number[] = [];
bus.subscribe('thread-1', (e) => received.push(e.id!));
bus.publish('thread-1', makeEvent('a', 'run_1'));
await flushDrain();
expect(received).toEqual([1]);
expect(publisher.publishCommand).toHaveBeenCalledTimes(1);
});
it('skips relay for oversized events but still delivers locally', async () => {
instanceSettings = { isMultiMain: true };
bus = buildBus();
const received: number[] = [];
bus.subscribe('thread-1', (e) => received.push(e.id!));
const huge = makeEvent('a', 'run_1');
(huge.payload as { text: string }).text = 'x'.repeat(6 * 1024 * 1024);
bus.publish('thread-1', huge);
await flushDrain();
// Relay skipped (would bloat pubsub), but the local SSE client still got it
// via the emit (even though the 2 MB store cap then evicts it).
expect(publisher.publishCommand).not.toHaveBeenCalled();
expect(received).toEqual([1]);
});
});
describe('handleRelayInstanceAiEvent', () => {
it('stores and re-emits a relayed event under its producer-assigned id', () => {
const received: number[] = [];
bus.subscribe('thread-1', (e) => received.push(e.id!));
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 42, event: makeEvent('a', 'run_1') },
});
expect(received).toEqual([42]);
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([42]);
// Re-emit must not re-relay (loop guard).
expect(publisher.publishCommand).not.toHaveBeenCalled();
});
it('ignores a relayed event when this main has no subscriber for the thread', () => {
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 1, event: makeEvent('a', 'run_1') },
});
// Nothing stored, since the thread has no local consumer here.
expect(bus.getEventsAfter('thread-1', 0)).toHaveLength(0);
});
it('keeps the store sorted when a concurrent producer relays a lower id', () => {
bus.subscribe('thread-1', () => {});
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 5, event: makeEvent('later', 'run_1') },
});
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 3, event: makeEvent('earlier', 'run_2') },
});
expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([3, 5]);
});
it('drops a duplicate id instead of storing or emitting it twice', () => {
const received: number[] = [];
bus.subscribe('thread-1', (e) => received.push(e.id!));
const storedEvent = { id: 5, event: makeEvent('a', 'run_1') };
bus.handleRelayInstanceAiEvent({ threadId: 'thread-1', storedEvent });
bus.handleRelayInstanceAiEvent({ threadId: 'thread-1', storedEvent });
expect(received).toEqual([5]);
expect(bus.getEventsAfter('thread-1', 0)).toHaveLength(1);
});
});
describe('hasSubscribers', () => {
it('reflects active subscriptions', () => {
it('reports whether this main holds a subscription', () => {
expect(bus.hasSubscribers('thread-1')).toBe(false);
const unsubscribe = bus.subscribe('thread-1', () => {});
expect(bus.hasSubscribers('thread-1')).toBe(true);
@@ -531,126 +125,54 @@ describe('InProcessEventBus', () => {
});
});
describe('durable log (flag on)', () => {
type EmitFn = (drained: { id?: number; event: InstanceAiEvent; live: boolean }) => void;
/** Route publish through the mocked drain and hand back its emit callback. */
function publishAndCaptureEmit(threadId: string, event: InstanceAiEvent): EmitFn {
bus.publish(threadId, event);
const call = eventLog.publish.mock.calls.at(-1)!;
expect(call[0]).toBe(threadId);
// publish() stamps `ts` onto a copy before handing it to the log
expect(call[1]).toEqual({ ...event, ts: expect.any(Number) });
return call[2] as EmitFn;
}
beforeEach(() => {
bus = buildBus({ durableLog: true });
});
it('routes publishes into the durable log instead of the Redis sequence', () => {
instanceSettings.isMultiMain = true;
bus = buildBus({ durableLog: true });
bus.publish('thread-1', makeEvent('a', 'run_1'));
expect(eventLog.publish).toHaveBeenCalledTimes(1);
expect(incrbyCalls).toHaveLength(0);
});
it('emits a drained durable fact with its DB seq and retains nothing', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
describe('cross-main relay', () => {
it('does not relay when single-main', () => {
const event = makeEvent('a', 'run_1');
const emit = publishAndCaptureEmit('thread-1', event);
emit({ id: 7, event, live: true });
publishAndCaptureEmit('thread-1', event)({ id: 1, event, live: true });
expect(received).toEqual([{ id: 7, event }]);
expect(bus.getEventsForRun('thread-1', 'run_1')).toHaveLength(0);
});
it('emits ephemeral events live without an id', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
const emit = publishAndCaptureEmit('thread-1', event);
emit({ event, live: true });
expect(received).toEqual([{ event }]);
expect(received[0]).not.toHaveProperty('id');
});
it('drops a coalesced block entirely (subscribers saw its deltas, the DB has the row)', () => {
const received: unknown[] = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
const emit = publishAndCaptureEmit('thread-1', event);
emit({ id: 3, event, live: false });
expect(received).toHaveLength(0);
expect(publisher.publishCommand).not.toHaveBeenCalled();
expect(bus.getEventsForRun('thread-1', 'run_1')).toHaveLength(0);
});
it('retains no per-thread state across a burst of drained facts', () => {
for (let thread = 0; thread < 50; thread++) {
const threadId = `thread-${thread}`;
const event = makeEvent('a', 'run_1');
const emit = publishAndCaptureEmit(threadId, event);
for (let seq = 1; seq <= 20; seq++) emit({ id: seq, event, live: true });
}
// Nothing to release on run completion, thread deletion or the TTL prune,
// because nothing was retained in the first place.
expect(bus.retainedThreadCount()).toBe(0);
});
it('reports a store read as a wiring bug, once per entry point', () => {
expect(bus.getEventsAfter('thread-1', 0)).toEqual([]);
expect(bus.getEventsAfter('thread-1', 0)).toEqual([]);
// Names the entry point the caller used, not the method it delegates to.
expect(bus.getEventsForRun('thread-1', 'run_1')).toEqual([]);
expect(bus.getEventsForRuns('thread-1', ['run_1'])).toEqual([]);
expect(logger.error.mock.calls).toEqual([
[expect.stringContaining('getEventsAfter'), { threadId: 'thread-1' }],
[expect.stringContaining('getEventsForRun '), { threadId: 'thread-1' }],
[expect.stringContaining('getEventsForRuns'), { threadId: 'thread-1' }],
]);
});
it('answers the next event id from the durable log, never from memory', async () => {
eventLog.getNextEventId.mockResolvedValue(42);
await expect(bus.getNextEventId('thread-1')).resolves.toBe(42);
expect(eventLog.getNextEventId).toHaveBeenCalledWith('thread-1');
});
it('relays live drained events to siblings with the DB seq passed through', () => {
it('relays live events, passing the DB seq through and omitting it when ephemeral', () => {
instanceSettings.isMultiMain = true;
bus = buildBus({ durableLog: true });
bus = buildBus();
const event = makeEvent('a', 'run_1');
const emit = publishAndCaptureEmit('thread-1', event);
emit({ id: 9, event, live: true });
emit({ event, live: true });
expect(publisher.publishCommand).toHaveBeenCalledTimes(2);
expect(publisher.publishCommand).toHaveBeenNthCalledWith(1, {
command: 'relay-instance-ai-event',
payload: { threadId: 'thread-1', storedEvent: { id: 9, event } },
});
// The ephemeral relay frame carries no id.
expect(publisher.publishCommand).toHaveBeenNthCalledWith(2, {
command: 'relay-instance-ai-event',
payload: { threadId: 'thread-1', storedEvent: { event } },
});
});
it('re-emits a relayed frame to subscribers without storing it, id-bearing or not', () => {
it('skips relay for an oversized event but still delivers it locally', () => {
instanceSettings.isMultiMain = true;
bus = buildBus();
const received: unknown[] = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('x'.repeat(6 * 1024 * 1024), 'run_1'); // > MAX_PUBSUB_PAYLOAD_BYTES
publishAndCaptureEmit('thread-1', event)({ id: 1, event, live: true });
expect(publisher.publishCommand).not.toHaveBeenCalled();
expect(received).toHaveLength(1);
expect(logger.warn).toHaveBeenCalledWith(
expect.stringContaining('Skipping cross-main relay'),
expect.objectContaining({ threadId: 'thread-1' }),
);
});
});
describe('handleRelayInstanceAiEvent', () => {
it('re-emits a relayed frame to subscribers, id-bearing or not', () => {
const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
const event = makeEvent('a', 'run_1');
@@ -659,24 +181,38 @@ describe('InProcessEventBus', () => {
bus.handleRelayInstanceAiEvent({ threadId: 'thread-1', storedEvent: { id: 4, event } });
expect(received).toEqual([{ event }, { id: 4, event }]);
expect(bus.retainedThreadCount()).toBe(0);
});
it('drops a relayed frame for a thread this main has no subscriber for', () => {
bus.handleRelayInstanceAiEvent({
threadId: 'thread-1',
storedEvent: { id: 4, event: makeEvent('a', 'run_1') },
});
const event = makeEvent('a', 'run_1');
// Nothing to assert beyond "does not throw and delivers nowhere": replay
// is served from the log, so an unsubscribed main needs no copy.
expect(() =>
bus.handleRelayInstanceAiEvent({ threadId: 'thread-1', storedEvent: { id: 4, event } }),
).not.toThrow();
});
});
expect(bus.retainedThreadCount()).toBe(0);
describe('teardown', () => {
it('clearThread drops the thread subscription and the log drain state', () => {
const received: unknown[] = [];
bus.subscribe('thread-1', (stored) => received.push(stored));
bus.clearThread('thread-1');
expect(eventLog.clearThread).toHaveBeenCalledWith('thread-1');
expect(bus.hasSubscribers('thread-1')).toBe(false);
});
it('clearThread and clear drop the durable log drain state too', () => {
bus.clearThread('thread-1');
expect(eventLog.clearThread).toHaveBeenCalledWith('thread-1');
it('clear drops every subscription and the log state', () => {
bus.subscribe('thread-1', () => {});
bus.subscribe('thread-2', () => {});
bus.clear();
expect(eventLog.clear).toHaveBeenCalledTimes(1);
expect(bus.hasSubscribers('thread-1')).toBe(false);
expect(bus.hasSubscribers('thread-2')).toBe(false);
});
});
});
packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts
-26
View File
@@ -1,6 +1,5 @@
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';
@@ -87,7 +86,6 @@ interface Setup {
checkpoints?: InstanceAiCheckpoint[];
isMultiMain?: boolean;
lastFactAt?: Date | null;
durableLog?: boolean;
host?: Partial<InterruptedRunResumeHost>;
}
@@ -142,7 +140,6 @@ function buildSweeper(setup: Setup) {
checkpointRepo,
eventBus as never,
metrics,
{ instanceAi: { durableLog: setup.durableLog ?? true } } as GlobalConfig,
{ isMultiMain: setup.isMultiMain ?? false } as InstanceSettings,
);
sweeper.setResumeHost(host);
@@ -194,18 +191,6 @@ describe('InterruptedRunSweeper', () => {
);
});
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')],
@@ -410,17 +395,6 @@ describe('InterruptedRunSweeper.cancelUnfinishedRuns', () => {
expect(activeSibling.published).toHaveLength(0);
});
it('does nothing when the durable log is off', async () => {
const { sweeper, published, eventLogRepo } = buildSweeper({
events: [runStart()],
durableLog: false,
});
expect(await sweeper.cancelUnfinishedRuns(THREAD)).toBe(0);
expect(eventLogRepo.findUnfinishedRuns).not.toHaveBeenCalled();
expect(published).toHaveLength(0);
});
it('never appends a second terminal fact when one landed mid-race', async () => {
const { sweeper, published, eventLogRepo } = buildSweeper({
events: [
packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts
+2 -5
View File
@@ -508,9 +508,8 @@ export class DurableEventLog {
* Append `events` with contiguous seqs, retrying on (threadId, seq) PK
* collision — another main won the range (multi-main only), so re-seed from
* the DB and try again. Returns the first assigned seq, or undefined when
* the batch had to be dropped (logged; live delivery still happens).
* INS-844 merges the shared-sequence drain (#33558) here: its Redis INCRBY
* becomes this batch-INSERT's id assignment, one round trip.
* the batch had to be dropped (logged; live delivery still happens). The
* batch INSERT is also what assigns the ids, so it is one round trip.
*/
private async persistWithRetry(
threadId: string,
@@ -726,8 +725,6 @@ export class DurableEventLog {
const cached = this.lastSeq.get(threadId);
if (cached !== undefined) return cached;
const max = await this.repo.maxSeq(threadId);
// Cutover note (RFC Q&A on cursors): INS-844 seeds from max(DB, Redis
// high-water mark) so cursors minted by the live shared sequence stay valid.
this.lastSeq.set(threadId, max);
return max;
}
packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts
+26 -409
View File
@@ -1,152 +1,59 @@
import { Logger } from '@n8n/backend-common';
import type { InstanceAiEvent } from '@n8n/api-types';
import { GlobalConfig } from '@n8n/config';
import { Logger } from '@n8n/backend-common';
import { OnPubSubEvent } from '@n8n/decorators';
import { Service } from '@n8n/di';
import type { InstanceAiEventBus, StoredEvent } from '@n8n/instance-ai';
import { EventEmitter } from 'node:events';
import { InstanceSettings } from 'n8n-core';
import { EventEmitter } from 'node:events';
import { MAX_PUBSUB_PAYLOAD_BYTES } from '@/scaling/constants';
import { Publisher } from '@/scaling/pubsub/publisher.service';
import { DurableEventLog, type DrainedEvent } from './durable-event-log';
// The store below is FLAG-OFF ONLY, and these caps bound it per thread. It has
// no cap on thread COUNT and is released only when a thread is cleared, so with
// the flag off a long-lived process retains up to 2MB per thread it has served.
// Accepted for the rollback switch rather than fixed there: a global eviction
// policy on the only store is data loss (the empty-agentTree bug class), and
// the path sunsets with the flag at Gate B.
//
// With the durable log ON nothing is stored here at all: instance_ai_events is
// the source of truth, every read goes through DurableEventLog, and this bus is
// a pure fan-out (local SSE subscribers + the cross-main relay).
const MAX_EVENTS_PER_THREAD = 500;
const MAX_BYTES_PER_THREAD = 2 * 1024 * 1024; // 2 MB
/**
* How long an idle thread's shared sequence key lives in Redis (refreshed on
* every assignment). Generous on purpose: if it ever expires and the sequence
* restarts at 1, clients holding stale high cursors get an empty replay
* (recovered via run-sync / hydration), and fresh page loads re-seed their
* cursor from `GET /messages` anyway.
* Live fan-out for Instance AI events. `instance_ai_events` is the source of
* truth: this bus persists nothing and reads nothing back. Every replay and
* run-scoped read goes to {@link DurableEventLog}, so all this owns is the
* local SSE emitter and the cross-main relay.
*/
const SEQ_KEY_TTL_SECONDS = 14 * 24 * 60 * 60;
/** Only id-bearing events enter the store; the id is the replay cursor. */
type SequencedEvent = StoredEvent & { id: number };
@Service()
export class InProcessEventBus implements InstanceAiEventBus {
private readonly emitter = new EventEmitter();
private readonly store = new Map<string, SequencedEvent[]>();
/** Approximate serialized size per thread for eviction. */
private readonly sizeBytes = new Map<string, number>();
/**
* Highest event id this main has assigned or observed per thread. The id
* source in single-main, and the fallback when Redis is unavailable in
* multi-main (kept bumped from relayed events so fallback ids stay above
* what siblings have already used).
*/
private readonly lastLocalId = new Map<string, number>();
/**
* Events awaiting a sequence number (multi-main only). `publish()` stays
* synchronous by enqueueing here; a single per-thread drain assigns ids.
*/
private readonly pendingByThread = new Map<string, InstanceAiEvent[]>();
/**
* The batch currently being sequenced (multi-main only): taken off the
* pending queue but not yet in the store. Kept visible so run-scoped reads
* see events across the Redis round trip.
*/
private readonly inFlightByThread = new Map<string, InstanceAiEvent[]>();
private readonly drainingThreads = new Set<string>();
private readonly seqKeyPrefix: string;
private readonly durableLogEnabled: boolean;
/** Store-read methods already reported under the durable log (log once). */
private readonly warnedStoreReads = new Set<string>();
constructor(
private readonly logger: Logger,
private readonly instanceSettings: InstanceSettings,
private readonly publisher: Publisher,
private readonly eventLog: DurableEventLog,
globalConfig: GlobalConfig,
) {
this.logger = this.logger.scoped('instance-ai');
this.seqKeyPrefix = `${globalConfig.redis.prefix}:instance-ai:event-seq:`;
this.durableLogEnabled = globalConfig.instanceAi.durableLog;
// Avoid warnings when many SSE clients connect (each adds a listener per thread)
this.emitter.setMaxListeners(0);
}
/**
* Publish an event for a thread.
* Publish an event for a thread: a synchronous enqueue into the durable
* log's per-thread drain, which assigns `seq` from the DB, persists durable
* facts, and hands each event back here ({@link onDrained}) for fan-out.
*
* Durable log ON: synchronous enqueue into the durable log's per-thread
* drain, which assigns `seq` from the DB, persists durable facts, and hands
* each event back here (`onDrained`) for fan-out only — live ones go to
* local SSE subscribers and — in multi-main — to sibling mains via the
* pubsub relay. Nothing is retained in this process. Ephemeral events
* (deltas, status) carry NO id, so their SSE frames have no `id:` line and
* the browser's replay cursor only ever points at durable facts. The Redis
* sequence machinery below is never touched: the flag picks exactly one drain
* (INS-844's composition was cancelled), and the flag-off paths below survive
* only as the rollback switch until they sunset at Gate B (INS-847).
*
* Flag OFF, single-main: assign the next local id and deliver in the same tick.
*
* Flag OFF, multi-main: enqueue and drain asynchronously — event ids come
* from a shared per-thread Redis sequence, so every main agrees on them and
* the frontend's replay cursor is valid against any main. The queue
* preserves publish order; each sequenced event is stored, delivered to
* local SSE subscribers, and relayed to sibling mains with its id.
* Ephemeral events (deltas, status) carry NO id, so their SSE frames have no
* `id:` line and the browser's replay cursor only ever points at durable
* facts.
*/
publish(threadId: string, event: InstanceAiEvent): void {
// Stamp publish time once — replays (SSE reconnect, snapshot rebuilds)
// rely on it to reconstruct real timing instead of processing time.
// Before the durable-log branch on purpose: persisted events must carry
// it too.
// rely on it to reconstruct real timing instead of processing time, and
// persisted events must carry it too.
if (event.ts === undefined) {
event = { ...event, ts: Date.now() };
}
if (this.durableLogEnabled) {
this.eventLog.publish(threadId, event, (drained) => this.onDrained(threadId, drained));
return;
}
if (!this.instanceSettings.isMultiMain) {
const id = (this.lastLocalId.get(threadId) ?? 0) + 1;
this.lastLocalId.set(threadId, id);
this.storeAndEmit(threadId, { id, event });
return;
}
const pending = this.pendingByThread.get(threadId);
if (pending) {
pending.push(event);
} else {
this.pendingByThread.set(threadId, [event]);
}
void this.drainQueue(threadId);
this.eventLog.publish(threadId, event, (drained) => this.onDrained(threadId, drained));
}
/**
* An event handed back by the durable log's drain (flag on): fan it out to
* local SSE subscribers and sibling mains. Nothing is retained — the row is
* already in instance_ai_events, and every replay/run-scoped read goes to
* the log, so keeping a copy here would only pin memory per thread for the
* process lifetime. Coalesced blocks are durable but NOT live
* An event handed back by the durable log's drain: fan it out to local SSE
* subscribers and sibling mains. Coalesced blocks are durable but NOT live
* (subscribers already saw their deltas), so they fan out to nobody.
*/
private onDrained(threadId: string, drained: DrainedEvent): void {
@@ -163,142 +70,6 @@ export class InProcessEventBus implements InstanceAiEventBus {
);
}
/**
* Assign sequence ids to queued events and dispatch them, preserving
* publish order. Only one drain runs per thread; events queued while a
* Redis round trip is in flight are picked up by the next loop iteration
* and sequenced as one batch (single sequence round trip).
*/
private async drainQueue(threadId: string): Promise<void> {
if (this.drainingThreads.has(threadId)) return;
this.drainingThreads.add(threadId);
try {
let batch = this.takePending(threadId);
while (batch.length > 0) {
this.inFlightByThread.set(threadId, batch);
// Never throws — falls back to local ids on Redis failure.
const firstId = await this.assignSequenceBlock(threadId, batch.length);
for (let i = 0; i < batch.length; i++) {
const stored: SequencedEvent = { id: firstId + i, event: batch[i] };
// Serialize once: reused for the store's size accounting and the relay guard.
const sizeBytes = Buffer.byteLength(JSON.stringify(batch[i]), 'utf8');
this.storeAndEmit(threadId, stored, sizeBytes);
this.relayToSiblings(threadId, stored, sizeBytes);
}
this.inFlightByThread.delete(threadId);
batch = this.takePending(threadId);
}
} finally {
this.inFlightByThread.delete(threadId);
this.drainingThreads.delete(threadId);
}
}
private takePending(threadId: string): InstanceAiEvent[] {
const pending = this.pendingByThread.get(threadId);
if (!pending) return [];
this.pendingByThread.delete(threadId);
return pending;
}
/**
* Reserve a contiguous block of `count` ids from the shared per-thread
* sequence (atomic INCRBY). On Redis failure, continue monotonically from the
* local high-water mark — ids stay usable for this main's connections, at the
* cost of possible overlap with siblings until Redis recovers.
*
* Accepted degradation: after a Redis outage the shared counter can briefly
* sit below this main's local high-water mark (the fallback advanced local ids
* that never reached Redis), so INCRBY on recovery may re-issue an id already
* in this main's store — `insertById` then drops it as a duplicate, i.e. a few
* events can be lost from the live stream during recovery. Not worth an atomic
* conditional-max (Lua/WATCH) here: it only bites during a Redis incident, and
* the persisted run snapshot reconciles the tree via `run-sync` on reconnect.
* (A single-main→multi-main flip mid-thread would collide the same way, but a
* thread only starts producing events once the license has settled isMultiMain
* at boot, so that path isn't reached in practice.)
*/
private async assignSequenceBlock(threadId: string, count: number): Promise<number> {
try {
const key = this.seqKey(threadId);
const results = await this.getRedisClient()
.multi()
.incrby(key, count)
.expire(key, SEQ_KEY_TTL_SECONDS)
.exec();
const [incrError, incrResult] = results?.[0] ?? [new Error('empty transaction result'), null];
if (incrError) throw incrError;
const endId = Number(incrResult);
if (!Number.isFinite(endId)) {
throw new Error(`non-numeric INCRBY result: ${String(incrResult)}`);
}
this.bumpLocalHighWaterMark(threadId, endId);
return endId - count + 1;
} catch (error) {
this.logger.error(
'Failed to assign Instance AI event sequence from Redis, falling back to local ids',
{ threadId, error },
);
const firstId = (this.lastLocalId.get(threadId) ?? 0) + 1;
this.lastLocalId.set(threadId, firstId + count - 1);
return firstId;
}
}
/**
* The shared sequence lives on the pubsub publisher's Redis client. Only ever
* reached in multi-main, which implies queue mode — where the publisher's
* client is initialized. Reusing it avoids a second persistent connection per
* main. Publishing never puts a client in subscriber mode, so running
* sequence commands on it is safe.
*/
private getRedisClient() {
return this.publisher.getClient();
}
private seqKey(threadId: string): string {
return `${this.seqKeyPrefix}${threadId}`;
}
private bumpLocalHighWaterMark(threadId: string, id: number): void {
if (id > (this.lastLocalId.get(threadId) ?? 0)) {
this.lastLocalId.set(threadId, id);
}
}
private storeAndEmit(threadId: string, stored: SequencedEvent, eventSizeBytes?: number): void {
const size = eventSizeBytes ?? Buffer.byteLength(JSON.stringify(stored.event), 'utf8');
const events = this.getOrCreateStore(threadId);
// Duplicate id (e.g. an event relayed twice): already stored and emitted.
if (!this.insertById(events, stored)) return;
this.sizeBytes.set(threadId, (this.sizeBytes.get(threadId) ?? 0) + size);
// Evict oldest events if count or size exceeds caps
this.evictIfNeeded(threadId, events);
this.emitter.emit(threadId, stored);
}
/**
* Insert keeping the store sorted by id. Local events always append, but a
* relayed event from a concurrent producer on another main (e.g. a
* background task while the orchestrator runs elsewhere) can arrive with a
* lower id than the latest stored one. Returns false for a duplicate id.
*/
private insertById(events: SequencedEvent[], stored: SequencedEvent): boolean {
if (events.length === 0 || events[events.length - 1].id < stored.id) {
events.push(stored);
return true;
}
let i = events.length - 1;
while (i >= 0 && events[i].id > stored.id) i--;
if (i >= 0 && events[i].id === stored.id) return false;
events.splice(i + 1, 0, stored);
return true;
}
private relayToSiblings(threadId: string, stored: StoredEvent, sizeBytes: number): void {
if (!this.instanceSettings.isMultiMain) return;
@@ -323,35 +94,18 @@ export class InProcessEventBus implements InstanceAiEventBus {
);
}
/** A relayed event from another main, carrying its producer-assigned id
* from the shared sequence (or the DB-assigned seq with the durable log
* on; id-less = ephemeral, live-only). Stored/re-emitted only if this main
* holds a subscription for the thread (avoids every main buffering every
* thread). */
/**
* A relayed event from another main, carrying its DB-assigned seq (id-less =
* ephemeral, live-only). Pure live delivery: reconnect replay reads the log,
* so a relayed frame is only ever needed by a subscriber attached right now,
* and a frame delivered twice is dropped client-side by its id.
*/
@OnPubSubEvent('relay-instance-ai-event', { instanceType: 'main' })
handleRelayInstanceAiEvent({
threadId,
storedEvent,
}: { threadId: string; storedEvent: StoredEvent }): void {
if (this.durableLogEnabled) {
// Pure live delivery: seqs come from the DB (no local high-water mark to
// track) and reconnect replay reads the log, so a relayed frame is only
// ever needed by a subscriber attached right now. A frame delivered
// twice is dropped client-side by its id.
if (this.hasSubscribers(threadId)) this.emitter.emit(threadId, storedEvent);
return;
}
// Track the shared-sequence high-water mark even without subscribers, so
// a Redis-outage fallback keeps assigning ids above what siblings used.
if (storedEvent.id !== undefined) {
this.bumpLocalHighWaterMark(threadId, storedEvent.id);
}
if (!this.hasSubscribers(threadId)) return;
if (storedEvent.id === undefined) {
this.emitter.emit(threadId, storedEvent);
return;
}
this.storeAndEmit(threadId, { id: storedEvent.id, event: storedEvent.event });
if (this.hasSubscribers(threadId)) this.emitter.emit(threadId, storedEvent);
}
subscribe(threadId: string, handler: (storedEvent: StoredEvent) => void): () => void {
@@ -364,152 +118,15 @@ export class InProcessEventBus implements InstanceAiEventBus {
return this.emitter.listenerCount(threadId) > 0;
}
/**
* Threads with events retained in this process. Always 0 under the durable
* log; with the flag off it grows with every thread this main has served
* until each is cleared.
*/
retainedThreadCount(): number {
return this.store.size;
}
/**
* FLAG-OFF ONLY. Events still awaiting a sequence number are intentionally
* excluded: they have no id yet, and once sequenced they reach subscribers
* live — the SSE bootstrap subscribes before calling this, so nothing is
* missed. With the durable log on, use DurableEventLog.getEventsAfter.
*/
getEventsAfter(threadId: string, afterId: number): StoredEvent[] {
if (this.assertNoStoreUnderDurableLog('getEventsAfter', threadId)) return [];
const events = this.store.get(threadId);
if (!events) return [];
return events.filter((e) => e.id > afterId);
}
/** FLAG-OFF ONLY — see {@link getEventsForRuns}. */
getEventsForRun(threadId: string, runId: string): InstanceAiEvent[] {
// Guarded before delegating so the report names the entry point the
// caller actually used, and so the two entry points dedupe separately.
if (this.assertNoStoreUnderDurableLog('getEventsForRun', threadId)) return [];
return this.getEventsForRuns(threadId, [runId]);
}
/**
* FLAG-OFF ONLY. With the durable log on nothing is stored here, so a caller
* must go through DurableEventLog.getEventsForRuns (via the service's
* `readRunEvents`, which flushes open coalesce buffers first).
*/
getEventsForRuns(threadId: string, runIds: string[]): InstanceAiEvent[] {
if (this.assertNoStoreUnderDurableLog('getEventsForRuns', threadId)) return [];
if (runIds.length === 0) return [];
const runIdSet = new Set(runIds);
const stored = (this.store.get(threadId) ?? [])
.filter((e) => runIdSet.has(e.event.runId))
.map((e) => e.event);
// Include events still awaiting a sequence number (both the batch being
// sequenced and the queue behind it) so same-main callers (terminal
// outcomes, tracing, snapshots) read their own writes. A run's events are
// produced on one main, so unsequenced ones are always newest.
const unsequenced = [
...(this.inFlightByThread.get(threadId) ?? []),
...(this.pendingByThread.get(threadId) ?? []),
].filter((e) => runIdSet.has(e.runId));
return [...stored, ...unsequenced];
}
async getNextEventId(threadId: string): Promise<number> {
// Delegated rather than guarded: unlike the store-scoped reads above this
// one is cheap to answer correctly, and the SSE cursor it seeds must never
// silently come back as 1.
if (this.durableLogEnabled) return await this.eventLog.getNextEventId(threadId);
if (this.instanceSettings.isMultiMain) {
try {
const value = await this.getRedisClient().get(this.seqKey(threadId));
if (value !== null) return Number(value) + 1;
} catch (error) {
this.logger.warn(
'Failed to read Instance AI event sequence from Redis, falling back to local high-water mark',
{ threadId, error },
);
}
}
return (this.lastLocalId.get(threadId) ?? 0) + 1;
}
/** Clear stored events for a specific thread (e.g. on thread expiration). */
/** Drop a thread's live state (e.g. on thread deletion or expiration). */
clearThread(threadId: string): void {
this.store.delete(threadId);
this.sizeBytes.delete(threadId);
this.lastLocalId.delete(threadId);
this.pendingByThread.delete(threadId);
this.inFlightByThread.delete(threadId);
this.eventLog.clearThread(threadId);
this.emitter.removeAllListeners(threadId);
if (this.instanceSettings.isMultiMain) {
// Every main clears on thread deletion (task-control broadcast), so the
// shared key DEL is idempotent across mains.
void this.getRedisClient()
.del(this.seqKey(threadId))
.catch((error: unknown) =>
this.logger.warn('Failed to delete Instance AI event sequence key', {
threadId,
error,
}),
);
}
}
/** Clear all stored events. Used during module shutdown. Leaves the shared
* Redis sequence keys untouched — sibling mains still rely on them. */
/** Drop every thread's live state. Used during module shutdown. */
clear(): void {
this.store.clear();
this.sizeBytes.clear();
this.lastLocalId.clear();
this.pendingByThread.clear();
this.inFlightByThread.clear();
this.eventLog.clear();
this.emitter.removeAllListeners();
}
/**
* The store is not populated under the durable log, so a read of it there is
* a wiring bug: the caller wants the log. Reported once per method rather
* than thrown — an empty result degrades a trace annotation or a replay that
* the log-backed path will serve anyway, where a throw would take down run
* finalization.
*/
private assertNoStoreUnderDurableLog(method: string, threadId: string): boolean {
if (!this.durableLogEnabled) return false;
// Once per entry point: any call is a bug, but a regression could sit in
// a per-group loop (the flag-off replay has one), and this is a read path.
if (!this.warnedStoreReads.has(method)) {
this.warnedStoreReads.add(method);
this.logger.error(
`InProcessEventBus.${method} was read with the durable event log enabled, where nothing is stored in memory — the caller must use DurableEventLog instead`,
{ threadId },
);
}
return true;
}
private evictIfNeeded(threadId: string, events: SequencedEvent[]): void {
let totalSize = this.sizeBytes.get(threadId) ?? 0;
while (events.length > MAX_EVENTS_PER_THREAD || totalSize > MAX_BYTES_PER_THREAD) {
const evicted = events.shift();
if (!evicted) break;
totalSize -= Buffer.byteLength(JSON.stringify(evicted.event), 'utf8');
}
this.sizeBytes.set(threadId, Math.max(0, totalSize));
}
private getOrCreateStore(threadId: string): SequencedEvent[] {
let events = this.store.get(threadId);
if (!events) {
events = [];
this.store.set(threadId, events);
}
return events;
}
}
packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts
+3 -12
View File
@@ -1,6 +1,5 @@
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';
@@ -76,19 +75,15 @@ export class InterruptedRunSweeper {
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 {
@@ -97,8 +92,6 @@ export class InterruptedRunSweeper {
/** Called from module init (startup). */
async sweep(): Promise<void> {
if (!this.durableLogEnabled) return;
let unfinished;
try {
unfinished = await this.eventLogRepo.findUnfinishedRuns();
@@ -154,8 +147,6 @@ export class InterruptedRunSweeper {
* window — a per-run claim would need the lease table this design avoids.
*/
async cancelUnfinishedRuns(threadId: string): Promise<number> {
if (!this.durableLogEnabled) return 0;
let unfinished;
try {
unfinished = await this.eventLogRepo.findUnfinishedRuns(threadId);
@@ -209,9 +200,9 @@ export class InterruptedRunSweeper {
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.
// Exact hostRunId match only: every logged run post-dates the column, 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,
);
packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts
+18 -22
View File
@@ -472,22 +472,19 @@ export class InstanceAiMemoryService {
// a tree to pair with, and hydrating it unbounded would read the whole
// thread to render nothing.
//
// Durable-log flag (fold-on-read): history trees derive from the event
// log; the stored snapshots (the flag-off and rollback path) are only
// loaded when the fold needs its pre-log/failure fallback, keeping the
// heaviest instance-ai table out of the flag-on hot path.
// Fold-on-read: history trees derive from the event log. Stored snapshots
// are only loaded for the fold's pre-log/failure fallback, keeping the
// heaviest instance-ai table out of the hot path.
const snapshots = !pageWindow
? []
: this.instanceAiConfig.durableLog
? await this.foldSnapshotsFromLog(
threadId,
loadStoredSnapshots,
collectSuspendedHostRunIds(activeCheckpoints),
pageWindow,
options?.excludeRunIds,
options?.excludeMessageGroupIds,
)
: await loadStoredSnapshots();
: await this.foldSnapshotsFromLog(
threadId,
loadStoredSnapshots,
collectSuspendedHostRunIds(activeCheckpoints),
pageWindow,
options?.excludeRunIds,
options?.excludeMessageGroupIds,
);
// Surface the in-flight messages from any suspended checkpoint. The
// user's prompt is persisted to memory on receipt, but the intermediate
@@ -511,10 +508,9 @@ export class InstanceAiMemoryService {
}
/**
* Durable-log fold-on-read: with the flag on, history agent trees derive
* from the event log. Stored snapshot rows keep being written (they are the
* flag-off and rollback path) but are neither read nor loaded here; the
* lazy loader runs only when the thread has no log rows or the read
* Fold-on-read: history agent trees derive from the event log. Stored
* snapshot rows keep being written but are neither read nor loaded here;
* the lazy loader runs only when the thread has no log rows or the read
* fails/derives nothing.
*
* Only the runs behind the requested page are read and folded, so a long
@@ -534,10 +530,10 @@ export class InstanceAiMemoryService {
const start = Date.now();
let rows;
try {
// Pre-log thread (instance ran before the flag): no run has a start
// fact, so stored snapshots still render. Production flips the flag
// together with the backfill migration (INS-851), so this branch is a
// dev-instance safety, not a design. Checked on run starts rather than
// Pre-log thread: no run has a start fact, so stored snapshots still
// render. The backfill migration gave every pre-existing run event
// rows, so this branch is a dev-instance safety, not a design.
// Checked on run starts rather than
// on the windowed rows, which are also empty for a thread whose log
// simply has nothing inside the page.
const runStarts = await this.eventLogRepository.getRunStarts(threadId);
packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts
+3 -13
View File
@@ -90,8 +90,8 @@ function appendTerminalOutcomeToAgentTree(
// The slice of each collaborator the terminal-outcome coordinator actually
// uses. Anchored to the concrete types via `Pick` so the signatures stay in
// sync with the source.
// Reads may be sync (in-memory bus, flag off) or async (durable log, flag on);
// the host injects a flag-resolved adapter.
// Reads are async: the host injects an adapter that flushes the thread's drain
// and then queries the durable log.
export type InstanceAiTerminalOutcomeEventBus = Pick<InProcessEventBus, 'publish'> & {
getEventsForRun(threadId: string, runId: string): InstanceAiEvent[] | Promise<InstanceAiEvent[]>;
getEventsForRuns(
@@ -126,13 +126,6 @@ export type InstanceAiTerminalOutcomeTracing = Pick<
export interface InstanceAiTerminalOutcomeServiceOptions {
eventBus: InstanceAiTerminalOutcomeEventBus;
/**
* Durable-log flag: outcome lines publish as `text-block` (a structural
* fact, persisted before it is emitted live) instead of a trailing
* `text-delta`, so a page reload right after a background outcome folds
* the line from the log instead of racing the coalescer's idle flush.
*/
durableLog: boolean;
dbSnapshotStorage: InstanceAiTerminalOutcomeSnapshotStorage;
agentMemory: PatchableThreadMemory;
telemetry: InstanceAiTerminalOutcomeTelemetry;
@@ -186,8 +179,6 @@ export class InstanceAiTerminalOutcomeService {
private readonly eventBus: InstanceAiTerminalOutcomeEventBus;
private readonly durableLog: boolean;
private readonly dbSnapshotStorage: InstanceAiTerminalOutcomeSnapshotStorage;
private readonly agentMemory: PatchableThreadMemory;
@@ -210,7 +201,6 @@ export class InstanceAiTerminalOutcomeService {
constructor(options: InstanceAiTerminalOutcomeServiceOptions) {
this.eventBus = options.eventBus;
this.durableLog = options.durableLog;
this.dbSnapshotStorage = options.dbSnapshotStorage;
this.agentMemory = options.agentMemory;
this.telemetry = options.telemetry;
@@ -530,7 +520,7 @@ export class InstanceAiTerminalOutcomeService {
if (alreadyPublished) return false;
this.eventBus.publish(outcome.threadId, {
type: this.durableLog ? 'text-block' : 'text-delta',
type: 'text-block',
runId: outcome.runId,
agentId: orchestratorAgentId(outcome.runId),
responseId,
packages/cli/src/modules/instance-ai/instance-ai.controller.ts
+136 -159
View File
@@ -94,10 +94,6 @@ const KEEP_ALIVE_INTERVAL_MS = 15_000;
export class InstanceAiController {
private readonly gatewayApiKey: string;
/** Durable-log prototype flag (N8N_INSTANCE_AI_DURABLE_LOG): replay and
* cursors come from the DB-backed log instead of the in-memory bus. */
private readonly durableLogEnabled: boolean;
private static getTreeRichnessScore(tree: InstanceAiAgentNode): number {
let score = 0;
const stack = [tree];
@@ -153,7 +149,6 @@ export class InstanceAiController {
globalConfig: GlobalConfig,
) {
this.gatewayApiKey = globalConfig.instanceAi.gatewayApiKey;
this.durableLogEnabled = globalConfig.instanceAi.durableLog;
}
private requireInstanceAiEnabled(): void {
@@ -442,160 +437,144 @@ export class InstanceAiController {
);
};
if (this.durableLogEnabled) {
// 6. Replay missed events from the DURABLE log — survives restarts and is
// valid on any main (the table is in the shared DB). The reads are
// async, so unlike the old synchronous memory-store replay, live
// events can land mid-bootstrap: buffer them across every await (the
// replay read, the run-sync tree reads, AND the gap read) and flush
// with seq dedupe only when no await remains before live delivery
// takes over (the drain persists before it emits, so a fact is never
// in neither place). A flushed event may already be folded into a
// run-sync tree; the shared reducer applies it idempotently, same as
// any live event arriving after a frame.
const arrivedDuringReplay: StoredEvent[] = [];
const stopBuffering = this.eventBus.subscribe(threadId, (stored) => {
arrivedDuringReplay.push(stored);
});
try {
const missed = await this.eventLog.getEventsAfter(threadId, cursor);
// The client may have disconnected during the read: stop before
// writing to the dead response or arming the keep-alive below.
if (closed) return;
let lastReplayedSeq = cursor;
for (const stored of missed) {
deliver(stored);
if (stored.id !== undefined) lastReplayedSeq = stored.id;
}
// Build each live group's bootstrap tree from the durable log, so the
// group renders fully even when the bus cache was evicted, the process
// restarted, or this main never buffered the thread (sibling main).
// Remember which coalesced blocks each delivered tree folds: the gap
// read below may return the same rows, and re-applying a block the
// tree already renders would append a duplicate timeline entry.
const blockKey = (event: {
type: string;
runId: string;
agentId: string;
responseId?: string;
payload: { text: string };
}) =>
`${event.type}:${event.runId}:${event.agentId}:${event.responseId ?? ''}:${event.payload.text}`;
const foldedBlockKeys = new Set<string>();
for (const [groupId, group] of liveGroups) {
const runEvents = await this.eventLog.getEventsForRuns(threadId, group.runIds);
if (closed) return;
writeRunSyncFrame(groupId, group, runEvents);
for (const event of runEvents) {
if (event.type === 'text-block' || event.type === 'reasoning-block') {
foldedBlockKeys.add(blockKey(event));
}
}
}
// One more durable read: coalesced blocks are persisted but never
// live-emitted (live clients saw the deltas), so a segment that closed
// during the awaits above exists only as rows the replay read predates
// — invisible to the buffering subscription. Without this read, the
// buffered fact that follows such a block would advance the browser
// cursor past it and no later replay would ever return it.
const gapRows = await this.eventLog.getEventsAfter(threadId, lastReplayedSeq);
if (closed) return;
// A still-streaming segment exists only in the log's coalesce buffer
// (deltas are never persisted), so a mid-stream refresh would render
// only the post-refresh tail. Serve each open segment as one ephemeral
// delta frame (no `id:` line — the cursor stays on durable facts), after
// the run-sync frames so live deltas keep appending to it and the
// segment's eventual block replaces it. Everything from this read to
// `bootstrapping = false` is synchronous, so a buffered delta of a
// served segment is exactly text inside the snapshot: skipping it loses
// nothing and delivering it would duplicate.
const openSegments = this.eventLog.getOpenSegments(threadId);
const segmentKey = (
kind: 'text' | 'reasoning',
event: { runId: string; agentId: string; responseId?: string },
) => `${kind}:${event.runId}:${event.agentId}:${event.responseId ?? ''}`;
const served = new Set(openSegments.map((segment) => segmentKey(segment.kind, segment)));
// Deliver the gap rows first. A block identical to one folded into a
// delivered run-sync tree is not re-applied (that would duplicate its
// text) but still counts as delivered for cursor contiguity — its
// content reached the client inside the frame. Buffered deltas of a
// gap block's segment are skipped below like served ones: their text
// is inside the block, and delivering them after it would duplicate.
const gapBlockSegments = new Set<string>();
for (const row of gapRows) {
if (row.id === undefined || row.id <= lastReplayedSeq) continue;
const { event } = row;
if (event.type === 'text-block' || event.type === 'reasoning-block') {
gapBlockSegments.add(
segmentKey(event.type === 'text-block' ? 'text' : 'reasoning', event),
);
if (foldedBlockKeys.has(blockKey(event))) {
lastReplayedSeq = row.id;
continue;
}
}
deliver(row);
lastReplayedSeq = row.id;
}
for (const stored of arrivedDuringReplay) {
if (stored.id !== undefined) {
if (stored.id <= lastReplayedSeq) continue;
if (stored.id === lastReplayedSeq + 1) {
deliver(stored);
lastReplayedSeq = stored.id;
continue;
}
// Rows between the cursor and this fact were persisted after the
// gap read (a segment closed while it was in flight): deliver the
// fact's content but strip its id line, so the cursor never
// crosses a row the client has not seen — the next replay returns
// the missing block and re-applies this fact idempotently.
deliver({ event: stored.event });
continue;
}
const { event } = stored;
if (
(event.type === 'text-delta' || event.type === 'reasoning-delta') &&
(served.has(segmentKey(event.type === 'text-delta' ? 'text' : 'reasoning', event)) ||
gapBlockSegments.has(
segmentKey(event.type === 'text-delta' ? 'text' : 'reasoning', event),
))
) {
continue;
}
deliver(stored);
}
for (const segment of openSegments) {
deliver({
event: {
type: segment.kind === 'text' ? 'text-delta' : 'reasoning-delta',
runId: segment.runId,
agentId: segment.agentId,
...(segment.responseId ? { responseId: segment.responseId } : {}),
payload: { text: segment.text },
},
});
}
this.durableLogMetrics.recordReplay(missed.length, Math.max(0, lastReplayedSeq - cursor));
} finally {
// The buffering subscription must not outlive the bootstrap, even when
// a durable read throws.
stopBuffering();
}
} else {
// 6. Replay missed events, emit run-sync frames, and flip to live delivery
// in one synchronous block. The event bus store and emitter are
// synchronous, so no event can slip between the replay and the live
// handler taking over. Events that arrived during the awaits above are
// already in the store (the early subscription in step 1 keeps relayed
// events flowing in multi-main) and are included in the replay here.
const missed = this.eventBus.getEventsAfter(threadId, cursor);
// 6. Replay missed events from the durable log — survives restarts and is
// valid on any main (the table is in the shared DB). The reads are
// async, so live events can land mid-bootstrap: buffer them across
// every await (the
// replay read, the run-sync tree reads, AND the gap read) and flush
// with seq dedupe only when no await remains before live delivery
// takes over (the drain persists before it emits, so a fact is never
// in neither place). A flushed event may already be folded into a
// run-sync tree; the shared reducer applies it idempotently, same as
// any live event arriving after a frame.
const arrivedDuringReplay: StoredEvent[] = [];
const stopBuffering = this.eventBus.subscribe(threadId, (stored) => {
arrivedDuringReplay.push(stored);
});
try {
const missed = await this.eventLog.getEventsAfter(threadId, cursor);
// The client may have disconnected during the read: stop before
// writing to the dead response or arming the keep-alive below.
if (closed) return;
let lastReplayedSeq = cursor;
for (const stored of missed) {
deliver(stored);
if (stored.id !== undefined) lastReplayedSeq = stored.id;
}
// Build each live group's bootstrap tree from the durable log, so the
// group renders fully even when the process
// restarted, or this main never buffered the thread (sibling main).
// Remember which coalesced blocks each delivered tree folds: the gap
// read below may return the same rows, and re-applying a block the
// tree already renders would append a duplicate timeline entry.
const blockKey = (event: {
type: string;
runId: string;
agentId: string;
responseId?: string;
payload: { text: string };
}) =>
`${event.type}:${event.runId}:${event.agentId}:${event.responseId ?? ''}:${event.payload.text}`;
const foldedBlockKeys = new Set<string>();
for (const [groupId, group] of liveGroups) {
writeRunSyncFrame(groupId, group, this.eventBus.getEventsForRuns(threadId, group.runIds));
const runEvents = await this.eventLog.getEventsForRuns(threadId, group.runIds);
if (closed) return;
writeRunSyncFrame(groupId, group, runEvents);
for (const event of runEvents) {
if (event.type === 'text-block' || event.type === 'reasoning-block') {
foldedBlockKeys.add(blockKey(event));
}
}
}
// One more durable read: coalesced blocks are persisted but never
// live-emitted (live clients saw the deltas), so a segment that closed
// during the awaits above exists only as rows the replay read predates
// — invisible to the buffering subscription. Without this read, the
// buffered fact that follows such a block would advance the browser
// cursor past it and no later replay would ever return it.
const gapRows = await this.eventLog.getEventsAfter(threadId, lastReplayedSeq);
if (closed) return;
// A still-streaming segment exists only in the log's coalesce buffer
// (deltas are never persisted), so a mid-stream refresh would render
// only the post-refresh tail. Serve each open segment as one ephemeral
// delta frame (no `id:` line — the cursor stays on durable facts), after
// the run-sync frames so live deltas keep appending to it and the
// segment's eventual block replaces it. Everything from this read to
// `bootstrapping = false` is synchronous, so a buffered delta of a
// served segment is exactly text inside the snapshot: skipping it loses
// nothing and delivering it would duplicate.
const openSegments = this.eventLog.getOpenSegments(threadId);
const segmentKey = (
kind: 'text' | 'reasoning',
event: { runId: string; agentId: string; responseId?: string },
) => `${kind}:${event.runId}:${event.agentId}:${event.responseId ?? ''}`;
const served = new Set(openSegments.map((segment) => segmentKey(segment.kind, segment)));
// Deliver the gap rows first. A block identical to one folded into a
// delivered run-sync tree is not re-applied (that would duplicate its
// text) but still counts as delivered for cursor contiguity — its
// content reached the client inside the frame. Buffered deltas of a
// gap block's segment are skipped below like served ones: their text
// is inside the block, and delivering them after it would duplicate.
const gapBlockSegments = new Set<string>();
for (const row of gapRows) {
if (row.id === undefined || row.id <= lastReplayedSeq) continue;
const { event } = row;
if (event.type === 'text-block' || event.type === 'reasoning-block') {
gapBlockSegments.add(
segmentKey(event.type === 'text-block' ? 'text' : 'reasoning', event),
);
if (foldedBlockKeys.has(blockKey(event))) {
lastReplayedSeq = row.id;
continue;
}
}
deliver(row);
lastReplayedSeq = row.id;
}
for (const stored of arrivedDuringReplay) {
if (stored.id !== undefined) {
if (stored.id <= lastReplayedSeq) continue;
if (stored.id === lastReplayedSeq + 1) {
deliver(stored);
lastReplayedSeq = stored.id;
continue;
}
// Rows between the cursor and this fact were persisted after the
// gap read (a segment closed while it was in flight): deliver the
// fact's content but strip its id line, so the cursor never
// crosses a row the client has not seen — the next replay returns
// the missing block and re-applies this fact idempotently.
deliver({ event: stored.event });
continue;
}
const { event } = stored;
if (
(event.type === 'text-delta' || event.type === 'reasoning-delta') &&
(served.has(segmentKey(event.type === 'text-delta' ? 'text' : 'reasoning', event)) ||
gapBlockSegments.has(
segmentKey(event.type === 'text-delta' ? 'text' : 'reasoning', event),
))
) {
continue;
}
deliver(stored);
}
for (const segment of openSegments) {
deliver({
event: {
type: segment.kind === 'text' ? 'text-delta' : 'reasoning-delta',
runId: segment.runId,
agentId: segment.agentId,
...(segment.responseId ? { responseId: segment.responseId } : {}),
payload: { text: segment.text },
},
});
}
this.durableLogMetrics.recordReplay(missed.length, Math.max(0, lastReplayedSeq - cursor));
} finally {
// The buffering subscription must not outlive the bootstrap, even when
// a durable read throws.
stopBuffering();
}
if (liveGroups.size > 0) res.flush?.();
@@ -987,11 +966,9 @@ export class InstanceAiController {
// Include the next SSE event ID so the frontend can skip past events
// already covered by these historical messages (prevents duplicates).
// Flag on: durable authority, valid across restarts and mains. Flag off:
// the shared sequence, so the cursor is valid against any main.
const nextEventId = this.durableLogEnabled
? await this.eventLog.getNextEventId(threadId)
: await this.eventBus.getNextEventId(threadId);
// Read from the log, so the cursor is valid across restarts and across
// mains sharing the database.
const nextEventId = await this.eventLog.getNextEventId(threadId);
return { ...result, nextEventId };
}
packages/cli/src/modules/instance-ai/instance-ai.module.ts
+11 -25
View File
@@ -36,31 +36,17 @@ 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 logger = Container.get(Logger).scoped('instance-ai');
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 });
});
} else {
// Surfaced at boot because the off switch changes this main's resource
// profile, not just where events are read from: the legacy buffer is
// held per thread until that thread is deleted or expires, so memory
// grows with the threads this main serves. Fixed only for the
// durable-log path; the legacy one sunsets with the flag at Gate B.
Container.get(Logger)
.scoped('instance-ai')
.warn(
'N8N_INSTANCE_AI_DURABLE_LOG is off: Instance AI events are held in a per-thread in-memory buffer (500 events / 2MB each) instead of the durable log. Memory scales with the number of threads this main has served, and replay does not survive a restart. Intended as a temporary rollback switch.',
);
}
// 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 { InterruptedRunSweeper } = await import('./event-bus/interrupted-run-sweeper.js');
const { InstanceAiService } = await import('./instance-ai.service.js');
const sweepLogger = Container.get(Logger).scoped('instance-ai');
const sweeper = Container.get(InterruptedRunSweeper);
sweeper.setResumeHost(Container.get(InstanceAiService));
void sweeper.sweep().catch((error: unknown) => {
sweepLogger.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');
packages/cli/src/modules/instance-ai/instance-ai.service.ts
+18 -26
View File
@@ -175,7 +175,6 @@ import {
buildInstanceAiObservabilityContext,
type InstanceAiObservabilityContext,
} from './observability';
import { resolveOutputRedaction } from './output-redaction-config';
import {
PlannedTaskActionRunner,
type PlannedBuildFollowUp,
@@ -880,9 +879,8 @@ export class InstanceAiService {
});
this.tracing = new InstanceAiTracingService({
logger: this.logger,
// `first_visible_state` has to see the run's streamed text, which under
// the durable log lives in the log (as coalesced blocks), never in the
// bus cache.
// `first_visible_state` has to see the run's streamed text, which lives
// in the log as coalesced blocks — the bus retains nothing.
eventReader: {
getEventsForRun: async (threadId, runId) => await this.readRunEvents(threadId, [runId]),
},
@@ -900,7 +898,6 @@ export class InstanceAiService {
aiService: this.aiService,
});
this.terminalOutcome = new InstanceAiTerminalOutcomeService({
durableLog: globalConfig.instanceAi.durableLog,
// The terminal guard and outcome-replay dedup must see the run's events
// after a restart too, which only the durable log can provide (the bus
// cache is empty in a fresh process).
@@ -1952,8 +1949,8 @@ export class InstanceAiService {
this.domainAccessTrackersByThread.clear();
this.tracing.clear();
// Durable-log flag: flush in-flight drains + open coalesce buffers so the
// tail of every streamed segment survives the restart. No-op when off.
// Flush in-flight drains + open coalesce buffers so the tail of every
// streamed segment survives the restart.
await this.eventLog.flushAll();
this.eventBus.clear();
@@ -2647,7 +2644,7 @@ export class InstanceAiService {
checkpointStore: this.checkpointStore,
eventBus: this.eventBus,
logger: this.logger,
outputRedaction: resolveOutputRedaction(this.instanceAiConfig),
outputRedaction: false, // raw-at-rest: the redactor defaults ON when omitted (INS-837)
trackTelemetry: (eventName, properties) => {
this.telemetry.track(eventName, redactTelemetryProperties(properties));
},
@@ -4104,7 +4101,7 @@ export class InstanceAiService {
logger: this.logger,
onActivity: () => this.runState.touchActiveRun(threadId),
stopSignal,
outputRedaction: resolveOutputRedaction(this.instanceAiConfig),
outputRedaction: false, // raw-at-rest: the redactor defaults ON when omitted (INS-837)
});
})
: await streamAgentRun(agent as StreamableAgent, streamInput, streamOptions, {
@@ -4116,7 +4113,7 @@ export class InstanceAiService {
logger: this.logger,
onActivity: () => this.runState.touchActiveRun(threadId),
stopSignal,
outputRedaction: resolveOutputRedaction(this.instanceAiConfig),
outputRedaction: false, // raw-at-rest: the redactor defaults ON when omitted (INS-837)
});
if (result.status === 'suspended') {
// finalizeRun only fires on terminal outcomes; record suspended-segment usage here.
@@ -5483,7 +5480,7 @@ export class InstanceAiService {
agentRunId: opts.agentRunId,
onActivity: () => this.runState.touchActiveRun(opts.threadId),
stopSignal,
outputRedaction: resolveOutputRedaction(this.instanceAiConfig),
outputRedaction: false, // raw-at-rest: the redactor defaults ON when omitted (INS-837)
});
})
: await resumeAgentRun(agent, resumeData, resumeOptions, {
@@ -5496,7 +5493,7 @@ export class InstanceAiService {
agentRunId: opts.agentRunId,
onActivity: () => this.runState.touchActiveRun(opts.threadId),
stopSignal,
outputRedaction: resolveOutputRedaction(this.instanceAiConfig),
outputRedaction: false, // raw-at-rest: the redactor defaults ON when omitted (INS-837)
});
if (!resumeClaimed) {
skipPostRunCleanup = true;
@@ -6873,18 +6870,13 @@ export class InstanceAiService {
}
/**
* The one place the run-event source is chosen. With the durable log on it
* is a read-own-writes barrier: settle the thread's drain (including open
* coalesce buffers) so everything published before this call is visible,
* then read the log. With it off the in-memory bus store is the only
* source. Every caller is a run boundary — terminal-guard inputs, trace
* metadata, snapshot builds — where closing the open segment early is
* correct anyway.
* Read-own-writes barrier for run-scoped reads: settle the thread's drain
* (including open coalesce buffers) so everything published before this call
* is visible, then read the log. Every caller is a run boundary —
* terminal-guard inputs, trace metadata, snapshot builds — where closing the
* open segment early is correct anyway.
*/
private async readRunEvents(threadId: string, runIds: string[]): Promise<InstanceAiEvent[]> {
if (!this.instanceAiConfig.durableLog) {
return this.eventBus.getEventsForRuns(threadId, runIds);
}
await this.eventLog.flush(threadId);
return await this.eventLog.getEventsForRuns(threadId, runIds);
}
@@ -6915,10 +6907,10 @@ export class InstanceAiService {
} else {
events = await this.readRunEvents(threadId, [runId]);
}
// Durable-log flag on: the tree input comes from the DB, so long runs can
// no longer out-evict their own snapshot input (the empty-agentTree bug
// class). The snapshot write itself stays during migration so pre-log
// threads keep rendering; INS-841 moves history to fold-on-read.
// The tree input comes from the DB, so long runs cannot out-evict their
// own snapshot input (the empty-agentTree bug class). The snapshot write
// itself stays for now so pre-log threads keep rendering; history moves
// to fold-on-read separately.
if (isUpdate && events.length === 0) {
this.logger.warn('Skipped updating empty Instance AI agent tree snapshot', {
threadId,
packages/cli/src/modules/instance-ai/message-parser.ts
+1 -1
View File
@@ -283,7 +283,7 @@ function buildFlatAgentTree(
/**
* Whether a snapshot tree carries anything worth rendering. An empty terminal tree —
* e.g. a `cancelled` run whose events were evicted from the in-memory bus before the
* e.g. a `cancelled` run whose events were lost before the
* snapshot was built — has none of these, so the message-derived flat tree is preferred.
*/
function isRenderableTree(tree: InstanceAiAgentNode): boolean {
packages/cli/src/modules/instance-ai/output-redaction-config.ts
-36
View File
@@ -1,36 +0,0 @@
import {
SUPPORTED_PII_CATEGORIES,
type PiiDetectionType,
type RedactionOptions,
} from '@n8n/agents';
import type { InstanceAiConfig } from '@n8n/config';
/**
* Resolve the Instance AI output-redaction policy from env config.
* Returns `false` when disabled so the redactor passes events through untouched.
*
* Durable-log flag: raw-at-rest storage policy (team decision, 2026-07-06) —
* the stream-side redactor moves off the publish path so the log captures raw
* values, consistent with workflow execution data. Redaction applies at egress
* boundaries instead; the stricter LangSmith telemetry redactor
* (trace-payloads.ts) is a separate layer and is unchanged.
*/
export function resolveOutputRedaction(config: InstanceAiConfig): RedactionOptions | false {
if (config.durableLog) return false;
if (!config.outputRedactionEnabled) return false;
const detect = config.outputRedactionPii
.split(',')
.map((value) => value.trim())
.filter((value): value is PiiDetectionType =>
(SUPPORTED_PII_CATEGORIES as readonly string[]).includes(value),
);
const placeholder = config.outputRedactionPlaceholder;
return {
secrets: config.outputRedactionSecrets,
detect,
// Fall back to the engine default when configured blank.
...(placeholder ? { placeholder } : {}),
};
}
packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts
+1 -3
View File
@@ -32,9 +32,7 @@ export class InstanceAiEventLogRepository extends Repository<InstanceAiEventLogE
* in one transaction. The (threadId, seq) PK makes a concurrent-writer race
* fail loudly instead of silently interleaving — the caller re-reads maxSeq
* and retries. Returns the serialized payload bytes written (instrumentation).
*
* INS-844 (compose with the shared-sequence drain): the live-id Redis INCRBY
* merges into this call, so id assignment and durable insert become one round trip.
* Id assignment and the durable insert are one round trip.
*/
async appendBatch(
threadId: string,
packages/cli/src/modules/instance-ai/tracing/instance-ai-tracing.service.ts
+3 -4
View File
@@ -58,10 +58,9 @@ export type OrchestratorResumeReason =
// The slice of each collaborator the tracing service actually uses. Anchored to
// the concrete types via `Pick` so the signatures stay in sync with the source.
/**
* Flag-resolved run-event read: the durable log with `instanceAi.durableLog` on,
* the in-memory bus store with it off. Async because the durable read flushes
* the thread's open coalesce buffers and then queries — which is what makes the
* streamed text of a run visible to `first_visible_state` at all.
* Run-event read from the durable log. Async because it flushes the thread's
* open coalesce buffers and then queries — which is what makes the streamed
* text of a run visible to `first_visible_state` at all.
*/
export type InstanceAiTracingEventReader = {
getEventsForRun: (threadId: string, runId: string) => Promise<InstanceAiEvent[]>;