import { randomUUID } from 'node:crypto'; import type { AgentRuntimeProvider, RuntimeAttachHandle, RuntimeAttachMode, RuntimeCapabilitySet, RuntimeHealth, RuntimeMessage, RuntimeScope, RuntimeSession, RuntimeSessionTree, RuntimeStreamEvent, } from '@mosaicstack/types'; const FLEET_PROVIDER_ID = 'fleet.tmux'; const ATTACHMENT_TTL_MS = 5 * 60 * 1_000; export type FleetRuntimeProviderErrorCode = | 'capability_unsupported' | 'forbidden' | 'invalid_request' | 'not_found'; /** A roster-bound target verified by the concrete fleet transport. */ export interface FleetRuntimeTarget { id: string; runtimeId: string; socketName: string; } /** * Narrow transport boundary implemented by the Mosaic tmux adapter. Keeping it * here prevents the runtime package from depending on the Mosaic CLI package. */ export interface FleetRuntimeTransport { verifySession(sessionId: string): Promise; listSessions(): Promise; sendMessage(sessionId: string, message: string, sourceLabel: string): Promise; terminate(sessionId: string): Promise; } export type FleetReadOperation = | 'runtime.health' | 'session.list' | 'session.tree' | 'session.attach'; export interface FleetReadAuthorization { operation: FleetReadOperation; scope: RuntimeScope; sessionId?: string; } /** Authorization for fleet inspection and read-only attachments. */ export interface FleetReadAuthority { canRead(authorization: FleetReadAuthorization): Promise; } export interface FleetWriteAuthorization { operation: 'session.send' | 'session.terminate'; sessionId: string; scope: RuntimeScope; /** Present only for terminate; authority adapters bind it to the exact action. */ approvalRef?: string; } /** * Mos is the only authority that may permit Tess write/control requests to a * fleet peer. Gateway records the request and denial/success around provider * invocation; the default authority prevents direct Tess writes by design. */ export interface FleetWriteAuthority { /** Non-consuming preflight used before probing the fleet transport. */ canWrite(authorization: FleetWriteAuthorization): Promise; /** Final exact-target authorization; may consume a Mos grant. */ assertAuthorized(authorization: FleetWriteAuthorization): Promise; } export interface TmuxFleetRuntimeProviderOptions { transport: FleetRuntimeTransport; readAuthority?: FleetReadAuthority; writeAuthority?: FleetWriteAuthority; sourceLabel?: string; attachmentIdFactory?: () => string; now?: () => Date; attachmentTtlMs?: number; } interface FleetAttachment { sessionId: string; scope: RuntimeScope; expiresAtMs: number; } /** A typed, fail-closed provider error that callers can normalize at the boundary. */ export class FleetRuntimeProviderError extends Error { constructor( readonly code: FleetRuntimeProviderErrorCode, message: string, ) { super(message); this.name = FleetRuntimeProviderError.name; } } class DenyFleetReadAuthority implements FleetReadAuthority { async canRead(_authorization: FleetReadAuthorization): Promise { return false; } } class DenyFleetWriteAuthority implements FleetWriteAuthority { async canWrite(_authorization: FleetWriteAuthorization): Promise { return false; } async assertAuthorized(_authorization: FleetWriteAuthorization): Promise { throw new FleetRuntimeProviderError( 'forbidden', 'Fleet writes require an explicit Mos authority decision', ); } } /** * A capability-limited provider for rostered local fleet peers. It never * permits raw tmux socket/target selection, interactive control attach, or * direct Tess writes; all side effects pass through exact transport checks. */ export class TmuxFleetRuntimeProvider implements AgentRuntimeProvider { readonly id = FLEET_PROVIDER_ID; private readonly attachments = new Map(); private readonly readAuthority: FleetReadAuthority; private readonly writeAuthority: FleetWriteAuthority; private readonly sourceLabel: string; private readonly attachmentIdFactory: () => string; private readonly now: () => Date; private readonly attachmentTtlMs: number; constructor(private readonly options: TmuxFleetRuntimeProviderOptions) { this.readAuthority = options.readAuthority ?? new DenyFleetReadAuthority(); this.writeAuthority = options.writeAuthority ?? new DenyFleetWriteAuthority(); this.sourceLabel = options.sourceLabel ?? 'tess'; 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.send', 'session.attach', 'session.terminate', ], }; } async health(scope: RuntimeScope): Promise { const targets = await this.readTargets('runtime.health', scope); return { status: targets.length > 0 ? 'healthy' : 'down', checkedAt: this.now().toISOString(), detail: targets.length > 0 ? 'Authorized rostered fleet peers are reachable' : 'No authorized rostered fleet peers are reachable', }; } async listSessions(scope: RuntimeScope): Promise { const targets = await this.readTargets('session.list', scope); return this.toRuntimeSessions(targets); } async getSessionTree(scope: RuntimeScope): Promise { const targets = await this.readTargets('session.tree', scope); return this.toRuntimeSessions(targets).map( (session): RuntimeSessionTree => ({ session, children: [], }), ); } async *streamSession( _sessionId: string, _cursor: string | undefined, _scope: RuntimeScope, ): AsyncIterable { throw new FleetRuntimeProviderError( 'capability_unsupported', 'Fleet session streaming is not supported by the tmux provider', ); } async sendMessage( sessionId: string, message: RuntimeMessage, scope: RuntimeScope, ): Promise { if (message.content.length === 0) { throw new FleetRuntimeProviderError('invalid_request', 'Fleet message content is required'); } await this.assertWritePermitted('session.send', sessionId, scope); const target = await this.options.transport.verifySession(sessionId); await this.assertWriteAuthorized('session.send', target.id, scope); await this.options.transport.sendMessage(target.id, message.content, this.sourceLabel); } async attach( sessionId: string, mode: RuntimeAttachMode, scope: RuntimeScope, ): Promise { if (mode !== 'read') { throw new FleetRuntimeProviderError('forbidden', 'Fleet control attach is not permitted'); } await this.assertReadAuthorized('session.attach', sessionId, scope); const target = await this.options.transport.verifySession(sessionId); await this.assertReadAuthorized('session.attach', target.id, scope); const attachmentId = this.attachmentIdFactory(); const nowMs = this.now().getTime(); this.pruneExpiredAttachments(nowMs); const expiresAtMs = nowMs + this.attachmentTtlMs; this.attachments.set(attachmentId, { sessionId: target.id, scope: snapshotScope(scope), expiresAtMs, }); return { attachmentId, sessionId: target.id, mode, expiresAt: new Date(expiresAtMs).toISOString(), }; } async detach(attachmentId: string, scope: RuntimeScope): Promise { const attachment = this.attachments.get(attachmentId); if (!attachment) { throw new FleetRuntimeProviderError('not_found', 'Fleet attachment is not active'); } if (this.now().getTime() >= attachment.expiresAtMs) { this.attachments.delete(attachmentId); throw new FleetRuntimeProviderError('forbidden', 'Fleet attachment has expired'); } if (!sameScope(attachment.scope, scope)) { throw new FleetRuntimeProviderError('forbidden', 'Fleet attachment scope does not match'); } this.attachments.delete(attachmentId); } async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise { if (approvalRef.trim().length === 0) { throw new FleetRuntimeProviderError( 'invalid_request', 'Fleet termination approval is required', ); } await this.assertWritePermitted('session.terminate', sessionId, scope, approvalRef); const target = await this.options.transport.verifySession(sessionId); await this.assertWriteAuthorized('session.terminate', target.id, scope, approvalRef); await this.options.transport.terminate(target.id); } private async readTargets( operation: FleetReadOperation, scope: RuntimeScope, ): Promise { await this.assertReadAuthorized(operation, undefined, scope); const targets = await this.options.transport.listSessions(); const authorization = await Promise.all( targets.map( async (target): Promise => this.readAuthority.canRead({ operation, sessionId: target.id, scope }), ), ); return targets.filter((_target, index): boolean => authorization[index] === true); } private toRuntimeSessions(targets: FleetRuntimeTarget[]): RuntimeSession[] { const timestamp = this.now().toISOString(); return targets.map( (target): RuntimeSession => ({ id: target.id, providerId: this.id, runtimeId: target.runtimeId, state: 'active', createdAt: timestamp, updatedAt: timestamp, }), ); } private async assertReadAuthorized( operation: FleetReadOperation, sessionId: string | undefined, scope: RuntimeScope, ): Promise { const allowed = await this.readAuthority.canRead({ operation, scope, ...(sessionId ? { sessionId } : {}), }); if (!allowed) { throw new FleetRuntimeProviderError('forbidden', 'Fleet read is not authorized'); } } private pruneExpiredAttachments(nowMs: number): void { for (const [attachmentId, attachment] of this.attachments) { if (attachment.expiresAtMs <= nowMs) { this.attachments.delete(attachmentId); } } } private async assertWritePermitted( operation: FleetWriteAuthorization['operation'], sessionId: string, scope: RuntimeScope, approvalRef?: string, ): Promise { const permitted = await this.writeAuthority.canWrite({ operation, sessionId, scope, ...(approvalRef ? { approvalRef } : {}), }); if (!permitted) { throw new FleetRuntimeProviderError('forbidden', 'Fleet write is not authorized'); } } private async assertWriteAuthorized( operation: FleetWriteAuthorization['operation'], sessionId: string, scope: RuntimeScope, approvalRef?: string, ): Promise { await this.writeAuthority.assertAuthorized({ operation, sessionId, scope, ...(approvalRef ? { approvalRef } : {}), }); } } function snapshotScope(scope: RuntimeScope): RuntimeScope { return Object.freeze({ actorId: scope.actorId, tenantId: scope.tenantId, channelId: scope.channelId, correlationId: scope.correlationId, }); } function sameScope(left: RuntimeScope, right: RuntimeScope): boolean { return ( left.actorId === right.actorId && left.tenantId === right.tenantId && left.channelId === right.channelId && left.correlationId === right.correlationId ); }