@emuanalytics/flow-agents

Support library for Flo.w agents: a Mastra agent process registers with flow-engine over a long-lived socket and serves invocations pushed back down it.

Modelled on @emuanalytics/flow-pipelines — same Context shape, same logger, same relationship to the platform. Where the two differ, the difference is the transport: a pipeline dials out to sources and sinks, an agent holds one connection to flow-engine and waits.

Spike code. Built for ISSUE-8bee under the flow-agents wayfinder map (ISSUE-438d). It is private: true — the permanent home and publish story are deliberately still open.

Node 22

This package requires Node >= 22.13.0, which is what the current Mastra line declares. So does the rest of the repo — the Node 20/22 split this package was built alongside is gone (ISSUE-326c), and one root .nvmrc now covers every package. There is no shell hook, so nvm use is still explicit:

nvm use    # reads the repo-root .nvmrc

The container image bases on node:22 unconditionally.

Writing an agent

Two entry points, deliberately (ISSUE-f6b8):

// src/mastra/index.ts — idiomatic Mastra, so `mastra dev` can bundle it
import { createAmazonBedrock } from '@ai-sdk/amazon-bedrock';
import { Agent } from '@mastra/core/agent';
import { Mastra } from '@mastra/core/mastra';
import { systemPrompt } from '@emuanalytics/flow-agents';
import config from 'config';

const bedrock = createAmazonBedrock({ region: config.get('bedrock.region') });

const sentiment = new Agent({
  id: 'sentiment',
  name: 'Sentiment Classifier',
  instructions: systemPrompt('Classify the sentiment of the input.', { key: 'agents.sentiment' }),
  model: bedrock(config.get('bedrock.model')),
});

export const mastra = new Mastra({ agents: { sentiment } });

// src/index.ts — the bootstrap the image runs
import { AgentContext, serveAgents } from '@emuanalytics/flow-agents';
import config from 'config';
import { mastra } from './mastra';

await serveAgents(mastra, {
  context: new AgentContext({
    environment: config.get('agent.environment'),
    logLevel: config.get('logger.level'),
    flow: config.get('flow'), // { host, apiKey } — app-scoped key with `write` on `agents`
  }),
  version: '1.0.0',
  onFatal: () => process.exit(1), // see Fatal conditions below
  healthcheck: { port: 8080 },    // opt-in; see The healthcheck port below
  agents: [
    {
      agent: 'sentiment',                              // id on the Mastra instance
      id: config.get('agents.sentiment.id'),           // flow-engine slug, PK with the app id
      description: 'Classifies sentiment',
      capabilities: { mode: 'single-shot', maxConcurrent: 8 },
    },
  ],
});

You build Mastra; this library serves it. Nothing is wrapped — @mastra/core is a peer dependency and appears in the public types, so Mastra's own docs and examples apply unchanged.

Two ids

agent names the agent on your Mastra instance. id is the flow-engine slug, which forms the primary key with the application id and carries the environment by convention (sentiment-classifier-prod). They are kept separate so a Mastra rename is not a flow-engine migration.

Overriding the system prompt

systemPrompt(base, { key }) resolves <key>.systemPrompt (replaces) and <key>.systemPromptAppend (appends) from node-config, per invocation. The effective prompt is logged at startup — deployed behaviour can differ from the repo, and that log line is the whole mitigation.

Operator configuration: metadata

An agent row in flow-engine has a metadata column that only operators write — the console, the CLI or PATCH /api/v1/agents/:id — and registration never clobbers it. The engine forwards it on every invocation, so it is how a deployed agent is reconfigured without a redeploy:

// PATCH /api/v1/agents/sentiment-classifier-prod
{ "metadata": { "tone": "formal", "limits": { "maxRefunds": 3 } } }

agentMetadata() reads it off the request context Mastra hands a dynamic instruction or a tool:

import { agentMetadata } from '@emuanalytics/flow-agents';

interface Config {
  tone: string;
  limits: { maxRefunds: number };
}

new Agent({
  id: 'assistant',
  name: 'Assistant',
  // Dynamic instructions, so the value is read per invocation rather than at construction.
  instructions: ({ requestContext }) => {
    const { tone = 'neutral' } = agentMetadata<Config>(requestContext);
    return `Answer briefly, in a ${tone} tone.`;
  },
  tools: {
    refund: createTool({
      // The same context, so a tool reads the same configuration.
      execute: async ({ context, requestContext }) => {
        const { limits } = agentMetadata<Config>(requestContext);
        return issueRefund(context.amount, limits?.maxRefunds ?? 1);
      },
      ...
    }),
  },
  model,
});

Three things are worth knowing:

• The type parameter is an assertion, not a validation. An operator can put anything in the column, so treat every key as optional and default it at the point of use.
• It is always an object. {} when nothing is configured, and also when there is no request context at all — a mastra dev playground call, or a unit test calling a tool directly — so a reader never distinguishes those from "configured empty".
• A client cannot reach it. The engine sends metadata as a sibling of the request body, so a client-supplied metadata lands at input.metadata where nothing reads it. It is the same structural protection principalId has.

This is capabilities pointed the other way: capabilities is what the agent declares about itself at registration, metadata is what an operator tells the agent.

What the library does

• One socket, N registrations. Registers every agent on connect and re-registers on every reconnect with reconnect: true, so the engine's startedAt keeps meaning "when this process started".
• Heartbeats every 30s. Advisory: routing liveness is the socket's own presence.
• Serves invocations. Request/response resolves an ack; streaming emits start, coalesced deltas and exactly one terminal done or error.
• Honours cancel by aborting the model call — the client is gone, so the point is to stop paying for tokens.
• Refuses over capabilities.maxConcurrent with busy, which the engine turns into a 429.
• Owns the logs and the telemetry switch — see below.
• Serves a liveness port, when you ask it to — see below.

It does not queue anything while disconnected, and it does not retry: an invocation costs tokens and may have run tools, so every failure is final and the client decides.

Logging and telemetry

Mastra logs through your logger. The library calls setLogger on your Mastra instance with an adapter, so the container ships one JSON stream rather than winston's plus Mastra's default ConsoleLogger; Mastra's lines carry component: "mastra" so they stay tellable apart. Its diagnostics are replaced, not silenced — discarding them buys nothing once the format matches.

invocationId rides this library's lines and only those, which is what makes them the authoritative correlation surface for a support question about one client request. Mastra's logger is one instance per Mastra, shared by concurrent invocations, so it has no per-call seam to bind it to.

Mastra's own PostHog telemetry is switched off: serveAgents defaults MASTRA_TELEMETRY_DISABLED=1 so you inherit it without remembering, and the template sets it in the Dockerfile and on the dev script as well. It is defaulted rather than set — export MASTRA_TELEMETRY_DISABLED=0 and you get it back.

mastra dev never calls serveAgents, so the playground keeps Mastra's own logger. That is correct: it is a dev-time sibling process, not the agent.

The healthcheck port

healthcheck: { port, host? } opens an HTTP port that answers 200 iff the socket is connected and every agent in the process is registered on it, and 503 otherwise. It is the one thing about an agent that is not dial-out, which is why it is off unless you set it: opening a listening socket is a deployment posture, and there is no default port to guess at. host defaults to every interface, since the orchestrator probes from outside the process.

Liveness, not readiness, and no routes. Every request gets the same verdict whatever the path or method — a liveness port reports one fact, and routing exists to tell several apart. Agent metrics on this port are a future decision, not an extension of this one.

registered goes false the moment the socket drops and again while a NotRegistered heartbeat is being repaired, so a brief 503 during a reconnect is the truth rather than a flap: a reconnected socket has no room membership until register has been replayed on it, and until then the engine cannot route to this process. A fatal condition also clears it, so a process on its way out never answers 200.

The listener is unref'd, like the heartbeat timer: reporting on the work must never be the thing keeping alive a process that has lost it.

Fatal conditions

Two states an agent process cannot wait its way out of: flow-engine explicitly closing the connection (no engine path does this today, so it is unexplained rather than a signal to obey), and a rejected registration (the key lacks write on agents for that id). Registration is all-or-nothing — a process serving one of its two agents is the same silent half-failure.

Either one aborts in-flight invocations and calls onFatal. Wire it to process.exit(1): the orchestrator is the agent's only supervisor, and non-zero is what makes it restart with backoff and a visible crashloop. Leave it unset and the library rethrows, so the fallback is a stack trace rather than silence.

Deliberately not fatal, and left to socket.io-client's own reconnection: an unreachable engine (transient, and crashlooping the estate would make an outage worse) and a transport close.

Testing your agent

@emuanalytics/flow-agents/testing exports a fake engine that accepts a real Socket.IO connection, plus a mock model so nothing needs Bedrock:

import { FakeFlowEngine, mockModel } from '@emuanalytics/flow-agents/testing';

const engine = await FakeFlowEngine.start();
const server = await serveAgents(mastra, { context: contextFor(engine.url), agents: [...] });

await engine.waitForRegistration('sentiment-test');
const ack = await engine.invoke('sentiment-test', { input: 'I love it' });

Set engine.metadata to stand in for the agent row's operator configuration, and read model.systemPrompts to assert on the prompt an agent actually built — which is the only way to check a dynamic one, since the mock answers the same text whatever it is asked:

engine.metadata = { tone: 'formal' };
await engine.invoke('assistant-test', { input: 'Say something' });

expect(model.systemPrompts.at(-1)).toContain('formal');

Scripts

| Script | | | --- | --- | | pnpm test | vitest, including the real-socket integration tests | | pnpm run build | unbuild → dist (ESM + CJS + types) | | pnpm run typecheck | tsc --noEmit | | pnpm run lint | eslint |