# Live connections

Some platforms only deliver messages over a connection the receiver keeps open for days: Discord's Gateway, IRC,
Matrix sync, Slack Socket Mode, a Home Assistant event stream. A plugin runs for seconds at a time and can't hold a
socket. So AgentParley holds it for you.

You describe **how to connect** and **what to do with what arrives**. A dedicated service opens the socket, keeps
it alive, sends your heartbeats, answers keep-alives, drops the noise you told it to drop, reconnects with backoff,
survives AgentParley's own deploys, and wakes your plugin only for frames you care about.

The official **Discord** plugin is built on this — see the [walkthrough](example-discord.md).

## The handlers

```ts
globalThis.connection = {
  open:     (request) => ConnectionOpenResult,     // required
  receive:  (request) => ConnectionStepResult,     // required
  timer?:   (request) => ConnectionStepResult,     // only if you use wakeAfterMs
};
```

Every call gets your stored `state` and returns what to do next. Nothing else survives between calls.

### `open` — connect (or reconnect)

```ts
request: {
  owner: string;                                   // "channel" or "install" — see "Two kinds of owner"
  reason: "start" | "reconnect" | "handoff";       // handoff = the platform moved you (deploy, rebalance)
  state: unknown | null;                           // your stored state; null on the very first open
  lastClose: { code: number | null; reason: string; by: "remote" | "plugin" | "platform"; link: string | null } | null;
  lastSequence: unknown | null;                    // the last sequence number seen (see `sequence`)
}
returns: ConnectionStepResult & {
  links: LinkSpec[];              // at most 5 sockets
  resumable?: boolean;            // your next open may resume the vendor session after a platform move
  connectsPerMinute?: number;     // 1–600; may only LOWER the platform's connect rate
}
```

Return `{ links: [], fail: "…" }` to stop for good (for example, a rejected token) — nothing reconnects until the
user acts.

### Links

```ts
type LinkSpec =
  | { transport: "websocket"; url: string; headers?: Record<string, string>; protocols?: string[] }
  | { transport: "tcp"; host: string; port: number; tls: boolean }         // line-based; never put CR/LF inside a frame
  | { transport: "sse"; url: string; headers?: Record<string, string> }
  | { transport: "poll"; everyMs: number; request: HttpRequest };    // everyMs: 30 000 – 86 400 000

type HttpRequest = {
  method?: "GET" | "POST" | "PUT"; url: string; headers?: Record<string, string>; body?: string; timeoutMs?: number;
  oauth?: string;     // a manifest `oauth` key: the service adds `Authorization: Bearer <fresh token>`; you never see it
  account?: string;   // the connection id from host.oauth.accounts(), when more than one is connected
};

// Every link also takes:
{
  key?: string;                  // name it if you have several; unnamed links are "link0", "link1", … in order
  idleTimeoutMs?: number;        // reconnect if nothing arrives for this long
  maxDetachedSeconds?: number;   // 30–300 (default 300): keep the socket while cut off from AgentParley
  filters?: FrameRule[];         // keep/drop rules, applied before any wake (on a poll link: to 2xx replies)
  autoReply?: { textPrefix: string; replacePrefixWith: string }[];   // e.g. IRC PING → PONG, costs no wake
  sequence?: { path: string };   // JSON path of the vendor's sequence number, tracked on every frame
}
```

### Poll links: quiet polling

The service repeats `request` every `everyMs` (the first poll comes after a random delay of up to `everyMs`) and does **not** wake you for a poll that finds nothing. A poll costs
one HTTPS request from the service and nothing else — no call to AgentParley, no wake, no allowance. Requests time
out at 120 s (set a lower `timeoutMs` if you like), redirects are not followed, a reply over 2 MB counts as a network
failure, and the request host must be on your `manifest.egress` list.

You are woken (with an `InboundFrame` of kind `response`) for:

- a **2xx** reply that your `filters` keep (rules match the first 256 KB of the body; a reply matching no rule is
  kept), and
- any **3xx or 4xx** except 401, 408 and 429.

After a wake the link pauses until `receive` returns; polling then resumes on its schedule. Return `next` to change
what it asks for (a new cursor, a new `pageToken`).

You are **not** woken for failures the service retries for you: **408, 429 and 5xx** replies, and network errors.
Retries back off from `everyMs` upward (doubling, at most 30 minutes between tries) and honour `Retry-After` (up to
one hour). After **5 quiet failures in a row** (408/429/5xx) the service posts one note to the connection card so the
user can see the outage; the next good reply clears the streak. **Five network failures in a row** (no HTTP reply at
all) close the connection and it reconnects through the normal back-off; any HTTP reply, even a 5xx, resets that
counter.

With `oauth` on the request, a **401** is not delivered to you: the service gets a fresh token and sends the request
once more. A second 401 stops the connection with "sign-in required" until the user reconnects the sign-in. A 401 on a
request without `oauth` is an ordinary 4xx and wakes you.

Example: a Gmail watcher that wakes only when the mailbox history moves.

```ts
{
  transport: "poll",
  everyMs: 120_000,
  request: { url: "https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=" + cursor
                  + "&historyTypes=messageAdded", oauth: "google", account },
  filters: [
    { when: [{ path: "history", exists: true }], action: "keep" },   // something changed
    { when: [{ textPrefix: "" }], action: "drop" },                   // nothing new: stay asleep
  ],
}
```

### `receive` — frames arrived

```ts
request: {
  state: unknown | null;
  link: string;              // which link
  frames: InboundFrame[];    // up to 50 per call
  lastSequence: unknown | null;
}

type InboundFrame =
  | { id: string; kind: "text"; text: string }
  | { id: string; kind: "binary"; base64: string }
  | { id: string; kind: "sse"; event: string; data: string; lastEventId: string | null }
  | { id: string; kind: "response"; status: number; headers: Record<string, string>; body: string };
```

A frame's `id` is the platform's de-duplication key across redelivery (not the vendor's id).

### `timer` — nothing arrived for `wakeAfterMs`

```ts
request: { state: unknown | null; lastSequence: unknown | null }
```

### What you return — `ConnectionStepResult`

Every field is optional; omit what you don't change.

| Field | Does |
| --- | --- |
| `state` | your new state (≤ 64 KB; a larger one is ignored and the old state kept). Omit to keep the old one. **Not encrypted — never store a secret in it.** |
| `send` | frames to write: `{ link?, text }` or `{ link?, base64 }` — ≤ 50 per step; frames beyond that are dropped. Leave out `link` for the link this step answers; in `open` or `timer`, it means the connection's only link, and with several links the frame is dropped |
| `heartbeat` | `{ link?, everyMs, jitterFirst?, frame }` — the service sends `frame` every `everyMs`, no wake. `null` clears it. A JSON frame may set `sequenceAt` to have the latest sequence written into it. |
| `filters` | `{ link?, rules }` — replace a link's keep/drop rules. `link` defaults as for `send`, and so does `heartbeat`'s |
| `wakeAfterMs` | call `timer` after this long with no inbound frame (periodic logic); `null` clears it |
| `next` | `{ link, request }[]` — replace a poll link's repeating request (`request` may carry `oauth`) |
| `messages` | chat messages (the `ChannelInboundMessage` shape) — for a **channel-owned** connection |
| `events` | `{ id, text }[]`, ≤ 50 per step — for an **install-owned** connection; stored once per `id`. `id` is up to 100 characters (an event without one is dropped); `text` is cut at 4,000 characters |
| `reconnect` | `{ reason, resetState? }` — ask for a reconnect (the platform decides when) |
| `fail` | a fatal error: stop until the user acts |

### Filters

The service applies your rules to every frame before deciding whether to wake you. A frame you drop costs nothing.

```ts
type FrameRule = { when: FrameCondition[]; action: "keep" | "drop" };

type FrameCondition =
  | { path: string; equals: string | number | boolean | null }   // dot path into JSON; "a[].b" = any element
  | { path: string; exists: boolean }
  | { textPrefix: string }
  | { binaryPrefixBase64: string };
```

The first rule whose conditions **all** match decides; a frame matching no rule is **kept**. Discord's rules keep
direct messages, @mentions and replies to the bot, and drop everything else in a server:

```ts
// plugins/discord/index.ts
function baseFilters(botId: string | undefined, respondToBots: boolean): FrameRule[] {
  const rules: FrameRule[] = [
    { when: [{ path: "op", equals: 11 }], action: "drop" },            // heartbeat ACKs
    { when: [{ path: "t", equals: "RESUMED" }], action: "drop" },
  ];
  if (!botId) {
    return rules;
  }
  if (!respondToBots) {
    rules.push({ when: [{ path: "d.author.bot", equals: true }], action: "drop" });
  }
  rules.push(
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.guild_id", exists: false }], action: "keep" },
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.mentions[].id", equals: botId }], action: "keep" },
    { when: [{ path: "t", equals: "MESSAGE_CREATE" }, { path: "d.referenced_message.author.id", equals: botId }], action: "keep" },
    { when: [{ path: "t", equals: "READY" }], action: "keep" },
    { when: [{ path: "op", equals: 0 }], action: "drop" },
  );
  return rules;
}
```

### Heartbeats, sequences and resume

A typical gateway: on HELLO, identify (or resume) and start a heartbeat; track the sequence number; on READY,
store the session id so the next `open` can resume.

```ts
// plugins/discord/index.ts (abridged)
if (payload.op === 10) {                                  // HELLO
  const identify = canResume
    ? { op: 6, d: { token, session_id: state.sessionId, seq: sequence } }
    : { op: 2, d: { token, intents: /* … */, properties: { os: "linux", browser: "agentparley", device: "agentparley" } } };
  send.push({ link: request.link, text: JSON.stringify(identify) });
  heartbeat = {
    link: request.link,
    everyMs: payload.d.heartbeat_interval,
    jitterFirst: true,
    frame: { json: { op: 1, d: null }, sequenceAt: "d" },  // the service writes the latest sequence into d
  };
}
```

With `sequence: { path: "s" }` on the link, the service tracks `s` on every frame — even dropped ones — and hands
it back as `lastSequence`. Declare `resumable: true` and, after a deploy moves your socket (`reason: "handoff"`),
you resume the vendor session instead of logging in again.

## Two kinds of owner

| Owner | Created by | Your `receive` returns | Used for |
| --- | --- | --- | --- |
| `"channel"` | the user connecting a channel (Channels → Connect a channel) | `messages` → chats with the agent | chat platforms: Discord, IRC |
| `"install"` | the user pressing **Connect** under **Live connection** on the installed plugin's page | `events` → fires **On event** schedules | event streams: "when the front door opens, tell my House agent" |

An install-owned connection requires `manifest.installOwnedConnection: true`. The same code can serve both —
`hello-live` branches on `request.owner`:

```ts
// infrastructure/plugins/hello-live/index.ts
if (state.owner === "install") {
  events.push({ id: dispatch.id, text: dispatch.text });
} else {
  messages.push({
    conversationId: dispatch.conversationId,
    messageId: dispatch.id,
    sender: { id: dispatch.senderId ?? "tester", displayName: dispatch.senderName ?? "Tester" },
    text: dispatch.text,
    isGroup: !!dispatch.guildId,
    mentionsBot: !!dispatch.mention,
  });
}
```

Each event you emit fires the account's schedules of kind **On event** that listen on that connection; the
schedule's prompt runs on its agent with your event text.

## Sending outside a step — `host.connection.send`

From a channel's `send` handler (a reply the agent wrote), write straight onto the open socket:

```ts
// infrastructure/plugins/hello-live/index.ts
globalThis.channel = {
  send: async (request) => {
    const frame = { op: 99, d: { conversationId: request.conversationId, text: request.text } };
    await host.connection!.send([{ text: JSON.stringify(frame) }], { idempotencyKey: request.deliveryId ?? undefined });
  },
};
```

At most 20 frames per call, accepted or refused whole; pass the same `idempotencyKey` on a retry so nothing is
written twice. Permission `host:connection:send` is derived from the call. (Discord sends replies over REST
instead, which is why it doesn't need this.)

## Where it may connect

Links may only reach hosts in your `manifest.egress` or a `hosts` setting the user filled in. The service also
refuses private, loopback, link-local and cloud-metadata addresses, checked on the address it actually dials.

## Limits

| | |
| --- | --- |
| Links per connection | 5 |
| Inbound frames per `receive` call | 50 |
| Frames in `send` per step | 50 |
| Frames per `host.connection.send` call | 20 |
| `state` | 64 KB, not encrypted |
| Inbound frame / TCP line | 1 MB / 64 KB |
| Outbound frame | 64 KB |
| Outbound frames | 120 per minute (heartbeats and auto-replies don't count) |
| Heartbeat interval / idle timeout | at least 5 s / at least 10 s |
| `wakeAfterMs` | 1 s – 24 h |
| Poll `everyMs` | 30 s – 24 h |
| Filters | 30 rules per link, 5 conditions per rule, 8 path segments; 5 auto-reply rules |
| Opens | 300 per day |
| `open` / `receive` / `timer` wall clock | 15 s / 10 s / 10 s |
| TCP ports | 25, 465 and 587 (mail) are refused |
| Detached budget | 30–300 s |
| Live connections per account | Free 1, Solo 5, Entrepreneur 20 |
| Wakes (open / receive / timer) | count against the hourly plugin-wake allowance; reconnects the platform causes are free |