From 6345dbfcf26272a3fb7ed96bfa80d9429fdcc188 Mon Sep 17 00:00:00 2001 From: "jason.woltje" Date: Mon, 13 Jul 2026 15:29:37 +0000 Subject: [PATCH] feat(agent): add Matrix native runtime provider (#744) --- packages/agent/src/index.ts | 1 + .../matrix-native-runtime-provider.test.ts | 153 ++++++++ .../src/matrix-native-runtime-provider.ts | 351 +++++++++++++++++ .../agent/src/runtime-provider-parity.test.ts | 176 +++++++++ .../matrix-native-runtime-transport.test.ts | 182 +++++++++ .../fleet/matrix-native-runtime-transport.ts | 367 ++++++++++++++++++ packages/mosaic/src/index.ts | 1 + 7 files changed, 1231 insertions(+) create mode 100644 packages/agent/src/matrix-native-runtime-provider.test.ts create mode 100644 packages/agent/src/matrix-native-runtime-provider.ts create mode 100644 packages/agent/src/runtime-provider-parity.test.ts create mode 100644 packages/mosaic/src/fleet/matrix-native-runtime-transport.test.ts create mode 100644 packages/mosaic/src/fleet/matrix-native-runtime-transport.ts diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index f7b1569..702d9d9 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -3,4 +3,5 @@ export const VERSION = '0.0.0'; export * from './runtime-provider-registry.js'; export * from './tmux-fleet-runtime-provider.js'; export * from './hermes-runtime-provider.js'; +export * from './matrix-native-runtime-provider.js'; export * from './tess-durable-session.js'; diff --git a/packages/agent/src/matrix-native-runtime-provider.test.ts b/packages/agent/src/matrix-native-runtime-provider.test.ts new file mode 100644 index 0000000..01631dc --- /dev/null +++ b/packages/agent/src/matrix-native-runtime-provider.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { RuntimeScope, RuntimeStreamEvent } from '@mosaicstack/types'; +import { + type MatrixRuntimeSession, + type MatrixRuntimeTransport, + MatrixNativeRuntimeProvider, + type MatrixReadAuthority, + type MatrixWriteAuthority, +} from './matrix-native-runtime-provider.js'; + +const scope: RuntimeScope = { + actorId: 'operator-1', + tenantId: 'tenant-a', + channelId: 'matrix-control', + correlationId: 'corr-1', +}; + +const session: MatrixRuntimeSession = { + id: 'native-1', + runtimeId: '@worker:example.test', + state: 'active', + createdAt: '2026-07-13T00:00:00.000Z', + updatedAt: '2026-07-13T00:00:00.000Z', +}; + +function transport(): MatrixRuntimeTransport { + return { + health: vi.fn(async () => ({ + status: 'healthy' as const, + checkedAt: '2026-07-13T00:00:00.000Z', + })), + listSessions: vi.fn(async () => [session]), + verifySession: vi.fn(async (sessionId) => { + if (sessionId !== session.id) throw new Error('unexpected session'); + return session; + }), + stream: vi.fn(async function* (): AsyncIterable { + yield { + type: 'message.delta', + sessionId: 'native-1', + cursor: 'cursor-1', + occurredAt: '2026-07-13T00:00:00.000Z', + content: 'hello', + }; + }), + send: vi.fn(async () => undefined), + terminate: vi.fn(async () => undefined), + }; +} + +function readAuthority(): MatrixReadAuthority { + return { canRead: vi.fn(async () => true) }; +} + +function writeAuthority(): MatrixWriteAuthority { + return { + canWrite: vi.fn(async () => true), + assertAuthorized: vi.fn(async () => undefined), + }; +} + +describe('MatrixNativeRuntimeProvider contract boundary', (): void => { + it('advertises the concrete Matrix operations and returns only normalized sessions', async (): Promise => { + const provider = new MatrixNativeRuntimeProvider({ + transport: transport(), + readAuthority: readAuthority(), + }); + + await expect(provider.capabilities(scope)).resolves.toEqual({ + supported: [ + 'session.list', + 'session.tree', + 'session.stream', + 'session.send', + 'session.attach', + 'session.terminate', + ], + }); + await expect(provider.listSessions(scope)).resolves.toEqual([ + expect.objectContaining({ id: 'native-1', providerId: 'runtime.matrix' }), + ]); + }); + + it('binds read attachments to immutable scope and rejects control mode before transport access', async (): Promise => { + const matrix = transport(); + const provider = new MatrixNativeRuntimeProvider({ + transport: matrix, + readAuthority: readAuthority(), + attachmentIdFactory: () => 'attachment-1', + now: () => new Date('2026-07-13T00:00:00.000Z'), + }); + + await expect(provider.attach('native-1', 'control', scope)).rejects.toMatchObject({ + code: 'forbidden', + }); + expect(matrix.verifySession).not.toHaveBeenCalled(); + + await provider.attach('native-1', 'read', scope); + await expect( + provider.detach('attachment-1', { ...scope, tenantId: 'other' }), + ).rejects.toMatchObject({ + code: 'forbidden', + }); + }); + + it('validates messages and applies exact Matrix authority after bound-session verification', async (): Promise => { + const matrix = transport(); + const writes = writeAuthority(); + const provider = new MatrixNativeRuntimeProvider({ + transport: matrix, + readAuthority: readAuthority(), + writeAuthority: writes, + }); + + await expect( + provider.sendMessage('native-1', { content: '', idempotencyKey: 'msg-1' }, scope), + ).rejects.toMatchObject({ + code: 'invalid_request', + }); + expect(matrix.verifySession).not.toHaveBeenCalled(); + + await provider.sendMessage('native-1', { content: 'hello', idempotencyKey: 'msg-1' }, scope); + expect(writes.assertAuthorized).toHaveBeenCalledWith({ + operation: 'session.send', + sessionId: 'native-1', + scope, + }); + expect(matrix.send).toHaveBeenCalledWith( + 'native-1', + { content: 'hello', idempotencyKey: 'msg-1' }, + scope, + ); + }); + + it('streams only after exact read authorization', async (): Promise => { + const matrix = transport(); + const provider = new MatrixNativeRuntimeProvider({ + transport: matrix, + readAuthority: readAuthority(), + }); + + await expect(collect(provider.streamSession('native-1', 'cursor-0', scope))).resolves.toEqual([ + expect.objectContaining({ type: 'message.delta', sessionId: 'native-1' }), + ]); + expect(matrix.stream).toHaveBeenCalledWith('native-1', 'cursor-0', scope); + }); +}); + +async function collect(stream: AsyncIterable): Promise { + const values: RuntimeStreamEvent[] = []; + for await (const value of stream) values.push(value); + return values; +} diff --git a/packages/agent/src/matrix-native-runtime-provider.ts b/packages/agent/src/matrix-native-runtime-provider.ts new file mode 100644 index 0000000..3398514 --- /dev/null +++ b/packages/agent/src/matrix-native-runtime-provider.ts @@ -0,0 +1,351 @@ +import { randomUUID } from 'node:crypto'; +import type { + AgentRuntimeProvider, + RuntimeAttachHandle, + RuntimeAttachMode, + RuntimeCapabilitySet, + RuntimeHealth, + RuntimeMessage, + RuntimeScope, + RuntimeSession, + RuntimeSessionTree, + RuntimeStreamEvent, +} from '@mosaicstack/types'; + +const MATRIX_PROVIDER_ID = 'runtime.matrix'; +const ATTACHMENT_TTL_MS = 5 * 60 * 1_000; + +/** A verified native runtime session. Matrix room and event details remain transport-local. */ +export interface MatrixRuntimeSession { + id: string; + runtimeId: string; + parentSessionId?: string; + state: RuntimeSession['state']; + createdAt: string; + updatedAt: string; +} + +/** + * Narrow native Matrix boundary. The concrete Mosaic transport owns homeserver + * authentication, exact room mapping, Matrix identity checks, and replay cursors. + */ +export interface MatrixRuntimeTransport { + health(scope: RuntimeScope): Promise; + listSessions(scope: RuntimeScope): Promise; + verifySession(sessionId: string, scope: RuntimeScope): Promise; + stream( + sessionId: string, + cursor: string | undefined, + scope: RuntimeScope, + ): AsyncIterable; + send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise; + terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise; +} + +export type MatrixRuntimeProviderErrorCode = + | 'capability_unsupported' + | 'forbidden' + | 'invalid_request' + | 'not_found'; + +export class MatrixRuntimeProviderError extends Error { + constructor( + readonly code: MatrixRuntimeProviderErrorCode, + message: string, + ) { + super(message); + this.name = MatrixRuntimeProviderError.name; + } +} + +export type MatrixReadOperation = + | 'runtime.health' + | 'session.list' + | 'session.tree' + | 'session.stream' + | 'session.attach'; + +export interface MatrixReadAuthority { + canRead(input: { + operation: MatrixReadOperation; + scope: RuntimeScope; + sessionId?: string; + }): Promise; +} + +export interface MatrixWriteAuthority { + canWrite(input: { + operation: 'session.send' | 'session.terminate'; + sessionId: string; + scope: RuntimeScope; + approvalRef?: string; + }): Promise; + assertAuthorized(input: { + operation: 'session.send' | 'session.terminate'; + sessionId: string; + scope: RuntimeScope; + approvalRef?: string; + }): Promise; +} + +export interface MatrixNativeRuntimeProviderOptions { + transport: MatrixRuntimeTransport; + readAuthority?: MatrixReadAuthority; + writeAuthority?: MatrixWriteAuthority; + attachmentIdFactory?: () => string; + now?: () => Date; + attachmentTtlMs?: number; +} + +interface Attachment { + sessionId: string; + scope: RuntimeScope; + expiresAtMs: number; +} + +class DenyMatrixReadAuthority implements MatrixReadAuthority { + async canRead(): Promise { + return false; + } +} + +class DenyMatrixWriteAuthority implements MatrixWriteAuthority { + async canWrite(): Promise { + return false; + } + + async assertAuthorized(): Promise { + throw new MatrixRuntimeProviderError( + 'forbidden', + 'Matrix runtime writes require Mos authority', + ); + } +} + +/** + * Native Matrix adapter behind the Mosaic runtime contract. It accepts only + * stable session IDs; room identifiers, Matrix event schemas, and credentials + * are deliberately confined to the concrete transport implementation. + */ +export class MatrixNativeRuntimeProvider implements AgentRuntimeProvider { + readonly id = MATRIX_PROVIDER_ID; + private readonly readAuthority: MatrixReadAuthority; + private readonly writeAuthority: MatrixWriteAuthority; + private readonly attachmentIdFactory: () => string; + private readonly now: () => Date; + private readonly attachmentTtlMs: number; + private readonly attachments = new Map(); + + constructor(private readonly options: MatrixNativeRuntimeProviderOptions) { + this.readAuthority = options.readAuthority ?? new DenyMatrixReadAuthority(); + this.writeAuthority = options.writeAuthority ?? new DenyMatrixWriteAuthority(); + this.attachmentIdFactory = options.attachmentIdFactory ?? randomUUID; + this.now = options.now ?? (() => new Date()); + this.attachmentTtlMs = options.attachmentTtlMs ?? ATTACHMENT_TTL_MS; + } + + async capabilities(_scope: RuntimeScope): Promise { + return { + supported: [ + 'session.list', + 'session.tree', + 'session.stream', + 'session.send', + 'session.attach', + 'session.terminate', + ], + }; + } + + async health(scope: RuntimeScope): Promise { + await this.assertRead('runtime.health', undefined, scope); + return this.options.transport.health(scope); + } + + async listSessions(scope: RuntimeScope): Promise { + await this.assertRead('session.list', undefined, scope); + const sessions = await this.options.transport.listSessions(scope); + const visible = await Promise.all( + sessions.map((session) => + this.readAuthority.canRead({ operation: 'session.list', sessionId: session.id, scope }), + ), + ); + return sessions + .filter((_session, index) => visible[index] === true) + .map((session) => this.runtimeSession(session)); + } + + async getSessionTree(scope: RuntimeScope): Promise { + await this.assertRead('session.tree', undefined, scope); + const sessions = await this.options.transport.listSessions(scope); + const visible = await Promise.all( + sessions.map((session) => + this.readAuthority.canRead({ operation: 'session.tree', sessionId: session.id, scope }), + ), + ); + const runtimeSessions = sessions + .filter((_session, index) => visible[index] === true) + .map((session) => this.runtimeSession(session)); + const nodes = new Map( + runtimeSessions.map((session) => [session.id, { session, children: [] }]), + ); + const roots: RuntimeSessionTree[] = []; + for (const session of runtimeSessions) { + const node = nodes.get(session.id)!; + const parent = session.parentSessionId ? nodes.get(session.parentSessionId) : undefined; + if (parent) parent.children.push(node); + else roots.push(node); + } + return roots; + } + + async *streamSession( + sessionId: string, + cursor: string | undefined, + scope: RuntimeScope, + ): AsyncIterable { + await this.assertRead('session.stream', sessionId, scope); + const session = await this.options.transport.verifySession(sessionId, scope); + await this.assertRead('session.stream', session.id, scope); + yield* this.options.transport.stream(session.id, cursor, scope); + } + + async sendMessage( + sessionId: string, + message: RuntimeMessage, + scope: RuntimeScope, + ): Promise { + if (!message.content.trim()) { + throw new MatrixRuntimeProviderError( + 'invalid_request', + 'Matrix runtime message content is required', + ); + } + await this.assertWritePermitted('session.send', sessionId, scope); + const session = await this.options.transport.verifySession(sessionId, scope); + await this.assertWriteAuthorized('session.send', session.id, scope); + await this.options.transport.send(session.id, message, scope); + } + + async attach( + sessionId: string, + mode: RuntimeAttachMode, + scope: RuntimeScope, + ): Promise { + if (mode !== 'read') { + throw new MatrixRuntimeProviderError('forbidden', 'Matrix control attach is not permitted'); + } + await this.assertRead('session.attach', sessionId, scope); + const session = await this.options.transport.verifySession(sessionId, scope); + await this.assertRead('session.attach', session.id, scope); + const nowMs = this.now().getTime(); + this.pruneExpired(nowMs); + const attachmentId = this.attachmentIdFactory(); + const expiresAtMs = nowMs + this.attachmentTtlMs; + this.attachments.set(attachmentId, { + sessionId: session.id, + scope: snapshotScope(scope), + expiresAtMs, + }); + return { + attachmentId, + sessionId: session.id, + mode, + expiresAt: new Date(expiresAtMs).toISOString(), + }; + } + + async detach(attachmentId: string, scope: RuntimeScope): Promise { + const attachment = this.attachments.get(attachmentId); + if (!attachment) + throw new MatrixRuntimeProviderError('not_found', 'Matrix attachment is not active'); + if (this.now().getTime() >= attachment.expiresAtMs) { + this.attachments.delete(attachmentId); + throw new MatrixRuntimeProviderError('forbidden', 'Matrix attachment has expired'); + } + if (!sameScope(attachment.scope, scope)) { + throw new MatrixRuntimeProviderError('forbidden', 'Matrix attachment scope does not match'); + } + this.attachments.delete(attachmentId); + } + + async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise { + if (!approvalRef.trim()) { + throw new MatrixRuntimeProviderError( + 'invalid_request', + 'Matrix termination approval is required', + ); + } + await this.assertWritePermitted('session.terminate', sessionId, scope, approvalRef); + const session = await this.options.transport.verifySession(sessionId, scope); + await this.assertWriteAuthorized('session.terminate', session.id, scope, approvalRef); + await this.options.transport.terminate(session.id, approvalRef, scope); + } + + private runtimeSession(session: MatrixRuntimeSession): RuntimeSession { + return { ...session, providerId: this.id }; + } + + private async assertRead( + operation: MatrixReadOperation, + sessionId: string | undefined, + scope: RuntimeScope, + ): Promise { + const allowed = await this.readAuthority.canRead({ + operation, + scope, + ...(sessionId ? { sessionId } : {}), + }); + if (!allowed) + throw new MatrixRuntimeProviderError('forbidden', 'Matrix runtime read is not authorized'); + } + + private async assertWritePermitted( + operation: 'session.send' | 'session.terminate', + sessionId: string, + scope: RuntimeScope, + approvalRef?: string, + ): Promise { + const allowed = await this.writeAuthority.canWrite({ + operation, + sessionId, + scope, + ...(approvalRef ? { approvalRef } : {}), + }); + if (!allowed) + throw new MatrixRuntimeProviderError('forbidden', 'Matrix runtime write is not authorized'); + } + + private async assertWriteAuthorized( + operation: 'session.send' | 'session.terminate', + sessionId: string, + scope: RuntimeScope, + approvalRef?: string, + ): Promise { + await this.writeAuthority.assertAuthorized({ + operation, + sessionId, + scope, + ...(approvalRef ? { approvalRef } : {}), + }); + } + + private pruneExpired(nowMs: number): void { + for (const [id, attachment] of this.attachments) { + if (attachment.expiresAtMs <= nowMs) this.attachments.delete(id); + } + } +} + +function snapshotScope(scope: RuntimeScope): RuntimeScope { + return Object.freeze({ ...scope }); +} + +function sameScope(left: RuntimeScope, right: RuntimeScope): boolean { + return ( + left.actorId === right.actorId && + left.tenantId === right.tenantId && + left.channelId === right.channelId && + left.correlationId === right.correlationId + ); +} diff --git a/packages/agent/src/runtime-provider-parity.test.ts b/packages/agent/src/runtime-provider-parity.test.ts new file mode 100644 index 0000000..e99a76f --- /dev/null +++ b/packages/agent/src/runtime-provider-parity.test.ts @@ -0,0 +1,176 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { AgentRuntimeProvider, RuntimeScope } from '@mosaicstack/types'; +import { + MatrixNativeRuntimeProvider, + type MatrixRuntimeTransport, +} from './matrix-native-runtime-provider.js'; +import { + TmuxFleetRuntimeProvider, + type FleetRuntimeTransport, +} from './tmux-fleet-runtime-provider.js'; + +const scope: RuntimeScope = { + actorId: 'operator-1', + tenantId: 'tenant-a', + channelId: 'cli', + correlationId: 'corr-1', +}; + +interface ProviderFixture { + name: string; + provider: AgentRuntimeProvider; + providerId: string; + verifySession: unknown; + send: unknown; +} + +function fixtures(): ProviderFixture[] { + const fleetTransport: FleetRuntimeTransport = { + verifySession: vi.fn(async () => ({ + id: 'session-1', + runtimeId: 'native-1', + socketName: 'fleet', + })), + listSessions: vi.fn(async () => [ + { id: 'session-1', runtimeId: 'native-1', socketName: 'fleet' }, + ]), + sendMessage: vi.fn(async () => undefined), + terminate: vi.fn(async () => undefined), + }; + const fleet = new TmuxFleetRuntimeProvider({ + transport: fleetTransport, + readAuthority: { canRead: vi.fn(async () => true) }, + writeAuthority: { + canWrite: vi.fn(async () => true), + assertAuthorized: vi.fn(async () => undefined), + }, + attachmentIdFactory: () => 'attachment-1', + now: () => new Date('2026-07-13T00:00:00.000Z'), + }); + + const matrixTransport: MatrixRuntimeTransport = { + health: vi.fn(async () => ({ + status: 'healthy' as const, + checkedAt: '2026-07-13T00:00:00.000Z', + })), + verifySession: vi.fn(async () => ({ + id: 'session-1', + runtimeId: 'native-1', + state: 'active' as const, + createdAt: '2026-07-13T00:00:00.000Z', + updatedAt: '2026-07-13T00:00:00.000Z', + })), + listSessions: vi.fn(async () => [ + { + id: 'session-1', + runtimeId: 'native-1', + state: 'active' as const, + createdAt: '2026-07-13T00:00:00.000Z', + updatedAt: '2026-07-13T00:00:00.000Z', + }, + ]), + stream: async function* () {}, + send: vi.fn(async () => undefined), + terminate: vi.fn(async () => undefined), + }; + const matrix = new MatrixNativeRuntimeProvider({ + transport: matrixTransport, + readAuthority: { canRead: vi.fn(async () => true) }, + writeAuthority: { + canWrite: vi.fn(async () => true), + assertAuthorized: vi.fn(async () => undefined), + }, + attachmentIdFactory: () => 'attachment-1', + now: () => new Date('2026-07-13T00:00:00.000Z'), + }); + + return [ + { + name: 'tmux/fleet', + provider: fleet, + providerId: 'fleet.tmux', + verifySession: fleetTransport.verifySession, + send: fleetTransport.sendMessage, + }, + { + name: 'Matrix/native', + provider: matrix, + providerId: 'runtime.matrix', + verifySession: matrixTransport.verifySession, + send: matrixTransport.send, + }, + ]; +} + +/** Shared contract tests for the migration-safe provider intersection. */ +describe('tmux/fleet and Matrix/native provider parity', (): void => { + it.each(fixtures())( + '%s exposes the shared runtime operations', + async (fixture): Promise => { + await expect(fixture.provider.capabilities(scope)).resolves.toEqual( + expect.objectContaining({ + supported: expect.arrayContaining([ + 'session.list', + 'session.tree', + 'session.send', + 'session.attach', + 'session.terminate', + ]), + }), + ); + await expect(fixture.provider.listSessions(scope)).resolves.toEqual([ + expect.objectContaining({ + id: 'session-1', + providerId: fixture.providerId, + runtimeId: 'native-1', + }), + ]); + }, + ); + + it.each(fixtures())( + '%s rejects empty messages before touching its transport', + async (fixture): Promise => { + await expect( + fixture.provider.sendMessage( + 'session-1', + { content: '', idempotencyKey: 'message-1' }, + scope, + ), + ).rejects.toMatchObject({ code: 'invalid_request' }); + expect(fixture.verifySession).not.toHaveBeenCalled(); + expect(fixture.send).not.toHaveBeenCalled(); + }, + ); + + it.each(fixtures())( + '%s rejects an empty termination approval before touching its transport', + async (fixture): Promise => { + await expect(fixture.provider.terminate('session-1', '', scope)).rejects.toMatchObject({ + code: 'invalid_request', + }); + expect(fixture.verifySession).not.toHaveBeenCalled(); + }, + ); + + it.each(fixtures())( + '%s creates a read-only handle bound to immutable scope', + async (fixture): Promise => { + await expect(fixture.provider.attach('session-1', 'control', scope)).rejects.toMatchObject({ + code: 'forbidden', + }); + expect(fixture.verifySession).not.toHaveBeenCalled(); + + await expect(fixture.provider.attach('session-1', 'read', scope)).resolves.toEqual( + expect.objectContaining({ + attachmentId: 'attachment-1', + sessionId: 'session-1', + mode: 'read', + }), + ); + await expect( + fixture.provider.detach('attachment-1', { ...scope, actorId: 'operator-2' }), + ).rejects.toMatchObject({ code: 'forbidden' }); + }, + ); +}); diff --git a/packages/mosaic/src/fleet/matrix-native-runtime-transport.test.ts b/packages/mosaic/src/fleet/matrix-native-runtime-transport.test.ts new file mode 100644 index 0000000..7b698cd --- /dev/null +++ b/packages/mosaic/src/fleet/matrix-native-runtime-transport.test.ts @@ -0,0 +1,182 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { RuntimeScope } from '@mosaicstack/types'; +import { + MatrixNativeRuntimeTransport, + type MatrixFetchLike, +} from './matrix-native-runtime-transport.js'; + +const scope: RuntimeScope = { + actorId: 'operator-1', + tenantId: 'tenant-a', + channelId: 'cli', + correlationId: 'corr-1', +}; + +const bindings = [ + { + id: 'native-1', + runtimeId: '@native-worker:example.test', + roomId: '!room:example.test', + remoteUserId: '@native-worker:example.test', + state: 'active' as const, + createdAt: '2026-07-13T00:00:00.000Z', + updatedAt: '2026-07-13T00:00:00.000Z', + }, +]; + +function response( + status: number, + body: unknown = {}, +): ReturnType extends Promise ? T : never { + return { + ok: status >= 200 && status < 300, + status, + json: async () => body, + text: async () => JSON.stringify(body), + } as never; +} + +function transport(fetchImpl: MatrixFetchLike): MatrixNativeRuntimeTransport { + return new MatrixNativeRuntimeTransport({ + homeserverUrl: 'https://matrix.example.test/base', + accessToken: 'test-token', + userId: '@mosaic:example.test', + sessions: bindings, + fetchImpl, + }); +} + +describe('MatrixNativeRuntimeTransport', (): void => { + it('uses only configured session-to-room bindings and idempotency-derived Matrix transactions', async (): Promise => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(response(200, { user_id: '@mosaic:example.test' })) + .mockResolvedValueOnce(response(200, { event_id: '$sent' })); + const matrix = transport(fetchImpl); + + await matrix.send('native-1', { content: 'hello', idempotencyKey: 'message-1' }, scope); + + expect(fetchImpl).toHaveBeenNthCalledWith( + 1, + 'https://matrix.example.test/base/_matrix/client/v3/account/whoami', + expect.objectContaining({ + headers: expect.objectContaining({ Authorization: 'Bearer test-token' }), + }), + ); + const [url, init] = fetchImpl.mock.calls[1]!; + expect(url).toMatch( + /^https:\/\/matrix\.example\.test\/base\/_matrix\/client\/v3\/rooms\/!room%3Aexample\.test\/send\/m\.room\.message\/mosaic-send-[A-Za-z0-9_-]+$/, + ); + expect(init?.method).toBe('PUT'); + expect(JSON.parse(init?.body ?? '')).toEqual({ + msgtype: 'm.text', + body: 'hello', + 'mosaic.runtime.v1': { + session_id: 'native-1', + runtime_id: '@native-worker:example.test', + actor_id: 'operator-1', + tenant_id: 'tenant-a', + channel_id: 'cli', + correlation_id: 'corr-1', + idempotency_key: 'message-1', + }, + }); + }); + + it('sends a termination only to the bound room with the exact approval reference', async (): Promise => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(response(200, { user_id: '@mosaic:example.test' })) + .mockResolvedValueOnce(response(200, { event_id: '$terminated' })); + const matrix = transport(fetchImpl); + + await matrix.terminate('native-1', 'approval-1', scope); + + const [url, init] = fetchImpl.mock.calls[1]!; + expect(url).toMatch( + /^https:\/\/matrix\.example\.test\/base\/_matrix\/client\/v3\/rooms\/!room%3Aexample\.test\/send\/mosaic\.runtime\.terminate\/mosaic-terminate-[A-Za-z0-9_-]+$/, + ); + expect(JSON.parse(init?.body ?? '')).toEqual({ + session_id: 'native-1', + runtime_id: '@native-worker:example.test', + actor_id: 'operator-1', + tenant_id: 'tenant-a', + channel_id: 'cli', + correlation_id: 'corr-1', + approval_ref: 'approval-1', + }); + }); + + it('fails closed when Matrix whoami does not match the configured native identity', async (): Promise => { + const fetchImpl = vi + .fn() + .mockResolvedValue(response(200, { user_id: '@other:example.test' })); + const matrix = transport(fetchImpl); + + await expect(matrix.listSessions(scope)).rejects.toMatchObject({ code: 'forbidden' }); + }); + + it('rejects unbound session IDs without using caller-supplied Matrix room data', async (): Promise => { + const fetchImpl = vi + .fn() + .mockResolvedValue(response(200, { user_id: '@mosaic:example.test' })); + const matrix = transport(fetchImpl); + + await expect(matrix.verifySession('!attacker-room:example.test', scope)).rejects.toMatchObject({ + code: 'not_found', + }); + }); + + it('maps replay-cursor events from only the configured remote identity', async (): Promise => { + const fetchImpl = vi + .fn() + .mockResolvedValueOnce(response(200, { user_id: '@mosaic:example.test' })) + .mockResolvedValueOnce( + response(200, { + next_batch: 'next-cursor', + rooms: { + join: { + '!room:example.test': { + timeline: { + events: [ + { + type: 'mosaic.runtime.event', + sender: '@intruder:example.test', + content: { 'mosaic.runtime.v1': { type: 'message.delta', content: 'nope' } }, + }, + { + type: 'mosaic.runtime.event', + sender: '@native-worker:example.test', + origin_server_ts: 1_784_246_400_000, + content: { + 'mosaic.runtime.v1': { + session_id: 'native-1', + type: 'message.delta', + content: 'accepted', + }, + }, + }, + ], + }, + }, + }, + }, + }), + ); + const matrix = transport(fetchImpl); + + const events = []; + for await (const event of matrix.stream('native-1', 'prior-cursor', scope)) events.push(event); + + expect(events).toEqual([ + { + type: 'message.delta', + sessionId: 'native-1', + cursor: 'next-cursor', + occurredAt: '2026-07-17T00:00:00.000Z', + content: 'accepted', + }, + ]); + expect(fetchImpl.mock.calls[1]?.[0]).toContain('since=prior-cursor'); + }); +}); diff --git a/packages/mosaic/src/fleet/matrix-native-runtime-transport.ts b/packages/mosaic/src/fleet/matrix-native-runtime-transport.ts new file mode 100644 index 0000000..c20b69b --- /dev/null +++ b/packages/mosaic/src/fleet/matrix-native-runtime-transport.ts @@ -0,0 +1,367 @@ +import { createHash } from 'node:crypto'; +import type { + RuntimeHealth, + RuntimeMessage, + RuntimeScope, + RuntimeSessionState, + RuntimeStreamEvent, +} from '@mosaicstack/types'; + +export type MatrixRuntimeTransportErrorCode = + | 'forbidden' + | 'invalid_request' + | 'not_found' + | 'unavailable'; + +export class MatrixRuntimeTransportError extends Error { + constructor( + readonly code: MatrixRuntimeTransportErrorCode, + message: string, + ) { + super(message); + this.name = MatrixRuntimeTransportError.name; + } +} + +/** Minimal injectable Matrix fetch surface; it avoids coupling this transport to an SDK. */ +export interface MatrixFetchLike { + ( + url: string, + init?: { method?: string; headers?: Record; body?: string }, + ): Promise<{ + ok: boolean; + status: number; + json(): Promise; + text(): Promise; + }>; +} + +/** A server-configured runtime session binding. Callers never select a Matrix room. */ +export interface MatrixNativeRuntimeSessionBinding { + id: string; + runtimeId: string; + roomId: string; + remoteUserId: string; + parentSessionId?: string; + state: RuntimeSessionState; + createdAt: string; + updatedAt: string; +} + +export interface MatrixNativeRuntimeTransportOptions { + homeserverUrl: string; + accessToken: string; + /** Matrix user authenticated by the service token. */ + userId: string; + sessions: readonly MatrixNativeRuntimeSessionBinding[]; + fetchImpl?: MatrixFetchLike; + now?: () => Date; +} + +interface MatrixSyncEvent { + type?: string; + sender?: string; + origin_server_ts?: number; + content?: Record; +} + +interface MatrixSyncResponse { + next_batch?: string; + rooms?: { join?: Record }; +} + +/** + * Concrete Matrix CS-API transport for the native provider. It authenticates + * the configured sender before each operation and resolves rooms exclusively + * from configured bindings, preventing client-selected room/identity routing. + */ +export class MatrixNativeRuntimeTransport { + private readonly baseUrl: string; + private readonly fetchImpl: MatrixFetchLike; + private readonly now: () => Date; + private readonly bindings: ReadonlyMap; + + constructor(private readonly options: MatrixNativeRuntimeTransportOptions) { + this.baseUrl = validatedHomeserverUrl(options.homeserverUrl); + if (!options.accessToken.trim()) { + throw new MatrixRuntimeTransportError('invalid_request', 'Matrix access token is required'); + } + if (!options.userId.trim()) { + throw new MatrixRuntimeTransportError('invalid_request', 'Matrix user identity is required'); + } + this.bindings = new Map( + options.sessions.map((binding) => [binding.id, Object.freeze({ ...binding })]), + ); + if (this.bindings.size !== options.sessions.length) { + throw new MatrixRuntimeTransportError( + 'invalid_request', + 'Matrix runtime session IDs must be unique', + ); + } + this.fetchImpl = options.fetchImpl ?? (globalThis.fetch as unknown as MatrixFetchLike); + this.now = options.now ?? (() => new Date()); + } + + async health(_scope: RuntimeScope): Promise { + try { + const versions = await this.request('/_matrix/client/versions', { method: 'GET' }); + if (!versions.ok) { + return this.healthResult('down', `versions HTTP ${versions.status}`); + } + await this.assertIdentity(); + return this.healthResult('healthy'); + } catch (error: unknown) { + return this.healthResult('down', message(error)); + } + } + + async listSessions(scope: RuntimeScope): Promise { + await this.assertIdentity(); + void scope; + return [...this.bindings.values()].map((binding) => ({ ...binding })); + } + + async verifySession( + sessionId: string, + scope: RuntimeScope, + ): Promise { + await this.assertIdentity(); + void scope; + const binding = this.bindings.get(sessionId); + if (!binding) { + throw new MatrixRuntimeTransportError( + 'not_found', + 'Matrix runtime session is not configured', + ); + } + return { ...binding }; + } + + async *stream( + sessionId: string, + cursor: string | undefined, + scope: RuntimeScope, + ): AsyncIterable { + const binding = await this.verifySession(sessionId, scope); + const query = new URLSearchParams({ timeout: '0' }); + if (cursor?.trim()) query.set('since', cursor); + const response = await this.request(`/_matrix/client/v3/sync?${query.toString()}`, { + method: 'GET', + headers: this.authHeaders(), + }); + if (!response.ok) { + throw new MatrixRuntimeTransportError( + 'unavailable', + `Matrix sync failed: HTTP ${response.status}`, + ); + } + const payload = (await response.json()) as MatrixSyncResponse; + const nextCursor = payload.next_batch?.trim() || cursor?.trim() || 'initial'; + const events = payload.rooms?.join?.[binding.roomId]?.timeline?.events ?? []; + for (const event of events) { + const normalized = normalizeEvent(event, binding, nextCursor); + if (normalized) yield normalized; + } + } + + async send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise { + const binding = await this.verifySession(sessionId, scope); + const txnId = transactionId('send', binding.id, message.idempotencyKey); + const response = await this.request( + `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/m.room.message/${encodeURIComponent(txnId)}`, + { + method: 'PUT', + headers: this.authHeaders(), + body: JSON.stringify({ + msgtype: 'm.text', + body: message.content, + 'mosaic.runtime.v1': metadata(binding, scope, message.idempotencyKey), + }), + }, + ); + if (!response.ok) { + throw new MatrixRuntimeTransportError( + 'unavailable', + `Matrix message delivery failed: HTTP ${response.status}`, + ); + } + } + + async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise { + const binding = await this.verifySession(sessionId, scope); + const txnId = transactionId('terminate', binding.id, `${approvalRef}:${scope.correlationId}`); + const response = await this.request( + `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/mosaic.runtime.terminate/${encodeURIComponent(txnId)}`, + { + method: 'PUT', + headers: this.authHeaders(), + body: JSON.stringify({ + ...metadata(binding, scope), + approval_ref: approvalRef, + }), + }, + ); + if (!response.ok) { + throw new MatrixRuntimeTransportError( + 'unavailable', + `Matrix termination delivery failed: HTTP ${response.status}`, + ); + } + } + + private async assertIdentity(): Promise { + let response: Awaited>; + try { + response = await this.request('/_matrix/client/v3/account/whoami', { + method: 'GET', + headers: this.authHeaders(), + }); + } catch (error: unknown) { + throw new MatrixRuntimeTransportError('unavailable', message(error)); + } + if (!response.ok) { + throw new MatrixRuntimeTransportError( + 'forbidden', + `Matrix whoami failed: HTTP ${response.status}`, + ); + } + const body = (await response.json()) as { user_id?: unknown }; + if (body.user_id !== this.options.userId) { + throw new MatrixRuntimeTransportError( + 'forbidden', + 'Matrix whoami identity does not match configuration', + ); + } + } + + private request( + path: string, + init: { method?: string; headers?: Record; body?: string }, + ) { + return this.fetchImpl(`${this.baseUrl}${path}`, init); + } + + private authHeaders(): Record { + return { + Authorization: `Bearer ${this.options.accessToken}`, + 'Content-Type': 'application/json', + }; + } + + private healthResult(status: RuntimeHealth['status'], detail?: string): RuntimeHealth { + return { status, checkedAt: this.now().toISOString(), ...(detail ? { detail } : {}) }; + } +} + +function metadata( + binding: MatrixNativeRuntimeSessionBinding, + scope: RuntimeScope, + idempotencyKey?: string, +): Record { + return { + session_id: binding.id, + runtime_id: binding.runtimeId, + actor_id: scope.actorId, + tenant_id: scope.tenantId, + channel_id: scope.channelId, + correlation_id: scope.correlationId, + ...(idempotencyKey ? { idempotency_key: idempotencyKey } : {}), + }; +} + +function transactionId(kind: 'send' | 'terminate', sessionId: string, identity: string): string { + return `mosaic-${kind}-${createHash('sha256').update(`${sessionId}\u0000${identity}`).digest('base64url')}`; +} + +function normalizeEvent( + event: MatrixSyncEvent, + binding: MatrixNativeRuntimeSessionBinding, + cursor: string, +): RuntimeStreamEvent | undefined { + if (event.type !== 'mosaic.runtime.event' || event.sender !== binding.remoteUserId) + return undefined; + const payload = event.content?.['mosaic.runtime.v1']; + if ( + !isRecord(payload) || + payload['session_id'] !== binding.id || + typeof payload['type'] !== 'string' + ) { + return undefined; + } + const occurredAt = + typeof payload['occurred_at'] === 'string' + ? payload['occurred_at'] + : new Date(event.origin_server_ts ?? 0).toISOString(); + const eventCursor = typeof payload['cursor'] === 'string' ? payload['cursor'] : cursor; + switch (payload['type']) { + case 'session.state': + return isState(payload['state']) + ? { + type: 'session.state', + sessionId: binding.id, + cursor: eventCursor, + occurredAt, + state: payload['state'], + } + : undefined; + case 'message.delta': + return typeof payload['content'] === 'string' + ? { + type: 'message.delta', + sessionId: binding.id, + cursor: eventCursor, + occurredAt, + content: payload['content'], + } + : undefined; + case 'message.complete': + return typeof payload['message_id'] === 'string' + ? { + type: 'message.complete', + sessionId: binding.id, + cursor: eventCursor, + occurredAt, + messageId: payload['message_id'], + } + : undefined; + default: + return undefined; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} + +function isState(value: unknown): value is RuntimeSessionState { + return ( + value === 'starting' || + value === 'active' || + value === 'idle' || + value === 'stopped' || + value === 'failed' + ); +} + +function validatedHomeserverUrl(value: string): string { + let url: URL; + try { + url = new URL(value); + } catch { + throw new MatrixRuntimeTransportError( + 'invalid_request', + 'Matrix homeserver URL must be absolute', + ); + } + if (url.protocol !== 'https:') { + throw new MatrixRuntimeTransportError( + 'invalid_request', + 'Matrix homeserver URL must use HTTPS', + ); + } + return url.toString().replace(/\/$/, ''); +} + +function message(error: unknown): string { + return error instanceof Error ? error.message : 'Matrix transport request failed'; +} diff --git a/packages/mosaic/src/index.ts b/packages/mosaic/src/index.ts index 4ff4162..99cdbf3 100644 --- a/packages/mosaic/src/index.ts +++ b/packages/mosaic/src/index.ts @@ -2,6 +2,7 @@ export const VERSION = '0.0.0'; export * from './fleet/interaction-service-profile.js'; export * from './fleet/tmux-runtime-transport.js'; +export * from './fleet/matrix-native-runtime-transport.js'; export { backgroundUpdateCheck,