fix(#707): scope session GC retention #720

Merged
jason.woltje merged 3 commits from fix/tess-session-gc-scope into main 2026-07-13 00:44:20 +00:00
10 changed files with 163 additions and 175 deletions

View File

@@ -103,7 +103,10 @@ describe('TESS-M1-SEC-001 command authorization abuse cases', () => {
expect(denied.success).toBe(false); expect(denied.success).toBe(false);
expect(denied.message).toContain('approval'); expect(denied.message).toContain('approval');
expect(approval).not.toBeNull(); expect(approval).not.toBeNull();
expect(approved.success).toBe(true); // A valid durable approval is consumed, but cannot authorize an unimplemented
expect(sessionGc.sweepOrphans).toHaveBeenCalledOnce(); // global retention operation. Session-scoped cleanup remains lifecycle-only.
expect(approved.success).toBe(false);
expect(approved.message).toContain('Global GC is disabled');
expect(sessionGc.sweepOrphans).not.toHaveBeenCalled();
}); });
}); });

View File

@@ -102,16 +102,15 @@ export class CommandExecutorService {
success: true, success: true,
message: 'Retry last message requested.', message: 'Retry last message requested.',
}; };
case 'gc': { case 'gc':
// Admin-only: system-wide GC sweep across all sessions // Global retention requires a separate, authorized and audited job.
const result = await this.sessionGC.sweepOrphans(); // Session cleanup is performed only through the session lifecycle.
return { return {
command: 'gc', command: 'gc',
success: true, success: false,
message: `GC sweep complete: ${result.orphanedSessions} orphaned sessions cleaned in ${result.duration}ms.`, message: 'Global GC is disabled pending an authorized retention job.',
conversationId, conversationId,
}; };
}
case 'agent': case 'agent':
return await this.handleAgent(args ?? null, conversationId, scope); return await this.handleAgent(args ?? null, conversationId, scope);
case 'provider': case 'provider':

View File

@@ -177,14 +177,12 @@ describe('CommandExecutorService — integration', () => {
expect(result.command).toBe('nonexistent'); expect(result.command).toBe('nonexistent');
}); });
// /gc handler calls SessionGCService.sweepOrphans (admin-only, no userId arg) it('/gc refuses an unaudited global sweep', async () => {
it('/gc calls SessionGCService.sweepOrphans without arguments', async () => {
const payload: SlashCommandPayload = { command: 'gc', conversationId }; const payload: SlashCommandPayload = { command: 'gc', conversationId };
const result = await executor.execute(payload, userScope); const result = await executor.execute(payload, userScope);
expect(mockSessionGC.sweepOrphans).toHaveBeenCalledWith(); expect(mockSessionGC.sweepOrphans).not.toHaveBeenCalled();
expect(result.success).toBe(true); expect(result.success).toBe(false);
expect(result.message).toContain('GC sweep complete'); expect(result.message).toContain('disabled pending an authorized retention job');
expect(result.message).toContain('3 orphaned sessions');
}); });
// /system with args calls SystemOverrideService.set // /system with args calls SystemOverrideService.set

View File

@@ -3,6 +3,7 @@ import { Logger } from '@nestjs/common';
import type { QueueHandle } from '@mosaicstack/queue'; import type { QueueHandle } from '@mosaicstack/queue';
import type { LogService } from '@mosaicstack/log'; import type { LogService } from '@mosaicstack/log';
import { SessionGCService } from './session-gc.service.js'; import { SessionGCService } from './session-gc.service.js';
import { CommandAuthorizationService } from '../commands/command-authorization.service.js';
type MockRedis = { type MockRedis = {
scan: ReturnType<typeof vi.fn>; scan: ReturnType<typeof vi.fn>;
@@ -12,7 +13,12 @@ type MockRedis = {
describe('SessionGCService', () => { describe('SessionGCService', () => {
let service: SessionGCService; let service: SessionGCService;
let mockRedis: MockRedis; let mockRedis: MockRedis;
let mockLogService: { logs: { promoteToWarm: ReturnType<typeof vi.fn> } }; let mockLogService: {
logs: {
promoteSessionToWarm: ReturnType<typeof vi.fn>;
promoteToWarm: ReturnType<typeof vi.fn>;
};
};
/** /**
* Helper: build a scan mock that returns all provided keys in a single * Helper: build a scan mock that returns all provided keys in a single
@@ -30,6 +36,7 @@ describe('SessionGCService', () => {
mockLogService = { mockLogService = {
logs: { logs: {
promoteSessionToWarm: vi.fn().mockResolvedValue(0),
promoteToWarm: vi.fn().mockResolvedValue(0), promoteToWarm: vi.fn().mockResolvedValue(0),
}, },
}; };
@@ -59,54 +66,76 @@ describe('SessionGCService', () => {
expect(result.cleaned.valkeyKeys).toBeUndefined(); expect(result.cleaned.valkeyKeys).toBeUndefined();
}); });
it('escapes glob metacharacters in a session identifier', async () => {
await service.collect('abc*?[tenant]\\escape');
expect(mockRedis.scan).toHaveBeenCalledWith(
'0',
'MATCH',
'mosaic:session:abc\\*\\?\\[tenant\\]\\\\escape:*',
'COUNT',
100,
);
});
it('preserves a valid durable approval after session GC', async () => {
const entries = new Map<string, string>();
const redis = {
scan: vi.fn().mockResolvedValue(['0', ['mosaic:session:owned:state']]),
get: vi.fn(async (key: string) => entries.get(key) ?? null),
set: vi.fn(async (key: string, value: string) => entries.set(key, value)),
del: vi.fn(async (...keys: string[]) => {
let deleted = 0;
for (const key of keys) deleted += Number(entries.delete(key));
return deleted;
}),
};
const authorization = new CommandAuthorizationService(
{
select: () => ({
from: () => ({ where: () => ({ limit: async () => [{ role: 'admin' }] }) }),
}),
} as never,
redis,
);
const command = {
name: 'gc',
description: 'System-wide garbage collection',
aliases: [],
scope: 'admin',
execution: 'socket',
available: true,
} as never;
const payload = { command: 'gc', conversationId: 'owned' };
const approval = await authorization.createApproval(command, payload, 'admin-1');
const approvalKey = `tess:command-approval:${approval!.approvalId}`;
const gc = new SessionGCService(redis as never, mockLogService as unknown as LogService);
await gc.collect('owned');
expect(entries.has(approvalKey)).toBe(true);
await expect(
authorization.authorize(command, payload, 'admin-1', approval!.approvalId),
).resolves.toEqual({ allowed: true });
});
it('collect() returns sessionId in result', async () => { it('collect() returns sessionId in result', async () => {
const result = await service.collect('test-session-id'); const result = await service.collect('test-session-id');
expect(result.sessionId).toBe('test-session-id'); expect(result.sessionId).toBe('test-session-id');
}); });
it('fullCollect() deletes all session keys', async () => { it('collect() demotes logs only for the requested session', async () => {
mockRedis.scan = makeScanMock(['mosaic:session:abc:system', 'mosaic:session:xyz:foo']); await service.collect('owned-session');
const result = await service.fullCollect();
expect(mockRedis.del).toHaveBeenCalled(); expect(mockLogService.logs.promoteSessionToWarm).toHaveBeenCalledWith(
expect(result.valkeyKeys).toBe(2); 'owned-session',
expect.any(Date),
);
expect(mockLogService.logs.promoteToWarm).not.toHaveBeenCalled();
}); });
it('fullCollect() with no keys returns 0 valkeyKeys', async () => { it('does not expose automatic global GC entry points', () => {
mockRedis.scan = makeScanMock([]); expect('fullCollect' in service).toBe(false);
const result = await service.fullCollect(); expect('sweepOrphans' in service).toBe(false);
expect(result.valkeyKeys).toBe(0);
expect(mockRedis.del).not.toHaveBeenCalled();
});
it('fullCollect() returns duration', async () => {
const result = await service.fullCollect();
expect(result.duration).toBeGreaterThanOrEqual(0);
});
it('sweepOrphans() extracts unique session IDs and collects them', async () => {
// First scan call returns the global session list; subsequent calls return
// per-session keys during collect().
mockRedis.scan = vi
.fn()
.mockResolvedValueOnce([
'0',
['mosaic:session:abc:system', 'mosaic:session:abc:messages', 'mosaic:session:xyz:system'],
])
// collect('abc') scan
.mockResolvedValueOnce(['0', ['mosaic:session:abc:system', 'mosaic:session:abc:messages']])
// collect('xyz') scan
.mockResolvedValueOnce(['0', ['mosaic:session:xyz:system']]);
mockRedis.del.mockResolvedValue(1);
const result = await service.sweepOrphans();
expect(result.orphanedSessions).toBeGreaterThanOrEqual(0);
expect(result.duration).toBeGreaterThanOrEqual(0);
});
it('sweepOrphans() returns empty when no session keys', async () => {
mockRedis.scan = makeScanMock([]);
const result = await service.sweepOrphans();
expect(result.orphanedSessions).toBe(0);
expect(result.totalCleaned).toHaveLength(0);
}); });
}); });

View File

@@ -1,4 +1,4 @@
import { Inject, Injectable, Logger, type OnModuleInit } from '@nestjs/common'; import { Inject, Injectable } from '@nestjs/common';
import type { QueueHandle } from '@mosaicstack/queue'; import type { QueueHandle } from '@mosaicstack/queue';
import type { LogService } from '@mosaicstack/log'; import type { LogService } from '@mosaicstack/log';
import { LOG_SERVICE } from '../log/log.tokens.js'; import { LOG_SERVICE } from '../log/log.tokens.js';
@@ -13,49 +13,18 @@ export interface GCResult {
}; };
} }
export interface GCSweepResult { /** Escape Redis glob metacharacters so a session identifier is always literal. */
orphanedSessions: number; function escapeRedisGlobLiteral(value: string): string {
totalCleaned: GCResult[]; return value.replace(/[\\*?\[\]]/g, '\\$&');
duration: number;
}
export interface FullGCResult {
valkeyKeys: number;
logsDemoted: number;
jobsPurged: number;
tempFilesRemoved: number;
duration: number;
} }
@Injectable() @Injectable()
export class SessionGCService implements OnModuleInit { export class SessionGCService {
private readonly logger = new Logger(SessionGCService.name);
constructor( constructor(
@Inject(REDIS) private readonly redis: QueueHandle['redis'], @Inject(REDIS) private readonly redis: QueueHandle['redis'],
@Inject(LOG_SERVICE) private readonly logService: LogService, @Inject(LOG_SERVICE) private readonly logService: LogService,
) {} ) {}
onModuleInit(): void {
// Fire-and-forget: run full GC asynchronously so it does not block the
// NestJS bootstrap chain. Cold-start GC typically takes 100500 ms
// depending on Valkey key count; deferring it removes that latency from
// the TTFB of the first HTTP request.
this.fullCollect()
.then((result) => {
this.logger.log(
`Full GC complete: ${result.valkeyKeys} Valkey keys, ` +
`${result.logsDemoted} logs demoted, ` +
`${result.jobsPurged} jobs purged, ` +
`${result.tempFilesRemoved} temp dirs removed ` +
`(${result.duration}ms)`,
);
})
.catch((err: unknown) => {
this.logger.error('Cold-start GC failed', err instanceof Error ? err.stack : String(err));
});
}
/** /**
* Scan Valkey for all keys matching a pattern using SCAN (non-blocking). * Scan Valkey for all keys matching a pattern using SCAN (non-blocking).
* KEYS is avoided because it blocks the Valkey event loop for the full scan * KEYS is avoided because it blocks the Valkey event loop for the full scan
@@ -79,86 +48,20 @@ export class SessionGCService implements OnModuleInit {
const result: GCResult = { sessionId, cleaned: {} }; const result: GCResult = { sessionId, cleaned: {} };
// 1. Valkey: delete all session-scoped keys // 1. Valkey: delete all session-scoped keys
const pattern = `mosaic:session:${sessionId}:*`; const pattern = `mosaic:session:${escapeRedisGlobLiteral(sessionId)}:*`;
const valkeyKeys = await this.scanKeys(pattern); const valkeyKeys = await this.scanKeys(pattern);
if (valkeyKeys.length > 0) { if (valkeyKeys.length > 0) {
await this.redis.del(...valkeyKeys); await this.redis.del(...valkeyKeys);
result.cleaned.valkeyKeys = valkeyKeys.length; result.cleaned.valkeyKeys = valkeyKeys.length;
} }
// 2. PG: demote hot-tier agent_logs for this session to warm // 2. PG: demote hot-tier agent logs for this session only.
const cutoff = new Date(); // demote all hot logs for this session const cutoff = new Date();
const logsDemoted = await this.logService.logs.promoteToWarm(cutoff); const logsDemoted = await this.logService.logs.promoteSessionToWarm(sessionId, cutoff);
if (logsDemoted > 0) { if (logsDemoted > 0) {
result.cleaned.logsDemoted = logsDemoted; result.cleaned.logsDemoted = logsDemoted;
} }
return result; return result;
} }
/**
* Sweep GC — find orphaned artifacts from dead sessions.
* System-wide operation: only call from admin-authorized paths or internal
* scheduled jobs. Individual session cleanup is handled by collect().
*/
async sweepOrphans(): Promise<GCSweepResult> {
const start = Date.now();
const cleaned: GCResult[] = [];
// 1. Find all session-scoped Valkey keys (non-blocking SCAN)
const allSessionKeys = await this.scanKeys('mosaic:session:*');
// Extract unique session IDs from keys
const sessionIds = new Set<string>();
for (const key of allSessionKeys) {
const match = key.match(/^mosaic:session:([^:]+):/);
if (match) sessionIds.add(match[1]!);
}
// 2. For each session ID, collect stale keys
for (const sessionId of sessionIds) {
const gcResult = await this.collect(sessionId);
if (Object.keys(gcResult.cleaned).length > 0) {
cleaned.push(gcResult);
}
}
return {
orphanedSessions: cleaned.length,
totalCleaned: cleaned,
duration: Date.now() - start,
};
}
/**
* Full GC — aggressive collection for cold start.
* Assumes no sessions survived the restart.
*/
async fullCollect(): Promise<FullGCResult> {
const start = Date.now();
// 1. Valkey: delete ALL session-scoped keys (non-blocking SCAN)
const sessionKeys = await this.scanKeys('mosaic:session:*');
if (sessionKeys.length > 0) {
await this.redis.del(...sessionKeys);
}
// 2. NOTE: channel keys are NOT collected on cold start
// (discord/telegram plugins may reconnect and resume)
// 3. PG: demote stale hot-tier logs older than 24h to warm
const hotCutoff = new Date(Date.now() - 24 * 60 * 60 * 1000);
const logsDemoted = await this.logService.logs.promoteToWarm(hotCutoff);
// 4. No summarization job purge API available yet
const jobsPurged = 0;
return {
valkeyKeys: sessionKeys.length,
logsDemoted,
jobsPurged,
tempFilesRemoved: 0,
duration: Date.now() - start,
};
}
} }

View File

@@ -6,11 +6,10 @@ import {
type OnModuleDestroy, type OnModuleDestroy,
} from '@nestjs/common'; } from '@nestjs/common';
import { SummarizationService } from './summarization.service.js'; import { SummarizationService } from './summarization.service.js';
import { SessionGCService } from '../gc/session-gc.service.js';
import { import {
QueueService, QueueService,
QUEUE_SUMMARIZATION,
QUEUE_GC, QUEUE_GC,
QUEUE_SUMMARIZATION,
QUEUE_TIER_MANAGEMENT, QUEUE_TIER_MANAGEMENT,
} from '../queue/queue.service.js'; } from '../queue/queue.service.js';
import type { Worker } from 'bullmq'; import type { Worker } from 'bullmq';
@@ -23,14 +22,12 @@ export class CronService implements OnModuleInit, OnModuleDestroy {
constructor( constructor(
@Inject(SummarizationService) private readonly summarization: SummarizationService, @Inject(SummarizationService) private readonly summarization: SummarizationService,
@Inject(SessionGCService) private readonly sessionGC: SessionGCService,
@Inject(QueueService) private readonly queueService: QueueService, @Inject(QueueService) private readonly queueService: QueueService,
) {} ) {}
async onModuleInit(): Promise<void> { async onModuleInit(): Promise<void> {
const summarizationSchedule = process.env['SUMMARIZATION_CRON'] ?? '0 */6 * * *'; // every 6 hours const summarizationSchedule = process.env['SUMMARIZATION_CRON'] ?? '0 */6 * * *'; // every 6 hours
const tierManagementSchedule = process.env['TIER_MANAGEMENT_CRON'] ?? '0 3 * * *'; // daily at 3am const tierManagementSchedule = process.env['TIER_MANAGEMENT_CRON'] ?? '0 3 * * *'; // daily at 3am
const gcSchedule = process.env['SESSION_GC_CRON'] ?? '0 4 * * *'; // daily at 4am
// M6-003: Summarization repeatable job // M6-003: Summarization repeatable job
await this.queueService.addRepeatableJob( await this.queueService.addRepeatableJob(
@@ -56,15 +53,12 @@ export class CronService implements OnModuleInit, OnModuleDestroy {
}); });
this.registeredWorkers.push(tierWorker); this.registeredWorkers.push(tierWorker);
// M6-004: GC repeatable job // Retire any repeatable global GC schedule created by older deployments.
await this.queueService.addRepeatableJob(QUEUE_GC, 'session-gc', {}, gcSchedule); // Session cleanup is now triggered only by an authorized session lifecycle operation.
const gcWorker = this.queueService.registerWorker(QUEUE_GC, async () => { await this.queueService.removeRepeatableJobs(QUEUE_GC, 'session-gc');
await this.sessionGC.sweepOrphans();
});
this.registeredWorkers.push(gcWorker);
this.logger.log( this.logger.log(
`BullMQ jobs scheduled: summarization="${summarizationSchedule}", tier="${tierManagementSchedule}", gc="${gcSchedule}"`, `BullMQ jobs scheduled: summarization="${summarizationSchedule}", tier="${tierManagementSchedule}"`,
); );
} }

View File

@@ -162,6 +162,23 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
); );
} }
/**
* Remove every existing repeatable schedule for a job name. This supports
* safe retirement of previously registered system-wide jobs.
*/
async removeRepeatableJobs(queueName: string, jobName: string): Promise<number> {
const queue = this.getQueue(queueName);
const jobs = await queue.getRepeatableJobs();
const matchingJobs = jobs.filter((job) => job.name === jobName);
await Promise.all(matchingJobs.map((job) => queue.removeRepeatableByKey(job.key)));
if (matchingJobs.length > 0) {
this.logger.log(
`Removed ${matchingJobs.length} repeatable "${jobName}" job(s) from "${queueName}"`,
);
}
return matchingJobs.length;
}
/** /**
* Register a Worker for the given queue name with error handling and * Register a Worker for the given queue name with error handling and
* exponential backoff. * exponential backoff.

View File

@@ -312,6 +312,10 @@ When `DISCORD_BOT_TOKEN` is configured, `DISCORD_SERVICE_TOKEN`, `DISCORD_SERVIC
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, 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.
### Session retention and garbage collection
Session cleanup is scoped to one session identifier and only removes that session's Valkey keys and demotes that session's hot logs. Gateway startup and scheduled jobs do not perform global session cleanup; startup removes legacy repeatable `session-gc` schedules created by older deployments. The `/gc` command is intentionally disabled until a distinct global-retention job supplies explicit authorization and audit evidence. This prevents one tenant or session's cleanup from changing another's retained data.
### Observability ### Observability
| Variable | Default | Description | | Variable | Default | Description |

View File

@@ -0,0 +1,22 @@
# Scratchpad — TESS-M1-SEC-006 Session GC scope
- **Task / issue:** TESS-M1-SEC-006 / #707
- **Branch:** `fix/tess-session-gc-scope` from `origin/main` at `59e49cfd`
- **Objective:** Make session cleanup session-scoped and prevent automatic global retention/GC without an authorized, auditable operation.
- **Scope:** `apps/gateway`, `packages/log`, admin/developer operations documentation.
- **Budget:** Task estimate 18K; no explicit hard cap supplied.
- **Assumption:** No authorized global retention service exists today. Existing full/sweep GC must therefore be disabled from startup and cron paths, while single-session cleanup remains available.
## Plan
1. Add failing isolation tests proving single-session cleanup only demotes its own logs and automatic startup/scheduled GC cannot globally delete session data.
2. Add session-scoped log repository retention and make `collect(sessionId)` use it.
3. Remove automatic full/sweep GC invocation; preserve any future global operation behind an explicit authorization/audit seam.
4. Document the operational boundary, run gates, review, and commit without push.
## Verification evidence
- Isolation TDD: `pnpm --filter @mosaicstack/gateway test -- session-gc.service.spec.ts commands.integration.spec.ts command-executor-p8012.spec.ts` — 61 passed.
- `pnpm typecheck` — passed.
- `pnpm lint` — passed.
- `pnpm format:check` remains red only on the known pre-existing Tess documentation debt; changed files are Prettier-clean.

View File

@@ -58,9 +58,28 @@ export function createAgentLogsRepo(db: Db) {
return rows[0]; return rows[0];
}, },
/**
* Transition hot logs for one session to warm tier. Session retention is
* default-deny: no other session's logs can be changed by this operation.
*/
async promoteSessionToWarm(sessionId: string, olderThan: Date): Promise<number> {
const result = await db
.update(agentLogs)
.set({ tier: 'warm', summarizedAt: new Date() })
.where(
and(
eq(agentLogs.sessionId, sessionId),
eq(agentLogs.tier, 'hot'),
lt(agentLogs.createdAt, olderThan),
),
)
.returning();
return result.length;
},
/** /**
* Transition hot logs older than the cutoff to warm tier. * Transition hot logs older than the cutoff to warm tier.
* Returns the number of logs transitioned. * Reserved for a separately authorized global retention job.
*/ */
async promoteToWarm(olderThan: Date): Promise<number> { async promoteToWarm(olderThan: Date): Promise<number> {
const result = await db const result = await db