Compare commits

..

7 Commits

Author SHA1 Message Date
Hermes Agent
b7302a94c3 fix(gateway): fence lifecycle authority against durable lease
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
2026-07-14 17:57:40 -05:00
Hermes Agent
894434ce1a docs(#755): record current-main reconciliation
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
2026-07-14 17:30:34 -05:00
Jarvis
d190732a55 feat(#755): add logical agent connector fencing
Add normalized runtime-neutral identity, durable PostgreSQL CAS leases, monotonic epochs, short-lived server grants, fail-closed adapter validation, credential-safe audit, and concurrency/restart/abuse coverage.\n\nRefs #755
2026-07-14 17:23:06 -05:00
Jarvis
8599fd27d8 docs(mos): gate logical identity fencing work 2026-07-14 17:22:50 -05:00
2e2280070a docs(#753): clear KBN-010 threat and schema gate (#765)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
2026-07-14 21:46:44 +00:00
aa5b43bba2 feat(fleet): add roster v2 structural compiler (#764)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
2026-07-14 20:37:52 +00:00
ba13c08890 Fixes #756 (#763)
Some checks failed
ci/woodpecker/push/ci-image Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
ci/woodpecker/push/publish Pipeline was canceled
2026-07-14 20:26:15 +00:00
54 changed files with 10911 additions and 365 deletions

View File

@@ -72,6 +72,7 @@ describe('interaction Discord/CLI durable-session integration', () => {
process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-1',
channelId: 'channel-1',
pairedUsers: {
@@ -133,6 +134,7 @@ describe('interaction Discord/CLI durable-session integration', () => {
interactionBindings: [
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-1',
channelId: 'channel-1',
pairedUsers: {

View File

@@ -115,6 +115,18 @@ describe('AgentService owner/tenant scope enforcement', () => {
).rejects.toBeInstanceOf(ForbiddenException);
await service.prompt(CONVERSATION_ID, 'owner prompt', OWNER_SCOPE);
expect(session.piSession.prompt).toHaveBeenCalledWith('owner prompt');
await service.prompt(CONVERSATION_ID, '', OWNER_SCOPE, [
{
id: 'attachment-001',
name: 'diagram.png',
url: 'https://cdn.example.test/diagram.png',
mimeType: 'image/png',
},
]);
expect(session.piSession.prompt).toHaveBeenLastCalledWith(
'\n\n[Untrusted channel attachments]\n' +
'{"id":"attachment-001","name":"diagram.png","mimeType":"image/png","url":"https://cdn.example.test/diagram.png"}',
);
await expect(service.destroySession(CONVERSATION_ID, FOREIGN_SCOPE)).rejects.toBeInstanceOf(
ForbiddenException,

View File

@@ -21,6 +21,12 @@ import { LogModule } from '../log/log.module.js';
import { CommandsModule } from '../commands/commands.module.js';
import { CommandRuntimeApprovalVerifier } from '../commands/runtime-approval-verifier.js';
import { GatewayHermesRuntimeTransport } from './hermes-runtime.transport.js';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
import {
CONNECTOR_LEASE_POLICY,
ConnectorLeaseService,
DenyConnectorLeasePolicy,
} from './connector-lease.service.js';
import {
AGENT_RUNTIME_PROVIDER_REGISTRY,
RUNTIME_APPROVAL_VERIFIER,
@@ -46,6 +52,13 @@ export function createGatewayRuntimeProviderRegistry(): AgentRuntimeProviderRegi
SkillLoaderService,
DurableSessionRepository,
DurableSessionService,
ConnectorLeaseRepository,
DenyConnectorLeasePolicy,
{
provide: CONNECTOR_LEASE_POLICY,
useExisting: DenyConnectorLeasePolicy,
},
ConnectorLeaseService,
{
provide: AGENT_RUNTIME_PROVIDER_REGISTRY,
useFactory: createGatewayRuntimeProviderRegistry,
@@ -78,6 +91,7 @@ export function createGatewayRuntimeProviderRegistry(): AgentRuntimeProviderRegi
SkillLoaderService,
DurableSessionService,
RuntimeProviderService,
ConnectorLeaseService,
AGENT_RUNTIME_PROVIDER_REGISTRY,
],
})

View File

@@ -15,6 +15,7 @@ import {
type ToolDefinition,
} from '@mariozechner/pi-coding-agent';
import type { Brain } from '@mosaicstack/brain';
import type { ChannelAttachmentDto } from '@mosaicstack/types';
import type { Memory, OperatorMemoryPlugin } from '@mosaicstack/memory';
import { BRAIN } from '../brain/brain.tokens.js';
import { MEMORY } from '../memory/memory.tokens.js';
@@ -43,6 +44,8 @@ export interface ConversationHistoryMessage {
role: 'user' | 'assistant' | 'system';
content: string;
createdAt: Date;
/** Validated, URI-referenced channel attachments preserved on session resume. */
attachments?: readonly ChannelAttachmentDto[];
}
export interface AgentSessionOptions {
@@ -428,7 +431,7 @@ export class AgentService implements OnModuleDestroy {
const formatMessage = (msg: ConversationHistoryMessage): string => {
const roleLabel =
msg.role === 'user' ? 'User' : msg.role === 'assistant' ? 'Assistant' : 'System';
return `**${roleLabel}:** ${msg.content}`;
return `**${roleLabel}:** ${msg.content}${this.attachmentContext(msg.attachments ?? [])}`;
};
const formatted = history.map((msg) => formatMessage(msg));
@@ -487,6 +490,21 @@ export class AgentService implements OnModuleDestroy {
return result;
}
private attachmentContext(attachments: readonly ChannelAttachmentDto[]): string {
if (attachments.length === 0) return '';
return `\n\n[Untrusted channel attachments]\n${attachments
.map((attachment: ChannelAttachmentDto): string =>
JSON.stringify({
id: attachment.id,
name: attachment.name,
mimeType: attachment.mimeType,
url: attachment.url,
...(attachment.sizeBytes !== undefined ? { sizeBytes: attachment.sizeBytes } : {}),
}),
)
.join('\n')}`;
}
private resolveModel(options?: AgentSessionOptions) {
if (!options?.provider && !options?.modelId) {
return this.providerService.getDefaultModel() ?? null;
@@ -673,7 +691,19 @@ export class AgentService implements OnModuleDestroy {
session.channels.delete(channel);
}
async prompt(sessionId: string, message: string, scope: ActorTenantScope): Promise<void> {
async prompt(sessionId: string, message: string, scope: ActorTenantScope): Promise<void>;
async prompt(
sessionId: string,
message: string,
scope: ActorTenantScope,
attachments: readonly ChannelAttachmentDto[] | undefined,
): Promise<void>;
async prompt(
sessionId: string,
message: string,
scope: ActorTenantScope,
attachments: readonly ChannelAttachmentDto[] = [],
): Promise<void> {
const session = this.sessions.get(sessionId);
if (!session) {
throw new Error(`No agent session found: ${sessionId}`);
@@ -681,12 +711,16 @@ export class AgentService implements OnModuleDestroy {
this.assertSessionScope(session, scope);
session.promptCount += 1;
// Channel attachments are untrusted URI references. Preserve exact,
// authenticated metadata for the agent without treating it as authority.
const attachmentContext = this.attachmentContext(attachments);
// Prepend session-scoped system override if present (renew TTL on each turn)
let effectiveMessage = message;
let effectiveMessage = `${message}${attachmentContext}`;
if (this.systemOverride) {
const override = await this.systemOverride.get(sessionId, scope);
if (override) {
effectiveMessage = `[System Override]\n${override}\n\n${message}`;
effectiveMessage = `[System Override]\n${override}\n\n${effectiveMessage}`;
await this.systemOverride.renew(sessionId, scope);
this.logger.debug(`Applied system override for session ${sessionId}`);
}

View File

@@ -0,0 +1,341 @@
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
import { Test, type TestingModule } from '@nestjs/testing';
import {
connectorLeaseAuditLog,
createPgliteDb,
eq,
runPgliteMigrations,
type DbHandle,
} from '@mosaicstack/db';
import type { ConnectorExecutionContext, FencedConnectorAdapter } from '@mosaicstack/types';
import { DB } from '../database/database.module.js';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
import {
CONNECTOR_LEASE_POLICY,
ConnectorLeaseService,
type ConnectorLeasePolicy,
type ConnectorLeasePolicySubject,
} from './connector-lease.service.js';
const authorize = vi.fn().mockResolvedValue(true);
const policy: ConnectorLeasePolicy = { authorize };
const context = {
actorScope: { userId: 'operator-a', tenantId: 'tenant-a' },
correlationId: 'correlation-acquire',
};
describe('gateway connector lease fencing integration', (): void => {
let dataDir: string;
let handle: DbHandle;
let moduleRef: TestingModule;
let service: ConnectorLeaseService;
let repository: ConnectorLeaseRepository;
beforeAll(async (): Promise<void> => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-07-14T17:00:00.000Z'));
dataDir = await mkdtemp(join(tmpdir(), 'mosaic-gateway-connector-lease-'));
handle = createPgliteDb(dataDir);
await runPgliteMigrations(handle);
moduleRef = await Test.createTestingModule({
providers: [
ConnectorLeaseRepository,
ConnectorLeaseService,
{ provide: DB, useValue: handle.db },
{ provide: CONNECTOR_LEASE_POLICY, useValue: policy },
],
}).compile();
service = moduleRef.get(ConnectorLeaseService);
repository = moduleRef.get(ConnectorLeaseRepository);
});
afterAll(async (): Promise<void> => {
vi.useRealTimers();
await moduleRef.close();
await handle.close();
await rm(dataDir, { recursive: true, force: true });
});
it('derives tenant authority at the gateway and validates a grant before side effects', async (): Promise<void> => {
const lease = await service.acquire(
{
logicalAgentId: 'Mos',
bindingId: 'operator-chat',
connectorId: 'pi-worker-a',
scopes: ['runtime.send'],
ttlMs: 60_000,
},
context,
);
const grant = await service.issueGrant(
{ lease, scopes: ['runtime.send'], ttlMs: 30_000 },
{ ...context, correlationId: 'correlation-grant' },
);
const execute = vi.fn(async (_message: string, leaseContext: ConnectorExecutionContext) => {
return leaseContext.leaseEpoch;
});
const adapter: FencedConnectorAdapter<string, string> = { execute };
await expect(service.executeGrant(grant, 'runtime.send', 'hello', adapter)).resolves.toBe('1');
expect(execute).toHaveBeenCalledOnce();
expect(authorize).toHaveBeenCalledWith(
expect.objectContaining({
action: 'grant.issue',
requestedScopes: ['runtime.send'],
requestedTtlMs: 30_000,
}),
);
expect(execute.mock.calls[0]?.[1]).toMatchObject({
identity: { tenantId: 'tenant-a', logicalAgentId: 'mos' },
bindingId: 'operator-chat',
connectorId: 'pi-worker-a',
});
});
it('normalizes lease-derived policy subjects before authorization', async (): Promise<void> => {
const lease = await service.acquire(
{
logicalAgentId: 'mos',
bindingId: 'operator-chat-policy',
connectorId: 'pi-worker-a',
scopes: ['runtime.send'],
ttlMs: 60_000,
},
{ ...context, correlationId: 'correlation-policy-setup' },
);
const aliasedLease = {
...lease,
identity: { ...lease.identity, logicalAgentId: ' MOS ' },
bindingId: ' Operator-Chat-Policy ',
connectorId: ' PI-Worker-A ',
scopes: [' Runtime.Send '],
leaseEpoch: `00${lease.leaseEpoch}`,
};
await service.heartbeat(aliasedLease, 30_000, {
...context,
correlationId: 'correlation-policy-heartbeat',
});
expect(authorize).toHaveBeenLastCalledWith(
expect.objectContaining({
action: 'lease.heartbeat',
logicalAgentId: 'mos',
bindingId: 'operator-chat-policy',
connectorId: 'pi-worker-a',
requestedScopes: ['runtime.send'],
}),
);
await service.issueGrant(
{ lease: aliasedLease, scopes: [' Runtime.Send '], ttlMs: 1_000 },
{ ...context, correlationId: 'correlation-policy-grant' },
);
expect(authorize).toHaveBeenLastCalledWith(
expect.objectContaining({
action: 'grant.issue',
logicalAgentId: 'mos',
bindingId: 'operator-chat-policy',
connectorId: 'pi-worker-a',
requestedScopes: ['runtime.send'],
}),
);
await service.release(aliasedLease, {
...context,
correlationId: 'correlation-policy-release',
});
expect(authorize).toHaveBeenLastCalledWith(
expect.objectContaining({
action: 'lease.release',
logicalAgentId: 'mos',
bindingId: 'operator-chat-policy',
connectorId: 'pi-worker-a',
requestedScopes: ['runtime.send'],
}),
);
});
it('denies stale, forged, expired, cross-tenant, and cross-binding grants before effects', async (): Promise<void> => {
const bindingId = 'operator-chat-denials';
const current = await service.acquire(
{
logicalAgentId: 'mos',
bindingId,
connectorId: 'pi-worker-a',
scopes: ['runtime.send'],
ttlMs: 60_000,
},
{ ...context, correlationId: 'correlation-denial-setup' },
);
const stale = await service.issueGrant(
{ lease: current, scopes: ['runtime.send'], ttlMs: 30_000 },
{ ...context, correlationId: 'correlation-stale' },
);
await service.takeover(
{
logicalAgentId: 'mos',
bindingId,
connectorId: 'pi-worker-b',
scopes: ['runtime.send'],
ttlMs: 60_000,
expectedEpoch: current.leaseEpoch,
},
{ ...context, correlationId: 'correlation-takeover' },
);
const adapter = { execute: vi.fn().mockResolvedValue(undefined) };
await expect(service.executeGrant(stale, 'runtime.send', undefined, adapter)).rejects.toThrow();
const active = await service.current('mos', bindingId, context);
if (!active) throw new Error('active lease fixture is unavailable');
const grant = await service.issueGrant(
{ lease: active, scopes: ['runtime.send'], ttlMs: 1_000 },
{ ...context, correlationId: 'correlation-active' },
);
await expect(
service.executeGrant({ ...grant }, 'runtime.send', undefined, adapter),
).rejects.toThrow();
await expect(
service.executeGrant(
{ ...grant, bindingId: 'other-binding' },
'runtime.send',
undefined,
adapter,
),
).rejects.toThrow();
await expect(
service.issueGrant(
{ lease: active, scopes: ['runtime.send'], ttlMs: 30_000 },
{
actorScope: { userId: 'operator-b', tenantId: 'tenant-b' },
correlationId: 'correlation-cross-tenant',
},
),
).rejects.toThrow();
const crossTenantAudit = await handle.db
.select()
.from(connectorLeaseAuditLog)
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-cross-tenant'));
expect(crossTenantAudit).toHaveLength(1);
expect(crossTenantAudit[0]).toMatchObject({
tenantId: 'tenant-b',
logicalAgentId: 'untrusted',
bindingId: 'untrusted',
connectorId: 'untrusted',
reason: 'policy_denied',
});
vi.setSystemTime(new Date('2026-07-14T17:00:02.000Z'));
await expect(service.executeGrant(grant, 'runtime.send', undefined, adapter)).rejects.toThrow();
expect(adapter.execute).not.toHaveBeenCalled();
});
it('rejects submitted lifecycle scopes that differ from durable authority before policy or mutation', async (): Promise<void> => {
authorize.mockResolvedValue(true);
const heartbeatLease = await service.acquire(
{
logicalAgentId: 'mos',
bindingId: 'operator-chat-heartbeat-scope',
connectorId: 'pi-worker-a',
scopes: ['runtime.send'],
ttlMs: 60_000,
},
{ ...context, correlationId: 'correlation-heartbeat-scope-setup' },
);
const releaseLease = await service.acquire(
{
logicalAgentId: 'mos',
bindingId: 'operator-chat-release-scope',
connectorId: 'pi-worker-a',
scopes: ['runtime.send'],
ttlMs: 60_000,
},
{ ...context, correlationId: 'correlation-release-scope-setup' },
);
const forgedHeartbeat = { ...heartbeatLease, scopes: ['tool.execute'] };
const forgedRelease = { ...releaseLease, scopes: ['tool.execute'] };
authorize.mockImplementation(async (subject: ConnectorLeasePolicySubject) => {
return subject.requestedScopes.length === 1 && subject.requestedScopes[0] === 'tool.execute';
});
authorize.mockClear();
await expect(
service.heartbeat(forgedHeartbeat, 30_000, {
...context,
correlationId: 'correlation-heartbeat-scope-forgery',
}),
).rejects.toThrow('Connector authority policy denied');
await expect(
service.release(forgedRelease, {
...context,
correlationId: 'correlation-release-scope-forgery',
}),
).rejects.toThrow('Connector authority policy denied');
expect(authorize).not.toHaveBeenCalled();
const currentHeartbeat = await repository.findCurrent({
identity: heartbeatLease.identity,
bindingId: heartbeatLease.bindingId,
});
const currentRelease = await repository.findCurrent({
identity: releaseLease.identity,
bindingId: releaseLease.bindingId,
});
expect(currentHeartbeat).toMatchObject({
leaseId: heartbeatLease.leaseId,
scopes: ['runtime.send'],
heartbeatAt: heartbeatLease.heartbeatAt,
expiresAt: heartbeatLease.expiresAt,
});
expect(currentRelease).toMatchObject({
leaseId: releaseLease.leaseId,
scopes: ['runtime.send'],
});
expect(currentRelease?.releasedAt).toBeUndefined();
const forgedAudits = await handle.db
.select()
.from(connectorLeaseAuditLog)
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-heartbeat-scope-forgery'));
expect(forgedAudits).toHaveLength(1);
expect(forgedAudits[0]).toMatchObject({
bindingId: heartbeatLease.bindingId,
connectorId: heartbeatLease.connectorId,
event: 'reject',
outcome: 'denied',
reason: 'policy_denied',
});
const forgedReleaseAudits = await handle.db
.select()
.from(connectorLeaseAuditLog)
.where(eq(connectorLeaseAuditLog.correlationId, 'correlation-release-scope-forgery'));
expect(forgedReleaseAudits).toHaveLength(1);
expect(forgedReleaseAudits[0]).toMatchObject({
bindingId: releaseLease.bindingId,
connectorId: releaseLease.connectorId,
event: 'reject',
outcome: 'denied',
reason: 'policy_denied',
});
authorize.mockImplementation(async (subject: ConnectorLeasePolicySubject) => {
return subject.requestedScopes.length === 1 && subject.requestedScopes[0] === 'runtime.send';
});
await expect(
service.heartbeat(heartbeatLease, 30_000, {
...context,
correlationId: 'correlation-heartbeat-scope-canonical',
}),
).resolves.toMatchObject({ scopes: ['runtime.send'] });
await expect(
service.release(releaseLease, {
...context,
correlationId: 'correlation-release-scope-canonical',
}),
).resolves.toBeUndefined();
});
});

View File

@@ -0,0 +1,76 @@
import { randomUUID } from 'node:crypto';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import {
connectorLeaseAuditLog,
createDb,
eq,
logicalAgentConnectorLeases,
type DbHandle,
} from '@mosaicstack/db';
import { ConnectorLeaseCoordinator } from '@mosaicstack/agent';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
const hasPostgres = Boolean(process.env['DATABASE_URL']);
const tenantId = `lease-test-${randomUUID()}`;
const identity = { tenantId, logicalAgentId: 'mos' } as const;
describe.skipIf(!hasPostgres)('ConnectorLeaseRepository real PostgreSQL integration', (): void => {
let handle: DbHandle;
beforeAll((): void => {
handle = createDb(process.env['DATABASE_URL']);
});
afterAll(async (): Promise<void> => {
if (!handle) return;
await handle.db
.delete(connectorLeaseAuditLog)
.where(eq(connectorLeaseAuditLog.tenantId, tenantId));
await handle.db
.delete(logicalAgentConnectorLeases)
.where(eq(logicalAgentConnectorLeases.tenantId, tenantId));
await handle.close();
});
it('preserves the exclusive CAS fence across a real pool close/reopen', async (): Promise<void> => {
const command = {
identity,
bindingId: 'operator-chat',
scopes: ['runtime.send'],
ttlMs: 60_000,
} as const;
const firstCoordinator = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db));
const contenders = await Promise.allSettled([
firstCoordinator.acquire({
...command,
connectorId: 'connector-a',
correlationId: 'postgres-acquire-a',
}),
firstCoordinator.acquire({
...command,
connectorId: 'connector-b',
correlationId: 'postgres-acquire-b',
}),
]);
const acquired = contenders.find((result) => result.status === 'fulfilled');
if (!acquired || acquired.status !== 'fulfilled') throw new Error('no lease contender won');
expect(contenders.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
await handle.close();
handle = createDb(process.env['DATABASE_URL']);
const reopened = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db));
const persisted = await reopened.current({ identity, bindingId: 'operator-chat' });
expect(persisted).toMatchObject({
leaseId: acquired.value.leaseId,
leaseEpoch: '1',
});
const takeover = await reopened.takeover({
...command,
connectorId: 'connector-c',
correlationId: 'postgres-takeover',
expectedEpoch: acquired.value.leaseEpoch,
});
expect(takeover).toMatchObject({ connectorId: 'connector-c', leaseEpoch: '2' });
});
});

View File

@@ -0,0 +1,149 @@
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import {
connectorLeaseAuditLog,
createPgliteDb,
eq,
runPgliteMigrations,
type DbHandle,
} from '@mosaicstack/db';
import { ConnectorLeaseCoordinator, ConnectorLeaseError } from '@mosaicstack/agent';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
const identity = { tenantId: 'tenant-a', logicalAgentId: 'mos' } as const;
function acquireCommand(connectorId: string, correlationId: string) {
return {
identity,
bindingId: 'operator-chat',
connectorId,
scopes: ['runtime.send', 'tool.execute'],
ttlMs: 60_000,
correlationId,
};
}
describe('ConnectorLeaseRepository PostgreSQL semantics', (): void => {
let dataDir: string;
let handle: DbHandle;
let now: Date;
let coordinator: ConnectorLeaseCoordinator;
beforeEach(async (): Promise<void> => {
dataDir = await mkdtemp(join(tmpdir(), 'mosaic-connector-lease-'));
handle = createPgliteDb(dataDir);
await runPgliteMigrations(handle);
now = new Date('2026-07-14T17:00:00.000Z');
coordinator = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db), {
now: (): Date => now,
});
});
afterEach(async (): Promise<void> => {
await handle.close();
await rm(dataDir, { recursive: true, force: true });
});
it('allows only one concurrent contender to acquire a binding', async (): Promise<void> => {
const outcomes = await Promise.allSettled([
coordinator.acquire(acquireCommand('connector-a', 'correlation-a')),
coordinator.acquire(acquireCommand('connector-b', 'correlation-b')),
]);
expect(outcomes.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
const rejected = outcomes.find((result) => result.status === 'rejected');
expect(rejected).toMatchObject({
reason: { code: 'lease_held' } satisfies Partial<ConnectorLeaseError>,
});
});
it('uses compare-and-swap takeover and increments the fencing epoch monotonically', async (): Promise<void> => {
const acquired = await coordinator.acquire(acquireCommand('connector-a', 'correlation-a'));
const results = await Promise.allSettled([
coordinator.takeover({
...acquireCommand('connector-b', 'correlation-b'),
expectedEpoch: acquired.leaseEpoch,
}),
coordinator.takeover({
...acquireCommand('connector-c', 'correlation-c'),
expectedEpoch: acquired.leaseEpoch,
}),
]);
const winner = results.find((result) => result.status === 'fulfilled');
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
expect(winner?.status === 'fulfilled' ? winner.value.leaseEpoch : null).toBe('2');
expect(results.find((result) => result.status === 'rejected')).toMatchObject({
reason: { code: 'cas_mismatch' } satisfies Partial<ConnectorLeaseError>,
});
});
it('heartbeats and releases only the current connector epoch', async (): Promise<void> => {
const acquired = await coordinator.acquire(acquireCommand('connector-a', 'correlation-a'));
now = new Date('2026-07-14T17:00:30.000Z');
const renewed = await coordinator.heartbeat({
lease: acquired,
ttlMs: 120_000,
correlationId: 'correlation-renew',
});
expect(renewed.expiresAt).toBe('2026-07-14T17:02:30.000Z');
await coordinator.release({ lease: renewed, correlationId: 'correlation-release' });
await expect(
coordinator.heartbeat({
lease: renewed,
ttlMs: 120_000,
correlationId: 'correlation-stale',
}),
).rejects.toMatchObject({ code: 'lease_released' } satisfies Partial<ConnectorLeaseError>);
});
it('survives close/reopen and requires CAS takeover to recover an expired lease', async (): Promise<void> => {
const acquired = await coordinator.acquire(acquireCommand('connector-a', 'correlation-a'));
await handle.close();
now = new Date('2026-07-14T17:02:00.000Z');
handle = createPgliteDb(dataDir);
await runPgliteMigrations(handle);
coordinator = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db), {
now: (): Date => now,
});
await expect(
coordinator.acquire(acquireCommand('connector-b', 'correlation-plain-acquire')),
).rejects.toMatchObject({ code: 'takeover_required' } satisfies Partial<ConnectorLeaseError>);
const recovered = await coordinator.takeover({
...acquireCommand('connector-b', 'correlation-takeover'),
expectedEpoch: acquired.leaseEpoch,
});
expect(recovered).toMatchObject({ connectorId: 'connector-b', leaseEpoch: '2' });
});
it('writes credential-safe lifecycle and rejection audit records', async (): Promise<void> => {
const acquired = await coordinator.acquire(acquireCommand('connector-a', 'correlation-a'));
await coordinator.heartbeat({
lease: acquired,
ttlMs: 60_000,
correlationId: 'correlation-renew',
});
await expect(
coordinator.acquire(acquireCommand('connector-b', 'correlation-reject')),
).rejects.toBeInstanceOf(ConnectorLeaseError);
const rows = await handle.db
.select()
.from(connectorLeaseAuditLog)
.where(eq(connectorLeaseAuditLog.tenantId, identity.tenantId));
expect(rows.map((row) => row.event)).toEqual(
expect.arrayContaining(['acquire', 'renew', 'reject']),
);
const serialized = JSON.stringify(rows, (_key: string, value: unknown): unknown =>
typeof value === 'bigint' ? value.toString(10) : value,
);
expect(serialized).not.toContain('tool.execute');
expect(serialized).not.toContain('runtime.send');
expect(serialized).not.toMatch(/token|secret|credential/i);
});
});

View File

@@ -0,0 +1,354 @@
import { Inject, Injectable } from '@nestjs/common';
import {
and,
connectorLeaseAuditLog,
eq,
gt,
isNull,
logicalAgentConnectorLeases,
sql,
type Db,
} from '@mosaicstack/db';
import { ConnectorLeaseError } from '@mosaicstack/agent';
import type {
ConnectorLease,
ConnectorLeaseAcquireMutation,
ConnectorLeaseAuditEvent,
ConnectorLeaseHeartbeatMutation,
ConnectorLeaseRejectReason,
ConnectorLeaseReleaseMutation,
ConnectorLeaseStore,
ConnectorLeaseTakeoverMutation,
LogicalAgentBinding,
} from '@mosaicstack/types';
import { DB } from '../database/database.module.js';
interface SuccessfulMutation {
readonly ok: true;
readonly lease: ConnectorLease;
}
interface FailedMutation {
readonly ok: false;
readonly reason: ConnectorLeaseRejectReason;
}
type MutationResult = SuccessfulMutation | FailedMutation;
@Injectable()
export class ConnectorLeaseRepository implements ConnectorLeaseStore {
constructor(@Inject(DB) private readonly db: Db) {}
async acquire(input: ConnectorLeaseAcquireMutation): Promise<ConnectorLease> {
const result: MutationResult = await this.db.transaction(
async (tx): Promise<MutationResult> => {
const inserted = await tx
.insert(logicalAgentConnectorLeases)
.values({
leaseId: input.leaseId,
tenantId: input.identity.tenantId,
logicalAgentId: input.identity.logicalAgentId,
bindingId: input.bindingId,
connectorId: input.connectorId,
scopes: [...input.scopes],
leaseEpoch: 1n,
acquiredAt: new Date(input.now),
heartbeatAt: new Date(input.now),
expiresAt: new Date(input.expiresAt),
updatedAt: new Date(input.now),
})
.onConflictDoNothing()
.returning();
const row = inserted[0];
if (row) {
const lease = toLease(row);
await insertAudit(tx, lifecycleAudit(input, lease, 'acquire'));
return { ok: true, lease };
}
const current = await findRow(tx, input);
if (current && current.expiresAt <= new Date(input.now) && !current.releasedAt) {
await insertAudit(tx, lifecycleAudit(input, toLease(current), 'expiry'));
}
const reason: ConnectorLeaseRejectReason =
current && (current.releasedAt || current.expiresAt <= new Date(input.now))
? 'takeover_required'
: 'lease_held';
await insertAudit(tx, rejectionAudit(input, current ? toLease(current) : null, reason));
return { ok: false, reason };
},
);
return unwrap(result);
}
async takeover(input: ConnectorLeaseTakeoverMutation): Promise<ConnectorLease> {
const result: MutationResult = await this.db.transaction(
async (tx): Promise<MutationResult> => {
const current = await findRow(tx, input);
if (!current || current.leaseEpoch.toString(10) !== input.expectedEpoch) {
await insertAudit(
tx,
rejectionAudit(input, current ? toLease(current) : null, 'cas_mismatch'),
);
return { ok: false, reason: 'cas_mismatch' };
}
if (current.expiresAt <= new Date(input.now) && !current.releasedAt) {
await insertAudit(tx, lifecycleAudit(input, toLease(current), 'expiry'));
}
const updated = await tx
.update(logicalAgentConnectorLeases)
.set({
leaseId: input.leaseId,
connectorId: input.connectorId,
scopes: [...input.scopes],
leaseEpoch: sql`${logicalAgentConnectorLeases.leaseEpoch} + 1`,
acquiredAt: new Date(input.now),
heartbeatAt: new Date(input.now),
expiresAt: new Date(input.expiresAt),
releasedAt: null,
updatedAt: new Date(input.now),
})
.where(
and(
bindingPredicate(input),
eq(logicalAgentConnectorLeases.leaseId, current.leaseId),
eq(logicalAgentConnectorLeases.leaseEpoch, BigInt(input.expectedEpoch)),
),
)
.returning();
const row = updated[0];
if (!row) {
await insertAudit(tx, rejectionAudit(input, toLease(current), 'cas_mismatch'));
return { ok: false, reason: 'cas_mismatch' };
}
const lease = toLease(row);
await insertAudit(tx, lifecycleAudit(input, lease, 'takeover'));
return { ok: true, lease };
},
);
return unwrap(result);
}
async heartbeat(input: ConnectorLeaseHeartbeatMutation): Promise<ConnectorLease> {
const result: MutationResult = await this.db.transaction(
async (tx): Promise<MutationResult> => {
const updated = await tx
.update(logicalAgentConnectorLeases)
.set({
heartbeatAt: new Date(input.now),
expiresAt: new Date(input.expiresAt),
updatedAt: new Date(input.now),
})
.where(
and(
bindingPredicate(input.lease),
eq(logicalAgentConnectorLeases.leaseId, input.lease.leaseId),
eq(logicalAgentConnectorLeases.connectorId, input.lease.connectorId),
eq(logicalAgentConnectorLeases.leaseEpoch, BigInt(input.lease.leaseEpoch)),
isNull(logicalAgentConnectorLeases.releasedAt),
gt(logicalAgentConnectorLeases.expiresAt, new Date(input.now)),
),
)
.returning();
const row = updated[0];
if (row) {
const lease = toLease(row);
await insertAudit(tx, lifecycleAudit(input, lease, 'renew'));
return { ok: true, lease };
}
const current = await findRow(tx, input.lease);
const reason = classifyAuthorityFailure(
current ? toLease(current) : null,
input.lease,
input.now,
);
if (reason === 'lease_expired' && current) {
await insertAudit(tx, lifecycleAudit(input, toLease(current), 'expiry'));
}
await insertAudit(
tx,
rejectionAudit(
{ ...input.lease, correlationId: input.correlationId, now: input.now },
current ? toLease(current) : null,
reason,
),
);
return { ok: false, reason };
},
);
return unwrap(result);
}
async release(input: ConnectorLeaseReleaseMutation): Promise<void> {
const result: MutationResult = await this.db.transaction(
async (tx): Promise<MutationResult> => {
const updated = await tx
.update(logicalAgentConnectorLeases)
.set({
releasedAt: new Date(input.now),
expiresAt: new Date(input.now),
updatedAt: new Date(input.now),
})
.where(
and(
bindingPredicate(input.lease),
eq(logicalAgentConnectorLeases.leaseId, input.lease.leaseId),
eq(logicalAgentConnectorLeases.connectorId, input.lease.connectorId),
eq(logicalAgentConnectorLeases.leaseEpoch, BigInt(input.lease.leaseEpoch)),
isNull(logicalAgentConnectorLeases.releasedAt),
gt(logicalAgentConnectorLeases.expiresAt, new Date(input.now)),
),
)
.returning();
const row = updated[0];
if (row) {
const lease = toLease(row);
await insertAudit(tx, lifecycleAudit(input, lease, 'release'));
return { ok: true, lease };
}
const current = await findRow(tx, input.lease);
const reason = classifyAuthorityFailure(
current ? toLease(current) : null,
input.lease,
input.now,
);
if (reason === 'lease_expired' && current) {
await insertAudit(tx, lifecycleAudit(input, toLease(current), 'expiry'));
}
await insertAudit(
tx,
rejectionAudit(
{ ...input.lease, correlationId: input.correlationId, now: input.now },
current ? toLease(current) : null,
reason,
),
);
return { ok: false, reason };
},
);
unwrap(result);
}
async findCurrent(binding: LogicalAgentBinding): Promise<ConnectorLease | null> {
const row = await findRow(this.db, binding);
return row ? toLease(row) : null;
}
async recordAudit(event: ConnectorLeaseAuditEvent): Promise<void> {
await insertAudit(this.db, event);
}
}
function unwrap(result: MutationResult): ConnectorLease {
if (!result.ok) throw new ConnectorLeaseError(result.reason, safeErrorMessage(result.reason));
return result.lease;
}
function safeErrorMessage(reason: ConnectorLeaseRejectReason): string {
return `Connector lease mutation denied: ${reason}`;
}
function bindingPredicate(binding: LogicalAgentBinding) {
return and(
eq(logicalAgentConnectorLeases.tenantId, binding.identity.tenantId),
eq(logicalAgentConnectorLeases.logicalAgentId, binding.identity.logicalAgentId),
eq(logicalAgentConnectorLeases.bindingId, binding.bindingId),
);
}
async function findRow(
db: Pick<Db, 'select'>,
binding: LogicalAgentBinding,
): Promise<typeof logicalAgentConnectorLeases.$inferSelect | null> {
const rows = await db
.select()
.from(logicalAgentConnectorLeases)
.where(bindingPredicate(binding))
.limit(1);
return rows[0] ?? null;
}
function toLease(row: typeof logicalAgentConnectorLeases.$inferSelect): ConnectorLease {
return Object.freeze({
identity: Object.freeze({ tenantId: row.tenantId, logicalAgentId: row.logicalAgentId }),
bindingId: row.bindingId,
leaseId: row.leaseId,
connectorId: row.connectorId,
scopes: Object.freeze([...row.scopes]),
leaseEpoch: row.leaseEpoch.toString(10),
acquiredAt: row.acquiredAt.toISOString(),
heartbeatAt: row.heartbeatAt.toISOString(),
expiresAt: row.expiresAt.toISOString(),
...(row.releasedAt ? { releasedAt: row.releasedAt.toISOString() } : {}),
});
}
function classifyAuthorityFailure(
current: ConnectorLease | null,
claimed: ConnectorLease,
now: string,
): ConnectorLeaseRejectReason {
if (!current) return 'lease_missing';
if (current.releasedAt) return 'lease_released';
if (new Date(current.expiresAt) <= new Date(now)) return 'lease_expired';
if (current.leaseEpoch !== claimed.leaseEpoch) return 'stale_epoch';
return 'connector_mismatch';
}
function lifecycleAudit(
input: { readonly correlationId: string; readonly now: string },
lease: ConnectorLease,
event: Exclude<ConnectorLeaseAuditEvent['event'], 'reject'>,
): ConnectorLeaseAuditEvent {
return {
identity: lease.identity,
bindingId: lease.bindingId,
connectorId: lease.connectorId,
leaseId: lease.leaseId,
leaseEpoch: lease.leaseEpoch,
event,
outcome: 'succeeded',
correlationId: input.correlationId,
occurredAt: input.now,
};
}
function rejectionAudit(
input: {
readonly identity: ConnectorLease['identity'];
readonly bindingId: string;
readonly connectorId: string;
readonly correlationId: string;
readonly now: string;
},
current: ConnectorLease | null,
reason: ConnectorLeaseRejectReason,
): ConnectorLeaseAuditEvent {
return {
identity: input.identity,
bindingId: input.bindingId,
connectorId: input.connectorId,
event: 'reject',
outcome: 'denied',
correlationId: input.correlationId,
occurredAt: input.now,
...(current ? { leaseId: current.leaseId, leaseEpoch: current.leaseEpoch } : {}),
reason,
};
}
async function insertAudit(db: Pick<Db, 'insert'>, event: ConnectorLeaseAuditEvent): Promise<void> {
await db.insert(connectorLeaseAuditLog).values({
tenantId: event.identity.tenantId,
logicalAgentId: event.identity.logicalAgentId,
bindingId: event.bindingId,
connectorId: event.connectorId,
...(event.leaseId ? { leaseId: event.leaseId } : {}),
...(event.leaseEpoch ? { leaseEpoch: BigInt(event.leaseEpoch) } : {}),
event: event.event,
outcome: event.outcome,
...(event.reason ? { reason: event.reason } : {}),
correlationId: event.correlationId,
occurredAt: new Date(event.occurredAt),
});
}

View File

@@ -0,0 +1,285 @@
import { ForbiddenException, Inject, Injectable } from '@nestjs/common';
import { ConnectorLeaseCoordinator, normalizeConnectorLease } from '@mosaicstack/agent';
import {
normalizeConnectorId,
normalizeConnectorScopes,
normalizeCorrelationId,
normalizeLogicalAgentIdentity,
normalizeLogicalBindingId,
type AcquireConnectorLeaseInput,
type ConnectorExecutionGrant,
type ConnectorLease,
type ConnectorLeaseAuditEvent,
type FencedConnectorAdapter,
} from '@mosaicstack/types';
import type { ActorTenantScope } from '../auth/session-scope.js';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
export const CONNECTOR_LEASE_POLICY = Symbol('CONNECTOR_LEASE_POLICY');
export type ConnectorLeasePolicyAction =
| 'lease.acquire'
| 'lease.takeover'
| 'lease.heartbeat'
| 'lease.release'
| 'lease.read'
| 'grant.issue';
export interface ConnectorLeaseRequestContext {
readonly actorScope: ActorTenantScope;
readonly correlationId: string;
}
export interface GatewayConnectorLeaseRequest {
readonly logicalAgentId: string;
readonly bindingId: string;
readonly connectorId: string;
readonly scopes: readonly string[];
readonly ttlMs: number;
}
export interface GatewayConnectorLeaseTakeoverRequest extends GatewayConnectorLeaseRequest {
readonly expectedEpoch: string;
}
export interface GatewayConnectorGrantRequest {
readonly lease: ConnectorLease;
readonly scopes: readonly string[];
readonly ttlMs: number;
}
export interface ConnectorLeasePolicySubject {
readonly action: ConnectorLeasePolicyAction;
readonly actorId: string;
readonly tenantId: string;
readonly logicalAgentId: string;
readonly bindingId: string;
readonly connectorId: string;
readonly requestedScopes: readonly string[];
readonly requestedTtlMs: number | null;
}
export interface ConnectorLeasePolicy {
authorize(subject: ConnectorLeasePolicySubject): Promise<boolean>;
}
/** M1 has no concrete cutover policy: unconfigured production use fails closed. */
@Injectable()
export class DenyConnectorLeasePolicy implements ConnectorLeasePolicy {
async authorize(_subject: ConnectorLeasePolicySubject): Promise<boolean> {
return false;
}
}
/** Gateway-owned policy surface for durable connector authority and fenced effects. */
@Injectable()
export class ConnectorLeaseService {
private readonly coordinator: ConnectorLeaseCoordinator;
constructor(
@Inject(ConnectorLeaseRepository) private readonly repository: ConnectorLeaseRepository,
@Inject(CONNECTOR_LEASE_POLICY) private readonly policy: ConnectorLeasePolicy,
) {
this.coordinator = new ConnectorLeaseCoordinator(repository);
}
async acquire(
request: GatewayConnectorLeaseRequest,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorLease> {
const command = this.command(request, context);
await this.assertPolicy('lease.acquire', command, context, command.scopes, command.ttlMs);
return this.coordinator.acquire({ ...command, correlationId: this.correlation(context) });
}
async takeover(
request: GatewayConnectorLeaseTakeoverRequest,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorLease> {
const command = this.command(request, context);
await this.assertPolicy('lease.takeover', command, context, command.scopes, command.ttlMs);
return this.coordinator.takeover({
...command,
expectedEpoch: request.expectedEpoch,
correlationId: this.correlation(context),
});
}
async heartbeat(
lease: ConnectorLease,
ttlMs: number,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorLease> {
const normalizedLease = normalizeConnectorLease(lease);
const durableLease = await this.durableLifecycleLease(normalizedLease, context);
await this.assertPolicy('lease.heartbeat', durableLease, context, durableLease.scopes, ttlMs);
return this.coordinator.heartbeat({
lease: durableLease,
ttlMs,
correlationId: this.correlation(context),
});
}
async release(lease: ConnectorLease, context: ConnectorLeaseRequestContext): Promise<void> {
const normalizedLease = normalizeConnectorLease(lease);
const durableLease = await this.durableLifecycleLease(normalizedLease, context);
await this.assertPolicy('lease.release', durableLease, context, durableLease.scopes, null);
await this.coordinator.release({
lease: durableLease,
correlationId: this.correlation(context),
});
}
async current(
logicalAgentId: string,
bindingId: string,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorLease | null> {
const binding = {
identity: normalizeLogicalAgentIdentity({
tenantId: context.actorScope.tenantId,
logicalAgentId,
}),
bindingId: normalizeLogicalBindingId(bindingId),
connectorId: 'gateway',
};
await this.assertPolicy('lease.read', binding, context, [], null);
return this.coordinator.current(binding);
}
async issueGrant(
request: GatewayConnectorGrantRequest,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorExecutionGrant> {
const lease = normalizeConnectorLease(request.lease);
await this.assertTenant(lease, context);
const scopes = normalizeConnectorScopes(request.scopes);
await this.assertPolicy('grant.issue', lease, context, scopes, request.ttlMs);
return this.coordinator.issueGrant({
lease,
scopes,
ttlMs: request.ttlMs,
correlationId: this.correlation(context),
});
}
async executeGrant<TInput, TOutput>(
grant: ConnectorExecutionGrant,
requiredScope: string,
input: TInput,
adapter: FencedConnectorAdapter<TInput, TOutput>,
): Promise<TOutput> {
return this.coordinator.executeGrant(grant, requiredScope, input, adapter);
}
private command(
request: GatewayConnectorLeaseRequest,
context: ConnectorLeaseRequestContext,
): Omit<AcquireConnectorLeaseInput, 'correlationId'> {
return {
identity: normalizeLogicalAgentIdentity({
tenantId: context.actorScope.tenantId,
logicalAgentId: request.logicalAgentId,
}),
bindingId: normalizeLogicalBindingId(request.bindingId),
connectorId: normalizeConnectorId(request.connectorId),
scopes: normalizeConnectorScopes(request.scopes),
ttlMs: request.ttlMs,
};
}
private async assertTenant(
lease: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
context: ConnectorLeaseRequestContext,
): Promise<void> {
if (lease.identity.tenantId !== context.actorScope.tenantId) {
await this.recordPolicyDenial(
{
identity: {
tenantId: context.actorScope.tenantId,
logicalAgentId: 'untrusted',
},
bindingId: 'untrusted',
connectorId: 'untrusted',
},
context,
);
throw new ForbiddenException('Connector authority tenant scope denied');
}
}
private async durableLifecycleLease(
submittedLease: ConnectorLease,
context: ConnectorLeaseRequestContext,
): Promise<ConnectorLease> {
await this.assertTenant(submittedLease, context);
const durableLease = await this.coordinator.current(submittedLease);
if (!durableLease || !hasSameLifecycleAuthority(submittedLease, durableLease)) {
await this.recordPolicyDenial(durableLease ?? submittedLease, context);
throw new ForbiddenException('Connector authority policy denied');
}
return durableLease;
}
private async assertPolicy(
action: ConnectorLeasePolicyAction,
subject: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
context: ConnectorLeaseRequestContext,
requestedScopes: readonly string[],
requestedTtlMs: number | null,
): Promise<void> {
const allowed = await this.policy.authorize({
action,
actorId: context.actorScope.userId,
tenantId: subject.identity.tenantId,
logicalAgentId: subject.identity.logicalAgentId,
bindingId: subject.bindingId,
connectorId: subject.connectorId,
requestedScopes: Object.freeze([...requestedScopes]),
requestedTtlMs,
});
if (!allowed) {
await this.recordPolicyDenial(subject, context);
throw new ForbiddenException('Connector authority policy denied');
}
}
private async recordPolicyDenial(
subject: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
context: ConnectorLeaseRequestContext,
): Promise<void> {
const event: ConnectorLeaseAuditEvent = {
identity: subject.identity,
bindingId: subject.bindingId,
connectorId: subject.connectorId,
event: 'reject',
outcome: 'denied',
reason: 'policy_denied',
correlationId: this.correlation(context),
occurredAt: new Date().toISOString(),
};
await this.repository.recordAudit(event);
}
private correlation(context: ConnectorLeaseRequestContext): string {
return normalizeCorrelationId(context.correlationId);
}
}
function hasSameLifecycleAuthority(
submittedLease: ConnectorLease,
durableLease: ConnectorLease,
): boolean {
return (
submittedLease.identity.tenantId === durableLease.identity.tenantId &&
submittedLease.identity.logicalAgentId === durableLease.identity.logicalAgentId &&
submittedLease.bindingId === durableLease.bindingId &&
submittedLease.leaseId === durableLease.leaseId &&
submittedLease.connectorId === durableLease.connectorId &&
submittedLease.leaseEpoch === durableLease.leaseEpoch &&
submittedLease.scopes.length === durableLease.scopes.length &&
submittedLease.scopes.every((scope: string, index: number): boolean => {
return scope === durableLease.scopes[index];
})
);
}

View File

@@ -1,3 +1,4 @@
import type { ChannelAttachmentDto } from '@mosaicstack/types';
import { IsOptional, IsString, IsUUID, MaxLength } from 'class-validator';
export class ChatRequestDto {
@@ -32,4 +33,7 @@ export class ChatSocketMessageDto {
@IsOptional()
@IsUUID()
agentId?: string;
/** Validated channel attachment references; binary content is not embedded. */
attachments?: readonly ChannelAttachmentDto[];
}

View File

@@ -4,6 +4,10 @@ import { ChatGateway } from './chat.gateway.js';
const CONVERSATION_ID = 'conversation-1';
const CANARY = 'sk_canary12345678';
function clientConversationKey(clientId: string, conversationId: string): string {
return `${clientId}\u0000${conversationId}`;
}
type GatewayInternals = {
clientSessions: Map<string, unknown>;
relayEvent(client: unknown, conversationId: string, event: unknown): void;
@@ -40,6 +44,7 @@ describe('ChatGateway redaction boundary', (): void => {
emit: vi.fn(),
};
const session = {
clientId: client.id,
conversationId: CONVERSATION_ID,
cleanup: vi.fn(),
assistantText: '',
@@ -47,7 +52,7 @@ describe('ChatGateway redaction boundary', (): void => {
pendingToolCalls: new Map(),
scope: { userId: 'user-1', tenantId: 'tenant-1' },
};
gateway.clientSessions.set(client.id, session);
gateway.clientSessions.set(clientConversationKey(client.id, CONVERSATION_ID), session);
gateway.relayEvent(client, CONVERSATION_ID, {
type: 'message_update',
@@ -139,6 +144,51 @@ describe('ChatGateway redaction boundary', (): void => {
});
});
it('isolates concurrent conversation streams sharing one Discord socket', (): void => {
const { gateway } = buildGateway();
const client = {
connected: true,
id: 'discord-client',
data: { user: { id: 'user-1' } },
emit: vi.fn(),
};
const firstConversation = 'Nova:discord:thread-1';
const secondConversation = 'Nova:discord:thread-2';
const createSession = (conversationId: string) => ({
clientId: client.id,
conversationId,
cleanup: vi.fn(),
assistantText: '',
toolCalls: [],
pendingToolCalls: new Map(),
scope: { userId: 'user-1', tenantId: 'tenant-1' },
});
const firstSession = createSession(firstConversation);
const secondSession = createSession(secondConversation);
gateway.clientSessions.set(clientConversationKey(client.id, firstConversation), firstSession);
gateway.clientSessions.set(clientConversationKey(client.id, secondConversation), secondSession);
gateway.relayEvent(client, firstConversation, {
type: 'message_update',
assistantMessageEvent: { type: 'text_delta', delta: 'first response ' },
});
gateway.relayEvent(client, secondConversation, {
type: 'message_update',
assistantMessageEvent: { type: 'text_delta', delta: 'second response ' },
});
expect(firstSession.assistantText).toBe('first response ');
expect(secondSession.assistantText).toBe('second response ');
expect(client.emit).toHaveBeenCalledWith('agent:text', {
conversationId: firstConversation,
text: 'first response ',
});
expect(client.emit).toHaveBeenCalledWith('agent:text', {
conversationId: secondConversation,
text: 'second response ',
});
});
it('persists only redacted assistant content with classifications', (): void => {
const { gateway, brain } = buildGateway();
const client = {
@@ -147,7 +197,8 @@ describe('ChatGateway redaction boundary', (): void => {
data: { user: { id: 'user-1' } },
emit: vi.fn(),
};
gateway.clientSessions.set(client.id, {
gateway.clientSessions.set(clientConversationKey(client.id, CONVERSATION_ID), {
clientId: client.id,
conversationId: CONVERSATION_ID,
cleanup: vi.fn(),
assistantText: CANARY,

View File

@@ -17,6 +17,7 @@ import {
parseDiscordInteractionBindings,
resolveDiscordInteractionActorId,
resolveDiscordInteractionBinding,
type DiscordAttachment,
type DiscordIngressEnvelope,
type DiscordIngressPayload,
} from '@mosaicstack/discord-plugin';
@@ -30,6 +31,7 @@ import type {
SystemReloadPayload,
RoutingDecisionInfo,
AbortPayload,
ChannelAttachmentDto,
} from '@mosaicstack/types';
import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js';
import {
@@ -56,6 +58,7 @@ import { DiscordReplayProtector } from '../plugin/discord-replay-protector.js';
/** Per-client state tracking streaming accumulation for persistence. */
interface ClientSession {
clientId: string;
conversationId: string;
cleanup: () => void;
/** Accumulated assistant response text for the current turn. */
@@ -76,6 +79,68 @@ interface ClientSession {
*/
const modelOverrides = new Map<string, string>();
const MAX_REDACTION_BUFFER_LENGTH = 8_192;
const MAX_CHANNEL_ATTACHMENTS = 10;
const MAX_ATTACHMENT_METADATA_BYTES = 16_384;
const MAX_ATTACHMENT_ID_LENGTH = 128;
const MAX_ATTACHMENT_NAME_LENGTH = 255;
const MAX_ATTACHMENT_URL_LENGTH = 2_048;
const MAX_ATTACHMENT_MIME_LENGTH = 255;
function isSafeAttachmentUrl(value: string): boolean {
if (value.length === 0 || value.length > MAX_ATTACHMENT_URL_LENGTH) return false;
try {
const url = new URL(value);
return (
url.protocol === 'https:' &&
!url.username &&
!url.password &&
!url.hash &&
url.search.length === 0
);
} catch {
return false;
}
}
function hasValidAttachmentBounds(value: {
id: string;
name: string;
url: string;
sizeBytes?: number;
}): boolean {
return (
value.id.length > 0 &&
value.id.length <= MAX_ATTACHMENT_ID_LENGTH &&
value.name.length > 0 &&
value.name.length <= MAX_ATTACHMENT_NAME_LENGTH &&
isSafeAttachmentUrl(value.url) &&
(value.sizeBytes === undefined || (Number.isFinite(value.sizeBytes) && value.sizeBytes >= 0))
);
}
function isDiscordAttachment(value: unknown): value is DiscordAttachment {
if (typeof value !== 'object' || value === null) return false;
const attachment = value as Partial<DiscordAttachment>;
return (
typeof attachment.id === 'string' &&
typeof attachment.name === 'string' &&
typeof attachment.url === 'string' &&
(attachment.contentType === null ||
(typeof attachment.contentType === 'string' &&
attachment.contentType.length <= MAX_ATTACHMENT_MIME_LENGTH)) &&
(attachment.sizeBytes === undefined || typeof attachment.sizeBytes === 'number') &&
hasValidAttachmentBounds(attachment as DiscordAttachment)
);
}
function hasValidAttachmentArray(value: unknown, guard: (attachment: unknown) => boolean): boolean {
return (
Array.isArray(value) &&
value.length <= MAX_CHANNEL_ATTACHMENTS &&
JSON.stringify(value).length <= MAX_ATTACHMENT_METADATA_BYTES &&
value.every(guard)
);
}
function isDiscordIngressEnvelope(value: unknown): value is DiscordIngressEnvelope {
if (typeof value !== 'object' || value === null) return false;
@@ -88,23 +153,49 @@ function isDiscordIngressEnvelope(value: unknown): value is DiscordIngressEnvelo
return false;
}
const payload = envelope.payload as Record<string, unknown>;
return [
payload['correlationId'],
payload['messageId'],
payload['guildId'],
payload['channelId'],
payload['userId'],
payload['conversationId'],
payload['content'],
].every((field: unknown): boolean => typeof field === 'string');
return (
[
payload['correlationId'],
payload['messageId'],
payload['guildId'],
payload['channelId'],
payload['userId'],
payload['conversationId'],
payload['content'],
].every((field: unknown): boolean => typeof field === 'string') &&
(payload['threadId'] === undefined || typeof payload['threadId'] === 'string') &&
(payload['attachments'] === undefined ||
hasValidAttachmentArray(payload['attachments'], isDiscordAttachment))
);
}
function isChannelAttachment(value: unknown): value is ChannelAttachmentDto {
if (typeof value !== 'object' || value === null) return false;
const attachment = value as Partial<ChannelAttachmentDto>;
return (
typeof attachment.id === 'string' &&
typeof attachment.name === 'string' &&
typeof attachment.url === 'string' &&
(attachment.mimeType === null ||
(typeof attachment.mimeType === 'string' &&
attachment.mimeType.length <= MAX_ATTACHMENT_MIME_LENGTH)) &&
(attachment.sizeBytes === undefined || typeof attachment.sizeBytes === 'number') &&
hasValidAttachmentBounds(attachment as ChannelAttachmentDto)
);
}
function isChatSocketMessage(value: unknown): value is ChatSocketMessageDto {
if (typeof value !== 'object' || value === null) return false;
const payload = value as { content?: unknown; conversationId?: unknown };
const payload = value as {
content?: unknown;
conversationId?: unknown;
attachments?: unknown;
};
return (
typeof payload.content === 'string' &&
(payload.conversationId === undefined || typeof payload.conversationId === 'string')
(payload.conversationId === undefined || typeof payload.conversationId === 'string') &&
(payload.attachments === undefined ||
hasValidAttachmentArray(payload.attachments, isChannelAttachment))
);
}
@@ -174,20 +265,24 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
handleDisconnect(client: Socket): void {
this.logger.log(`Client disconnected: ${client.id}`);
const session = this.clientSessions.get(client.id);
if (session) {
for (const [key, session] of this.clientSessions) {
if (session.clientId !== client.id) continue;
session.cleanup();
this.agentService.removeChannel(
session.conversationId,
`websocket:${client.id}`,
session.scope,
);
this.clientSessions.delete(client.id);
this.clientSessions.delete(key);
this.textEgressBuffers.delete(key);
this.thinkingEgressBuffers.delete(key);
this.overflowedEgress.delete(`${key}:agent:text`);
this.overflowedEgress.delete(`${key}:agent:thinking`);
}
this.textEgressBuffers.delete(client.id);
this.thinkingEgressBuffers.delete(client.id);
this.overflowedEgress.delete(this.egressKey(client, 'agent:text'));
this.overflowedEgress.delete(this.egressKey(client, 'agent:thinking'));
}
private clientConversationKey(client: Pick<Socket, 'id'>, conversationId: string): string {
return `${client.id}\u0000${conversationId}`;
}
private getClientScope(client: Socket): ActorTenantScope | null {
@@ -218,7 +313,25 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
discordIngress = this.resolveDiscordIngress(client, rawData);
if (!discordIngress) return;
data = { conversationId: discordIngress.conversationId, content: discordIngress.content };
data = {
conversationId: discordIngress.conversationId,
content: discordIngress.content,
...(discordIngress.attachments
? {
attachments: discordIngress.attachments.map(
(attachment): ChannelAttachmentDto => ({
id: attachment.id,
name: attachment.name,
url: attachment.url,
mimeType: attachment.contentType,
...(attachment.sizeBytes !== undefined
? { sizeBytes: attachment.sizeBytes }
: {}),
}),
),
}
: {}),
};
} else {
if (!isChatSocketMessage(rawData)) {
this.logger.warn(`Rejected malformed chat message from ${client.id}`);
@@ -227,6 +340,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
data = rawData;
}
const conversationId = data.conversationId ?? uuid();
const clientConversationKey = this.clientConversationKey(client, conversationId);
const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID'];
if (discordIngress && !discordServiceUserId) {
this.logger.warn(
@@ -281,7 +395,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
this.logger.log(
`Using /model override "${modelOverride}" for conversation=${conversationId}`,
);
} else if (!resolvedProvider && !resolvedModelId) {
} else if (!resolvedProvider && !resolvedModelId && !discordIngress) {
// No explicit provider/model from client — use routing engine (M4-012)
try {
const routingDecision = await this.routingEngine.resolve(data.content, userId);
@@ -304,12 +418,24 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
}
let resolvedAgentConfigId = data.agentId;
if (discordIngress) {
const binding = this.discordBindingFor(discordIngress, 'send');
const agentConfig = binding
? await this.brain.agents.findById(binding.agentConfigId)
: undefined;
if (!binding || !agentConfig || agentConfig.name !== binding.instanceId) {
throw new Error('Configured Discord logical agent is not provisioned');
}
resolvedAgentConfigId = agentConfig.id;
}
// M5-004: Use existingSessionId as sessionId when available (session reuse)
const sessionIdToCreate = existingSessionId ?? conversationId;
agentSession = await this.agentService.createSession(sessionIdToCreate, {
provider: resolvedProvider,
modelId: resolvedModelId,
agentConfigId: data.agentId,
agentConfigId: resolvedAgentConfigId,
userId,
tenantId: scope.tenantId,
conversationHistory: conversationHistory.length > 0 ? conversationHistory : undefined,
@@ -360,6 +486,17 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
discordUserId: discordIngress?.userId,
}
: {}),
...(data.attachments && data.attachments.length > 0
? {
channelAttachments: data.attachments.map(
(attachment): ChannelAttachmentDto => ({
...attachment,
name: redactSensitiveContent(attachment.name).content,
url: redactSensitiveContent(attachment.url).content,
}),
),
}
: {}),
classifications: redactSensitiveContent(data.content).classifications,
},
},
@@ -374,7 +511,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
// Always clean up previous listener to prevent leak
const existing = this.clientSessions.get(client.id);
const existing = this.clientSessions.get(clientConversationKey);
if (existing) {
existing.cleanup();
}
@@ -389,10 +526,11 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
);
// Preserve routing decision from the existing client session if we didn't get a new one
const prevClientSession = this.clientSessions.get(client.id);
const prevClientSession = this.clientSessions.get(clientConversationKey);
const routingDecisionToStore = sessionRoutingDecision ?? prevClientSession?.lastRoutingDecision;
this.clientSessions.set(client.id, {
this.clientSessions.set(clientConversationKey, {
clientId: client.id,
conversationId,
cleanup,
assistantText: '',
@@ -438,7 +576,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
// Dispatch to agent
try {
await this.agentService.prompt(conversationId, data.content, scope);
await this.agentService.prompt(conversationId, data.content, scope, data.attachments);
} catch (err) {
this.logger.error(
`Agent prompt failed for client=${client.id}, conversation=${conversationId}`,
@@ -645,9 +783,9 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
};
// Emit to all clients currently subscribed to this conversation
for (const [clientId, session] of this.clientSessions) {
for (const session of this.clientSessions.values()) {
if (session.conversationId === conversationId && this.scopesEqual(session.scope, scope)) {
const socket = this.server.sockets.sockets.get(clientId);
const socket = this.server.sockets.sockets.get(session.clientId);
if (socket?.connected) {
socket.emit('session:info', payload);
}
@@ -677,16 +815,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
!this.durableSessions
)
return;
const binding = resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
ingress.guildId,
ingress.channelId,
ingress.userId,
'approve',
);
const binding = this.discordBindingFor(ingress, 'approve');
const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
const agentName = process.env['MOSAIC_AGENT_NAME']?.trim();
if (!actorId || !agentName || binding.instanceId !== agentName) {
const agentName = binding?.instanceId;
if (!actorId || !agentName) {
this.logger.warn(
`Rejected Discord approval without a matching runtime agent from ${client.id}`,
);
@@ -774,13 +906,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
if (!ingress || !approvalRef || !tenantId || !this.runtimeRegistry || !this.durableSessions)
return;
const binding = resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
ingress.guildId,
ingress.channelId,
ingress.userId,
'stop',
);
const binding = this.discordBindingFor(ingress, 'stop');
const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
if (!actorId) return;
@@ -854,17 +980,18 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
return null;
}
try {
const binding = resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
payload.guildId,
payload.channelId,
payload.userId,
operation,
);
const binding = this.discordBindingFor(payload, operation);
if (!binding) {
this.logger.warn(`Rejected unpaired Discord ingress from ${client.id}`);
return null;
}
const expectedConversationId = `${binding.instanceId}:discord:${payload.threadId ?? payload.channelId}`;
if (payload.conversationId !== expectedConversationId) {
this.logger.warn(
`Rejected Discord ingress for a different logical agent from ${client.id}`,
);
return null;
}
} catch {
this.logger.warn(
`Rejected Discord ingress without valid binding configuration from ${client.id}`,
@@ -880,6 +1007,19 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
return payload;
}
private discordBindingFor(
payload: DiscordIngressPayload,
operation: 'send' | 'approve' | 'stop',
) {
return resolveDiscordInteractionBinding(
parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
payload.guildId,
payload.channelId,
payload.userId,
operation,
);
}
private readDiscordAllowlist(name: string): string[] {
return (process.env[name] ?? '')
.split(',')
@@ -958,11 +1098,15 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
const messages = await this.brain.conversations.findMessages(conversationId, userId);
if (messages.length === 0) return [];
return messages.map((msg) => ({
role: msg.role as 'user' | 'assistant' | 'system',
content: msg.content,
createdAt: msg.createdAt,
}));
return messages.map((msg) => {
const attachments = this.persistedChannelAttachments(msg.metadata);
return {
role: msg.role as 'user' | 'assistant' | 'system',
content: msg.content,
createdAt: msg.createdAt,
...(attachments ? { attachments } : {}),
};
});
} catch (err) {
this.logger.error(
`Failed to load conversation history for conversation=${conversationId}`,
@@ -972,6 +1116,14 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
}
private persistedChannelAttachments(metadata: unknown): readonly ChannelAttachmentDto[] | null {
if (typeof metadata !== 'object' || metadata === null) return null;
const attachments = (metadata as { channelAttachments?: unknown }).channelAttachments;
return hasValidAttachmentArray(attachments, isChannelAttachment)
? (attachments as readonly ChannelAttachmentDto[])
: null;
}
private appendAndFlushRedactedEgress(
client: Socket,
conversationId: string,
@@ -979,18 +1131,19 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
buffers: Map<string, string>,
delta: string,
): void {
const key = this.egressKey(client, eventName);
const sessionKey = this.clientConversationKey(client, conversationId);
const key = this.egressKey(client, conversationId, eventName);
if (this.overflowedEgress.has(key)) return;
const buffered = `${buffers.get(client.id) ?? ''}${delta}`;
const buffered = `${buffers.get(sessionKey) ?? ''}${delta}`;
if (buffered.length > MAX_REDACTION_BUFFER_LENGTH) {
buffers.delete(client.id);
buffers.delete(sessionKey);
this.overflowedEgress.add(key);
client.emit(eventName, { conversationId, text: '[REDACTED_STREAM_OVERFLOW]' });
return;
}
buffers.set(client.id, buffered);
buffers.set(sessionKey, buffered);
this.flushRedactedEgress(client, conversationId, eventName, buffers, false);
}
@@ -1006,21 +1159,22 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
buffers: Map<string, string>,
final: boolean,
): void {
const key = this.egressKey(client, eventName);
const sessionKey = this.clientConversationKey(client, conversationId);
const key = this.egressKey(client, conversationId, eventName);
if (this.overflowedEgress.has(key)) {
if (final) this.overflowedEgress.delete(key);
return;
}
const buffered = buffers.get(client.id) ?? '';
const buffered = buffers.get(sessionKey) ?? '';
const releaseLength = final ? buffered.length : this.safeRedactionPrefixLength(buffered);
const released = buffered.slice(0, releaseLength);
const pending = buffered.slice(releaseLength);
if (pending) {
buffers.set(client.id, pending);
buffers.set(sessionKey, pending);
} else {
buffers.delete(client.id);
buffers.delete(sessionKey);
}
if (released) {
@@ -1089,8 +1243,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
return retainedFrom;
}
private egressKey(client: Socket, eventName: 'agent:text' | 'agent:thinking'): string {
return `${client.id}:${eventName}`;
private egressKey(
client: Socket,
conversationId: string,
eventName: 'agent:text' | 'agent:thinking',
): string {
return `${this.clientConversationKey(client, conversationId)}:${eventName}`;
}
private relayEvent(client: Socket, conversationId: string, event: AgentSessionEvent): void {
@@ -1101,26 +1259,27 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
return;
}
const sessionKey = this.clientConversationKey(client, conversationId);
switch (event.type) {
case 'agent_start': {
// Reset accumulation buffers for the new turn
const cs = this.clientSessions.get(client.id);
const cs = this.clientSessions.get(sessionKey);
if (cs) {
cs.assistantText = '';
cs.toolCalls = [];
cs.pendingToolCalls.clear();
}
this.textEgressBuffers.set(client.id, '');
this.thinkingEgressBuffers.set(client.id, '');
this.overflowedEgress.delete(this.egressKey(client, 'agent:text'));
this.overflowedEgress.delete(this.egressKey(client, 'agent:thinking'));
this.textEgressBuffers.set(sessionKey, '');
this.thinkingEgressBuffers.set(sessionKey, '');
this.overflowedEgress.delete(this.egressKey(client, conversationId, 'agent:text'));
this.overflowedEgress.delete(this.egressKey(client, conversationId, 'agent:thinking'));
client.emit('agent:start', { conversationId });
break;
}
case 'agent_end': {
// Gather usage stats from the Pi session
const activeClientSession = this.clientSessions.get(client.id);
const activeClientSession = this.clientSessions.get(sessionKey);
const agentSession = activeClientSession
? this.agentService.getSession(conversationId, activeClientSession.scope)
: undefined;
@@ -1173,7 +1332,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
// Persist the assistant message with metadata
const cs = this.clientSessions.get(client.id);
const cs = this.clientSessions.get(sessionKey);
const userId = (client.data.user as { id: string } | undefined)?.id;
if (cs && userId && cs.assistantText.trim().length > 0) {
const metadata: Record<string, unknown> = {
@@ -1225,7 +1384,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
const assistantEvent = event.assistantMessageEvent;
if (assistantEvent.type === 'text_delta') {
// Keep raw stream material in memory only; persist and emit only redacted text.
const cs = this.clientSessions.get(client.id);
const cs = this.clientSessions.get(sessionKey);
if (cs) {
cs.assistantText += assistantEvent.delta;
}
@@ -1250,7 +1409,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
case 'tool_execution_start': {
// Track pending tool call for later recording
const cs = this.clientSessions.get(client.id);
const cs = this.clientSessions.get(sessionKey);
if (cs) {
cs.pendingToolCalls.set(event.toolCallId, {
toolName: event.toolName,
@@ -1267,7 +1426,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
case 'tool_execution_end': {
// Finalise tool call record
const cs = this.clientSessions.get(client.id);
const cs = this.clientSessions.get(sessionKey);
if (cs) {
const pending = cs.pendingToolCalls.get(event.toolCallId);
cs.toolCalls.push({

View File

@@ -24,6 +24,7 @@ const ENV_KEYS = [
'DISCORD_ALLOWED_CHANNEL_IDS',
'DISCORD_ALLOWED_USER_IDS',
'MOSAIC_AGENT_NAME',
'MOSAIC_AGENT_CONFIG_ID',
] as const;
const savedEnv = new Map<string, string | undefined>();
@@ -33,12 +34,14 @@ function configureDiscordEnv(role: 'admin' | 'member' = 'admin'): void {
process.env['DISCORD_SERVICE_USER_ID'] = 'discord-service';
process.env['DISCORD_SERVICE_TENANT_ID'] = 'tenant-discord';
process.env['MOSAIC_AGENT_NAME'] = 'Nova';
process.env['MOSAIC_AGENT_CONFIG_ID'] = 'agent-config-nova';
process.env['DISCORD_ALLOWED_GUILD_IDS'] = 'guild-001';
process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001';
process.env['DISCORD_ALLOWED_USER_IDS'] = 'user-001';
process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: {
@@ -141,7 +144,7 @@ function createPayload(overrides: Partial<DiscordIngressPayload> = {}): DiscordI
guildId: 'guild-001',
channelId: 'channel-001',
userId: 'user-001',
conversationId: 'discord-channel-001',
conversationId: 'Nova:discord:channel-001',
content: 'hello Tess',
...overrides,
};
@@ -153,6 +156,7 @@ describe('Discord ingress security', () => {
JSON.stringify([
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: { 'user-001': 'admin' },
@@ -170,6 +174,7 @@ describe('Discord ingress security', () => {
[
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: { 'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' } },
@@ -307,41 +312,46 @@ describe('Discord ingress security', () => {
);
});
it.each([
[
'binding',
() => {
process.env['MOSAIC_AGENT_NAME'] = 'Other';
},
],
[
'durable session',
(durable: { getSnapshot: ReturnType<typeof vi.fn> }) => {
durable.getSnapshot.mockResolvedValueOnce({
identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
});
},
],
])(
'rejects approval when the %s targets a different runtime agent',
async (_source, configure) => {
configureDiscordEnv();
const { gateway, client, durable } = discordGateway('admin');
configure(durable);
it('rejects approval when the durable session targets a different logical agent', async () => {
configureDiscordEnv();
const { gateway, client, durable } = discordGateway('admin');
durable.getSnapshot.mockResolvedValueOnce({
identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' },
});
await gateway.handleDiscordApproval(
client as never,
ingressEnvelope('/approve', 'mismatched-agent-approve'),
);
await gateway.handleDiscordApproval(
client as never,
ingressEnvelope('/approve', 'mismatched-agent-approve'),
);
expect(client.emit).toHaveBeenCalledWith('discord:approval', {
correlationId: 'correlation-001',
success: false,
approvalId: undefined,
expiresAt: undefined,
});
},
);
expect(client.emit).toHaveBeenCalledWith('discord:approval', {
correlationId: 'correlation-001',
success: false,
approvalId: undefined,
expiresAt: undefined,
});
});
it('rejects privileged envelopes with a forged current conversation route', async () => {
configureDiscordEnv();
const { gateway, client } = discordGateway('admin');
await gateway.handleDiscordApproval(
client as never,
ingressEnvelope('/approve', 'forged-approval-route', {
conversationId: 'Nova:discord:other-channel',
}),
);
await gateway.handleDiscordStop(
client as never,
ingressEnvelope('/stop forged', 'forged-stop-route', {
conversationId: 'Nova:discord:other-channel',
}),
);
expect(client.emit).not.toHaveBeenCalledWith('discord:approval', expect.anything());
expect(client.emit).not.toHaveBeenCalledWith('discord:stop', expect.anything());
});
it('rejects unpaired and non-admin Discord users for approval and stop', async () => {
configureDiscordEnv();
@@ -398,6 +408,267 @@ describe('Discord ingress security', () => {
]);
});
it.each([
'https://user:password@cdn.example.test/diagram.png',
'https://cdn.example.test/diagram.png?token=secret',
'https://cdn.example.test/diagram.png?X-Amz-Signature=secret',
'https://cdn.example.test/diagram.png?auth=secret',
'https://cdn.example.test/diagram.png?hm=secret',
])('rejects credential-bearing attachment URLs before gateway dispatch', async (url) => {
configureDiscordEnv();
const { gateway, client } = discordGateway('admin');
await gateway.handleMessage(
client as never,
ingressEnvelope('', `credential-url-${url.length}`, {
conversationId: 'Nova:discord:channel-001',
attachments: [
{ id: 'attachment-credential', name: 'diagram.png', url, contentType: 'image/png' },
],
}),
);
expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything());
});
it("selects each binding's trusted logical-agent config when creating Discord sessions", async () => {
configureDiscordEnv();
process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001,channel-002';
process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: {
'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' },
},
},
{
instanceId: 'Orion',
agentConfigId: 'agent-config-orion',
guildId: 'guild-001',
channelId: 'channel-002',
pairedUsers: {
'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' },
},
},
]);
const session = {
provider: 'configured-provider',
modelId: 'configured-model',
piSession: {
thinkingLevel: 'medium',
getAvailableThinkingLevels: (): string[] => ['medium'],
},
};
const createSession = vi.fn().mockResolvedValue(session);
const agentService = {
getSession: vi.fn().mockReturnValue(undefined),
createSession,
recordMessage: vi.fn(),
onEvent: vi.fn().mockReturnValue((): void => undefined),
addChannel: vi.fn(),
prompt: vi.fn().mockResolvedValue(undefined),
};
const brain = {
agents: {
findById: vi.fn((id: string) =>
Promise.resolve({
id,
name: id === 'agent-config-orion' ? 'Orion' : 'Nova',
}),
),
},
conversations: {
findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }),
findMessages: vi.fn().mockResolvedValue([]),
create: vi.fn().mockResolvedValue(undefined),
update: vi.fn().mockResolvedValue(undefined),
addMessage: vi.fn().mockResolvedValue(undefined),
},
};
const routingEngine = { resolve: vi.fn() };
const gateway = new ChatGateway(
agentService as never,
{} as never,
brain as never,
{} as never,
{} as never,
routingEngine as never,
);
const client = {
id: 'discord-client-new-session',
data: { discordService: true },
emit: vi.fn(),
};
await gateway.handleMessage(
client as never,
ingressEnvelope('start configured session', 'configured-session-001', {
conversationId: 'Nova:discord:channel-001',
}),
);
await gateway.handleMessage(
client as never,
ingressEnvelope('start second configured session', 'configured-session-002', {
channelId: 'channel-002',
conversationId: 'Orion:discord:channel-002',
}),
);
expect(createSession).toHaveBeenCalledWith(
'Nova:discord:channel-001',
expect.objectContaining({
agentConfigId: 'agent-config-nova',
userId: 'discord-service',
tenantId: 'tenant-discord',
}),
);
expect(createSession).toHaveBeenCalledWith(
'Orion:discord:channel-002',
expect.objectContaining({ agentConfigId: 'agent-config-orion' }),
);
expect(routingEngine.resolve).not.toHaveBeenCalled();
});
it('retains validated persisted attachments in resumed conversation history', async () => {
const attachment = {
id: 'attachment-history',
name: 'diagram.png',
url: 'https://cdn.example.test/diagram.png',
mimeType: 'image/png',
sizeBytes: 4_096,
};
const gateway = new ChatGateway(
{} as never,
{} as never,
{
conversations: {
findMessages: vi.fn().mockResolvedValue([
{
role: 'user',
content: '',
createdAt: new Date('2026-07-14T12:00:00.000Z'),
metadata: { channelAttachments: [attachment] },
},
]),
},
} as never,
{} as never,
{} as never,
{} as never,
) as unknown as {
loadConversationHistory(
conversationId: string,
userId: string,
): Promise<Array<{ attachments?: readonly (typeof attachment)[] }>>;
};
await expect(
gateway.loadConversationHistory('Nova:discord:channel-001', 'discord-service'),
).resolves.toEqual([expect.objectContaining({ attachments: [attachment] })]);
});
it('rejects malformed signed attachment payloads before gateway dispatch', async () => {
configureDiscordEnv();
const { gateway, client } = discordGateway('admin');
const malformedPayload: Record<string, unknown> = {
...createPayload({
messageId: 'malformed-attachments-001',
conversationId: 'Nova:discord:channel-001',
}),
attachments: { id: 'not-an-array' },
};
const envelope = createDiscordIngressEnvelope(
malformedPayload as unknown as DiscordIngressPayload,
SERVICE_TOKEN,
);
await gateway.handleMessage(client as never, envelope);
expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything());
});
it('preserves authenticated attachment metadata through persistence and agent dispatch', async () => {
configureDiscordEnv();
const prompt = vi.fn().mockResolvedValue(undefined);
const addMessage = vi.fn().mockResolvedValue(undefined);
const session = {
provider: 'test-provider',
modelId: 'test-model',
piSession: {
thinkingLevel: 'medium',
getAvailableThinkingLevels: (): string[] => ['medium'],
},
};
const agentService = {
getSession: vi.fn().mockReturnValue(session),
recordMessage: vi.fn(),
onEvent: vi.fn().mockReturnValue((): void => undefined),
addChannel: vi.fn(),
prompt,
};
const brain = {
conversations: {
findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }),
create: vi.fn().mockResolvedValue(undefined),
update: vi.fn().mockResolvedValue(undefined),
addMessage,
},
};
const gateway = new ChatGateway(
agentService as never,
{} as never,
brain as never,
{} as never,
{} as never,
{} as never,
);
const client = {
id: 'discord-client-001',
data: { discordService: true },
emit: vi.fn(),
};
const attachment = {
id: 'attachment-001',
name: 'diagram.png',
url: 'https://cdn.example.test/diagram.png',
contentType: 'image/png',
sizeBytes: 4_096,
};
await gateway.handleMessage(
client as never,
ingressEnvelope('', 'attachment-message-001', {
conversationId: 'Nova:discord:channel-001',
attachments: [attachment],
}),
);
const expectedAttachment = {
id: attachment.id,
name: attachment.name,
url: attachment.url,
mimeType: attachment.contentType,
sizeBytes: attachment.sizeBytes,
};
expect(prompt).toHaveBeenCalledWith(
'Nova:discord:channel-001',
'',
{ userId: 'discord-service', tenantId: 'tenant-discord' },
[expectedAttachment],
);
expect(addMessage).toHaveBeenCalledWith(
expect.objectContaining({
conversationId: 'Nova:discord:channel-001',
metadata: expect.objectContaining({ channelAttachments: [expectedAttachment] }),
}),
'discord-service',
);
});
it('accepts a thread message through its allowed bound parent channel', () => {
const emitted = vi.fn();
const plugin = new DiscordPlugin({
@@ -410,6 +681,7 @@ describe('Discord ingress security', () => {
interactionBindings: [
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: { 'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' } },

View File

@@ -61,6 +61,16 @@ function requiredDiscordAllowlist(name: string): string[] {
return value;
}
function optionalPositiveInteger(name: string): number | undefined {
const raw = process.env[name];
if (raw === undefined) return undefined;
const value = Number(raw);
if (!Number.isInteger(value) || value <= 0) {
throw new Error(`${name} must be a positive integer when configured`);
}
return value;
}
function createPluginRegistry(): IChannelPlugin[] {
const plugins: IChannelPlugin[] = [];
const discordToken = process.env['DISCORD_BOT_TOKEN'];
@@ -82,6 +92,10 @@ function createPluginRegistry(): IChannelPlugin[] {
guildId: discordGuildId,
gatewayUrl: discordGatewayUrl,
serviceToken: discordServiceToken,
messageRateLimitPerMinute: optionalPositiveInteger(
'DISCORD_MESSAGE_RATE_LIMIT_PER_MINUTE',
),
threadRateLimitPerMinute: optionalPositiveInteger('DISCORD_THREAD_RATE_LIMIT_PER_MINUTE'),
allowedGuildIds: requiredDiscordAllowlist('DISCORD_ALLOWED_GUILD_IDS'),
allowedChannelIds: requiredDiscordAllowlist('DISCORD_ALLOWED_CHANNEL_IDS'),
allowedUserIds: requiredDiscordAllowlist('DISCORD_ALLOWED_USER_IDS'),

View File

@@ -221,6 +221,100 @@ Delivery uses five gated milestones: runtime contracts/security; Pi service/stat
---
## Official Channel Plugin Workstream (#756)
### Problem and Objective
The Discord plugin currently couples Discord event handling, gateway bridging, and reply routing in one implementation and activates only on mentions. Mosaic needs an official channel adapter that behaves the same no matter whether the bound logical agent currently runs through Claude, Codex, Pi, OpenCode, or a future harness. The Discord connection and conversation address must remain stable while the gateway changes the runtime provider behind that logical session.
The objective is to make Discord the first implementation of a transport-neutral official channel contract, with explicit authorization and deterministic channel/thread routing that future Matrix, Slack, and other adapters can share.
### Scope
#### In Scope
1. `CHN-001`: Transport-neutral channel adapter, route, message, attachment, authorization-principal, response-target, and health contracts in `@mosaicstack/types`, including trusted per-binding logical-agent configuration selection.
2. `CHN-002`: Stable channel conversation addresses based on logical agent plus channel/thread identity; harness, model, and runtime-provider IDs are forbidden from channel session keys.
3. `DSC-001`: An authorized untagged message in a configured agent-bound channel routes to the agent and receives its response in that channel.
4. `DSC-002`: A bot mention in a configured parent channel creates a Discord thread, or reuses the thread already attached to that same native message; the mentioned turn and subsequent thread turns route and respond in that thread.
5. `DSC-003`: A message already inside an authorized thread inherits authorization from its configured parent and never attempts a nested thread.
6. `DSC-004`: Guild, parent channel, user, pairing, and role authorization remains default-deny before thread creation or gateway dispatch.
7. `DSC-005`: Discord service authentication, HMAC envelope integrity, replay protection, attachments, approvals, response chunking, and correlation behavior remain intact.
8. `DSC-006`: The Discord adapter exposes lifecycle and health behavior through the shared channel contract without importing a harness SDK.
#### Out of Scope
1. The logical-agent lease, fencing epoch, execution grant, checkpoint, or cross-harness takeover implementation tracked by #754/#755.
2. Dynamic Discord authorization administration in the web UI.
3. Multi-guild tenant isolation, DMs, slash commands, voice, reactions, or production bot deployment.
4. Implementing Matrix or Slack adapters in this slice.
### Non-Functional Requirements
1. **Security:** no thread or dispatch side effect occurs until guild, parent channel, user, pairing, role, and bounded per-user/channel rate checks pass; attachment metadata is shape- and size-bounded; credentials never enter source, messages, session keys, or logs.
2. **Portability:** channel contracts and stable conversation IDs contain no Claude, Codex, Pi, OpenCode, model, process, or provider-specific field; each configuration-owned binding selects its trusted logical agent without changing the channel identity.
3. **Reliability:** repeated messages for one channel/thread resolve the same conversation handle; reconnecting the adapter does not require a harness-specific rebinding.
4. **Maintainability:** Discord-specific API translation stays in the Discord package; gateway and future adapters depend on transport-neutral contracts.
5. **Observability:** thread creation or routing failure is reported without message content or credential material.
### Acceptance Criteria
1. `AC-CHN-01`: Contract and behavior tests prove the plugin route contains only logical agent plus channel/thread identity and produces the same stable conversation handle regardless of underlying harness selection.
2. `AC-CHN-02`: A mentioned authorized parent-channel message creates a thread (or reuses its already-attached thread), dispatches to the thread conversation, and targets the response to that thread.
3. `AC-CHN-03`: An untagged authorized parent-channel message dispatches to the parent conversation and targets the response to the parent channel.
4. `AC-CHN-04`: Untagged follow-ups inside an authorized thread dispatch and respond in that same thread without creating a nested thread.
5. `AC-CHN-05`: Unauthorized guilds, channels, users, unpaired users, insufficient roles, and rate-limited senders produce no thread and no gateway dispatch.
6. `AC-CHN-06`: Shared channel contracts are exported from `@mosaicstack/types`, Discord implements the lifecycle/health seam, and no harness SDK is imported by the plugin.
7. `AC-CHN-07`: Focused routing/auth tests, package tests, typecheck, lint, formatting, coverage, independent code/security review, and terminal-green CI pass.
### Constraints, Risks, and Assumptions
- Dependency: Mosaic gateway remains the policy, durable-session, audit, and runtime-provider boundary.
- Constraint: This work must not modify orchestrator-to-Pi migration or #754/#755 lease/fencing files.
- Risk: accepting untagged messages could create noisy or unintended agent input. Mitigation: only explicitly configured channels and paired, role-authorized users are accepted, with bounded per-user/channel message and thread rates.
- Risk: Discord thread creation can fail because of channel permissions, archived state, or API rate limits. Mitigation: fail without dispatching a turn whose response destination cannot be honored, and emit sanitized diagnostics.
- `ASSUMPTION:` Configured channels are dedicated agent interaction surfaces, so authorized untagged human messages are intentional agent input.
- `ASSUMPTION:` Mention in a parent channel selects a public thread; messages already in a thread remain there because Discord has no nested threads.
- `ASSUMPTION:` One Discord bot may serve multiple configuration-owned logical-agent bindings.
- `ASSUMPTION:` Static allowlists and paired-user roles are the authorization administration surface for this slice.
### Testing and Delivery Intent
Use TDD for remote-ingress routing and permission boundaries. Required evidence includes parent-channel mention, untagged parent message, existing-thread follow-up, existing-thread mention, thread reuse, unauthorized side-effect denial, stable harness-neutral conversation identity, adapter health, and regression coverage for signed envelopes and approvals. Deliver through issue #756, a reviewed squash PR to `main`, terminal-green CI, and issue closure.
---
## Mos Runtime Portability Workstream (MOS-PORT)
### Problem and Objective
Mos is currently identified partly by a harness-native session and communication process. Replacement/rebinding exists, but no gateway-enforced logical identity or fencing prevents a stale harness from continuing to reply or execute effects after takeover.
The objective is to make Mos a server-derived logical Mosaic identity whose authority can move safely among runtime connectors. The gateway owns identity, lease, policy, and audit; harnesses remain replaceable adapters.
### M1 Requirements
1. `MOS-PORT-ID-001`: Define a normalized logical-agent identity independent of Claude Code, Pi, Codex, tmux, Matrix, and provider-native session IDs.
2. `MOS-PORT-LEASE-001`: Persist one exclusive connector lease per tenant/logical-agent/binding with CAS acquisition, monotonic fencing epoch, TTL, heartbeat, explicit release, and takeover.
3. `MOS-PORT-FENCE-001`: Bind every connector dispatch/execution grant to the current server-derived tenant, logical identity, binding, connector, scopes, expiry, and lease epoch.
4. `MOS-PORT-FENCE-002`: Reject and audit stale, expired, forged, cross-tenant, cross-binding, and unauthorized grants before connector, channel, provider, or tool side effects.
5. `MOS-PORT-OBS-001`: Emit credential-safe correlation/audit events for lease acquire, renew, takeover, reject, release, and expiry.
6. `MOS-PORT-ARCH-001`: Runtime/provider adapters consume normalized lease context without adding harness-native schemas to Mosaic core.
### M1 Acceptance Criteria
1. `AC-MOS-PORT-01`: Two contenders for one binding cannot simultaneously hold current authority under concurrency.
2. `AC-MOS-PORT-02`: Successful takeover increments the fencing epoch and every operation from the old epoch fails closed before side effects.
3. `AC-MOS-PORT-03`: Gateway/database restart preserves lease and epoch state; expired leases can be recovered only through the authorized takeover path.
4. `AC-MOS-PORT-04`: Cross-tenant, cross-agent, cross-binding, forged, and expired lease/grant cases are denied and audited.
5. `AC-MOS-PORT-05`: Unit, migration, repository close/reopen, concurrency, abuse, gateway integration, independent security review, CI, and documentation gates pass.
### Deferred to Later #754 Milestones
Canonical checkpoint/handoff payloads, exactly-once connector receipts, concrete Claude/Pi/Codex adapters, channel cutover, and full cross-harness failover/rollback E2E are explicitly out of M1 scope.
---
## Architecture
### High-Level System Diagram
@@ -575,7 +669,8 @@ Discord remote control channel. Architecture inspired by OpenClaw (https://githu
- Single-guild binding only (v0.1.0) — prevents data leaks between servers
- Receives Discord messages, dispatches through gateway routing
- Streams agent responses back to Discord (chunked for 2000-char limit)
- Supports mention-based activation, thread management for multi-turn
- Routes authorized untagged messages in-channel; mentions create threads (or reuse the same message's attached thread) for multi-turn topics
- Uses stable logical-agent/channel conversation addresses independent of the active harness/provider
- Bot pairing and permission management (Discord user → Mosaic user mapping)
- DM support for private conversations
@@ -748,10 +843,12 @@ Telegram remote control channel.
### FR-9: Remote Control — Discord
- Discord bot that connects to the gateway
- Mention-based activation in channels
- Discord bot that connects to the gateway through a transport-neutral channel adapter contract
- Authorized messages in configured agent-bound channels work without a mention and respond in-channel
- Mentions in parent channels create threads, or reuse a thread already attached to that same native message, for multi-turn conversations
- Messages already in a thread remain there without requiring repeated mentions
- Stable logical-agent/channel conversation identity survives underlying harness/provider changes
- DM support for private conversations
- Thread creation for multi-turn conversations
- Chunked message delivery (Discord 2000-char limit)
- Bot configuration via web dashboard
- Permission management (which Discord users/roles can interact)
@@ -935,10 +1032,13 @@ Telegram remote control channel.
### AC-3: Discord Remote Control
- [ ] Discord bot connects and responds to mentions
- [ ] Messages route through gateway to agent pool
- [ ] Discord bot connects through the harness-neutral channel contract
- [ ] Authorized untagged channel messages route through the gateway and respond in-channel
- [ ] Mentioned parent-channel messages create a thread (or reuse their already-attached thread) and respond there
- [ ] Existing-thread follow-ups stay in the thread without repeated mentions
- [ ] Channel/session identity remains stable while the underlying harness/provider changes
- [ ] Responses stream back to Discord (chunked)
- [ ] Thread creation for multi-turn conversations
- [ ] Unauthorized guilds, channels, users, pairings, and roles create no thread and dispatch no message
### AC-4: Gateway Orchestration
@@ -1137,7 +1237,7 @@ All work is **alpha** (< 0.1.0) until Jason approves 0.1.0 beta release.
6. ASSUMPTION: **Log summarization uses Haiku-tier LLM by default, configurable.** Haiku is well-suited for summarization (compression, not generation — source material is in context). Guardrails: structured output via Zod schema (force extraction of decisions/tools/outcomes/errors as discrete fields), chunked per-session processing (no bulk conflation), extraction-focused prompts. Raw logs stay in hot tier (7 days) as safety net. Users can override the summarization model via routing engine config if they want higher fidelity. Rationale: Haiku is 10-20x cheaper than Sonnet; log summarization runs on schedule against large volumes where cost matters.
7. ASSUMPTION: **Discord plugin starts minimal and single-guild only**DM support, mention-based channel activation, thread management, chunked responses. Single guild binding to prevent data leaks between servers. Advanced features (voice, components, slash commands, multi-guild) are post-beta. Rationale: Proven pattern from OpenClaw; ship core interaction first; data isolation is non-negotiable.
7. ASSUMPTION: **Discord plugin starts minimal and single-guild only**explicitly configured agent-bound channels accept authorized untagged messages in-channel, while mentions create threads or reuse a thread already attached to that same native message; responses are chunked. Single guild binding prevents data leaks between servers. DM support, voice, components, slash commands, and multi-guild operation are post-beta. Rationale: Ship the requested core interaction model while preserving default-deny data isolation.
8. ASSUMPTION: **Telegram plugin is lower priority than Discord** and may ship as v0.0.7 or later if Discord takes longer than expected. Rationale: Jason indicated Discord as the high-priority remote channel.

View File

@@ -1,5 +1,13 @@
# Documentation Sitemap
## Official channel plugins
- [Channel protocol architecture](architecture/channel-protocol.md) — shared lifecycle, message, stable-route, authorization, and response-target contracts.
- [Discord administrator configuration](guides/admin-guide.md#discord-ingress-security) — secrets, allowlists, bindings, role policy, and thread permissions.
- [Discord user workflow](tess/USER-GUIDE.md#discord-conversations) — in-channel messages, mention-created threads, and runtime-transparent continuity.
- [Channel plugin authoring](tess/PLUGIN-GUIDE.md#official-channel-adapter-contract) — requirements for future Matrix, Slack, and other official adapters.
- [Discord package guide](../plugins/discord/README.md) — package behavior, configuration shape, and development commands.
## Native Kanban and canonical task SOT
- [Canonical requirements](requirements/native-kanban-sot.md) — ratified P0P3 requirements and acceptance criteria.
@@ -48,3 +56,5 @@
- [Optional AI egress gateway ADR](architecture/ADR-MOS-EGRESS-GATEWAYS.md) — placement and gates for LiteLLM, Bifrost, and purpose-built translation proxies.
- [Runtime-neutral Mos identity and failover mission](https://git.mosaicstack.dev/mosaicstack/stack/issues/754)
- [Logical identity and connector lease/fencing implementation](https://git.mosaicstack.dev/mosaicstack/stack/issues/755)
- [M1 logical identity and fencing architecture](architecture/mos-runtime-portability-m1.md)
- [M1 connector lease operations](guides/mos-connector-lease-operations.md)

View File

@@ -1,9 +1,9 @@
# Channel Protocol Architecture
**Status:** Draft
**Status:** Official adapter baseline implemented by #756; extended registry/multiplexing remains iterative
**Authors:** Mosaic Core Team
**Last Updated:** 2026-03-22
**Covers:** M7-001 (IChannelAdapter interface), M7-002 (ChannelMessage protocol), M7-003 (Matrix integration design), M7-004 (conversation multiplexing), M7-005 (remote auth bridging), M7-006 (agent-to-agent communication via Matrix), M7-007 (multi-user isolation in Matrix)
**Last Updated:** 2026-07-14
**Covers:** M7-001 (OfficialChannelAdapter interface), M7-002 (ChannelMessageDto protocol), M7-003 (Matrix integration design), M7-004 (conversation multiplexing), M7-005 (remote auth bridging), M7-006 (agent-to-agent communication via Matrix), M7-007 (multi-user isolation in Matrix)
---
@@ -11,93 +11,80 @@
The channel protocol defines a unified abstraction layer between Mosaic's core messaging infrastructure and the external communication channels it supports (Matrix, Discord, Telegram, TUI, WebUI, and future channels).
The protocol consists of two main contracts:
The implemented baseline is exported from `@mosaicstack/types` and consists of four contract groups:
1. `IChannelAdapter` — the interface each channel driver must implement.
2. `ChannelMessage` — the canonical message format that flows through the system.
1. `OfficialChannelAdapter` — transport lifecycle and connection health.
2. `ChannelMessageDto` / `ChannelAttachmentDto` — canonical transport data.
3. `ChannelConversationRouteDto` — stable logical-agent conversation and authorization address.
4. `ChannelResponseTargetDto` — channel/thread destination for replies.
All channel-specific translation logic lives inside the adapter implementation. The rest of Mosaic works exclusively with `ChannelMessage` objects.
All channel-specific translation logic lives inside the adapter implementation. Runtime selection does not: gateway durable-session and provider services may rebind the logical session from Claude to Codex, Pi, OpenCode, or another harness without reconnecting the channel adapter.
---
## M7-001: IChannelAdapter Interface
## M7-001: OfficialChannelAdapter Interface
```typescript
interface IChannelAdapter {
/**
* Stable, lowercase identifier for this channel (e.g. "matrix", "discord").
* Used as a namespace key in registry lookups and log metadata.
*/
interface OfficialChannelAdapter {
/** Stable, lowercase adapter identifier such as "discord" or "matrix". */
readonly name: string;
/**
* Establish a connection to the external channel backend.
* Called once at application startup. Must be idempotent (safe to call
* when already connected).
*/
connect(): Promise<void>;
/**
* Gracefully disconnect from the channel backend.
* Must flush in-flight sends and release resources before resolving.
*/
disconnect(): Promise<void>;
/**
* Return the current health of the adapter connection.
* Used by the admin health endpoint and alerting.
*
* - "connected" — fully operational
* - "degraded" — partial connectivity (e.g. read-only, rate-limited)
* - "disconnected" — no connection to channel backend
*/
health(): Promise<{ status: 'connected' | 'degraded' | 'disconnected' }>;
/**
* Register an inbound message handler.
* The adapter calls `handler` for every message received from the channel.
* Multiple calls replace the previous handler (last-write-wins).
* The handler is async; the adapter must not deliver new messages until
* the previous handler promise resolves (back-pressure).
*/
onMessage(handler: (msg: ChannelMessage) => Promise<void>): void;
/**
* Send a ChannelMessage to the given channel/room/conversation.
* `channelId` is the channel-native identifier (e.g. Matrix room ID,
* Discord channel snowflake, Telegram chat ID).
*/
sendMessage(channelId: string, msg: ChannelMessage): Promise<void>;
/**
* Map a channel-native user identifier to the Mosaic internal userId.
* Returns null when no matching Mosaic account exists for the given
* channelUserId (anonymous or unlinked user).
*/
mapIdentity(channelUserId: string): Promise<string | null>;
/** Establish both native-channel and gateway connections. */
start(): Promise<void>;
/** Gracefully close connections and release resources. */
stop(): Promise<void>;
/** Best-effort health; ordinary disconnection is a result, not an exception. */
health(): Promise<{
status: 'connected' | 'degraded' | 'disconnected';
detail?: string;
}>;
}
```
The small lifecycle seam lets the gateway host official plugins uniformly without moving native message translation into gateway core. Message ingress remains adapter-owned; gateway policy, durable session routing, auditing, and runtime/provider selection remain gateway-owned.
### Stable conversation route
```typescript
interface ChannelConversationRouteDto {
bindingId: string;
logicalAgentId: string;
conversationId: string;
channelName: string;
authorizationChannelId: string;
responseTarget: { channelId: string; threadId?: string };
}
```
Harness, provider, model, process, and native runtime-session identifiers are forbidden from this route. Runtime adapters consume the gateway's durable logical-session binding; channel adapters consume only the stable route and response target.
### Typed ingress and egress ports
`ChannelIngressPort` is the transport-neutral direct-integration seam for official adapters. The current deployed Discord adapter preserves its existing HMAC-signed Socket.IO compatibility ingress so gateway-side service authentication, replay protection, approval handling, and correlation semantics remain unchanged; it normalizes the same `ChannelIngressDto` before signing. The adapter uses a supplied `ChannelIngressPort` directly when a future gateway registration provides one. New adapters must use the shared ports rather than adding channel branches to gateway core.
`ChannelBindingDto` contains the configuration-owned workspace/channel→logical-agent mapping and paired external principals; credentials are absent. After native allowlist, pairing, and role checks pass, an adapter submits `ChannelIngressDto` to `ChannelIngressPort.receive()`. It includes the normalized message, `ChannelAuthorizedPrincipalDto`, operation, correlation ID, native message ID, and stable route. Unauthorized input never reaches the port.
Gateway policy and runtime routing produce `ChannelEgressDto`, which `ChannelEgressPort.send()` delivers to the route's response target. Discord's existing HMAC envelope is its authenticated wire encoding of this boundary; future Matrix/Slack adapters use their native authenticated transports while preserving the same actor/operation/correlation semantics.
### Adapter Registration
Adapters are registered with the `ChannelRegistry` service at startup. The registry calls `connect()` on each adapter and monitors `health()` on a configurable interval (default: 30 s).
Adapters are registered with the gateway plugin host at startup. The host calls `start()`/`stop()` and may monitor `health()` on a configurable interval. A richer dynamic `ChannelRegistry` remains a compatible future extension of this lifecycle contract.
```
ChannelRegistry
└── register(adapter: IChannelAdapter): void
└── getAdapter(name: string): IChannelAdapter | null
└── listAdapters(): IChannelAdapter[]
└── register(adapter: OfficialChannelAdapter): void
└── getAdapter(name: string): OfficialChannelAdapter | null
└── listAdapters(): OfficialChannelAdapter[]
└── healthAll(): Promise<Record<string, AdapterHealth>>
```
---
## M7-002: ChannelMessage Protocol
## M7-002: ChannelMessageDto Protocol
### Canonical Message Format
```typescript
interface ChannelMessage {
interface ChannelMessageDto {
/**
* Globally unique message ID.
* Format: UUID v4. Generated by the adapter when receiving, or by Mosaic
@@ -110,6 +97,7 @@ interface ChannelMessage {
* The adapter populates this from the inbound message.
* For outbound messages, the caller supplies the target channel.
*/
channelName: string;
channelId: string;
/**
@@ -119,7 +107,7 @@ interface ChannelMessage {
senderId: string;
/** Sender classification. */
senderType: 'user' | 'agent' | 'system';
senderKind: 'user' | 'agent' | 'system';
/**
* Textual content of the message.
@@ -136,7 +124,7 @@ interface ChannelMessage {
* - "image" — binary image; content is empty, see attachments
* - "file" — binary file; content is empty, see attachments
*/
contentType: 'text' | 'markdown' | 'code' | 'image' | 'file';
contentKind: 'text' | 'markdown' | 'code' | 'image' | 'file';
/**
* Arbitrary key-value metadata for channel-specific extension fields.
@@ -144,7 +132,7 @@ interface ChannelMessage {
* Adapters should store channel-native IDs here so round-trip correlation
* is possible without altering the canonical fields.
*/
metadata: Record<string, unknown>;
metadata: Readonly<Record<string, ChannelMetadataValue>>;
/**
* Optional thread or reply-chain identifier.
@@ -163,18 +151,21 @@ interface ChannelMessage {
* Binary or URI-referenced attachments.
* Each attachment carries its MIME type and a URL or base64 payload.
*/
attachments?: ChannelAttachment[];
attachments?: readonly ChannelAttachmentDto[];
/** Wall-clock timestamp when the message was sent/received. */
timestamp: Date;
/** ISO-8601 wall-clock timestamp when the message was sent/received. */
timestamp: string;
}
interface ChannelAttachment {
/** Filename or identifier. */
interface ChannelAttachmentDto {
/** Channel-native attachment identifier. */
id: string;
/** Filename or display name. */
name: string;
/** MIME type (e.g. "image/png", "application/pdf"). */
mimeType: string;
/** MIME type when supplied by the channel. */
mimeType: string | null;
/**
* URL pointing to the attachment, OR a `data:` URI with base64 payload.
@@ -192,23 +183,23 @@ interface ChannelAttachment {
## Channel Translation Reference
The following sections document how each supported channel maps its native message format to and from `ChannelMessage`.
The following sections document how each supported channel maps its native message format to and from `ChannelMessageDto`.
### Matrix
| ChannelMessage field | Matrix equivalent |
| -------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Matrix event ID (`$...`) |
| `channelId` | Matrix room ID (`!roomid:homeserver`) |
| `senderId` | Matrix user ID (`@user:homeserver`) |
| `senderType` | Always `"user"` for inbound; `"agent"` or `"system"` for outbound |
| `content` | `event.content.body` |
| `contentType` | `"markdown"` if `msgtype = m.text` and body contains markdown; `"text"` otherwise; `"image"` for `m.image`; `"file"` for `m.file` |
| `threadId` | `event.content['m.relates_to']['event_id']` when `rel_type = m.thread` |
| `replyToId` | Mosaic ID looked up from `event.content['m.relates_to']['m.in_reply_to']['event_id']` |
| `attachments` | Populated from `url` in `m.image` / `m.file` events |
| `timestamp` | `new Date(event.origin_server_ts)` |
| `metadata` | `{ channelMessageId, roomId, eventType, unsigned }` |
| ChannelMessageDto field | Matrix equivalent |
| ----------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Matrix event ID (`$...`) |
| `channelId` | Matrix room ID (`!roomid:homeserver`) |
| `senderId` | Matrix user ID (`@user:homeserver`) |
| `senderKind` | Always `"user"` for inbound; `"agent"` or `"system"` for outbound |
| `content` | `event.content.body` |
| `contentKind` | `"markdown"` if `msgtype = m.text` and body contains markdown; `"text"` otherwise; `"image"` for `m.image`; `"file"` for `m.file` |
| `threadId` | `event.content['m.relates_to']['event_id']` when `rel_type = m.thread` |
| `replyToId` | Mosaic ID looked up from `event.content['m.relates_to']['m.in_reply_to']['event_id']` |
| `attachments` | Populated from `url` in `m.image` / `m.file` events |
| `timestamp` | `new Date(event.origin_server_ts)` |
| `metadata` | `{ channelMessageId, roomId, eventType, unsigned }` |
**Outbound:** Adapter sends `m.room.message` with `msgtype = m.text` (or `m.notice` for system messages). Markdown content is sent with `format = org.matrix.custom.html` and a rendered HTML body.
@@ -216,21 +207,34 @@ The following sections document how each supported channel maps its native messa
### Discord
| ChannelMessage field | Discord equivalent |
| -------------------- | ----------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Discord message snowflake |
| `channelId` | Discord channel ID (snowflake string) |
| `senderId` | Discord user ID (snowflake) |
| `senderType` | `"user"` for human members; `"agent"` for bot messages |
| `content` | `message.content` |
| `contentType` | `"markdown"` (Discord uses a markdown-like syntax natively) |
| `threadId` | `message.thread.id` when the message is inside a thread channel |
| `replyToId` | Mosaic ID looked up from `message.referenced_message.id` |
| `attachments` | `message.attachments` mapped to `ChannelAttachment` |
| `timestamp` | `new Date(message.timestamp)` |
| `metadata` | `{ channelMessageId, guildId, channelType, mentions, embeds }` |
| ChannelMessageDto field | Discord equivalent |
| ----------------------- | ----------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Discord message snowflake |
| `channelId` | Discord channel ID (snowflake string) |
| `senderId` | Discord user ID (snowflake) |
| `senderKind` | `"user"` for human members; `"agent"` for bot messages |
| `content` | `message.content` |
| `contentKind` | `"markdown"` (Discord uses a markdown-like syntax natively) |
| `threadId` | `message.thread.id` when the message is inside a thread channel |
| `replyToId` | Mosaic ID looked up from `message.referenced_message.id` |
| `attachments` | `message.attachments` mapped to `ChannelAttachmentDto` |
| `timestamp` | `new Date(message.timestamp)` |
| `metadata` | `{ channelMessageId, guildId, channelType, mentions, embeds }` |
**Outbound:** Adapter calls Discord REST `POST /channels/{id}/messages`. Markdown content is sent as-is (Discord renders it). For `contentType = "code"` the adapter wraps in triple-backtick fences with the `metadata.language` tag.
**Outbound:** Adapter calls Discord REST `POST /channels/{id}/messages`. Markdown content is sent as-is (Discord renders it). For `contentKind = "code"` the adapter wraps in triple-backtick fences with the `metadata.language` tag.
### Discord routing and thread policy
A configured Discord binding maps `(guildId, parentChannelId)` to a stable logical agent and a trusted gateway agent-config ID. Gateway verifies that configuration's name matches the binding logical agent before session creation. The stable conversation handle is derived from logical agent plus response channel/thread and never includes the active harness, provider, model, process, or agent-config ID.
| Inbound location/trigger | Conversation and response target |
| ------------------------------------------ | --------------------------------------------------------------- |
| Authorized untagged parent-channel message | Parent channel; response is sent in-channel |
| Authorized bot mention in parent channel | Thread already attached to that message, or a new public thread |
| Authorized message already in a thread | Existing thread; no repeated mention and no nested thread |
| `/approve` or `/stop <approval>` | Current parent/thread durable session; no new topic is created |
Authorization order is fixed: guild allowlist → parent-channel allowlist → user allowlist → configured binding/pairing → operation role → per-user/channel message and thread rate limits → thread creation/dispatch. A normal Discord channel's category parent is never treated as the thread authorization parent. If requested thread creation fails, dispatch does not occur because the adapter cannot honor the response target.
### Discord service ingress security
@@ -240,21 +244,21 @@ The Discord adapter is an authenticated gateway service, not an anonymous Socket
### Telegram
| ChannelMessage field | Telegram equivalent |
| -------------------- | ------------------------------------------------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Telegram `message_id` (integer) |
| `channelId` | Telegram `chat_id` (integer as string) |
| `senderId` | Telegram `from.id` (integer as string) |
| `senderType` | `"user"` for human senders; `"agent"` for bot-originated messages |
| `content` | `message.text` or `message.caption` |
| `contentType` | `"text"` for plain; `"markdown"` if `parse_mode = MarkdownV2`; `"image"` for `photo`; `"file"` for `document` |
| `threadId` | `message.message_thread_id` (for supergroup topics) |
| `replyToId` | Mosaic ID looked up from `message.reply_to_message.message_id` |
| `attachments` | `photo`, `document`, `video` fields mapped to `ChannelAttachment` |
| `timestamp` | `new Date(message.date * 1000)` |
| `metadata` | `{ channelMessageId, chatType, fromUsername, forwardFrom }` |
| ChannelMessageDto field | Telegram equivalent |
| ----------------------- | ------------------------------------------------------------------------------------------------------------- |
| `id` | Generated UUID; `metadata.channelMessageId` = Telegram `message_id` (integer) |
| `channelId` | Telegram `chat_id` (integer as string) |
| `senderId` | Telegram `from.id` (integer as string) |
| `senderKind` | `"user"` for human senders; `"agent"` for bot-originated messages |
| `content` | `message.text` or `message.caption` |
| `contentKind` | `"text"` for plain; `"markdown"` if `parse_mode = MarkdownV2`; `"image"` for `photo`; `"file"` for `document` |
| `threadId` | `message.message_thread_id` (for supergroup topics) |
| `replyToId` | Mosaic ID looked up from `message.reply_to_message.message_id` |
| `attachments` | `photo`, `document`, `video` fields mapped to `ChannelAttachmentDto` |
| `timestamp` | `new Date(message.date * 1000)` |
| `metadata` | `{ channelMessageId, chatType, fromUsername, forwardFrom }` |
**Outbound:** Adapter calls Telegram Bot API `sendMessage` with `parse_mode = MarkdownV2` for markdown content. For `contentType = "image"` or `"file"` it uses `sendPhoto` / `sendDocument`.
**Outbound:** Adapter calls Telegram Bot API `sendMessage` with `parse_mode = MarkdownV2` for markdown content. For `contentKind = "image"` or `"file"` it uses `sendPhoto` / `sendDocument`.
---
@@ -262,19 +266,19 @@ The Discord adapter is an authenticated gateway service, not an anonymous Socket
The TUI adapter bridges Mosaic's terminal interface (`packages/cli`) to the channel protocol so that TUI sessions can be treated as a first-class channel.
| ChannelMessage field | TUI equivalent |
| -------------------- | ------------------------------------------------------------------ |
| `id` | Generated UUID (TUI has no native message IDs) |
| `channelId` | `"tui:<conversationId>"` — the active conversation ID |
| `senderId` | Authenticated Mosaic `userId` |
| `senderType` | `"user"` for human input; `"agent"` for agent replies |
| `content` | Raw text from stdin / agent output |
| `contentType` | `"text"` for input; `"markdown"` for agent responses |
| `threadId` | Not used (TUI sessions are linear) |
| `replyToId` | Not used |
| `attachments` | File paths dragged/pasted into the TUI; resolved to `file://` URLs |
| `timestamp` | `new Date()` at the moment of send |
| `metadata` | `{ conversationId, sessionId, ttyWidth, colorSupport }` |
| ChannelMessageDto field | TUI equivalent |
| ----------------------- | ------------------------------------------------------------------ |
| `id` | Generated UUID (TUI has no native message IDs) |
| `channelId` | `"tui:<conversationId>"` — the active conversation ID |
| `senderId` | Authenticated Mosaic `userId` |
| `senderKind` | `"user"` for human input; `"agent"` for agent replies |
| `content` | Raw text from stdin / agent output |
| `contentKind` | `"text"` for input; `"markdown"` for agent responses |
| `threadId` | Not used (TUI sessions are linear) |
| `replyToId` | Not used |
| `attachments` | File paths dragged/pasted into the TUI; resolved to `file://` URLs |
| `timestamp` | `new Date()` at the moment of send |
| `metadata` | `{ conversationId, sessionId, ttyWidth, colorSupport }` |
**Outbound:** The adapter writes rendered content to stdout. Markdown is rendered via a terminal markdown renderer (e.g. `marked-terminal`). Code blocks are syntax-highlighted when `metadata.colorSupport = true`.
@@ -284,19 +288,19 @@ The TUI adapter bridges Mosaic's terminal interface (`packages/cli`) to the chan
The WebUI adapter connects the Next.js frontend (`apps/web`) to the channel protocol over the existing Socket.IO gateway (`apps/gateway`).
| ChannelMessage field | WebUI equivalent |
| -------------------- | ------------------------------------------------------------ |
| `id` | Generated UUID; echoed back in the WebSocket event |
| `channelId` | `"webui:<conversationId>"` |
| `senderId` | Authenticated Mosaic `userId` |
| `senderType` | `"user"` for browser input; `"agent"` for agent responses |
| `content` | Message text from the input field |
| `contentType` | `"text"` or `"markdown"` |
| `threadId` | Not used (conversation model handles threading) |
| `replyToId` | Message ID the user replied to (UI reply affordance) |
| `attachments` | Files uploaded via the file picker; stored to object storage |
| `timestamp` | `new Date()` at send, or server timestamp from event |
| `metadata` | `{ conversationId, sessionId, clientTimezone, userAgent }` |
| ChannelMessageDto field | WebUI equivalent |
| ----------------------- | ------------------------------------------------------------ |
| `id` | Generated UUID; echoed back in the WebSocket event |
| `channelId` | `"webui:<conversationId>"` |
| `senderId` | Authenticated Mosaic `userId` |
| `senderKind` | `"user"` for browser input; `"agent"` for agent responses |
| `content` | Message text from the input field |
| `contentKind` | `"text"` or `"markdown"` |
| `threadId` | Not used (conversation model handles threading) |
| `replyToId` | Message ID the user replied to (UI reply affordance) |
| `attachments` | Files uploaded via the file picker; stored to object storage |
| `timestamp` | `new Date()` at send, or server timestamp from event |
| `metadata` | `{ conversationId, sessionId, clientTimezone, userAgent }` |
**Outbound:** Adapter emits a `chat:message` Socket.IO event. The WebUI React component receives it and appends to the conversation list. Markdown content is rendered client-side via the existing markdown renderer component.
@@ -304,7 +308,7 @@ The WebUI adapter connects the Next.js frontend (`apps/web`) to the channel prot
## Identity Mapping
`mapIdentity(channelUserId)` resolves a channel-native user identifier to a Mosaic `userId`. This is required to attribute inbound messages to authenticated Mosaic accounts.
Gateway identity-linking policy resolves a channel-native user identifier to a Mosaic `userId` and produces `ChannelAuthorizedPrincipalDto`. Adapters provide native identity evidence but cannot self-authorize Mosaic scope. Discord currently uses configuration-owned paired users; database-backed linking remains the canonical direction for dynamic Matrix/Slack identity.
The implementation must query a `channel_identities` table (or equivalent) keyed on `(channel_name, channel_user_id)`. When no mapping exists the method returns `null` and the message is treated as anonymous (no Mosaic session context).
@@ -323,8 +327,8 @@ Identity linking flows (OAuth dance, deep-link verification token, etc.) are out
## Error Handling Conventions
- `connect()` must throw a structured error (subclass of `ChannelConnectError`) if the initial connection cannot be established within a reasonable timeout (default: 10 s).
- `sendMessage()` must throw `ChannelSendError` on terminal failures (auth revoked, channel not found). Transient failures (rate limit, network blip) should be retried internally with exponential backoff before throwing.
- `start()` must establish the native channel transport or throw a structured connection error. An adapter hosted inside the gateway must not wait for a loopback connection to that same not-yet-listening process; it starts the native transport, lets Socket.IO reconnect, and reports `degraded` until both links are ready.
- `ChannelEgressPort.send()` implementations must throw a typed terminal error for revoked auth, an invalid route, or a missing channel. Only transient rate/network/server failures are retried with bounded exponential backoff; Discord retries reuse a stable enforced nonce to prevent duplicate chunks, while permanent 4xx failures are not retried.
- `health()` must never throw — it returns `{ status: 'disconnected' }` on error.
- Adapters must emit structured logs with `{ channel: adapter.name, event, ... }` metadata for observability.
@@ -332,7 +336,7 @@ Identity linking flows (OAuth dance, deep-link verification token, etc.) are out
## Versioning
The `ChannelMessage` protocol follows semantic versioning. Non-breaking field additions (new optional fields) are minor version bumps. Breaking changes (type changes, required field additions) require a major version bump and a migration guide.
The `ChannelMessageDto` protocol follows semantic versioning. Non-breaking field additions (new optional fields) are minor version bumps. Breaking changes (type changes, required field additions) require a major version bump and a migration guide.
Current version: **1.0.0**
@@ -473,7 +477,7 @@ A single Mosaic conversation can be accessed simultaneously from multiple surfac
### Real-Time Sync Flow
1. A message arrives on any surface (TUI keystroke, browser send, Matrix event).
2. The surface's adapter normalizes the message to `ChannelMessage` and delivers it to `ConversationService`.
2. The surface's adapter normalizes the message to `ChannelMessageDto` and delivers it to `ConversationService`.
3. `ConversationService` persists the message to PostgreSQL, assigns a canonical `id`, and publishes a `message:new` event to the Valkey pub/sub channel keyed by `conversationId`.
4. All active surfaces subscribed to that `conversationId` receive the fanout event and push it to their respective clients:
- TUI adapter: writes rendered output to the connected terminal session.
@@ -561,7 +565,7 @@ Matrix sessions for linked users are persistent and long-lived. Unlike TUI sessi
- Their `channel_identities` row exists (link not revoked).
- They remain members of the relevant Matrix rooms.
Revoking a Matrix link (`DELETE /auth/channel-link/matrix/<matrixUserId>`) removes the `channel_identities` row and causes `mapIdentity()` to return `null`. The appservice optionally kicks the Matrix user from all Mosaic-managed rooms as part of the revocation flow (configurable, default: off).
Revoking a Matrix link (`DELETE /auth/channel-link/matrix/<matrixUserId>`) removes the `channel_identities` row and causes gateway principal resolution to deny the identity. The appservice optionally kicks the Matrix user from all Mosaic-managed rooms as part of the revocation flow (configurable, default: off).
---
@@ -733,7 +737,7 @@ room_retention_policies
created_at TIMESTAMP
```
The retention policy is enforced by a background job in the gateway that calls Conduit's admin API to purge events older than the configured threshold. Purged events are removed from the Conduit store but Mosaic's PostgreSQL message store retains the canonical `ChannelMessage` record unless the Mosaic retention policy also covers it.
The retention policy is enforced by a background job in the gateway that calls Conduit's admin API to purge events older than the configured threshold. Purged events are removed from the Conduit store but Mosaic's PostgreSQL message store retains the canonical `ChannelMessageDto` record unless the Mosaic retention policy also covers it.
Default retention values:

View File

@@ -0,0 +1,49 @@
# Mos Runtime Portability M1 — Logical Identity and Fencing
## Boundary
M1 separates the logical Mosaic agent from any Claude, Pi, Codex, tmux, Matrix, or provider-native session. The normalized identity is:
```text
(tenant_id, logical_agent_id, binding_id)
```
`logical_agent_id` is a server-owned stable identifier. A connector is a replaceable holder of a lease for one binding; it is not the agent identity.
## Durable lease model
PostgreSQL table `logical_agent_connector_leases` has one unique row per identity/binding tuple. The current row records:
- an opaque lease UUID;
- connector ID and normalized allowed scopes;
- a positive decimal fencing epoch stored as PostgreSQL `bigint`;
- acquired, heartbeat, expiry, release, and update timestamps.
Initial acquisition is insert-only. An existing active row causes `lease_held`. An expired or released row causes `takeover_required`; ordinary acquisition cannot recover it. Authorized takeover uses compare-and-swap against the expected epoch, rotates the lease UUID, and increments the epoch atomically. Heartbeat and release match the full identity, binding, connector, lease UUID, and epoch.
The companion `connector_lease_audit_log` is append-only metadata. It stores lifecycle event, outcome/reason, identity/binding/connector, epoch, correlation ID, and timestamp. It deliberately excludes scopes, grant objects, payloads, approval references, tokens, and credentials.
## Execution grants
`ConnectorLeaseCoordinator` issues a short-lived internal grant only after rereading the durable current lease. Defense-in-depth caps leases at 5 minutes and grants at 30 seconds by default; constructor options may tighten these limits. A grant is bound to tenant, logical agent, binding, connector, lease UUID, scope subset, expiry, and epoch.
Validation occurs immediately before adapter invocation and rereads PostgreSQL. The adapter receives only `ConnectorExecutionContext`; harness-native schemas remain behind the adapter. Validation denies:
- grants not minted by the current gateway process (including cloned/forged objects);
- expired grants or leases;
- released leases;
- stale epochs or replaced connector/lease UUIDs;
- missing/cross-tenant/cross-agent/cross-binding leases;
- scopes not authorized by both grant and current lease.
A gateway restart intentionally invalidates process-local grants. The durable lease and epoch survive, and a fresh grant may be issued only after current-lease and gateway-policy validation.
## Concurrency and side-effect rule
The database CAS determines the sole current holder. A successful takeover makes every old-epoch validation fail. Connector adapters must consume and propagate the normalized lease epoch/context so downstream effect boundaries can also fence races that occur after gateway validation.
M1 does not provide exactly-once receipts or a side-effect journal. Those remain later #754 work; callers must not infer exactly-once delivery from lease fencing.
## Extension boundary
`ConnectorLeaseService` is the gateway-owned policy surface. Every policy decision receives the normalized requested scopes and TTL (or explicit `null` where no TTL applies), so a concrete policy can enforce least privilege and duration limits. Its production default policy denies every lease/grant operation until a server-configured connector policy is supplied. No M1 HTTP endpoint accepts caller-controlled tenant or logical identity, and no concrete Claude/Pi/Codex adapter or channel cutover is included.

View File

@@ -0,0 +1,95 @@
# Fleet Roster v2 Structural Contract
**Status:** FCM-M1-001 local-tmux structural compiler contract. This document describes parsing,
strict structural validation, normalized in-memory representation, and deterministic rendering only.
It does not authorize role resolution, lifecycle reconciliation, mutation, migration, remote
placement, connector configuration, secret references, arbitrary commands, channels, gateway
mapping, or any live-fleet change.
The executable schema is [`roster-v2.schema.json`](./roster-v2.schema.json). The compiler exports
the same schema and its test parses this file and compares it structurally with the executable contract.
## Format and canonical shape
The compiler accepts YAML or JSON. It reads only snake_case source fields and renders canonical,
snake_case YAML. Rendering sorts runtime keys and agents by stable name. Agent names, class names,
and tool-policy names are structural identifiers; whether a class or policy resolves is a later
shared-resolver concern.
```yaml
version: 2
generation: 1
transport: tmux
tmux:
socket_name: mosaic-fleet
holder_session: _holder
defaults:
working_directory: ~/src
runtime: pi
runtimes:
pi:
reset_command: /new
agents:
- name: coder0
alias: Coder 0
class: code
runtime: pi
provider: openai
model: gpt-5.6-sol
reasoning: high
tool_policy: code
working_directory: ~/src
persistent_persona: false
reset_between_tasks: true
lifecycle:
enabled: true
desired_state: stopped
launch:
yolo: true
```
## Root fields
| Field | Required | Constraint | Meaning |
| ------------ | -------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------------- |
| `version` | yes | integer constant `2` | Identifies this contract. Version `1` is explicitly rejected by this compiler and remains on the existing v1 path until M4 migration. |
| `generation` | yes | positive safe integer | Desired-state generation. M2 uses it for mutation guards; M1 does not mutate it. |
| `transport` | yes | constant `tmux` | M1M5 support local tmux only. |
| `tmux` | yes | strict object | Explicit local socket and holder-session configuration. |
| `defaults` | yes | strict object | Default work directory and one supported local runtime. |
| `runtimes` | yes | non-empty object | Declared local runtime reset policy map. |
| `agents` | yes | non-empty array | Local fleet entries. Duplicate stable names are rejected. |
## Nested fields
| Path | Required | Constraint |
| ---------------------------------------------------- | -------- | ---------------------------------------------------------------------------------------------------- |
| `tmux.socket_name` | yes | non-empty `[A-Za-z0-9_.-]+`; an explicit named socket prevents default-versus-named socket ambiguity |
| `tmux.holder_session` | yes | non-empty `[A-Za-z0-9_.-]+` |
| `defaults.working_directory` | yes | non-empty string |
| `defaults.runtime` | yes | `claude`, `codex`, `opencode`, or `pi`; it must be declared in `runtimes` |
| `runtimes.<runtime>.reset_command` | yes | non-empty string; runtime key must be a supported local runtime |
| `agents[].name` | yes | unique `[A-Za-z0-9][A-Za-z0-9_.-]*` stable machine identity |
| `agents[].alias` | yes | non-empty display string |
| `agents[].class` | yes | `[a-z][a-z0-9-]*`; structural only in M1, semantic role resolution is FCM-M1-002 |
| `agents[].runtime` | yes | `claude`, `codex`, `opencode`, or `pi`; it must be declared in `runtimes` |
| `agents[].provider`, `model`, `working_directory` | yes | non-empty strings; provider/model capability resolution is a later card |
| `agents[].reasoning` | yes | `low`, `medium`, or `high` |
| `agents[].tool_policy` | yes | `[a-z][a-z0-9-]*`; structural only in M1 |
| `agents[].persistent_persona`, `reset_between_tasks` | yes | booleans |
| `agents[].lifecycle.enabled` | yes | boolean; stored now, reconciled in FCM-M3-001 |
| `agents[].lifecycle.desired_state` | yes | `running` or `stopped` |
| `agents[].launch.yolo` | yes | boolean; structured data only, not an arbitrary command escape hatch |
## Fail-closed boundary
Every object is `additionalProperties: false`. The compiler rejects unknown, missing, malformed,
and wrong-type fields before producing a model. It specifically rejects remote/SSH/host/socket
per-agent fields, connector blocks, secret references, channel fields, arbitrary command fields,
and gateway fields because they are unsupported in the local-tmux M1 contract. It does not silently
ignore v1 camelCase input, version `1`, or a source that does not parse to an object.
The v2 compiler is intentionally isolated from the existing v1 loader. Existing v1 rosters and
current examples/profiles continue on their current path; FCM-M4 owns explicit inventory, preview,
migration, and rollback. FCM-M2 owns generated-file/local-override quarantine, and FCM-M3 owns
runtime lifecycle and reconciliation.

View File

@@ -0,0 +1,156 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://mosaicstack.dev/schemas/fleet/roster-v2.schema.json",
"title": "Mosaic local tmux fleet roster v2",
"type": "object",
"additionalProperties": false,
"required": ["version", "generation", "transport", "tmux", "defaults", "runtimes", "agents"],
"properties": {
"version": {
"const": 2
},
"generation": {
"type": "integer",
"minimum": 1,
"maximum": 9007199254740991
},
"transport": {
"const": "tmux"
},
"tmux": {
"type": "object",
"additionalProperties": false,
"required": ["socket_name", "holder_session"],
"properties": {
"socket_name": {
"type": "string",
"pattern": "^[A-Za-z0-9_.-]+$"
},
"holder_session": {
"type": "string",
"pattern": "^[A-Za-z0-9_.-]+$"
}
}
},
"defaults": {
"type": "object",
"additionalProperties": false,
"required": ["working_directory", "runtime"],
"properties": {
"working_directory": {
"type": "string",
"minLength": 1
},
"runtime": {
"enum": ["claude", "codex", "opencode", "pi"]
}
}
},
"runtimes": {
"type": "object",
"minProperties": 1,
"propertyNames": {
"enum": ["claude", "codex", "opencode", "pi"]
},
"additionalProperties": {
"type": "object",
"additionalProperties": false,
"required": ["reset_command"],
"properties": {
"reset_command": {
"type": "string",
"minLength": 1
}
}
}
},
"agents": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"additionalProperties": false,
"required": [
"name",
"alias",
"class",
"runtime",
"provider",
"model",
"reasoning",
"tool_policy",
"working_directory",
"persistent_persona",
"reset_between_tasks",
"lifecycle",
"launch"
],
"properties": {
"name": {
"type": "string",
"pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]*$"
},
"alias": {
"type": "string",
"minLength": 1
},
"class": {
"type": "string",
"pattern": "^[a-z][a-z0-9-]*$"
},
"runtime": {
"enum": ["claude", "codex", "opencode", "pi"]
},
"provider": {
"type": "string",
"minLength": 1
},
"model": {
"type": "string",
"minLength": 1
},
"reasoning": {
"enum": ["low", "medium", "high"]
},
"tool_policy": {
"type": "string",
"pattern": "^[a-z][a-z0-9-]*$"
},
"working_directory": {
"type": "string",
"minLength": 1
},
"persistent_persona": {
"type": "boolean"
},
"reset_between_tasks": {
"type": "boolean"
},
"lifecycle": {
"type": "object",
"additionalProperties": false,
"required": ["enabled", "desired_state"],
"properties": {
"enabled": {
"type": "boolean"
},
"desired_state": {
"enum": ["running", "stopped"]
}
}
},
"launch": {
"type": "object",
"additionalProperties": false,
"required": ["yolo"],
"properties": {
"yolo": {
"type": "boolean"
}
}
}
}
}
}
}
}

View File

@@ -293,24 +293,64 @@ Each OIDC provider requires its client ID, client secret, and issuer URL togethe
### Plugins
| Variable | Description |
| ----------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
| `DISCORD_BOT_TOKEN` | Discord bot token (enables Discord plugin) |
| `DISCORD_SERVICE_TOKEN` | Required high-entropy service credential used to authenticate and sign Discord ingress; inject through the approved secret mechanism only |
| `DISCORD_SERVICE_USER_ID` | Required Mosaic service-principal user ID that owns persisted Discord conversations; the original Discord user ID remains audit metadata |
| `DISCORD_GUILD_ID` | Discord guild/server ID |
| `DISCORD_GATEWAY_URL` | Gateway URL for Discord plugin to call (default: `http://localhost:14242`) |
| `DISCORD_ALLOWED_GUILD_IDS` | Required comma-separated Discord guild snowflake allowlist; default-deny |
| `DISCORD_ALLOWED_CHANNEL_IDS` | Required comma-separated Discord channel snowflake allowlist; default-deny |
| `DISCORD_ALLOWED_USER_IDS` | Required comma-separated Discord user snowflake allowlist; default-deny |
| `TELEGRAM_BOT_TOKEN` | Telegram bot token (enables Telegram plugin) |
| `TELEGRAM_GATEWAY_URL` | Gateway URL for Telegram plugin to call |
| Variable | Description |
| --------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
| `DISCORD_BOT_TOKEN` | Discord bot token (enables Discord plugin) |
| `DISCORD_SERVICE_TOKEN` | Required high-entropy service credential used to authenticate and sign Discord ingress; inject through the approved secret mechanism only |
| `DISCORD_SERVICE_USER_ID` | Required Mosaic service-principal user ID that owns persisted Discord conversations; the original Discord user ID remains audit metadata |
| `DISCORD_GUILD_ID` | Discord guild/server ID |
| `DISCORD_GATEWAY_URL` | Gateway URL for Discord plugin to call (default: `http://localhost:14242`) |
| `DISCORD_ALLOWED_GUILD_IDS` | Required comma-separated Discord guild snowflake allowlist; default-deny |
| `DISCORD_ALLOWED_CHANNEL_IDS` | Required comma-separated Discord channel snowflake allowlist; default-deny |
| `DISCORD_ALLOWED_USER_IDS` | Required comma-separated Discord user snowflake allowlist; default-deny |
| `DISCORD_INTERACTION_BINDINGS` | Required JSON bindings from guild/channel to logical agent and paired Discord users with `viewer`, `operator`, or `admin` roles |
| `DISCORD_MESSAGE_RATE_LIMIT_PER_MINUTE` | Optional positive integer; authorized turns per guild/channel/user each minute (default: `30`) |
| `DISCORD_THREAD_RATE_LIMIT_PER_MINUTE` | Optional positive integer; mention-thread routes per guild/channel/user each minute (default: `5`) |
| `TELEGRAM_BOT_TOKEN` | Telegram bot token (enables Telegram plugin) |
| `TELEGRAM_GATEWAY_URL` | Gateway URL for Telegram plugin to call |
### Discord ingress security
When `DISCORD_BOT_TOKEN` is configured, `DISCORD_SERVICE_TOKEN`, `DISCORD_SERVICE_USER_ID`, and all three Discord allowlists are required. Gateway startup fails rather than enabling a broad or unauthenticated remote-control surface. The service user ID identifies a provisioned Mosaic service principal for persistence; the original Discord user ID is retained in ingress audit metadata. The service token is a secret supplied by the approved runtime secret mechanism and is never committed or logged.
Inbound Discord messages must originate from an allowed guild, channel, and user, mention the bot, and carry a signed envelope containing the native Discord message ID and a generated correlation ID. The gateway validates the service identity, envelope signature, and allowlists again before dispatching. Replayed Discord message IDs are rejected during the bounded ingress replay window. Durable inbox/idempotency retention is introduced with Tess durable state.
Inbound Discord messages must originate from an allowed guild and configured parent channel, come from an allowed and paired user whose role permits sending, and carry a signed envelope containing the native Discord message ID and a generated correlation ID. Attachment references are limited in count, metadata size, field length, and declared size; only query-free HTTPS URLs without credentials or fragments are accepted, so bearer or presigned URLs never reach persistence or an agent prompt. The gateway validates the service identity, envelope signature, allowlists, pairing, and role again before dispatching. Replayed Discord message IDs are rejected during the bounded ingress replay window. Durable inbox/idempotency retention is introduced with Tess durable state.
Configured channels are dedicated agent interaction surfaces. An authorized untagged message routes to the bound logical agent and the response returns in that channel. Mentioning the bot on a normal channel message creates a public Discord thread, or reuses the thread already attached to that same message; the response and later thread messages stay in that thread without repeated mentions. A normal channel's category is not an authorization parent—only a Discord thread inherits authorization from its configured parent channel. Runtime control commands such as `/approve` and `/stop <approval>` remain on the current channel/thread because they target that durable session rather than opening a new topic.
Authorization and per-user/channel rate limits are evaluated before thread creation, so an unlisted guild/channel/user, unpaired user, `viewer`, or rate-limited sender cannot create bot threads or dispatch gateway work. The bot needs Discord permissions to view/send in configured channels and create/send in public threads. If thread creation fails, the turn is not dispatched because the requested response destination cannot be honored.
Conversation handles use the configured logical agent plus Discord channel/thread identity. They do not contain a Claude, Codex, Pi, OpenCode, model, process, or runtime-provider identifier; changing the runtime behind the logical session therefore does not require reconnecting the Discord bot.
#### Interaction binding format
`DISCORD_INTERACTION_BINDINGS` must be a non-empty JSON array. Each item requires `instanceId`, trusted `agentConfigId`, `guildId`, `channelId`, and a non-empty `pairedUsers` object. `instanceId` is the configured logical-agent name; its trusted database `agentConfigId` must resolve to an agent configuration with exactly that name, preserving provider/model/prompt/tool selection per binding. The guild/channel must also appear in their corresponding allowlists. IDs below are placeholders:
```json
[
{
"instanceId": "interaction-agent",
"agentConfigId": "agent-config-id",
"guildId": "guild-id",
"channelId": "channel-id",
"pairedUsers": {
"discord-user-id": {
"role": "operator",
"mosaicUserId": "mosaic-user-id"
}
}
}
]
```
| Pairing role | Send message | Create/continue thread | Approve | Stop |
| ------------ | ------------ | ---------------------- | ------- | ---- |
| `viewer` | No | No | No | No |
| `operator` | Yes | Yes | No | No |
| `admin` | Yes | Yes | Yes | Yes |
A legacy role-only value such as `"discord-user-id": "operator"` remains valid for non-privileged ingress. Approval and stop require the object form with a provisioned `mosaicUserId`; gateway policy checks that Mosaic identity and consumes one exact-action approval once. Do not make the Discord service principal an approving administrator.
After changing bindings or allowlists, restart the gateway/plugin through the normal service manager and verify both Discord and gateway connectivity. The adapter reports `connected` only when both links are ready, `degraded` when one is ready, and `disconnected` when neither is ready. Test one authorized untagged channel turn, one mention-created thread, one thread follow-up, and one unauthorized user denial without using production credential values in logs or evidence.
### Session retention and garbage collection

View File

@@ -0,0 +1,43 @@
# Mos Connector Lease Operations — M1
## Operational status
M1 installs the durable schema and gateway policy/adapter boundary. It does **not** activate a connector, expose a lease administration endpoint, or cut over a channel. The default gateway connector-lease policy is deny-all until a later work package supplies an authorized server-side policy and concrete adapter.
## Events to monitor
Use correlation IDs to follow `connector_lease_audit_log` events:
| Event | Meaning |
| ---------- | --------------------------------------------------------------------- |
| `acquire` | First holder inserted for an unused binding |
| `renew` | Current holder heartbeat extended the TTL |
| `takeover` | Authorized CAS replaced the holder and incremented epoch |
| `release` | Current holder explicitly relinquished authority |
| `expiry` | An expired current lease was observed |
| `reject` | Policy, CAS, expiry, scope, or fencing validation denied an operation |
Audit data is metadata-only. Raw grant objects, connector payloads, scopes, tokens, approval references, and credentials must never be added to audit output.
## Incident checks
For suspected duplicate/stale connector effects:
1. Correlate the attempted operation with its `reject`, `takeover`, or `expiry` event.
2. Compare the current row's connector ID, lease UUID, epoch, expiry, and release time with the adapter's normalized execution context.
3. Treat an old epoch, old lease UUID, expired lease, or released lease as non-authoritative. Do not retry it as the old holder.
4. Recovery uses the authorized takeover path with the observed expected epoch. Ordinary acquire is intentionally rejected for expired/released rows.
5. If an external effect may already have happened, preserve evidence and do not assume lease fencing provides exactly-once replay safety.
## Migration and rollback safety
Migration `0016_salty_morlocks.sql` is additive: it creates two new tables and indexes without modifying existing authorization/session tables. Before rollout, normal database backup and migration verification still apply. Rolling application code back leaves unused additive tables in place; dropping tables is not part of automated rollback because it would destroy lease/audit evidence.
## Security constraints
- Tenant comes from authenticated gateway context, never a connector request field.
- Logical agent, binding, connector, and scope identifiers use normalized constrained forms.
- Takeover requires explicit gateway policy authorization and an expected epoch.
- Default defense-in-depth TTL caps are 5 minutes for leases and 30 seconds for grants; policy may enforce stricter limits.
- Validation and rejection audit complete before adapter side effects.
- Existing authz and exact-action approval controls remain additional required gates; a valid connector lease does not bypass them.

View File

@@ -0,0 +1,27 @@
# Independent Code Review — #756 Official Discord Channel Plugin
**Verdict: APPROVE**
## Scope reviewed
Complete current uncommitted #756 delta: multi-binding trusted agent selection, privileged ingress-route validation, attachment validation/persistence/resume, egress route lifecycle and idempotency, concurrent gateway stream state, Discord lifecycle/thread/rate behavior, compatibility ingress, and tests.
## Review result
No blocking or change-request finding remains.
- **Trusted multi-agent routing:** each binding requires a provisioned `agentConfigId`; the gateway resolves that config server-side, verifies its logical-agent name, and never accepts a Discord-controlled agent selection or applies generic routing to Discord ingress.
- **Auth and route integrity:** allowlist, pairing, role, and canonical logical-agent/channel-or-thread conversation-route validation occur before gateway processing. Privileged approval/stop paths use the same binding and route validation.
- **Attachments:** ingress rejects malformed, over-bounded, credential-bearing, fragment-bearing, or query-bearing URLs. Valid attachment metadata, including `sizeBytes`, persists and is reconstructed into resume history as explicitly untrusted context.
- **Discord reliability:** the adapter supports degraded Socket.IO reconnect, parent/thread routing semantics, bounded pre-side-effect ingress rates, and terminal response-route cleanup. Egress validates route/message alignment, sends deterministic nonces, distinguishes permanent from transient errors, and uses bounded retries.
- **Concurrent state:** per-client/conversation keys isolate listener, redaction, tool, and stream state for simultaneous threads sharing a Discord socket; disconnect cleanup covers all associated conversations.
- **Harness neutrality:** contracts retain logical-agent/channel data only; no harness/provider identity leaks into adapter routes or message boundaries.
## Verification performed
- `git diff --check` — passed.
- `pnpm --filter @mosaicstack/discord-plugin typecheck` — passed.
- `pnpm --filter @mosaicstack/discord-plugin lint` — passed.
- `pnpm --filter @mosaicstack/discord-plugin test` — passed: 44 tests; coverage 92.18% statements/lines, 86.55% branches, 100% functions (all ≥85% threshold).
- `pnpm --filter @mosaicstack/gateway typecheck` — passed.
- `cd apps/gateway && pnpm exec vitest run src/plugin/discord-ingress.security.spec.ts src/chat/chat.gateway-redaction.spec.ts src/__tests__/integration/tess-cross-surface.integration.test.ts` — passed: 32 tests.

View File

@@ -0,0 +1,36 @@
# Documentation Completion Checklist — #756 Official Discord plugin
## Required artifacts
- [x] `docs/PRD.md` includes the #756 workstream, assumptions, and acceptance criteria.
- [x] User workflow updated in `docs/tess/USER-GUIDE.md`.
- [x] Administrator configuration and authorization policy updated in `docs/guides/admin-guide.md`.
- [x] Developer/plugin authoring guidance updated in `docs/tess/PLUGIN-GUIDE.md`.
- [x] Channel architecture updated in `docs/architecture/channel-protocol.md`.
- [x] Package operations/development guide added at `plugins/discord/README.md`.
- [x] `docs/SITEMAP.md` links the official channel plugin documentation.
## API coverage
- [x] No HTTP or WebSocket endpoint was added, removed, or changed.
- [x] No OpenAPI update is needed.
- [x] Shared TypeScript contracts are documented in architecture and plugin-authoring guides.
- [x] Discord authentication, authorization, thread failure, and control-command behavior are documented.
## Structural standards
- [x] Working notes remain under `docs/scratchpads/`.
- [x] Review and checklist artifacts remain under `docs/reports/`.
- [x] No generated publishing output was added.
- [x] Existing repository documentation structure was preserved; no unrelated root cleanup was attempted.
## Review gate
- [x] Independent documentation/contract review passes (shared-contract review plus final code review of the current documentation delta).
- [x] Independent code review verifies documentation matches implementation (`docs/reports/code-review/756-code-review.md`: APPROVE).
- [x] Independent security review verifies documented controls (`docs/reports/security/756-security-review.md`: APPROVE).
## Publishing
- [x] Canonical source remains in-repository.
- [x] No external publishing action is in scope for this slice.

View File

@@ -0,0 +1,37 @@
# Security Review — Issue #756
**Scope:** final current uncommitted Discord plugin, shared channel contract, gateway ingress, AgentService, and plugin registration delta
**Snapshot:** `plugins/discord/src/index.ts` SHA-256 `5ee6b6aa4e2ff349f137f76b918d02c1254e12066cb48b136048e57bef36fdb2`
**Verdict:** **APPROVE**
## Final remediation verification
| Area | Current evidence | Result |
|---|---|---|
| Attachment confidentiality and integrity | Ingress accepts bounded attachment metadata only when URL is HTTPS, query-free, fragment-free, and credential-free. Count, aggregate size, field length, MIME, and finite non-negative size validation apply before dispatch; `sizeBytes` is signed and retained. | Pass |
| Trusted agent selection | Each binding names a required trusted `agentConfigId`; gateway resolves that config server-side and requires its configured name to equal the binding logical-agent ID. Client/provider input cannot select the agent for Discord ingress. | Pass |
| Privileged operation routing | Gateway revalidates the binding and requires the signed conversation identity to match the configured logical agent before approve/stop actions. Paired admin identity and one-time approval checks remain enforced. | Pass |
| Egress containment and delivery | Egress requires an aligned message/route, configured parent or exact observed thread target, and cleans response routes after completion, errors, typed-ingress failure, or bounded-map pressure. | Pass |
| Side-effect limits | Per guild/authorized-parent/user rolling message and thread limits execute before thread creation or dispatch. | Pass |
## Security controls reviewed
- Default-deny guild, parent-channel, user, configured pairing, and role checks precede rate consumption and all side effects.
- Gateway independently enforces Discord service authentication, HMAC integrity, allowlists, binding/role checks, attachment validation, and replay-ID rejection.
- Thread authorization uses only the Discord thread parent; category parents cannot authorize ingress.
- Agent-visible attachment references are explicitly labeled untrusted; binary content is not embedded. Persisted attachment name/URL values are redacted.
- Egress uses deterministic nonces, bounded transient retry, typed terminal errors, and does not send to forged targets.
- No secrets or message content were added to plugin logs.
## Verification evidence
| Command | Result |
|---|---|
| `pnpm --filter @mosaicstack/discord-plugin test` | PASS — 44 tests; v8 coverage: 92.18% statements/lines, 86.55% branches, 100% functions |
| `pnpm --filter @mosaicstack/discord-plugin typecheck` | PASS |
| `pnpm --filter @mosaicstack/types typecheck` | PASS |
| `pnpm --filter @mosaicstack/gateway typecheck` | PASS |
| `pnpm --filter @mosaicstack/gateway exec vitest run src/plugin/discord-ingress.security.spec.ts src/agent/__tests__/agent-service-ownership.test.ts` | PASS — 29 tests |
| `git diff --check` | PASS |
No unresolved critical, high, medium, or low security findings were identified in the reviewed final delta.

View File

@@ -0,0 +1,148 @@
# Issue #755 — Logical Mos identity and connector lease fencing
- Task: `MOS-PORT-M1-001`
- Branch: `feat/mos-logical-identity-fencing`
- Base: `origin/main`
- Started: 2026-07-14
- Working budget: 38K tokens (task ledger estimate); one implementation lane, bounded to M1.
## Objective
Implement the first runtime-portability security boundary: normalized logical-agent identity plus a PostgreSQL-durable exclusive connector lease and server-validated fencing grants.
## Scope
- Normalized identity contract independent of harness/provider-native session IDs.
- DB migration/schema/repository for one lease per tenant/logical-agent/binding.
- CAS acquire/takeover, monotonic epoch, TTL, heartbeat, release, expiry handling.
- Server-derived grants bound to tenant, logical agent, binding, connector, scopes, expiry, and lease epoch.
- Reject and credential-safely audit stale, expired, forged, unauthorized, cross-tenant, and cross-binding grants before adapter side effects.
- Runtime adapter boundary consumes normalized lease context.
- Unit, migration, close/reopen, concurrency, abuse, and gateway integration tests.
- Required developer/operations documentation for schema and security behavior.
## Explicit exclusions
No checkpoint/handoff payloads, exactly-once journal/receipts, concrete Claude/Pi/Codex harness adapter, channel cutover, or full cross-harness failover E2E.
## Reconciliation baseline
- Prospective remediation handoff: `web1:coder1`; PR [#757](https://git.mosaicstack.dev/mosaicstack/stack/pulls/757), issue [#755](https://git.mosaicstack.dev/mosaicstack/stack/issues/755).
- Before rebase: `dff8ce4f79ef90370c29d925002118a708010091`; required base: `origin/main` at `2e2280070ae67288be45f41743cf67052a8ca5a6`; original branch base: `d0771835542deab048ad8e79f271e3abdb6151f7`.
- Provider metadata: #757 is open, targets `main`, head is `feat/mos-logical-identity-fencing`, prior CI is green, and the provider reports it is not mergeable because of conflicts.
- Affected delivery paths: `apps/gateway/src/agent/agent.module.ts`; connector-lease gateway repository/service and three focused tests; `packages/agent/src/connector-lease.ts` plus test/export; `packages/types/src/agent/connector-lease.dto.ts` plus test/export; `packages/db/src/schema.ts`, migration `0016_salty_morlocks.sql`, Drizzle snapshot/journal; `docs/PRD.md`, `docs/SITEMAP.md`, MOS architecture/operations pages, this scratchpad, and `docs/tess/TASKS.md`.
- Read-only merge-tree inspection found only `docs/PRD.md` and `docs/SITEMAP.md` conflicts. Current-main #756 channel contracts, #758 roster-v2 structural compiler, and Native Kanban SOT use distinct contract domains; no substantive architecture collision was identified before mechanical reconciliation.
- Reconciliation constraints: retain all current-main #752/#756/#758/KBN content; add only nonduplicative #755 references; do not alter semantics or conflate connector leases/grants with Kanban task leases/fences, local Fleet leases, auth sessions, ResetSession generations, or federation grants.
## Plan (TDD RED → GREEN → REFACTOR)
1. Map existing contracts, DB/migration conventions, gateway authorization/audit boundaries, and test infrastructure.
2. Add failing contract/repository/concurrency/restart/abuse/gateway tests and capture RED evidence.
3. Implement the smallest normalized contracts, schema/migration/repository, grant validator, audit sink, and gateway service/adapter boundary needed to pass.
4. Refactor for clear invariants and credential-safe observability; rerun focused suites.
5. Run package/repo typecheck, lint, format, and appropriate tests.
6. Run independent code + security review, remediate, and re-review.
7. Inspect the final diff for security/scope drift; commit; queue guard; push; open PR with `Refs #755` and exact verification; stop without merge/issue closure.
## Constraints and safety notes
- `docs/tess/TASKS.md` is orchestrator-only and will not be edited.
- Existing dirty `.mosaic/orchestrator/mission.json` and `.mosaic/orchestrator/session.lock` are launcher/orchestrator state and will not be staged or altered intentionally.
- No client-supplied identity may confer authority.
- No credential, token, or raw grant material may be persisted to audit/log output.
- Existing authorization checks remain intact; fencing is an additional fail-closed layer.
## Assumptions resolved from existing architecture
- `ASSUMPTION:` M1 exposes no public lease endpoint. The gateway service is an internal policy surface with deny-all default policy because concrete connector activation/cutover is explicitly deferred.
- `ASSUMPTION:` Fencing epochs use PostgreSQL `bigint` and cross-module decimal strings, preserving JSON portability without JavaScript number precision loss.
- `ASSUMPTION:` Process-local grant provenance intentionally fails closed across restart; durable lease/epoch state survives and fresh grants require current policy + lease validation.
## TDD evidence
RED observed before implementation:
- `corepack pnpm --filter @mosaicstack/types exec vitest run src/agent/connector-lease.dto.spec.ts` → failed to load missing `connector-lease.dto.js`.
- `corepack pnpm --filter @mosaicstack/agent exec vitest run src/connector-lease.test.ts` → failed to load missing `connector-lease.js`.
- Gateway focused tests failed before implementation because the new repository/service boundaries did not exist (workspace dependencies were then built before behavioral GREEN runs).
GREEN to date:
- Types contract: 6/6 passed.
- Agent grant/fencing unit suite: 5/5 passed.
- Gateway PGlite repository + policy/side-effect integration: 7/7 passed; 1 real-PostgreSQL test skipped when `DATABASE_URL` absent.
- Real PostgreSQL focused run with configured `DATABASE_URL`: 1/1 passed (credential value not emitted in reports).
## Documentation checklist
- [x] `docs/PRD.md` contains current MOS-PORT M1 scope and acceptance criteria.
- [x] Developer architecture: `docs/architecture/mos-runtime-portability-m1.md`.
- [x] Admin/operations guidance: `docs/guides/mos-connector-lease-operations.md`.
- [x] `docs/SITEMAP.md` links both pages.
- [x] No user-guide change: M1 exposes no user-facing flow or channel cutover.
- [x] No OpenAPI/endpoint-index change: M1 adds no HTTP endpoint.
- [x] Migration/restart/rollback safety and credential-safe audit constraints documented.
- [x] Canonical source remains in-repo; no external publishing action is in scope.
- [x] Independent review confirms documentation matches implementation; implementation-specific findings were remediated.
## Independent review and remediation
Codex code/security review ran in multiple rounds. Findings and root-cause remediations:
1. Policy could not inspect requested scope/TTL → policy subject now receives normalized requested scopes and explicit requested TTL.
2. Unbounded authority lifetime → hard defaults cap leases at 5 minutes and grants at 30 seconds; overrides may only tighten; over-limit tests added.
3. Cross-tenant denial could audit under submitted tenant → mismatch audit uses authenticated tenant plus sanitized `untrusted` target metadata; integration assertion added.
4. Malformed forged grant could break the denial/audit path → runtime-safe shape validation with sanitized fallback audit; malformed-input test added.
5. Gateway integration test depended on prior test state → denial test now seeds a unique binding itself; isolated `-t` run passed.
6. Reviewer repeatedly identified launcher-generated `.mosaic/orchestrator/*` state; those files remain unstaged and excluded from the implementation commit.
Latest independent security review: no critical/high/medium/low findings. Final commit-level code review remains to run after the intended diff is committed without launcher state.
## Verification evidence
- Focused contracts/fencing: types 6/6; agent 9/9.
- Gateway focused PGlite repository/policy integration: 7/7; isolated denial test 1/1.
- Real PostgreSQL close/reopen/CAS test: 1/1 with configured `DATABASE_URL`.
- Root `corepack pnpm typecheck`: 42/42 Turbo tasks passed.
- Root `corepack pnpm lint`: 23/23 Turbo tasks passed.
- Root `corepack pnpm format:check`: all matched files passed.
- Root `corepack pnpm test`: 42/42 Turbo tasks passed; gateway 616 passed / 12 environment-gated skipped; DB 19 passed / 7 environment-gated skipped; Mosaic 650 passed.
## Known residual risks
- Concrete connector policies and Claude/Pi/Codex adapters are intentionally deferred; production policy defaults deny-all.
- Gateway pre-side-effect validation cannot make an external system exactly-once. Adapters must propagate/enforce the epoch at downstream effect boundaries; receipts/journaling are later #754 scope.
- Migration rollback is additive-only; dropping lease/audit tables is intentionally manual to avoid destroying authority/audit evidence.
## Commit-level review remediation
- Commit-level Codex code review found one `should-fix`: heartbeat, release, and grant issuance authorized caller-supplied lease fields before canonical normalization.
- TDD RED: the isolated gateway policy-boundary test showed mixed-case/padded logical agent, binding, connector, scope, and epoch values reaching policy unchanged.
- Remediation: exported the coordinator's canonical lease normalizer and applied it at the gateway boundary before tenant/policy checks and coordinator dispatch for heartbeat, release, and grant issuance.
- GREEN: isolated policy test 1/1; focused types 6/6, agent 9/9, gateway 8/8; root typecheck 42/42, lint 23/23, format check passed, and root tests 42/42 (gateway 617 passed / 12 environment-gated skipped).
- Commit-level security review remained clean: no critical/high/medium/low findings.
## Durable grant-expiry review remediation
- Final commit review found a second `should-fix`: grant expiry was capped against submitted lease metadata after current-authority validation, rather than the durable lease row.
- TDD RED: a crafted same-authority lease with a later submitted expiry produced a grant expiring after the durable row.
- Remediation: grant authority fields and expiry now derive from the durable current lease; submitted scopes remain an additional narrowing constraint.
- GREEN: focused agent fencing suite 10/10.
## Current-main reconciliation (2026-07-14)
- Rebased the existing PR branch from `dff8ce4f79ef90370c29d925002118a708010091` (old base `d0771835542deab048ad8e79f271e3abdb6151f7`) onto `origin/main` `2e2280070ae67288be45f41743cf67052a8ca5a6`; current uncommitted reconciliation head is `d190732a550918161b91d3eb54640f4ea0e2e499`.
- Resolved only `docs/PRD.md` and `docs/SITEMAP.md`: preserved current-main #752 Native Kanban, #756 official-channel, and #758 FCM material; retained the nonduplicative #755 M1 workstream and placed its two documentation links in the existing Runtime-neutral Mos section. No #755 source semantics changed.
- Compatibility review confirmed separate authority domains: connector lease/epoch/grant remains distinct from KBN task leases/fences, local Fleet roster lifecycle, auth sessions, ResetSession context, and federation grants. No channel cutover, adapter activation, checkpoint/exactly-once behavior, UI convergence, or #754 expansion was added.
- Focused verification after building the required workspace dependencies: types contract 6/6; agent fencing/grant 10/10; gateway repository/PGlite and policy integration 8/8. The focused real-PostgreSQL test was skipped because `DATABASE_URL` was not configured; no credentials were inspected or emitted.
- Generated schema check: `pnpm --filter @mosaicstack/db db:generate` reported no schema changes; migration `0016_salty_morlocks`, snapshot, and journal were unchanged by generation.
- Full verification: `pnpm typecheck` 42/42; `pnpm lint` 23/23; `pnpm format:check` passed; `pnpm test` 42/42 (gateway 627 passed / 12 environment-gated skipped). Scoped diff check and local documentation-link scan passed.
- Pending only: commit this reconciliation record, queue guard, force-with-lease push of the rebased existing branch, then fresh independent DB/code/security review and Ultron. Do not claim merge or issue closure.
## Durable lifecycle-authority remediation (2026-07-14)
- Independent review found that heartbeat and release authorized the submitted lease, allowing forged lifecycle scope data to influence policy before the durable row was consulted.
- Remediation: lifecycle operations tenant-check the submitted lease, load the durable current lease, compare tenant, logical agent, binding, lease UUID, connector, epoch, and canonical ordered scopes, audit and deny any absent/mismatched authority, then run policy and coordinator lifecycle calls with the durable lease. Coordinator CAS/fencing checks remain unchanged.
- Adversarial gateway integration coverage proves forged `tool.execute` scopes on a durable `runtime.send` lease deny before policy or mutation, preserve heartbeat/release state, and write a denial audit for both lifecycle actions; canonical heartbeat and release remain accepted.
- Verification: focused types 6/6; agent 10/10; gateway PGlite integration/repository 9/9 with one `DATABASE_URL`-gated PostgreSQL test skipped; Drizzle check passed; root typecheck 42/42; lint 23/23; format check passed; root test 42/42 (gateway 628 passed / 12 environment-gated skipped).
- Pending: commit, queue-guard, force-with-lease push, then independent DB/code/security rereview and CI. Do not merge or close #755.

View File

@@ -0,0 +1,57 @@
# Scratchpad — #756 Official Discord channel plugin
- **Task / issue:** Official Discord channel plugin / #756
- **Branch:** `feat/756-official-discord-plugin`
- **Worktree:** `/home/hermes/agent-work/stack-discord-plugin`
- **Base:** `origin/main` at `49e8a54105eddf41e8e0e44603ded616ee76044f`
- **Objective:** Deliver harness-neutral Discord routing, native mention-to-thread behavior, untagged in-channel interaction, fail-closed channel/user authorization, and transport-neutral contracts for future official channel plugins.
- **Collision boundary:** Do not modify orchestrator-to-Pi migration, logical-agent lease/fencing (#754/#755), runtime provider implementations, or orchestrator-owned `docs/TASKS.md`.
- **Working budget:** 55K tokens for requirements, implementation, tests, documentation, independent review, and delivery. No user-specified hard cap. Reduce optional refactoring before reducing acceptance coverage.
## Assumptions
1. **ASSUMPTION:** Every configured Discord channel is intentionally agent-bound, so an authorized human's untagged message is agent input and receives an in-channel response. Rationale: this satisfies the requested no-tag behavior without activating the bot in arbitrary channels.
2. **ASSUMPTION:** Mentioning the bot in a parent channel creates or reuses a public Discord thread; messages already in a thread remain in that thread without repeated mentions. Rationale: Discord does not support nested threads and the request describes tags as the thread-selection signal.
3. **ASSUMPTION:** One bot process may host multiple configured channel-to-logical-agent bindings. Rationale: bindings are already configuration-owned and this avoids per-agent Discord credentials.
4. **ASSUMPTION:** Static guild/channel/user allowlists and paired-user roles remain the administration surface for this slice. Rationale: dynamic admin UI is larger and can be added without changing adapter contracts.
5. **ASSUMPTION:** The stable conversation handle contains logical agent plus Discord channel/thread identity and never a harness/provider identifier. Rationale: runtime re-enrollment and the active lease work can change Claude/Codex/Pi/OpenCode behind the same channel connection.
6. **ASSUMPTION:** Canonical documentation remains in-repository for this slice; no external publishing is performed.
## Plan
1. Update `docs/PRD.md` before code with #756 scope, constraints, assumptions, and acceptance criteria.
2. Add failing Discord behavior and authorization tests first.
3. Add transport-neutral channel DTO/contracts in `@mosaicstack/types`.
4. Implement mention-to-thread, untagged in-channel, existing-thread, stable conversation, and adapter health behavior without runtime-specific imports.
5. Update admin/developer/channel protocol docs and documentation checklist.
6. Run focused tests, typecheck, lint, formatting, full applicable baseline, and coverage.
7. Run independent code and security review; remediate and re-review.
8. Commit, queue-guard, push, open PR to `main`, wait for green CI, squash merge, verify merged CI, and close #756.
## Progress
- 2026-07-14: Loaded mission state (none active), global/project guidance, current `origin/main`, existing Discord/Tess/fleet connector architecture, issue #709 history, and active portability issues #754/#755.
- 2026-07-14: Created issue #756 through the Mosaic wrapper.
- 2026-07-14: Created isolated worktree/branch from current `origin/main`; the stale root checkout and unrelated QA artifacts remain untouched.
## TDD decision
Required and applied. This change modifies authorization-sensitive remote ingress and routing behavior. Failing permission and routing tests will be captured before implementation.
## Risks / blockers
- Active logical-agent lease work may later add stronger fencing fields. This slice must expose a clean, harness-neutral seam without duplicating that schema.
- Discord thread creation is an external side effect. Unit tests use a typed fake; live credential smoke testing is out of scope and must not use committed secrets.
- Repository-wide checks may expose unrelated baseline debt; changed-scope evidence and CI remain mandatory.
## Verification evidence
- `pnpm format:check` — passed.
- `git diff --check` — passed.
- `pnpm typecheck` — passed (42 Turbo tasks).
- `pnpm lint` — passed (23 Turbo tasks).
- `pnpm build` — passed (23 Turbo tasks).
- `pnpm --filter @mosaicstack/discord-plugin test` — passed: 44 tests; V8 coverage 92.18% statements/lines, 86.55% branches, 100% functions (all configured thresholds ≥85%).
- Focused gateway verification passed: Discord ingress/security, cross-surface, ownership, redaction/concurrency, and agent attachment tests.
- `pnpm test` — changed-scope suites passed; repository baseline remains blocked by `apps/gateway/src/__tests__/cross-user-isolation.test.ts` requiring PostgreSQL at localhost port 5433 (`ECONNREFUSED`). The failure is unrelated to this change.
- Independent code/security re-reviews requested after final remediation; reports are stored under ignored `docs/reports/` evidence paths.

View File

@@ -1,3 +1,28 @@
# Tess Plugin Authoring
Plugins are replaceable adapters. Declare capabilities, derive scope from trusted context, preserve correlation IDs, redact before persistence/egress, and return unsupported operations as fail-closed results. Names and identities are configuration data, not literals in keys or defaults.
## Official channel adapter contract
Official Discord, Matrix, Slack, and future channel adapters share contracts exported from `@mosaicstack/types` under `channel/`:
- `OfficialChannelAdapter` provides `name`, `start()`, `stop()`, and non-throwing connection `health()`.
- `ChannelMessageDto` and `ChannelAttachmentDto` normalize transport data with JSON-safe metadata.
- `ChannelBindingDto` and `ChannelAuthorizedPrincipalDto` normalize configuration-owned logical-agent binding and the already-allowlisted/paired external actor.
- `ChannelIngressDto` carries operation, correlation, native message ID, authorized principal, normalized message, and stable route into `ChannelIngressPort`.
- `ChannelConversationRouteDto` binds a configured channel to `logicalAgentId`, stable `conversationId`, authorization parent, and response target.
- `ChannelEgressDto` and `ChannelEgressPort` separate where a response is delivered from the gateway's runtime/provider selection.
`ChannelConversationRouteDto` deliberately has no harness, provider, model, process, or native runtime-session field. The gateway owns runtime selection, durable enrollment, authorization, audit, and lease/fencing. A channel adapter must not call Claude, Codex, Pi, OpenCode, tmux, or Matrix runtime providers directly. Discord currently preserves its signed Socket.IO compatibility ingress for established gateway authentication/replay/approval controls while normalizing the same ingress DTO; supplied direct ports are the future registration path.
## Adapter requirements
1. Resolve configuration-owned channel and logical-agent bindings before dispatch. A binding may carry a trusted gateway agent-config reference, but the stable route contains only the logical agent; gateway verifies the reference resolves to that agent before runtime selection.
2. Apply channel-native allowlists and paired-user roles before any external side effect such as thread creation.
3. Preserve native message ID, correlation ID, channel/thread address, attachments, and response target.
4. Treat normal channel parents (for example Discord categories) separately from thread parents.
5. Keep reconnect and conversation identity independent of the active runtime provider.
6. Report sanitized connection/routing failures without message bodies or credentials.
7. Pass the shared route/authorization contract suite plus adapter-specific translation tests.
Discord establishes the first policy: authorized untagged messages respond in the configured channel; a mention creates a thread or reuses the thread already attached to that message; existing thread messages stay there. Runtime control commands remain on the current durable session. Matrix and Slack should translate native rooms/threads into the same route and response-target semantics rather than adding transport branches to gateway core.

View File

@@ -43,3 +43,4 @@
| TESS-M5-002 | done | Complete migration inventory, cutover, rollback, retention and deprecation evidence | #711 | coder3 | docs/tess | feat/tess-migration-docs | TESS-M4-V | 18K | TESS-MIG-001. **Mos-DISPATCHED to coder3 2026-07-13** ("M4 complete; advancing to M5") — dispatched AHEAD of TESS-M4-V passing; the M4-V-status-vs-#710-CLOSED dependency reconciliation is pending Mos ruling (tracked, not orchestrator-decided). **DELIVERED as PR #742** — 4 new files docs/tess/M5-MIGRATION-{INVENTORY,CUTOVER,ROLLBACK,RETENTION-DEPRECATION}.md, base main b7b0f508, frozen head b5e9d0e528a50aae2e916cc7202bacc2e8db67dd. Docs-only; tracking-control trio (MISSION-MANIFEST/TASKS/VERIFICATION-MATRIX) UNTOUCHED; command-authz byte-identical a9f829e7; no live creds. **CI pipeline 1773 SUCCESS** (repo 47, refs/pull/742/head, commit==head). **Independent non-author ROR COMPLETE: reviewer VERIFIED APPROVE at exact head b5e9d0e528a50aae2e916cc7202bacc2e8db67dd (Gitea comment 17108)** — evidence claims verified to trace to landed Hermes adapter / capability matrix, gateway registry/reachability, operator-memory scope path, Mos coordination boundary; docs do NOT over-claim transcript/profile import, schema migration, unsupported-capability enablement, production cutover, or deprecation completion. Head independently verified UNMOVED at b5e9d0e528a5 post-ROR (live Gitea), base main, mergeable=true. **#742 MERGED by Mos → main 5789711e. Deliverable docs LANDED. GATE: TESS-M4-V/M5-V verification-gate reconciliation remains Mos-owned/pending (this row was dispatched ahead of M4-V passing).** |
| TESS-M5-003 | done | Complete OpenAPI, user/admin/developer/plugin/operations docs and checklist | #711 | codex | docs | feat/tess-docs | TESS-M5-001,TESS-M5-002 | 22K | Documentation hard gate. **DELIVERED as PR #746** (branch feat/tess-docs, base main, 7 docs-only files: docs/openapi-tess.yaml + docs/tess/{ADMIN,DEVELOPER,OPERATIONS,PLUGIN,USER}-GUIDE.md + M5-003-DOCUMENTATION-CHECKLIST.md). 4-round revise-loop (heads 470eb911→c9f69300→7aea94e2→25b9d642) converged: OpenAPI covers interaction routes + SSE /sessions/{sessionId}/stream + Mos coord (/api/coord/mos/handoff,/observe,/result) + memory preferences/insights/search; request-body schemas aligned to real DTOs (Send requires content+idempotencyKey, Stop requires approvalRef, Insight requires only content, MosHandoff body requires idempotencyKey+summary); checklist accurate. **CI pipeline 1786 SUCCESS** (repo 47, refs/pull/746/head, commit==head 25b9d642). **Independent non-author ROR COMPLETE: reviewer VERIFIED APPROVE at exact head 25b9d642b939014b3efd61826e7524cafc6ffc2e (Gitea comment 17170)** — docs-only, tracking-trio untouched, command-authz byte-identical a9f829e7, no false coverage claims. Head verified UNMOVED at 25b9d642 (live Gitea, not worker-reported), base main, mergeable=true. **#746 MERGED by Mos → main bc8016c8314ec3a4b6ebc2fec5d9f276fca3327a. Documentation gate LANDED.** GATE: TESS-M4-V/M5-V verification-gate reconciliation remains Mos-owned/pending. |
| TESS-M5-V | not-started | Full baseline, contract, integration, Discord/CLI E2E, security review, recovery drill and rollback qualification | #711 | sonnet | apps/gateway, packages/agent, plugins/discord, packages/mosaic | review/tess-final | TESS-M5-003 | 35K | Maps AC-TESS-01..11 to evidence |
| MOS-PORT-M1-001 | in-progress | Implement logical Mos identity, PostgreSQL connector lease, monotonic fencing, server-bound execution grants, audit, migrations, concurrency/restart/abuse/integration tests | #755 | codex | packages/types, packages/agent, packages/db, apps/gateway | feat/mos-logical-identity-fencing | — | 38K | Requirements MOS-PORT-ID-001, MOS-PORT-LEASE-001, MOS-PORT-FENCE-001..002, MOS-PORT-OBS-001, MOS-PORT-ARCH-001. One Sol/Pi worker; TDD; PR-open STOP; worker must not edit this ledger. |

View File

@@ -1,5 +1,13 @@
# Tess User Guide
## Discord conversations
In a configured Tess/interaction channel, an authorized untagged message is sent to the bound logical agent and its response appears in the channel. Mention the bot when starting a separate topic: Mosaic reuses a thread already attached to that same Discord message, or creates a new thread for the message, and responds there. Continue in that thread without tagging the bot again. Messages from unconfigured channels or users without an authorized pairing are ignored without creating a thread.
`/approve` and `/stop <approval>` operate on the current channel/thread session and do not open a new thread. The Discord connection is bound to the logical agent conversation, not Claude, Codex, Pi, OpenCode, or another harness; a runtime handoff behind Mosaic does not change where you continue the conversation.
## CLI and HTTP interaction
All HTTP interaction calls require authenticated session credentials and `X-Correlation-Id`. Use `GET /api/interaction/{agentName}/sessions?provider=...` to list only visible runtime sessions, then enroll with `POST .../sessions/{sessionId}/enroll` body `{providerId,runtimeSessionId}`. Attach uses `{mode:"read"}`; send uses `{content,idempotencyKey}`. Stop requires `{approvalRef}` and fails with 403 without the exact durable approval. Recovery only requeues interrupted durable work.
Memory is user-scoped: preferences support list/get/upsert/delete; insights support list/get/create/delete; search body is `{query,limit?,maxDistance?}`. Mos work is handed off with `POST /api/coord/mos/handoff`; observe and result use the returned handoff ID.

View File

@@ -28,6 +28,7 @@ export default tseslint.config(
'apps/web/e2e/helpers/*.ts',
'apps/web/playwright.config.ts',
'apps/gateway/vitest.config.ts',
'plugins/discord/vitest.config.ts',
'packages/db/vitest.config.ts',
'packages/storage/vitest.config.ts',
'packages/mosaic/vitest.config.ts',

View File

@@ -0,0 +1,261 @@
import { describe, expect, it, vi } from 'vitest';
import type {
ConnectorExecutionContext,
ConnectorExecutionGrant,
ConnectorLease,
ConnectorLeaseAuditEvent,
ConnectorLeaseStore,
FencedConnectorAdapter,
LogicalAgentBinding,
} from '@mosaicstack/types';
import {
ConnectorLeaseCoordinator,
MAX_CONNECTOR_GRANT_TTL_MS,
MAX_CONNECTOR_LEASE_TTL_MS,
} from './connector-lease.js';
import type { ConnectorLeaseError } from './connector-lease.js';
const identity = { tenantId: 'tenant-a', logicalAgentId: 'mos' } as const;
const binding: LogicalAgentBinding = { identity, bindingId: 'operator-chat' };
const activeLease: ConnectorLease = {
...binding,
leaseId: '00000000-0000-4000-8000-000000000001',
connectorId: 'connector-a',
scopes: ['runtime.send', 'tool.execute'],
leaseEpoch: '3',
acquiredAt: '2026-07-14T17:00:00.000Z',
heartbeatAt: '2026-07-14T17:00:00.000Z',
expiresAt: '2026-07-14T17:10:00.000Z',
};
class FakeLeaseStore implements ConnectorLeaseStore {
lease: ConnectorLease | null = activeLease;
readonly audits: ConnectorLeaseAuditEvent[] = [];
async acquire(): Promise<ConnectorLease> {
if (!this.lease) throw new Error('fixture has no lease');
return this.lease;
}
async takeover(): Promise<ConnectorLease> {
if (!this.lease) throw new Error('fixture has no lease');
return this.lease;
}
async heartbeat(): Promise<ConnectorLease> {
if (!this.lease) throw new Error('fixture has no lease');
return this.lease;
}
async release(): Promise<void> {}
async findCurrent(): Promise<ConnectorLease | null> {
return this.lease;
}
async recordAudit(event: ConnectorLeaseAuditEvent): Promise<void> {
this.audits.push(event);
}
}
describe('ConnectorLeaseCoordinator fencing', (): void => {
it('validates a server-minted grant immediately before invoking an adapter side effect', async (): Promise<void> => {
const store = new FakeLeaseStore();
const coordinator = new ConnectorLeaseCoordinator(store, {
now: (): Date => new Date('2026-07-14T17:01:00.000Z'),
});
const grant = await coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: 30_000,
correlationId: 'correlation-1',
});
const execute = vi.fn(async (_input: string, context: ConnectorExecutionContext) => context);
const adapter: FencedConnectorAdapter<string, ConnectorExecutionContext> = { execute };
const context = await coordinator.executeGrant(grant, 'runtime.send', 'hello', adapter);
expect(execute).toHaveBeenCalledOnce();
expect(context).toMatchObject({
identity,
bindingId: 'operator-chat',
connectorId: 'connector-a',
leaseEpoch: '3',
scopes: ['runtime.send'],
});
});
it('caps grant expiry to the durable current lease instead of submitted metadata', async (): Promise<void> => {
const store = new FakeLeaseStore();
store.lease = { ...activeLease, expiresAt: '2026-07-14T17:01:05.000Z' };
const coordinator = new ConnectorLeaseCoordinator(store, {
now: (): Date => new Date('2026-07-14T17:01:00.000Z'),
});
const grant = await coordinator.issueGrant({
lease: { ...activeLease, expiresAt: '2026-07-14T18:00:00.000Z' },
scopes: ['runtime.send'],
ttlMs: 30_000,
correlationId: 'correlation-durable-expiry',
});
expect(grant.expiresAt).toBe('2026-07-14T17:01:05.000Z');
});
it.each([
['forged clone', (grant: ConnectorExecutionGrant): ConnectorExecutionGrant => ({ ...grant })],
[
'cross-tenant clone',
(grant: ConnectorExecutionGrant): ConnectorExecutionGrant => ({
...grant,
identity: { ...grant.identity, tenantId: 'tenant-b' },
}),
],
[
'cross-agent clone',
(grant: ConnectorExecutionGrant): ConnectorExecutionGrant => ({
...grant,
identity: { ...grant.identity, logicalAgentId: 'other-agent' },
}),
],
[
'cross-binding clone',
(grant: ConnectorExecutionGrant): ConnectorExecutionGrant => ({
...grant,
bindingId: 'other-binding',
}),
],
[
'cross-connector clone',
(grant: ConnectorExecutionGrant): ConnectorExecutionGrant => ({
...grant,
connectorId: 'connector-b',
}),
],
])('denies and audits a %s before adapter invocation', async (_label, forge): Promise<void> => {
const store = new FakeLeaseStore();
const coordinator = new ConnectorLeaseCoordinator(store, {
now: (): Date => new Date('2026-07-14T17:01:00.000Z'),
});
const grant = await coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: 30_000,
correlationId: 'correlation-forged',
});
const adapter = { execute: vi.fn().mockResolvedValue(undefined) };
await expect(
coordinator.executeGrant(forge(grant), 'runtime.send', undefined, adapter),
).rejects.toMatchObject({ code: 'forged_grant' } satisfies Partial<ConnectorLeaseError>);
expect(adapter.execute).not.toHaveBeenCalled();
expect(store.audits.at(-1)).toMatchObject({
event: 'reject',
outcome: 'denied',
reason: 'forged_grant',
correlationId: 'correlation-forged',
});
});
it('denies malformed forged grants with sanitized audit metadata', async (): Promise<void> => {
const store = new FakeLeaseStore();
const coordinator = new ConnectorLeaseCoordinator(store, {
now: (): Date => new Date('2026-07-14T17:01:00.000Z'),
});
const adapter = { execute: vi.fn().mockResolvedValue(undefined) };
await expect(
coordinator.executeGrant(
// @ts-expect-error Deliberately exercise malformed runtime input at the trust boundary.
{},
'runtime.send',
undefined,
adapter,
),
).rejects.toMatchObject({ code: 'forged_grant' } satisfies Partial<ConnectorLeaseError>);
expect(adapter.execute).not.toHaveBeenCalled();
expect(store.audits).toContainEqual(
expect.objectContaining({
identity: { tenantId: 'untrusted', logicalAgentId: 'untrusted' },
bindingId: 'untrusted',
connectorId: 'untrusted',
correlationId: 'untrusted',
reason: 'forged_grant',
}),
);
});
it('rejects lease and grant TTLs above server-side safety caps', async (): Promise<void> => {
const store = new FakeLeaseStore();
const coordinator = new ConnectorLeaseCoordinator(store, {
now: (): Date => new Date('2026-07-14T17:01:00.000Z'),
});
expect(
() =>
new ConnectorLeaseCoordinator(store, {
maxLeaseTtlMs: MAX_CONNECTOR_LEASE_TTL_MS + 1,
}),
).toThrow(/no greater than/);
await expect(
coordinator.acquire({
identity,
bindingId: 'operator-chat',
connectorId: 'connector-a',
scopes: ['runtime.send'],
ttlMs: MAX_CONNECTOR_LEASE_TTL_MS + 1,
correlationId: 'correlation-ttl',
}),
).rejects.toThrow(/no greater than/);
await expect(
coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: MAX_CONNECTOR_GRANT_TTL_MS + 1,
correlationId: 'correlation-ttl',
}),
).rejects.toThrow(/no greater than/);
});
it('denies stale epoch, expired grant, and unauthorized scope before side effects', async (): Promise<void> => {
let now = new Date('2026-07-14T17:01:00.000Z');
const store = new FakeLeaseStore();
const coordinator = new ConnectorLeaseCoordinator(store, { now: (): Date => now });
const stale = await coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: 30_000,
correlationId: 'correlation-stale',
});
store.lease = { ...activeLease, leaseEpoch: '4', connectorId: 'connector-b' };
const adapter = { execute: vi.fn().mockResolvedValue(undefined) };
await expect(
coordinator.executeGrant(stale, 'runtime.send', undefined, adapter),
).rejects.toMatchObject({ code: 'stale_epoch' } satisfies Partial<ConnectorLeaseError>);
store.lease = activeLease;
const expiring = await coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: 1_000,
correlationId: 'correlation-expired',
});
now = new Date('2026-07-14T17:01:02.000Z');
await expect(
coordinator.executeGrant(expiring, 'runtime.send', undefined, adapter),
).rejects.toMatchObject({ code: 'grant_expired' } satisfies Partial<ConnectorLeaseError>);
now = new Date('2026-07-14T17:01:00.000Z');
const scoped = await coordinator.issueGrant({
lease: activeLease,
scopes: ['runtime.send'],
ttlMs: 30_000,
correlationId: 'correlation-scope',
});
await expect(
coordinator.executeGrant(scoped, 'tool.execute', undefined, adapter),
).rejects.toMatchObject({ code: 'scope_denied' } satisfies Partial<ConnectorLeaseError>);
expect(adapter.execute).not.toHaveBeenCalled();
});
});

View File

@@ -0,0 +1,381 @@
import { randomUUID } from 'node:crypto';
import {
normalizeConnectorId,
normalizeConnectorScope,
normalizeConnectorScopes,
normalizeCorrelationId,
normalizeLeaseEpoch,
normalizeLogicalAgentIdentity,
normalizeLogicalBindingId,
type AcquireConnectorLeaseInput,
type ConnectorExecutionContext,
type ConnectorExecutionGrant,
type ConnectorLease,
type ConnectorLeaseAuditEvent,
type ConnectorLeaseRejectReason,
type ConnectorLeaseStore,
type FencedConnectorAdapter,
type HeartbeatConnectorLeaseInput,
type IssueConnectorExecutionGrantInput,
type LogicalAgentBinding,
type ReleaseConnectorLeaseInput,
type TakeoverConnectorLeaseInput,
} from '@mosaicstack/types';
export const MAX_CONNECTOR_LEASE_TTL_MS = 5 * 60 * 1000;
export const MAX_CONNECTOR_GRANT_TTL_MS = 30 * 1000;
export interface ConnectorLeaseCoordinatorOptions {
readonly now?: () => Date;
readonly maxLeaseTtlMs?: number;
readonly maxGrantTtlMs?: number;
}
export class ConnectorLeaseError extends Error {
constructor(
readonly code: ConnectorLeaseRejectReason,
message: string,
) {
super(message);
this.name = ConnectorLeaseError.name;
}
}
/**
* Runtime-neutral lease coordinator. Durable CAS lives in the store adapter;
* grant provenance remains process-local so a restart fails closed and mints
* fresh grants from the durable current lease.
*/
export class ConnectorLeaseCoordinator {
private readonly issuedGrants = new WeakSet<object>();
private readonly now: () => Date;
private readonly maxLeaseTtlMs: number;
private readonly maxGrantTtlMs: number;
constructor(
private readonly store: ConnectorLeaseStore,
options: ConnectorLeaseCoordinatorOptions = {},
) {
this.now = options.now ?? (() => new Date());
this.maxLeaseTtlMs = normalizeTtlLimit(
options.maxLeaseTtlMs ?? MAX_CONNECTOR_LEASE_TTL_MS,
MAX_CONNECTOR_LEASE_TTL_MS,
'lease',
);
this.maxGrantTtlMs = normalizeTtlLimit(
options.maxGrantTtlMs ?? MAX_CONNECTOR_GRANT_TTL_MS,
MAX_CONNECTOR_GRANT_TTL_MS,
'grant',
);
}
async acquire(input: AcquireConnectorLeaseInput): Promise<ConnectorLease> {
const command = normalizeAcquireInput(input, this.maxLeaseTtlMs);
const now = this.now();
return this.store.acquire({
...command,
leaseId: randomUUID(),
now: now.toISOString(),
expiresAt: expiresAt(now, command.ttlMs),
});
}
async takeover(input: TakeoverConnectorLeaseInput): Promise<ConnectorLease> {
const command = normalizeAcquireInput(input, this.maxLeaseTtlMs);
const now = this.now();
return this.store.takeover({
...command,
expectedEpoch: normalizeLeaseEpoch(input.expectedEpoch),
leaseId: randomUUID(),
now: now.toISOString(),
expiresAt: expiresAt(now, command.ttlMs),
});
}
async heartbeat(input: HeartbeatConnectorLeaseInput): Promise<ConnectorLease> {
const now = this.now();
const lease = normalizeConnectorLease(input.lease);
const ttlMs = normalizeTtl(input.ttlMs, this.maxLeaseTtlMs, 'lease');
return this.store.heartbeat({
lease,
ttlMs,
correlationId: normalizeCorrelationId(input.correlationId),
now: now.toISOString(),
expiresAt: expiresAt(now, ttlMs),
});
}
async release(input: ReleaseConnectorLeaseInput): Promise<void> {
await this.store.release({
lease: normalizeConnectorLease(input.lease),
correlationId: normalizeCorrelationId(input.correlationId),
now: this.now().toISOString(),
});
}
async current(binding: LogicalAgentBinding): Promise<ConnectorLease | null> {
return this.store.findCurrent(normalizeBinding(binding));
}
async issueGrant(input: IssueConnectorExecutionGrantInput): Promise<ConnectorExecutionGrant> {
const now = this.now();
const lease = normalizeConnectorLease(input.lease);
const scopes = normalizeConnectorScopes(input.scopes);
const correlationId = normalizeCorrelationId(input.correlationId);
const ttlMs = normalizeTtl(input.ttlMs, this.maxGrantTtlMs, 'grant');
const current = await this.store.findCurrent(lease);
await this.assertCurrentLease(current, lease, now, correlationId);
if (!current) throw new ConnectorLeaseError('lease_missing', 'Connector lease is unavailable');
if (!isScopeSubset(scopes, lease.scopes) || !isScopeSubset(scopes, current.scopes)) {
await this.reject(lease, correlationId, now, 'scope_denied');
}
const requestedExpiry = new Date(now.getTime() + ttlMs);
const leaseExpiry = new Date(current.expiresAt);
const grantExpiry = requestedExpiry < leaseExpiry ? requestedExpiry : leaseExpiry;
const grant: ConnectorExecutionGrant = Object.freeze({
identity: current.identity,
bindingId: current.bindingId,
leaseId: current.leaseId,
connectorId: current.connectorId,
scopes,
leaseEpoch: current.leaseEpoch,
issuedAt: now.toISOString(),
expiresAt: grantExpiry.toISOString(),
correlationId,
});
this.issuedGrants.add(grant);
return grant;
}
async executeGrant<TInput, TOutput>(
grant: ConnectorExecutionGrant,
requiredScope: string,
input: TInput,
adapter: FencedConnectorAdapter<TInput, TOutput>,
): Promise<TOutput> {
const now = this.now();
const normalizedScope = normalizeConnectorScope(requiredScope);
if (!this.issuedGrants.has(grant)) {
await this.rejectForgedGrant(grant, now);
}
if (new Date(grant.expiresAt) <= now) {
await this.rejectGrant(grant, now, 'grant_expired');
}
const current = await this.store.findCurrent(normalizeBinding(grant));
await this.assertCurrentLease(current, grant, now, grant.correlationId);
if (!grant.scopes.includes(normalizedScope) || !current?.scopes.includes(normalizedScope)) {
await this.rejectGrant(grant, now, 'scope_denied');
}
if (!current) throw new ConnectorLeaseError('lease_missing', 'Connector lease is unavailable');
const context: ConnectorExecutionContext = Object.freeze({
identity: current.identity,
bindingId: current.bindingId,
leaseId: current.leaseId,
connectorId: current.connectorId,
scopes: Object.freeze([...grant.scopes]),
leaseEpoch: current.leaseEpoch,
correlationId: grant.correlationId,
grantExpiresAt: grant.expiresAt,
});
return adapter.execute(input, context);
}
private async assertCurrentLease(
current: ConnectorLease | null,
authority: ConnectorLease | ConnectorExecutionGrant,
now: Date,
correlationId: string,
): Promise<void> {
if (!current) await this.reject(authority, correlationId, now, 'lease_missing');
if (!current) throw new ConnectorLeaseError('lease_missing', 'Connector lease is unavailable');
if (current.releasedAt) await this.reject(authority, correlationId, now, 'lease_released');
if (new Date(current.expiresAt) <= now) {
await this.store.recordAudit(auditEvent(current, correlationId, now, 'expiry', 'succeeded'));
await this.reject(authority, correlationId, now, 'lease_expired');
}
if (current.leaseEpoch !== authority.leaseEpoch) {
await this.reject(authority, correlationId, now, 'stale_epoch');
}
if (current.leaseId !== authority.leaseId || current.connectorId !== authority.connectorId) {
await this.reject(authority, correlationId, now, 'connector_mismatch');
}
}
private async rejectGrant(
grant: ConnectorExecutionGrant,
now: Date,
reason: ConnectorLeaseRejectReason,
): Promise<never> {
return this.reject(grant, grant.correlationId, now, reason);
}
private async rejectForgedGrant(grant: unknown, now: Date): Promise<never> {
const event = safeForgedGrantAudit(grant, now);
await this.store.recordAudit(event);
throw new ConnectorLeaseError('forged_grant', safeReasonMessage('forged_grant'));
}
private async reject(
authority: LogicalAgentBinding & {
readonly connectorId: string;
readonly leaseId?: string;
readonly leaseEpoch?: string;
},
correlationId: string,
now: Date,
reason: ConnectorLeaseRejectReason,
): Promise<never> {
await this.store.recordAudit(
auditEvent(authority, correlationId, now, 'reject', 'denied', reason),
);
throw new ConnectorLeaseError(reason, safeReasonMessage(reason));
}
}
function normalizeAcquireInput(
input: AcquireConnectorLeaseInput,
maxLeaseTtlMs: number,
): AcquireConnectorLeaseInput {
return {
identity: normalizeLogicalAgentIdentity(input.identity),
bindingId: normalizeLogicalBindingId(input.bindingId),
connectorId: normalizeConnectorId(input.connectorId),
scopes: normalizeConnectorScopes(input.scopes),
ttlMs: normalizeTtl(input.ttlMs, maxLeaseTtlMs, 'lease'),
correlationId: normalizeCorrelationId(input.correlationId),
};
}
function normalizeBinding(input: LogicalAgentBinding): LogicalAgentBinding {
return {
identity: normalizeLogicalAgentIdentity(input.identity),
bindingId: normalizeLogicalBindingId(input.bindingId),
};
}
export function normalizeConnectorLease(lease: ConnectorLease): ConnectorLease {
const binding = normalizeBinding(lease);
return Object.freeze({
...binding,
leaseId: lease.leaseId,
connectorId: normalizeConnectorId(lease.connectorId),
scopes: normalizeConnectorScopes(lease.scopes),
leaseEpoch: normalizeLeaseEpoch(lease.leaseEpoch),
acquiredAt: normalizeTimestamp(lease.acquiredAt, 'lease acquisition'),
heartbeatAt: normalizeTimestamp(lease.heartbeatAt, 'lease heartbeat'),
expiresAt: normalizeTimestamp(lease.expiresAt, 'lease expiry'),
...(lease.releasedAt
? { releasedAt: normalizeTimestamp(lease.releasedAt, 'lease release') }
: {}),
});
}
function normalizeTtl(ttlMs: number, maximum: number, kind: 'lease' | 'grant'): number {
if (!Number.isSafeInteger(ttlMs) || ttlMs <= 0 || ttlMs > maximum) {
throw new Error(
`Connector ${kind} TTL must be a positive safe integer no greater than ${maximum}ms`,
);
}
return ttlMs;
}
function normalizeTtlLimit(ttlMs: number, hardMaximum: number, kind: 'lease' | 'grant'): number {
if (!Number.isSafeInteger(ttlMs) || ttlMs <= 0 || ttlMs > hardMaximum) {
throw new Error(
`Maximum connector ${kind} TTL must be a positive safe integer no greater than ${hardMaximum}ms`,
);
}
return ttlMs;
}
function normalizeTimestamp(value: string, label: string): string {
const date = new Date(value);
if (Number.isNaN(date.getTime())) throw new Error(`${label} timestamp is invalid`);
return date.toISOString();
}
function expiresAt(now: Date, ttlMs: number): string {
const expiry = new Date(now.getTime() + ttlMs);
if (Number.isNaN(expiry.getTime())) throw new Error('Connector lease TTL exceeds date range');
return expiry.toISOString();
}
function isScopeSubset(requested: readonly string[], allowed: readonly string[]): boolean {
return requested.every((scope) => allowed.includes(scope));
}
function auditEvent(
authority: LogicalAgentBinding & {
readonly connectorId: string;
readonly leaseId?: string;
readonly leaseEpoch?: string;
},
correlationId: string,
now: Date,
event: ConnectorLeaseAuditEvent['event'],
outcome: ConnectorLeaseAuditEvent['outcome'],
reason?: ConnectorLeaseRejectReason,
): ConnectorLeaseAuditEvent {
return {
identity: authority.identity,
bindingId: authority.bindingId,
connectorId: authority.connectorId,
correlationId,
occurredAt: now.toISOString(),
event,
outcome,
...(authority.leaseId ? { leaseId: authority.leaseId } : {}),
...(authority.leaseEpoch ? { leaseEpoch: authority.leaseEpoch } : {}),
...(reason ? { reason } : {}),
};
}
function safeForgedGrantAudit(grant: unknown, now: Date): ConnectorLeaseAuditEvent {
const fallback: ConnectorLeaseAuditEvent = {
identity: { tenantId: 'untrusted', logicalAgentId: 'untrusted' },
bindingId: 'untrusted',
connectorId: 'untrusted',
correlationId: 'untrusted',
occurredAt: now.toISOString(),
event: 'reject',
outcome: 'denied',
reason: 'forged_grant',
};
if (typeof grant !== 'object' || grant === null || !('identity' in grant)) return fallback;
const identity = grant.identity;
if (typeof identity !== 'object' || identity === null) return fallback;
if (!('tenantId' in identity) || !('logicalAgentId' in identity)) return fallback;
if (!('bindingId' in grant) || !('connectorId' in grant) || !('correlationId' in grant)) {
return fallback;
}
if (
typeof identity.tenantId !== 'string' ||
typeof identity.logicalAgentId !== 'string' ||
typeof grant.bindingId !== 'string' ||
typeof grant.connectorId !== 'string' ||
typeof grant.correlationId !== 'string'
) {
return fallback;
}
try {
return {
identity: normalizeLogicalAgentIdentity({
tenantId: identity.tenantId,
logicalAgentId: identity.logicalAgentId,
}),
bindingId: normalizeLogicalBindingId(grant.bindingId),
connectorId: normalizeConnectorId(grant.connectorId),
correlationId: normalizeCorrelationId(grant.correlationId),
occurredAt: now.toISOString(),
event: 'reject',
outcome: 'denied',
reason: 'forged_grant',
};
} catch {
return fallback;
}
}
function safeReasonMessage(reason: ConnectorLeaseRejectReason): string {
return `Connector authority denied: ${reason}`;
}

View File

@@ -5,3 +5,4 @@ export * from './tmux-fleet-runtime-provider.js';
export * from './hermes-runtime-provider.js';
export * from './matrix-native-runtime-provider.js';
export * from './durable-session.js';
export * from './connector-lease.js';

View File

@@ -0,0 +1,36 @@
CREATE TABLE "connector_lease_audit_log" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"tenant_id" text NOT NULL,
"logical_agent_id" text NOT NULL,
"binding_id" text NOT NULL,
"connector_id" text NOT NULL,
"lease_id" uuid,
"lease_epoch" bigint,
"event" text NOT NULL,
"outcome" text NOT NULL,
"reason" text,
"correlation_id" text NOT NULL,
"occurred_at" timestamp with time zone NOT NULL
);
--> statement-breakpoint
CREATE TABLE "logical_agent_connector_leases" (
"lease_id" uuid PRIMARY KEY NOT NULL,
"tenant_id" text NOT NULL,
"logical_agent_id" text NOT NULL,
"binding_id" text NOT NULL,
"connector_id" text NOT NULL,
"scopes" jsonb NOT NULL,
"lease_epoch" bigint NOT NULL,
"acquired_at" timestamp with time zone NOT NULL,
"heartbeat_at" timestamp with time zone NOT NULL,
"expires_at" timestamp with time zone NOT NULL,
"released_at" timestamp with time zone,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"updated_at" timestamp with time zone DEFAULT now() NOT NULL
);
--> statement-breakpoint
CREATE INDEX "connector_lease_audit_binding_occurred_idx" ON "connector_lease_audit_log" USING btree ("tenant_id","logical_agent_id","binding_id","occurred_at" DESC NULLS LAST);--> statement-breakpoint
CREATE INDEX "connector_lease_audit_correlation_idx" ON "connector_lease_audit_log" USING btree ("correlation_id");--> statement-breakpoint
CREATE UNIQUE INDEX "logical_agent_connector_lease_binding_idx" ON "logical_agent_connector_leases" USING btree ("tenant_id","logical_agent_id","binding_id");--> statement-breakpoint
CREATE INDEX "logical_agent_connector_lease_expiry_idx" ON "logical_agent_connector_leases" USING btree ("expires_at");--> statement-breakpoint
CREATE INDEX "logical_agent_connector_lease_connector_idx" ON "logical_agent_connector_leases" USING btree ("connector_id");

File diff suppressed because it is too large Load Diff

View File

@@ -113,6 +113,13 @@
"when": 1783942610000,
"tag": "0015_interaction_checkpoint_payload_digest",
"breakpoints": true
},
{
"idx": 16,
"version": "7",
"when": 1784050648841,
"tag": "0016_salty_morlocks",
"breakpoints": true
}
]
}

View File

@@ -15,6 +15,7 @@ import {
uniqueIndex,
real,
integer,
bigint,
customType,
} from 'drizzle-orm/pg-core';
@@ -487,6 +488,68 @@ export const agentLogs = pgTable(
],
);
// ─── Logical agent connector authority ──────────────────────────────────────
// One durable row is the current authority for a tenant/logical-agent/binding.
// Runtime-native session identifiers never enter these core tables.
export const logicalAgentConnectorLeases = pgTable(
'logical_agent_connector_leases',
{
leaseId: uuid('lease_id').primaryKey(),
tenantId: text('tenant_id').notNull(),
logicalAgentId: text('logical_agent_id').notNull(),
bindingId: text('binding_id').notNull(),
connectorId: text('connector_id').notNull(),
scopes: jsonb('scopes').notNull().$type<string[]>(),
leaseEpoch: bigint('lease_epoch', { mode: 'bigint' }).notNull(),
acquiredAt: timestamp('acquired_at', { withTimezone: true }).notNull(),
heartbeatAt: timestamp('heartbeat_at', { withTimezone: true }).notNull(),
expiresAt: timestamp('expires_at', { withTimezone: true }).notNull(),
releasedAt: timestamp('released_at', { withTimezone: true }),
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(),
},
(t) => [
uniqueIndex('logical_agent_connector_lease_binding_idx').on(
t.tenantId,
t.logicalAgentId,
t.bindingId,
),
index('logical_agent_connector_lease_expiry_idx').on(t.expiresAt),
index('logical_agent_connector_lease_connector_idx').on(t.connectorId),
],
);
/** Append-only, credential-safe lease lifecycle and fencing denial metadata. */
export const connectorLeaseAuditLog = pgTable(
'connector_lease_audit_log',
{
id: uuid('id').primaryKey().defaultRandom(),
tenantId: text('tenant_id').notNull(),
logicalAgentId: text('logical_agent_id').notNull(),
bindingId: text('binding_id').notNull(),
connectorId: text('connector_id').notNull(),
leaseId: uuid('lease_id'),
leaseEpoch: bigint('lease_epoch', { mode: 'bigint' }),
event: text('event', {
enum: ['acquire', 'renew', 'takeover', 'reject', 'release', 'expiry'],
}).notNull(),
outcome: text('outcome', { enum: ['succeeded', 'denied'] }).notNull(),
reason: text('reason'),
correlationId: text('correlation_id').notNull(),
occurredAt: timestamp('occurred_at', { withTimezone: true }).notNull(),
},
(t) => [
index('connector_lease_audit_binding_occurred_idx').on(
t.tenantId,
t.logicalAgentId,
t.bindingId,
t.occurredAt.desc(),
),
index('connector_lease_audit_correlation_idx').on(t.correlationId),
],
);
// ─── Tess durable session state ─────────────────────────────────────────────
// PostgreSQL is canonical for restart-safe Tess session recovery. The state
// machine lives in @mosaicstack/agent; these records are its durable adapter.

View File

@@ -0,0 +1,242 @@
import { readFile } from 'node:fs/promises';
import { fileURLToPath } from 'node:url';
import { describe, expect, it } from 'vitest';
import {
ROSTER_V2_JSON_SCHEMA,
RosterV2ValidationError,
parseRosterV2,
renderRosterV2Yaml,
} from './roster-v2.js';
const validRoster = `
version: 2
generation: 7
transport: tmux
tmux:
socket_name: mosaic-fleet
holder_session: _holder
defaults:
working_directory: ~/src
runtime: pi
runtimes:
pi:
reset_command: /new
agents:
- name: coder0
alias: Coder 0
class: code
runtime: pi
provider: openai
model: gpt-5.6-sol
reasoning: high
tool_policy: code
working_directory: ~/src
persistent_persona: false
reset_between_tasks: true
lifecycle:
enabled: true
desired_state: stopped
launch:
yolo: true
`;
describe('roster v2 structural compiler', (): void => {
it('parses YAML into a normalized typed model and renders canonical YAML', (): void => {
const roster = parseRosterV2(validRoster, 'yaml');
expect(roster).toEqual({
version: 2,
generation: 7,
transport: 'tmux',
tmux: { socketName: 'mosaic-fleet', holderSession: '_holder' },
defaults: { workingDirectory: '~/src', runtime: 'pi' },
runtimes: { pi: { resetCommand: '/new' } },
agents: [
{
name: 'coder0',
alias: 'Coder 0',
className: 'code',
runtime: 'pi',
provider: 'openai',
model: 'gpt-5.6-sol',
reasoning: 'high',
toolPolicy: 'code',
workingDirectory: '~/src',
persistentPersona: false,
resetBetweenTasks: true,
lifecycle: { enabled: true, desiredState: 'stopped' },
launch: { yolo: true },
},
],
});
expect(renderRosterV2Yaml(roster)).toBe(validRoster.trimStart());
});
it('parses JSON and produces the same normalized model', (): void => {
const yaml = parseRosterV2(validRoster, 'yaml');
const json = JSON.stringify({
version: 2,
generation: 7,
transport: 'tmux',
tmux: { socket_name: 'mosaic-fleet', holder_session: '_holder' },
defaults: { working_directory: '~/src', runtime: 'pi' },
runtimes: { pi: { reset_command: '/new' } },
agents: [
{
name: 'coder0',
alias: 'Coder 0',
class: 'code',
runtime: 'pi',
provider: 'openai',
model: 'gpt-5.6-sol',
reasoning: 'high',
tool_policy: 'code',
working_directory: '~/src',
persistent_persona: false,
reset_between_tasks: true,
lifecycle: { enabled: true, desired_state: 'stopped' },
launch: { yolo: true },
},
],
});
expect(parseRosterV2(json, 'json')).toEqual(yaml);
});
it('sorts runtime and agent maps in the deterministic renderer', (): void => {
const roster = parseRosterV2(
validRoster
.replace(
'runtimes:\n pi:\n reset_command: /new',
'runtimes:\n pi:\n reset_command: /new\n codex:\n reset_command: /clear',
)
.replace(
'agents:\n - name: coder0',
'agents:\n - name: reviewer\n alias: Reviewer\n class: review\n runtime: pi\n provider: openai\n model: gpt-5.6-sol\n reasoning: medium\n tool_policy: review\n working_directory: ~/src\n persistent_persona: false\n reset_between_tasks: true\n lifecycle:\n enabled: true\n desired_state: stopped\n launch:\n yolo: true\n - name: coder0',
),
'yaml',
);
const rendered = renderRosterV2Yaml(roster);
expect(rendered.indexOf(' codex:')).toBeLessThan(rendered.indexOf(' pi:'));
expect(rendered.indexOf(' - name: coder0')).toBeLessThan(
rendered.indexOf(' - name: reviewer'),
);
});
it.each([
['v1 document', validRoster.replace('version: 2', 'version: 1'), /v1.*existing v1 path/i],
[
'unknown connector',
`${validRoster}\nconnector:\n kind: matrix\n`,
/unsupported field.*connector/i,
],
[
'remote host',
validRoster.replace(
'runtime: pi\n provider',
'runtime: pi\n host: remote\n provider',
),
/unsupported field.*host/i,
],
[
'secret reference',
validRoster.replace('model: gpt-5.6-sol', 'model: gpt-5.6-sol\n secret_ref: vault://x'),
/unsupported field.*secret_ref/i,
],
[
'channel override',
validRoster.replace('model: gpt-5.6-sol', 'model: gpt-5.6-sol\n channels: discord'),
/unsupported field.*channels/i,
],
[
'arbitrary command',
validRoster.replace('model: gpt-5.6-sol', 'model: gpt-5.6-sol\n command: whoami'),
/unsupported field.*command/i,
],
[
'gateway field',
`${validRoster}\ngateway:\n url: https://gateway.example\n`,
/unsupported field.*gateway/i,
],
[
'missing required field',
validRoster.replace(' model: gpt-5.6-sol\n', ''),
/model.*required/i,
],
[
'invalid type',
validRoster.replace('generation: 7', 'generation: seven'),
/generation.*integer/i,
],
[
'unsafe generation',
validRoster.replace('generation: 7', 'generation: 9007199254740992'),
/generation.*integer/i,
],
[
'duplicate names',
validRoster.replace(
' - name: coder0',
' - name: coder0\n alias: Duplicate\n class: code\n runtime: pi\n provider: openai\n model: gpt-5.6-sol\n reasoning: high\n tool_policy: code\n working_directory: ~/src\n persistent_persona: false\n reset_between_tasks: true\n lifecycle:\n enabled: true\n desired_state: stopped\n launch:\n yolo: true\n - name: coder0',
),
/duplicate agent name/i,
],
['invalid name', validRoster.replace('name: coder0', 'name: ../coder0'), /invalid agent name/i],
[
'invalid transport',
validRoster.replace('transport: tmux', 'transport: matrix'),
/transport.*tmux/i,
],
[
'invalid runtime',
validRoster.replace('runtime: pi', 'runtime: matrix'),
/runtime.*supported/i,
],
[
'invalid reasoning',
validRoster.replace('reasoning: high', 'reasoning: extreme'),
/reasoning.*low.*medium.*high/i,
],
[
'ambiguous socket',
validRoster.replace('socket_name: mosaic-fleet', 'socket_name: default/socket'),
/socket_name/i,
],
[
'agent socket override',
validRoster.replace(
'runtime: pi\n provider',
'runtime: pi\n socket: another\n provider',
),
/unsupported field.*socket/i,
],
])('rejects %s', (_name: string, source: string, expected: RegExp): void => {
expect((): void => {
parseRosterV2(source, 'yaml');
}).toThrow(expected);
});
it('rejects malformed JSON as a validation error', (): void => {
expect((): void => {
parseRosterV2('{', 'json');
}).toThrow(RosterV2ValidationError);
});
it('declares supported runtime map keys in the executable schema', (): void => {
expect(ROSTER_V2_JSON_SCHEMA).toMatchObject({
properties: {
runtimes: { propertyNames: { enum: ['claude', 'codex', 'opencode', 'pi'] } },
},
});
});
it('keeps the checked-in documentation schema structurally identical to the executable schema', async (): Promise<void> => {
const schemaPath = fileURLToPath(
new URL('../../../../docs/fleet/reference/roster-v2.schema.json', import.meta.url),
);
const documented = await readFile(schemaPath, 'utf8');
expect(JSON.parse(documented) as unknown).toEqual(ROSTER_V2_JSON_SCHEMA);
});
});

View File

@@ -0,0 +1,473 @@
import YAML from 'yaml';
export const ROSTER_V2_SUPPORTED_RUNTIMES = ['claude', 'codex', 'opencode', 'pi'] as const;
export const ROSTER_V2_REASONING_LEVELS = ['low', 'medium', 'high'] as const;
export const ROSTER_V2_DESIRED_STATES = ['running', 'stopped'] as const;
export type RosterV2RuntimeName = (typeof ROSTER_V2_SUPPORTED_RUNTIMES)[number];
export type RosterV2ReasoningLevel = (typeof ROSTER_V2_REASONING_LEVELS)[number];
export type RosterV2DesiredState = (typeof ROSTER_V2_DESIRED_STATES)[number];
export type RosterV2InputFormat = 'json' | 'yaml';
export interface FleetRosterV2Tmux {
readonly socketName: string;
readonly holderSession: string;
}
export interface FleetRosterV2Defaults {
readonly workingDirectory: string;
readonly runtime: RosterV2RuntimeName;
}
export interface FleetRosterV2Runtime {
readonly resetCommand: string;
}
export interface FleetRosterV2Lifecycle {
readonly enabled: boolean;
readonly desiredState: RosterV2DesiredState;
}
export interface FleetRosterV2Launch {
readonly yolo: boolean;
}
export interface FleetRosterV2Agent {
readonly name: string;
readonly alias: string;
readonly className: string;
readonly runtime: RosterV2RuntimeName;
readonly provider: string;
readonly model: string;
readonly reasoning: RosterV2ReasoningLevel;
readonly toolPolicy: string;
readonly workingDirectory: string;
readonly persistentPersona: boolean;
readonly resetBetweenTasks: boolean;
readonly lifecycle: FleetRosterV2Lifecycle;
readonly launch: FleetRosterV2Launch;
}
export interface FleetRosterV2 {
readonly version: 2;
readonly generation: number;
readonly transport: 'tmux';
readonly tmux: FleetRosterV2Tmux;
readonly defaults: FleetRosterV2Defaults;
readonly runtimes: Readonly<Record<string, FleetRosterV2Runtime>>;
readonly agents: readonly FleetRosterV2Agent[];
}
export class RosterV2ValidationError extends Error {
constructor(message: string) {
super(message);
this.name = 'RosterV2ValidationError';
}
}
type JsonSchema = string | number | boolean | null | JsonSchema[] | { [key: string]: JsonSchema };
/**
* Executable v2 structural contract. The checked-in documentation schema is
* structurally compared to this value in roster-v2.spec.ts.
*/
export const ROSTER_V2_JSON_SCHEMA: JsonSchema = {
$schema: 'https://json-schema.org/draft/2020-12/schema',
$id: 'https://mosaicstack.dev/schemas/fleet/roster-v2.schema.json',
title: 'Mosaic local tmux fleet roster v2',
type: 'object',
additionalProperties: false,
required: ['version', 'generation', 'transport', 'tmux', 'defaults', 'runtimes', 'agents'],
properties: {
version: { const: 2 },
generation: { type: 'integer', minimum: 1, maximum: Number.MAX_SAFE_INTEGER },
transport: { const: 'tmux' },
tmux: {
type: 'object',
additionalProperties: false,
required: ['socket_name', 'holder_session'],
properties: {
socket_name: { type: 'string', pattern: '^[A-Za-z0-9_.-]+$' },
holder_session: { type: 'string', pattern: '^[A-Za-z0-9_.-]+$' },
},
},
defaults: {
type: 'object',
additionalProperties: false,
required: ['working_directory', 'runtime'],
properties: {
working_directory: { type: 'string', minLength: 1 },
runtime: { enum: [...ROSTER_V2_SUPPORTED_RUNTIMES] },
},
},
runtimes: {
type: 'object',
minProperties: 1,
propertyNames: { enum: [...ROSTER_V2_SUPPORTED_RUNTIMES] },
additionalProperties: {
type: 'object',
additionalProperties: false,
required: ['reset_command'],
properties: { reset_command: { type: 'string', minLength: 1 } },
},
},
agents: {
type: 'array',
minItems: 1,
items: {
type: 'object',
additionalProperties: false,
required: [
'name',
'alias',
'class',
'runtime',
'provider',
'model',
'reasoning',
'tool_policy',
'working_directory',
'persistent_persona',
'reset_between_tasks',
'lifecycle',
'launch',
],
properties: {
name: { type: 'string', pattern: '^[A-Za-z0-9][A-Za-z0-9_.-]*$' },
alias: { type: 'string', minLength: 1 },
class: { type: 'string', pattern: '^[a-z][a-z0-9-]*$' },
runtime: { enum: [...ROSTER_V2_SUPPORTED_RUNTIMES] },
provider: { type: 'string', minLength: 1 },
model: { type: 'string', minLength: 1 },
reasoning: { enum: [...ROSTER_V2_REASONING_LEVELS] },
tool_policy: { type: 'string', pattern: '^[a-z][a-z0-9-]*$' },
working_directory: { type: 'string', minLength: 1 },
persistent_persona: { type: 'boolean' },
reset_between_tasks: { type: 'boolean' },
lifecycle: {
type: 'object',
additionalProperties: false,
required: ['enabled', 'desired_state'],
properties: {
enabled: { type: 'boolean' },
desired_state: { enum: [...ROSTER_V2_DESIRED_STATES] },
},
},
launch: {
type: 'object',
additionalProperties: false,
required: ['yolo'],
properties: { yolo: { type: 'boolean' } },
},
},
},
},
},
};
const ROOT_KEYS = ['version', 'generation', 'transport', 'tmux', 'defaults', 'runtimes', 'agents'];
const TMUX_KEYS = ['socket_name', 'holder_session'];
const DEFAULT_KEYS = ['working_directory', 'runtime'];
const RUNTIME_KEYS = ['reset_command'];
const AGENT_KEYS = [
'name',
'alias',
'class',
'runtime',
'provider',
'model',
'reasoning',
'tool_policy',
'working_directory',
'persistent_persona',
'reset_between_tasks',
'lifecycle',
'launch',
];
const LIFECYCLE_KEYS = ['enabled', 'desired_state'];
const LAUNCH_KEYS = ['yolo'];
const IDENTIFIER = /^[A-Za-z0-9][A-Za-z0-9_.-]*$/;
const TMUX_IDENTIFIER = /^[A-Za-z0-9_.-]+$/;
const POLICY_IDENTIFIER = /^[a-z][a-z0-9-]*$/;
/** Parses YAML or JSON and compiles only the local-tmux roster v2 contract. */
export function parseRosterV2(source: string, format?: RosterV2InputFormat): FleetRosterV2 {
const parsed = parseSource(source, format);
return normalizeRosterV2(parsed);
}
/** Renders canonical snake_case YAML with sorted runtime and agent entries. */
export function renderRosterV2Yaml(roster: FleetRosterV2): string {
const normalized = normalizeRosterV2(toSourceShape(roster));
return YAML.stringify(toSourceShape(normalized));
}
function parseSource(source: string, format?: RosterV2InputFormat): unknown {
const resolvedFormat = format ?? (source.trimStart().startsWith('{') ? 'json' : 'yaml');
try {
if (resolvedFormat === 'json') return JSON.parse(source) as unknown;
return YAML.parse(source) as unknown;
} catch (error: unknown) {
const detail = error instanceof Error ? error.message : String(error);
throw new RosterV2ValidationError(`Roster v2 ${resolvedFormat} parse failed: ${detail}`);
}
}
export function normalizeRosterV2(raw: unknown): FleetRosterV2 {
const root = requiredObject(raw, 'Roster v2');
assertKnownKeys(root, 'Roster v2', ROOT_KEYS);
if (root.version === 1) {
throw new RosterV2ValidationError(
'Roster v2 compiler rejects v1 input; use the existing v1 path until migration.',
);
}
if (root.version !== 2) throw new RosterV2ValidationError('Roster v2 version must be 2.');
const generation = requiredPositiveInteger(root.generation, 'Roster v2 generation');
const transport = requiredEnum(root.transport, 'Roster v2 transport', ['tmux'] as const);
const tmux = normalizeTmux(root.tmux);
const defaults = normalizeDefaults(root.defaults);
const runtimes = normalizeRuntimes(root.runtimes);
const agents = normalizeAgents(root.agents, runtimes);
if (!runtimes[defaults.runtime]) {
throw new RosterV2ValidationError(
`Roster v2 defaults runtime "${defaults.runtime}" must be declared in runtimes.`,
);
}
return { version: 2, generation, transport, tmux, defaults, runtimes, agents };
}
function normalizeTmux(value: unknown): FleetRosterV2Tmux {
const raw = requiredObject(value, 'Roster v2 tmux');
assertKnownKeys(raw, 'Roster v2 tmux', TMUX_KEYS);
return {
socketName: requiredTmuxIdentifier(raw.socket_name, 'Roster v2 tmux socket_name'),
holderSession: requiredTmuxIdentifier(raw.holder_session, 'Roster v2 tmux holder_session'),
};
}
function normalizeDefaults(value: unknown): FleetRosterV2Defaults {
const raw = requiredObject(value, 'Roster v2 defaults');
assertKnownKeys(raw, 'Roster v2 defaults', DEFAULT_KEYS);
return {
workingDirectory: requiredString(raw.working_directory, 'Roster v2 defaults working_directory'),
runtime: requiredRuntime(raw.runtime, 'Roster v2 defaults runtime'),
};
}
function normalizeRuntimes(value: unknown): Readonly<Record<string, FleetRosterV2Runtime>> {
const raw = requiredObject(value, 'Roster v2 runtimes');
const names = Object.keys(raw);
if (names.length === 0)
throw new RosterV2ValidationError('Roster v2 runtimes must not be empty.');
const result: Record<string, FleetRosterV2Runtime> = {};
for (const name of names.sort()) {
const runtime = requiredRuntime(name, 'Roster v2 runtime name');
const config = requiredObject(raw[name], `Roster v2 runtime "${runtime}"`);
assertKnownKeys(config, `Roster v2 runtime "${runtime}"`, RUNTIME_KEYS);
result[runtime] = {
resetCommand: requiredString(
config.reset_command,
`Roster v2 runtime "${runtime}" reset_command`,
),
};
}
return result;
}
function normalizeAgents(
value: unknown,
runtimes: Readonly<Record<string, FleetRosterV2Runtime>>,
): readonly FleetRosterV2Agent[] {
if (!Array.isArray(value) || value.length === 0) {
throw new RosterV2ValidationError('Roster v2 agents must be a non-empty array.');
}
const seen = new Set<string>();
const agents = value.map((candidate: unknown, index: number): FleetRosterV2Agent => {
const raw = requiredObject(candidate, `Roster v2 agents[${index}]`);
assertKnownKeys(raw, `Roster v2 agents[${index}]`, AGENT_KEYS);
const name = requiredIdentifier(raw.name, `Roster v2 agents[${index}] name`);
if (seen.has(name))
throw new RosterV2ValidationError(`Roster v2 has duplicate agent name: ${name}.`);
seen.add(name);
const runtime = requiredRuntime(raw.runtime, `Roster v2 agent "${name}" runtime`);
if (!runtimes[runtime]) {
throw new RosterV2ValidationError(
`Roster v2 agent "${name}" runtime "${runtime}" must be declared in runtimes.`,
);
}
return {
name,
alias: requiredString(raw.alias, `Roster v2 agent "${name}" alias`),
className: requiredPolicyIdentifier(raw.class, `Roster v2 agent "${name}" class`),
runtime,
provider: requiredString(raw.provider, `Roster v2 agent "${name}" provider`),
model: requiredString(raw.model, `Roster v2 agent "${name}" model`),
reasoning: requiredEnum(
raw.reasoning,
`Roster v2 agent "${name}" reasoning`,
ROSTER_V2_REASONING_LEVELS,
),
toolPolicy: requiredPolicyIdentifier(
raw.tool_policy,
`Roster v2 agent "${name}" tool_policy`,
),
workingDirectory: requiredString(
raw.working_directory,
`Roster v2 agent "${name}" working_directory`,
),
persistentPersona: requiredBoolean(
raw.persistent_persona,
`Roster v2 agent "${name}" persistent_persona`,
),
resetBetweenTasks: requiredBoolean(
raw.reset_between_tasks,
`Roster v2 agent "${name}" reset_between_tasks`,
),
lifecycle: normalizeLifecycle(raw.lifecycle, name),
launch: normalizeLaunch(raw.launch, name),
};
});
return agents.sort((left: FleetRosterV2Agent, right: FleetRosterV2Agent): number =>
left.name.localeCompare(right.name),
);
}
function normalizeLifecycle(value: unknown, agentName: string): FleetRosterV2Lifecycle {
const raw = requiredObject(value, `Roster v2 agent "${agentName}" lifecycle`);
assertKnownKeys(raw, `Roster v2 agent "${agentName}" lifecycle`, LIFECYCLE_KEYS);
return {
enabled: requiredBoolean(raw.enabled, `Roster v2 agent "${agentName}" lifecycle enabled`),
desiredState: requiredEnum(
raw.desired_state,
`Roster v2 agent "${agentName}" lifecycle desired_state`,
ROSTER_V2_DESIRED_STATES,
),
};
}
function normalizeLaunch(value: unknown, agentName: string): FleetRosterV2Launch {
const raw = requiredObject(value, `Roster v2 agent "${agentName}" launch`);
assertKnownKeys(raw, `Roster v2 agent "${agentName}" launch`, LAUNCH_KEYS);
return { yolo: requiredBoolean(raw.yolo, `Roster v2 agent "${agentName}" launch yolo`) };
}
function requiredObject(value: unknown, label: string): Record<string, unknown> {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
throw new RosterV2ValidationError(`${label} must be an object.`);
}
return value as Record<string, unknown>;
}
function assertKnownKeys(
value: Record<string, unknown>,
label: string,
allowedKeys: readonly string[],
): void {
const allowed = new Set(allowedKeys);
const unknown = Object.keys(value).filter((key: string): boolean => !allowed.has(key));
if (unknown.length > 0) {
throw new RosterV2ValidationError(`${label} has unsupported field(s): ${unknown.join(', ')}.`);
}
}
function requiredString(value: unknown, label: string): string {
if (typeof value !== 'string' || value.trim() === '') {
throw new RosterV2ValidationError(`${label} is required and must be a non-empty string.`);
}
return value.trim();
}
function requiredIdentifier(value: unknown, label: string): string {
const result = requiredString(value, label);
if (!IDENTIFIER.test(result)) {
throw new RosterV2ValidationError(`Invalid agent name (${label}): ${result}.`);
}
return result;
}
function requiredTmuxIdentifier(value: unknown, label: string): string {
const result = requiredString(value, label);
if (!TMUX_IDENTIFIER.test(result))
throw new RosterV2ValidationError(`Invalid ${label}: ${result}.`);
return result;
}
function requiredPolicyIdentifier(value: unknown, label: string): string {
const result = requiredString(value, label);
if (!POLICY_IDENTIFIER.test(result))
throw new RosterV2ValidationError(`Invalid ${label}: ${result}.`);
return result;
}
function requiredPositiveInteger(value: unknown, label: string): number {
if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 1) {
throw new RosterV2ValidationError(`${label} must be a positive integer.`);
}
return value;
}
function requiredBoolean(value: unknown, label: string): boolean {
if (typeof value !== 'boolean') throw new RosterV2ValidationError(`${label} must be a boolean.`);
return value;
}
function requiredRuntime(value: unknown, label: string): RosterV2RuntimeName {
return requiredEnum(value, label, ROSTER_V2_SUPPORTED_RUNTIMES);
}
function requiredEnum<T extends string>(value: unknown, label: string, allowed: readonly T[]): T {
if (typeof value !== 'string' || !allowed.includes(value as T)) {
throw new RosterV2ValidationError(
`${label} must be one of the supported values: ${allowed.join(', ')}.`,
);
}
return value as T;
}
function toSourceShape(roster: FleetRosterV2): Record<string, unknown> {
return {
version: roster.version,
generation: roster.generation,
transport: roster.transport,
tmux: { socket_name: roster.tmux.socketName, holder_session: roster.tmux.holderSession },
defaults: {
working_directory: roster.defaults.workingDirectory,
runtime: roster.defaults.runtime,
},
runtimes: Object.fromEntries(
Object.entries(roster.runtimes)
.sort(([left], [right]): number => left.localeCompare(right))
.map(([name, runtime]): [string, unknown] => [
name,
{ reset_command: runtime.resetCommand },
]),
),
agents: [...roster.agents]
.sort((left: FleetRosterV2Agent, right: FleetRosterV2Agent): number =>
left.name.localeCompare(right.name),
)
.map(
(agent: FleetRosterV2Agent): Record<string, unknown> => ({
name: agent.name,
alias: agent.alias,
class: agent.className,
runtime: agent.runtime,
provider: agent.provider,
model: agent.model,
reasoning: agent.reasoning,
tool_policy: agent.toolPolicy,
working_directory: agent.workingDirectory,
persistent_persona: agent.persistentPersona,
reset_between_tasks: agent.resetBetweenTasks,
lifecycle: {
enabled: agent.lifecycle.enabled,
desired_state: agent.lifecycle.desiredState,
},
launch: { yolo: agent.launch.yolo },
}),
),
};
}

View File

@@ -0,0 +1,59 @@
import { describe, expect, it } from 'vitest';
import {
normalizeConnectorId,
normalizeConnectorScopes,
normalizeLeaseEpoch,
normalizeLogicalAgentIdentity,
normalizeLogicalBindingId,
type ConnectorExecutionContext,
type LogicalAgentIdentity,
} from './connector-lease.dto.js';
describe('logical agent connector lease contract', (): void => {
it('normalizes a runtime-neutral logical identity and binding vocabulary', (): void => {
const identity = normalizeLogicalAgentIdentity({
tenantId: ' tenant-01 ',
logicalAgentId: ' MOS.Primary ',
});
expect(identity).toEqual({ tenantId: 'tenant-01', logicalAgentId: 'mos.primary' });
expect(normalizeLogicalBindingId(' Discord:Operations ')).toBe('discord:operations');
expect(normalizeConnectorId(' PI.Worker-01 ')).toBe('pi.worker-01');
expect(Object.isFrozen(identity)).toBe(true);
expect(Object.keys(identity).sort()).toEqual(['logicalAgentId', 'tenantId']);
});
it('canonicalizes scopes and decimal fencing epochs', (): void => {
expect(normalizeConnectorScopes([' Runtime.Send ', 'tool.execute', 'runtime.send'])).toEqual([
'runtime.send',
'tool.execute',
]);
expect(normalizeLeaseEpoch('00042')).toBe('42');
});
it.each([
['', 'mos'],
['tenant', ''],
['tenant', 'claude session/123'],
])('rejects ambiguous identity values tenant=%j agent=%j', (tenantId, logicalAgentId): void => {
expect(() => normalizeLogicalAgentIdentity({ tenantId, logicalAgentId })).toThrow();
});
it('defines an adapter context without harness-native identity fields', (): void => {
const identity: LogicalAgentIdentity = { tenantId: 'tenant-01', logicalAgentId: 'mos' };
const context: ConnectorExecutionContext = {
identity,
bindingId: 'operator-chat',
connectorId: 'connector-a',
leaseId: '00000000-0000-4000-8000-000000000001',
leaseEpoch: '7',
scopes: ['runtime.send'],
correlationId: 'correlation-1',
grantExpiresAt: '2026-07-14T18:00:00.000Z',
};
expect(context).not.toHaveProperty('sessionId');
expect(context).not.toHaveProperty('tmuxSession');
expect(context).not.toHaveProperty('providerSessionId');
});
});

View File

@@ -0,0 +1,199 @@
const ID_PATTERN = /^[a-z0-9][a-z0-9._:@-]{0,127}$/;
const TENANT_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$/;
const SCOPE_PATTERN = /^[a-z][a-z0-9._:-]{0,127}$/;
/** Stable Mosaic identity. It intentionally contains no runtime/provider session identifier. */
export interface LogicalAgentIdentity {
readonly tenantId: string;
readonly logicalAgentId: string;
}
export interface LogicalAgentBinding {
readonly identity: LogicalAgentIdentity;
readonly bindingId: string;
}
export interface ConnectorLease extends LogicalAgentBinding {
readonly leaseId: string;
readonly connectorId: string;
readonly scopes: readonly string[];
readonly leaseEpoch: string;
readonly acquiredAt: string;
readonly heartbeatAt: string;
readonly expiresAt: string;
readonly releasedAt?: string;
}
export interface AcquireConnectorLeaseInput {
readonly identity: LogicalAgentIdentity;
readonly bindingId: string;
readonly connectorId: string;
readonly scopes: readonly string[];
readonly ttlMs: number;
readonly correlationId: string;
}
export interface TakeoverConnectorLeaseInput extends AcquireConnectorLeaseInput {
readonly expectedEpoch: string;
}
export interface HeartbeatConnectorLeaseInput {
readonly lease: ConnectorLease;
readonly ttlMs: number;
readonly correlationId: string;
}
export interface ReleaseConnectorLeaseInput {
readonly lease: ConnectorLease;
readonly correlationId: string;
}
export interface IssueConnectorExecutionGrantInput {
readonly lease: ConnectorLease;
readonly scopes: readonly string[];
readonly ttlMs: number;
readonly correlationId: string;
}
/** Internal server grant. Object provenance is checked in addition to these fields. */
export interface ConnectorExecutionGrant extends LogicalAgentBinding {
readonly leaseId: string;
readonly connectorId: string;
readonly scopes: readonly string[];
readonly leaseEpoch: string;
readonly issuedAt: string;
readonly expiresAt: string;
readonly correlationId: string;
}
/** Normalized context passed to an adapter only after current-lease validation. */
export interface ConnectorExecutionContext extends LogicalAgentBinding {
readonly leaseId: string;
readonly connectorId: string;
readonly scopes: readonly string[];
readonly leaseEpoch: string;
readonly correlationId: string;
readonly grantExpiresAt: string;
}
export interface FencedConnectorAdapter<TInput, TOutput> {
execute(input: TInput, context: ConnectorExecutionContext): Promise<TOutput>;
}
export type ConnectorLeaseAuditEventType =
| 'acquire'
| 'renew'
| 'takeover'
| 'reject'
| 'release'
| 'expiry';
export type ConnectorLeaseAuditOutcome = 'succeeded' | 'denied';
export type ConnectorLeaseRejectReason =
| 'policy_denied'
| 'lease_held'
| 'takeover_required'
| 'cas_mismatch'
| 'lease_missing'
| 'lease_released'
| 'lease_expired'
| 'stale_epoch'
| 'connector_mismatch'
| 'scope_denied'
| 'forged_grant'
| 'grant_expired';
/** Credential-safe metadata only: no grant object, scope set, payload, token, or approval ref. */
export interface ConnectorLeaseAuditEvent extends LogicalAgentBinding {
readonly event: ConnectorLeaseAuditEventType;
readonly outcome: ConnectorLeaseAuditOutcome;
readonly connectorId: string;
readonly correlationId: string;
readonly occurredAt: string;
readonly leaseId?: string;
readonly leaseEpoch?: string;
readonly reason?: ConnectorLeaseRejectReason;
}
export interface ConnectorLeaseAcquireMutation extends AcquireConnectorLeaseInput {
readonly leaseId: string;
readonly now: string;
readonly expiresAt: string;
}
export interface ConnectorLeaseTakeoverMutation extends TakeoverConnectorLeaseInput {
readonly leaseId: string;
readonly now: string;
readonly expiresAt: string;
}
export interface ConnectorLeaseHeartbeatMutation extends HeartbeatConnectorLeaseInput {
readonly now: string;
readonly expiresAt: string;
}
export interface ConnectorLeaseReleaseMutation extends ReleaseConnectorLeaseInput {
readonly now: string;
}
export interface ConnectorLeaseStore {
acquire(input: ConnectorLeaseAcquireMutation): Promise<ConnectorLease>;
takeover(input: ConnectorLeaseTakeoverMutation): Promise<ConnectorLease>;
heartbeat(input: ConnectorLeaseHeartbeatMutation): Promise<ConnectorLease>;
release(input: ConnectorLeaseReleaseMutation): Promise<void>;
findCurrent(binding: LogicalAgentBinding): Promise<ConnectorLease | null>;
recordAudit(event: ConnectorLeaseAuditEvent): Promise<void>;
}
export function normalizeLogicalAgentIdentity(input: LogicalAgentIdentity): LogicalAgentIdentity {
const tenantId = requiredIdentifier(input.tenantId, 'tenant ID', TENANT_PATTERN, false);
const logicalAgentId = requiredIdentifier(
input.logicalAgentId,
'logical agent ID',
ID_PATTERN,
true,
);
return Object.freeze({ tenantId, logicalAgentId });
}
export function normalizeLogicalBindingId(value: string): string {
return requiredIdentifier(value, 'logical binding ID', ID_PATTERN, true);
}
export function normalizeConnectorId(value: string): string {
return requiredIdentifier(value, 'connector ID', ID_PATTERN, true);
}
export function normalizeCorrelationId(value: string): string {
return requiredIdentifier(value, 'correlation ID', TENANT_PATTERN, false);
}
export function normalizeConnectorScope(value: string): string {
return requiredIdentifier(value, 'connector scope', SCOPE_PATTERN, true);
}
export function normalizeConnectorScopes(values: readonly string[]): readonly string[] {
if (values.length === 0) throw new Error('At least one connector scope is required');
return Object.freeze(Array.from(new Set(values.map(normalizeConnectorScope))).sort());
}
export function normalizeLeaseEpoch(value: string): string {
const trimmed = value.trim();
if (!/^\d+$/.test(trimmed)) throw new Error('Lease epoch must be a positive decimal integer');
const epoch = BigInt(trimmed);
if (epoch < 1n) throw new Error('Lease epoch must be a positive decimal integer');
return epoch.toString(10);
}
function requiredIdentifier(
value: string,
label: string,
pattern: RegExp,
lowerCase: boolean,
): string {
const trimmed = value.trim();
const normalized = lowerCase ? trimmed.toLowerCase() : trimmed;
if (!pattern.test(normalized)) {
throw new Error(`${label} has an invalid normalized format`);
}
return normalized;
}

View File

@@ -4,3 +4,4 @@ export interface AgentSessionHandle {
}
export * from './agent-runtime-provider.js';
export * from './connector-lease.dto.js';

View File

@@ -0,0 +1,43 @@
import type {
ChannelAdapterHealthDto,
ChannelEgressDto,
ChannelIngressDto,
} from './channel.dto.js';
export type ChannelDeliveryErrorCode =
| 'invalid_route'
| 'destination_unavailable'
| 'delivery_failed';
/** Terminal adapter delivery failure surfaced to the gateway/caller. */
export class ChannelDeliveryError extends Error {
readonly name = 'ChannelDeliveryError';
constructor(
readonly code: ChannelDeliveryErrorCode,
message: string,
readonly retryable = false,
options?: ErrorOptions,
) {
super(message, options);
}
}
/** Gateway policy boundary consumed by official channel adapters. */
export interface ChannelIngressPort {
receive(ingress: ChannelIngressDto): Promise<void>;
}
/** Adapter egress boundary used by the gateway after agent output is ready. */
export interface ChannelEgressPort {
send(egress: ChannelEgressDto): Promise<void>;
}
/** Shared lifecycle seam implemented by every official channel adapter. */
export interface OfficialChannelAdapter {
readonly name: string;
start(): Promise<void>;
stop(): Promise<void>;
/** Health is best-effort and never throws for ordinary disconnected state. */
health(): Promise<ChannelAdapterHealthDto>;
}

View File

@@ -0,0 +1,97 @@
/** JSON-safe metadata carried across channel adapter boundaries. */
export type ChannelMetadataValue =
| string
| number
| boolean
| null
| readonly ChannelMetadataValue[]
| { readonly [key: string]: ChannelMetadataValue };
export type ChannelSenderKind = 'user' | 'agent' | 'system';
export type ChannelContentKind = 'text' | 'markdown' | 'code' | 'image' | 'file';
export type ChannelAdapterStatus = 'connected' | 'degraded' | 'disconnected';
export type ChannelAuthorizationRole = 'viewer' | 'operator' | 'admin';
export type ChannelOperation = 'message.send' | 'approval.create' | 'session.stop';
export interface ChannelAttachmentDto {
id: string;
name: string;
mimeType: string | null;
url: string;
sizeBytes?: number;
}
/** Canonical transport-neutral message shape for official channel adapters. */
export interface ChannelMessageDto {
id: string;
channelName: string;
channelId: string;
senderId: string;
senderKind: ChannelSenderKind;
content: string;
contentKind: ChannelContentKind;
timestamp: string;
threadId?: string;
replyToId?: string;
attachments?: readonly ChannelAttachmentDto[];
metadata: Readonly<Record<string, ChannelMetadataValue>>;
}
/** Where an adapter must deliver a response for one normalized conversation turn. */
/** Provisioned external identity after adapter allowlist/pairing checks pass. */
export interface ChannelAuthorizedPrincipalDto {
channelUserId: string;
role: ChannelAuthorizationRole;
/** Needed when gateway policy must authorize a privileged Mosaic operation. */
mosaicUserId?: string;
}
/** Configuration-owned binding. Credentials are intentionally absent. */
export interface ChannelBindingDto {
bindingId: string;
channelName: string;
workspaceId: string;
channelId: string;
logicalAgentId: string;
principals: Readonly<Record<string, ChannelAuthorizedPrincipalDto>>;
}
export interface ChannelResponseTargetDto {
channelId: string;
threadId?: string;
}
/**
* Stable channel-to-session route. Runtime provider, harness, model, process,
* and native runtime session identifiers are intentionally absent.
*/
export interface ChannelConversationRouteDto {
bindingId: string;
logicalAgentId: string;
conversationId: string;
channelName: string;
authorizationChannelId: string;
responseTarget: ChannelResponseTargetDto;
}
/** Authorized adapter-to-gateway ingress after native translation. */
export interface ChannelIngressDto {
correlationId: string;
nativeMessageId: string;
operation: ChannelOperation;
principal: ChannelAuthorizedPrincipalDto;
message: ChannelMessageDto;
route: ChannelConversationRouteDto;
}
/** Gateway-to-adapter egress; runtime/provider identity remains gateway-internal. */
export interface ChannelEgressDto {
correlationId: string;
message: ChannelMessageDto;
route: ChannelConversationRouteDto;
}
export interface ChannelAdapterHealthDto {
status: ChannelAdapterStatus;
detail?: string;
}

View File

@@ -0,0 +1,2 @@
export * from './channel-adapter.js';
export * from './channel.dto.js';

View File

@@ -1,3 +1,4 @@
import type { ChannelAttachmentDto } from '../channel/index.js';
import type {
CommandManifestPayload,
SlashCommandApprovalResultPayload,
@@ -73,6 +74,7 @@ export interface ChatMessagePayload {
provider?: string;
modelId?: string;
agentId?: string;
attachments?: readonly ChannelAttachmentDto[];
}
/** Routing decision summary included in session:info for transparency */

View File

@@ -1,5 +1,6 @@
export const VERSION = '0.0.0';
export * from './channel/index.js';
export * from './chat/index.js';
export * from './agent/index.js';
export * from './provider/index.js';

67
plugins/discord/README.md Normal file
View File

@@ -0,0 +1,67 @@
# @mosaicstack/discord-plugin
Official Discord channel adapter for Mosaic Stack.
## Behavior
- Runs independently of the bound agent's Claude, Codex, Pi, OpenCode, or future harness.
- Routes authorized untagged messages in configured agent channels and replies in-channel.
- Creates a public Discord thread when the bot is mentioned in a parent channel, or reuses the thread already attached to that same message.
- Keeps follow-ups in existing threads without requiring repeated mentions.
- Keeps `/approve` and `/stop <approval>` on the current durable session.
- Applies guild, parent-channel, user, pairing, role, and per-minute abuse limits before thread creation or gateway dispatch.
- Authenticates to the gateway and signs ingress envelopes with the injected service token.
## Required configuration
| Variable | Purpose |
| --------------------------------------- | -------------------------------------------------------------------- |
| `DISCORD_BOT_TOKEN` | Discord bot credential |
| `DISCORD_SERVICE_TOKEN` | High-entropy plugin-to-gateway credential |
| `DISCORD_SERVICE_USER_ID` | Provisioned Mosaic service principal |
| `DISCORD_ALLOWED_GUILD_IDS` | Comma-separated guild allowlist |
| `DISCORD_ALLOWED_CHANNEL_IDS` | Comma-separated parent-channel allowlist |
| `DISCORD_ALLOWED_USER_IDS` | Comma-separated Discord user allowlist |
| `DISCORD_INTERACTION_BINDINGS` | JSON channel→logical-agent bindings and paired-user roles |
| `DISCORD_GATEWAY_URL` | Gateway base URL; defaults to the gateway's local development URL |
| `DISCORD_MESSAGE_RATE_LIMIT_PER_MINUTE` | Authorized turns per guild/channel/user per minute; default `30` |
| `DISCORD_THREAD_RATE_LIMIT_PER_MINUTE` | Mention-thread routes per guild/channel/user per minute; default `5` |
Supply credentials through the approved runtime secret mechanism. Never commit tokens or binding data containing secrets.
Example binding shape (identifiers are placeholders):
```json
[
{
"instanceId": "interaction-agent",
"agentConfigId": "agent-config-id",
"guildId": "guild-id",
"channelId": "channel-id",
"pairedUsers": {
"discord-user-id": {
"role": "operator",
"mosaicUserId": "mosaic-user-id"
}
}
}
]
```
Each binding's trusted `agentConfigId` must identify a provisioned database agent configuration whose name exactly matches its `instanceId`. Roles are `viewer`, `operator`, and `admin`. `viewer` cannot send agent turns. Runtime approval and stop operations require an `admin` pairing with a provisioned `mosaicUserId`.
The bot needs Discord permissions to view/send messages in configured channels and create/send in public threads. A channel category is not an authorization boundary; threads inherit authorization only from their configured parent text channel.
## Shared contract
The adapter implements `OfficialChannelAdapter` from `@mosaicstack/types`. `ChannelConversationRouteDto` carries only stable logical-agent/channel identity and a response target. Gateway durable-session and provider layers own runtime selection and handoff; Discord code must not import a harness SDK.
## Development
```bash
pnpm --filter @mosaicstack/types build
pnpm --filter @mosaicstack/discord-plugin typecheck
pnpm --filter @mosaicstack/discord-plugin lint
pnpm --filter @mosaicstack/discord-plugin test
pnpm --filter @mosaicstack/discord-plugin build
```

View File

@@ -19,13 +19,16 @@
"dev": "tsx watch src/index.ts",
"lint": "eslint src",
"typecheck": "tsc --noEmit",
"test": "vitest run --passWithNoTests"
"test": "vitest run --coverage",
"test:coverage": "vitest run --coverage"
},
"dependencies": {
"@mosaicstack/types": "workspace:^",
"discord.js": "^14.16.0",
"socket.io-client": "^4.8.0"
},
"devDependencies": {
"@vitest/coverage-v8": "^2.0.0",
"tsx": "^4.0.0",
"typescript": "^5.8.0",
"vitest": "^2.0.0"

View File

@@ -0,0 +1,950 @@
import { EventEmitter } from 'node:events';
import type {
ChannelConversationRouteDto,
ChannelEgressDto,
ChannelIngressDto,
ChannelIngressPort,
} from '@mosaicstack/types';
import { afterEach, describe, expect, it, vi } from 'vitest';
import {
createDiscordIngressEnvelope,
DiscordPlugin,
parseDiscordInteractionBindings,
resolveDiscordInteractionActorId,
resolveDiscordInteractionBinding,
verifyDiscordIngressEnvelope,
type DiscordIngressEnvelope,
type DiscordInteractionRole,
type DiscordPluginConfig,
} from './index.js';
const SERVICE_TOKEN = 'test-service-token';
interface FakeDiscordMessageOptions {
id?: string;
guildId?: string;
content: string;
mentioned?: boolean;
userId?: string;
channelId?: string;
parentChannelId?: string | null;
isThread?: boolean;
hasThread?: boolean;
existingThreadId?: string;
fetchedThreadId?: string;
createdThreadId?: string;
attachments?: Map<string, FakeDiscordAttachment>;
}
interface FakeDiscordAttachment {
id: string;
name: string;
url: string;
contentType: string | null;
size?: number;
}
interface FakeDiscordMessage {
id: string;
guildId: string;
channelId: string;
author: { id: string; bot: boolean };
mentions: { has(user: { id: string }): boolean };
content: string;
createdAt: Date;
channel: {
parentId: string | null;
isThread(): boolean;
threads?: { fetch(id: string): Promise<{ id: string }> };
};
attachments: Map<string, FakeDiscordAttachment>;
hasThread: boolean;
thread: { id: string } | null;
startThread: ReturnType<typeof vi.fn>;
}
interface FakeGatewaySocket extends EventEmitter {
connected: boolean;
disconnect?: ReturnType<typeof vi.fn>;
}
interface FakeLifecycleClient extends EventEmitter {
user: { id: string; tag: string };
login: ReturnType<typeof vi.fn>;
destroy: ReturnType<typeof vi.fn>;
isReady(): boolean;
guilds: { cache: Map<string, object> };
channels: { cache: Map<string, object> };
}
interface DiscordPluginInternals {
client: {
user: { id: string };
isReady(): boolean;
channels?: {
cache: { get(id: string): { send(options: unknown): Promise<void> } | undefined };
};
};
socket: {
connected: boolean;
emit: ReturnType<typeof vi.fn>;
} | null;
conversationRoutes: Map<string, ChannelConversationRouteDto>;
handleDiscordMessage(message: FakeDiscordMessage): void | Promise<void>;
sendToDiscord(conversationId: string, text: string): Promise<void>;
}
function lifecyclePlugin(): {
plugin: DiscordPlugin;
socket: FakeGatewaySocket;
client: FakeLifecycleClient;
} {
const socket = Object.assign(new EventEmitter(), {
connected: false,
disconnect: vi.fn(),
});
const client = Object.assign(new EventEmitter(), {
user: { id: 'bot-001', tag: 'mosaic-bot' },
login: vi.fn().mockResolvedValue('token'),
destroy: vi.fn().mockResolvedValue(undefined),
isReady: (): boolean => true,
guilds: { cache: new Map<string, object>() },
channels: { cache: new Map<string, object>() },
});
const plugin = new DiscordPlugin(
{
token: 'unused',
gatewayUrl: 'http://gateway.invalid',
serviceToken: SERVICE_TOKEN,
allowedGuildIds: ['guild-001'],
allowedChannelIds: ['channel-001'],
allowedUserIds: ['user-001'],
interactionBindings: [
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: {
'user-001': { role: 'operator', mosaicUserId: 'mosaic-user-001' },
},
},
],
},
{
client: client as never,
socketFactory: vi.fn().mockReturnValue(socket) as never,
},
);
return { plugin, socket, client };
}
function createPlugin(
role: DiscordInteractionRole = 'operator',
paired = true,
ingressPort?: ChannelIngressPort,
configOverrides: Partial<DiscordPluginConfig> = {},
): {
plugin: DiscordPlugin;
internals: DiscordPluginInternals;
emit: ReturnType<typeof vi.fn>;
} {
const plugin = new DiscordPlugin(
{
token: 'unused',
gatewayUrl: 'http://unused',
serviceToken: SERVICE_TOKEN,
allowedGuildIds: ['guild-001'],
allowedChannelIds: ['channel-001'],
allowedUserIds: ['user-001'],
interactionBindings: [
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: paired
? {
'user-001': { role, mosaicUserId: 'mosaic-user-001' },
}
: {},
},
],
...configOverrides,
},
{ ingressPort },
);
const emit = vi.fn();
const internals = plugin as unknown as DiscordPluginInternals;
internals.client = { user: { id: 'bot-001' }, isReady: (): boolean => true };
internals.socket = { connected: true, emit };
return { plugin, internals, emit };
}
function fakeMessage(options: FakeDiscordMessageOptions): FakeDiscordMessage {
const channelId = options.channelId ?? 'channel-001';
const parentChannelId = options.parentChannelId;
const existingThreadId = options.existingThreadId;
return {
id: options.id ?? 'message-001',
guildId: options.guildId ?? 'guild-001',
channelId,
author: { id: options.userId ?? 'user-001', bot: false },
mentions: { has: (): boolean => options.mentioned ?? false },
content: options.content,
createdAt: new Date('2026-07-14T12:00:00.000Z'),
channel: {
parentId: parentChannelId ?? null,
isThread: (): boolean =>
options.isThread ?? (parentChannelId !== undefined && parentChannelId !== null),
...(options.fetchedThreadId
? {
threads: {
fetch: vi.fn().mockResolvedValue({ id: options.fetchedThreadId }),
},
}
: {}),
},
attachments: options.attachments ?? new Map<string, FakeDiscordAttachment>(),
hasThread: options.hasThread ?? existingThreadId !== undefined,
thread: existingThreadId ? { id: existingThreadId } : null,
startThread: vi.fn().mockResolvedValue({ id: options.createdThreadId ?? 'thread-created-001' }),
};
}
function emittedEnvelope(emit: ReturnType<typeof vi.fn>): DiscordIngressEnvelope {
const call = emit.mock.calls[0] as [string, DiscordIngressEnvelope] | undefined;
expect(call?.[0]).toBe('message');
expect(call?.[1]).toBeDefined();
return (
call?.[1] ??
createDiscordIngressEnvelope(
{
correlationId: 'unreachable',
messageId: 'unreachable',
guildId: 'unreachable',
channelId: 'unreachable',
userId: 'unreachable',
conversationId: 'unreachable',
content: 'unreachable',
},
SERVICE_TOKEN,
)
);
}
afterEach((): void => {
vi.useRealTimers();
vi.restoreAllMocks();
});
describe('Discord binding configuration', () => {
it('parses role-only and Mosaic-linked pairings', () => {
const bindings = parseDiscordInteractionBindings(
JSON.stringify([
{
instanceId: 'Nova',
agentConfigId: 'agent-config-nova',
guildId: 'guild-001',
channelId: 'channel-001',
pairedUsers: {
'legacy-user': 'operator',
'linked-user': { role: 'admin', mosaicUserId: 'mosaic-admin-001' },
'linked-without-id': { role: 'operator' },
},
},
]),
);
expect(bindings).toHaveLength(1);
expect(resolveDiscordInteractionActorId(bindings[0]!, 'legacy-user')).toBeNull();
expect(resolveDiscordInteractionActorId(bindings[0]!, 'linked-user')).toBe('mosaic-admin-001');
expect(resolveDiscordInteractionActorId(bindings[0]!, 'linked-without-id')).toBeNull();
expect(resolveDiscordInteractionActorId(bindings[0]!, 'missing-user')).toBeNull();
expect(
resolveDiscordInteractionBinding(bindings, 'guild-001', 'channel-001', 'linked-user', 'bind'),
).toBe(bindings[0]);
expect(
resolveDiscordInteractionBinding(
bindings,
'guild-001',
'channel-001',
'linked-without-id',
'attach',
),
).toBe(bindings[0]);
expect(
resolveDiscordInteractionBinding(
bindings,
'other-guild',
'channel-001',
'linked-user',
'send',
),
).toBeNull();
});
it.each([
['missing value', undefined],
['empty array', '[]'],
['non-object binding', '[null]'],
['missing binding fields', '[{"instanceId":"Nova"}]'],
[
'empty Discord user ID',
'[{"instanceId":"Nova","guildId":"g","channelId":"c","pairedUsers":{"":"operator"}}]',
],
[
'invalid role-only pairing',
'[{"instanceId":"Nova","guildId":"g","channelId":"c","pairedUsers":{"u":"owner"}}]',
],
[
'non-object pairing',
'[{"instanceId":"Nova","guildId":"g","channelId":"c","pairedUsers":{"u":42}}]',
],
[
'invalid linked pairing',
'[{"instanceId":"Nova","guildId":"g","channelId":"c","pairedUsers":{"u":{"role":"admin","mosaicUserId":" "}}}]',
],
])('rejects %s', (_case: string, value: string | undefined) => {
expect(() => parseDiscordInteractionBindings(value)).toThrow();
});
});
describe('Discord ingress integrity', () => {
it.each([
['guild', { guildIds: ['other'], channelIds: ['channel-001'], userIds: ['user-001'] }],
['channel', { guildIds: ['guild-001'], channelIds: ['other'], userIds: ['user-001'] }],
['user', { guildIds: ['guild-001'], channelIds: ['channel-001'], userIds: ['other'] }],
])('rejects an unallowlisted %s', (_field: string, allowlists) => {
const payload = {
correlationId: 'correlation-001',
messageId: 'message-001',
guildId: 'guild-001',
channelId: 'channel-001',
userId: 'user-001',
conversationId: 'Nova:discord:channel-001',
content: 'hello',
};
const envelope = createDiscordIngressEnvelope(payload, SERVICE_TOKEN);
expect(verifyDiscordIngressEnvelope(envelope, SERVICE_TOKEN, allowlists)).toBeNull();
});
});
describe('Discord adapter lifecycle', () => {
it('starts while the in-process gateway is still connecting, then reports connected', async () => {
const { plugin, socket, client } = lifecyclePlugin();
await plugin.start();
expect(client.login).toHaveBeenCalledWith('unused');
expect(await plugin.health()).toEqual({ status: 'degraded' });
socket.connected = true;
socket.emit('connect');
expect(await plugin.health()).toEqual({ status: 'connected' });
await plugin.stop();
expect(socket.disconnect).toHaveBeenCalledOnce();
expect(client.destroy).toHaveBeenCalledOnce();
});
it('keeps running while Socket.IO reconnects after an initial gateway error', async () => {
const { plugin, socket } = lifecyclePlugin();
const error = vi.spyOn(console, 'error').mockImplementation((): void => undefined);
await plugin.start();
socket.emit('connect_error', new Error('gateway not listening yet'));
expect(error).toHaveBeenCalledWith(
'[discord] Gateway connection error: gateway not listening yet',
);
expect(await plugin.health()).toEqual({ status: 'degraded' });
});
it('cleans up when Discord login fails', async () => {
const { plugin, socket, client } = lifecyclePlugin();
client.login.mockRejectedValueOnce(new Error('Discord authentication rejected'));
await expect(plugin.start()).rejects.toThrow('Discord authentication rejected');
expect(socket.disconnect).toHaveBeenCalledOnce();
expect(client.destroy).toHaveBeenCalledOnce();
});
});
describe('Discord project channel provisioning', () => {
it('returns null without a configured guild or visible guild', async () => {
const { plugin } = createPlugin();
await expect(
plugin.createProjectChannel({ id: 'project-1', name: 'Alpha' }),
).resolves.toBeNull();
const { client } = lifecyclePlugin();
const configured = new DiscordPlugin(
{
token: 'unused',
gatewayUrl: 'http://unused',
serviceToken: SERVICE_TOKEN,
guildId: 'missing-guild',
allowedGuildIds: ['guild-001'],
allowedChannelIds: ['channel-001'],
allowedUserIds: ['user-001'],
interactionBindings: [],
},
{ client: client as never },
);
await expect(
configured.createProjectChannel({ id: 'project-1', name: 'Alpha' }),
).resolves.toBeNull();
});
it('creates a normalized Discord project channel', async () => {
const create = vi.fn().mockResolvedValue({ id: 'created-channel-001' });
const { client } = lifecyclePlugin();
client.guilds.cache.set('guild-001', { channels: { create } });
const plugin = new DiscordPlugin(
{
token: 'unused',
gatewayUrl: 'http://unused',
serviceToken: SERVICE_TOKEN,
guildId: 'guild-001',
allowedGuildIds: ['guild-001'],
allowedChannelIds: ['channel-001'],
allowedUserIds: ['user-001'],
interactionBindings: [],
},
{ client: client as never },
);
await expect(
plugin.createProjectChannel({
id: 'project-1',
name: ' Project Alpha! ',
description: 'Alpha workspace',
}),
).resolves.toEqual({ channelId: 'created-channel-001' });
expect(create).toHaveBeenCalledWith(
expect.objectContaining({ name: 'mosaic-project-alpha', topic: 'Alpha workspace' }),
);
await plugin.createProjectChannel({ id: 'project-2', name: 'Beta' });
expect(create).toHaveBeenLastCalledWith(
expect.objectContaining({ topic: 'Mosaic project: Beta' }),
);
});
});
function egressFor(
route: ChannelConversationRouteDto,
content = 'agent response',
): ChannelEgressDto {
return {
correlationId: 'egress-correlation-001',
route,
message: {
id: 'egress-message-001',
channelName: 'discord',
channelId: route.responseTarget.channelId,
senderId: route.logicalAgentId,
senderKind: 'agent',
content,
contentKind: 'markdown',
timestamp: '2026-07-14T12:00:00.000Z',
metadata: {},
},
};
}
describe('official Discord channel routing', () => {
it('normalizes an authorized turn through the shared channel ingress port', async () => {
const receive = vi.fn<(ingress: ChannelIngressDto) => Promise<void>>().mockResolvedValue();
const { internals, emit } = createPlugin('operator', true, { receive });
internals.socket = null;
await internals.handleDiscordMessage(fakeMessage({ content: 'normalized turn' }));
expect(emit).not.toHaveBeenCalled();
expect(receive).toHaveBeenCalledWith(
expect.objectContaining({
operation: 'message.send',
principal: {
channelUserId: 'user-001',
role: 'operator',
mosaicUserId: 'mosaic-user-001',
},
message: expect.objectContaining({
channelName: 'discord',
channelId: 'channel-001',
content: 'normalized turn',
senderKind: 'user',
}),
route: expect.objectContaining({
logicalAgentId: 'Nova',
conversationId: 'Nova:discord:channel-001',
}),
}),
);
});
it('releases a response route when typed ingress rejects', async () => {
const receive = vi.fn().mockRejectedValue(new Error('gateway rejected ingress'));
const { internals } = createPlugin('operator', true, { receive });
await expect(
internals.handleDiscordMessage(fakeMessage({ content: 'rejected typed ingress' })),
).rejects.toThrow('gateway rejected ingress');
expect(internals.conversationRoutes).toHaveLength(0);
});
it('routes an authorized untagged parent-channel message and responds in that channel', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({ content: 'channel conversation' });
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
const payload = verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN);
expect(payload).toMatchObject({
channelId: 'channel-001',
conversationId: 'Nova:discord:channel-001',
content: 'channel conversation',
});
expect(payload?.threadId).toBeUndefined();
});
it('creates a thread for a mentioned parent-channel message and routes the response there', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: '<@bot-001> investigate this topic',
mentioned: true,
createdThreadId: 'thread-created-001',
});
await internals.handleDiscordMessage(message);
expect(message.startThread).toHaveBeenCalledOnce();
const payload = verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN);
expect(payload).toMatchObject({
channelId: 'channel-001',
conversationId: 'Nova:discord:thread-created-001',
content: 'investigate this topic',
threadId: 'thread-created-001',
});
});
it('fetches and reuses an existing thread that is missing from the cache', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: '<@bot-001> continue uncached topic',
mentioned: true,
hasThread: true,
fetchedThreadId: 'thread-fetched-001',
});
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN)).toMatchObject({
conversationId: 'Nova:discord:thread-fetched-001',
threadId: 'thread-fetched-001',
});
});
it('delivers the agent response to the thread selected by the mentioned turn', async () => {
const { internals } = createPlugin();
const send = vi.fn().mockResolvedValue(undefined);
internals.client = {
user: { id: 'bot-001' },
isReady: (): boolean => true,
channels: {
cache: {
get: (id: string): { send(options: unknown): Promise<void> } | undefined =>
id === 'thread-response-001' ? { send } : undefined,
},
},
};
await internals.handleDiscordMessage(
fakeMessage({
content: '<@bot-001> threaded response',
mentioned: true,
createdThreadId: 'thread-response-001',
}),
);
await internals.sendToDiscord('Nova:discord:thread-response-001', 'agent answer');
expect(send).toHaveBeenCalledWith(
expect.objectContaining({ content: 'agent answer', enforceNonce: true }),
);
});
it('reuses a thread already attached to a mentioned message instead of creating another', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: '<@bot-001> continue existing topic',
mentioned: true,
existingThreadId: 'thread-existing-001',
});
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN)).toMatchObject({
conversationId: 'Nova:discord:thread-existing-001',
threadId: 'thread-existing-001',
});
});
it('keeps an untagged follow-up inside an authorized thread without nesting threads', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: 'thread follow-up',
channelId: 'thread-001',
parentChannelId: 'channel-001',
});
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN)).toMatchObject({
channelId: 'channel-001',
conversationId: 'Nova:discord:thread-001',
content: 'thread follow-up',
threadId: 'thread-001',
});
});
it.each([
['guild', { guildId: 'guild-not-allowed' }],
['channel', { channelId: 'channel-not-allowed' }],
['user', { userId: 'user-not-allowed' }],
])(
'rejects an unauthorized %s before thread creation or gateway dispatch',
async (_boundary: string, override: Partial<FakeDiscordMessageOptions>) => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: '<@bot-001> unauthorized topic',
mentioned: true,
...override,
});
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(emit).not.toHaveBeenCalled();
},
);
it.each([
['parent channel', {}],
['existing thread', { channelId: 'thread-unpaired-001', parentChannelId: 'channel-001' }],
])(
'rejects an allowlisted but unpaired user in %s',
async (_location: string, override: Partial<FakeDiscordMessageOptions>) => {
const { internals, emit } = createPlugin('operator', false);
const message = fakeMessage({ content: 'unpaired message', ...override });
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(emit).not.toHaveBeenCalled();
},
);
it('rejects a paired viewer before thread creation or gateway dispatch', async () => {
const { internals, emit } = createPlugin('viewer');
const message = fakeMessage({
content: '<@bot-001> viewer cannot send',
mentioned: true,
});
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(emit).not.toHaveBeenCalled();
});
it('uses a normal channel ID rather than its category parent for authorization', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({
content: 'message from categorized channel',
parentChannelId: 'category-001',
isThread: false,
});
await internals.handleDiscordMessage(message);
expect(verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN)).toMatchObject({
channelId: 'channel-001',
conversationId: 'Nova:discord:channel-001',
});
});
it.each([
['parent channel', {}],
['existing thread', { channelId: 'thread-attachment-001', parentChannelId: 'channel-001' }],
])(
'preserves an attachment-only turn in an authorized %s',
async (_location: string, override: Partial<FakeDiscordMessageOptions>) => {
const { internals, emit } = createPlugin();
const attachments = new Map<string, FakeDiscordAttachment>([
[
'attachment-001',
{
id: 'attachment-001',
name: 'diagram.png',
url: 'https://cdn.example.invalid/diagram.png',
contentType: 'image/png',
size: 4_096,
},
],
]);
await internals.handleDiscordMessage(fakeMessage({ content: '', attachments, ...override }));
expect(verifyDiscordIngressEnvelope(emittedEnvelope(emit), SERVICE_TOKEN)).toMatchObject({
content: '',
attachments: [
{
id: 'attachment-001',
name: 'diagram.png',
contentType: 'image/png',
sizeBytes: 4_096,
},
],
});
},
);
it('does not dispatch when Discord cannot create the requested thread', async () => {
const { internals, emit } = createPlugin();
const message = fakeMessage({ content: '<@bot-001> new topic', mentioned: true });
message.startThread.mockRejectedValueOnce(new Error('missing thread permission'));
await expect(internals.handleDiscordMessage(message)).rejects.toThrow(
'missing thread permission',
);
expect(emit).not.toHaveBeenCalled();
});
it.each(['/approve', '/stop approval-001'])(
'rejects operator use of privileged command %s before gateway dispatch',
async (command: string) => {
const { internals, emit } = createPlugin('operator');
const message = fakeMessage({ content: `<@bot-001> ${command}`, mentioned: true });
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(emit).not.toHaveBeenCalled();
},
);
it('keeps runtime control commands on the current durable session', async () => {
const { internals, emit } = createPlugin('admin');
const message = fakeMessage({ content: '<@bot-001> /approve', mentioned: true });
await internals.handleDiscordMessage(message);
expect(message.startThread).not.toHaveBeenCalled();
expect(emit).toHaveBeenCalledWith('discord:approve', expect.any(Object));
});
it('keeps the stable conversation address independent of a runtime harness', async () => {
const first = createPlugin();
const second = createPlugin();
await first.internals.handleDiscordMessage(
fakeMessage({ id: 'message-claude', content: 'before runtime handoff' }),
);
await second.internals.handleDiscordMessage(
fakeMessage({ id: 'message-pi', content: 'after runtime handoff' }),
);
const firstPayload = verifyDiscordIngressEnvelope(emittedEnvelope(first.emit), SERVICE_TOKEN);
const secondPayload = verifyDiscordIngressEnvelope(emittedEnvelope(second.emit), SERVICE_TOKEN);
expect(firstPayload?.conversationId).toBe('Nova:discord:channel-001');
expect(secondPayload?.conversationId).toBe(firstPayload?.conversationId);
});
it('rate-limits authorized turns before thread creation or dispatch', async () => {
const { internals, emit } = createPlugin('operator', true, undefined, {
messageRateLimitPerMinute: 1,
threadRateLimitPerMinute: 1,
});
const error = vi.spyOn(console, 'error').mockImplementation((): void => undefined);
const first = fakeMessage({ id: 'rate-first', content: 'first turn' });
const second = fakeMessage({
id: 'rate-second',
content: '<@bot-001> second turn',
mentioned: true,
});
await internals.handleDiscordMessage(first);
await internals.handleDiscordMessage(second);
expect(emit).toHaveBeenCalledOnce();
expect(second.startThread).not.toHaveBeenCalled();
expect(error).toHaveBeenCalledWith(expect.stringContaining('Message rate limit reached'));
});
it('applies a stricter mention-thread rate limit before Discord side effects', async () => {
const { internals, emit } = createPlugin('operator', true, undefined, {
messageRateLimitPerMinute: 10,
threadRateLimitPerMinute: 1,
});
vi.spyOn(console, 'error').mockImplementation((): void => undefined);
const first = fakeMessage({ id: 'thread-rate-first', content: 'first topic', mentioned: true });
const second = fakeMessage({
id: 'thread-rate-second',
content: 'second topic',
mentioned: true,
});
await internals.handleDiscordMessage(first);
await internals.handleDiscordMessage(second);
expect(first.startThread).toHaveBeenCalledOnce();
expect(second.startThread).not.toHaveBeenCalled();
expect(emit).toHaveBeenCalledOnce();
});
it('rejects unconfigured egress and handles missing or failed Discord destinations', async () => {
const { plugin, internals } = createPlugin();
const route: ChannelConversationRouteDto = {
bindingId: 'guild-001:channel-001:Nova',
logicalAgentId: 'Nova',
conversationId: 'Nova:discord:channel-001',
channelName: 'discord',
authorizationChannelId: 'channel-001',
responseTarget: { channelId: 'channel-001' },
};
await expect(
plugin.send(
egressFor({
...route,
bindingId: 'forged-binding',
}),
),
).rejects.toMatchObject({ code: 'invalid_route' });
await expect(
plugin.send(
egressFor({
...route,
conversationId: 'Nova:discord:unrelated-conversation',
}),
),
).rejects.toMatchObject({ code: 'invalid_route' });
await expect(
plugin.send({
...egressFor(route),
message: { ...egressFor(route).message, channelId: 'other-channel' },
}),
).rejects.toMatchObject({ code: 'invalid_route' });
const forgedSend = vi.fn().mockResolvedValue(undefined);
internals.client = {
user: { id: 'bot-001' },
isReady: (): boolean => true,
channels: {
cache: {
get: (): { send(options: unknown): Promise<void> } => ({ send: forgedSend }),
},
},
};
await expect(
plugin.send(
egressFor({
...route,
conversationId: 'Nova:discord:forged-channel',
responseTarget: { channelId: 'forged-channel', threadId: 'forged-channel' },
}),
),
).rejects.toMatchObject({ code: 'invalid_route' });
expect(forgedSend).not.toHaveBeenCalled();
internals.client = {
user: { id: 'bot-001' },
isReady: (): boolean => true,
channels: { cache: { get: (): undefined => undefined } },
};
await expect(plugin.send(egressFor(route))).rejects.toMatchObject({
code: 'destination_unavailable',
});
await expect(
internals.sendToDiscord('missing-conversation', 'missing route'),
).rejects.toMatchObject({ code: 'invalid_route' });
vi.useFakeTimers();
const send = vi
.fn()
.mockRejectedValue(Object.assign(new Error('Discord unavailable'), { status: 503 }));
internals.client.channels = {
cache: { get: (): { send(options: unknown): Promise<void> } => ({ send }) },
};
const delivery = plugin.send(egressFor(route));
const rejection = expect(delivery).rejects.toMatchObject({
code: 'delivery_failed',
retryable: true,
});
await vi.runAllTimersAsync();
await rejection;
expect(send).toHaveBeenCalledTimes(3);
expect(new Set(send.mock.calls.map(([options]) => options.nonce)).size).toBe(1);
expect(send).toHaveBeenCalledWith(
expect.objectContaining({ enforceNonce: true, content: 'agent response' }),
);
const permanentSend = vi
.fn()
.mockRejectedValue(Object.assign(new Error('Discord forbidden'), { status: 403 }));
internals.client.channels = {
cache: {
get: (): { send(options: unknown): Promise<void> } => ({ send: permanentSend }),
},
};
await expect(plugin.send(egressFor(route))).rejects.toMatchObject({
code: 'delivery_failed',
retryable: false,
});
expect(permanentSend).toHaveBeenCalledOnce();
});
it('chunks long egress responses at Discord-safe boundaries', async () => {
const { plugin, internals } = createPlugin();
const send = vi.fn().mockResolvedValue(undefined);
internals.client = {
user: { id: 'bot-001' },
isReady: (): boolean => true,
channels: {
cache: { get: (): { send(options: unknown): Promise<void> } => ({ send }) },
},
};
const route: ChannelConversationRouteDto = {
bindingId: 'guild-001:channel-001:Nova',
logicalAgentId: 'Nova',
conversationId: 'Nova:discord:channel-001',
channelName: 'discord',
authorizationChannelId: 'channel-001',
responseTarget: { channelId: 'channel-001' },
};
await plugin.send(egressFor(route, `${'a'.repeat(1_500)}\n${'b'.repeat(1_500)}`));
expect(send).toHaveBeenCalledTimes(2);
});
it('reports channel adapter health without exposing runtime-provider state', async () => {
const { plugin, internals } = createPlugin();
expect(await plugin.health()).toEqual({ status: 'connected' });
internals.socket = { connected: false, emit: vi.fn() };
expect(await plugin.health()).toEqual({ status: 'degraded' });
internals.client = { user: { id: 'bot-001' }, isReady: (): boolean => false };
expect(await plugin.health()).toEqual({ status: 'disconnected' });
internals.socket = { connected: true, emit: vi.fn() };
expect(await plugin.health()).toEqual({ status: 'degraded' });
});
});

View File

@@ -1,10 +1,38 @@
import { createHmac, randomUUID, timingSafeEqual } from 'node:crypto';
import { ChannelType, Client, GatewayIntentBits, type Message as DiscordMessage } from 'discord.js';
import { createHash, createHmac, randomUUID, timingSafeEqual } from 'node:crypto';
import { ChannelDeliveryError } from '@mosaicstack/types';
import type {
ChannelAdapterHealthDto,
ChannelAuthorizedPrincipalDto,
ChannelConversationRouteDto,
ChannelEgressDto,
ChannelEgressPort,
ChannelIngressDto,
ChannelIngressPort,
ChannelMessageDto,
OfficialChannelAdapter,
} from '@mosaicstack/types';
import {
ChannelType,
Client,
GatewayIntentBits,
ThreadAutoArchiveDuration,
type Message as DiscordMessage,
} from 'discord.js';
import { io, type Socket } from 'socket.io-client';
export interface DiscordPluginDependencies {
ingressPort?: ChannelIngressPort;
client?: Client;
socketFactory?: (url: string, options: Parameters<typeof io>[1]) => Socket;
}
export interface DiscordPluginConfig {
token: string;
gatewayUrl: string;
/** Maximum authorized turns per Discord user/channel per minute. */
messageRateLimitPerMinute?: number;
/** Maximum mention-triggered thread routes per Discord user/channel per minute. */
threadRateLimitPerMinute?: number;
/** Shared service credential injected by the approved secret mechanism. */
serviceToken: string;
/** Which guild to bind to (single-guild only for v0.1.0). */
@@ -30,13 +58,23 @@ export interface DiscordInteractionUserBinding {
export type DiscordInteractionPairing = DiscordInteractionRole | DiscordInteractionUserBinding;
export interface DiscordInteractionBinding {
/** Stable logical agent identity used in channel conversation routes. */
instanceId: string;
/** Trusted gateway database agent-config ID selected for this binding. */
agentConfigId: string;
guildId: string;
channelId: string;
/** Pairing roster keyed by Discord user ID. */
pairedUsers: Readonly<Record<string, DiscordInteractionPairing>>;
}
const DEFAULT_MESSAGE_RATE_LIMIT_PER_MINUTE = 30;
const DEFAULT_THREAD_RATE_LIMIT_PER_MINUTE = 5;
const RATE_LIMIT_WINDOW_MS = 60_000;
const DELIVERY_MAX_ATTEMPTS = 3;
const DELIVERY_RETRY_BASE_MS = 50;
const MAX_CONVERSATION_ROUTES = 1_000;
const operationRoles: Readonly<
Record<DiscordInteractionOperation, readonly DiscordInteractionRole[]>
> = {
@@ -90,6 +128,7 @@ export function parseDiscordInteractionBindings(
const candidate = binding as Partial<DiscordInteractionBinding>;
if (
!candidate.instanceId ||
!candidate.agentConfigId ||
!candidate.guildId ||
!candidate.channelId ||
!candidate.pairedUsers ||
@@ -130,6 +169,7 @@ export function parseDiscordInteractionBindings(
) as Record<string, DiscordInteractionPairing>;
return {
instanceId: candidate.instanceId,
agentConfigId: candidate.agentConfigId,
guildId: candidate.guildId,
channelId: candidate.channelId,
pairedUsers,
@@ -154,6 +194,7 @@ export interface DiscordAttachment {
name: string;
url: string;
contentType: string | null;
sizeBytes?: number;
}
export interface DiscordIngressEnvelope {
@@ -229,27 +270,37 @@ export function verifyDiscordIngressEnvelope(
return envelope.payload;
}
export class DiscordPlugin {
export class DiscordPlugin implements OfficialChannelAdapter, ChannelEgressPort {
readonly name = 'discord';
private client: Client;
private socket: Socket | null = null;
/** Map Discord channel ID → Mosaic conversation ID. */
private channelConversations = new Map<string, string>();
/** Bounded last-authorized routes for response-target egress validation. */
private conversationRoutes = new Map<string, ChannelConversationRouteDto>();
/** Track in-flight responses to avoid duplicate streaming. */
private pendingResponses = new Map<string, string>();
private readonly messageRateWindows = new Map<string, number[]>();
private readonly threadRateWindows = new Map<string, number[]>();
constructor(private readonly config: DiscordPluginConfig) {
this.client = new Client({
intents: [
GatewayIntentBits.Guilds,
GatewayIntentBits.GuildMessages,
GatewayIntentBits.MessageContent,
GatewayIntentBits.DirectMessages,
],
});
constructor(
private readonly config: DiscordPluginConfig,
private readonly dependencies: DiscordPluginDependencies = {},
) {
this.client =
dependencies.client ??
new Client({
intents: [
GatewayIntentBits.Guilds,
GatewayIntentBits.GuildMessages,
GatewayIntentBits.MessageContent,
GatewayIntentBits.DirectMessages,
],
});
}
async start(): Promise<void> {
this.socket = io(`${this.config.gatewayUrl}/chat`, {
const socketFactory = this.dependencies.socketFactory ?? io;
this.socket = socketFactory(`${this.config.gatewayUrl}/chat`, {
auth: { discordServiceToken: this.config.serviceToken },
transports: ['websocket'],
});
@@ -276,11 +327,13 @@ export class DiscordPlugin {
this.socket.on('agent:end', (data: { conversationId: string }) => {
const text = this.pendingResponses.get(data.conversationId);
this.pendingResponses.delete(data.conversationId);
if (text) {
this.pendingResponses.delete(data.conversationId);
this.sendToDiscord(data.conversationId, text).catch((err: unknown) => {
this.sendAgentResponse(data.conversationId, text).catch((err: unknown) => {
console.error(`[discord] Error sending response for ${data.conversationId}:`, err);
});
} else {
this.conversationRoutes.delete(data.conversationId);
}
});
@@ -288,15 +341,33 @@ export class DiscordPlugin {
this.pendingResponses.set(data.conversationId, '');
});
this.client.on('messageCreate', (message: DiscordMessage) =>
this.handleDiscordMessage(message),
);
this.socket.on('error', (data: { conversationId?: unknown }) => {
if (typeof data.conversationId !== 'string') return;
this.pendingResponses.delete(data.conversationId);
this.conversationRoutes.delete(data.conversationId);
});
this.client.on('messageCreate', (message: DiscordMessage) => {
void Promise.resolve(this.handleDiscordMessage(message)).catch((error: unknown): void => {
const errorName = error instanceof Error ? error.name : 'UnknownError';
console.error(
`[discord] Message routing failed. channel=${message.channelId} message=${message.id} error=${errorName}`,
);
});
});
this.client.on('ready', () => {
console.log(`[discord] Bot logged in as ${this.client.user?.tag}`);
});
await this.client.login(this.config.token);
try {
await this.client.login(this.config.token);
} catch (error: unknown) {
this.socket.disconnect();
this.socket = null;
await this.client.destroy();
throw error;
}
}
async stop(): Promise<void> {
@@ -304,6 +375,14 @@ export class DiscordPlugin {
await this.client.destroy();
}
async health(): Promise<ChannelAdapterHealthDto> {
const discordReady = this.client.isReady();
const gatewayConnected = this.socket?.connected === true;
if (discordReady && gatewayConnected) return { status: 'connected' };
if (discordReady || gatewayConnected) return { status: 'degraded' };
return { status: 'disconnected' };
}
async createProjectChannel(project: {
id: string;
name: string;
@@ -325,77 +404,224 @@ export class DiscordPlugin {
topic: project.description ?? `Mosaic project: ${project.name}`,
});
this.channelConversations.set(channel.id, `discord-${channel.id}`);
// A project channel has no logical-agent conversation until a configured
// binding authorizes a message. Do not seed a legacy channel-only key.
return { channelId: channel.id };
}
private handleDiscordMessage(message: DiscordMessage): void {
if (message.author.bot || !this.client.user) return;
if (!message.guildId || !this.isAllowedMessage(message)) return;
private handleDiscordMessage(message: DiscordMessage): void | Promise<void> {
if (message.author.bot || !this.client.user || !message.guildId) return;
const authorizationChannelId = this.authorizationChannelId(message);
if (!this.isAllowedMessage(message, authorizationChannelId)) return;
const isMention = message.mentions.has(this.client.user);
if (!isMention) return;
const content = isMention
? message.content.replace(new RegExp(`<@!?${this.client.user.id}>`, 'g'), '').trim()
: message.content.trim();
if (!content && message.attachments.size === 0) return;
const operation = this.interactionOperation(content);
const binding = resolveDiscordInteractionBinding(
this.config.interactionBindings ?? [],
message.guildId,
authorizationChannelId,
message.author.id,
operation,
);
// Pairing and operation-specific role checks happen before thread creation
// or any gateway dispatch, including privileged control events.
if (!binding) return;
const content = message.content
.replace(new RegExp(`<@!?${this.client.user.id}>`, 'g'), '')
.trim();
if (!content) return;
if (!this.socket?.connected) {
const rateKey = `${message.guildId}:${authorizationChannelId}:${message.author.id}`;
if (
!this.consumeRateLimit(
this.messageRateWindows,
rateKey,
this.config.messageRateLimitPerMinute ?? DEFAULT_MESSAGE_RATE_LIMIT_PER_MINUTE,
)
) {
console.error(
`[discord] Message rate limit reached. guild=${message.guildId} channel=${authorizationChannelId} user=${message.author.id}`,
);
return;
}
const createThread = isMention && operation === 'send';
if (
createThread &&
!this.consumeRateLimit(
this.threadRateWindows,
rateKey,
this.config.threadRateLimitPerMinute ?? DEFAULT_THREAD_RATE_LIMIT_PER_MINUTE,
)
) {
console.error(
`[discord] Thread rate limit reached. guild=${message.guildId} channel=${authorizationChannelId} user=${message.author.id}`,
);
return;
}
if (!this.dependencies.ingressPort && !this.socket?.connected) {
console.error(
`[discord] Cannot forward message: not connected to gateway. channel=${message.channelId} message=${message.id}`,
);
return;
}
const channelId = message.channelId;
const parentChannelId = 'parentId' in message.channel ? message.channel.parentId : null;
const bindingChannelId = parentChannelId ?? channelId;
const binding = resolveDiscordInteractionBinding(
this.config.interactionBindings ?? [],
message.guildId,
bindingChannelId,
message.author.id,
'send',
// Approval/stop commands act on the current durable session. They remain
// at the current channel/thread target rather than creating a new topic.
const route = this.resolveConversationRoute(
message,
binding,
authorizationChannelId,
createThread,
);
if (!binding) return;
const conversationId =
this.channelConversations.get(channelId) ?? `${binding.instanceId}:discord:${channelId}`;
this.channelConversations.set(channelId, conversationId);
if (route instanceof Promise) {
return route.then((resolved: ChannelConversationRouteDto): void | Promise<void> =>
this.dispatchDiscordIngress(message, content, operation, binding, resolved),
);
}
return this.dispatchDiscordIngress(message, content, operation, binding, route);
}
private dispatchDiscordIngress(
message: DiscordMessage,
content: string,
operation: 'send' | 'approve' | 'stop',
binding: DiscordInteractionBinding,
route: ChannelConversationRouteDto,
): void | Promise<void> {
const guildId = message.guildId;
if (!guildId) return;
if (operation === 'send') this.rememberConversationRoute(route);
const correlationId = randomUUID();
const ingress: ChannelIngressDto = {
correlationId,
nativeMessageId: message.id,
operation:
operation === 'approve'
? 'approval.create'
: operation === 'stop'
? 'session.stop'
: 'message.send',
principal: this.authorizedPrincipal(binding, message.author.id),
message: this.channelMessage(message, content, route),
route,
};
if (this.dependencies.ingressPort) {
return this.dependencies.ingressPort.receive(ingress).catch((error: unknown) => {
if (operation === 'send') this.conversationRoutes.delete(route.conversationId);
throw error;
});
}
const socket = this.socket;
if (!socket?.connected) {
console.error(
`[discord] Cannot dispatch routed message: gateway disconnected. channel=${message.channelId} message=${message.id}`,
);
return;
}
this.emitDiscordIngress(socket, ingress);
}
private authorizedPrincipal(
binding: DiscordInteractionBinding,
channelUserId: string,
): ChannelAuthorizedPrincipalDto {
const pairing = binding.pairedUsers[channelUserId];
if (!pairing) throw new Error('Authorized Discord pairing is unavailable');
if (typeof pairing === 'string') return { channelUserId, role: pairing };
return {
channelUserId,
role: pairing.role,
...(pairing.mosaicUserId ? { mosaicUserId: pairing.mosaicUserId } : {}),
};
}
private channelMessage(
message: DiscordMessage,
content: string,
route: ChannelConversationRouteDto,
): ChannelMessageDto {
const attachments = Array.from(message.attachments.values()).map((attachment) => ({
id: attachment.id,
name: attachment.name,
url: attachment.url,
mimeType: attachment.contentType,
sizeBytes: attachment.size,
}));
const firstContentType = attachments[0]?.mimeType;
return {
id: randomUUID(),
channelName: this.name,
channelId: route.responseTarget.channelId,
senderId: message.author.id,
senderKind: 'user',
content,
contentKind:
content.length > 0 ? 'markdown' : firstContentType?.startsWith('image/') ? 'image' : 'file',
timestamp:
message.createdAt instanceof Date
? message.createdAt.toISOString()
: new Date().toISOString(),
...(route.responseTarget.threadId ? { threadId: route.responseTarget.threadId } : {}),
...(attachments.length > 0 ? { attachments } : {}),
metadata: {
channelMessageId: message.id,
guildId: message.guildId ?? '',
},
};
}
private emitDiscordIngress(socket: Socket, ingress: ChannelIngressDto): void {
const envelope = createDiscordIngressEnvelope(
{
correlationId: randomUUID(),
messageId: message.id,
guildId: message.guildId,
channelId: bindingChannelId,
userId: message.author.id,
conversationId,
content,
threadId: parentChannelId ? channelId : undefined,
attachments: Array.from(message.attachments.values()).map((attachment) => ({
correlationId: ingress.correlationId,
messageId: ingress.nativeMessageId,
guildId: String(ingress.message.metadata['guildId'] ?? ''),
channelId: ingress.route.authorizationChannelId,
userId: ingress.principal.channelUserId,
conversationId: ingress.route.conversationId,
content: ingress.message.content,
...(ingress.route.responseTarget.threadId
? { threadId: ingress.route.responseTarget.threadId }
: {}),
attachments: ingress.message.attachments?.map((attachment) => ({
id: attachment.id,
name: attachment.name,
url: attachment.url,
contentType: attachment.contentType,
contentType: attachment.mimeType,
...(attachment.sizeBytes !== undefined ? { sizeBytes: attachment.sizeBytes } : {}),
})),
},
this.config.serviceToken,
);
this.socket.emit(
/^\/approve$/i.test(content)
socket.emit(
ingress.operation === 'approval.create'
? 'discord:approve'
: content.startsWith('/stop ')
: ingress.operation === 'session.stop'
? 'discord:stop'
: 'message',
envelope,
);
}
private isAllowedMessage(message: DiscordMessage): boolean {
private authorizationChannelId(message: DiscordMessage): string {
// A normal guild channel can itself have a category parent. Only Discord
// threads inherit authorization from a configured parent text channel.
const channel = message.channel as DiscordMessage['channel'] & {
isThread?: () => boolean;
parentId?: string | null;
};
const isThread =
typeof channel.isThread === 'function'
? channel.isThread()
: channel.parentId !== undefined && channel.parentId !== null;
return isThread && channel.parentId ? channel.parentId : message.channelId;
}
private isAllowedMessage(message: DiscordMessage, authorizationChannelId: string): boolean {
const guildId = message.guildId;
const parentChannelId = 'parentId' in message.channel ? message.channel.parentId : null;
// Threads inherit their authorization boundary from their configured parent.
const authorizationChannelId = parentChannelId ?? message.channelId;
return (
guildId !== null &&
includesId(this.config.allowedGuildIds, guildId) &&
@@ -404,31 +630,282 @@ export class DiscordPlugin {
);
}
private async sendToDiscord(conversationId: string, text: string): Promise<void> {
const channelId = Array.from(this.channelConversations.entries()).find(
([, convId]) => convId === conversationId,
)?.[0];
if (!channelId) {
console.error(`[discord] No channel found for conversation ${conversationId}`);
return;
private resolveConversationRoute(
message: DiscordMessage,
binding: DiscordInteractionBinding,
authorizationChannelId: string,
createThread: boolean,
): ChannelConversationRouteDto | Promise<ChannelConversationRouteDto> {
if (authorizationChannelId !== message.channelId) {
return this.createConversationRoute(
binding,
authorizationChannelId,
message.channelId,
message.channelId,
);
}
if (!createThread) {
return this.createConversationRoute(binding, authorizationChannelId, message.channelId);
}
if (message.hasThread) {
const cachedThread = message.thread;
if (cachedThread) {
return this.createConversationRoute(
binding,
authorizationChannelId,
cachedThread.id,
cachedThread.id,
);
}
if (!('threads' in message.channel)) {
return Promise.reject(new Error('Existing Discord thread manager is unavailable'));
}
return message.channel.threads
.fetch(message.id)
.then((thread): ChannelConversationRouteDto => {
if (!thread) throw new Error('Existing Discord thread is unavailable');
return this.createConversationRoute(
binding,
authorizationChannelId,
thread.id,
thread.id,
);
});
}
return message
.startThread({
name: `Mosaic conversation ${message.id.slice(-8)}`,
autoArchiveDuration: ThreadAutoArchiveDuration.OneHour,
reason: 'Authorized Mosaic mention',
})
.then(
(thread): ChannelConversationRouteDto =>
this.createConversationRoute(binding, authorizationChannelId, thread.id, thread.id),
);
}
private createConversationRoute(
binding: DiscordInteractionBinding,
authorizationChannelId: string,
responseChannelId: string,
threadId?: string,
): ChannelConversationRouteDto {
// Recompute from configuration on every turn so a stale in-memory map can
// never carry a channel-only or differently bound agent identity forward.
const conversationId = `${binding.instanceId}:discord:${responseChannelId}`;
return {
bindingId: `${binding.guildId}:${binding.channelId}:${binding.instanceId}`,
logicalAgentId: binding.instanceId,
conversationId,
channelName: this.name,
authorizationChannelId,
responseTarget: {
channelId: responseChannelId,
...(threadId ? { threadId } : {}),
},
};
}
private consumeRateLimit(
windows: Map<string, number[]>,
key: string,
limit: number,
now = Date.now(),
): boolean {
const active = (windows.get(key) ?? []).filter(
(timestamp: number): boolean => now - timestamp < RATE_LIMIT_WINDOW_MS,
);
if (active.length >= Math.max(1, limit)) {
windows.set(key, active);
return false;
}
active.push(now);
windows.set(key, active);
return true;
}
private interactionOperation(content: string): 'send' | 'approve' | 'stop' {
if (/^\/approve$/i.test(content)) return 'approve';
if (/^\/stop\s+\S+/i.test(content)) return 'stop';
return 'send';
}
async send(egress: ChannelEgressDto): Promise<void> {
if (!this.isConfiguredRoute(egress.route) || !this.isMessageAlignedWithRoute(egress)) {
throw new ChannelDeliveryError(
'invalid_route',
`Discord egress route is not authorized for conversation ${egress.route.conversationId}`,
);
}
const channelId = egress.route.responseTarget.channelId;
const channel = this.client.channels.cache.get(channelId);
if (!channel || !('send' in channel)) {
console.error(
`[discord] Channel ${channelId} not sendable for conversation ${conversationId}`,
throw new ChannelDeliveryError(
'destination_unavailable',
`Discord destination is unavailable for conversation ${egress.route.conversationId}`,
);
return;
}
for (const chunk of this.chunkText(text, 1900)) {
const chunks = this.chunkText(egress.message.content, 1900);
for (const [chunkIndex, chunk] of chunks.entries()) {
await this.sendChunkWithRetry(
channel as {
send(options: { content: string; nonce: string; enforceNonce: true }): Promise<unknown>;
},
chunk,
egress.correlationId,
chunkIndex,
egress.route.conversationId,
);
}
}
private async sendChunkWithRetry(
channel: {
send(options: { content: string; nonce: string; enforceNonce: true }): Promise<unknown>;
},
chunk: string,
correlationId: string,
chunkIndex: number,
conversationId: string,
): Promise<void> {
const nonce = createHash('sha256')
.update(`${correlationId}:${chunkIndex}`)
.digest('hex')
.slice(0, 25);
let lastError: unknown;
for (let attempt = 1; attempt <= DELIVERY_MAX_ATTEMPTS; attempt += 1) {
try {
await (channel as { send: (content: string) => Promise<unknown> }).send(chunk);
} catch (err: unknown) {
console.error(`[discord] Failed to send message to channel ${channelId}:`, err);
await channel.send({ content: chunk, nonce, enforceNonce: true });
return;
} catch (error: unknown) {
lastError = error;
const retryable = this.isTransientDeliveryError(error);
if (!retryable) {
throw new ChannelDeliveryError(
'delivery_failed',
`Discord delivery failed for conversation ${conversationId}`,
false,
{ cause: error },
);
}
if (attempt < DELIVERY_MAX_ATTEMPTS) {
await new Promise<void>((resolve): void => {
setTimeout(resolve, DELIVERY_RETRY_BASE_MS * 2 ** (attempt - 1));
});
}
}
}
throw new ChannelDeliveryError(
'delivery_failed',
`Discord delivery failed for conversation ${conversationId}`,
true,
{ cause: lastError },
);
}
private isTransientDeliveryError(error: unknown): boolean {
if (typeof error !== 'object' || error === null) return false;
const candidate = error as { status?: unknown; code?: unknown };
if (
typeof candidate.status === 'number' &&
(candidate.status === 429 || candidate.status >= 500)
) {
return true;
}
return (
typeof candidate.code === 'string' &&
['ECONNRESET', 'ETIMEDOUT', 'EAI_AGAIN', 'UND_ERR_CONNECT_TIMEOUT'].includes(candidate.code)
);
}
private async sendAgentResponse(conversationId: string, text: string): Promise<void> {
const route = this.conversationRoutes.get(conversationId);
if (!route) {
throw new ChannelDeliveryError(
'invalid_route',
`Discord response route is unavailable for conversation ${conversationId}`,
);
}
try {
await this.send({
correlationId: randomUUID(),
route,
message: {
id: randomUUID(),
channelName: this.name,
channelId: route.responseTarget.channelId,
senderId: route.logicalAgentId,
senderKind: 'agent',
content: text,
contentKind: 'markdown',
timestamp: new Date().toISOString(),
...(route.responseTarget.threadId ? { threadId: route.responseTarget.threadId } : {}),
metadata: {},
},
});
} finally {
this.conversationRoutes.delete(conversationId);
}
}
/** Compatibility wrapper while Socket.IO agent events carry only conversation ID. */
private async sendToDiscord(conversationId: string, text: string): Promise<void> {
await this.sendAgentResponse(conversationId, text);
}
private rememberConversationRoute(route: ChannelConversationRouteDto): void {
if (
!this.conversationRoutes.has(route.conversationId) &&
this.conversationRoutes.size >= MAX_CONVERSATION_ROUTES
) {
throw new ChannelDeliveryError(
'delivery_failed',
'Discord has reached its active response-route limit',
true,
);
}
this.conversationRoutes.set(route.conversationId, route);
}
private isConfiguredRoute(route: ChannelConversationRouteDto): boolean {
if (
route.channelName !== this.name ||
route.conversationId !==
`${route.logicalAgentId}:${this.name}:${route.responseTarget.channelId}`
) {
return false;
}
const bindingMatches = (this.config.interactionBindings ?? []).some(
(binding): boolean =>
route.bindingId === `${binding.guildId}:${binding.channelId}:${binding.instanceId}` &&
route.logicalAgentId === binding.instanceId &&
route.authorizationChannelId === binding.channelId,
);
if (!bindingMatches) return false;
if (
route.responseTarget.channelId === route.authorizationChannelId &&
route.responseTarget.threadId === undefined
) {
return true;
}
const observed = this.conversationRoutes.get(route.conversationId);
return (
observed?.bindingId === route.bindingId &&
observed.logicalAgentId === route.logicalAgentId &&
observed.authorizationChannelId === route.authorizationChannelId &&
observed.responseTarget.channelId === route.responseTarget.channelId &&
observed.responseTarget.threadId === route.responseTarget.threadId
);
}
private isMessageAlignedWithRoute(egress: ChannelEgressDto): boolean {
return (
egress.message.channelName === this.name &&
egress.message.channelId === egress.route.responseTarget.channelId &&
egress.message.threadId === egress.route.responseTarget.threadId
);
}
private chunkText(text: string, maxLength: number): string[] {

View File

@@ -4,5 +4,15 @@ export default defineConfig({
test: {
globals: true,
environment: 'node',
coverage: {
provider: 'v8',
reporter: ['text', 'json', 'html'],
thresholds: {
lines: 85,
functions: 85,
branches: 85,
statements: 85,
},
},
},
});

41
pnpm-lock.yaml generated
View File

@@ -732,6 +732,9 @@ importers:
plugins/discord:
dependencies:
'@mosaicstack/types':
specifier: workspace:^
version: link:../../packages/types
discord.js:
specifier: ^14.16.0
version: 14.25.1
@@ -739,6 +742,9 @@ importers:
specifier: ^4.8.0
version: 4.8.3
devDependencies:
'@vitest/coverage-v8':
specifier: ^2.0.0
version: 2.1.9(vitest@2.1.9(@types/node@24.12.0)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1))
tsx:
specifier: ^4.0.0
version: 4.21.0
@@ -1169,11 +1175,11 @@ packages:
'@esbuild-kit/core-utils@3.3.2':
resolution: {integrity: sha512-sPRAnw9CdSsRmEtnsl2WXWdyquogVpB3yZ3dgwJfe8zrOzTsV7cJvmwrKVa+0ma5BoiGJ+BoqkMvawbayKUsqQ==}
deprecated: 'Merged into tsx: https://tsx.is'
deprecated: 'Merged into tsx: https://tsx.hirok.io'
'@esbuild-kit/esm-loader@2.6.5':
resolution: {integrity: sha512-FxEMIkJKnodyA1OaCUoEvbYRkoZlLZ4d/eXFu9Fh8CbBBgP5EmZxrfTRyN0qpXZ4vOvqnE5YdRdcrmUUXuU+dA==}
deprecated: 'Merged into tsx: https://tsx.is'
deprecated: 'Merged into tsx: https://tsx.hirok.io'
'@esbuild/aix-ppc64@0.21.5':
resolution: {integrity: sha512-1SDgH6ZSPTlggy1yI6+Dbkiz8xzpHJEVAlF/AM1tHPLsf5STom9rwtjE4hKAF20FfXXNTFqEYXyJNWh1GiZedQ==}
@@ -2160,47 +2166,57 @@ packages:
'@mariozechner/pi-agent-core@0.63.1':
resolution: {integrity: sha512-h0B20xfs/iEVR2EC4gwiE8hKI1TPeB8REdRJMgV+uXKH7gpeIZ9+s8Dp9nX35ZR0QUjkNey2+ULk2DxQtdg14Q==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-agent-core instead going forward
'@mariozechner/pi-agent-core@0.63.2':
resolution: {integrity: sha512-9QTS7ylcmoAIWXk0EVpwCCop3fK4NIqTAN8TiRuXvuKYx+wYmUJc+P5+RfehIZhwsy7g9O/rktz0c1YEUBFB0g==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-agent-core instead going forward
'@mariozechner/pi-agent-core@0.65.0':
resolution: {integrity: sha512-QCDqkgxvCkizCgJOl0aFekT1gURppznzuBIGXS8dXWZMour/xX6YF7chxX56mZ0p0DXkILM1ixf5jXYBfDsP5w==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-agent-core instead going forward
'@mariozechner/pi-ai@0.63.1':
resolution: {integrity: sha512-wjgwY+yfrFO6a9QdAfjWpH7iSrDean6GsKDDMohNcLCy6PreMxHOZvNM0NwJARL1tZoZovr7ikAQfLGFZbnjsw==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-ai instead going forward
hasBin: true
'@mariozechner/pi-ai@0.63.2':
resolution: {integrity: sha512-EJNPyzeZeifTJmkD8PPYQmSO4P4h8kFCrhUqU4NvFUkug+GNYr954KlxhYnXH0f77MpdIEpf/O5zdDrYJQyafA==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-ai instead going forward
hasBin: true
'@mariozechner/pi-ai@0.65.0':
resolution: {integrity: sha512-MsCsCHlHIlBYbg6jB2PJBeCNKbjzVZge7ddBNUJN2gsFY8sdjFh482+GB+r5Ou6k9Fnhi3nO779YDymo5+t89w==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-ai instead going forward
hasBin: true
'@mariozechner/pi-coding-agent@0.63.1':
resolution: {integrity: sha512-XSoMyLtuMA7ePK1UBWqSJ/BBdtBdJUHY9nbtnNyG6GeW7Gbgd+iqljIuwmAUf8wlYL981UIfYM/WIPQ6t/dIxw==}
engines: {node: '>=20.6.0'}
deprecated: please use @earendil-works/pi-coding-agent instead going forward
hasBin: true
'@mariozechner/pi-coding-agent@0.65.0':
resolution: {integrity: sha512-IEBZ74n17w8NxnG/X2ixErsSYcvLm/h5WKALNbPgPWJZqvafNtJ0GcrCfLCS6RVIq2o+O/a2QwsbSI6bgJ6W/A==}
engines: {node: '>=20.6.0'}
deprecated: please use @earendil-works/pi-coding-agent instead going forward
hasBin: true
'@mariozechner/pi-tui@0.63.1':
resolution: {integrity: sha512-G5p+eh1EPkFCNaaggX6vRrqttnDscK6npgmEOknoCQXZtch8XNgh9Lf3VJ0A2lZXSgR7IntG5dfXHPH/Ki64wA==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-tui instead going forward
'@mariozechner/pi-tui@0.65.0':
resolution: {integrity: sha512-P5Uuf4x1sTplMNQw8NrC1Hyz0N/tZq9kC6CDRkTT7rZuxZEeXl9uhKvlLEGigdKVOVrWnPE7ip0jrO81POYy3g==}
engines: {node: '>=20.0.0'}
deprecated: please use @earendil-works/pi-tui instead going forward
'@matrix-org/matrix-sdk-crypto-nodejs@0.4.0':
resolution: {integrity: sha512-+qqgpn39XFSbsD0dFjssGO9vHEP7sTyfs8yTpt8vuqWpUpF20QMwpCZi0jpYw7GxjErNTsMshopuo8677DfGEA==}
@@ -3915,6 +3931,7 @@ packages:
'@ungap/structured-clone@1.3.0':
resolution: {integrity: sha512-WmoN8qaIAo7WTYWbAZuG8PYEhn5fkz7dZrqTBZ7dtt//lL2Gwms1IcnQ5yHqjDfX8Ft5j4YzDM23f87zBfDe9g==}
deprecated: Potential CWE-502 - Update to 1.3.1 or higher
'@vitest/coverage-v8@2.1.9':
resolution: {integrity: sha512-Z2cOr0ksM00MpEfyVE8KXIYPEcBFxdbLSs56L8PO0QQMxt/6bDj45uQfxoc96v05KW3clk7vvgP0qfDit9DmfQ==}
@@ -4095,6 +4112,7 @@ packages:
basic-ftp@5.2.0:
resolution: {integrity: sha512-VoMINM2rqJwJgfdHq6RiUudKt2BV+FY5ZFezP/ypmwayk68+NzzAQy4XXLlqsGD4MCzq3DrmNFD/uUmBJuGoXw==}
engines: {node: '>=10.0.0'}
deprecated: Security vulnerability fixed in 5.2.1, please upgrade
better-auth@1.5.5:
resolution: {integrity: sha512-GpVPaV1eqr3mOovKfghJXXk6QvlcVeFbS3z+n+FPDid5rK/2PchnDtiaVCzWyXA9jH2KkirOfl+JhAUvnja0Eg==}
@@ -7087,6 +7105,7 @@ packages:
uuid@9.0.1:
resolution: {integrity: sha512-b+1eJOlsR9K8HJpow9Ok3fiWOWSIcIzXodvv0rQjVoOVNpWMpxf1wZNpt4y9h10odCNrqnYp1OBzRktckBe3sA==}
deprecated: uuid@10 and below is no longer supported. For ESM codebases, update to uuid@latest. For CommonJS codebases, use uuid@11 (but be aware this version will likely be deprecated in 2028).
hasBin: true
validator@13.15.26:
@@ -10977,6 +10996,24 @@ snapshots:
transitivePeerDependencies:
- supports-color
'@vitest/coverage-v8@2.1.9(vitest@2.1.9(@types/node@24.12.0)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1))':
dependencies:
'@ampproject/remapping': 2.3.0
'@bcoe/v8-coverage': 0.2.3
debug: 4.4.3
istanbul-lib-coverage: 3.2.2
istanbul-lib-report: 3.0.1
istanbul-lib-source-maps: 5.0.6
istanbul-reports: 3.2.0
magic-string: 0.30.21
magicast: 0.3.5
std-env: 3.10.0
test-exclude: 7.0.2
tinyrainbow: 1.2.0
vitest: 2.1.9(@types/node@24.12.0)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1)
transitivePeerDependencies:
- supports-color
'@vitest/expect@2.1.9':
dependencies:
'@vitest/spy': 2.1.9