De-hardcode orchestrator and interaction agent names (#748)

apps/gateway/src/agent/durable-session.repository.ts

@@ -0,0 +1,529 @@
+import { createHash, createHmac } from 'node:crypto';
+import { Inject, Injectable } from '@nestjs/common';
+import {
+  and,
+  asc,
+  desc,
+  eq,
+  interactionCheckpoints,
+  interactionHandoffs,
+  interactionInbox,
+  interactionOutbox,
+  interactionSessions,
+  type Db,
+} from '@mosaicstack/db';
+import { seal, unseal } from '@mosaicstack/auth';
+import { redactSensitiveContent } from '@mosaicstack/log';
+import type {
+  DurableCheckpoint,
+  DurableCheckpointInput,
+  DurableEnqueueResult,
+  DurableHandoff,
+  DurableHandoffInput,
+  DurableInboxEntry,
+  DurableInboxInput,
+  DurableInboxStatus,
+  DurableOutboxEntry,
+  DurableOutboxInput,
+  DurableOutboxStatus,
+  DurableSessionIdentity,
+  DurableSessionSnapshot,
+  DurableSessionStore,
+} from '@mosaicstack/agent';
+import { DB } from '../database/database.module.js';
+
+@Injectable()
+export class DurableSessionRepository implements DurableSessionStore {
+  constructor(@Inject(DB) private readonly db: Db) {}
+
+  async create(identity: DurableSessionIdentity): Promise<void> {
+    await this.db
+      .insert(interactionSessions)
+      .values({
+        id: identity.sessionId,
+        agentName: identity.agentName,
+        tenantId: identity.tenantId,
+        ownerId: identity.ownerId,
+        providerId: identity.providerId,
+        runtimeSessionId: identity.runtimeSessionId,
+      })
+      .onConflictDoNothing();
+
+    const existing = await this.session(identity.sessionId);
+    if (!existing || !sameEnrollmentScope(existing, identity)) {
+      throw new Error(`Durable session identity conflict: ${identity.sessionId}`);
+    }
+    // A recovered/re-enrolled runtime can receive a new provider session ID;
+    // the conversation handle and owner scope remain immutable.
+    if (
+      existing.providerId !== identity.providerId ||
+      existing.runtimeSessionId !== identity.runtimeSessionId
+    ) {
+      await this.db
+        .update(interactionSessions)
+        .set({ providerId: identity.providerId, runtimeSessionId: identity.runtimeSessionId })
+        .where(eq(interactionSessions.id, identity.sessionId));
+    }
+  }
+
+  async snapshot(sessionId: string): Promise<DurableSessionSnapshot | null> {
+    const identity = await this.session(sessionId);
+    if (!identity) return null;
+
+    const [inbox, outbox, checkpoints, handoffs] = await Promise.all([
+      this.db
+        .select()
+        .from(interactionInbox)
+        .where(eq(interactionInbox.sessionId, sessionId))
+        .orderBy(asc(interactionInbox.createdAt)),
+      this.db
+        .select()
+        .from(interactionOutbox)
+        .where(eq(interactionOutbox.sessionId, sessionId))
+        .orderBy(asc(interactionOutbox.createdAt)),
+      this.db
+        .select()
+        .from(interactionCheckpoints)
+        .where(eq(interactionCheckpoints.sessionId, sessionId))
+        .orderBy(
+          desc(interactionCheckpoints.compactionEpoch),
+          desc(interactionCheckpoints.createdAt),
+        )
+        .limit(1),
+      this.db
+        .select()
+        .from(interactionHandoffs)
+        .where(eq(interactionHandoffs.sessionId, sessionId))
+        .orderBy(asc(interactionHandoffs.createdAt)),
+    ]);
+
+    const checkpoint = checkpoints[0];
+    return {
+      identity,
+      inbox: inbox.map(toInbox),
+      outbox: outbox.map(toOutbox),
+      ...(checkpoint ? { checkpoint: toCheckpoint(checkpoint) } : {}),
+      handoffs: handoffs.map(toHandoff),
+    };
+  }
+
+  async enqueueInbox(input: DurableInboxInput): Promise<DurableEnqueueResult<DurableInboxStatus>> {
+    const digest = contentDigest(input.content);
+    const record: DurableInboxInput = {
+      ...input,
+      content: redactSensitiveContent(input.content).content,
+    };
+    const inserted = await this.db
+      .insert(interactionInbox)
+      .values({
+        ...record,
+        content: seal(record.content),
+        contentDigest: digest,
+        status: 'pending',
+      })
+      .onConflictDoNothing()
+      .returning({ status: interactionInbox.status });
+    if (inserted[0]) return { accepted: true, status: inserted[0].status };
+
+    const existing = await this.db
+      .select()
+      .from(interactionInbox)
+      .where(
+        and(
+          eq(interactionInbox.sessionId, input.sessionId),
+          eq(interactionInbox.idempotencyKey, input.idempotencyKey),
+        ),
+      )
+      .limit(1);
+    if (!existing[0]) throw new Error(`Durable inbox enqueue failed: ${input.idempotencyKey}`);
+    const entry = toInbox(existing[0]);
+    if (
+      !sameInbox(entry, record) ||
+      !matchesContentDigest(existing[0].contentDigest, input.content)
+    ) {
+      throw new Error(`Durable inbox idempotency conflict: ${input.idempotencyKey}`);
+    }
+    return { accepted: false, status: entry.status };
+  }
+
+  async claimInbox(sessionId: string): Promise<DurableInboxEntry | null> {
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      const candidate = await this.db
+        .select()
+        .from(interactionInbox)
+        .where(
+          and(eq(interactionInbox.sessionId, sessionId), eq(interactionInbox.status, 'pending')),
+        )
+        .orderBy(asc(interactionInbox.createdAt))
+        .limit(1);
+      const entry = candidate[0];
+      if (!entry) return null;
+      const claimed = await this.db
+        .update(interactionInbox)
+        .set({ status: 'processing', updatedAt: new Date() })
+        .where(and(eq(interactionInbox.id, entry.id), eq(interactionInbox.status, 'pending')))
+        .returning();
+      if (claimed[0]) return toInbox(claimed[0]);
+    }
+    return null;
+  }
+
+  async completeInbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(interactionInbox)
+      .set({ status: 'processed', updatedAt: new Date() })
+      .where(
+        and(
+          eq(interactionInbox.sessionId, sessionId),
+          eq(interactionInbox.idempotencyKey, idempotencyKey),
+          eq(interactionInbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async releaseInbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(interactionInbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(
+        and(
+          eq(interactionInbox.sessionId, sessionId),
+          eq(interactionInbox.idempotencyKey, idempotencyKey),
+          eq(interactionInbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async enqueueOutbox(
+    input: DurableOutboxInput,
+  ): Promise<DurableEnqueueResult<DurableOutboxStatus>> {
+    const digest = contentDigest(input.content);
+    const record: DurableOutboxInput = {
+      ...input,
+      content: redactSensitiveContent(input.content).content,
+    };
+    const inserted = await this.db
+      .insert(interactionOutbox)
+      .values({
+        ...record,
+        content: seal(record.content),
+        contentDigest: digest,
+        status: 'pending',
+      })
+      .onConflictDoNothing()
+      .returning({ status: interactionOutbox.status });
+    if (inserted[0]) return { accepted: true, status: inserted[0].status };
+
+    const existing = await this.db
+      .select()
+      .from(interactionOutbox)
+      .where(
+        and(
+          eq(interactionOutbox.sessionId, input.sessionId),
+          eq(interactionOutbox.idempotencyKey, input.idempotencyKey),
+        ),
+      )
+      .limit(1);
+    if (!existing[0]) throw new Error(`Durable outbox enqueue failed: ${input.idempotencyKey}`);
+    const entry = toOutbox(existing[0]);
+    if (
+      !sameOutbox(entry, record) ||
+      !matchesContentDigest(existing[0].contentDigest, input.content)
+    ) {
+      throw new Error(`Durable outbox idempotency conflict: ${input.idempotencyKey}`);
+    }
+    return { accepted: false, status: entry.status };
+  }
+
+  async claimOutbox(sessionId: string): Promise<DurableOutboxEntry | null> {
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      const candidate = await this.db
+        .select()
+        .from(interactionOutbox)
+        .where(
+          and(eq(interactionOutbox.sessionId, sessionId), eq(interactionOutbox.status, 'pending')),
+        )
+        .orderBy(asc(interactionOutbox.createdAt))
+        .limit(1);
+      const entry = candidate[0];
+      if (!entry) return null;
+      const claimed = await this.db
+        .update(interactionOutbox)
+        .set({ status: 'processing', updatedAt: new Date() })
+        .where(and(eq(interactionOutbox.id, entry.id), eq(interactionOutbox.status, 'pending')))
+        .returning();
+      if (claimed[0]) return toOutbox(claimed[0]);
+    }
+    return null;
+  }
+
+  async claimOutboxByKey(
+    sessionId: string,
+    idempotencyKey: string,
+  ): Promise<DurableOutboxEntry | null> {
+    const claimed = await this.db
+      .update(interactionOutbox)
+      .set({ status: 'processing', updatedAt: new Date() })
+      .where(
+        and(
+          eq(interactionOutbox.sessionId, sessionId),
+          eq(interactionOutbox.idempotencyKey, idempotencyKey),
+          eq(interactionOutbox.status, 'pending'),
+        ),
+      )
+      .returning();
+    return claimed[0] ? toOutbox(claimed[0]) : null;
+  }
+
+  async completeOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(interactionOutbox)
+      .set({ status: 'delivered', updatedAt: new Date() })
+      .where(
+        and(
+          eq(interactionOutbox.sessionId, sessionId),
+          eq(interactionOutbox.idempotencyKey, idempotencyKey),
+          eq(interactionOutbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async releaseOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(interactionOutbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(
+        and(
+          eq(interactionOutbox.sessionId, sessionId),
+          eq(interactionOutbox.idempotencyKey, idempotencyKey),
+          eq(interactionOutbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async checkpoint(input: DurableCheckpointInput): Promise<void> {
+    // Compute identity before redaction.  The persisted digest is keyed so a database
+    // reader cannot use it as an offline oracle for sensitive cursor/summary values.
+    const digest = contentDigest(JSON.stringify([input.cursor, input.summary]));
+    const checkpoint: DurableCheckpointInput = {
+      ...input,
+      cursor: redactSensitiveContent(input.cursor).content,
+      summary: redactSensitiveContent(input.summary).content,
+    };
+    const inserted = await this.db
+      .insert(interactionCheckpoints)
+      .values({
+        ...checkpoint,
+        contentDigest: digest,
+        cursor: seal(checkpoint.cursor),
+        summary: seal(checkpoint.summary),
+      })
+      .onConflictDoNothing()
+      .returning({ checkpointId: interactionCheckpoints.checkpointId });
+    if (inserted[0]) return;
+
+    const existing = await this.db
+      .select()
+      .from(interactionCheckpoints)
+      .where(
+        and(
+          eq(interactionCheckpoints.sessionId, input.sessionId),
+          eq(interactionCheckpoints.checkpointId, input.checkpointId),
+        ),
+      )
+      .limit(1);
+    if (
+      !existing[0] ||
+      !sameCheckpoint(toCheckpoint(existing[0]), checkpoint) ||
+      !matchesCheckpointDigest(existing[0].contentDigest, digest)
+    ) {
+      throw new Error(`Durable checkpoint identity conflict: ${input.checkpointId}`);
+    }
+  }
+
+  async findCheckpoint(sessionId: string, checkpointId: string): Promise<DurableCheckpoint | null> {
+    const checkpoints = await this.db
+      .select()
+      .from(interactionCheckpoints)
+      .where(
+        and(
+          eq(interactionCheckpoints.sessionId, sessionId),
+          eq(interactionCheckpoints.checkpointId, checkpointId),
+        ),
+      )
+      .limit(1);
+    const checkpoint = checkpoints[0];
+    return checkpoint ? toCheckpoint(checkpoint) : null;
+  }
+
+  async handoff(input: DurableHandoffInput): Promise<void> {
+    const checkpoint = await this.findCheckpoint(input.sessionId, input.checkpointId);
+    if (!checkpoint) {
+      throw new Error(`Durable handoff checkpoint is unavailable: ${input.checkpointId}`);
+    }
+    const inserted = await this.db
+      .insert(interactionHandoffs)
+      .values({ ...input })
+      .onConflictDoNothing()
+      .returning({ handoffId: interactionHandoffs.handoffId });
+    if (inserted[0]) return;
+
+    const existing = await this.findHandoff(input.handoffId);
+    if (!existing || !sameHandoff(existing, input)) {
+      throw new Error(`Durable handoff identity conflict: ${input.handoffId}`);
+    }
+  }
+
+  async findHandoff(handoffId: string): Promise<DurableHandoff | null> {
+    const handoffs = await this.db
+      .select()
+      .from(interactionHandoffs)
+      .where(eq(interactionHandoffs.handoffId, handoffId))
+      .limit(1);
+    const handoff = handoffs[0];
+    return handoff ? toHandoff(handoff) : null;
+  }
+
+  async requeueInFlight(sessionId: string): Promise<void> {
+    // Inbox handlers are process-local work. A provider outbox claim may have
+    // reached an external target before a crash, so it is deliberately not
+    // replayed by generic recovery.
+    await this.db
+      .update(interactionInbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(
+        and(eq(interactionInbox.sessionId, sessionId), eq(interactionInbox.status, 'processing')),
+      );
+  }
+
+  private async session(sessionId: string): Promise<DurableSessionIdentity | null> {
+    const sessions = await this.db
+      .select()
+      .from(interactionSessions)
+      .where(eq(interactionSessions.id, sessionId))
+      .limit(1);
+    const session = sessions[0];
+    return session
+      ? {
+          agentName: session.agentName,
+          sessionId: session.id,
+          tenantId: session.tenantId,
+          ownerId: session.ownerId,
+          providerId: session.providerId,
+          runtimeSessionId: session.runtimeSessionId,
+        }
+      : null;
+  }
+}
+
+function contentDigest(content: string): string {
+  const secret = process.env['BETTER_AUTH_SECRET'];
+  if (!secret) {
+    throw new Error('BETTER_AUTH_SECRET is required for durable idempotency digests');
+  }
+  return `hmac:v1:${createHmac('sha256', secret).update(content).digest('hex')}`;
+}
+
+function matchesContentDigest(stored: string, content: string): boolean {
+  return (
+    stored === contentDigest(content) ||
+    stored === createHash('sha256').update(content).digest('hex')
+  );
+}
+
+function matchesCheckpointDigest(stored: string, digest: string): boolean {
+  // Legacy rows predate any pre-redaction identity and cannot safely prove equality.
+  // Reject rather than let redaction collapse distinct sensitive checkpoint payloads.
+  return stored === digest;
+}
+
+function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
+  return (
+    left.agentName === right.agentName &&
+    left.sessionId === right.sessionId &&
+    left.tenantId === right.tenantId &&
+    left.ownerId === right.ownerId
+  );
+}
+
+function sameInbox(left: DurableInboxEntry, right: DurableInboxInput): boolean {
+  return (
+    left.sessionId === right.sessionId &&
+    left.idempotencyKey === right.idempotencyKey &&
+    left.correlationId === right.correlationId &&
+    left.content === right.content
+  );
+}
+
+function sameOutbox(left: DurableOutboxEntry, right: DurableOutboxInput): boolean {
+  return (
+    left.sessionId === right.sessionId &&
+    left.idempotencyKey === right.idempotencyKey &&
+    left.correlationId === right.correlationId &&
+    left.channelId === right.channelId &&
+    left.kind === right.kind &&
+    left.content === right.content
+  );
+}
+
+function sameCheckpoint(left: DurableCheckpoint, right: DurableCheckpointInput): boolean {
+  return (
+    left.sessionId === right.sessionId &&
+    left.checkpointId === right.checkpointId &&
+    left.compactionEpoch === right.compactionEpoch
+  );
+}
+
+function sameHandoff(left: DurableHandoff, right: DurableHandoffInput): boolean {
+  return (
+    left.sessionId === right.sessionId &&
+    left.handoffId === right.handoffId &&
+    left.destination === right.destination &&
+    left.correlationId === right.correlationId &&
+    left.checkpointId === right.checkpointId &&
+    left.status === right.status
+  );
+}
+
+function toInbox(row: typeof interactionInbox.$inferSelect): DurableInboxEntry {
+  return {
+    sessionId: row.sessionId,
+    idempotencyKey: row.idempotencyKey,
+    correlationId: row.correlationId,
+    content: unseal(row.content),
+    status: row.status,
+  };
+}
+
+function toOutbox(row: typeof interactionOutbox.$inferSelect): DurableOutboxEntry {
+  return {
+    sessionId: row.sessionId,
+    idempotencyKey: row.idempotencyKey,
+    correlationId: row.correlationId,
+    channelId: row.channelId,
+    kind: row.kind,
+    content: unseal(row.content),
+    status: row.status,
+  };
+}
+
+function toCheckpoint(row: typeof interactionCheckpoints.$inferSelect): DurableCheckpoint {
+  return {
+    sessionId: row.sessionId,
+    checkpointId: row.checkpointId,
+    cursor: unseal(row.cursor),
+    summary: unseal(row.summary),
+    compactionEpoch: row.compactionEpoch,
+  };
+}
+
+function toHandoff(row: typeof interactionHandoffs.$inferSelect): DurableHandoff {
+  return {
+    sessionId: row.sessionId,
+    handoffId: row.handoffId,
+    destination: row.destination,
+    correlationId: row.correlationId,
+    checkpointId: row.checkpointId,
+    status: row.status,
+  };
+}