diff --git a/.github/workflows/sandboxd.yml b/.github/workflows/sandboxd.yml
index 8a1c4ddc0..ab627b58b 100644
--- a/.github/workflows/sandboxd.yml
+++ b/.github/workflows/sandboxd.yml
@@ -73,7 +73,7 @@ jobs:
         env:
           SANDBOXD_TEST_PG_URL: postgres://postgres:sbxci@localhost:5432/postgres?sslmode=disable
         run: |
-          GOWORK=off go test -race -count=1 -v ./tenants/... ./poolset/... | tee "$RUNNER_TEMP/pg-test.log"
+          GOWORK=off go test -race -count=1 -v ./tenants/... ./poolset/... ./archives/... ./usage/... | tee "$RUNNER_TEMP/pg-test.log"
           GOWORK=off go test -race -count=1 -v -run 'TestTwoNodesOnOneMetaStore|TestNodesOfACell|TestAMetaStoreNodeWithoutACell' ./pool/ | tee -a "$RUNNER_TEMP/pg-test.log"
           ! grep -- '--- SKIP' "$RUNNER_TEMP/pg-test.log"
 
diff --git a/README.md b/README.md
index 10eccfd47..e260e7e28 100644
--- a/README.md
+++ b/README.md
@@ -45,8 +45,8 @@ performance) — source in
   restart reconcile with VMM cold restart, memberlist mesh with redirect
   placement, a checkpoint store on a shared directory or S3 (sparse,
   content-addressed chunks), and an optional shared PostgreSQL meta store
-  for tenants, pools, node leases and the cross-node adoption of archived
-  claims and of running claims a pool replicates (`replicate_seconds`)
+  for tenants, pools, node leases, usage events and the cross-node adoption of
+  archived claims and of running claims a pool replicates (`replicate_seconds`)
 - `sdk/go/` — Go SDK: `Connect/New/Lookup`, `Exec/Run`, files,
   `Push/Pull`, sessions, `Find/Replace`, `Watch`, git verbs, `OpenPty`,
   `Renew/Fork/Hibernate/Promote/Checkpoint`, claim env, `DialPort/ProxyPort/PreviewURL`,
diff --git a/docs/cluster.md b/docs/cluster.md
index a94f4ded7..f40a47523 100644
--- a/docs/cluster.md
+++ b/docs/cluster.md
@@ -238,8 +238,11 @@ replication rounds write it, so the claim path is unchanged.
   replica retirement keeps its tombstone until the hourly sweep after that.
 - **Usage journal.** The journal stays per node: an adopted claim's `archive`
   is in the old owner's `usage.jsonl`, and its `unarchive`, `release` or
-  `archive_delete` is in the adopter's or the purging node's. Fold the journals
-  of all nodes by sandbox id.
+  `archive_delete` is in the adopter's or the purging node's. Fold the events
+  of all nodes by sandbox id. With `meta_store.usage` every node lands its
+  journal in the cell's [shared usage table](deploy.md#shared-usage-table), and
+  [`GET /v1/usage`](sandboxd-api.md#get-v1usage) on any node returns the
+  events of all of them.
 
 The node lease decides adoption, so a partitioned owner can lose an archived
 claim it still lists once its lease runs out; the epoch fences it out. A claim
diff --git a/docs/deploy.md b/docs/deploy.md
index aba0df01e..95dd46a8a 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -129,7 +129,7 @@ gets the intersection of the two.
 | `egress_usage_bytes` | `false` | append an `egress_bytes` usage event with the payload bytes of every allowed egress tunnel or request when it ends; one extra usage-journal write per connection |
 | `egress_ca` | unset | [HTTPS-interception](egress.md#https-interception) PKI: `root_cert` (the cluster root baked into intercepted guests; may bundle old+new roots during rotation) plus this node's `intermediate_cert`/`intermediate_key` from `sandboxd ca issue-intermediate`. Required when any pool rule sets `intercept` |
 | `api_token` | unset | the operator (root) credential: when set, guards the node-level endpoints (Bearer) with full access, including release-by-id cleanup. Per-sandbox tokens guard ordinary sandbox-scoped calls |
-| `meta_store` | unset | `{"kind": "pg", "dsn_env": "SANDBOXD_PG_DSN"}`: keep the tenant set in one shared PostgreSQL instead of each node's `tenants.json` (see [shared tenant database](#shared-tenant-database)). `dsn_env` names the node env holding the connection string. An optional `cell` also keeps the API-applied pool set there, one per cell, instead of in each node's `pools.json` (see [shared pool set](#shared-pool-set)); without it pool targets stay per node. Needs `api_token`. Restart only |
+| `meta_store` | unset | `{"kind": "pg", "dsn_env": "SANDBOXD_PG_DSN"}`: keep the tenant set in one shared PostgreSQL instead of each node's `tenants.json` (see [shared tenant database](#shared-tenant-database)). `dsn_env` names the node env holding the connection string. An optional `cell` also keeps the API-applied pool set there, one per cell, instead of in each node's `pools.json` (see [shared pool set](#shared-pool-set)); without it pool targets stay per node. An optional `usage: true` also lands this node's usage journal in one shared table and serves [`GET /v1/usage`](sandboxd-api.md#get-v1usage) from it (see [shared usage table](#shared-usage-table)). Needs `api_token`. Restart only |
 | `max_fork_count` | 16 | children a single `fork` may create; each is a full-RAM VM, so this bounds one request's memory blast radius to the node's capacity |
 | `refill_concurrency` | 0 (auto) | concurrent VM provisioning budget, shared by warm-pool refills, fork clones, and the reap/hibernate/reconcile engine batches; the batches never take the last quarter of it (at least one slot), so a mass expiry or idle sweep cannot stop refills. 0 sizes it from the node: `NumCPU*2/3` clamped to [4, 256] — a 384-core node gets 256; small nodes keep a floor of 4 |
 | `release_delay_seconds` | 0 | seconds a released VM waits in the removal queue before `cocoon vm rm`. The claim is dropped and journaled and its egress listener closed at release; the VM removal, the volume hold release and the egress tap unlock run on the first reap tick (5 s) after the delay, on the `refill_concurrency` budget like every other batch teardown, so a burst of releases does not compete with the claims still running. Until then the VM holds its memory and its volume reservations without counting as a claim. 0 removes inline |
@@ -600,14 +600,14 @@ sandbox and checkpoint listings filter to the caller's tenant, and a tenant
 can delete only its own checkpoints and templates (root sees everything).
 Operator surfaces stay root-only — per-id sandbox reads and stats,
 `GET /v1/checkpoints/{id}/blob`, `GET /v1/info`, `PUT /v1/pools`,
-`/v1/tenants`, `/v1/config/reload`, `POST/DELETE /v1/drain`, `/metrics`; a
+`/v1/tenants`, `/v1/config/reload`, `POST/DELETE /v1/drain`, `/metrics`, `GET /v1/usage`; a
 tenant token there is authenticated but not authorized, so it answers 403 (a
 wrong token stays 401). Per-sandbox
 tokens are unchanged: whoever holds a sandbox's token drives that sandbox.
 Fork children inherit the parent's tenant and count against its
 `max_claims` (0 = unlimited, a per-node cap, so a tenant's cluster limit is
 `max_claims` × nodes); a tenant at its cap gets 429 exactly like a node at
-`max_claims`, and the usage journal's claim events carry the tenant for
+`max_claims`, and every usage event of a tenant's claim carries the tenant for
 per-tenant billing. `config.json` has no `tenants` field: a node starts with
 no tenants until the API adds them.
 
@@ -748,6 +748,9 @@ through `/v1/tenants` on any node.
   - The pool defaults to `max(4, CPUs)` connections; set `pool_max_conns` in
     the DSN (4 suffices) so a fleet stays within the database's
     `max_connections`.
+  - Every other shared table a node opens adds a query pool of the same size:
+    the pool set with a `cell`, the archive registry with `archive_failover`,
+    and the usage table with `usage`. Count each in `max_connections`.
   - LISTEN needs a session-mode connection, so a connection pooler in front
     of the database must not run in transaction mode for that DSN.
   - A write runs in one transaction with a `pg_notify`.
@@ -809,6 +812,98 @@ the tenant set.
 - **Reverting** to config-owned pools is a `PUT` of the config pools; changing
   a node's `cell` takes a restart.
 
+### Shared usage table
+
+With `meta_store.usage: true`, each node also lands its usage journal in one
+table, and root reads the events of every node of the cell through
+[`GET /v1/usage`](sandboxd-api.md#get-v1usage) on any of them. It is off by
+default, so an upgrade adds no table and no writes. The table holds raw events
+only; computing durations or prices stays with the consumer.
+
+- **Landing:** a background loop reads `usage.jsonl` and its `.1` backup every
+  5 s and lands the lines it has not landed yet, up to 1000 per transaction,
+  until it catches up. The journal append is unchanged, so a claim, release,
+  hibernate or wake never waits on the database.
+- **Node:** rows carry the node's mesh `node_id` (its `bind` when unset), or
+  its host name without a mesh. The name must be unique within the cell.
+- **Schema:** sandboxd creates the table at boot when it is missing; create it
+  ahead with the same DDL when the node's role has no `CREATE` right:
+
+  ```sql
+  CREATE TABLE IF NOT EXISTS sandboxd_usage (
+      cell        text        NOT NULL,
+      cursor      bigint      GENERATED ALWAYS AS IDENTITY,
+      node        text        NOT NULL,
+      line_sha256 bytea       NOT NULL,
+      t           timestamptz NOT NULL,
+      ev          text        NOT NULL,
+      id          text        NOT NULL,
+      vm          text        NOT NULL DEFAULT '',
+      key         text        NOT NULL DEFAULT '',
+      tenant      text        NOT NULL DEFAULT '',
+      volumes     text[],
+      volumes_rw  text[],
+      children    text[],
+      ref         text        NOT NULL DEFAULT '',
+      upstream    text        NOT NULL DEFAULT '',
+      secret      text        NOT NULL DEFAULT '',
+      tx          bigint      NOT NULL DEFAULT 0,
+      rx          bigint      NOT NULL DEFAULT 0,
+      PRIMARY KEY (cell, cursor),
+      UNIQUE (cell, node, line_sha256)
+  );
+  CREATE INDEX IF NOT EXISTS sandboxd_usage_node ON sandboxd_usage (cell, node, cursor);
+  CREATE INDEX IF NOT EXISTS sandboxd_usage_tenant ON sandboxd_usage (cell, tenant, cursor);
+  CREATE INDEX IF NOT EXISTS sandboxd_usage_t ON sandboxd_usage (cell, t);
+  ```
+
+- **Columns:** the [journal line](sandboxd-api.md#usage-journal-usagejsonl)'s
+  keys, with an absent key as `''`, `0` or `NULL`, plus `cell` (`''` without
+  one), `node` and `line_sha256`, the SHA-256 of the line. `t` keeps
+  microseconds, and a NUL in `ref` lands as U+FFFD.
+- **At most once per line:** `(cell, node, line_sha256)` is unique, and a
+  land skips a line whose digest already landed for its node, so no journal
+  line lands twice. The node's newest row (by `cursor`) is its resume point:
+  at boot and after a failed land, the node reads that row's `line_sha256`
+  and resumes after that line in
+  `usage.jsonl.1` or `usage.jsonl`. When neither file holds the line any
+  more (a crash lost a journal tail the node had already landed, or the
+  journal was reset), the node reads both files from the start and lands only
+  the lines not landed yet. A restart, a retry or a commit whose reply was
+  lost therefore lands no line twice. Two byte-identical lines of one node
+  (same nanosecond `t` and fields) land once. A line that does not decode is skipped with
+  a warning, and a last line without its newline waits for it. A line a
+  crash cut short stays a line of its own: the node ends it with a newline
+  when it opens the journal, so the first event after the restart is whole.
+- **Order:** a cell's lands commit one at a time under a transaction-scoped
+  advisory lock, so `cursor` grows in commit order and a reader that pages by
+  cursor never skips a row that commits later. A node lands its lines in
+  journal order, so its rows follow that order by `cursor`. A backfilled row
+  gets a newer cursor than rows with a later `t`.
+- **Connection:** the table has its own query pool, sized like the others (see
+  [shared tenant database](#shared-tenant-database)), and no LISTEN session.
+  The node lands on one connection at a time; `GET /v1/usage` reads take
+  more.
+- **Outage:** claims and the journal carry on; the node logs one warning,
+  retries every 5 s, and backfills when the database returns. Nothing is lost
+  while the unlanded lines are still in `usage.jsonl` or `usage.jsonl.1`, so
+  an outage loses lines only after the journal rotates twice past the last
+  landed one (64 MiB each); the node no longer holds those either. A node
+  needs the database at boot, like the other shared tables.
+- **Retention:** sandboxd never deletes a row. Prune only rows every consumer
+  has paged past, and keep each node's newest row: it is the node's resume
+  point, and a node that finds none lands its whole journal again. A crash
+  that loses that line from the journal also lands again the pruned lines the
+  journal still holds.
+
+  ```sql
+  DELETE FROM sandboxd_usage u
+  WHERE cell = $1 AND cursor <= $2
+    AND cursor < (SELECT max(cursor) FROM sandboxd_usage WHERE cell = u.cell AND node = u.node);
+  ```
+- **Renaming** a node, or moving it to another cell, starts it with no rows:
+  the node lands what its journal still holds again under the new name.
+
 ### Reloading the config
 
 `kill -HUP <sandboxd pid>` or a root `POST /v1/config/reload`
diff --git a/docs/index.md b/docs/index.md
index 4005a401c..d3aaec111 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -27,7 +27,7 @@ vsock-only I/O (hardened default); `net=egress` attaches a bridge/CNI NIC.
   renew, hibernate, fork, promote, checkpoints, preview, the guest-port
   relay, claim env, runtime volume registration, live volume changes,
   runtime tenants, config reload, online pool reconfigure, metrics, the usage
-  journal
+  journal and its shared table
 - [Replace a dataset in a running sandbox](live-volumes.md) — swapping a
   read-only catalog volume on a running claim with the volume API
 - [Hibernate with an external volume controller](volume-controller.md) — detach,
diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md
index 98b166394..8760c3084 100644
--- a/docs/sandboxd-api.md
+++ b/docs/sandboxd-api.md
@@ -9,7 +9,7 @@ All bodies are JSON. Three token kinds:
   tenant creates is stamped with its name. Operator surfaces
   (the per-id sandbox reads, `PUT /v1/sandboxes/{id}/instance-metadata`, `GET /v1/info`,
   `PUT /v1/pools`, `PUT/DELETE /v1/volumes/{name}`, `POST/DELETE /v1/drain`,
-  `GET /metrics`, `GET /v1/checkpoints/{id}/blob`)
+  `GET /metrics`, `GET /v1/usage`, `GET /v1/checkpoints/{id}/blob`)
   answer a tenant token `403` — authenticated but not authorized; an unknown
   token stays `401`. On a node with a shared
   [`meta_store`](deploy.md#shared-tenant-database), a tenant token the node
@@ -1015,8 +1015,8 @@ Always on: every lifecycle transition appends one JSONL event to
 `<data_dir>/usage.jsonl` — `{"t": <RFC3339>, "ev":
 "claim|hibernate|wake|fork|checkpoint|promote|release|reap|archive|unarchive|archive_delete|egress|egress_bytes|vmm_exit|vmm_restart|vmm_failed",
 "id": "sb_…", "vm": "sbx-…"}` plus `key` (the pool key's stable hash, claim
-events), `tenant` (the owning tenant, on claim, egress, archive, unarchive and
-archive_delete), `children` (fork) and `ref` (the promoted
+events), `tenant` (the owning tenant, on every event of a tenant's claim),
+`children` (fork) and `ref` (the promoted
 template / checkpoint id, the egress host, or a `vmm_failed` reason). An egress event that injected credentials names
 them in `secret` (`gh,claim:<entry>`, never a value). An egress event routed through an
 [upstream proxy](egress.md#upstream-proxies) carries `upstream` (its `host:port`), and with
@@ -1025,7 +1025,8 @@ them in `secret` (`gh,claim:<entry>`, never a value). An egress event routed thr
 the subset of those names claimed `rw`, so billing can discriminate write
 access (mounts and host paths are not billing dimensions). The file rotates at
 64 MiB keeping one `.1` backup, so a tailing collector never loses a window
-silently. Folding rules: billable compute seconds per sandbox =
+silently. A last line a crash cut short is ended with a newline when the node
+reopens the journal, so it never swallows the next event. Folding rules: billable compute seconds per sandbox =
 Σ(claim→release/reap) − Σ(hibernate→wake) − Σ(vmm_exit→vmm_restart/release/reap);
 `vmm_exit` is stamped when the node first finds the VMM gone, so an outage
 before sandboxd runs again (a host that was down) stays billed up to that
@@ -1035,7 +1036,39 @@ point, and `vmm_failed` is informational; hibernated storage seconds =
 VM's RAM. An interval left open by a crash clamps to the claim's
 deadline; reconcile records nothing for the claims it drops (those whose VM
 record is gone), so a restart leaves those intervals without a terminal event. The `vm` name joins
-cocoon's machine-level metering ledger for audit cross-checks.
+cocoon's machine-level metering ledger for audit cross-checks. With
+`meta_store.usage`, the node also lands every line in the cell's
+[shared usage table](deploy.md#shared-usage-table), read through
+[`GET /v1/usage`](#get-v1usage).
+
+## GET /v1/usage
+
+Auth: root only (tenant tokens get 403). Reads the raw usage events of every
+node of the cell from the [shared usage table](deploy.md#shared-usage-table).
+
+- `GET /v1/usage?tenant=<name>&since=<time>&until=<time>&after=<cursor>&limit=<n>` →
+  `{"events": [{"cursor": 41, "node": "n1", "t": "2026-10-30T08:00:00.123456Z", "ev": "claim", "id": "sb_…", "vm": "sbx-…", "key": "…", "tenant": "acme"}], "next": 41}`.
+- **Events:** each is one journal line with its keys as in
+  [usage.jsonl](#usage-journal-usagejsonl), `t` in UTC to the microsecond, plus
+  `cursor` (its place in the table; one node's events follow its journal
+  order) and `node` (the node that recorded it).
+- **Filters:** `tenant` keeps one tenant's events; `since` (inclusive) and
+  `until` (exclusive) are RFC 3339 times compared with `t`. Each is optional.
+  A journal written by v0.1.20 or earlier stamps `tenant` only on `claim`,
+  `archive`, `unarchive`, `archive_delete`, `egress`, `egress_bytes` and
+  `vmm_*`; join its other events on `id` to the `claim`.
+- **Order and paging:**
+  - events come in `cursor` order, the order they landed, which is not `t`
+    order: a node that was cut off from the database lands its backlog later;
+  - `after` returns only events with a larger cursor (default 0);
+  - `limit` defaults to 1000, at most 10000;
+  - `next` is the last event's cursor when the page is full, absent otherwise;
+    pass it as the following page's `after`;
+  - the cell's lands commit in cursor order, so polling with the last cursor seen
+    as `after` returns every event exactly once, backfill included.
+- 400 on a malformed `after`, `limit`, `since` or `until`; 501
+  `no shared usage store configured` on a node without `meta_store.usage`; 503
+  `usage store down; retry` while the database is unreachable.
 
 ## GET /v1/sandboxes/{id}/agent
 
diff --git a/sandboxd/backends.go b/sandboxd/backends.go
index d2f147fe7..307a438d0 100644
--- a/sandboxd/backends.go
+++ b/sandboxd/backends.go
@@ -17,6 +17,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/tenants"
 	"github.com/cocoonstack/sandbox/sandboxd/tenants/file"
 	tenantspg "github.com/cocoonstack/sandbox/sandboxd/tenants/pg"
+	usagepg "github.com/cocoonstack/sandbox/sandboxd/usage/pg"
 )
 
 func openBackends(ctx context.Context, cfg *config.Config) (pool.Backends, error) {
@@ -47,6 +48,21 @@ func openBackends(ctx context.Context, cfg *config.Config) (pool.Backends, error
 	return b, nil
 }
 
+func openUsage(ctx context.Context, cfg *config.Config) (*usagepg.Store, error) {
+	ms := cfg.MetaStore
+	if ms == nil || !ms.Usage {
+		return nil, nil
+	}
+	return usagepg.Open(ctx, os.Getenv(ms.DSNEnv), ms.Cell)
+}
+
+func usageNode(cfg *config.Config) (string, error) {
+	if cfg.Mesh != nil {
+		return cfg.Mesh.ID(), nil
+	}
+	return os.Hostname()
+}
+
 func newStoreView(ctx context.Context, cfg *config.Config, staging string, idRe *regexp.Regexp) (store.Store, error) {
 	if cs := cfg.CheckpointStore; cs != nil && cs.Kind == "s3" {
 		return s3.New(ctx, *cs.S3, filepath.Join(cfg.DataDir, staging), idRe)
diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go
index afd49fc68..6fed35b5e 100644
--- a/sandboxd/config/config.go
+++ b/sandboxd/config/config.go
@@ -158,6 +158,8 @@ type MetaStoreConfig struct {
 	DSNEnv string `json:"dsn_env"`
 	// Cell opts the node into one pool set for every node of the cell; empty keeps the pool set per node.
 	Cell string `json:"cell,omitempty"`
+	// Usage lands the node's usage journal in the shared usage table and serves GET /v1/usage from it.
+	Usage bool `json:"usage,omitzero"`
 }
 
 // VolumeSpec declares one operator-managed dataset disk, held by exactly one node.
diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go
index 1b5411bef..3fbf15166 100644
--- a/sandboxd/config/config_test.go
+++ b/sandboxd/config/config_test.go
@@ -195,6 +195,13 @@ func TestLoadMetaStore(t *testing.T) {
 	if _, _, err := cfg.ReloadDiff(&next); err == nil || !strings.Contains(err.Error(), "meta_store") {
 		t.Errorf("a reload that drops meta_store: %v, want a restart refusal", err)
 	}
+	withUsage, usageErr := Load(writeConfig(t, `{"api_token":"r","pools":[],"meta_store":{"kind":"pg","dsn_env":"SANDBOX_TEST_DSN","usage":true}}`))
+	if usageErr != nil || !withUsage.MetaStore.Usage || cfg.MetaStore.Usage {
+		t.Fatalf("Load with usage: %+v %v, want usage only where set", withUsage, usageErr)
+	}
+	if _, _, err := cfg.ReloadDiff(withUsage); err == nil || !strings.Contains(err.Error(), "meta_store") {
+		t.Errorf("a reload that turns usage on: %v, want a restart refusal", err)
+	}
 }
 
 func TestLoadArchiveFailover(t *testing.T) {
diff --git a/sandboxd/main.go b/sandboxd/main.go
index a74f8ef50..7e080d6da 100644
--- a/sandboxd/main.go
+++ b/sandboxd/main.go
@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"runtime/debug"
 	"slices"
 	"syscall"
@@ -26,6 +27,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/pool"
 	"github.com/cocoonstack/sandbox/sandboxd/server"
 	"github.com/cocoonstack/sandbox/sandboxd/store/peer"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -86,6 +88,10 @@ func main() {
 	if err != nil {
 		logger.Fatalf(ctx, err, "open backends")
 	}
+	usageStore, err := openUsage(ctx, cfg)
+	if err != nil {
+		logger.Fatalf(ctx, err, "open usage store")
+	}
 	mgr, err := pool.NewManager(ctx, cfg, eng, secrets, backends)
 	if err != nil {
 		logger.Fatalf(ctx, err, "init pool manager")
@@ -154,6 +160,15 @@ func main() {
 	if cfg.NoRxSplice {
 		srv.NoRxSplice()
 	}
+	if usageStore != nil {
+		node, err := usageNode(cfg)
+		if err != nil {
+			logger.Fatalf(ctx, err, "name the usage node")
+		}
+		go usage.Ship(ctx, filepath.Join(cfg.DataDir, usage.JournalFile), node, usageStore)
+		logger.Infof(ctx, "usage journal lands in the shared table as node %s", node)
+		srv.SetUsage(usageStore)
+	}
 	httpSrv := &http.Server{
 		Addr:              cfg.Listen,
 		Handler:           srv.Handler(),
diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go
index e551bdfe6..2b379f680 100644
--- a/sandboxd/pool/archive.go
+++ b/sandboxd/pool/archive.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/cocoonstack/sandbox/sandboxd/store"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 	"github.com/cocoonstack/sandbox/sandboxd/utils"
 )
 
@@ -175,7 +176,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error {
 		m.deleteOrphanArchiveCk(ctx, replica)
 	}
 	m.counters.archives.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant})
+	m.recordUsage(ctx, usage.Event{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant})
 	return nil
 }
 
@@ -244,7 +245,7 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string,
 		return "", ErrUnknownSandbox
 	}
 	m.counters.unarchives.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "unarchive", ID: sb.ID, Reference: ck, Tenant: sb.Tenant})
+	m.recordUsage(ctx, usage.Event{Event: "unarchive", ID: sb.ID, Reference: ck, Tenant: sb.Tenant})
 	return built.VsockSocket, nil
 }
 
diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go
index aed5f7c13..d80cd293e 100644
--- a/sandboxd/pool/checkpoint.go
+++ b/sandboxd/pool/checkpoint.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cocoonstack/sandbox/sandboxd/store"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 // DeleteScope selects how far a checkpoint delete reaches; fleet-wide is the zero value.
@@ -53,7 +54,7 @@ func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, te
 		return types.Checkpoint{}, err
 	}
 	m.counters.checkpoints.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "checkpoint", ID: sb.ID, VMName: lockedVMName(sb), Reference: ckpt.ID})
+	m.recordUsage(ctx, usage.Event{Event: "checkpoint", ID: sb.ID, VMName: lockedVMName(sb), Tenant: sb.Tenant, Reference: ckpt.ID})
 	return ckpt, nil
 }
 
diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go
index c13e8e75f..9efedfa9e 100644
--- a/sandboxd/pool/claim.go
+++ b/sandboxd/pool/claim.go
@@ -13,6 +13,7 @@ import (
 	"github.com/projecteru2/core/log"
 
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -357,7 +358,7 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand
 	m.out.Disarm(id, removed)
 	m.dropSnap(ctx, snap)
 	m.counters.releases.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: vmName})
+	m.recordUsage(ctx, usage.Event{Event: "release", ID: id, VMName: vmName, Tenant: sb.Tenant})
 	return err
 }
 
@@ -502,7 +503,7 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t
 	}
 	// usage lands only after the batch armed, so a rollback leaves no unterminated claim event
 	for _, sb := range sbs {
-		m.recordUsage(ctx, usageEvent{
+		m.recordUsage(ctx, usage.Event{
 			Event: "claim",
 			ID:    sb.ID, VMName: sb.VMName,
 			KeyHash: sb.Key.Hash(), Tenant: sb.Tenant,
@@ -659,7 +660,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 			m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant)
 			m.untrack(m.pendingCks, v.ck)
 			m.counters.reaps.Add(1)
-			m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id})
+			m.recordUsage(ctx, usage.Event{Event: "reap", ID: v.id, Tenant: v.tenant})
 			logger.Infof(ctx, "purged archived sandbox %s", v.id)
 		case reapArchive:
 			logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id)
@@ -671,7 +672,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 			m.out.Disarm(v.id, m.removeOrRetry(ctx, v.vmName, v.id, td))
 			m.dropSnap(ctx, v.snap)
 			m.counters.reaps.Add(1)
-			m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName})
+			m.recordUsage(ctx, usage.Event{Event: "reap", ID: v.id, VMName: v.vmName, Tenant: v.tenant})
 			logger.Infof(ctx, "reaped expired sandbox %s (%s)", v.id, v.vmName)
 		}
 	})
@@ -704,7 +705,7 @@ func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) {
 		logger.Warnf(ctx, "delete archive ck %s: %v", ck, err)
 	}
 	m.counters.archiveDeletes.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant})
+	m.recordUsage(ctx, usage.Event{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant})
 }
 
 func (m *Manager) setIfLive(sb *types.Sandbox, mutate func()) (claimSnapshot, bool) {
diff --git a/sandboxd/pool/egress.go b/sandboxd/pool/egress.go
index c9098756f..c956b670a 100644
--- a/sandboxd/pool/egress.go
+++ b/sandboxd/pool/egress.go
@@ -10,6 +10,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/egress"
 	"github.com/cocoonstack/sandbox/sandboxd/outbound"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 // NetRoute reports sb's route as its outbound host sees it.
@@ -80,7 +81,7 @@ func (m *Manager) outboundOptions(ca *egress.CA) outbound.Options {
 		Pooled: m.claimPooled,
 		Record: m.recordEgress,
 		Transfer: func(ctx context.Context, id, tenant string, ev egress.Event, sent, received int64) {
-			m.recordUsage(ctx, usageEvent{Event: "egress_bytes", ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Sent: sent, Received: received})
+			m.recordUsage(ctx, usage.Event{Event: "egress_bytes", ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Sent: sent, Received: received})
 		},
 		CA:         ca,
 		LockNIC:    len(m.cfg.Bridges) > 0,
diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go
index dfcb60caa..432c26327 100644
--- a/sandboxd/pool/failover.go
+++ b/sandboxd/pool/failover.go
@@ -16,6 +16,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/store"
 	"github.com/cocoonstack/sandbox/sandboxd/tenants"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 // AdoptArchived takes over a claim whose owner's lease expired and reports whether this node owns it now.
@@ -502,7 +503,7 @@ func (m *Manager) purgeArchiveRow(ctx context.Context, rec archives.Record) {
 	}
 	if !rec.Consumed {
 		m.counters.archiveDeletes.Add(1)
-		m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: rec.ID, Reference: rec.Checkpoint, Tenant: rec.Tenant})
+		m.recordUsage(ctx, usage.Event{Event: "archive_delete", ID: rec.ID, Reference: rec.Checkpoint, Tenant: rec.Tenant})
 	}
 	logger.Infof(ctx, "purged archive %s of %s", rec.ID, rec.Owner)
 }
diff --git a/sandboxd/pool/fork.go b/sandboxd/pool/fork.go
index 83c2ad8d9..5197d70c4 100644
--- a/sandboxd/pool/fork.go
+++ b/sandboxd/pool/fork.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 // Fork clones a claimed sandbox into count children, each a fresh claim; all-or-nothing.
@@ -56,7 +57,7 @@ func (m *Manager) Fork(ctx context.Context, id string, cred Cred, count int, ttl
 	for i, c := range children {
 		ids[i] = c.ID
 	}
-	m.recordUsage(ctx, usageEvent{Event: "fork", ID: sb.ID, VMName: lockedVMName(sb), Children: ids})
+	m.recordUsage(ctx, usage.Event{Event: "fork", ID: sb.ID, VMName: lockedVMName(sb), Tenant: sb.Tenant, Children: ids})
 	return children, nil
 }
 
diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go
index 4e464fbf1..9e91ddef3 100644
--- a/sandboxd/pool/hibernate.go
+++ b/sandboxd/pool/hibernate.go
@@ -12,6 +12,7 @@ import (
 	"github.com/projecteru2/core/log"
 
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 func (m *Manager) Hibernate(ctx context.Context, id string, cred Cred) error {
@@ -207,7 +208,7 @@ func (m *Manager) wakeOrRead(ctx context.Context, sb *types.Sandbox, readerOK bo
 	}
 	m.counters.wakes.Add(1)
 	m.counters.wakeNanos.Add(uint64(time.Since(wakeStart))) //nolint:gosec // durations are positive
-	m.recordUsage(ctx, usageEvent{Event: "wake", ID: sb.ID, VMName: sb.VMName})
+	m.recordUsage(ctx, usage.Event{Event: "wake", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
 	if proxyErr := m.out.ArmProxy(ctx, sb); proxyErr != nil {
 		log.WithFunc("pool.wakeOrRead").Errorf(ctx, proxyErr, "arm egress proxy %s", sb.ID)
 	}
@@ -372,7 +373,7 @@ func (m *Manager) resolvePendingSnap(ctx context.Context, sb *types.Sandbox) (ad
 
 func (m *Manager) recordHibernate(ctx context.Context, sb *types.Sandbox) {
 	m.counters.hibernates.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "hibernate", ID: sb.ID, VMName: sb.VMName})
+	m.recordUsage(ctx, usage.Event{Event: "hibernate", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
 }
 
 func skipIdle(sb *types.Sandbox) bool {
diff --git a/sandboxd/pool/journal.go b/sandboxd/pool/journal.go
index b4ba6d692..d5db13b4f 100644
--- a/sandboxd/pool/journal.go
+++ b/sandboxd/pool/journal.go
@@ -6,30 +6,11 @@ import (
 	"errors"
 	"os"
 	"sync"
-	"time"
 )
 
 // journalMaxBytes rotates a journal once it crosses this size; one .1 backup is kept.
 const journalMaxBytes = 64 * 1024 * 1024
 
-// usageEvent is one line of usage.jsonl, the billing stream the platform collector folds.
-type usageEvent struct {
-	Time      time.Time `json:"t"`
-	Event     string    `json:"ev"`
-	ID        string    `json:"id"`
-	VMName    string    `json:"vm,omitempty"`
-	KeyHash   string    `json:"key,omitempty"`
-	Tenant    string    `json:"tenant,omitempty"`
-	Volumes   []string  `json:"volumes,omitempty"`
-	VolumesRW []string  `json:"volumes_rw,omitempty"`
-	Children  []string  `json:"children,omitempty"`
-	Reference string    `json:"ref,omitempty"`
-	Upstream  string    `json:"upstream,omitempty"`
-	Secret    string    `json:"secret,omitempty"`
-	Sent      int64     `json:"tx,omitzero"`
-	Received  int64     `json:"rx,omitzero"`
-}
-
 // journal is an append-only JSONL writer with size rotation; its lock only orders appends.
 type journal struct {
 	mu   sync.Mutex
@@ -39,7 +20,7 @@ type journal struct {
 }
 
 func newJournal(path string) (*journal, error) {
-	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) //nolint:gosec // under data_dir
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) //nolint:gosec // under data_dir
 	if err != nil {
 		return nil, err
 	}
@@ -48,7 +29,12 @@ func newJournal(path string) (*journal, error) {
 		_ = f.Close()
 		return nil, err
 	}
-	return &journal{path: path, f: f, size: st.Size()}, nil
+	j := &journal{path: path, f: f, size: st.Size()}
+	if err = j.endTornLine(); err != nil {
+		_ = f.Close()
+		return nil, err
+	}
+	return j, nil
 }
 
 func (j *journal) append(v any) error {
@@ -68,6 +54,22 @@ func (j *journal) append(v any) error {
 	return errors.Join(rotateErr, err)
 }
 
+func (j *journal) endTornLine() error {
+	if j.size == 0 {
+		return nil
+	}
+	last := make([]byte, 1)
+	if _, err := j.f.ReadAt(last, j.size-1); err != nil {
+		return err
+	}
+	if last[0] == '\n' {
+		return nil
+	}
+	n, err := j.f.Write([]byte{'\n'})
+	j.size += int64(n)
+	return err
+}
+
 // rotate moves the live file to .1 and reopens; the old descriptor closes last.
 func (j *journal) rotate() error {
 	renameErr := os.Rename(j.path, j.path+".1")
diff --git a/sandboxd/pool/journal_test.go b/sandboxd/pool/journal_test.go
index 017034f20..ccd02dfd5 100644
--- a/sandboxd/pool/journal_test.go
+++ b/sandboxd/pool/journal_test.go
@@ -1,12 +1,15 @@
 package pool
 
 import (
+	"encoding/json/v2"
 	"errors"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 func TestJournalRotateFailureDegradesAndRecovers(t *testing.T) {
@@ -91,6 +94,46 @@ func TestJournalRotateReopenFailureKeepsWriting(t *testing.T) {
 	}
 }
 
+func TestJournalKeepsTheFirstAppendAfterATornLine(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "usage.jsonl")
+	if err := os.WriteFile(path, []byte(`{"ev":"claim","id":"sb_a"}`+"\n"+`{"t":"2026-10-30T00:00:00Z","ev":"rel`), 0o600); err != nil {
+		t.Fatalf("setup: %v", err)
+	}
+	j, err := newJournal(path)
+	if err != nil {
+		t.Fatalf("new journal: %v", err)
+	}
+	if err = j.append(usage.Event{Event: "claim", ID: "sb_b"}); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	lines := strings.Split(strings.TrimSuffix(string(raw), "\n"), "\n")
+	var ev usage.Event
+	if len(lines) != 3 || json.Unmarshal([]byte(lines[2]), &ev) != nil || ev.ID != "sb_b" {
+		t.Errorf("journal after a torn line %q, want the append on a line of its own", raw)
+	}
+	if st, _ := os.Stat(path); j.size != st.Size() {
+		t.Errorf("journal size %d, file %d", j.size, st.Size())
+	}
+}
+
+func TestJournalLeavesAWholeLastLine(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "usage.jsonl")
+	if err := os.WriteFile(path, []byte(`{"ev":"claim"}`+"\n"), 0o600); err != nil {
+		t.Fatalf("setup: %v", err)
+	}
+	j, err := newJournal(path)
+	if err != nil {
+		t.Fatalf("new journal: %v", err)
+	}
+	if j.size != 15 {
+		t.Errorf("journal size %d after opening a whole file, want 15", j.size)
+	}
+}
+
 func tail(t *testing.T, path string) string {
 	t.Helper()
 	f, err := os.Open(path)
diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go
index c87398ba7..640746831 100644
--- a/sandboxd/pool/liveness.go
+++ b/sandboxd/pool/liveness.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/cocoonstack/sandbox/sandboxd/engine"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -149,7 +150,7 @@ func (m *Manager) recoverVMM(ctx context.Context, sb *types.Sandbox) {
 	}
 	if !sb.ExitRecorded {
 		sb.ExitRecorded = true
-		m.recordUsage(ctx, usageEvent{Event: "vmm_exit", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
+		m.recordUsage(ctx, usage.Event{Event: "vmm_exit", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
 	}
 	reason := coldBootBlocker(sb, found, volumes)
 	if reason == "" && m.vmmRestart == types.VMMRestartNone {
@@ -243,7 +244,7 @@ func (m *Manager) restartLocked(ctx context.Context, sb *types.Sandbox) error {
 	}
 	sb.ExitRecorded = false
 	sb.Touch()
-	m.recordUsage(ctx, usageEvent{Event: "vmm_restart", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
+	m.recordUsage(ctx, usage.Event{Event: "vmm_restart", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant})
 	log.WithFunc("pool.restartLocked").Warnf(ctx, "restarted %s (%s) from its disk; guest memory is lost", sb.ID, sb.VMName)
 	return nil
 }
@@ -253,7 +254,7 @@ func (m *Manager) failLocked(ctx context.Context, sb *types.Sandbox, reason stri
 	if live, _ := m.commitIfLive(ctx, sb, func() { sb.Failed = reason }); !live {
 		return
 	}
-	m.recordUsage(ctx, usageEvent{Event: "vmm_failed", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Reference: reason})
+	m.recordUsage(ctx, usage.Event{Event: "vmm_failed", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Reference: reason})
 	log.WithFunc("pool.failLocked").Warnf(ctx, "sandbox %s failed: %s", sb.ID, reason)
 }
 
diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go
index 679b7f71b..cd14662e2 100644
--- a/sandboxd/pool/pool.go
+++ b/sandboxd/pool/pool.go
@@ -32,6 +32,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/store/peer"
 	"github.com/cocoonstack/sandbox/sandboxd/tenants"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -505,11 +506,11 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg
 			m.tplSet[rec.ID] = info
 		}
 	}
-	usage, err := newJournal(filepath.Join(cfg.DataDir, "usage.jsonl"))
+	usageJournal, err := newJournal(filepath.Join(cfg.DataDir, usage.JournalFile))
 	if err != nil {
 		return nil, fmt.Errorf("open usage journal: %w", err)
 	}
-	m.usage = usage
+	m.usage = usageJournal
 	if cfg.AuditLog {
 		audit, err := newJournal(filepath.Join(cfg.DataDir, "audit.jsonl"))
 		if err != nil {
diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go
index ada7fd5f4..5ba945c45 100644
--- a/sandboxd/pool/telemetry.go
+++ b/sandboxd/pool/telemetry.go
@@ -12,6 +12,7 @@ import (
 	"github.com/projecteru2/core/log"
 
 	"github.com/cocoonstack/sandbox/sandboxd/egress"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -66,15 +67,7 @@ type requestFrame struct {
 }
 
 type auditFrame struct {
-	Op      string   `json:"op"`
-	Argv    []string `json:"argv,omitempty"`
-	Path    string   `json:"path,omitempty"`
-	Dest    string   `json:"dest,omitempty"`
-	From    string   `json:"from,omitempty"`
-	To      string   `json:"to,omitempty"`
-	URL     string   `json:"url,omitempty"`
-	Session string   `json:"session,omitempty"`
-	Port    uint16   `json:"port,omitzero"`
+	requestFrame
 	Added   []string `json:"added,omitempty"`
 	Changed []string `json:"changed,omitempty"`
 	Removed []string `json:"removed,omitempty"`
@@ -135,10 +128,7 @@ func (m *Manager) Audit(ctx context.Context, id string, line []byte) {
 	if frame.Op == "" {
 		return // torn cap boundary or a non-frame line: nothing to record
 	}
-	m.recordAudit(ctx, id, auditFrame{
-		Op: frame.Op, Argv: frame.Argv, Path: frame.Path, Dest: frame.Dest,
-		From: frame.From, To: frame.To, URL: frame.URL, Session: frame.Session, Port: frame.Port,
-	})
+	m.recordAudit(ctx, id, auditFrame{requestFrame: frame})
 }
 
 // AuditEnabled reports whether the relay should tap request frames at all.
@@ -182,11 +172,11 @@ func (m *Manager) recordEgress(ctx context.Context, id, tenant string, ev egress
 			Tenant: tenant, Src: ev.Local, Dst: ev.Remote,
 		}})
 	}
-	m.recordUsage(ctx, usageEvent{Event: egressEvent, ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected})
+	m.recordUsage(ctx, usage.Event{Event: egressEvent, ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected})
 }
 
 // recordUsage appends one billing event; failures are logged, never propagated.
-func (m *Manager) recordUsage(ctx context.Context, ev usageEvent) {
+func (m *Manager) recordUsage(ctx context.Context, ev usage.Event) {
 	ev.Time = time.Now()
 	if err := m.usage.append(ev); err != nil {
 		log.WithFunc("pool.recordUsage").Errorf(ctx, err, "append usage event %s %s", ev.Event, ev.ID)
diff --git a/sandboxd/pool/telemetry_test.go b/sandboxd/pool/telemetry_test.go
index e2715401c..15b5b4fd9 100644
--- a/sandboxd/pool/telemetry_test.go
+++ b/sandboxd/pool/telemetry_test.go
@@ -15,7 +15,9 @@ import (
 
 	"github.com/cocoonstack/sandbox/sandboxd/config"
 	"github.com/cocoonstack/sandbox/sandboxd/egress"
+	"github.com/cocoonstack/sandbox/sandboxd/tenants"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 func TestUsageJournalRecordsLifecycle(t *testing.T) {
@@ -53,7 +55,7 @@ func TestJournalRotationKeepsOneBackup(t *testing.T) {
 	}
 	j.size = journalMaxBytes
 
-	if err := j.append(usageEvent{Event: "claim", ID: "sb_1"}); err != nil {
+	if err := j.append(usage.Event{Event: "claim", ID: "sb_1"}); err != nil {
 		t.Fatalf("append: %v", err)
 	}
 	if _, err := os.Stat(path + ".1"); err != nil {
@@ -137,16 +139,8 @@ func TestTenantStampedInJournals(t *testing.T) {
 		t.Errorf("persisted claim %+v, want tenant acme", claims[sb.ID])
 	}
 
-	usage, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl"))
-	if err != nil {
-		t.Fatalf("read usage journal: %v", err)
-	}
 	found := false
-	for line := range strings.SplitSeq(strings.TrimSpace(string(usage)), "\n") {
-		var ev usageEvent
-		if err := json.Unmarshal([]byte(line), &ev); err != nil {
-			t.Fatalf("bad journal line %q: %v", line, err)
-		}
+	for _, ev := range usageJournal(t, m) {
 		if ev.Event == "claim" && ev.ID == sb.ID {
 			found = true
 			if ev.Tenant != "acme" {
@@ -159,6 +153,54 @@ func TestTenantStampedInJournals(t *testing.T) {
 	}
 }
 
+func TestEveryClaimScopedUsageEventCarriesTheTenant(t *testing.T) {
+	m := tenantManager(t, t.TempDir())
+	ctx := t.Context()
+	if err := m.PutTenant(ctx, tenants.TenantSpec{Name: "acme", Token: "acme-tok"}); err != nil {
+		t.Fatalf("put tenant: %v", err)
+	}
+	sb, err := m.ClaimProvision(ctx, testKey, ClaimOptions{TTL: time.Hour, Tenant: "acme"})
+	if err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+	cred := Cred{Token: sb.Token}
+	if err = m.Hibernate(ctx, sb.ID, cred); err != nil {
+		t.Fatalf("hibernate: %v", err)
+	}
+	if _, _, _, err = m.WakeAgentSocket(ctx, sb.ID, sb.Token); err != nil {
+		t.Fatalf("wake: %v", err)
+	}
+	children, err := m.Fork(ctx, sb.ID, cred, 1, time.Hour, "", "")
+	if err != nil {
+		t.Fatalf("fork: %v", err)
+	}
+	if _, err = m.Checkpoint(ctx, sb.ID, cred, "", ""); err != nil {
+		t.Fatalf("checkpoint: %v", err)
+	}
+	if _, _, err = m.Promote(ctx, sb.ID, cred, "tpl:acme", "acme"); err != nil {
+		t.Fatalf("promote: %v", err)
+	}
+	if err = m.Release(ctx, sb.ID, cred); err != nil {
+		t.Fatalf("release: %v", err)
+	}
+	expire(m, children[0])
+	m.reapOnce(ctx)
+	waitFor(t, func() bool { return strings.HasSuffix(usageEventsOf(t, m, children[0].ID), ",reap") })
+
+	kinds := map[string]bool{}
+	for _, ev := range usageJournal(t, m) {
+		kinds[ev.Event] = true
+		if ev.Tenant != "acme" {
+			t.Errorf("%s event of %s carries tenant %q, want acme", ev.Event, ev.ID, ev.Tenant)
+		}
+	}
+	for _, kind := range []string{"claim", "hibernate", "wake", "fork", "checkpoint", "promote", "release", "reap"} {
+		if !kinds[kind] {
+			t.Errorf("no %s event in the journal", kind)
+		}
+	}
+}
+
 func TestSandboxesIndexOmitsTokens(t *testing.T) {
 	eng := newFakeEngine()
 	m := newTestManager(t, eng)
@@ -332,16 +374,8 @@ func TestClaimMetadataSurvivesARestart(t *testing.T) {
 func TestEgressUsageEventNamesTheInjectedCredentials(t *testing.T) {
 	m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey})
 	m.recordEgress(t.Context(), "sb_v", "acme", egress.Event{Method: http.MethodGet, Host: "graph.example.com", Port: 443, Decision: egress.DecisionAllow, Injected: "claim:FB"})
-	raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl"))
-	if err != nil {
-		t.Fatalf("read journal: %v", err)
-	}
-	var ev usageEvent
-	if err = json.Unmarshal(raw, &ev); err != nil {
-		t.Fatalf("decode %q: %v", raw, err)
-	}
-	if ev.Event != "egress" || ev.Reference != "graph.example.com" || ev.Secret != "claim:FB" {
-		t.Errorf("usage event %+v, want egress to graph.example.com naming claim:FB", ev)
+	if evs := usageJournal(t, m); len(evs) != 1 || evs[0].Event != "egress" || evs[0].Reference != "graph.example.com" || evs[0].Secret != "claim:FB" {
+		t.Errorf("usage events %+v, want one egress to graph.example.com naming claim:FB", evs)
 	}
 }
 
@@ -414,18 +448,27 @@ func TestARelayedFrameCannotForgeOrDropItsAuditLine(t *testing.T) {
 	}
 }
 
-func usageEventsOf(t *testing.T, m *Manager, id string) string {
+func usageJournal(t *testing.T, m *Manager) []usage.Event {
 	t.Helper()
-	raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl"))
+	raw, err := os.ReadFile(filepath.Join(m.dataDir, usage.JournalFile))
 	if err != nil {
-		t.Fatalf("read journal: %v", err)
+		t.Fatalf("read usage journal: %v", err)
 	}
-	var events []string
+	var events []usage.Event
 	for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") {
-		var ev usageEvent
+		var ev usage.Event
 		if err := json.Unmarshal([]byte(line), &ev); err != nil {
 			t.Fatalf("bad journal line %q: %v", line, err)
 		}
+		events = append(events, ev)
+	}
+	return events
+}
+
+func usageEventsOf(t *testing.T, m *Manager, id string) string {
+	t.Helper()
+	var events []string
+	for _, ev := range usageJournal(t, m) {
 		if ev.ID == id {
 			events = append(events, ev.Event)
 		}
diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go
index b2cc7d251..528bf4d06 100644
--- a/sandboxd/pool/template.go
+++ b/sandboxd/pool/template.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cocoonstack/sandbox/sandboxd/store"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 // TemplateInfo is the ops view of one promoted template; an empty Tenant means the operator (root).
@@ -85,7 +86,7 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t
 		m.notifyTemplates()
 	}
 	m.counters.promotes.Add(1)
-	m.recordUsage(ctx, usageEvent{Event: "promote", ID: sb.ID, VMName: vmName, Reference: key.Template})
+	m.recordUsage(ctx, usage.Event{Event: "promote", ID: sb.ID, VMName: vmName, Tenant: sb.Tenant, Reference: key.Template})
 	return key, digest, nil
 }
 
diff --git a/sandboxd/pool/volume_rw_test.go b/sandboxd/pool/volume_rw_test.go
index fd3c40407..c0030381f 100644
--- a/sandboxd/pool/volume_rw_test.go
+++ b/sandboxd/pool/volume_rw_test.go
@@ -1,7 +1,6 @@
 package pool
 
 import (
-	"encoding/json/v2"
 	"errors"
 	"os"
 	"path/filepath"
@@ -631,15 +630,7 @@ func TestWritableDiscoveryAndUsageEvent(t *testing.T) {
 	if err != nil {
 		t.Fatalf("ClaimProvision: %v", err)
 	}
-	raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl"))
-	if err != nil {
-		t.Fatalf("read usage journal: %v", err)
-	}
-	for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") {
-		var event usageEvent
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode usage event: %v", err)
-		}
+	for _, event := range usageJournal(t, m) {
 		if event.Event != "claim" || event.ID != sb.ID {
 			continue
 		}
diff --git a/sandboxd/pool/volume_test.go b/sandboxd/pool/volume_test.go
index f79b637a3..a7c430709 100644
--- a/sandboxd/pool/volume_test.go
+++ b/sandboxd/pool/volume_test.go
@@ -2,7 +2,6 @@ package pool
 
 import (
 	"context"
-	"encoding/json/v2"
 	"errors"
 	"os"
 	"path/filepath"
@@ -498,15 +497,7 @@ func TestVolumeClaimUsageAndScopedSummaries(t *testing.T) {
 		t.Errorf("sandbox summary=%+v ok=%v, want applied volume", got, ok)
 	}
 
-	raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl"))
-	if err != nil {
-		t.Fatalf("read usage journal: %v", err)
-	}
-	for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") {
-		var event usageEvent
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode usage event: %v", err)
-		}
+	for _, event := range usageJournal(t, m) {
 		if event.Event == "claim" && event.ID == acme.ID {
 			if !slices.Equal(event.Volumes, []string{"data"}) {
 				t.Errorf("claim event volumes=%v, want [data]", event.Volumes)
diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go
index c26137baa..d60674e4e 100644
--- a/sandboxd/server/server.go
+++ b/sandboxd/server/server.go
@@ -14,7 +14,6 @@ import (
 	"net/netip"
 	"net/url"
 	"slices"
-	"strconv"
 	"sync"
 	"time"
 
@@ -26,6 +25,7 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/store/peer"
 	"github.com/cocoonstack/sandbox/sandboxd/tenants"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
 )
 
 const (
@@ -36,8 +36,8 @@ const (
 	maxMetadataDoc  = 4 << 10
 	previewTTL      = time.Hour
 
-	defaultTenantPage = 1000
-	maxTenantPage     = 10000
+	defaultPage = 1000
+	maxPage     = 10000
 )
 
 var poolErrHTTP = []struct {
@@ -166,6 +166,11 @@ type CheckpointProber interface {
 	Forget(id string)
 }
 
+// UsageReader reads the cell's shared usage table.
+type UsageReader interface {
+	List(ctx context.Context, q usage.Query) ([]usage.Record, error)
+}
+
 // ConfigReloader re-reads the node config and applies its reloadable settings.
 type ConfigReloader func(ctx context.Context) (pool.ReloadResult, error)
 
@@ -221,6 +226,12 @@ type PoolUpdateRequest struct {
 	Pools []config.PoolSpec `json:"pools"`
 }
 
+// UsageResponse is the wire reply of GET /v1/usage; Next is set when the page is full.
+type UsageResponse struct {
+	Events []usage.Record `json:"events"`
+	Next   int64          `json:"next,omitzero"`
+}
+
 // Server serves the control plane for one node.
 type Server struct {
 	mgr       Manager
@@ -232,6 +243,7 @@ type Server struct {
 	advertise string
 	preview   *PreviewServer
 	reload    ConfigReloader
+	usage     UsageReader
 	noSplice  bool
 
 	execBudget *semaphore.Weighted
@@ -269,6 +281,11 @@ func (s *Server) SetConfigReload(f ConfigReloader) {
 	s.reload = f
 }
 
+// SetUsage serves GET /v1/usage from u; set it before serving.
+func (s *Server) SetUsage(u UsageReader) {
+	s.usage = u
+}
+
 // NoRxSplice copies client bytes to guests in user space; call it before serving.
 func (s *Server) NoRxSplice() { s.noSplice = true }
 
@@ -311,6 +328,7 @@ func (s *Server) Handler() http.Handler {
 	mux.HandleFunc("DELETE /v1/tenants/{name}", s.requireRoot(s.handleDeleteTenant))
 	mux.HandleFunc("POST /v1/drain", s.requireRoot(s.handleDrain))
 	mux.HandleFunc("DELETE /v1/drain", s.requireRoot(s.handleUncordon))
+	mux.HandleFunc("GET /v1/usage", s.requireRoot(s.handleUsage))
 	mux.HandleFunc("GET /v1/sandboxes/{id}/agent", s.handleAgent)
 	mux.HandleFunc("GET /v1/sandboxes/{id}/ports/{port}", s.handlePort)
 	mux.HandleFunc("POST /v1/sandboxes/{id}/exec", s.handleExec)
@@ -749,14 +767,10 @@ func (s *Server) handleReload(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) handleTenants(w http.ResponseWriter, r *http.Request) {
 	q := r.URL.Query()
-	limit := defaultTenantPage
-	if raw := q.Get("limit"); raw != "" {
-		n, err := strconv.Atoi(raw)
-		if err != nil || n < 1 || n > maxTenantPage {
-			writeErr(w, http.StatusBadRequest, fmt.Sprintf("limit must be 1..%d", maxTenantPage))
-			return
-		}
-		limit = n
+	limit, err := pageLimit(q.Get("limit"))
+	if err != nil {
+		writeErr(w, http.StatusBadRequest, err.Error())
+		return
 	}
 	page, err := s.mgr.Tenants(r.Context(), q.Get("after"), limit)
 	writeResult(w, r, "tenants", "", "list tenants failed", err, func() { writeJSON(w, http.StatusOK, page) })
diff --git a/sandboxd/server/server_http.go b/sandboxd/server/server_http.go
index 48bf2a949..ac2aeb3cc 100644
--- a/sandboxd/server/server_http.go
+++ b/sandboxd/server/server_http.go
@@ -11,6 +11,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/projecteru2/core/log"
@@ -169,6 +170,17 @@ func metadataFilter(pairs []string) (types.Metadata, error) {
 	return filter, nil
 }
 
+func pageLimit(raw string) (int, error) {
+	if raw == "" {
+		return defaultPage, nil
+	}
+	n, err := strconv.Atoi(raw)
+	if err != nil || n < 1 || n > maxPage {
+		return 0, fmt.Errorf("limit must be 1..%d", maxPage)
+	}
+	return n, nil
+}
+
 func writePoolErr(w http.ResponseWriter, err error) bool {
 	code, msg, ok := mapPoolErr(w, err)
 	if ok {
diff --git a/sandboxd/server/usage.go b/sandboxd/server/usage.go
new file mode 100644
index 000000000..01b6ab75d
--- /dev/null
+++ b/sandboxd/server/usage.go
@@ -0,0 +1,69 @@
+package server
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/projecteru2/core/log"
+
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
+)
+
+func (s *Server) handleUsage(w http.ResponseWriter, r *http.Request) {
+	if s.usage == nil {
+		writeErr(w, http.StatusNotImplemented, "no shared usage store configured")
+		return
+	}
+	q, err := usageQuery(r.URL.Query())
+	if err != nil {
+		writeErr(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	recs, err := s.usage.List(r.Context(), q)
+	if err != nil {
+		log.WithFunc("server.handleUsage").Error(r.Context(), err, "list usage")
+		writeErr(w, http.StatusServiceUnavailable, "usage store down; retry")
+		return
+	}
+	resp := UsageResponse{Events: recs}
+	if len(recs) == q.Limit {
+		resp.Next = recs[len(recs)-1].Cursor
+	}
+	writeJSON(w, http.StatusOK, resp)
+}
+
+func usageQuery(v url.Values) (usage.Query, error) {
+	q := usage.Query{Tenant: v.Get("tenant")}
+	var err error
+	if raw := v.Get("after"); raw != "" {
+		if q.After, err = strconv.ParseInt(raw, 10, 64); err != nil || q.After < 0 {
+			return usage.Query{}, errors.New("after must be a non-negative cursor")
+		}
+	}
+	if q.Limit, err = pageLimit(v.Get("limit")); err != nil {
+		return usage.Query{}, err
+	}
+	if q.Since, err = usageTime(v, "since"); err != nil {
+		return usage.Query{}, err
+	}
+	if q.Until, err = usageTime(v, "until"); err != nil {
+		return usage.Query{}, err
+	}
+	return q, nil
+}
+
+func usageTime(v url.Values, name string) (time.Time, error) {
+	raw := v.Get(name)
+	if raw == "" {
+		return time.Time{}, nil
+	}
+	t, err := time.Parse(time.RFC3339Nano, raw)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("%s must be an RFC 3339 time", name)
+	}
+	return t, nil
+}
diff --git a/sandboxd/server/usage_test.go b/sandboxd/server/usage_test.go
new file mode 100644
index 000000000..e9809e5d2
--- /dev/null
+++ b/sandboxd/server/usage_test.go
@@ -0,0 +1,103 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/cocoonstack/sandbox/sandboxd/tenants"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
+)
+
+func TestUsageIsRootOnlyAndNamesAMissingStore(t *testing.T) {
+	ts := newTenantTestServer(t, "root", []tenants.TenantSpec{{Name: "acme", Token: "acme-tok"}}, &fakeManager{}, nil)
+	for token, want := range map[string]int{"acme-tok": http.StatusForbidden, "": http.StatusUnauthorized, "root": http.StatusNotImplemented} {
+		resp := doReq(t, http.MethodGet, ts.URL+"/v1/usage", token, "")
+		body, _ := io.ReadAll(resp.Body)
+		_ = resp.Body.Close()
+		if resp.StatusCode != want {
+			t.Errorf("GET /v1/usage with %q: %d %s, want %d", token, resp.StatusCode, body, want)
+		}
+		if want == http.StatusNotImplemented && !strings.Contains(string(body), "no shared usage store configured") {
+			t.Errorf("GET /v1/usage without a store: %s", body)
+		}
+	}
+}
+
+func TestUsagePassesTheQueryAndPages(t *testing.T) {
+	t0 := time.Date(2026, 10, 30, 1, 2, 3, 400000000, time.UTC)
+	reader := &fakeUsage{recs: []usage.Record{
+		{Cursor: 7, Node: "n1", Digest: [32]byte{1}, Event: usage.Event{Time: t0, Event: "claim", ID: "sb_1", Tenant: "acme"}},
+		{Cursor: 9, Node: "n2", Event: usage.Event{Time: t0, Event: "release", ID: "sb_1"}},
+	}}
+	ts := usageTestServer(t, reader)
+	resp := doReq(t, http.MethodGet, ts.URL+"/v1/usage?tenant=acme&since=2026-10-30T00:00:00Z&until=2026-10-31T00:00:00.5%2B08:00&after=5&limit=2", "root", "")
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	want := `{"events":[{"cursor":7,"node":"n1","t":"2026-10-30T01:02:03.4Z","ev":"claim","id":"sb_1","tenant":"acme"},` +
+		`{"cursor":9,"node":"n2","t":"2026-10-30T01:02:03.4Z","ev":"release","id":"sb_1"}],"next":9}`
+	if resp.StatusCode != http.StatusOK || string(body) != want {
+		t.Errorf("full page: %d %s, want %s", resp.StatusCode, body, want)
+	}
+	until := time.Date(2026, 10, 31, 0, 0, 0, 500000000, time.FixedZone("", 8*3600))
+	if q := reader.got; q.Tenant != "acme" || !q.Since.Equal(time.Date(2026, 10, 30, 0, 0, 0, 0, time.UTC)) || !q.Until.Equal(until) || q.After != 5 || q.Limit != 2 {
+		t.Errorf("reader got %+v", q)
+	}
+
+	resp = doReq(t, http.MethodGet, ts.URL+"/v1/usage?after=9", "root", "")
+	body, _ = io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK || strings.Contains(string(body), "next") || reader.got != (usage.Query{After: 9, Limit: 1000}) {
+		t.Errorf("default partial page: %d %s, reader got %+v", resp.StatusCode, body, reader.got)
+	}
+
+	reader.recs = nil
+	resp = doReq(t, http.MethodGet, ts.URL+"/v1/usage", "root", "")
+	body, _ = io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if string(body) != `{"events":[]}` {
+		t.Errorf("empty page: %s", body)
+	}
+}
+
+func TestUsageRefusesABadQueryAndADownStore(t *testing.T) {
+	reader := &fakeUsage{}
+	ts := usageTestServer(t, reader)
+	for _, q := range []string{"after=-1", "after=x", "limit=0", "limit=10001", "limit=x", "since=yesterday", "until=2026-10-30"} {
+		resp := doReq(t, http.MethodGet, ts.URL+"/v1/usage?"+q, "root", "")
+		_ = resp.Body.Close()
+		if resp.StatusCode != http.StatusBadRequest {
+			t.Errorf("%s: %d, want 400", q, resp.StatusCode)
+		}
+	}
+	reader.err = errors.New("connection refused")
+	resp := doReq(t, http.MethodGet, ts.URL+"/v1/usage", "root", "")
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusServiceUnavailable || strings.Contains(string(body), "refused") {
+		t.Errorf("down store: %d %s, want 503 without the cause", resp.StatusCode, body)
+	}
+}
+
+func usageTestServer(t *testing.T, reader UsageReader) *httptest.Server {
+	t.Helper()
+	srv := New("root", "node:7777", &fakeManager{}, &fakeDialer{}, nil, nil, nil, nil)
+	srv.SetUsage(reader)
+	return serveTestServer(t, srv)
+}
+
+type fakeUsage struct {
+	recs []usage.Record
+	err  error
+	got  usage.Query
+}
+
+func (f *fakeUsage) List(_ context.Context, q usage.Query) ([]usage.Record, error) {
+	f.got = q
+	return f.recs, f.err
+}
diff --git a/sandboxd/usage/pg/pg.go b/sandboxd/usage/pg/pg.go
new file mode 100644
index 000000000..ac8c8afb5
--- /dev/null
+++ b/sandboxd/usage/pg/pg.go
@@ -0,0 +1,122 @@
+// Package pg keeps the usage events of every node of a cell in one shared PostgreSQL table.
+package pg
+
+import (
+	"context"
+	_ "embed"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/cocoonstack/sandbox/sandboxd/metastore"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
+)
+
+const (
+	table = "sandboxd_usage"
+
+	// cellLock serializes a cell's lands, so cursors commit in order and a reader paging by cursor never skips a row.
+	cellLock = `SELECT pg_advisory_xact_lock(hashtextextended('sandboxd_usage/' || $1, 0))`
+
+	insertRow = `INSERT INTO sandboxd_usage (cell, node, line_sha256, t, ev, id, vm, key, tenant, volumes, volumes_rw, children, ref, upstream, secret, tx, rx)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
+		ON CONFLICT (cell, node, line_sha256) DO NOTHING`
+
+	listColumns = `cursor, node, t, ev, id, vm, key, tenant, volumes, volumes_rw, children, ref, upstream, secret, tx, rx`
+)
+
+//go:embed schema.sql
+var schema string
+
+var _ usage.Store = (*Store)(nil)
+
+// Store is the shared usage table.
+type Store struct {
+	pool *pgxpool.Pool
+	cell string
+}
+
+// Open connects to cell's rows and creates the table when it is missing.
+func Open(ctx context.Context, dsn, cell string) (*Store, error) {
+	pool, _, err := metastore.Open(ctx, dsn, table, schema)
+	if err != nil {
+		return nil, err
+	}
+	return &Store{pool: pool, cell: cell}, nil
+}
+
+func (s *Store) Last(ctx context.Context, node string) ([32]byte, error) {
+	qctx, cancel := context.WithTimeout(ctx, metastore.QueryTimeout)
+	defer cancel()
+	var mark [32]byte
+	var digest []byte
+	err := s.pool.QueryRow(qctx, `SELECT line_sha256 FROM sandboxd_usage WHERE cell = $1 AND node = $2 ORDER BY cursor DESC LIMIT 1`, s.cell, node).Scan(&digest)
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		return mark, err
+	}
+	copy(mark[:], digest)
+	return mark, nil
+}
+
+func (s *Store) Land(ctx context.Context, recs []usage.Record) error {
+	qctx, cancel := context.WithTimeout(ctx, metastore.QueryTimeout)
+	defer cancel()
+	return pgx.BeginFunc(qctx, s.pool, func(tx pgx.Tx) error {
+		if _, err := tx.Exec(qctx, cellLock, s.cell); err != nil {
+			return err
+		}
+		var batch pgx.Batch
+		for _, r := range recs {
+			batch.Queue(insertRow, s.cell, r.Node, r.Digest[:], r.Time, r.Event.Event, r.ID, r.VMName, r.KeyHash, r.Tenant, r.Volumes, r.VolumesRW, r.Children,
+				strings.ReplaceAll(r.Reference, "\x00", "\uFFFD"), r.Upstream, r.Secret, r.Sent, r.Received)
+		}
+		return tx.SendBatch(qctx, &batch).Close()
+	})
+}
+
+func (s *Store) List(ctx context.Context, q usage.Query) ([]usage.Record, error) {
+	sql := `SELECT ` + listColumns + ` FROM sandboxd_usage WHERE cell = $1 AND cursor > $2`
+	args := []any{s.cell, q.After}
+	where := func(cond string, arg any) {
+		args = append(args, arg)
+		sql += " AND " + cond + " $" + strconv.Itoa(len(args))
+	}
+	if q.Tenant != "" {
+		where("tenant =", q.Tenant)
+	}
+	if !q.Since.IsZero() {
+		where("t >=", q.Since)
+	}
+	if !q.Until.IsZero() {
+		where("t <", q.Until)
+	}
+	args = append(args, q.Limit)
+	sql += " ORDER BY cursor LIMIT $" + strconv.Itoa(len(args))
+	qctx, cancel := context.WithTimeout(ctx, metastore.QueryTimeout)
+	defer cancel()
+	rows, err := s.pool.Query(qctx, sql, args...)
+	if err != nil {
+		return nil, err
+	}
+	return pgx.CollectRows(rows, scan)
+}
+
+// Close releases the connection pool.
+func (s *Store) Close() {
+	s.pool.Close()
+}
+
+func scan(row pgx.CollectableRow) (usage.Record, error) {
+	var rec usage.Record
+	err := row.Scan(&rec.Cursor, &rec.Node, &rec.Time, &rec.Event.Event, &rec.ID, &rec.VMName, &rec.KeyHash, &rec.Tenant,
+		&rec.Volumes, &rec.VolumesRW, &rec.Children, &rec.Reference, &rec.Upstream, &rec.Secret, &rec.Sent, &rec.Received)
+	if err != nil {
+		return usage.Record{}, fmt.Errorf("scan usage row: %w", err)
+	}
+	rec.Time = rec.Time.UTC()
+	return rec, nil
+}
diff --git a/sandboxd/usage/pg/pg_test.go b/sandboxd/usage/pg/pg_test.go
new file mode 100644
index 000000000..ca9c33ccf
--- /dev/null
+++ b/sandboxd/usage/pg/pg_test.go
@@ -0,0 +1,289 @@
+package pg
+
+import (
+	"cmp"
+	"context"
+	"crypto/sha256"
+	"encoding/json/v2"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/cocoonstack/sandbox/sandboxd/metastore/metastoretest"
+	"github.com/cocoonstack/sandbox/sandboxd/usage"
+)
+
+func TestLandListsInCursorOrderAndResumesFromTheMark(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	ctx := t.Context()
+	if mark, err := s.Last(ctx, "n1"); err != nil || mark != ([32]byte{}) {
+		t.Fatalf("mark of a node with nothing landed: %+v %v", mark, err)
+	}
+	t0 := time.Date(2026, 10, 30, 0, 0, 0, 0, time.UTC)
+	full := usage.Event{
+		Time: t0.Add(time.Microsecond), Event: "claim", ID: "sb_1", VMName: "sbx-1", KeyHash: "k", Tenant: "acme",
+		Volumes: []string{"data", "ro"}, VolumesRW: []string{"data"}, Reference: "ref", Upstream: "up:1", Secret: "gh", Sent: 7, Received: 9,
+	}
+	land(t, s, record("n1", 1, full), record("n2", 1, usage.Event{Time: t0, Event: "claim", ID: "sb_2"}))
+	land(t, s, record("n1", 2, usage.Event{Time: t0.Add(time.Hour), Event: "fork", ID: "sb_1", Children: []string{"sb_3"}}))
+
+	recs, err := s.List(ctx, usage.Query{Limit: 10})
+	if err != nil || len(recs) != 3 {
+		t.Fatalf("list: %d %v", len(recs), err)
+	}
+	if got := recs[0]; got.Node != "n1" || !equalEvents(got.Event, full) {
+		t.Errorf("first record %+v, want n1 %+v", got, full)
+	}
+	if !slices.IsSortedFunc(recs, func(a, b usage.Record) int { return cmp.Compare(a.Cursor, b.Cursor) }) || recs[2].Children[0] != "sb_3" {
+		t.Errorf("records out of cursor order or lost children: %+v", recs)
+	}
+	if mark, _ := s.Last(ctx, "n1"); mark != digest("n1", 2) {
+		t.Errorf("mark of n1 %x, want the digest of its second line", mark)
+	}
+}
+
+func TestListFiltersAndPages(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	ctx := t.Context()
+	t0 := time.Date(2026, 10, 30, 0, 0, 0, 0, time.UTC)
+	for i, tenant := range []string{"acme", "beta", "acme", "acme", "beta"} {
+		land(t, s, record("n1", int64(i)+1, usage.Event{Time: t0.Add(time.Duration(i) * time.Minute), Event: "claim", ID: fmt.Sprintf("%s%d", tenant, i+1), Tenant: tenant}))
+	}
+	all, err := s.List(ctx, usage.Query{Limit: 10})
+	if err != nil || len(all) != 5 {
+		t.Fatalf("list: %d %v", len(all), err)
+	}
+	ids := func(q usage.Query) string {
+		t.Helper()
+		q.Limit = max(q.Limit, 10)
+		recs, listErr := s.List(ctx, q)
+		if listErr != nil {
+			t.Fatalf("list %+v: %v", q, listErr)
+		}
+		var out []string
+		for _, r := range recs {
+			out = append(out, r.ID)
+		}
+		return strings.Join(out, ",")
+	}
+	for _, tt := range []struct {
+		name string
+		q    usage.Query
+		want string
+	}{
+		{"tenant", usage.Query{Tenant: "acme"}, "acme1,acme3,acme4"},
+		{"since", usage.Query{Since: t0.Add(2 * time.Minute)}, "acme3,acme4,beta5"},
+		{"until", usage.Query{Until: t0.Add(2 * time.Minute)}, "acme1,beta2"},
+		{"window and tenant", usage.Query{Tenant: "beta", Since: t0.Add(time.Minute), Until: t0.Add(4 * time.Minute)}, "beta2"},
+		{"after", usage.Query{After: all[2].Cursor}, "acme4,beta5"},
+		{"after and tenant", usage.Query{Tenant: "acme", After: all[0].Cursor}, "acme3,acme4"},
+	} {
+		if got := ids(tt.q); got != tt.want {
+			t.Errorf("%s: %s, want %s", tt.name, got, tt.want)
+		}
+	}
+	page, err := s.List(ctx, usage.Query{Limit: 2})
+	if err != nil || len(page) != 2 || page[1].Cursor != all[1].Cursor {
+		t.Errorf("first page of 2: %+v %v", page, err)
+	}
+	if empty, err := s.List(ctx, usage.Query{After: all[4].Cursor, Limit: 10}); err != nil || len(empty) != 0 {
+		t.Errorf("page past the last cursor: %+v %v", empty, err)
+	}
+}
+
+func TestCellsKeepTheirOwnRows(t *testing.T) {
+	dsn := metastoretest.PGSchema(t)
+	c1, c2 := open(t, dsn, "c1"), open(t, dsn, "c2")
+	ev := usage.Event{Time: time.Now(), Event: "claim", ID: "sb_1"}
+	land(t, c1, record("n1", 1, ev))
+	land(t, c2, record("n1", 1, ev))
+	if recs, err := c2.List(t.Context(), usage.Query{Limit: 10}); err != nil || len(recs) != 1 {
+		t.Errorf("c2 lists %d rows %v, want its own 1", len(recs), err)
+	}
+	if mark, _ := c2.Last(t.Context(), "n1"); mark != digest("n1", 1) {
+		t.Errorf("c2's mark of n1 %x, want its one line", mark)
+	}
+}
+
+func TestANulInAReferenceLands(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	land(t, s, record("n1", 1, usage.Event{Time: time.Now(), Event: "egress", ID: "sb_1", Reference: "a\x00b"}))
+	if recs, err := s.List(t.Context(), usage.Query{Limit: 1}); err != nil || len(recs) != 1 || recs[0].Reference != "a\uFFFDb" {
+		t.Errorf("list: %+v %v", recs, err)
+	}
+}
+
+func TestLandWaitsForTheCellLock(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	ctx := t.Context()
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("begin: %v", err)
+	}
+	if _, err = tx.Exec(ctx, cellLock, "c1"); err != nil {
+		t.Fatalf("take the cell lock: %v", err)
+	}
+	landed := make(chan error, 1)
+	go func() {
+		landed <- s.Land(ctx, []usage.Record{record("n1", 1, usage.Event{Time: time.Now(), Event: "claim", ID: "sb_1"})})
+	}()
+	select {
+	case err = <-landed:
+		t.Fatalf("land under a held cell lock returned %v", err)
+	case <-time.After(300 * time.Millisecond):
+	}
+	_ = tx.Rollback(ctx)
+	if err = <-landed; err != nil {
+		t.Errorf("land after the lock: %v", err)
+	}
+}
+
+func TestSchemaIndexesTheQueries(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	var n int
+	if err := s.pool.QueryRow(t.Context(), `SELECT count(*) FROM pg_indexes
+		WHERE schemaname = current_schema() AND tablename = 'sandboxd_usage'`).Scan(&n); err != nil || n != 5 {
+		t.Errorf("usage index count %d %v, want 5 (primary key, node digest, node cursor, tenant, time)", n, err)
+	}
+}
+
+func TestShipLandsAJournalOnceAcrossRestarts(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	path := filepath.Join(t.TempDir(), usage.JournalFile)
+	writeJournal(t, path, "sb_1", "sb_2")
+	ship(t, s, path, 2)
+	writeJournal(t, path, "sb_3")
+	if err := os.Rename(path, path+".1"); err != nil {
+		t.Fatalf("rotate: %v", err)
+	}
+	writeJournal(t, path, "sb_4")
+	ship(t, s, path, 4)
+	ship(t, s, path, 4)
+	recs, err := s.List(t.Context(), usage.Query{Limit: 10})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	for i, r := range recs {
+		if want := fmt.Sprintf("sb_%d", i+1); r.ID != want || r.Node != "n1" {
+			t.Errorf("record %d: %s node %s, want %s n1", i, r.ID, r.Node, want)
+		}
+	}
+}
+
+func TestShipLandsNoLineTwiceAfterTheJournalLosesItsLandedTail(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	path := filepath.Join(t.TempDir(), usage.JournalFile)
+	writeJournal(t, path, "sb_1", "sb_2", "sb_3")
+	ship(t, s, path, 3)
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read journal: %v", err)
+	}
+	if err = os.WriteFile(path, []byte(strings.Join(strings.SplitAfter(string(raw), "\n")[:2], "")), 0o600); err != nil {
+		t.Fatalf("cut journal: %v", err)
+	}
+	writeJournal(t, path, "sb_4")
+	ship(t, s, path, 4)
+	recs, err := s.List(t.Context(), usage.Query{Limit: 10})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	for i, r := range recs {
+		if want := fmt.Sprintf("sb_%d", i+1); r.ID != want {
+			t.Errorf("record %d: %s, want %s", i, r.ID, want)
+		}
+	}
+}
+
+func TestLandSkipsALandedLine(t *testing.T) {
+	s := open(t, metastoretest.PGSchema(t), "c1")
+	first, second := usage.Event{Time: time.Now(), Event: "claim", ID: "sb_1"}, usage.Event{Time: time.Now(), Event: "release", ID: "sb_1"}
+	land(t, s, record("n1", 1, first))
+	land(t, s, record("n1", 1, first), record("n1", 2, second))
+	recs, err := s.List(t.Context(), usage.Query{Limit: 10})
+	if err != nil || len(recs) != 2 || recs[0].Event.Event != "claim" || recs[1].Event.Event != "release" {
+		t.Errorf("rows after relanding a digest: %+v %v, want the claim and the release once each", recs, err)
+	}
+	if mark, _ := s.Last(t.Context(), "n1"); mark != digest("n1", 2) {
+		t.Errorf("mark %x, want line 2", mark)
+	}
+}
+
+func open(t *testing.T, dsn, cell string) *Store {
+	t.Helper()
+	s, err := Open(t.Context(), dsn, cell)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	t.Cleanup(s.Close)
+	return s
+}
+
+func land(t *testing.T, s *Store, recs ...usage.Record) {
+	t.Helper()
+	if err := s.Land(t.Context(), recs); err != nil {
+		t.Fatalf("land: %v", err)
+	}
+}
+
+func record(node string, line int64, ev usage.Event) usage.Record {
+	return usage.Record{Node: node, Digest: digest(node, line), Event: ev}
+}
+
+func digest(node string, line int64) [32]byte {
+	return sha256.Sum256(fmt.Appendf(nil, "%s/%d", node, line))
+}
+
+func equalEvents(a, b usage.Event) bool {
+	x, _ := json.Marshal(a)
+	y, _ := json.Marshal(b)
+	return string(x) == string(y)
+}
+
+func writeJournal(t *testing.T, path string, ids ...string) {
+	t.Helper()
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		t.Fatalf("open journal: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	for _, id := range ids {
+		line, _ := json.Marshal(usage.Event{Time: time.Now(), Event: "claim", ID: id})
+		if _, err = f.Write(append(line, '\n')); err != nil {
+			t.Fatalf("append: %v", err)
+		}
+	}
+}
+
+func ship(t *testing.T, s *Store, path string, want int) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		usage.Ship(ctx, path, "n1", s)
+	}()
+	deadline := time.Now().Add(10 * time.Second)
+	for count(t, s) < want && time.Now().Before(deadline) {
+		time.Sleep(20 * time.Millisecond)
+	}
+	time.Sleep(300 * time.Millisecond)
+	cancel()
+	<-done
+	if n := count(t, s); n != want {
+		t.Fatalf("%d rows landed, want %d", n, want)
+	}
+}
+
+func count(t *testing.T, s *Store) int {
+	t.Helper()
+	recs, err := s.List(t.Context(), usage.Query{Limit: 100})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	return len(recs)
+}
diff --git a/sandboxd/usage/pg/schema.sql b/sandboxd/usage/pg/schema.sql
new file mode 100644
index 000000000..de81d6562
--- /dev/null
+++ b/sandboxd/usage/pg/schema.sql
@@ -0,0 +1,25 @@
+CREATE TABLE IF NOT EXISTS sandboxd_usage (
+    cell        text        NOT NULL,
+    cursor      bigint      GENERATED ALWAYS AS IDENTITY,
+    node        text        NOT NULL,
+    line_sha256 bytea       NOT NULL,
+    t           timestamptz NOT NULL,
+    ev          text        NOT NULL,
+    id          text        NOT NULL,
+    vm          text        NOT NULL DEFAULT '',
+    key         text        NOT NULL DEFAULT '',
+    tenant      text        NOT NULL DEFAULT '',
+    volumes     text[],
+    volumes_rw  text[],
+    children    text[],
+    ref         text        NOT NULL DEFAULT '',
+    upstream    text        NOT NULL DEFAULT '',
+    secret      text        NOT NULL DEFAULT '',
+    tx          bigint      NOT NULL DEFAULT 0,
+    rx          bigint      NOT NULL DEFAULT 0,
+    PRIMARY KEY (cell, cursor),
+    UNIQUE (cell, node, line_sha256)
+);
+CREATE INDEX IF NOT EXISTS sandboxd_usage_node ON sandboxd_usage (cell, node, cursor);
+CREATE INDEX IF NOT EXISTS sandboxd_usage_tenant ON sandboxd_usage (cell, tenant, cursor);
+CREATE INDEX IF NOT EXISTS sandboxd_usage_t ON sandboxd_usage (cell, t);
diff --git a/sandboxd/usage/ship.go b/sandboxd/usage/ship.go
new file mode 100644
index 000000000..827d14341
--- /dev/null
+++ b/sandboxd/usage/ship.go
@@ -0,0 +1,223 @@
+package usage
+
+import (
+	"bufio"
+	"cmp"
+	"context"
+	"crypto/sha256"
+	"encoding/json/jsontext"
+	"encoding/json/v2"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"iter"
+	"math"
+	"os"
+	"time"
+
+	"github.com/projecteru2/core/log"
+)
+
+const (
+	shipEvery = 5 * time.Second
+	shipBatch = 1000
+)
+
+// Ship lands the journal at path in store as node's events until ctx ends; it only reads the journal.
+func Ship(ctx context.Context, path, node string, store Store) {
+	s := &shipper{path: path, node: node, store: store}
+	tick := time.Tick(shipEvery)
+	for {
+		s.drain(ctx)
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick:
+		}
+	}
+}
+
+// shipper resumes after the line whose digest the store's mark names; at is the last line read, a hint that survives a rename.
+type shipper struct {
+	path    string
+	node    string
+	store   Store
+	mark    [32]byte
+	synced  bool
+	at      anchor
+	failing bool
+}
+
+func (s *shipper) drain(ctx context.Context) {
+	logger := log.WithFunc("usage.Ship")
+	for {
+		n, err := s.shipOnce(ctx)
+		switch {
+		case err != nil:
+			s.synced = false
+			if !s.failing {
+				logger.Warnf(ctx, "usage events stop landing (%v); the journal keeps them, retrying every %s", err, shipEvery)
+			}
+			s.failing = true
+			return
+		case s.failing:
+			logger.Info(ctx, "usage events land again")
+			s.failing = false
+		}
+		if n < shipBatch {
+			return
+		}
+	}
+}
+
+// shipOnce lands the next batch of lines and returns how many it read.
+func (s *shipper) shipOnce(ctx context.Context) (int, error) {
+	if !s.synced {
+		mark, err := s.store.Last(ctx, s.node)
+		if err != nil {
+			return 0, fmt.Errorf("read mark: %w", err)
+		}
+		if mark != s.mark {
+			s.at = anchor{}
+		}
+		s.mark, s.synced = mark, true
+	}
+	live, err := os.Open(s.path)
+	if err != nil {
+		return 0, err
+	}
+	defer func() { _ = live.Close() }()
+	backup, err := os.Open(s.path + ".1")
+	if err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return 0, err
+	}
+	if backup != nil {
+		defer func() { _ = backup.Close() }()
+		if sameFile(live, backup) {
+			return 0, nil // rotated between the two opens
+		}
+	}
+	f, off := s.locate(live, backup)
+	files := []*os.File{f}
+	if f == backup {
+		files = append(files, live)
+	}
+	recs, last, n := s.read(ctx, files, off)
+	if len(recs) > 0 {
+		if err = s.store.Land(ctx, recs); err != nil {
+			return 0, fmt.Errorf("land %d events: %w", len(recs), err)
+		}
+		s.mark = recs[len(recs)-1].Digest
+	}
+	if n > 0 {
+		s.at = last
+	}
+	return n, nil
+}
+
+// locate returns where the first unlanded line starts: past the anchor when a file still holds it, else past the marked line, else at the oldest line.
+func (s *shipper) locate(live, backup *os.File) (*os.File, int64) {
+	for _, f := range []*os.File{live, backup} {
+		if f != nil && s.at.in(f) {
+			return f, s.at.end
+		}
+	}
+	oldest := cmp.Or(backup, live)
+	if s.mark == ([32]byte{}) {
+		return oldest, 0
+	}
+	var found *os.File
+	for _, f := range []*os.File{backup, live} {
+		if f == nil {
+			continue
+		}
+		for line, end := range lines(f, 0) {
+			if a := newAnchor(line, end); a.digest == s.mark {
+				found, s.at = f, a
+			}
+		}
+	}
+	if found == nil {
+		return oldest, 0
+	}
+	return found, s.at.end
+}
+
+// read decodes up to shipBatch lines of files, the first from off, into s.node's next records.
+func (s *shipper) read(ctx context.Context, files []*os.File, off int64) ([]Record, anchor, int) {
+	var (
+		recs []Record
+		last anchor
+		n    int
+	)
+	for _, f := range files {
+		for line, end := range lines(f, off) {
+			n++
+			last = newAnchor(line, end)
+			ev, err := decode(line)
+			if err != nil {
+				log.WithFunc("usage.Ship").Warnf(ctx, "skip undecodable line ending at %s:%d: %v", f.Name(), end, err)
+			} else {
+				recs = append(recs, Record{Node: s.node, Digest: last.digest, Event: ev})
+			}
+			if n == shipBatch {
+				return recs, last, n
+			}
+		}
+		off = 0
+	}
+	return recs, last, n
+}
+
+// anchor is one journal line by content and by the offset just past it.
+type anchor struct {
+	digest [32]byte
+	end    int64
+	size   int64
+}
+
+func newAnchor(line []byte, end int64) anchor {
+	return anchor{digest: sha256.Sum256(line[:len(line)-1]), end: end, size: int64(len(line))}
+}
+
+// in reports whether f still holds the anchor's line at its offset.
+func (a anchor) in(f *os.File) bool {
+	if a.size == 0 {
+		return false
+	}
+	buf := make([]byte, a.size)
+	if _, err := f.ReadAt(buf, a.end-a.size); err != nil {
+		return false
+	}
+	return buf[a.size-1] == '\n' && sha256.Sum256(buf[:a.size-1]) == a.digest
+}
+
+// lines yields each newline-terminated line of f from off, newline included, with the offset just past it.
+func lines(f *os.File, off int64) iter.Seq2[[]byte, int64] {
+	return func(yield func([]byte, int64) bool) {
+		r := bufio.NewReader(io.NewSectionReader(f, off, math.MaxInt64-off))
+		for {
+			line, err := r.ReadBytes('\n')
+			if err != nil {
+				return
+			}
+			off += int64(len(line))
+			if !yield(line, off) {
+				return
+			}
+		}
+	}
+}
+
+func decode(line []byte) (Event, error) {
+	var ev Event
+	err := json.Unmarshal(line[:len(line)-1], &ev, jsontext.AllowInvalidUTF8(true))
+	return ev, err
+}
+
+func sameFile(a, b *os.File) bool {
+	sa, errA := a.Stat()
+	sb, errB := b.Stat()
+	return errA == nil && errB == nil && os.SameFile(sa, sb)
+}
diff --git a/sandboxd/usage/ship_test.go b/sandboxd/usage/ship_test.go
new file mode 100644
index 000000000..4ad144062
--- /dev/null
+++ b/sandboxd/usage/ship_test.go
@@ -0,0 +1,306 @@
+package usage
+
+import (
+	"context"
+	"encoding/json/v2"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+var errDown = errors.New("store down")
+
+func TestShipLandsEachLineOnceAcrossRestarts(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	appendEvents(t, path, "a", "b", "c")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+	appendEvents(t, path, "d", "e")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c,d,e")
+}
+
+func TestShipFollowsARotation(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	s := newShipper(path, store)
+	appendEvents(t, path, "a", "b")
+	s.drain(t.Context())
+	appendEvents(t, path, "c")
+	rotate(t, path)
+	appendEvents(t, path, "d")
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b,c,d")
+	appendEvents(t, path, "e")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c,d,e")
+}
+
+func TestShipBackfillsOnceTheStoreReturns(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	s := newShipper(path, store)
+	appendEvents(t, path, "a")
+	s.drain(t.Context())
+	store.setDown(true)
+	appendEvents(t, path, "b")
+	s.drain(t.Context())
+	rotate(t, path)
+	appendEvents(t, path, "c")
+	s.drain(t.Context())
+	if !s.failing {
+		t.Error("shipper with the store down is not failing")
+	}
+	assertLanded(t, store, "a")
+	store.setDown(false)
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+	if s.failing {
+		t.Error("shipper still failing after the store returned")
+	}
+}
+
+func TestShipConvergesAfterALostReply(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	s := newShipper(path, store)
+	appendEvents(t, path, "a", "b")
+	store.loseReplies(1)
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b")
+	appendEvents(t, path, "c")
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+	if s.failing {
+		t.Error("shipper failing after the resync")
+	}
+}
+
+func TestShipLandsWhatTheJournalKeepsAfterTwoRotations(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	s := newShipper(path, store)
+	appendEvents(t, path, "a")
+	s.drain(t.Context())
+	store.setDown(true)
+	appendEvents(t, path, "b")
+	rotate(t, path)
+	appendEvents(t, path, "c")
+	rotate(t, path)
+	appendEvents(t, path, "d")
+	s.drain(t.Context())
+	store.setDown(false)
+	s.drain(t.Context())
+	assertLanded(t, store, "a,c,d")
+}
+
+func TestShipStartsOverOnANewJournal(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	appendEvents(t, path, "a", "b")
+	newShipper(path, store).drain(t.Context())
+	if err := os.Remove(path); err != nil {
+		t.Fatalf("remove journal: %v", err)
+	}
+	appendEvents(t, path, "c")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+}
+
+func TestShipLandsNoLineTwiceAfterTheJournalLosesItsLandedTail(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	appendEvents(t, path, "a", "b", "c")
+	newShipper(path, store).drain(t.Context())
+	keepLines(t, path, 2)
+	appendEvents(t, path, "d")
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, "a,b,c,d")
+}
+
+func TestShipSkipsAnUndecodableLineAndWaitsForATornOne(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	s := newShipper(path, store)
+	appendEvents(t, path, "a")
+	appendRaw(t, path, "{\"t\":\"x\"\n")
+	appendEvents(t, path, "b")
+	appendRaw(t, path, `{"t":"2026-10-30T00:00:00Z","ev":"claim",`)
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b")
+	appendRaw(t, path, `"id":"c"}`+"\n")
+	s.drain(t.Context())
+	assertLanded(t, store, "a,b,c")
+}
+
+func TestShipDrainsABacklogPastOneBatch(t *testing.T) {
+	path, store := journalAt(t), &memStore{}
+	ids := make([]string, shipBatch+5)
+	for i := range ids {
+		ids[i] = fmt.Sprintf("sb_%d", i)
+	}
+	appendEvents(t, path, ids...)
+	newShipper(path, store).drain(t.Context())
+	assertLanded(t, store, strings.Join(ids, ","))
+	if store.lands != 2 {
+		t.Errorf("%d lands for %d lines, want 2", store.lands, len(ids))
+	}
+}
+
+func TestShipLandsEveryTickUntilItsContextEnds(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		path, store := journalAt(t), &memStore{}
+		appendEvents(t, path, "a")
+		ctx, cancel := context.WithCancel(t.Context())
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			Ship(ctx, path, "n1", store)
+		}()
+		synctest.Wait()
+		assertLanded(t, store, "a")
+		appendEvents(t, path, "b")
+		time.Sleep(shipEvery)
+		synctest.Wait()
+		assertLanded(t, store, "a,b")
+		cancel()
+		<-done
+	})
+}
+
+func journalAt(t *testing.T) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), JournalFile)
+	appendRaw(t, path, "")
+	return path
+}
+
+func newShipper(path string, store Store) *shipper {
+	return &shipper{path: path, node: "n1", store: store}
+}
+
+func appendEvents(t *testing.T, path string, ids ...string) {
+	t.Helper()
+	var b strings.Builder
+	for _, id := range ids {
+		line, err := json.Marshal(Event{Time: time.Now(), Event: "claim", ID: id})
+		if err != nil {
+			t.Fatalf("marshal: %v", err)
+		}
+		b.Write(line)
+		b.WriteByte('\n')
+	}
+	appendRaw(t, path, b.String())
+}
+
+func appendRaw(t *testing.T, path, s string) {
+	t.Helper()
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		t.Fatalf("open journal: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	if _, err = f.WriteString(s); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+}
+
+func keepLines(t *testing.T, path string, n int) {
+	t.Helper()
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read journal: %v", err)
+	}
+	lines := strings.SplitAfter(string(raw), "\n")
+	if err = os.WriteFile(path, []byte(strings.Join(lines[:n], "")), 0o600); err != nil {
+		t.Fatalf("cut journal: %v", err)
+	}
+}
+
+func rotate(t *testing.T, path string) {
+	t.Helper()
+	if err := os.Rename(path, path+".1"); err != nil {
+		t.Fatalf("rotate: %v", err)
+	}
+	appendRaw(t, path, "")
+}
+
+func assertLanded(t *testing.T, store *memStore, want string) {
+	t.Helper()
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	ids := make([]string, len(store.recs))
+	for i, r := range store.recs {
+		ids[i] = r.ID
+		if r.Cursor != int64(i)+1 || r.Node != "n1" {
+			t.Errorf("record %d: cursor %d node %s, want cursor %d and node n1", i, r.Cursor, r.Node, i+1)
+		}
+	}
+	if got := strings.Join(ids, ","); got != want {
+		t.Errorf("landed %s, want %s", got, want)
+	}
+}
+
+// memStore lands like the PostgreSQL store: whole batches that skip a landed digest.
+type memStore struct {
+	mu    sync.Mutex
+	recs  []Record
+	down  bool
+	lose  int
+	lands int
+}
+
+func (m *memStore) Last(_ context.Context, node string) ([32]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var mark [32]byte
+	if m.down {
+		return mark, errDown
+	}
+	for _, r := range m.recs {
+		if r.Node == node {
+			mark = r.Digest
+		}
+	}
+	return mark, nil
+}
+
+func (m *memStore) Land(_ context.Context, recs []Record) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.down {
+		return errDown
+	}
+	for _, r := range recs {
+		if !slices.ContainsFunc(m.recs, func(o Record) bool { return o.Node == r.Node && o.Digest == r.Digest }) {
+			r.Cursor = int64(len(m.recs)) + 1
+			m.recs = append(m.recs, r)
+		}
+	}
+	m.lands++
+	if m.lose > 0 {
+		m.lose--
+		return errDown
+	}
+	return nil
+}
+
+func (m *memStore) List(context.Context, Query) ([]Record, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return slices.Clone(m.recs), nil
+}
+
+func (m *memStore) setDown(down bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.down = down
+}
+
+func (m *memStore) loseReplies(n int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lose = n
+}
diff --git a/sandboxd/usage/usage.go b/sandboxd/usage/usage.go
new file mode 100644
index 000000000..44f82a51b
--- /dev/null
+++ b/sandboxd/usage/usage.go
@@ -0,0 +1,55 @@
+// Package usage defines the usage journal's events and lands a node's journal in a store the whole cell shares.
+package usage
+
+import (
+	"context"
+	"time"
+)
+
+// JournalFile is the usage journal's name under data_dir; it rotates to JournalFile+".1".
+const JournalFile = "usage.jsonl"
+
+// Event is one line of usage.jsonl, the billing stream the platform collector folds.
+type Event struct {
+	Time      time.Time `json:"t"`
+	Event     string    `json:"ev"`
+	ID        string    `json:"id"`
+	VMName    string    `json:"vm,omitempty"`
+	KeyHash   string    `json:"key,omitempty"`
+	Tenant    string    `json:"tenant,omitempty"`
+	Volumes   []string  `json:"volumes,omitempty"`
+	VolumesRW []string  `json:"volumes_rw,omitempty"`
+	Children  []string  `json:"children,omitempty"`
+	Reference string    `json:"ref,omitempty"`
+	Upstream  string    `json:"upstream,omitempty"`
+	Secret    string    `json:"secret,omitempty"`
+	Sent      int64     `json:"tx,omitzero"`
+	Received  int64     `json:"rx,omitzero"`
+}
+
+// Record is one landed event; Cursor orders the store and Digest is the journal line's SHA-256.
+type Record struct {
+	Cursor int64    `json:"cursor"`
+	Node   string   `json:"node"`
+	Digest [32]byte `json:"-"`
+	Event
+}
+
+// Query selects events with a cursor above After, of Tenant when set, timed in [Since, Until) where set, at most Limit.
+type Query struct {
+	Tenant string
+	Since  time.Time
+	Until  time.Time
+	After  int64
+	Limit  int
+}
+
+// Store keeps the landed events of every node of a cell.
+type Store interface {
+	// Last returns the digest of node's newest landed line; zero when none landed.
+	Last(ctx context.Context, node string) ([32]byte, error)
+	// Land appends recs in one transaction, skipping a line whose node and digest already landed; cursors grow in commit order.
+	Land(ctx context.Context, recs []Record) error
+	// List returns the events q selects in cursor order.
+	List(ctx context.Context, q Query) ([]Record, error)
+}
diff --git a/sdk/python/tests/test_volume_controller.py b/sdk/python/tests/test_volume_controller.py
index c43036be7..e0215f72d 100644
--- a/sdk/python/tests/test_volume_controller.py
+++ b/sdk/python/tests/test_volume_controller.py
@@ -7,16 +7,17 @@
 from examples.volume_controller import VolumeHooks, hibernate_volume, resume_volume
 
 PATH = "/v1/sandboxes/sb_1"
+HOME = {"name": "home", "mode": "rw", "attach_only": True}
 EMPTY = {"volumes_want": [], "volumes": [], "states": [], "converging": False}
 ATTACHED = {
-    "volumes_want": [{"name": "home", "mode": "rw", "attach_only": True}],
-    "volumes": [{"name": "home", "mode": "rw", "attach_only": True}],
+    "volumes_want": [HOME],
+    "volumes": [HOME],
     "states": [{"name": "home", "state": "attached"}],
     "converging": False,
 }
 
 REMOVE = ("patch", {"remove": ["home"]})
-ADD = ("patch", {"add": [{"name": "home", "mode": "rw", "attach_only": True}]})
+ADD = ("patch", {"add": [HOME]})
 REMOVE_STEPS = ["close", "reserved", "unmount", REMOVE]
 ADD_STEPS = ["close", "reserved", "wake", ADD]
 
@@ -75,7 +76,7 @@ def test_controller_cycle_opens_after_verification(controller):
     resume_volume(controller.sb, "home", controller.hooks)
     assert controller.events == REMOVE_STEPS + ["hibernate"] + ADD_STEPS + ["verify", "open"]
     assert controller.gate["open"]
-    assert controller.sb.volumes == [{"name": "home", "mode": "rw", "attach_only": True}]
+    assert controller.sb.volumes == [HOME]
 
 
 @pytest.mark.parametrize(