Agents and events

Long Tail publishes an event every time a workflow, task, escalation, activity or agent changes state. An agent subscribes to those events, runs a workflow for each match, and keeps goals, rules and a knowledge domain alongside. This page covers declaring an agent, the subject hierarchy, the transport adapters and the topic catalog.

Declare an agent

Agents are declared in start() next to your workers. This one reacts to failed workflows and runs an hourly sweep:

index.ts

import { start } from '@hotmeshio/long-tail';

await start({
  database: { connectionString: process.env.DATABASE_URL },
  workers: [{ taskQueue: 'ops', workflow: captureFailure }],
  agents: [
    {
      name: 'health-monitor',
      description: 'Captures diagnostics when a workflow fails.',
      goals: 'Detect failures early and capture state before they cascade.',
      rules: 'Never restart a failed workflow. Capture state and escalate to a person.',
      knowledge_domain: 'system-health',
      subscriptions: [
        {
          topic: 'system.workflow.*.failed',
          reaction_type: 'durable',
          workflow_type: 'captureFailure',
          input_mapping: {
            data: { workflowId: '{event.workflowId}', error: '{event.data.error}' },
          },
          execute_as: 'ops-bot',
        },
      ],
      schedules: [
        { cron: '0 * * * *', reaction_type: 'durable', workflow_type: 'captureFailure', envelope: { data: { sweep: true } } },
      ],
    },
  ],
});

Every agent has the same five parts:

Part	Field	What it does
Identity	name, description	Lowercase kebab-case name shown in the sidebar, event payloads and logs.
Motivation	goals, rules	Goals say what the agent is trying to achieve. Rules are the guardrails it keeps even when a goal points elsewhere.
Knowledge	knowledge_domain	The domain the agent accumulates context into. Other agents can read it.
Subscriptions	subscriptions[]	Topic pattern to reaction: when this event arrives, run that.
Schedule	schedules[]	5-field cron expressions (UTC), each targeting a workflow with a static envelope.

An agent can be event-driven, scheduled, or both. Pausing an agent stops every subscription and schedule and keeps its knowledge and history. Reactivating it re-arms them.

Note

Configuring an agent or subscription takes builder access and the right to act as the identity it runs as. An agent or subscription that runs as lt-system, and any capability subscription, is configured by a superadmin.

Subscriptions

A subscription has a topic pattern, a reaction, an optional filter and an optional input mapping.

Topic patterns

Topics are dot-delimited. Patterns use NATS-style wildcards:

Pattern	Matches
system.workflow.*.failed	Any workflow failure. * matches exactly one token.
system.escalation.reviewer.>	Every event on the reviewer queue. > matches one or more trailing tokens and must be last.
system.escalation.*.*.created	Every new escalation on every queue.
app.orders.>	Every application event under app.orders.

Reactions

reaction_type	Runs
durable	A registered TypeScript workflow (workflow_type). Supports conditional waits.
pipeline	A compiled YAML workflow (pipeline_id).
mcp_query	An AI-driven tool sequence from mcp_prompt.

The run identity is the subscription’s execute_as, then the agent’s own service account, then lt-system. That identity must be allowed to start the target workflow.

Input mapping and filters

input_mapping builds the workflow envelope from {event.*} templates that resolve at runtime. Without a mapping, the workflow receives event.data as its data.

filter matches against event.data. Every key must match:

{
  "status": 422,
  "extension": { "$in": ["png", "jpg"] },
  "trace_id": { "$exists": true },
  "source": { "$ne": "test" }
}

Exactly once across containers

Every container receives every event. Each one computes the same workflow ID from the subscription and the event, then starts the workflow with that ID:

agent-{agentId}-{subscriptionId[0:8]}-{eventKey}

eventKey is the event’s workflowId, taskId or escalationId. A custom event without those IDs uses the first 12 hex characters of a SHA-256 over its timestamp, type and data. The first container to start the workflow wins and the others receive a duplicate-ID error, which the registry logs and skips. One event, one workflow run, however many containers you run.

Cron schedules run through HotMesh’s Virtual.cron, which fires each tick once.

Tip

When you publish your own events, put a stable identifier in data (an orderId, a serialNumber). Two events with identical payloads and timestamps hash to the same key and run one reaction.

Agent lifecycle events (system.agent.{name}.started, completed, failed) are published by each container that handles the event. They are observability signals.

The subject hierarchy

System subjects start with system. and carry the identifiers you filter on:

Family	Subject	Adds to the envelope
workflow	system.workflow.{id}.{started|completed|failed}	workflowId, workflowName, taskQueue, status
task	system.task.{taskId}.{created|started|completed|escalated|failed}	taskId
escalation	system.escalation.{role}.{id}.{created|resolved|claimed|released|cancelled|reassigned|expired}	escalationId, role, status
activity	system.activity.{wfId}.{activity}.{started|completed|failed}	activityName
milestone	system.milestone.{wfId}	milestones
agent	system.agent.{name}.{started|completed|failed|status_changed}	agent id and name
knowledge, file	system.knowledge.{domain}.{stored|deleted}, system.file.{stored|deleted}	payload only

Escalation subjects put the role (the queue) second so you can scope three ways:

system.escalation.reviewer.>          one queue
system.escalation.*.*.created         one verb across queues
system.escalation.*.{id}.>            one row across role moves

The role token is sanitized to [A-Za-z0-9_-]. A reassigned event publishes under both the old and the new role. An admin reassign publishes claimed with reassigned_from, and an unassign publishes released with released_by.

The envelope

Three fields are universal: type (the subject), timestamp, and an id minted when the publisher omits one. source and data are optional. System families add the fields in the table above. A custom app.* event is the base plus data.

An escalation event is a notification. Its data is the row’s routing scalars plus every metadata facet (orderId, serialNumber, schema_version), with provenance such as resolved_by or released_by. envelope, escalation_payload, resolver_payload and metadata.form_schema stay in the database. A subscriber that needs them reads the row by id through the authenticated API.

Events are fire-and-forget. A failed adapter is logged and skipped, and events are not replayed when a workflow recovers. Treat the row as the record and the event as the nudge.

Transports

Events fan out to every registered adapter. Socket.IO is built in for the dashboard. Add NATS or your own:

NATS

index.ts

await start({
  database: { connectionString: process.env.DATABASE_URL },
  workers: [/* ... */],
  events: { nats: { url: 'nats://localhost:4222' } },
});

Events publish to {subjectPrefix}.{event.type}. The default prefix is lt.events, so a new escalation on the reviewer queue lands on lt.events.system.escalation.reviewer.{id}.created. On shutdown the adapter drains in-flight publishes.

SNS

sns-adapter.ts

import { SNSClient, PublishCommand } from '@aws-sdk/client-sns';
import type { LTEventAdapter, LTEvent } from '@hotmeshio/long-tail';

export class SnsEventAdapter implements LTEventAdapter {
  private client = new SNSClient({ region: 'us-east-1' });
  constructor(private topicArn: string) {}

  async connect() {}

  async publish(event: LTEvent) {
    await this.client.send(new PublishCommand({
      TopicArn: this.topicArn,
      Message: JSON.stringify(event),
      MessageAttributes: { eventType: { DataType: 'String', StringValue: event.type } },
    }));
  }

  async disconnect() { this.client.destroy(); }
}

Webhook

webhook-adapter.ts

import type { LTEventAdapter, LTEvent } from '@hotmeshio/long-tail';

export class WebhookEventAdapter implements LTEventAdapter {
  constructor(private url: string) {}
  async connect() {}
  async publish(event: LTEvent) {
    await fetch(this.url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(event),
    });
  }
  async disconnect() {}
}

Register custom adapters with events: { adapters: [new SnsEventAdapter(arn), new WebhookEventAdapter(url)] }. start() connects them at boot and disconnects them on shutdown.

Listening in process

start() registers a CallbackEventAdapter. From the SDK, subscribe with lt.events.on, which takes the same patterns and returns an unsubscribe function:

const off = lt.events.on('system.escalation.reviewer.>', (event) => {
  console.log(event.type, event.escalationId);
});

For tests, register InMemoryEventAdapter and assert against adapter.events.

Dashboard transport

The dashboard listens on Socket.IO by default. When the API and workers run as separate processes, set EVENT_TRANSPORT=nats with NATS_URL, NATS_WS_URL and NATS_TOKEN. With events.nats.wsProxy set, browsers reach NATS through Long Tail at <basePath>/nats-ws with a short-lived ticket (NATS_WS_TICKET_TTL_SECONDS, default 300) and subscribe only. Builders receive events whole. Everyone else receives escalation routing fields and reads records through the scoped API.

The topic catalog

The catalog lists every known topic with a description, a JSON Schema for event.data, an example payload and the agents that subscribe. Browse it in the dashboard at /topics. When you create a subscription, the topic field searches the catalog and previews the {event.data.*} fields available to your input mapping.

Topics enter the catalog three ways:

Source	How	Ownership
System	Seeded at startup for every system.* subject	Read-only
Code	topics[] in start()	Follows configSource, or the entry’s reset
Runtime	publish_event with an unknown topic registers it, using the payload as the example	The database

index.ts

await start({
  database: { connectionString: process.env.DATABASE_URL },
  workers: [{ taskQueue: 'orders', workflow: processOrder }],
  topics: [
    {
      topic: 'app.orders.created',
      description: 'Fired when a new order is placed.',
      payload_schema: {
        type: 'object',
        properties: { orderId: { type: 'string' }, total: { type: 'number' } },
      },
      example_payload: { orderId: 'ord-123', total: 99.99 },
      tags: ['orders'],
      reset: true,
    },
  ],
});

With reset: true, every boot writes the description, schema and tags from code, so the definition lives in git. Without it, the first boot seeds the row and dashboard edits persist. See Code-owned configuration.

Application topics follow app.{namespace}.{entity}.{action}. The publish_event MCP tool adds the app. prefix when it is missing. A topic’s category is its first segment.

Topics over MCP

Tool	Does	Access
list_topics	Browse topics by category or search term, with schemas and subscriber counts	Any account
register_topic	Declare a topic and its schema before the first publish	Builder
publish_event	Publish an app.* event to every matching subscriber	Builder

An agent can list topics, read a schema and create a subscription entirely through tool calls. Publishing a system. or agent. subject over REST takes superadmin.

Related

MCPConnect agents and expose people as tools.

IdentityService accounts and execute_as.

Exactly-once event subscriptions

Autonomous topic wiring