diff --git a/.agents/skills/create-instance-ai-eval/SKILL.md b/.agents/skills/create-instance-ai-eval/SKILL.md index b76a1ef21c1..2da38503610 100644 --- a/.agents/skills/create-instance-ai-eval/SKILL.md +++ b/.agents/skills/create-instance-ai-eval/SKILL.md @@ -140,7 +140,7 @@ discover → verify → encode workflow. ## Pick the case shape first -The corpus is four archetypes. Decide which you're writing before you draft — it +The corpus is five archetypes. Decide which you're writing before you draft — it determines the fields, the grading, and how you validate. They compose (a seeded case can still assert outcome), but the primary shape drives the work. @@ -150,10 +150,25 @@ case can still assert outcome), but the primary shape drives the work. | **Behaviour / process** | Does the agent *converse* correctly (ask the right clarifying question, not re-ask, honour a correction, respect plan approval)? | `processExpectations` + multi-turn director script; often **build-only** | | **Credential** | Does the build behave correctly given a specific credential view? | `credentials[]` | | **Seeded** | Start mid-thread, with prior work already in place, and drive the turn under test | `seed` (authored `mode: "inline"`; `"replay"` for a local check) | +| **Context** | Does the agent still work once the conversation is long — does it reuse what it already read, and keep what matters after observational memory compacts the history? | `processExpectations` + a long `seed`; `requiresMemoryCompaction` for the compaction ones | -**Build** is documented in full below. The other three, the director-script +**Build** is documented in full below. The other four, the director-script vocabulary, and the seeding modes are in [`case-shapes.md`](case-shapes.md). +The judge also sees **token ground truth** for workflow-build cases: per-turn +input and output tokens inline in each transcript turn header, a build-wide +total, a cache read/write split, and the opening step's cost. So an expectation +may reference cost or consumption directly ("does not re-read the same node's +schema in turn 2"). The numbers cover the orchestrator's own LLM steps only. A +delegated Agent build runs in a sub-agent whose steps are not in the snapshot, +so do not write cost expectations on an Agent case. Those numbers come from run-debug snapshots, so they need +`N8N_INSTANCE_AI_RUN_DEBUG_ENABLED=true` on the instance under test; without it +the judge reads `(no run debug captured)` and cost expectations are ungradeable. + +The judge also sees the thread's **observation rows** — what observational +memory kept after it compacted — so a memory case can grade the summary itself. +That block is a separate REST read and needs no flag. + ## Core principle (all shapes) **Write expectations from intent, then calibrate against a real build.** Decide diff --git a/.agents/skills/create-instance-ai-eval/case-shapes.md b/.agents/skills/create-instance-ai-eval/case-shapes.md index b7ddbd4bd5f..9b5da730762 100644 --- a/.agents/skills/create-instance-ai-eval/case-shapes.md +++ b/.agents/skills/create-instance-ai-eval/case-shapes.md @@ -1,7 +1,7 @@ # Case shapes beyond the plain build The [SKILL](SKILL.md) covers the **build** archetype. This file covers the other -three — **behaviour/process**, **credential**, **seeded** — plus the +four — **behaviour/process**, **credential**, **seeded**, **context** — plus the director-note vocabulary multi-turn cases rely on. Field-level docs live in the eval [README](../../../packages/@n8n/instance-ai/evaluations/README.md); this is the opinionated *how* and the traps. @@ -678,3 +678,139 @@ build". A seeded thread starts with an empty sandbox, so the agent re-reads the workflow from the database and re-derives SDK source a real resumed session would still have on disk. The bias is *harder* than reality, so those numbers read worse for a reason that has nothing to do with the builder. + +--- + +## Context cases (long conversations, token cost, compaction) + +A context case asks whether the agent still works once the conversation is +long. Two questions, two mechanics. + +### The judge already has the token numbers + +Every case's judge context carries token ground truth, so no new field is +needed to grade cost: + +- Each transcript turn header reads + `Turn 2 (3 steps, 47446 tokens in [41220 cached], 812 tokens out)`. +- A **Total** block sums every LLM step across the whole build, main run and + resumes, plus a cache read/write split and a **Fixed overhead** line — the + opening step's input, which is instructions and tool schemas plus one user + message, so it reads as the floor cost of any turn on that instance. Use it + to separate "the prompt got bigger" from "this conversation got expensive". + +So `processExpectations` may reference consumption directly. Write them as +statements about *behaviour the numbers reveal*, not as thresholds: + +```json +"processExpectations": [ + "The agent answers the second question without calling get_node_details on the Notion node again — it read that schema in turn 1 and the answer needs nothing new.", + "Turn 2's input tokens are mostly cache reads, which is what an unchanged prefix looks like. A turn that re-reads the history from scratch shows a cache-read share near zero." +] +``` + +Absolute token budgets rot: a prompt edit moves every number and the case goes +red for a reason that has nothing to do with the agent. Grade the *shape* — +what got re-read, what came from cache, which turn is the expensive one. + +Two limits to know before you write a cost expectation: + +- It needs `N8N_INSTANCE_AI_RUN_DEBUG_ENABLED=true` on the instance under test. + Without it the blocks render `(no run debug captured)` and the expectation is + unjudgeable — it fails or passes at random. +- The capture hooks **only the orchestrator's own stream**. A workflow build + runs inside that loop, so a workflow case's numbers are its real cost. A + delegated **Agent** build (the `build-agent` tool) runs on a separate stream + with no hooks, so its tokens are missing from every block. Don't write a cost + expectation on an `data/agents/` case. + +### `requiresMemoryCompaction` — grade the post-compaction window + +Observational memory compacts a long thread: an Observer summarises it into +observation rows, the summary is rendered into the system prompt, and the early +turns are masked out of the agent's window. A case that tests whether a decision +*survives* that has a setup problem — production compacts at 30k tokens of +visible message content, which is a conversation too long to hand-author. + +Set the flag and the harness handles it: + +```json +"requiresMemoryCompaction": true +``` + +Two things happen, both per-thread, so no other case in the run is affected: + +1. The harness sends a low observer threshold with every turn of that thread + only: **1,000 tokens** of visible message content, the request field's floor. + The seed plus the live turns must cross it, or the observer never runs — a + short seed does not test memory, it reports **not judged** (below). Size the + seed generously; a few real turns of tool output cross 1,000 tokens easily. + The Observer's prompt, masking and cursor logic are the production ones at + any threshold. +2. After the build, the harness checks the premise actually held. If it did + not, every expectation on the case is reported **not judged** — an + `incomplete` verdict, excluded from scoring, not a red. + +The premise check is the point of the flag. Uncompacted, the raw early turns +are still in the window, so the agent answers off them and every expectation +passes for free — a green that tests nothing. + +The evidence is structural. `GET /rest/instance-ai/eval/threads/:threadId/memory` +returns the observation rows and the compaction cursor from their own tables. A +cursor means the observer ran and everything up to `lastObservedMessageId` is +masked out; the rows are what replaced it. Missing either, the case had nothing +to test and is reported not judged. Nothing reads the system prompt, so this +needs **no debug flag** and a prompt or SDK rename cannot quietly turn "never +compacted" into the answer. + +The judge never sees the premise check — it grades the conversation, not whether +the harness configured the scenario. A misconfigured lane must not read as a +quality regression. + +### Grade the summary, not just the reply + +The judge *does* see the observation rows, as ground truth: + +``` +## Observational memory after compaction (ground truth — do not recount) + +- [CRITICAL] User ruled OUT the native Slack node; will post through the HTTP + Request node instead, because the workspace is pinned to a legacy Slack API + version. Do not revisit. +- [CRITICAL] Alert fires when queue depth goes above 4700 items. +``` + +So write expectations at **both** layers: + +```json +"processExpectations": [ + "The observational-memory block retains the alert threshold as 4700 queued items. A summary that keeps the topic but loses the number is the specific failure this looks for.", + "In its reply to the final turn, the agent gives the alert threshold as 4700 queued items." +] +``` + +The first grades the memory system, the second grades the agent. Grading only +the reply is weaker than it looks: the agent can answer correctly from the +unmasked tail, or guess well, and the case goes green while memory dropped the +detail. The `[MARKER]` is the Observer's own priority label, so "retained as +CRITICAL" is checkable too. + +### Writing the conversation + +The shape that makes a compaction case real: + +1. **State the anchors once, in a `user` turn near the start of the seed**, and + never restate them. Anything the agent repeats later lands *after* the + cursor and survives for the wrong reason. +2. **Make live turn 1 off-topic.** Its answer must not re-surface the anchors + into the post-cursor window. Compaction runs at the end of it. +3. **Ask for the anchors back in live turn 2.** That turn runs against the + compacted context. +4. **Force the antecedent** — a lost anchor must produce a *confident wrong* + answer, not a vague one. "Posting goes through HTTP Request, not the native + Slack node" works because the Slack node is the obvious guess. "The + threshold is 4700" works because a guess reaches for 5000. See + [negative expectations need a forced antecedent](SKILL.md#core-principle-all-shapes). + +Tag these `context-consumption` and put them in a `context` dataset, so a +tier that needs run debug enabled can select them. diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index df928085e29..5931319a1c6 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/packages/@n8n/api-types/src/index.ts @@ -597,6 +597,8 @@ export type { InstanceAiRunDebugStep, InstanceAiRunDebugWorkflowCodeSnapshot, InstanceAiRunDebugResponse, + InstanceAiEvalThreadMemoryResponse, + InstanceAiEvalObservation, InstanceAiThreadDebugRunsResponse, InstanceAiRichMessagesResponse, InstanceAiMemoryTaskKind, 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 b03956d42bc..b1a74e066f3 100644 --- a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts +++ b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts @@ -1567,6 +1567,9 @@ export class InstanceAiSendMessageRequest extends Z.class({ mode: instanceAiBuildModeSchema.optional(), /** Pin a published prompt profile. Takes precedence over mode. */ promptVersion: z.string().trim().min(1).max(128).optional(), + /** Eval override: observer threshold for THIS thread, so driving compaction + * for one case does not lower it for every conversation on the instance. */ + observerThresholdTokens: z.number().int().min(1000).max(1_000_000).optional(), }) {} export class InstanceAiCorrectTaskRequest extends Z.class({ @@ -2671,6 +2674,22 @@ export interface InstanceAiEvalAgentSkippedFeature { reason: string; } +export interface InstanceAiEvalObservation { + marker: 'critical' | 'important' | 'info' | 'completion'; + text: string; + tokenCount: number; +} + +/** What observational memory holds for a thread. The rows themselves, not the rendered + * system prompt — a prompt or SDK rename must not read as "never compacted". */ +export interface InstanceAiEvalThreadMemoryResponse { + /** Live observations, oldest first. Empty when nothing has been observed. */ + observations: InstanceAiEvalObservation[]; + /** Null until the observer runs; `lastObservedMessageId` is the compaction + * cursor, so everything up to it is masked out of the agent's window. */ + cursor: { lastObservedMessageId: string; lastObservedAt: string } | null; +} + export interface InstanceAiEvalAgentExecutionResult { runId: string; /** The run completed without framework/model errors. Tool-level errors live on toolCalls[].error. */ diff --git a/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md b/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md index 86feb127343..b6ac43523d2 100644 --- a/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md +++ b/packages/@n8n/instance-ai/evaluations/ARCHITECTURE.md @@ -59,6 +59,57 @@ split them out of the old `runner.ts` monolith): - `harness/cleanup.ts` — `cleanupBuild`, per-case timeout policy, bounded concurrency, binary workflow checks and shared failure summaries. +## Expectation judging: context blocks and preconditions + +`build-expectations/verifier.ts` assembles one prompt per build. Everything the +judge may treat as fact goes in under a **"Ground truth — do not recount"** +heading, so a block is never something the judge re-derives from prose: + +- The transcript (`utils/conversation-text.ts`), with per-turn token usage + inlined in each turn header — step count, input tokens, cached share, output + tokens — joined to the turn by its `runIds` (one turn spans many runs, because + every resume emits its own `run-start`). +- A build-wide token **Total**, a cache read/write split, and a **Fixed + overhead** line (the opening step's input — instructions and tool schemas + plus one user message, before the thread had any history). +- The **observation rows** for the thread (markers + text), read from + `instance_ai_observations` — so an expectation can grade the compacted summary + itself, not just whether the reply happened to be right. +- The built workflow, tool traces, and rendered artifacts. + +The **token** numbers come from `RunDebugBuffer` snapshots, so they need +`N8N_INSTANCE_AI_RUN_DEBUG_ENABLED=true` on the instance under test; without it +those blocks render `(no run debug captured)`. The **memory** block does not — +it is a separate REST read, so a compaction case runs with the flag off. + +The buffer keys records by `runId` and hooks only the orchestrator's own stream +(`buildOrchestratorAgentStreamOptions` and its resume twin are the sole call +sites of `createRunDebugStepHooks`). A workflow build runs in that loop, so a +workflow case's totals are its whole cost. A delegated **Agent** build reuses +the same `runId` on a stream with no hooks, and a step carries no `agentId`, so +its tokens are absent and could not be attributed even if the hooks existed. +Per-agent cost attribution needs hooks on the builder stream plus an `agentId` +on `RunDebugStep` / `InstanceAiRunDebugStep`. + +`DEBUG_JUDGE_CONTEXT=` dumps the assembled prompt, which is the only +practical way to check a new block renders as intended. + +**Preconditions belong in the harness, not the prompt.** When a case's premise +can fail to materialise (`requiresMemoryCompaction` — no compaction cursor, or a +cursor with no observations), `run/build-orchestrator.ts` replaces the judge call +with `allFailVerdicts(…)`, producing `incomplete` verdicts that scoring +**excludes**. The judge never learns the premise was checked: it grades the +conversation, and a misconfigured lane must not read as a quality regression. +Same mechanism as `priorRunFailed`. + +**Read state from its own store, not from a rendering.** The premise check used +to regex an `` tag out of the debug snapshot's system prompt, +through a *display* helper. That is the same surface that silently emptied when +the AI SDK renamed `system` to `instructions` (#38887) — and it fails quietly, so +a rename would report every compaction case "not judged" forever while looking +fine. `GET /rest/instance-ai/eval/threads/:threadId/memory` serves the rows and +the cursor instead. + ## Where to add things | You want to… | Touch exactly | @@ -74,6 +125,15 @@ split them out of the old `runner.ts` monolith): - **`eval-results.json`** is ingested by the LangTracer dispatcher, which runs this CLI keyless per case. The exact field set is pinned by `__tests__/eval-results-dispatcher-contract.test.ts`. + + **The CLI is per case; the n8n instance is not.** The dispatcher never starts + n8n — it targets a long-lived container per slot, `restart: unless-stopped`, + booted once for the whole sweep and never reset between cases. The nightly + runs ~12 of them (runners x slots), so one n8n process serves dozens of cases + back to back. Isolation is per-case *user* (`run/lane-users.ts` + `provisionCaseBuildUser`), not per-case instance. Assume anything the backend + holds in memory outlives the case that created it, and size it for a + multi-hour process rather than one run. - **`eval-pr-comment.md`** is posted verbatim by CI. The comment uses an `### Instance AI Workflow Eval` or `### Instance AI Agent Eval` prefix. - **LangSmith feedback keys** (`scenario_pass`, `failure_category`, diff --git a/packages/@n8n/instance-ai/evaluations/README.md b/packages/@n8n/instance-ai/evaluations/README.md index df101b51a32..2dbcd018b61 100644 --- a/packages/@n8n/instance-ai/evaluations/README.md +++ b/packages/@n8n/instance-ai/evaluations/README.md @@ -744,7 +744,7 @@ The corpus lives in **LangTracer**. Workflow cases use the `baseline` suite. Age } ``` -`conversation` (≥1 turn, first must be `user`), plus `complexity` and `tags`, are required. `executionScenarios`, `description`, `triggerType`, `messageBudget`, `processExpectations`, `outcomeExpectations`, `credentials`, and `datasets` (default `["full"]`) are optional — but **a case must declare at least one `executionScenario`, or one process/outcome expectation** (a case that asserts nothing is rejected at load). A _build-only_ case omits `executionScenarios` and is graded by its `processExpectations`/`outcomeExpectations` plus the always-on workflow checks: the workflow is still built, only the mock-execution `successCriteria` pass is skipped. A turn’s `text` may be a string or an array of strings joined with newlines — handy for long stage directions. +`conversation` (≥1 turn, first must be `user`), plus `complexity` and `tags`, are required. `executionScenarios`, `description`, `triggerType`, `messageBudget`, `processExpectations`, `outcomeExpectations`, `credentials`, `requiresMemoryCompaction`, and `datasets` (default `["full"]`) are optional — but **a case must declare at least one `executionScenario`, or one process/outcome expectation** (a case that asserts nothing is rejected at load). A _build-only_ case omits `executionScenarios` and is graded by its `processExpectations`/`outcomeExpectations` plus the always-on workflow checks: the workflow is still built, only the mock-execution `successCriteria` pass is skipped. A turn’s `text` may be a string or an array of strings joined with newlines — handy for long stage directions. **One case = one LangSmith split**, named from the case slug (the LangTracer case name; for disk-loaded files, the filename without `.json`). Pick a slug you're happy to also use as a `--filter` target. @@ -812,6 +812,21 @@ writes anything. Keep these cases on disk until LangTracer stores descriptions. The existing `blank` field also needs server support. The export check catches a server that drops it after a write. +### `requiresMemoryCompaction` — grade the window after observational memory compacts + +```json +"requiresMemoryCompaction": true +``` + +For a case that is only meaningful once observational memory has compacted the thread — the Observer has written its observations and the early turns are masked out of the agent's window. Production compacts at 30k tokens of visible message content (`N8N_INSTANCE_AI_OBSERVER_MESSAGE_TOKENS`), which is a conversation too long to hand-author, so the flag does two things per-thread, leaving every other case in the run alone: + +1. The harness sends a low `observerThresholdTokens` override with **every** turn of that thread. The flag means "compact as soon as there is anything to compact" — the seed does not have to hit a token number. The Observer's prompt, masking and cursor logic are the production ones at any threshold. +2. After the build, the harness checks the premise held, and reports every expectation on the case **not judged** if it did not. That is an `incomplete` verdict — excluded from scoring, not a red. + +The check is the point of the flag: uncompacted, the raw early turns are still in the window, so the agent answers off them and every expectation passes for free. The evidence is structural — `GET /rest/instance-ai/eval/threads/:threadId/memory` returns the observation rows and the compaction cursor straight from `instance_ai_observations` / `instance_ai_observation_cursors`. A cursor means the observer ran and everything up to `lastObservedMessageId` is masked out; the rows are what replaced it. Both, or the case had nothing to test. Nothing parses the system prompt, so a prompt or SDK rename cannot quietly turn "never compacted" into the answer, and the case needs no debug flag. + +The judge never sees the premise check — it grades the conversation, not whether the harness configured the scenario. It *does* see the observation rows, as a ground-truth block, so an expectation can grade the summary itself ("memory retained the 4700 threshold") rather than only the agent's reply, which passes just as well on a lucky guess. Authoring guidance — where to state the anchors, why live turn 1 must be off-topic — is in the [skill's context case shape](../../../../.agents/skills/create-instance-ai-eval/case-shapes.md#context-cases-long-conversations-token-cost-compaction). + ### Seeded cases (conversation pre-seeding) A seeded case starts **mid-conversation**: prior history is restored into the build thread before the live turn, so the eval drives only the turn under test. Use it to replicate a real misbehaviour — restore the conversation up to the moment it went wrong, re-drive that turn, and assert what should happen instead. diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/build-expectations-verifier.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/build-expectations-verifier.test.ts index 3dcc76de253..1a664f9a86e 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/build-expectations-verifier.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/build-expectations-verifier.test.ts @@ -1,3 +1,4 @@ +import type { InstanceAiRunDebugResponse } from '@n8n/api-types'; import type { Mock, MockedFunction } from 'vitest'; import { vi } from 'vitest'; @@ -171,6 +172,164 @@ describe('verifyBuildExpectations', () => { expect(buildBlock.providerOptions).toEqual({}); }); + it('renders the build-wide token total when run debug is captured', async () => { + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + const runDebug: InstanceAiRunDebugResponse[] = [ + { + threadId: 't1', + runId: 'r1', + startedAt: 0, + workflowCode: [], + steps: [ + { stepNumber: 0, output: { usage: { inputTokens: 100, outputTokens: 20 } } }, + { stepNumber: 1, output: { usage: { inputTokens: 300, outputTokens: 40 } } }, + ], + }, + ]; + await verifyBuildExpectations(['expectation zero'], { transcript: TRANSCRIPT, runDebug }); + + const sentMessages = JSON.stringify(generate.mock.calls[0]?.[0]); + expect(sentMessages).toContain('Token usage totals'); + expect(sentMessages).toContain( + 'Total: 400 tokens in / 60 tokens out across 2 LLM steps, 1 run', + ); + expect(sentMessages).toContain('Cache: 0 tokens read / 0 tokens written'); + // The opening step alone, separable from what the turns added. + expect(sentMessages).toContain('Opening step: 100 input tokens on the first LLM call'); + }); + + it('counts OpenAI cache hits reported only through provider metadata', async () => { + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + const runDebug: InstanceAiRunDebugResponse[] = [ + { + threadId: 't1', + runId: 'run-1', + startedAt: 0, + workflowCode: [], + steps: [ + { + stepNumber: 0, + output: { + usage: { inputTokens: 100, outputTokens: 20 }, + providerMetadata: { openai: { cachedPromptTokens: 80 } }, + }, + }, + ], + }, + ]; + await verifyBuildExpectations(['expectation zero'], { transcript: TRANSCRIPT, runDebug }); + + const sentMessages = JSON.stringify(generate.mock.calls[0]?.[0]); + expect(sentMessages).toContain('Cache: 80 tokens read / 0 tokens written'); + }); + + it('sums cache read/write tokens across steps, from the nested inputTokenDetails shape', async () => { + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + const runDebug: InstanceAiRunDebugResponse[] = [ + { + threadId: 't1', + runId: 'r1', + startedAt: 0, + workflowCode: [], + steps: [ + { + stepNumber: 0, + output: { + usage: { + inputTokens: 100, + outputTokens: 20, + inputTokenDetails: { cacheReadTokens: 8000, cacheWriteTokens: 0 }, + }, + }, + }, + { + stepNumber: 1, + output: { + usage: { + inputTokens: 300, + outputTokens: 40, + inputTokenDetails: { cacheReadTokens: 0, cacheWriteTokens: 2500 }, + }, + }, + }, + ], + }, + ]; + await verifyBuildExpectations(['expectation zero'], { transcript: TRANSCRIPT, runDebug }); + + const sentMessages = JSON.stringify(generate.mock.calls[0]?.[0]); + expect(sentMessages).toContain('Cache: 8000 tokens read / 2500 tokens written'); + }); + + it('renders the observation rows so an expectation can grade the summary itself', async () => { + // The point of reading the rows: without them a memory case can only check the + // agent's reply, which passes just as well on a lucky guess. + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + await verifyBuildExpectations(['memory kept the node decision'], { + transcript: TRANSCRIPT, + threadMemory: { + observations: [ + { + marker: 'critical', + text: 'Posting via HTTP Request, not the Slack node', + tokenCount: 9, + }, + { marker: 'info', text: 'Timezone America/New_York', tokenCount: 4 }, + ], + cursor: { lastObservedMessageId: 'm137', lastObservedAt: '2020-01-01T00:00:00.000Z' }, + }, + }); + + const sent = JSON.stringify(generate.mock.calls[0]?.[0]); + expect(sent).toContain('Observational memory after compaction'); + expect(sent).toContain('[CRITICAL] Posting via HTTP Request, not the Slack node'); + expect(sent).toContain('[INFO] Timezone America/New_York'); + }); + + it('says so when memory never compacted, rather than showing an empty block', async () => { + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + await verifyBuildExpectations(['anything'], { + transcript: TRANSCRIPT, + threadMemory: { observations: [], cursor: null }, + }); + + expect(JSON.stringify(generate.mock.calls[0]?.[0])).toContain( + 'observational memory has not compacted this conversation', + ); + }); + + it('falls back to a placeholder when no run debug was captured', async () => { + const generate: GenerateMock = vi.fn().mockResolvedValue({ + structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, + }); + mockJudge(generate); + + await verifyBuildExpectations(['expectation zero'], { transcript: TRANSCRIPT }); + + const sentMessages = JSON.stringify(generate.mock.calls[0]?.[0]); + expect(sentMessages).toContain('(no run debug captured)'); + }); + it('sends only the workflow block when there is no artifact context', async () => { const generate: GenerateMock = vi.fn().mockResolvedValue({ structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] }, diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts index 29fe997efc7..c8f4f8919c6 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/build-orchestrator.test.ts @@ -97,10 +97,21 @@ function failedBuild(error: string): BuildResult { }; } -function makeLane(num: number, tracedBuild: LaneState['tracedBuild']): LaneState { +function makeLane( + num: number, + tracedBuild: LaneState['tracedBuild'], + // stashThreadMemory reads this for a case that asserts on compaction, so the + // stub must answer it. An Error makes the read reject, the way a failed request does. + threadMemory: unknown = { observations: [], cursor: null }, +): LaneState { return { runner: { - client: {} as unknown as N8nClient, + client: { + getThreadMemory: + threadMemory instanceof Error + ? vi.fn().mockRejectedValue(threadMemory) + : vi.fn().mockResolvedValue(threadMemory), + } as unknown as N8nClient, baseUrl: `http://lane${String(num)}.test`, preRunWorkflowIds: new Set(), preRunDataTableIds: new Set(), @@ -145,6 +156,7 @@ function makeDeps( transcriptByThreadId: new Map(), buildExpectationsByKey: new Map(), runDebugByThreadId: new Map(), + threadMemoryByThreadId: new Map(), agentContextByKey: new Map(), // The provider-outage backoff is minutes long in production — never slept here. sleep: vi.fn().mockResolvedValue(undefined), @@ -160,6 +172,9 @@ async function settleMicrotasks(): Promise { afterEach(() => { vi.unstubAllGlobals(); + // The module-level verifier/judge mocks are shared, and several tests assert + // they were NOT called — leaked calls make those pass or fail by test order. + vi.clearAllMocks(); }); describe('createBuildOrchestrator', () => { @@ -183,6 +198,217 @@ describe('createBuildOrchestrator', () => { ); }); + it("forwards the case's requiresMemoryCompaction to the build", async () => { + // Same invisible-to-tsc hazard: dropped, it never compacts and reads as an agent miss. + const tracedBuild = vi.fn().mockResolvedValue(okBuild()); + const orchestrator = createBuildOrchestrator( + makeDeps([makeLane(1, tracedBuild)], { + testCaseByFileSlug: new Map([['case-a', baseCase({ requiresMemoryCompaction: true })]]), + }), + ); + + await orchestrator.getOrBuild(0, 'case-a'); + + expect(tracedBuild).toHaveBeenCalledWith( + expect.objectContaining({ requiresMemoryCompaction: true }), + ); + }); + + describe('requiresMemoryCompaction premise', () => { + /** A build whose thread memory is whatever the endpoint returned. */ + function depsWithMemory(memory: unknown) { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true })); + return makeDeps( + [ + makeLane( + 1, + vi.fn().mockResolvedValue( + okBuild({ + threadId: 'thread-1', + transcript: [ + { + userMessage: 'remind me what we decided', + steps: [{ kind: 'agent-text', text: 'The HTTP Request node.' }], + }, + ], + }), + ), + memory, + ), + ], + { + testCaseByFileSlug: new Map([ + [ + 'case-a', + baseCase({ + requiresMemoryCompaction: true, + processExpectations: ['recalls the decision'], + }), + ], + ]), + }, + ); + } + + const OBSERVATION = { marker: 'critical', text: 'Posting via HTTP Request', tokenCount: 7 }; + const CURSOR = { lastObservedMessageId: 'm137', lastObservedAt: '2020-01-01T00:00:00.000Z' }; + + it('reports it unjudged when the observer never ran', async () => { + // No cursor: uncompacted, the raw turns are still in the window and every + // expectation passes for free — so this must never reach the judge. + const deps = depsWithMemory({ observations: [], cursor: null }); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].incomplete).toBe(true); + expect(verdicts?.[0].reason).toContain('never compacted'); + // framework_issue, not verification_gap: the judge was fine, our setup was + // not. LangTracer stores this verbatim, so the wrong label misdirects triage. + expect(verdicts?.[0].attribution).toBe('framework_issue'); + }); + + it('reports it unjudged when it compacted but kept nothing', async () => { + // A cursor with no rows masked the history and remembered none of it, so + // there is no summary for the case to assert on. + const deps = depsWithMemory({ observations: [], cursor: CURSOR }); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].incomplete).toBe(true); + expect(verdicts?.[0].attribution).toBe('framework_issue'); + }); + + it('reports it unjudged when the memory read failed', async () => { + // The stash swallows the error into undefined: "no evidence", never "it compacted". + const deps = depsWithMemory(new Error('memory read failed')); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].incomplete).toBe(true); + expect(verdicts?.[0].attribution).toBe('framework_issue'); + expect(vi.mocked(verifyBuildExpectations)).not.toHaveBeenCalled(); + }); + + it('judges it normally once the cursor and the observations are both there', async () => { + const deps = depsWithMemory({ observations: [OBSERVATION], cursor: CURSOR }); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].incomplete).toBeUndefined(); + expect(verdicts?.[0].pass).toBe(true); + expect(verdicts?.[0].attribution).toBeUndefined(); + // The rows reach the judge: a dropped `threadMemory` would still pass above. + expect(vi.mocked(verifyBuildExpectations)).toHaveBeenCalledWith( + ['recalls the decision'], + expect.objectContaining({ + threadMemory: { observations: [OBSERVATION], cursor: CURSOR }, + }), + ); + }); + + it('does not read thread memory for a case that does not assert on compaction', async () => { + // The read is a REST call plus two queries per build; only a case with the + // flag pays it. + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true })); + const lane = makeLane(1, vi.fn().mockResolvedValue(okBuild({ threadId: 'thread-1' }))); + await createBuildOrchestrator(makeDeps([lane])).getOrBuild(0, 'case-a'); + + expect(lane.runner.client.getThreadMemory).not.toHaveBeenCalled(); + }); + + it('reads thread memory for a case that does', async () => { + const deps = depsWithMemory({ observations: [OBSERVATION], cursor: CURSOR }); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + expect(deps.laneStates[0].runner.client.getThreadMemory).toHaveBeenCalledWith('thread-1'); + }); + + it('names the outage, not the seed, when the build died on infra', async () => { + // Memory never compacted because the provider was down, not because the seed + // was too short; the infra attribution has to win over the premise reason. + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true })); + vi.mocked(verifyBuildExpectations).mockResolvedValueOnce([ + { expectation: 'recalls the decision', pass: false, reason: 'no reply to judge' }, + ]); + const deps = makeDeps( + [ + makeLane( + 1, + vi.fn().mockResolvedValue( + okBuild({ + success: false, + providerOutage: 'Agent error: Overloaded; No output generated.', + threadId: 'thread-1', + transcript: [ + { + userMessage: 'remind me what we decided', + steps: [{ kind: 'agent-text', text: 'Working on it.' }], + }, + ], + }), + ), + { observations: [], cursor: null }, + ), + ], + { + testCaseByFileSlug: new Map([ + [ + 'case-a', + baseCase({ + requiresMemoryCompaction: true, + processExpectations: ['recalls the decision'], + }), + ], + ]), + }, + ); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].reason).not.toContain('never compacted'); + expect(verdicts?.[0].attribution).toBe('framework_issue'); + }); + + it('skips the premise check for a prebuilt workflow', async () => { + // No conversation ran, so there is nothing to compact; the outcome + // expectations grade the workflow and are judged as for any prebuilt case. + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true })); + const lane = makeLane(1, vi.fn().mockResolvedValue(okBuild())); + lane.runner.client = { + getWorkflow: vi.fn().mockResolvedValue({ + id: 'wf-123', + name: 'Prebuilt', + active: false, + versionId: 'v1', + nodes: [], + connections: {}, + }), + } as unknown as N8nClient; + const deps = makeDeps([lane], { + prebuiltManifest: { 'case-a': ['wf-123'] }, + testCaseByFileSlug: new Map([ + [ + 'case-a', + baseCase({ + requiresMemoryCompaction: true, + processExpectations: ['recalls the decision'], + outcomeExpectations: ['sends a digest'], + }), + ], + ]), + }); + await createBuildOrchestrator(deps).getOrBuild(0, 'case-a'); + + const verdicts = await deps.buildExpectationsByKey.get('0:case-a'); + expect(verdicts?.[0].incomplete).toBeUndefined(); + expect(verdicts?.[0].pass).toBe(true); + expect(vi.mocked(verifyBuildExpectations)).toHaveBeenCalledWith( + ['sends a digest'], + expect.anything(), + ); + }); + }); + it('forwards the case identity so the build can stamp its trace', async () => { // Same class of bug as credentialFixture above, different symptom: the // harness passes these to ensureThread as sourceContext, which is what @@ -413,6 +639,9 @@ describe('createBuildOrchestrator', () => { // `incomplete` is the ONLY thing that keeps a verdict out of the pass rate. expect(verdicts?.[0].incomplete).toBe(true); expect(verdicts?.[0].reason).toContain('premise is missing'); + // This branch bypasses `attribute`, so without stamping it here a failed unit + // ships with no owner at all. + expect(verdicts?.[0].attribution).toBe('framework_issue'); }); it('serves prebuilt workflows by fetching them, never invoking the builder', async () => { @@ -550,11 +779,7 @@ describe('expectation judging context', () => { }); describe('credential-setup check wiring', () => { - // This file has no global mock reset; without it the second test counts the - // first test's call. - beforeEach(() => { - vi.mocked(runCredentialSetupChecks).mockClear(); - }); + // Call counts between tests are reset by the file-level afterEach. const SECRET = 'sk-ant-api03-LEAKED-abcdefghijklmnop'; diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/build-workflow-observer-threshold.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/build-workflow-observer-threshold.test.ts new file mode 100644 index 00000000000..f17a8394524 --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/__tests__/build-workflow-observer-threshold.test.ts @@ -0,0 +1,113 @@ +import { vi } from 'vitest'; + +import type { N8nClient } from '../clients/n8n-client'; +import { buildWorkflow } from '../harness/build-workflow'; +import { runMultiTurnConversation } from '../harness/chat-loop'; +import type { EvalLogger } from '../harness/logger'; + +// Only the message send matters here; the loop and the proxy are stubbed as in +// build-workflow-conversation-seed. +vi.mock('../harness/chat-loop', () => ({ + SSE_SETTLE_DELAY_MS: 0, + startSseConnection: vi.fn().mockResolvedValue(undefined), + waitForAllActivity: vi.fn().mockResolvedValue(undefined), + runMultiTurnConversation: vi.fn().mockResolvedValue(undefined), + recordUserTurn: vi.fn(), +})); + +vi.mock('../utils/user-proxy', () => ({ + UserProxyLlm: class { + respondToConfirmation = vi.fn().mockResolvedValue({ approve: true }); + ingestEvents = vi.fn(); + decideFollowUp = vi.fn().mockResolvedValue({ kind: 'done' }); + getDecisionStats = vi.fn().mockReturnValue({}); + }, +})); + +vi.mock('../outcome/workflow-discovery', () => ({ + buildAgentOutcome: vi.fn().mockResolvedValue({ + workflowsCreated: [], + executionsRun: [], + dataTablesCreated: [], + finalText: 'done', + workflowJsons: [], + }), + extractWorkflowIdsFromMessages: vi.fn().mockReturnValue([]), +})); + +const silentLogger: EvalLogger = { + info: () => {}, + verbose: () => {}, + success: () => {}, + warn: () => {}, + error: () => {}, + isVerbose: false, +}; + +function makeClient(sendMessage: ReturnType): N8nClient { + return { + getPersonalProjectId: vi.fn().mockResolvedValue('project-1'), + ensureThread: vi.fn().mockResolvedValue(undefined), + setThreadCredentialAllowlist: vi.fn().mockResolvedValue(undefined), + sendMessage, + getThreadMessages: vi.fn().mockResolvedValue({ messages: [] }), + listWorkflows: vi.fn().mockResolvedValue([]), + } as unknown as N8nClient; +} + +const baseConfig = { + skipWorkflowChecks: true, + preRunWorkflowIds: new Set(), + claimedWorkflowIds: new Set(), + logger: silentLogger, +}; + +const singleTurn = [{ role: 'user' as const, text: 'build a thing' }]; +const twoTurns = [ + { role: 'user' as const, text: 'build a thing' }, + { role: 'user' as const, text: 'now add a filter' }, +]; + +/** `sendMessage(threadId, message, attachments, mode, promptVersion, handoff, threshold)`. */ +function thresholdSentBy(sendMessage: ReturnType): unknown { + return sendMessage.mock.calls[0]?.[6]; +} + +describe('buildWorkflow observer threshold', () => { + it('sends the compaction floor with the opening message when the case asks for it', async () => { + const sendMessage = vi.fn().mockResolvedValue({ runId: 'run-1' }); + await buildWorkflow({ + client: makeClient(sendMessage), + ...baseConfig, + conversation: singleTurn, + requiresMemoryCompaction: true, + }); + expect(thresholdSentBy(sendMessage)).toBe(1000); + }); + + it('sends no threshold otherwise, so the thread keeps the instance default', async () => { + const sendMessage = vi.fn().mockResolvedValue({ runId: 'run-1' }); + await buildWorkflow({ + client: makeClient(sendMessage), + ...baseConfig, + conversation: singleTurn, + }); + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(thresholdSentBy(sendMessage)).toBeUndefined(); + }); + + it('hands the threshold to the multi-turn loop as well', async () => { + const sendMessage = vi.fn().mockResolvedValue({ runId: 'run-1' }); + await buildWorkflow({ + client: makeClient(sendMessage), + ...baseConfig, + conversation: twoTurns, + requiresMemoryCompaction: true, + }); + // The opening turn goes out directly; the loop sends every later turn. + expect(thresholdSentBy(sendMessage)).toBe(1000); + expect(vi.mocked(runMultiTurnConversation)).toHaveBeenCalledWith( + expect.objectContaining({ observerThresholdTokens: 1000 }), + ); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/conversation-text.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/conversation-text.test.ts index dfdb49ec2a8..cd8a943b63f 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/conversation-text.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/conversation-text.test.ts @@ -1,3 +1,5 @@ +import type { InstanceAiRunDebugResponse } from '@n8n/api-types'; + import type { CaseSeed } from '../harness/schema'; import type { ConversationTurn, TranscriptTurn } from '../types'; import { @@ -8,6 +10,7 @@ import { perTurnToolCallCounts, transcriptAsText, userTurnsAsText, + usageTokens, } from '../utils/conversation-text'; describe('userTurnsAsText', () => { @@ -292,6 +295,79 @@ describe('transcriptAsText', () => { 'Q (single): Which service? [RocketChat / Zulip]', ); }); + + it('inlines a turn heading with tokens summed across its runs (main + resume)', () => { + const transcript: TranscriptTurn[] = [ + { + userMessage: 'build it', + steps: [{ kind: 'agent-text', text: 'Done.' }], + runIds: ['run-1', 'run-2'], + }, + ]; + const runDebug: InstanceAiRunDebugResponse[] = [ + { + threadId: 't1', + runId: 'run-1', + startedAt: 0, + workflowCode: [], + steps: [{ stepNumber: 0, output: { usage: { inputTokens: 100, outputTokens: 20 } } }], + }, + { + threadId: 't1', + runId: 'run-2', + startedAt: 1, + workflowCode: [], + steps: [{ stepNumber: 0, output: { usage: { inputTokens: 300, outputTokens: 40 } } }], + }, + ]; + const text = transcriptAsText(transcript, runDebug); + expect(text).toContain('### Turn 1 (2 steps, 400 tokens in, 60 tokens out)'); + }); + + it('adds a cached clause to the heading only when the provider reported cache reads', () => { + const transcript: TranscriptTurn[] = [ + { userMessage: 'follow-up', steps: [], runIds: ['run-1'] }, + { userMessage: 'another', steps: [], runIds: ['run-2'] }, + ]; + const runDebug: InstanceAiRunDebugResponse[] = [ + { + threadId: 't1', + runId: 'run-1', + startedAt: 0, + workflowCode: [], + steps: [ + { + stepNumber: 0, + output: { + usage: { + inputTokens: 60000, + outputTokens: 500, + inputTokenDetails: { cacheReadTokens: 58000, cacheWriteTokens: 0 }, + }, + }, + }, + ], + }, + { + threadId: 't1', + runId: 'run-2', + startedAt: 1, + workflowCode: [], + steps: [{ stepNumber: 0, output: { usage: { inputTokens: 900, outputTokens: 30 } } }], + }, + ]; + const text = transcriptAsText(transcript, runDebug); + expect(text).toContain('### Turn 1 (1 step, 60000 tokens in [58000 cached], 500 tokens out)'); + // No cache reads reported for run-2 — no clause, rather than "[0 cached]". + expect(text).toContain('### Turn 2 (1 step, 900 tokens in, 30 tokens out)'); + }); + + it('leaves the heading plain when no run-debug matches the turn', () => { + const transcript: TranscriptTurn[] = [ + { userMessage: 'build it', steps: [{ kind: 'agent-text', text: 'Done.' }] }, + ]; + expect(transcriptAsText(transcript)).toContain('### Turn 1\n'); + }); }); describe('perTurnToolCallCounts', () => { @@ -359,3 +435,32 @@ describe('lastAgentText', () => { expect(text.length).toBeLessThan(answer.length + 6000); }); }); + +describe('usageTokens', () => { + it('reads cache tokens from the AI SDK details when present', () => { + expect( + usageTokens({ + inputTokens: 100, + outputTokens: 5, + inputTokenDetails: { cacheReadTokens: 80, cacheWriteTokens: 10 }, + }), + ).toEqual({ input: 100, output: 5, cacheRead: 80, cacheWrite: 10 }); + }); + + it("falls back to OpenAI's cachedPromptTokens, the way toTokenUsage does", () => { + // OpenAI reports cache hits only through provider metadata; without this the + // judge reads "0 tokens read" on a run that cached most of its prompt. + expect( + usageTokens({ inputTokens: 100, outputTokens: 5 }, { openai: { cachedPromptTokens: 80 } }), + ).toEqual({ input: 100, output: 5, cacheRead: 80, cacheWrite: 0 }); + }); + + it('prefers the SDK details over the provider fallback', () => { + expect( + usageTokens( + { inputTokens: 100, outputTokens: 5, inputTokenDetails: { cacheReadTokens: 30 } }, + { openai: { cachedPromptTokens: 80 } }, + ).cacheRead, + ).toBe(30); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/n8n-client-thread-memory.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/n8n-client-thread-memory.test.ts new file mode 100644 index 00000000000..46c3d899014 --- /dev/null +++ b/packages/@n8n/instance-ai/evaluations/__tests__/n8n-client-thread-memory.test.ts @@ -0,0 +1,75 @@ +import { afterEach, vi } from 'vitest'; + +import { N8nClient } from '../clients/n8n-client'; + +/** + * The compaction premise check reads memory through `getThreadMemory`, and the + * harness drives compaction through `sendMessage`'s threshold field. Both are + * mocked everywhere else, so this is the only place the wire shape is pinned. + */ + +function mockFetch(body: unknown = { data: {} }) { + const fn = vi.fn((_url: string | URL, _init?: RequestInit) => ({ + ok: true, + status: 200, + headers: { get: () => null }, + json: () => body, + text: () => '', + })); + vi.stubGlobal('fetch', fn); + return fn; +} + +afterEach(() => vi.unstubAllGlobals()); + +function requestOf(fn: ReturnType): { path: string; init: RequestInit } { + const [url, init] = fn.mock.calls[0] as [string, RequestInit]; + return { path: url.replace('http://n8n.test', ''), init }; +} + +describe('N8nClient.getThreadMemory', () => { + it('reads the eval memory route and unwraps the REST envelope', async () => { + const memory = { + observations: [{ marker: 'critical', text: 'Posting via HTTP Request', tokenCount: 7 }], + cursor: { lastObservedMessageId: 'm137', lastObservedAt: '2020-01-01T00:00:00.000Z' }, + }; + const fetchMock = mockFetch({ data: memory }); + + const result = await new N8nClient('http://n8n.test').getThreadMemory('thread-1'); + + expect(result).toEqual(memory); + const { path, init } = requestOf(fetchMock); + expect(path).toBe('/rest/instance-ai/eval/threads/thread-1/memory'); + expect(init.method).toBe('GET'); + }); +}); + +describe('N8nClient.sendMessage observer threshold', () => { + it('sends the threshold when the harness asks for compaction', async () => { + const fetchMock = mockFetch({ data: { runId: 'run-1' } }); + + await new N8nClient('http://n8n.test').sendMessage( + 'thread-1', + 'build', + undefined, + 'default', + undefined, + undefined, + 1000, + ); + + const { path, init } = requestOf(fetchMock); + expect(path).toBe('/rest/instance-ai/chat/thread-1'); + expect(JSON.parse(init.body as string)).toMatchObject({ observerThresholdTokens: 1000 }); + }); + + it('omits the field otherwise, so the instance default applies', async () => { + const fetchMock = mockFetch({ data: { runId: 'run-1' } }); + + await new N8nClient('http://n8n.test').sendMessage('thread-1', 'build'); + + expect(JSON.parse(requestOf(fetchMock).init.body as string)).not.toHaveProperty( + 'observerThresholdTokens', + ); + }); +}); diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/transcript-from-events.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/transcript-from-events.test.ts index 468ec16af9d..a847d08edcb 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/transcript-from-events.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/transcript-from-events.test.ts @@ -42,6 +42,30 @@ describe('buildTranscriptFromEvents', () => { ]); }); + describe('run-id tracking', () => { + it('collects every run-id (main run + any resumes) that fall inside one user turn', () => { + const turns = buildTranscriptFromEvents({ + events: [ + evt(USER_TURN_EVENT, { payload: { text: 'build it' } }), + evt('run-start', { runId: 'run-1' }), + evt('text-delta', { runId: 'run-1', text: 'Working on it.' }), + evt('run-start', { runId: 'run-2' }), + evt('text-delta', { runId: 'run-2', text: 'Done.' }), + ], + }); + expect(turns).toHaveLength(1); + expect(turns[0].runIds).toEqual(['run-1', 'run-2']); + }); + + it('omits runIds when no event carries one', () => { + const turns = buildTranscriptFromEvents({ + events: [RUN_START, evt('text-delta', { text: 'hi' })], + openingMessage: 'hi', + }); + expect(turns[0].runIds).toBeUndefined(); + }); + }); + describe('secret redaction', () => { it('redacts secret-shaped keys in tool-call args', () => { const turns = buildTranscriptFromEvents({ diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/user-execution.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/user-execution.test.ts index ef2f3f577b6..965a9a3e47b 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/user-execution.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/user-execution.test.ts @@ -92,6 +92,8 @@ describe('user execution during a conversation', () => { undefined, 'progressive', 'progressive@1', + undefined, + undefined, ); }); diff --git a/packages/@n8n/instance-ai/evaluations/build-expectations/verifier.ts b/packages/@n8n/instance-ai/evaluations/build-expectations/verifier.ts index 1e1a361015b..eacf6be07e6 100644 --- a/packages/@n8n/instance-ai/evaluations/build-expectations/verifier.ts +++ b/packages/@n8n/instance-ai/evaluations/build-expectations/verifier.ts @@ -1,3 +1,7 @@ +import type { + InstanceAiEvalThreadMemoryResponse, + InstanceAiRunDebugResponse, +} from '@n8n/api-types'; import type { Message } from '@n8n/agents'; import { buildAssertionsBlock, judgeExpectations } from './assertion-judge'; @@ -5,7 +9,12 @@ import { EPHEMERAL_CACHE } from '../../src/utils/eval-agents'; import type { WorkflowResponse } from '../clients/n8n-client'; import { buildWorkflowContextBlock } from '../harness/workflow-context'; import type { BuildExpectationResult, ConversationMetrics, TranscriptTurn } from '../types'; -import { perTurnToolCallCounts, transcriptAsText } from '../utils/conversation-text'; +import { + perTurnToolCallCounts, + sumUsage, + transcriptAsText, + usageTokens, +} from '../utils/conversation-text'; // Re-exported for import-site stability — cli/index.ts and runner.ts import it from here. export { allFailVerdicts } from './assertion-judge'; @@ -19,6 +28,10 @@ export interface BuildExpectationsInput { transcript: TranscriptTurn[]; workflowJson?: WorkflowResponse; metrics?: ConversationMetrics; + /** Per-step debug from the build; source of every token number below. */ + runDebug?: InstanceAiRunDebugResponse[]; + /** Observational memory for the thread: rows plus the compaction cursor. */ + threadMemory?: InstanceAiEvalThreadMemoryResponse; /** Rendered agent/config-eval sections (each with a "(no … produced)" fallback), appended * to the cached build context so outcome expectations can be judged against them. */ artifactContext?: string; @@ -53,15 +66,42 @@ export async function verifyBuildExpectations( }, { type: 'text', - text: buildConversationContext(expectations, build.transcript, build.metrics), + text: buildConversationContext( + expectations, + build.transcript, + build.metrics, + build.runDebug, + build.threadMemory, + ), }, ], }, ]; + await dumpJudgeContext(messages); + return await judgeExpectations(messages, expectations); } +/** Dump the assembled judge prompt to `DEBUG_JUDGE_CONTEXT`; nothing else shows it. */ +async function dumpJudgeContext(messages: Message[]): Promise { + const target = process.env.DEBUG_JUDGE_CONTEXT; + if (!target) return; + const text = messages + .flatMap((message) => + Array.isArray(message.content) + ? message.content.flatMap((part) => ('text' in part ? [part.text] : [])) + : [], + ) + .join('\n\n--- block ---\n\n'); + try { + const { appendFile } = await import('fs/promises'); + await appendFile(target, `\n\n===== JUDGE CALL =====\n\n${text}\n`); + } catch { + // Diagnostics must never fail a run. + } +} + // --------------------------------------------------------------------------- // Artifact assembly // --------------------------------------------------------------------------- @@ -70,6 +110,8 @@ function buildConversationContext( expectations: string[], transcript: TranscriptTurn[], metrics: ConversationMetrics | undefined, + runDebug: InstanceAiRunDebugResponse[] | undefined, + threadMemory: InstanceAiEvalThreadMemoryResponse | undefined, ): string { const metricsBlock = metrics ? `\`\`\`json\n${JSON.stringify(metrics, null, 2)}\n\`\`\`` @@ -77,7 +119,7 @@ function buildConversationContext( return [ '## Conversation transcript', '', - transcriptAsText(transcript), + transcriptAsText(transcript, runDebug), '', '## Conversation metrics (ground truth — do not recount)', '', @@ -87,6 +129,58 @@ function buildConversationContext( '', perTurnToolCallCounts(transcript), '', + '## Token usage totals (ground truth — do not recount)', + '', + tokenUsageTotals(runDebug), + '', + // Fetched only for a case that asserts on compaction. Absent, the section is + // left out rather than telling the judge memory was "not captured". + ...(threadMemory + ? [ + '## Observational memory after compaction (ground truth — do not recount)', + '', + observationLogBlock(threadMemory), + '', + ] + : []), buildAssertionsBlock(expectations), ].join('\n'); } + +/** What the agent remembers after compaction, so an expectation can grade the summary + * itself. Markers are the Observer's own priority labels. */ +function observationLogBlock(memory: InstanceAiEvalThreadMemoryResponse): string { + if (!memory.cursor) return '(observational memory has not compacted this conversation)'; + if (memory.observations.length === 0) return '(compacted, but no observations were kept)'; + return memory.observations + .map(({ marker, text }) => `- [${marker.toUpperCase()}] ${text}`) + .join('\n'); +} + +// --------------------------------------------------------------------------- +// Token usage totals +// --------------------------------------------------------------------------- + +/** Build-wide sum, the one number the turn headers don't carry. Orchestrator steps + * only, so a delegated `build-agent` leg is absent (workflow builds are not). */ +function tokenUsageTotals(runDebug: InstanceAiRunDebugResponse[] | undefined): string { + if (!runDebug || runDebug.length === 0) return '(no run debug captured)'; + + const { + input, + output, + cacheRead, + cacheWrite, + steps: stepCount, + } = sumUsage(runDebug.flatMap((run) => run.steps)); + + // Instructions + tool schemas + the first message. A seeded case also carries + // its restored history here, so this is the first call, not a history-free floor. + const opening = usageTokens(runDebug[0]?.steps[0]?.output?.usage).input; + const runWord = runDebug.length === 1 ? 'run' : 'runs'; + return [ + `Total: ${String(input)} tokens in / ${String(output)} tokens out across ${String(stepCount)} LLM steps, ${String(runDebug.length)} ${runWord}`, + `Cache: ${String(cacheRead)} tokens read / ${String(cacheWrite)} tokens written`, + `Opening step: ${String(opening)} input tokens on the first LLM call (instructions, tool schemas and the first message; a seeded case also carries its restored history here)`, + ].join('\n'); +} diff --git a/packages/@n8n/instance-ai/evaluations/cli/mcp-builder.ts b/packages/@n8n/instance-ai/evaluations/cli/mcp-builder.ts index 533229eec8b..301160b842d 100644 --- a/packages/@n8n/instance-ai/evaluations/cli/mcp-builder.ts +++ b/packages/@n8n/instance-ai/evaluations/cli/mcp-builder.ts @@ -187,6 +187,7 @@ export const MCP_BUILD_KEY_SUPPORT: Record< buildMode: 'orchestrator-only', promptVersion: 'orchestrator-only', allowUserExecution: 'orchestrator-only', + requiresMemoryCompaction: 'orchestrator-only', // Judged by the harness after the build (processExpectations are skipped for // transcript-less MCP builds there); declaring them needs no build-side setup. processExpectations: 'supported', diff --git a/packages/@n8n/instance-ai/evaluations/clients/n8n-client.ts b/packages/@n8n/instance-ai/evaluations/clients/n8n-client.ts index 69cfada6ce3..023e9d47b2d 100644 --- a/packages/@n8n/instance-ai/evaluations/clients/n8n-client.ts +++ b/packages/@n8n/instance-ai/evaluations/clients/n8n-client.ts @@ -16,6 +16,7 @@ import type { InstanceAiEvalAgentExecutionResult, InstanceAiEvalExecutionResult, InstanceAiRunDebugResponse, + InstanceAiEvalThreadMemoryResponse, InstanceAiThreadDebugRunsResponse, InstanceAiThreadStatusResponse, InstanceAiEvalSeedAgent, @@ -348,6 +349,7 @@ export class N8nClient { mode: InstanceAiBuildMode = 'default', promptVersion?: string, handoffContext?: InstanceAiHandoffContext, + observerThresholdTokens?: number, ): Promise<{ runId: string }> { const result = await this.fetch(`/rest/instance-ai/chat/${threadId}`, { method: 'POST', @@ -357,6 +359,8 @@ export class N8nClient { mode, ...(promptVersion ? { promptVersion } : {}), ...(handoffContext ? { context: handoffContext } : {}), + // Per-thread, so other cases in the suite keep the instance default. + ...(observerThresholdTokens ? { observerThresholdTokens } : {}), } satisfies InstanceAiSendMessageRequest, }); return this.unwrapRestData<{ runId: string }>(result); @@ -439,6 +443,16 @@ export class N8nClient { ); } + /** Live observations + the compaction cursor for a thread. */ + async getThreadMemory( + threadId: string, + timeoutMs?: number, + ): Promise { + return this.unwrapRestData( + await this.fetch(`/rest/instance-ai/eval/threads/${threadId}/memory`, { timeoutMs }), + ); + } + // -- Computer-use gateway (pairing + status) ----------------------------- /** diff --git a/packages/@n8n/instance-ai/evaluations/harness/build-workflow.ts b/packages/@n8n/instance-ai/evaluations/harness/build-workflow.ts index 597c684164a..df4f6b6125b 100644 --- a/packages/@n8n/instance-ai/evaluations/harness/build-workflow.ts +++ b/packages/@n8n/instance-ai/evaluations/harness/build-workflow.ts @@ -110,6 +110,10 @@ const DEFAULT_TIMEOUT_MS = 900_000; // Multi-turn driver — wires UserProxyLlm into runMultiTurnConversation // --------------------------------------------------------------------------- +/** The request field's floor, so the flag means "compact as soon as there is + * anything to compact" — a case sized for the 30k default is unauthorable. */ +const COMPACTION_OBSERVER_THRESHOLD_TOKENS = 1_000; + interface MultiTurnDriverConfig { client: N8nClient; threadId: string; @@ -118,6 +122,7 @@ interface MultiTurnDriverConfig { /** Resolved wire value sent with every message (see `resolveEvalBuildMode`). */ buildMode?: InstanceAiBuildMode; promptVersion?: string; + observerThresholdTokens?: number; allowUserExecution?: boolean; beforeUserExecution?: (deadline: number) => Promise; events: CapturedEvent[]; @@ -212,6 +217,7 @@ async function driveMultiTurnConversation( config.buildMode, config.promptVersion, config.openingHandoffContext, + config.observerThresholdTokens, ); await runMultiTurnConversation({ @@ -227,6 +233,7 @@ async function driveMultiTurnConversation( proxyResponses: config.proxyResponses, buildMode: config.buildMode, promptVersion: config.promptVersion, + observerThresholdTokens: config.observerThresholdTokens, allowUserExecution: config.allowUserExecution, beforeUserExecution: config.beforeUserExecution, }); @@ -462,6 +469,7 @@ export interface BuildWorkflowConfig { /** Case-declared build style; resolved via `resolveEvalBuildMode` (absent → default). */ buildMode?: WorkflowTestCase['buildMode']; promptVersion?: string; + requiresMemoryCompaction?: boolean; allowUserExecution?: boolean; /** Credentials this build should see (created for real, view pinned to them). */ credentials?: TestCaseCredential[]; @@ -529,6 +537,9 @@ export function workflowExpectedForCase( export async function buildWorkflow(config: BuildWorkflowConfig): Promise { const { client, logger } = config; const { buildMode, promptVersion } = resolveEvalPromptSettings(config); + const observerThresholdTokens = config.requiresMemoryCompaction + ? COMPACTION_OBSERVER_THRESHOLD_TOKENS + : undefined; const threadId = crypto.randomUUID(); const startTime = Date.now(); const timeoutMs = config.timeoutMs ?? DEFAULT_TIMEOUT_MS; @@ -1089,6 +1100,7 @@ export async function buildWorkflow(config: BuildWorkflowConfig): Promise { @@ -348,6 +349,9 @@ export async function runMultiTurnConversation(config: MultiTurnConfig): Promise undefined, config.buildMode, config.promptVersion, + undefined, + // Re-sent per message: the backend clears an omitted override. + config.observerThresholdTokens, ); } catch (error: unknown) { const msg = error instanceof Error ? error.message : String(error); diff --git a/packages/@n8n/instance-ai/evaluations/harness/schema.ts b/packages/@n8n/instance-ai/evaluations/harness/schema.ts index 55b3d97d8f6..b8d7b53ab0d 100644 --- a/packages/@n8n/instance-ai/evaluations/harness/schema.ts +++ b/packages/@n8n/instance-ai/evaluations/harness/schema.ts @@ -174,6 +174,9 @@ const evalTestCaseObjectSchema = z buildMode: z.enum(['progressive', 'default']).optional(), promptVersion: z.string().trim().min(1).max(128).optional(), allowUserExecution: z.boolean().optional(), + /** Harness lowers the observer threshold for this thread, and reports the case + * not judged if compaction never ran. */ + requiresMemoryCompaction: z.boolean().optional(), /** Optional NL assertions about the build CONVERSATION (process: clarifications, push-back, * ordering). LLM-judged from the transcript, so skipped in prebuilt/MCP runs. Counted as units. */ processExpectations: z.array(z.string().min(1)).optional(), diff --git a/packages/@n8n/instance-ai/evaluations/outcome/transcript-from-events.ts b/packages/@n8n/instance-ai/evaluations/outcome/transcript-from-events.ts index b7972de4ef2..13dbd90fd62 100644 --- a/packages/@n8n/instance-ai/evaluations/outcome/transcript-from-events.ts +++ b/packages/@n8n/instance-ai/evaluations/outcome/transcript-from-events.ts @@ -148,6 +148,7 @@ function buildTurn( } flushText(); + const runIds = collectRunIds(events); // The transcript leaves the machine via eval-results.json — content-scrub // the user message HERE, the boundary both capture paths share (the // marker path's extracted turn text AND the legacy fallback's raw @@ -155,9 +156,19 @@ function buildTurn( return { userMessage: userMessage === undefined ? undefined : redactSecretsInText(userMessage), steps, + ...(runIds.length > 0 ? { runIds } : {}), }; } +/** Every distinct `runId` in this turn — the main run plus any resumes. */ +function collectRunIds(events: CapturedEvent[]): string[] { + return [ + ...new Set( + events.map((e) => getString(e.data, 'runId')).filter((id): id is string => Boolean(id)), + ), + ]; +} + interface ToolOutcome { result?: unknown; error?: string; diff --git a/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts b/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts index 7aa37ad75f8..f0c5423ac1b 100644 --- a/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts +++ b/packages/@n8n/instance-ai/evaluations/run/build-orchestrator.ts @@ -8,7 +8,10 @@ // `LaneState`, so tracing stays a caller concern. // --------------------------------------------------------------------------- -import type { InstanceAiRunDebugResponse } from '@n8n/api-types'; +import type { + InstanceAiEvalThreadMemoryResponse, + InstanceAiRunDebugResponse, +} from '@n8n/api-types'; import { sleep } from '@n8n/utils/sleep'; import type { LaneAllocator } from './lane-allocator'; @@ -123,6 +126,9 @@ export type BuildArgs = Pick< // callback's parameter type, so tsc cannot catch a dropped field here; the // orchestrator test pins it. | 'credentialFixture' + // Same hazard: dropped, the case runs with the instance's own observer + // threshold, never compacts, and reads as an agent miss. + | 'requiresMemoryCompaction' > & { timeoutMs: number; /** Which case this build is, and which repeat of it. Not used by the build @@ -298,6 +304,7 @@ export interface BuildOrchestratorDeps { transcriptByThreadId: Map; buildExpectationsByKey: Map>; runDebugByThreadId: Map>; + threadMemoryByThreadId: Map>; agentContextByKey: Map>; /** Injectable delay for the provider-outage retry backoff — tests pass a no-op. */ sleep?: (ms: number) => Promise; @@ -326,6 +333,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche transcriptByThreadId, buildExpectationsByKey, runDebugByThreadId, + threadMemoryByThreadId, agentContextByKey, } = deps; const delay = deps.sleep ?? sleep; @@ -392,6 +400,28 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche agentContextByKey.set(key, fetchAgentScenarioContext(client, agentRef, logger)); } + /** Only for a case that asserts on compaction: the read is a REST call plus two + * queries per build, and the nightly suite runs dozens of builds back to back + * on one instance. Undefined on failure — the premise check reads that as "no + * evidence", never as "it compacted". */ + function stashThreadMemory( + client: N8nClient, + build: BuildResult, + wanted: boolean | undefined, + ): void { + if (!wanted || !build.threadId) return; + const threadId = build.threadId; + threadMemoryByThreadId.set( + threadId, + client.getThreadMemory(threadId).catch((error: unknown) => { + logger.warn( + ` Dropped thread memory for ${threadId}: ${error instanceof Error ? error.message : String(error)}`, + ); + return undefined; + }), + ); + } + function stashRunDebug(client: N8nClient, build: BuildResult): void { if (!build.threadId) return; // Re-read from n8n AFTER the build was scrubbed, so it arrives raw and the @@ -446,7 +476,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche allFailVerdicts( collectExpectations(testCase), `not judged — prior run staging did not land, so the case premise is missing: ${build.priorRunFailed}`, - ), + ).map((verdict) => ({ ...verdict, attribution: 'framework_issue' as const })), ), ); return; @@ -484,7 +514,12 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche // and reshape's side band) then carry the same verdict (TRUST-375). const infraFailed = buildFailedOnInfra(build); const attribute = (verdicts: BuildExpectationResult[]): BuildExpectationResult[] => - verdicts.map((v) => ({ ...v, attribution: attributionForExpectation(v, infraFailed) })); + verdicts.map((v) => ({ + // An attribution already set is a decision the caller made with more + // context than this closure has; don't overwrite it. + ...v, + attribution: v.attribution ?? attributionForExpectation(v, infraFailed), + })); // The lane's deterministic verdicts ride along on EVERY path, including the // unjudged one: they describe what the run actually did to the provider and // to n8n, which stays true whether or not the author expectations got judged. @@ -507,11 +542,39 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche buildExpectationsByKey.set( key, withInjected( - (async () => - await verifyBuildExpectations(expectations, { + (async () => { + const runDebug = build.threadId + ? await runDebugByThreadId.get(build.threadId) + : undefined; + const threadMemory = build.threadId + ? await threadMemoryByThreadId.get(build.threadId) + : undefined; + // Premise, not an expectation. `framework_issue`, not `verification_gap`: + // the judge was fine, the harness failed to set the scenario up, and + // that distinction is what tells triage where to look. + // A prebuilt run has no conversation to compact: only its outcome + // expectations are judged, and those grade the workflow, not memory. + // A build that died on infra never reached compaction either; `attribute` + // below names the outage, where this reason would blame the seed. The read + // itself is settled: `waitForAllActivity` polls the thread's memory tasks to + // a terminal state before the build returns (`waitForMemoryTasks`). + if ( + testCase.requiresMemoryCompaction && + !isPrebuilt && + !infraFailed && + !memoryWasCompacted(threadMemory) + ) { + return allFailVerdicts(expectations, NOT_COMPACTED_REASON).map((verdict) => ({ + ...verdict, + attribution: 'framework_issue' as const, + })); + } + return await verifyBuildExpectations(expectations, { transcript, workflowJson: build.workflowJsons[0], metrics: build.conversationMetrics, + runDebug, + threadMemory, // Rendered non-workflow artifacts (agent AND config-eval), sectioned // with "(no produced)" fallbacks, so outcome expectations can // judge artifact existence, absence and content — parity with the @@ -521,7 +584,8 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche client, logger, }), - }))() + }); + })() .catch((error: unknown) => allFailVerdicts( expectations, @@ -583,8 +647,12 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche stashTranscript(build); // isPrebuilt=true: MCP builds have no build transcript, so only // outcome expectations are judged (against the workflow), like prebuilt. - stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); + // Ordered before stashBuildExpectations so runDebugByThreadId already has + // this build's promise stashed by the time that call reads it. stashRunDebug(lane.runner.client, build); + // No premise check on this path (isPrebuilt), so memory is not read. + stashThreadMemory(lane.runner.client, build, false); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); if (build.success && !build.workflowChecks) { build.workflowChecks = await runWorkflowChecks({ workflow: build.workflowJsons[0], @@ -611,8 +679,12 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche const buildDurationMs = Date.now() - start; buildDurations.set(key, buildDurationMs); stashTranscript(build); - stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); + // Ordered before stashBuildExpectations so runDebugByThreadId already has + // this build's promise stashed by the time that call reads it. stashRunDebug(lane.runner.client, build); + // No premise check on this path (isPrebuilt), so memory is not read. + stashThreadMemory(lane.runner.client, build, false); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, true); if (build.success && !build.workflowChecks) { // No transcript in prebuilt mode, but the authored conversation still // carries the user's request — feed it so prompt-aware checks (e.g. @@ -653,6 +725,7 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche messageBudget: entry.messageBudget, buildMode: entry.buildMode, promptVersion: entry.promptVersion, + requiresMemoryCompaction: entry.requiresMemoryCompaction, allowUserExecution: entry.allowUserExecution, credentials: entry.credentials, seed: entry.seed, @@ -685,8 +758,11 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche buildDurations.set(key, buildDurationMs); stashTranscript(build); stashAgentContext(key, lane.runner.client, build); - stashBuildExpectations(key, fileSlug, lane.runner.client, build, false); + // Ordered before stashBuildExpectations so runDebugByThreadId already has + // this build's promise stashed by the time that call reads it. stashRunDebug(lane.runner.client, build); + stashThreadMemory(lane.runner.client, build, entry.requiresMemoryCompaction); + stashBuildExpectations(key, fileSlug, lane.runner.client, build, false); logger.info( `[lane ${String(lane.laneNum)}] built ${fileSlug} (iteration ${String(iteration)}) thread=${build.threadId ?? 'none'} success=${String(build.success)}`, ); @@ -715,3 +791,14 @@ export function createBuildOrchestrator(deps: BuildOrchestratorDeps): BuildOrche return { getOrBuild, buildCache, orphanedBuilds, buildDurations }; } + +const NOT_COMPACTED_REASON = + 'not judged — observational memory never compacted this thread, so the case premise is ' + + 'absent. Either the seed is too short to cross the observer threshold, or the compaction ' + + 'the harness asked for never reached the run.'; + +/** A cursor means the observer ran and everything up to it is masked out; the rows are + * what replaced it. Both, or the case had nothing to test. */ +function memoryWasCompacted(memory: InstanceAiEvalThreadMemoryResponse | undefined): boolean { + return Boolean(memory?.cursor) && (memory?.observations.length ?? 0) > 0; +} diff --git a/packages/@n8n/instance-ai/evaluations/run/eval-session.ts b/packages/@n8n/instance-ai/evaluations/run/eval-session.ts index 948026b6bfb..51e6926a378 100644 --- a/packages/@n8n/instance-ai/evaluations/run/eval-session.ts +++ b/packages/@n8n/instance-ai/evaluations/run/eval-session.ts @@ -8,7 +8,10 @@ // shape stays a driver concern and the assembly exists exactly once. // --------------------------------------------------------------------------- -import type { InstanceAiRunDebugResponse } from '@n8n/api-types'; +import type { + InstanceAiEvalThreadMemoryResponse, + InstanceAiRunDebugResponse, +} from '@n8n/api-types'; import { createBuildOrchestrator, @@ -115,6 +118,10 @@ export function createEvalSession(config: EvalSessionConfig): EvalSession { // Fired during getOrBuild, awaited in resolveSideBand. const buildExpectationsByKey = new Map>(); const runDebugByThreadId = new Map>(); + const threadMemoryByThreadId = new Map< + string, + Promise + >(); // Rows carry only per-scenario fields. The build-side fields (conversation, // expectations, declared credentials) are sourced locally, keyed by fileSlug. @@ -147,6 +154,7 @@ export function createEvalSession(config: EvalSessionConfig): EvalSession { messageBudget: buildArgs.messageBudget, buildMode: buildArgs.buildMode, promptVersion: buildArgs.promptVersion, + requiresMemoryCompaction: buildArgs.requiresMemoryCompaction, allowUserExecution: buildArgs.allowUserExecution, credentials: buildArgs.credentials, seed: buildArgs.seed, @@ -251,6 +259,7 @@ export function createEvalSession(config: EvalSessionConfig): EvalSession { transcriptByThreadId, buildExpectationsByKey, runDebugByThreadId, + threadMemoryByThreadId, agentContextByKey, }); diff --git a/packages/@n8n/instance-ai/evaluations/types.ts b/packages/@n8n/instance-ai/evaluations/types.ts index 1786807157e..03c7b848444 100644 --- a/packages/@n8n/instance-ai/evaluations/types.ts +++ b/packages/@n8n/instance-ai/evaluations/types.ts @@ -233,6 +233,8 @@ export interface WorkflowTestCase { promptVersion?: string; /** Enable the user-run action for credential-free execution cases. */ allowUserExecution?: boolean; + /** Harness drives compaction, and reports the case not judged if it never ran. */ + requiresMemoryCompaction?: boolean; /** Optional NL assertions about the build CONVERSATION (process: clarifications, push-back, * ordering). LLM-judged from the transcript; requires a transcript, so skipped in * prebuilt/MCP runs. Counted toward the per-case + headline pass rate alongside scenarios. */ @@ -375,6 +377,8 @@ export interface TranscriptTurn { /** True for turns restored from a conversation seed — context that predates * the evaluated run, as opposed to behaviour captured live. */ seeded?: boolean; + /** Main run plus any resumes — the join key for this turn's token usage. */ + runIds?: string[]; } /** One ordered step within a turn: a slice of agent narration or a tool interaction. */ diff --git a/packages/@n8n/instance-ai/evaluations/utils/conversation-text.ts b/packages/@n8n/instance-ai/evaluations/utils/conversation-text.ts index b873b7a0188..161886c5e7d 100644 --- a/packages/@n8n/instance-ai/evaluations/utils/conversation-text.ts +++ b/packages/@n8n/instance-ai/evaluations/utils/conversation-text.ts @@ -1,3 +1,4 @@ +import type { InstanceAiRunDebugResponse } from '@n8n/api-types'; import { isRecord } from '@n8n/utils/is-record'; import type { CaseSeed } from '../harness/schema'; @@ -110,12 +111,104 @@ export function conversationUserTurnsAsText( return turns.map((text, i) => `Turn ${String(i + 1)}: ${text}`).join('\n\n'); } -/** Full transcript (agent narration + tool interactions, in order) as plain text for LLM-judged checks. */ -export function transcriptAsText(transcript: TranscriptTurn[]): string { +export interface UsageTokens { + input: number; + output: number; + cacheRead: number; + cacheWrite: number; +} + +function numberOr0(value: unknown): number { + return typeof value === 'number' ? value : 0; +} + +/** A step's usage is one number for the whole LLM call — real granularity bottoms + * out here, not at the individual tool-call/message level. `inputTokens`/`promptTokens` + * naming varies by SDK version, matching `parseUsageSummary`'s own fallback; cache + * fields nest under `inputTokenDetails` per the AI SDK's `LanguageModelUsage` shape. */ +export function usageTokens(usage: unknown, providerMetadata?: unknown): UsageTokens { + if (!isRecord(usage)) return { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }; + const details = isRecord(usage.inputTokenDetails) ? usage.inputTokenDetails : undefined; + const cacheRead = numberOr0(details?.cacheReadTokens); + const cacheWrite = numberOr0(details?.cacheWriteTokens); + return { + input: numberOr0(usage.inputTokens ?? usage.promptTokens), + output: numberOr0(usage.outputTokens ?? usage.completionTokens), + // OpenAI's provider reports cache hits only in `providerMetadata` and leaves + // the SDK's `inputTokenDetails` at zero, so an explicit zero there is not + // evidence of no cache. The provider figure is read only when the SDK fields + // carry nothing, the same fallback the runtime's `toTokenUsage` + // (`@n8n/agents`, `runtime/streaming/stream.ts`) applies, so the two agree. + cacheRead: cacheRead || cacheWrite ? cacheRead : openAiCachedPromptTokens(providerMetadata), + cacheWrite, + }; +} + +function openAiCachedPromptTokens(providerMetadata: unknown): number { + if (!isRecord(providerMetadata) || !isRecord(providerMetadata.openai)) return 0; + return numberOr0(providerMetadata.openai.cachedPromptTokens); +} + +/** A run's usage plus its step count — totals sum ACROSS steps, so without the count + * "76033 tokens in" reads as one huge prompt when it was two ordinary ones. */ +export type RunUsage = UsageTokens & { steps: number }; + +/** The four usage fields summed over LLM steps, with the step count. The one + * accumulator: the turn headings and the verifier's totals both read it, so the + * two cannot drift. */ +export function sumUsage(steps: InstanceAiRunDebugResponse['steps']): RunUsage { + const total: RunUsage = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, steps: 0 }; + for (const step of steps) { + const usage = usageTokens(step.output?.usage, step.output?.providerMetadata); + total.input += usage.input; + total.output += usage.output; + total.cacheRead += usage.cacheRead; + total.cacheWrite += usage.cacheWrite; + total.steps++; + } + return total; +} + +/** Token usage per run-id, the join target for a turn's `runIds`. */ +function usageByRunId(runDebug: InstanceAiRunDebugResponse[] | undefined): Map { + return new Map((runDebug ?? []).map((run) => [run.runId, sumUsage(run.steps)])); +} + +/** `" (N steps, N tokens in [C cached], M tokens out)"` for a turn heading. The cached + * clause is omitted when the provider reported none, and '' when nothing matched — + * a plain heading beats a false zero. */ +function turnUsageSuffix(turn: TranscriptTurn, usageMap: Map): string { + if (!turn.runIds || turn.runIds.length === 0) return ''; + let input = 0; + let output = 0; + let cacheRead = 0; + let steps = 0; + let matched = false; + for (const runId of turn.runIds) { + const usage = usageMap.get(runId); + if (!usage) continue; + matched = true; + input += usage.input; + output += usage.output; + cacheRead += usage.cacheRead; + steps += usage.steps; + } + if (!matched) return ''; + const cached = cacheRead > 0 ? ` [${String(cacheRead)} cached]` : ''; + const stepWord = steps === 1 ? 'step' : 'steps'; + return ` (${String(steps)} ${stepWord}, ${String(input)} tokens in${cached}, ${String(output)} tokens out)`; +} + +/** Transcript as plain text for LLM-judged checks; `runDebug` inlines per-turn usage. */ +export function transcriptAsText( + transcript: TranscriptTurn[], + runDebug?: InstanceAiRunDebugResponse[], +): string { + const usageMap = usageByRunId(runDebug); return transcript .map((turn, i) => { // No seeded label: the judge evaluates the whole conversation as one. - const lines: string[] = [`### Turn ${String(i + 1)}`]; + const lines: string[] = [`### Turn ${String(i + 1)}${turnUsageSuffix(turn, usageMap)}`]; if (turn.userMessage) lines.push(`User: ${turn.userMessage}`); for (const step of turn.steps) { const line = describeStep(step); diff --git a/packages/@n8n/instance-ai/src/runtime/__tests__/run-state-registry.test.ts b/packages/@n8n/instance-ai/src/runtime/__tests__/run-state-registry.test.ts index 7ef101be4f3..4f333421136 100644 --- a/packages/@n8n/instance-ai/src/runtime/__tests__/run-state-registry.test.ts +++ b/packages/@n8n/instance-ai/src/runtime/__tests__/run-state-registry.test.ts @@ -86,6 +86,28 @@ describe('RunStateRegistry', () => { expect(registry.getBuildMode('thread-1')).toBeUndefined(); }); + it('scopes the observer threshold override to its own thread and clears it', () => { + registry.setObserverThresholdTokens('thread-1', 8000); + // A second thread on the same process keeps the instance default. + expect(registry.getObserverThresholdTokens('thread-1')).toBe(8000); + expect(registry.getObserverThresholdTokens('thread-2')).toBeUndefined(); + + registry.setObserverThresholdTokens('thread-1', undefined); + expect(registry.getObserverThresholdTokens('thread-1')).toBeUndefined(); + }); + + it('drops the observer threshold override when a thread is cleared', () => { + registry.setObserverThresholdTokens('thread-1', 8000); + registry.clearThread('thread-1'); + expect(registry.getObserverThresholdTokens('thread-1')).toBeUndefined(); + }); + + it('drops every observer threshold override on shutdown', () => { + registry.setObserverThresholdTokens('thread-1', 8000); + registry.shutdown(); + expect(registry.getObserverThresholdTokens('thread-1')).toBeUndefined(); + }); + it('retains the selected prompt version and metadata until the next explicit selection', () => { const metadata = { version: 'progressive@1', diff --git a/packages/@n8n/instance-ai/src/runtime/run-state-registry.ts b/packages/@n8n/instance-ai/src/runtime/run-state-registry.ts index 60f4667d8ee..e44ef458f3f 100644 --- a/packages/@n8n/instance-ai/src/runtime/run-state-registry.ts +++ b/packages/@n8n/instance-ai/src/runtime/run-state-registry.ts @@ -157,6 +157,8 @@ export class RunStateRegistry { /** Build mode captured at user-run entry and reused by follow-up runs. */ private readonly threadBuildModes = new Map(); + + private readonly threadObserverThresholds = new Map(); private readonly threadPromptSelections = new Map< string, { version: string; metadata?: InstanceAiPromptConfiguration } @@ -521,6 +523,16 @@ export class RunStateRegistry { return this.threadBuildModes.get(threadId); } + /** Per-thread observer threshold; an omitted value clears it. */ + setObserverThresholdTokens(threadId: string, tokens: number | undefined): void { + if (tokens === undefined) this.threadObserverThresholds.delete(threadId); + else this.threadObserverThresholds.set(threadId, tokens); + } + + getObserverThresholdTokens(threadId: string): number | undefined { + return this.threadObserverThresholds.get(threadId); + } + setPromptVersion(threadId: string, version: string | undefined): void { if (version === undefined) this.threadPromptSelections.delete(threadId); else this.threadPromptSelections.set(threadId, { version }); @@ -690,6 +702,7 @@ export class RunStateRegistry { this.threadTimeZones.delete(threadId); this.threadComputerUseChannels.delete(threadId); this.threadBuildModes.delete(threadId); + this.threadObserverThresholds.delete(threadId); this.threadPromptSelections.delete(threadId); const groupId = this.threadMessageGroupId.get(threadId); @@ -741,6 +754,7 @@ export class RunStateRegistry { this.threadTimeZones.clear(); this.threadComputerUseChannels.clear(); this.threadBuildModes.clear(); + this.threadObserverThresholds.clear(); this.threadPromptSelections.clear(); this.threadMessageGroupId.clear(); this.runIdsByMessageGroup.clear(); 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 8512b7e1db5..96a4d2f8464 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 @@ -40,6 +40,7 @@ import type { InstanceAiAdminSettingsUpdateRequest, InstanceAiEvalCredentialAllowlistRequest, InstanceAiEvalRestoreThreadRequest, + InstanceAiEvalThreadMemoryResponse, InstanceAiSendMessageRequest, InstanceAiCorrectTaskRequest, InstanceAiConfirmRequest, @@ -182,6 +183,7 @@ describe('InstanceAiController', () => { describe('chat', () => { const payload = mock({ + observerThresholdTokens: undefined, message: 'hello', timeZone: 'Europe/Helsinki', attachments: undefined, @@ -211,11 +213,13 @@ describe('InstanceAiController', () => { payload.promptVersion, payload.computerUseChannels, payload.threadArtifacts, + payload.observerThresholdTokens, ); }); it('should forward thread artifacts to startRun', async () => { const payloadWithArtifacts = mock({ + observerThresholdTokens: undefined, message: 'Change this', timeZone: 'UTC', attachments: undefined, @@ -242,9 +246,48 @@ describe('InstanceAiController', () => { payloadWithArtifacts.promptVersion, payloadWithArtifacts.computerUseChannels, payloadWithArtifacts.threadArtifacts, + payloadWithArtifacts.observerThresholdTokens, ); }); + it('forwards the observer threshold override for a caller with the eval scope', async () => { + const evalPayload = mock({ + message: 'build', + attachments: undefined, + observerThresholdTokens: 1000, + }); + const evalReq = mock({ + user: { id: USER_ID, role: { scopes: [{ slug: 'instanceAi:eval' }] } }, + }); + memoryService.checkThreadOwnership.mockResolvedValue('owned'); + instanceAiService.hasActiveRun.mockReturnValue(false); + instanceAiService.startRun.mockReturnValue('run-7'); + + await controller.chat(evalReq, res, THREAD_ID, evalPayload); + + const args = instanceAiService.startRun.mock.calls[0]; + expect(args[args.length - 1]).toBe(1000); + }); + + it('rejects the observer threshold override from a caller without the eval scope', async () => { + // The override changes how often the observer runs, so it stays an eval-only knob. + const chatPayload = mock({ + message: 'build', + attachments: undefined, + observerThresholdTokens: 1000, + }); + const chatReq = mock({ + user: { id: USER_ID, role: { scopes: [{ slug: 'instanceAi:message' }] } }, + }); + memoryService.checkThreadOwnership.mockResolvedValue('owned'); + instanceAiService.hasActiveRun.mockReturnValue(false); + + await expect(controller.chat(chatReq, res, THREAD_ID, chatPayload)).rejects.toThrow( + ForbiddenError, + ); + expect(instanceAiService.startRun).not.toHaveBeenCalled(); + }); + it('should allow new threads', async () => { memoryService.checkThreadOwnership.mockResolvedValue('not_found'); instanceAiService.hasActiveRun.mockReturnValue(false); @@ -257,6 +300,7 @@ describe('InstanceAiController', () => { it('should forward pushRef to startRun', async () => { const payloadWithPushRef = mock({ + observerThresholdTokens: undefined, message: 'build me a workflow', pushRef: 'iframe-push-ref-123', timeZone: 'UTC', @@ -280,11 +324,13 @@ describe('InstanceAiController', () => { payloadWithPushRef.promptVersion, payloadWithPushRef.computerUseChannels, payloadWithPushRef.threadArtifacts, + payloadWithPushRef.observerThresholdTokens, ); }); it('should forward the build mode and prompt version to startRun', async () => { const payloadWithMode = mock({ + observerThresholdTokens: undefined, message: 'build me a workflow', timeZone: 'UTC', attachments: undefined, @@ -309,11 +355,13 @@ describe('InstanceAiController', () => { 'progressive@1', payloadWithMode.computerUseChannels, payloadWithMode.threadArtifacts, + payloadWithMode.observerThresholdTokens, ); }); it('should forward handoff context to startRun', async () => { const payloadWithContext = mock({ + observerThresholdTokens: undefined, message: 'How do I set up Gmail OAuth?', context: { source: 'credential-modal', @@ -345,6 +393,7 @@ describe('InstanceAiController', () => { payloadWithContext.promptVersion, payloadWithContext.computerUseChannels, payloadWithContext.threadArtifacts, + payloadWithContext.observerThresholdTokens, ); }); @@ -376,6 +425,7 @@ describe('InstanceAiController', () => { memoryService.checkThreadOwnership.mockResolvedValue('owned'); instanceAiService.hasActiveRun.mockReturnValue(false); const badPayload = mock({ + observerThresholdTokens: undefined, message: 'see attached', attachments: [ { type: 'file', data: '', mimeType: 'application/zip', fileName: 'archive.zip' }, @@ -394,6 +444,7 @@ describe('InstanceAiController', () => { instanceAiService.hasActiveRun.mockReturnValue(false); instanceAiService.startRun.mockReturnValue('run-3'); const goodPayload = mock({ + observerThresholdTokens: undefined, message: 'see attached', attachments: [ { type: 'file', data: '', mimeType: 'application/pdf', fileName: 'doc.pdf' }, @@ -424,6 +475,7 @@ describe('InstanceAiController', () => { }, ]; const nodesPayload = mock({ + observerThresholdTokens: undefined, message: 'what do these nodes do?', timeZone: 'UTC', }); @@ -447,6 +499,7 @@ describe('InstanceAiController', () => { nodesPayload.promptVersion, nodesPayload.computerUseChannels, nodesPayload.threadArtifacts, + nodesPayload.observerThresholdTokens, ); }); @@ -454,6 +507,7 @@ describe('InstanceAiController', () => { memoryService.checkThreadOwnership.mockResolvedValue('owned'); instanceAiService.hasActiveRun.mockReturnValue(false); const oversizedPayload = mock({ + observerThresholdTokens: undefined, message: 'see screenshot', attachments: [ { @@ -477,6 +531,7 @@ describe('InstanceAiController', () => { instanceAiService.hasActiveRun.mockReturnValue(false); const halfBudget = Math.floor(MAX_TOTAL_ATTACHMENT_BASE64_BYTES / 2) + 1; const payloadOverBudget = mock({ + observerThresholdTokens: undefined, message: 'two screenshots', attachments: [ { @@ -983,6 +1038,47 @@ describe('InstanceAiController', () => { }); }); + describe('getEvalThreadMemory', () => { + const memory: InstanceAiEvalThreadMemoryResponse = { + observations: [{ marker: 'critical', text: 'Posting via HTTP Request', tokenCount: 7 }], + cursor: { lastObservedMessageId: 'm137', lastObservedAt: '2020-01-01T00:00:00.000Z' }, + }; + + it('should require instanceAi:eval scope', () => { + expect(scopeOf('getEvalThreadMemory')).toEqual({ + scope: 'instanceAi:eval', + globalOnly: true, + }); + }); + + it('should return the memory of an owned thread', async () => { + memoryService.checkThreadOwnership.mockResolvedValue('owned'); + instanceAiService.getThreadMemory.mockResolvedValue(memory); + + const result = await controller.getEvalThreadMemory(req, res, THREAD_ID); + + expect(result).toEqual(memory); + expect(instanceAiService.getThreadMemory).toHaveBeenCalledWith(USER_ID, THREAD_ID); + }); + + it("should reject another user's thread", async () => { + memoryService.checkThreadOwnership.mockResolvedValue('other_user'); + + await expect(controller.getEvalThreadMemory(req, res, THREAD_ID)).rejects.toThrow( + ForbiddenError, + ); + expect(instanceAiService.getThreadMemory).not.toHaveBeenCalled(); + }); + + it('should reject a thread that does not exist', async () => { + memoryService.checkThreadOwnership.mockResolvedValue('not_found'); + + await expect(controller.getEvalThreadMemory(req, res, THREAD_ID)).rejects.toThrow( + NotFoundError, + ); + }); + }); + describe('restoreEvalThread', () => { const seedMessages = [ { id: 'm1', type: 'llm', role: 'user', content: [], createdAt: '2026-01-01T00:00:00.000Z' }, 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 8412efef15f..86469813dcf 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 @@ -268,6 +268,7 @@ import { UserError } from 'n8n-workflow'; import type { Mock, MockedFunction } from 'vitest'; import { InstanceAiBuilderDelegateAdapterService } from '@/modules/agents/instance-ai-builder-delegate.adapter'; +import { ForbiddenError } from '@/errors/response-errors/forbidden.error'; import { userHasScopes } from '@/permissions.ee/check-access'; import { EvalThreadCredentialAllowlistService } from '../eval/thread-credential-allowlist.service'; @@ -298,6 +299,7 @@ type StartRunServiceInternals = { getComputerUseChannels: Mock; setBuildMode: MockedFunction<(threadId: string, mode: string | undefined) => void>; setPromptVersion: Mock; + setObserverThresholdTokens: Mock; activeRunCount: MockedFunction<() => number>; activeRunCountForUser: MockedFunction<(userId: string) => number>; }; @@ -325,6 +327,7 @@ function createStartRunService(): StartRunServiceInternals { getComputerUseChannels: vi.fn(() => undefined), setBuildMode: vi.fn(), setPromptVersion: vi.fn(), + setObserverThresholdTokens: vi.fn(), activeRunCount: vi.fn(() => 0), activeRunCountForUser: vi.fn(() => 0), }; @@ -790,6 +793,7 @@ describe('InstanceAiService — runtime workspace setup', () => { setPromptConfiguration: Mock; setBuildMode: Mock; setPromptVersion: Mock; + setObserverThresholdTokens: Mock; getComputerUseChannels: Mock; }; cancelBackgroundTask: Mock; @@ -864,6 +868,7 @@ describe('InstanceAiService — runtime workspace setup', () => { setPromptConfiguration: vi.fn(), setBuildMode: vi.fn(), setPromptVersion: vi.fn(), + setObserverThresholdTokens: vi.fn(), getComputerUseChannels: vi.fn(() => undefined), }; service.cancelBackgroundTask = vi.fn(); @@ -1121,6 +1126,7 @@ describe('InstanceAiService — runtime workspace setup', () => { setPromptConfiguration: Mock; setBuildMode: Mock; setPromptVersion: Mock; + setObserverThresholdTokens: Mock; getComputerUseChannels: Mock; }; cancelBackgroundTask: Mock; @@ -1191,6 +1197,7 @@ describe('InstanceAiService — runtime workspace setup', () => { setPromptConfiguration: vi.fn(), setBuildMode: vi.fn(), setPromptVersion: vi.fn(), + setObserverThresholdTokens: vi.fn(), getComputerUseChannels: vi.fn(() => undefined), }; service.cancelBackgroundTask = vi.fn(); @@ -6057,11 +6064,17 @@ describe('createAgentMemoryOptions', () => { user: User, threadId: string, runId: string, - ) => { observationalMemory: { onTaskUsage: (report: MemoryTaskUsageReport) => Promise } }; + ) => { + observationalMemory: { + observerThresholdTokens: number; + onTaskUsage: (report: MemoryTaskUsageReport) => Promise; + }; + }; instanceAiConfig: Pick< InstanceAiConfig, 'observerMessageTokens' | 'reflectorObservationTokens' >; + runState: { getObserverThresholdTokens: Mock }; creditService: { claimRunUsage: Mock; ensureQuotaLockApplied: Mock }; logger: { warn: Mock }; }; @@ -6069,6 +6082,7 @@ describe('createAgentMemoryOptions', () => { function buildService(): MemoryOptionsInternals { const service = Object.create(InstanceAiService.prototype) as unknown as MemoryOptionsInternals; service.instanceAiConfig = { observerMessageTokens: 8_000, reflectorObservationTokens: 12_000 }; + service.runState = { getObserverThresholdTokens: vi.fn().mockReturnValue(undefined) }; service.creditService = { claimRunUsage: vi.fn(async () => {}), ensureQuotaLockApplied: vi.fn(async () => {}), @@ -6077,6 +6091,28 @@ describe('createAgentMemoryOptions', () => { return service; } + it('uses the instance observer threshold when the thread has no override', () => { + const service = buildService(); + const { observerThresholdTokens } = service.createAgentMemoryOptions( + { id: 'user-1' } as User, + 'thread-1', + 'run-1', + ).observationalMemory; + expect(observerThresholdTokens).toBe(8_000); + expect(service.runState.getObserverThresholdTokens).toHaveBeenCalledWith('thread-1'); + }); + + it('uses the per-thread override when an eval set one', () => { + const service = buildService(); + service.runState.getObserverThresholdTokens.mockReturnValue(1_000); + const { observerThresholdTokens } = service.createAgentMemoryOptions( + { id: 'user-1' } as User, + 'thread-1', + 'run-1', + ).observationalMemory; + expect(observerThresholdTokens).toBe(1_000); + }); + it('claims converted usage under the orchestrator dedupe key, and skips claiming when usage is zero', async () => { const service = buildService(); const user = { id: 'user-1' } as User; @@ -6363,3 +6399,73 @@ describe('InstanceAiService — resolveAiPreferencesBlock', () => { ); }); }); + +describe('getThreadMemory', () => { + type MemoryInternals = { + getThreadMemory: InstanceAiService['getThreadMemory']; + agentMemory: { getThread: Mock }; + observationRepo: { findActiveForThread: Mock }; + observationCursorRepo: { findForThread: Mock }; + }; + + function buildService(): MemoryInternals { + const service = Object.create(InstanceAiService.prototype) as unknown as MemoryInternals; + service.agentMemory = { + getThread: vi.fn().mockResolvedValue({ id: 'thread-1', resourceId: 'user-1' }), + }; + service.observationRepo = { findActiveForThread: vi.fn().mockResolvedValue([]) }; + service.observationCursorRepo = { findForThread: vi.fn().mockResolvedValue(null) }; + return service; + } + + it("refuses another user's thread before it reads anything", async () => { + // The controller checks ownership too; this keeps the service safe for any other caller. + const service = buildService(); + await expect(service.getThreadMemory('user-2', 'thread-1')).rejects.toThrow(ForbiddenError); + expect(service.observationRepo.findActiveForThread).not.toHaveBeenCalled(); + expect(service.observationCursorRepo.findForThread).not.toHaveBeenCalled(); + }); + + it('refuses a thread that does not exist', async () => { + const service = buildService(); + service.agentMemory.getThread.mockResolvedValue(null); + await expect(service.getThreadMemory('user-1', 'thread-1')).rejects.toThrow(ForbiddenError); + }); + + it('returns the live rows and the cursor as an ISO timestamp', async () => { + const service = buildService(); + service.observationRepo.findActiveForThread.mockResolvedValue([ + { + id: 'obs-1', + marker: 'critical', + text: 'Posting via HTTP Request', + tokenCount: 7, + status: 'active', + observationScopeId: 'thread-1', + }, + ]); + service.observationCursorRepo.findForThread.mockResolvedValue({ + observationScopeId: 'thread-1', + lastObservedMessageId: 'm137', + lastObservedAt: new Date('2020-01-01T00:00:00.000Z'), + }); + + const memory = await service.getThreadMemory('user-1', 'thread-1'); + + // Only the three fields the eval grades on; ids and status stay server-side. + expect(memory).toEqual({ + observations: [{ marker: 'critical', text: 'Posting via HTTP Request', tokenCount: 7 }], + cursor: { lastObservedMessageId: 'm137', lastObservedAt: '2020-01-01T00:00:00.000Z' }, + }); + expect(service.observationRepo.findActiveForThread).toHaveBeenCalledWith('thread-1'); + expect(service.observationCursorRepo.findForThread).toHaveBeenCalledWith('thread-1'); + }); + + it('returns a null cursor and no rows when the observer never ran', async () => { + const service = buildService(); + expect(await service.getThreadMemory('user-1', 'thread-1')).toEqual({ + observations: [], + cursor: null, + }); + }); +}); 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 1ed00a634aa..2d7f19bf448 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.controller.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.controller.ts @@ -26,7 +26,11 @@ import { findSeedFolderIssues, findUnbackedSeedWorkflowTools, } from '@n8n/api-types'; -import type { InstanceAiAdminSettingsResponse, InstanceAiEvent } from '@n8n/api-types'; +import type { + InstanceAiAdminSettingsResponse, + InstanceAiEvalThreadMemoryResponse, + InstanceAiEvent, +} from '@n8n/api-types'; import { ModuleRegistry } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; import { AuthenticatedRequest, User, UserRepository } from '@n8n/db'; @@ -46,6 +50,7 @@ import { Query, } from '@n8n/decorators'; import type { StoredEvent } from '@n8n/instance-ai'; +import { hasGlobalScope } from '@n8n/permissions'; import { buildAgentTreeFromEvents, clearedAgentBuilderTargetMetadata, @@ -220,6 +225,15 @@ export class InstanceAiController { throw new ConflictError('A run is already active for this thread'); } + // The override is an eval knob. It changes how often the observer runs, so a + // plain chat caller must not be able to set it. + if ( + payload.observerThresholdTokens !== undefined && + !hasGlobalScope(req.user, 'instanceAi:eval') + ) { + throw new ForbiddenError('observerThresholdTokens requires the instanceAi:eval scope'); + } + const runId = this.instanceAiService.startRun( req.user, threadId, @@ -232,6 +246,7 @@ export class InstanceAiController { payload.promptVersion, payload.computerUseChannels, payload.threadArtifacts, + payload.observerThresholdTokens, ); return { runId }; } @@ -1074,6 +1089,25 @@ export class InstanceAiController { return { ok: true }; } + /** + * Observational memory for a thread, for a context eval to assert on. + * + * Returns the observation text, an LLM-written summary of the user's conversation. + * Gated like every other `/eval/` route: `instanceAi:eval` is owner/admin-only, + * and `assertThreadAccess` keeps a caller to threads they can already read in full. + */ + @Get('/eval/threads/:threadId/memory') + @GlobalScope('instanceAi:eval') + async getEvalThreadMemory( + req: AuthenticatedRequest, + _res: Response, + @Param('threadId') threadId: string, + ): Promise { + this.requireInstanceAiEnabled(); + await this.assertThreadAccess(req.user.id, threadId); + return await this.instanceAiService.getThreadMemory(req.user.id, threadId); + } + /** * Seed an existing (owned) thread with a previously exported conversation: * recreate the artifacts the history references — workflows (node credentials 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 6f94afcb749..b722fc873a9 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.service.ts @@ -29,6 +29,7 @@ import { type InstanceAiConfirmResponse, type InstanceAiEvent, type InstanceAiThreadStatusResponse, + type InstanceAiEvalThreadMemoryResponse, type InstanceAiThreadArtifactsContext, } from '@n8n/api-types'; import { Logger, ModuleRegistry } from '@n8n/backend-common'; @@ -219,6 +220,8 @@ import { type PlannedWorkflowVerificationTracker, } from './planned-task-action-runner'; import { InstanceAiEventLogRepository } from './repositories/instance-ai-event-log.repository'; +import { InstanceAiObservationCursorRepository } from './repositories/instance-ai-observation-cursor.repository'; +import { InstanceAiObservationRepository } from './repositories/instance-ai-observation.repository'; import { InstanceAiPendingConfirmationRepository } from './repositories/instance-ai-pending-confirmation.repository'; import { InstanceAiThreadGrantRepository } from './repositories/instance-ai-thread-grant.repository'; import { InstanceAiSandboxService, type RuntimeSandboxEntry } from './sandbox'; @@ -746,6 +749,8 @@ export class InstanceAiService { private readonly aiService: AiService, private readonly threadGrantRepo: InstanceAiThreadGrantRepository, private readonly pendingConfirmationRepo: InstanceAiPendingConfirmationRepository, + private readonly observationRepo: InstanceAiObservationRepository, + private readonly observationCursorRepo: InstanceAiObservationCursorRepository, private readonly urlService: UrlService, private readonly eventLogRepository: InstanceAiEventLogRepository, private readonly dbIterationLogStorage: DbIterationLogStorage, @@ -1277,6 +1282,35 @@ export class InstanceAiService { }; } + /** What observational memory holds for a thread: the live observations and the + * compaction cursor. An eval asserts on these rows instead of parsing the + * rendered system prompt. Refuses a thread the caller does not own, so the + * check does not depend on the route. */ + async getThreadMemory( + userId: string, + threadId: string, + ): Promise { + const thread = await this.agentMemory.getThread(threadId); + if (!thread || thread.resourceId !== userId) { + throw new ForbiddenError('Not authorized for this thread'); + } + const [observations, cursor] = await Promise.all([ + this.observationRepo.findActiveForThread(threadId), + this.observationCursorRepo.findForThread(threadId), + ]); + return { + observations: observations.map(({ marker, text, tokenCount }) => ({ + marker, + text, + tokenCount, + })), + cursor: cursor && { + lastObservedMessageId: cursor.lastObservedMessageId, + lastObservedAt: cursor.lastObservedAt.toISOString(), + }, + }; + } + getRunDebug(runId: string) { return this.runDebugBuffer.get(runId); } @@ -1377,6 +1411,7 @@ export class InstanceAiService { promptVersion?: string, computerUseChannels?: ComputerUseChannel[], threadArtifacts?: InstanceAiThreadArtifactsContext, + observerThresholdTokens?: number, ): string { if ( promptVersion !== undefined && @@ -1406,6 +1441,7 @@ export class InstanceAiService { // otherwise environment creation selects and stores the backend assignment. this.runState.setBuildMode(threadId, mode); this.runState.setPromptVersion(threadId, promptVersion); + this.runState.setObserverThresholdTokens(threadId, observerThresholdTokens); if (pushRef !== undefined) { this.threadPushRef.set(threadId, pushRef); @@ -2124,7 +2160,9 @@ export class InstanceAiService { private createAgentMemoryOptions(user: User, threadId: string, runId: string) { return { observationalMemory: { - observerThresholdTokens: this.instanceAiConfig.observerMessageTokens, + observerThresholdTokens: + this.runState.getObserverThresholdTokens(threadId) ?? + this.instanceAiConfig.observerMessageTokens, reflectorThresholdTokens: this.instanceAiConfig.reflectorObservationTokens, midRunObservation: this.instanceAiConfig.midRunObservation, // Observer/reflector calls run in the background outside the run's diff --git a/packages/cli/src/modules/instance-ai/repositories/__tests__/instance-ai-observation.repository.integration.test.ts b/packages/cli/src/modules/instance-ai/repositories/__tests__/instance-ai-observation.repository.integration.test.ts new file mode 100644 index 00000000000..7dd919dca87 --- /dev/null +++ b/packages/cli/src/modules/instance-ai/repositories/__tests__/instance-ai-observation.repository.integration.test.ts @@ -0,0 +1,78 @@ +import { getPersonalProject, testDb, testModules } from '@n8n/backend-test-utils'; +import { Container } from '@n8n/di'; + +import { createOwner } from '@test-integration/db/users'; + +import { InstanceAiObservationRepository } from '../instance-ai-observation.repository'; +import { InstanceAiThreadRepository } from '../instance-ai-thread.repository'; + +/** + * `findActiveForThread` feeds the eval memory endpoint. A superseded or dropped + * row is no longer what the agent remembers, and order is what a judge reads as + * the story, so both are pinned against a real database. + */ +describe('InstanceAiObservationRepository', () => { + let repository: InstanceAiObservationRepository; + + const THREAD = '11111111-1111-4111-8111-111111111111'; + const OTHER_THREAD = '22222222-2222-4222-8222-222222222222'; + + beforeAll(async () => { + await testModules.loadModules(['instance-ai']); + await testDb.init(); + repository = Container.get(InstanceAiObservationRepository); + // Observations hang off a thread, and a thread off a project. + const project = await getPersonalProject(await createOwner()); + const threads = Container.get(InstanceAiThreadRepository); + await threads.save( + [THREAD, OTHER_THREAD].map((id) => + threads.create({ id, resourceId: `res-${id}`, projectId: project.id, title: '' }), + ), + ); + }); + + afterEach(async () => { + // Not in the shared truncate list; a module table cleans up through its repository. + await repository.delete({}); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + function row( + text: string, + status: 'active' | 'superseded' | 'dropped', + createdAt: string, + scope = THREAD, + ) { + return repository.create({ + observationScopeId: scope, + marker: 'info' as const, + text, + parentId: null, + tokenCount: 3, + status, + supersededBy: null, + createdAt: new Date(createdAt), + }); + } + + it('returns only the live rows of the thread, oldest first', async () => { + await repository.save([ + row('third', 'active', '2026-01-01T00:00:03.000Z'), + row('first', 'active', '2026-01-01T00:00:01.000Z'), + row('replaced', 'superseded', '2026-01-01T00:00:02.000Z'), + row('forgotten', 'dropped', '2026-01-01T00:00:02.500Z'), + row('elsewhere', 'active', '2026-01-01T00:00:00.000Z', OTHER_THREAD), + ]); + + const rows = await repository.findActiveForThread(THREAD); + + expect(rows.map((r) => r.text)).toEqual(['first', 'third']); + }); + + it('returns an empty list when the observer never ran', async () => { + expect(await repository.findActiveForThread(THREAD)).toEqual([]); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation-cursor.repository.ts b/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation-cursor.repository.ts index a01d0c34dbc..fc7d9f60cf1 100644 --- a/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation-cursor.repository.ts +++ b/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation-cursor.repository.ts @@ -8,4 +8,10 @@ export class InstanceAiObservationCursorRepository extends Repository { + return await this.findOneBy({ observationScopeId: threadId }); + } } diff --git a/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation.repository.ts b/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation.repository.ts index bfaac23b080..6b882146259 100644 --- a/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation.repository.ts +++ b/packages/cli/src/modules/instance-ai/repositories/instance-ai-observation.repository.ts @@ -8,4 +8,13 @@ export class InstanceAiObservationRepository extends Repository { + return await this.find({ + where: { observationScopeId: threadId, status: 'active' }, + order: { createdAt: 'ASC', id: 'ASC' }, + }); + } }