Input & output processors

A turn has exactly two boundaries with the provider: the prompt going out and the answer coming back. Processors are the seams on each of them.

AgentModule.forRoot({
  model,
  actorResolver,
  inputProcessors: [maskIdentifiers],
  outputProcessors: [refuseRawCustomerRows],
});

Both lists are module-wide and apply to every agent. There is no per-agent override on purpose: a control one persona can opt out of is not a control.

Input: rewriting the promptCopy Anchor Link

import type { InputProcessor } from '@dudousxd/nestjs-agent-core';

const maskIdentifiers: InputProcessor = {
  name: 'mask-identifiers',
  process: (prompt) => ({
    system: prompt.system,
    messages: prompt.messages.map((message) => ({
      ...message,
      content: message.content.replace(/\b\d{3}-\d{2}-\d{4}\b/g, '[ssn]'),
    })),
  }),
};

It runs before every model call of a turn, not once per run. The transcript grows between steps — a tool result is appended and the loop calls the model again — so a redactor that only ever saw the opening prompt would wave through whatever that tool brought back.

What it returns is a derived prompt. The loop's own transcript is untouched, and so is the store: a redaction is what leaves the process, never the thread's memory of what was said.

Output: gating the answerCopy Anchor Link

import type { OutputProcessor } from '@dudousxd/nestjs-agent-core';

const refuseRawCustomerRows: OutputProcessor = {
  name: 'sql-guard',
  process: async (answer) => {
    if (looksLikeADump(answer.text)) {
      return { action: 'reject', reason: 'answer contains raw customer rows' };
    }
    return { action: 'replace', text: redactEmails(answer.text) };
  },
};

Each processor sees the answer as the previous one left it and returns one of three verdicts:

A rejection fails the run with an OutputRejectedError carrying the processor and the reason, and the subscriber gets an output_rejected stream error — its own code, distinct from run_failed, so a client that retries on a model failure does not retry a refusal and nobody is paged for the control doing its job. The first rejection stops the chain: a later processor has nothing to add about text that is going nowhere, and running it would bill a moderation call for a turn already refused.

The refused turn still records its chat usage row. Those tokens were genuinely spent, and a gate that hid its own cost could burn a budget invisibly.

What the chain actually seesCopy Anchor Link

A turn can produce model text in four places, not one. Three of them are calls the loop makes for itself, and they are not streamed — which is why they read as invisible until you go looking.

The structured passCopy Anchor Link

outputSchema asks the model to restate the finished answer as a schema. That is a second route out of the model, and an ungated one would undo the chain completely: anything held back from the prose can be restated into a field. So the chain runs over the restated text before it is validated, and what the schema validates — and what is persisted as the synthetic structured_output call and pushed to the client — is the gated version.

One consequence worth stating: a provider that constrained its own generation reports a parsed object alongside the text. Once the chain rewrites the text the two disagree, so the gated text is re-parsed and the provider's object is discarded. Trusting it would hand the reader exactly the value the gate had just removed.

A rejection here fails the run with OutputRejectedError, same as on the streamed answer.

Follow-up suggestionsCopy Anchor Link

Each suggestion goes through the chain on its own — one at a time, so a refusal is confined to the one that earned it and a replace cannot smear across a neighbour.

The history summary is not gatedCopy Anchor Link

When a historyPolicy folds old messages into a summary, that summary is model-generated — and it is deliberately not put through the output chain.

It never reaches a reader. It is re-injected as the leading system message of the next prompt, which makes it input, and the input seam already owns it: inputProcessors see the summary on every step of every turn that carries one, because it is part of prompt.messages. Running answer-shaped controls over prompt text would apply a contract the processor's author never agreed to — and the seam that does apply is already there.

If you want to constrain what a summary may say, write an InputProcessor.

The streaming tradeCopy Anchor Link

By default — any processor that does not declare otherwise — the call writes to a buffer, and the loop releases it only once the chain has passed, as a single text frame carrying the gated answer.

What a subscriber still receives live: step-start / step-finish boundaries, and the turn's tool-call frames (tool-input-start, tool-input-delta, tool-input-available), in their original order relative to the answer. What it loses:

• Token-by-token text. The answer arrives in one frame, once the gate has decided.
• Anything the gate cannot classify. The sink is a byte channel, and a model provider outside the AgentStreamEvent vocabulary may write bare bytes into it. Those frames are dropped, and the text is reconstructed from the gated answer instead — forwarding bytes it cannot read would not be a gate. Adapters that speak the vocabulary (including the bundled AI SDK one) are unaffected.

Register no output processor and the turn streams exactly as it always did, frame for frame.

Incremental gating: keeping the streamCopy Anchor Link

An LLM judge has to read the whole answer. A pattern redactor does not — it can rule on a prefix and let that prefix go. Declaring incremental says so, and the turn keeps streaming:

const redactEmails: OutputProcessor = {
  name: 'redact-emails',
  // Wide enough for the longest pattern this matches. Costs ~81 tokens of latency before the
  // reader sees anything, and an answer shorter than this streams nothing at all — see below.
  incremental: { lookbackChars: 320 },
  process: (answer) => ({
    action: 'replace',
    text: answer.text.replace(EMAIL, '[email]'),
  }),
};

The chain sees the growing prefix, not each new chunk, so it always gets well-formed text and never has to reassemble a pattern split across frames. After every prefix, the loop releases what the chain produced minus the trailing lookbackChars characters, which stay held so the processor can still change them. The rest arrives when the whole-answer pass runs.

What you promise by declaring itCopy Anchor Link

1. A rejection is decidable from a prefix. The refusal you would return for the whole answer is already returned for the first prefix that contains the reason.
2. A replacement is prefix-stable outside the window. Once a character of your output is more than lookbackChars from the end of that output, it never changes as the prefix grows.

Both promises are checked rather than trusted, because the whole-answer pass is still authoritative for the stream and the store — the incremental release only emits its prefix early. Once it settles, the loop asserts the final text startsWith what was already released and raises ProcessorFailedError if it does not:

output processor "redact-emails" failed: the gated answer is not an extension of the 412 characters
this chain already released — a processor that declares "incremental" promises its result on a
prefix stays a prefix of its result on the whole answer, outside the last lookbackChars

That is what a window too short for your pattern looks like: a failed run, not a redaction that quietly did not happen. A window that truncates silently would be worse than having no incremental path at all.

Choosing lookbackCharsCopy Anchor Link

It is per-processor, defaults to DEFAULT_INCREMENTAL_LOOKBACK_CHARS (64), and the gate uses the widest value in the chain — a release safe for the shortest-sighted processor is not safe for the one matching longer patterns. Set it to the length of the longest text your processor can act on.

It is also the answer's minimum latency tail: those characters are only released once the turn's last model chunk has arrived, so "as large as possible" is not the safe choice either. Measured, with the first text frame reaching the reader only after that many characters have been gated:

Non-text frames (tool cards, reasoning) are forwarded live in arrival order, and frames the gate cannot decode are dropped — the same rule the buffered path applies, for the same reason. The chain runs once per streamed text frame, over the prefix so far, which is why an incremental processor should be cheap and local.

Under the durable runnerCopy Anchor Link

Both chains run inside a checkpoint — process:input:<step> and process:output:<step> — so a processor may call a model, and a replay reads the result back rather than deciding again. Three consequences worth knowing:

• The held frames ride the llm:<step> checkpoint, not a variable in the process that ran the model. A run that suspends between the model call and the gate resumes in a process that never saw the stream; a buffer held in memory would leave the turn's whole answer unstreamed.
• The release happens inside the gate's own checkpoint, so "refused" and "nothing was streamed" are one journaled fact rather than two a suspend could separate — and a replay does not flush the same answer into the stream a second time.
• Under durable: true the model runs on another worker, whose sink the loop cannot wrap. The envelope therefore carries a bufferOutput flag and AgentRunSteps.llm hands the held frames back on its step result. Without that, the gate would be bypassed on exactly the deployment that needs it. An incremental chain falls back to full buffering here — there is no prefix to release, because the model wrote its chunks into a sink on the far side of the dispatch.
• An incremental turn journals its released prefix on llm:<step> as releasedText, and the gate step reads which release it still owes from that value rather than from the chain the resuming process happens to be configured with. So a run that suspended under an incremental chain cannot flush the whole answer again under a re-declared one. A refusal reached from a prefix rides the same checkpoint as gateRejection, and is raised only after persist:usage:<step> and quota:bump:<step> — the same ordering a whole-answer refusal has always had.

process:output:<step> keeps its name and its position in all three modes, so changing a declaration never moves a checkpoint. Configure neither list and the loop's checkpoint names and positions are byte-identical to before, so a run already in flight when you ship the config keeps replaying.

The two non-streamed passes take checkpoints of their own — process:output:structured:<step>:<attempt> and process:output:followups:<step> — and both are reachable only when an output chain is registered and the feature they gate is configured. Each sits immediately after the usage row for the call it is ruling on, so a refusal can never hide the tokens it already spent.

When a processor throwsCopy Anchor Link

It surfaces as ProcessorFailedError, carrying phase ('input' | 'output'), processor (the processor's name) and the original error as cause:

output processor "moderation" failed: judge timed out

Wrapped deliberately: the model call is the only other thing that can fail at that point in the turn, and a bare TypeError from a classifier that would otherwise read as the provider going down is the kind of ambiguity that costs an hour at 3am.

GuardrailsCopy Anchor Link

@dudousxd/nestjs-agent-core/guardrails (re-exported as @dudousxd/nestjs-agent/guardrails) ships PII, secret, prompt-injection and tool-poisoning detection as a ready-made pair of processors:

import { createGuardrails } from '@dudousxd/nestjs-agent/guardrails';

const guardrails = createGuardrails({
  pii: 'redact',                 // reversible: the model sees [EMAIL_1], the reader the address
  secrets: 'block',
  injection: { threshold: 0.6 }, // default action: block
  toolPoisoning: true,           // for guardrails.screenTool, see below
  rules: (ctx) => rulesForTenant(ctx.actor?.tenantRef), // more rules, resolved per scan
  onEvent: (event) => audit.write(event),               // hits carry fingerprints, never values
  fingerprint: (value) => hmac(value),
});

AgentModule.forRoot({
  model,
  inputProcessors: [guardrails.input],
  outputProcessors: [guardrails.output],
});

Each shorthand takes an action ('allow' | 'log' | 'redact' | 'approve' | 'block') or an object ({ action, types?, stages? }; pii also takes restore, default true). The loop has no approval flow for guardrails, so approve is enforced as block.

• Streaming stays on. guardrails.output declares incremental with a 256-character lookback. incremental: { lookbackChars } widens it; incremental: false gates the whole answer.
• Placeholders live per thread in a VaultStore (option vaults, default InMemoryVaultStore, bounded to 1,000 threads). A run resumed in another process finds no vault, so its placeholders reach the reader unrestored; the values still never reach the model. Vault.toJSON() / Vault.fromJSON() back a shared store, which then holds the raw values.
• wrapTool is the only place a tool's arguments can be rewritten: it restores the thread's placeholders (the tool needs the address, not [EMAIL_1]) and runs the tool_args rules on what the tool receives. That is why the pii shorthand does not cover tool_args; secrets does.
• Injection is scored on what the user typed and what tools returned, not on your system prompt.
• Custom rules are { id, stages, detectors, action, match?: { sources?, tools? }, when?, options? }. Detectors are pii, secrets, injection, tool_poisoning, regex, keywords, or { kind: 'custom', name, detect } for a classifier or LLM judge. A detector that throws follows the rule's options.failMode ('open' by default).

onEvent is awaited, and a throw from it is swallowed: an audit sink cannot change a decision. Each hit is reported once per turn, since every step re-sends the prompt.

Tool poisoning on MCPCopy Anchor Link

A remote tool's description never enters the transcript, so no processor sees it. AgentMcpModule takes a screen (module-wide, or per server) that runs on every listed definition before import; screenTool has its shape:

import { createGuardrails } from '@dudousxd/nestjs-agent-core/guardrails';

const guardrails = createGuardrails({ toolPoisoning: true });

AgentMcpModule.forRoot({
  servers: [{ name: 'docs', transport: { type: 'http', url } }],
  screen: guardrails.screenTool, // bound to its instance, so it can be passed as is
});

A refused tool is skipped and logged, and a screen that throws skips the tool too. McpToolsService.refresh() re-screens. It is a heuristic, not a proof.

Without the loopCopy Anchor Link

Everything but Guardrails is standalone (no NestJS, no Node-only API), for a gateway proxying raw provider traffic: scan, Vault, StreamGuard (OpenAI / Anthropic stream chunks), the *Slots helpers, and the detectors detectPii, detectSecrets, scoreInjection, scoreToolText, screenToolDefinition.

Building blocksCopy Anchor Link

Where to go nextCopy Anchor Link