feat(agent): add Matrix native runtime provider #744

Merged
jason.woltje merged 1 commits from feat/tess-matrix-provider into main 2026-07-13 15:29:38 +00:00
7 changed files with 1231 additions and 0 deletions

View File

@@ -3,4 +3,5 @@ export const VERSION = '0.0.0';
export * from './runtime-provider-registry.js'; export * from './runtime-provider-registry.js';
export * from './tmux-fleet-runtime-provider.js'; export * from './tmux-fleet-runtime-provider.js';
export * from './hermes-runtime-provider.js'; export * from './hermes-runtime-provider.js';
export * from './matrix-native-runtime-provider.js';
export * from './tess-durable-session.js'; export * from './tess-durable-session.js';

View File

@@ -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<RuntimeStreamEvent> {
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<void> => {
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<void> => {
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<void> => {
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<void> => {
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<RuntimeStreamEvent>): Promise<RuntimeStreamEvent[]> {
const values: RuntimeStreamEvent[] = [];
for await (const value of stream) values.push(value);
return values;
}

View File

@@ -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<RuntimeHealth>;
listSessions(scope: RuntimeScope): Promise<MatrixRuntimeSession[]>;
verifySession(sessionId: string, scope: RuntimeScope): Promise<MatrixRuntimeSession>;
stream(
sessionId: string,
cursor: string | undefined,
scope: RuntimeScope,
): AsyncIterable<RuntimeStreamEvent>;
send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise<void>;
terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise<void>;
}
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<boolean>;
}
export interface MatrixWriteAuthority {
canWrite(input: {
operation: 'session.send' | 'session.terminate';
sessionId: string;
scope: RuntimeScope;
approvalRef?: string;
}): Promise<boolean>;
assertAuthorized(input: {
operation: 'session.send' | 'session.terminate';
sessionId: string;
scope: RuntimeScope;
approvalRef?: string;
}): Promise<void>;
}
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<boolean> {
return false;
}
}
class DenyMatrixWriteAuthority implements MatrixWriteAuthority {
async canWrite(): Promise<boolean> {
return false;
}
async assertAuthorized(): Promise<void> {
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<string, Attachment>();
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<RuntimeCapabilitySet> {
return {
supported: [
'session.list',
'session.tree',
'session.stream',
'session.send',
'session.attach',
'session.terminate',
],
};
}
async health(scope: RuntimeScope): Promise<RuntimeHealth> {
await this.assertRead('runtime.health', undefined, scope);
return this.options.transport.health(scope);
}
async listSessions(scope: RuntimeScope): Promise<RuntimeSession[]> {
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<RuntimeSessionTree[]> {
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<string, RuntimeSessionTree>(
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<RuntimeStreamEvent> {
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<void> {
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<RuntimeAttachHandle> {
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<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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
);
}

View File

@@ -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<void> => {
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<void> => {
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<void> => {
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<void> => {
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' });
},
);
});

View File

@@ -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<MatrixFetchLike> extends Promise<infer T> ? 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<void> => {
const fetchImpl = vi
.fn<MatrixFetchLike>()
.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<void> => {
const fetchImpl = vi
.fn<MatrixFetchLike>()
.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<void> => {
const fetchImpl = vi
.fn<MatrixFetchLike>()
.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<void> => {
const fetchImpl = vi
.fn<MatrixFetchLike>()
.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<void> => {
const fetchImpl = vi
.fn<MatrixFetchLike>()
.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');
});
});

View File

@@ -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<string, string>; body?: string },
): Promise<{
ok: boolean;
status: number;
json(): Promise<unknown>;
text(): Promise<string>;
}>;
}
/** 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<string, unknown>;
}
interface MatrixSyncResponse {
next_batch?: string;
rooms?: { join?: Record<string, { timeline?: { events?: MatrixSyncEvent[] } }> };
}
/**
* 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<string, MatrixNativeRuntimeSessionBinding>;
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<RuntimeHealth> {
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<MatrixNativeRuntimeSessionBinding[]> {
await this.assertIdentity();
void scope;
return [...this.bindings.values()].map((binding) => ({ ...binding }));
}
async verifySession(
sessionId: string,
scope: RuntimeScope,
): Promise<MatrixNativeRuntimeSessionBinding> {
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<RuntimeStreamEvent> {
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<void> {
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<void> {
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<void> {
let response: Awaited<ReturnType<MatrixFetchLike>>;
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<string, string>; body?: string },
) {
return this.fetchImpl(`${this.baseUrl}${path}`, init);
}
private authHeaders(): Record<string, string> {
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<string, string> {
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<string, unknown> {
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';
}

View File

@@ -2,6 +2,7 @@ export const VERSION = '0.0.0';
export * from './fleet/interaction-service-profile.js'; export * from './fleet/interaction-service-profile.js';
export * from './fleet/tmux-runtime-transport.js'; export * from './fleet/tmux-runtime-transport.js';
export * from './fleet/matrix-native-runtime-transport.js';
export { export {
backgroundUpdateCheck, backgroundUpdateCheck,