Concorde

Appearance

@shutter-network/concorde/signals ​

The Signal Worker owns the Signal queue, Signal Handler dispatch and Run execution. A Signal is something that arrived and may make the agent act, emitted by a Producer, which is anything inside the Gateway trusted to write one. A Run is one execution of the agent over one Prompt. The Worker claims Signals in the order they arrived and runs one Run at a time, whatever Session that Run is in.

Most deployments meet this subpath as a vocabulary rather than as a constructor. SignalHandler is what an Operator writes, and it is the framework's primary extension point in the way an endpoint handler is a web framework's: it takes a Signal and answers with Prompts, and SignalHandlers is the map from a kind to one of them. Runtime is the other seam, the single method an Agent Implementation is driven through. createSignalWorker builds the Worker itself, and SignalWorker is what comes back. Its programmatic API is emit, which a Producer writes a Signal through.

templateHandler is a Handler most deployments start from rather than writing SignalHandler by hand: it renders one Prompt per Signal from a Handlebars template, and TemplateHandlerOptions is where the template source, the Session and the values it substitutes are stated. It is an ordinary Handler built by an ordinary function, so a deployment that outgrows it returns one of its own from the same place and unwires nothing.

createGateway builds a Worker already, so reach for the constructor only when you assemble a Gateway by hand. Either way the Handler map is a construction option, so a Handler that emits back into the same Worker is built after the Worker and assigned in.

The two tables are here beside the constructor, for the schema an Operator generates their migrations from. They reference no other component's table, so that schema can carry them alone.

Example ​

A Signal Handler, and a Producer that emits into it.

ts
import type { Db } from "@shutter-network/concorde/db";
import type { Signal, SignalHandler, SignalWorker } from "@shutter-network/concorde/signals";

const greet: SignalHandler<{ name: string }> = {
  handle: (signal: Signal<{ name: string }>) => [
    { session: `user_${signal.payload.name}`, text: `Say hello to ${signal.payload.name}.` },
  ],
  // Runs once the Run above has finished, however it finished.
  post: (signal, outcome) => {
    if (outcome.failed) console.error(`nobody greeted ${signal.id}`);
  },
};

// A Producer emits inside its own transaction, so the row and the wakeup commit together.
async function greetSomebody(db: Db, worker: SignalWorker, name: string): Promise<string> {
  return db.tx((tx) => worker.emit(tx, { kind: "greet", payload: { name } }));
}

Type Aliases ​

EmittedSignal ​

ts
type EmittedSignal = {
  readonly kind: string;
  readonly payload: unknown;
};

What a Producer hands to SignalWorker's emit.

Properties ​

kind ​
ts
readonly kind: string;

Selects exactly one Signal Handler, and is the whole of what dispatch looks at.

One kind never reaches two Handlers. Fanning out is one Handler answering with several Prompts, so there is no second mechanism for it here.

payload ​
ts
readonly payload: unknown;

Arbitrary JSON, taken as fact and never interpreted.

The Signal Worker believes whatever a Producer writes, including any claim about who the Signal came from. That is why a Producer is a part of the Gateway rather than a peer outside it, and why attribution is a term in a Producer's payload contract rather than a column here.

PostOutcome ​

ts
type PostOutcome = {
  readonly failed: boolean;
};

What the post phase is told about the Signal it is closing out.

failed is true if any Run failed, and true as well if handle threw before there were any, so it says the Signal came to nothing rather than that the agent ran and came back unhappy.

Properties ​

failed ​
ts
readonly failed: boolean;

Prompt ​

ts
type Prompt = {
  readonly session: string | null;
  readonly text: string;
};

What a Handler produces from a Signal, and the only form in which anything reaches the agent.

session names the Session this Prompt continues, and null asks for a fresh one. A Session is the agent's own conversational state, kept by the Agent Implementation and continued by being named again, so it is what makes one Prompt remember an earlier one. It organises context and does not partition it: the agent reads every Signal, Run and Message over the Agent server whatever Session it is in.

The framework fixes no session topology and validates no name. One shared Session, one per User, one per Run and any mixture of those are written here and nowhere else. Any string reaches the Runtime as it was written, and a name the Agent Implementation refuses fails that one Prompt's Run, carrying that program's own message, while the Prompts beside it still run.

Properties ​

session ​
ts
readonly session: string | null;
text ​
ts
readonly text: string;

RunOutcome ​

ts
type RunOutcome =
  | {
      readonly ok: true;
    }
  | {
      readonly error: string;
      readonly ok: false;
    };

How a Run ended, and the whole of what a Runtime reports.

It carries none of the agent's output. Nothing in the framework reads what the agent said: the Agent Implementation keeps its own Session files, and the agent records anything it wants kept through the Agent server or the Workspace.

A failure carries a message because that message is what RunRecord's error holds and what every later read of the Run answers with. Nothing parses it, so it is written for a person.

RunPrompt ​

ts
type RunPrompt = Omit<Prompt, "session"> & {
  readonly session: string;
};

The Prompt as a Runtime receives it: what a Handler wrote, with the Session resolved.

Two types rather than one nullable one. The null a Handler may write is a request for a fresh Session rather than a value, and the Signal Worker answers it before any Runtime is called, naming that Session after the Run it belongs to. So there is no fresh-Session case to handle here, no naming convention for a Runtime to invent, and every Run records the name it really ran under.

Type Declaration ​

session ​
ts
readonly session: string;

RunRecord ​

ts
type RunRecord = {
  readonly endedAt: string | null;
  readonly error: string | null;
  readonly id: string;
  readonly prompt: string;
  readonly session: string | null;
  readonly signalId: string;
  readonly startedAt: string | null;
  readonly state: RunState;
};

A Run as the agent reads it: one Prompt, in one Session, and how it went.

signalId is the Signal whose Handler wrote the Prompt, and prompt is the text the agent was given rather than the template it came from. session is a plain name and a reference to nothing. Every Run the Worker records now carries one, and it stays nullable because rows written before that still hold null.

The timings are ISO 8601 strings, or null for a Run that has not reached that point, so a running Run has a startedAt and no endedAt.

Properties ​

endedAt ​
ts
readonly endedAt: string | null;
error ​
ts
readonly error: string | null;
id ​
ts
readonly id: string;
prompt ​
ts
readonly prompt: string;
session ​
ts
readonly session: string | null;
signalId ​
ts
readonly signalId: string;
startedAt ​
ts
readonly startedAt: string | null;
state ​
ts
readonly state: RunState;

RunState ​

ts
type RunState = typeof runStates[number];

Runtime ​

ts
type Runtime = {
  run: (prompt: RunPrompt) => Promise<RunOutcome>;
};

Starts one Run and reports how it ended: the one method an Agent Implementation is driven through.

Called one Run at a time and never concurrently, whatever Session each is in. An implementation therefore needs no locking of its own, and a Workspace shared with every Signal Handler is safe.

Throwing instead of answering a failure comes to the same thing: the Run fails carrying the thrown message. Neither form takes the Signal Worker down.

There is no timeout and no cancellation, here or anywhere else in the framework. A call that never settles halts the Gateway for every Party, so a Runtime that waits on something remote brings its own bound.

Methods ​

run() ​
ts
run(prompt: RunPrompt): Promise<RunOutcome>;
Parameters ​
prompt ​

RunPrompt

Returns ​

Promise<RunOutcome>

Signal ​

ts
type Signal<TPayload> = {
  readonly emittedAt: Date;
  readonly id: string;
  readonly kind: string;
  readonly payload: TPayload;
};

What a Handler is given: the arrival record a Producer wrote, and the whole of it.

The payload is whatever that Producer wrote and is never interpreted on the way here, so what a given kind carries is the Producer's contract with this Handler rather than anything the framework settles.

The state and error that SignalRecord carries are absent. A Handler runs because the Signal is being processed, and how it ends is the framework's to record.

Type Parameters ​

TPayload ​

TPayload = unknown

What this Handler expects to find in the payload. The Signal Worker itself carries unknown, having no opinion about any of it, so the narrowing is the Handler's.

Properties ​

emittedAt ​
ts
readonly emittedAt: Date;
id ​
ts
readonly id: string;
kind ​
ts
readonly kind: string;
payload ​
ts
readonly payload: TPayload;

SignalHandler ​

ts
type SignalHandler<TPayload> = {
  handle: (signal: Signal<TPayload>) => readonly Prompt[] | Promise<readonly Prompt[]>;
  post?: (signal: Signal<TPayload>, outcome: PostOutcome) => void | Promise<void>;
};

A Signal Handler: handle, and optionally post.

There is no context object and no second argument. A Handler closes over the logger, the Workspace path, the Messenger and its prompt template, and its own factory in the entry point supplies them, which is also why a Handler under test is a function of a Signal and needs no harness. The one thing it cannot close over is the Signal Worker it runs under, that Worker being constructed with the map this Handler is in.

handle returns zero, one or many Prompts. Declining, answering and fanning out are one mechanism, and an empty array is not a special case: the Signal is done with no Runs, and the arrival record stays behind, which is what makes a refusal auditable. Fanning out runs every Prompt in the order returned, and one that fails does not stop the rest. Throwing fails this Signal alone, and the Worker carries on.

post runs once, after every Run arising from the Signal has finished, whether they succeeded, failed or were never created. It produces no Prompts, and it is the whole of the framework's failure handling: notifying somebody, cleaning up, or emitting the work again is written here or nowhere. Throwing here fails the Signal too, beside whatever else went wrong.

Neither is given a timeout, and neither is ever run twice for one Signal.

Type Parameters ​

TPayload ​

TPayload = unknown

Methods ​

handle() ​
ts
handle(signal: Signal<TPayload>): readonly Prompt[] | Promise<readonly Prompt[]>;
Parameters ​
signal ​

Signal<TPayload>

Returns ​

| readonly Prompt[] | Promise<readonly Prompt[]>

post()? ​
ts
post?(signal: Signal<TPayload>, outcome: PostOutcome): void | Promise<void>;
Parameters ​
signal ​

Signal<TPayload>

outcome ​

PostOutcome

Returns ​

void | Promise<void>

SignalHandlers ​

ts
type SignalHandlers = Readonly<Record<string, SignalHandler>>;

The kind-to-Handler map: what a Gateway can act on, and the whole of it.

A Signal whose kind is not a key here fails permanently. There is no Handler to run a post phase, and nothing re-runs it, so a typo in a kind is loud and one-way rather than silent.

SignalRecord ​

ts
type SignalRecord = {
  readonly emittedAt: string;
  readonly error: string | null;
  readonly id: string;
  readonly kind: string;
  readonly payload: unknown;
  readonly state: SignalState;
};

A Signal as the agent reads it, and the JSON two of these routes answer with.

The payload arrives as the Producer wrote it, and emittedAt as an ISO 8601 string, JSON having no date.

state and error are here, where Signal has neither: what there is to know about a prior arrival is mostly how it ended, and since a failed Signal is failed for good, the reason has to be readable by whoever finds it.

Properties ​

emittedAt ​
ts
readonly emittedAt: string;
error ​
ts
readonly error: string | null;
id ​
ts
readonly id: string;
kind ​
ts
readonly kind: string;
payload ​
ts
readonly payload: unknown;
state ​
ts
readonly state: SignalState;

SignalState ​

ts
type SignalState = typeof signalStates[number];

SignalWorker ​

ts
type SignalWorker = Component & {
  readonly agentRoutes: FastifyPluginAsync;
  emit: <TSchema>(tx: Handle<TSchema>, signal: EmittedSignal) => Promise<string>;
  start: () => Promise<void>;
  stop: () => Promise<void>;
};

The Signal queue as a Component: something for a Producer to emit into, a Run loop, and a read of both that the agent reaches over HTTP.

A Signal is a durable row with a processing state rather than an event, so the queue survives the process and a Gateway that stops mid-queue starts again with the pending ones still in it. Nothing in flight survives. A Signal left processing by a stopped worker is failed at the next start and never re-run, because its Runs may already have sent Messages, written the Workspace or called something outside, and its Prompt is already in the Session on disk. Its Runs are failed with it, a running row with nothing running being a lie.

It holds no identity and knows nothing about messaging. No method takes a User, and nothing here is scoped by one.

There is nothing that cancels, retries, reprioritises or removes a Signal, and no way to ask whether the queue is empty. A failed Signal is failed for good, and doing the work after all means emitting another.

Type Declaration ​

agentRoutes ​
ts
readonly agentRoutes: FastifyPluginAsync;

The Agent server routes, as a Fastify plugin to register yourself.

Register it under a prefix of your own, or inside your own encapsulated plugin, or behind a hook you share with your own routes. Passing no server and never registering this is how the group is switched off.

The routes read these two tables and no other component's, and the whole surface is read-only and unscoped: every Signal and every Run, whatever Session the reading Run is in.

emit() ​
ts
emit<TSchema>(tx: Handle<TSchema>, signal: EmittedSignal): Promise<string>;

Records a Signal as pending, wakes the worker when the caller's transaction commits, and answers with the new Signal's id.

It takes that transaction rather than opening one, so recording something and telling the agent about it cannot come apart: a rollback loses both, and no wakeup is sent for a Signal that never existed. The transaction may carry any component's schema, this write naming its own table.

It waits for nothing beyond the insert. The Signal is queued, not run, and the id is for reading the outcome back later rather than for awaiting it.

Type Parameters ​
TSchema ​

TSchema extends Record<string, unknown>

Parameters ​
tx ​

Handle<TSchema>

signal ​

EmittedSignal

Returns ​

Promise<string>

start() ​
ts
start(): Promise<void>;

Starts looking for Signals, with the Handlers this Worker was constructed with.

It resolves immediately, and the first thing the worker does after that is fail whatever a previous worker left processing. Nothing an Operator does next waits on that: a Signal emitted meanwhile is a row in a queue, drained once the recovery is done.

Returns ​

Promise<void>

Throws ​

If it has already been called. One Signal Worker drains one queue, Runs being serial across the whole Gateway.

stop() ​
ts
stop(): Promise<void>;

Stops looking for Signals and waits for the Run in flight to finish.

Not a shutdown protocol: the framework installs no SIGTERM handling of its own. There is no cancellation either. The Run in flight runs to completion, because abandoning it would leave partial effects nothing retries, so this takes as long as the slowest Run the agent can have started.

Whatever is still pending stays pending, for the next worker over this database to drain.

Returns ​

Promise<void>

SignalWorkerOptions ​

ts
type SignalWorkerOptions = {
  readonly agentServer?: {
    readonly fastify: FastifyInstance;
  };
  readonly db: Db;
  readonly handlers: SignalHandlers;
  readonly logger?: Logger;
  readonly runtime: Runtime;
  readonly sweepIntervalMs?: number;
};

Properties ​

agentServer? ​
ts
readonly agentServer?: {
  readonly fastify: FastifyInstance;
};

The Agent server, if the agent is to read prior Signals and Runs.

Given one, the constructor registers agentRoutes on its Fastify instance at no prefix: /signals, /signals/:id, /runs and /runs/:id. Omit it and nothing is registered anywhere, which is how the group is switched off.

Structural: anything carrying a Fastify instance satisfies it. A server built on http2 does not, and takes the agentRoutes plugin instead.

fastify ​
ts
readonly fastify: FastifyInstance;
db ​
ts
readonly db: Db;
handlers ​
ts
readonly handlers: SignalHandlers;

The kind-to-Handler map: what this Gateway can act on, and the whole of it.

A construction option rather than an argument to start, so a Signal Worker with no Handlers is unconstructable rather than merely unstartable.

Held rather than copied. The Worker looks a kind up in this same object at dispatch, so an entry written into it before start is dispatched on, and that is the way out of the knot a Handler that emits back into this Worker ties: it cannot close over an object that does not exist yet, so it is built after construction and assigned in.

logger? ​
ts
readonly logger?: Logger;

Defaults to a pino instance on stdout.

One info line as each Signal is claimed and as it finishes, one as each Run starts and ends carrying the Session it ran in, and a debug line for every wakeup saying what caused it. A Signal Handler's failure and a Run's are logged at error whether or not anything else notices them, and a Signal a stopped worker left behind at warn. No Prompt text and no payload is ever written.

runtime ​
ts
readonly runtime: Runtime;

What a Prompt is handed to. createPiRuntime, on @shutter-network/concorde/pi, builds the one this framework ships.

sweepIntervalMs? ​
ts
readonly sweepIntervalMs?: number;

How often the worker looks for pending Signals regardless of notifications, in milliseconds. Defaults to 5000.

Not the latency of a Signal: emitting one wakes the worker at once. This is the safety net for a notification sent while the listening connection was down, so what the number bounds is how long a Signal can sit unnoticed after a database restart. There is no correctness in it, and the cost of lowering it is a query per interval forever.

TemplateHandlerOptions ​

ts
type TemplateHandlerOptions<TPayload> = {
  readonly data: (signal: Signal<TPayload>) => unknown;
  readonly helpers?: Readonly<Record<string, Handlebars.HelperDelegate>>;
  readonly partials?: Readonly<Record<string, string>>;
  readonly session: (signal: Signal<TPayload>) => string | null | Promise<string | null>;
  readonly template: string;
};

Type Parameters ​

TPayload ​

TPayload = unknown

Properties ​

data ​
ts
readonly data: (signal: Signal<TPayload>) => unknown;

The values the template substitutes. A name the template references and this does not supply fails the Signal rather than rendering as nothing.

A returned Promise is awaited, so this can be async, and it is where a Handler reads what the Prompt needs: the Message log, the Workspace, or tables of your own.

Parameters ​
signal ​

Signal<TPayload>

Returns ​

unknown

helpers? ​
ts
readonly helpers?: Readonly<Record<string, Handlebars.HelperDelegate>>;

Handlebars helpers, registered on an environment belonging to this Handler alone. Another Handler built by another call cannot see them, and neither can the shared Handlebars instance. What a helper returns is substituted unescaped, like everything else.

partials? ​
ts
readonly partials?: Readonly<Record<string, string>>;

Handlebars partials, as template source rather than as templates already compiled.

They are compiled here with the same options as the template itself, so noEscape and strict hold inside them too.

session ​
ts
readonly session: (signal: Signal<TPayload>) => string | null | Promise<string | null>;

Which Session this Signal's Prompt continues, or null to ask for a fresh one.

The topology is yours: one Session per User, one per Run, or one for the whole agent. A returned Promise is awaited.

Parameters ​
signal ​

Signal<TPayload>

Returns ​

string | null | Promise<string | null>

template ​
ts
readonly template: string;

The Handlebars source, compiled once when the Handler is built.

Source, and not a path or a file: URL. A deployment that keeps its wording in a file reads the file itself: template: readFileSync(new URL("./prompt.hbs", import.meta.url), "utf8"). Nothing reads it again after that, so an edit reaches no Prompt until the process starts again. In exchange a template Handlebars cannot compile throws from templateHandler, before the Gateway listens, instead of failing a Signal that nothing retries.

It is compiled with noEscape, so nothing substituted is HTML-escaped, and with strict, which fails the Signal on a variable data did not supply. strict also disables inverse sections: a caret block such as ^absent throws, and the unless helper is what to write in its place. The if, each and else helpers behave as usual.

Variables ​

runStates ​

ts
const runStates: readonly ["pending", "running", "done", "failed"];

A Run's state. There is no timed_out, the framework imposing no timeout on a Run or on anything else, so a Run that never ends stays running and holds the queue.

signalStates ​

ts
const signalStates: readonly ["pending", "processing", "done", "failed"];

A Signal's processing state. One-way: nothing returns to pending, and a failed Signal is never re-run, so error is the whole of what became of it.

Functions ​

createSignalWorker() ​

ts
function createSignalWorker(options: SignalWorkerOptions): SignalWorker;

Builds a Signal Worker over a Db, a Runtime and a map of Signal Handlers, and registers the read routes on the Agent server if one was passed.

createGateway builds a Worker already. Reach for this only when assembling a Gateway by hand with createBareGateway.

Parameters ​

options ​

SignalWorkerOptions

Returns ​

SignalWorker

templateHandler() ​

ts
function templateHandler<TPayload>(options: TemplateHandlerOptions<TPayload>): SignalHandler<TPayload>;

Builds a Signal Handler that renders one Prompt per Signal from a Handlebars template.

One Prompt, always. It never fans a Signal out across several Sessions and never declines one, although the Handler contract allows both. It has no post phase either, and gains one by being spread: { ...templateHandler(options), post } is a Handler.

A template that compiles and then does not render fails the Signal, with a message naming the Signal's kind. Handlebars names the variable, the line and the column, and never says which Handler was rendering, which is the one thing an Operator running several of them needs.

Type Parameters ​

TPayload ​

TPayload = unknown

Parameters ​

options ​

TemplateHandlerOptions<TPayload>

Returns ​

SignalHandler<TPayload>

Throws ​

if template does not compile. The message is Handlebars' own and names the line and the column. A helper's complaint about its own arguments is not among these, because a helper runs only with a context.