# `modules/agent` — the `3pm agent` module

The second bundled module (`modules/agent/`, its own workspace and
lockfile, bin `3pm-agent`, `module.toml` name `agent`). Two roles in one
binary: the CLI 3pm dispatches to (`3pm agent link|unlink|status`; `3pm
link` / `3pm unlink` are built-in sugar over the first two, arguments
passed through untouched), and
`__run`, the supervised app that holds the link. Lib + bin: the binary,
the unit tests and the test backend share one implementation.

## Usage

```
$ 3pm link 3pmc_<cluster> 3pmk_<token> wss://fleet.example.com [flags]   # = 3pm agent link …
$ 3pm agent link 3pmc_<cluster> 3pmk_<token> wss://fleet.example.com [flags]
$ 3pm link 3pmc_<cluster> 3pmk_<token>   # no server: wss://fleet.edgecr4ft.net (DEFAULT_SERVER)
$ 3pm link [--json]               # the link; exit 0 only when established (= 3pm agent link, alone)
$ 3pm link stop [--json]          # pause: stop the app, keep the enrollment (= 3pm agent link stop)
$ 3pm link start [--json]         # resume; registers the app again if it was deleted
$ 3pm agent status [--json]       # also `3pm link status`
$ 3pm link delete [--json]        # = 3pm unlink (also `rm`, `remove`)
$ 3pm agent unlink [--json]
```

`link` flags: `--insecure` (accept `ws://` to a non-loopback host; over
`wss://` it is refused until the TLS hardening step), `--name` (app
name, default `agent`), `--cap` (repeatable, comma-friendly:
`metrics:read`, `proc:restart`, `proc:stop`, `proc:action`; default
all — a `link.toml` written before `proc:action` existed lists the
three of its day and is read as all), `--no-env-keys`
(do not ship environment variable names), `--force` (enroll again over an
existing link: fresh key, hence a new node id; the app is restarted in
place), `--json`.

`3pm link` alone — no positional and no `THREEPM_LINK_*` in the
environment — is a check: the `status` view plus a `link` line,
`connected for 3m 2s (session 2, server 0.1.0)` or `NOT established —
<why>`, and exit 0 only when the link is up (1 with `link not
established: <why>` on stderr: not linked, app stopped / not registered,
daemon unreachable, no session reported yet, `link down for … (<last
disconnect>), retrying`, `parked: the node was revoked`). The running
agent writes `session.json` in the state directory on every change of
its session (pid, connected, since, sessions, retries, parked, last
disconnect, server version) and removes it on a clean exit; readers
trust it only when its pid is the app's current one.

The same two files feed the line `3pm ls` prints above its table on a
linked host — `⇆ 3pm link | node 3pmn_… | dash: <bucket page>`, green,
or red with the reason (`crates/threepm-cli/src/link_state.rs` +
`render/linkline.rs`; the verdict itself is
`threepm_link::local::status`, shared with `status_cmd`). The CLI reads
the files directly: no module is spawned for a table. The bucket page —
`http(s)://<server>/b/<cluster-id>`, `ServerUrl::bucket_url` — is also
the `dashboard` line of `3pm agent status` and the last line of a
successful `3pm link` (`"dashboard"` in `--json`).

`link stop` stops the supervised app and saves the process list, so
the pause survives a reboot; the key and `link.toml` stay.
`link start` restarts the app or, when it was deleted by hand,
registers it again from `link.toml` (same spec as `link`). Both refuse
a host that is not linked, before the daemon is asked anything.

The three positionals can come from the environment, which **wins**
(a provisioning script over a pasted line): `THREEPM_LINK_CLUSTER`,
`THREEPM_LINK_TOKEN`, `THREEPM_LINK_SERVER`, `THREEPM_LINK_INSECURE`.
Empty variables do not shadow the positionals. The server is optional:
given nowhere (a blank value counts as none), it is
`inputs::DEFAULT_SERVER`, `wss://fleet.edgecr4ft.net`. A missing input names its
position and its variable; a malformed token is reported **without
echoing it**; clear text to a host that is not loopback is refused with
`ERR_INSECURE_SERVER` — the agent's own code string, not one of the
link's: the wire has no such failure to name.

`--json` contracts (also `THREEPM_JSON=1`, which 3pm's dispatcher sets
when it sees the module's `--json`):

| Command | Document |
|---|---|
| `link` | `{node_id, cluster_id, state: active\|pending, affected: [ProcessRecord]}` |
| `status`, bare `link` | `{linked, app, cluster_id?, node_id?, server?, process?: RedactedRecord, session?: {connected, for_ms, sessions, retries, parked, frames_invalid, last_disconnect?, server_version?}, daemon_error?}` — `session` only when the app runs and has reported |
| `unlink`, `link delete` | `{affected: [ProcessRecord], removed: bool}` |
| `link stop` | `{stopped: true, app, affected: [ProcessRecord]}` |
| `link start` | `{started: true, app, registered: bool, affected: [ProcessRecord]}` — `registered` when the app had to be created again |

Exit code 1 with `[3PM][ERROR] …` on stderr for every refusal
(`ERR_TOKEN_EXHAUSTED`, `ERR_TOKEN_INVALID`, a dead port, "already
linked — pass --force"…), and **nothing is written** before the backend
said yes.

## The supervised app

`link` starts this binary as `<name>` with `args = ["__run"]`, `cwd` =
the state directory, `wait_ready = true`, tagged `spec.module = {agent,
version}` (so it lists under **Modules** and `stop|restart|delete all`
leave it alone), then `Save`s the process list. Its environment is only
the `THREEPM_AGENT_*` options below — identity and server live in
`link.toml`, never in env (`3pm show` prints env).

| Variable | Default | Meaning |
|---|---|---|
| `THREEPM_AGENT_STATE_DIR` | `<home>/agent` | Holds `node.key`, `link.toml`, and `session.json` while the agent runs |
| `THREEPM_AGENT_NAME` | `agent` | App name (what `unlink` deletes) |
| `THREEPM_AGENT_LOG` | `info` | `tracing` filter for the `__run` role (stdout, so it lands in the app's out log). Read by the app, **not** injected by `link` — set it with `3pm env`/`--env` on the `agent` app if needed |

The pace of the link is the backend's (`welcome.pace`), not an option: the
`THREEPM_AGENT_INTERVAL_MS` / `THREEPM_AGENT_FULL_EVERY_S` an older
`3pm link` left on the app are no longer read, whatever they say.

Readiness is announced on the control channel **as soon as the identity
is loaded**: ready means "the agent runs", not "the backend answers" —
otherwise a backend outage would put the app in `errored` and 3pm would
restart it in a loop.

Then the session and its three workers (`run.rs::spawn_workers`: the
node's state, the events, and `publish_session` keeping `session.json`)
until a shutdown (control-channel `shutdown`, SIGINT — 3pm's default
kill signal — or SIGTERM, or the daemon's own `daemon_shutdown` event):

- **session**: connect with the node's two ids in `Authorization` →
  handshake → writer + reader → classify the disconnect → wait the
  schedule says → again. The agent speaks link 5 only: a server whose
  challenge offers an older version alone is refused before anything is
  signed (`ERR_LINK_VERSION`, retried on the Auth schedule — link
  protocol §8). Revoked parks with an hourly probe (logged once).
  Shutdown sends `Close(Normal)` and waits 2 s for the echo.
  A call is acked by whatever accepts it, as its first act — so one the
  reader could not hand over is never acked and the backend's view says
  so. Calls run in `commands.rs`, one task each: the capability against
  what this host allows (`link.toml` `caps`), four in flight at most (`ping` excepted), the
  daemon asked with a bounded budget (`DaemonLink`, a connection per
  call), an action's result gzip inline within the reply's budget and
  the file it returned uploaded to the backend's artifact store over
  HTTP before the reply names it (link protocol §5); a call whose
  connection dropped withholds its reply (at-most-once).
- **the node's state** (`node_state.rs`): the one stream that carries
  the node — link protocol §7 is the reference, not repeated here. One
  task, one daemon connection (`DaemonLink`, each call bounded to 3 s,
  reconnection 1 → 30 s), one `NodeWriter`. Every `welcome.pace.state_s`
  seconds (1 s floor) it asks the daemon for complete states — `List`
  and `HostMetrics`; `AppMetrics` (`id` alone) for the instances
  whose `(gen, app_metrics_rev)` moved, every `pace.metrics_s`;
  `ProcessTrees` for the instances with more than one process whose
  `(gen, tree_rev)` moved, plus the figures on `pace.trees`
  (`shape` = the shape only, `off` = no tree) —, shapes each into its
  section (`threepm_link::node_state`), cuts it against what the backend
  holds and sends what moved in **one** tracked `state` frame, empty
  when nothing did: the heartbeat. A section is committed only once the
  writer confirmed the bytes on the socket (a frame not written is cut
  again next tick, larger, at the same revisions); owed in full on the
  first tick of a session, on `state_resync {secs}` and when the daemon
  was replaced (another `boot_id`); said `gone` when its instance left
  the table, its tree is back to one process, or the host agent is down
  — never while the daemon itself is down. A section alone over the
  1 MiB cap is not sent (warned once) and commits nothing; a daemon
  answering `unknown_request` to `ProcessTrees` is asked again 60 s
  later. While the daemon is down the frame still goes: an empty table,
  no daemon identity.
- **events**: subscription to the daemon bus filtered to `process`,
  `daemon_shutdown` and `app`, forwarded as `emit`; lifecycle events
  wait for room on the control channel (never dropped), `app` events
  are `try_send` (a burst never holds the channel full and starves the
  `state` frames); `daemon_shutdown` then closes the link politely;
  resubscription 1 → 30 s.

## State directory

`<home>/agent/` (0700 on Unix), written only after a successful
enrollment, in this order: `node.key` first (base64url seed, **created
0600 before any byte lands**, temp file in the same directory + fsync +
rename), then `link.toml`:

```toml
# Written by `3pm agent link`. No secret here: the node key is node.key.
version = 4
cluster_id = "3pmc_…"
node_id = "3pmn_…"
server = "wss://fleet.example.com:443"
insecure = false
caps = ["metrics:read", "proc:restart", "proc:stop", "proc:action", "node:shell", "node:forward"]
env_keys = true
app = "agent"
linked_at_ms = 1788523845154

# Never written by `3pm agent link`: the operator's, by hand.
[shell]
gateway_ports = true
```

`version` (`threepm_link::local::LINK_FILE_VERSION`, currently 4) says how
`caps` reads: since 4 the list is literal; a file with `version = 3`
listing exactly the five capabilities of its day reads as all of them
(so a default link gains `node:forward`), any other v3 list is literal;
a file with `version = 2`
(written before `node:shell` existed) has exactly the four capabilities
of its day read as all — any other v2 list is literal —, a file with
`version = 1` (written before `proc:action` existed) has its three
capabilities of the day read as all, and a file with no `version` key —
never written by any build — is read literally, like today's. So a node
linked with the defaults offers `node:shell` (a root shell where the
agent runs as root) after its update, without a re-link: take it out of
`caps` to refuse `3pm fleet ssh`. `[shell] gateway_ports`
(`threepm_link::local::ShellSettings`, off when absent) lets `3pm fleet
ssh -R` bind an address other than the loopback, as ssh's
`GatewayPorts`: the agent reads it, the backend can neither read nor
change it, and a re-link keeps it. Unknown keys are ignored,
missing ones default, unknown capabilities are dropped at parse time —
except in `[logs]`, where an unknown key (a misspelt `file = false`)
makes the section malformed: it then ships nothing
(`LogsSettings::closed`), and `LinkConfig::logs_problem` says why.
The shape of `link.toml` and of `session.json` lives in
`threepm_link::local`, not here: the 3pm CLI reads both to print
its link line above `3pm ls` (`crates/threepm-cli/src/link_state.rs`),
without spawning this module. `shell.key` (X25519, 0600) appears
beside them at the first end-to-end `3pm fleet ssh` session the node
serves: its static key for that channel, shown as `shell key SHA256:…`
by `3pm link status`; the caller's own is `cli.key` beside its
`known_nodes` (`$THREEPM_HOME/cli.key`, moved from `~/.3pm/agent/` on
first use). `unlink` deletes the app (a missing one is not an
error), saves, then removes the directory except `authorized_keys` and
`shell.key`; not linked means the daemon is not even asked. `logs/`
holds the log pipeline's positions, source ids and miner state ("Logs"
below), so an unlink forgets them too.

## Fleet ssh access: `authorized_keys`

A node opens a `3pm fleet ssh` session (every mode: shell, exec, `cp`,
attach, `-L`, `-W`, `-R`, `--user`) only for a caller key listed in
`<state dir>/authorized_keys` — `$THREEPM_HOME/agent/` or
`THREEPM_AGENT_STATE_DIR`, `/root/.3pm/agent/` on a root node. Lines
`SHA256:<fingerprint> [comment]`, `#` comments, a bad line skipped with
a warning. sshd's `StrictModes`: owned by the agent's euid and not
group/other-writable (0644 accepted), else refused. **Absent or empty:
every session refused.** Re-read per session (64 KiB cap); written
atomically under `flock` of a sibling lock file by `3pm link
authorize|deauthorize`, `3pm link … --authorize` or
`THREEPM_LINK_AUTHORIZE`; listed by `3pm link status` (JSON
`authorized_keys {path, keys[], closed?}`). A file posed by hand by an
agent ≤ 0.48 is honoured as is once the agent is updated.

**Where the check sits.** `shell/host.rs`: the session takes one of 16
handshake slots (5 s for the caller's first message, 60 s for its
answer: the first-contact prompt), at most 8 of them per caller as the
backend stamped it (`from`, `shell/gate.rs`: the key itself is only
known at the handshake's last message, so this is what keeps one
stolen `3pmu_` token from refusing everyone else `unavailable`; the
same account's own token stolen still can, until it is revoked), the Noise handshake runs on the legs
(relay or LAN direct), the initiator's static key is checked against
the file, and only then is the session's permit taken, `--user`
resolved, a socket dialed or bound, or a process spawned. The proof is
a type: `shell::run` takes an `Authorized` that only the check builds.
Refusal: `AgentErrCode::KeyNotAuthorized` (`key_not_authorized`, the
backend's `forbidden`), the message naming the fingerprint and the
`sudo 3pm link authorize …` to type; the caller exits 255.

**The request is bound (prologue v2).** Prologue = `3pm fleet ssh e2e
v2` ‖ call id ‖ node id ‖ `Bound::encode` (`e2e/bind.rs`): tagged,
length-prefixed fields for the operation, argv and `pty`, the `RunAs`
user, the attach target/instance/mode/`read_only`/peek, `host:port`, an
`-R` takeover's listen and connection, `bind:port`, `term` and first
size, `record`. The command line encodes what it asked for (its
`nodeShell`/`nodeForward`/`nodeListen` body and the `record` the
backend answered), the node what it received in the `AgentRequest`;
any difference fails the handshake before anything starts. An `-R`
connection (`Forward{accepted}`) is taken only by the key that opened
the listen. A 3pm ≤ 0.48 cannot talk to a v2 agent: update callers and
nodes together.

**Recording.** `record` is the caller's (`--record`; without it the
command line sends `false`), lowered only
by the backend (bucket `record_shells` off, `cp`, forwards), bound by
the handshake; `[shell] record = "never"` in `link.toml` wins locally
and the node says so in its answer. The audit's command lines follow
the effective value (`shell/host.rs`: asked and the node records): an
unrecorded session builds no `vt100` audit (`shell/pump.rs`) nor typed
lines (`shell/attach.rs`), and emits no `line`/`screen`, only `open`,
`exec`/`attach` and `close {lines: 0}`.

**Recovering from a lockout** (lost key, file badly posed, node
closed): with local access to the machine (real ssh, the console,
`sshpass` as `envision@` on a tablet), `sudo 3pm link authorize
SHA256:…`; or, from a machine holding the offline rescue key made by
`3pm fleet key new --out`, `THREEPM_CLI_KEY=/Volumes/USB/secours.key 3pm
fleet key push <node> SHA256:…`. That rescue key is a fleet **caller**
key, not the release signing rescue key of `~/.3pm-release/`. Before an
agent that enforces the file reaches a node, pose the file with the
running agent (`3pm-rd/rd/fleet-ssh-authorized-keys/scripts/fleet-authorize.sh`, then
`fleet-authorize-check.sh`).

## Fleet identity, data socket and gate

Full behaviour and the CLI surface: `FEATURES.md` → *Fleet RPC* and
*Fleet link agent*. This is only the map of where it lives.

`session.json` carries the fleet half next to the link status
(`state::fleet::FleetSession`, `#[serde(flatten)]` over `SessionFile`):
`fleet_token` (the node's bearer, `3pma_<id>_<secret>`, redacted in
`Debug` and so safe to log) and `fleet_token_at_ms`, both replaced in
block at every `welcome` (a `welcome` with no bearer erases the
previous one) in the same file `publish_session` (`run.rs`) already
rewrites on every session change, 0600. `3pm agent fleet …` is a
*different process*: it reads the file rather than asking the running
agent, and reports `TokenProblem::Abandoned(pid)` when the file says
connected but its writer's `pid` is gone (`kill(pid, 0)`, `EPERM`
counting as alive): the file is rewritten only when the session
changes, so its age says nothing about a healthy agent — and judging
by it warned on every node linked for more than ten minutes.
`staleness()` speaks of the token's age only when that writer is gone
or the link is down.

`agent/data.sock` (`$THREEPM_DATA_SOCK` overrides) is opened by
`run_fleet::open` alongside the gate, never on its own — a platform without Unix sockets (`#[cfg(not(unix))]`) opens
neither, and a call that needs either reads `node_offline`. It serves
the SDK's `PUT`/`GET` contract (`packages/node/README.md` → *`data.sock`*)
and relays to the backend on the agent's own warm connections, without
ever handing the SDK a network address or a backend token.

The **`fleet_ready` gate** is a second control connection the module
opens to its own daemon line: `run_fleet::gate_config` reads the
32-byte secret the daemon drew for this launch of the `agent` app from
`$THREEPM_FLEET_SECRET` (`threepm_protocol::control::FLEET_SECRET_ENV`,
cleared from the process's own environment right after) and the pid
identity (`ctl::Identity::from_env`), then sends `hello{source:
"fleet"}` followed by `fleet_ready{node_id, cluster_id, data_sock,
secret}`. A root daemon that hosts users answers an opened gate with
`hosted_socks{socks: [{uid, gid, path}]}` — `<hosted home>/agent/data-<uid>.sock`,
the path the SDK of a hosted app looks for first — and `data::hosted`
binds each one on the same server as `data.sock` (same `handler::Ctx`,
so the same tokens), `lchown`s it to the user, 0600. The list is whole
(empty when the daemon hosts nobody) and the agent serves exactly it: a
reopened gate keeps what it already serves and closes what the list no
longer names — a daemon handoff re-reads `[users]` and cuts the gate. A path it
cannot take (a file or a link there, no directory) costs that user
alone. No secret in this launch's environment — an old daemon, or a
process that is not the daemon's own `agent` row — means no gate and no
route for a node-placed call; `fleet_gate::run` is what the daemon then
believes.

`3pm agent fleet nodes | apps | inspect | call | top` (`3pm fleet …` is
sugar over it, like `3pm link`) live in `src/fleet/` (`nodes`, `apps`,
`inspect` and `call` over `http::FleetHttp` with the `session.json`
bearer, `top` in `src/fleet/top/` fed by the backend's SSE) and
`src/fleet_gate/` (the server side that receives a `Forward` for this
node and places it). All five need only this host's own link — no
admin token, no separate credential.

## Logs

The agent reads this host's logs and posts them to the backend over HTTP,
never on the link (`link-protocol.md` §6b; user view: `FEATURES.md`, "Log
shipping"). It runs while the last `welcome` carried `logs` and
`link.toml` `[logs]` agrees on at least one switch
(`LogsSettings::effective`); a `welcome` that changes what is wanted
restarts it, one without `logs` stops it. A disconnect does not: the
sender retries, the sources keep their positions.

```text
apps (bus) ─┐
journalctl ─┼─▶ fold → redact → prep → mine → policy → budget → ring ─▶ cut every 10 s ─▶ POST /logs
files ──────┘                                └─▶ catalog, counts ──┘
```

| File (`src/logs/`) | Role |
|---|---|
| `mod.rs` | The tree and its map; `[logs] exclude` globs (`*`, `?` match `/` too) |
| `raw.rs` | `Item` / `Raw` / `Mark`: a line cleaned (ANSI stripped, cut at 8 KiB), not yet redacted, with the position its batch commits |
| `ids.rs` | Short source ids (`a1`, `j`, `f3`), kept in `logs/ids`, never handed to another source |
| `commit.rs` | Read vs committed positions: `offsets` and `journal.cursor` advance only on `200` |
| `sources.rs` | The entry: spawns the producers `LogsWanted` enables, one bounded stream (`ITEMS_CAPACITY` 1 024) |
| `apps.rs` | The 3pm apps' output: a second bus subscription (`log`), its own cursor; `lagged` counted as `lag`; the agent's own app skipped |
| `journal.rs` | One `journalctl --follow --output=json --priority=<n>` child (Linux; `THREEPM_JOURNALCTL` in tests), respawned 1 s → 60 s, resumes after the committed cursor, first start `--since=now`, entries older than 24 h dropped (`stale`) |
| `merge.rs` | Discovery every 60 s: the host agent's `LogSources` (through the daemon) merged with automonit's by (dev, inode), `exclude` applied, sticky set |
| `automonit.rs` | The automonit `logs` action (version 1) every 5 min, 15 s timeout; asks only the root-installed module row, never an app a user named `automonit` |
| `vet.rs` | What a file automonit adds alone must look like: a log by its path, never `/etc`, `/root` or a private name, normalised, parent not a symlink, the declared owner uid |
| `files.rs` | The poll tailer (1 s): first open at the declared inode from its end, rotation and `copytruncate`, 256 KiB per tick, 512 files, parked after 10 min idle, `EACCES` skipped until the next pass |
| `tail.rs` | One file: `O_NOFOLLOW \| O_NONBLOCK \| O_CLOEXEC` + `fstat` against the expected (dev, inode, uid; a successor: same owner, mode, one link), 64-byte fingerprint, lines with a stable `pos` |
| `redact.rs` | Secrets replaced by `<REDACTED>` before anything else reads the line; a pattern list, linear (`regex`), one line at a time (a YAML env `value:` on the line after its `name:` is not caught) |
| `prep.rs` | The PoC's prep V2: process tag kept, timestamp and level stripped, exception heads read as errors, logfmt/JSON as fields, ids masked; continuation lines folded |
| `drain.rs` | Drain (similarity 0.5 on informative tokens, level in the leaf key, bounded) |
| `mine.rs` | The miners of every source and their state (`logs/miner`, atomic, bounded); template ids never reused (an epoch in every wire source id, `f3.k2x9`) |
| `policy.rs` | What leaves line by line: counts per 5-min window always, a raw line only for `new`, `error`, `back`, `spike`, `return`, `context` |
| `policy_counts.rs` | The count windows and parameter stats (distinct + top 3), closed into `cnt` records within the wire's bounds |
| `budget.rs` | 4 KiB/s per source (256 KiB burst) and the 4 MiB RAM ring while the backend is away; over either, `dropped {budget\|ring}` |
| `gather.rs` | What the pipeline holds between two cuts: catalog, declarations, drops, positions; empty lines neither mined nor counted |
| `pipeline.rs` | The loop: items in, a cut every 10 s to the sender; `logs/status.json` for `3pm agent status`; the stop (`STOP_BUDGET`) |
| `ship.rs` | Cuts into parts under 1 MiB gzip / 4 MiB inflated, each opening with `batch {run, seq, ts}`; one in flight; `200` commit, `429` wait, `413` split, `400` drop, `401` next bearer, `503`/5xx/timeout retry 1 s → 60 s |
| `start.rs` | When the pipeline runs (`desired`), `HttpPoster` (the session's current bearer at each post) |
| `../http_logs.rs` | `FleetHttp::push_logs`: one `POST /api/buckets/<cluster>/logs`, the answer as a status code |

State, under `<home>/agent/logs/` (written atomically, `write_private`):

| File | Holds |
|---|---|
| `offsets` | per file `<dev> <ino> <offset> <fingerprint> <path>`, committed on `200` |
| `journal.cursor` | the journal cursor of the last accepted batch |
| `ids` | the source ids handed out, and the counter |
| `miner` | every source's templates and id counter, with its epoch; saved before a cut that carries a new template |
| `status.json` | the counters of the `logs` row in `3pm agent status` (not state: rewritten while running) |

A line read but not accepted before a stop is read again at the next
start and stored once by the backend (`(node, src, pos)`); app lines
carry no `pos` (the bus is never re-read).

**Stop.** On `shutdown` the session ends first, then `stop_logs`: the
pipeline closes every open window and tries a last batch within
`STOP_BUDGET` (5 s; `run` gives it 9 s at most before aborting), then
the other workers stop. The agent's row has `kill_timeout_ms = 10000`
(`start.rs` `KILL_TIMEOUT_MS`, and `[module] kill_timeout_ms` in
`module.toml` raises rows linked before at the next module refresh), so
the daemon does not SIGKILL it at the default 1.6 s. What cannot go out
in time is lost as counts (they are deltas) and read again as lines.

## Link traffic

`src/link_traffic.rs`: process-wide counters per (direction, kind) —
the closed list in `threepm_link::traffic`. Bytes and messages are
added where they pass: the writer and `send_frame` (a frame's compressed
payload; a `state` frame split pro rata of its patches' JSON length,
`Deflater::frame_sized` giving the JSON length), `next_frame` (a Text
frame's payload, by what it decoded to; pings 0 B), `FleetHttp::send_bytes`
(the one counted path of `get`/`post`/`put_artifact`/`put_blob`, the
caller's kind), `push_logs` (every attempt), and the relay's blob legs
(`link_traffic::Counted`, per chunk). `node_state/traffic_cut.rs` cuts
the `traffic` section every `max(1, 5 / state_s)` ticks through the
same engine; `link_traffic/file.rs` writes `<state>/traffic.json`
(counters + the last minute's rates) once a minute for the `traffic`
row of `3pm agent status`.

## Files

| File | Role |
|---|---|
| `main.rs` | Role detection (`argv[1] == "__run"`), clap dispatch, `[3PM][ERROR]` on failure |
| `lib.rs` | Module tree, `VERSION`, `AGENT_NAME`, `MODULE_NAME`, `now_ms` |
| `cli.rs` | `Cli { Link(LinkArgs) \| Unlink {json} \| Status {json} }`, `json_wanted(flag, lookup)` |
| `inputs.rs` | Pure `LinkInputs::resolve(&LinkArgs, lookup)`: env > positionals, parsing, the insecure guard, caps sorted/unique/known, `InputError` |
| `options.rs` | `AgentOptions {state_dir, name}`: `from_env(lookup, home)` (infallible: an empty value is no value) / `to_env()` round trip, `truthy` |
| `state/mod.rs` | `StateDir`: paths, `is_linked` (`link.toml` **and** `node.key`), `create`, `remove`, `remove_stale_files` (`node.crt` / `ca.pub`, unused since link 3), typed `read_*` / `write_*`; re-exports `LinkConfig` / `SessionFile` from `threepm_link::local`, whose shape the CLI reads too |
| `state/fs.rs` | `write_private` (0600 before any byte) / `write_public`: temp + fsync + rename |
| `identity.rs` | `generate_key`, `pubkey_b64`, `sign_enroll_proof(key, token_id)`; `NodeIdentity {key}`: `save`, `load` (which drops what an older link left beside the key), `pubkey_b64`, `sign_link_proof`, `direct_identity`. The key only: the `Authorization` header is two public ids, built where it is sent (`threepm_link::LinkAuth` in `session.rs::open`) |
| `tls.rs` | `connector(&ServerUrl, insecure) -> Option<Connector>`: Plain → `None`; Tls → rustls (ring) + `webpki-roots`; `--insecure` over TLS → refused loudly until the hardening step |
| `codec.rs` | `Deflater` (one stream per connection, `Sync` flush per frame), `Inflater` (for the test backend; cap inclusive, exactly 1 MiB passes like the real backend), `decode_server` (capped; a tag this build does not know is `Unknown`, a body it cannot read is `Invalid`) |
| `transport/mod.rs` | `ws_config()` (1 MiB messages/frames, 32 KiB write buffer → 1 MiB max), `connect(url, headers, connector)` (sub-protocol, `User-Agent`, 10 s, HTTP rejection → `ConnectError::Rejected {status, code, message}`), `Frame {Link, Close, Eof, Other}`, `next_frame` / `read_frame`, `send_link`, `close_reason_code` |
| `transport/handshake.rs` | `HelloFacts {boot_id, caps}`, `Welcomed` (`pace`, `fleet_token`, `pending_calls`), `check_challenge` (version overlap → the negotiated version, then skew; the agent speaks link 5 only, so an older server is `HandshakeError::LinkVersion` — `ERR_LINK_VERSION`, `Disconnect::LinkVersion` — before anything is signed), `hello_frame` (the signature over the challenge, the facts and the negotiated version: three frames, no separate `proof`), `read_link` (skips `unknown` frames within the timeout), `handshake(…)`, `HandshakeError::disconnect()` |
| `transport/writer.rs` | `Outbound {Link, Tracked(Link, oneshot::Sender<bool>), Scoped(Link, ConnId, Option<oneshot::Sender<bool>>), Close}` (`Tracked` answers `true` once the bytes went out, `false` when the codec dropped the frame, nothing when the connection died first; `Scoped` is a frame of one connection — a call's ack, its reply — skipped once that connection ended or its sender gave up, answered like `Tracked` when someone waits; a `Tracked`/`Scoped` frame nobody waits for any more is skipped, never written stale on the next connection), `run(sink, deflater, ctrl_rx) -> WriterEnd {Closed, Dead}` (every frame rides the one control channel), `encode_or_drop` (a frame over the cap is dropped and its tracked sender told `false` — never `Dead`; the node's state is packed under the cap before it gets here); `CTRL_CAPACITY` 256, `FLUSH_EVERY` 5 s, `PING_EVERY` 20 s |
| `transport/reader.rs` | the frames handed to the session (a call, a `state_resync`, a call result). The reader sends no `ack` of its own; a tag this build does not know is skipped and counted, and a body it cannot read is counted too and, for a `call`, answered `err malformed`, `run(stream, ctrl_tx, inbound_tx) -> ReaderEnd {Silence, Disconnect}`; `SILENCE` 60 s |
| `session.rs` | `SessionConfig {…, stop_probe, stable_session}` (defaults = the constants, injectable for real-time tests), `SessionStatus` (watch: connected, sessions, retries, attempt, last_disconnect, parked, server_version, pace, fleet_token, frames_invalid; in `session_types.rs`), `SessionHandle {ctrl_tx, status_rx}` + `take_state_resync()` (the `state_resync` orders for the node-state task, once; `RESYNC_CAPACITY` 16, `try_send`: a task not keeping up loses the order rather than stalling the link), `Session::new` / `run(shutdown)` (connect and handshake abort on shutdown; no backoff sleep after a shutdown), `random_boot_id`, `STOP_PROBE` 1 h, `CLOSE_GRACE` 2 s, `STABLE_SESSION` 60 s |
| `daemon.rs` | `DaemonLink::new(paths, role)`: `request(&Request)` (connects on the schedule with an identity `Ping`, bounded, a failure drops the client), `info()`, `boot_id()`, `replaced()` (another `boot_id` since the last call, said once: the node-state task then owes every section), `is_connected()`, `lost(why)`, `with_timeout`; `DAEMON_TIMEOUT` 3 s, reconnect 1 → 30 s — one for the node-state task, a pool of them for the commands; `daemon/testing.rs`: `FakeDaemon` (scripted daemon for the in-crate tests: `list`, `host`, `revs`, `trees`, `trees_unsupported`, `answers_per_connection`) and `fake_writer` (confirms every tracked frame with a verdict) |
| `node_state.rs` | `NodeStateTask::new(paths, redact).run(status_rx, resync_rx, ctrl_tx, shutdown)` (`gen` = the process's start in ms; ends with the session's `state_resync` sender; a period change from `welcome` is taken at once). A tick is `ask` (the daemon's complete states, a `Snapshot`; `plan(procs, paces, now)` says which `AppMetrics` / `ProcessTrees` to ask from the table's counters) then `round` (`cut` every section and the `gone` ones, `pack` them in frames under the cap — a section never split —, `stream::write`, `commit` on the writer's `true` only, which also records the `(gen, rev)` the section was shaped from). `CRC_EVERY` 60 s per section, each at its own moment of the minute (`paces::crc_minute`: a hash of the section's name shifts where its minute turns, and a patch carries a CRC when that minute is not the one of the section's last CRC: the sections that move every tick are not all serialized and checked on the same one); a patch's `p` moves into the frame, measured without being written (`json::json_len`); `TREES_RETRY` 60 s |
| `node_state/paces.rs` | What the `welcome`'s `Pace` asks of that task: the tick (1 s floor — the stream is always on), `Mode {Silent, ShapeOnly, Figures(n)}` from `pace.trees`, `Paces::of(status)` (`metrics_every` and the figures counted in ticks) |
| `stream.rs` | What a streaming task is made of: `ticker`, `Wake<T>` / `wake` (what woke the loop: the tick, a status change, what the backend asked, a stop), `write` (a tracked frame, confirmed by the writer; `CONFIRM_TIMEOUT` 5 s) |
| `events.rs` | `filter()` (`process`, `daemon_shutdown`, `app`), `emit_of(line)`, `run(paths, ctrl_tx, shutdown_tx, shutdown)`; lifecycle events `send().await`, `app` events `try_send` (dropped with a debug line when the channel is full); backoff after an EOF too, reset only after a `STABLE_BUS` (10 s) subscription, "bus closed" warned once per outage |
| `enroll.rs` | `HostFacts::detect`, `Enrollment {cluster, token, key, facts}`, pure `enroll_body(req) -> threepm_link::EnrollRequest` and `check_answer(answer, now_ms)` (the version picked must be one this build speaks, then the 5 min skew ⇒ `EnrollError::ClockSkew {delta_ms}`), `enroll(&reqwest::Client, &ServerUrl, &Enrollment) -> EnrollResponse` (one `POST` on `ServerUrl::enroll_url()`, the client being the agent's own — `http::client`, which refuses `--insecure` over TLS), `EnrollError {Transport, Timeout, Refused {code, message}, Status, LinkVersion, ClockSkew, Shape}`; `ENROLL_TIMEOUT` 10 s |
| `start.rs` | `spec(&AgentOptions, exe)` via `SpecPatch`, `register(daemon, spec)` = `Start {if_changed: false}` then `Save`, `affected_of` |
| `link_cmd.rs` | The `link` command; `refuse_existing_link`, pure `link_config(inputs, enrolled, app, now_ms, direct)` |
| `unlink_cmd.rs` | The `unlink` command; `delete_app` (NotFound tolerant), `save` |
| `status_cmd.rs` | Pure `view(cfg, app, process, session, now)` / `render` (with its `dashboard` line), `dashboard_url(server, cluster)`; the verdict is `threepm_link::local::status`; `Describe` answers `ProcessList` |
| `run.rs` | The `__run` role: options, state, identity, readiness, signals, tracing, the session and its three workers (`spawn_workers`: `publish_session` keeping `session.json`, the node-state task, the event forwarder), the fleet side from `run_fleet.rs` |
| `logs/`, `http_logs.rs` | Log shipping: sources, pipeline, sender, one row per file in "Logs" above |
| `e2e/` | The end-to-end channel of `fleet ssh`: `Noise_XX` handshake bound to call + node (`mod.rs`), records `[u16][Noise message]` and their `Sealer`/`Opener` (`record.rs`), compressed frames of a command on pipes, agreed in the handshake's payloads (`compress.rs`), legs as streams (`stream.rs`), `shell.key` / `cli.key` and their `SHA256:` fingerprints (`key.rs`) |
| `shell/forward.rs` | The target of `forward` (`-L`, `-W`, a `-R` takeover): connect (5 s) or take an `accepted` connection, `pump` both ways, then `pump_then_send` — the caller's leg held until the output leg's `PUT` ended, or the relay drops the answer's tail |
| `shell/listen.rs` | The target of `listen` (`-R`): `bind_addr` (loopback unless `gateway_ports`), the accept loop, the parked connections (`LISTEN_CLAIM_MS` 10 s, taken once by the same `from` and caller key), `bound` / `accepted` / `keepalive` on the legs, the end on `END`, a cut leg, `LISTEN_IDLE_MS` 60 s of silence or the agent's stop |
| `fleet/ssh/forward.rs` | The command line's `-L` (`serve`, `tunnel`), `-W` (`proxy`: stdin/stdout, stdout closed for ssh's EOF, exit 0 / 255) and `carry` (one `nodeForward`, `To::Dial` or `To::Accepted`, the output leg held until the input leg's `PUT` ended) |
| `fleet/ssh/reverse.rs` | The command line's `-R`: one sealed `nodeListen` per spec, keepalive every 20 s, each `accepted` taken over by a `nodeForward {accepted}` and `carry`, at most `TAKEOVERS_MAX` at once per `-R`, `END` on close |
| `shell/record.rs` | The node's recording of a sealed session: asciicast before sealing, bounded queue, upload before `exited {recording}`, stale files swept at start |
| `fleet/known_nodes.rs` | `~/.3pm/known_nodes` (TOFU): lookup, pin under a lock, the `/dev/tty` question, `--accept-new`, the changed-key refusal, `known-nodes ls` / `rm` |
| _(control channel)_ | `threepm_protocol::module::ctl` — `hello` then `ready`; the `shutdown` callback flips the agent's `watch` instead of exiting the process |

Registration sites (to duplicate for any new module): `justfile`
(`fmt`, `msrv`, `modules`, `e2e-modules`, `bundle`),
`.github/workflows/ci.yml` (`modules`, `msrv`), `release.yml`,
`distribution/nfpm.yml`, `docs/modules.md`, `FEATURES.md`,
`CHANGELOG.md`, `CLAUDE.md`. `install.sh` copies `modules/` as a whole.

## Tests

| Suite | Command | What it proves |
|---|---|---|
| unit | `cargo test --manifest-path modules/agent/Cargo.toml --lib --bins` | Per file; writer and reader under paused tokio time (ping at 20 s, flushes at 5/10/15/20 s, frames in order, a tracked frame confirmed and a stale one skipped, an oversized frame dropped and its sender told so, silence at exactly 60 s, a pong resets it; `state_resync` handed over, link 1's `resync` / `trees_request` / `host_resync` skipped); the handshake (a server one version behind refused before anything is signed, with `ERR_LINK_VERSION`); event mapping; `DaemonLink` schedule and bounded calls; the node-state task from the daemon's answers on (`node_state_tests.rs`, against the writer stand-in: every section whole in one frame first, then what moved, then a heartbeat, the mock `NodeReader` rebuilding what the daemon said; a frame not written commits nothing and is cut again at the same revisions; a resync, a new session and a replaced daemon owe full sections; what the counters say to ask — metrics every n ticks, shape only, silent; what is no longer there said `gone`, nothing while the daemon is down; a CRC once a minute per section, each at its own moment of it; frames under the cap, a section over it alone left out, warned once and asked again) and asking a scripted daemon (`node_state_ask_tests.rs`: whole states, no host agent, a daemon predating trees asked again only after the retry, a daemon down an empty tick, a replaced one owing everything); the paces (`period` never silent, the modes, ticks); `stream.rs` (what woke the loop, a closed writer channel is not a write) |
| session | `… --test session` (`tests/session/{main,shutdown,enrolling,reconnect,state,…}.rs`) | The session in process against the mock: handshake facts, reconnect ≤ 3 s after a cut, 4403 revoked ⇒ parked with no retry, 403 at the upgrade ⇒ probe on the stop schedule then relink, ack before reply, unknown frame skipped (`sessions == 1`, `retries == 0`), polite close 1000; shutdown ends a session stuck in connect / TCP-accepted-but-silent / held handshake within 1 s; enrollment refused on clock skew (nothing written) and accepted within tolerance; `attempt` back to 1 after a stable session; `error` instead of `challenge` ⇒ `Closed {4401, ProofInvalid}`; backend up late ⇒ retries grow then connected; the `welcome`'s `pace` in the status; the node's state (`state.rs`), the real task reading a scripted daemon (`tests/support/fake_daemon.rs`): every section whole with its CRC in the session's first frame, the mock's `NodeReader` rebuilding the table, the host, `app/1` and `tree/1` from the frames alone with never a gap; patches when the host and a registry move; heartbeats and no other frame of the agent's own when nothing does; one section whole again on `state_resync`, every one on a new session; an instance deleted said `gone`; a `pace` asking for less than a second still gets a frame a second |
| e2e | `THREEPM_BIN=target/debug/3pm cargo test --manifest-path modules/agent/Cargo.toml --test e2e --test e2e_trees` (or `just e2e-modules`; `tests/support/e2e.rs` stages the bundle for both) | A real 3pm dispatching to the staged module against the mock: status/unlink before any link, bad inputs refused without state, enroll ⇒ app online and in `dump.json`, state files and modes, `status`, refusals write nothing, pending + `--force`, unlink; the running app: hello with its signature, reconnect with the same boot id, Close 1000 on `stop` (no SIGKILL), new boot id on `restart`, parked when revoked; the node's state with a real sleeper: ≥ 3 `state` frames of one `gen` with a growing `ts`, every section whole with its CRC the first time, the host and the table rebuilt through `NodeReader` (the daemon's pid and protocol, the sleeper with its env key names), no env value / home path / script path in any frame, `emit {process}` on restart, `emit {daemon_shutdown}` then Close 1000 on `3pm kill`; the Node SDK under the real daemon (skipped without `node`): the instance's `app/<id>` section whole, then chained patches naming the counter's value and defining nothing again, the `deploy` event as `emit {app}`; a shell with two sleeps: a `tree/<id>` section with `sh` and its two `sleep`s, the same `id`/`gen` as the table's row, whose `proc_count` says 3 (the agent's own row says 1: no tree section is ever sent for it), held while the link goes on with heartbeats |

`tests/support/` is the mock backend: `mock_backend.rs` (its own thread
and runtime on `ws://127.0.0.1:0`, per-instance cluster, tokens
with uses and pre-approval, every agent frame recorded, `Order::{Cut,
Close(code, reason), Push(Link)}` to every live link, node state
switch, the `pace` it welcomes with (a tick a second and no trees by
default), `state_frames()` and `node()`
(the frames applied in order to a `NodeReader`, a gap failing the test), `hold_handshake`,
`refuse_handshake`, `enroll_skew_ms`, `link_range` (a backend of another link generation),
`unbound()` / `listen()` / `stop_listening()` (the same nodes across a
restart), `wait_for` / `wait_for_async`), `mock_serve.rs` (the accept
loop, and the `Authorization` ids looked up **before** the upgrade with
the same 401/403 answers), `mock_enroll.rs` (the `POST` handler, the
real backend's check order in miniature; its requests are kept apart
from the bucket ones — `enroll_requests()` / `enroll_seen()`),
`mock_link.rs` (challenge/hello/welcome, pings and agent close
codes counted).

## Notes and caveats

- Paused-time tests never involve the network: tokio auto-advances the
  clock while a task waits on I/O, so a paused test against the mock
  would skip its own timeouts. Session-level tests run in real time with
  short schedules (`stop_probe` is configurable for that).
- In a current-thread test, block only with `wait_for_async`: a
  `std::thread::sleep` starves the session task.
- The dispatcher fix that came with step 8: `3pm <module> … --json` is
  honored by `crates/threepm-cli/src/modules/dispatch.rs` (spawn notice
  to stderr, `THREEPM_JSON` set), so a first run's stdout stays one
  JSON document.
- `--insecure` over `wss://` is
  refused with a clear message until the TLS hardening step; `ws://` to
  loopback needs no flag.
- The `daemon_shutdown` path: the agent is itself a child of the daemon,
  so it forwards the event, closes the link and exits before the daemon
  reaps it.