mosaicstack/stack
1
0
Fork 0
Code Issues 42 Pull Requests Actions Packages Projects Releases 15 Wiki Activity

fix(tess): harden durable recovery namespaces
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
Details

Browse Source
This commit is contained in:
Jarvis
2026-07-12 23:36:14 -05:00
parent 102a7b606b
commit 444988d23b
22 changed files with 449 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/gateway/src/agent/__tests__/runtime-provider-registry.service.test.ts
View File

@@ -22,6 +22,8 @@ import {
type RuntimeApprovalVerifier,
} from '../runtime-provider-registry.service.js';
process.env['MOSAIC_AGENT_NAME'] ??= 'test-runtime-agent';
const OWNER_SCOPE: ActorTenantScope = { userId: 'owner-1', tenantId: 'tenant-1' };
const CONTEXT = {
actorScope: OWNER_SCOPE,
@@ -243,6 +245,7 @@ describe('RuntimeProviderService security boundary', (): void => {
tenantId: OWNER_SCOPE.tenantId,
channelId: CONTEXT.channelId,
correlationId: CONTEXT.correlationId,
agentName: process.env['MOSAIC_AGENT_NAME'],
});
expect(provider.terminateCalls).toBe(1);
});

8
apps/gateway/src/agent/runtime-provider-registry.service.ts
View File

@@ -64,12 +64,19 @@ export interface RuntimeTerminationAction {
tenantId: string;
channelId: string;
correlationId: string;
agentName: string;
}
export interface RuntimeApprovalVerifier {
consume(approvalRef: string, action: RuntimeTerminationAction): Promise<boolean>;
}
function configuredAgentName(): string {
const agentName = process.env['MOSAIC_AGENT_NAME']?.trim();
if (!agentName) throw new RuntimeApprovalDeniedError();
return agentName;
}
class RuntimeApprovalDeniedError extends Error {
constructor() {
super('Runtime termination approval denied');
@@ -261,6 +268,7 @@ export class RuntimeProviderService {
tenantId: scope.tenantId,
channelId: scope.channelId,
correlationId: scope.correlationId,
agentName: configuredAgentName(),
});
if (!approved) {
throw new RuntimeApprovalDeniedError();

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

@@ -1,14 +1,15 @@
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { eq, sql, tessInbox } from '@mosaicstack/db';
import { eq, sql, interactionInbox } from '@mosaicstack/db';
import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { afterAll, beforeAll, 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 = {
agentName: 'Nova',
sessionId: 'tess-pglite-session',
tenantId: 'tenant-pglite',
ownerId: 'tess-owner',
@@ -18,27 +19,34 @@ const IDENTITY: DurableSessionIdentity = {
describe('TessDurableSessionRepository', () => {
let dataDir: string | undefined;
let handle: DbHandle | undefined;
let handle: DbHandle;
let previousAuthSecret: string | undefined;
beforeEach((): void => {
beforeAll(async (): Promise<void> => {
previousAuthSecret = process.env['BETTER_AUTH_SECRET'];
process.env['BETTER_AUTH_SECRET'] = 'tess-durable-state-test-sealing-key';
dataDir = mkdtempSync(join(tmpdir(), 'tess-durable-state-'));
handle = createPgliteDb(dataDir);
await runPgliteMigrations(handle);
await seedOwner(handle);
}, 30_000);
beforeEach(async (): Promise<void> => {
await handle.db.execute(sql`DELETE FROM interaction_handoffs`);
await handle.db.execute(sql`DELETE FROM interaction_checkpoints`);
await handle.db.execute(sql`DELETE FROM interaction_inbox`);
await handle.db.execute(sql`DELETE FROM interaction_outbox`);
await handle.db.execute(sql`DELETE FROM interaction_sessions`);
});
afterEach(async (): Promise<void> => {
await handle?.close();
afterAll(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),
);
@@ -81,7 +89,7 @@ describe('TessDurableSessionRepository', () => {
});
await handle.close();
handle = createPgliteDb(dataDir);
handle = createPgliteDb(dataDir!);
const afterRestart = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
const recovered = await afterRestart.recover(IDENTITY.sessionId);
@@ -111,11 +119,6 @@ describe('TessDurableSessionRepository', () => {
}, 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({
@@ -142,9 +145,9 @@ describe('TessDurableSessionRepository', () => {
const snapshot = await coordinator.snapshot(IDENTITY.sessionId);
const [persisted] = await handle.db
.select({ content: tessInbox.content })
.from(tessInbox)
.where(eq(tessInbox.idempotencyKey, 'redacted-inbox'));
.select({ content: interactionInbox.content })
.from(interactionInbox)
.where(eq(interactionInbox.idempotencyKey, 'redacted-inbox'));
expect(JSON.stringify(snapshot)).not.toContain('super-secret-canary');
expect(JSON.stringify(snapshot)).not.toContain('operator@example.test');
@@ -153,11 +156,6 @@ describe('TessDurableSessionRepository', () => {
}, 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({
@@ -196,11 +194,6 @@ describe('TessDurableSessionRepository', () => {
}, 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) };
@@ -241,11 +234,6 @@ describe('TessDurableSessionRepository', () => {
}, 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) };
@@ -279,11 +267,6 @@ describe('TessDurableSessionRepository', () => {
}, 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) };

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

@@ -1,14 +1,15 @@
import { createHash } from 'node:crypto';
import { Inject, Injectable } from '@nestjs/common';
import {
and,
asc,
desc,
eq,
tessCheckpoints,
tessHandoffs,
tessInbox,
tessOutbox,
tessSessions,
interactionCheckpoints,
interactionHandoffs,
interactionInbox,
interactionOutbox,
interactionSessions,
type Db,
} from '@mosaicstack/db';
import { seal, unseal } from '@mosaicstack/auth';
@@ -37,9 +38,10 @@ export class TessDurableSessionRepository implements DurableSessionStore {
async create(identity: DurableSessionIdentity): Promise<void> {
await this.db
.insert(tessSessions)
.insert(interactionSessions)
.values({
id: identity.sessionId,
agentName: identity.agentName,
tenantId: identity.tenantId,
ownerId: identity.ownerId,
providerId: identity.providerId,
@@ -60,25 +62,28 @@ export class TessDurableSessionRepository implements DurableSessionStore {
const [inbox, outbox, checkpoints, handoffs] = await Promise.all([
this.db
.select()
.from(tessInbox)
.where(eq(tessInbox.sessionId, sessionId))
.orderBy(asc(tessInbox.createdAt)),
.from(interactionInbox)
.where(eq(interactionInbox.sessionId, sessionId))
.orderBy(asc(interactionInbox.createdAt)),
this.db
.select()
.from(tessOutbox)
.where(eq(tessOutbox.sessionId, sessionId))
.orderBy(asc(tessOutbox.createdAt)),
.from(interactionOutbox)
.where(eq(interactionOutbox.sessionId, sessionId))
.orderBy(asc(interactionOutbox.createdAt)),
this.db
.select()
.from(tessCheckpoints)
.where(eq(tessCheckpoints.sessionId, sessionId))
.orderBy(desc(tessCheckpoints.compactionEpoch), desc(tessCheckpoints.createdAt))
.from(interactionCheckpoints)
.where(eq(interactionCheckpoints.sessionId, sessionId))
.orderBy(
desc(interactionCheckpoints.compactionEpoch),
desc(interactionCheckpoints.createdAt),
)
.limit(1),
this.db
.select()
.from(tessHandoffs)
.where(eq(tessHandoffs.sessionId, sessionId))
.orderBy(asc(tessHandoffs.createdAt)),
.from(interactionHandoffs)
.where(eq(interactionHandoffs.sessionId, sessionId))
.orderBy(asc(interactionHandoffs.createdAt)),
]);
const checkpoint = checkpoints[0];
@@ -92,30 +97,36 @@ export class TessDurableSessionRepository implements DurableSessionStore {
}
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(tessInbox)
.values({ ...record, content: seal(record.content), status: 'pending' })
.insert(interactionInbox)
.values({
...record,
content: seal(record.content),
contentDigest: digest,
status: 'pending',
})
.onConflictDoNothing()
.returning({ status: tessInbox.status });
.returning({ status: interactionInbox.status });
if (inserted[0]) return { accepted: true, status: inserted[0].status };
const existing = await this.db
.select()
.from(tessInbox)
.from(interactionInbox)
.where(
and(
eq(tessInbox.sessionId, input.sessionId),
eq(tessInbox.idempotencyKey, input.idempotencyKey),
eq(interactionInbox.sessionId, input.sessionId),
eq(interactionInbox.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)) {
if (!sameInbox(entry, record) || existing[0].contentDigest !== digest) {
throw new Error(`Durable Tess inbox idempotency conflict: ${input.idempotencyKey}`);
}
return { accepted: false, status: entry.status };
@@ -125,16 +136,18 @@ export class TessDurableSessionRepository implements DurableSessionStore {
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))
.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(tessInbox)
.update(interactionInbox)
.set({ status: 'processing', updatedAt: new Date() })
.where(and(eq(tessInbox.id, entry.id), eq(tessInbox.status, 'pending')))
.where(and(eq(interactionInbox.id, entry.id), eq(interactionInbox.status, 'pending')))
.returning();
if (claimed[0]) return toInbox(claimed[0]);
}
@@ -143,26 +156,26 @@ export class TessDurableSessionRepository implements DurableSessionStore {
async completeInbox(sessionId: string, idempotencyKey: string): Promise<void> {
await this.db
.update(tessInbox)
.update(interactionInbox)
.set({ status: 'processed', updatedAt: new Date() })
.where(
and(
eq(tessInbox.sessionId, sessionId),
eq(tessInbox.idempotencyKey, idempotencyKey),
eq(tessInbox.status, 'processing'),
eq(interactionInbox.sessionId, sessionId),
eq(interactionInbox.idempotencyKey, idempotencyKey),
eq(interactionInbox.status, 'processing'),
),
);
}
async releaseInbox(sessionId: string, idempotencyKey: string): Promise<void> {
await this.db
.update(tessInbox)
.update(interactionInbox)
.set({ status: 'pending', updatedAt: new Date() })
.where(
and(
eq(tessInbox.sessionId, sessionId),
eq(tessInbox.idempotencyKey, idempotencyKey),
eq(tessInbox.status, 'processing'),
eq(interactionInbox.sessionId, sessionId),
eq(interactionInbox.idempotencyKey, idempotencyKey),
eq(interactionInbox.status, 'processing'),
),
);
}
@@ -170,31 +183,37 @@ export class TessDurableSessionRepository implements DurableSessionStore {
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(tessOutbox)
.values({ ...record, content: seal(record.content), status: 'pending' })
.insert(interactionOutbox)
.values({
...record,
content: seal(record.content),
contentDigest: digest,
status: 'pending',
})
.onConflictDoNothing()
.returning({ status: tessOutbox.status });
.returning({ status: interactionOutbox.status });
if (inserted[0]) return { accepted: true, status: inserted[0].status };
const existing = await this.db
.select()
.from(tessOutbox)
.from(interactionOutbox)
.where(
and(
eq(tessOutbox.sessionId, input.sessionId),
eq(tessOutbox.idempotencyKey, input.idempotencyKey),
eq(interactionOutbox.sessionId, input.sessionId),
eq(interactionOutbox.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)) {
if (!sameOutbox(entry, record) || existing[0].contentDigest !== digest) {
throw new Error(`Durable Tess outbox idempotency conflict: ${input.idempotencyKey}`);
}
return { accepted: false, status: entry.status };
@@ -204,16 +223,18 @@ export class TessDurableSessionRepository implements DurableSessionStore {
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))
.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(tessOutbox)
.update(interactionOutbox)
.set({ status: 'processing', updatedAt: new Date() })
.where(and(eq(tessOutbox.id, entry.id), eq(tessOutbox.status, 'pending')))
.where(and(eq(interactionOutbox.id, entry.id), eq(interactionOutbox.status, 'pending')))
.returning();
if (claimed[0]) return toOutbox(claimed[0]);
}
@@ -225,13 +246,13 @@ export class TessDurableSessionRepository implements DurableSessionStore {
idempotencyKey: string,
): Promise<DurableOutboxEntry | null> {
const claimed = await this.db
.update(tessOutbox)
.update(interactionOutbox)
.set({ status: 'processing', updatedAt: new Date() })
.where(
and(
eq(tessOutbox.sessionId, sessionId),
eq(tessOutbox.idempotencyKey, idempotencyKey),
eq(tessOutbox.status, 'pending'),
eq(interactionOutbox.sessionId, sessionId),
eq(interactionOutbox.idempotencyKey, idempotencyKey),
eq(interactionOutbox.status, 'pending'),
),
)
.returning();
@@ -240,26 +261,26 @@ export class TessDurableSessionRepository implements DurableSessionStore {
async completeOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
await this.db
.update(tessOutbox)
.update(interactionOutbox)
.set({ status: 'delivered', updatedAt: new Date() })
.where(
and(
eq(tessOutbox.sessionId, sessionId),
eq(tessOutbox.idempotencyKey, idempotencyKey),
eq(tessOutbox.status, 'processing'),
eq(interactionOutbox.sessionId, sessionId),
eq(interactionOutbox.idempotencyKey, idempotencyKey),
eq(interactionOutbox.status, 'processing'),
),
);
}
async releaseOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
await this.db
.update(tessOutbox)
.update(interactionOutbox)
.set({ status: 'pending', updatedAt: new Date() })
.where(
and(
eq(tessOutbox.sessionId, sessionId),
eq(tessOutbox.idempotencyKey, idempotencyKey),
eq(tessOutbox.status, 'processing'),
eq(interactionOutbox.sessionId, sessionId),
eq(interactionOutbox.idempotencyKey, idempotencyKey),
eq(interactionOutbox.status, 'processing'),
),
);
}
@@ -271,14 +292,14 @@ export class TessDurableSessionRepository implements DurableSessionStore {
summary: redactSensitiveContent(input.summary).content,
};
const inserted = await this.db
.insert(tessCheckpoints)
.insert(interactionCheckpoints)
.values({
...checkpoint,
cursor: seal(checkpoint.cursor),
summary: seal(checkpoint.summary),
})
.onConflictDoNothing()
.returning({ checkpointId: tessCheckpoints.checkpointId });
.returning({ checkpointId: interactionCheckpoints.checkpointId });
if (inserted[0]) return;
const existing = await this.findCheckpoint(input.sessionId, input.checkpointId);
@@ -290,11 +311,11 @@ export class TessDurableSessionRepository implements DurableSessionStore {
async findCheckpoint(sessionId: string, checkpointId: string): Promise<DurableCheckpoint | null> {
const checkpoints = await this.db
.select()
.from(tessCheckpoints)
.from(interactionCheckpoints)
.where(
and(
eq(tessCheckpoints.sessionId, sessionId),
eq(tessCheckpoints.checkpointId, checkpointId),
eq(interactionCheckpoints.sessionId, sessionId),
eq(interactionCheckpoints.checkpointId, checkpointId),
),
)
.limit(1);
@@ -308,10 +329,10 @@ export class TessDurableSessionRepository implements DurableSessionStore {
throw new Error(`Durable Tess handoff checkpoint is unavailable: ${input.checkpointId}`);
}
const inserted = await this.db
.insert(tessHandoffs)
.insert(interactionHandoffs)
.values({ ...input })
.onConflictDoNothing()
.returning({ handoffId: tessHandoffs.handoffId });
.returning({ handoffId: interactionHandoffs.handoffId });
if (inserted[0]) return;
const existing = await this.findHandoff(input.handoffId);
@@ -323,8 +344,8 @@ export class TessDurableSessionRepository implements DurableSessionStore {
async findHandoff(handoffId: string): Promise<DurableHandoff | null> {
const handoffs = await this.db
.select()
.from(tessHandoffs)
.where(eq(tessHandoffs.handoffId, handoffId))
.from(interactionHandoffs)
.where(eq(interactionHandoffs.handoffId, handoffId))
.limit(1);
const handoff = handoffs[0];
return handoff ? toHandoff(handoff) : null;
@@ -335,20 +356,23 @@ export class TessDurableSessionRepository implements DurableSessionStore {
// reached an external target before a crash, so it is deliberately not
// replayed by generic recovery.
await this.db
.update(tessInbox)
.update(interactionInbox)
.set({ status: 'pending', updatedAt: new Date() })
.where(and(eq(tessInbox.sessionId, sessionId), eq(tessInbox.status, 'processing')));
.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(tessSessions)
.where(eq(tessSessions.id, sessionId))
.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,
@@ -359,8 +383,13 @@ export class TessDurableSessionRepository implements DurableSessionStore {
}
}
function contentDigest(content: string): string {
return createHash('sha256').update(content).digest('hex');
}
function sameIdentity(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
return (
left.agentName === right.agentName &&
left.sessionId === right.sessionId &&
left.tenantId === right.tenantId &&
left.ownerId === right.ownerId &&
@@ -410,7 +439,7 @@ function sameHandoff(left: DurableHandoff, right: DurableHandoffInput): boolean
);
}
function toInbox(row: typeof tessInbox.$inferSelect): DurableInboxEntry {
function toInbox(row: typeof interactionInbox.$inferSelect): DurableInboxEntry {
return {
sessionId: row.sessionId,
idempotencyKey: row.idempotencyKey,
@@ -420,7 +449,7 @@ function toInbox(row: typeof tessInbox.$inferSelect): DurableInboxEntry {
};
}
function toOutbox(row: typeof tessOutbox.$inferSelect): DurableOutboxEntry {
function toOutbox(row: typeof interactionOutbox.$inferSelect): DurableOutboxEntry {
return {
sessionId: row.sessionId,
idempotencyKey: row.idempotencyKey,
@@ -432,7 +461,7 @@ function toOutbox(row: typeof tessOutbox.$inferSelect): DurableOutboxEntry {
};
}
function toCheckpoint(row: typeof tessCheckpoints.$inferSelect): DurableCheckpoint {
function toCheckpoint(row: typeof interactionCheckpoints.$inferSelect): DurableCheckpoint {
return {
sessionId: row.sessionId,
checkpointId: row.checkpointId,
@@ -442,7 +471,7 @@ function toCheckpoint(row: typeof tessCheckpoints.$inferSelect): DurableCheckpoi
};
}
function toHandoff(row: typeof tessHandoffs.$inferSelect): DurableHandoff {
function toHandoff(row: typeof interactionHandoffs.$inferSelect): DurableHandoff {
return {
sessionId: row.sessionId,
handoffId: row.handoffId,

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

@@ -70,11 +70,12 @@ describe('CommandAuthorizationService', () => {
tenantId: 'tenant-1',
channelId: 'discord:operator',
correlationId: 'correlation-malformed-expiry',
agentName: 'Nova',
};
const service = createService('admin', entries);
const approval = await service.createRuntimeTerminationApproval(action);
expect(approval).not.toBeNull();
const key = `tess:command-approval:${approval!.approvalId}`;
const key = `agent:Nova:command-approval:${approval!.approvalId}`;
const stored = entries.get(key);
expect(stored).toBeDefined();
entries.set(key, JSON.stringify({ ...JSON.parse(stored!), expiresAt: 'not-a-date' }));
@@ -93,6 +94,7 @@ describe('CommandAuthorizationService', () => {
tenantId: 'tenant-1',
channelId: 'discord:operator',
correlationId: 'correlation-1',
agentName: 'Nova',
};
const beforeRestart = createService('admin', entries);
const approval = await beforeRestart.createRuntimeTerminationApproval(action);

24
apps/gateway/src/commands/command-authorization.service.ts
View File

@@ -23,6 +23,8 @@ export interface RuntimeTerminationApprovalAction {
tenantId: string;
channelId: string;
correlationId: string;
/** Provisioned roster identity; isolates approvals between interaction agents. */
agentName: string;
}
export interface RuntimeTerminationApproval extends RuntimeTerminationApprovalAction {
@@ -92,7 +94,7 @@ export class CommandAuthorizationService {
}
/**
* Uses the same `tess:command-approval:*` store and one-time deletion rule as
* Uses the same `interaction:command-approval:*` store and one-time deletion rule as
* command approvals. This deliberately avoids a parallel approval database.
*/
async createRuntimeTerminationApproval(
@@ -108,7 +110,12 @@ export class CommandAuthorizationService {
...action,
expiresAt: new Date(Date.now() + 5 * 60_000).toISOString(),
};
await this.redis.set(this.key(approval.approvalId), JSON.stringify(approval), 'EX', '300');
await this.redis.set(
this.runtimeKey(action.agentName, approval.approvalId),
JSON.stringify(approval),
'EX',
'300',
);
return approval;
}
@@ -116,7 +123,7 @@ export class CommandAuthorizationService {
approvalId: string,
action: RuntimeTerminationApprovalAction,
): Promise<boolean> {
const encoded = await this.redis.get(this.key(approvalId));
const encoded = await this.redis.get(this.runtimeKey(action.agentName, approvalId));
if (!encoded) return false;
let approval: unknown;
try {
@@ -134,7 +141,7 @@ export class CommandAuthorizationService {
return false;
}
if ((await this.resolveRole(approval.actorId)) !== 'admin') return false;
return (await this.redis.del(this.key(approvalId))) === 1;
return (await this.redis.del(this.runtimeKey(action.agentName, approvalId))) === 1;
}
private async resolveRole(actorId: string): Promise<CommandRole | null> {
@@ -196,6 +203,7 @@ export class CommandAuthorizationService {
action.tenantId,
action.channelId,
action.correlationId,
action.agentName,
].every((value: string): boolean => value.trim().length > 0);
}
@@ -209,6 +217,7 @@ export class CommandAuthorizationService {
tenantId: action.tenantId,
channelId: action.channelId,
correlationId: action.correlationId,
agentName: action.agentName,
}),
)
.digest('hex');
@@ -244,11 +253,16 @@ export class CommandAuthorizationService {
'sessionId' in value &&
'channelId' in value &&
'correlationId' in value &&
'agentName' in value &&
'expiresAt' in value
);
}
private key(approvalId: string): string {
return `tess:command-approval:${approvalId}`;
return `interaction:command-approval:${approvalId}`;
}
private runtimeKey(agentName: string, approvalId: string): string {
return `agent:${encodeURIComponent(agentName)}:command-approval:${approvalId}`;
}
}

2
apps/gateway/src/commands/runtime-approval-verifier.ts
View File

@@ -7,7 +7,7 @@ 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
* Redis-backed `interaction:command-approval:*` store. It has no separate approval
* persistence or replay semantics.
*/
@Injectable()

2
apps/gateway/src/gc/session-gc.service.spec.ts
View File

@@ -108,7 +108,7 @@ describe('SessionGCService', () => {
} as never;
const payload = { command: 'gc', conversationId: 'owned' };
const approval = await authorization.createApproval(command, payload, 'admin-1');
const approvalKey = `tess:command-approval:${approval!.approvalId}`;
const approvalKey = `interaction:command-approval:${approval!.approvalId}`;
const gc = new SessionGCService(redis as never, mockLogService as unknown as LogService);
await gc.collect('owned');