feat(tess): persist durable session state

apps/gateway/src/agent/agent.module.ts

@@ -10,14 +10,17 @@ import { ProvidersController } from './providers.controller.js';
 import { SessionsController } from './sessions.controller.js';
 import { AgentConfigsController } from './agent-configs.controller.js';
 import { RoutingController } from './routing/routing.controller.js';
+import { TessDurableSessionRepository } from './tess-durable-session.repository.js';
+import { TessDurableSessionService } from './tess-durable-session.service.js';
 import { CoordModule } from '../coord/coord.module.js';
 import { McpClientModule } from '../mcp-client/mcp-client.module.js';
 import { SkillsModule } from '../skills/skills.module.js';
 import { GCModule } from '../gc/gc.module.js';
 import { LogModule } from '../log/log.module.js';
+import { CommandsModule } from '../commands/commands.module.js';
+import { CommandRuntimeApprovalVerifier } from '../commands/runtime-approval-verifier.js';
 import {
   AGENT_RUNTIME_PROVIDER_REGISTRY,
-  DenyRuntimeApprovalVerifier,
   RUNTIME_APPROVAL_VERIFIER,
   RUNTIME_PROVIDER_AUDIT_SINK,
   RuntimeProviderAuditService,
@@ -26,13 +29,15 @@ import {
 
 @Global()
 @Module({
-  imports: [CoordModule, McpClientModule, SkillsModule, GCModule, LogModule],
+  imports: [CoordModule, McpClientModule, SkillsModule, GCModule, LogModule, CommandsModule],
   providers: [
     ProviderService,
     ProviderCredentialsService,
     RoutingService,
     RoutingEngineService,
     SkillLoaderService,
+    TessDurableSessionRepository,
+    TessDurableSessionService,
     {
       provide: AGENT_RUNTIME_PROVIDER_REGISTRY,
       useFactory: (): AgentRuntimeProviderRegistry => new AgentRuntimeProviderRegistry(),
@@ -42,10 +47,9 @@ import {
       provide: RUNTIME_PROVIDER_AUDIT_SINK,
       useExisting: RuntimeProviderAuditService,
     },
-    DenyRuntimeApprovalVerifier,
     {
       provide: RUNTIME_APPROVAL_VERIFIER,
-      useExisting: DenyRuntimeApprovalVerifier,
+      useExisting: CommandRuntimeApprovalVerifier,
     },
     RuntimeProviderService,
     AgentService,
@@ -58,6 +62,7 @@ import {
     RoutingService,
     RoutingEngineService,
     SkillLoaderService,
+    TessDurableSessionService,
     RuntimeProviderService,
     AGENT_RUNTIME_PROVIDER_REGISTRY,
   ],

apps/gateway/src/agent/tess-durable-session.dto.ts

@@ -0,0 +1,10 @@
+import type { RuntimeProviderRequestContext } from './runtime-provider-registry.service.js';
+
+/** Server-side request for a replay-safe provider message. */
+export interface TessProviderOutboxDto {
+  sessionId: string;
+  idempotencyKey: string;
+  correlationId: string;
+  content: string;
+  context: RuntimeProviderRequestContext;
+}

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

@@ -0,0 +1,336 @@
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { eq, sql, tessInbox } from '@mosaicstack/db';
+import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { createPgliteDb, runPgliteMigrations, type DbHandle } from '@mosaicstack/db';
+import { TessDurableSessionRepository } from './tess-durable-session.repository.js';
+import { TessDurableSessionService } from './tess-durable-session.service.js';
+
+const IDENTITY: DurableSessionIdentity = {
+  sessionId: 'tess-pglite-session',
+  tenantId: 'tenant-pglite',
+  ownerId: 'tess-owner',
+  providerId: 'fleet',
+  runtimeSessionId: 'nova',
+};
+
+describe('TessDurableSessionRepository', () => {
+  let dataDir: string | undefined;
+  let handle: DbHandle | undefined;
+  let previousAuthSecret: string | undefined;
+
+  beforeEach((): void => {
+    previousAuthSecret = process.env['BETTER_AUTH_SECRET'];
+    process.env['BETTER_AUTH_SECRET'] = 'tess-durable-state-test-sealing-key';
+  });
+
+  afterEach(async (): Promise<void> => {
+    await handle?.close();
+    if (dataDir) rmSync(dataDir, { recursive: true, force: true });
+    if (previousAuthSecret === undefined) delete process.env['BETTER_AUTH_SECRET'];
+    else process.env['BETTER_AUTH_SECRET'] = previousAuthSecret;
+  });
+
+  it('survives a full PGlite close/reopen mid-session without duplicate inbox or outbox side effects', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-recovery-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const beforeRestart = new DurableSessionCoordinator(
+      new TessDurableSessionRepository(handle.db),
+    );
+    await beforeRestart.create(IDENTITY);
+    await beforeRestart.receive({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'inbox-before-kill',
+      correlationId: 'correlation-before-kill',
+      content: 'resume after a kill',
+    });
+    await beforeRestart.enqueueOutbox({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'outbox-before-kill',
+      correlationId: 'correlation-before-kill',
+      channelId: 'cli',
+      kind: 'provider.send',
+      content: 'one response only',
+    });
+    await beforeRestart.checkpoint({
+      sessionId: IDENTITY.sessionId,
+      checkpointId: 'checkpoint-before-kill',
+      cursor: 'cursor-before-kill',
+      summary: 'restart-safe state',
+      compactionEpoch: 1,
+    });
+    await beforeRestart.handoff({
+      sessionId: IDENTITY.sessionId,
+      handoffId: 'handoff-before-kill',
+      destination: 'mos',
+      correlationId: 'correlation-before-kill',
+      checkpointId: 'checkpoint-before-kill',
+      status: 'pending',
+    });
+    await beforeRestart.checkpoint({
+      sessionId: IDENTITY.sessionId,
+      checkpointId: 'checkpoint-after-handoff',
+      cursor: 'cursor-after-handoff',
+      summary: 'newer state cannot strand the portable handoff',
+      compactionEpoch: 2,
+    });
+
+    await handle.close();
+    handle = createPgliteDb(dataDir);
+
+    const afterRestart = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    const recovered = await afterRestart.recover(IDENTITY.sessionId);
+    const resumedHandoff = await afterRestart.resumeHandoff('handoff-before-kill');
+    const handled: string[] = [];
+    const effects: string[] = [];
+
+    await afterRestart.drainInbox(IDENTITY.sessionId, async (entry): Promise<void> => {
+      handled.push(entry.idempotencyKey);
+    });
+    await afterRestart.dispatchOutbox(IDENTITY.sessionId, async (entry): Promise<void> => {
+      effects.push(entry.idempotencyKey);
+    });
+    await afterRestart.drainInbox(IDENTITY.sessionId, async (entry): Promise<void> => {
+      handled.push(entry.idempotencyKey);
+    });
+    await afterRestart.dispatchOutbox(IDENTITY.sessionId, async (entry): Promise<void> => {
+      effects.push(entry.idempotencyKey);
+    });
+
+    expect(recovered.identity).toEqual(IDENTITY);
+    expect(recovered.checkpoint).toMatchObject({ checkpointId: 'checkpoint-after-handoff' });
+    expect(recovered.handoffs).toMatchObject([{ handoffId: 'handoff-before-kill' }]);
+    expect(resumedHandoff.checkpoint).toMatchObject({ checkpointId: 'checkpoint-before-kill' });
+    expect(handled).toEqual(['inbox-before-kill']);
+    expect(effects).toEqual(['outbox-before-kill']);
+  }, 30_000);
+
+  it('redacts sensitive durable payloads before persistence', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-redaction-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    await coordinator.create(IDENTITY);
+    await coordinator.receive({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'redacted-inbox',
+      correlationId: 'correlation-redaction',
+      content: 'api_key=super-secret-canary',
+    });
+    await coordinator.enqueueOutbox({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'redacted-outbox',
+      correlationId: 'correlation-redaction',
+      channelId: 'cli',
+      kind: 'provider.send',
+      content: 'email operator@example.test api_key=super-secret-canary',
+    });
+    await coordinator.checkpoint({
+      sessionId: IDENTITY.sessionId,
+      checkpointId: 'redacted-checkpoint',
+      cursor: 'bearer super-secret-canary',
+      summary: 'email operator@example.test',
+      compactionEpoch: 0,
+    });
+
+    const snapshot = await coordinator.snapshot(IDENTITY.sessionId);
+    const [persisted] = await handle.db
+      .select({ content: tessInbox.content })
+      .from(tessInbox)
+      .where(eq(tessInbox.idempotencyKey, 'redacted-inbox'));
+
+    expect(JSON.stringify(snapshot)).not.toContain('super-secret-canary');
+    expect(JSON.stringify(snapshot)).not.toContain('operator@example.test');
+    expect(persisted?.content).not.toContain('super-secret-canary');
+    expect(persisted?.content).not.toContain('[REDACTED]');
+  }, 30_000);
+
+  it('rejects database inbox and outbox idempotency-key conflicts', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-idempotency-conflict-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    await coordinator.create(IDENTITY);
+    await coordinator.receive({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'inbox-conflict',
+      correlationId: 'correlation-inbox',
+      content: 'original inbox',
+    });
+    await coordinator.enqueueOutbox({
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'outbox-conflict',
+      correlationId: 'correlation-outbox',
+      channelId: 'cli',
+      kind: 'provider.send',
+      content: 'original outbox',
+    });
+
+    await expect(
+      coordinator.receive({
+        sessionId: IDENTITY.sessionId,
+        idempotencyKey: 'inbox-conflict',
+        correlationId: 'forged-correlation',
+        content: 'original inbox',
+      }),
+    ).rejects.toThrow(/idempotency conflict/);
+    await expect(
+      coordinator.enqueueOutbox({
+        sessionId: IDENTITY.sessionId,
+        idempotencyKey: 'outbox-conflict',
+        correlationId: 'correlation-outbox',
+        channelId: 'forged-channel',
+        kind: 'provider.send',
+        content: 'original outbox',
+      }),
+    ).rejects.toThrow(/idempotency conflict/);
+  }, 30_000);
+
+  it('does not requeue a live outbox claim during a normal scoped dispatch', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-live-claim-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const repository = new TessDurableSessionRepository(handle.db);
+    const coordinator = new DurableSessionCoordinator(repository);
+    const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
+    const service = new TessDurableSessionService(repository, runtimeProviders as never);
+    const input = {
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'live-effect',
+      correlationId: 'correlation-live',
+      content: 'must not duplicate',
+      context: {
+        actorScope: { userId: IDENTITY.ownerId, tenantId: IDENTITY.tenantId },
+        channelId: 'cli',
+        correlationId: 'correlation-live',
+      },
+    };
+
+    await coordinator.create(IDENTITY);
+    await service.queueProviderSend(input);
+    expect(await repository.claimOutbox(IDENTITY.sessionId)).toMatchObject({
+      status: 'processing',
+    });
+
+    await service.dispatchProviderOutbox(IDENTITY.sessionId, input);
+    await expect(
+      service.recoverProviderSession(IDENTITY.sessionId, {
+        ...input,
+        context: {
+          ...input.context,
+          actorScope: { userId: 'intruder', tenantId: 'tenant-pglite' },
+        },
+      }),
+    ).rejects.toThrow(/scope or correlation mismatch/);
+
+    expect(runtimeProviders.sendMessage).not.toHaveBeenCalled();
+    expect(await coordinator.snapshot(IDENTITY.sessionId)).toMatchObject({
+      outbox: [{ idempotencyKey: 'live-effect', status: 'processing' }],
+    });
+  }, 30_000);
+
+  it('rejects an outbox correlation mismatch before claiming the pending effect', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-outbox-mismatch-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const repository = new TessDurableSessionRepository(handle.db);
+    const coordinator = new DurableSessionCoordinator(repository);
+    const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
+    const service = new TessDurableSessionService(repository, runtimeProviders as never);
+    const input = {
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'mismatch-effect',
+      correlationId: 'correlation-expected',
+      content: 'must remain pending',
+      context: {
+        actorScope: { userId: IDENTITY.ownerId, tenantId: IDENTITY.tenantId },
+        channelId: 'cli',
+        correlationId: 'correlation-expected',
+      },
+    };
+
+    await coordinator.create(IDENTITY);
+    await service.queueProviderSend(input);
+    await expect(
+      service.dispatchProviderOutbox(IDENTITY.sessionId, {
+        ...input,
+        correlationId: 'correlation-forged',
+        context: { ...input.context, correlationId: 'correlation-forged' },
+      }),
+    ).rejects.toThrow(/scope or correlation mismatch/);
+
+    expect(runtimeProviders.sendMessage).not.toHaveBeenCalled();
+    expect(await coordinator.snapshot(IDENTITY.sessionId)).toMatchObject({
+      outbox: [{ idempotencyKey: 'mismatch-effect', status: 'pending' }],
+    });
+  }, 30_000);
+
+  it('dispatches only the outbox record bound to the supplied correlation and channel', async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-outbox-scope-'));
+    handle = createPgliteDb(dataDir);
+    await runPgliteMigrations(handle);
+    await seedOwner(handle);
+
+    const repository = new TessDurableSessionRepository(handle.db);
+    const coordinator = new DurableSessionCoordinator(repository);
+    const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
+    const service = new TessDurableSessionService(repository, runtimeProviders as never);
+    const first = {
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'scoped-effect-one',
+      correlationId: 'correlation-one',
+      content: 'first result',
+      context: {
+        actorScope: { userId: IDENTITY.ownerId, tenantId: IDENTITY.tenantId },
+        channelId: 'cli',
+        correlationId: 'correlation-one',
+      },
+    };
+    const second = {
+      ...first,
+      idempotencyKey: 'scoped-effect-two',
+      correlationId: 'correlation-two',
+      content: 'second result',
+      context: { ...first.context, correlationId: 'correlation-two' },
+    };
+
+    await coordinator.create(IDENTITY);
+    await service.queueProviderSend(first);
+    await service.queueProviderSend(second);
+    await service.dispatchProviderOutbox(IDENTITY.sessionId, first);
+
+    expect(runtimeProviders.sendMessage).toHaveBeenCalledTimes(1);
+    expect(runtimeProviders.sendMessage).toHaveBeenCalledWith(
+      IDENTITY.providerId,
+      IDENTITY.runtimeSessionId,
+      { content: 'first result', idempotencyKey: 'scoped-effect-one' },
+      first.context,
+    );
+    expect(await coordinator.snapshot(IDENTITY.sessionId)).toMatchObject({
+      outbox: [
+        { idempotencyKey: 'scoped-effect-one', status: 'delivered' },
+        { idempotencyKey: 'scoped-effect-two', status: 'pending' },
+      ],
+    });
+  }, 30_000);
+});
+
+async function seedOwner(handle: DbHandle): Promise<void> {
+  await handle.db.execute(sql`
+    INSERT INTO users (id, name, email, email_verified, created_at, updated_at)
+    VALUES ('tess-owner', 'Tess Owner', 'tess-owner@example.test', false, now(), now())
+  `);
+}

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

@@ -0,0 +1,454 @@
+import { Inject, Injectable } from '@nestjs/common';
+import {
+  and,
+  asc,
+  desc,
+  eq,
+  tessCheckpoints,
+  tessHandoffs,
+  tessInbox,
+  tessOutbox,
+  tessSessions,
+  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 TessDurableSessionRepository implements DurableSessionStore {
+  constructor(@Inject(DB) private readonly db: Db) {}
+
+  async create(identity: DurableSessionIdentity): Promise<void> {
+    await this.db
+      .insert(tessSessions)
+      .values({
+        id: identity.sessionId,
+        tenantId: identity.tenantId,
+        ownerId: identity.ownerId,
+        providerId: identity.providerId,
+        runtimeSessionId: identity.runtimeSessionId,
+      })
+      .onConflictDoNothing();
+
+    const existing = await this.session(identity.sessionId);
+    if (!existing || !sameIdentity(existing, identity)) {
+      throw new Error(`Durable Tess session identity conflict: ${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(tessInbox)
+        .where(eq(tessInbox.sessionId, sessionId))
+        .orderBy(asc(tessInbox.createdAt)),
+      this.db
+        .select()
+        .from(tessOutbox)
+        .where(eq(tessOutbox.sessionId, sessionId))
+        .orderBy(asc(tessOutbox.createdAt)),
+      this.db
+        .select()
+        .from(tessCheckpoints)
+        .where(eq(tessCheckpoints.sessionId, sessionId))
+        .orderBy(desc(tessCheckpoints.compactionEpoch), desc(tessCheckpoints.createdAt))
+        .limit(1),
+      this.db
+        .select()
+        .from(tessHandoffs)
+        .where(eq(tessHandoffs.sessionId, sessionId))
+        .orderBy(asc(tessHandoffs.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 record: DurableInboxInput = {
+      ...input,
+      content: redactSensitiveContent(input.content).content,
+    };
+    const inserted = await this.db
+      .insert(tessInbox)
+      .values({ ...record, content: seal(record.content), status: 'pending' })
+      .onConflictDoNothing()
+      .returning({ status: tessInbox.status });
+    if (inserted[0]) return { accepted: true, status: inserted[0].status };
+
+    const existing = await this.db
+      .select()
+      .from(tessInbox)
+      .where(
+        and(
+          eq(tessInbox.sessionId, input.sessionId),
+          eq(tessInbox.idempotencyKey, input.idempotencyKey),
+        ),
+      )
+      .limit(1);
+    if (!existing[0]) throw new Error(`Durable Tess inbox enqueue failed: ${input.idempotencyKey}`);
+    const entry = toInbox(existing[0]);
+    if (!sameInbox(entry, record)) {
+      throw new Error(`Durable Tess 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(tessInbox)
+        .where(and(eq(tessInbox.sessionId, sessionId), eq(tessInbox.status, 'pending')))
+        .orderBy(asc(tessInbox.createdAt))
+        .limit(1);
+      const entry = candidate[0];
+      if (!entry) return null;
+      const claimed = await this.db
+        .update(tessInbox)
+        .set({ status: 'processing', updatedAt: new Date() })
+        .where(and(eq(tessInbox.id, entry.id), eq(tessInbox.status, 'pending')))
+        .returning();
+      if (claimed[0]) return toInbox(claimed[0]);
+    }
+    return null;
+  }
+
+  async completeInbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(tessInbox)
+      .set({ status: 'processed', updatedAt: new Date() })
+      .where(
+        and(
+          eq(tessInbox.sessionId, sessionId),
+          eq(tessInbox.idempotencyKey, idempotencyKey),
+          eq(tessInbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async releaseInbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(tessInbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(
+        and(
+          eq(tessInbox.sessionId, sessionId),
+          eq(tessInbox.idempotencyKey, idempotencyKey),
+          eq(tessInbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async enqueueOutbox(
+    input: DurableOutboxInput,
+  ): Promise<DurableEnqueueResult<DurableOutboxStatus>> {
+    const record: DurableOutboxInput = {
+      ...input,
+      content: redactSensitiveContent(input.content).content,
+    };
+    const inserted = await this.db
+      .insert(tessOutbox)
+      .values({ ...record, content: seal(record.content), status: 'pending' })
+      .onConflictDoNothing()
+      .returning({ status: tessOutbox.status });
+    if (inserted[0]) return { accepted: true, status: inserted[0].status };
+
+    const existing = await this.db
+      .select()
+      .from(tessOutbox)
+      .where(
+        and(
+          eq(tessOutbox.sessionId, input.sessionId),
+          eq(tessOutbox.idempotencyKey, input.idempotencyKey),
+        ),
+      )
+      .limit(1);
+    if (!existing[0])
+      throw new Error(`Durable Tess outbox enqueue failed: ${input.idempotencyKey}`);
+    const entry = toOutbox(existing[0]);
+    if (!sameOutbox(entry, record)) {
+      throw new Error(`Durable Tess 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(tessOutbox)
+        .where(and(eq(tessOutbox.sessionId, sessionId), eq(tessOutbox.status, 'pending')))
+        .orderBy(asc(tessOutbox.createdAt))
+        .limit(1);
+      const entry = candidate[0];
+      if (!entry) return null;
+      const claimed = await this.db
+        .update(tessOutbox)
+        .set({ status: 'processing', updatedAt: new Date() })
+        .where(and(eq(tessOutbox.id, entry.id), eq(tessOutbox.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(tessOutbox)
+      .set({ status: 'processing', updatedAt: new Date() })
+      .where(
+        and(
+          eq(tessOutbox.sessionId, sessionId),
+          eq(tessOutbox.idempotencyKey, idempotencyKey),
+          eq(tessOutbox.status, 'pending'),
+        ),
+      )
+      .returning();
+    return claimed[0] ? toOutbox(claimed[0]) : null;
+  }
+
+  async completeOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(tessOutbox)
+      .set({ status: 'delivered', updatedAt: new Date() })
+      .where(
+        and(
+          eq(tessOutbox.sessionId, sessionId),
+          eq(tessOutbox.idempotencyKey, idempotencyKey),
+          eq(tessOutbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async releaseOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
+    await this.db
+      .update(tessOutbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(
+        and(
+          eq(tessOutbox.sessionId, sessionId),
+          eq(tessOutbox.idempotencyKey, idempotencyKey),
+          eq(tessOutbox.status, 'processing'),
+        ),
+      );
+  }
+
+  async checkpoint(input: DurableCheckpointInput): Promise<void> {
+    const checkpoint: DurableCheckpointInput = {
+      ...input,
+      cursor: redactSensitiveContent(input.cursor).content,
+      summary: redactSensitiveContent(input.summary).content,
+    };
+    const inserted = await this.db
+      .insert(tessCheckpoints)
+      .values({
+        ...checkpoint,
+        cursor: seal(checkpoint.cursor),
+        summary: seal(checkpoint.summary),
+      })
+      .onConflictDoNothing()
+      .returning({ checkpointId: tessCheckpoints.checkpointId });
+    if (inserted[0]) return;
+
+    const existing = await this.findCheckpoint(input.sessionId, input.checkpointId);
+    if (!existing || !sameCheckpoint(existing, checkpoint)) {
+      throw new Error(`Durable Tess checkpoint identity conflict: ${input.checkpointId}`);
+    }
+  }
+
+  async findCheckpoint(sessionId: string, checkpointId: string): Promise<DurableCheckpoint | null> {
+    const checkpoints = await this.db
+      .select()
+      .from(tessCheckpoints)
+      .where(
+        and(
+          eq(tessCheckpoints.sessionId, sessionId),
+          eq(tessCheckpoints.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 Tess handoff checkpoint is unavailable: ${input.checkpointId}`);
+    }
+    const inserted = await this.db
+      .insert(tessHandoffs)
+      .values({ ...input })
+      .onConflictDoNothing()
+      .returning({ handoffId: tessHandoffs.handoffId });
+    if (inserted[0]) return;
+
+    const existing = await this.findHandoff(input.handoffId);
+    if (!existing || !sameHandoff(existing, input)) {
+      throw new Error(`Durable Tess handoff identity conflict: ${input.handoffId}`);
+    }
+  }
+
+  async findHandoff(handoffId: string): Promise<DurableHandoff | null> {
+    const handoffs = await this.db
+      .select()
+      .from(tessHandoffs)
+      .where(eq(tessHandoffs.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(tessInbox)
+      .set({ status: 'pending', updatedAt: new Date() })
+      .where(and(eq(tessInbox.sessionId, sessionId), eq(tessInbox.status, 'processing')));
+  }
+
+  private async session(sessionId: string): Promise<DurableSessionIdentity | null> {
+    const sessions = await this.db
+      .select()
+      .from(tessSessions)
+      .where(eq(tessSessions.id, sessionId))
+      .limit(1);
+    const session = sessions[0];
+    return session
+      ? {
+          sessionId: session.id,
+          tenantId: session.tenantId,
+          ownerId: session.ownerId,
+          providerId: session.providerId,
+          runtimeSessionId: session.runtimeSessionId,
+        }
+      : null;
+  }
+}
+
+function sameIdentity(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
+  return (
+    left.sessionId === right.sessionId &&
+    left.tenantId === right.tenantId &&
+    left.ownerId === right.ownerId &&
+    left.providerId === right.providerId &&
+    left.runtimeSessionId === right.runtimeSessionId
+  );
+}
+
+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.cursor === right.cursor &&
+    left.summary === right.summary &&
+    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 tessInbox.$inferSelect): DurableInboxEntry {
+  return {
+    sessionId: row.sessionId,
+    idempotencyKey: row.idempotencyKey,
+    correlationId: row.correlationId,
+    content: unseal(row.content),
+    status: row.status,
+  };
+}
+
+function toOutbox(row: typeof tessOutbox.$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 tessCheckpoints.$inferSelect): DurableCheckpoint {
+  return {
+    sessionId: row.sessionId,
+    checkpointId: row.checkpointId,
+    cursor: unseal(row.cursor),
+    summary: unseal(row.summary),
+    compactionEpoch: row.compactionEpoch,
+  };
+}
+
+function toHandoff(row: typeof tessHandoffs.$inferSelect): DurableHandoff {
+  return {
+    sessionId: row.sessionId,
+    handoffId: row.handoffId,
+    destination: row.destination,
+    correlationId: row.correlationId,
+    checkpointId: row.checkpointId,
+    status: row.status,
+  };
+}

apps/gateway/src/agent/tess-durable-session.service.ts

@@ -0,0 +1,93 @@
+import { ForbiddenException, Inject, Injectable } from '@nestjs/common';
+import { DurableSessionCoordinator } from '@mosaicstack/agent';
+import type { TessProviderOutboxDto } from './tess-durable-session.dto.js';
+import { TessDurableSessionRepository } from './tess-durable-session.repository.js';
+import { RuntimeProviderService } from './runtime-provider-registry.service.js';
+
+/**
+ * Scoped gateway boundary for the canonical Tess state machine. It deliberately
+ * uses composition: raw state methods cannot be injected into channel, CLI, or
+ * MCP adapters without a server-derived actor/tenant/correlation context.
+ */
+@Injectable()
+export class TessDurableSessionService {
+  private readonly coordinator: DurableSessionCoordinator;
+
+  constructor(
+    @Inject(TessDurableSessionRepository) repository: TessDurableSessionRepository,
+    @Inject(RuntimeProviderService) private readonly runtimeProviders: RuntimeProviderService,
+  ) {
+    this.coordinator = new DurableSessionCoordinator(repository);
+  }
+
+  async queueProviderSend(input: TessProviderOutboxDto): Promise<void> {
+    const snapshot = await this.coordinator.snapshot(input.sessionId);
+    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
+    await this.coordinator.enqueueOutbox({
+      sessionId: input.sessionId,
+      idempotencyKey: input.idempotencyKey,
+      correlationId: input.correlationId,
+      channelId: input.context.channelId,
+      kind: 'provider.send',
+      content: input.content,
+    });
+  }
+
+  async dispatchProviderOutbox(sessionId: string, input: TessProviderOutboxDto): Promise<void> {
+    if (sessionId !== input.sessionId) {
+      throw new ForbiddenException('Durable Tess outbox session mismatch');
+    }
+    const snapshot = await this.coordinator.snapshot(sessionId);
+    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
+    const pendingEntry = snapshot.outbox.find(
+      (entry): boolean => entry.idempotencyKey === input.idempotencyKey,
+    );
+    if (!pendingEntry) return;
+    // Validate immutable routing before claiming. A caller with a mismatched
+    // correlation/channel must not strand a pending external side effect.
+    this.assertOutboxScope(pendingEntry, input);
+    await this.coordinator.dispatchOutboxEntry(
+      sessionId,
+      input.idempotencyKey,
+      async (entry): Promise<void> => {
+        this.assertOutboxScope(entry, input);
+        await this.runtimeProviders.sendMessage(
+          snapshot.identity.providerId,
+          snapshot.identity.runtimeSessionId,
+          { content: entry.content, idempotencyKey: entry.idempotencyKey },
+          input.context,
+        );
+      },
+    );
+  }
+
+  /** Startup/recovery-only path; normal queue/dispatch methods never requeue live work. */
+  async recoverProviderSession(sessionId: string, input: TessProviderOutboxDto): Promise<void> {
+    const snapshot = await this.coordinator.snapshot(sessionId);
+    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
+    await this.coordinator.recover(sessionId);
+  }
+
+  private assertOutboxScope(
+    entry: { kind: string; correlationId: string; channelId: string },
+    input: TessProviderOutboxDto,
+  ): void {
+    if (
+      entry.kind !== 'provider.send' ||
+      entry.correlationId !== input.correlationId ||
+      entry.channelId !== input.context.channelId
+    ) {
+      throw new ForbiddenException('Durable Tess outbox scope or correlation mismatch');
+    }
+  }
+
+  private assertScope(ownerId: string, tenantId: string, input: TessProviderOutboxDto): void {
+    if (
+      input.context.actorScope.userId !== ownerId ||
+      input.context.actorScope.tenantId !== tenantId ||
+      input.context.correlationId !== input.correlationId
+    ) {
+      throw new ForbiddenException('Durable Tess session scope or correlation mismatch');
+    }
+  }
+}

apps/gateway/src/commands/command-authorization.service.spec.ts

@@ -12,8 +12,10 @@ const adminCommand: CommandDef = {
 };
 const payload: SlashCommandPayload = { command: 'gc', conversationId: 'conversation-1' };
 
-function createService(role: string): CommandAuthorizationService {
-  const entries = new Map<string, string>();
+function createService(
+  role: string,
+  entries: Map<string, string> = new Map<string, string>(),
+): CommandAuthorizationService {
   const db = {
     select: () => ({ from: () => ({ where: () => ({ limit: async () => [{ role }] }) }) }),
   };
@@ -58,4 +60,55 @@ describe('CommandAuthorizationService', () => {
       (await service.authorize(adminCommand, payload, 'member-1', 'forged-approval-id')).allowed,
     ).toBe(false);
   });
+
+  it('denies a malformed durable approval expiry instead of treating it as unexpired', async (): Promise<void> => {
+    const entries = new Map<string, string>();
+    const action = {
+      providerId: 'fleet',
+      sessionId: 'nova',
+      actorId: 'admin-1',
+      tenantId: 'tenant-1',
+      channelId: 'discord:operator',
+      correlationId: 'correlation-malformed-expiry',
+    };
+    const service = createService('admin', entries);
+    const approval = await service.createRuntimeTerminationApproval(action);
+    expect(approval).not.toBeNull();
+    const key = `tess:command-approval:${approval!.approvalId}`;
+    const stored = entries.get(key);
+    expect(stored).toBeDefined();
+    entries.set(key, JSON.stringify({ ...JSON.parse(stored!), expiresAt: 'not-a-date' }));
+
+    expect(await service.consumeRuntimeTerminationApproval(approval!.approvalId, action)).toBe(
+      false,
+    );
+  });
+
+  it('persists and consumes one exact runtime termination approval across a service restart', async (): Promise<void> => {
+    const entries = new Map<string, string>();
+    const action = {
+      providerId: 'fleet',
+      sessionId: 'nova',
+      actorId: 'admin-1',
+      tenantId: 'tenant-1',
+      channelId: 'discord:operator',
+      correlationId: 'correlation-1',
+    };
+    const beforeRestart = createService('admin', entries);
+    const approval = await beforeRestart.createRuntimeTerminationApproval(action);
+
+    const afterRestart = createService('admin', entries);
+    expect(
+      await afterRestart.consumeRuntimeTerminationApproval(approval!.approvalId, {
+        ...action,
+        sessionId: 'forged-session',
+      }),
+    ).toBe(false);
+    expect(await afterRestart.consumeRuntimeTerminationApproval(approval!.approvalId, action)).toBe(
+      true,
+    );
+    expect(await afterRestart.consumeRuntimeTerminationApproval(approval!.approvalId, action)).toBe(
+      false,
+    );
+  });
 });

apps/gateway/src/commands/command-authorization.service.ts

@@ -15,6 +15,22 @@ export interface CommandApproval {
   expiresAt: string;
 }
 
+/** Exact immutable binding for a privileged runtime termination. */
+export interface RuntimeTerminationApprovalAction {
+  providerId: string;
+  sessionId: string;
+  actorId: string;
+  tenantId: string;
+  channelId: string;
+  correlationId: string;
+}
+
+export interface RuntimeTerminationApproval extends RuntimeTerminationApprovalAction {
+  approvalId: string;
+  actionDigest: string;
+  expiresAt: string;
+}
+
 export interface CommandAuthorizationResult {
   allowed: boolean;
   reason?: string;
@@ -75,6 +91,52 @@ export class CommandAuthorizationService {
     return approval;
   }
 
+  /**
+   * Uses the same `tess:command-approval:*` store and one-time deletion rule as
+   * command approvals. This deliberately avoids a parallel approval database.
+   */
+  async createRuntimeTerminationApproval(
+    action: RuntimeTerminationApprovalAction,
+  ): Promise<RuntimeTerminationApproval | null> {
+    if (!this.hasRuntimeTerminationAction(action)) return null;
+    const role = await this.resolveRole(action.actorId);
+    if (role !== 'admin') return null;
+
+    const approval: RuntimeTerminationApproval = {
+      approvalId: randomUUID(),
+      actionDigest: this.runtimeActionDigest(action),
+      ...action,
+      expiresAt: new Date(Date.now() + 5 * 60_000).toISOString(),
+    };
+    await this.redis.set(this.key(approval.approvalId), JSON.stringify(approval), 'EX', '300');
+    return approval;
+  }
+
+  async consumeRuntimeTerminationApproval(
+    approvalId: string,
+    action: RuntimeTerminationApprovalAction,
+  ): Promise<boolean> {
+    const encoded = await this.redis.get(this.key(approvalId));
+    if (!encoded) return false;
+    let approval: unknown;
+    try {
+      approval = JSON.parse(encoded);
+    } catch {
+      return false;
+    }
+    if (
+      !this.isRuntimeTerminationApproval(approval) ||
+      approval.actionDigest !== this.runtimeActionDigest(action) ||
+      approval.actorId !== action.actorId ||
+      approval.tenantId !== action.tenantId ||
+      !this.isUnexpired(approval.expiresAt)
+    ) {
+      return false;
+    }
+    if ((await this.resolveRole(approval.actorId)) !== 'admin') return false;
+    return (await this.redis.del(this.key(approvalId))) === 1;
+  }
+
   private async resolveRole(actorId: string): Promise<CommandRole | null> {
     const [user] = await this.db
       .select({ role: usersTable.role })
@@ -98,12 +160,17 @@ export class CommandAuthorizationService {
     const key = this.key(approvalId);
     const encoded = await this.redis.get(key);
     if (!encoded) return false;
-    const parsed: unknown = JSON.parse(encoded);
+    let parsed: unknown;
+    try {
+      parsed = JSON.parse(encoded);
+    } catch {
+      return false;
+    }
     if (
-      !this.isApproval(parsed) ||
+      !this.isCommandApproval(parsed) ||
       parsed.actorId !== actorId ||
       parsed.actionDigest !== actionDigest ||
-      Date.parse(parsed.expiresAt) <= Date.now()
+      !this.isUnexpired(parsed.expiresAt)
     )
       return false;
     return (await this.redis.del(key)) === 1;
@@ -121,13 +188,62 @@ export class CommandAuthorizationService {
       .digest('hex');
   }
 
-  private isApproval(value: unknown): value is CommandApproval {
+  private hasRuntimeTerminationAction(action: RuntimeTerminationApprovalAction): boolean {
+    return [
+      action.providerId,
+      action.sessionId,
+      action.actorId,
+      action.tenantId,
+      action.channelId,
+      action.correlationId,
+    ].every((value: string): boolean => value.trim().length > 0);
+  }
+
+  private runtimeActionDigest(action: RuntimeTerminationApprovalAction): string {
+    return createHash('sha256')
+      .update(
+        JSON.stringify({
+          providerId: action.providerId,
+          sessionId: action.sessionId,
+          actorId: action.actorId,
+          tenantId: action.tenantId,
+          channelId: action.channelId,
+          correlationId: action.correlationId,
+        }),
+      )
+      .digest('hex');
+  }
+
+  private isUnexpired(expiresAt: unknown): expiresAt is string {
+    if (typeof expiresAt !== 'string') return false;
+    const expiresAtMs = Date.parse(expiresAt);
+    return Number.isFinite(expiresAtMs) && expiresAtMs > Date.now();
+  }
+
+  private isCommandApproval(value: unknown): value is CommandApproval {
     return (
       typeof value === 'object' &&
       value !== null &&
       'approvalId' in value &&
       'actionDigest' in value &&
       'actorId' in value &&
+      'expiresAt' in value &&
+      'command' in value
+    );
+  }
+
+  private isRuntimeTerminationApproval(value: unknown): value is RuntimeTerminationApproval {
+    return (
+      typeof value === 'object' &&
+      value !== null &&
+      'approvalId' in value &&
+      'actionDigest' in value &&
+      'actorId' in value &&
+      'tenantId' in value &&
+      'providerId' in value &&
+      'sessionId' in value &&
+      'channelId' in value &&
+      'correlationId' in value &&
       'expiresAt' in value
     );
   }

apps/gateway/src/commands/commands.module.ts

@@ -6,6 +6,7 @@ import { ReloadModule } from '../reload/reload.module.js';
 import { CommandAuthorizationService } from './command-authorization.service.js';
 import { CommandExecutorService } from './command-executor.service.js';
 import { CommandRegistryService } from './command-registry.service.js';
+import { CommandRuntimeApprovalVerifier } from './runtime-approval-verifier.js';
 import { COMMANDS_REDIS } from './commands.tokens.js';
 
 const COMMANDS_QUEUE_HANDLE = 'COMMANDS_QUEUE_HANDLE';
@@ -26,9 +27,15 @@ const COMMANDS_QUEUE_HANDLE = 'COMMANDS_QUEUE_HANDLE';
     },
     CommandRegistryService,
     CommandAuthorizationService,
+    CommandRuntimeApprovalVerifier,
+    CommandExecutorService,
+  ],
+  exports: [
+    CommandRegistryService,
+    CommandAuthorizationService,
+    CommandRuntimeApprovalVerifier,
     CommandExecutorService,
   ],
-  exports: [CommandRegistryService, CommandExecutorService],
 })
 export class CommandsModule implements OnApplicationShutdown {
   constructor(@Inject(COMMANDS_QUEUE_HANDLE) private readonly handle: QueueHandle) {}

apps/gateway/src/commands/runtime-approval-verifier.ts

@@ -0,0 +1,23 @@
+import { Inject, Injectable } from '@nestjs/common';
+import type {
+  RuntimeApprovalVerifier,
+  RuntimeTerminationAction,
+} from '../agent/runtime-provider-registry.service.js';
+import { CommandAuthorizationService } from './command-authorization.service.js';
+
+/**
+ * Adapter from the provider registry's exact termination action to the shared,
+ * Redis-backed `tess:command-approval:*` store. It has no separate approval
+ * persistence or replay semantics.
+ */
+@Injectable()
+export class CommandRuntimeApprovalVerifier implements RuntimeApprovalVerifier {
+  constructor(
+    @Inject(CommandAuthorizationService)
+    private readonly authorization: CommandAuthorizationService,
+  ) {}
+
+  async consume(approvalRef: string, action: RuntimeTerminationAction): Promise<boolean> {
+    return this.authorization.consumeRuntimeTerminationApproval(approvalRef, action);
+  }
+}