Add a usage/event ledger for per-namespace metering, audit, and on-chain attribution

[harrymove-ctrl]

Context

The hosted Worker has zero metering or audit. No usage table exists in D1 (migrations 0001–0006 cover accounts, namespaces, growth loops, schedules — none record per-action usage). Authed users are hardcoded quota: { ... unlimited: true } (apps/api/src/worker.ts:1503) and the only counter is the 1/day demo-IP quota (consumeDemoQuota, worker.ts:4296; table contextmem_demo_limits). The roadmap calls this ledger "the shared spine feeding billing, observability, AND on-chain attribution — build it first."

Goal / user story

As the platform owner, I want every billable/auditable action (extract, recall, remember, mcp_read, ai_query, chat) recorded as a durable per-namespace event so I can meter usage, power an observability dashboard, enforce quotas, and later anchor on-chain attribution receipts — built once.

Acceptance criteria

• New migration 0007_usage_events.sql creates contextmem_usage_events (id, ts, owner_id, namespace, action with a CHECK on the enum, units INTEGER, request_id, source, metadata_json) + indexes on (namespace, ts) and (owner_id, ts).
• A single recordUsage(env, { ownerId, namespace, action, units, requestId, source, metadata }) helper inserts one row, never throws (try/catch + ctx.waitUntil), and is unit-tested.
• Events are emitted from the hosted handlers: extraction/build (createNamespaceBuildJob), recall (/api/memwal/recall, worker.ts:631), chat (memwalChat, worker.ts:634/3498), ai-query (aiQueryRun, worker.ts:1349), hosted MCP reads (/mcp, isMcpPath worker.ts:800), and demo extractions.
• Each event carries a stable request_id for idempotency so retries don't double-count.
• GET /api/usage?namespace=&from=&to= returns aggregated counts/units per action (owner-scoped) using the index, not a full-table scan.
• No request bodies or secrets are stored: a test asserts Authorization/delegate keys never leak into metadata_json.

Implementation notes

Reuse the D1 upsert pattern from consumeDemoQuota (worker.ts:4304). Wrap every emit in ctx.waitUntil(recordUsage(...)) so logging never blocks responses. Define the action union in a shared module so packages/mcp tool calls can emit the same events. Keep the table append-only; aggregate with SELECT action, COUNT(*), SUM(units) ... GROUP BY action. A daily rollup table can come later if volume grows — not required now. No wrangler.jsonc change (uses the existing CONTEXTMEM_DB binding).

Sui Overflow angle

This ledger is the precondition for on-chain attribution receipts: each metered extract/recall/remember becomes the exact payload a Talus/Nexus node anchors on Sui (caller, namespace, action, artifactDigest, ts). Building it now turns the later receipt work into a thin write-through over existing rows rather than a from-scratch capture layer during the demo crunch.

Dependencies

Owner-scoping of /api/usage depends on the "hosted accounts + owner-scoped authorization" issue (otherwise ownerId is spoofable). The on-chain attribution-receipt issue consumes this ledger.

Part of the ContextMEM roadmap (#4) • Sui Overflow build.

[harrymove-ctrl]

Implementation Plan

Approach
Add one append-only D1 table (contextmem_usage_events) plus a single recordUsage(env, {...}) helper that never throws and is always wrapped in ctx.waitUntil(...), modeled on the existing consumeDemoQuota upsert pattern (apps/api/src/worker.ts:4296). Emit one event from each hosted handler (extract/build, recall, chat, ai-query, hosted MCP reads, demo extractions), keyed by a stable per-operation request_id that an INSERT … ON CONFLICT(request_id, action) DO NOTHING uses for idempotency. Expose GET /api/usage that aggregates with a GROUP BY action query served off the (owner_id, ts) / (namespace, ts) indexes. The action union lives in @contextmem/core so packages/mcp can reference the same enum; the D1-bound recordUsage helper lives in the Worker (D1 only exists there). This is intentionally built once as the shared spine for billing, observability, and the later on-chain attribution receipt (roadmap §6 "build it first").

Prerequisites & dependencies

• No new bindings. Reuses the existing CONTEXTMEM_DB D1 binding (wrangler.jsonc:22-28); the migration auto-applies via migrations_dir. No wrangler.jsonc edit.
• Soft dependency — owner-scoped auth: truly trustworthy owner_id on /api/usage depends on the roadmap's "Real hosted accounts + owner-scoped authorization" item (§6 P1). Until that lands, owner_id is derived from readHostedRunAuth → hostedOwnerId(accountId) (worker.ts:1461,1512) / demoOwnerId (worker.ts:4317), which is spoofable via the x-memwal-account-id header. Ship the ledger now; the /api/usage scoping tightens automatically when accounts land. Call this out in the PR.
• Consumed by: the future on-chain attribution-receipt issue reads these rows; keep the table append-only and the columns receipt-shaped (caller=owner_id, namespace, action, artifact digest in metadata, ts).
• No external creds, no Sui/Harbor work — this issue is pure D1 + Worker.

Step-by-step

1. Migration 0007_usage_events.sql (new, apps/api/migrations/0007_usage_events.sql). Follow the style of 0001_account_gate.sql (TEXT timestamps, IF NOT EXISTS):

   CREATE TABLE IF NOT EXISTS contextmem_usage_events (
     id TEXT PRIMARY KEY,
     ts TEXT NOT NULL,
     owner_id TEXT NOT NULL,
     namespace TEXT,
     action TEXT NOT NULL CHECK (action IN ('extract','recall','remember','mcp_read','ai_query','chat')),
     units INTEGER NOT NULL DEFAULT 1,
     request_id TEXT NOT NULL,
     source TEXT NOT NULL DEFAULT 'api',
     metadata_json TEXT
   );
   CREATE INDEX IF NOT EXISTS contextmem_usage_events_ns_ts_idx    ON contextmem_usage_events(namespace, ts);
   CREATE INDEX IF NOT EXISTS contextmem_usage_events_owner_ts_idx ON contextmem_usage_events(owner_id, ts);
   CREATE UNIQUE INDEX IF NOT EXISTS contextmem_usage_events_req_action_uidx ON contextmem_usage_events(request_id, action);

   The UNIQUE (request_id, action) index is what makes ON CONFLICT DO NOTHING idempotent on retries.
2. Shared action union in packages/core (new packages/core/src/usage.ts; add export * from "./usage.js"; to packages/core/src/index.ts:13):

   export const USAGE_ACTIONS = ['extract','recall','remember','mcp_read','ai_query','chat'] as const;
   export type UsageAction = (typeof USAGE_ACTIONS)[number];
   export type UsageSource = 'web' | 'api' | 'mcp' | 'cron' | 'demo';

   This is the single source of truth the Worker CHECK enum, recordUsage, and packages/mcp all reference.
3. recordUsage + aggregateUsage helper (new apps/api/src/usage.ts, imported by worker.ts). recordUsage must mirror the consumeDemoQuota shape (worker.ts:4296-4311) but be fail-safe:

   export async function recordUsage(env, e: { ownerId: string; namespace?: string|null; action: UsageAction; units?: number; requestId: string; source?: UsageSource; metadata?: Record<string, unknown>; }): Promise<void> {
     try {
       await env.CONTEXTMEM_DB.prepare(
         `INSERT INTO contextmem_usage_events (id, ts, owner_id, namespace, action, units, request_id, source, metadata_json)
          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
          ON CONFLICT(request_id, action) DO NOTHING`
       ).bind(crypto.randomUUID(), new Date().toISOString(), e.ownerId, e.namespace ?? null, e.action, e.units ?? 1, e.requestId, e.source ?? 'api', e.metadata ? JSON.stringify(sanitizeUsageMetadata(e.metadata)) : null).run();
     } catch { /* metering must never break a response */ }
   }

   Add sanitizeUsageMetadata() that drops any key matching /authorization|bearer|token|secret|private[-_]?key|delegate|cookie|x-memwal/i and caps value length — this is what the leak test asserts against (AC Build HarborClient: reserve→sign→finalize bucket, multipart upload, status poll, download #6). Add aggregateUsage(env, ownerId, namespace?, from?, to?) running SELECT action, COUNT(*) AS count, SUM(units) AS units FROM contextmem_usage_events WHERE owner_id = ? [AND namespace = ?] [AND ts >= ?] [AND ts < ?] GROUP BY action — owner_id-first so the (owner_id, ts) index is used (no full scan, AC Spike: run @mysten/seal encrypt+decrypt inside a Cloudflare Worker #5).
4. Emit extract from build/extract in createNamespaceBuildJob (worker.ts:1901-1941). It already has input.ownerId, namespace, the stable jobId (worker.ts:1907), and ctx. After the insert + queue send, add if (ctx.waitUntil) ctx.waitUntil(recordUsage(env, { ownerId: input.ownerId, namespace, action: 'extract', units: sources.length, requestId: jobId, source: input.ownerId.startsWith('demo:') ? 'demo' : 'web', metadata: { buildKind, sourceCount: sources.length } })). jobId as request_id means queue retries / re-invocation don't double-count (AC docs: add product roadmap (Harbor storage + Talus on-chain direction) #4). This single site covers both createExtraction/createNamespaceBuild and demo extractions (demo flows through createExtractionJob → createNamespaceBuildJob with demoOwnerId, AC bullet "demo extractions").
5. Emit recall from memwalRecall (worker.ts:3408). Thread ctx into the handler (update call site worker.ts:632 to memwalRecall(request, env, ctx) and the signature). Resolve owner: readHostedRunAuth(request)?.ownerId ?? demoOwnerId(request). On a returned response, ctx.waitUntil(recordUsage(env, { ownerId, namespace, action: 'recall', requestId: usageRequestId(request), source: 'api', metadata: { source: <'walrus-memory'|'facts'> } })). Add a usageRequestId(request) helper (new in usage.ts): honor an inbound Idempotency-Key/x-request-id header, else crypto.randomUUID().
6. Emit chat from memwalChat (worker.ts:3498). Thread ctx (call site worker.ts:635 → memwalChat(request, env, ctx)). Owner resolution as in step 5. Emit action: 'chat', metadata: { source, recallHits: recallHits.length } (counts only, never message text).
7. Emit ai_query from aiQueryRun (worker.ts:1349). Thread ctx (call site worker.ts:652 → aiQueryRun(request, env, hostedRunRoute.runId, ctx)). It already resolves job.ownerId and the namespace; use requestId: runId (stable). Emit on the success path, metadata: { model?: env.OPENAI_MODEL }.
8. Emit mcp_read in the MCP dispatch block (worker.ts:772-794). After authorizeNamespace succeeds (auth.ok), ctx.waitUntil(recordUsage(env, { ownerId: auth.summary.owner_id ?? demoOwnerId(request), namespace, action: 'mcp_read', requestId: usageRequestId(request), source: 'mcp', metadata: { method: request.method } })). Owner = the namespace owner from auth.summary (the read happens against their namespace). Gateway (namespace-less) MCP calls: emit with namespace: null, owner = demoOwnerId/caller.
9. remember (enum-complete, optional emit): wire it where the extraction job performs the MemWal index write inside processExtractionJob if reachable in this pass; otherwise the enum value is reserved for the on-chain-receipt issue. Not required by the AC emit list — keep it in the CHECK so no migration is needed later.
10. GET /api/usage?namespace=&from=&to= route. Add to routeWorkerRequest near the other GET /api/* routes (e.g. after worker.ts:592 /api/me): require readHostedRunAuth(request) (401 if absent — owner-scoped), then return json(await aggregateUsage(env, auth.ownerId, namespace, from, to)). Validate from/to as ISO dates; clamp range. Returns { namespace, from, to, totals: [{ action, count, units }], total: {count, units} }.
11. CORS: if adopting an inbound Idempotency-Key/x-request-id header, add it to the access-control-allow-headers list (worker.ts:5307).

New / changed files

• apps/api/migrations/0007_usage_events.sql (new)
• packages/core/src/usage.ts (new) — USAGE_ACTIONS, UsageAction, UsageSource
• packages/core/src/index.ts (edit) — add barrel export
• apps/api/src/usage.ts (new) — recordUsage, aggregateUsage, sanitizeUsageMetadata, usageRequestId
• apps/api/src/worker.ts (edit) — import helpers; emit at createNamespaceBuildJob (1901), memwalRecall (3408), memwalChat (3498), aiQueryRun (1349), MCP dispatch (772); thread ctx into recall/chat/ai-query call sites (632/635/652); add GET /api/usage route; CORS header (5307)
• apps/api/src/usage.test.ts (new) and/or additions to apps/api/src/worker.test.ts
• apps/api/src/worker.test.ts (edit) — extend MemoryD1Database (add a usageEvents array + handle the INSERT … contextmem_usage_events and the GROUP BY action SELECT, mirroring the existing SQL-string matching at worker.test.ts:845-935)

Test plan

• Unit (recordUsage): inserts exactly one row; calling twice with the same (request_id, action) yields one row (idempotency, AC Memory knowledge graph: 3D neural constellation + grounded entity panel #2 + docs: add product roadmap (Harbor storage + Talus on-chain direction) #4); a throwing/undefined D1 stub does NOT reject the promise (never-throws, AC Memory knowledge graph: 3D neural constellation + grounded entity panel #2).
• Unit (sanitizeUsageMetadata / leak test, AC Build HarborClient: reserve→sign→finalize bucket, multipart upload, status poll, download #6): pass metadata containing Authorization, x-memwal-authorization, a 64-char hex delegate, Bearer … → assert the persisted metadata_json contains none of them. Also a handler-level test: fire a recall/chat with delegate headers set, then read back the usage row and assert no header/secret substring appears.
• Integration (handlers): extend the existing Worker test harness — run a build via /api/namespaces/build, then assert one extract row with request_id === jobId; POST /api/memwal/recall and /api/memwal/chat → one recall/chat row each; a hosted MCP GET/POST against an authorized namespace → one mcp_read row; aiQueryRun → one ai_query row.
• Integration (/api/usage, AC Spike: run @mysten/seal encrypt+decrypt inside a Cloudflare Worker #5): seed several events across two namespaces/actions; assert GET /api/usage?namespace=…&from=…&to= returns correct per-action count/units, is owner-scoped (a different owner sees nothing / 401 without auth), and respects the from/to window.
• Manual (proves it works in prod): wrangler d1 migrations apply contextmem; deploy the Worker (wrangler deploy — note: CI does not ship the Worker per repo deploy notes); run one real hosted build + one chat against contextmem-backend.petlofi.workers.dev; wrangler d1 execute contextmem --command "SELECT action, COUNT(*) FROM contextmem_usage_events GROUP BY action" shows the rows; GET /api/usage returns matching aggregates.

Risks & gotchas

• Async ordering: ctx.waitUntil runs after the response; in unit tests waitUntil may be undefined — recordUsage must be awaited in the no-ctx test path (the harness passes ctx = {}), or assert after awaiting the returned promise. Mirror the createNamespaceBuildJob pattern (worker.ts:1934-1937).
• D1 write amplification: one INSERT per metered action. Acceptable now; the ON CONFLICT DO NOTHING + indexes keep it cheap. A daily rollup table is explicitly out of scope (issue note).
• /api/usage scoping is only as strong as identity — header-derived owner_id is spoofable until real accounts land (see Prerequisites). Do not advertise it as billing-grade yet.
• Test harness is a hand-rolled SQL-string matcher (MemoryD1Database), not real SQLite — it will silently return null/empty for any query it doesn't pattern-match, so the new INSERT and GROUP BY branches MUST be added or integration assertions will pass vacuously. Consider a focused unit test against a real better-sqlite3/bun:sqlite for the aggregation SQL if higher fidelity is wanted.
• metadata_json is the leak surface — only ever pass computed scalars (counts, durations, model name, status, source); never spread request.headers, the body, or creds. The sanitizer is defense-in-depth, not the primary guard.
• CHECK enum drift: the SQL CHECK list and USAGE_ACTIONS in core must stay in lockstep; a value present in one but not the other fails inserts. Keep both in this PR and reference the core constant in code.
• remember not yet emitted — leaving it in the enum is intentional (no future migration); just don't claim it's wired unless step 9 lands.

Acceptance-criteria mapping

• Migration 0007_usage_events.sql w/ CHECK enum + both indexes → Step 1.
• Single recordUsage(...) helper, never throws, ctx.waitUntil, unit-tested → Steps 3, 4-8 (all emits wrapped in waitUntil), Test plan (unit).
• Events emitted from build/recall/chat/ai-query/MCP reads/demo → Steps 4 (build+demo), 5 (recall), 6 (chat), 7 (ai-query), 8 (mcp_read).
• Stable request_id for idempotency → Step 1 UNIQUE (request_id, action) + Step 3 ON CONFLICT DO NOTHING + Steps 4/7 use natural ids (jobId/runId), Steps 5/6/8 use usageRequestId; Test plan idempotency case.
• GET /api/usage aggregates per action via index, owner-scoped → Steps 3 (aggregateUsage, owner_id-first), 10; Test plan /api/usage case.
• No bodies/secrets in metadata_json; test asserts no leak → Step 3 sanitizeUsageMetadata + Steps 5-8 passing only scalars; Test plan leak test.

Effort: M — ~2 person-days. Migration + two small modules are trivial; the bulk is threading ctx into three existing handlers and extending the bespoke MemoryD1Database test mock to cover the new INSERT/GROUP BY so the integration assertions are real.

---

Implementation plan generated for execution (planning phase). Part of the roadmap (#4).