feat(tess): add roster-bound tmux fleet provider
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
export const VERSION = '0.0.0';
|
||||
|
||||
export * from './runtime-provider-registry.js';
|
||||
export * from './tmux-fleet-runtime-provider.js';
|
||||
|
||||
261
packages/agent/src/tmux-fleet-runtime-provider.test.ts
Normal file
261
packages/agent/src/tmux-fleet-runtime-provider.test.ts
Normal file
@@ -0,0 +1,261 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import type { RuntimeScope } from '@mosaicstack/types';
|
||||
import {
|
||||
type FleetReadAuthority,
|
||||
type FleetWriteAuthority,
|
||||
TmuxFleetRuntimeProvider,
|
||||
} from './tmux-fleet-runtime-provider.js';
|
||||
import type {
|
||||
FleetRuntimeProviderError,
|
||||
FleetRuntimeTarget,
|
||||
FleetRuntimeTransport,
|
||||
} from './tmux-fleet-runtime-provider.js';
|
||||
|
||||
const scope: RuntimeScope = {
|
||||
actorId: 'operator-1',
|
||||
tenantId: 'tenant-a',
|
||||
channelId: 'discord-1',
|
||||
correlationId: 'corr-1',
|
||||
};
|
||||
|
||||
const target: FleetRuntimeTarget = {
|
||||
id: 'coder0',
|
||||
runtimeId: 'codex',
|
||||
socketName: 'tess-fleet',
|
||||
};
|
||||
|
||||
function transport(): FleetRuntimeTransport {
|
||||
return {
|
||||
verifySession: vi.fn(async (): Promise<FleetRuntimeTarget> => target),
|
||||
listSessions: vi.fn(async (): Promise<FleetRuntimeTarget[]> => [target]),
|
||||
sendMessage: vi.fn(async (): Promise<void> => undefined),
|
||||
terminate: vi.fn(async (): Promise<void> => undefined),
|
||||
};
|
||||
}
|
||||
|
||||
function readAuthority(): FleetReadAuthority {
|
||||
return { canRead: vi.fn(async (): Promise<boolean> => true) };
|
||||
}
|
||||
|
||||
describe('TmuxFleetRuntimeProvider security policy', (): void => {
|
||||
it('advertises only fleet operations it can safely implement', async (): Promise<void> => {
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: transport() });
|
||||
|
||||
await expect(provider.capabilities(scope)).resolves.toEqual({
|
||||
supported: [
|
||||
'session.list',
|
||||
'session.tree',
|
||||
'session.send',
|
||||
'session.attach',
|
||||
'session.terminate',
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects control attach without consulting the tmux transport', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet });
|
||||
|
||||
await expect(provider.attach('coder0', 'control', scope)).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
expect(fleet.verifySession).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('denies fleet listing without an exact-scope read authority decision', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet });
|
||||
|
||||
await expect(provider.listSessions(scope)).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
expect(fleet.listSessions).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('denies read attachment without an exact-scope read authority decision', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet });
|
||||
|
||||
await expect(provider.attach('coder0', 'read', scope)).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
expect(fleet.verifySession).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates a read-only attachment only after exact target verification', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: fleet,
|
||||
readAuthority: readAuthority(),
|
||||
attachmentIdFactory: (): string => 'attachment-1',
|
||||
now: (): Date => new Date('2026-07-12T00:00:00.000Z'),
|
||||
});
|
||||
|
||||
await expect(provider.attach('coder0', 'read', scope)).resolves.toEqual({
|
||||
attachmentId: 'attachment-1',
|
||||
sessionId: 'coder0',
|
||||
mode: 'read',
|
||||
expiresAt: '2026-07-12T00:05:00.000Z',
|
||||
});
|
||||
expect(fleet.verifySession).toHaveBeenCalledWith('coder0');
|
||||
});
|
||||
|
||||
it('denies attachment-handle replay from another immutable actor scope', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: fleet,
|
||||
readAuthority: readAuthority(),
|
||||
attachmentIdFactory: (): string => 'attachment-1',
|
||||
now: (): Date => new Date('2026-07-12T00:00:00.000Z'),
|
||||
});
|
||||
await provider.attach('coder0', 'read', scope);
|
||||
|
||||
await expect(
|
||||
provider.detach('attachment-1', { ...scope, actorId: 'operator-2' }),
|
||||
).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
});
|
||||
|
||||
it('keeps attachment scope immutable after caller-side scope mutation', async (): Promise<void> => {
|
||||
const mutableScope = { ...scope };
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: transport(),
|
||||
readAuthority: readAuthority(),
|
||||
attachmentIdFactory: (): string => 'attachment-1',
|
||||
now: (): Date => new Date('2026-07-12T00:00:00.000Z'),
|
||||
});
|
||||
await provider.attach('coder0', 'read', mutableScope);
|
||||
mutableScope.actorId = 'operator-2';
|
||||
|
||||
await expect(provider.detach('attachment-1', scope)).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it('denies expired attachment handles and removes them', async (): Promise<void> => {
|
||||
let now = new Date('2026-07-12T00:00:00.000Z');
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: transport(),
|
||||
readAuthority: readAuthority(),
|
||||
attachmentIdFactory: (): string => 'attachment-1',
|
||||
now: (): Date => now,
|
||||
});
|
||||
await provider.attach('coder0', 'read', scope);
|
||||
now = new Date('2026-07-12T00:05:00.001Z');
|
||||
|
||||
await expect(provider.detach('attachment-1', scope)).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
await expect(provider.detach('attachment-1', scope)).rejects.toMatchObject({
|
||||
code: 'not_found',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
});
|
||||
|
||||
it('prunes expired attachment handles before creating a new handle', async (): Promise<void> => {
|
||||
let now = new Date('2026-07-12T00:00:00.000Z');
|
||||
let attachmentSequence = 0;
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: transport(),
|
||||
readAuthority: readAuthority(),
|
||||
attachmentIdFactory: (): string => `attachment-${++attachmentSequence}`,
|
||||
now: (): Date => now,
|
||||
});
|
||||
await provider.attach('coder0', 'read', scope);
|
||||
now = new Date('2026-07-12T00:05:00.001Z');
|
||||
await provider.attach('coder0', 'read', scope);
|
||||
|
||||
await expect(provider.detach('attachment-1', scope)).rejects.toMatchObject({
|
||||
code: 'not_found',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
});
|
||||
|
||||
it('rejects an empty message before consulting write authority or tmux', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const writeAuthority: FleetWriteAuthority = {
|
||||
canWrite: vi.fn(async (): Promise<boolean> => true),
|
||||
assertAuthorized: vi.fn(async (): Promise<void> => undefined),
|
||||
};
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet, writeAuthority });
|
||||
|
||||
await expect(
|
||||
provider.sendMessage('coder0', { content: '', idempotencyKey: 'message-1' }, scope),
|
||||
).rejects.toMatchObject({
|
||||
code: 'invalid_request',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
expect(writeAuthority.canWrite).not.toHaveBeenCalled();
|
||||
expect(writeAuthority.assertAuthorized).not.toHaveBeenCalled();
|
||||
expect(fleet.verifySession).not.toHaveBeenCalled();
|
||||
expect(fleet.sendMessage).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('denies fleet writes by default before probing the tmux transport', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet });
|
||||
|
||||
await expect(
|
||||
provider.sendMessage('coder0', { content: 'hello', idempotencyKey: 'message-1' }, scope),
|
||||
).rejects.toMatchObject({ code: 'forbidden' } satisfies Partial<FleetRuntimeProviderError>);
|
||||
await expect(provider.terminate('coder0', 'approval-1', scope)).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
expect(fleet.verifySession).not.toHaveBeenCalled();
|
||||
expect(fleet.sendMessage).not.toHaveBeenCalled();
|
||||
expect(fleet.terminate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects an unverified target before consulting Mos write authority', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
fleet.verifySession = vi.fn(async (): Promise<FleetRuntimeTarget> => {
|
||||
throw new Error('target identity mismatch');
|
||||
});
|
||||
const writeAuthority: FleetWriteAuthority = {
|
||||
canWrite: vi.fn(async (): Promise<boolean> => true),
|
||||
assertAuthorized: vi.fn(async (): Promise<void> => undefined),
|
||||
};
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: fleet, writeAuthority });
|
||||
|
||||
await expect(
|
||||
provider.sendMessage('coder', { content: 'hello', idempotencyKey: 'message-1' }, scope),
|
||||
).rejects.toThrow('target identity mismatch');
|
||||
expect(writeAuthority.assertAuthorized).not.toHaveBeenCalled();
|
||||
expect(fleet.sendMessage).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('passes an exact session ID to the fleet transport only through authorized Mos writes', async (): Promise<void> => {
|
||||
const fleet = transport();
|
||||
const writeAuthority: FleetWriteAuthority = {
|
||||
canWrite: vi.fn(async (): Promise<boolean> => true),
|
||||
assertAuthorized: vi.fn(async (): Promise<void> => undefined),
|
||||
};
|
||||
const provider = new TmuxFleetRuntimeProvider({
|
||||
transport: fleet,
|
||||
sourceLabel: 'tess',
|
||||
writeAuthority,
|
||||
});
|
||||
|
||||
await provider.sendMessage('coder0', { content: 'hello', idempotencyKey: 'message-1' }, scope);
|
||||
await provider.terminate('coder0', 'approval-1', scope);
|
||||
|
||||
expect(writeAuthority.assertAuthorized).toHaveBeenCalledWith({
|
||||
operation: 'session.send',
|
||||
sessionId: 'coder0',
|
||||
scope,
|
||||
});
|
||||
expect(writeAuthority.assertAuthorized).toHaveBeenCalledWith({
|
||||
operation: 'session.terminate',
|
||||
sessionId: 'coder0',
|
||||
scope,
|
||||
approvalRef: 'approval-1',
|
||||
});
|
||||
expect(fleet.sendMessage).toHaveBeenCalledWith('coder0', 'hello', 'tess');
|
||||
expect(fleet.terminate).toHaveBeenCalledWith('coder0');
|
||||
});
|
||||
|
||||
it('fails closed when consumers ask for session streaming', async (): Promise<void> => {
|
||||
const provider = new TmuxFleetRuntimeProvider({ transport: transport() });
|
||||
const stream = provider.streamSession('coder0', undefined, scope)[Symbol.asyncIterator]();
|
||||
|
||||
await expect(stream.next()).rejects.toMatchObject({
|
||||
code: 'capability_unsupported',
|
||||
} satisfies Partial<FleetRuntimeProviderError>);
|
||||
});
|
||||
});
|
||||
368
packages/agent/src/tmux-fleet-runtime-provider.ts
Normal file
368
packages/agent/src/tmux-fleet-runtime-provider.ts
Normal file
@@ -0,0 +1,368 @@
|
||||
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<FleetRuntimeTarget>;
|
||||
listSessions(): Promise<FleetRuntimeTarget[]>;
|
||||
sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void>;
|
||||
terminate(sessionId: string): Promise<void>;
|
||||
}
|
||||
|
||||
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<boolean>;
|
||||
}
|
||||
|
||||
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<boolean>;
|
||||
/** Final exact-target authorization; may consume a Mos grant. */
|
||||
assertAuthorized(authorization: FleetWriteAuthorization): Promise<void>;
|
||||
}
|
||||
|
||||
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<boolean> {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
class DenyFleetWriteAuthority implements FleetWriteAuthority {
|
||||
async canWrite(_authorization: FleetWriteAuthorization): Promise<boolean> {
|
||||
return false;
|
||||
}
|
||||
|
||||
async assertAuthorized(_authorization: FleetWriteAuthorization): Promise<void> {
|
||||
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<string, FleetAttachment>();
|
||||
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<RuntimeCapabilitySet> {
|
||||
return {
|
||||
supported: [
|
||||
'session.list',
|
||||
'session.tree',
|
||||
'session.send',
|
||||
'session.attach',
|
||||
'session.terminate',
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
async health(scope: RuntimeScope): Promise<RuntimeHealth> {
|
||||
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<RuntimeSession[]> {
|
||||
const targets = await this.readTargets('session.list', scope);
|
||||
return this.toRuntimeSessions(targets);
|
||||
}
|
||||
|
||||
async getSessionTree(scope: RuntimeScope): Promise<RuntimeSessionTree[]> {
|
||||
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<RuntimeStreamEvent> {
|
||||
throw new FleetRuntimeProviderError(
|
||||
'capability_unsupported',
|
||||
'Fleet session streaming is not supported by the tmux provider',
|
||||
);
|
||||
}
|
||||
|
||||
async sendMessage(
|
||||
sessionId: string,
|
||||
message: RuntimeMessage,
|
||||
scope: RuntimeScope,
|
||||
): Promise<void> {
|
||||
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<RuntimeAttachHandle> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<FleetRuntimeTarget[]> {
|
||||
await this.assertReadAuthorized(operation, undefined, scope);
|
||||
const targets = await this.options.transport.listSessions();
|
||||
const authorization = await Promise.all(
|
||||
targets.map(
|
||||
async (target): Promise<boolean> =>
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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
|
||||
);
|
||||
}
|
||||
160
packages/mosaic/src/fleet/tmux-runtime-transport.test.ts
Normal file
160
packages/mosaic/src/fleet/tmux-runtime-transport.test.ts
Normal file
@@ -0,0 +1,160 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import type { CommandResult, CommandRunner, FleetRoster } from '../commands/fleet.js';
|
||||
import { FleetTmuxRuntimeTransport } from './tmux-runtime-transport.js';
|
||||
import type { FleetRuntimeTransportError } from './tmux-runtime-transport.js';
|
||||
|
||||
const roster: FleetRoster = {
|
||||
version: 1,
|
||||
transport: 'tmux',
|
||||
tmux: { socketName: 'tess-fleet', holderSession: '_holder' },
|
||||
defaults: { workingDirectory: '~/src' },
|
||||
runtimes: { codex: { resetCommand: '/clear' } },
|
||||
agents: [{ name: 'coder0', runtime: 'codex', className: 'code' }],
|
||||
};
|
||||
|
||||
function commandResult(stdout = '', exitCode = 0, stderr = ''): CommandResult {
|
||||
return { stdout, stderr, exitCode };
|
||||
}
|
||||
|
||||
describe('FleetTmuxRuntimeTransport security boundary', (): void => {
|
||||
it('rejects a prefix target before it invokes tmux', async (): Promise<void> => {
|
||||
const runner = vi.fn<CommandRunner>(async (): Promise<CommandResult> => commandResult());
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await expect(transport.verifySession('coder')).rejects.toMatchObject({
|
||||
code: 'not_found',
|
||||
} satisfies Partial<FleetRuntimeTransportError>);
|
||||
expect(runner).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('uses the roster socket and exact pane target while verifying peer identity', async (): Promise<void> => {
|
||||
const runner = vi.fn<CommandRunner>(
|
||||
async (): Promise<CommandResult> => commandResult('111 codex 0 0 0 0\n'),
|
||||
);
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await expect(transport.verifySession('coder0')).resolves.toEqual({
|
||||
id: 'coder0',
|
||||
runtimeId: 'codex',
|
||||
socketName: 'tess-fleet',
|
||||
});
|
||||
expect(runner).toHaveBeenCalledWith('tmux', [
|
||||
'-L',
|
||||
'tess-fleet',
|
||||
'list-panes',
|
||||
'-t',
|
||||
'=coder0:0.0',
|
||||
'-F',
|
||||
'#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}',
|
||||
]);
|
||||
});
|
||||
|
||||
it('denies a live pane whose runtime does not match the roster identity', async (): Promise<void> => {
|
||||
const runner = vi.fn<CommandRunner>(
|
||||
async (): Promise<CommandResult> => commandResult('111 python3 0 0 0 0\n'),
|
||||
);
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await expect(transport.verifySession('coder0')).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeTransportError>);
|
||||
expect(runner).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('surfaces a runtime identity mismatch instead of hiding it from fleet status', async (): Promise<void> => {
|
||||
const rosterWithMismatchedPeer: FleetRoster = {
|
||||
...roster,
|
||||
agents: [...roster.agents, { name: 'coder1', runtime: 'codex', className: 'code' }],
|
||||
};
|
||||
const runner = vi.fn<CommandRunner>(
|
||||
async (_command: string, args: string[]): Promise<CommandResult> =>
|
||||
commandResult(
|
||||
args.includes('=coder1:0.0') ? '222 python3 0 0 0 0\n' : '111 codex 0 0 0 0\n',
|
||||
),
|
||||
);
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => rosterWithMismatchedPeer,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await expect(transport.listSessions()).rejects.toMatchObject({
|
||||
code: 'forbidden',
|
||||
} satisfies Partial<FleetRuntimeTransportError>);
|
||||
});
|
||||
|
||||
it('denies sending when the roster socket does not contain the exact target', async (): Promise<void> => {
|
||||
const runner = vi.fn<CommandRunner>(
|
||||
async (): Promise<CommandResult> => commandResult('', 1, "can't find session: coder0"),
|
||||
);
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await expect(transport.sendMessage('coder0', 'hello', 'tess')).rejects.toMatchObject({
|
||||
code: 'unavailable',
|
||||
} satisfies Partial<FleetRuntimeTransportError>);
|
||||
expect(runner).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('sends only through the maintained sender after exact target and identity verification', async (): Promise<void> => {
|
||||
const runner = vi
|
||||
.fn<CommandRunner>()
|
||||
.mockResolvedValueOnce(commandResult('111 codex 0 0 0 0\n'))
|
||||
.mockResolvedValueOnce(commandResult());
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await transport.sendMessage('coder0', 'hello', 'tess');
|
||||
|
||||
expect(runner).toHaveBeenNthCalledWith(2, '/mosaic/tools/tmux/agent-send.sh', [
|
||||
'-L',
|
||||
'tess-fleet',
|
||||
'-S',
|
||||
'tess',
|
||||
'-s',
|
||||
'coder0',
|
||||
'-m',
|
||||
'hello',
|
||||
]);
|
||||
});
|
||||
|
||||
it('terminates only the exact roster target after identity verification', async (): Promise<void> => {
|
||||
const runner = vi
|
||||
.fn<CommandRunner>()
|
||||
.mockResolvedValueOnce(commandResult('111 codex 0 0 0 0\n'))
|
||||
.mockResolvedValueOnce(commandResult());
|
||||
const transport = new FleetTmuxRuntimeTransport({
|
||||
rosterLoader: async (): Promise<FleetRoster> => roster,
|
||||
runner,
|
||||
mosaicHome: '/mosaic',
|
||||
});
|
||||
|
||||
await transport.terminate('coder0');
|
||||
|
||||
expect(runner).toHaveBeenNthCalledWith(2, 'tmux', [
|
||||
'-L',
|
||||
'tess-fleet',
|
||||
'kill-session',
|
||||
'-t',
|
||||
'=coder0',
|
||||
]);
|
||||
});
|
||||
});
|
||||
200
packages/mosaic/src/fleet/tmux-runtime-transport.ts
Normal file
200
packages/mosaic/src/fleet/tmux-runtime-transport.ts
Normal file
@@ -0,0 +1,200 @@
|
||||
import {
|
||||
buildAgentSendCommand,
|
||||
buildTmuxListPanesCommand,
|
||||
getRosterAgent,
|
||||
loadFleetRoster,
|
||||
parseTmuxListPanes,
|
||||
resolveFleetPaths,
|
||||
RUNTIME_ACCEPTABLE_COMMANDS,
|
||||
socketArgs,
|
||||
type CommandResult,
|
||||
type CommandRunner,
|
||||
type FleetRoster,
|
||||
} from '../commands/fleet.js';
|
||||
|
||||
export type FleetRuntimeTransportErrorCode =
|
||||
| 'forbidden'
|
||||
| 'invalid_request'
|
||||
| 'not_found'
|
||||
| 'unavailable';
|
||||
|
||||
/** A roster-bound, verified tmux target. It never accepts a caller-selected socket. */
|
||||
export interface FleetRuntimeTarget {
|
||||
id: string;
|
||||
runtimeId: string;
|
||||
socketName: string;
|
||||
}
|
||||
|
||||
/** Narrow transport boundary consumed by the runtime provider. */
|
||||
export interface FleetRuntimeTransport {
|
||||
verifySession(sessionId: string): Promise<FleetRuntimeTarget>;
|
||||
listSessions(): Promise<FleetRuntimeTarget[]>;
|
||||
sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void>;
|
||||
terminate(sessionId: string): Promise<void>;
|
||||
}
|
||||
|
||||
export interface FleetTmuxRuntimeTransportOptions {
|
||||
mosaicHome: string;
|
||||
rosterPath?: string;
|
||||
rosterLoader?: () => Promise<FleetRoster>;
|
||||
runner: CommandRunner;
|
||||
}
|
||||
|
||||
/** A typed, fail-closed error for roster, target, identity, and tmux failures. */
|
||||
export class FleetRuntimeTransportError extends Error {
|
||||
constructor(
|
||||
readonly code: FleetRuntimeTransportErrorCode,
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = FleetRuntimeTransportError.name;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Roster-bound transport for the local fleet tmux server. Every side-effecting
|
||||
* operation verifies an exact roster name, the roster socket, and the runtime
|
||||
* command in the exact pane before it invokes the maintained sender or tmux.
|
||||
*/
|
||||
export class FleetTmuxRuntimeTransport implements FleetRuntimeTransport {
|
||||
private readonly rosterLoader: () => Promise<FleetRoster>;
|
||||
|
||||
constructor(private readonly options: FleetTmuxRuntimeTransportOptions) {
|
||||
this.rosterLoader =
|
||||
options.rosterLoader ??
|
||||
(() =>
|
||||
loadFleetRoster(options.rosterPath ?? resolveFleetPaths(options.mosaicHome).rosterPath));
|
||||
}
|
||||
|
||||
async verifySession(sessionId: string): Promise<FleetRuntimeTarget> {
|
||||
const roster = await this.loadRoster();
|
||||
return this.verifyRosterSession(roster, sessionId);
|
||||
}
|
||||
|
||||
async listSessions(): Promise<FleetRuntimeTarget[]> {
|
||||
const roster = await this.loadRoster();
|
||||
const targets = await Promise.all(
|
||||
roster.agents.map(async (agent): Promise<FleetRuntimeTarget | undefined> => {
|
||||
try {
|
||||
return await this.verifyRosterSession(roster, agent.name);
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof FleetRuntimeTransportError && error.code === 'unavailable') {
|
||||
return undefined;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}),
|
||||
);
|
||||
return targets.filter((target): target is FleetRuntimeTarget => target !== undefined);
|
||||
}
|
||||
|
||||
async sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void> {
|
||||
if (message.length === 0) {
|
||||
throw new FleetRuntimeTransportError('invalid_request', 'Fleet message content is required');
|
||||
}
|
||||
if (sourceLabel.length === 0) {
|
||||
throw new FleetRuntimeTransportError(
|
||||
'invalid_request',
|
||||
'Fleet message source label is required',
|
||||
);
|
||||
}
|
||||
const target = await this.verifySession(sessionId);
|
||||
const command = buildAgentSendCommand(
|
||||
resolveFleetPaths(this.options.mosaicHome),
|
||||
target.id,
|
||||
message,
|
||||
target.socketName,
|
||||
sourceLabel,
|
||||
);
|
||||
await this.runChecked(command, 'Fleet message delivery failed');
|
||||
}
|
||||
|
||||
async terminate(sessionId: string): Promise<void> {
|
||||
const target = await this.verifySession(sessionId);
|
||||
await this.runChecked(
|
||||
['tmux', ...socketArgs(target.socketName), 'kill-session', '-t', `=${target.id}`],
|
||||
'Fleet session termination failed',
|
||||
);
|
||||
}
|
||||
|
||||
private async loadRoster(): Promise<FleetRoster> {
|
||||
try {
|
||||
return await this.rosterLoader();
|
||||
} catch (error: unknown) {
|
||||
const message = error instanceof Error ? error.message : 'Fleet roster is unavailable';
|
||||
throw new FleetRuntimeTransportError(
|
||||
'unavailable',
|
||||
`Fleet roster is unavailable: ${message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async verifyRosterSession(
|
||||
roster: FleetRoster,
|
||||
sessionId: string,
|
||||
): Promise<FleetRuntimeTarget> {
|
||||
const agent = this.exactRosterAgent(roster, sessionId);
|
||||
const socketName = roster.tmux.socketName;
|
||||
const result = await this.run(buildTmuxListPanesCommand(agent.name, socketName));
|
||||
if (result.exitCode !== 0) {
|
||||
throw new FleetRuntimeTransportError(
|
||||
'unavailable',
|
||||
`Fleet session is unavailable: ${agent.name}`,
|
||||
);
|
||||
}
|
||||
const pane = parseTmuxListPanes(result.stdout);
|
||||
if (pane.dead || pane.command === null) {
|
||||
throw new FleetRuntimeTransportError(
|
||||
'unavailable',
|
||||
`Fleet session is unavailable: ${agent.name}`,
|
||||
);
|
||||
}
|
||||
if (!hasExactRuntimeIdentity(agent.runtime, pane.command)) {
|
||||
throw new FleetRuntimeTransportError(
|
||||
'forbidden',
|
||||
`Fleet runtime identity mismatch: ${agent.name}`,
|
||||
);
|
||||
}
|
||||
return { id: agent.name, runtimeId: agent.runtime, socketName };
|
||||
}
|
||||
|
||||
private exactRosterAgent(roster: FleetRoster, sessionId: string): FleetRoster['agents'][number] {
|
||||
try {
|
||||
const agent = getRosterAgent(roster, sessionId);
|
||||
if (agent.name !== sessionId) {
|
||||
throw new FleetRuntimeTransportError('not_found', 'Fleet session target must be exact');
|
||||
}
|
||||
return agent;
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof FleetRuntimeTransportError) {
|
||||
throw error;
|
||||
}
|
||||
throw new FleetRuntimeTransportError('not_found', 'Fleet session target is not roster-bound');
|
||||
}
|
||||
}
|
||||
|
||||
private async run(command: string[]): Promise<CommandResult> {
|
||||
const [executable, ...args] = command;
|
||||
if (executable === undefined) {
|
||||
throw new FleetRuntimeTransportError('invalid_request', 'Fleet command is required');
|
||||
}
|
||||
try {
|
||||
return await this.options.runner(executable, args);
|
||||
} catch (error: unknown) {
|
||||
const message = error instanceof Error ? error.message : 'Fleet command failed';
|
||||
throw new FleetRuntimeTransportError('unavailable', message);
|
||||
}
|
||||
}
|
||||
|
||||
private async runChecked(command: string[], failureMessage: string): Promise<void> {
|
||||
const result = await this.run(command);
|
||||
if (result.exitCode !== 0) {
|
||||
throw new FleetRuntimeTransportError('unavailable', failureMessage);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function hasExactRuntimeIdentity(runtimeId: string, paneCommand: string): boolean {
|
||||
const expectedCommands = RUNTIME_ACCEPTABLE_COMMANDS[runtimeId];
|
||||
return expectedCommands !== undefined && expectedCommands.includes(paneCommand);
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
export const VERSION = '0.0.0';
|
||||
|
||||
export * from './fleet/tmux-runtime-transport.js';
|
||||
|
||||
export {
|
||||
backgroundUpdateCheck,
|
||||
checkForUpdate,
|
||||
|
||||
Reference in New Issue
Block a user