fix: enforce Tess session ownership scope #715

Merged
jason.woltje merged 1 commits from fix/tess-session-ownership into main 2026-07-12 22:14:18 +00:00
13 changed files with 750 additions and 145 deletions

View File

@@ -20,7 +20,7 @@ export class AdminHealthController {
async check(): Promise<HealthStatusDto> {
const [database, cache] = await Promise.all([this.checkDatabase(), this.checkCache()]);
const sessions = this.agentService.listSessions();
const sessions = this.agentService.listAllSessionsForSystem();
const providers = this.providerService.listProviders();
const allOk = database.status === 'ok' && cache.status === 'ok';

View File

@@ -0,0 +1,142 @@
import { ForbiddenException } from '@nestjs/common';
import { describe, expect, it, vi } from 'vitest';
import { AgentService, type AgentSession } from '../agent.service.js';
import type { ActorTenantScope } from '../../auth/session-scope.js';
const CONVERSATION_ID = '22222222-2222-4222-8222-222222222222';
const OWNER_SCOPE: ActorTenantScope = { userId: 'owner-user', tenantId: 'owner-tenant' };
const FOREIGN_SCOPE: ActorTenantScope = { userId: 'foreign-user', tenantId: 'foreign-tenant' };
type AgentServiceInternals = {
sessions: Map<string, AgentSession>;
creating: Map<string, Promise<AgentSession>>;
};
function makeService(): AgentService {
return new AgentService(
{} as never,
{} as never,
{} as never,
{ available: false } as never,
{} as never,
{} as never,
{} as never,
null,
null,
{ collect: vi.fn().mockResolvedValue(undefined) } as never,
);
}
function internals(service: AgentService): AgentServiceInternals {
return service as unknown as AgentServiceInternals;
}
function makeSession(scope: ActorTenantScope = OWNER_SCOPE): AgentSession {
return {
id: CONVERSATION_ID,
provider: 'test-provider',
modelId: 'test-model',
piSession: {
thinkingLevel: 'off',
getAvailableThinkingLevels: vi.fn().mockReturnValue(['off', 'low', 'high']),
setThinkingLevel: vi.fn(),
abort: vi.fn().mockResolvedValue(undefined),
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
getSessionStats: vi.fn(),
getContextUsage: vi.fn(),
} as unknown as AgentSession['piSession'],
listeners: new Set(),
unsubscribe: vi.fn(),
createdAt: Date.now(),
promptCount: 0,
channels: new Set(),
skillPromptAdditions: [],
sandboxDir: '/tmp/tess-session-ownership-test',
allowedTools: null,
userId: scope.userId,
tenantId: scope.tenantId,
metrics: {
tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
modelSwitches: 0,
messageCount: 0,
lastActivityAt: new Date('2026-07-12T00:00:00Z').toISOString(),
},
};
}
describe('AgentService owner/tenant scope enforcement', () => {
it('allows owner-scoped operations and rejects foreign scopes for seeded sessions', async () => {
const service = makeService();
const session = makeSession();
internals(service).sessions.set(CONVERSATION_ID, session);
expect(service.getSession(CONVERSATION_ID, OWNER_SCOPE)).toBe(session);
expect(service.getSession(CONVERSATION_ID, FOREIGN_SCOPE)).toBeUndefined();
expect(service.getSessionInfo(CONVERSATION_ID, FOREIGN_SCOPE)).toBeUndefined();
expect(service.listSessions(OWNER_SCOPE)).toHaveLength(1);
expect(service.listSessions(FOREIGN_SCOPE)).toEqual([]);
service.addChannel(CONVERSATION_ID, 'websocket:owner', OWNER_SCOPE);
expect(session.channels.has('websocket:owner')).toBe(true);
expect(() => service.addChannel(CONVERSATION_ID, 'websocket:foreign', FOREIGN_SCOPE)).toThrow(
ForbiddenException,
);
expect(() => service.removeChannel(CONVERSATION_ID, 'websocket:owner', FOREIGN_SCOPE)).toThrow(
ForbiddenException,
);
expect(() =>
service.updateSessionModel(CONVERSATION_ID, 'foreign-model', FOREIGN_SCOPE),
).toThrow(ForbiddenException);
service.updateSessionModel(CONVERSATION_ID, 'owner-model', OWNER_SCOPE);
expect(session.modelId).toBe('owner-model');
expect(() =>
service.applyAgentConfig(CONVERSATION_ID, 'agent-foreign', 'Foreign Agent', FOREIGN_SCOPE),
).toThrow(ForbiddenException);
service.applyAgentConfig(CONVERSATION_ID, 'agent-owner', 'Owner Agent', OWNER_SCOPE);
expect(session.agentConfigId).toBe('agent-owner');
expect(() => service.onEvent(CONVERSATION_ID, vi.fn(), FOREIGN_SCOPE)).toThrow(
ForbiddenException,
);
const cleanup = service.onEvent(CONVERSATION_ID, vi.fn(), OWNER_SCOPE);
cleanup();
await expect(
service.prompt(CONVERSATION_ID, 'foreign prompt', FOREIGN_SCOPE),
).rejects.toBeInstanceOf(ForbiddenException);
await service.prompt(CONVERSATION_ID, 'owner prompt', OWNER_SCOPE);
expect(session.piSession.prompt).toHaveBeenCalledWith('owner prompt');
await expect(service.destroySession(CONVERSATION_ID, FOREIGN_SCOPE)).rejects.toBeInstanceOf(
ForbiddenException,
);
expect(internals(service).sessions.has(CONVERSATION_ID)).toBe(true);
await service.destroySession(CONVERSATION_ID, OWNER_SCOPE);
expect(session.piSession.dispose).toHaveBeenCalled();
expect(internals(service).sessions.has(CONVERSATION_ID)).toBe(false);
});
it('checks owner/tenant scope before returning an in-flight session creation', async () => {
const service = makeService();
const session = makeSession();
internals(service).creating.set(CONVERSATION_ID, Promise.resolve(session));
await expect(
service.createSession(CONVERSATION_ID, {
userId: FOREIGN_SCOPE.userId,
tenantId: FOREIGN_SCOPE.tenantId,
}),
).rejects.toBeInstanceOf(ForbiddenException);
await expect(
service.createSession(CONVERSATION_ID, {
userId: OWNER_SCOPE.userId,
tenantId: OWNER_SCOPE.tenantId,
}),
).resolves.toBe(session);
});
});

View File

@@ -0,0 +1,262 @@
import { readFileSync } from 'node:fs';
import { resolve } from 'node:path';
import { ForbiddenException, NotFoundException } from '@nestjs/common';
import { describe, expect, it, vi } from 'vitest';
vi.mock('../agent.service.js', () => ({ AgentService: class AgentService {} }));
vi.mock('../../commands/command-executor.service.js', () => ({
CommandExecutorService: class CommandExecutorService {},
}));
vi.mock('../routing/routing-engine.service.js', () => ({
RoutingEngineService: class RoutingEngineService {},
}));
import { SessionsController } from '../sessions.controller.js';
import { ChatController } from '../../chat/chat.controller.js';
import { ChatGateway } from '../../chat/chat.gateway.js';
import type { AgentSession } from '../agent.service.js';
import type { SessionInfoDto } from '../session.dto.js';
const USER_A = { id: 'user-a', tenantId: 'tenant-a' };
const USER_B = { id: 'user-b', tenantId: 'tenant-b' };
const CONVERSATION_ID = '11111111-1111-4111-8111-111111111111';
function makeSessionInfo(overrides?: Partial<SessionInfoDto>): SessionInfoDto {
return {
id: CONVERSATION_ID,
provider: 'test-provider',
modelId: 'test-model',
createdAt: new Date('2026-07-12T00:00:00Z').toISOString(),
promptCount: 0,
channels: [],
durationMs: 0,
metrics: {
tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
modelSwitches: 0,
messageCount: 0,
lastActivityAt: new Date('2026-07-12T00:00:00Z').toISOString(),
},
...overrides,
};
}
function makeAgentSession(owner = USER_A): AgentSession {
return {
id: CONVERSATION_ID,
provider: 'test-provider',
modelId: 'test-model',
piSession: {
thinkingLevel: 'off',
getAvailableThinkingLevels: vi.fn().mockReturnValue(['off', 'low', 'high']),
setThinkingLevel: vi.fn(),
abort: vi.fn().mockResolvedValue(undefined),
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
getSessionStats: vi.fn(),
getContextUsage: vi.fn(),
} as unknown as AgentSession['piSession'],
listeners: new Set(),
unsubscribe: vi.fn(),
createdAt: Date.now(),
promptCount: 0,
channels: new Set(),
skillPromptAdditions: [],
sandboxDir: '/tmp',
allowedTools: null,
userId: owner.id,
tenantId: owner.tenantId,
metrics: {
tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
modelSwitches: 0,
messageCount: 0,
lastActivityAt: new Date('2026-07-12T00:00:00Z').toISOString(),
},
};
}
function makeScopedAgentService() {
const foreign = makeAgentSession(USER_A);
return {
listSessions: vi.fn((scope?: { userId: string; tenantId?: string }) =>
scope?.userId === USER_B.id ? [] : [makeSessionInfo({ id: foreign.id })],
),
getSessionInfo: vi.fn((_id: string, scope?: { userId: string; tenantId?: string }) =>
scope?.userId === USER_B.id ? undefined : makeSessionInfo({ id: foreign.id }),
),
destroySession: vi.fn(),
getSession: vi.fn((_id: string, scope?: { userId: string; tenantId?: string }) =>
scope?.userId === USER_B.id ? undefined : foreign,
),
createSession: vi.fn().mockRejectedValue(new ForbiddenException('Session scope mismatch')),
onEvent: vi.fn(() => vi.fn()),
addChannel: vi.fn(),
removeChannel: vi.fn(),
recordMessage: vi.fn(),
prompt: vi.fn().mockResolvedValue(undefined),
};
}
describe('TESS-M1-SEC-002 AgentService ownership boundary', () => {
it('requires explicit owner+tenant scope on protected session operations', () => {
const source = readFileSync(resolve('src/agent/agent.service.ts'), 'utf8');
expect(source).toContain('getSession(sessionId: string, scope: ActorTenantScope)');
expect(source).toContain('listSessions(scope: ActorTenantScope)');
expect(source).toContain('getSessionInfo(sessionId: string, scope: ActorTenantScope)');
expect(source).toContain(
'addChannel(sessionId: string, channel: string, scope: ActorTenantScope)',
);
expect(source).toContain(
'removeChannel(sessionId: string, channel: string, scope: ActorTenantScope)',
);
expect(source).toContain(
'async prompt(sessionId: string, message: string, scope: ActorTenantScope)',
);
expect(source).toContain('scope: ActorTenantScope,');
expect(source).toContain('async destroySession(sessionId: string, scope: ActorTenantScope)');
expect(source).not.toContain('scope?: ActorTenantScope');
});
});
describe('TESS-M1-SEC-002 REST session ownership and tenant binding', () => {
it('lists only sessions owned by the authenticated owner+tenant scope', () => {
const agentService = makeScopedAgentService();
const controller = new SessionsController(agentService as never);
expect(controller.list(USER_B)).toEqual({ sessions: [], total: 0 });
expect(agentService.listSessions).toHaveBeenCalledWith({
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
});
it('does not reveal another owner/tenant session by guessed id', () => {
const agentService = makeScopedAgentService();
const controller = new SessionsController(agentService as never);
expect(() => controller.findOne(CONVERSATION_ID, USER_B)).toThrow(NotFoundException);
expect(agentService.getSessionInfo).toHaveBeenCalledWith(CONVERSATION_ID, {
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
});
it('does not terminate another owner/tenant session by guessed id', async () => {
const agentService = makeScopedAgentService();
const controller = new SessionsController(agentService as never);
await expect(controller.destroy(CONVERSATION_ID, USER_B)).rejects.toBeInstanceOf(
NotFoundException,
);
expect(agentService.destroySession).not.toHaveBeenCalled();
});
});
describe('TESS-M1-SEC-002 REST chat send ownership and tenant binding', () => {
it('does not send a prompt into another owner/tenant session by guessed conversationId', async () => {
const agentService = makeScopedAgentService();
const controller = new ChatController(agentService as never);
await expect(
controller.chat({ conversationId: CONVERSATION_ID, content: 'take over' }, USER_B),
).rejects.toMatchObject({ status: 404 });
expect(agentService.getSession).toHaveBeenCalledWith(CONVERSATION_ID, {
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
expect(agentService.prompt).not.toHaveBeenCalled();
});
});
describe('TESS-M1-SEC-002 WebSocket session ownership and tenant binding', () => {
function makeGateway(agentService = makeScopedAgentService()) {
const brain = {
conversations: {
findById: vi.fn().mockResolvedValue(undefined),
create: vi.fn().mockResolvedValue(undefined),
update: vi.fn().mockResolvedValue(undefined),
findMessages: vi.fn().mockResolvedValue([]),
addMessage: vi.fn().mockResolvedValue(undefined),
},
};
const commandRegistry = { getManifest: vi.fn().mockReturnValue([]) };
const commandExecutor = { execute: vi.fn() };
const routingEngine = {
resolve: vi.fn().mockResolvedValue({ provider: 'test', model: 'test-model' }),
};
const gateway = new ChatGateway(
agentService as never,
{} as never,
brain as never,
commandRegistry as never,
commandExecutor as never,
routingEngine as never,
);
return { gateway, agentService };
}
function makeSocket() {
return {
id: 'socket-b',
connected: true,
data: { user: USER_B, session: { id: 'auth-session-b', userId: USER_B.id } },
emit: vi.fn(),
disconnect: vi.fn(),
};
}
it('does not attach or send to another owner/tenant session by guessed conversationId', async () => {
const { gateway, agentService } = makeGateway();
const socket = makeSocket();
await gateway.handleMessage(socket as never, {
conversationId: CONVERSATION_ID,
content: 'attach to foreign session',
});
expect(agentService.getSession).toHaveBeenCalledWith(CONVERSATION_ID, {
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
expect(agentService.onEvent).not.toHaveBeenCalled();
expect(agentService.addChannel).not.toHaveBeenCalled();
expect(agentService.prompt).not.toHaveBeenCalled();
expect(socket.emit).toHaveBeenCalledWith(
'error',
expect.objectContaining({ conversationId: CONVERSATION_ID }),
);
});
it('does not mutate thinking level on another owner/tenant session', () => {
const { gateway, agentService } = makeGateway();
const socket = makeSocket();
gateway.handleSetThinking(socket as never, { conversationId: CONVERSATION_ID, level: 'high' });
expect(agentService.getSession).toHaveBeenCalledWith(CONVERSATION_ID, {
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
expect(socket.emit).toHaveBeenCalledWith(
'error',
expect.objectContaining({ conversationId: CONVERSATION_ID }),
);
});
it('does not terminate another owner/tenant session over WebSocket abort', async () => {
const { gateway, agentService } = makeGateway();
const socket = makeSocket();
await gateway.handleAbort(socket as never, { conversationId: CONVERSATION_ID });
expect(agentService.getSession).toHaveBeenCalledWith(CONVERSATION_ID, {
userId: USER_B.id,
tenantId: USER_B.tenantId,
});
expect(socket.emit).toHaveBeenCalledWith(
'error',
expect.objectContaining({ conversationId: CONVERSATION_ID }),
);
});
});

View File

@@ -1,4 +1,11 @@
import { Inject, Injectable, Logger, Optional, type OnModuleDestroy } from '@nestjs/common';
import {
ForbiddenException,
Inject,
Injectable,
Logger,
Optional,
type OnModuleDestroy,
} from '@nestjs/common';
import {
createAgentSession,
DefaultResourceLoader,
@@ -28,6 +35,7 @@ import type { SessionInfoDto, SessionMetrics } from './session.dto.js';
import { SystemOverrideService } from '../preferences/system-override.service.js';
import { PreferencesService } from '../preferences/preferences.service.js';
import { SessionGCService } from '../gc/session-gc.service.js';
import type { ActorTenantScope } from '../auth/session-scope.js';
/** A single message from DB conversation history, used for context injection. */
export interface ConversationHistoryMessage {
@@ -68,6 +76,8 @@ export interface AgentSessionOptions {
agentConfigId?: string;
/** ID of the user who owns this session. Used for preferences and system override lookups. */
userId?: string;
/** Server-derived tenant scope that owns this session. Falls back to userId for solo users. */
tenantId?: string;
/**
* Prior conversation messages to inject as context when resuming a session.
* These messages are formatted and prepended to the system prompt so the
@@ -94,6 +104,8 @@ export interface AgentSession {
allowedTools: string[] | null;
/** User ID that owns this session, used for preference lookups. */
userId?: string;
/** Server-derived tenant scope that owns this session. Falls back to userId for solo users. */
tenantId?: string;
/** Agent config ID applied to this session, if any (M5-001). */
agentConfigId?: string;
/** Human-readable agent name applied to this session, if any (M5-001). */
@@ -174,12 +186,20 @@ export class AgentService implements OnModuleDestroy {
.filter((t) => t.length > 0);
}
async createSession(sessionId: string, options?: AgentSessionOptions): Promise<AgentSession> {
async createSession(sessionId: string, options: AgentSessionOptions): Promise<AgentSession> {
const scope = this.scopeFromOptions(options);
const existing = this.sessions.get(sessionId);
if (existing) return existing;
if (existing) {
this.assertSessionScope(existing, scope);
return existing;
}
const inflight = this.creating.get(sessionId);
if (inflight) return inflight;
if (inflight) {
const session = await inflight;
this.assertSessionScope(session, scope);
return session;
}
const promise = this.doCreateSession(sessionId, options).finally(() => {
this.creating.delete(sessionId);
@@ -342,6 +362,7 @@ export class AgentService implements OnModuleDestroy {
sandboxDir,
allowedTools,
userId: mergedOptions?.userId,
tenantId: this.tenantIdFor(mergedOptions?.userId, mergedOptions?.tenantId),
agentConfigId: mergedOptions?.agentConfigId,
agentName: resolvedAgentName,
metrics: {
@@ -473,38 +494,70 @@ export class AgentService implements OnModuleDestroy {
return this.providerService.getDefaultModel() ?? null;
}
getSession(sessionId: string): AgentSession | undefined {
return this.sessions.get(sessionId);
getSession(sessionId: string, scope: ActorTenantScope): AgentSession | undefined {
const session = this.sessions.get(sessionId);
if (!session || !this.sessionMatchesScope(session, scope)) return undefined;
return session;
}
listSessions(): SessionInfoDto[] {
listSessions(scope: ActorTenantScope): SessionInfoDto[] {
const now = Date.now();
return Array.from(this.sessions.values()).map((s) => ({
id: s.id,
provider: s.provider,
modelId: s.modelId,
...(s.agentName ? { agentName: s.agentName } : {}),
createdAt: new Date(s.createdAt).toISOString(),
promptCount: s.promptCount,
channels: Array.from(s.channels),
durationMs: now - s.createdAt,
metrics: { ...s.metrics },
}));
return Array.from(this.sessions.values())
.filter((s) => this.sessionMatchesScope(s, scope))
.map((s) => this.toSessionInfo(s, now));
}
getSessionInfo(sessionId: string): SessionInfoDto | undefined {
listAllSessionsForSystem(): SessionInfoDto[] {
const now = Date.now();
return Array.from(this.sessions.values()).map((s) => this.toSessionInfo(s, now));
}
getSessionInfo(sessionId: string, scope: ActorTenantScope): SessionInfoDto | undefined {
const s = this.sessions.get(sessionId);
if (!s) return undefined;
if (!s || !this.sessionMatchesScope(s, scope)) return undefined;
return this.toSessionInfo(s);
}
private scopeFromOptions(options: AgentSessionOptions): ActorTenantScope {
if (!options.userId) {
throw new ForbiddenException('Session owner scope is required');
}
return {
id: s.id,
provider: s.provider,
modelId: s.modelId,
...(s.agentName ? { agentName: s.agentName } : {}),
createdAt: new Date(s.createdAt).toISOString(),
promptCount: s.promptCount,
channels: Array.from(s.channels),
durationMs: Date.now() - s.createdAt,
metrics: { ...s.metrics },
userId: options.userId,
tenantId: this.tenantIdFor(options.userId, options.tenantId) ?? options.userId,
};
}
private tenantIdFor(
userId: string | undefined,
tenantId: string | undefined,
): string | undefined {
return tenantId ?? userId;
}
private sessionMatchesScope(session: AgentSession, scope: ActorTenantScope): boolean {
return (
session.userId === scope.userId && (session.tenantId ?? session.userId) === scope.tenantId
);
}
private assertSessionScope(session: AgentSession, scope: ActorTenantScope): void {
if (!this.sessionMatchesScope(session, scope)) {
throw new ForbiddenException('Session does not belong to the current owner/tenant scope');
}
}
private toSessionInfo(session: AgentSession, now = Date.now()): SessionInfoDto {
return {
id: session.id,
provider: session.provider,
modelId: session.modelId,
...(session.agentName ? { agentName: session.agentName } : {}),
createdAt: new Date(session.createdAt).toISOString(),
promptCount: session.promptCount,
channels: Array.from(session.channels),
durationMs: now - session.createdAt,
metrics: { ...session.metrics },
};
}
@@ -553,9 +606,10 @@ export class AgentService implements OnModuleDestroy {
* not reconstructed — the model is used on the next createSession call for
* the same conversationId when the session is torn down or a new one is created.
*/
updateSessionModel(sessionId: string, modelId: string): void {
updateSessionModel(sessionId: string, modelId: string, scope: ActorTenantScope): void {
const session = this.sessions.get(sessionId);
if (!session) return;
this.assertSessionScope(session, scope);
const prev = session.modelId;
session.modelId = modelId;
this.recordModelSwitch(sessionId);
@@ -572,48 +626,51 @@ export class AgentService implements OnModuleDestroy {
sessionId: string,
agentConfigId: string,
agentName: string,
scope: ActorTenantScope,
modelId?: string,
): void {
const session = this.sessions.get(sessionId);
if (!session) return;
this.assertSessionScope(session, scope);
session.agentConfigId = agentConfigId;
session.agentName = agentName;
if (modelId) {
this.updateSessionModel(sessionId, modelId);
this.updateSessionModel(sessionId, modelId, scope);
}
this.logger.log(
`Session ${sessionId}: agent switched to "${agentName}" (${agentConfigId}) (M5-003)`,
);
}
addChannel(sessionId: string, channel: string): void {
addChannel(sessionId: string, channel: string, scope: ActorTenantScope): void {
const session = this.sessions.get(sessionId);
if (session) {
if (!session) return;
this.assertSessionScope(session, scope);
session.channels.add(channel);
}
}
removeChannel(sessionId: string, channel: string): void {
removeChannel(sessionId: string, channel: string, scope: ActorTenantScope): void {
const session = this.sessions.get(sessionId);
if (session) {
if (!session) return;
this.assertSessionScope(session, scope);
session.channels.delete(channel);
}
}
async prompt(sessionId: string, message: string): Promise<void> {
async prompt(sessionId: string, message: string, scope: ActorTenantScope): Promise<void> {
const session = this.sessions.get(sessionId);
if (!session) {
throw new Error(`No agent session found: ${sessionId}`);
}
this.assertSessionScope(session, scope);
session.promptCount += 1;
// Prepend session-scoped system override if present (renew TTL on each turn)
let effectiveMessage = message;
if (this.systemOverride) {
const override = await this.systemOverride.get(sessionId);
const override = await this.systemOverride.get(sessionId, scope);
if (override) {
effectiveMessage = `[System Override]\n${override}\n\n${message}`;
await this.systemOverride.renew(sessionId);
await this.systemOverride.renew(sessionId, scope);
this.logger.debug(`Applied system override for session ${sessionId}`);
}
}
@@ -629,16 +686,28 @@ export class AgentService implements OnModuleDestroy {
}
}
onEvent(sessionId: string, listener: (event: AgentSessionEvent) => void): () => void {
onEvent(
sessionId: string,
listener: (event: AgentSessionEvent) => void,
scope: ActorTenantScope,
): () => void {
const session = this.sessions.get(sessionId);
if (!session) {
throw new Error(`No agent session found: ${sessionId}`);
}
this.assertSessionScope(session, scope);
session.listeners.add(listener);
return () => session.listeners.delete(listener);
}
async destroySession(sessionId: string): Promise<void> {
async destroySession(sessionId: string, scope: ActorTenantScope): Promise<void> {
const session = this.sessions.get(sessionId);
if (!session) return;
this.assertSessionScope(session, scope);
await this.destroySessionForSystem(sessionId);
}
private async destroySessionForSystem(sessionId: string): Promise<void> {
const session = this.sessions.get(sessionId);
if (!session) return;
this.logger.log(`Destroying agent session ${sessionId}`);
@@ -667,7 +736,7 @@ export class AgentService implements OnModuleDestroy {
async onModuleDestroy(): Promise<void> {
this.logger.log('Shutting down all agent sessions');
const stops = Array.from(this.sessions.keys()).map((id) => this.destroySession(id));
const stops = Array.from(this.sessions.keys()).map((id) => this.destroySessionForSystem(id));
const results = await Promise.allSettled(stops);
for (const result of results) {
if (result.status === 'rejected') {

View File

@@ -10,6 +10,8 @@ import {
UseGuards,
} from '@nestjs/common';
import { AuthGuard } from '../auth/auth.guard.js';
import { CurrentUser } from '../auth/current-user.decorator.js';
import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js';
import { AgentService } from './agent.service.js';
@Controller('api/sessions')
@@ -18,23 +20,24 @@ export class SessionsController {
constructor(@Inject(AgentService) private readonly agentService: AgentService) {}
@Get()
list() {
const sessions = this.agentService.listSessions();
list(@CurrentUser() user: AuthenticatedUserLike) {
const sessions = this.agentService.listSessions(scopeFromUser(user));
return { sessions, total: sessions.length };
}
@Get(':id')
findOne(@Param('id') id: string) {
const info = this.agentService.getSessionInfo(id);
findOne(@Param('id') id: string, @CurrentUser() user: AuthenticatedUserLike) {
const info = this.agentService.getSessionInfo(id, scopeFromUser(user));
if (!info) throw new NotFoundException('Session not found');
return info;
}
@Delete(':id')
@HttpCode(HttpStatus.NO_CONTENT)
async destroy(@Param('id') id: string) {
const info = this.agentService.getSessionInfo(id);
async destroy(@Param('id') id: string, @CurrentUser() user: AuthenticatedUserLike) {
const scope = scopeFromUser(user);
const info = this.agentService.getSessionInfo(id, scope);
if (!info) throw new NotFoundException('Session not found');
await this.agentService.destroySession(id);
await this.agentService.destroySession(id, scope);
}
}

View File

@@ -0,0 +1,24 @@
export interface AuthenticatedUserLike {
id: string;
tenantId?: string | null;
teamId?: string | null;
organizationId?: string | null;
orgId?: string | null;
}
export interface ActorTenantScope {
userId: string;
tenantId: string;
}
/**
* Build the immutable server-derived scope used for Tess session operations.
* Current Mosaic auth is user-scoped; future org/team claims can populate one
* of the tenant fields without allowing clients to choose another tenant.
*/
export function scopeFromUser(user: AuthenticatedUserLike): ActorTenantScope {
return {
userId: user.id,
tenantId: user.tenantId ?? user.teamId ?? user.organizationId ?? user.orgId ?? user.id,
};
}

View File

@@ -12,7 +12,8 @@ describe('Chat controller source hardening', () => {
const source = readFileSync(resolve('src/chat/chat.controller.ts'), 'utf8');
expect(source).toContain('@UseGuards(AuthGuard)');
expect(source).toContain('@CurrentUser() user: { id: string }');
expect(source).toContain('@CurrentUser() user: AuthenticatedUserLike');
expect(source).toContain('const scope = scopeFromUser(user);');
});
});

View File

@@ -3,8 +3,10 @@ import {
Post,
Body,
Logger,
ForbiddenException,
HttpException,
HttpStatus,
NotFoundException,
Inject,
UseGuards,
} from '@nestjs/common';
@@ -13,6 +15,7 @@ import { Throttle } from '@nestjs/throttler';
import { AgentService } from '../agent/agent.service.js';
import { AuthGuard } from '../auth/auth.guard.js';
import { CurrentUser } from '../auth/current-user.decorator.js';
import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js';
import { v4 as uuid } from 'uuid';
import { ChatRequestDto } from './chat.dto.js';
@@ -32,16 +35,23 @@ export class ChatController {
@Throttle({ default: { limit: 10, ttl: 60_000 } })
async chat(
@Body() body: ChatRequestDto,
@CurrentUser() user: { id: string },
@CurrentUser() user: AuthenticatedUserLike,
): Promise<ChatResponse> {
const conversationId = body.conversationId ?? uuid();
const scope = scopeFromUser(user);
try {
let agentSession = this.agentService.getSession(conversationId);
let agentSession = this.agentService.getSession(conversationId, scope);
if (!agentSession) {
agentSession = await this.agentService.createSession(conversationId);
agentSession = await this.agentService.createSession(conversationId, {
userId: scope.userId,
tenantId: scope.tenantId,
});
}
} catch (err) {
if (err instanceof ForbiddenException) {
throw new NotFoundException('Session not found');
}
this.logger.error(
`Session creation failed for conversation=${conversationId}`,
err instanceof Error ? err.stack : String(err),
@@ -60,8 +70,13 @@ export class ChatController {
reject(new Error('Agent response timed out'));
}, 120_000);
const cleanup = this.agentService.onEvent(conversationId, (event: AgentSessionEvent) => {
if (event.type === 'message_update' && event.assistantMessageEvent.type === 'text_delta') {
const cleanup = this.agentService.onEvent(
conversationId,
(event: AgentSessionEvent) => {
if (
event.type === 'message_update' &&
event.assistantMessageEvent.type === 'text_delta'
) {
responseText += event.assistantMessageEvent.delta;
}
if (event.type === 'agent_end') {
@@ -69,11 +84,13 @@ export class ChatController {
cleanup();
resolve();
}
});
},
scope,
);
});
try {
await this.agentService.prompt(conversationId, body.content);
await this.agentService.prompt(conversationId, body.content, scope);
await done;
} catch (err) {
if (err instanceof HttpException) throw err;

View File

@@ -22,6 +22,11 @@ import type {
} from '@mosaicstack/types';
import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js';
import { AUTH } from '../auth/auth.tokens.js';
import {
scopeFromUser,
type ActorTenantScope,
type AuthenticatedUserLike,
} from '../auth/session-scope.js';
import { BRAIN } from '../brain/brain.tokens.js';
import { CommandRegistryService } from '../commands/command-registry.service.js';
import { CommandExecutorService } from '../commands/command-executor.service.js';
@@ -40,6 +45,8 @@ interface ClientSession {
toolCalls: Array<{ toolCallId: string; toolName: string; args: unknown; isError: boolean }>;
/** Tool calls in-flight (started but not ended yet). */
pendingToolCalls: Map<string, { toolName: string; args: unknown }>;
/** Server-derived owner/tenant scope for this socket's conversation attachment. */
scope: ActorTenantScope;
/** Last routing decision made for this session (M4-008) */
lastRoutingDecision?: RoutingDecisionInfo;
}
@@ -97,25 +104,48 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
const session = this.clientSessions.get(client.id);
if (session) {
session.cleanup();
this.agentService.removeChannel(session.conversationId, `websocket:${client.id}`);
this.agentService.removeChannel(
session.conversationId,
`websocket:${client.id}`,
session.scope,
);
this.clientSessions.delete(client.id);
}
}
private getClientScope(client: Socket): ActorTenantScope | null {
const user = client.data.user as AuthenticatedUserLike | undefined;
if (!user?.id) return null;
return scopeFromUser(user);
}
private modelOverrideKey(conversationId: string, scope: ActorTenantScope): string {
return `${scope.tenantId}:${scope.userId}:${conversationId}`;
}
private scopesEqual(a: ActorTenantScope, b: ActorTenantScope): boolean {
return a.userId === b.userId && a.tenantId === b.tenantId;
}
@SubscribeMessage('message')
async handleMessage(
@ConnectedSocket() client: Socket,
@MessageBody() data: ChatSocketMessageDto,
): Promise<void> {
const conversationId = data.conversationId ?? uuid();
const userId = (client.data.user as { id: string } | undefined)?.id;
const scope = this.getClientScope(client);
if (!scope) {
client.emit('error', { conversationId, error: 'Authenticated user scope is required.' });
return;
}
const userId = scope.userId;
this.logger.log(`Message from ${client.id} in conversation ${conversationId}`);
// Ensure agent session exists for this conversation
let sessionRoutingDecision: RoutingDecisionInfo | undefined;
try {
let agentSession = this.agentService.getSession(conversationId);
let agentSession = this.agentService.getSession(conversationId, scope);
if (!agentSession) {
// When resuming an existing conversation, load prior messages to inject as context (M1-004)
const conversationHistory = await this.loadConversationHistory(conversationId, userId);
@@ -135,7 +165,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
let resolvedProvider = data.provider;
let resolvedModelId = data.modelId;
const modelOverride = modelOverrides.get(conversationId);
const modelOverride = modelOverrides.get(this.modelOverrideKey(conversationId, scope));
if (modelOverride) {
// /model override bypasses routing engine (M4-007)
resolvedModelId = modelOverride;
@@ -172,6 +202,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
modelId: resolvedModelId,
agentConfigId: data.agentId,
userId,
tenantId: scope.tenantId,
conversationHistory: conversationHistory.length > 0 ? conversationHistory : undefined,
});
@@ -232,9 +263,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
}
// Subscribe to agent events and relay to client
const cleanup = this.agentService.onEvent(conversationId, (event: AgentSessionEvent) => {
const cleanup = this.agentService.onEvent(
conversationId,
(event: AgentSessionEvent) => {
this.relayEvent(client, conversationId, event);
});
},
scope,
);
// Preserve routing decision from the existing client session if we didn't get a new one
const prevClientSession = this.clientSessions.get(client.id);
@@ -246,16 +281,17 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
assistantText: '',
toolCalls: [],
pendingToolCalls: new Map(),
scope,
lastRoutingDecision: routingDecisionToStore,
});
// Track channel connection
this.agentService.addChannel(conversationId, `websocket:${client.id}`);
this.agentService.addChannel(conversationId, `websocket:${client.id}`, scope);
// Send session info so the client knows the model/provider (M4-008: include routing decision)
// Include agentName when a named agent config is active (M5-001)
{
const agentSession = this.agentService.getSession(conversationId);
const agentSession = this.agentService.getSession(conversationId, scope);
if (agentSession) {
const piSession = agentSession.piSession;
client.emit('session:info', {
@@ -275,7 +311,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
// Dispatch to agent
try {
await this.agentService.prompt(conversationId, data.content);
await this.agentService.prompt(conversationId, data.content, scope);
} catch (err) {
this.logger.error(
`Agent prompt failed for client=${client.id}, conversation=${conversationId}`,
@@ -293,7 +329,16 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
@ConnectedSocket() client: Socket,
@MessageBody() data: SetThinkingPayload,
): void {
const session = this.agentService.getSession(data.conversationId);
const scope = this.getClientScope(client);
if (!scope) {
client.emit('error', {
conversationId: data.conversationId,
error: 'Authenticated user scope is required.',
});
return;
}
const session = this.agentService.getSession(data.conversationId, scope);
if (!session) {
client.emit('error', {
conversationId: data.conversationId,
@@ -334,7 +379,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
const conversationId = data.conversationId;
this.logger.log(`Abort requested by ${client.id} for conversation ${conversationId}`);
const session = this.agentService.getSession(conversationId);
const scope = this.getClientScope(client);
if (!scope) {
client.emit('error', { conversationId, error: 'Authenticated user scope is required.' });
return;
}
const session = this.agentService.getSession(conversationId, scope);
if (!session) {
client.emit('error', {
conversationId,
@@ -363,8 +414,18 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
@ConnectedSocket() client: Socket,
@MessageBody() payload: SlashCommandPayload,
): Promise<void> {
const userId = (client.data.user as { id: string } | undefined)?.id ?? 'unknown';
const result = await this.commandExecutor.execute(payload, userId);
const scope = this.getClientScope(client);
if (!scope) {
client.emit('command:result', {
command: payload.command,
conversationId: payload.conversationId,
success: false,
message: 'Authenticated user scope is required.',
});
return;
}
const result = await this.commandExecutor.execute(payload, scope);
client.emit('command:result', result);
}
@@ -380,18 +441,23 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
* M5-005: Emits session:info to clients subscribed to this conversation when a model is set.
* M5-007: Records a model switch in session metrics.
*/
setModelOverride(conversationId: string, modelName: string | null): void {
setModelOverride(
conversationId: string,
modelName: string | null,
scope: ActorTenantScope,
): void {
const key = this.modelOverrideKey(conversationId, scope);
if (modelName) {
modelOverrides.set(conversationId, modelName);
modelOverrides.set(key, modelName);
this.logger.log(`Model override set: conversation=${conversationId} model="${modelName}"`);
// M5-002: Update the live session's modelId so session:info reflects the new model immediately
this.agentService.updateSessionModel(conversationId, modelName);
this.agentService.updateSessionModel(conversationId, modelName, scope);
// M5-005: Broadcast session:info to all clients subscribed to this conversation
this.broadcastSessionInfo(conversationId);
this.broadcastSessionInfo(conversationId, scope);
} else {
modelOverrides.delete(conversationId);
modelOverrides.delete(key);
this.logger.log(`Model override cleared: conversation=${conversationId}`);
}
}
@@ -399,8 +465,8 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
/**
* Return the active model override for a conversation, or undefined if none.
*/
getModelOverride(conversationId: string): string | undefined {
return modelOverrides.get(conversationId);
getModelOverride(conversationId: string, scope: ActorTenantScope): string | undefined {
return modelOverrides.get(this.modelOverrideKey(conversationId, scope));
}
/**
@@ -409,9 +475,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
*/
broadcastSessionInfo(
conversationId: string,
scope: ActorTenantScope,
extra?: { agentName?: string; routingDecision?: RoutingDecisionInfo },
): void {
const agentSession = this.agentService.getSession(conversationId);
const agentSession = this.agentService.getSession(conversationId, scope);
if (!agentSession) return;
const piSession = agentSession.piSession;
@@ -428,7 +495,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
// Emit to all clients currently subscribed to this conversation
for (const [clientId, session] of this.clientSessions) {
if (session.conversationId === conversationId) {
if (session.conversationId === conversationId && this.scopesEqual(session.scope, scope)) {
const socket = this.server.sockets.sockets.get(clientId);
if (socket?.connected) {
socket.emit('session:info', payload);
@@ -550,7 +617,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
case 'agent_end': {
// Gather usage stats from the Pi session
const agentSession = this.agentService.getSession(conversationId);
const activeClientSession = this.clientSessions.get(client.id);
const agentSession = activeClientSession
? this.agentService.getSession(conversationId, activeClientSession.scope)
: undefined;
const piSession = agentSession?.piSession;
const stats = piSession?.getSessionStats();
const contextUsage = piSession?.getContextUsage();

View File

@@ -89,6 +89,7 @@ function buildService(): CommandExecutorService {
describe('CommandExecutorService — P8-012 commands', () => {
let service: CommandExecutorService;
const userId = 'user-123';
const userScope = { userId, tenantId: userId };
const conversationId = 'conv-456';
beforeEach(() => {
@@ -99,7 +100,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /provider login — missing provider name
it('/provider login with no provider name returns usage error', async () => {
const payload: SlashCommandPayload = { command: 'provider', args: 'login', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(false);
expect(result.message).toContain('Usage: /provider login');
expect(result.command).toBe('provider');
@@ -112,7 +113,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
args: 'login anthropic',
conversationId,
};
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('provider');
expect(result.message).toContain('anthropic');
@@ -138,7 +139,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /provider with no args — returns usage
it('/provider with no args returns usage message', async () => {
const payload: SlashCommandPayload = { command: 'provider', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('Usage: /provider');
});
@@ -146,7 +147,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /provider list
it('/provider list returns success', async () => {
const payload: SlashCommandPayload = { command: 'provider', args: 'list', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('provider');
});
@@ -154,7 +155,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /provider logout with no name — usage error
it('/provider logout with no name returns error', async () => {
const payload: SlashCommandPayload = { command: 'provider', args: 'logout', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(false);
expect(result.message).toContain('Usage: /provider logout');
});
@@ -166,7 +167,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
args: 'unknown',
conversationId,
};
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(false);
expect(result.message).toContain('Unknown subcommand');
});
@@ -174,7 +175,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /mission status
it('/mission status returns stub message', async () => {
const payload: SlashCommandPayload = { command: 'mission', args: 'status', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('mission');
expect(result.message).toContain('Mission status');
@@ -183,7 +184,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /mission with no args
it('/mission with no args returns status stub', async () => {
const payload: SlashCommandPayload = { command: 'mission', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('Mission status');
});
@@ -195,7 +196,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
args: 'set my-mission-123',
conversationId,
};
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('my-mission-123');
});
@@ -203,7 +204,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /agent list
it('/agent list returns stub message', async () => {
const payload: SlashCommandPayload = { command: 'agent', args: 'list', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('agent');
expect(result.message).toContain('agent');
@@ -212,7 +213,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /agent with no args
it('/agent with no args returns usage', async () => {
const payload: SlashCommandPayload = { command: 'agent', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('Usage: /agent');
});
@@ -224,7 +225,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
args: 'my-agent-id',
conversationId,
};
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('my-agent-id');
});
@@ -232,7 +233,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /prdy
it('/prdy returns PRD wizard message', async () => {
const payload: SlashCommandPayload = { command: 'prdy', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('prdy');
expect(result.message).toContain('mosaic prdy');
@@ -241,7 +242,7 @@ describe('CommandExecutorService — P8-012 commands', () => {
// /tools
it('/tools returns tools stub message', async () => {
const payload: SlashCommandPayload = { command: 'tools', conversationId };
const result = await service.execute(payload, userId);
const result = await service.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('tools');
expect(result.message).toContain('tools');

View File

@@ -3,6 +3,7 @@ import type { QueueHandle } from '@mosaicstack/queue';
import type { Brain } from '@mosaicstack/brain';
import type { SlashCommandPayload, SlashCommandResultPayload } from '@mosaicstack/types';
import { AgentService } from '../agent/agent.service.js';
import type { ActorTenantScope } from '../auth/session-scope.js';
import { ChatGateway } from '../chat/chat.gateway.js';
import { SessionGCService } from '../gc/session-gc.service.js';
import { SystemOverrideService } from '../preferences/system-override.service.js';
@@ -34,8 +35,12 @@ export class CommandExecutorService {
private readonly mcpClient: McpClientService | null,
) {}
async execute(payload: SlashCommandPayload, userId: string): Promise<SlashCommandResultPayload> {
async execute(
payload: SlashCommandPayload,
scope: ActorTenantScope,
): Promise<SlashCommandResultPayload> {
const { command, args, conversationId } = payload;
const userId = scope.userId;
const def = this.registry.getManifest().commands.find((c) => c.name === command);
if (!def) {
@@ -50,11 +55,11 @@ export class CommandExecutorService {
try {
switch (command) {
case 'model':
return await this.handleModel(args ?? null, conversationId);
return await this.handleModel(args ?? null, conversationId, scope);
case 'thinking':
return await this.handleThinking(args ?? null, conversationId);
case 'system':
return await this.handleSystem(args ?? null, conversationId);
return await this.handleSystem(args ?? null, conversationId, scope);
case 'new':
return {
command,
@@ -94,7 +99,7 @@ export class CommandExecutorService {
};
}
case 'agent':
return await this.handleAgent(args ?? null, conversationId, userId);
return await this.handleAgent(args ?? null, conversationId, scope);
case 'provider':
return await this.handleProvider(args ?? null, userId, conversationId);
case 'mission':
@@ -146,10 +151,11 @@ export class CommandExecutorService {
private async handleModel(
args: string | null,
conversationId: string,
scope: ActorTenantScope,
): Promise<SlashCommandResultPayload> {
if (!args || args.trim().length === 0) {
// Show current override or usage hint
const currentOverride = this.chatGateway?.getModelOverride(conversationId);
const currentOverride = this.chatGateway?.getModelOverride(conversationId, scope);
if (currentOverride) {
return {
command: 'model',
@@ -171,7 +177,7 @@ export class CommandExecutorService {
// /model clear removes the override and re-enables automatic routing
if (modelName === 'clear') {
this.chatGateway?.setModelOverride(conversationId, null);
this.chatGateway?.setModelOverride(conversationId, null, scope);
return {
command: 'model',
conversationId,
@@ -181,9 +187,9 @@ export class CommandExecutorService {
}
// Set the sticky per-session override (M4-007)
this.chatGateway?.setModelOverride(conversationId, modelName);
this.chatGateway?.setModelOverride(conversationId, modelName, scope);
const session = this.agentService.getSession(conversationId);
const session = this.agentService.getSession(conversationId, scope);
if (!session) {
return {
command: 'model',
@@ -224,10 +230,11 @@ export class CommandExecutorService {
private async handleSystem(
args: string | null,
conversationId: string,
scope: ActorTenantScope,
): Promise<SlashCommandResultPayload> {
if (!args || args.trim().length === 0) {
// Clear the override when called with no args
await this.systemOverride.clear(conversationId);
await this.systemOverride.clear(conversationId, scope);
return {
command: 'system',
conversationId,
@@ -236,7 +243,7 @@ export class CommandExecutorService {
};
}
await this.systemOverride.set(conversationId, args.trim());
await this.systemOverride.set(conversationId, args.trim(), scope);
return {
command: 'system',
conversationId,
@@ -248,8 +255,9 @@ export class CommandExecutorService {
private async handleAgent(
args: string | null,
conversationId: string,
userId: string,
scope: ActorTenantScope,
): Promise<SlashCommandResultPayload> {
const userId = scope.userId;
if (!args) {
return {
command: 'agent',
@@ -338,11 +346,14 @@ export class CommandExecutorService {
conversationId,
agentConfig.id,
agentConfig.name,
scope,
agentConfig.model ?? undefined,
);
// Broadcast updated session:info so TUI TopBar reflects new agent/model
this.chatGateway?.broadcastSessionInfo(conversationId, { agentName: agentConfig.name });
this.chatGateway?.broadcastSessionInfo(conversationId, scope, {
agentName: agentConfig.name,
});
this.logger.log(
`Agent switched to "${agentConfig.name}" (${agentConfig.id}) for conversation ${conversationId} (M5-003)`,

View File

@@ -159,6 +159,7 @@ describe('CommandExecutorService — integration', () => {
let registry: CommandRegistryService;
let executor: CommandExecutorService;
const userId = 'user-integ-001';
const userScope = { userId, tenantId: userId };
const conversationId = 'conv-integ-001';
beforeEach(() => {
@@ -170,7 +171,7 @@ describe('CommandExecutorService — integration', () => {
// Unknown command returns error
it('unknown command returns success:false with descriptive message', async () => {
const payload: SlashCommandPayload = { command: 'nonexistent', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(false);
expect(result.message).toContain('nonexistent');
expect(result.command).toBe('nonexistent');
@@ -179,7 +180,7 @@ describe('CommandExecutorService — integration', () => {
// /gc handler calls SessionGCService.sweepOrphans (admin-only, no userId arg)
it('/gc calls SessionGCService.sweepOrphans without arguments', async () => {
const payload: SlashCommandPayload = { command: 'gc', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(mockSessionGC.sweepOrphans).toHaveBeenCalledWith();
expect(result.success).toBe(true);
expect(result.message).toContain('GC sweep complete');
@@ -190,8 +191,8 @@ describe('CommandExecutorService — integration', () => {
it('/system with text calls SystemOverrideService.set', async () => {
const override = 'You are a helpful assistant.';
const payload: SlashCommandPayload = { command: 'system', args: override, conversationId };
const result = await executor.execute(payload, userId);
expect(mockSystemOverride.set).toHaveBeenCalledWith(conversationId, override);
const result = await executor.execute(payload, userScope);
expect(mockSystemOverride.set).toHaveBeenCalledWith(conversationId, override, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('override set');
});
@@ -199,8 +200,8 @@ describe('CommandExecutorService — integration', () => {
// /system with no args clears the override
it('/system with no args calls SystemOverrideService.clear', async () => {
const payload: SlashCommandPayload = { command: 'system', conversationId };
const result = await executor.execute(payload, userId);
expect(mockSystemOverride.clear).toHaveBeenCalledWith(conversationId);
const result = await executor.execute(payload, userScope);
expect(mockSystemOverride.clear).toHaveBeenCalledWith(conversationId, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('cleared');
});
@@ -212,7 +213,7 @@ describe('CommandExecutorService — integration', () => {
args: 'claude-3-opus',
conversationId,
};
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('model');
expect(result.message).toContain('claude-3-opus');
@@ -221,7 +222,7 @@ describe('CommandExecutorService — integration', () => {
// /thinking with valid level returns success
it('/thinking with valid level returns success', async () => {
const payload: SlashCommandPayload = { command: 'thinking', args: 'high', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('high');
});
@@ -229,7 +230,7 @@ describe('CommandExecutorService — integration', () => {
// /thinking with invalid level returns usage message
it('/thinking with invalid level returns usage message', async () => {
const payload: SlashCommandPayload = { command: 'thinking', args: 'invalid', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.message).toContain('Usage:');
});
@@ -237,7 +238,7 @@ describe('CommandExecutorService — integration', () => {
// /new command returns success
it('/new returns success', async () => {
const payload: SlashCommandPayload = { command: 'new', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe('new');
});
@@ -245,7 +246,7 @@ describe('CommandExecutorService — integration', () => {
// /reload without reloadService returns failure
it('/reload without ReloadService returns failure', async () => {
const payload: SlashCommandPayload = { command: 'reload', conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(false);
expect(result.message).toContain('ReloadService');
});
@@ -255,7 +256,7 @@ describe('CommandExecutorService — integration', () => {
for (const cmd of stubCommands) {
it(`/${cmd} returns success (stub)`, async () => {
const payload: SlashCommandPayload = { command: cmd, conversationId };
const result = await executor.execute(payload, userId);
const result = await executor.execute(payload, userScope);
expect(result.success).toBe(true);
expect(result.command).toBe(cmd);
});

View File

@@ -1,9 +1,13 @@
import { Injectable, Logger } from '@nestjs/common';
import { createQueue, type QueueHandle } from '@mosaicstack/queue';
import type { ActorTenantScope } from '../auth/session-scope.js';
const SESSION_SYSTEM_KEY = (sessionId: string) => `mosaic:session:${sessionId}:system`;
const SESSION_SYSTEM_FRAGMENTS_KEY = (sessionId: string) =>
`mosaic:session:${sessionId}:system:fragments`;
const scopedSessionId = (sessionId: string, scope: ActorTenantScope) =>
`${scope.tenantId}:${scope.userId}:${sessionId}`;
const SESSION_SYSTEM_KEY = (sessionId: string, scope: ActorTenantScope) =>
`mosaic:session:${scopedSessionId(sessionId, scope)}:system`;
const SESSION_SYSTEM_FRAGMENTS_KEY = (sessionId: string, scope: ActorTenantScope) =>
`mosaic:session:${scopedSessionId(sessionId, scope)}:system:fragments`;
const SYSTEM_OVERRIDE_TTL_SECONDS = 604800; // 7 days
interface OverrideFragment {
@@ -20,9 +24,9 @@ export class SystemOverrideService {
this.handle = createQueue();
}
async set(sessionId: string, override: string): Promise<void> {
async set(sessionId: string, override: string, scope: ActorTenantScope): Promise<void> {
// Load existing fragments
const existing = await this.handle.redis.get(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId));
const existing = await this.handle.redis.get(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId, scope));
const fragments: OverrideFragment[] = existing
? (JSON.parse(existing) as OverrideFragment[])
: [];
@@ -37,11 +41,11 @@ export class SystemOverrideService {
// Store both: fragments array and condensed result
const pipeline = this.handle.redis.pipeline();
pipeline.setex(
SESSION_SYSTEM_FRAGMENTS_KEY(sessionId),
SESSION_SYSTEM_FRAGMENTS_KEY(sessionId, scope),
SYSTEM_OVERRIDE_TTL_SECONDS,
JSON.stringify(fragments),
);
pipeline.setex(SESSION_SYSTEM_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS, condensed);
pipeline.setex(SESSION_SYSTEM_KEY(sessionId, scope), SYSTEM_OVERRIDE_TTL_SECONDS, condensed);
await pipeline.exec();
this.logger.debug(
@@ -49,21 +53,21 @@ export class SystemOverrideService {
);
}
async get(sessionId: string): Promise<string | null> {
return this.handle.redis.get(SESSION_SYSTEM_KEY(sessionId));
async get(sessionId: string, scope: ActorTenantScope): Promise<string | null> {
return this.handle.redis.get(SESSION_SYSTEM_KEY(sessionId, scope));
}
async renew(sessionId: string): Promise<void> {
async renew(sessionId: string, scope: ActorTenantScope): Promise<void> {
const pipeline = this.handle.redis.pipeline();
pipeline.expire(SESSION_SYSTEM_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS);
pipeline.expire(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId), SYSTEM_OVERRIDE_TTL_SECONDS);
pipeline.expire(SESSION_SYSTEM_KEY(sessionId, scope), SYSTEM_OVERRIDE_TTL_SECONDS);
pipeline.expire(SESSION_SYSTEM_FRAGMENTS_KEY(sessionId, scope), SYSTEM_OVERRIDE_TTL_SECONDS);
await pipeline.exec();
}
async clear(sessionId: string): Promise<void> {
async clear(sessionId: string, scope: ActorTenantScope): Promise<void> {
await this.handle.redis.del(
SESSION_SYSTEM_KEY(sessionId),
SESSION_SYSTEM_FRAGMENTS_KEY(sessionId),
SESSION_SYSTEM_KEY(sessionId, scope),
SESSION_SYSTEM_FRAGMENTS_KEY(sessionId, scope),
);
this.logger.debug(`Cleared system override for session ${sessionId}`);
}