diff --git a/docs/contributing/architecture/entitlements.md b/docs/contributing/architecture/entitlements.md index 09cdd9a251..30117e2fc6 100644 --- a/docs/contributing/architecture/entitlements.md +++ b/docs/contributing/architecture/entitlements.md @@ -375,6 +375,40 @@ soak/parity review. waits until reporting-off-D1 work merges and mirror parity is verified in production. +### Admin UserMeter parity gates (`admin_user_meter_parity`) + +Production verification for mirror retirement and authority flips uses the +admin-only read-only capability `admin_user_meter_parity` (input: +`stable_user_id`). It compares production-shaped D1 rows for one account against +direct UserMeter RPCs and never bootstraps or writes parity state. Opening a +cold UserMeter stub may still run Durable Object constructor schema maintenance +and opportunistic stale daily-counter pruning. Cold meter rows surface as +`needsBootstrap` with `meterCount`/`meterBytes` null. + +Interpret the structured report as independent gates: + +| Gate | Pass condition | +| -------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Daily counters (current UTC day) | Each of the four daily resources has `parity: true` (`d1Count === meterCount`); aggregate `daily.mismatchCount === 0`. | +| Storage bytes | `storage.parity` — D1 `users.d1_storage_bytes` equals UserMeter `readStorageBytes` (not `needsBootstrap`). | +| Package services | `packageServices.parity` — inventory mismatch category counts are all zero (`d1Only` / `meterOnly` / `statusMismatch` / `startedAtMismatch` / `sourceUpdatedAtMismatch`), fresh-running counts match under the shared 24h stale window, and the meter page walk is not `truncated`. | +| Deletion tombstone | `deletion.deletingAtParity` — D1 `users.deleting_at` matches the meter tombstone. | +| Temporary D1 lease mirror | `deletion.mirrorLeaseParity` — `doOnly === 0`, `legacyWithoutD1 === 0`, inventory not truncated, and `d1ActiveLeaseCount >= doAuthorityLeaseCount` (same-token mirror coverage). `tokenSetMismatches.d1Only` is reported but does **not** fail this gate. | + +**D1-only leases:** email and other transition paths that omit `env` still take +exact D1 leases, so `d1Only > 0` is expected until that handoff. Mirror-removal +readiness therefore ignores `d1Only`. Operators separately confirm those rows +are known email/transition holders via `admin_account_write_lease_list` and +holder classification before retiring the temporary D1 mirror inventory. + +**Threshold:** treat unexplained mismatches as blocking for the corresponding +cutover (daily mirror retirement, storage authority flip, package-service +authority flip, or temporary D1 lease-mirror removal). Expected cold accounts +may report `needsBootstrap` until live traffic or an intentional bootstrap path +seeds the DO; that is a bootstrap gap, not a silent pass. Truncated inventories +fail closed (`parity` / `mirrorLeaseParity` false) so operators re-run or raise +the bounded page cap rather than approve a partial compare. + Module wiring: `consumeDailyEntitlement`, `refundDailyEntitlement`, and `readDailyEntitlementResourceUsage` require `env.USER_METER` and fail closed when the binding is missing. Storage-byte helpers (`readUserD1StorageBytes`, diff --git a/packages/worker/src/admin/user-meter-parity.node.test.ts b/packages/worker/src/admin/user-meter-parity.node.test.ts new file mode 100644 index 0000000000..49b6b8a0e5 --- /dev/null +++ b/packages/worker/src/admin/user-meter-parity.node.test.ts @@ -0,0 +1,766 @@ +import { DatabaseSync } from 'node:sqlite' +import { expect, test, vi } from 'vitest' +import { utcDayKey } from '@kody-internal/shared/date-keys.ts' +import { createD1FromSqlite } from '#worker/test-support/create-d1-from-sqlite.ts' +import { testStableUserIdFromEmail } from '#worker/test-support/stable-user-id.ts' +import { createInMemoryUserMeterEnv } from '#worker/test-support/user-meter.ts' +import { + userMeterRpc, + type UserMeterEnv, +} from '#worker/entitlements/user-meter-client.ts' +import { + dailyEntitlementResources, + type UserMeterPackageServiceState, + type UserMeterWriteLeaseShadow, +} from '#worker/entitlements/user-meter-do.ts' +import { loadAdminUserMeterParityReport } from './user-meter-parity.ts' + +const stableUserId = testStableUserIdFromEmail('parity@example.com') +const now = new Date('2026-08-01T12:00:00.000Z') +const day = utcDayKey(now) + +function createParityTestDb() { + const sqlite = new DatabaseSync(':memory:') + sqlite.exec(` + CREATE TABLE users ( + id INTEGER PRIMARY KEY, + stable_user_id TEXT UNIQUE NOT NULL, + username TEXT NOT NULL, + email TEXT NOT NULL, + password_hash TEXT, + d1_storage_bytes INTEGER NOT NULL DEFAULT 0, + deleting_at TEXT, + active_write_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000Z', + updated_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000Z' + ); + CREATE TABLE entitlement_daily_counters ( + user_id TEXT NOT NULL, + resource TEXT NOT NULL, + day TEXT NOT NULL, + count INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (user_id, resource, day) + ); + CREATE TABLE package_service_states ( + user_id TEXT NOT NULL, + package_id TEXT NOT NULL, + service_name TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('running', 'idle', 'stopped', 'error')), + started_at TEXT, + updated_at TEXT NOT NULL, + PRIMARY KEY (user_id, package_id, service_name) + ); + CREATE TABLE account_write_leases ( + token TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + holder TEXT NOT NULL, + acquired_at TEXT NOT NULL, + released_at TEXT + ); + `) + return { sqlite, db: createD1FromSqlite(sqlite) } +} + +function insertUser( + sqlite: DatabaseSync, + input: { + stableUserId: string + d1StorageBytes?: number + deletingAt?: string | null + }, +) { + sqlite + .prepare( + `INSERT INTO users ( + stable_user_id, username, email, d1_storage_bytes, deleting_at + ) VALUES (?, ?, ?, ?, ?)`, + ) + .run( + input.stableUserId, + 'parity-user', + 'parity@example.com', + input.d1StorageBytes ?? 0, + input.deletingAt ?? null, + ) +} + +async function seedMeterParityBaseline(input: { + meter: ReturnType + stableUserId: string + dailyCounts: Record<(typeof dailyEntitlementResources)[number], number> + storageBytes: number + packageServices?: Array<{ + packageId: string + serviceName: string + status: 'running' | 'idle' | 'stopped' | 'error' + startedAt?: string | null + sourceUpdatedAt: string + }> + deletingAt?: string | null + leases?: Array<{ + token: string + holder: string + acquiredAt: string + authority: 'do' | 'legacy' + }> +}) { + const meter = userMeterRpc({ + env: input.meter.env, + userId: input.stableUserId, + }) + for (const resource of dailyEntitlementResources) { + await meter.initialize({ + resource, + day, + count: input.dailyCounts[resource], + updatedAt: now.toISOString(), + }) + } + await meter.initializeStorageBytes({ + bytes: input.storageBytes, + updatedAt: now.toISOString(), + }) + for (const service of input.packageServices ?? []) { + await meter.upsertPackageServiceState({ + packageId: service.packageId, + serviceName: service.serviceName, + status: service.status, + startedAt: service.startedAt ?? null, + sourceUpdatedAt: service.sourceUpdatedAt, + updatedAt: now.toISOString(), + }) + } + await meter.bootstrapDeletionState({ + deletingAt: input.deletingAt ?? null, + leases: input.leases, + }) +} + +function seedD1ParityBaseline(input: { + sqlite: DatabaseSync + stableUserId: string + dailyCounts: Record<(typeof dailyEntitlementResources)[number], number> + storageBytes: number + packageServices?: Array<{ + packageId: string + serviceName: string + status: 'running' | 'idle' | 'stopped' | 'error' + startedAt?: string | null + updatedAt: string + }> + deletingAt?: string | null + leases?: Array<{ + token: string + holder: string + acquiredAt: string + }> +}) { + insertUser(input.sqlite, { + stableUserId: input.stableUserId, + d1StorageBytes: input.storageBytes, + deletingAt: input.deletingAt ?? null, + }) + for (const resource of dailyEntitlementResources) { + input.sqlite + .prepare( + `INSERT INTO entitlement_daily_counters ( + user_id, resource, day, count, updated_at + ) VALUES (?, ?, ?, ?, ?)`, + ) + .run( + input.stableUserId, + resource, + day, + input.dailyCounts[resource], + now.toISOString(), + ) + } + for (const service of input.packageServices ?? []) { + input.sqlite + .prepare( + `INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?)`, + ) + .run( + input.stableUserId, + service.packageId, + service.serviceName, + service.status, + service.status === 'running' ? (service.startedAt ?? null) : null, + service.updatedAt, + ) + } + for (const lease of input.leases ?? []) { + input.sqlite + .prepare( + `INSERT INTO account_write_leases ( + token, user_id, holder, acquired_at, released_at + ) VALUES (?, ?, ?, ?, NULL)`, + ) + .run(lease.token, input.stableUserId, lease.holder, lease.acquiredAt) + } +} + +function assertNoLeaseSecrets(value: unknown) { + const serialized = JSON.stringify(value) + expect(serialized).not.toMatch(/lease-token|holder-secret|secret-holder/i) + expect(serialized).not.toContain('"token"') + expect(serialized).not.toContain('"holder"') +} + +type MeterListPageInput = { + pageSize?: number + startAfter?: string | null +} + +function withMeterListOverrides( + meter: ReturnType, + overrides: { + listPackageServiceStates?: (input: MeterListPageInput) => Promise<{ + states: Array + nextStartAfter: string | null + truncated: boolean + }> + listWriteLeases?: (input: MeterListPageInput) => Promise<{ + leases: Array + nextStartAfter: string | null + truncated: boolean + }> + }, +): UserMeterEnv { + const namespace = meter.env.USER_METER + if (!namespace) throw new Error('expected USER_METER binding') + const originalGet = namespace.get.bind(namespace) + return { + USER_METER: { + idFromName: namespace.idFromName.bind(namespace), + get: (id: DurableObjectId) => { + const stub = originalGet(id) as unknown as Record & { + listPackageServiceStates: (input: MeterListPageInput) => Promise<{ + states: Array + nextStartAfter: string | null + truncated: boolean + }> + listWriteLeases: (input: MeterListPageInput) => Promise<{ + leases: Array + nextStartAfter: string | null + truncated: boolean + }> + } + return { + ...stub, + listPackageServiceStates: + overrides.listPackageServiceStates ?? + stub.listPackageServiceStates.bind(stub), + listWriteLeases: + overrides.listWriteLeases ?? stub.listWriteLeases.bind(stub), + } + }, + }, + } as unknown as UserMeterEnv +} + +test('loadAdminUserMeterParityReport reports full parity across daily/storage/services/deletion', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + const dailyCounts = { + email_sends_per_day: 3, + email_receives_per_day: 5, + execute_calls_per_day: 11, + outbound_fetches_per_day: 7, + } + const serviceUpdatedAt = '2026-08-01T11:00:00.000Z' + const leaseToken = 'lease-token-shared' + seedD1ParityBaseline({ + sqlite, + stableUserId, + dailyCounts, + storageBytes: 4096, + packageServices: [ + { + packageId: 'pkg-a', + serviceName: 'web', + status: 'running', + startedAt: '2026-08-01T10:00:00.000Z', + updatedAt: serviceUpdatedAt, + }, + ], + leases: [ + { + token: leaseToken, + holder: 'holder-secret', + acquiredAt: '2026-08-01T09:00:00.000Z', + }, + ], + }) + await seedMeterParityBaseline({ + meter, + stableUserId, + dailyCounts, + storageBytes: 4096, + packageServices: [ + { + packageId: 'pkg-a', + serviceName: 'web', + status: 'running', + startedAt: '2026-08-01T10:00:00.000Z', + sourceUpdatedAt: serviceUpdatedAt, + }, + ], + leases: [ + { + token: leaseToken, + holder: 'holder-secret', + acquiredAt: '2026-08-01T09:00:00.000Z', + authority: 'do', + }, + ], + }) + + const report = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId, + now, + }) + expect(report).toMatchObject({ + generatedAt: now.toISOString(), + stableUserId, + daily: { + day, + mismatchCount: 0, + }, + storage: { + d1Bytes: 4096, + meterBytes: 4096, + needsBootstrap: false, + delta: 0, + parity: true, + }, + packageServices: { + d1Count: 1, + meterCount: 1, + truncated: false, + mismatchCategories: { + d1Only: 0, + meterOnly: 0, + statusMismatch: 0, + startedAtMismatch: 0, + sourceUpdatedAtMismatch: 0, + }, + running: { + d1FreshRunningCount: 1, + meterRunningCount: 1, + parity: true, + }, + parity: true, + }, + deletion: { + d1DeletingAt: null, + meterDeletingAt: null, + deletingAtParity: true, + d1ActiveLeaseCount: 1, + doAuthorityLeaseCount: 1, + doLegacyLeaseCount: 0, + tokenSetMismatches: { + d1Only: 0, + doOnly: 0, + legacyWithoutD1: 0, + }, + truncated: false, + mirrorLeaseParity: true, + }, + }) + expect(report?.daily.resources).toHaveLength(4) + expect(report?.daily.resources.every((row) => row.parity)).toBe(true) + assertNoLeaseSecrets(report) +}) + +test('loadAdminUserMeterParityReport treats D1-only leases as mirrorLeaseParity pass', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + const dailyCounts = { + email_sends_per_day: 0, + email_receives_per_day: 0, + execute_calls_per_day: 0, + outbound_fetches_per_day: 0, + } + seedD1ParityBaseline({ + sqlite, + stableUserId, + dailyCounts, + storageBytes: 0, + leases: [ + { + token: 'lease-token-email-only', + holder: 'holder-secret', + acquiredAt: '2026-08-01T09:00:00.000Z', + }, + ], + }) + await seedMeterParityBaseline({ + meter, + stableUserId, + dailyCounts, + storageBytes: 0, + }) + + const report = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId, + now, + }) + expect(report?.deletion).toMatchObject({ + d1ActiveLeaseCount: 1, + doAuthorityLeaseCount: 0, + doLegacyLeaseCount: 0, + tokenSetMismatches: { + d1Only: 1, + doOnly: 0, + legacyWithoutD1: 0, + }, + truncated: false, + mirrorLeaseParity: true, + }) + assertNoLeaseSecrets(report) +}) + +test('loadAdminUserMeterParityReport classifies mismatches and needsBootstrap without writing parity state', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + insertUser(sqlite, { stableUserId, d1StorageBytes: 100 }) + sqlite + .prepare( + `INSERT INTO entitlement_daily_counters ( + user_id, resource, day, count, updated_at + ) VALUES (?, 'email_sends_per_day', ?, 9, ?)`, + ) + .run(stableUserId, day, now.toISOString()) + sqlite + .prepare( + `INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES (?, 'pkg-a', 'web', 'running', ?, ?)`, + ) + .run(stableUserId, '2026-08-01T10:00:00.000Z', '2026-08-01T11:00:00.000Z') + sqlite + .prepare( + `INSERT INTO account_write_leases ( + token, user_id, holder, acquired_at, released_at + ) VALUES ('d1-only-token', ?, 'holder-secret', ?, NULL)`, + ) + .run(stableUserId, '2026-08-01T09:00:00.000Z') + + const meterStub = userMeterRpc({ env: meter.env, userId: stableUserId }) + await meterStub.initialize({ + resource: 'email_sends_per_day', + day, + count: 2, + updatedAt: now.toISOString(), + }) + await meterStub.upsertPackageServiceState({ + packageId: 'pkg-b', + serviceName: 'worker', + status: 'idle', + sourceUpdatedAt: '2026-08-01T11:30:00.000Z', + }) + await meterStub.bootstrapDeletionState({ + leases: [ + { + token: 'do-only-token', + holder: 'secret-holder', + acquiredAt: '2026-08-01T08:00:00.000Z', + authority: 'do', + }, + { + token: 'legacy-only-token', + holder: 'secret-holder', + acquiredAt: '2026-08-01T07:00:00.000Z', + authority: 'legacy', + }, + ], + }) + + const beforeRead = await meterStub.read({ + resource: 'email_receives_per_day', + day, + now: now.toISOString(), + }) + expect(beforeRead).toEqual({ outcome: 'needs_bootstrap' }) + + const report = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId, + now, + }) + expect(report?.daily.mismatchCount).toBe(4) + expect( + report?.daily.resources.find( + (row) => row.resource === 'email_sends_per_day', + ), + ).toMatchObject({ + d1Count: 9, + meterCount: 2, + needsBootstrap: false, + delta: 7, + parity: false, + }) + expect( + report?.daily.resources.find( + (row) => row.resource === 'email_receives_per_day', + ), + ).toMatchObject({ + d1Count: 0, + meterCount: null, + needsBootstrap: true, + delta: null, + parity: false, + }) + expect(report?.storage).toMatchObject({ + d1Bytes: 100, + meterBytes: null, + needsBootstrap: true, + delta: null, + parity: false, + }) + expect(report?.packageServices).toMatchObject({ + d1Count: 1, + meterCount: 1, + mismatchCategories: { + d1Only: 1, + meterOnly: 1, + statusMismatch: 0, + startedAtMismatch: 0, + sourceUpdatedAtMismatch: 0, + }, + running: { + d1FreshRunningCount: 1, + meterRunningCount: 0, + parity: false, + }, + parity: false, + }) + expect(report?.deletion).toMatchObject({ + d1ActiveLeaseCount: 1, + doAuthorityLeaseCount: 1, + doLegacyLeaseCount: 1, + tokenSetMismatches: { + d1Only: 1, + doOnly: 1, + legacyWithoutD1: 1, + }, + mirrorLeaseParity: false, + }) + + const afterRead = await meterStub.read({ + resource: 'email_receives_per_day', + day, + now: now.toISOString(), + }) + expect(afterRead).toEqual({ outcome: 'needs_bootstrap' }) + expect(await meterStub.readStorageBytes()).toEqual({ + outcome: 'needs_bootstrap', + }) + assertNoLeaseSecrets(report) +}) + +test('loadAdminUserMeterParityReport fails mirrorLeaseParity for doOnly or legacyWithoutD1', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + insertUser(sqlite, { stableUserId, d1StorageBytes: 0 }) + const meterStub = userMeterRpc({ env: meter.env, userId: stableUserId }) + await meterStub.initializeStorageBytes({ + bytes: 0, + updatedAt: now.toISOString(), + }) + for (const resource of dailyEntitlementResources) { + await meterStub.initialize({ + resource, + day, + count: 0, + updatedAt: now.toISOString(), + }) + } + await meterStub.bootstrapDeletionState({ + leases: [ + { + token: 'do-only-token', + holder: 'secret-holder', + acquiredAt: '2026-08-01T08:00:00.000Z', + authority: 'do', + }, + ], + }) + + const doOnlyReport = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId, + now, + }) + expect(doOnlyReport?.deletion).toMatchObject({ + d1ActiveLeaseCount: 0, + doAuthorityLeaseCount: 1, + tokenSetMismatches: { + d1Only: 0, + doOnly: 1, + legacyWithoutD1: 0, + }, + mirrorLeaseParity: false, + }) + + await meterStub.releaseWriteLease({ token: 'do-only-token' }) + await meterStub.bootstrapDeletionState({ + leases: [ + { + token: 'legacy-only-token', + holder: 'secret-holder', + acquiredAt: '2026-08-01T07:00:00.000Z', + authority: 'legacy', + }, + ], + }) + const legacyReport = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId, + now, + }) + expect(legacyReport?.deletion).toMatchObject({ + d1ActiveLeaseCount: 0, + doAuthorityLeaseCount: 0, + doLegacyLeaseCount: 1, + tokenSetMismatches: { + d1Only: 0, + doOnly: 0, + legacyWithoutD1: 1, + }, + mirrorLeaseParity: false, + }) + assertNoLeaseSecrets(doOnlyReport) + assertNoLeaseSecrets(legacyReport) +}) + +test('loadAdminUserMeterParityReport returns null for missing users', async () => { + const { db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + const report = await loadAdminUserMeterParityReport({ + db, + env: meter.env, + stableUserId: testStableUserIdFromEmail('missing@example.com'), + now, + }) + expect(report).toBeNull() +}) + +test('loadAdminUserMeterParityReport fails closed on repeated package-service cursor', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + const dailyCounts = { + email_sends_per_day: 0, + email_receives_per_day: 0, + execute_calls_per_day: 0, + outbound_fetches_per_day: 0, + } + seedD1ParityBaseline({ + sqlite, + stableUserId, + dailyCounts, + storageBytes: 0, + }) + await seedMeterParityBaseline({ + meter, + stableUserId, + dailyCounts, + storageBytes: 0, + }) + + const stuckCursor = '["pkg-a","web"]' + const listPackageServiceStates = vi.fn( + async (_input: MeterListPageInput) => ({ + states: [ + { + packageId: 'pkg-a', + serviceName: 'web', + status: 'idle' as const, + startedAt: null, + sourceUpdatedAt: '2026-08-01T11:00:00.000Z', + revision: 1, + updatedAt: now.toISOString(), + mirrorUpdatedAt: '1', + }, + ], + nextStartAfter: stuckCursor, + truncated: true, + }), + ) + const env = withMeterListOverrides(meter, { + listPackageServiceStates, + }) + + const report = await loadAdminUserMeterParityReport({ + db, + env, + stableUserId, + now, + }) + expect(listPackageServiceStates).toHaveBeenCalledTimes(2) + expect(listPackageServiceStates.mock.calls[0]?.[0]?.startAfter ?? null).toBe( + null, + ) + expect(listPackageServiceStates.mock.calls[1]?.[0]?.startAfter).toBe( + stuckCursor, + ) + expect(report?.packageServices).toMatchObject({ + truncated: true, + parity: false, + }) +}) + +test('loadAdminUserMeterParityReport fails closed on zero-item write-lease cursor', async () => { + const { sqlite, db } = createParityTestDb() + const meter = createInMemoryUserMeterEnv() + const dailyCounts = { + email_sends_per_day: 0, + email_receives_per_day: 0, + execute_calls_per_day: 0, + outbound_fetches_per_day: 0, + } + seedD1ParityBaseline({ + sqlite, + stableUserId, + dailyCounts, + storageBytes: 0, + }) + await seedMeterParityBaseline({ + meter, + stableUserId, + dailyCounts, + storageBytes: 0, + }) + + const listWriteLeases = vi.fn(async (_input: MeterListPageInput) => ({ + leases: [] as Array, + nextStartAfter: 'stuck-lease-cursor', + truncated: true, + })) + const env = withMeterListOverrides(meter, { + listWriteLeases, + }) + + const report = await loadAdminUserMeterParityReport({ + db, + env, + stableUserId, + now, + }) + expect(listWriteLeases).toHaveBeenCalledTimes(1) + expect(report?.deletion).toMatchObject({ + truncated: true, + mirrorLeaseParity: false, + }) +}) diff --git a/packages/worker/src/admin/user-meter-parity.ts b/packages/worker/src/admin/user-meter-parity.ts new file mode 100644 index 0000000000..f5dd00b59a --- /dev/null +++ b/packages/worker/src/admin/user-meter-parity.ts @@ -0,0 +1,584 @@ +import { utcDayKey } from '@kody-internal/shared/date-keys.ts' +import { + userMeterRpc, + type UserMeterEnv, +} from '#worker/entitlements/user-meter-client.ts' +import { + dailyEntitlementResources, + type DailyEntitlementResource, + type UserMeterPackageServiceState, + type UserMeterWriteLeaseShadow, +} from '#worker/entitlements/user-meter-do.ts' +import { + countRunningPackageServices, + packageServiceStateStaleMs, +} from '#worker/entitlements/service.ts' +import { planLimits } from '#worker/entitlements/plans.ts' +import { isStableUserId, normalizeStableUserId } from '#worker/user-id.ts' + +/** Cap paged UserMeter inventory pulls (matches max plan package-service ceiling). */ +export const userMeterParityInventoryMaxRows = planLimits.max.maxPackageServices + +/** Page size for UserMeter list RPCs (matches DO max export/list page). */ +export const userMeterParityPageSize = 500 + +type DailyResourceParity = { + resource: DailyEntitlementResource + d1Count: number + meterCount: number | null + needsBootstrap: boolean + delta: number | null + parity: boolean +} + +type StorageParity = { + d1Bytes: number + meterBytes: number | null + needsBootstrap: boolean + delta: number | null + parity: boolean +} + +type PackageServiceMismatchCategories = { + d1Only: number + meterOnly: number + statusMismatch: number + startedAtMismatch: number + sourceUpdatedAtMismatch: number +} + +type PackageServicesParity = { + d1Count: number + meterCount: number + truncated: boolean + mismatchCategories: PackageServiceMismatchCategories + running: { + d1FreshRunningCount: number + meterRunningCount: number + parity: boolean + } + parity: boolean +} + +type DeletionLeaseTokenMismatches = { + d1Only: number + doOnly: number + legacyWithoutD1: number +} + +type DeletionParity = { + d1DeletingAt: string | null + meterDeletingAt: string | null + deletingAtParity: boolean + d1ActiveLeaseCount: number + doAuthorityLeaseCount: number + doLegacyLeaseCount: number + tokenSetMismatches: DeletionLeaseTokenMismatches + truncated: boolean + /** + * Temporary same-token D1 mirror readiness for DO-authority leases. + * Passes when every DO-authority and legacy meter lease is mirrored in D1 + * (`doOnly` / `legacyWithoutD1` are zero), the meter page walk is complete, + * and D1 active lease count covers DO-authority count. `d1Only` is reported + * but does not fail this gate — expected for email/transition D1 leases + * until that path hands off. + */ + mirrorLeaseParity: boolean +} + +export type AdminUserMeterParityReport = { + generatedAt: string + stableUserId: string + daily: { + day: string + resources: Array + mismatchCount: number + } + storage: StorageParity + packageServices: PackageServicesParity + deletion: DeletionParity +} + +type D1PackageServiceStateRow = { + package_id: string + service_name: string + status: string + started_at: string | null + updated_at: string +} + +type D1WriteLeaseTokenRow = { + token: string +} + +function countDeltaParity(input: { + d1Value: number + meterValue: number | null + needsBootstrap: boolean +}): { delta: number | null; parity: boolean } { + if (input.needsBootstrap || input.meterValue == null) { + return { delta: null, parity: false } + } + const delta = input.d1Value - input.meterValue + return { delta, parity: delta === 0 } +} + +async function userExists(db: D1Database, stableUserId: string) { + const row = await db + .prepare(`SELECT 1 AS present FROM users WHERE stable_user_id = ?`) + .bind(stableUserId) + .first<{ present: number }>() + return row != null +} + +async function readD1DailyCounts(input: { + db: D1Database + stableUserId: string + day: string +}): Promise> { + const rows = await input.db + .prepare( + `SELECT resource, count + FROM entitlement_daily_counters + WHERE user_id = ? AND day = ?`, + ) + .bind(input.stableUserId, input.day) + .all<{ resource: string; count: number }>() + const counts = new Map() + for (const resource of dailyEntitlementResources) { + counts.set(resource, 0) + } + for (const row of rows.results ?? []) { + if ( + (dailyEntitlementResources as ReadonlyArray).includes( + row.resource, + ) + ) { + counts.set( + row.resource as DailyEntitlementResource, + Math.max(0, Number(row.count ?? 0)), + ) + } + } + return counts +} + +async function readDailyParity(input: { + db: D1Database + env: UserMeterEnv + stableUserId: string + day: string + generatedAt: string +}): Promise { + const d1Counts = await readD1DailyCounts(input) + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const resources: Array = [] + for (const resource of dailyEntitlementResources) { + const d1Count = d1Counts.get(resource) ?? 0 + const meterRead = await meter.read({ + resource, + day: input.day, + now: input.generatedAt, + }) + const needsBootstrap = meterRead.outcome === 'needs_bootstrap' + const meterCount = needsBootstrap ? null : meterRead.count + const { delta, parity } = countDeltaParity({ + d1Value: d1Count, + meterValue: meterCount, + needsBootstrap, + }) + resources.push({ + resource, + d1Count, + meterCount, + needsBootstrap, + delta, + parity, + }) + } + return { + day: input.day, + resources, + mismatchCount: resources.filter((row) => !row.parity).length, + } +} + +async function readStorageParity(input: { + db: D1Database + env: UserMeterEnv + stableUserId: string +}): Promise { + const d1Row = await input.db + .prepare( + `SELECT d1_storage_bytes AS bytes + FROM users + WHERE stable_user_id = ?`, + ) + .bind(input.stableUserId) + .first<{ bytes: number }>() + const d1Bytes = Math.max(0, Number(d1Row?.bytes ?? 0)) + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const meterRead = await meter.readStorageBytes() + const needsBootstrap = meterRead.outcome === 'needs_bootstrap' + const meterBytes = needsBootstrap ? null : meterRead.bytes + const { delta, parity } = countDeltaParity({ + d1Value: d1Bytes, + meterValue: meterBytes, + needsBootstrap, + }) + return { + d1Bytes, + meterBytes, + needsBootstrap, + delta, + parity, + } +} + +/** + * Bounded UserMeter list walk. Stops at maxRows, a missing cursor, a + * repeated/non-advancing cursor, or a zero-item page that still returns a + * cursor. Stalled/invalid progress marks `truncated` so callers fail closed. + */ +async function collectBoundedUserMeterPages(input: { + maxRows: number + fetchPage: (args: { + pageSize: number + startAfter: string | null + }) => Promise<{ + items: ReadonlyArray + nextStartAfter: string | null + truncated: boolean + }> +}): Promise<{ items: Array; truncated: boolean }> { + const items: Array = [] + let startAfter: string | null = null + let truncated = false + for (;;) { + const remaining = input.maxRows - items.length + if (remaining <= 0) { + truncated = true + break + } + const page = await input.fetchPage({ + pageSize: Math.min(userMeterParityPageSize, remaining), + startAfter, + }) + items.push(...page.items) + if (!page.nextStartAfter) { + truncated = truncated || page.truncated + break + } + if (page.items.length === 0 || page.nextStartAfter === startAfter) { + truncated = true + break + } + if (items.length >= input.maxRows) { + truncated = true + break + } + startAfter = page.nextStartAfter + } + return { items, truncated } +} + +async function listAllMeterPackageServiceStates(input: { + env: UserMeterEnv + stableUserId: string + maxRows: number +}): Promise<{ + states: Array + truncated: boolean +}> { + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const { items: states, truncated } = await collectBoundedUserMeterPages({ + maxRows: input.maxRows, + fetchPage: async ({ pageSize, startAfter }) => { + const page = await meter.listPackageServiceStates({ + pageSize, + startAfter, + }) + return { + items: page.states, + nextStartAfter: page.nextStartAfter, + truncated: page.truncated, + } + }, + }) + return { states, truncated } +} + +async function listAllMeterWriteLeases(input: { + env: UserMeterEnv + stableUserId: string + maxRows: number +}): Promise<{ + leases: Array + truncated: boolean +}> { + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const { items: leases, truncated } = await collectBoundedUserMeterPages({ + maxRows: input.maxRows, + fetchPage: async ({ pageSize, startAfter }) => { + const page = await meter.listWriteLeases({ + pageSize, + startAfter, + }) + return { + items: page.leases, + nextStartAfter: page.nextStartAfter, + truncated: page.truncated, + } + }, + }) + return { leases, truncated } +} + +function packageServiceKey(packageId: string, serviceName: string) { + return `${packageId}\0${serviceName}` +} + +function comparePackageServiceStates(input: { + d1Rows: ReadonlyArray + meterStates: ReadonlyArray +}): PackageServiceMismatchCategories { + const d1ByKey = new Map( + input.d1Rows.map((row) => [ + packageServiceKey(row.package_id, row.service_name), + row, + ]), + ) + const meterByKey = new Map( + input.meterStates.map((row) => [ + packageServiceKey(row.packageId, row.serviceName), + row, + ]), + ) + let d1Only = 0 + let meterOnly = 0 + let statusMismatch = 0 + let startedAtMismatch = 0 + let sourceUpdatedAtMismatch = 0 + for (const [key, d1Row] of d1ByKey) { + const meterRow = meterByKey.get(key) + if (!meterRow) { + d1Only += 1 + continue + } + if (d1Row.status !== meterRow.status) statusMismatch += 1 + const d1StartedAt = d1Row.started_at ?? null + const meterStartedAt = meterRow.startedAt ?? null + if (d1StartedAt !== meterStartedAt) startedAtMismatch += 1 + if (d1Row.updated_at !== meterRow.sourceUpdatedAt) { + sourceUpdatedAtMismatch += 1 + } + } + for (const key of meterByKey.keys()) { + if (!d1ByKey.has(key)) meterOnly += 1 + } + return { + d1Only, + meterOnly, + statusMismatch, + startedAtMismatch, + sourceUpdatedAtMismatch, + } +} + +async function readPackageServicesParity(input: { + db: D1Database + env: UserMeterEnv + stableUserId: string + now: Date +}): Promise { + const d1RowsResult = await input.db + .prepare( + `SELECT package_id, service_name, status, started_at, updated_at + FROM package_service_states + WHERE user_id = ? + ORDER BY package_id ASC, service_name ASC`, + ) + .bind(input.stableUserId) + .all() + const d1Rows = d1RowsResult.results ?? [] + const { states: meterStates, truncated } = + await listAllMeterPackageServiceStates({ + env: input.env, + stableUserId: input.stableUserId, + maxRows: userMeterParityInventoryMaxRows, + }) + const mismatchCategories = comparePackageServiceStates({ + d1Rows, + meterStates, + }) + const d1FreshRunningCount = await countRunningPackageServices({ + db: input.db, + userId: input.stableUserId, + now: input.now, + }) + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const meterRunning = await meter.countRunningPackageServices({ + staleAfterMs: packageServiceStateStaleMs, + now: input.now.toISOString(), + }) + const runningParity = d1FreshRunningCount === meterRunning.count + const inventoryMismatchCount = + mismatchCategories.d1Only + + mismatchCategories.meterOnly + + mismatchCategories.statusMismatch + + mismatchCategories.startedAtMismatch + + mismatchCategories.sourceUpdatedAtMismatch + return { + d1Count: d1Rows.length, + meterCount: meterStates.length, + truncated, + mismatchCategories, + running: { + d1FreshRunningCount, + meterRunningCount: meterRunning.count, + parity: runningParity, + }, + parity: inventoryMismatchCount === 0 && runningParity && !truncated, + } +} + +function compareDeletionLeaseTokens(input: { + d1Tokens: ReadonlyArray + meterLeases: ReadonlyArray +}): DeletionLeaseTokenMismatches { + const d1TokenSet = new Set(input.d1Tokens) + const doAuthorityTokens = new Set() + const legacyTokens = new Set() + for (const lease of input.meterLeases) { + if (lease.authority === 'do') doAuthorityTokens.add(lease.token) + else legacyTokens.add(lease.token) + } + let d1Only = 0 + for (const token of d1TokenSet) { + if (!doAuthorityTokens.has(token) && !legacyTokens.has(token)) { + d1Only += 1 + } + } + let doOnly = 0 + for (const token of doAuthorityTokens) { + if (!d1TokenSet.has(token)) doOnly += 1 + } + let legacyWithoutD1 = 0 + for (const token of legacyTokens) { + if (!d1TokenSet.has(token)) legacyWithoutD1 += 1 + } + return { d1Only, doOnly, legacyWithoutD1 } +} + +async function readDeletionParity(input: { + db: D1Database + env: UserMeterEnv + stableUserId: string +}): Promise { + const d1DeletingRow = await input.db + .prepare(`SELECT deleting_at FROM users WHERE stable_user_id = ?`) + .bind(input.stableUserId) + .first<{ deleting_at: string | null }>() + const d1DeletingAt = d1DeletingRow?.deleting_at ?? null + const d1LeaseRows = await input.db + .prepare( + `SELECT token + FROM account_write_leases + WHERE user_id = ? AND released_at IS NULL + ORDER BY acquired_at, token`, + ) + .bind(input.stableUserId) + .all() + const d1Tokens = (d1LeaseRows.results ?? []).map((row) => row.token) + const meter = userMeterRpc({ env: input.env, userId: input.stableUserId }) + const meterDeletion = await meter.readDeletionState() + const { leases: meterLeases, truncated } = await listAllMeterWriteLeases({ + env: input.env, + stableUserId: input.stableUserId, + maxRows: userMeterParityInventoryMaxRows, + }) + let doAuthorityLeaseCount = 0 + let doLegacyLeaseCount = 0 + for (const lease of meterLeases) { + if (lease.authority === 'do') doAuthorityLeaseCount += 1 + else doLegacyLeaseCount += 1 + } + const tokenSetMismatches = compareDeletionLeaseTokens({ + d1Tokens, + meterLeases, + }) + const deletingAtParity = d1DeletingAt === meterDeletion.deletingAt + const d1ActiveLeaseCount = d1Tokens.length + const mirrorLeaseParity = + tokenSetMismatches.doOnly === 0 && + tokenSetMismatches.legacyWithoutD1 === 0 && + !truncated && + d1ActiveLeaseCount >= doAuthorityLeaseCount + return { + d1DeletingAt, + meterDeletingAt: meterDeletion.deletingAt, + deletingAtParity, + d1ActiveLeaseCount, + doAuthorityLeaseCount, + doLegacyLeaseCount, + tokenSetMismatches, + truncated, + mirrorLeaseParity, + } +} + +/** + * Read-only D1 ↔ UserMeter parity report for one account. Never bootstraps or + * writes parity state. Opening a cold UserMeter stub may still run Durable + * Object constructor schema maintenance and opportunistic stale daily-counter + * pruning; cold meter rows surface as `needsBootstrap`. + */ +export async function loadAdminUserMeterParityReport(input: { + db: D1Database + env: UserMeterEnv + stableUserId: string + now?: Date +}): Promise { + const stableUserId = normalizeStableUserId(input.stableUserId) + if (!isStableUserId(stableUserId)) return null + if (!(await userExists(input.db, stableUserId))) return null + + const now = input.now ?? new Date() + const generatedAt = now.toISOString() + const day = utcDayKey(now) + + const [daily, storage, packageServices, deletion] = await Promise.all([ + readDailyParity({ + db: input.db, + env: input.env, + stableUserId, + day, + generatedAt, + }), + readStorageParity({ + db: input.db, + env: input.env, + stableUserId, + }), + readPackageServicesParity({ + db: input.db, + env: input.env, + stableUserId, + now, + }), + readDeletionParity({ + db: input.db, + env: input.env, + stableUserId, + }), + ]) + + return { + generatedAt, + stableUserId, + daily, + storage, + packageServices, + deletion, + } +} diff --git a/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.node.test.ts b/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.node.test.ts new file mode 100644 index 0000000000..ceccd65157 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.node.test.ts @@ -0,0 +1,154 @@ +import { DatabaseSync } from 'node:sqlite' +import { expect, test, vi } from 'vitest' +import type * as AuditLog from '#worker/audit-log.ts' +import { createD1FromSqlite } from '#worker/test-support/create-d1-from-sqlite.ts' +import { createMcpCallerContext } from '#mcp/context.ts' +import { testStableUserIdFromEmail } from '#worker/test-support/stable-user-id.ts' +import { createInMemoryUserMeterEnv } from '#worker/test-support/user-meter.ts' +import { userMeterRpc } from '#worker/entitlements/user-meter-client.ts' + +const mockModule = vi.hoisted(() => ({ + logAuditEvent: vi.fn(async () => undefined), +})) + +vi.mock('#worker/audit-log.ts', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + logAuditEvent: (...args: Array) => + mockModule.logAuditEvent(...args), + } +}) + +const { adminDomain } = await import('./domain.ts') +const { adminUserMeterParityCapability } = + await import('./admin-user-meter-parity.ts') + +const stableUserId = testStableUserIdFromEmail('parity-cap@example.com') + +function createCapabilityTestDb() { + const sqlite = new DatabaseSync(':memory:') + sqlite.exec(` + CREATE TABLE users ( + id INTEGER PRIMARY KEY, + stable_user_id TEXT UNIQUE NOT NULL, + username TEXT NOT NULL, + email TEXT NOT NULL, + d1_storage_bytes INTEGER NOT NULL DEFAULT 0, + deleting_at TEXT, + created_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000Z', + updated_at TEXT NOT NULL DEFAULT '2026-01-01T00:00:00.000Z' + ); + CREATE TABLE entitlement_daily_counters ( + user_id TEXT NOT NULL, + resource TEXT NOT NULL, + day TEXT NOT NULL, + count INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (user_id, resource, day) + ); + CREATE TABLE package_service_states ( + user_id TEXT NOT NULL, + package_id TEXT NOT NULL, + service_name TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT, + updated_at TEXT NOT NULL, + PRIMARY KEY (user_id, package_id, service_name) + ); + CREATE TABLE account_write_leases ( + token TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + holder TEXT NOT NULL, + acquired_at TEXT NOT NULL, + released_at TEXT + ); + INSERT INTO users (stable_user_id, username, email, d1_storage_bytes) + VALUES ('${stableUserId}', 'parity-cap', 'parity-cap@example.com', 1); + INSERT INTO account_write_leases ( + token, user_id, holder, acquired_at, released_at + ) VALUES ( + 'lease-token-privacy', '${stableUserId}', 'holder-secret', + '2026-08-01T09:00:00.000Z', NULL + ); + `) + return createD1FromSqlite(sqlite) +} + +test('admin_user_meter_parity is registered admin-only read-only with parity keywords', () => { + const capability = adminDomain.capabilities.find( + (entry) => entry.name === 'admin_user_meter_parity', + ) + expect(capability).toMatchObject({ + requiredRole: 'admin', + readOnly: true, + idempotent: true, + destructive: false, + }) + expect(adminDomain.keywords).toEqual( + expect.arrayContaining(['user meter', 'parity', 'cutover']), + ) + expect(adminUserMeterParityCapability.name).toBe('admin_user_meter_parity') +}) + +test('admin_user_meter_parity returns null for missing users and omits lease secrets', async () => { + const db = createCapabilityTestDb() + const meter = createInMemoryUserMeterEnv() + const meterStub = userMeterRpc({ env: meter.env, userId: stableUserId }) + await meterStub.bootstrapDeletionState({ + leases: [ + { + token: 'lease-token-privacy', + holder: 'holder-secret', + acquiredAt: '2026-08-01T09:00:00.000Z', + authority: 'do', + }, + ], + }) + const ctx = { + env: { APP_DB: db, ...meter.env } as Env, + callerContext: createMcpCallerContext({ + baseUrl: 'https://heykody.dev', + user: { + userId: testStableUserIdFromEmail('admin@example.com'), + email: 'admin@example.com', + displayName: 'Admin', + roles: ['admin'], + }, + }), + } + + const missing = await adminUserMeterParityCapability.handler( + { + stable_user_id: testStableUserIdFromEmail('missing@example.com'), + }, + ctx, + ) + expect(missing).toEqual({ report: null }) + expect(mockModule.logAuditEvent).toHaveBeenCalledWith( + expect.objectContaining({ + action: 'admin_user_meter_parity', + result: 'success', + reason: `target_stable_user_id=${testStableUserIdFromEmail('missing@example.com')}`, + }), + ) + + const present = await adminUserMeterParityCapability.handler( + { stable_user_id: stableUserId }, + ctx, + ) + expect(present.report?.stableUserId).toBe(stableUserId) + expect(present.report?.deletion.mirrorLeaseParity).toBe(true) + expect(mockModule.logAuditEvent).toHaveBeenCalledWith( + expect.objectContaining({ + action: 'admin_user_meter_parity', + result: 'success', + reason: `target_stable_user_id=${stableUserId}`, + }), + ) + const serialized = JSON.stringify(present) + expect(serialized).not.toContain('lease-token-privacy') + expect(serialized).not.toContain('holder-secret') + expect(serialized).not.toContain('"token"') + expect(serialized).not.toContain('"holder"') +}) diff --git a/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.ts b/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.ts new file mode 100644 index 0000000000..2d40113791 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/admin/admin-user-meter-parity.ts @@ -0,0 +1,134 @@ +import { z } from 'zod' +import { loadAdminUserMeterParityReport } from '#worker/admin/user-meter-parity.ts' +import { dailyEntitlementResources } from '#worker/entitlements/user-meter-do.ts' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { + adminCapabilityAccess, + auditAdminCapabilityInvocation, + stableUserIdSchema, +} from './admin-shared.ts' + +const dailyResourceSchema = z.enum(dailyEntitlementResources) + +const dailyResourceParitySchema = z.object({ + resource: dailyResourceSchema, + d1Count: z.number().int().nonnegative(), + meterCount: z.number().int().nonnegative().nullable(), + needsBootstrap: z.boolean(), + delta: z.number().int().nullable(), + parity: z.boolean(), +}) + +const storageParitySchema = z.object({ + d1Bytes: z.number().int().nonnegative(), + meterBytes: z.number().int().nonnegative().nullable(), + needsBootstrap: z.boolean(), + delta: z.number().int().nullable(), + parity: z.boolean(), +}) + +const packageServiceMismatchCategoriesSchema = z.object({ + d1Only: z.number().int().nonnegative(), + meterOnly: z.number().int().nonnegative(), + statusMismatch: z.number().int().nonnegative(), + startedAtMismatch: z.number().int().nonnegative(), + sourceUpdatedAtMismatch: z.number().int().nonnegative(), +}) + +const packageServicesParitySchema = z.object({ + d1Count: z.number().int().nonnegative(), + meterCount: z.number().int().nonnegative(), + truncated: z.boolean(), + mismatchCategories: packageServiceMismatchCategoriesSchema, + running: z.object({ + d1FreshRunningCount: z.number().int().nonnegative(), + meterRunningCount: z.number().int().nonnegative(), + parity: z.boolean(), + }), + parity: z.boolean(), +}) + +const deletionLeaseTokenMismatchesSchema = z.object({ + d1Only: z.number().int().nonnegative(), + doOnly: z.number().int().nonnegative(), + legacyWithoutD1: z.number().int().nonnegative(), +}) + +const deletionParitySchema = z.object({ + d1DeletingAt: z.string().nullable(), + meterDeletingAt: z.string().nullable(), + deletingAtParity: z.boolean(), + d1ActiveLeaseCount: z.number().int().nonnegative(), + doAuthorityLeaseCount: z.number().int().nonnegative(), + doLegacyLeaseCount: z.number().int().nonnegative(), + tokenSetMismatches: deletionLeaseTokenMismatchesSchema, + truncated: z.boolean(), + mirrorLeaseParity: z + .boolean() + .describe( + 'Temporary same-token D1 mirror readiness: doOnly and legacyWithoutD1 are zero, inventory is not truncated, and d1ActiveLeaseCount covers doAuthorityLeaseCount. d1Only is reported separately and does not fail this gate.', + ), +}) + +const reportSchema = z.object({ + generatedAt: z.string(), + stableUserId: stableUserIdSchema, + daily: z.object({ + day: z.string(), + resources: z.array(dailyResourceParitySchema), + mismatchCount: z.number().int().nonnegative(), + }), + storage: storageParitySchema, + packageServices: packageServicesParitySchema, + deletion: deletionParitySchema, +}) + +const inputSchema = z.object({ + stable_user_id: stableUserIdSchema, +}) + +const outputSchema = z.object({ + report: reportSchema.nullable(), +}) + +export const adminUserMeterParityCapability = defineDomainCapability( + capabilityDomainNames.admin, + { + ...adminCapabilityAccess, + name: 'admin_user_meter_parity', + description: + 'Read-only production verification report comparing one user D1 entitlement mirrors against UserMeter for daily counters, storage bytes, package-service liveness, and deletion write leases. Never bootstraps or writes parity state (DO constructor schema maintenance and stale daily pruning may still run); returns counts and parity only (no lease tokens/holders or email content). Admin-only.', + keywords: [ + 'admin', + 'user meter', + 'parity', + 'cutover', + 'daily counters', + 'storage bytes', + 'package services', + 'write leases', + 'deletion', + 'mirror', + 'bootstrap', + ], + inputSchema, + outputSchema, + async handler(args, ctx) { + return auditAdminCapabilityInvocation( + ctx, + 'admin_user_meter_parity', + async () => ({ + report: await loadAdminUserMeterParityReport({ + db: ctx.env.APP_DB, + env: ctx.env, + stableUserId: args.stable_user_id, + }), + }), + { + successReason: () => `target_stable_user_id=${args.stable_user_id}`, + }, + ) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/admin/domain.ts b/packages/worker/src/mcp/capabilities/admin/domain.ts index 1cc69be789..f562e7d0a5 100644 --- a/packages/worker/src/mcp/capabilities/admin/domain.ts +++ b/packages/worker/src/mcp/capabilities/admin/domain.ts @@ -28,11 +28,12 @@ import { adminUserListCapability } from './admin-user-list.ts' import { adminUserUpdateCapability } from './admin-user-update.ts' import { adminAccountWriteLeaseListCapability } from './admin-account-write-lease-list.ts' import { adminAccountWriteLeaseRepairCapability } from './admin-account-write-lease-repair.ts' +import { adminUserMeterParityCapability } from './admin-user-meter-parity.ts' export const adminDomain = defineDomain({ name: capabilityDomainNames.admin, description: - 'Admin-only operator capabilities for account metadata, platform accounts, package scope grants, fleet package-codemod scan/dry-run/apply/revert over published package trees, feature flags, operator-owned system email, attributed platform feedback users explicitly submit for admin review, and metadata about activity on public community listings; never exposes private package source or unrelated user content such as secrets, memories, jobs, or user inbox email.', + 'Admin-only operator capabilities for account metadata, platform accounts, package scope grants, fleet package-codemod scan/dry-run/apply/revert over published package trees, feature flags, operator-owned system email, attributed platform feedback users explicitly submit for admin review, UserMeter↔D1 parity verification, and metadata about activity on public community listings; never exposes private package source or unrelated user content such as secrets, memories, jobs, or user inbox email.', keywords: [ 'admin', 'rbac', @@ -51,12 +52,16 @@ export const adminDomain = defineDomain({ 'package codemod', 'fleet', 'migration', + 'user meter', + 'parity', + 'cutover', ], capabilities: [ adminUserListCapability, adminUserGetCapability, adminUserCreateCapability, adminUserUpdateCapability, + adminUserMeterParityCapability, adminAccountWriteLeaseListCapability, adminAccountWriteLeaseRepairCapability, adminPlatformAccountCreateCapability, diff --git a/packages/worker/src/mcp/capabilities/registry.node.test.ts b/packages/worker/src/mcp/capabilities/registry.node.test.ts index 2633fcca93..ab53ac3ba4 100644 --- a/packages/worker/src/mcp/capabilities/registry.node.test.ts +++ b/packages/worker/src/mcp/capabilities/registry.node.test.ts @@ -110,10 +110,12 @@ test('getCapabilityRegistryForContext filters admin capabilities by current call }) expect(adminRegistry.capabilityMap.admin_user_list).toBeTruthy() + expect(adminRegistry.capabilityMap.admin_user_meter_parity).toBeTruthy() expect( adminRegistry.capabilityDomains.some((domain) => domain.name === 'admin'), ).toBe(true) expect(regularRegistry.capabilityMap.admin_user_list).toBeUndefined() + expect(regularRegistry.capabilityMap.admin_user_meter_parity).toBeUndefined() expect( regularRegistry.capabilityDomains.some((domain) => domain.name === 'admin'), ).toBe(false)