attend

Decide when an event should wake your agent.

An app emits a firehose of events: log lines, messages, clicks, alerts. An agent that wakes on every one is noisy, and one behind a debounce is deaf to the message that mattered. attend sits between the two. It settles each scope's events into a window, asks a judge what the agent should do about the window, and hands the decision to your app, which keeps the final say in how that route runs.

  app events (the firehose)
        |   push() in-process, or NDJSON over HTTP to attend serve
        v
  +----------------------------------------------------------+
  |  attend                                                  |
  |   window    one per scope; events pile up                |
  |   settle    the act in it is finished, or the window     |
  |             is old, or full, or you flushed it           |
  |   judge     reads the settled window, picks a route:     |
  |             interrupt | queue | note | ask | skip | wait |
  |   log       every step and decision is persisted         |
  +----------------------------------------------------------+
        |  ~4 to 8 decisions a minute, not thousands of events
        v
  decisions out
        |-- in-process: your sink's handler for the route
        '-- attend serve: MCP Events poll, stream and tools, for any MCP client
        v
  your agent reacts (barge in, take it next turn, add context, offer the user a prompt);
  your app decides how each route is carried out

The judge picks the route and the app executes it: interrupt means "the agent should hear this now", and your handler decides whether that cancels the turn, steers it or shows a banner. Raw events never leave attend. Only settled, judged decisions do.

npm install attend

attend needs Node 22.13 or later and ships ESM only. The core has no runtime dependencies and imports nothing from Node.

Two ways to run it

One engine, run either way.

Raw events never cross MCP, so the MCP hop carries a few decisions a minute and adds about a millisecond to a judge call that takes 140 ms or more.

attend serve

npx attend serve --store sqlite:attend.db --judge jev --kinds kinds.json
# attend serving http://127.0.0.1:8787  (ingest POST http://127.0.0.1:8787/v1/ingest, MCP http://127.0.0.1:8787/mcp)

One command, no image. Every flag has an environment variable, and a flag wins.

Send events from anything that can POST:

curl -H 'content-type: application/x-ndjson' --data-binary @events.ndjson http://127.0.0.1:8787/v1/ingest
# {"scope":"ops/db","id":"a1","at":1760570000123,"name":"alert.fired","data":{},"text":"replica lag 40s on db-2"}
# {"scope":"ops/db","end":true}

A line is an event, {"scope", "flush": true} or {"scope", "end": true}. The reply is {accepted, duplicates, rejected: [{line, scope, error}]} once every accepted line has committed, and a 429 with Retry-After when attend is over its limit. POST /v1/ingest/stream keeps one NDJSON request open and answers with acks and credits.

An agent hears the decisions by subscribing to attention.decision on /mcp. Hosts that do not subscribe call the attend_wait tool in a loop. In TypeScript, decisionFeed from attend/mcp keeps the cursor for you. This is examples/serve.ts, which runs the server and a subscriber in one process:

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
import { createAttend, memoryStore } from 'attend';
import { decisionFeed } from 'attend/mcp';
import { serve } from 'attend/server';

const store = memoryStore();
const attend = createAttend({
  store,
  kind: (event) => (event.name === 'page.acked' ? 'flush' : event.name === 'alert.fired' ? 'closing' : 'consequence'),
  judge: async ({ window }) => ({ route: window.reason === 'flush' ? 'interrupt' : 'note' }),
  sink: 'log',
  deadLetter: 'log',
});

const server = serve(attend, store, { listen: { port: 0 } });
await server.ready;
const url = server.url();
console.log(`attend serving ${url}`);

const client = new Client({ name: 'pager-agent', version: '1.0.0' });
// The SDK's class types sessionId as optional, which exactOptionalPropertyTypes rejects.
await client.connect(new StreamableHTTPClientTransport(new URL(`${url}/mcp`)) as Transport);
const stop = new AbortController();
const feed = decisionFeed(client, { cursor: null, signal: stop.signal })[Symbol.asyncIterator]();

serve(attend, store, options) is the same thing as a library call: it runs attend.run, the events server over store.log and the ingest routes on one HTTP listener. Give it principal to authorize by scope prefix, and close it with await server.close().

Related MCP server: voice-gate

Quickstart

import { createAttend, defaultNotice, memoryStore, routes, type Decision } from 'attend';

const tell = (decision: Decision) => console.log(`[${decision.route}]\n${defaultNotice(decision)}`);

const attend = createAttend({
  store: memoryStore(),
  kind: (event) => (event.name === 'user.message' ? 'flush' : 'consequence'),
  judge: async ({ window }) => ({ route: window.reason === 'flush' ? 'interrupt' : 'note' }),
  sink: routes({ interrupt: tell, queue: tell, note: tell, ask: tell }),
  deadLetter: (letter) => console.error('dead letter', letter.id, letter.message),
});

void attend.run({ signal: AbortSignal.timeout(1000) });
await attend.push('session-1', [
  { id: 'e1', at: Date.now(), name: 'build.failed', data: {}, text: 'Build 4521 failed: 1 test' },
  { id: 'e2', at: Date.now(), name: 'user.message', data: {}, text: 'can you look?' },
]);

The build failure alone never wakes the agent. The user's message settles the window at once, and the agent hears both lines in one notice:

[interrupt]
Build 4521 failed: 1 test
can you look?
Attend to this now, even mid-task.

Before production, replace the function judge with jevJudge or llmJudge and memoryStore with a database store.

Demo

pnpm demo starts a real MCP Events server in the same process, streams a short CI story into attend, and prints what the agent hears. It needs no API key and no database. The source is examples/demo.ts; the script builds the package and runs it with node --experimental-strip-types, so Node 22.13 or later runs it as TypeScript.

attend demo: a CI server streams MCP Events, and attend decides what the agent hears.

 0.0s  event    ci.started {"run":4521,"branch":"main"}
 0.1s  event    ci.log {"line":"pnpm install"}
 0.2s  event    ci.log {"line":"pnpm test"}
 0.3s  event    ci.log {"line":"211 passed, 1 failed"}
 0.4s  event    ci.test_failed {"test":"checkout applies the coupon"}
 0.6s  settle   5 events (closing)
 0.6s  judge    wait (hold the window and judge it again)
 1.0s  event    comment.created {"author":"sam","body":"can you look?"}
 1.0s  settle   6 events (flush), merged with the held window
 1.0s  judge    interrupt
 1.0s  deliver  interrupt
                ci.started {"run":4521,"branch":"main"}
                ci.log {"line":"pnpm install"}
                ci.log {"line":"pnpm test"}
                ci.log {"line":"211 passed, 1 failed"}
                ci.test_failed {"test":"checkout applies the coupon"}
                comment.created {"author":"sam","body":"can you look?"}
                Attend to this now, even mid-task.
 1.8s  event    deploy.finished {"env":"staging","sha":"9f3c2e1"}
 2.0s  settle   1 event (closing)
 2.0s  judge    note
 2.0s  deliver  note
                deploy.finished {"env":"staging","sha":"9f3c2e1"}
                Background only. No reply is needed.

7 events in, 2 deliveries out. Waking on every event would have run the agent 7 times.

attend held the log lines until the failed test closed the window. The judge chose wait for the failure, so the user's comment merged into the held window, and one interrupt carried all six events to the agent. The deploy arrived as a note.

How it works

  MCP Events, webhooks, push()
        |
        v
  kind(event)        closing | flush | continuous | passive | consequence | context
        |
        v
  window             one per scope, persisted with the source's cursor
        |            settles when its earliest deadline passes
        v
  judge(window)      interrupt | queue | note | ask | skip | wait
        |                                                      |
        v                                                      +-- held, judged again later
  sink.deliver(decision)    at least once, idempotent on the decision id

1. A source hands events to a scope. A scope is whatever one agent attends to: a conversation, a repo or a device. Each scope has one persisted state: a cursor per attached source, the open window, the judge slot and the outbox.

2. kind(event) gives each event a settle role when it arrives. attend stores the kind on the event, so a later change to kind never reinterprets an open window.

3. The window settles at the earliest deadline among its events. A window that holds only consequence and context events never settles.

4. The judge sees the settled window and picks a route. One judge call runs per scope at a time. Events that arrive during the call wait in the next window.

5. The decision is appended to the store's decision log in the same write as the state change, so a restart loses nothing. In-process, the sink receives it; over attend serve, subscribers read it from the log. It carries the route, the window, a probability for each route and the answers to any predicates. The judge never writes text. defaultNotice(decision) builds the agent's notice from the events.

Kinds

closing, flush, continuous and passive events are acts: things that happen in the scope and can call for the agent, whether the user did them or an outside party did. A window also settles when its first act is maxWindowMs old (reason age), when it fills its share of the judge's context window (reason size, see the window budget), when you call attend.end(scope) because the producer knows the unit is over (reason closing, wait still offered), or when you call attend.flush(scope) (reason forced, with wait not offered). When a window passes maxEvents, attend drops its oldest consequence and context events and counts them in window.dropped. If the window is still too big, it settles with reason overflow. attend never drops an act. Consequence and context events that wait longer than maxWindowMs for an act are dropped and counted the same way.

A system event that should wake the agent on its own, such as a failed deploy or a fired alert, needs an act kind. closing is the usual choice.

Routes

attend picks the route, and your sink carries it out. routes() makes you write a handler for each of the four delivered routes.

MCP Events

mcpEventsSource from attend/mcp turns one MCP Events subscription into a source. Install the peer dependencies with npm install @modelcontextprotocol/sdk zod; zod 3.25 or later and zod 4 both work. This excerpt is from examples/mcp-events.ts, with attend set up as in the quickstart and controller an AbortController:

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js';
import { mcpEventsSource } from 'attend/mcp';

const client = new Client({ name: 'my-agent', version: '1.0.0' });
await client.connect(new StdioClientTransport({ command: 'node', args: ['./ci-events-server.js'] }));

const source = mcpEventsSource(client, {
  name: 'repo.activity',
  arguments: { repo: 'acme/shop' },
  onError: (error) => console.warn('mcp events', error),
});

await Promise.all([
  attend.run({ signal: controller.signal }),
  attend.attach('repo:acme/shop', source, { signal: controller.signal }),
]);

The source covers these parts of the draft:

• In the default auto mode it reads events/list, streams with events/stream when the event lists push delivery, and polls with events/poll otherwise. A -32026 (Unsupported) reply to a stream falls back to polling.

• Every batch carries the server's cursor, and attend commits it in the same write as the events, under the source's key (mcp:<name>:<arguments>). After a restart, attach resumes from the committed cursor.

• eventId becomes Event.id, so the scope drops redelivered events.

• truncated: true sets window.gap. The judge sees it, and defaultNotice says that events were lost.

• When a poll result sets hasMore, the source polls again at once. Otherwise it waits nextPollMs, or pollMs (default 10000) when the server sends none.

• A terminated stream reopens from the committed cursor. So does a stream with more than maxBuffered (default 1000) notifications waiting on a slow store, which is reported as overflow, and a stream that sends nothing, heartbeats included, for idleTimeoutMs (default 300000), which is reported as idle.

• A -32025 (ResourceExhausted) reply or a request timeout is thrown as a plain error, so attach backs off through retryBackoffMs and resubscribes from the committed cursor. Any other error ends the subscription: attach rejects with a PermanentError whose cause is the SDK error.

• When the Client's transport closes, attach rejects with a PermanentError that says the Client closed. Reconnect the Client and attach again.

• onError receives each notifications/events/error, each payload that fails validation, each overflow, each idle, and each unrouted notification, one that names a subscription no stream here opened. None of them ends the subscription.

• Several streams can share one Client. attend installs one handler for each of the five notifications/events/* methods on the Client, records the JSON-RPC id of each events/stream request and routes notifications by _meta["io.modelcontextprotocol/subscriptionId"]. Do not replace those handlers.

MCP events carry no one-line summary, so Event.text stays unset and defaultNotice prints each event's name and JSON data.

Judges

The speed default is a system-one classifier. jevJudge (TypeSafe's Jev, with your own TypeSafe key) and decisionsJudge (the OpenAI Decisions API) answer a fixed multiple-choice question with probabilities in roughly 140 ms or more, which is what a real-time agent can afford on every settled window. llmJudge calls a general model you already use; it is slower, and its self-reported probabilities are close to one-hot. Use it when you want reasoning over speed.

A judge is a function, (input: JudgeInput, signal: AbortSignal) => Promise<Judgment>. It gets the settled window, the routes on offer with your criteria, your guidance, the predicates to answer, live facts from context() and the scope's recent verdicts. It returns { route, probabilities?, predicates? }.

• Throw to retry with backoff. Throw a PermanentError to dead-letter the window at once.

• A route that was not offered counts as a retryable failure. Check input.routes before you answer wait.

• Stop when signal aborts. attend aborts the call after judgeTimeoutMs.

llmJudge

llmJudge from attend/llm works with any model that answers in JSON matching a schema. You pass that one call as complete. attend writes the prompt and the schema and validates the reply. With Claude:

import Anthropic from '@anthropic-ai/sdk';
import { PermanentError } from 'attend';
import { llmJudge } from 'attend/llm';

const anthropic = new Anthropic();

export const judge = llmJudge({
  complete: async ({ system, prompt, images, schema, signal }) => {
    const res = await anthropic.messages.create(
      {
        model: 'claude-opus-5-5',
        max_tokens: 1024,
        system,
        messages: [
          {
            role: 'user',
            content: [
              { type: 'text', text: prompt },
              ...images.map((image) => ({ type: 'image' as const, source: { type: 'base64' as const, media_type: image.mimeType, data: image.data } })),
            ],
          },
        ],
        output_config: { effort: 'low', format: { type: 'json_schema', schema } },
      },
      { signal },
    );
    if (res.stop_reason === 'refusal') throw new PermanentError('judge refused');
    const text = res.content.find((b) => b.type === 'text');
    return JSON.parse(text?.type === 'text' ? text.text : 'null');
  },
});

skills/attend/references/judges.md has the same judge on the OpenAI Responses API and the Vercel AI SDK. The default prompt is DEFAULT_SYSTEM, and each route's default criterion is in DEFAULT_CRITERIA. Override them with system and criteria.

jevJudge

jevJudge from attend/jev calls TypeSafe's Jev, a classifier built for this kind of question, with your own TypeSafe key. It asks one choice question for the route and one noul question per predicate, and sends the state as JSON. The model defaults to jev-latest; jev-preview, jev-1.14.0 and jev-1.13.0 are the others. Pass endpoint to go through OpenRouter or LiteLLM, and longContext: true for 262k tokens of context at some cost in speed and accuracy. Its probabilities are real, so the act threshold works with it.

import { jevJudge } from 'attend/jev';

export const judge = jevJudge({ apiKey: process.env.TYPESAFE_API_KEY ?? '' });

decisionsJudge

decisionsJudge from attend/decisions calls the OpenAI Decisions API. It asks one choice question for the route and one predicate question per predicate.

import { decisionsJudge } from 'attend/decisions';

export const judge = decisionsJudge({ apiKey: process.env.OPENAI_API_KEY ?? '', model: '<model id>' });

Both hosted judges retry network errors and 408, 429 and 5xx responses, and dead-letter the window at once on any other 4xx, on a billing or quota 429, and when the model refuses the route question. Every error from a response carries status, requestId and retryAfterMs. Both return the classifier's confidence in the route on decision.confidence.

The window budget

A window must never overload the judge that reads it. A judge declares its input budget (jevJudge declares 32k tokens, or 262k with longContext; llmJudge and decisionsJudge take contextTokens), and attend turns it into a size limit on the open window: (contextTokens - reservedTokens) * windowShare * bytesPerToken, with windowShare defaulting to 0.5. attend keeps a running size estimate for each window as events arrive. A window that reaches its share settles early with reason size, and a judge that still gets more than fits drops sheddable events (consequence and context, oldest first) when it renders and never an act or a flush. As a backstop, a provider's context-length rejection triggers exactly one retry with the window halved. A judge that declares no budget leaves the limit off. Set settle.maxWindowBytes yourself to override.

The act threshold

By default the judge's route stands. A judge that returns real probabilities (jevJudge, decisionsJudge) also ranks how much the agent should act, and only your app knows whether an interruption or a missed moment costs more. Set actThreshold in (0, 1] on createAttend, and the route becomes an acting route (interrupt or ask, whichever is likelier and offered) exactly when p(interrupt) + p(ask) is at least the threshold, and a quieter route (skip, note, queue or wait, by probability) otherwise. It is off by default. A judgment without probabilities is never changed, and a changed route emits a routeThresholded trace. The threshold is applied before the decision is logged, so replays stay deterministic.

Pick it from your own traffic instead of guessing. Replay a recorded stream with the threshold off, record each judgment, and suggestThreshold from attend/testing returns the smallest threshold whose acts fit an interruptions-per-hour budget, with the rate at every candidate:

import { suggestThreshold } from 'attend/testing';
import type { Judge, Judgment } from 'attend';

export const judgments: Judgment[] = [];
export const recording = (judge: Judge): Judge => async (input, signal) => {
  const judgment = await judge(input, signal);
  judgments.push(judgment);
  return judgment;
};

// After replaying an hour of recorded events through createAttend({ judge: recording(judge), ... }):
export const { threshold, candidates } = suggestThreshold(judgments, { durationMs: 3_600_000, maxActsPerHour: 6 });

threshold is null when no threshold fits the budget. llmJudge's probabilities are near one-hot, so a threshold changes little there.

Images

Give createAttend a capture(scope, signal) that returns a JudgeImage (base64 data and a mimeType) or a list of them, such as a screenshot of what the user is looking at, and every judge call carries them as input.images. A capture that throws or takes longer than captureTimeoutMs (default 500) never fails the call: the judge runs without images and an observer sees a captureFailed trace. jevJudge sends PNG and JPEG within the provider's limits (280 to 1568 px per side, 16 MiB per file) and drops the rest, reporting each through onImageError; decisionsJudge sends up to 128 of any format. Either one retries once without images when the provider rejects them. llmJudge hands them to complete as images, as the snippet above shows.

Your own judge

Any function with the judge signature works, and judges compose as functions. This one answers a direct message without calling a model:

import type { Judge } from 'attend';

export const withShortcut =
  (judge: Judge): Judge =>
  async (input, signal) =>
    input.window.reason === 'flush' ? { route: 'interrupt' } : judge(input, signal);

To ask the judge yes-or-no questions beside the route, pass predicates to createAttend, such as { asked: 'Did the user ask the agent for this?' }. Each answer arrives on decision.predicates as a probability.

Sinks

A sink delivers decisions to the agent. attend calls sink.deliver(decision, signal) at least once per decision, one decision at a time per scope, oldest first. After a crash it can deliver the same decision again with the same id, so make the sink idempotent on decision.id, which is unique across scopes. Throw to retry, or throw a PermanentError to dead-letter at once.

routes() builds a sink from one handler per delivered route. A sink never sees skip or wait. This excerpt from examples/postgres-worker.ts writes every decision to an inbox table that the agent's runtime reads:

async function enqueue(decision: Decision): Promise<void> {
  await pool.query('INSERT INTO agent_inbox (scope, decision_id, route, notice) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING', [
    decision.scope,
    decision.id,
    decision.route,
    defaultNotice(decision),
  ]);
}

const attend = createAttend({
  store: postgresStore(pool),
  kind: (event) => (event.name === 'message.created' ? 'flush' : event.name === 'job.completed' ? 'closing' : 'consequence'),
  judge: async ({ window }) => ({ route: window.reason === 'flush' ? 'interrupt' : 'queue' }),
  sink: routes({ interrupt: enqueue, queue: enqueue, note: enqueue, ask: enqueue }),
  deadLetter: async (letter) => {
    await pool.query('INSERT INTO agent_dead_letters (scope, id, letter) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING', [letter.scope, letter.id, JSON.stringify(letter)]);
  },
});

deadLetter is required. It receives a window that no judge or sink could take, with the stage that gave up, the error message and the attempt count. attend retries a dead letter with capped backoff until the handler succeeds, so write it to durable storage and keep the handler simple. Dead letters travel in their own lane, so a failing handler never holds back decisions; once the lane holds maxDeadLetters of them, push and a source's emit reject with OutboxFullError until it drains, and nothing is dropped.

Storage

One ScopeStore holds each scope's latest snapshot, the inputs committed since it, the due index and the decision log. attend replays the short tail of inputs over the snapshot to recover a scope, so a restart loses nothing.

Each store also takes retain (decision rows kept by count or age), so the log does not grow without bound. Every store passes the same conformance and crash suites.

Several processes can share one store. Any of them can push, and any number can run or serve the log: a write names the stored version it read, a writer that loses reloads and applies its input again, and every input is idempotent. Give every process the same kind function, because the process that ingests an event is the one that kinds it. A scope can take several sources at once; the record keeps one cursor per Source.key. Attach a given source to a scope from one process only, since two processes running the same source would overwrite each other's cursor, and a second attend.attach of the same key in one process throws.

Each adapter also has an EventLog: memoryLog, sqliteLog, postgresLog and redisLog. When producers live in other processes and you want a durable shared log in front of attend, append to the log and attach logSource(log, scope) instead of calling push.

Guarantees

• One snapshot and a short tail per scope. The snapshot holds the open window, the judge slot and the outbox. A write appends the inputs that changed state, and a snapshot replaces the tail once it grows past the snapshot's own size. A source's cursor commits in the same write as the events it covers.

• Versioned writes. Each write names the version it read. A writer that loses waits a few jittered milliseconds, reloads and applies its input again. After 8 losses in a row it gives up: push and flush reject with ConflictError, and work in flight recovers when its lease expires.

• Effects only after a committed write. attend calls the judge and the sink only for state it has persisted. push and flush persist and return; the runner settles, judges and delivers. If a process crashes after a write and before its effect, the work stays in the store, and a runner retries it when its lease expires.

• The decision log is the delivery for subscribers. The decision row commits with the state change that made it. Subscribers read it by cursor, so a reconnect resumes where it stopped; a cursor that fell behind retention is reported as truncated, never silently skipped.

• At-least-once delivery with stable ids. A decision's id is a hash of the scope and its window's event ids, so a redelivery carries the same id and no two scopes share one. Deliveries in one scope run one at a time, oldest first.

• Dead letters are never dropped. The judge and the sink each get retryBackoffMs.length retries. A window that still fails becomes a dead letter, which attend retries with capped backoff until deadLetter accepts it. A lane that fills to its cap (maxOutbox decisions, maxDeadLetters dead letters) makes push and a source's emit reject until it drains.

• Crash-tested at every commit boundary. crashSuite runs a scripted session once, then once more for each commit with a crash just before it and again just after it. Every run must deliver the same decision ids and dead-letter ids. CI runs it against the memory, SQLite, Postgres and Redis stores.

Testing with a fake clock

attend/testing makes settling deterministic. Inject fakeClock(), answer with scriptedJudge, and assert on recordingSink. This excerpt is from examples/testing.ts, which runs with node --test:

import assert from 'node:assert/strict';
import { test } from 'node:test';
import { createAttend, memoryStore, type DeadLetter, type Event, type Kind } from 'attend';
import { fakeClock, recordingSink, scriptedJudge, type ScriptedAnswer } from 'attend/testing';

const kind = (event: Event): Kind => {
  if (event.name === 'user.message') return 'flush';
  if (event.name === 'build.finished') return 'closing';
  return 'consequence';
};

function setup(script: ScriptedAnswer[]) {
  const clock = fakeClock();
  const judge = scriptedJudge(script);
  const sink = recordingSink();
  const letters: DeadLetter[] = [];
  const attend = createAttend({
    store: memoryStore(),
    kind,
    judge,
    sink,
    clock,
    deadLetter: (letter) => {
      letters.push(letter);
    },
  });
  const controller = new AbortController();
  const running = attend.run({ signal: controller.signal });
  const stop = async () => {
    controller.abort();
    await running;
  };
  return { clock, judge, sink, letters, attend, stop };
}

test('a finished build is judged after the closing grace, with its log lines', async () => {
  const { clock, judge, sink, letters, attend, stop } = setup([{ route: 'queue' }]);
  await attend.push('repo', [
    { id: 'log-1', at: 0, name: 'build.log', data: { line: 'pnpm test' } },
    { id: 'done-1', at: 0, name: 'build.finished', data: { ok: false } },
  ]);
  await clock.advance(499);
  assert.equal(judge.calls.length, 0);
  await clock.advance(1);
  assert.deepEqual(judge.calls[0]?.window.events.map((e) => e.id), ['log-1', 'done-1']);
  assert.deepEqual(sink.delivered.map((d) => d.route), ['queue']);
  assert.deepEqual(letters, []);
  await stop();
});

• clock.advance(ms) fires every timer due on the way and lets each one's work finish before it returns.

• scriptedJudge answers in order, repeats its last answer, and records every input in calls. An answer can also be { fail: 'message' }, { fail: 'message', permanent: true } or { hang: true }.

• recordingSink records delivered and attempts. failNext(n) makes the next n deliveries throw.

• Pass the same clock to mcpEventsSource and logSource through their clock option, so their polling runs on fake time too.

• harness runs attend on a fake clock over a store that it can crash at any commit. Use it for restart tests.

Writing an adapter

A store implements ScopeStore (see src/store/types.ts, where each member's guarantee is in its doc comment):

• load(scope) returns the latest snapshot and the inputs committed after it, or null.

• commit(writes) applies each scope's write only if its stored version equals expectedVersion (0 means no stored scope), and returns 'ok' or 'conflict' for each. The inputs, the snapshot, the due value and the decision rows of one write land together or not at all. Of N concurrent writes at one version, exactly one is 'ok'.

• due(now, limit, after?) pages the scopes whose due value has passed. It reads and claims nothing.

• remove(scope) forgets the scope.

• log is the DecisionLog: the epoch, bounds, reads by position, find, wait and trim.

Run the conformance suites and the crash suites against your adapter with any test runner. Each returns named async cases. examples/adapter-suites.ts runs them against memoryStore with node:test:

import { test } from 'node:test';
import { memoryStore } from 'attend';
import { crashSuite, multiScopeCrashSuite, scopeStoreConformance, decisionLogConformance } from 'attend/testing';

for (const c of scopeStoreConformance(() => memoryStore())) test(c.name, c.run);
for (const c of crashSuite(() => memoryStore())) test(c.name, c.run);
for (const c of decisionLogConformance(() => memoryStore())) test(c.name, c.run);
for (const c of multiScopeCrashSuite(() => memoryStore())) test(c.name, c.run);

The factory must return a fresh, empty store for each case. For a database, give each case its own table prefix or key prefix. An EventLog adapter runs logConformance(make, { canBlock }), where canBlock says whether read waits for blockMs.

Defaults

These are round starting points, not tuned values. Override any of them with settle in createAttend. Times are in milliseconds.

run() sweeps the store for due scopes every sweepMs (1000), up to sweepLimit (100) scopes per sweep.

Agent skill

The repo ships an Agent Skill that leads your coding agent through wiring attend into your app. It works out which events the app emits, which of them are signals, and which deserve which reaction, then wires it with one small module and a few call sites at your existing event chokepoints. It tells you in a few lines what it will do, asks one batch of questions with a recommended default for each, and tunes the act threshold from a replay of your own events. To install it in Claude Code:

/plugin marketplace add acquiredintelligence/attend
/plugin install attend@attend

For Codex or Cursor, copy skills/attend into your project's .agents/skills/. The npm package includes the skill at node_modules/attend/skills/attend. The third-party Vercel Labs CLI also installs it with npx skills add acquiredintelligence/attend. That CLI is not part of this project.

Status

attend is v0, and its API can change before 1.0. MCP Events is a working-group draft, so its wire shapes can change too. attend/mcp follows the current draft.

License

MIT. Built by the team behind Zettel.