refactor(#747): de-hardcode interaction coordination names
This commit is contained in:
Hermes Agent
416
apps/gateway/src/agent/durable-session.repository.test.ts
Normal file
416
apps/gateway/src/agent/durable-session.repository.test.ts
Normal file
@@ -0,0 +1,416 @@
|
||||
import { mkdtempSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { eq, sql, interactionCheckpoints, interactionInbox } from '@mosaicstack/db';
|
||||
import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { createPgliteDb, runPgliteMigrations, type DbHandle } from '@mosaicstack/db';
|
||||
import { DurableSessionRepository } from './durable-session.repository.js';
|
||||
import { DurableSessionService } from './durable-session.service.js';
|
||||
|
||||
const IDENTITY: DurableSessionIdentity = {
|
||||
agentName: 'Nova',
|
||||
sessionId: 'tess-pglite-session',
|
||||
tenantId: 'tenant-pglite',
|
||||
ownerId: 'tess-owner',
|
||||
providerId: 'fleet',
|
||||
runtimeSessionId: 'nova',
|
||||
};
|
||||
|
||||
describe('DurableSessionRepository', () => {
|
||||
let dataDir: string | undefined;
|
||||
let handle: DbHandle;
|
||||
let previousAuthSecret: string | undefined;
|
||||
|
||||
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`);
|
||||
});
|
||||
|
||||
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 () => {
|
||||
const beforeRestart = new DurableSessionCoordinator(new DurableSessionRepository(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 DurableSessionRepository(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 () => {
|
||||
const coordinator = new DurableSessionCoordinator(new DurableSessionRepository(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: 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');
|
||||
expect(persisted?.content).not.toContain('super-secret-canary');
|
||||
expect(persisted?.content).not.toContain('[REDACTED]');
|
||||
}, 30_000);
|
||||
|
||||
it('fails closed when the configured idempotency secret is unavailable', async () => {
|
||||
const coordinator = new DurableSessionCoordinator(new DurableSessionRepository(handle.db));
|
||||
await coordinator.create(IDENTITY);
|
||||
const secret = process.env['BETTER_AUTH_SECRET'];
|
||||
delete process.env['BETTER_AUTH_SECRET'];
|
||||
try {
|
||||
await expect(
|
||||
coordinator.receive({
|
||||
sessionId: IDENTITY.sessionId,
|
||||
idempotencyKey: 'requires-idempotency-secret',
|
||||
correlationId: 'correlation-secret',
|
||||
content: 'sensitive payload',
|
||||
}),
|
||||
).rejects.toThrow(/required for durable idempotency digests/);
|
||||
} finally {
|
||||
if (secret === undefined) delete process.env['BETTER_AUTH_SECRET'];
|
||||
else process.env['BETTER_AUTH_SECRET'] = secret;
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it('uses keyed pre-redaction digests to reject distinct sensitive checkpoint payloads', async () => {
|
||||
const coordinator = new DurableSessionCoordinator(new DurableSessionRepository(handle.db));
|
||||
await coordinator.create(IDENTITY);
|
||||
const input = {
|
||||
sessionId: IDENTITY.sessionId,
|
||||
checkpointId: 'checkpoint-secret-conflict',
|
||||
cursor: 'api_key=secret-one',
|
||||
summary: 'bearer secret-one',
|
||||
compactionEpoch: 1,
|
||||
};
|
||||
await coordinator.checkpoint(input);
|
||||
|
||||
await expect(
|
||||
coordinator.checkpoint({
|
||||
...input,
|
||||
cursor: 'api_key=secret-two',
|
||||
summary: 'bearer secret-two',
|
||||
}),
|
||||
).rejects.toThrow(/checkpoint identity conflict/);
|
||||
|
||||
const [persisted] = await handle.db
|
||||
.select({
|
||||
digest: interactionCheckpoints.contentDigest,
|
||||
cursor: interactionCheckpoints.cursor,
|
||||
})
|
||||
.from(interactionCheckpoints)
|
||||
.where(eq(interactionCheckpoints.checkpointId, input.checkpointId));
|
||||
expect(persisted?.cursor).not.toContain('secret-one');
|
||||
expect(persisted?.digest).not.toBe(
|
||||
createHash('sha256')
|
||||
.update(JSON.stringify([input.cursor, input.summary]))
|
||||
.digest('hex'),
|
||||
);
|
||||
|
||||
await coordinator.checkpoint({
|
||||
...input,
|
||||
checkpointId: 'checkpoint-delimiter-conflict',
|
||||
cursor: 'a\u0000b',
|
||||
summary: 'c',
|
||||
});
|
||||
await expect(
|
||||
coordinator.checkpoint({
|
||||
...input,
|
||||
checkpointId: 'checkpoint-delimiter-conflict',
|
||||
cursor: 'a',
|
||||
summary: 'b\u0000c',
|
||||
}),
|
||||
).rejects.toThrow(/checkpoint identity conflict/);
|
||||
}, 30_000);
|
||||
|
||||
it('rejects distinct sensitive inbox and outbox payloads under reused idempotency keys', async () => {
|
||||
const coordinator = new DurableSessionCoordinator(new DurableSessionRepository(handle.db));
|
||||
await coordinator.create(IDENTITY);
|
||||
const inbox = {
|
||||
sessionId: IDENTITY.sessionId,
|
||||
idempotencyKey: 'inbox-secret-conflict',
|
||||
correlationId: 'correlation-inbox-secret',
|
||||
content: 'api_key=secret-one',
|
||||
};
|
||||
const outbox = {
|
||||
sessionId: IDENTITY.sessionId,
|
||||
idempotencyKey: 'outbox-secret-conflict',
|
||||
correlationId: 'correlation-outbox-secret',
|
||||
channelId: 'cli',
|
||||
kind: 'provider.send',
|
||||
content: 'api_key=secret-one',
|
||||
};
|
||||
await coordinator.receive(inbox);
|
||||
await coordinator.enqueueOutbox(outbox);
|
||||
|
||||
await expect(coordinator.receive({ ...inbox, content: 'api_key=secret-two' })).rejects.toThrow(
|
||||
/idempotency conflict/,
|
||||
);
|
||||
await expect(
|
||||
coordinator.enqueueOutbox({ ...outbox, content: 'api_key=secret-two' }),
|
||||
).rejects.toThrow(/idempotency conflict/);
|
||||
}, 30_000);
|
||||
|
||||
it('rejects database inbox and outbox idempotency-key conflicts', async () => {
|
||||
const coordinator = new DurableSessionCoordinator(new DurableSessionRepository(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 () => {
|
||||
const repository = new DurableSessionRepository(handle.db);
|
||||
const coordinator = new DurableSessionCoordinator(repository);
|
||||
const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new DurableSessionService(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 () => {
|
||||
const repository = new DurableSessionRepository(handle.db);
|
||||
const coordinator = new DurableSessionCoordinator(repository);
|
||||
const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new DurableSessionService(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 () => {
|
||||
const repository = new DurableSessionRepository(handle.db);
|
||||
const coordinator = new DurableSessionCoordinator(repository);
|
||||
const runtimeProviders = { sendMessage: vi.fn().mockResolvedValue(undefined) };
|
||||
const service = new DurableSessionService(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())
|
||||
`);
|
||||
}
|
||||
Reference in New Issue
Block a user