feat(tess): wire durable interaction surfaces #732

Merged
jason.woltje merged 1 commits from feat/tess-m3-integration into main 2026-07-13 10:05:30 +00:00
16 changed files with 782 additions and 58 deletions

View File

@@ -0,0 +1,159 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { InMemoryDurableSessionStore } from '@mosaicstack/agent';
import {
createDiscordIngressEnvelope,
type DiscordIngressPayload,
} from '@mosaicstack/discord-plugin';
import { InteractionController } from '../../agent/interaction.controller.js';
import { RuntimeProviderService } from '../../agent/runtime-provider-registry.service.js';
import { TessDurableSessionService } from '../../agent/tess-durable-session.service.js';
import { ChatGateway } from '../../chat/chat.gateway.js';
import { CommandAuthorizationService } from '../../commands/command-authorization.service.js';
const SERVICE_TOKEN = 'test-discord-service-token';
const envKeys = [
'DISCORD_SERVICE_TOKEN',
'DISCORD_SERVICE_TENANT_ID',
'DISCORD_INTERACTION_BINDINGS',
'DISCORD_ALLOWED_GUILD_IDS',
'DISCORD_ALLOWED_CHANNEL_IDS',
'DISCORD_ALLOWED_USER_IDS',
'MOSAIC_AGENT_NAME',
] as const;
const priorEnv = new Map<string, string | undefined>();
function payload(content: string, messageId: string, correlationId: string): DiscordIngressPayload {
return {
content,
messageId,
correlationId,
guildId: 'guild-1',
channelId: 'channel-1',
userId: 'discord-admin-1',
conversationId: 'conversation-1',
};
}
function authorization(): CommandAuthorizationService {
const entries = new Map<string, string>();
return new CommandAuthorizationService(
{
select: () => ({
from: () => ({ where: () => ({ limit: async () => [{ role: 'admin' }] }) }),
}),
} as never,
{
get: async (key: string) => entries.get(key) ?? null,
set: async (key: string, value: string) => entries.set(key, value),
del: async (key: string) => Number(entries.delete(key)),
},
);
}
describe('Tess Discord/CLI durable-session integration', () => {
afterEach(() => {
for (const key of envKeys) {
const value = priorEnv.get(key);
if (value === undefined) delete process.env[key];
else process.env[key] = value;
}
priorEnv.clear();
});
it('enrolls through the CLI surface then resolves the same durable session from Discord', async () => {
for (const key of envKeys) priorEnv.set(key, process.env[key]);
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
process.env['DISCORD_SERVICE_TOKEN'] = SERVICE_TOKEN;
process.env['DISCORD_SERVICE_TENANT_ID'] = 'tenant-1';
process.env['DISCORD_ALLOWED_GUILD_IDS'] = 'guild-1';
process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-1';
process.env['DISCORD_ALLOWED_USER_IDS'] = 'discord-admin-1';
process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([
{
instanceId: 'Nova',
guildId: 'guild-1',
channelId: 'channel-1',
pairedUsers: {
'discord-admin-1': { role: 'admin', mosaicUserId: 'mosaic-admin-1' },
},
},
]);
const durable = new TessDurableSessionService(
new InMemoryDurableSessionStore() as never,
{} as never,
);
const enrollmentRuntime = {
listSessions: vi.fn().mockResolvedValue([{ id: 'runtime-1' }]),
};
const controller = new InteractionController(enrollmentRuntime as never, durable);
await controller.enroll(
'Nova',
'conversation-1',
{ providerId: 'fleet', runtimeSessionId: 'runtime-1' },
{ id: 'mosaic-admin-1', tenantId: 'tenant-1' },
'cli-enrollment-correlation',
);
const authz = authorization();
const terminated = vi.fn().mockResolvedValue(undefined);
const runtime = new RuntimeProviderService(
{
require: () => ({
capabilities: async () => ({ supported: ['session.terminate'] }),
terminate: terminated,
}),
} as never,
{ record: async () => undefined } as never,
{
consume: (approvalId, action) =>
authz.consumeRuntimeTerminationApproval(approvalId, action),
},
);
const gateway = new ChatGateway(
{} as never,
{} as never,
{} as never,
{} as never,
{} as never,
{} as never,
authz,
runtime,
durable,
);
const client = { data: { discordService: true }, emit: vi.fn() };
await gateway.handleDiscordApproval(
client as never,
createDiscordIngressEnvelope(
payload('/approve', 'approve-1', 'discord-approve-correlation'),
SERVICE_TOKEN,
),
);
const approval = client.emit.mock.calls.find(
([event]) => event === 'discord:approval',
)?.[1] as {
approvalId: string;
success: boolean;
};
expect(approval.success).toBe(true);
await gateway.handleDiscordStop(
client as never,
createDiscordIngressEnvelope(
payload(`/stop ${approval.approvalId}`, 'stop-1', 'discord-stop-correlation'),
SERVICE_TOKEN,
),
);
expect(terminated).toHaveBeenCalledWith(
'runtime-1',
approval.approvalId,
expect.objectContaining({ actorId: 'mosaic-admin-1' }),
);
expect(client.emit).toHaveBeenCalledWith('discord:stop', {
correlationId: 'discord-stop-correlation',
success: true,
});
});
});

View File

@@ -37,6 +37,7 @@ class RecordingRuntimeProvider implements AgentRuntimeProvider {
readonly sentMessages: RuntimeMessage[] = []; readonly sentMessages: RuntimeMessage[] = [];
terminateCalls = 0; terminateCalls = 0;
throwAfterSend = false; throwAfterSend = false;
throwAuthorization = false;
constructor(private readonly supported: RuntimeCapability[]) {} constructor(private readonly supported: RuntimeCapability[]) {}
@@ -76,6 +77,9 @@ class RecordingRuntimeProvider implements AgentRuntimeProvider {
): Promise<void> { ): Promise<void> {
this.receivedScopes.push(scope); this.receivedScopes.push(scope);
this.sentMessages.push(message); this.sentMessages.push(message);
if (this.throwAuthorization) {
throw Object.assign(new Error('provider authorization denied'), { code: 'forbidden' });
}
if (this.throwAfterSend) { if (this.throwAfterSend) {
throw new Error('provider acknowledgement failed'); throw new Error('provider acknowledgement failed');
} }
@@ -295,6 +299,23 @@ describe('RuntimeProviderService security boundary', (): void => {
}); });
}); });
it('records a provider authorization rejection as denied rather than provider failure', async (): Promise<void> => {
const provider = new RecordingRuntimeProvider(['session.send']);
provider.throwAuthorization = true;
const audit = new RecordingAuditSink();
const service = makeService(provider, audit);
await expect(
service.sendMessage(
'fleet',
'session-1',
{ content: 'hello', idempotencyKey: 'key-1' },
CONTEXT,
),
).rejects.toThrow(/provider authorization denied/);
expect(audit.events.at(-1)).toMatchObject({ outcome: 'denied', errorCode: 'policy_denied' });
});
it('persists only metadata-only runtime audit fields', async (): Promise<void> => { it('persists only metadata-only runtime audit fields', async (): Promise<void> => {
let persisted: unknown; let persisted: unknown;
const ingest = async (entry: unknown): Promise<unknown> => { const ingest = async (entry: unknown): Promise<unknown> => {

View File

@@ -1,9 +1,27 @@
import { firstValueFrom } from 'rxjs';
import { afterEach, describe, expect, it, vi } from 'vitest'; import { afterEach, describe, expect, it, vi } from 'vitest';
import { RuntimeApprovalDeniedError } from './runtime-provider-registry.service.js';
import { RuntimeApprovalDeniedFilter } from './runtime-approval-denied.filter.js';
import { InteractionController } from './interaction.controller.js'; import { InteractionController } from './interaction.controller.js';
describe('InteractionController', (): void => { describe('InteractionController', (): void => {
afterEach(() => vi.restoreAllMocks()); afterEach(() => vi.restoreAllMocks());
it('maps a denied runtime approval to Fastify HTTP 403', () => {
const send = vi.fn();
const status = vi.fn().mockReturnValue({ send });
const response = { status };
const host = { switchToHttp: () => ({ getResponse: () => response }) };
new RuntimeApprovalDeniedFilter().catch(new RuntimeApprovalDeniedError(), host as never);
expect(status).toHaveBeenCalledWith(403);
expect(send).toHaveBeenCalledWith({
statusCode: 403,
message: 'Runtime termination approval denied',
});
});
it('honors a differently named configured instance without a code change', async () => { it('honors a differently named configured instance without a code change', async () => {
const prior = process.env['MOSAIC_AGENT_NAME']; const prior = process.env['MOSAIC_AGENT_NAME'];
process.env['MOSAIC_AGENT_NAME'] = 'Nova'; process.env['MOSAIC_AGENT_NAME'] = 'Nova';
@@ -30,6 +48,36 @@ describe('InteractionController', (): void => {
); );
}); });
it('enrolls a visible runtime session under the cross-surface conversation handle', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const runtime = {
listSessions: vi.fn().mockResolvedValue([{ id: 'runtime-1' }]),
};
const durable = { enroll: vi.fn().mockResolvedValue(undefined) };
const controller = new InteractionController(runtime as never, durable as never);
await expect(
controller.enroll(
'Nova',
'conversation-1',
{ providerId: 'fleet', runtimeSessionId: 'runtime-1' },
{ id: 'owner', tenantId: 'team' },
'corr-1',
),
).resolves.toEqual({ status: 'enrolled', sessionId: 'conversation-1' });
expect(durable.enroll).toHaveBeenCalledWith(
{
agentName: 'Nova',
sessionId: 'conversation-1',
tenantId: 'team',
ownerId: 'owner',
providerId: 'fleet',
runtimeSessionId: 'runtime-1',
},
expect.objectContaining({ correlationId: 'corr-1' }),
);
});
it('rejects an invalid attach mode before invoking a provider', async () => { it('rejects an invalid attach mode before invoking a provider', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova'; process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const controller = new InteractionController({ attach: vi.fn() } as never, {} as never); const controller = new InteractionController({ attach: vi.fn() } as never, {} as never);
@@ -39,6 +87,121 @@ describe('InteractionController', (): void => {
).rejects.toThrow('Interaction attach mode is invalid'); ).rejects.toThrow('Interaction attach mode is invalid');
}); });
it('resumes a durable session by attaching and streaming its runtime events', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const runtimeEvent = {
type: 'message.delta' as const,
sessionId: 'runtime-1',
cursor: 'cursor-1',
occurredAt: '2026-07-13T00:00:00.000Z',
content: 'resumed',
};
const runtime = {
attach: vi.fn().mockResolvedValue({ attachmentId: 'attach-1', sessionId: 'runtime-1' }),
streamSession: vi.fn(async function* () {
yield runtimeEvent;
}),
};
const durable = {
getSnapshot: vi.fn().mockResolvedValue({
identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
}),
};
const controller = new InteractionController(runtime as never, durable as never);
await controller.attach('Nova', 'conversation-1', { mode: 'read' }, { id: 'owner' }, 'corr-1');
await expect(
firstValueFrom(
controller.stream('Nova', 'conversation-1', undefined, { id: 'owner' }, 'corr-1'),
),
).resolves.toEqual({ data: runtimeEvent });
expect(runtime.attach).toHaveBeenCalledWith(
'fleet',
'runtime-1',
'read',
expect.objectContaining({ correlationId: 'corr-1' }),
);
expect(runtime.streamSession).toHaveBeenCalledWith(
'fleet',
'runtime-1',
undefined,
expect.objectContaining({ correlationId: 'corr-1' }),
);
});
it('does not create a runtime stream after the SSE subscriber disconnects during snapshot lookup', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
let resolveSnapshot!: (value: { identity: Record<string, string> }) => void;
const snapshot = new Promise<{ identity: Record<string, string> }>((resolve) => {
resolveSnapshot = resolve;
});
const runtime = { streamSession: vi.fn() };
const durable = { getSnapshot: vi.fn().mockReturnValue(snapshot) };
const controller = new InteractionController(runtime as never, durable as never);
const subscription = controller
.stream('Nova', 'conversation-1', undefined, { id: 'owner' }, 'corr-1')
.subscribe();
subscription.unsubscribe();
resolveSnapshot({
identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
});
await new Promise((resolve) => setTimeout(resolve, 0));
expect(runtime.streamSession).not.toHaveBeenCalled();
});
it.each([
['wrong actor', { getSnapshot: vi.fn().mockRejectedValue(new Error('scope mismatch')) }],
[
'session-agent mismatch',
{
getSnapshot: vi.fn().mockResolvedValue({
identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
}),
},
],
])('denies a CLI stop for %s', async (_reason, durable) => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const runtime = { terminate: vi.fn().mockResolvedValue(undefined) };
const controller = new InteractionController(runtime as never, durable as never);
await expect(
controller.stop(
'Nova',
'durable-1',
{ approvalRef: 'approval-1' },
{ id: 'owner' },
'corr-1',
),
).rejects.toBeDefined();
expect(runtime.terminate).not.toHaveBeenCalled();
});
it('surfaces a denied runtime approval to the CLI interaction surface', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const runtime = {
terminate: vi.fn().mockRejectedValue(new Error('Runtime termination approval denied')),
};
const durable = {
getSnapshot: vi.fn().mockResolvedValue({
identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
}),
};
const controller = new InteractionController(runtime as never, durable as never);
await expect(
controller.stop(
'Nova',
'durable-1',
{ approvalRef: 'approval-1' },
{ id: 'owner' },
'corr-1',
),
).rejects.toThrow('Runtime termination approval denied');
});
it('uses the durable session identity and runtime registry for an approved stop', async () => { it('uses the durable session identity and runtime registry for an approved stop', async () => {
process.env['MOSAIC_AGENT_NAME'] = 'Nova'; process.env['MOSAIC_AGENT_NAME'] = 'Nova';
const runtime = { terminate: vi.fn().mockResolvedValue(undefined) }; const runtime = { terminate: vi.fn().mockResolvedValue(undefined) };

View File

@@ -4,17 +4,21 @@ import {
ForbiddenException, ForbiddenException,
Get, Get,
Headers, Headers,
Sse,
Inject, Inject,
Param, Param,
Post, Post,
Query, Query,
UseGuards, UseGuards,
UseFilters,
} from '@nestjs/common'; } from '@nestjs/common';
import type { RuntimeAttachMode } from '@mosaicstack/types'; import type { RuntimeAttachMode, RuntimeStreamEvent } from '@mosaicstack/types';
import { Observable } from 'rxjs';
import { AuthGuard } from '../auth/auth.guard.js'; import { AuthGuard } from '../auth/auth.guard.js';
import { CurrentUser } from '../auth/current-user.decorator.js'; import { CurrentUser } from '../auth/current-user.decorator.js';
import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js'; import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js';
import { TessDurableSessionService } from './tess-durable-session.service.js'; import { TessDurableSessionService } from './tess-durable-session.service.js';
import { RuntimeApprovalDeniedFilter } from './runtime-approval-denied.filter.js';
import { import {
RuntimeProviderService, RuntimeProviderService,
type RuntimeProviderRequestContext, type RuntimeProviderRequestContext,
@@ -26,6 +30,7 @@ import {
*/ */
@Controller('api/interaction/:agentName') @Controller('api/interaction/:agentName')
@UseGuards(AuthGuard) @UseGuards(AuthGuard)
@UseFilters(RuntimeApprovalDeniedFilter)
export class InteractionController { export class InteractionController {
constructor( constructor(
@Inject(RuntimeProviderService) private readonly runtime: RuntimeProviderService, @Inject(RuntimeProviderService) private readonly runtime: RuntimeProviderService,
@@ -60,6 +65,41 @@ export class InteractionController {
); );
} }
/**
* Bind an existing, authorized runtime session to the stable conversation ID.
* This is the lifecycle boundary where both runtime identifiers are known.
*/
@Post('sessions/:sessionId/enroll')
async enroll(
@Param('agentName') agentName: string,
@Param('sessionId') sessionId: string,
@Body() body: { providerId?: string; runtimeSessionId?: string } = {},
@CurrentUser() user: AuthenticatedUserLike,
@Headers('x-correlation-id') correlationId?: string,
) {
this.assertConfiguredAgent(agentName);
const providerId = this.requiredProvider(body.providerId ?? '');
const runtimeSessionId = body.runtimeSessionId?.trim();
if (!runtimeSessionId) throw new ForbiddenException('Runtime session identity is required');
const context = this.context(user, correlationId);
const sessions = await this.runtime.listSessions(providerId, context);
if (!sessions.some((session): boolean => session.id === runtimeSessionId)) {
throw new ForbiddenException('Runtime session is not visible to this actor');
}
await this.durable.enroll(
{
agentName,
sessionId,
tenantId: context.actorScope.tenantId,
ownerId: context.actorScope.userId,
providerId,
runtimeSessionId,
},
context,
);
return { status: 'enrolled', sessionId };
}
@Post('sessions/:sessionId/attach') @Post('sessions/:sessionId/attach')
async attach( async attach(
@Param('agentName') agentName: string, @Param('agentName') agentName: string,
@@ -84,6 +124,53 @@ export class InteractionController {
); );
} }
@Sse('sessions/:sessionId/stream')
stream(
@Param('agentName') agentName: string,
@Param('sessionId') sessionId: string,
@Query('cursor') cursor: string | undefined,
@CurrentUser() user: AuthenticatedUserLike,
@Headers('x-correlation-id') correlationId?: string,
): Observable<{ data: RuntimeStreamEvent }> {
this.assertConfiguredAgent(agentName);
const context = this.context(user, correlationId);
return new Observable((subscriber) => {
let iterator: AsyncIterator<RuntimeStreamEvent> | undefined;
let cancelled = false;
void (async (): Promise<void> => {
try {
const snapshot = await this.durable.getSnapshot(sessionId, context);
if (cancelled || subscriber.closed) return;
this.assertSessionAgent(snapshot.identity.agentName, agentName);
iterator = this.runtime
.streamSession(
snapshot.identity.providerId,
snapshot.identity.runtimeSessionId,
cursor?.trim() || undefined,
context,
)
[Symbol.asyncIterator]();
if (cancelled || subscriber.closed) {
await iterator.return?.();
return;
}
while (!cancelled && !subscriber.closed) {
const next = await iterator.next();
if (next.done || cancelled || subscriber.closed) break;
subscriber.next({ data: next.value });
}
if (!subscriber.closed) subscriber.complete();
} catch (error: unknown) {
if (!subscriber.closed) subscriber.error(error);
}
})();
return (): void => {
cancelled = true;
void iterator?.return?.();
};
});
}
@Post('sessions/:sessionId/send') @Post('sessions/:sessionId/send')
async send( async send(
@Param('agentName') agentName: string, @Param('agentName') agentName: string,

View File

@@ -0,0 +1,13 @@
import { Catch, type ArgumentsHost, type ExceptionFilter } from '@nestjs/common';
import { RuntimeApprovalDeniedError } from './runtime-provider-registry.service.js';
/** Maps a consumed/missing runtime approval to a stable HTTP authorization response. */
@Catch(RuntimeApprovalDeniedError)
export class RuntimeApprovalDeniedFilter implements ExceptionFilter {
catch(_exception: RuntimeApprovalDeniedError, host: ArgumentsHost): void {
const response = host.switchToHttp().getResponse<{
status(code: number): { send(body: { statusCode: number; message: string }): void };
}>();
response.status(403).send({ statusCode: 403, message: 'Runtime termination approval denied' });
}
}

View File

@@ -77,7 +77,7 @@ function configuredAgentName(): string {
return agentName; return agentName;
} }
class RuntimeApprovalDeniedError extends Error { export class RuntimeApprovalDeniedError extends Error {
constructor() { constructor() {
super('Runtime termination approval denied'); super('Runtime termination approval denied');
} }
@@ -301,7 +301,7 @@ export class RuntimeProviderService {
return result; return result;
} catch (error: unknown) { } catch (error: unknown) {
const durationMs = Date.now() - startedAt; const durationMs = Date.now() - startedAt;
if (invocationStarted && !(error instanceof RuntimeApprovalDeniedError)) { if (invocationStarted && !this.isAuthorizationDenied(error)) {
await this.recordFailure(providerId, operation, scope, resourceId, durationMs); await this.recordFailure(providerId, operation, scope, resourceId, durationMs);
} else { } else {
await this.record( await this.record(
@@ -360,6 +360,17 @@ export class RuntimeProviderService {
} }
} }
private isAuthorizationDenied(error: unknown): boolean {
return (
error instanceof RuntimeApprovalDeniedError ||
error instanceof ForbiddenException ||
(typeof error === 'object' &&
error !== null &&
'code' in error &&
(error as { code?: unknown }).code === 'forbidden')
);
}
private provider(providerId: string): AgentRuntimeProvider { private provider(providerId: string): AgentRuntimeProvider {
try { try {
return this.registry.require(providerId); return this.registry.require(providerId);

View File

@@ -50,9 +50,20 @@ export class TessDurableSessionRepository implements DurableSessionStore {
.onConflictDoNothing(); .onConflictDoNothing();
const existing = await this.session(identity.sessionId); const existing = await this.session(identity.sessionId);
if (!existing || !sameIdentity(existing, identity)) { if (!existing || !sameEnrollmentScope(existing, identity)) {
throw new Error(`Durable Tess session identity conflict: ${identity.sessionId}`); 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> { async snapshot(sessionId: string): Promise<DurableSessionSnapshot | null> {
@@ -427,14 +438,12 @@ function matchesCheckpointDigest(stored: string, digest: string): boolean {
return stored === digest; return stored === digest;
} }
function sameIdentity(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean { function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
return ( return (
left.agentName === right.agentName && left.agentName === right.agentName &&
left.sessionId === right.sessionId && left.sessionId === right.sessionId &&
left.tenantId === right.tenantId && left.tenantId === right.tenantId &&
left.ownerId === right.ownerId && left.ownerId === right.ownerId
left.providerId === right.providerId &&
left.runtimeSessionId === right.runtimeSessionId
); );
} }

View File

@@ -1,5 +1,5 @@
import { ForbiddenException, Inject, Injectable } from '@nestjs/common'; import { ForbiddenException, Inject, Injectable } from '@nestjs/common';
import { DurableSessionCoordinator } from '@mosaicstack/agent'; import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
import type { TessProviderOutboxDto } from './tess-durable-session.dto.js'; import type { TessProviderOutboxDto } from './tess-durable-session.dto.js';
import { TessDurableSessionRepository } from './tess-durable-session.repository.js'; import { TessDurableSessionRepository } from './tess-durable-session.repository.js';
import { import {
@@ -23,6 +23,20 @@ export class TessDurableSessionService {
this.coordinator = new DurableSessionCoordinator(repository); this.coordinator = new DurableSessionCoordinator(repository);
} }
/** Enroll a verified runtime session under the stable cross-surface conversation handle. */
async enroll(
identity: DurableSessionIdentity,
context: RuntimeProviderRequestContext,
): Promise<void> {
if (
identity.ownerId !== context.actorScope.userId ||
identity.tenantId !== context.actorScope.tenantId
) {
throw new ForbiddenException('Durable Tess enrollment scope mismatch');
}
await this.coordinator.create(identity);
}
async queueProviderSend(input: TessProviderOutboxDto): Promise<void> { async queueProviderSend(input: TessProviderOutboxDto): Promise<void> {
const snapshot = await this.coordinator.snapshot(input.sessionId); const snapshot = await this.coordinator.snapshot(input.sessionId);
this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input); this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);

View File

@@ -32,7 +32,12 @@ import type {
AbortPayload, AbortPayload,
} from '@mosaicstack/types'; } from '@mosaicstack/types';
import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js'; import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js';
import { RuntimeProviderService } from '../agent/runtime-provider-registry.service.js'; import {
RUNTIME_PROVIDER_AUDIT_SINK,
RuntimeProviderService,
type RuntimeAuditSink,
} from '../agent/runtime-provider-registry.service.js';
import { TessDurableSessionService } from '../agent/tess-durable-session.service.js';
import { AUTH } from '../auth/auth.tokens.js'; import { AUTH } from '../auth/auth.tokens.js';
import { import {
scopeFromUser, scopeFromUser,
@@ -134,6 +139,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
@Optional() @Optional()
@Inject(RuntimeProviderService) @Inject(RuntimeProviderService)
private readonly runtimeRegistry: RuntimeProviderService | null = null, private readonly runtimeRegistry: RuntimeProviderService | null = null,
@Optional()
@Inject(TessDurableSessionService)
private readonly durableSessions: TessDurableSessionService | null = null,
@Optional()
@Inject(RUNTIME_PROVIDER_AUDIT_SINK)
private readonly runtimeAudit: RuntimeAuditSink | null = null,
) {} ) {}
afterInit(): void { afterInit(): void {
@@ -656,9 +667,16 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
): Promise<void> { ): Promise<void> {
if (!client.data.discordService) return; if (!client.data.discordService) return;
const ingress = this.resolveDiscordIngress(client, envelope, 'approve'); const ingress = this.resolveDiscordIngress(client, envelope, 'approve');
const actionParts = ingress?.content.match(/^\/approve\s+([^\s]+)\s+([^\s]+)$/i); const isApprovalCommand = /^\/approve\s*$/i.test(ingress?.content ?? '');
const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim(); const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
if (!ingress || !actionParts || !tenantId || !this.commandAuthorization) return; if (
!ingress ||
!isApprovalCommand ||
!tenantId ||
!this.commandAuthorization ||
!this.durableSessions
)
return;
const binding = resolveDiscordInteractionBinding( const binding = resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']), parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
ingress.guildId, ingress.guildId,
@@ -680,22 +698,63 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}); });
return; return;
} }
const providerId = actionParts[1]!; let snapshot;
const sessionId = actionParts[2]!; try {
snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, {
actorScope: { userId: actorId, tenantId },
channelId: ingress.channelId,
correlationId: ingress.correlationId,
});
} catch {
client.emit('discord:approval', {
correlationId: ingress.correlationId,
success: false,
approvalId: undefined,
expiresAt: undefined,
});
return;
}
if (snapshot.identity.agentName !== agentName) {
client.emit('discord:approval', {
correlationId: ingress.correlationId,
success: false,
approvalId: undefined,
expiresAt: undefined,
});
return;
}
const approval = await this.commandAuthorization.createRuntimeTerminationApproval({ const approval = await this.commandAuthorization.createRuntimeTerminationApproval({
providerId, providerId: snapshot.identity.providerId,
sessionId, sessionId: snapshot.identity.runtimeSessionId,
actorId, actorId,
tenantId, tenantId,
channelId: ingress.channelId, channelId: ingress.channelId,
correlationId: this.discordRuntimeActionCorrelation( correlationId: this.discordRuntimeActionCorrelation(
binding.instanceId, binding.instanceId,
ingress, ingress,
providerId, snapshot.identity.providerId,
sessionId, snapshot.identity.runtimeSessionId,
), ),
agentName, agentName,
}); });
if (!approval) {
await this.runtimeAudit?.record({
providerId: snapshot.identity.providerId,
operation: 'session.terminate',
outcome: 'denied',
actorId,
tenantId,
channelId: ingress.channelId,
correlationId: this.discordRuntimeActionCorrelation(
binding.instanceId,
ingress,
snapshot.identity.providerId,
snapshot.identity.runtimeSessionId,
),
resourceId: snapshot.identity.runtimeSessionId,
errorCode: 'policy_denied',
});
}
client.emit('discord:approval', { client.emit('discord:approval', {
correlationId: ingress.correlationId, correlationId: ingress.correlationId,
success: approval !== null, success: approval !== null,
@@ -711,9 +770,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
): Promise<void> { ): Promise<void> {
if (!client.data.discordService) return; if (!client.data.discordService) return;
const ingress = this.resolveDiscordIngress(client, envelope, 'stop'); const ingress = this.resolveDiscordIngress(client, envelope, 'stop');
const actionParts = ingress?.content.match(/^\/stop\s+([^\s]+)\s+([^\s]+)\s+([^\s]+)$/i); const approvalRef = /^\/stop\s+([^\s]+)$/i.exec(ingress?.content ?? '')?.[1];
const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim(); const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
if (!ingress || !actionParts || !tenantId || !this.runtimeRegistry) return; if (!ingress || !approvalRef || !tenantId || !this.runtimeRegistry || !this.durableSessions)
return;
const binding = resolveDiscordInteractionBinding( const binding = resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']), parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
ingress.guildId, ingress.guildId,
@@ -723,25 +783,31 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
); );
const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId); const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
if (!actorId) return; if (!actorId) return;
const providerId = actionParts[1]!;
const sessionId = actionParts[2]!;
try { try {
const context = {
actorScope: { userId: actorId, tenantId },
channelId: ingress.channelId,
correlationId: ingress.correlationId,
};
const snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, context);
if (snapshot.identity.agentName !== binding.instanceId) throw new Error('agent mismatch');
// RuntimeProviderService consumes the durable approval exactly once using the // RuntimeProviderService consumes the durable approval exactly once using the
// provisioned approving-admin identity, never the Discord service account. // provisioned approving-admin identity, never the Discord service account.
await this.runtimeRegistry.terminate(providerId, sessionId, actionParts[3]!, { await this.runtimeRegistry.terminate(
actorScope: { snapshot.identity.providerId,
userId: actorId, snapshot.identity.runtimeSessionId,
tenantId, approvalRef,
}, {
channelId: ingress.channelId, ...context,
correlationId: this.discordRuntimeActionCorrelation( correlationId: this.discordRuntimeActionCorrelation(
binding.instanceId, binding.instanceId,
ingress, ingress,
providerId, snapshot.identity.providerId,
sessionId, snapshot.identity.runtimeSessionId,
), ),
}); },
);
client.emit('discord:stop', { correlationId: ingress.correlationId, success: true }); client.emit('discord:stop', { correlationId: ingress.correlationId, success: true });
} catch { } catch {
client.emit('discord:stop', { correlationId: ingress.correlationId, success: false }); client.emit('discord:stop', { correlationId: ingress.correlationId, success: false });

View File

@@ -77,9 +77,17 @@ function discordGateway(role: 'admin' | 'member'): {
gateway: ChatGateway; gateway: ChatGateway;
client: { data: { discordService: boolean }; emit: ReturnType<typeof vi.fn> }; client: { data: { discordService: boolean }; emit: ReturnType<typeof vi.fn> };
consumedActions: Array<{ actorId: string; correlationId: string }>; consumedActions: Array<{ actorId: string; correlationId: string }>;
durable: { getSnapshot: ReturnType<typeof vi.fn> };
audit: { record: ReturnType<typeof vi.fn> };
} { } {
const authorization = commandAuthorization(role); const authorization = commandAuthorization(role);
const consumedActions: Array<{ actorId: string; correlationId: string }> = []; const consumedActions: Array<{ actorId: string; correlationId: string }> = [];
const durable = {
getSnapshot: vi.fn().mockResolvedValue({
identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
}),
};
const audit = { record: vi.fn().mockResolvedValue(undefined) };
const runtimeRegistry = new RuntimeProviderService( const runtimeRegistry = new RuntimeProviderService(
{ {
require: () => ({ require: () => ({
@@ -105,9 +113,13 @@ function discordGateway(role: 'admin' | 'member'): {
{} as never, {} as never,
authorization, authorization,
runtimeRegistry, runtimeRegistry,
durable as never,
audit as never,
), ),
client: { data: { discordService: true }, emit: vi.fn() }, client: { data: { discordService: true }, emit: vi.fn() },
consumedActions, consumedActions,
durable,
audit,
}; };
} }
@@ -241,7 +253,7 @@ describe('Discord ingress security', () => {
const { gateway, client, consumedActions } = discordGateway('admin'); const { gateway, client, consumedActions } = discordGateway('admin');
await gateway.handleDiscordApproval( await gateway.handleDiscordApproval(
client as never, client as never,
ingressEnvelope('/approve fleet runtime-1', 'approve-message', { ingressEnvelope('/approve', 'approve-message', {
correlationId: 'approval-ingress-correlation', correlationId: 'approval-ingress-correlation',
}), }),
); );
@@ -255,7 +267,7 @@ describe('Discord ingress security', () => {
await gateway.handleDiscordStop( await gateway.handleDiscordStop(
client as never, client as never,
ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'stop-message', { ingressEnvelope(`/stop ${approval.approvalId}`, 'stop-message', {
correlationId: 'stop-ingress-correlation', correlationId: 'stop-ingress-correlation',
}), }),
); );
@@ -271,14 +283,13 @@ describe('Discord ingress security', () => {
]); ]);
}); });
it('rejects approval when the binding targets a different runtime agent', async () => { it('audits a Discord mint-side authorization denial', async () => {
configureDiscordEnv(); configureDiscordEnv();
process.env['MOSAIC_AGENT_NAME'] = 'Other'; const { gateway, client, audit } = discordGateway('member');
const { gateway, client } = discordGateway('admin');
await gateway.handleDiscordApproval( await gateway.handleDiscordApproval(
client as never, client as never,
ingressEnvelope('/approve fleet runtime-1', 'mismatched-agent-approve'), ingressEnvelope('/approve', 'denied-approve'),
); );
expect(client.emit).toHaveBeenCalledWith('discord:approval', { expect(client.emit).toHaveBeenCalledWith('discord:approval', {
@@ -287,14 +298,57 @@ describe('Discord ingress security', () => {
approvalId: undefined, approvalId: undefined,
expiresAt: undefined, expiresAt: undefined,
}); });
expect(audit.record).toHaveBeenCalledWith(
expect.objectContaining({
outcome: 'denied',
operation: 'session.terminate',
errorCode: 'policy_denied',
}),
);
}); });
it.each([
[
'binding',
() => {
process.env['MOSAIC_AGENT_NAME'] = 'Other';
},
],
[
'durable session',
(durable: { getSnapshot: ReturnType<typeof vi.fn> }) => {
durable.getSnapshot.mockResolvedValueOnce({
identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
});
},
],
])(
'rejects approval when the %s targets a different runtime agent',
async (_source, configure) => {
configureDiscordEnv();
const { gateway, client, durable } = discordGateway('admin');
configure(durable);
await gateway.handleDiscordApproval(
client as never,
ingressEnvelope('/approve', 'mismatched-agent-approve'),
);
expect(client.emit).toHaveBeenCalledWith('discord:approval', {
correlationId: 'correlation-001',
success: false,
approvalId: undefined,
expiresAt: undefined,
});
},
);
it('rejects unpaired and non-admin Discord users for approval and stop', async () => { it('rejects unpaired and non-admin Discord users for approval and stop', async () => {
configureDiscordEnv(); configureDiscordEnv();
const { gateway, client } = discordGateway('member'); const { gateway, client } = discordGateway('member');
await gateway.handleDiscordApproval( await gateway.handleDiscordApproval(
client as never, client as never,
ingressEnvelope('/approve fleet runtime-1', 'member-approve'), ingressEnvelope('/approve', 'member-approve'),
); );
expect(client.emit).toHaveBeenCalledWith('discord:approval', { expect(client.emit).toHaveBeenCalledWith('discord:approval', {
correlationId: 'correlation-001', correlationId: 'correlation-001',
@@ -306,7 +360,7 @@ describe('Discord ingress security', () => {
process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([]); process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([]);
await gateway.handleDiscordStop( await gateway.handleDiscordStop(
client as never, client as never,
ingressEnvelope('/stop fleet runtime-1 forged', 'unpaired-stop'), ingressEnvelope('/stop forged', 'unpaired-stop'),
); );
expect(client.emit).not.toHaveBeenCalledWith('discord:stop', expect.anything()); expect(client.emit).not.toHaveBeenCalledWith('discord:stop', expect.anything());
}); });
@@ -316,7 +370,7 @@ describe('Discord ingress security', () => {
const { gateway, client } = discordGateway('admin'); const { gateway, client } = discordGateway('admin');
await gateway.handleDiscordApproval( await gateway.handleDiscordApproval(
client as never, client as never,
ingressEnvelope('/approve fleet runtime-1', 'replay-approve', { ingressEnvelope('/approve', 'replay-approve', {
correlationId: 'replay-approval-correlation', correlationId: 'replay-approval-correlation',
}), }),
); );
@@ -327,13 +381,13 @@ describe('Discord ingress security', () => {
}; };
await gateway.handleDiscordStop( await gateway.handleDiscordStop(
client as never, client as never,
ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-one', { ingressEnvelope(`/stop ${approval.approvalId}`, 'replay-stop-one', {
correlationId: 'replay-stop-correlation-one', correlationId: 'replay-stop-correlation-one',
}), }),
); );
await gateway.handleDiscordStop( await gateway.handleDiscordStop(
client as never, client as never,
ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-two', { ingressEnvelope(`/stop ${approval.approvalId}`, 'replay-stop-two', {
correlationId: 'replay-stop-correlation-two', correlationId: 'replay-stop-correlation-two',
}), }),
); );

View File

@@ -56,3 +56,11 @@
**Final focused review:** PASS. Deterministic audit validated all task repository roots with zero missing paths; no planning placeholders remained; security prerequisites still gate Tess exposure; observability traceability is explicit. **Final focused review:** PASS. Deterministic audit validated all task repository roots with zero missing paths; no planning placeholders remained; security prerequisites still gate Tess exposure; observability traceability is explicit.
**Current gate:** planning PR must merge to `main` with terminal-green CI before any source-code worker starts. **Current gate:** planning PR must merge to `main` with terminal-green CI before any source-code worker starts.
## 2026-07-13 — M3 cross-surface delivery
**Branch:** `feat/tess-m3-integration` from `main` at `84d884b9`.
**Delivered:** Stable Discord `conversationId` enrollment after a visible provider/runtime session is known; idempotent provider-session rebinding that preserves agent/tenant/owner scope; Discord approval/stop target resolution through the durable snapshot; SSE runtime streaming after CLI attach; denial/audit parity including provider authorization denials and HTTP 403 approval-denial mapping.
**Evidence:** Gateway targeted suite: 37 tests passed; Mosaic CLI interaction test passed; agent durable-session test passed; gateway and CLI typechecks passed; changed-file format and whitespace checks passed. Codex security review found no confident vulnerability. Code review identified a Fastify exception-response mismatch and two UX/acknowledgement issues; all were corrected before the final validation run.

View File

@@ -64,6 +64,23 @@ describe('DurableSessionCoordinator', () => {
expect(recovered.handoffs).toMatchObject([{ handoffId: 'handoff-1', status: 'pending' }]); 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 () => { it('deduplicates duplicate ingress and never reprocesses an inbox record after restart or compaction', async () => {
const store = new InMemoryDurableSessionStore(); const store = new InMemoryDurableSessionStore();
const firstProcess = new DurableSessionCoordinator(store); const firstProcess = new DurableSessionCoordinator(store);

View File

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

View File

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

View File

@@ -3,6 +3,7 @@ import type { Command } from 'commander';
import { withAuth } from './with-auth.js'; import { withAuth } from './with-auth.js';
import { import {
attachInteractionSession, attachInteractionSession,
enrollInteractionSession,
fetchInteractionHealth, fetchInteractionHealth,
fetchInteractionSessions, fetchInteractionSessions,
fetchInteractionStatus, fetchInteractionStatus,
@@ -10,6 +11,7 @@ import {
recoverInteractionSession, recoverInteractionSession,
sendInteractionMessage, sendInteractionMessage,
stopInteractionSession, stopInteractionSession,
streamInteractionSession,
} from '../tui/gateway-api.js'; } from '../tui/gateway-api.js';
interface InteractionOptions { interface InteractionOptions {
@@ -103,18 +105,48 @@ export function registerInteractionCommand(program: Command): Command {
}); });
options( 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)') .option('--control', 'Request control mode (provider policy may deny it)')
.action(async (sessionId: string, opts: InteractionOptions & { control?: boolean }) => { .action(async (sessionId: string, opts: InteractionOptions & { control?: boolean }) => {
const request = await authRequest(opts); const request = await authRequest(opts);
print( const attachment = await attachInteractionSession(request.gateway, request.cookie, {
await attachInteractionSession(request.gateway, request.cookie, {
...request, ...request,
sessionId, sessionId,
mode: opts.control ? 'control' : 'read', mode: opts.control ? 'control' : 'read',
}), });
); print(attachment);
for await (const event of streamInteractionSession(request.gateway, request.cookie, {
...request,
sessionId,
})) {
print(event);
}
}); });
const send = options( const send = options(

View File

@@ -406,6 +406,32 @@ export async function fetchInteractionTree(
return handleResponse<unknown[]>(res, 'Failed to get interaction session tree'); 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( export async function attachInteractionSession(
gatewayUrl: string, gatewayUrl: string,
sessionCookie: string, sessionCookie: string,
@@ -422,6 +448,49 @@ export async function attachInteractionSession(
return handleResponse<unknown>(res, 'Failed to attach interaction session'); 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( export async function sendInteractionMessage(
gatewayUrl: string, gatewayUrl: string,
sessionCookie: string, sessionCookie: string,