feat(tess): wire durable interaction surfaces
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful

This commit is contained in:
Jarvis
2026-07-13 04:28:27 -05:00
parent 84d884b932
commit 451f7e04ec
16 changed files with 782 additions and 58 deletions

View File

@@ -64,6 +64,23 @@ describe('DurableSessionCoordinator', () => {
expect(recovered.handoffs).toMatchObject([{ handoffId: 'handoff-1', status: 'pending' }]);
});
it('rebinds a recovered runtime while preserving the immutable conversation owner scope', async () => {
const coordinator = new DurableSessionCoordinator(new InMemoryDurableSessionStore());
await coordinator.create(IDENTITY);
await coordinator.create({
...IDENTITY,
providerId: 'fleet-next',
runtimeSessionId: 'nova-next',
});
await expect(coordinator.snapshot(IDENTITY.sessionId)).resolves.toMatchObject({
identity: { ...IDENTITY, providerId: 'fleet-next', runtimeSessionId: 'nova-next' },
});
await expect(coordinator.create({ ...IDENTITY, ownerId: 'other-owner' })).rejects.toThrow(
/identity conflict/,
);
});
it('deduplicates duplicate ingress and never reprocesses an inbox record after restart or compaction', async () => {
const store = new InMemoryDurableSessionStore();
const firstProcess = new DurableSessionCoordinator(store);

View File

@@ -256,9 +256,11 @@ export class InMemoryDurableSessionStore implements DurableSessionStore {
async create(identity: DurableSessionIdentity): Promise<void> {
const existing = this.sessions.get(identity.sessionId);
if (existing) {
if (!identitiesEqual(existing.identity, identity)) {
if (!sameEnrollmentScope(existing.identity, identity)) {
throw new Error(`Durable Tess session identity conflict: ${identity.sessionId}`);
}
existing.identity.providerId = identity.providerId;
existing.identity.runtimeSessionId = identity.runtimeSessionId;
return;
}
this.sessions.set(identity.sessionId, {
@@ -416,14 +418,12 @@ export class InMemoryDurableSessionStore implements DurableSessionStore {
}
}
function identitiesEqual(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
return (
left.agentName === right.agentName &&
left.sessionId === right.sessionId &&
left.tenantId === right.tenantId &&
left.ownerId === right.ownerId &&
left.providerId === right.providerId &&
left.runtimeSessionId === right.runtimeSessionId
left.ownerId === right.ownerId
);
}

View File

@@ -10,6 +10,7 @@ describe('generic interaction CLI', (): void => {
expect(command.commands.map((item) => item.name()).sort()).toEqual([
'attach',
'chat',
'enroll',
'health',
'recover',
'send',

View File

@@ -3,6 +3,7 @@ import type { Command } from 'commander';
import { withAuth } from './with-auth.js';
import {
attachInteractionSession,
enrollInteractionSession,
fetchInteractionHealth,
fetchInteractionSessions,
fetchInteractionStatus,
@@ -10,6 +11,7 @@ import {
recoverInteractionSession,
sendInteractionMessage,
stopInteractionSession,
streamInteractionSession,
} from '../tui/gateway-api.js';
interface InteractionOptions {
@@ -103,18 +105,48 @@ export function registerInteractionCommand(program: Command): Command {
});
options(
command.command('attach <sessionId>').description('Create a scoped read or write attachment'),
command
.command('enroll <sessionId> <provider> <runtimeSessionId>')
.description('Bind an authorized runtime session to a durable conversation handle'),
).action(
async (
sessionId: string,
providerId: string,
runtimeSessionId: string,
opts: InteractionOptions,
) => {
const request = await authRequest(opts);
print(
await enrollInteractionSession(request.gateway, request.cookie, {
...request,
sessionId,
providerId,
runtimeSessionId,
}),
);
},
);
options(
command
.command('attach <sessionId>')
.description('Create a scoped read or write attachment and stream the session'),
)
.option('--control', 'Request control mode (provider policy may deny it)')
.action(async (sessionId: string, opts: InteractionOptions & { control?: boolean }) => {
const request = await authRequest(opts);
print(
await attachInteractionSession(request.gateway, request.cookie, {
...request,
sessionId,
mode: opts.control ? 'control' : 'read',
}),
);
const attachment = await attachInteractionSession(request.gateway, request.cookie, {
...request,
sessionId,
mode: opts.control ? 'control' : 'read',
});
print(attachment);
for await (const event of streamInteractionSession(request.gateway, request.cookie, {
...request,
sessionId,
})) {
print(event);
}
});
const send = options(

View File

@@ -406,6 +406,32 @@ export async function fetchInteractionTree(
return handleResponse<unknown[]>(res, 'Failed to get interaction session tree');
}
export async function enrollInteractionSession(
gatewayUrl: string,
sessionCookie: string,
request: InteractionRequest & {
sessionId: string;
providerId: string;
runtimeSessionId: string;
},
): Promise<{ status: string; sessionId: string }> {
const res = await fetch(
`${gatewayUrl}${interactionPath(request.agentName, `/sessions/${encodeURIComponent(request.sessionId)}/enroll`)}`,
{
method: 'POST',
headers: interactionHeaders(sessionCookie, gatewayUrl, request.correlationId),
body: JSON.stringify({
providerId: request.providerId,
runtimeSessionId: request.runtimeSessionId,
}),
},
);
return handleResponse<{ status: string; sessionId: string }>(
res,
'Failed to enroll interaction session',
);
}
export async function attachInteractionSession(
gatewayUrl: string,
sessionCookie: string,
@@ -422,6 +448,49 @@ export async function attachInteractionSession(
return handleResponse<unknown>(res, 'Failed to attach interaction session');
}
export async function* streamInteractionSession(
gatewayUrl: string,
sessionCookie: string,
request: InteractionRequest & { sessionId: string; cursor?: string },
): AsyncIterable<unknown> {
const params = request.cursor?.trim()
? `?${new URLSearchParams({ cursor: request.cursor.trim() }).toString()}`
: '';
const res = await fetch(
`${gatewayUrl}${interactionPath(request.agentName, `/sessions/${encodeURIComponent(request.sessionId)}/stream${params}`)}`,
{ headers: interactionHeaders(sessionCookie, gatewayUrl, request.correlationId) },
);
if (!res.ok || !res.body) {
const body = await res.text().catch(() => '');
throw new Error(`Failed to stream interaction session (${res.status}): ${body}`);
}
const reader = res.body.getReader();
const decoder = new TextDecoder();
let pending = '';
try {
for (;;) {
const chunk = await reader.read();
if (chunk.done) break;
pending += decoder.decode(chunk.value, { stream: true });
for (;;) {
const separator = pending.indexOf('\n\n');
if (separator < 0) break;
const frame = pending.slice(0, separator);
pending = pending.slice(separator + 2);
const data = frame
.split('\n')
.find((line: string): boolean => line.startsWith('data:'))
?.slice('data:'.length)
.trim();
if (data) yield JSON.parse(data) as unknown;
}
}
} finally {
reader.releaseLock();
}
}
export async function sendInteractionMessage(
gatewayUrl: string,
sessionCookie: string,