diff --git a/.mise/config.toml b/.mise/config.toml
index 558ae9e0..1c1ede9d 100644
--- a/.mise/config.toml
+++ b/.mise/config.toml
@@ -1240,12 +1240,23 @@ shell = "{{ vars.task_shell }}"
 
 [tasks.cargo-clippy-check-pkg]
 alias = ["clippy-pkg"]
-description = "cargo clippy on a single workspace package (narrow inner-loop check)"
-# Same flags as cargo-clippy-check minus --workspace.
-# Usage-spec captures the package name and exposes it as $usage_package; a missing arg fails up front with
-# mise's auto-generated usage message instead of clippy's generic "--package <SPEC> not found" error.
-run = 'cargo clippy --keep-going --all-targets -p "$usage_package"'
-usage = 'arg "<package>" help="Workspace package name (e.g. edge-toolkit)"'
+description = "cargo clippy on one or more workspace packages (narrow inner-loop check)"
+# Same flags as cargo-clippy-check minus --workspace, plus --no-deps.
+# Only the named packages are linted, not the workspace crates they depend on. Those still build, but their lints
+# are cargo-clippy-check's to report.
+# Usage-spec captures the package names and exposes them space-joined in $usage_package, split here into one `-p`
+# each; a missing arg fails up front with mise's auto-generated usage message instead of clippy's generic
+# "--package <SPEC> not found" error.
+run = """
+set --
+# shellcheck disable=SC2086 # $usage_package is a space-joined package list, split by design.
+for package in $usage_package; do
+  set -- "$@" -p "$package"
+done
+cargo clippy --keep-going --all-targets --no-deps "$@"
+"""
+shell = "{{ vars.task_shell }}"
+usage = 'arg "<package>..." var=#true help="Workspace package names (e.g. edge-toolkit et-ws-service)"'
 
 [tasks.cargo-doc-check]
 description = "Type-check rustdoc with -D warnings to fail on any doc issues"
@@ -2569,20 +2580,20 @@ description = "Regenerate HELP.md for every binary crate using clap or reading s
 run = "cargo run -q -p et-int-gen --bin et-int-gen -- help-md"
 
 [tasks."gen:checks"]
-description = "Regenerate CHECKS.md, the catalogue of every check across every MISE_ENV"
+description = "Regenerate CHECKS.md, the catalogue of every check across every MISE_ENV, stamped with today's date"
 run = "cargo run -q -p et-int-gen --bin et-int-gen -- checks"
 
+# Holds CHECKS.md to its generation date rather than re-rendering it, so it costs no build on any lane.
+# `--parser ignore` because Markdown has no parser; it hands the file over as its lines, which is all this reads.
 [tasks.checks-md-check]
+description = "Fail if CHECKS.md was last generated more than 7 days ago (run `mise run gen:checks`) (conftest)"
+run = "conftest test --combine --parser ignore --namespace checks_md -p config/conftest/policy CHECKS.md"
+
+[tasks.gen-help-check]
 # Ordered after the workspace builds so its `cargo run` reuses their artifacts.
 # Run beside them it is a fourth concurrent build, which doubled every cargo task's time on the three-core macOS
 # runner and took the macOS check lane past its budget.
 depends = ["cargo-check", "cargo-clippy-check", "cargo-doc-check", "regenerate-verification"]
-description = "Fail if CHECKS.md is stale vs the mise task tables and rule files (run `mise run gen:checks`)"
-run = "cargo run -q -p et-int-gen --bin et-int-gen -- checks --check"
-
-[tasks.gen-help-check]
-# After checks-md-check, whose et-int-gen build it reuses.
-depends = ["checks-md-check"]
 description = "Fail if any checked-in HELP.md is stale vs its clap tree or env config (run `mise run gen-help-all`)"
 run = "cargo run -q -p et-int-gen --bin et-int-gen -- help-md --check"
 
diff --git a/AGENTS.md b/AGENTS.md
index 8207e5a1..76cb06b3 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -252,6 +252,17 @@ confirm you see the identical failure -> fix -> confirm the fix turns that same
 push. If you genuinely cannot reproduce locally (e.g. a lane you have no access to), say so explicitly and do not
 pass off an unverified change as a fix.
 
+That loop is for a failure whose cause the output does not settle. When you have been given the failing check's
+output and it already names the problem -- a lint at a file and line, a compile error, a failed assertion -- fix it
+first, and only then run the check, once, to confirm the fix. Running the check beforehand just to watch it fail
+again buys nothing: the checks are expensive and slow, and that run reproduces what the output already said.
+
+Once a lint fix to code is in, decide whether it could have changed what the tests see before running any. Renaming
+a constructor call, adding a justified `#[expect]` or reflowing a comment cannot fail a test; changing a signature,
+a branch or a lookup can. Run the tests only when the answer is yes, and then as a step of their own rather than
+chained onto the lint and check commands, so a slow test run never holds up the check results and a failure is
+never lost in their output.
+
 ## Keep lines <= 120 characters
 
 Assume every file you touch is bound by the 120-char limit, and keep each line under it on the first draft rather
@@ -419,11 +430,12 @@ Formatter task(s) per file type:
 env) breaks ties differently from `dprint-fmt` -- it collapses a short array onto one line where dprint leaves it
 expanded -- so a file dprint accepts can still fail `oxfmt-check` in CI.
 
-For Rust inner-loop iteration on a single crate, use `mise run cargo-clippy-check-pkg <package>` (alias
-`clippy-pkg`) instead of `cargo-clippy-check` -- it runs `cargo clippy --keep-going --tests -p <package>` so you
-only compile that crate plus its deps, not the whole workspace. Same lint config; just narrower scope. Switch back
-to `cargo-clippy-check` for the final verification pass before declaring done, since cross-crate
-`feature-unification` differences can fire workspace-wide lints that don't fire on a single `-p` build.
+For Rust inner-loop iteration, use `mise run cargo-clippy-check-pkg <package>...` (alias `clippy-pkg`) instead of
+`cargo-clippy-check`, naming the workspace package of every crate you changed in one call -- it compiles only those
+packages plus their deps, not the whole workspace, and lints only the named packages, not the workspace packages
+they depend on. Same lint config; just narrower scope. Switch back to `cargo-clippy-check` for the final
+verification pass before declaring done: it is the pass that lints a package you edited but did not name, and
+cross-crate `feature-unification` differences can fire workspace-wide lints that don't fire on a narrower `-p` build.
 
 ## Architecture
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2114330d..f6453e51 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -43,6 +43,8 @@ Changelog 1.1.0.
   storage, so an upload through one hub reads back through any other. `services.backplane` and `services.storage`
   take `limits:` for the two. `secrets.env` carries the store's secret as `AWS_SECRET_ACCESS_KEY` and
   `RUSTFS_SECRET_KEY`. A scenario without `services.hub.machines` renders as before.
+- `WS_MESSAGE_RETENTION=all` makes the hub keep every unacknowledged direct message a recipient has from a sender,
+  rather than only the newest. The default, `latest-per-sender`, behaves as before.
 
 ### Changed
 
@@ -59,6 +61,12 @@ Changelog 1.1.0.
 - **Breaking:** `et-ws-server` takes no command-line arguments. Its one flag, `-a` / `--agent-registry`, is now the
   `WS_AGENT_REGISTRY` environment variable, still defaulting to `registry.yaml`, and `et_ws_server::cli` is gone.
   Generated k3s manifests set the variable instead of overriding the container's `command` and `args`.
+- **Breaking:** `PUT /storage/{agent_id}/{filename}` accepts a write only while the agent is connected, and only
+  from the address it is connected from; anything else is answered `403`. An upload made on an agent's behalf must
+  now come from the agent's own host while it is connected.
+- **Breaking:** `AgentRegistry::queue_direct` and `NatsBackplane::queue` take a `MessageRetention`.
+  `Backplane::knows` is replaced by `Backplane::presence`, and `Presence::Connected` carries the agent's
+  `last_known_ip`.
 
 ## [0.3.0] - 2026-10-15
 
diff --git a/CHECKS.md b/CHECKS.md
index fa33cc40..c3321e57 100644
--- a/CHECKS.md
+++ b/CHECKS.md
@@ -1,16 +1,18 @@
 # Checks
 
+- **Generated:** 2026-10-27
 - **Checks:** 75
 - **Tools:** 47
 - **Shell-script checks:** 10
 - **Checks with a repair pass:** 21
-- **Rules:** 343 custom local rules, or non-default strict linter settings
+- **Rules:** 349 custom local rules, or non-default strict linter settings
 
 Every check is listed across every `MISE_ENV`, filed under the tool it runs. Each entry names the env whose config
 declares it, the aggregate that runs it as part of `mise run check`, the `fix` or `fmt` pass that repairs what it
 reports, where one does, and the config files it reads. `mise run fix` runs every such pass. Generated by `mise run
 gen:checks` from the task tables in `.mise/config*.toml`, the rule files under `config/` and the `[workspace.lints]` in
-`Cargo.toml`; `mise run checks-md-check` fails when this file drifts from them.
+`Cargo.toml`. It is not held to them on every change: `mise run checks-md-check` fails only once the generation date
+above is more than 7 days old.
 
 ## `action-validator`
 
@@ -235,6 +237,12 @@ Rules, from `config/ast-grep/rules`:
 - `no-string-new` -- `String::new()` is banned (`DeepSource RS-W1079`): construct empty strings with `String::default()`
   instead, so the zero value routes through the `Default` impl like every other default-constructed value in the
   codebase.
+- `no-test-in-python-docstrings` -- Docstring in non-test code mentions tests. Document what the code does for every
+  caller; why a test needs it belongs in the test that relies on it. A name containing it (`test`, `ws-test-server`,
+  `ET_TEST_X`) is allowed.
+- `no-test-in-rust-docs` -- Doc comment in non-test code mentions tests. Document what the item does for every caller;
+  why a test needs it belongs in the test that relies on it. A name containing it (`test`, `ws-test-server`,
+  `ET_TEST_X`) is allowed.
 - `no-uds-listener` -- HTTP servers here listen on TCP only. A Unix-domain-socket listener gives requests no peer
   address, and the handlers take the client address from the socket peer rather than from client-supplied `Forwarded` /
   `X-Forwarded-For` headers -- so `ws_handler` refuses every request arriving on one. Bind a TCP address instead.
@@ -307,8 +315,8 @@ Rules, from `config/ast-grep/rules`:
 
 - `cargo-clippy-check` (`default` env; run by `check:rust`). Repaired by `cargo-clippy-fix`. Config: `Cargo.toml`,
   `config/clippy.toml`.
-- `cargo-clippy-check-pkg` (`default` env; not run by `mise run check`) -- cargo clippy on a single workspace package
-  (narrow inner-loop check). Config: `Cargo.toml`, `config/clippy.toml`.
+- `cargo-clippy-check-pkg` (`default` env; not run by `mise run check`) -- cargo clippy on one or more workspace
+  packages (narrow inner-loop check). Config: `Cargo.toml`, `config/clippy.toml`.
 
 Rules, from `Cargo.toml`:
 
@@ -505,6 +513,8 @@ Rules, from `Cargo.toml`:
 
 ## `conftest`
 
+- `checks-md-check` (`default` env; run by `check:rust`) -- Fail if CHECKS.md was last generated more than 7 days ago
+  (run `mise run gen:checks`) (conftest). Config: `config/conftest/policy`.
 - `conftest-check-checks-badge` (`default` env; run by `conftest-check`) -- Check the README checks badge keeps step
   with CHECKS.md's rules count (conftest). Config: `config/conftest/policy`.
 - `conftest-check-dockerfile` (`default` env; run by `conftest-check`) -- Cross-check Dockerfiles against mise [tools]
@@ -566,6 +576,11 @@ Rules, from `config/conftest/policy`:
   - CHECKS.md must record its rules count, which the badge is held to.
   - README.md must define the checks badge.
   - The checks badge records the rules count rounded down to a multiple of 20, so it moves with every 20 rules added.
+- `checks_md` (`config/conftest/policy/checks_md/checks_md.rego`):
+  - CHECKS.md must say when it was generated, which only the generator writes.
+  - A document stamped more than once was not written by the generator, which stamps it exactly once.
+  - The document is regenerated at least once every `max_age_days` days.
+  - The generator stamps today's UTC date, so a later one was written by hand.
 - `checksums` (`config/conftest/policy/checksums/checksums.rego`):
   - Every `_asset` var in the mise configs must have an [asset] table in config/upstream-cache/data.toml.
   - Every [asset] table in config/upstream-cache/data.toml must be referenced by a mise `_asset` var.
@@ -686,8 +701,6 @@ Rules, from `config/conftest/policy`:
 
 ## `et-int-gen`
 
-- `checks-md-check` (`default` env; run by `check:rust`) -- Fail if CHECKS.md is stale vs the mise task tables and rule
-  files (run `mise run gen:checks`).
 - `gen-help-all` (`default` env; run by `check:rust`) -- Regenerate HELP.md for every binary crate using clap or reading
   serde-env config.
 - `gen-help-check` (`default` env; run by `check:rust`) -- Fail if any checked-in HELP.md is stale vs its clap tree or
diff --git a/Cargo.lock b/Cargo.lock
index 8db4b08a..c6914e14 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -455,7 +455,7 @@ version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -466,7 +466,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
 dependencies = [
  "anstyle",
  "once_cell_polyfill",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -1281,7 +1281,7 @@ dependencies = [
  "maybe-owned",
  "rustix 1.1.5",
  "rustix-linux-procfs",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
  "winx",
 ]
 
@@ -2155,9 +2155,9 @@ dependencies = [
 
 [[package]]
 name = "crypto-bigint"
-version = "0.7.8"
+version = "0.7.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d726facc8b27e604391b232b970cdc19b5468c69f08b466b9734c759216c7415"
+checksum = "9c035cb64cba03dad2fb921f3e218785e8cf9a41790da5eb3a1400f62fa2be11"
 dependencies = [
  "cpubits",
  "ctutils",
@@ -2225,9 +2225,9 @@ dependencies = [
 
 [[package]]
 name = "ctutils"
-version = "0.4.5"
+version = "0.4.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "38d28fb6aa7dd680b3ffd1a3e64e115c177bdba63c0ae7f9cd48129ff05c4fa0"
+checksum = "fe3e9d21d39e6f39ff46e03d5b621424f88ee23d8a1672c6bfc9ccc7f2da4f6a"
 dependencies = [
  "cmov",
  "subtle",
@@ -4338,7 +4338,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d65aa39b3a5c1c9c1b745c9a019234bb7a21b77abcb4f4d266d706e2d577d65"
 dependencies = [
  "base16ct 1.0.0",
- "crypto-bigint 0.7.8",
+ "crypto-bigint 0.7.10",
  "crypto-common 0.2.2",
  "ff 0.14.0",
  "group 0.14.0",
@@ -4402,7 +4402,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4507,6 +4507,7 @@ dependencies = [
  "et-ws-server",
  "fs-err",
  "heck",
+ "jiff",
  "kdl 6.7.1",
  "openapiv3",
  "pretty_yaml",
@@ -7870,7 +7871,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -9177,7 +9178,7 @@ dependencies = [
  "quinn-udp",
  "rustc-hash 2.1.3",
  "rustls",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "thiserror 2.0.21",
  "tokio",
  "tracing",
@@ -9217,9 +9218,9 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.6.5",
+ "socket2 0.5.10",
  "tracing",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -9837,7 +9838,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.12.1",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -9927,7 +9928,7 @@ dependencies = [
  "security-framework 3.7.0",
  "security-framework-sys",
  "webpki-root-certs 1.0.10",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -10706,7 +10707,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -10808,7 +10809,7 @@ dependencies = [
  "cfg-if",
  "libc",
  "psm",
- "windows-sys 0.61.2",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]
@@ -11462,7 +11463,7 @@ dependencies = [
  "getrandom 0.3.4",
  "once_cell",
  "rustix 1.1.5",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -11856,9 +11857,9 @@ checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
 
 [[package]]
 name = "tonic"
-version = "0.14.6"
+version = "0.14.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
+checksum = "16ea7e687c08008d22518056608ad0c1cc721204d30ed48c8439a88ae5b9fc73"
 dependencies = [
  "async-trait",
  "axum",
@@ -11878,11 +11879,12 @@ dependencies = [
  "sync_wrapper",
  "tokio",
  "tokio-stream",
+ "tokio-util",
  "tower",
  "tower-layer",
  "tower-service",
  "tracing",
- "zstd 0.13.2",
+ "zstd 0.14.3",
 ]
 
 [[package]]
@@ -13425,7 +13427,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
diff --git a/config/ast-grep/rules/no-test-in-python-docstrings.yaml b/config/ast-grep/rules/no-test-in-python-docstrings.yaml
new file mode 100644
index 00000000..0efe95bf
--- /dev/null
+++ b/config/ast-grep/rules/no-test-in-python-docstrings.yaml
@@ -0,0 +1,21 @@
+id: no-test-in-python-docstrings
+language: Python
+severity: error
+message: |
+  Docstring in non-test code mentions tests. Document what the code does for every caller; why a test needs it
+  belongs in the test that relies on it. A name containing it (`test`, `ws-test-server`, `ET_TEST_X`) is allowed.
+rule:
+  kind: string
+  regex: "(?i)(^|[^`\\w-])test(s|ed|ing|able)?([^`\\w-]|$)"
+  inside:
+    kind: expression_statement
+    nthChild: 1
+    inside:
+      any:
+        - kind: module
+        - kind: block
+ignores:
+  - generated/**
+  - "**/tests/**"
+  - "**/*_tests/**"
+  - "**/test_*.py"
diff --git a/config/ast-grep/rules/no-test-in-rust-docs.yaml b/config/ast-grep/rules/no-test-in-rust-docs.yaml
new file mode 100644
index 00000000..4793025c
--- /dev/null
+++ b/config/ast-grep/rules/no-test-in-rust-docs.yaml
@@ -0,0 +1,18 @@
+id: no-test-in-rust-docs
+language: Rust
+severity: error
+message: |
+  Doc comment in non-test code mentions tests. Document what the item does for every caller; why a test needs it
+  belongs in the test that relies on it. A name containing it (`test`, `ws-test-server`, `ET_TEST_X`) is allowed.
+rule:
+  all:
+    - kind: line_comment
+      regex: "^//[/!]"
+    - kind: line_comment
+      regex: "(?i)(^|[^`\\w-])test(s|ed|ing|able)?([^`\\w-]|$)"
+ignores:
+  - generated/**
+  - "**/tests/**"
+  - "**/test_*.rs"
+  - libs/test-*/**
+  - services/ws-test-server/**
diff --git a/config/ast-grep/tests/no-test-in-python-docstrings.yaml b/config/ast-grep/tests/no-test-in-python-docstrings.yaml
new file mode 100644
index 00000000..60fe4bf5
--- /dev/null
+++ b/config/ast-grep/tests/no-test-in-python-docstrings.yaml
@@ -0,0 +1,23 @@
+id: no-test-in-python-docstrings
+valid:
+  - |
+    """Organisation helpers, at their latest attested versions."""
+  - |
+    def run():
+        """Run the healthcheck's `test` command against ws-test-server."""
+  - |
+    def run():
+        x = 1
+        "a test string that is not a docstring"
+invalid:
+  - |
+    """Helpers shared with the tests."""
+  - |
+    def run():
+        """Return the sum, as the test harness expects."""
+  - |
+    class Model:
+        """Kept small so it stays testable."""
+  - |
+    def run():
+        """Delegate to client.test() and return its exit status."""
diff --git a/config/ast-grep/tests/no-test-in-rust-docs.yaml b/config/ast-grep/tests/no-test-in-rust-docs.yaml
new file mode 100644
index 00000000..f3f7436e
--- /dev/null
+++ b/config/ast-grep/tests/no-test-in-rust-docs.yaml
@@ -0,0 +1,32 @@
+id: no-test-in-rust-docs
+valid:
+  - |
+    /// Wrap a populated agent map.
+    fn from_agents() {}
+  - |
+    /// The latest attested contest result.
+    fn latest() {}
+  - |
+    /// A healthcheck running `test`, on the shared cadence.
+    fn healthcheck() {}
+  - |
+    //! Embeds the input from ws-test-server's data, read under ET_TEST_COVERAGE.
+  - |
+    fn run() {
+        // A plain comment may say why a test needs this.
+    }
+invalid:
+  - |
+    /// Wrap a populated agent map; useful for tests building fixtures.
+    fn from_agents() {}
+  - |
+    //! Stores the model where the test harness reads it.
+  - |
+    /// Independently testable with an in-memory backend.
+    trait Backend {}
+  - |
+    /// Public so a Test can drive the reader pump.
+    fn pump() {}
+  - |
+    /// Delegate to Client::test() and return its exit status.
+    fn run() {}
diff --git a/config/conftest/policy/checks_md/checks_md.rego b/config/conftest/policy/checks_md/checks_md.rego
new file mode 100644
index 00000000..94ca91ed
--- /dev/null
+++ b/config/conftest/policy/checks_md/checks_md.rego
@@ -0,0 +1,59 @@
+# CHECKS.md's generation date, held to at most a week old instead of re-rendering the document on every change.
+# Run with `--namespace checks_md` over CHECKS.md, `--combine`d and parsed with `ignore`, which hands the file over as
+# its lines; the `Original` field of each is the line's text.
+package checks_md
+
+# How many whole days may pass since generation before the document has to be regenerated.
+max_age_days := 7
+
+ns_per_day := ((24 * 60) * 60) * 1000000000
+
+# The whole header line the generator writes, capturing its date.
+stamp_pattern := `^- \*\*Generated:\*\* ([0-9]{4}-[0-9]{2}-[0-9]{2})$`
+
+# The date on each `- **Generated:** YYYY-MM-DD` line of CHECKS.md, one entry per line, so repeats are counted.
+stamps := [match[1] |
+	some file in input
+	endswith(file.path, "CHECKS.md")
+	some line in file.contents
+	some match in regex.find_all_string_submatch_n(stamp_pattern, line.Original, 1)
+]
+
+# Midnight UTC on `date`, in nanoseconds since the epoch; undefined when `date` is not a calendar date.
+midnight_ns(date) := time.parse_ns("2006-01-02", date)
+
+# Whole days from midnight UTC on `date` to now; meaningful only for a date not in the future.
+age_days(date) := floor((time.now_ns() - midnight_ns(date)) / ns_per_day)
+
+# CHECKS.md must say when it was generated, which only the generator writes.
+deny contains msg if {
+	count(stamps) == 0
+	msg := "CHECKS.md: no `- **Generated:** YYYY-MM-DD` line; regenerate it"
+}
+
+# A document stamped more than once was not written by the generator, which stamps it exactly once.
+deny contains msg if {
+	count(stamps) > 1
+	msg := $"CHECKS.md: {count(stamps)} `- **Generated:**` lines; regenerate it"
+}
+
+# A stamp the shape allows but the calendar does not, such as 2026-02-30, was written by hand and has no age.
+deny contains msg if {
+	some date in stamps
+	not midnight_ns(date)
+	msg := $"CHECKS.md: generated {date}, which is not a calendar date; regenerate it"
+}
+
+# The document is regenerated at least once every `max_age_days` days.
+deny contains msg if {
+	some date in stamps
+	age_days(date) > max_age_days
+	msg := $"CHECKS.md: generated {date}, more than {max_age_days} days ago; regenerate it"
+}
+
+# The generator stamps today's UTC date, so a later one was written by hand.
+deny contains msg if {
+	some date in stamps
+	midnight_ns(date) > time.now_ns()
+	msg := $"CHECKS.md: generated {date}, which is in the future; regenerate it"
+}
diff --git a/config/conftest/policy/checks_md/checks_md_test.rego b/config/conftest/policy/checks_md/checks_md_test.rego
new file mode 100644
index 00000000..bf95ac6c
--- /dev/null
+++ b/config/conftest/policy/checks_md/checks_md_test.rego
@@ -0,0 +1,61 @@
+# Unit tests for the CHECKS.md freshness policy, run by `conftest verify`.
+#
+# The real task only ever sees today's document, so the clock is pinned here to reach every boundary.
+package checks_md_test
+
+import data.checks_md
+
+lines(texts) := [{"Kind": "Path", "Original": text, "Value": text} | some text in texts]
+
+document(header) := [{"path": "CHECKS.md", "contents": lines(array.concat(["# Checks", ""], header))}]
+
+stamp(date) := $"- **Generated:** {date}"
+
+# Noon UTC on 2026-10-27, so a whole-day boundary is never within an hour of the clock.
+now := time.parse_rfc3339_ns("2026-10-27T12:00:00Z")
+
+reports(msgs, fragment) if {
+	some msg in msgs
+	contains(msg, fragment)
+}
+
+test_a_document_generated_today_is_accepted if {
+	count(checks_md.deny) == 0 with input as document([stamp("2026-10-27")]) with time.now_ns as now
+}
+
+test_a_document_exactly_seven_days_old_is_accepted if {
+	count(checks_md.deny) == 0 with input as document([stamp("2026-10-20")]) with time.now_ns as now
+}
+
+test_a_document_eight_days_old_is_rejected if {
+	reports(checks_md.deny, "more than 7 days ago") with input as document([stamp("2026-10-19")])
+		with time.now_ns as now
+}
+
+test_a_document_dated_tomorrow_is_rejected if {
+	reports(checks_md.deny, "in the future") with input as document([stamp("2026-10-28")]) with time.now_ns as now
+}
+
+test_a_document_without_a_stamp_is_rejected if {
+	reports(checks_md.deny, "no `- **Generated:**") with input as document(["- **Checks:** 75"]) with time.now_ns as now
+}
+
+test_a_stamp_with_trailing_text_is_not_a_stamp if {
+	reports(checks_md.deny, "no `- **Generated:**") with input as document([$"{stamp("2026-10-27")} by hand"])
+		with time.now_ns as now
+}
+
+test_two_different_stamps_are_rejected if {
+	header := [stamp("2026-10-27"), stamp("2026-10-26")]
+	reports(checks_md.deny, "2 `- **Generated:**` lines") with input as document(header) with time.now_ns as now
+}
+
+test_two_identical_stamps_are_rejected if {
+	header := [stamp("2026-10-27"), stamp("2026-10-27")]
+	reports(checks_md.deny, "2 `- **Generated:**` lines") with input as document(header) with time.now_ns as now
+}
+
+test_a_stamp_that_is_not_a_calendar_date_is_rejected if {
+	reports(checks_md.deny, "not a calendar date") with input as document([stamp("2026-02-30")])
+		with time.now_ns as now
+}
diff --git a/config/deny.toml b/config/deny.toml
index 4fc21cf7..5e13ce44 100644
--- a/config/deny.toml
+++ b/config/deny.toml
@@ -139,6 +139,61 @@ ignore = [
   # (serde-saphyr is the advisory's suggestion) is in our hands, but progenitor-macro would still carry it, so
   # drop this only once both have moved.
   { id = "RUSTSEC-2024-0449", reason = "expires 2026-10-31" },
+  # zstd 0.13.2 / zstd-safe 7.3.0 -- the zstd entries below share one blocker, so it is described once, here.
+  # The fixes are only in zstd 0.14.3 and zstd-safe 8.2.0, and Cargo.lock pins those for every consumer that
+  # accepts them (actix-http, compression-codecs, tonic). Two cannot: deno_node pins `zstd = "=0.13.2"` by equality
+  # through its newest release, 0.198.0, and wasmtime-internal-cache requires ^0.13 through 50.0.0. cargo-deny
+  # ignores by id only, so these also mute the patched 0.14 copy -- re-check it when touching them. All of them go
+  # when both move to zstd 0.14. Our resolution enables zstd-safe's arrays,legacy,std,zdict_builder features only
+  # (`cargo tree -i zstd-safe@7.3.0 -e features`); `experimental`, `seekable` and `zstdmt` are all off.
+  #
+  # Leftover input from an abandoned stream leaks into the next one on a reused context, fixed in zstd 0.14.3.
+  # Reached only through the `with_context` stream constructors. deno_node's node:zlib ops build their raw
+  # encoders and decoders with `new`, and wasmtime-internal-cache calls only `encode_all` / `decode_all`.
+  { id = "RUSTSEC-2026-0391", reason = "expires 2026-10-31" },
+  # Three memory-safety flaws in `Seekable`, all fixed in 8.2.0 and all in the `seekable` feature, which is off.
+  # `decompress` sets a `Vec` length past its capacity for an offset past the end, an uninitialized or failed
+  # `Seekable` reads a NULL seek table, and `seekable_compressStream` reads past an `InBuffer` whose `pos` is out
+  # of range.
+  { id = "RUSTSEC-2026-0396", reason = "expires 2026-10-31" },
+  { id = "RUSTSEC-2026-0398", reason = "expires 2026-10-31" },
+  { id = "RUSTSEC-2026-0405", reason = "expires 2026-10-31" },
+  # Three memory-safety flaws in `experimental` APIs, all fixed in 8.2.0; that feature is off, so none is compiled in.
+  # `CParameter::StableInBuffer` compresses freed input, `init_using_cdict` / `init_using_ddict` outlive their
+  # dictionary, and `DCtx::try_clone` double-frees one.
+  { id = "RUSTSEC-2026-0395", reason = "expires 2026-10-31" },
+  { id = "RUSTSEC-2026-0399", reason = "expires 2026-10-31" },
+  { id = "RUSTSEC-2026-0404", reason = "expires 2026-10-31" },
+  # Two memory-safety flaws in multithreaded compression, both fixed in 8.2.0 and both needing `zstdmt` (off).
+  # `CCtx::try_clone` overflows the heap after a multithreaded frame, and an unfinished one lets a reset free a
+  # dictionary or job table a worker still uses. deno_node does pass guest JS's `NbWorkers` through, but without
+  # `zstdmt` the bundled library has no worker threads, so no multithreaded frame is ever started.
+  { id = "RUSTSEC-2026-0402", reason = "expires 2026-10-31" },
+  { id = "RUSTSEC-2026-0406", reason = "expires 2026-10-31" },
+  # `OutBuffer::around_pos` marks a `Vec` initialized up to a position past its length, fixed in 8.2.0.
+  # deno_node wraps plain slices with `OutBuffer::around`, and wasmtime-internal-cache calls only
+  # `encode_all` / `decode_all`, so neither consumer ever builds an `OutBuffer` over a `Vec` with a position.
+  { id = "RUSTSEC-2026-0397", reason = "expires 2026-10-31" },
+  # A one-shot call on a context with an unfinished stream leaves it pointing at a freed dictionary, fixed in 8.2.0.
+  # Needs one context used for both streaming and one-shot calls. deno_node's node:zlib ops only stream, on a
+  # context per JS stream, and wasmtime-internal-cache only makes one-shot calls, each on a context of its own.
+  { id = "RUSTSEC-2026-0400", reason = "expires 2026-10-31" },
+  # `train_from_buffer` reads past its samples buffer when the sample sizes overflow, fixed in 8.2.0.
+  # `zdict_builder` is on, but neither consumer trains a dictionary, so the trainer is compiled in and never called.
+  { id = "RUSTSEC-2026-0401", reason = "expires 2026-10-31" },
+  # A failed streaming decompression of a legacy (v0.4 to v0.7) frame exposes uninitialized memory, fixed in 8.2.0.
+  # Live: `legacy` is on, and deno_node's raw decoder runs `decompress_stream`. It is reached only when guest JS in
+  # the web runner decompresses zstd through node:zlib, which our modules never do. wasmtime-internal-cache
+  # decompresses only its own compilation cache, and ws-wasi-runner configures none, so that path never runs.
+  { id = "RUSTSEC-2026-0403", reason = "expires 2026-10-31" },
+  # block-buffer 0.10.4: a panic caught inside the compress callback leaves the buffer's cursor corrupted.
+  # Fixed only in 0.12.1, and the 0.10 line is pulled by digest 0.10, which the RustCrypto hashes under deno_crypto,
+  # deno_node_crypto and curve25519-dalek require through their newest releases, so no bump reaches it; Cargo.lock
+  # already pins 0.12.1 for the consumers that accept it, and since cargo-deny ignores by id only this also mutes
+  # that copy. Exploiting it needs the callback to panic, the panic to be caught, and the same buffer to be used
+  # again. The callbacks are the hashes' fixed-size block functions, which do not panic, and nothing here catches
+  # a panic mid-hash and keeps the hasher. Drop this when digest 0.10 leaves the tree.
+  { id = "GHSA-qwgh-2vcv-g2f7", reason = "expires 2026-10-31" },
 ]
 
 [licenses]
diff --git a/config/generated-trees.toml b/config/generated-trees.toml
index 894cbe5f..ab2a5565 100644
--- a/config/generated-trees.toml
+++ b/config/generated-trees.toml
@@ -60,6 +60,8 @@ required_in = [
   "config/ast-grep/rules/no-shadow-result.yaml",
   "config/ast-grep/rules/no-string-literal-line-continuation.yaml",
   "config/ast-grep/rules/no-string-new.yaml",
+  "config/ast-grep/rules/no-test-in-python-docstrings.yaml",
+  "config/ast-grep/rules/no-test-in-rust-docs.yaml",
   "config/ast-grep/rules/prefer-self-mod-use.yaml",
   "config/ast-grep/rules/python-process-control-except.yaml",
   "config/ast-grep/rules/use-mod-order.yaml",
diff --git a/config/jscpd-baseline.json b/config/jscpd-baseline.json
index 16e9b83c..a80c35a5 100644
--- a/config/jscpd-baseline.json
+++ b/config/jscpd-baseline.json
@@ -18,7 +18,6 @@
     "0875c25e5ebb12cc": 1,
     "0c04b2e97c016262": 1,
     "0c8f78b8c0d0f6b4": 1,
-    "0de69d245e3a7ff8": 1,
     "0ea84f1f5db88fe7": 1,
     "0f033dd72cb08a76": 1,
     "0f10e9243723cb05": 3,
@@ -66,7 +65,6 @@
     "31a1985df946271b": 1,
     "31a98e49914dbdda": 1,
     "32268a277147a0a2": 1,
-    "331841e078587935": 1,
     "3355d5d4a61ff011": 1,
     "3369118fac56660d": 1,
     "34b2a7c1a1a932c1": 1,
@@ -86,7 +84,6 @@
     "3e5aadc3e77078b6": 1,
     "3e6e51f257b8453d": 1,
     "3ef0ad9e8ac464b8": 1,
-    "406b435c57117487": 1,
     "4268c1cf96611f46": 1,
     "43e742aef05399f9": 1,
     "43f0e3fca21daafa": 1,
@@ -127,7 +124,6 @@
     "608f556637a1f4db": 1,
     "61b19feee0aa5bb0": 1,
     "61b28209f41edecf": 1,
-    "62257586054b0689": 1,
     "62dd1cdba0d53531": 2,
     "62f4c358b07348db": 1,
     "635fb32c899e4df6": 1,
@@ -142,7 +138,6 @@
     "6ab191050bb8943d": 1,
     "6b7ef9701aa8a94b": 1,
     "6bad88d8a503b406": 1,
-    "6bafbd78440b4958": 1,
     "6d80cc58c007a238": 1,
     "6e7055e0693626a4": 1,
     "6f84b76a00168521": 1,
@@ -195,7 +190,6 @@
     "960909be0a867187": 1,
     "96825238ca07ac74": 1,
     "96fdb2ccaf1ea7d5": 1,
-    "983245de70d60f9f": 1,
     "985af6ffa52d3359": 1,
     "9868e80d502c44d3": 1,
     "98ace481ce842e9b": 1,
@@ -244,7 +238,6 @@
     "ae590b605758bec9": 1,
     "afc243bb6427e6d9": 1,
     "b07cc7ec51bc1602": 1,
-    "b0d913f1fde4e8e3": 1,
     "b29077918336343b": 1,
     "b32ec6344a29f094": 1,
     "b3b79293ad6e0885": 1,
@@ -257,11 +250,9 @@
     "ba18dcc601d42b5e": 1,
     "ba8145a95d4bf884": 1,
     "bab02c8fb19dca60": 1,
-    "bbce73eb21524ef5": 1,
     "bbd85a05647fdf5e": 1,
     "bbda58a7774990a3": 1,
     "bd684492ef8fdd33": 1,
-    "be1b0a5fd14b77a4": 1,
     "c01708e08c252d4a": 1,
     "c10acaf9d504846e": 1,
     "c10d9673e6b919c8": 1,
@@ -303,6 +294,7 @@
     "d72424c1bba450b5": 1,
     "d7954652649a6f2c": 1,
     "d7ad53a1ef1dc59c": 1,
+    "d8915fed4856b3df": 1,
     "d8a0ad2f6e337e91": 1,
     "da7b27ae8c89cb73": 1,
     "da812f4e54fefbf1": 1,
@@ -327,7 +319,6 @@
     "ef6dd6376769473d": 1,
     "f1f99624666ed51c": 1,
     "f20e3a3f28708194": 1,
-    "f2167d54ff8b56ea": 1,
     "f2a086bcedd018a8": 1,
     "f2e5c8aacb2d0d6e": 1,
     "f2f40a2a067eb8ed": 1,
@@ -345,7 +336,6 @@
     "fb78fa42d28ddc83": 1,
     "fb8747afe01f9463": 1,
     "fb8c4a24a3cbfb47": 1,
-    "fcad2abbc735eba6": 1,
     "ff0bc219f7f029b9": 1,
     "ffa1257e338b1953": 1
   }
diff --git a/config/osv-scanner.toml b/config/osv-scanner.toml
index ae29aae4..48d55f8e 100644
--- a/config/osv-scanner.toml
+++ b/config/osv-scanner.toml
@@ -3,6 +3,7 @@
 IgnoredVulns = [
   { id = "GHSA-5j98-2g5x-46v6", ignoreUntil = 2026-10-31 },
   { id = "GHSA-6f2x-v7q7-m7m5", ignoreUntil = 2026-10-31 },
+  { id = "GHSA-qwgh-2vcv-g2f7", ignoreUntil = 2026-10-31 },
   { id = "RUSTSEC-2023-0071", ignoreUntil = 2026-10-31 },
   { id = "RUSTSEC-2024-0436", ignoreUntil = 2026-10-31 },
   { id = "RUSTSEC-2024-0449", ignoreUntil = 2026-10-31 },
@@ -20,4 +21,17 @@ IgnoredVulns = [
   { id = "RUSTSEC-2026-0381", ignoreUntil = 2026-10-31 },
   { id = "RUSTSEC-2026-0383", ignoreUntil = 2026-10-31 },
   { id = "RUSTSEC-2026-0387", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0391", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0395", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0396", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0397", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0398", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0399", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0400", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0401", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0402", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0403", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0404", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0405", ignoreUntil = 2026-10-31 },
+  { id = "RUSTSEC-2026-0406", ignoreUntil = 2026-10-31 },
 ]
diff --git a/generated/dart-rest/lib/clients/storage.dart b/generated/dart-rest/lib/clients/storage.dart
index 845c5259..c9bdbd53 100644
--- a/generated/dart-rest/lib/clients/storage.dart
+++ b/generated/dart-rest/lib/clients/storage.dart
@@ -27,9 +27,8 @@ abstract class Storage {
 
   /// Upload a file to an agent's storage bucket.
   ///
-  /// Only the agent that owns the bucket may write to it (the agent must.
-  /// currently be connected); the path component must be a single.
-  /// filename, not a nested path.
+  /// Only the agent that owns the bucket may write to it: the agent must be connected, and the request must come from.
+  /// the address it is connected from. The path component must be a single filename, not a nested path.
   ///
   /// [agentId] - Agent identifier (must be a connected agent).
   ///
diff --git a/generated/python-rest/et_rest_client/api/storage/put_file.py b/generated/python-rest/et_rest_client/api/storage/put_file.py
index 4eacddfa..df6e794c 100644
--- a/generated/python-rest/et_rest_client/api/storage/put_file.py
+++ b/generated/python-rest/et_rest_client/api/storage/put_file.py
@@ -39,6 +39,9 @@ def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Res
     if response.status_code == 400:
         return None
 
+    if response.status_code == 403:
+        return None
+
     if response.status_code == 404:
         return None
 
@@ -66,9 +69,9 @@ def sync_detailed(
 ) -> Response[Any]:
     """Upload a file to an agent's storage bucket.
 
-     Only the agent that owns the bucket may write to it (the agent must
-    currently be connected); the path component must be a single
-    filename, not a nested path.
+     Only the agent that owns the bucket may write to it: the agent must be connected, and the request
+    must come from
+    the address it is connected from. The path component must be a single filename, not a nested path.
 
     Args:
         agent_id (str):
@@ -109,9 +112,9 @@ async def asyncio_detailed(
 ) -> Response[Any]:
     """Upload a file to an agent's storage bucket.
 
-     Only the agent that owns the bucket may write to it (the agent must
-    currently be connected); the path component must be a single
-    filename, not a nested path.
+     Only the agent that owns the bucket may write to it: the agent must be connected, and the request
+    must come from
+    the address it is connected from. The path component must be a single filename, not a nested path.
 
     Args:
         agent_id (str):
diff --git a/generated/rust-rest/src/lib.rs b/generated/rust-rest/src/lib.rs
index d21a574d..bf3f6b38 100644
--- a/generated/rust-rest/src/lib.rs
+++ b/generated/rust-rest/src/lib.rs
@@ -382,9 +382,8 @@ impl Client {
     }
     /**Upload a file to an agent's storage bucket
 
-    Only the agent that owns the bucket may write to it (the agent must
-    currently be connected); the path component must be a single
-    filename, not a nested path.
+    Only the agent that owns the bucket may write to it: the agent must be connected, and the request must come from
+    the address it is connected from. The path component must be a single filename, not a nested path.
 
     Sends a `PUT` request to `/storage/{agent_id}/{filename}`
 
@@ -450,6 +449,7 @@ impl Client {
         match response.status().as_u16() {
             200u16 => Ok(ResponseValue::empty(response)),
             400u16 => Err(Error::ErrorResponse(ResponseValue::empty(response))),
+            403u16 => Err(Error::ErrorResponse(ResponseValue::empty(response))),
             404u16 => Err(Error::ErrorResponse(ResponseValue::empty(response))),
             _ => Err(Error::UnexpectedResponse(response)),
         }
diff --git a/generated/specs/rest.yaml b/generated/specs/rest.yaml
index a14fb89f..25786f23 100644
--- a/generated/specs/rest.yaml
+++ b/generated/specs/rest.yaml
@@ -117,9 +117,8 @@ paths:
         - storage
       summary: Upload a file to an agent's storage bucket.
       description: |-
-        Only the agent that owns the bucket may write to it (the agent must
-        currently be connected); the path component must be a single
-        filename, not a nested path.
+        Only the agent that owns the bucket may write to it: the agent must be connected, and the request must come from
+        the address it is connected from. The path component must be a single filename, not a nested path.
       operationId: put_file
       parameters:
         - in: path
@@ -160,6 +159,8 @@ paths:
                 type: string
         "400":
           description: Invalid filename
+        "403":
+          description: Agent not connected, or request not from its address
         "404":
           description: Agent not found
     head:
diff --git a/generated/zig-rest/src/et_rest_client.zig b/generated/zig-rest/src/et_rest_client.zig
index 7f073afe..1e022c2d 100644
--- a/generated/zig-rest/src/et_rest_client.zig
+++ b/generated/zig-rest/src/et_rest_client.zig
@@ -1,7 +1,7 @@
 // <auto-generated>
 //   This code was generated by openapi2zig 0.6.0 (13a4354)
 //   Changes to this file may cause incorrect behavior and will be lost if the code is regenerated
-//   Checksum: 3e1a49aa14019014
+//   Checksum: 192b90fe0bb313d
 // </auto-generated>
 
 const std = @import("std");
@@ -574,9 +574,8 @@ pub fn get_fileRaw(client: *Client, agent_id: []const u8, filename: []const u8)
 // Upload a file to an agent's storage bucket.
 //
 // Description:
-// Only the agent that owns the bucket may write to it (the agent must
-// currently be connected); the path component must be a single
-// filename, not a nested path.
+// Only the agent that owns the bucket may write to it: the agent must be connected, and the request must come from
+// the address it is connected from. The path component must be a single filename, not a nested path.
 //
 pub fn put_file(client: *Client, agent_id: []const u8, filename: []const u8, requestBody: []const u8) !void {
     var raw = try put_fileRaw(client, agent_id, filename, requestBody);
diff --git a/libs/edge-toolkit/src/config.rs b/libs/edge-toolkit/src/config.rs
index b4a22463..aff1db91 100644
--- a/libs/edge-toolkit/src/config.rs
+++ b/libs/edge-toolkit/src/config.rs
@@ -244,12 +244,12 @@ impl Language {
 ///   the developer's `mise install` covered everything; treat every language
 ///   as available.
 /// - **`MISE_ENV=""`** (set, empty): CI/Docker explicitly narrowed the env
-///   to "no guest languages". Every language returns false; tests that
-///   depend on a guest-env tool skip cleanly.
+///   to "no guest languages". Every language returns false; work that
+///   depends on a guest-env tool skips cleanly.
 ///
-/// Used to gate live-installed-tool tests: when CI narrows `MISE_ENV` for faster feedback (e.g. `dotnet,rust`), tests
-/// that depend on a tool in a dropped env (e.g. `http:pyodide` from the `python` env) can skip cleanly instead of
-/// panicking.
+/// Used to gate work on live-installed tools: when CI narrows `MISE_ENV` for faster feedback (e.g. `dotnet,rust`),
+/// work that depends on a tool in a dropped env (e.g. `http:pyodide` from the `python` env) can skip cleanly instead
+/// of panicking.
 ///
 /// When the language is absent (`MISE_ENV=""` or a list that omits it), a skip line naming the language is emitted to
 /// stderr, so callers only need to branch on the returned bool -- they don't repeat the message themselves.
diff --git a/libs/edge-toolkit/src/ws_server.rs b/libs/edge-toolkit/src/ws_server.rs
index 84f457ff..3624ce45 100644
--- a/libs/edge-toolkit/src/ws_server.rs
+++ b/libs/edge-toolkit/src/ws_server.rs
@@ -105,6 +105,33 @@ impl PendingDirectMessage {
     }
 }
 
+/// How many unacknowledged direct messages a recipient holds from each sender.
+///
+/// The default keeps one: a recipient's pending messages then number at most its senders, so the memory they take is
+/// bounded and known in advance. Agents that must not lose a message wait for the acknowledgement of one before sending
+/// the next; that discipline is theirs, not the hub's.
+#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "kebab-case")]
+#[non_exhaustive]
+pub enum MessageRetention {
+    /// Only the newest message from each sender; a second one replaces the first, which is never delivered.
+    #[default]
+    LatestPerSender,
+    /// Every message until it is acknowledged. Nothing bounds how many a recipient holds.
+    All,
+}
+
+impl MessageRetention {
+    /// The key `pending` is filed under in its recipient's pending messages.
+    #[must_use]
+    pub fn key(self, pending: &PendingDirectMessage) -> String {
+        match self {
+            Self::LatestPerSender => pending.from_agent_id.clone(),
+            Self::All => pending.message_id.clone(),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[non_exhaustive]
 pub struct AgentRecord<S> {
@@ -112,6 +139,10 @@ pub struct AgentRecord<S> {
     pub last_known_ip: IpAddr,
     #[serde(skip)]
     pub session: Option<S>,
+    /// Unacknowledged direct messages, keyed by [`MessageRetention::key`].
+    ///
+    /// Entries are found by the message id they carry rather than by key, so a registry saved under one retention mode
+    /// still acknowledges after a restart under the other.
     #[serde(default)]
     pub pending_direct_messages: BTreeMap<String, PendingDirectMessage>,
 }
@@ -143,7 +174,7 @@ pub struct AgentRegistry<S> {
 }
 
 impl<S> AgentRegistry<S> {
-    /// Wrap a populated agent map; useful for tests building fixtures.
+    /// Wrap a populated agent map.
     #[must_use]
     pub fn from_agents(agents: BTreeMap<String, AgentRecord<S>>) -> Self {
         Self {
@@ -223,13 +254,13 @@ impl<S: Clone + Send + 'static> AgentRegistry<S> {
     ///
     /// A connection that takes over an id another connection still holds -- the newest claim wins, so a reconnect whose
     /// old socket has not closed yet keeps its identity -- replaces that session. When the displaced connection later
-    /// closes, it must not disconnect its replacement, so a closing connection passes a test for its own session. A
-    /// record whose session is already gone is disconnected whatever the test says, since nothing live holds it.
+    /// closes, it must not disconnect its replacement, so a closing connection passes a predicate for its own session.
+    /// A record whose session is already gone is disconnected whatever the predicate says, since nothing live holds it.
     ///
-    /// The test is a trait object rather than a generic so there is one copy of this per session type. Every closure is
-    /// its own type, and a generic would give each caller its own copy with only that caller's branches taken -- which
-    /// the branch-coverage gate scores copy by copy, so no single copy would ever read as fully covered:
-    /// `libs/edge-toolkit/src/ws_server.rs 19/20 branches` on commit
+    /// The predicate is a trait object rather than a generic so there is one copy of this per session type.
+    /// Every closure is its own type, and a generic would give each caller its own copy with only that caller's
+    /// branches taken -- which the branch-coverage gate scores copy by copy, so no single copy would ever read
+    /// as fully covered: `libs/edge-toolkit/src/ws_server.rs 19/20 branches` on commit
     /// <https://github.com/edge-toolkit/core/commit/1782e4517b8c16d714d5c5e6e83d0078f2a869a9> at
     /// <https://github.com/edge-toolkit/core/actions/runs/36699637919/job/109835793913>.
     ///
@@ -266,6 +297,8 @@ impl<S: Clone + Send + 'static> AgentRegistry<S> {
 
     /// Queue a direct message for `to_agent_id`, returning the stored message and the recipient's session.
     ///
+    /// `retention` decides whether an earlier unacknowledged message from the same sender is replaced or kept.
+    ///
     /// Returns `None` when `to_agent_id` is not in the registry. The inner `Option<S>` is the recipient's live session
     /// -- `Some` when connected, `None` when the message was queued for a disconnected agent.
     #[must_use]
@@ -276,6 +309,7 @@ impl<S: Clone + Send + 'static> AgentRegistry<S> {
         to_agent_id: &str,
         server_received_at: String,
         message: serde_json::Value,
+        retention: MessageRetention,
     ) -> Option<(PendingDirectMessage, Option<S>)> {
         let mut agents = lock_agents(&self.agents);
         let recipient = agents.get_mut(to_agent_id)?;
@@ -284,7 +318,7 @@ impl<S: Clone + Send + 'static> AgentRegistry<S> {
         let session = recipient.session.clone();
         let _previous: Option<PendingDirectMessage> = recipient
             .pending_direct_messages
-            .insert(from_agent_id.to_string(), pending.clone());
+            .insert(retention.key(&pending), pending.clone());
         drop(agents);
 
         Some((pending, session))
@@ -314,18 +348,19 @@ impl<S: Clone + Send + 'static> AgentRegistry<S> {
             .get_mut(recipient_agent_id)
             .ok_or_else(|| AcknowledgeError::UnknownAgent(recipient_agent_id.to_string()))?;
 
-        let sender_agent_id = recipient
+        let key = recipient
             .pending_direct_messages
             .iter()
-            .find_map(|(id, pending)| (pending.message_id == message_id).then(|| id.clone()))
+            .find_map(|(key, pending)| (pending.message_id == message_id).then(|| key.clone()))
             .ok_or(AcknowledgeError::NoPendingMessage)?;
         // The `find_map` above drops its iterator before we re-borrow `pending_direct_messages` mutably for the
         // removal. The double lookup costs O(log n) but lets us share the `NoPendingMessage` error with the find-side
         // case instead of asserting an invariant.
         let pending = recipient
             .pending_direct_messages
-            .remove(&sender_agent_id)
+            .remove(&key)
             .ok_or(AcknowledgeError::NoPendingMessage)?;
+        let sender_agent_id = pending.from_agent_id;
         let sender_session = agents.get(&sender_agent_id).and_then(|record| record.session.clone());
         drop(agents);
 
diff --git a/libs/edge-toolkit/tests/registry.rs b/libs/edge-toolkit/tests/registry.rs
index 567a4f69..c0ea1f2d 100644
--- a/libs/edge-toolkit/tests/registry.rs
+++ b/libs/edge-toolkit/tests/registry.rs
@@ -8,7 +8,9 @@ use std::collections::BTreeMap;
 use std::net::{IpAddr, Ipv4Addr};
 
 use edge_toolkit::ws::{AgentConnectionState, ConnectStatus};
-use edge_toolkit::ws_server::{AgentRecord, AgentRegistry, MAX_AGENT_ID_LEN, is_valid_agent_id};
+use edge_toolkit::ws_server::{
+    AcknowledgeError, AgentRecord, AgentRegistry, MAX_AGENT_ID_LEN, MessageRetention, is_valid_agent_id,
+};
 use tempfile::tempdir;
 
 const FIRST_IP: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
@@ -174,3 +176,96 @@ fn agent_record_with_pending_builder_replaces_the_map() {
         .with_pending_direct_messages(BTreeMap::new());
     assert!(record.pending_direct_messages.is_empty());
 }
+
+/// A registry holding a connected `recipient` and a connected `sender`, each with a session named after it.
+fn sender_and_recipient() -> AgentRegistry<String> {
+    let registry = AgentRegistry::<String>::default();
+    let (_recipient, _assigned) = registry.connect_agent(
+        Some("recipient".to_string()),
+        String::default(),
+        FIRST_IP,
+        "r".to_string(),
+    );
+    let (_sender, _assigned) = registry.connect_agent(
+        Some("sender".to_string()),
+        String::default(),
+        SECOND_IP,
+        "s".to_string(),
+    );
+    registry
+}
+
+/// Queue two messages from `sender` to `recipient` under `retention`, returning the ids left pending.
+fn queue_two(registry: &AgentRegistry<String>, retention: MessageRetention) -> Vec<String> {
+    for message_id in ["m1", "m2"] {
+        let (pending, session) = registry
+            .queue_direct(
+                message_id.to_string(),
+                "sender",
+                "recipient",
+                "2026-10-26T00:00:00Z".to_string(),
+                serde_json::json!({"id": message_id}),
+                retention,
+            )
+            .unwrap();
+        assert_eq!(pending.message_id, message_id);
+        assert_eq!(session.as_deref(), Some("r"));
+    }
+    let pending = registry.pending_messages_for("recipient");
+    pending.into_iter().map(|message| message.message_id).collect()
+}
+
+#[test]
+fn latest_per_sender_retention_replaces_the_earlier_message() {
+    let registry = sender_and_recipient();
+    assert_eq!(MessageRetention::default(), MessageRetention::LatestPerSender);
+
+    assert_eq!(queue_two(&registry, MessageRetention::LatestPerSender), ["m2"]);
+    assert!(matches!(
+        registry.acknowledge_message("recipient", "m1"),
+        Err(AcknowledgeError::NoPendingMessage)
+    ));
+}
+
+#[test]
+fn all_retention_keeps_every_message_until_acknowledged() {
+    let registry = sender_and_recipient();
+
+    assert_eq!(queue_two(&registry, MessageRetention::All), ["m1", "m2"]);
+    let (acknowledged, sender_session, sender) = registry.acknowledge_message("recipient", "m1").unwrap();
+    assert_eq!(
+        (acknowledged.as_str(), sender_session.as_deref(), sender.as_str()),
+        ("m1", Some("s"), "sender")
+    );
+    let left = registry.pending_messages_for("recipient");
+    assert_eq!(left.len(), 1);
+    assert_eq!(left[0].message_id, "m2");
+}
+
+#[test]
+fn messages_queued_under_one_retention_acknowledge_under_the_other() {
+    let registry = sender_and_recipient();
+    let _kept = queue_two(&registry, MessageRetention::All);
+    // The earlier entries are keyed by message id, so a later per-sender one sits beside them rather than replacing.
+    let _queued = registry.queue_direct(
+        "m3".to_string(),
+        "sender",
+        "recipient",
+        "2026-10-26T00:00:01Z".to_string(),
+        serde_json::Value::Null,
+        MessageRetention::LatestPerSender,
+    );
+
+    for message_id in ["m3", "m1", "m2"] {
+        let (_acknowledged, _session, sender) = registry.acknowledge_message("recipient", message_id).unwrap();
+        assert_eq!(sender, "sender");
+    }
+    assert!(registry.pending_messages_for("recipient").is_empty());
+}
+
+#[test]
+fn retention_reads_its_kebab_case_names() {
+    let parse = |name: &str| serde_json::from_value::<MessageRetention>(serde_json::json!(name)).unwrap();
+    assert_eq!(parse("latest-per-sender"), MessageRetention::LatestPerSender);
+    assert_eq!(parse("all"), MessageRetention::All);
+}
diff --git a/libs/et-otlp/src/lib.rs b/libs/et-otlp/src/lib.rs
index 542d1642..48cc30d8 100644
--- a/libs/et-otlp/src/lib.rs
+++ b/libs/et-otlp/src/lib.rs
@@ -51,9 +51,9 @@ impl OtelHandles {
 
 /// The HTTP headers every OTLP exporter is built with, carrying basic auth when the config supplies it.
 ///
-/// Split out of [`init`] so both shapes are reachable from a test. `init` installs a process-global
+/// Split out of [`init`] so both shapes can be exercised. `init` installs a process-global
 /// subscriber and can therefore run at most once per process, which leaves anything decided inside it
-/// testable only in whichever configuration that single call happens to use -- and a collector reached
+/// reachable only in whichever configuration that single call happens to use -- and a collector reached
 /// without its credentials rejects every export, silently, for the life of the process.
 #[must_use]
 pub fn exporter_headers(auth: Option<&edge_toolkit::auth::BasicAuth>) -> std::collections::HashMap<String, String> {
@@ -67,8 +67,8 @@ pub fn exporter_headers(auth: Option<&edge_toolkit::auth::BasicAuth>) -> std::co
 /// The resource attributes describing this service, given whatever the host's name resolved to.
 ///
 /// Takes the hostname rather than reading it, for the same reason as [`exporter_headers`] and one more:
-/// the `None` case is a host whose name does not resolve or is not UTF-8, which no test can bring about
-/// by asking the real machine. Passing it in is what makes the attribute's absence an asserted behaviour
+/// the `None` case is a host whose name does not resolve or is not UTF-8, which nothing can bring about on
+/// demand by asking the real machine. Passing it in is what makes the attribute's absence an asserted behaviour
 /// instead of an assumption -- the resource must still carry the version, and simply omit the instance.
 #[must_use]
 pub fn service_descriptors(hostname: Option<String>) -> Vec<KeyValue> {
diff --git a/libs/python-org/et_org/__init__.py b/libs/python-org/et_org/__init__.py
index 16a6feff..08bc5ef2 100644
--- a/libs/python-org/et_org/__init__.py
+++ b/libs/python-org/et_org/__init__.py
@@ -1,7 +1,7 @@
 """Who this project is, as the strings that carry its identity into registries, images and manifests.
 
 The Python twin of the Rust `et-org` crate. The organisation name is written exactly once, in `ORG`, and every
-longer string is assembled from it; a test in the Rust crate fails if `ORG` here stops matching the Rust one.
+longer string is assembled from it; the Rust crate checks that `ORG` here keeps matching its own.
 """
 
 ORG = "edge-toolkit"
diff --git a/libs/web/src/lib.rs b/libs/web/src/lib.rs
index 03384d05..00fb850e 100644
--- a/libs/web/src/lib.rs
+++ b/libs/web/src/lib.rs
@@ -116,8 +116,8 @@ pub fn websocket_url() -> Result<String, JsValue> {
 
 /// The `/ws` endpoint URL for a page at `location`, upgrading to `wss:` when its protocol is `https:`.
 ///
-/// Takes the location rather than reading `window.location`: a test page is only ever served over plain
-/// http, so the `wss:` upgrade can be asserted only against a location handed in. `location` needs just the
+/// Takes the location rather than reading `window.location`, so the `wss:` upgrade can be asserted against a
+/// location handed in, whatever protocol served the page. `location` needs just the
 /// `protocol` and `host` properties, which is what makes a plain object stand in for the real one.
 pub fn websocket_url_from_location(location: &JsValue) -> Result<String, JsValue> {
     let protocol = js_sys::Reflect::get(location, &JsValue::from_str("protocol"))?
diff --git a/libs/ws-runner-common/src/web_config.rs b/libs/ws-runner-common/src/web_config.rs
index 6193b1b8..b74eb79b 100644
--- a/libs/ws-runner-common/src/web_config.rs
+++ b/libs/ws-runner-common/src/web_config.rs
@@ -17,9 +17,9 @@ pub struct Config {
     ///
     /// Only present under the `coverage` cargo feature -- the capture code is compiled in only there. When the feature
     /// is on, `ET_TEST_COVERAGE=true` makes each module PUT its minicov `.profraw` / coverage.py data to its own
-    /// storage bucket for the web-runner test to gather; `false` (the default) leaves the compiled-in capture inert.
-    /// serde-env parses the value with `str::parse::<bool>`, so it must be `true`/`false`. Left out of the env schema,
-    /// so a runner's HELP.md documents only what every build reads.
+    /// storage bucket for the coverage collection to gather; `false` (the default) leaves the compiled-in capture
+    /// inert. serde-env parses the value with `str::parse::<bool>`, so it must be `true`/`false`. Left out of the env
+    /// schema, so a runner's HELP.md documents only what every build reads.
     #[cfg(feature = "coverage")]
     #[cfg_attr(feature = "env-schema", schemars(skip))]
     #[serde(default)]
diff --git a/services/hub-backplane/src/lib.rs b/services/hub-backplane/src/lib.rs
index 361ff6f2..fad7465d 100644
--- a/services/hub-backplane/src/lib.rs
+++ b/services/hub-backplane/src/lib.rs
@@ -4,8 +4,8 @@
 //!
 //! - an **agent directory** (`JetStream` KV bucket `et-agents`) recording, per agent, its connection state, its last
 //!   known IP and the hub currently holding its socket;
-//! - a **mailbox** (KV bucket `et-pending`) of unacknowledged direct messages per recipient, keeping the hub's "newest
-//!   message per sender wins" rule, so a message survives its recipient reconnecting to a different hub;
+//! - a **mailbox** (KV bucket `et-pending`) of unacknowledged direct messages per recipient, filed by the hub's
+//!   [`MessageRetention`], so a message survives its recipient reconnecting to a different hub;
 //! - a **delivery fabric** over core NATS subjects: `et.hub.<hub_id>` carries frames for one hub's agents and
 //!   `et.broadcast` carries fan-out to every hub. The target agent and payload kind travel as headers and the frame
 //!   bytes as the body, so binary relay needs no re-encoding.
@@ -22,7 +22,7 @@ use async_nats::jetstream::kv::{self, CreateErrorKind, Operation, Store, UpdateE
 use async_nats::{Client, HeaderMap, Message};
 use bytes::Bytes;
 use edge_toolkit::ws::{AgentConnectionState, AgentSummary};
-use edge_toolkit::ws_server::PendingDirectMessage;
+use edge_toolkit::ws_server::{MessageRetention, PendingDirectMessage};
 use futures_util::{Stream, StreamExt as _, TryStreamExt as _};
 use serde::{Deserialize, Serialize};
 use tracing::warn;
@@ -47,12 +47,14 @@ const KIND_HEADER: &str = "Et-Kind";
 const CONFIRM_TIMEOUT: Duration = Duration::from_secs(10);
 
 /// The backplane a hub routes through: nothing beyond this process, or a shared NATS server.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 #[expect(
     clippy::exhaustive_enums,
     reason = "the hub routes on each mode; a new one must make it say how"
 )]
 pub enum Backplane {
+    /// Also what an app that never joined a backplane -- any embedding that predates them -- runs as.
+    #[default]
     Local,
     Nats(Box<NatsBackplane>),
 }
@@ -66,13 +68,13 @@ impl Backplane {
         }
     }
 
-    /// Whether any hub on this backplane has registered `agent_id`.
+    /// Where the hubs on this backplane place `agent_id`.
     ///
-    /// Always false for `Local`, whose only registry is the caller's own.
-    pub async fn knows(&self, agent_id: &str) -> Result<bool, BackplaneError> {
+    /// Always `Unknown` for `Local`, whose only registry is the caller's own.
+    pub async fn presence(&self, agent_id: &str) -> Result<Presence, BackplaneError> {
         match self {
-            Self::Local => Ok(false),
-            Self::Nats(nats) => Ok(!matches!(nats.presence(agent_id).await?, Presence::Unknown)),
+            Self::Local => Ok(Presence::Unknown),
+            Self::Nats(nats) => nats.presence(agent_id).await,
         }
     }
 }
@@ -86,8 +88,8 @@ impl Backplane {
 pub enum Presence {
     /// No hub has ever registered it.
     Unknown,
-    /// Its socket is open on the hub with this id.
-    Connected { hub_id: String },
+    /// Its socket is open on the hub with id `hub_id`, from `last_known_ip`.
+    Connected { hub_id: String, last_known_ip: IpAddr },
     /// It registered once and has since gone.
     Disconnected,
 }
@@ -182,7 +184,7 @@ struct DirectoryRecord {
     hub_id: String,
 }
 
-/// One recipient's mailbox: its newest unacknowledged message from each sender, keyed by sender.
+/// One recipient's unacknowledged messages, keyed by [`MessageRetention::key`].
 type Mailbox = BTreeMap<String, PendingDirectMessage>;
 
 /// A hub's connection to a shared NATS backplane.
@@ -280,9 +282,10 @@ impl NatsBackplane {
     pub async fn presence(&self, agent_id: &str) -> Result<Presence, BackplaneError> {
         Ok(match self.directory_entry(agent_id).await? {
             None => Presence::Unknown,
-            Some((record, _revision)) if record.state == AgentConnectionState::Connected => {
-                Presence::Connected { hub_id: record.hub_id }
-            }
+            Some((record, _revision)) if record.state == AgentConnectionState::Connected => Presence::Connected {
+                hub_id: record.hub_id,
+                last_known_ip: record.last_known_ip,
+            },
             Some(_disconnected) => Presence::Disconnected,
         })
     }
@@ -308,10 +311,15 @@ impl NatsBackplane {
         Ok(agents)
     }
 
-    /// Put `message` in `to_agent_id`'s mailbox, replacing any earlier one from the same sender.
-    pub async fn queue(&self, to_agent_id: &str, message: PendingDirectMessage) -> Result<(), BackplaneError> {
+    /// Put `message` in `to_agent_id`'s mailbox, replacing an earlier one from the same sender if `retention` says to.
+    pub async fn queue(
+        &self,
+        to_agent_id: &str,
+        message: PendingDirectMessage,
+        retention: MessageRetention,
+    ) -> Result<(), BackplaneError> {
         self.modify_mailbox(to_agent_id, |mailbox| {
-            let _previous = mailbox.insert(message.from_agent_id.clone(), message.clone());
+            let _previous = mailbox.insert(retention.key(&message), message.clone());
             Ok(())
         })
         .await
@@ -328,11 +336,13 @@ impl NatsBackplane {
     /// Take message `message_id` out of `recipient_agent_id`'s mailbox, returning the id of the agent that sent it.
     pub async fn acknowledge(&self, recipient_agent_id: &str, message_id: &str) -> Result<String, BackplaneError> {
         self.modify_mailbox(recipient_agent_id, |mailbox| {
-            let sender = mailbox
+            let (key, sender) = mailbox
                 .iter()
-                .find_map(|(sender, pending)| (pending.message_id == message_id).then(|| sender.clone()))
+                .find_map(|(key, pending)| {
+                    (pending.message_id == message_id).then(|| (key.clone(), pending.from_agent_id.clone()))
+                })
                 .ok_or_else(|| BackplaneError::NoPendingMessage(message_id.to_owned()))?;
-            let _acknowledged = mailbox.remove(&sender);
+            let _acknowledged = mailbox.remove(&key);
             Ok(sender)
         })
         .await
diff --git a/services/hub-backplane/tests/backplane.rs b/services/hub-backplane/tests/backplane.rs
index e616e65a..10462ba4 100644
--- a/services/hub-backplane/tests/backplane.rs
+++ b/services/hub-backplane/tests/backplane.rs
@@ -7,12 +7,13 @@ use std::time::Duration;
 use async_nats::HeaderMap;
 use bytes::Bytes;
 use edge_toolkit::ws::AgentConnectionState;
-use edge_toolkit::ws_server::PendingDirectMessage;
+use edge_toolkit::ws_server::{MessageRetention, PendingDirectMessage};
 use et_hub_backplane::{Backplane, BackplaneConfig, BackplaneError, Frame, Inbound, NatsBackplane, Presence};
 use et_test_nats::NatsServer;
 use futures_util::{Stream, StreamExt as _};
 
 const LOOPBACK: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
+const NEWEST: MessageRetention = MessageRetention::LatestPerSender;
 
 fn pending(message_id: &str, from: &str) -> PendingDirectMessage {
     PendingDirectMessage::new(
@@ -23,6 +24,10 @@ fn pending(message_id: &str, from: &str) -> PendingDirectMessage {
     )
 }
 
+fn ids(messages: Vec<PendingDirectMessage>) -> Vec<String> {
+    messages.into_iter().map(|message| message.message_id).collect()
+}
+
 /// The next inbound frame, failing the test if none arrives within five seconds.
 async fn next_inbound(stream: &mut (impl Stream<Item = Inbound> + Unpin)) -> Inbound {
     tokio::time::timeout(Duration::from_secs(5), stream.next())
@@ -64,7 +69,7 @@ fn config_names_local_or_a_nats_url() {
 async fn a_local_backplane_knows_nobody() {
     let backplane = Backplane::connect(&BackplaneConfig::Local).await.unwrap();
     assert!(matches!(backplane, Backplane::Local));
-    assert!(!backplane.knows("anyone").await.unwrap());
+    assert_eq!(backplane.presence("anyone").await.unwrap(), Presence::Unknown);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -125,6 +130,7 @@ async fn the_directory_follows_an_agent_across_hubs() {
     first.announce("a.b", LOOPBACK).await.unwrap();
     let on_first = Presence::Connected {
         hub_id: first.hub_id().to_owned(),
+        last_known_ip: LOOPBACK,
     };
     assert_eq!(second.presence("a.b").await.unwrap(), on_first);
 
@@ -133,6 +139,7 @@ async fn the_directory_follows_an_agent_across_hubs() {
     first.withdraw("a.b").await.unwrap();
     let on_second = Presence::Connected {
         hub_id: second.hub_id().to_owned(),
+        last_known_ip: LOOPBACK,
     };
     assert_eq!(first.presence("a.b").await.unwrap(), on_second);
 
@@ -150,9 +157,13 @@ async fn the_directory_follows_an_agent_across_hubs() {
     assert_eq!(listed[0].state, AgentConnectionState::Disconnected);
     assert_eq!(listed[1].state, AgentConnectionState::Connected);
 
+    let z_on_first = Presence::Connected {
+        hub_id: first.hub_id().to_owned(),
+        last_known_ip: LOOPBACK,
+    };
     let backplane = Backplane::Nats(Box::new(first));
-    assert!(backplane.knows("z").await.unwrap());
-    assert!(!backplane.knows("ghost").await.unwrap());
+    assert_eq!(backplane.presence("z").await.unwrap(), z_on_first);
+    assert_eq!(backplane.presence("ghost").await.unwrap(), Presence::Unknown);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -198,7 +209,7 @@ async fn corrupt_records_are_reported_not_ignored() {
         .unwrap();
     assert!(matches!(hub.pending_for("bad").await, Err(BackplaneError::Codec(_))));
     assert!(matches!(
-        hub.queue("bad", pending("m", "x")).await,
+        hub.queue("bad", pending("m", "x"), NEWEST).await,
         Err(BackplaneError::Codec(_))
     ));
 }
@@ -210,15 +221,9 @@ async fn the_mailbox_keeps_the_newest_message_per_sender_until_acknowledged() {
     let second = NatsBackplane::connect(&server.url()).await.unwrap();
 
     assert!(first.pending_for("r").await.unwrap().is_empty());
-    first.queue("r", pending("m1", "alice")).await.unwrap();
-    first.queue("r", pending("m2", "alice")).await.unwrap();
-    second.queue("r", pending("m3", "bob")).await.unwrap();
-    let ids = |messages: Vec<PendingDirectMessage>| {
-        messages
-            .into_iter()
-            .map(|message| message.message_id)
-            .collect::<Vec<_>>()
-    };
+    first.queue("r", pending("m1", "alice"), NEWEST).await.unwrap();
+    first.queue("r", pending("m2", "alice"), NEWEST).await.unwrap();
+    second.queue("r", pending("m3", "bob"), NEWEST).await.unwrap();
     assert_eq!(ids(second.pending_for("r").await.unwrap()), ["m2", "m3"]);
 
     assert_eq!(second.acknowledge("r", "m2").await.unwrap(), "alice");
@@ -235,10 +240,27 @@ async fn the_mailbox_keeps_the_newest_message_per_sender_until_acknowledged() {
         .await
         .unwrap();
     assert!(first.pending_for("r").await.unwrap().is_empty());
-    first.queue("r", pending("m4", "carol")).await.unwrap();
+    first.queue("r", pending("m4", "carol"), NEWEST).await.unwrap();
     assert_eq!(ids(first.pending_for("r").await.unwrap()), ["m4"]);
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn the_mailbox_keeps_every_message_under_all_retention() {
+    let server = NatsServer::start();
+    let hub = NatsBackplane::connect(&server.url()).await.unwrap();
+
+    hub.queue("r", pending("m1", "alice"), MessageRetention::All)
+        .await
+        .unwrap();
+    hub.queue("r", pending("m2", "alice"), MessageRetention::All)
+        .await
+        .unwrap();
+    assert_eq!(ids(hub.pending_for("r").await.unwrap()), ["m1", "m2"]);
+
+    assert_eq!(hub.acknowledge("r", "m1").await.unwrap(), "alice");
+    assert_eq!(ids(hub.pending_for("r").await.unwrap()), ["m2"]);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn concurrent_senders_all_land_in_one_mailbox() {
     let server = NatsServer::start();
@@ -251,8 +273,12 @@ async fn concurrent_senders_all_land_in_one_mailbox() {
     for (index, hub) in (0..40_u32).zip(hubs.iter().cycle()) {
         let hub = hub.clone();
         writers.push(tokio::spawn(async move {
-            hub.queue("busy", pending(&format!("m{index:02}"), &format!("s{index:02}")))
-                .await
+            hub.queue(
+                "busy",
+                pending(&format!("m{index:02}"), &format!("s{index:02}")),
+                NEWEST,
+            )
+            .await
         }));
     }
     for writer in writers {
diff --git a/services/recovery-supervisor/src/kubernetes.rs b/services/recovery-supervisor/src/kubernetes.rs
index 4e0559cc..d8d12f61 100644
--- a/services/recovery-supervisor/src/kubernetes.rs
+++ b/services/recovery-supervisor/src/kubernetes.rs
@@ -9,7 +9,7 @@ use crate::RecoveryError;
 use crate::config::{Config, Workload};
 use crate::policy::Health;
 
-/// Deployment-specific operations consumed by the supervisor, independently testable with an in-memory backend.
+/// Deployment-specific operations consumed by the supervisor, which an in-memory backend can stand in for.
 pub trait Backend: Send {
     /// Observe readiness; failures must remain errors, not be interpreted as application failure.
     fn health(&mut self, workload: &Workload) -> impl Future<Output = Result<Health, RecoveryError>> + Send;
diff --git a/services/storage/src/lib.rs b/services/storage/src/lib.rs
index f1c141ed..7d04a268 100644
--- a/services/storage/src/lib.rs
+++ b/services/storage/src/lib.rs
@@ -92,6 +92,18 @@ pub enum StorageError {
     #[response(status = 404, reason = "NOT_FOUND")]
     AgentNotFound,
 
+    #[error("agent is not connected")]
+    #[response(status = 403, reason = "FORBIDDEN")]
+    AgentNotConnected,
+
+    #[error("requester is not the agent that owns this bucket")]
+    #[response(status = 403, reason = "FORBIDDEN")]
+    RequesterNotBucketOwner,
+
+    /// Every listener is TCP, so a request without a peer means the server was bound wrongly.
+    #[error("request has no peer address")]
+    NoPeerAddress,
+
     #[error("no such object")]
     #[response(status = 404, reason = "NOT_FOUND")]
     ObjectNotFound,
diff --git a/services/storage/src/routes.rs b/services/storage/src/routes.rs
index b18fd7ce..56224b4d 100644
--- a/services/storage/src/routes.rs
+++ b/services/storage/src/routes.rs
@@ -21,12 +21,14 @@
     )
 )]
 
+use std::net::IpAddr;
 use std::path::PathBuf;
 
 use actix_web::http::header;
 use actix_web::{HttpRequest, HttpResponse, HttpResponseBuilder, web};
+use edge_toolkit::ws::AgentConnectionState;
 use edge_toolkit::ws_server::AgentRegistry;
-use et_hub_backplane::Backplane;
+use et_hub_backplane::{Backplane, Presence};
 use futures_util::StreamExt as _;
 use object_store::{ObjectStore, ObjectStoreExt as _, PutPayload};
 use tracing::info;
@@ -82,9 +84,8 @@ pub struct BinaryBlob(#[expect(dead_code)] Vec<u8>);
 
 /// Upload a file to an agent's storage bucket.
 ///
-/// Only the agent that owns the bucket may write to it (the agent must
-/// currently be connected); the path component must be a single
-/// filename, not a nested path.
+/// Only the agent that owns the bucket may write to it: the agent must be connected, and the request must come from
+/// the address it is connected from. The path component must be a single filename, not a nested path.
 #[cfg_attr(
     feature = "openapi-spec",
     utoipa::path(
@@ -105,6 +106,7 @@ pub struct BinaryBlob(#[expect(dead_code)] Vec<u8>);
                 ("ETag" = String, description = "Entity tag of the stored object")
             )),
             (status = 400, description = "Invalid filename"),
+            (status = 403, description = "Agent not connected, or request not from its address"),
             (status = 404, description = "Agent not found")
         )
     )
@@ -124,16 +126,10 @@ where
 {
     let (agent_id, filename) = agent_and_filename(&req)?;
 
-    // An agent this hub has never seen may be registered on another hub sharing its backplane, and uploading here.
-    let registered_here = registry.agents.lock()?.contains_key(&agent_id);
-    let registered = registered_here
-        || match req.app_data::<web::Data<Backplane>>() {
-            Some(backplane) => backplane.knows(&agent_id).await?,
-            None => false,
-        };
-    if !registered {
-        return Err(StorageError::AgentNotFound);
-    }
+    // The socket peer, as the hub takes an agent's address from when it connects, so the two compare like for like.
+    let peer_ip = req.peer_addr().ok_or(StorageError::NoPeerAddress)?.ip();
+    let backplane = req.app_data::<web::Data<Backplane>>().cloned().unwrap_or_default();
+    require_bucket_owner(&backplane, &registry, &agent_id, peer_ip).await?;
 
     // The body is buffered rather than streamed to the store.
     // `ObjectStore::put` takes a whole payload, and the alternative (`put_multipart`) buys streaming at the
@@ -144,6 +140,8 @@ where
     while let Some(chunk) = payload.next().await {
         body.extend_from_slice(&chunk?);
     }
+    // Asked again now the body is in, since the agent may have disconnected or moved while it streamed.
+    require_bucket_owner(&backplane, &registry, &agent_id, peer_ip).await?;
 
     let path = object_path(&agent_id, &filename);
     info!("Agent {} storing file: {}", agent_id, path);
@@ -157,6 +155,58 @@ where
     Ok(response.finish())
 }
 
+/// Fail unless `peer_ip` is the address `agent_id` is connected from, the one its bucket accepts writes from.
+async fn require_bucket_owner<S>(
+    backplane: &Backplane,
+    registry: &AgentRegistry<S>,
+    agent_id: &str,
+    peer_ip: IpAddr,
+) -> Result<(), StorageError>
+where
+    S: Send,
+{
+    let owner_ip = connected_owner_ip(backplane, registry, agent_id).await?;
+    if peer_ip.to_canonical() == owner_ip.to_canonical() {
+        Ok(())
+    } else {
+        Err(StorageError::RequesterNotBucketOwner)
+    }
+}
+
+/// The address `agent_id` is connected from.
+///
+/// A socket open on this hub is current by definition, so a local record holding a live session answers first.
+/// Otherwise the shared directory decides, since the agent may be connected through another hub, or may have moved to
+/// one since its record here was written. A local record that says connected with no session behind it -- one
+/// restored from a saved registry -- is not proof of a connection, and a hub with no backplane treats it as offline.
+#[expect(
+    clippy::single_call_fn,
+    reason = "the ownership lookup of require_bucket_owner, kept apart from the address comparison"
+)]
+async fn connected_owner_ip<S>(
+    backplane: &Backplane,
+    registry: &AgentRegistry<S>,
+    agent_id: &str,
+) -> Result<IpAddr, StorageError>
+where
+    S: Send,
+{
+    let local = registry
+        .agents
+        .lock()?
+        .get(agent_id)
+        .map(|record| (record.state.clone(), record.last_known_ip, record.session.is_some()));
+    if let Some((AgentConnectionState::Connected, ip, true)) = local {
+        return Ok(ip);
+    }
+    match backplane.presence(agent_id).await? {
+        Presence::Connected { last_known_ip, .. } => Ok(last_known_ip),
+        Presence::Disconnected => Err(StorageError::AgentNotConnected),
+        Presence::Unknown if local.is_some() => Err(StorageError::AgentNotConnected),
+        Presence::Unknown => Err(StorageError::AgentNotFound),
+    }
+}
+
 /// Log a stored image, and show it on the terminal when the `tty-image` feature is on.
 ///
 /// `put_file` is the only path a file reaches storage through, so it is where every stored file can be watched
diff --git a/services/storage/tests/put.rs b/services/storage/tests/put.rs
index 5a6387ad..1c9cc2da 100644
--- a/services/storage/tests/put.rs
+++ b/services/storage/tests/put.rs
@@ -1,6 +1,6 @@
 //! Integration tests for the `PUT /storage/{agent_id}/{filename}` route,
 //! focused on the `StorageError` paths (`InvalidFilename` -> 400,
-//! `AgentNotFound` -> 404, `Io` -> 500) plus the happy path.
+//! `AgentNotFound` -> 404, ownership -> 403, `Io` -> 500) plus the happy path.
 //!
 //! Each test brings up a fresh `App` with a tempdir-backed `StorageConfig`
 //! and an `AgentRegistry` it controls explicitly, then hits the route via
@@ -10,50 +10,164 @@
 #![cfg(test)]
 
 use std::collections::BTreeMap;
-use std::net::Ipv4Addr;
+use std::net::{Ipv4Addr, SocketAddr};
+use std::pin::Pin;
 
-use actix_web::dev::Payload as DevPayload;
-use actix_web::error::ResponseError as _;
+use actix_web::dev::{Payload as DevPayload, ServiceResponse};
+use actix_web::error::{PayloadError, ResponseError as _};
 use actix_web::http::{Method, StatusCode, header};
 use actix_web::{App, FromRequest as _, test, web};
 use edge_toolkit::ws::AgentConnectionState;
 use edge_toolkit::ws_server::{AgentRecord, AgentRegistry};
 use et_storage_service::{StorageConfig, StorageError, build_store, configure, put_file};
+use futures_util::Stream;
 use tempfile::TempDir;
 
-/// Build a registry with a single connected agent.
-fn registry_with_agent(agent_id: &str) -> AgentRegistry<()> {
+/// Build a registry holding only `agent_id`, in `state` at the loopback address, with `session` behind it.
+fn registry_with_record(agent_id: &str, state: AgentConnectionState, session: Option<()>) -> AgentRegistry<()> {
     let mut agents = BTreeMap::new();
     let _previous: Option<AgentRecord<()>> = agents.insert(
         agent_id.to_string(),
-        AgentRecord::new(AgentConnectionState::Connected, Ipv4Addr::LOCALHOST.into(), Some(())),
+        AgentRecord::new(state, Ipv4Addr::LOCALHOST.into(), session),
     );
     AgentRegistry::from_agents(agents)
 }
 
+/// Build a registry with a single connected agent.
+fn registry_with_agent(agent_id: &str) -> AgentRegistry<()> {
+    registry_with_record(agent_id, AgentConnectionState::Connected, Some(()))
+}
+
 fn storage_config(tmp: &TempDir) -> StorageConfig {
     StorageConfig::local(tmp.path())
 }
 
-#[actix_rt::test]
-async fn rejects_unknown_agent_with_404() {
-    let tmp = tempfile::tempdir().unwrap();
-    let config = storage_config(&tmp);
+/// A PUT of `uri` arriving from the address [`registry_with_agent`] registers its agent at.
+fn put_from_agent(uri: &str) -> test::TestRequest {
+    test::TestRequest::put()
+        .uri(uri)
+        .peer_addr(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 40123))
+}
+
+/// Send `reqs` in turn to a fresh app over `registry` that stores under `tmp`, returning their responses.
+#[expect(
+    clippy::future_not_send,
+    reason = "actix test requests and services are Rc-backed and !Send; they run on actix's single-threaded runtime"
+)]
+async fn send<const N: usize>(
+    registry: AgentRegistry<()>,
+    tmp: &TempDir,
+    reqs: [test::TestRequest; N],
+) -> [ServiceResponse; N] {
+    let config = storage_config(tmp);
     let app = test::init_service(
         App::new()
-            .app_data(web::Data::new(AgentRegistry::<()>::default()))
+            .app_data(web::Data::new(registry))
             .app_data(web::Data::new(config.clone()))
             .configure(|cfg| configure::<()>(cfg, &config)),
     )
     .await;
+    let mut responses = Vec::with_capacity(N);
+    for req in reqs {
+        responses.push(test::call_service(&app, req.to_request()).await);
+    }
+    responses.try_into().unwrap()
+}
+
+/// Send `req` with a small body to a fresh app over `registry`, returning its status and the storage root.
+#[expect(clippy::future_not_send, reason = "awaits `send`, whose future is not Send")]
+async fn put_into(registry: AgentRegistry<()>, req: test::TestRequest) -> (StatusCode, TempDir) {
+    let tmp = tempfile::tempdir().unwrap();
+    let [resp] = send(registry, &tmp, [req.set_payload(b"hello".as_ref())]).await;
+    (resp.status(), tmp)
+}
+
+#[actix_rt::test]
+async fn rejects_unknown_agent_with_404() {
+    let (status, _tmp) = put_into(
+        AgentRegistry::default(),
+        put_from_agent("/storage/missing-agent/file.txt"),
+    )
+    .await;
+    assert_eq!(status, StatusCode::NOT_FOUND);
+}
 
+/// An agent's id is not enough to write to its bucket: the request must come from where the agent is connected.
+#[actix_rt::test]
+async fn rejects_a_peer_other_than_the_agents_address_with_403() {
     let req = test::TestRequest::put()
-        .uri("/storage/missing-agent/file.txt")
-        .set_payload(b"hello".as_ref())
-        .to_request();
-    let resp = test::call_service(&app, req).await;
+        .uri("/storage/agent-1/payload.txt")
+        .peer_addr(SocketAddr::new(Ipv4Addr::new(192, 0, 2, 9).into(), 40123));
+    let (status, tmp) = put_into(registry_with_agent("agent-1"), req).await;
 
-    assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert!(!tmp.path().join("agent-1").exists());
+}
+
+#[actix_rt::test]
+async fn rejects_a_disconnected_agent_with_403() {
+    let registry = registry_with_record("agent-1", AgentConnectionState::Disconnected, None);
+    let (status, tmp) = put_into(registry, put_from_agent("/storage/agent-1/payload.txt")).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert!(!tmp.path().join("agent-1").exists());
+}
+
+/// A record restored from a saved registry says connected but has no socket behind it, so it owns nothing.
+#[actix_rt::test]
+async fn rejects_a_connected_record_without_a_session_with_403() {
+    let registry = registry_with_record("agent-1", AgentConnectionState::Connected, None);
+    let (status, tmp) = put_into(registry, put_from_agent("/storage/agent-1/payload.txt")).await;
+
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert!(!tmp.path().join("agent-1").exists());
+}
+
+/// Ownership is asked again once the body is in, so an agent that disconnects while its upload streams loses it.
+#[actix_rt::test]
+async fn rejects_an_upload_whose_agent_disconnects_mid_body() {
+    let tmp = tempfile::tempdir().unwrap();
+    let config = storage_config(&tmp);
+    let registry = web::Data::new(registry_with_agent("agent-1"));
+    let http_req = put_from_agent("/storage/agent-1/payload.txt")
+        .param("agent_id", "agent-1")
+        .param("filename", "payload.txt")
+        .to_http_request();
+
+    let disconnecting = registry.clone();
+    let body = futures_util::stream::once(async move {
+        disconnecting.mark_disconnected("agent-1");
+        Ok(web::Bytes::from_static(b"hello"))
+    });
+    let body: Pin<Box<dyn Stream<Item = Result<web::Bytes, PayloadError>>>> = Box::pin(body);
+    let mut payload = DevPayload::from(body);
+    let payload = web::Payload::from_request(&http_req, &mut payload).await.unwrap();
+
+    let store = web::Data::<dyn object_store::ObjectStore>::from(build_store(&config).unwrap());
+    let err = put_file::<()>(http_req, payload, registry, store).await.unwrap_err();
+
+    assert!(matches!(err, StorageError::AgentNotConnected));
+    assert!(!tmp.path().join("agent-1").exists());
+}
+
+/// A dual-stack listener reports an IPv4 client as `::ffff:a.b.c.d`; it is still the same address.
+#[actix_rt::test]
+async fn accepts_the_agents_address_written_as_ipv4_mapped_ipv6() {
+    let mapped = Ipv4Addr::LOCALHOST.to_ipv6_mapped();
+    let req = test::TestRequest::put()
+        .uri("/storage/agent-1/payload.txt")
+        .peer_addr(SocketAddr::new(mapped.into(), 40123));
+    let (status, tmp) = put_into(registry_with_agent("agent-1"), req).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert!(tmp.path().join("agent-1").join("payload.txt").exists());
+}
+
+#[actix_rt::test]
+async fn refuses_a_request_without_a_peer_address_with_500() {
+    let req = test::TestRequest::put().uri("/storage/agent-1/payload.txt");
+    let (status, _tmp) = put_into(registry_with_agent("agent-1"), req).await;
+    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
 }
 
 /// Invokes the handler directly, since a multi-component filename can't otherwise reach it.
@@ -86,27 +200,12 @@ async fn rejects_multi_component_filename_with_400() {
 
 #[actix_rt::test]
 async fn writes_file_for_registered_agent() {
-    let tmp = tempfile::tempdir().unwrap();
-    let config = storage_config(&tmp);
-    let registry = registry_with_agent("agent-1");
-    let app = test::init_service(
-        App::new()
-            .app_data(web::Data::new(registry))
-            .app_data(web::Data::new(config.clone()))
-            .configure(|cfg| configure::<()>(cfg, &config)),
-    )
-    .await;
-
-    let body = b"hello from agent-1".as_ref();
-    let req = test::TestRequest::put()
-        .uri("/storage/agent-1/payload.txt")
-        .set_payload(body)
-        .to_request();
-    let resp = test::call_service(&app, req).await;
+    let req = put_from_agent("/storage/agent-1/payload.txt");
+    let (status, tmp) = put_into(registry_with_agent("agent-1"), req).await;
 
-    assert_eq!(resp.status(), StatusCode::OK);
+    assert_eq!(status, StatusCode::OK);
     let written = fs_err::read(tmp.path().join("agent-1").join("payload.txt")).unwrap();
-    assert_eq!(written, body);
+    assert_eq!(written, b"hello");
 }
 
 /// Storing an image must succeed even though tty rendering cannot, since it is a best-effort side effect.
@@ -117,22 +216,9 @@ async fn writes_file_for_registered_agent() {
 #[actix_rt::test]
 async fn stores_an_image_and_returns_200_even_though_tty_rendering_cannot_succeed_in_tests() {
     let tmp = tempfile::tempdir().unwrap();
-    let config = storage_config(&tmp);
-    let registry = registry_with_agent("agent-1");
-    let app = test::init_service(
-        App::new()
-            .app_data(web::Data::new(registry))
-            .app_data(web::Data::new(config.clone()))
-            .configure(|cfg| configure::<()>(cfg, &config)),
-    )
-    .await;
-
     let body = b"\x89PNG\r\n\x1a\nnot a real png, just image-extension-shaped bytes".as_ref();
-    let req = test::TestRequest::put()
-        .uri("/storage/agent-1/capture.png")
-        .set_payload(body)
-        .to_request();
-    let resp = test::call_service(&app, req).await;
+    let req = put_from_agent("/storage/agent-1/capture.png").set_payload(body);
+    let [resp] = send(registry_with_agent("agent-1"), &tmp, [req]).await;
 
     assert_eq!(resp.status(), StatusCode::OK);
     let written = fs_err::read(tmp.path().join("agent-1").join("capture.png")).unwrap();
@@ -148,41 +234,24 @@ async fn stores_an_image_and_returns_200_even_though_tty_rendering_cannot_succee
 #[actix_rt::test]
 async fn get_and_head_expose_etag_and_head_reports_size() {
     let tmp = tempfile::tempdir().unwrap();
-    let config = storage_config(&tmp);
-    let registry = registry_with_agent("agent-1");
-    let app = test::init_service(
-        App::new()
-            .app_data(web::Data::new(registry))
-            .app_data(web::Data::new(config.clone()))
-            .configure(|cfg| configure::<()>(cfg, &config)),
-    )
-    .await;
-
     let body = b"etag round-trip payload".as_ref();
-    let put = test::TestRequest::put()
-        .uri("/storage/agent-1/payload.txt")
-        .set_payload(body)
-        .to_request();
-    let put_resp = test::call_service(&app, put).await;
+    let put = put_from_agent("/storage/agent-1/payload.txt").set_payload(body);
+    let get = test::TestRequest::get().uri("/storage/agent-1/payload.txt");
+    let head = test::TestRequest::default()
+        .method(Method::HEAD)
+        .uri("/storage/agent-1/payload.txt");
+    let [put_resp, get_resp, head_resp] = send(registry_with_agent("agent-1"), &tmp, [put, get, head]).await;
+
     assert_eq!(put_resp.status(), StatusCode::OK);
     assert!(
         put_resp.headers().contains_key(header::ETAG),
         "PUT should return an ETag"
     );
 
-    let get = test::TestRequest::get()
-        .uri("/storage/agent-1/payload.txt")
-        .to_request();
-    let get_resp = test::call_service(&app, get).await;
     assert_eq!(get_resp.status(), StatusCode::OK);
     let get_etag = get_resp.headers().get(header::ETAG).cloned();
     assert!(get_etag.is_some(), "GET should return an ETag");
 
-    let head = test::TestRequest::default()
-        .method(Method::HEAD)
-        .uri("/storage/agent-1/payload.txt")
-        .to_request();
-    let head_resp = test::call_service(&app, head).await;
     assert_eq!(head_resp.status(), StatusCode::OK);
     assert_eq!(
         head_resp.headers().get(header::ETAG).cloned(),
@@ -203,21 +272,10 @@ async fn get_and_head_expose_etag_and_head_reports_size() {
 #[actix_rt::test]
 async fn head_missing_object_returns_404() {
     let tmp = tempfile::tempdir().unwrap();
-    let config = storage_config(&tmp);
-    let registry = registry_with_agent("agent-1");
-    let app = test::init_service(
-        App::new()
-            .app_data(web::Data::new(registry))
-            .app_data(web::Data::new(config.clone()))
-            .configure(|cfg| configure::<()>(cfg, &config)),
-    )
-    .await;
-
     let head = test::TestRequest::default()
         .method(Method::HEAD)
-        .uri("/storage/agent-1/absent.txt")
-        .to_request();
-    let resp = test::call_service(&app, head).await;
+        .uri("/storage/agent-1/absent.txt");
+    let [resp] = send(registry_with_agent("agent-1"), &tmp, [head]).await;
     assert_eq!(resp.status(), StatusCode::NOT_FOUND);
 }
 
@@ -232,21 +290,8 @@ async fn surfaces_io_failure_as_500() {
     // Blocking one level down keeps the test on the request path it is actually about.
     let tmp = tempfile::tempdir().unwrap();
     fs_err::write(tmp.path().join("agent-1"), b"i am a file, not a directory").unwrap();
-    let config = StorageConfig::local(tmp.path());
-    let registry = registry_with_agent("agent-1");
-    let app = test::init_service(
-        App::new()
-            .app_data(web::Data::new(registry))
-            .app_data(web::Data::new(config.clone()))
-            .configure(|cfg| configure::<()>(cfg, &config)),
-    )
-    .await;
-
-    let req = test::TestRequest::put()
-        .uri("/storage/agent-1/payload.txt")
-        .set_payload(b"hello".as_ref())
-        .to_request();
-    let resp = test::call_service(&app, req).await;
+    let req = put_from_agent("/storage/agent-1/payload.txt").set_payload(b"hello".as_ref());
+    let [resp] = send(registry_with_agent("agent-1"), &tmp, [req]).await;
 
     assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
 }
diff --git a/services/storage/tests/s3_backend.rs b/services/storage/tests/s3_backend.rs
index 068d8f2b..ebb9b864 100644
--- a/services/storage/tests/s3_backend.rs
+++ b/services/storage/tests/s3_backend.rs
@@ -21,7 +21,7 @@
 #![cfg(test)]
 
 use std::collections::BTreeMap;
-use std::net::Ipv4Addr;
+use std::net::{Ipv4Addr, SocketAddr};
 use std::path::Path;
 use std::process::{Command, Stdio};
 
@@ -162,6 +162,7 @@ async fn round_trips_put_and_get_through_the_s3_backend() {
 
         let put = test::TestRequest::put()
             .uri(&format!("/storage/{AGENT}/{FILENAME}"))
+            .peer_addr(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 40123))
             .set_payload(BODY)
             .to_request();
         let put_status = test::call_service(&app, put).await.status();
diff --git a/services/ws-modules/face-detection/src/lib.rs b/services/ws-modules/face-detection/src/lib.rs
index 9f0f34e2..7f1242f3 100644
--- a/services/ws-modules/face-detection/src/lib.rs
+++ b/services/ws-modules/face-detection/src/lib.rs
@@ -426,8 +426,8 @@ fn update_face_status(input_name: &str, output_names: &[String], summary: &Detec
 
 /// Build the status-panel lines for a detection summary.
 ///
-/// Pure (no DOM/interop calls) and parameterised on plain values, so the host-side test can cover
-/// the formatting -- including the blank separator ahead of the best-box section.
+/// Pure (no DOM/interop calls) and parameterised on plain values, so the formatting -- including the blank
+/// separator ahead of the best-box section -- runs off the browser too.
 #[must_use]
 pub fn face_status_lines(
     input_name: &str,
diff --git a/services/ws-modules/har1/src/lib.rs b/services/ws-modules/har1/src/lib.rs
index 1ddf09fa..e248ecc9 100644
--- a/services/ws-modules/har1/src/lib.rs
+++ b/services/ws-modules/har1/src/lib.rs
@@ -513,8 +513,8 @@ async fn run_inner(client: &WsClient, sensors: &mut DeviceSensors) -> Result<(),
 
 /// Append a blank-line-separated section header to the status-panel lines.
 ///
-/// Pure (no DOM/interop calls), so the host-side test can cover the separator formatting the
-/// sensor display builds its orientation/motion sections through.
+/// Pure (no DOM/interop calls), so the separator formatting the sensor display builds its orientation/motion
+/// sections through runs off the browser too.
 pub fn push_section(lines: &mut Vec<String>, title: &str) {
     lines.push(String::default());
     lines.push(String::from(title));
diff --git a/services/ws-modules/math1-sender/build.rs b/services/ws-modules/math1-sender/build.rs
index 81f8bcec..97a5d5f8 100644
--- a/services/ws-modules/math1-sender/build.rs
+++ b/services/ws-modules/math1-sender/build.rs
@@ -1,5 +1,5 @@
 //! Emit `ET_MATH1_INPUT_PATH` so `include_str!` in `src/lib.rs` embeds the canonical math1 input -- the same
-//! bytes every math1 test harness injects.
+//! bytes every math1 harness injects.
 
 fn main() {
     et_path::emit_repo_file_env("ET_MATH1_INPUT_PATH", "services/ws-test-server/data/math1-input.json");
diff --git a/services/ws-modules/math1-sender/src/lib.rs b/services/ws-modules/math1-sender/src/lib.rs
index 789de7f3..972b05e4 100644
--- a/services/ws-modules/math1-sender/src/lib.rs
+++ b/services/ws-modules/math1-sender/src/lib.rs
@@ -2,7 +2,7 @@
 //!
 //! Plays the fake-agent side of the math1 storage exchange from a browser: uploads the canonical
 //! input JSON (embedded at build time from ws-test-server's `data/math1-input.json`, the same bytes
-//! every test harness injects) into this agent's own storage bucket, then broadcasts the
+//! every harness injects) into this agent's own storage bucket, then broadcasts the
 //! `math1-input` pointer once a second for a minute so math1 twins started in other tabs (or on
 //! other machines connected to the same hub) pick it up, compute, and store their models. This
 //! module only sends; each twin's stored `math1-output.json` is the operator's to inspect.
@@ -20,7 +20,7 @@ use wasm_bindgen::prelude::*;
 
 /// The canonical input bytes, embedded from the committed file so there is one source of truth.
 const MATH1_INPUT_JSON: &str = include_str!(env!("ET_MATH1_INPUT_PATH"));
-/// Storage object name the twins' broadcast pointer names, matching the test harnesses.
+/// Storage object name the twins' broadcast pointer names, matching the harnesses.
 const INPUT_FILENAME: &str = "math1-input.json";
 /// How many one-second-spaced pointer broadcasts to send before completing.
 const BROADCASTS: u32 = 60;
diff --git a/services/ws-modules/math1/src/lib.rs b/services/ws-modules/math1/src/lib.rs
index fc6b4501..9dfb1809 100644
--- a/services/ws-modules/math1/src/lib.rs
+++ b/services/ws-modules/math1/src/lib.rs
@@ -5,7 +5,7 @@
 //! This module waits for that pointer, reads the input from storage, runs the `FedAvg` kernel --
 //! rounds of local full-batch gradient-descent epochs per client merged with a sample-count-weighted
 //! average, only `+ - * /` on f64 so the result is bit-identical on every IEEE-754 runtime -- and
-//! stores the global model to `math1-output.json` in its own bucket, where the test harness reads
+//! stores the global model to `math1-output.json` in its own bucket, where the harness reads
 //! and verifies it. The math1 twins in the other guest languages run the same protocol and kernel.
 
 #![expect(
diff --git a/services/ws-modules/pic-viewer/build.rs b/services/ws-modules/pic-viewer/build.rs
index 12fc4f4f..9bff7bb3 100644
--- a/services/ws-modules/pic-viewer/build.rs
+++ b/services/ws-modules/pic-viewer/build.rs
@@ -1,6 +1,6 @@
-//! Stage the browser tests' image fixture (the ws-server page favicon) into `OUT_DIR`.
+//! Stage the `show_image` harness's sample image (the ws-server page favicon) into `OUT_DIR`.
 //!
-//! The `show_image` browser tests embed a real PNG via `include_bytes!`, and the repo's
+//! The `show_image` harness embeds a real PNG via `include_bytes!`, and the repo's
 //! `no-relative-path-literal` rule (rightly) forbids reaching it through a `../..` literal. Copying it here,
 //! anchored on `et_path::find_project_root_from_manifest()`, keeps the checked-in favicon as the single
 //! source of truth while giving the macro a stable `OUT_DIR` path to include.
diff --git a/services/ws-modules/pymath1/pymath1/__init__.py b/services/ws-modules/pymath1/pymath1/__init__.py
index d198742a..444c1505 100644
--- a/services/ws-modules/pymath1/pymath1/__init__.py
+++ b/services/ws-modules/pymath1/pymath1/__init__.py
@@ -1,7 +1,7 @@
 """math1 twin in Python (Pyodide): a storage-driven FedAvg simulation.
 
 The JS shim hands this module the raw input JSON it fetched from ws-server storage (client
-datasets + hyperparameters, injected by the test harness's fake agent). fed_avg() runs rounds of
+datasets + hyperparameters, injected by the harness's fake agent). fed_avg() runs rounds of
 local full-batch gradient-descent epochs per client and merges the local models with a
 sample-count-weighted average. Only + - * / on floats (no math-module calls), so the result is
 bit-identical to the other math1 language twins. run() returns the output JSON the shim stores to
diff --git a/services/ws-modules/wasi-comm1/src/lib.rs b/services/ws-modules/wasi-comm1/src/lib.rs
index 2547a08f..c979e38a 100644
--- a/services/ws-modules/wasi-comm1/src/lib.rs
+++ b/services/ws-modules/wasi-comm1/src/lib.rs
@@ -2,7 +2,7 @@
 //!
 //! Browser comm1 (`services/ws-modules/comm1`) waits for a second agent to
 //! be connected, then exchanges broadcast and direct messages with it. The
-//! integration test only spins up a single runner, so the WASI port instead
+//! runner it is exercised under is alone on its hub, so the WASI port instead
 //! exercises the message round-trip with the server itself:
 //!   1. Connect and capture our `agent_id`.
 //!   2. Send `list-agents`, recv a `list-agents-response`, assert the list
diff --git a/services/ws-modules/wasi-math1-sender/build.rs b/services/ws-modules/wasi-math1-sender/build.rs
index 81f8bcec..97a5d5f8 100644
--- a/services/ws-modules/wasi-math1-sender/build.rs
+++ b/services/ws-modules/wasi-math1-sender/build.rs
@@ -1,5 +1,5 @@
 //! Emit `ET_MATH1_INPUT_PATH` so `include_str!` in `src/lib.rs` embeds the canonical math1 input -- the same
-//! bytes every math1 test harness injects.
+//! bytes every math1 harness injects.
 
 fn main() {
     et_path::emit_repo_file_env("ET_MATH1_INPUT_PATH", "services/ws-test-server/data/math1-input.json");
diff --git a/services/ws-modules/wasi-math1-sender/src/lib.rs b/services/ws-modules/wasi-math1-sender/src/lib.rs
index a633ba35..1c29824e 100644
--- a/services/ws-modules/wasi-math1-sender/src/lib.rs
+++ b/services/ws-modules/wasi-math1-sender/src/lib.rs
@@ -1,7 +1,7 @@
 //! WASI Preview 2 trigger for the math1 family: the native counterpart of the browser `math1-sender`.
 //!
 //! Plays the fake-agent side of the math1 storage exchange under wasmtime: uploads the canonical input JSON
-//! (embedded at build time from ws-test-server's `data/math1-input.json`, the same bytes every test harness
+//! (embedded at build time from ws-test-server's `data/math1-input.json`, the same bytes every harness
 //! injects) into this agent's own storage bucket, then broadcasts the `math1-input` pointer once a second so
 //! math1 twins pick it up, compute, and store their models. This module only sends; each twin's stored
 //! `math1-output.json` is the harness's (or the operator's) to inspect.
@@ -34,7 +34,7 @@ const LOG_CONTEXT: &str = env!("CARGO_PKG_NAME");
 /// The canonical input bytes, embedded from the committed file so there is one source of truth.
 const MATH1_INPUT_JSON: &str = include_str!(env!("ET_MATH1_INPUT_PATH"));
 
-/// Storage object name the twins' broadcast pointer names, matching the test harnesses.
+/// Storage object name the twins' broadcast pointer names, matching the harnesses.
 const INPUT_FILENAME: &str = "math1-input.json";
 
 /// How many one-second-spaced pointer broadcasts to send before completing.
diff --git a/services/ws-modules/wasi-math1/src/lib.rs b/services/ws-modules/wasi-math1/src/lib.rs
index b518e5f4..7e406fc8 100644
--- a/services/ws-modules/wasi-math1/src/lib.rs
+++ b/services/ws-modules/wasi-math1/src/lib.rs
@@ -4,7 +4,7 @@
 //! broadcast `math1-input` pointer (relayed to the guest through `ws.recv`), reads the input JSON
 //! (client datasets + hyperparameters) from storage via `wasi:keyvalue/store`, runs the kernel --
 //! only `+ - * /` on f64 in a fixed evaluation order, bit-identical to the browser twins -- and
-//! stores the global model to `math1-output.json` in its own bucket, where the test harness reads
+//! stores the global model to `math1-output.json` in its own bucket, where the harness reads
 //! and verifies it.
 //!
 //! Crate-level cfg gate: the wit-bindgen-generated extern declarations
diff --git a/services/ws-server/HELP.md b/services/ws-server/HELP.md
index 57f25ec9..4d4d1882 100644
--- a/services/ws-server/HELP.md
+++ b/services/ws-server/HELP.md
@@ -234,3 +234,13 @@ Largest single WebSocket frame the hub will accept.
 Frames above this are dropped by actix-ws before they reach the handler, so callers shipping big tensors / blobs need to
 raise it above their payload size. `WS_MAX_FRAME_SIZE` takes a human byte size (e.g. `64MiB`, `64MB`, `512KiB`) or a
 plain byte count; unset defaults to 64 MiB.
+
+### `WS_MESSAGE_RETENTION`
+
+Type: string. Default: `latest-per-sender`.
+
+Which unacknowledged direct messages a recipient holds: `latest-per-sender` (the default) or `all`.
+
+`latest-per-sender` keeps only the newest message from each sender, replacing the earlier one, so a recipient holds at
+most one message per sender and the hub's memory for them is bounded. `all` keeps every message until it is
+acknowledged, with no bound on how many pile up for a recipient that stays away.
diff --git a/services/ws-test-server/tests/horizontal_scaling.rs b/services/ws-test-server/tests/horizontal_scaling.rs
index b81dc52c..0c0b582f 100644
--- a/services/ws-test-server/tests/horizontal_scaling.rs
+++ b/services/ws-test-server/tests/horizontal_scaling.rs
@@ -350,6 +350,18 @@ async fn storage_accepts_an_upload_through_a_hub_the_agent_never_joined() {
     assert_eq!(stranger.status(), reqwest::StatusCode::NOT_FOUND);
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn storage_refuses_an_upload_for_an_agent_that_left_another_hub() {
+    let cluster = cluster(2);
+    let [(alice, alice_id), (mut bob, _bob_id)] = <[_; 2]>::try_from(one_agent_per_hub(&cluster).await).unwrap();
+    drop(alice);
+    wait_for_state(&mut bob, &alice_id, &AgentConnectionState::Disconnected).await;
+
+    let put_url = format!("{}/storage/{alice_id}/result.txt", cluster.hubs[1].base_url);
+    let put = reqwest::Client::new().put(put_url).body("42").send().await.unwrap();
+    assert_eq!(put.status(), reqwest::StatusCode::FORBIDDEN);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn thirty_agents_spread_over_three_hubs_all_reach_each_other() {
     const AGENTS: usize = 30;
diff --git a/services/ws-wasi-runner/src/host/ws.rs b/services/ws-wasi-runner/src/host/ws.rs
index 89395033..aecfc9ed 100644
--- a/services/ws-wasi-runner/src/host/ws.rs
+++ b/services/ws-wasi-runner/src/host/ws.rs
@@ -158,7 +158,7 @@ impl WsBackend {
     /// Take the next inbound message, or `None` if none arrives within `timeout`.
     ///
     /// The guest-facing `recv` converts to WIT types on the way out; this is the same drain without that step, so a
-    /// test can assert on the `ServerMessage` the reader pump actually produced.
+    /// caller sees the `ServerMessage` the reader pump actually produced.
     pub async fn next_message(&self, timeout: Duration) -> Option<ServerMessage> {
         let mut rx = self.inbox.lock().await;
         tokio::time::timeout(timeout, rx.recv()).await.ok().flatten()
diff --git a/services/ws-web-runner/src/runtime.rs b/services/ws-web-runner/src/runtime.rs
index 75efb166..a167b56c 100644
--- a/services/ws-web-runner/src/runtime.rs
+++ b/services/ws-web-runner/src/runtime.rs
@@ -152,7 +152,7 @@ fn shim_js(http_base: &str, ws_url: &str, coverage: bool) -> String {
 /// that awaits the module's `default`/`run` export. Under the `coverage` feature the `finally` also captures the
 /// module's minicov `.profraw` via et-web's `__et_capture_coverage` export (browser wasm has no filesystem) and PUTs it
 /// to the module's own agent bucket -- a registered agent (its id sniffed from et-connect-ack by the ws-agent-id shim),
-/// so `put_file`'s agent check passes; the web-runner test then scans every bucket for the .profraw. The `finally`
+/// so `put_file`'s agent check passes; coverage collection then scans every bucket for the .profraw. The `finally`
 /// means fail-fast modules still dump what they exercised. Without the feature the capture is empty, leaving a bare
 /// `finally {}`.
 #[expect(
diff --git a/services/ws/src/connection.rs b/services/ws/src/connection.rs
index 5b0a7c2f..e21ce65c 100644
--- a/services/ws/src/connection.rs
+++ b/services/ws/src/connection.rs
@@ -7,7 +7,7 @@ use actix_ws::{AggregatedMessage, AggregatedMessageStream, CloseCode, CloseReaso
 use bytes::Bytes;
 use chrono::Utc;
 use edge_toolkit::ws::{ClientMessage, ConnectStatus, MessageDeliveryStatus, MessageScope, ServerMessage};
-use edge_toolkit::ws_server::PendingDirectMessage;
+use edge_toolkit::ws_server::{MessageRetention, PendingDirectMessage};
 use et_hub_backplane::{Backplane, BackplaneError, Frame, NatsBackplane, Presence};
 use futures_util::StreamExt as _;
 use opentelemetry::global;
@@ -32,6 +32,8 @@ pub(crate) struct Connection {
     outbox: AgentSession,
     /// Idle timeout for this connection, or `None` to never time out.
     idle_timeout: Option<Duration>,
+    /// Whether a direct message this agent sends replaces its earlier unacknowledged one to the same recipient.
+    retention: MessageRetention,
 }
 
 impl Connection {
@@ -43,6 +45,7 @@ impl Connection {
         session: Session,
         outbox: AgentSession,
         idle_timeout: Option<Duration>,
+        retention: MessageRetention,
     ) -> Self {
         info!("New WebSocket connection for client IP {}", client_ip);
         Self {
@@ -54,6 +57,7 @@ impl Connection {
             session,
             outbox,
             idle_timeout,
+            retention,
         }
     }
 
@@ -180,7 +184,7 @@ impl Connection {
             Utc::now().to_rfc3339(),
             message,
         );
-        if let Err(error) = nats.queue(&to_agent_id, pending.clone()).await {
+        if let Err(error) = nats.queue(&to_agent_id, pending.clone(), self.retention).await {
             error!("Backplane mailbox write for {to_agent_id} failed: {error}");
             self.send_backplane_unavailable(&error).await;
             return;
@@ -199,7 +203,7 @@ impl Connection {
             },
         };
 
-        if let Presence::Connected { hub_id } = destination {
+        if let Presence::Connected { hub_id, .. } = destination {
             self.deliver_live(nats, &hub_id, &to_agent_id, pending).await;
             return;
         }
@@ -307,7 +311,7 @@ impl Connection {
             detail: format!("agent {recipient_agent_id} acknowledged receipt"),
         };
         let delivered = match nats.presence(sender_agent_id).await {
-            Ok(Presence::Connected { hub_id }) => {
+            Ok(Presence::Connected { hub_id, .. }) => {
                 self.deliver_via_backplane(nats, &hub_id, sender_agent_id, receipt)
                     .await
             }
@@ -419,6 +423,7 @@ impl Connection {
             &to_agent_id,
             server_received_at,
             message,
+            self.retention,
         ) else {
             warn!("direct message target {to_agent_id} is not a connected agent");
             self.send_invalid(None, format!("unknown target agent {to_agent_id}"))
diff --git a/services/ws/src/lib.rs b/services/ws/src/lib.rs
index 875eab21..2a457555 100644
--- a/services/ws/src/lib.rs
+++ b/services/ws/src/lib.rs
@@ -11,6 +11,7 @@ use std::time::Duration;
 use actix_web::{Error, HttpRequest, HttpResponse, web};
 use bytes::Bytes;
 use edge_toolkit::ws::{CloudEvent, ServerMessage, client_event_type};
+pub use edge_toolkit::ws_server::MessageRetention;
 use edge_toolkit::ws_server::{AgentRecord, AgentRegistry, PendingDirectMessage, RegistryError};
 pub use et_hub_backplane::{Backplane, BackplaneConfig, BackplaneError};
 use et_hub_backplane::{Frame, Inbound};
@@ -98,6 +99,18 @@ pub struct WsConfig {
     #[cfg_attr(feature = "env-schema", schemars(with = "String", extend("default" = "local")))]
     pub backplane: BackplaneConfig,
 
+    /// Which unacknowledged direct messages a recipient holds: `latest-per-sender` (the default) or `all`.
+    ///
+    /// `latest-per-sender` keeps only the newest message from each sender, replacing the earlier one, so a recipient
+    /// holds at most one message per sender and the hub's memory for them is bounded. `all` keeps every message until
+    /// it is acknowledged, with no bound on how many pile up for a recipient that stays away.
+    #[serde(default)]
+    #[cfg_attr(
+        feature = "env-schema",
+        schemars(with = "String", extend("default" = "latest-per-sender"))
+    )]
+    pub message_retention: MessageRetention,
+
     /// The YAML file a local hub loads its agent registry from at startup and saves it to on shutdown.
     #[serde_inline_default(PathBuf::from("registry.yaml"))]
     #[cfg_attr(feature = "env-schema", schemars(extend("default" = "registry.yaml")))]
@@ -259,10 +272,10 @@ pub async fn ws_handler(
         .max_frame_size(config.max_frame_size)
         .aggregate_continuations();
 
-    // An app that never joined a backplane -- most tests, and any embedding that predates them -- is a local hub.
     let backplane = req
         .app_data::<web::Data<Backplane>>()
-        .map_or(Backplane::Local, |backplane| backplane.get_ref().clone());
+        .map(|data| data.get_ref().clone())
+        .unwrap_or_default();
 
     let (tx, rx) = mpsc::unbounded_channel::<SessionMessage>();
     let conn = Connection::new(
@@ -272,6 +285,7 @@ pub async fn ws_handler(
         session,
         tx,
         config.connection_timeout,
+        config.message_retention,
     );
 
     let _join = actix_web::rt::spawn(async move {
diff --git a/services/ws/tests/config.rs b/services/ws/tests/config.rs
index 38f8bca1..a1f3f3e1 100644
--- a/services/ws/tests/config.rs
+++ b/services/ws/tests/config.rs
@@ -1,6 +1,6 @@
 //! `WS_*` deserialization, through the real `serde_env::from_env` path the ws-server uses, with the process env set
 //! via `temp-env`. Verifies the human-byte-size parsing (`64MiB`, plain byte counts), the idle timeout, the backplane
-//! choice, and each default when its variable is absent.
+//! and message-retention choices, and each default when its variable is absent.
 #![cfg(test)]
 #![expect(
     clippy::decimal_literal_representation,
@@ -8,7 +8,7 @@
 )]
 
 use et_test_helpers::temp_env;
-use et_ws_service::{BackplaneConfig, WsConfig};
+use et_ws_service::{BackplaneConfig, MessageRetention, WsConfig};
 use serde::Deserialize;
 
 // Mirror the ws-server's nesting so the env key is `WS_MAX_FRAME_SIZE`.
@@ -71,6 +71,20 @@ fn backplane_absent_defaults_to_local() {
     });
 }
 
+#[test]
+fn message_retention_absent_defaults_to_latest_per_sender() {
+    temp_env::with_var_unset("WS_MESSAGE_RETENTION", || {
+        assert_eq!(load().message_retention, MessageRetention::LatestPerSender);
+    });
+}
+
+#[test]
+fn message_retention_parses_all() {
+    temp_env::with_var("WS_MESSAGE_RETENTION", Some("all"), || {
+        assert_eq!(load().message_retention, MessageRetention::All);
+    });
+}
+
 #[test]
 fn backplane_parses_a_nats_url() {
     temp_env::with_var("WS_BACKPLANE", Some("nats://broker:4222"), || {
diff --git a/services/ws/tests/load_registry.rs b/services/ws/tests/load_registry.rs
index 8f4e73a3..92f03599 100644
--- a/services/ws/tests/load_registry.rs
+++ b/services/ws/tests/load_registry.rs
@@ -6,7 +6,7 @@
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
 use edge_toolkit::ws::{AgentConnectionState, AgentSummary, ConnectStatus};
-use edge_toolkit::ws_server::RegistryError;
+use edge_toolkit::ws_server::{MessageRetention, RegistryError};
 use et_ws_service::{WsAgentRegistry, load_registry};
 use tempfile::tempdir;
 
@@ -47,6 +47,7 @@ fn round_trips_state_ip_and_pending_messages_but_not_sessions() {
         "agent-2",
         "2026-07-22T00:00:00Z".to_string(),
         serde_json::json!({"hello": "world"}),
+        MessageRetention::default(),
     );
     assert!(queued.is_some(), "agent-2 must exist to receive the queued message");
     registry.mark_disconnected("agent-2");
diff --git a/utilities/cli/src/deployment_types/k3s.rs b/utilities/cli/src/deployment_types/k3s.rs
index 54607891..ba3fdc9d 100644
--- a/utilities/cli/src/deployment_types/k3s.rs
+++ b/utilities/cli/src/deployment_types/k3s.rs
@@ -64,7 +64,7 @@ use crate::{
 /// `artifact_source` a scenario asks for. A locally built image never reaches a registry at all, so an `Always`
 /// pull would send k3s looking for it in one that does not have it. A published image is there to be found, and
 /// `IfNotPresent` is what stops every restart re-pulling a tag the node already holds. It also lets a published-image
-/// scenario be tested against a local build of the same tag, which is how CI checks a change to an image before it
+/// scenario run against a local build of the same tag, which is how CI checks a change to an image before it
 /// is published.
 pub(super) const IMAGE_PULL_POLICY: &str = "IfNotPresent";
 
diff --git a/utilities/cli/src/deployment_types/mise.rs b/utilities/cli/src/deployment_types/mise.rs
index 6125bf76..f1e18894 100644
--- a/utilities/cli/src/deployment_types/mise.rs
+++ b/utilities/cli/src/deployment_types/mise.rs
@@ -133,9 +133,9 @@ fn insert_entry_tasks(tasks: &mut Table, cluster: &ClusterInput, runners: &[Runn
 
 /// Add one `machine-<name>` task per machine, each starting what that machine runs and nothing else.
 ///
-/// The server's also starts the collector and the hub, and any other machine running a hub starts its own.
-/// `generated-scenario` stays alongside them, still starting everything, because the whole cluster on one host remains
-/// the quickest way to run it.
+/// The server's also starts the collector and the hub, and any other machine running a hub starts its own. `generated-
+/// scenario` stays alongside them, still starting everything, because the whole cluster on one host remains the
+/// quickest way to run it.
 #[expect(clippy::single_call_fn, reason = "the per-machine step of insert_entry_tasks")]
 fn insert_machine_tasks(tasks: &mut Table, cluster: &ClusterInput, runners: &[RunnerInstance]) {
     let hubs = hubs(cluster);
@@ -361,7 +361,8 @@ fn ws_server_run_body(
 /// layout (`services/ws-server/storage`) under whatever it takes for the project root -- outside this repository,
 /// the directory it was started in. A published hub starts in the deployment directory, so the store is named there
 /// directly, as the compose file does with its `/app/storage` volume. A local hub keeps the default, which is where
-/// this repository's tests read agent output from. Hubs that `shared` storage are pointed at it by the caller instead.
+/// this repository's harnesses read agent output from. Hubs that `shared` storage are pointed at it by the caller
+/// instead.
 #[expect(
     clippy::single_call_fn,
     reason = "distinct step of generate_mise_deployment, split out to keep it readable"
diff --git a/utilities/cli/src/lib.rs b/utilities/cli/src/lib.rs
index b852d2f9..729f0e61 100644
--- a/utilities/cli/src/lib.rs
+++ b/utilities/cli/src/lib.rs
@@ -613,8 +613,8 @@ fn generate_format(
             generate_docker_compose_deployment(cluster, output_dir)?;
             generate_scenario_image(cluster, output_dir)
         }
-        // The scenario image is emitted here too, not just for compose. Both formats reference it: compose builds it as
-        // a service, and the manifests name it as the hub's image with the README giving the `docker build` for it.
+        // The scenario image is emitted here too, not just for compose. Both formats reference it: compose builds it
+        // as a service, and the manifests name it as the hub's image with the README giving the `docker build` for it.
         // Generating k3s alone without it produced a README pointing at a Dockerfile that was never written.
         OutputType::K3s => {
             generate_k3s_deployment(cluster, output_dir)?;
@@ -669,8 +669,8 @@ fn write_readme(cluster: &ClusterInput, output_dir: &Path, output_types: &[Outpu
 /// path, which is what the command needs, since it runs from the repository root.
 ///
 /// A single-component output directory has a parent, and it is the empty path rather than `None` -- so this cannot lean
-/// on `unwrap_or` and has to test the rendered parent. Prefixing an empty one would produce `/$scenario/Dockerfile`, an
-/// absolute path to a directory nobody has.
+/// on `unwrap_or` and has to check the rendered parent. Prefixing an empty one would produce `/$scenario/Dockerfile`,
+/// an absolute path to a directory nobody has.
 #[must_use]
 pub fn scenario_dockerfile_path(output_dir: &Path) -> String {
     let parent = output_dir
@@ -735,10 +735,9 @@ fn write_deployment_gitignore(output_dir: &Path) -> Result<(), CliError> {
 /// takes `ZO_ROOT_USER_PASSWORD` as its root credential, and the ws-server authenticates its OTLP exports with
 /// `OTLP_AUTH_PASSWORD`. A cluster of several hubs adds the secret of the object store they share, under the name the
 /// hubs' S3 client reads and the one rustfs reads. Nothing else belongs here. The accounts these authenticate as are
-/// not secrets and each
-/// format states it outright, so keeping it in this file would have meant an uncommitted file standing between a reader
-/// and a value there was never any reason to withhold -- and, in the Kubernetes case, a `Secret` holding something that
-/// is not one.
+/// not secrets and each format states it outright, so keeping it in this file would have meant an uncommitted file
+/// standing between a reader and a value there was never any reason to withhold -- and, in the Kubernetes case, a
+/// `Secret` holding something that is not one.
 ///
 /// Each value carries a `skipcq` pragma for `DeepSource SCT-A000`: the copies committed under `verification/` are
 /// read as hardcoded credentials, and the path excludes that keep the rest of that tree out of analysis do not reach
@@ -999,8 +998,8 @@ where
 /// `NPM_CONFIG_USERCONFIG` is exported rather than left to the `[env]` beside it, which carries the same value. mise
 /// computes that entry but does not apply it to its own tool resolution, so the npm client it embeds reads no user
 /// config, falls back to registry.npmjs.org, and reports the scoped module packages as `package not found` -- they
-/// exist only on GitHub Packages. Exported into the process, the same file resolves against the right registry, for
-/// the lock and the install alike. The `[env]` entry stays because the tasks do get it.
+/// exist only on GitHub Packages. Exported into the process, the same file resolves against the right registry, for the
+/// lock and the install alike. The `[env]` entry stays because the tasks do get it.
 const fn mise_install_note(artifacts: ArtifactSource) -> &'static str {
     if matches!(artifacts, ArtifactSource::Published) {
         return concat!(
diff --git a/utilities/int-gen/Cargo.toml b/utilities/int-gen/Cargo.toml
index 0a46a770..294b67e2 100644
--- a/utilities/int-gen/Cargo.toml
+++ b/utilities/int-gen/Cargo.toml
@@ -30,6 +30,7 @@ et-ws-runner-common = { workspace = true, features = ["env-schema"] }
 et-ws-server = { workspace = true, features = ["env-schema", "openapi-spec"] }
 fs-err.workspace = true
 heck.workspace = true
+jiff.workspace = true
 kdl.workspace = true
 openapiv3.workspace = true
 pretty_yaml.workspace = true
diff --git a/utilities/int-gen/HELP.md b/utilities/int-gen/HELP.md
index f5903a3a..29736d2a 100644
--- a/utilities/int-gen/HELP.md
+++ b/utilities/int-gen/HELP.md
@@ -20,7 +20,7 @@ Generate checked-in artifacts (generated/, CHECKS.md, HELP.md) from in-repo sour
 
 - `generate` — Emit the generated artifacts for one target (default: all)
 - `fetch-deps` — Fetch upstream WASI WIT packages into generated/specs/wit/ at pinned versions
-- `checks` — Write CHECKS.md, the catalogue of every check across every `MISE_ENV`
+- `checks` — Write CHECKS.md, the catalogue of every check across every `MISE_ENV`, stamped with today's UTC date
 - `help-md` — Write each binary crate's HELP.md from its clap command tree and its environment config
 
 ## `et-int-gen generate`
@@ -55,13 +55,9 @@ Fetch upstream WASI WIT packages into generated/specs/wit/ at pinned versions
 
 ## `et-int-gen checks`
 
-Write CHECKS.md, the catalogue of every check across every `MISE_ENV`
+Write CHECKS.md, the catalogue of every check across every `MISE_ENV`, stamped with today's UTC date
 
-**Usage:** `et-int-gen checks [OPTIONS]`
-
-### Options
-
-- `--check` — Compare against the committed CHECKS.md and fail on drift instead of writing it
+**Usage:** `et-int-gen checks`
 
 ## `et-int-gen help-md`
 
diff --git a/utilities/int-gen/src/bin/int-gen.rs b/utilities/int-gen/src/bin/int-gen.rs
index b727deac..d647690a 100644
--- a/utilities/int-gen/src/bin/int-gen.rs
+++ b/utilities/int-gen/src/bin/int-gen.rs
@@ -5,8 +5,8 @@ use clap::Parser as _;
 use edge_toolkit::config::get_project_root;
 use et_int_gen::cli::{Cli, Command, Target};
 use et_int_gen::{
-    check_checks, check_help, generate, generate_bindings, generate_checks, generate_core, generate_help,
-    generate_rust, generate_zig, wit::upstream,
+    check_help, generate, generate_bindings, generate_checks, generate_core, generate_help, generate_rust,
+    generate_zig, wit::upstream,
 };
 
 fn main() -> Result<(), et_int_gen::Error> {
@@ -21,8 +21,7 @@ fn main() -> Result<(), et_int_gen::Error> {
             Target::All => generate(),
         },
         Command::FetchDeps => upstream::run(&get_project_root()),
-        Command::Checks { check: true } => check_checks(),
-        Command::Checks { check: false } => generate_checks(),
+        Command::Checks => generate_checks(),
         Command::HelpMd { check: true } => check_help(),
         Command::HelpMd { check: false } => generate_help(),
     }
diff --git a/utilities/int-gen/src/checks.rs b/utilities/int-gen/src/checks.rs
index c58ce169..87a965bd 100644
--- a/utilities/int-gen/src/checks.rs
+++ b/utilities/int-gen/src/checks.rs
@@ -163,11 +163,18 @@ pub fn render(root: &Path) -> Result<String, Error> {
         shell: shell.as_ref().map_or(0, Vec::len),
         tools: sections.len(),
     };
-    let mut out = header(&counts)?;
+    let today = jiff::Timestamp::now().to_zoned(jiff::tz::TimeZone::UTC).date();
+    let mut out = header(&counts, today)?;
     out.push_str(&body);
     Ok(out)
 }
 
+/// The header line recording the UTC day the document was generated, which the freshness policy reads.
+#[must_use]
+pub fn generated_stamp(date: jiff::civil::Date) -> String {
+    format!("- **Generated:** {date}")
+}
+
 /// The totals the document opens with.
 struct Counts {
     checks: usize,
@@ -217,8 +224,9 @@ fn render_section(
 }
 
 /// The document's title, its totals and the paragraph saying what it lists and how it is kept current.
-fn header(counts: &Counts) -> Result<String, Error> {
+fn header(counts: &Counts, generated: jiff::civil::Date) -> Result<String, Error> {
     let mut out = String::from("# Checks\n\n");
+    writeln!(out, "{}", generated_stamp(generated))?;
     writeln!(out, "- **Checks:** {}", counts.checks)?;
     writeln!(out, "- **Tools:** {}", counts.tools)?;
     writeln!(out, "- **Shell-script checks:** {}", counts.shell)?;
@@ -233,8 +241,8 @@ fn header(counts: &Counts) -> Result<String, Error> {
         "whose config declares it, the aggregate that runs it as part of `mise run check`, the `fix` or `fmt` pass",
         "that repairs what it reports, where one does, and the config files it reads. `mise run fix` runs every",
         "such pass. Generated by `mise run gen:checks` from the task tables in `.mise/config*.toml`, the rule",
-        "files under `config/` and the `[workspace.lints]` in `Cargo.toml`; `mise run checks-md-check` fails when",
-        "this file drifts from them.",
+        "files under `config/` and the `[workspace.lints]` in `Cargo.toml`. It is not held to them on every",
+        "change: `mise run checks-md-check` fails only once the generation date above is more than 7 days old.",
     ];
     out.push_str(&wrap("", "", &intro.join(" ")));
     Ok(out)
@@ -772,7 +780,7 @@ pub fn rule_sources(body: &str) -> Result<BTreeSet<String>, Error> {
 
 /// The rules a section's checks are configured with, and the deepest directory every rule file shares.
 ///
-/// Rule files are recognised by format rather than by tool: a `.rego` policy (its tests excepted), a `.schema.json`
+/// Rule files are recognised by format rather than by tool: a `.rego` policy (not a `_test.rego`), a `.schema.json`
 /// schema, and a YAML document that is a rule (`id` + `message`), lists rules (`rules:` as a sequence), or points at
 /// directories of them (`ruleDirs:`), and a TOML manifest's `[workspace.lints]` tables. Sources are the paths the
 /// checks pass as configuration plus the configs the section's tool finds by itself. A directory is searched
diff --git a/utilities/int-gen/src/cli.rs b/utilities/int-gen/src/cli.rs
index f3522694..597dbd12 100644
--- a/utilities/int-gen/src/cli.rs
+++ b/utilities/int-gen/src/cli.rs
@@ -24,12 +24,8 @@ pub enum Command {
     },
     /// Fetch upstream WASI WIT packages into generated/specs/wit/ at pinned versions.
     FetchDeps,
-    /// Write CHECKS.md, the catalogue of every check across every `MISE_ENV`.
-    Checks {
-        /// Compare against the committed CHECKS.md and fail on drift instead of writing it.
-        #[arg(long)]
-        check: bool,
-    },
+    /// Write CHECKS.md, the catalogue of every check across every `MISE_ENV`, stamped with today's UTC date.
+    Checks,
     /// Write each binary crate's HELP.md from its clap command tree and its environment config.
     HelpMd {
         /// Compare against the committed HELP.md files and fail on drift instead of writing them.
diff --git a/utilities/int-gen/src/error.rs b/utilities/int-gen/src/error.rs
index 9e45f4e5..fe00cb67 100644
--- a/utilities/int-gen/src/error.rs
+++ b/utilities/int-gen/src/error.rs
@@ -36,8 +36,6 @@ pub enum Error {
     // The message is carried as data rather than only in the `#[error]` string because the binary's `main` reports
     // errors through `Debug`, which would otherwise print the bare variant name.
     #[error("{0}")]
-    ChecksStale(&'static str),
-    #[error("{0}")]
     UndescribedRule(String),
     #[error("{0}")]
     DuplicateLinkLabel(String),
diff --git a/utilities/int-gen/src/help.rs b/utilities/int-gen/src/help.rs
index 48ccc2cc..cacee4a1 100644
--- a/utilities/int-gen/src/help.rs
+++ b/utilities/int-gen/src/help.rs
@@ -95,8 +95,8 @@ pub fn env_schemas() -> Result<BTreeMap<&'static str, Value>, Error> {
 /// Every workspace member with a documented binary target, read from the manifests and sorted by name.
 ///
 /// The manifests are read directly rather than through `cargo metadata`, which takes the workspace lock. Only normal
-/// dependencies count, target-specific ones included, so a crate using clap or serde-env in its tests alone is left
-/// out.
+/// dependencies count, target-specific ones included, so a crate using clap or serde-env in its dev-dependencies alone
+/// is left out.
 pub fn binary_crates(root: &Path) -> Result<Vec<BinaryCrate>, Error> {
     let workspace = read_manifest(&root.join("Cargo.toml"))?;
     let members = workspace
diff --git a/utilities/int-gen/src/lib.rs b/utilities/int-gen/src/lib.rs
index 3dda9beb..1e0ce634 100644
--- a/utilities/int-gen/src/lib.rs
+++ b/utilities/int-gen/src/lib.rs
@@ -71,19 +71,6 @@ pub fn generate_checks() -> Result<(), Error> {
     write_if_changed(&project_root.join("CHECKS.md"), &checks::render(&project_root)?)
 }
 
-/// Fail when the committed `CHECKS.md` differs from what [`generate_checks`] would write.
-pub fn check_checks() -> Result<(), Error> {
-    let project_root = get_project_root();
-    let committed = fs::read_to_string(project_root.join("CHECKS.md"))?;
-    if committed == checks::render(&project_root)? {
-        Ok(())
-    } else {
-        Err(Error::ChecksStale(
-            "CHECKS.md is out of date with the mise configs and rule files; run `mise run gen:checks`",
-        ))
-    }
-}
-
 /// Write each binary crate's `HELP.md` from its clap command tree and its environment config.
 pub fn generate_help() -> Result<(), Error> {
     let project_root = get_project_root();
diff --git a/utilities/int-gen/tests/checks.rs b/utilities/int-gen/tests/checks.rs
index 416d6498..4c3e69ab 100644
--- a/utilities/int-gen/tests/checks.rs
+++ b/utilities/int-gen/tests/checks.rs
@@ -6,7 +6,8 @@
 use std::collections::{BTreeMap, BTreeSet};
 
 use et_int_gen::checks::{
-    config_paths, link_label, lint_opposites, pair_repairs, rego_summaries, rule_sources, runner_label, wrap,
+    config_paths, generated_stamp, link_label, lint_opposites, pair_repairs, rego_summaries, rule_sources,
+    runner_label, wrap,
 };
 
 /// The set of paths a path-collecting function is expected to return.
@@ -242,3 +243,9 @@ fn a_pass_no_check_pairs_with_is_named() {
         format!("{passes_named}, which no check pairs with by name: add the check, or take the pass out")
     );
 }
+
+#[test]
+fn generated_stamp_writes_a_zero_padded_iso_date() {
+    let date = jiff::civil::date(2026, 3, 7);
+    assert_eq!(generated_stamp(date), "- **Generated:** 2026-03-07");
+}