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