mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat(ai-builder): Grade token consumption and memory compaction in Instance AI evals (no-changelog) (#38974)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
José Braulio González Valido
GitHub
parent
340d36c860
commit
5b149495c5
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -597,6 +597,8 @@ export type {
|
|||
InstanceAiRunDebugStep,
|
||||
InstanceAiRunDebugWorkflowCodeSnapshot,
|
||||
InstanceAiRunDebugResponse,
|
||||
InstanceAiEvalThreadMemoryResponse,
|
||||
InstanceAiEvalObservation,
|
||||
InstanceAiThreadDebugRunsResponse,
|
||||
InstanceAiRichMessagesResponse,
|
||||
InstanceAiMemoryTaskKind,
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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=<file>` 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 `<observations>` 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`,
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<GenerateFn>().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<GenerateFn>().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<GenerateFn>().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<GenerateFn>().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<GenerateFn>().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<GenerateFn>().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<GenerateFn>().mockResolvedValue({
|
||||
structuredOutput: { results: [{ index: 0, pass: true, reason: 'ok' }] },
|
||||
|
|
|
|||
|
|
@ -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<string>(),
|
||||
preRunDataTableIds: new Set<string>(),
|
||||
|
|
@ -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<void> {
|
|||
|
||||
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';
|
||||
|
||||
|
|
|
|||
|
|
@ -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<typeof vi.fn>): 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<string>(),
|
||||
claimedWorkflowIds: new Set<string>(),
|
||||
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<typeof vi.fn>): 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 }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<typeof mockFetch>): { 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',
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -92,6 +92,8 @@ describe('user execution during a conversation', () => {
|
|||
undefined,
|
||||
'progressive',
|
||||
'progressive@1',
|
||||
undefined,
|
||||
undefined,
|
||||
);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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<InstanceAiEvalThreadMemoryResponse> {
|
||||
return this.unwrapRestData<InstanceAiEvalThreadMemoryResponse>(
|
||||
await this.fetch(`/rest/instance-ai/eval/threads/${threadId}/memory`, { timeoutMs }),
|
||||
);
|
||||
}
|
||||
|
||||
// -- Computer-use gateway (pairing + status) -----------------------------
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
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<BuildResult> {
|
||||
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<BuildR
|
|||
timeoutMs,
|
||||
logger,
|
||||
proxyResponses,
|
||||
observerThresholdTokens,
|
||||
followUpMessagesOut: followUpMessages,
|
||||
// Only wired when the credential view is actually pinned — the
|
||||
// allowlist endpoint a mid-run creation depends on isn't available
|
||||
|
|
@ -1117,6 +1129,7 @@ export async function buildWorkflow(config: BuildWorkflowConfig): Promise<BuildR
|
|||
buildMode,
|
||||
promptVersion,
|
||||
openingHandoffContext,
|
||||
observerThresholdTokens,
|
||||
);
|
||||
await waitForAllActivity({
|
||||
client,
|
||||
|
|
|
|||
|
|
@ -302,6 +302,7 @@ export interface MultiTurnConfig extends WaitConfig {
|
|||
/** Repeat the eval override on each message to bypass the backend assignment. */
|
||||
buildMode?: InstanceAiBuildMode;
|
||||
promptVersion?: string;
|
||||
observerThresholdTokens?: number;
|
||||
}
|
||||
|
||||
export async function runMultiTurnConversation(config: MultiTurnConfig): Promise<void> {
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<string, TranscriptTurn[]>;
|
||||
buildExpectationsByKey: Map<string, Promise<BuildExpectationResult[]>>;
|
||||
runDebugByThreadId: Map<string, Promise<InstanceAiRunDebugResponse[]>>;
|
||||
threadMemoryByThreadId: Map<string, Promise<InstanceAiEvalThreadMemoryResponse | undefined>>;
|
||||
agentContextByKey: Map<string, Promise<AgentScenarioContext>>;
|
||||
/** Injectable delay for the provider-outage retry backoff — tests pass a no-op. */
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
|
|
@ -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 <type> 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string, Promise<BuildExpectationResult[]>>();
|
||||
const runDebugByThreadId = new Map<string, Promise<InstanceAiRunDebugResponse[]>>();
|
||||
const threadMemoryByThreadId = new Map<
|
||||
string,
|
||||
Promise<InstanceAiEvalThreadMemoryResponse | undefined>
|
||||
>();
|
||||
|
||||
// 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,
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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<string, RunUsage> {
|
||||
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, RunUsage>): 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);
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -157,6 +157,8 @@ export class RunStateRegistry<TUser = unknown> {
|
|||
|
||||
/** Build mode captured at user-run entry and reused by follow-up runs. */
|
||||
private readonly threadBuildModes = new Map<string, InstanceAiBuildMode>();
|
||||
|
||||
private readonly threadObserverThresholds = new Map<string, number>();
|
||||
private readonly threadPromptSelections = new Map<
|
||||
string,
|
||||
{ version: string; metadata?: InstanceAiPromptConfiguration }
|
||||
|
|
@ -521,6 +523,16 @@ export class RunStateRegistry<TUser = unknown> {
|
|||
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<TUser = unknown> {
|
|||
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<TUser = unknown> {
|
|||
this.threadTimeZones.clear();
|
||||
this.threadComputerUseChannels.clear();
|
||||
this.threadBuildModes.clear();
|
||||
this.threadObserverThresholds.clear();
|
||||
this.threadPromptSelections.clear();
|
||||
this.threadMessageGroupId.clear();
|
||||
this.runIdsByMessageGroup.clear();
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ import type {
|
|||
InstanceAiAdminSettingsUpdateRequest,
|
||||
InstanceAiEvalCredentialAllowlistRequest,
|
||||
InstanceAiEvalRestoreThreadRequest,
|
||||
InstanceAiEvalThreadMemoryResponse,
|
||||
InstanceAiSendMessageRequest,
|
||||
InstanceAiCorrectTaskRequest,
|
||||
InstanceAiConfirmRequest,
|
||||
|
|
@ -182,6 +183,7 @@ describe('InstanceAiController', () => {
|
|||
|
||||
describe('chat', () => {
|
||||
const payload = mock<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
message: 'build',
|
||||
attachments: undefined,
|
||||
observerThresholdTokens: 1000,
|
||||
});
|
||||
const evalReq = mock<AuthenticatedRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
message: 'build',
|
||||
attachments: undefined,
|
||||
observerThresholdTokens: 1000,
|
||||
});
|
||||
const chatReq = mock<AuthenticatedRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
observerThresholdTokens: undefined,
|
||||
message: 'see attached',
|
||||
attachments: [
|
||||
{ type: 'file', data: '', mimeType: 'application/pdf', fileName: 'doc.pdf' },
|
||||
|
|
@ -424,6 +475,7 @@ describe('InstanceAiController', () => {
|
|||
},
|
||||
];
|
||||
const nodesPayload = mock<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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<InstanceAiSendMessageRequest>({
|
||||
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' },
|
||||
|
|
|
|||
|
|
@ -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<void> } };
|
||||
) => {
|
||||
observationalMemory: {
|
||||
observerThresholdTokens: number;
|
||||
onTaskUsage: (report: MemoryTaskUsageReport) => Promise<void>;
|
||||
};
|
||||
};
|
||||
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,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<InstanceAiEvalThreadMemoryResponse> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<InstanceAiEvalThreadMemoryResponse> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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([]);
|
||||
});
|
||||
});
|
||||
|
|
@ -8,4 +8,10 @@ export class InstanceAiObservationCursorRepository extends Repository<InstanceAi
|
|||
constructor(dataSource: DataSource) {
|
||||
super(InstanceAiObservationCursor, dataSource.manager);
|
||||
}
|
||||
|
||||
/** Observational memory is thread-scoped, so the scope id is the thread id.
|
||||
* Null means the observer never ran for this thread. */
|
||||
async findForThread(threadId: string): Promise<InstanceAiObservationCursor | null> {
|
||||
return await this.findOneBy({ observationScopeId: threadId });
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,4 +8,13 @@ export class InstanceAiObservationRepository extends Repository<InstanceAiObserv
|
|||
constructor(dataSource: DataSource) {
|
||||
super(InstanceAiObservation, dataSource.manager);
|
||||
}
|
||||
|
||||
/** Live observations for a thread, oldest first. Superseded and dropped rows are
|
||||
* excluded: they are no longer what the agent remembers. */
|
||||
async findActiveForThread(threadId: string): Promise<InstanceAiObservation[]> {
|
||||
return await this.find({
|
||||
where: { observationScopeId: threadId, status: 'active' },
|
||||
order: { createdAt: 'ASC', id: 'ASC' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user