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

Activity

1. harrymove-ctrl

   added this to the Sui Overflow milestone on Jun 16, 2026
2. harrymove-ctrl

   added

   P0Demo-blocking: required for a working Sui Overflow demoDemo-blocking: required for a working Sui Overflow demo

   featureUser- or agent-facing capabilityUser- or agent-facing capability

   platformBackend platform plumbing: Worker, D1, queues, secrets, meteringBackend platform plumbing: Worker, D1, queues, secrets, metering

   agent-talusTalus/Nexus agent integration and attribution receiptsTalus/Nexus agent integration and attribution receipts

   on Jun 16, 2026
3. harrymove-ctrl commented on Jun 19, 2026

   

   harrymove-ctrl

   on Jun 19, 2026

   OwnerAuthor

   More actions

   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).