# @3pm/api

Node.js client for the [3pm](https://github.com/Unitech/3pm) process manager,
plus `io`, the application-metrics SDK (see below).
Same shape as pm2's programmatic API — `connect`, `start`, `list`, `stop`,
`restart`, `delete`, `describe`, `launchBus` — but promise-based, with zero
dependencies. It talks straight to the daemon's Unix sockets
(`rpc.sock` for commands, `pub.sock` for events) and auto-spawns the daemon
when none is running.

```console
npm install https://3pm.edgecr4ft.net/sdk/node/3pm-api-0.64.0.tgz
```

Published on <https://3pm.edgecr4ft.net> only, not on the npm registry:
every version and its sha256 in <https://3pm.edgecr4ft.net/sdk/sdk.json>.

Requires Node ≥ 18 and a `3pm` binary on `PATH` (or `THREEPM_BIN`) for
auto-spawn — not needed when a daemon is already up.

## Quickstart

```js
const { ThreePM } = require('@3pm/api');

const pm = await ThreePM.connect();          // spawns the daemon if needed

await pm.start({ script: 'app.js', name: 'api', instances: 2, env: { PORT: 3000 } });
console.table((await pm.list()).map(({ id, name, status, pid, restarts }) => ({ id, name, status, pid, restarts })));

const bus = await pm.launchBus({ process: 'api' });
bus.on('log:out',        (e) => process.stdout.write(`${e.name}[${e.instance}] ${e.line}\n`));
bus.on('log:err',        (e) => process.stderr.write(`${e.name}[${e.instance}] ${e.line}\n`));
bus.on('process:exit',   (e) => console.log('exit', e.exit_code, e.exit_signal, e.stderr_tail));
bus.on('process:online', (e) => console.log('online', e.name));
bus.on('daemon:shutdown',(e) => console.log('daemon going down:', e.reason));

await pm.restart('api');
await pm.stop('api');
await pm.delete('api');
pm.disconnect();                              // closes the rpc socket and every bus
```

## API

`new ThreePM({ home?, bin?, autoSpawn? = true, timeout? = 30000 })` — or `await ThreePM.connect(opts)`.

`home` resolves like the CLI (`resolvePaths`): the option, else `$THREEPM_HOME`, else the home a root daemon
hosts for this account (`/var/lib/3pm/users/<user>`, marked by a `hosted` file — `hostedHome()` returns it or
`null`), else `~/.3pm`. On a hosted home the client never spawns a daemon: when the root daemon is down it
throws `daemon_unreachable` with `err.hosted === true`. `start()` accepts `user` (run the app as a declared
account, root daemon) and `persist: false` (kept out of `dump.json`, never relaunched at boot); a hosted
account's daemon-level calls (`save`, `killDaemon`, `resurrect`, `internals`, `hostMetrics`) fail with
`forbidden`.

| Method | Wire request | Returns |
|---|---|---|
| `connect()` | `ping` (checks `protocol === 2`) | `this` |
| `ping()` | `ping` | `{pid, version, protocol}` |
| `start(opts \| opts[], {ifChanged?})` | `start` | `ProcessRecord[]` |
| `scale(name, n)` | `scale` | `ProcessRecord[]` |
| `stop(target)` | `stop` | `ProcessRecord[]` (returns once dead) |
| `restart(target, {toggleWatch?})` | `restart` | `ProcessRecord[]` |
| `delete(target, {flush?})` / `del` | `delete` | `ProcessRecord[]` |
| `reset(target)` | `reset` | `ProcessRecord[]` |
| `sendSignal(sig, target, {group?})` | `send_signal` | `ProcessRecord[]` |
| `list()` / `ls()` | `list` | `ProcessRecord[]` |
| `describe(target)` / `show()` | `describe` | `ProcessRecord[]` (every instance) |
| `save()` / `resurrect()` | `save` / `resurrect` | `{path, count}` / `ProcessRecord[]` |
| `killDaemon()` | `kill_daemon` | — (buses stay open to see `daemon:shutdown`) |
| `internals()` / `hostMetrics()` | `internals` / `host_metrics` | growable snapshots |
| `flushLogs(t?)` / `rotateLogs(t?)` / `reloadLogs()` | `*_logs` | touched file paths |
| `request(cmd, fields)` | anything | raw response — escape hatch for new requests |
| `launchBus(filter?)` | — | `EventBus` |

`target` is an id (`3` / `"3"`), a name (`"api"`), `"all"`, `{id}` or `{name}`.

### Start options

pm2 names, pm2 units: `script` (required; resolved against `cwd`), `name`, `cwd`, `args`
(string or array), `interpreter` (inferred from the extension — `.js/.ts→node`,
`.py→python3`, `.rb→ruby`, `.sh→bash`, `.pl→perl` — then resolved to an absolute path;
`'none'` = direct exec), `env`, `instances` (`'max'` = one per CPU), `autorestart` /
`restart_policy`, `max_memory_restart` (`"200M"`), `min_uptime`, `kill_timeout`,
`restart_delay`, `listen_timeout`, `wait_ready`, `watch` (`true`, path, list),
`ignore_watch`, `out_file`/`error_file`/`log_file` (`'NULL'` disables), `merge_logs`,
`log_date_format`, `stop_exit_codes`, `cron_restart`, `depends_on`, `max_restarts`,
`backoff_*`, `log_rotate`, … — the full list is in `index.d.ts` (`StartOptions`).
Unknown keys throw `invalid_spec` with a did-you-mean.

### Events

`launchBus({ subscribe?, process?, ids?, reconnect? })` sends the daemon an
`event_filter` so unwanted events never cross the socket:

- `subscribe`: globs over event names — `['process:*']`, `['log:err']`, `['process:exit', 'log*']`
- `process`: one app name; `ids`: process ids
- `reconnect: true | {delayMs, maxAttempts}`: re-subscribe after an EOF (e.g. across `3pm update`)

| Emitted as | Also as | Payload |
|---|---|---|
| `log:out`, `log:err` (from `stream`) | `log` | `{id, name, instance, line, stream, ts}` |
| `process:event` + `process:<kind>` (`start`, `online`, `exit`, `restart`, `stop`, `delete`, `errored`) | `process` | `{id, name, kind, exit_code?, exit_signal?, stderr_tail?}` |
| `log:rotated` | `log_rotated` | `{id, name, instance, path}` |
| `lagged` | — | `{missed}` — the subscriber fell behind |
| `daemon:shutdown` | `daemon_shutdown` | `{reason: 'kill' \| 'signal' \| 'home_removed'}` |
| `event` | — | every decoded event, unknown kinds included |
| `connect`, `reconnect`, `close({reason})`, `error` | — | bus lifecycle |

A `close` with `reason: 'eof'` (no `daemon_shutdown` first) is an unannounced daemon death.

### Errors

Every failure is a `ThreePMError` with `.code`: the daemon's codes (`not_found`,
`invalid_spec`, `spawn_failed`, `unknown_request`, `shutting_down`, `internal`) or
client-side `daemon_unreachable`, `protocol_skew` (run `3pm update`), `timeout`,
`disconnected`, `protocol`. Codes are exported as `CODES`.

## Application metrics and events: `io`

The other half of the package: what a process **started by 3pm** reports
about itself (pm2's `io.metric`/`io.counter`/`io.notifyError`, for 3pm).
Zero dependencies, never blocking, never throwing; outside 3pm (no
`THREEPM_CTL_SOCK` in the environment) every call is a no-op.

```js
const { io } = require('@3pm/api');
io.init({ source: 'worker', builtins: true });      // one connection per process of the app

const requests = io.counter('http.requests', { labels: { method: 'GET' } });
const latency  = io.histogram('http.latency', { unit: 'ms' });
const temp     = io.gauge('room.temp', { unit: 'Cel', description: 'studio temperature' });
io.metric('queue.depth', () => queue.length);        // read at every flush

requests.labels({ status: 200 }).inc();
latency.observe(12.5);
temp.set(21.5);
io.event('deploy', { level: 'warn', attrs: { version: '1.2' } });
io.notifyError(err, { attrs: { where: 'worker' } });  // an `error` event with the stack
```

Then `3pm show <app>` (section "Custom metrics") and `3pm events -f app`.
Full reference, bounds and the three languages: `docs/app-metrics.md` in
the 3pm repository.

| Call | Notes |
|---|---|
| `io.init({ source?, flush_ms? = 1000, builtins?, profile?, on_shutdown?, keep_alive? = true, connect_wait_ms? = 5000 })` | Connects in the background; idempotent. While the application declares an action of its own (`io.action`, `io.stream`; the SDK's `3pm:profile` does not count) the process is held up to serve it, so a worker that only answers calls needs no `setInterval`: `keep_alive: false` opts out, and the daemon's `shutdown` lets go. A call or a session placed before the connection is up waits for it up to `connect_wait_ms`; `await io.ready()` says when it is (`false` after the wait, or when not under 3pm). `source` defaults to a name unique per connection: `main` for the process 3pm spawned, `worker-<pid>-<threadId>` in a worker thread, `pid-<pid>` in a forked child (`child_process.fork`, cluster) or a child of the process holding `main`; two processes that both look spawned (`sh -c 'node a & node b'`) both claim `main`, and the one refused falls back to `pid-<pid>` — two connections of one instance under one source replace each other. The built-in `3pm:profile` action is declared on `main` only (a bare `3pm profile <app>` is never ambiguous); `profile: true` adds it on any source, `profile: false` withholds it. `builtins: true` adds the event loop (`nodejs.eventloop.delay` histogram, `.p99`/`.max`, `.utilization`), GC pauses (`nodejs.gc.duration{kind}`), memory (`nodejs.heap.used`/`.committed`/`.limit`, `nodejs.memory.external`/`.array_buffers`), `nodejs.handles.active`, `process.cpu.utilization`, and the counters `process.page_faults.major`, `process.context_switches.involuntary`; `builtins: { heap_spaces: true }` adds `nodejs.heap.space.used{space}` |
| `io.electron(require('electron'))` | Electron main process: `electron.processes`/`.process.cpu`/`.process.memory` by `type`, `electron.power.*` (battery, thermal state, speed limit), counters and events for crashed processes and hung windows, sleep/wake. Returns `stop()` |
| `createRendererIo(ipcRenderer, { source?, builtins? })` (`@3pm/api/renderer`) | Renderer/preload side of `io.ipcBridge(ipcMain)`. `builtins: true` adds `renderer.eventloop.delay`, `renderer.longtask.duration`, `renderer.heap.*`, `renderer.blink.*`, `renderer.cpu.utilization`, `renderer.dom.nodes` (what the runtime offers) |
| `gauge` / `counter` / `meter` / `histogram(name, { unit?, description?, labels?, scale? })` | Same name + labels = same object. Counters are cumulative; histograms are base-2 exponential (`scale` 0..4, default 1) and lower their scale by one (buckets merge in pairs) whenever a 129th bucket would open, so an entry always fits the daemon's cap. `.labels({...})` derives a child — cache it rather than calling it per request; past 50 label sets a metric folds into `overflow="true"`, and the fold's `.labels()` returns the fold |
| `metric(name, fn, opts?)` | Gauge by callback, evaluated on the event loop at flush. Callbacks and metrics share the 200-name budget: a name is one or the other |
| `event(name, { level?, attrs?, exc?, fp? })` | Sent at once (queued while the connection comes up) |
| `notifyError(err, { attrs?, name? })` | `error`-level event carrying `exc.{type,message,stacktrace}` |
| `action(name, { description?, remote? = true, timeout_ms? }, handler)` | Expose `handler(params, ctx)` to `3pm trigger <app> <name> [params]` and, unless `remote: false`, to the fleet backend. What it returns (JSON, ≤ 512 KiB, ≤ 64 levels deep) is the result; past either bound the call fails with code `too_large` — a bigger one is a file: write `ctx.file + '.tmp'` then `await ctx.commitFile(ext)` (past the call's timeout it deletes the `.tmp` and throws). Whatever the handler throws is the failure, even a value with no readable message. Returns `unregister`. A `3pm:` name throws (the SDK's prefix). At most 32 actions per source, `3pm:profile` included. `3pm:profile` (CPU/heap sampling profile, `{kind: cpu\|heap, duration_ms: 1000..60000, sampling}`) is declared on `main` by default; invalid params, or a timeout leaving less than 15 s after the sampling, fail the call before anything is sampled |
| `flush({ wait_ms? })`, `shutdown()` | The whole registry is sent every `flush_ms` anyway. `flush` resolves `true` when every batch was written (and, with `wait_ms`, left the process); `shutdown` is safe to call twice |

Strings are bounded in UTF-8 bytes, as the daemon measures them: label
values ≤ 64 bytes (else the series is refused), unit/description cut to
16/128 bytes, event attributes to 256, exception fields to 128/1024/8192.
Cuts land on character boundaries and lone surrogates become U+FFFD, so
nothing sent is ever refused. Everything else that goes wrong (bad name,
reserved label, wrong kind for a known name) returns a no-op metric
(`enabled: false`) and logs under `THREEPM_IO_DEBUG=1`.

`init` also accepts transport knobs, advanced and mainly for tests:
`reconnect_min_ms`, `reconnect_max_ms`, `max_buffer_bytes` (outbound
budget past which metrics and events are dropped, 64 KiB; the actions
declaration and replies have their own and are never crowded out),
`max_refusals`, `refusal_window_ms`. `io.stats` is a copy of `{ sent,
dropped, connects, closes, refusals }` (metrics and events lines).

Then `3pm trigger <app>` lists the actions, `3pm trigger <app> <name>
'{"json": true}'` runs one on every instance, `3pm profile <app>` takes a
profile and prints the hot frames; every call is a `proc_action` event
(`3pm events -f proc_action`).

Electron: the renderer has no Unix socket. In the main process,
`io.init({ source: 'electron-main' }); io.ipcBridge(ipcMain, { flush_ms?,
max_sources? = 16 })`; in a renderer, `const { createRendererIo } =
require('@3pm/api/renderer'); const rio = createRendererIo(ipcRenderer,
{ source: 'renderer' })` — same surface, one control connection per
renderer source. The renderer entry loads no Node builtin; the main
process re-validates every relayed entry and bounds what it keeps per
source; a renderer cannot take an automatic name (`main`, `pid-*`,
`worker-*`) or the main process's own source. A renderer's
`rio.action()` needs an `ipcRenderer` with `on` (calls come back on
`3pm:action`, acknowledgements on `3pm:actions_ack`: the declaration is
sent again until the bridge has it, so one made before `ipcBridge()`
still arrives; `shutdown()` withdraws the actions and removes the
listeners). Several windows may share a source: each keeps its own
actions — a call goes to the window that declared it (the latest, when
several declared one name) and only its reply is taken — until it
declares again, commits a navigation (a reload too; the next page is
asked for its own), crashes or is destroyed; a navigation that never
commits (prevented, a download, an HTTP 204) keeps them. The bridge profiles a renderer over
`webContents.debugger` under `3pm:profile` when exactly one live window
holds the source (give each window its own source to profile it), and
fails at once when the debugger is already attached (DevTools open).

## Fleet calls: `io.call`, `io.open`, `io.stream`

An action can take bytes and answer with bytes, on this machine or on
another node of the bucket. Declare what it takes:

```js
io.action('infer', { timeout_ms: 60_000, input: { max_bytes: 100e6, ext: ['jpg', 'png'] } },
  async (params, ctx) => {                    // ctx.input: { path, bytes, ext, sha256 } — fetched and checked
    return await model.run(await fs.promises.readFile(ctx.input.path), params)
  })

io.action('upscale', { input: { max_bytes: 50e6 } }, async (params, ctx) => {
  await fs.promises.copyFile(await render(ctx.input.path), `${ctx.file}.tmp`)
  await ctx.commitFile('png')                 // with an output ticket: PUT on the agent's socket
  return { seed: params.seed }
})
```

`input.max_bytes` (≤ 8 GiB) refuses a bigger call before a byte travels;
`ext` narrows what it accepts. The bytes arrive as a file the daemon
named, fetched from the agent's socket, cut at the size the call
declared and checked against its digest — a mismatch fails the call
rather than handing over half a file. The SDK then sends
`action_started`, so the transfer is not the handler's timeout, and
deletes the file once the reply is out. `concurrent` (1 by default, at
most 16) says how many calls of that action may run at once; past it a
call is `busy`.

`output.max_bytes` (an integer, 1 byte to 256 MiB) bounds the file the
handler commits: past it `ctx.commitFile` throws `too_large` naming both
sizes, deletes the `.tmp`, and nothing is sent. On the fleet, a caller
that asks for the file reserves that much of its blob quota while the
call runs — 256 MiB when the action declares nothing — so a small bound
lets many calls run at once:

```js
io.action('screenshot', { output: { max_bytes: 2_000_000 } }, async (params, ctx) => {
  await capture(`${ctx.file}.tmp`)
  await ctx.commitFile('png')
})
```

Any other `output` value leaves the action unregistered (logged), since
the daemon would refuse the source's whole list over it.

> **Not in `@3pm/api` 0.1.3** (the published SDK): `output` comes with
> the next published version (`[Unreleased]` in `CHANGELOG.md`); 0.1.3
> ignores it without a word — no check in `ctx.commitFile`, 256 MiB
> reserved. For the bound to reach the fleet, the node's daemon and
> agent must be of that release too, and the backend: an older one
> drops it and reserves 256 MiB, as before.

Place one:

```js
const res = await io.call('gpu-1', 'vision', 'infer', { threshold: 0.7 }, { file: '/tmp/capture.jpg' })
// { call_id, ok: true, data: { count: 3 }, duration_ms: 840, results: [...] }
const up = await io.call('gpu-1', 'vision', 'upscale', {}, { file: 'in.png', output: '/tmp/out.png' })
// up.file → { path: '/tmp/out.png', bytes, ext: 'png' }
await io.call(null, 'vision', 'infer', {}, { file })    // same machine: neither agent nor backend
```

`node` is a node id (`3pmn_…`) or a unique hostname; `null` is this
machine, and the file is then taken by path — nothing leaves it.
`'self'` is this machine through the fleet, by its own node id. `io.node`
is that node as `3pm link` recorded it (`{ id, cluster_id, server,
linked_at_ms }` from the agent's `link.toml`, `null` when the host is not
linked); name it by `io.node.id`, never by `os.hostname()`, which need not
match the hostname it enrolled under. One
instance answers (0 by default; `instance: null` asks every one, and is
refused when the call carries bytes). `opts`: `file`, `ext`, `sha256`,
`bytes`, `output` (a path, or `true` to be told the blob only),
`instance`, `source`, `timeout_ms`.

### One pass over a large input, not three

A hundred megabytes of input is mostly SHA-256 in JavaScript and a copy
to disk, not the road: `io.call` reads the file once to find its digest,
the callee writes it once into a staging file, and a handler that wants
the bytes reads it a third time. Two opt-ins remove two of those passes.
Both are off by default and neither changes what goes on the wire.

```js
// The caller already holds the digest — from a manifest, from whatever
// produced the file — so the SDK does not read the file to find it again.
await io.call('gpu-1', 'vision', 'infer', {}, { file, sha256, bytes })

// The handler reads the body as it arrives: nothing is staged on disk.
io.action('scan', { timeout_ms: 300_000, input: { max_bytes: 8e9, stream: true } },
  async (params, ctx) => {
    const h = crypto.createHash('sha256')
    for await (const chunk of ctx.input.stream()) h.update(chunk)
    return h.digest('hex')                       // only now are the bytes vouched for
  })
```

`sha256` and `bytes` go together; the size is checked against the file
(one `stat`, which catches a digest kept after the file changed under
it) and the digest is taken at your word — verifying it is the work
being skipped, and the callee checks it against what lands as ever.
Without `file`, or on a call to this machine (`node: null`, where the
daemon takes the file by path), they are ignored rather than refused.

`input.stream` changes three things worth knowing.

1. **`ctx.input.path` is `null` on a call from another node** — nothing
   was written. A call placed on this machine always names a path, and
   `stream()` there reads the daemon's own file, so one handler serves
   both without asking which it is.
2. **The size and the digest are checked at the end of the body**, not
   before the handler runs: a truncated or altered input throws out of
   the iteration. Nothing has vouched for the bytes until that iteration
   ends, so do not act on them before it does — a handler that `break`s
   early, or swallows the throw, has used bytes nobody checked.
3. **The transfer counts against `timeout_ms`.** The handler starts
   before a byte has moved, so the budget must cover the download as
   well as the work; the default path hands the handler its whole
   budget with the bytes already in. A hundred megabytes over a ten
   megabyte per second link is ten seconds of it.

The body is read once — a second `stream()` is `refused` — and a handler
that never calls it never fetches the bytes at all. The SDK still hashes
the body to check it (about 88 ms per hundred megabytes), so a handler
that also hashes pays for two passes; take the digest from the iteration
if that is all you wanted.

A failure is a `CallError` with a `code`: `timeout`, `refused`,
`too_large`, `busy`, `gone`, `failed`, `lost`, `node_offline`,
`node_unknown`, `ambiguous`, `forbidden`, `agent_too_old`,
`daemon_too_old`. `node_offline` is also what a call gets when this host
has no agent at all, and `daemon_too_old` when the daemon answered no
ticket within five seconds. A `CallError` prints with its code:
`String(err)` is `node_offline: …`, `console.error(err)` shows
`CallError [node_offline]: …`; `err.name` and `err.message` are unchanged.

### Streams

A stream is a call with two framed legs instead of one result:

```js
io.stream('analyze', { idle_ms: 30_000, concurrent: 8 }, async (session, params) => {
  for await (const f of session.input.frames())               // one JPEG per frame
    session.output.writeFrame(await model.annotate(f.payload), { key: true })
  return { done: true }
})

const s = await io.open('gpu-1', 'vision', 'analyze', { fps: 15 })
s.send(io.mjpeg(ffmpeg.stdout))                                // late frames are dropped, never queued
for await (const f of s.output.frames()) display(f.payload)    // ends on the callee's END
await s.close()
const { data } = await s.result                                // rejects if either leg died
```

A live session always goes through the agent's data socket, so
`io.open(null, …)` (or `'self'`) reaches this machine through the fleet,
as `io.node.id`; on a host that is not linked it rejects `node_offline`
(`this host is not linked: …`).

`idle_ms` (a stream only) is how long a session may stay silent, at most
24 h; it is the same wire field as `timeout_ms`, which the daemon reads as
a session's idle period, and wins over it.

**A leg that dies never kills the process.** A refusal from the agent — a
stale token, a quota, a lost link to the backend — is a routine event: the
sink keeps the failure, `await s.result` rejects with it, and `s.failure`
names it. `send()` and `pipe()` are both safe, with no `'error'` handler to
remember. Listen for `'error'` on `s.input` (or use `stream.pipeline`)
only when you want to know before the reply.

`await s.close()` means the bytes are in: it ends this side's leg, waits
for the agent to answer it (3 s by default, `opts.sink.settle_ms`; an
agent that does not answer by then has its leg taken back rather than
leaking a socket) and then stops reading. It never rejects — the failure
is on `s.result` and `s.failure`. On the caller's side, closed, the
session gives its slots back at once — its place among the 16 sessions
and the 16 calls in flight of the process — so the next `io.open` is not
`busy`; its call stays routed, so a reply that comes late is still
`s.result`, for `reply_grace_ms` (10 s at most), after which `result`
rejects `lost`. The daemon keeps its own count: a call placed on the
fleet holds one of the host's 64 forwards until it is answered or its
deadline passes, whatever its legs did — the callee answers once it sees
the `END`, which is what normally frees it. On the serving side the same rule
holds: a handler that returns happily while its output leg was refused
answers a failure, not `ok`. A stream handler must return: the session
holds one of the action's `concurrent` slots until it does, and nothing
reclaims it before `ACTION_STREAM_MAX_MS` (24 h).

**Ending a session.** `s.close()` also cuts the reading leg: a
`for await` over `frames()` in progress simply returns — it never throws
`ERR_STREAM_PREMATURE_CLOSE` — while a call that failed still throws its
failure there. A leg is read once: a second `frames()` while the first
reads throws (a closed or failed leg says why instead). To stop sending
but hear the rest of the answer, half-close
with `await s.end()` (both sides have it): it lets go of whatever is piped
in, writes `END`, waits for the agent like `close()`, and leaves the
reading leg open until the other side's own `END`. After the end a
`writeFrame` is ignored and returns `false`: a producer racing the close
is not an error. So is a frame the wire cannot carry — a payload over
1 MiB, a `ts_us` that is not an integer: `writeFrame` returns `false`,
the sink counts it in `stats.refused` and says it on `'refused'`, and
the leg goes on (a frame `pipe()`d in is refused the same way, never
failing the stream; `send()` counts only the frames that went).

**Feeding a source.** `s.send(readable, { end })` writes one frame per
chunk (`io.mjpeg(…)`'s images, or objects shaped like `writeFrame`'s)
and ends the leg with `END` when the source ends, fails or is closed —
`ffmpeg` killed mid-image included — or when `sending.stop()` is called
(the source is paused, never destroyed). `{ end: false }` keeps the leg
open for the next source. `await sending.done` never rejects: it is
`{ frames, error }`, `error` naming what broke the source (`null` for an
end, a close or a stop). `pipe()` still works; it just ends the session
with its source and says nothing when the source dies. `io.mjpeg` itself
never emits `'error'`: a source that fails or closes early ends the
images, drops the partial one, and leaves the reason on
`out.source_error`.

**Relaying.** A received frame is written on another leg as it came —
`session.output.writeFrame(f)` keeps its payload, flags (unknown bits
included) and `ts_us`; the sink gives it its own `seq`. A named flag
passed explicitly wins: `writeFrame(f, { key: false })`.

**What the path lost, and how late.** `s.stats.dropped` counts only this
process's abandon sink; the agents and the backend's window drop too.
Every one of them leaves a hole in `seq`, so the reading side counts
them: `s.stats` is `{ sent, dropped, bytes_out, received, bytes_in,
missing, discont }` (`missing` from the gaps, wrapping at 2³²; an `INIT`
replayed to a late reader is not a gap). The reading leg's own
`stats.latency` (`s.output.stats` on the caller, `session.input.stats`
on the callee) is `now − ts_us` over the last 256 frames, as `{ samples,
p50_ms, p95_ms, max_ms }` at the wall clock's millisecond resolution. It
is exact on a round trip, where one clock stamped and read the frame;
between two hosts it includes their clock skew, and can be negative. A
producer that wants to react sooner listens on the sink:
`s.input.on('drop', ({ seq, key, bytes, dropped }) => lowerFps())`.

Both sides write through an **abandon sink**: frames go out as fast as
the socket takes them and, when it says no more, they queue: a frame
that waited past a 100 ms window is dropped on its own age, never the
newest one (`s.stats.dropped` counts them) — `opts.sink.window_ms` moves
it, and it is this process's own window, unrelated to the backend's
`--live-window-ms`; past 8 MiB queued the oldest frame that is not a
`KEY` goes first, then the oldest `KEY` or superseded `INIT`, until the
queue fits. The newest `INIT` frame is always kept, as the agent's sink
keeps it: without it a reader that joins cannot decode at all; an older
one is an ordinary frame. Abandon starts at the source; back-pressure never travels
up a live stream. At most 16 sessions per process, as the daemon allows,
and at most 16 calls in flight on one connection.

A message is a 20-byte big-endian header and a payload of at most 1 MiB:
`u32 len | u8 flags | u8 hdr | u16 rsv | u32 seq | u64 ts_us`. Flags are
`KEY 0x01` (a message a reader may join on), `END 0x02` (the close),
`INIT 0x04` (decoding parameters — SPS/PPS, `ftyp+moov` — kept and given
to every reader that joins) and `DISCONT 0x08` (a gap: reset the
decoder). A longer header is accepted and skipped, a shorter one
refused, and unknown flag bits are carried. `ts_us` is a BigInt: the top
of its range does not fit a double. `io.mjpeg(readable)` cuts JPEG
images on `FFD8`/`FFD9` as they arrive — the `FFD9` looked for past the
headers, up to the start of the scan (`SOS`), so the thumbnail an EXIF
block carries never cuts its image short — and marks each one `KEY` (give
`ffmpeg` `-flush_packets 1`, or it adds 33 ms). It holds at most one
message: an image that never ends — a dead encoder, or a stream that is
not MJPEG but happens to carry `FF D8` — is given up on, the next start
marker resynchronises, and the frame that follows carries `DISCONT` so
the decoder resets. For H.264, keep the GOP under a second, drop
B-frames, and send SPS/PPS as an `INIT` frame.

`io.open` takes no `file`: a session carries frames, and a file belongs
to `io.call` (passing one is refused rather than ignored).

### `data.sock`

The SDK knows one path and never a network address: the agent's Unix
socket. This is the contract it speaks, and the whole of it.

**Where.** `$THREEPM_DATA_SOCK` when set. Otherwise, under
`$THREEPM_HOME` (or the home `resolvePaths` finds): `agent/data-<uid>.sock`
when one exists for this uid — a root daemon running a hosted user's app
leaves one socket per account, mode 0600, owned by that uid — else
`agent/data.sock`. A candidate counts only when `stat()` says it is a
socket. None of them: there is no agent here (and never one on Windows,
where a call with bytes is `node_offline` and nothing crashes).

**Requests.** HTTP/1.1 on that socket, `Authorization: Bearer <token>`
on every one — the local token the daemon put in the call's ticket, not
the node's. The blob id is percent-encoded in the path. Keep-alive is
asked for by everything except a tail, which takes a connection of its
own, and must be allowed, never required — a `PUT` asks for it so that
a refusal landing mid-body survives the answer.

| Request | Body | Answer |
|---|---|---|
| `PUT /blobs/<blob>` | the file, with `Content-Length` **and** `X-3pm-Sha256: <64 lowercase hex>` | `201` |
| `PUT /blobs/<blob>` | frames, `Transfer-Encoding: chunked`, no digest | `201` when the body ends |
| `GET /blobs/<blob>` | — | `200`, the bytes |
| `GET /blobs/<blob>?live=1` | — | `200`, the frames, until the writer's `END` |

Those four are the whole surface. `X-3pm-Sha256` is a **request header,
never a trailer**: the SDK hashes a file before it places the call (the
digest travels in the call itself, so the callee can check it), and sends
no HTTP trailer anywhere. A `Content-Type` of
`application/octet-stream` is sent and need not be read. The SDK never
sends `Expect: 100-continue`.

**Statuses.** `200` a `GET`; `201` a `PUT` taken. `401` the token is
unknown or expired; `404` no such blob, or not one of this call's; `409`
the blob is settled (a second `PUT`); `413` over the size the call
declared; `422` a malformed request (see below); `429` over a quota of
the bucket or the node; `503` the agent has no link to the backend, or
the backend no longer knows the agent's bearer (it takes a fresh one at
its next `welcome`, so the caller retries), or the agent is already
holding every connection it can — 512 for the host, shared by every app
on it (`this agent holds all the connections it can`). One connection
per transfer in flight is the rule, so that last one is a host running
hundreds of streams at once, not a busy app. Every other status the
backend answers with is relayed as it is when it is one of `403`, `404`,
`408`, `409`, `413`, `422`, `429`, and becomes `502` otherwise — that
is the agent's one rule, not a list of special cases. The SDK reads them as
`forbidden` (401, 403), `gone` (404, 409), `timeout` (408), `too_large`
(413, 422), `busy` (429), `node_offline` (503) and `failed` for anything
else; `ENOENT`, `ECONNREFUSED`, `EACCES` and `ENOTSOCK` on the socket
are `node_offline`. A refusal's body is read up to 1 KiB and its first
512 bytes go into the error message, so it is worth one plain line.

**Timeouts, per direction.** A response header is waited for 10 s on
every request **except a `PUT`**, sized or chunked: both are answered
only when the body ends — hours later for a stream, minutes for a file
of a few gigabytes — so neither is given a head budget at all. The agent
must therefore send a `GET`'s header as soon as it accepts the request —
`200`, flushed before the first byte of the body — and must not answer a
`PUT` early. That budget starts when the request gets a socket, not when
it is created; a request still queuing behind the pool has its own
budget of 120 s, after which it fails `timeout` rather than waiting for
ever. What holds a `PUT` instead is silence, twice over: 5 min without
the body moving ends it, and 5 min between the last byte and the status
ends it too. Both are long on purpose. A `GET` body gets 60 s
(`BLOB_IDLE`), but an upload stops moving for a reason that is not a
fault — the backend's window fills until a reader attaches — so the
write budgets guard against an agent that has gone away, never against
a reader who is slow. The SDK
sets `TCP_NODELAY` and keeps a pool of 8 keep-alive sockets for
requests that are over in a round trip and those alone: a read the
caller knows is settled **and** under a mebibyte. Every `PUT`, sized or
chunked, every read of a blob still being written, and every settled
read larger than that go on an unbounded agent instead, so that the
ninth of them never waits out its queue budget behind eight transfers
that have not finished. A `PUT` keeps `Connection: keep-alive` even so:
a refusal that lands while the body is still going out must survive,
and a connection closed on the spot takes the answer with it. Pooled
sockets are dropped after 8 s, under the agent's own 10 s, so that the
SDK is the side that closes first.

**Sizes.** Header lines stay under 8 KiB and the whole header block
under 16 KiB (the SDK sends about 300 bytes). The 16 KiB is enforced on
the **first** head of a connection, which the agent reads itself; the
heads 2..n of a keep-alive connection are bounded by the agent's own
read buffer, 256 KiB, so a later head between the two sizes is served
rather than refused. A `file` `PUT` is cut at the size the call
declared; a `live` frame is at most 1 MiB.

Sixteen details an implementation needs, settled here so both sides agree:

1. **No `Range`.** A blob is read whole. The SDK never sends one, so
   there is no `206` and no `416` to implement; an agent that wants to
   answer a `Range` from some other client may, but nothing here depends
   on it.
2. **A malformed `PUT` is `422`.** That means: neither `Content-Length`
   nor `Transfer-Encoding`, an `X-3pm-Sha256` without a `Content-Length`
   (a chunked body has no digest to check against), a digest that is not
   64 lowercase hex, a body that does not match the digest, a `live`
   frame over 1 MiB, and a chunked `live` body whose bytes are not
   frames at all or that stops in the middle of one (the agent says how
   many bytes into the message it ended). A body longer than the
   `Content-Length` it announced is what the backend refuses `413`, and
   the agent relays that as it relays every backend status (see
   *Statuses*). The SDK produces none of these; the agent refuses them
   rather than guessing.
3. **A tail ends with its body.** After relaying the writer's `END`
   frame the agent closes the response. The SDK's reader stops on the
   `END` flag *and* on the end of the body, so both work — the rule is
   written down so the agent and the backend do not pick different ones.
4. **`GET ?live=1` late.** A blob whose writer finished cleanly and
   whose ring the agent still holds answers `200` and an immediate `END`:
   a clean end, not an error. A blob that is unknown, or whose ring is
   gone, is `404` — which the caller reads as `gone`.
5. **A chunked `PUT` cut short** (the writing process died mid-body)
   seals the blob as truncated, never as complete: the reader's body ends
   without an `END` frame. A `file` `PUT` that stopped short of its
   `Content-Length` is discarded, so the reader sees fewer bytes than the
   call declared and fails the call rather than handing a half file to a
   handler.
6. **No `403`.** `401` covers the token and `404` covers "not a blob of
   this call". The SDK still maps `403` to `forbidden`, for a proxy that
   invents one; the agent does not use it.
7. **Response headers** stay under 8 KiB in total. Node's parser stops
   at 16 KiB and the SDK would read the failure as `failed`; it reads no
   response header other than the status.
8. **A blob id** is `^[A-Za-z0-9_-]{1,64}$` (the backend mints
   `bl_<hex>`). The SDK percent-encodes the path segment anyway, so the
   agent percent-decodes it and answers `404` to anything outside that
   alphabet.
9. **The agent's own request-header budget** is 10 s, answered `408`
   (which the SDK reads as `timeout`).
10. **Keep-alive.** The agent may close an idle socket whenever it likes
    — the SDK's pool holds them for 15 s and tolerates a close — but it
    must not answer `Connection: close` to a short request, or the pool
    of 8 is rebuilt at every call.

## The fleet backend's HTTP API: `@3pm/api/fleet`

Everything above is this host's daemon. A *fleet backend* — the server
side of `3pm link` — is a different thing: an HTTP API over every node
that reports to it. Its client is generated from that backend's own
route table and ships here as a subpath:

```js
const { Client } = require('@3pm/api/fleet');

const api = new Client({ baseUrl: 'https://fleet.example.com', token: '3pmu_…' });
const buckets = await api.listBuckets();
const nodes = await api.listNodes(buckets[0].cluster_id);

for await (const [kind, payload] of api.follow(buckets[0].cluster_id)) {
  if (kind === 'telemetry') console.log(payload);
}
```

A method per endpoint, `api.url(name, …)` for the routes that are bytes
rather than JSON (an artifact's file, a blob), `ApiError {status, code,
detail}` carrying the backend's own `ERR_*`, and `api.discover()` to
follow the table of whichever backend answers. A token is how you
authenticate — there is no browser session to lean on. The live feed is
read with `fetch` and a `ReadableStream`, the only way Node has.

The feed is cut on the server before it leaves. It comes **without**
`init.history` unless you ask for it, and takes the rest as a query:

```js
// Only the nodes whose hostname contains "envision", each telemetry frame
// cut to the host's gauges and a short row per process.
api.events(clusterId, handlers, { match: 'envision', view: 'live' });
for await (const item of api.follow(clusterId, { query: { view: 'live' } })) …
api.events(clusterId, handlers, { history: 1 }); // the per-process history back
```

On a bucket of nine hosts reporting every second that is ~100 KB/s down
to ~14 KB/s (five hosts, `view: 'live'`), and ~1 KB/s through a proxy
that gzips `text/event-stream`.

It is the browser client (`GET /api/v1/client.js`) generated a second
time as CommonJS, with the browser half — the token prompt, the
`EventSource` reader — cut out. Both come from
`3pm-backend/src/api/client.tmpl.js` via `just api` in that repository,
so neither is edited here.

## Compatibility

The wire is growable: unknown events are still surfaced on `event`, unknown
response tags become a `protocol` error rather than a hang, and the frozen
fixtures of `threepm-protocol` are part of this package's test suite.
`PROTOCOL_VERSION` is checked on every `connect()`. The `live` framing has
frozen vectors of its own
(`crates/threepm-protocol/tests/wire_samples/live_frames.json`), replayed
here.

A daemon that does not know how to place a call ignores the line: the
call fails `daemon_too_old` after five seconds, and nothing else
changes. A daemon that dies mid-call fails it `lost` at once, rather
than at its deadline. An older SDK never receives an input, since it
declares none.

With no agent on the host there is no socket to find, and a call that
carries bytes is `node_offline` before it is even placed; only the local
call (`node: null`) works. Every other fleet code — `agent_too_old`,
`node_unknown`, `ambiguous`, `forbidden`, `lost` — comes from the daemon
on the reply and is passed through as it arrived, newer ones included:
the SDK produces only `node_offline` (no socket, no daemon),
`daemon_too_old`, `busy` (its own caps), `refused` (what it can check
before placing a call) and `failed`.

## Development

```console
npm test                       # unit + wire fixtures + e2e (uses ../../target/{debug,release}/3pm or THREEPM_BIN)
npm run coverage               # the same, with per-file coverage
npm run check-types            # compile the published declarations and a consumer under --strict
```

`check-types` compiles `test/types/consumer.ts` — which imports only from
`index.d.ts` and exercises the documented lines, `io.mjpeg(…).pipe(s.input)`
included — together with the declarations, **without** `--skipLibCheck`,
which would skip the `.d.ts` files and so the whole point. The package has
no dependencies and gains none for this: TypeScript is taken from a local,
workspace or `npx`-cached install, and when there is none (or no
`@types/node`) the gate says so and exits 0. `node --test` type-checks
nothing, so without this a broken declaration ships with 217 green tests.