feat(iios): projection-cursor ledger + ProjectionCursorService (P9)

Adds the doc-mandated IiosProjectionCursor (ordered high-water-mark +
rolling checksum per projection/topic/partition) and a reusable
ProjectionCursorService.advance(). The checksum is a deterministic fold over
the ordered event stream, so replaying the same events reproduces it — the
KG-06 proof that replay reproduces projection outcomes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/iios-service/src/app.module.ts

@@ -4,6 +4,7 @@ import { PlatformModule } from './platform/platform.module';
 import { IdentityModule } from './identity/identity.module';
 import { InteractionsModule } from './interactions/interactions.module';
 import { IdempotencyModule } from './idempotency/idempotency.module';
+import { ProjectionModule } from './projection/projection.module';
 import { OutboxModule } from './outbox/outbox.module';
 import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
@@ -25,6 +26,7 @@ import { DevController } from './dev/dev.controller';
     IdentityModule,
     InteractionsModule,
     IdempotencyModule,
+    ProjectionModule,
     OutboxModule,
     ThreadsModule,
     MessageModule,

packages/iios-service/src/projection/projection-cursor.service.spec.ts

@@ -0,0 +1,88 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import type { CloudEvent } from '@insignia/iios-contracts';
+import { resetDb } from '../test-utils/reset-db';
+import { ProjectionCursorService } from './projection-cursor.service';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
+const prisma = new PrismaClient({ datasources: { db: { url } } });
+const asService = prisma as unknown as PrismaService;
+const svc = () => new ProjectionCursorService(asService);
+
+const TOPIC = 'com.insignia.iios.test.v1';
+const ce = (id: string, data: Record<string, unknown> = {}, scope = 'scope-1', type = TOPIC): CloudEvent =>
+  ({
+    specversion: '1.0',
+    id,
+    type,
+    source: 'iios/test',
+    time: '2026-01-01T00:00:00.000Z',
+    insignia: { scopeSnapshotId: scope, idempotencyKey: id, dataClass: 'internal' },
+    data,
+  }) as unknown as CloudEvent;
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('ProjectionCursorService (P9 — replay cursor, KG-06)', () => {
+  it('advance creates a cursor at offset 1 with the event id and a 64-hex checksum', async () => {
+    const row = await svc().advance('p', ce('e1', { n: 1 }));
+    expect(row.lastOffset).toBe(1);
+    expect(row.lastEventId).toBe('e1');
+    expect(row.checksum).toMatch(/^[0-9a-f]{64}$/);
+  });
+
+  it('three distinct events advance the offset to 3 and track the latest event id', async () => {
+    const s = svc();
+    await s.advance('p', ce('e1'));
+    await s.advance('p', ce('e2'));
+    const third = await s.advance('p', ce('e3'));
+    expect(third.lastOffset).toBe(3);
+    expect(third.lastEventId).toBe('e3');
+  });
+
+  it('replay determinism: re-running the same ordered stream reproduces the checksum (KG-06)', async () => {
+    const s = svc();
+    await s.advance('p', ce('e1', { a: 1 }));
+    await s.advance('p', ce('e2', { b: 2 }));
+    const c1 = (await s.advance('p', ce('e3', { c: 3 }))).checksum;
+
+    // Wipe the cursor and replay the identical stream.
+    await prisma.iiosProjectionCursor.deleteMany({});
+    await s.advance('p', ce('e1', { a: 1 }));
+    await s.advance('p', ce('e2', { b: 2 }));
+    const c2 = (await s.advance('p', ce('e3', { c: 3 }))).checksum;
+
+    expect(c2).toBe(c1);
+  });
+
+  it('order sensitivity: a different event order yields a different checksum', async () => {
+    const s = svc();
+    await s.advance('p', ce('e1'));
+    const ab = (await s.advance('p', ce('e2'))).checksum;
+
+    await prisma.iiosProjectionCursor.deleteMany({});
+    await s.advance('p', ce('e2'));
+    const ba = (await s.advance('p', ce('e1'))).checksum;
+
+    expect(ba).not.toBe(ab);
+  });
+
+  it('a different source_topic for the same projection is a separate cursor row', async () => {
+    const s = svc();
+    await s.advance('p', ce('e1', {}, 'scope-1', 'topic.a'));
+    await s.advance('p', ce('e2', {}, 'scope-1', 'topic.b'));
+    expect(await prisma.iiosProjectionCursor.count({ where: { projectionName: 'p' } })).toBe(2);
+  });
+
+  it('two partitions (scopes) track independently', async () => {
+    const s = svc();
+    const a = await s.advance('p', ce('e1', {}, 'scope-A'));
+    const b = await s.advance('p', ce('e2', {}, 'scope-B'));
+    expect(a.partitionKey).toBe('scope-A');
+    expect(b.partitionKey).toBe('scope-B');
+    expect(await prisma.iiosProjectionCursor.count()).toBe(2);
+  });
+});

packages/iios-service/src/projection/projection-cursor.service.ts

@@ -0,0 +1,51 @@
+import { Injectable } from '@nestjs/common';
+import { createHash } from 'node:crypto';
+import type { CloudEvent } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+
+const sha = (s: string) => createHash('sha256').update(s).digest('hex');
+// Stable stringify: sort object keys so {a,b} and {b,a} hash equal.
+const stable = (v: unknown): string =>
+  v === null || typeof v !== 'object'
+    ? JSON.stringify(v) ?? 'null'
+    : Array.isArray(v)
+      ? `[${v.map(stable).join(',')}]`
+      : `{${Object.keys(v as object)
+          .sort()
+          .map((k) => `${JSON.stringify(k)}:${stable((v as Record<string, unknown>)[k])}`)
+          .join(',')}}`;
+
+/**
+ * Projection replay cursor (P9, KG-06). Each projection worker folds every event it
+ * consumes into an ordered high-water-mark (lastEventId + monotonic lastOffset) and a
+ * rolling checksum, keyed per (projection, source_topic, partition). Because the checksum
+ * is a deterministic function of the ordered (eventId, eventHash) stream, replaying the
+ * same events reproduces the same checksum — the proof that replay reproduces outcomes.
+ */
+@Injectable()
+export class ProjectionCursorService {
+  constructor(private readonly prisma: PrismaService) {}
+
+  /** Fold one consumed event into the projection's ordered cursor + rolling checksum. */
+  async advance(projectionName: string, event: CloudEvent) {
+    const sourceTopic = event.type;
+    const partitionKey = event.insignia?.scopeSnapshotId ?? 'global';
+    const eventHash = sha(stable({ id: event.id, type: event.type, data: event.data }));
+    const key = { projectionName_sourceTopic_partitionKey: { projectionName, sourceTopic, partitionKey } };
+
+    const prev = await this.prisma.iiosProjectionCursor.findUnique({ where: key });
+    const checksum = sha(`${prev?.checksum ?? ''}${event.id}${eventHash}`);
+    const lastOffset = (prev?.lastOffset ?? 0) + 1;
+
+    return this.prisma.iiosProjectionCursor.upsert({
+      where: key,
+      create: { projectionName, sourceTopic, partitionKey, lastEventId: event.id, lastOffset, checksum },
+      update: { lastEventId: event.id, lastOffset, checksum },
+    });
+  }
+
+  /** All cursors (global ops read for /metrics + replay/repair tooling). */
+  async list() {
+    return this.prisma.iiosProjectionCursor.findMany({ orderBy: [{ projectionName: 'asc' }, { sourceTopic: 'asc' }] });
+  }
+}

packages/iios-service/src/projection/projection.module.ts

@@ -0,0 +1,10 @@
+import { Global, Module } from '@nestjs/common';
+import { ProjectionCursorService } from './projection-cursor.service';
+
+/** Projection replay-cursor ledger, global so every projector can checkpoint. */
+@Global()
+@Module({
+  providers: [ProjectionCursorService],
+  exports: [ProjectionCursorService],
+})
+export class ProjectionModule {}

packages/iios-service/src/test-utils/reset-db.ts

@@ -8,7 +8,7 @@ import type { PrismaClient } from '@prisma/client';
 export async function resetDb(prisma: PrismaClient): Promise<void> {
   await prisma.$executeRawUnsafe(
     `TRUNCATE TABLE
-      "IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink",
+      "IiosProjectionCursor","IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink",
       "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest",
       "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow",
       "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",