diff --git a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts index 337674d607e..804cd16eb1a 100644 --- a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts +++ b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts @@ -185,8 +185,8 @@ export const instanceAiEventTypeSchema = z.enum([ export type InstanceAiEventType = z.infer; /** - * 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 diff --git a/packages/@n8n/config/src/configs/instance-ai.config.ts b/packages/@n8n/config/src/configs/instance-ai.config.ts index 283e7c7eb20..612d43ead69 100644 --- a/packages/@n8n/config/src/configs/instance-ai.config.ts +++ b/packages/@n8n/config/src/configs/instance-ai.config.ts @@ -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; diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index eadd247fa94..08af5339adc 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -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, diff --git a/packages/@n8n/instance-ai/docs/architecture.md b/packages/@n8n/instance-ai/docs/architecture.md index de6fd91ddb7..a378cb4b5b5 100644 --- a/packages/@n8n/instance-ai/docs/architecture.md +++ b/packages/@n8n/instance-ai/docs/architecture.md @@ -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
main n8n database] Memory --> SQLite[SQLite
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) diff --git a/packages/@n8n/instance-ai/docs/configuration.md b/packages/@n8n/instance-ai/docs/configuration.md index 9744f48237e..bb9e0db7486 100644 --- a/packages/@n8n/instance-ai/docs/configuration.md +++ b/packages/@n8n/instance-ai/docs/configuration.md @@ -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 diff --git a/packages/@n8n/instance-ai/docs/streaming-protocol.md b/packages/@n8n/instance-ai/docs/streaming-protocol.md index 16514f09f26..cd90ca3caa5 100644 --- a/packages/@n8n/instance-ai/docs/streaming-protocol.md +++ b/packages/@n8n/instance-ai/docs/streaming-protocol.md @@ -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 diff --git a/packages/@n8n/instance-ai/evaluations/harness/in-memory-event-bus.ts b/packages/@n8n/instance-ai/evaluations/harness/in-memory-event-bus.ts index dc46295dda8..b56c573c6a5 100644 --- a/packages/@n8n/instance-ai/evaluations/harness/in-memory-event-bus.ts +++ b/packages/@n8n/instance-ai/evaluations/harness/in-memory-event-bus.ts @@ -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); - }, }; } diff --git a/packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts b/packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts index f3fa698d9ab..fa2e2b460b4 100644 --- a/packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts +++ b/packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts @@ -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; } diff --git a/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts b/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts index 2d0fbd55882..630bb4fb6bb 100644 --- a/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts +++ b/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts @@ -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([]), }; diff --git a/packages/@n8n/instance-ai/src/runtime/__tests__/stream-runner.test.ts b/packages/@n8n/instance-ai/src/runtime/__tests__/stream-runner.test.ts index b9f33892133..f4220bc5d79 100644 --- a/packages/@n8n/instance-ai/src/runtime/__tests__/stream-runner.test.ts +++ b/packages/@n8n/instance-ai/src/runtime/__tests__/stream-runner.test.ts @@ -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([]), }; diff --git a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts index d10443239af..e0802a76978 100644 --- a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts +++ b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts @@ -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; } diff --git a/packages/@n8n/instance-ai/src/runtime/stream-runner.ts b/packages/@n8n/instance-ai/src/runtime/stream-runner.ts index d6687ce7d05..69fb6977992 100644 --- a/packages/@n8n/instance-ai/src/runtime/stream-runner.ts +++ b/packages/@n8n/instance-ai/src/runtime/stream-runner.ts @@ -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; } diff --git a/packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts b/packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts index 42acf1c7c35..2939343f9ba 100644 --- a/packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts +++ b/packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts @@ -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 diff --git a/packages/@n8n/instance-ai/src/stream/__tests__/consume-with-hitl.test.ts b/packages/@n8n/instance-ai/src/stream/__tests__/consume-with-hitl.test.ts index 728526ce3aa..d3ae3de5006 100644 --- a/packages/@n8n/instance-ai/src/stream/__tests__/consume-with-hitl.test.ts +++ b/packages/@n8n/instance-ai/src/stream/__tests__/consume-with-hitl.test.ts @@ -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([]), }; } diff --git a/packages/@n8n/instance-ai/src/stream/consume-with-hitl.ts b/packages/@n8n/instance-ai/src/stream/consume-with-hitl.ts index a8a7110be3d..6c02322dbb5 100644 --- a/packages/@n8n/instance-ai/src/stream/consume-with-hitl.ts +++ b/packages/@n8n/instance-ai/src/stream/consume-with-hitl.ts @@ -35,7 +35,11 @@ export interface ConsumeWithHitlOptions { resumeOptions?: Record; /** 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; } diff --git a/packages/@n8n/instance-ai/src/stream/output-redaction.ts b/packages/@n8n/instance-ai/src/stream/output-redaction.ts index 5154940abdb..df2a4a746bf 100644 --- a/packages/@n8n/instance-ai/src/stream/output-redaction.ts +++ b/packages/@n8n/instance-ai/src/stream/output-redaction.ts @@ -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; } diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.cascade-restart.test.ts b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.cascade-restart.test.ts index 66530d06c92..17e4eab0604 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.cascade-restart.test.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.cascade-restart.test.ts @@ -314,10 +314,6 @@ function createEventBusStub(): InstanceAiEventBus { return { publish: () => {}, subscribe: () => () => {}, - getEventsAfter: () => [], - getEventsForRun: () => [], - getEventsForRuns: () => [], - getNextEventId: async () => await Promise.resolve(1), }; } diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/complete-checkpoint.tool.test.ts b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/complete-checkpoint.tool.test.ts index c83ea77648e..7105e31f2de 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/complete-checkpoint.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/complete-checkpoint.tool.test.ts @@ -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(), diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/plan.tool.test.ts b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/plan.tool.test.ts index 3d6a044c37e..2510af5b800 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/plan.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/plan.tool.test.ts @@ -25,10 +25,6 @@ function createMockContext(overrides: Partial = {}): 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(), diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/report-verification-verdict.tool.test.ts b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/report-verification-verdict.tool.test.ts index d85158a90a7..258212d5fda 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/report-verification-verdict.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/report-verification-verdict.tool.test.ts @@ -31,10 +31,6 @@ function createMockContext(overrides: Partial = {}): 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(), diff --git a/packages/@n8n/instance-ai/src/types.ts b/packages/@n8n/instance-ai/src/types.ts index 79f558bd056..fb184845dce 100644 --- a/packages/@n8n/instance-ai/src/types.ts +++ b/packages/@n8n/instance-ai/src/types.ts @@ -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) => void; /** diff --git a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts index 55e8450c6e2..33dd37b21c5 100644 --- a/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus/instance-ai-metrics.service.ts @@ -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).', diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts index 9f922eb2010..469954a30db 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts @@ -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', { diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts index 4ccb4c8c6f2..ee36c9225b8 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts @@ -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: {}, diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts index a7609221608..3993e0c8c35 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts @@ -121,7 +121,7 @@ describe('InstanceAiController', () => { const publisher = mock(); const urlService = mock(); const globalConfig = mock({ - 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>(); memoryService.getRichMessages.mockResolvedValue(richResult); - eventBus.getNextEventId.mockResolvedValue(42); + eventLog.getNextEventId.mockResolvedValue(42); const query = mock({ 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(); const memoryService = mock(); const settingsService = mock(); @@ -1993,7 +2004,7 @@ describe('InstanceAiController — durable-log SSE replay (flag on)', () => { const eventLog = mock(); const durableLogMetrics = mock(); const globalConfig = mock({ - instanceAi: { gatewayApiKey: 'static-key', durableLog: true }, + instanceAi: { gatewayApiKey: 'static-key' }, editorBaseUrl: 'http://localhost:5678', port: 5678, }); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts index cd7034ab0dd..7b50d9fce25 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts @@ -673,12 +673,7 @@ type TerminalGuardOrderServiceInternals = { suspendedThreads: { dropPendingConfirmationsForThread: Mock; persistPendingConfirmation: Mock }; logger: { warn: Mock; error: Mock }; instanceAiErrorReporter: ReturnType; - 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'); diff --git a/packages/cli/src/modules/instance-ai/__tests__/output-redaction-config.test.ts b/packages/cli/src/modules/instance-ai/__tests__/output-redaction-config.test.ts deleted file mode 100644 index e3d95ec9e4a..00000000000 --- a/packages/cli/src/modules/instance-ai/__tests__/output-redaction-config.test.ts +++ /dev/null @@ -1,65 +0,0 @@ -import type { InstanceAiConfig } from '@n8n/config'; - -import { resolveOutputRedaction } from '../output-redaction-config'; - -function config(overrides: Partial = {}): 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', - ); - }); -}); diff --git a/packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts b/packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts index 692ed92ab49..25de275edf5 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts @@ -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>; @@ -30,499 +26,97 @@ describe('InProcessEventBus', () => { let logger: ReturnType>; 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; - 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.scoped.mockReturnValue(logger); publisher = mock(); 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(); - const globalConfig = mock({ - 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); }); }); }); diff --git a/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts b/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts index a1e9eb1dd71..140d706e729 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/__tests__/interrupted-run-sweeper.test.ts @@ -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; } @@ -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: [ diff --git a/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts b/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts index 0a8ca9050a3..48b50a813cb 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/durable-event-log.ts @@ -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; } diff --git a/packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts b/packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts index 6864a72bbf5..889ee1b2f2d 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts @@ -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(); - - /** Approximate serialized size per thread for eviction. */ - private readonly sizeBytes = new Map(); - - /** - * 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(); - - /** - * 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(); - - /** - * 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(); - - private readonly drainingThreads = new Set(); - - private readonly seqKeyPrefix: string; - - private readonly durableLogEnabled: boolean; - - /** Store-read methods already reported under the durable log (log once). */ - private readonly warnedStoreReads = new Set(); - 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 { - 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 { - 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 { - // 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; - } } diff --git a/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts b/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts index 00108c166ac..301c7c04839 100644 --- a/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts +++ b/packages/cli/src/modules/instance-ai/event-bus/interrupted-run-sweeper.ts @@ -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 { - 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 { - 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, ); diff --git a/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts b/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts index 26bf52eeda8..cea0e17e456 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts @@ -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); diff --git a/packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts b/packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts index 7401130b732..8a0c65873d2 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts @@ -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 & { getEventsForRun(threadId: string, runId: string): InstanceAiEvent[] | Promise; 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, diff --git a/packages/cli/src/modules/instance-ai/instance-ai.controller.ts b/packages/cli/src/modules/instance-ai/instance-ai.controller.ts index 04b1a793ae0..a853f3a49cf 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.controller.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.controller.ts @@ -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(); - 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(); - 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(); 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(); + 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 }; } diff --git a/packages/cli/src/modules/instance-ai/instance-ai.module.ts b/packages/cli/src/modules/instance-ai/instance-ai.module.ts index f02475b15e5..b8bcb5303aa 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.module.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.module.ts @@ -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'); diff --git a/packages/cli/src/modules/instance-ai/instance-ai.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.service.ts index 58d203f6e61..daed30f5b64 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -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 { - 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, diff --git a/packages/cli/src/modules/instance-ai/message-parser.ts b/packages/cli/src/modules/instance-ai/message-parser.ts index f5d7203671d..742b6a46294 100644 --- a/packages/cli/src/modules/instance-ai/message-parser.ts +++ b/packages/cli/src/modules/instance-ai/message-parser.ts @@ -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 { diff --git a/packages/cli/src/modules/instance-ai/output-redaction-config.ts b/packages/cli/src/modules/instance-ai/output-redaction-config.ts deleted file mode 100644 index 86e21a64cc8..00000000000 --- a/packages/cli/src/modules/instance-ai/output-redaction-config.ts +++ /dev/null @@ -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 } : {}), - }; -} diff --git a/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts index ce7ac133f3b..7609a2b6ffb 100644 --- a/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts +++ b/packages/cli/src/modules/instance-ai/repositories/instance-ai-event-log.repository.ts @@ -32,9 +32,7 @@ export class InstanceAiEventLogRepository extends Repository Promise;