417 lines
16 KiB
TypeScript
417 lines
16 KiB
TypeScript
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())
|
|
`);
|
|
}
|