Bridge a Threaded Channel to the Agent Team

You build an adapter that relays messages between a human channel and an agent team's dispatch workflow. The channel can be GitHub Discussions, Microsoft Teams, or the next chat platform someone asks for. The first time you do this, you reach for last project's callback registry, rate limiter, and history-bound prompt builder. @forwardimpact/libbridge gives you those primitives. The host service can then focus on the channel-specific SDK glue. It leaves thread state, callback verification, prompt construction, and workflow dispatch to a shared library.

Prerequisites

• Node.js 22+
• Install the library and its peers:

npm install @forwardimpact/libbridge @forwardimpact/libstorage @forwardimpact/libindex

• A workflow on the target repository that accepts the channel-bridge payload through workflow_dispatch. The Kata agent team ships the reference implementation of that workflow.
• A GitHub token with actions:write on that repository.

What libbridge owns

libbridge is channel-agnostic. It never imports botbuilder, @octokit/*, or any channel-specific SDK. The host service (services/ghbridge, services/msbridge, your next adapter) owns the SDK glue, signature verification, and channel-shaped responses. libbridge owns the shared primitives every adapter needs:

Primitive	Purpose
createBridgeServer	Hono server that wires a channel webhook route and /api/callback/:tenant_id/:token together
Acknowledgement	Reaction-plus-optional-typing-verb lifecycle for "I received your message" feedback
Dispatcher	Composes callback registration, acknowledgement, workflow dispatch, history append, and rollback-on-failure into one call
createCallbackHandler	Inbound-callback skeleton with verdict routing (adjourned / failed / recessed) and span instrumentation
ResumeScheduler	Channel-agnostic suspend/resume lifecycle for recessed verdicts. Wraps ElapsedScheduler
CallbackRegistry	In-memory token registry with tenant-bound entries, TTL enforced at lookup, periodic sweep, and atomic consume
DiscussionAdapter (typedef)	The persistence contract every bridge implements: loadByChannel, loadByCorrelation, listOpenRecesses, add, flush, shutdown (plus optional putPendingDispatch / resolvePendingDispatch)
newDiscussionContext	Channel-agnostic factory for a fresh per-thread record, keyed by (channel, discussion_id)
RateLimiter	Sliding-window per-thread rate limit so a noisy channel cannot DoS the workflow
ProgressTicker	Tick-and-stop timer so the host can show progress while the workflow runs
appendHistory	Bounded message history. The default cap is 10 entries, and the oldest entry drops on overflow
buildPrompt	Prompt builder that prepends recent history bounded by exchange count and char cap
dispatchWorkflow	GitHub Actions workflow_dispatch POST with the agreed input shape
evaluateTrigger	Caller-clock resume-trigger evaluation (kinds: missing_input, elapsed, escalation_needed)
parseIsoDuration	ISO-8601 duration parser (P1D, PT12H, P1DT6H) that evaluateTrigger uses

Four primitives form the composition layer: Acknowledgement, Dispatcher, createCallbackHandler, and ResumeScheduler. A real bridge wires the channel SDK into these constructors, and each one owns its slice of the dance. The primitives below them are still available when you need to step outside the shared composition.

Two injection rules keep the surface testable from any host. Persistence is contract-injected. Every libbridge primitive that touches per-thread state (Dispatcher, ResumeScheduler, createCallbackHandler, createLinkCompleteHandler) takes a store parameter that satisfies the DiscussionAdapter typedef. The library never constructs persistence on its own. The trigger evaluator is clock-injected. evaluateTrigger(trigger, observed, now) takes now as a parameter. The library never calls Date.now().

Compose a bridge server

At minimum, a channel adapter needs a Hono server with a channel-shaped webhook route and a workflow callback route. createBridgeServer mounts both routes on a Hono app and returns lifecycle handles. Both routes hand the raw Hono Context to host-supplied callbacks. The host owns signature verification, token redemption, and channel-shaped responses:

import {
  createBridgeServer,
  CallbackRegistry,
} from "@forwardimpact/libbridge";

const store = createDiscussionAdapter();   // see "Persist per-thread context" below
const registry = new CallbackRegistry({ ttlMs: 60 * 60 * 1000, clock });
registry.startSweepTimer();                // periodic eviction of expired tokens

const bridge = createBridgeServer({
  config: { host: "0.0.0.0", port: 8080 },
  logger,
  webhookPath: "/api/messages",
  onWebhook: async (c) => {
    const event = await verifyChannelSignature(c);
    await handleChannelEvent({ event, store, registry });
    return c.body(null, 200);
  },
  onCallback: async (c) => {
    const tenantId = c.req.param("tenant_id");
    const entry = registry.consume(c.req.param("token"), { tenant_id: tenantId });
    if (!entry) return c.json({ error: "Unknown token" }, 404);
    const payload = await c.req.json();
    if (payload.correlation_id !== entry.correlationId) {
      return c.json({ error: "Correlation ID mismatch" }, 400);
    }
    const ctx = await store.loadByChannel("example", entry.meta.discussionId);
    if (payload.verdict === "adjourned") {
      for (const reply of payload.replies) {
        await postChannelMessage(ctx.discussion_id, reply.body);
      }
    } else if (payload.verdict === "failed") {
      await postChannelMessage(ctx.discussion_id, `Failed: ${payload.summary}`);
    }
    return c.json({ ok: true }, 200);
  },
});

await bridge.start();

createBridgeServer mounts POST <webhookPath> and POST /api/callback/:tenant_id/:token on a Hono app. It captures the raw POST body on c.get("rawBody") for signature verification. It returns { start, stop, app, address }. The host owns lifecycle, the channel SDK, and the verdict-to-channel translation (a GraphQL addDiscussionComment for GitHub, a botbuilder activity for Teams, etc.).

Persist per-thread context

Each thread (a Discussion, a Teams conversation) carries its own context record, keyed by (channel, discussion_id). newDiscussionContext builds a fresh record so every bridge agrees on the shape:

import { newDiscussionContext } from "@forwardimpact/libbridge";

const ctx = newDiscussionContext({
  clock,
  channel: "github-discussions",
  discussionId,
  participant: { name: "octocat", kind: "human", external_id: "1234" },
});
// {
//   id: "github-discussions:<discussion_id>",
//   channel, discussion_id,
//   history: [], participants: [participant],
//   open_rfcs: {}, lead: "release-engineer",
//   pending_callbacks: {}, dispatches: [],
//   active_requester: null, last_posted_seq: -1,
//   last_active_at: <clock.now()>,
// }

The host owns persistence. It implements the DiscussionAdapter typedef. It passes the instance as store to Dispatcher, ResumeScheduler, createCallbackHandler, and createLinkCompleteHandler. The contract:

/**
 * @typedef {object} DiscussionAdapter
 * @property {(channel: string, discussionId: string) => Promise<object|null>} loadByChannel
 * @property {(correlationId: string) => Promise<object|null>} loadByCorrelation
 * @property {() => Promise<Array<{correlationId: string, dueAt: number}>>} listOpenRecesses
 * @property {(ctx: object) => Promise<void>} add
 * @property {() => Promise<void>} flush
 * @property {() => Promise<void>} shutdown
 * @property {(target: object) => Promise<void>} [putPendingDispatch]
 * @property {(linkToken: string, expectedSurfaceUserId?: string) => Promise<object|null>} [resolvePendingDispatch]
 */

The next example is a minimal in-process adapter. It stores durable JSONL with @forwardimpact/libindex and @forwardimpact/libstorage. It suits single-process bridges:

import { BufferedIndex } from "@forwardimpact/libindex";
import { createStorage } from "@forwardimpact/libstorage";
import { appendHistory } from "@forwardimpact/libbridge";

function createInProcessAdapter({ clock }) {
  const storage = createStorage("bridges/example");
  const index = new BufferedIndex(storage, "discussions.jsonl", {}, { clock });

  return {
    async loadByChannel(channel, id) {
      await index.loadData();
      return index.index.get(`${channel}:${id}`) ?? null;
    },
    async loadByCorrelation(correlationId) {
      await index.loadData();
      for (const rec of index.index.values()) {
        if (Object.values(rec.pending_callbacks ?? {}).includes(correlationId)) {
          return rec;
        }
        if (rec.open_rfcs?.[correlationId]) return rec;
      }
      return null;
    },
    async listOpenRecesses() {
      await index.loadData();
      const refs = [];
      for (const rec of index.index.values()) {
        for (const [cid, rfc] of Object.entries(rec.open_rfcs ?? {})) {
          if (typeof rfc.due_at === "number") {
            refs.push({ correlationId: cid, dueAt: rfc.due_at });
          }
        }
      }
      return refs;
    },
    add: (ctx) => index.add(ctx),
    flush: () => index.flush(),
    shutdown: () => index.flush(),
  };
}

const store = createInProcessAdapter({ clock });
const ctx = (await store.loadByChannel("github-discussions", discussionId))
  ?? newDiscussionContext({ clock, channel: "github-discussions", discussionId, participant });
appendHistory(ctx.history, { role: "user", text: "Should we add nested levels?" });
ctx.last_active_at = clock.now();
await store.add(ctx);
await store.flush();

For multi-process bridges, point the adapter at a shared backend such as Redis, Postgres, or a dedicated persistence service. Every bridge replica then sees the same (channel, discussion_id) records. The pending_callbacks tokens also survive restarts. The reference deployment runs a small gRPC service that owns the JSONL files and the TTL sweep. services/ghbridge and services/msbridge wrap a generated client in a DiscussionAdapter to talk to it. Implementations swap freely. libbridge only sees the contract.

Issue and verify callback tokens

A bridge dispatches a workflow run. It then waits for the workflow to POST back its verdict. The host registers a (correlationId, meta) pair, and meta.tenant_id is required. The host receives a randomly generated token. The host embeds the token in the callback URL. The workflow echoes it. The host consumes the token once and rejects every later attempt. consume(token, { tenant_id }) is atomic. It removes the entry and returns it in one call. It returns null when the token is unknown, expired, or bound to a different tenant. The default TTL is two hours. A lookup drops an expired entry when it observes it. startSweepTimer() evicts tokens whose dispatch never calls back, every 10 minutes by default. stopSweepTimer() cancels the sweep. Use peek(token, { tenant_id }) to inspect an entry. peek does not consume it.

import { randomUUID } from "node:crypto";
import {
  CallbackRegistry,
  dispatchWorkflow,
} from "@forwardimpact/libbridge";

const registry = new CallbackRegistry({ ttlMs: 60 * 60 * 1000, clock });
registry.startSweepTimer();

const correlationId = randomUUID();
const token = registry.register(correlationId, { tenant_id: tenantId, discussionId });
await dispatchWorkflow({
  workflowFile: "agent-dispatch.yml",
  ref: "main",
  repo: "owner/repo",
  token: ghInstallationToken,
  prompt,
  callbackUrl: `${publicUrl}/api/callback/${tenantId}/${token}`,
  correlationId,
  discussionId,
});

// In the `onCallback` handler passed to createBridgeServer:
async function onCallback(c) {
  const entry = registry.consume(c.req.param("token"), {
    tenant_id: c.req.param("tenant_id"),
  });
  if (!entry) return c.json({ error: "Unknown token" }, 404);
  const payload = await c.req.json();
  if (payload.correlation_id !== entry.correlationId) {
    return c.json({ error: "Correlation ID mismatch" }, 400);
  }
  // …deliver replies, recess, or fail per payload.verdict…
  return c.json({ ok: true }, 200);
}

The registry is in-memory. For multi-process bridges, persist pending_callbacks on each discussion-context record through the adapter's add() call. The host can then re-register tokens on restart. The correlation_id echoes through the workflow. The host checks it against the consumed entry's correlationId to defend against token-and-payload mismatches. The tenant binding makes sure a token issued for one tenant cannot redeem a callback addressed to another.

Evaluate recess triggers

Long-running RFCs use the libharness Recess verdict to wait for an external signal. A trigger is one of three shapes, named for the lead's intent:

• { kind: "missing_input", replies: N } — fire when at least N new replies arrive on the dispatching thread after the recess opens.
• { kind: "elapsed", elapsed: "P1D" } — fire after an ISO-8601 duration passes. The parser supports days, hours, minutes, and seconds (P14D, PT12H, P1DT6H).
• { kind: "escalation_needed", signal: "<name>" } — reserved for future use. The schema accepts this shape, but the scheduler throws until signal-based resume support ships.

evaluateTrigger(trigger, observed, now) returns { fired: boolean, due_at?: number }. due_at is the absolute ms-epoch when an elapsed arm fires. You can use it to schedule a wake-up. The host owns now, so unit tests stay deterministic:

import { evaluateTrigger } from "@forwardimpact/libbridge";

const trigger = { kind: "elapsed", elapsed: "P1D" };
const observed = { opened_at: Date.now() - 25 * 60 * 60 * 1000 };

const result = evaluateTrigger(trigger, observed, Date.now());
if (result.fired) {
  await dispatchWorkflow({
    workflowFile: "agent-dispatch.yml",
    ref: "main",
    repo: "owner/repo",
    token: ghInstallationToken,
    prompt: "Resume requested.",
    callbackUrl,
    correlationId: newCorrelationId,
    discussionId,
    resumeContext: JSON.stringify({
      correlation_id: priorCorrelationId,
      history_since: historySliceSinceRecess,
    }),
  });
}

evaluateTrigger is pure. It takes a trigger, an observation ({ replies?, opened_at? }), and a clock reading. It returns whether the observation satisfies the trigger. The host calls it whenever a candidate event arrives. For missing_input, the host calls it on every new channel message. For elapsed, the host calls it on a host-scheduled wake-up at due_at. escalation_needed throws today. It will integrate with channel signal intake once that spec lands.

Verify

You have reached the outcome of this guide when:

• You can stand up a Hono server with channel-webhook and /api/callback/:tenant_id/:token routes through createBridgeServer. The host's channel-specific SDK glue sits only inside onWebhook and onCallback.
• You can persist per-thread state. You implement the DiscussionAdapter contract (loadByChannel, loadByCorrelation, listOpenRecesses, add, flush, shutdown). You build fresh records with newDiscussionContext, keyed by (channel, discussion_id).
• You can register tenant-bound tokens, dispatch, and one-shot consume(token, { tenant_id }) through CallbackRegistry. The correlation_id echoes end-to-end. A lookup rejects expired tokens.
• You can evaluate missing_input and elapsed recess triggers against a caller-supplied clock and route the resume back through dispatchWorkflow with a JSON-encoded resume_context. escalation_needed triggers parse but throw at evaluation until signal-based resume ships.

What's next

The agent team that receives a bridged message keeps its own memory and its own control charts. Gemba documents that side in Operate a Predictable Agent Team.