feat(iios): idempotency-command ledger + IdempotencyService (P9)

Adds the doc-mandated IiosIdempotencyCommand table + a reusable
IdempotencyService.run() that writes IN_FLIGHT at command start, replays the
cached response on a same-key/same-body retry, rejects a same-key/different-body
reuse (409), and retries a FAILED command. Tenant-scoped via
@@unique([scopeId, commandName, idempotencyKey]) (KG-02/KG-06).

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

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

@@ -3,6 +3,7 @@ import { PrismaModule } from './prisma/prisma.module';
 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 { OutboxModule } from './outbox/outbox.module';
 import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
@@ -23,6 +24,7 @@ import { DevController } from './dev/dev.controller';
     PlatformModule,
     IdentityModule,
     InteractionsModule,
+    IdempotencyModule,
     OutboxModule,
     ThreadsModule,
     MessageModule,

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

@@ -0,0 +1,10 @@
+import { Global, Module } from '@nestjs/common';
+import { IdempotencyService } from './idempotency.service';
+
+/** Command-idempotency ledger, global so any feature module can wrap its commands. */
+@Global()
+@Module({
+  providers: [IdempotencyService],
+  exports: [IdempotencyService],
+})
+export class IdempotencyModule {}

packages/iios-service/src/idempotency/idempotency.service.spec.ts

@@ -0,0 +1,89 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { ConflictException } from '@nestjs/common';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import { IdempotencyService } from './idempotency.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 IdempotencyService(asService);
+const CMD = 'test.command';
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('IdempotencyService (P9 — command ledger)', () => {
+  it('runs fn once, returns its result, and records a COMPLETED row with the response', async () => {
+    let calls = 0;
+    const result = await svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => {
+      calls++;
+      return { id: 'r1', ok: true };
+    });
+    expect(result).toEqual({ id: 'r1', ok: true });
+    expect(calls).toBe(1);
+    const row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } });
+    expect(row.status).toBe('COMPLETED');
+    expect(row.response).toEqual({ id: 'r1', ok: true });
+    expect(row.completedAt).toBeTruthy();
+  });
+
+  it('replays the cached response on a duplicate (same request) without re-running fn', async () => {
+    let calls = 0;
+    const params = { scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } };
+    const first = await svc().run(params, async () => { calls++; return { id: 'r1' }; });
+    const second = await svc().run(params, async () => { calls++; return { id: 'r2-should-not-run' }; });
+    expect(second).toEqual(first);
+    expect(calls).toBe(1); // fn never ran the second time
+    expect(await prisma.iiosIdempotencyCommand.count()).toBe(1);
+  });
+
+  it('rejects the same key reused with a different request (fn not called)', async () => {
+    let calls = 0;
+    await svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => { calls++; return { id: 'r1' }; });
+    await expect(
+      svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 2 } }, async () => { calls++; return { id: 'r2' }; }),
+    ).rejects.toBeInstanceOf(ConflictException);
+    expect(calls).toBe(1);
+  });
+
+  it('marks FAILED on throw + propagates, and a later run with the same key retries to COMPLETED', async () => {
+    let calls = 0;
+    const params = { scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } };
+    await expect(
+      svc().run(params, async () => { calls++; throw new Error('boom'); }),
+    ).rejects.toThrow('boom');
+    let row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } });
+    expect(row.status).toBe('FAILED');
+
+    const retry = await svc().run(params, async () => { calls++; return { id: 'r-ok' }; });
+    expect(retry).toEqual({ id: 'r-ok' });
+    expect(calls).toBe(2); // failed attempt + successful retry
+    row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } });
+    expect(row.status).toBe('COMPLETED');
+  });
+
+  it('a concurrent IN_FLIGHT row → conflict (command in progress)', async () => {
+    await prisma.iiosIdempotencyCommand.create({
+      data: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1', requestHash: 'irrelevant', status: 'IN_FLIGHT' },
+    });
+    // Same requestHash so we reach the IN_FLIGHT branch, not the mismatch branch.
+    const { createHash } = await import('node:crypto');
+    const rh = createHash('sha256').update('{"a":1}').digest('hex');
+    await prisma.iiosIdempotencyCommand.updateMany({ where: { idempotencyKey: 'k1' }, data: { requestHash: rh } });
+    await expect(
+      svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'r' })),
+    ).rejects.toThrow(/in progress/);
+  });
+
+  it('the same command+key under two different scopes are independent (tenant isolation)', async () => {
+    const a = await svc().run({ scopeId: 'scope-A', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'A' }));
+    const b = await svc().run({ scopeId: 'scope-B', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'B' }));
+    expect(a).toEqual({ id: 'A' });
+    expect(b).toEqual({ id: 'B' }); // B did not replay A's response
+    expect(await prisma.iiosIdempotencyCommand.count()).toBe(2);
+  });
+});

packages/iios-service/src/idempotency/idempotency.service.ts

@@ -0,0 +1,86 @@
+import { ConflictException, Injectable } from '@nestjs/common';
+import { createHash } from 'node:crypto';
+import { PrismaService } from '../prisma/prisma.service';
+
+const hash = (v: unknown) => createHash('sha256').update(stable(v)).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(',')}}`;
+
+export interface IdempotencyParams {
+  scopeId?: string;
+  commandName: string;
+  key: string;
+  request: unknown;
+  actorRefId?: string;
+}
+
+/**
+ * Idempotent-command ledger (P9, KG-06). Wraps an externally-visible mutation so a
+ * retry with the same (scope, command, key) replays the cached response instead of
+ * re-executing; the same key with a different request body is rejected as a conflict.
+ * Write IN_FLIGHT at command start → COMPLETED (+ cached response) or FAILED on throw.
+ */
+@Injectable()
+export class IdempotencyService {
+  constructor(private readonly prisma: PrismaService) {}
+
+  async run<T>(p: IdempotencyParams, fn: () => Promise<T>): Promise<T> {
+    const requestHash = hash(p.request);
+    const scopeId = p.scopeId ?? null;
+    const where = { scopeId, commandName: p.commandName, idempotencyKey: p.key };
+
+    const existing = await this.prisma.iiosIdempotencyCommand.findFirst({ where });
+    if (existing) return this.resolveExisting<T>(where, existing, requestHash, fn);
+
+    try {
+      await this.prisma.iiosIdempotencyCommand.create({
+        data: { scopeId, commandName: p.commandName, idempotencyKey: p.key, requestHash, actorRefId: p.actorRefId },
+      });
+    } catch {
+      // Lost the create race: another caller inserted the same (scope, command, key).
+      const row = await this.prisma.iiosIdempotencyCommand.findFirst({ where });
+      if (row) return this.resolveExisting<T>(where, row, requestHash, fn);
+      throw new ConflictException('idempotent command conflict');
+    }
+
+    return this.executeAndRecord<T>(where, fn);
+  }
+
+  private async resolveExisting<T>(
+    where: { scopeId: string | null; commandName: string; idempotencyKey: string },
+    row: { status: string; requestHash: string; response: unknown },
+    requestHash: string,
+    fn: () => Promise<T>,
+  ): Promise<T> {
+    if (row.requestHash !== requestHash) throw new ConflictException('idempotency key reused with a different request');
+    if (row.status === 'COMPLETED') return row.response as T; // replay — no re-execution
+    if (row.status === 'IN_FLIGHT') throw new ConflictException('command in progress');
+    return this.executeAndRecord<T>(where, fn); // FAILED → retry + re-record the outcome
+  }
+
+  /** Run the command, then record COMPLETED (+ cached response) or FAILED against its ledger row. */
+  private async executeAndRecord<T>(
+    where: { scopeId: string | null; commandName: string; idempotencyKey: string },
+    fn: () => Promise<T>,
+  ): Promise<T> {
+    try {
+      const result = await fn();
+      await this.prisma.iiosIdempotencyCommand.updateMany({
+        where,
+        data: { status: 'COMPLETED', responseHash: hash(result), response: result as object, completedAt: new Date() },
+      });
+      return result;
+    } catch (err) {
+      await this.prisma.iiosIdempotencyCommand.updateMany({ where, data: { status: 'FAILED', completedAt: new Date() } });
+      throw err;
+    }
+  }
+}

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
-      "IiosDlqItem","IiosAuditLink",
+      "IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink",
       "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest",
       "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow",
       "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",