diff --git a/docs/chat-chain-changes/2026-07-05-group-chat-workspace-diff.md b/docs/chat-chain-changes/2026-07-05-group-chat-workspace-diff.md new file mode 100644 index 0000000000..a530082eac --- /dev/null +++ b/docs/chat-chain-changes/2026-07-05-group-chat-workspace-diff.md @@ -0,0 +1,29 @@ +--- +date: 2026-07-05 +pr: pending +feature: Group Chat workspace diff audit messages +impact: Group Chat room-agent workspace runs persist bounded workspace_diff audit cards using Bridge-assigned run ids, exclude those audit cards from future model context, and fence stale or aborted run output before it can be saved. +--- + +Group Chat workspace runs now start a workspace diff checkpoint after Agent +Bridge returns its canonical `run_id`. WUI does not supply or override Bridge +run ids; the persisted audit row and tool-style room message follow the +Bridge-assigned run id. + +When a room-agent run finishes, the server persists the bounded +`workspace_diff` message and matching `workspace_run_changes` row in one +database transaction. The persisted payload stores only the workspace basename, +keeps bounded file summaries and patches for the chat card, and avoids adding a +lazy group-chat file-detail endpoint. + +Workspace diff audit messages are excluded from future Group Chat model context +and context token estimates, while still rendering as visible audit cards even +when generic tool traces are hidden. Client-supplied `workspace_diff` tool rows +are sanitized so only server-created audit cards keep that protected tool name. + +Clear-context, delete-room, workspace-switch, and interrupt flows now fence the +current room-agent Bridge session before stale assistant/tool/workspace-diff +output can be persisted. Synchronized interrupts can finalize in-flight +diffs as aborted, unsynchronized interrupt failures leave the diff state +pending, and room interrupt pauses mention-queue draining so a queued mention +cannot start a new old-workspace run while the room is being reset. diff --git a/packages/client/src/components/hermes/group-chat/GroupMessageItem.vue b/packages/client/src/components/hermes/group-chat/GroupMessageItem.vue index 3049729382..1a470fdf56 100644 --- a/packages/client/src/components/hermes/group-chat/GroupMessageItem.vue +++ b/packages/client/src/components/hermes/group-chat/GroupMessageItem.vue @@ -174,6 +174,23 @@ const copyableContent = computed(() => { const toolExpanded = ref(false) const isToolMessage = computed(() => props.message.role === 'tool') +const workspaceDiffPayload = computed(() => { + if ((props.message.toolName || props.message.tool_name) !== 'workspace_diff') return null + const raw = props.message.toolResult ?? props.message.content + if (!raw) return null + if (typeof raw === 'object' && (raw as any)?.kind === 'workspace_diff') return raw as any + if (typeof raw === 'string') { + try { + const parsed = JSON.parse(raw) + return parsed?.kind === 'workspace_diff' ? parsed : null + } catch { + return null + } + } + return null +}) +const workspaceDiffFiles = computed(() => Array.isArray(workspaceDiffPayload.value?.files) ? workspaceDiffPayload.value.files : []) +const workspaceDiffLabel = computed(() => workspaceDiffPayload.value?.workspace_basename || t('chat.workspace')) const toolArgsPayload = computed(() => formatToolPayload(props.message.toolArgs)) const toolResultPayload = computed(() => formatToolPayload(props.message.toolResult, true)) const hasToolDetails = computed(() => !!(toolArgsPayload.value.full || toolResultPayload.value.full)) @@ -490,7 +507,33 @@ onBeforeUnmount(() => { {{ message.senderName }} {{ agentInfo.description }} -
+
+
+ {{ t('chat.workspaceChanges') }} + {{ workspaceDiffPayload.status }} +
+
+ {{ workspaceDiffLabel }} + {{ t('chat.changedFiles', { files: workspaceDiffPayload.files_changed ?? workspaceDiffFiles.length }) }} + +{{ workspaceDiffPayload.additions || 0 }} + -{{ workspaceDiffPayload.deletions || 0 }} + {{ t('chat.truncated') }} +
+
+
+
+ {{ file.path }} + {{ file.change_type }} + +{{ file.additions || 0 }} + -{{ file.deletions || 0 }} + {{ t('chat.binaryFileDiffUnavailable') }} + {{ t('chat.truncated') }} +
+
+
+
+
+
{ {{ t('chat.error') }}
-
+
{{ t('chat.arguments') }}
@@ -756,6 +799,81 @@ onBeforeUnmount(() => { padding-left: 10px; } +.workspace-diff-card { + width: min(760px, 100%); + border: 1px solid $border-color; + border-radius: $radius-sm; + background: rgba(var(--accent-primary-rgb), 0.04); + overflow: hidden; +} + +.workspace-diff-head, +.workspace-diff-meta, +.workspace-diff-file-head { + display: flex; + align-items: center; + gap: 8px; + min-width: 0; +} + +.workspace-diff-head { + justify-content: space-between; + padding: 8px 10px; + border-bottom: 1px solid $border-color; +} + +.workspace-diff-title { + font-weight: 700; + color: $text-primary; +} + +.workspace-diff-status { + font-size: 11px; + color: $text-secondary; + font-family: $font-code; +} + +.workspace-diff-meta { + padding: 6px 10px; + font-size: 12px; + color: $text-secondary; + flex-wrap: wrap; +} + +.workspace-diff-files { + display: grid; + gap: 8px; + padding: 8px 10px 10px; +} + +.workspace-diff-file { + min-width: 0; +} + +.workspace-diff-file-head { + padding: 4px 0; + font-size: 11px; + color: $text-muted; + font-family: $font-code; +} + +.workspace-diff-path { + flex: 1 1 auto; + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + color: $text-primary; +} + +.diff-add { + color: $success; +} + +.diff-del { + color: $error; +} + .tool-detail-section { margin-bottom: 6px; } diff --git a/packages/client/src/components/hermes/group-chat/GroupMessageList.vue b/packages/client/src/components/hermes/group-chat/GroupMessageList.vue index b4f7c6ae42..9165ae909f 100644 --- a/packages/client/src/components/hermes/group-chat/GroupMessageList.vue +++ b/packages/client/src/components/hermes/group-chat/GroupMessageList.vue @@ -11,7 +11,13 @@ const { t } = useI18n() const { toolTraceVisible } = useToolTraceVisibility() const listRef = ref | null>(null) const showScrollBottomButton = ref(false) -const displayMessages = computed(() => store.sortedMessages.filter(msg => msg.role !== 'tool' || toolTraceVisible.value || msg.toolStatus === 'running')) +const displayMessages = computed(() => store.sortedMessages.filter(msg => + msg.role !== 'tool' || + toolTraceVisible.value || + msg.toolStatus === 'running' || + msg.toolName === 'workspace_diff' || + msg.tool_name === 'workspace_diff', +)) const listPadding = computed(() => store.activePendingApproval ? '16px 20px 260px' : '16px 20px') let pendingInitialBottomRoomId: string | null = store.currentRoomId diff --git a/packages/client/src/stores/hermes/group-chat.ts b/packages/client/src/stores/hermes/group-chat.ts index 49125f8c32..17ee46b910 100644 --- a/packages/client/src/stores/hermes/group-chat.ts +++ b/packages/client/src/stores/hermes/group-chat.ts @@ -951,6 +951,16 @@ function runtimePayloadText(value: unknown): string { return String(value) } +function parseWorkspaceDiffPayload(value: unknown): unknown { + const text = runtimePayloadText(value) + if (!text) return undefined + try { + const parsed = JSON.parse(text) + return parsed?.kind === 'workspace_diff' ? parsed : value + } catch { + return value + } +} function mapGroupMessages(msgs: ChatMessage[]): ChatMessage[] { const toolNameMap = new Map() @@ -998,7 +1008,9 @@ function mapGroupMessages(msgs: ChatMessage[]): ChatMessage[] { const toolName = msg.tool_name || toolNameMap.get(tcId) || undefined const toolArgs = toolArgsMap.has(tcId) ? toolArgsMap.get(tcId) : undefined let preview = '' - const toolResult = runtimeToolPayloadOrUndefined((msg as any).content) + const toolResult = toolName === 'workspace_diff' + ? parseWorkspaceDiffPayload((msg as any).content) + : runtimeToolPayloadOrUndefined((msg as any).content) const contentText = runtimePayloadText((msg as any).content) if (contentText) { try { diff --git a/packages/server/src/controllers/hermes/sessions.ts b/packages/server/src/controllers/hermes/sessions.ts index 78181f43dc..d6c3ca8996 100644 --- a/packages/server/src/controllers/hermes/sessions.ts +++ b/packages/server/src/controllers/hermes/sessions.ts @@ -19,7 +19,13 @@ import type { UsageStatsModelRow, UsageStatsDailyRow } from '../../db/hermes/usa import { deleteWorkspaceRunChangesForSession, getWorkspaceRunChangeFile as getWorkspaceRunChangeFileFromDb, listWorkspaceRunChangesForSession } from '../../db/hermes/workspace-run-changes-store' import { getModelContextLength } from '../../services/hermes/model-context' import { getActiveProfileName, listProfileNamesFromDisk } from '../../services/hermes/hermes-profile' -import { isNearestExistingRealPathWithin, isPathWithin, isRealPathWithin } from '../../services/hermes/hermes-path' +import { isNearestExistingRealPathWithin, isPathWithin } from '../../services/hermes/hermes-path' +import { + isWorkspaceListPathAllowed, + normalizeWindowsWorkspacePath, + useWindowsDriveWorkspaceMode, + workspaceBaseOverride, +} from '../../services/hermes/workspace-path' import { getGroupChatServer } from '../../routes/hermes/group-chat' import { logger } from '../../services/logger' import type { ConversationSummary } from '../../services/hermes/conversations' @@ -30,7 +36,7 @@ import { AgentBridgeClient, getAgentBridgeManager } from '../../services/hermes/ import { ensureHermesRunWorkspace } from '../../services/hermes/run-chat/workspace' import { isSensitivePath, MAX_EDIT_SIZE } from '../../services/hermes/file-provider' import { readFile, stat as fsStat, writeFile } from 'fs/promises' -import { normalize as pathNormalize, resolve as pathResolve, win32 as pathWin32 } from 'path' +import { normalize as pathNormalize, resolve as pathResolve } from 'path' function getPendingDeletedSessionIds(): Set { return getGroupChatServer()?.getStorage().getPendingDeletedSessionIds() || new Set() @@ -1103,30 +1109,6 @@ export async function usageStats(ctx: any) { } } -function workspaceBaseOverride(): string { - return process.env.WORKSPACE_BASE?.trim() || '' -} - -function useWindowsDriveWorkspaceMode(): boolean { - return process.platform === 'win32' && !workspaceBaseOverride() -} - -function windowsDriveRoot(pathValue: string): string | null { - const match = /^([a-zA-Z]:)[\\/]?$/.exec(pathValue.trim()) - return match ? `${match[1].toUpperCase()}\\` : null -} - -function normalizeWindowsWorkspacePath(inputPath: string): { base: string; fullPath: string } | null { - const raw = String(inputPath || '').trim() - if (!/^[a-zA-Z]:[\\/]/.test(raw)) return null - const fullPath = pathWin32.resolve(raw) - const root = windowsDriveRoot(pathWin32.parse(fullPath).root) - if (!root) return null - const rel = pathWin32.relative(root, fullPath) - if (rel.startsWith('..') || pathWin32.isAbsolute(rel)) return null - return { base: root, fullPath } -} - async function listWindowsWorkspaceDrives() { const { existsSync } = await import('fs') const drives = [] @@ -1143,23 +1125,12 @@ async function listWindowsWorkspaceDrives() { return drives } -async function isWorkspaceListPathAllowed(fullPath: string, basePath: string, statFn: any): Promise { - try { - const info = await statFn(fullPath) - if (!info.isDirectory()) return false - if (process.platform === 'win32') return true - return await isRealPathWithin(fullPath, basePath) - } catch { - return false - } -} - -async function isSafeWorkspaceFolderEntry(entry: any, fullPath: string, basePath: string, statFn: any): Promise { +async function isSafeWorkspaceFolderEntry(entry: any, fullPath: string, basePath: string, statFn: any, options?: { trustWindowsJunctions?: boolean }): Promise { if (!entry.isDirectory() && !(typeof entry.isSymbolicLink === 'function' && entry.isSymbolicLink())) { return false } - return isWorkspaceListPathAllowed(fullPath, basePath, statFn) + return isWorkspaceListPathAllowed(fullPath, basePath, statFn, options) } /** diff --git a/packages/server/src/db/hermes/schemas.ts b/packages/server/src/db/hermes/schemas.ts index e4a5e593fa..fb7f36512e 100644 --- a/packages/server/src/db/hermes/schemas.ts +++ b/packages/server/src/db/hermes/schemas.ts @@ -93,6 +93,8 @@ export const WORKSPACE_RUN_CHANGES_TABLE = 'workspace_run_changes' export const WORKSPACE_RUN_CHANGES_SCHEMA: Record = { change_id: 'TEXT PRIMARY KEY', + room_id: "TEXT NOT NULL DEFAULT ''", + message_id: "TEXT NOT NULL DEFAULT ''", session_id: 'TEXT NOT NULL', run_id: 'TEXT NOT NULL DEFAULT \'\'', source: 'TEXT NOT NULL DEFAULT \'run\'', @@ -131,6 +133,7 @@ export const WORKSPACE_RUN_CHANGE_FILES_SCHEMA: Record = { export const WORKSPACE_RUN_CHANGES_INDEXES = { idx_workspace_run_changes_session: 'CREATE INDEX IF NOT EXISTS idx_workspace_run_changes_session ON workspace_run_changes(session_id, created_at)', idx_workspace_run_changes_run: 'CREATE INDEX IF NOT EXISTS idx_workspace_run_changes_run ON workspace_run_changes(run_id)', + idx_workspace_run_changes_room: 'CREATE INDEX IF NOT EXISTS idx_workspace_run_changes_room ON workspace_run_changes(room_id, created_at)', } export const WORKSPACE_RUN_CHANGE_FILES_INDEXES = { diff --git a/packages/server/src/db/hermes/workspace-run-changes-store.ts b/packages/server/src/db/hermes/workspace-run-changes-store.ts index 448e4c37f3..0584d7acc6 100644 --- a/packages/server/src/db/hermes/workspace-run-changes-store.ts +++ b/packages/server/src/db/hermes/workspace-run-changes-store.ts @@ -27,6 +27,8 @@ export interface WorkspaceRunChangeFileDetail extends WorkspaceRunChangeFileSumm export interface WorkspaceRunChangeSummary { change_id: string + room_id: string + message_id: string session_id: string run_id: string source: 'run' @@ -45,6 +47,8 @@ export interface WorkspaceRunChangeSummary { export interface SaveWorkspaceRunChangeInput { change_id: string + room_id?: string + message_id?: string session_id: string run_id?: string source?: 'run' @@ -101,6 +105,8 @@ function mapFileDetail(row: Record): WorkspaceRunChangeFileDeta function mapSummary(row: Record, files: WorkspaceRunChangeFileSummary[]): WorkspaceRunChangeSummary { return { change_id: String(row.change_id || ''), + room_id: String(row.room_id || ''), + message_id: String(row.message_id || ''), session_id: String(row.session_id || ''), run_id: String(row.run_id || ''), source: 'run', @@ -118,62 +124,86 @@ function mapSummary(row: Record, files: WorkspaceRunChangeFileS } } -export function saveWorkspaceRunChange(change: SaveWorkspaceRunChangeInput): WorkspaceRunChangeSummary | null { - if (!isSqliteAvailable()) return null - const db = getDb() - if (!db) return null +type HermesDb = NonNullable> + +function readWorkspaceRunChange(db: HermesDb, sessionId: string, changeId: string): WorkspaceRunChangeSummary | null { + const row = db.prepare( + `SELECT * FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE session_id = ? AND change_id = ?`, + ).get(sessionId, changeId) as Record | undefined + if (!row) return null + const files = db.prepare( + `SELECT id, change_id, session_id, path, old_path, change_type, additions, deletions, + size_before, size_after, patch_bytes, truncated, binary, created_at + FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} + WHERE session_id = ? AND change_id = ? + ORDER BY path COLLATE NOCASE ASC`, + ).all(sessionId, changeId) as Record[] + return mapSummary(row, files.map(mapFileSummary)) +} +export function insertWorkspaceRunChange(db: HermesDb, change: SaveWorkspaceRunChangeInput): WorkspaceRunChangeSummary | null { const createdAt = Math.floor(Date.now() / 1000) - db.exec('BEGIN') - try { - db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} WHERE change_id = ?`).run(change.change_id) - db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE change_id = ?`).run(change.change_id) - db.prepare( - `INSERT INTO ${WORKSPACE_RUN_CHANGES_TABLE} ( - change_id, session_id, run_id, source, workspace, workspace_kind, started_at, finished_at, - files_changed, additions, deletions, truncated, total_patch_bytes, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} WHERE change_id = ?`).run(change.change_id) + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE change_id = ?`).run(change.change_id) + db.prepare( + `INSERT INTO ${WORKSPACE_RUN_CHANGES_TABLE} ( + change_id, room_id, message_id, session_id, run_id, source, workspace, workspace_kind, started_at, finished_at, + files_changed, additions, deletions, truncated, total_patch_bytes, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + change.change_id, + change.room_id || '', + change.message_id || '', + change.session_id, + change.run_id || '', + change.source || 'run', + change.workspace, + change.workspace_kind || 'git', + change.started_at, + change.finished_at, + change.files_changed, + change.additions, + change.deletions, + change.truncated ? 1 : 0, + change.total_patch_bytes, + createdAt, + ) + + const insertFile = db.prepare( + `INSERT INTO ${WORKSPACE_RUN_CHANGE_FILES_TABLE} ( + change_id, session_id, path, old_path, change_type, additions, deletions, + size_before, size_after, patch, patch_bytes, truncated, binary, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + for (const file of change.files) { + insertFile.run( change.change_id, change.session_id, - change.run_id || '', - change.source || 'run', - change.workspace, - change.workspace_kind || 'git', - change.started_at, - change.finished_at, - change.files_changed, - change.additions, - change.deletions, - change.truncated ? 1 : 0, - change.total_patch_bytes, + file.path, + file.old_path || null, + file.change_type, + file.additions, + file.deletions, + file.size_before ?? null, + file.size_after ?? null, + file.patch || null, + file.patch_bytes, + file.truncated ? 1 : 0, + file.binary ? 1 : 0, createdAt, ) + } + return readWorkspaceRunChange(db, change.session_id, change.change_id) +} - const insertFile = db.prepare( - `INSERT INTO ${WORKSPACE_RUN_CHANGE_FILES_TABLE} ( - change_id, session_id, path, old_path, change_type, additions, deletions, - size_before, size_after, patch, patch_bytes, truncated, binary, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ) - for (const file of change.files) { - insertFile.run( - change.change_id, - change.session_id, - file.path, - file.old_path || null, - file.change_type, - file.additions, - file.deletions, - file.size_before ?? null, - file.size_after ?? null, - file.patch || null, - file.patch_bytes, - file.truncated ? 1 : 0, - file.binary ? 1 : 0, - createdAt, - ) - } +export function saveWorkspaceRunChange(change: SaveWorkspaceRunChangeInput): WorkspaceRunChangeSummary | null { + if (!isSqliteAvailable()) return null + const db = getDb() + if (!db) return null + + db.exec('BEGIN') + try { + insertWorkspaceRunChange(db, change) db.exec('COMMIT') } catch (err) { db.exec('ROLLBACK') @@ -187,18 +217,7 @@ export function getWorkspaceRunChange(sessionId: string, changeId: string): Work if (!isSqliteAvailable()) return null const db = getDb() if (!db) return null - const row = db.prepare( - `SELECT * FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE session_id = ? AND change_id = ?`, - ).get(sessionId, changeId) as Record | undefined - if (!row) return null - const files = db.prepare( - `SELECT id, change_id, session_id, path, old_path, change_type, additions, deletions, - size_before, size_after, patch_bytes, truncated, binary, created_at - FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} - WHERE session_id = ? AND change_id = ? - ORDER BY path COLLATE NOCASE ASC`, - ).all(sessionId, changeId) as Record[] - return mapSummary(row, files.map(mapFileSummary)) + return readWorkspaceRunChange(db, sessionId, changeId) } export function listWorkspaceRunChangesForSession(sessionId: string): WorkspaceRunChangeSummary[] { @@ -263,6 +282,32 @@ export function deleteWorkspaceRunChangesForSession(sessionId: string): void { } } +export function deleteWorkspaceRunChangesForRoom(db: any, roomId: string, beforeTimestamp?: number): void { + const room = String(roomId || '').trim() + if (!room) return + const rows = beforeTimestamp == null + ? db.prepare(`SELECT change_id FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE room_id = ?`).all(room) + : db.prepare( + `SELECT c.change_id + FROM ${WORKSPACE_RUN_CHANGES_TABLE} c + INNER JOIN gc_messages m ON m.id = c.message_id AND m.roomId = c.room_id + WHERE c.room_id = ? AND m.timestamp < ?`, + ).all(room, beforeTimestamp) + const ids = rows.map((row: any) => String(row.change_id || '').trim()).filter(Boolean) + if (!ids.length) return + const placeholders = ids.map(() => '?').join(',') + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} WHERE change_id IN (${placeholders})`).run(...ids) + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE change_id IN (${placeholders})`).run(...ids) +} + +export function deleteWorkspaceRunChangesByChangeIds(db: any, changeIds: string[]): void { + const ids = [...new Set(changeIds.map(id => String(id || '').trim()).filter(Boolean))] + if (!ids.length) return + const placeholders = ids.map(() => '?').join(',') + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGE_FILES_TABLE} WHERE change_id IN (${placeholders})`).run(...ids) + db.prepare(`DELETE FROM ${WORKSPACE_RUN_CHANGES_TABLE} WHERE change_id IN (${placeholders})`).run(...ids) +} + function isOptionalCleanupSqliteError(err: unknown): boolean { return err instanceof Error && /(no such table|database is locked)/i.test(err.message) } diff --git a/packages/server/src/routes/hermes/group-chat.ts b/packages/server/src/routes/hermes/group-chat.ts index 03e48284f7..fd94d37ad9 100644 --- a/packages/server/src/routes/hermes/group-chat.ts +++ b/packages/server/src/routes/hermes/group-chat.ts @@ -503,8 +503,14 @@ groupChatRoutes.delete('/api/hermes/group-chat/rooms/:roomId', async (ctx) => { ctx.body = { error: 'Access denied' } return } - // Disconnect all agents in room - chatServer.agentClients.disconnectRoom(roomId) + // Interrupt active bridge runs, then evict sockets and disconnect agents before deleting persisted data. + try { + await chatServer.deleteRoomRuntimeState(roomId) + } catch (err: any) { + ctx.status = Number(err?.status || 409) + ctx.body = { error: err?.message || 'Room interrupt did not complete' } + return + } // Delete all data storage.deleteRoom(roomId) ctx.body = { success: true } @@ -531,8 +537,14 @@ groupChatRoutes.post('/api/hermes/group-chat/rooms/:roomId/clear-context', async ctx.body = { error: 'Access denied' } return } + try { + await chatServer.clearRoomRuntimeState(roomId) + } catch (err: any) { + ctx.status = Number(err?.status || 409) + ctx.body = { error: err?.message || 'Room interrupt did not complete' } + return + } storage.clearRoomContext(roomId) - chatServer.clearRoomRuntimeState(roomId) ctx.body = { success: true, room: serializeRoom(storage.getRoom(roomId), true) } }) @@ -599,6 +611,15 @@ groupChatRoutes.put('/api/hermes/group-chat/rooms/:roomId/workspace', async (ctx try { const rawWorkspace = workspace.trim() const normalized = rawWorkspace ? (await assertAllowedWorkspaceFolder(rawWorkspace)).fullPath : '' + if (normalized !== String(room.workspace || '')) { + const releaseSessionFence = chatServer.fenceCurrentRoomAgentSessions(roomId) + try { + await chatServer.agentClients.interruptRoom(roomId) + } catch (err) { + releaseSessionFence() + throw err + } + } ctx.body = { room: serializeRoom(storage.updateRoomWorkspace(roomId, normalized), true) } } catch (err: any) { ctx.status = Number(err?.status || 403) diff --git a/packages/server/src/services/hermes/group-chat/agent-clients.ts b/packages/server/src/services/hermes/group-chat/agent-clients.ts index f28e4b6da9..591ce1b7c8 100644 --- a/packages/server/src/services/hermes/group-chat/agent-clients.ts +++ b/packages/server/src/services/hermes/group-chat/agent-clients.ts @@ -1,5 +1,5 @@ import { io, Socket } from 'socket.io-client' -import { randomBytes } from 'crypto' +import { createHash, randomBytes } from 'crypto' import { getToken } from '../../../services/auth' import { logger } from '../../../services/logger' import { updateUsage } from '../../../db/hermes/usage-store' @@ -7,9 +7,14 @@ import { countTokens } from '../../../lib/context-compressor' import { AgentBridgeClient, type AgentBridgeContextEstimate, type AgentBridgeMessage, type AgentBridgeOutput } from '../agent-bridge' import { convertContentBlocksForAgent, isContentBlockArray } from '../run-chat/content-blocks' import { resolveBridgeRunModelConfig } from '../run-chat/model-config' +import { + completeWorkspaceRunCheckpointDraft, + discardWorkspaceRunCheckpoint, + startWorkspaceRunCheckpoint, +} from '../run-chat/workspace-diff-tracker' import type { ContentBlock } from '../run-chat/types' import type { StoredMessage } from '../context-engine/types' -import { buildProjectedGroupChatHistory, projectGroupChatMessage } from './context-projection' +import { buildProjectedGroupChatHistory, isWorkspaceDiffToolMessage, projectGroupChatMessage } from './context-projection' import { sliceGroupMessagesForSnapshotTail } from './group-message-ordering' import { isAllAgentsMentioned, @@ -63,6 +68,22 @@ export function mentionMessageToStoredContextMessage(roomId: string, msg: Mentio type GroupEstimateMessage = { role: 'user' | 'assistant'; content: string } export type GroupModelContext = { model: string; provider: string } +type WorkspaceDiffTerminalStatus = 'completed' | 'failed' | 'aborted' +type WorkspaceDiffBroadcaster = (roomId: string, message: MessageData & Record, totalTokens: number) => void + +function isUnknownBridgeSessionError(err: unknown): boolean { + const message = String((err as any)?.message || err || '').toLowerCase() + return message.includes('unknown session') || message.includes('session not found') +} + +interface WorkspaceDiffRunState { + roomId: string + sessionId: string + runId: string + workspace: string + abortRequested: boolean + finalized: boolean +} interface BridgeContextCache { fixedContextTokens: number @@ -149,6 +170,9 @@ class AgentClient { private pendingToolCallIds = new Map() private pendingToolBaseIds = new Map() private bridgeContextCache = new Map() + private workspaceDiffRuns = new Map() + private interruptVersions = new Map() + private workspaceDiffBroadcaster: WorkspaceDiffBroadcaster | null = null constructor(config: AgentConfig, handlers: AgentEventHandler = {}) { this.agentId = config.agentId || Date.now().toString(36) + Math.random().toString(36).slice(2, 8) @@ -174,6 +198,10 @@ class AgentClient { this.storage = storage } + setWorkspaceDiffBroadcaster(broadcaster: WorkspaceDiffBroadcaster | null): void { + this.workspaceDiffBroadcaster = broadcaster + } + async connect(port?: number): Promise { const actualPort = port ?? parseInt(process.env.PORT || '8648', 10) const token = await getToken() @@ -238,10 +266,10 @@ class AgentClient { }) } - sendMessage(roomId: string, content: string, messageId?: string, extra?: Record): Promise { + sendMessage(roomId: string, content: string, messageId?: string, extra?: Record, agentSessionId?: string): Promise { this.ensureConnected() return new Promise((resolve, reject) => { - this.socket!.emit('message', { roomId, content, id: messageId, ...extra }, (res: { id?: string; error?: string }) => { + this.socket!.emit('message', { roomId, content, id: messageId, ...extra, ...(agentSessionId ? { agentSessionId } : {}) }, (res: { id?: string; error?: string }) => { if (res.error) { reject(new Error(res.error)) } else { @@ -261,9 +289,9 @@ class AgentClient { this.socket!.emit('stop_typing', { roomId }) } - emitContextStatus(roomId: string, status: 'compressing' | 'replying' | 'ready', extra?: Record): void { + emitContextStatus(roomId: string, status: 'compressing' | 'replying' | 'ready', extra?: Record, agentSessionId?: string): void { this.ensureConnected() - this.socket!.emit('context_status', { roomId, agentName: this.name, status, ...extra }) + this.socket!.emit('context_status', { roomId, agentName: this.name, status, ...extra, ...(agentSessionId ? { agentSessionId } : {}) }) } emitApprovalRequested(roomId: string, payload: Record): void { @@ -276,15 +304,40 @@ class AgentClient { this.socket!.emit('approval.resolved', { roomId, agentName: this.name, ...payload }) } - async interrupt(roomId: string): Promise { + async interrupt(roomId: string): Promise { const sessionSeed = String(this.storage?.getRoom?.(roomId)?.sessionSeed || '0') const sessionId = groupBridgeSessionId(roomId, this.profile, this.name, sessionSeed) - await new AgentBridgeClient().interrupt(sessionId, 'Interrupted by group chat user', this.profile) - this.stopTyping(roomId) - this.emitContextStatus(roomId, 'ready') + let result: Awaited> | null = null + try { + result = await new AgentBridgeClient().interrupt(sessionId, 'Interrupted by group chat user', this.profile) + } catch (err) { + if (!isUnknownBridgeSessionError(err)) throw err + logger.info(`[AgentClients] ${this.name}: bridge session ${sessionId} was already idle/missing during interrupt`) + } + const synced = result?.synced !== false + if (!synced) return false + this.markSessionInterrupted(sessionId) + const abortedStates = this.markWorkspaceDiffAborted(roomId) + try { + for (const state of abortedStates) { + await this.finalizeWorkspaceDiffOnce(state, 'aborted', null) + } + } finally { + try { + this.stopTyping(roomId) + } catch (err: any) { + logger.warn(`[AgentClients] ${this.name}: failed to emit stop_typing after interrupt: ${err.message || err}`) + } + try { + this.emitContextStatus(roomId, 'ready', undefined, sessionId) + } catch (err: any) { + logger.warn(`[AgentClients] ${this.name}: failed to emit ready status after interrupt: ${err.message || err}`) + } + } + return true } - emitMessageStreamStart(roomId: string, messageId: string): void { + emitMessageStreamStart(roomId: string, messageId: string, agentSessionId?: string): void { this.ensureConnected() this.socket!.emit('message_stream_start', { roomId, @@ -292,24 +345,25 @@ class AgentClient { senderId: this.socket?.id || this.agentId, senderName: this.name, timestamp: Date.now(), + ...(agentSessionId ? { agentSessionId } : {}), }) } - emitMessageStreamDelta(roomId: string, messageId: string, delta: string): void { + emitMessageStreamDelta(roomId: string, messageId: string, delta: string, agentSessionId?: string): void { if (!delta) return this.ensureConnected() - this.socket!.emit('message_stream_delta', { roomId, id: messageId, delta }) + this.socket!.emit('message_stream_delta', { roomId, id: messageId, delta, ...(agentSessionId ? { agentSessionId } : {}) }) } - emitMessageReasoningDelta(roomId: string, messageId: string, delta: string): void { + emitMessageReasoningDelta(roomId: string, messageId: string, delta: string, agentSessionId?: string): void { if (!delta) return this.ensureConnected() - this.socket!.emit('message_reasoning_delta', { roomId, id: messageId, delta }) + this.socket!.emit('message_reasoning_delta', { roomId, id: messageId, delta, ...(agentSessionId ? { agentSessionId } : {}) }) } - emitMessageStreamEnd(roomId: string, messageId: string): void { + emitMessageStreamEnd(roomId: string, messageId: string, agentSessionId?: string): void { this.ensureConnected() - this.socket!.emit('message_stream_end', { roomId, id: messageId }) + this.socket!.emit('message_stream_end', { roomId, id: messageId, ...(agentSessionId ? { agentSessionId } : {}) }) } getJoinedRooms(): string[] { @@ -416,6 +470,106 @@ class AgentClient { } } + private workspaceDiffKey(roomId: string, sessionId: string, runId: string): string { + return `${roomId}\u0000${sessionId}\u0000${runId}` + } + + private beginWorkspaceDiffIfNeeded(args: { roomId: string; sessionId: string; runId: string; workspace: string }): WorkspaceDiffRunState | null { + if (!args.workspace) return null + startWorkspaceRunCheckpoint({ + sessionId: args.sessionId, + runId: args.runId, + workspace: args.workspace, + }) + const state: WorkspaceDiffRunState = { ...args, abortRequested: false, finalized: false } + this.workspaceDiffRuns.set(this.workspaceDiffKey(args.roomId, args.sessionId, args.runId), state) + return state + } + + private discardWorkspaceDiffRun(state: WorkspaceDiffRunState | null): void { + if (!state) return + this.workspaceDiffRuns.delete(this.workspaceDiffKey(state.roomId, state.sessionId, state.runId)) + discardWorkspaceRunCheckpoint({ sessionId: state.sessionId, runId: state.runId }) + } + + private interruptVersion(sessionId: string): number { + return this.interruptVersions.get(sessionId) || 0 + } + + private markSessionInterrupted(sessionId: string): void { + this.interruptVersions.set(sessionId, this.interruptVersion(sessionId) + 1) + } + + private replySessionIsCurrent(roomId: string, sessionId: string, interruptVersion: number): boolean { + return this.roomSessionIsCurrent(roomId, sessionId) && this.interruptVersion(sessionId) === interruptVersion + } + + private roomSessionIsCurrent(roomId: string, sessionId: string): boolean { + const room = this.storage?.getRoom?.(roomId) + if (!room) return false + const seed = String(room.sessionSeed || '0') + return groupBridgeSessionId(roomId, this.profile, this.name, seed) === sessionId + } + + private markWorkspaceDiffAborted(roomId: string): WorkspaceDiffRunState[] { + const aborted: WorkspaceDiffRunState[] = [] + for (const state of this.workspaceDiffRuns.values()) { + if (state.roomId === roomId) { + state.abortRequested = true + aborted.push(state) + } + } + return aborted + } + + private async finalizeWorkspaceDiffOnce( + state: WorkspaceDiffRunState | null, + status: WorkspaceDiffTerminalStatus, + parentMessageId?: string | null, + ): Promise { + if (!state) return + const key = this.workspaceDiffKey(state.roomId, state.sessionId, state.runId) + const current = this.workspaceDiffRuns.get(key) + if (!current || current.finalized) return + if (!this.roomSessionIsCurrent(current.roomId, current.sessionId)) { + this.discardWorkspaceDiffRun(current) + return + } + current.finalized = true + this.workspaceDiffRuns.delete(key) + const finalStatus = current.abortRequested ? 'aborted' : status + let draft + try { + draft = completeWorkspaceRunCheckpointDraft({ + sessionId: current.sessionId, + runId: current.runId, + workspace: current.workspace, + }) + } catch (err) { + logger.warn({ err, roomId: current.roomId, sessionId: current.sessionId, runId: current.runId }, '[GroupChat] failed to complete workspace diff draft') + return + } + if (!draft) return + try { + const saved = this.storage?.saveWorkspaceDiffMessageForRun?.({ + roomId: current.roomId, + senderId: this.agentId, + senderName: this.name, + sessionId: current.sessionId, + runId: current.runId, + status: finalStatus, + workspace: current.workspace, + draft, + parentMessageId, + }) + if (saved?.message) { + this.workspaceDiffBroadcaster?.(current.roomId, saved.message, saved.totalTokens) + } + } catch (err) { + logger.warn({ err, roomId: current.roomId, sessionId: current.sessionId, runId: current.runId }, '[GroupChat] failed to persist workspace diff message') + } + } + // ─── Hermes Agent Bridge Integration ─────────────────────── /** @@ -436,6 +590,12 @@ class AgentClient { let totalContent = '' let reasoningContent = '' let streamStarted = false + let bridgeStarted = false + let workspaceRunState: WorkspaceDiffRunState | null = null + let activeSessionId = '' + let activeReplyInterruptVersion = 0 + let staleStartedRunStopped = false + let stopStaleStartedRun: ((reason?: string) => Promise) | null = null try { // Notify room that agent is typing this.startTyping(roomId) @@ -446,6 +606,48 @@ class AgentClient { const bridge = new AgentBridgeClient() const sessionSeed = String(this.storage?.getRoom?.(roomId)?.sessionSeed || '0') const sessionId = groupBridgeSessionId(roomId, this.profile, this.name, sessionSeed) + const replyInterruptVersion = this.interruptVersion(sessionId) + const reportStatus = (status: 'compressing' | 'replying' | 'ready', extra?: Record) => { + onStatus?.(status, { ...extra, agentSessionId: sessionId }) + } + activeSessionId = sessionId + activeReplyInterruptVersion = replyInterruptVersion + stopStaleStartedRun = async (reason = 'Interrupted because group chat room state changed') => { + if (staleStartedRunStopped) return + staleStartedRunStopped = true + if (bridgeStarted) { + let destroySession = false + try { + const result = await bridge.interrupt(sessionId, reason, this.profile) + destroySession = result?.synced === false + } catch (err: any) { + destroySession = true + logger.warn(`[AgentClients] ${this.name}: failed to interrupt stale bridge run: ${err.message || err}`) + } + if (destroySession) { + try { + await bridge.destroy(sessionId, this.profile) + } catch (err: any) { + logger.warn(`[AgentClients] ${this.name}: failed to destroy stale bridge session: ${err.message || err}`) + } + } + if (streamStarted) { + try { + this.emitMessageStreamEnd(roomId, streamMessageId, sessionId) + } catch (err: any) { + logger.warn(`[AgentClients] ${this.name}: failed to end stale stream: ${err.message || err}`) + } + } + } + this.discardWorkspaceDiffRun(workspaceRunState) + workspaceRunState = null + try { + this.stopTyping(roomId) + } catch (err: any) { + logger.warn(`[AgentClients] ${this.name}: failed to stop typing after stale bridge run: ${err.message || err}`) + } + reportStatus('ready') + } const modelContext = await resolveGroupAgentModelContext(this.profile) if (this.contextEngine && this.storage) { @@ -479,7 +681,7 @@ class AgentClient { compression, profile: this.profile, onProgress: (event: { status: 'compressing'; messageCount: number; tokenCount: number }) => { - onStatus?.('compressing', { + reportStatus('compressing', { messageCount: event.messageCount, totalTokens: event.tokenCount, }) @@ -496,17 +698,21 @@ class AgentClient { ) }, }) + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } conversationHistory = ctx.conversationHistory instructions = ctx.instructions if (typeof ctx.meta.contextTokenEstimate === 'number' && Number.isFinite(ctx.meta.contextTokenEstimate)) { this.storage.updateRoomTotalTokens?.(roomId, ctx.meta.contextTokenEstimate) - onStatus?.('replying', { totalTokens: ctx.meta.contextTokenEstimate }) + reportStatus('replying', { totalTokens: ctx.meta.contextTokenEstimate }) } logger.debug(`[AgentClients] ${this.name}: context built — historyLen=${conversationHistory.length}, meta=%j`, ctx.meta) - onStatus?.('replying') + reportStatus('replying') } catch (err: any) { logger.warn(`[AgentClients] ${this.name}: context engine failed: ${err.message}`) - onStatus?.('replying') + reportStatus('replying') // Degrade: continue without context } } @@ -530,6 +736,10 @@ class AgentClient { const bridgeInput: AgentBridgeMessage = isContentBlockArray(input) ? await convertContentBlocksForAgent(input) : input + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } const flushedAssistantParts = new Set() let lastChunk: AgentBridgeOutput | null = null const roomWorkspace = String(this.storage?.getRoom?.(roomId)?.workspace || '').trim() @@ -546,43 +756,74 @@ class AgentClient { ...(roomWorkspace ? { workspace: roomWorkspace } : {}), }, ) + bridgeStarted = true + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } + if (roomWorkspace) { + workspaceRunState = this.beginWorkspaceDiffIfNeeded({ + roomId, + sessionId, + runId: started.run_id, + workspace: roomWorkspace, + }) + } - this.emitMessageStreamStart(roomId, streamMessageId) + this.emitMessageStreamStart(roomId, streamMessageId, sessionId) streamStarted = true for await (const chunk of bridge.streamOutput(started.run_id, { timeoutMs: 120000 })) { + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } lastChunk = chunk - reasoningContent += await this.recordBridgeEvents(roomId, sessionId, instructions, modelContext, chunk, () => streamMessageId, async () => { + reasoningContent += await this.recordBridgeEvents(roomId, sessionId, replyInterruptVersion, instructions, modelContext, chunk, () => streamMessageId, async () => { const toolBaseId = streamMessageId if (currentContent.trim()) { + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + currentContent = '' + return toolBaseId + } await this.sendMessage(roomId, currentContent, streamMessageId, { role: 'assistant', mentionDepth: nextMentionDepth(msg), reasoning: reasoningContent || null, reasoning_content: reasoningContent || null, - }) + }, sessionId) flushedAssistantParts.add(streamMessageId) currentContent = '' } - this.emitMessageStreamEnd(roomId, toolBaseId) + this.emitMessageStreamEnd(roomId, toolBaseId, sessionId) partIndex += 1 streamMessageId = groupMessagePartId(runMessageId, partIndex) - this.emitMessageStreamStart(roomId, streamMessageId) + this.emitMessageStreamStart(roomId, streamMessageId, sessionId) streamStarted = true return toolBaseId }) + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } if (chunk.delta) { currentContent += chunk.delta totalContent += chunk.delta - this.emitMessageStreamDelta(roomId, streamMessageId, chunk.delta) + this.emitMessageStreamDelta(roomId, streamMessageId, chunk.delta, sessionId) } } if (lastChunk?.status === 'error') { logger.error(`[AgentClients] ${this.name}: bridge response failed: ${lastChunk.error || 'unknown error'}`) - await this.sendAgentErrorMessage(roomId, streamMessageId, lastChunk.error || 'Run failed', msg, reasoningContent) - this.emitMessageStreamEnd(roomId, streamMessageId) + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } + await this.sendAgentErrorMessage(roomId, streamMessageId, lastChunk.error || 'Run failed', msg, reasoningContent, sessionId) + await this.finalizeWorkspaceDiffOnce(workspaceRunState, 'failed', streamStarted ? streamMessageId : null) + this.emitMessageStreamEnd(roomId, streamMessageId, sessionId) this.stopTyping(roomId) - onStatus?.('ready') + reportStatus('ready') return } @@ -590,35 +831,62 @@ class AgentClient { currentContent = extractBridgeFinalText(lastChunk) totalContent = currentContent } + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } recordBridgeUsage(roomId, this.profile, lastChunk?.result) logger.debug(`[AgentClients] ${this.name}: bridge response completed, content length=${totalContent.length}`) if (currentContent) { + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } this.stopTyping(roomId) await this.sendMessage(roomId, currentContent, streamMessageId, { role: 'assistant', mentionDepth: nextMentionDepth(msg), reasoning: reasoningContent || null, reasoning_content: reasoningContent || null, - }) - this.emitMessageStreamEnd(roomId, streamMessageId) + }, sessionId) + this.emitMessageStreamEnd(roomId, streamMessageId, sessionId) + await this.finalizeWorkspaceDiffOnce(workspaceRunState, 'completed', streamMessageId) await this.refreshRoomFullContextEstimate(roomId, sessionId, bridge, instructions, modelContext) - onStatus?.('ready') + reportStatus('ready') return } logger.warn(`[AgentClients] ${this.name}: bridge response completed without content`) - this.emitMessageStreamEnd(roomId, streamMessageId) + if (!this.replySessionIsCurrent(roomId, sessionId, replyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } + this.emitMessageStreamEnd(roomId, streamMessageId, sessionId) + await this.finalizeWorkspaceDiffOnce(workspaceRunState, 'completed', streamStarted ? streamMessageId : null) this.stopTyping(roomId) - onStatus?.('ready') + reportStatus('ready') } catch (err: any) { logger.error(`[AgentClients] ${this.name}: error handling message: ${err.message}`) + if (activeSessionId && !this.replySessionIsCurrent(roomId, activeSessionId, activeReplyInterruptVersion)) { + await stopStaleStartedRun?.() + return + } + if (workspaceRunState && !bridgeStarted) { + await stopStaleStartedRun?.('Interrupted after group chat bridge launch failed') + } else { + await this.finalizeWorkspaceDiffOnce(workspaceRunState, 'failed', streamStarted ? streamMessageId : null) + } try { - await this.sendAgentErrorMessage(roomId, streamMessageId, err, msg, reasoningContent) - if (streamStarted) this.emitMessageStreamEnd(roomId, streamMessageId) + await this.sendAgentErrorMessage(roomId, streamMessageId, err, msg, reasoningContent, activeSessionId || undefined) + if (streamStarted) this.emitMessageStreamEnd(roomId, streamMessageId, activeSessionId || undefined) } catch (sendErr: any) { logger.warn(`[AgentClients] ${this.name}: failed to send error message: ${sendErr.message}`) } this.stopTyping(roomId) - onStatus?.('ready') + if (activeSessionId) { + onStatus?.('ready', { agentSessionId: activeSessionId }) + } else { + onStatus?.('ready') + } } } @@ -642,9 +910,10 @@ class AgentClient { 'final', ) if (cachedTokens == null || cachedTokens <= 0) return + if (!this.roomSessionIsCurrent(roomId, sessionId)) return const rounded = Math.floor(cachedTokens) this.storage.updateRoomTotalTokens?.(roomId, rounded) - this.emitContextStatus(roomId, 'replying', { totalTokens: rounded }) + this.emitContextStatus(roomId, 'replying', { totalTokens: rounded }, sessionId) } catch (err: any) { logger.warn(`[GroupChat] failed to refresh final context estimate room=${roomId} agent=${this.name}: ${err.message}`) } @@ -657,7 +926,9 @@ class AgentClient { const tail = sliceGroupMessagesForSnapshotTail(messages, snapshot.lastMessageId).messages return buildProjectedGroupChatHistory(snapshot.summary, tail, { agentId: this.agentId, socketId: this.socket?.id, name: this.name }) } - return messages.map((message: any) => this.mapRoomMessageForEstimate(message)) + return messages + .filter((message: any) => !isWorkspaceDiffToolMessage(message)) + .map((message: any) => this.mapRoomMessageForEstimate(message)) } private mapRoomMessageForEstimate(message: any): { role: 'user' | 'assistant'; content: string } { @@ -670,6 +941,7 @@ class AgentClient { error: unknown, sourceMsg: MentionMessage, reasoningContent = '', + sessionId?: string, ): Promise { const detail = error instanceof Error ? error.message : String(error || 'Run failed') const content = detail.startsWith('Error:') ? detail : `Error: ${detail}` @@ -679,12 +951,13 @@ class AgentClient { finish_reason: 'error', reasoning: reasoningContent || null, reasoning_content: reasoningContent || null, - }) + }, sessionId) } private async recordBridgeEvents( roomId: string, sessionId: string, + interruptVersion: number, instructions: string | undefined, modelContext: GroupModelContext, chunk: AgentBridgeOutput, @@ -693,17 +966,21 @@ class AgentClient { ): Promise { let reasoning = '' for (const ev of chunk.events || []) { + if (!this.replySessionIsCurrent(roomId, sessionId, interruptVersion)) return reasoning const eventType = String((ev as any)?.event || '') if (eventType === 'bridge.context.ready') { this.cacheBridgeContext(sessionId, ev as Record, instructions, modelContext) } else if (eventType === 'tool.started') { const toolBaseId = await beforeToolStarted() - this.recordToolStarted(roomId, ev as Record, toolBaseId) + if (!this.replySessionIsCurrent(roomId, sessionId, interruptVersion)) return reasoning + this.recordToolStarted(roomId, sessionId, ev as Record, toolBaseId) } else if (eventType === 'tool.completed') { - this.recordToolCompleted(roomId, ev as Record) + if (!this.replySessionIsCurrent(roomId, sessionId, interruptVersion)) return reasoning + this.recordToolCompleted(roomId, sessionId, ev as Record) } else if (eventType === 'approval.requested') { this.emitApprovalRequested(roomId, { event: 'approval.requested', + agentSessionId: sessionId, approval_id: (ev as any).approval_id, command: (ev as any).command, description: (ev as any).description, @@ -713,6 +990,7 @@ class AgentClient { } else if (eventType === 'approval.resolved') { this.emitApprovalResolved(roomId, { event: 'approval.resolved', + agentSessionId: sessionId, approval_id: (ev as any).approval_id, choice: (ev as any).choice, }) @@ -720,14 +998,14 @@ class AgentClient { const text = groupBridgeReasoningDeltaFromEvent(ev as Record) if (text) { reasoning += text - this.emitMessageReasoningDelta(roomId, getCurrentMessageId(), text) + this.emitMessageReasoningDelta(roomId, getCurrentMessageId(), text, sessionId) } } } return reasoning } - private recordToolStarted(roomId: string, ev: Record, runMessageId: string): void { + private recordToolStarted(roomId: string, sessionId: string, ev: Record, runMessageId: string): void { const toolName = String(ev.tool_name || ev.tool || ev.name || '') const toolCallId = groupToolCallId(ev.tool_call_id, toolName, this.nextToolIndex(roomId, toolName)) this.trackPendingToolCall(roomId, toolName, toolCallId) @@ -759,10 +1037,10 @@ class AgentClient { tool_calls: msg.tool_calls, finish_reason: 'tool_calls', timestamp, - }).catch((err: any) => logger.warn(`[AgentClients] failed to record tool call: ${err.message}`)) + }, sessionId).catch((err: any) => logger.warn(`[AgentClients] failed to record tool call: ${err.message}`)) } - private recordToolCompleted(roomId: string, ev: Record): void { + private recordToolCompleted(roomId: string, sessionId: string, ev: Record): void { const toolName = String(ev.tool_name || ev.tool || ev.name || '') const rawId = String(ev.tool_call_id || '').trim() const toolCallId = rawId || this.takePendingToolCall(roomId, toolName) || groupToolCallId(null, toolName, this.nextToolIndex(roomId, toolName)) @@ -786,7 +1064,7 @@ class AgentClient { tool_call_id: toolCallId, tool_name: toolName || null, timestamp, - }).catch((err: any) => logger.warn(`[AgentClients] failed to record tool result: ${err.message}`)) + }, sessionId).catch((err: any) => logger.warn(`[AgentClients] failed to record tool result: ${err.message}`)) } private pendingToolKey(roomId: string, toolName: string): string { @@ -852,9 +1130,12 @@ class AgentClient { } } -function groupBridgeSessionId(roomId: string, profile: string, name: string, sessionSeed: string): string { - const raw = `gc_${roomId}_${profile}_${name}_${sessionSeed || '0'}` - return raw.replace(/[^a-zA-Z0-9_-]/g, '_').slice(0, 120) +export function groupBridgeSessionId(roomId: string, profile: string, name: string, sessionSeed: string): string { + const rawKey = `gc_${roomId}_${profile}_${name}_${sessionSeed || '0'}` + const safePrefix = rawKey.replace(/[^a-zA-Z0-9_-]/g, '_') + const keyHash = createHash('sha256').update(rawKey).digest('hex').slice(0, 16) + const suffix = `_h_${keyHash}` + return `${safePrefix.slice(0, Math.max(0, 120 - suffix.length))}${suffix}` } function groupMessageId(roomId: string, profile: string, name: string): string { @@ -921,10 +1202,12 @@ export class AgentClients { private rooms = new Map>() private _contextEngine: any = null private _storage: any = null + private _workspaceDiffBroadcaster: WorkspaceDiffBroadcaster | null = null // Per-room processing lock + mention queue private _processingRooms = new Set() private _mentionQueue = new Map>() + private _pausedRooms = new Set() /** * Create an agent client and connect it to the server. @@ -937,6 +1220,7 @@ export class AgentClients { // Auto-apply stored references (fixes propagation for agents created after set*) if (this._contextEngine) client.setContextEngine(this._contextEngine) if (this._storage) client.setStorage(this._storage) + client.setWorkspaceDiffBroadcaster(this._workspaceDiffBroadcaster) logger.info(`[AgentClients] Connected: ${client.name} (${client.agentId})`) return client @@ -1030,11 +1314,56 @@ export class AgentClients { return Promise.all(agents.map((agent) => agent.sendMessage(roomId, content))) } + private buildUnsyncedInterruptError(roomId: string): Error { + const err = new Error(`Room "${roomId}" still has running bridge sessions; try again after the interrupt completes`) as Error & { status?: number } + err.status = 409 + return err + } + + private mentionQueueKeysForRoom(roomId: string): string[] { + return Array.from(this._mentionQueue.keys()).filter(key => key === roomId || key.startsWith(`${roomId}:`)) + } + + private clearMentionQueuesForRoom(roomId: string): void { + for (const key of this.mentionQueueKeysForRoom(roomId)) this._mentionQueue.delete(key) + } + + private queueMention(agentKey: string, agent: AgentClient, msg: MentionMessage): void { + let queue = this._mentionQueue.get(agentKey) + if (!queue) { + queue = [] + this._mentionQueue.set(agentKey, queue) + } + queue.push({ agent, msg }) + } + async interruptAgent(roomId: string, agentName: string): Promise { const agent = this.getAgents(roomId).find(a => a.name === agentName) if (!agent) throw new Error(`Agent "${agentName}" not found in room "${roomId}"`) + const synced = await agent.interrupt(roomId) + if (!synced) throw this.buildUnsyncedInterruptError(roomId) this._mentionQueue.delete(`${roomId}:${agent.name}`) - await agent.interrupt(roomId) + } + + async interruptRoom(roomId: string): Promise { + const agents = this.getAgents(roomId) + this._pausedRooms.add(roomId) + const results = await Promise.allSettled(agents.map(agent => agent.interrupt(roomId))) + let unsynced = false + for (const result of results) { + if (result.status === 'rejected') { + unsynced = true + logger.warn(`[AgentClients] failed to interrupt room ${roomId}: ${result.reason?.message || result.reason}`) + } else if (result.value === false) { + unsynced = true + logger.warn(`[AgentClients] bridge interrupt for room ${roomId} was not synchronized`) + } + } + this._pausedRooms.delete(roomId) + if (unsynced) { + throw this.buildUnsyncedInterruptError(roomId) + } + this.clearMentionQueuesForRoom(roomId) } /** @@ -1046,6 +1375,8 @@ export class AgentClients { room.forEach((client) => client.disconnect()) this.rooms.delete(roomId) + this.clearMentionQueuesForRoom(roomId) + this._pausedRooms.delete(roomId) logger.info(`[AgentClients] All agents disconnected from room: ${roomId}`) // Invalidate context engine cache for this room @@ -1055,10 +1386,8 @@ export class AgentClients { } resetRoomContext(roomId: string): void { - this._mentionQueue.delete(roomId) - for (const key of Array.from(this._mentionQueue.keys())) { - if (key.startsWith(`${roomId}:`)) this._mentionQueue.delete(key) - } + this.clearMentionQueuesForRoom(roomId) + this._pausedRooms.delete(roomId) for (const key of Array.from(this._processingRooms)) { if (key.startsWith(`${roomId}:`)) this._processingRooms.delete(key) } @@ -1098,6 +1427,13 @@ export class AgentClients { }) } + setWorkspaceDiffBroadcaster(broadcaster: WorkspaceDiffBroadcaster | null): void { + this._workspaceDiffBroadcaster = broadcaster + this.rooms.forEach((room) => { + room.forEach((client) => client.setWorkspaceDiffBroadcaster(broadcaster)) + }) + } + /** * Server-side: parse @mentions and forward to matching agents directly. @@ -1126,14 +1462,13 @@ export class AgentClients { msg: MentionMessage, ): Promise { const agentKey = `${roomId}:${agent.name}` + if (this._pausedRooms.has(roomId)) { + this.queueMention(agentKey, agent, msg) + logger.debug(`[AgentClients] room ${roomId} is interrupting, queued mention for agent ${agent.name}`) + return + } if (this._processingRooms.has(agentKey)) { - // Queue for this specific agent - let queue = this._mentionQueue.get(agentKey) - if (!queue) { - queue = [] - this._mentionQueue.set(agentKey, queue) - } - queue.push({ agent, msg }) + this.queueMention(agentKey, agent, msg) logger.debug(`[AgentClients] agent ${agent.name} is processing, queued mention in room ${roomId}`) return } @@ -1148,7 +1483,9 @@ export class AgentClients { await agent.replyToMention(roomId, msg, onStatus) } finally { this._processingRooms.delete(agentKey) - await this._drainQueue(agentKey, roomId) + if (!this._pausedRooms.has(roomId)) { + await this._drainQueue(agentKey, roomId) + } } } diff --git a/packages/server/src/services/hermes/group-chat/context-projection.ts b/packages/server/src/services/hermes/group-chat/context-projection.ts index 0052fa6769..44e162919b 100644 --- a/packages/server/src/services/hermes/group-chat/context-projection.ts +++ b/packages/server/src/services/hermes/group-chat/context-projection.ts @@ -13,6 +13,10 @@ export type GroupChatProjectionAgent = { name: string } +export function isWorkspaceDiffToolMessage(message: Pick): boolean { + return String(message.role || '') === 'tool' && String(message.tool_name || '') === 'workspace_diff' +} + export function projectGroupChatMessage( message: ProjectableGroupChatMessage, ownAgent: GroupChatProjectionAgent, @@ -69,7 +73,9 @@ export function buildProjectedGroupChatHistory( ) } - history.push(...messages.map((message) => projectGroupChatMessage(message, ownAgent))) + history.push(...messages + .filter((message) => !isWorkspaceDiffToolMessage(message)) + .map((message) => projectGroupChatMessage(message, ownAgent))) return history } diff --git a/packages/server/src/services/hermes/group-chat/index.ts b/packages/server/src/services/hermes/group-chat/index.ts index 8cfc7f990a..28699a1ce6 100644 --- a/packages/server/src/services/hermes/group-chat/index.ts +++ b/packages/server/src/services/hermes/group-chat/index.ts @@ -1,13 +1,15 @@ import { Server, Socket, Namespace } from 'socket.io' import type { Server as HttpServer } from 'http' +import { basename } from 'path' import { logger } from '../../../services/logger' import { getDb } from '../../../db' import { normalizeMessageContentForStorage, normalizeMessageContentForStorageRole } from '../../../db/hermes/message-content' -import { AgentClients, GROUP_CHAT_AGENT_SOCKET_SECRET } from './agent-clients' +import { AgentClients, GROUP_CHAT_AGENT_SOCKET_SECRET, groupBridgeSessionId } from './agent-clients' import { ContextEngine } from '../context-engine/compressor' import { SessionDeleter } from '../session-deleter' import { countTokens, SUMMARY_PREFIX } from '../../../lib/context-compressor' import { AgentBridgeClient } from '../agent-bridge' +import { insertWorkspaceRunChange, deleteWorkspaceRunChangesForRoom, type SaveWorkspaceRunChangeInput, type WorkspaceRunChangeSummary } from '../../../db/hermes/workspace-run-changes-store' import { authenticateUserToken, isAuthEnabled, type AuthenticatedUser } from '../../../middleware/user-auth' import { findUserByUsername, getUserAvatar } from '../../../db/hermes/users-store' import { config } from '../../../config' @@ -32,6 +34,7 @@ interface ChatMessage { reasoning_details?: string | null reasoning_content?: string | null mentionDepth?: number + agentSessionId?: string } function contentToStorageString(content: unknown): string { @@ -89,6 +92,17 @@ interface RoomInfo { ownerAuthUserId: number | null } +interface SaveWorkspaceDiffMessageArgs { + roomId: string + senderId: string + senderName: string + sessionId: string + runId: string + status: 'completed' | 'failed' | 'aborted' + workspace: string + draft: SaveWorkspaceRunChangeInput + parentMessageId?: string | null +} interface Member { id: string @@ -442,7 +456,9 @@ class ChatStorage { getMessagesForContext(roomId: string, cutoff?: GroupMessageCursorCutoff): ChatMessage[] { const rows = (this.db()?.prepare( - 'SELECT id, roomId, senderId, senderName, content, timestamp, role, tool_call_id, tool_calls, tool_name, finish_reason, reasoning, reasoning_details, reasoning_content FROM gc_messages WHERE roomId = ?' + `SELECT id, roomId, senderId, senderName, content, timestamp, role, tool_call_id, tool_calls, tool_name, finish_reason, reasoning, reasoning_details, reasoning_content + FROM gc_messages + WHERE roomId = ? AND COALESCE(tool_name, '') <> 'workspace_diff'` ).all(roomId) || []) as any[] return sliceGroupMessagesCanonical(rows.map(row => this.mapStoredMessageRow(row)), cutoff).messages } @@ -498,13 +514,103 @@ class ChatStorage { ) } + saveWorkspaceDiffMessageForRun(args: SaveWorkspaceDiffMessageArgs): { message: ChatMessage; totalTokens: number; change: WorkspaceRunChangeSummary } | null { + const db = this.db() + if (!db) return null + const idPrefix = 'gcmsg_workspace_diff_' + const runIdPart = args.runId.replace(/[^a-zA-Z0-9_-]/g, '_').slice(-64) || 'run' + const roomIdBudget = Math.max(24, 180 - idPrefix.length - runIdPart.length - 1) + const roomIdPart = args.roomId.replace(/[^a-zA-Z0-9_-]/g, '_').slice(0, roomIdBudget) || 'room' + const messageId = `${idPrefix}${roomIdPart}_${runIdPart}` + db.exec('BEGIN IMMEDIATE') + try { + const roomExists = db.prepare('SELECT 1 FROM gc_rooms WHERE id = ?').get(args.roomId) + if (!roomExists) { + db.exec('ROLLBACK') + return null + } + const workspaceLabel = basename(args.workspace) || 'workspace' + const redactedDraft: SaveWorkspaceRunChangeInput = { + ...args.draft, + room_id: args.roomId, + message_id: messageId, + workspace: workspaceLabel, + } + const change = insertWorkspaceRunChange(db, redactedDraft) + if (!change) { + db.exec('ROLLBACK') + return null + } + const files = change.files.map((file) => { + const draftFile = redactedDraft.files.find(candidate => candidate.path === file.path && candidate.change_type === file.change_type) + return { + id: file.id, + path: file.path, + change_type: file.change_type, + additions: file.additions, + deletions: file.deletions, + patch: draftFile?.patch || null, + binary: file.binary, + truncated: file.truncated, + } + }) + const payload = { + kind: 'workspace_diff', + version: 1, + room_id: args.roomId, + session_id: args.sessionId, + run_id: args.runId, + status: args.status, + change_id: change.change_id, + workspace_basename: workspaceLabel, + files_changed: change.files_changed, + additions: change.additions, + deletions: change.deletions, + truncated: change.truncated, + files, + ...(args.parentMessageId ? { parent_message_id: args.parentMessageId } : {}), + } + const message: ChatMessage = { + id: messageId, + roomId: args.roomId, + senderId: args.senderId, + senderName: args.senderName, + content: JSON.stringify(payload), + timestamp: Date.now(), + role: 'tool', + tool_call_id: `workspace_diff:${args.runId}`, + tool_calls: null, + tool_name: 'workspace_diff', + } + this.upsertMessage(message) + this.pruneMessages(args.roomId) + const messages = this.getMessagesForContext(args.roomId) + const totalTokens = this.estimateRoomTotalTokens(args.roomId, messages) + this.updateRoomTotalTokens(args.roomId, totalTokens) + db.exec('COMMIT') + return { message, totalTokens, change } + } catch (err) { + try { db.exec('ROLLBACK') } catch { /* ignore */ } + throw err + } + } + saveMessageAndRefreshRoom(msg: ChatMessage, options: { preserveExistingTimestamp?: boolean } = {}): { message: ChatMessage; totalTokens: number } { const db = this.db() if (!db) return { message: msg, totalTokens: 0 } db.exec('BEGIN IMMEDIATE') try { const existing = this.getMessage(msg.id) - const message = existing && options.preserveExistingTimestamp ? { ...msg, timestamp: existing.timestamp } : msg + if (existing?.tool_name === 'workspace_diff') { + const messages = this.getMessagesForContext(existing.roomId) + const totalTokens = this.estimateRoomTotalTokens(existing.roomId, messages) + db.exec('COMMIT') + return { message: existing, totalTokens } + } + const safeMsg = msg.tool_name === 'workspace_diff' + ? { ...msg, role: 'user', tool_call_id: null, tool_calls: null, tool_name: null } + : msg + const message = existing && options.preserveExistingTimestamp ? { ...safeMsg, timestamp: existing.timestamp } : safeMsg this.upsertMessage(message) this.pruneMessages(msg.roomId) const messages = this.getMessagesForContext(msg.roomId) @@ -518,12 +624,36 @@ class ChatStorage { } } + private deleteWorkspaceDiffChanges(roomId: string, beforeTimestamp?: number): void { + const db = this.db() + if (!db) return + deleteWorkspaceRunChangesForRoom(db, roomId, beforeTimestamp) + } + + private withImmediateTransaction(db: any, fn: () => void): void { + if (db.inTransaction || db.isTransaction) { + fn() + return + } + db.exec('BEGIN IMMEDIATE') + try { + fn() + db.exec('COMMIT') + } catch (err) { + try { db.exec('ROLLBACK') } catch { /* ignore */ } + throw err + } + } + clearRoomContext(roomId: string): void { const db = this.db() if (!db) return - db.prepare('DELETE FROM gc_messages WHERE roomId = ?').run(roomId) - db.prepare('DELETE FROM gc_context_snapshots WHERE roomId = ?').run(roomId) - db.prepare('UPDATE gc_rooms SET totalTokens = 0, sessionSeed = ? WHERE id = ?').run(`${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}`, roomId) + this.withImmediateTransaction(db, () => { + this.deleteWorkspaceDiffChanges(roomId) + db.prepare('DELETE FROM gc_messages WHERE roomId = ?').run(roomId) + db.prepare('DELETE FROM gc_context_snapshots WHERE roomId = ?').run(roomId) + db.prepare('UPDATE gc_rooms SET totalTokens = 0, sessionSeed = ? WHERE id = ?').run(`${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}`, roomId) + }) } pruneMessages(roomId: string, keep = 500): void { @@ -535,8 +665,11 @@ class ChatStorage { 'SELECT timestamp FROM gc_messages WHERE roomId = ? ORDER BY timestamp DESC LIMIT 1 OFFSET ?' ).get(roomId, keep - 1) as any if (cutoff) { - const result = db.prepare('DELETE FROM gc_messages WHERE roomId = ? AND timestamp < ?').run(roomId, cutoff.timestamp) - logger.info(`[GroupChat] pruned ${result.changes} messages from room ${roomId} (had ${count}, keeping ${keep})`) + this.withImmediateTransaction(db, () => { + this.deleteWorkspaceDiffChanges(roomId, cutoff.timestamp) + const result = db.prepare('DELETE FROM gc_messages WHERE roomId = ? AND timestamp < ?').run(roomId, cutoff.timestamp) + logger.info(`[GroupChat] pruned ${result.changes} messages from room ${roomId} (had ${count}, keeping ${keep})`) + }) } } } @@ -594,11 +727,14 @@ class ChatStorage { deleteRoom(roomId: string): void { const db = this.db() if (!db) return - db.prepare('DELETE FROM gc_messages WHERE roomId = ?').run(roomId) - db.prepare('DELETE FROM gc_room_agents WHERE roomId = ?').run(roomId) - db.prepare('DELETE FROM gc_room_members WHERE roomId = ?').run(roomId) - db.prepare('DELETE FROM gc_context_snapshots WHERE roomId = ?').run(roomId) - db.prepare('DELETE FROM gc_rooms WHERE id = ?').run(roomId) + this.withImmediateTransaction(db, () => { + this.deleteWorkspaceDiffChanges(roomId) + db.prepare('DELETE FROM gc_messages WHERE roomId = ?').run(roomId) + db.prepare('DELETE FROM gc_room_agents WHERE roomId = ?').run(roomId) + db.prepare('DELETE FROM gc_room_members WHERE roomId = ?').run(roomId) + db.prepare('DELETE FROM gc_context_snapshots WHERE roomId = ?').run(roomId) + db.prepare('DELETE FROM gc_rooms WHERE id = ?').run(roomId) + }) } // ─── Room Members ────────────────────────────────────── @@ -787,6 +923,8 @@ export class GroupChatServer { private typingState = new Map }>>() /** roomId -> (agentName -> { agentName, status }) */ private contextStatusState = new Map>() + /** roomId -> blocked Bridge session ids from room-level interrupts/rotations. */ + private fencedRoomAgentSessions = new Map>() constructor(httpServers: HttpServer | HttpServer[]) { this.storage = new ChatStorage() @@ -838,6 +976,10 @@ export class GroupChatServer { }) this.agentClients.setContextEngine(contextEngine) this.agentClients.setStorage(this.storage) + this.agentClients.setWorkspaceDiffBroadcaster((roomId, msg, totalTokens) => { + this.nsp.to(roomId).emit('message', msg) + this.nsp.to(roomId).emit('room_updated', { roomId, totalTokens }) + }) this._contextEngine = contextEngine // Restore agent connections — call restoreAgents() after server is listening @@ -860,18 +1002,75 @@ export class GroupChatServer { return Array.from(this.rooms.keys()) } - clearRoomRuntimeState(roomId: string): void { + fenceCurrentRoomAgentSessions(roomId: string): () => void { + const room = typeof this.storage.getRoom === 'function' ? this.storage.getRoom(roomId) : undefined + if (!room) return () => {} + const ids = new Set() + for (const agent of this.storage.getRoomAgents(roomId) || []) { + ids.add(groupBridgeSessionId(roomId, agent.profile, agent.name, String(room.sessionSeed || '0'))) + } + if (!ids.size) return () => {} + if (!this.fencedRoomAgentSessions) this.fencedRoomAgentSessions = new Map>() + let fenced = this.fencedRoomAgentSessions.get(roomId) + if (!fenced) { + fenced = new Set() + this.fencedRoomAgentSessions.set(roomId, fenced) + } + for (const id of ids) fenced.add(id) + let released = false + return () => { + if (released) return + released = true + const current = this.fencedRoomAgentSessions.get(roomId) + if (!current) return + for (const id of ids) current.delete(id) + if (!current.size) this.fencedRoomAgentSessions.delete(roomId) + } + } + + private isRoomAgentSessionFenced(roomId: string, sessionId: string): boolean { + return this.fencedRoomAgentSessions?.get(roomId)?.has(sessionId) === true + } + + async clearRoomRuntimeState(roomId: string): Promise { const roomTyping = this.typingState.get(roomId) if (roomTyping) { for (const entry of roomTyping.values()) clearTimeout(entry.timer) this.typingState.delete(roomId) } this.contextStatusState.delete(roomId) + const releaseSessionFence = this.fenceCurrentRoomAgentSessions(roomId) + try { + await this.agentClients.interruptRoom(roomId) + } catch (err) { + releaseSessionFence() + throw err + } this.agentClients.resetRoomContext(roomId) this.nsp.to(roomId).emit('room_cleared', { roomId, totalTokens: 0 }) this.nsp.to(roomId).emit('room_updated', { roomId, totalTokens: 0 }) } + async deleteRoomRuntimeState(roomId: string): Promise { + const roomTyping = this.typingState.get(roomId) + if (roomTyping) { + for (const entry of roomTyping.values()) clearTimeout(entry.timer) + this.typingState.delete(roomId) + } + this.contextStatusState.delete(roomId) + const releaseSessionFence = this.fenceCurrentRoomAgentSessions(roomId) + try { + await this.agentClients.interruptRoom(roomId) + } catch (err) { + releaseSessionFence() + throw err + } + this.agentClients.disconnectRoom(roomId) + this.rooms.delete(roomId) + this.nsp.in(roomId).socketsLeave(roomId) + this.fencedRoomAgentSessions?.delete(roomId) + } + // ─── Restore Agents ───────────────────────────────────────── /** @@ -953,7 +1152,7 @@ export class GroupChatServer { logger.debug(`[GroupChat] Connected: ${userName} (socket=${socket.id}, user=${userId})`) - socket.on('join', (data: { roomId?: string; name?: string; description?: string; inviteCode?: string }, ack?: (response?: unknown) => void) => this.handleJoin(socket, data, ack)) + socket.on('join', (data: { roomId?: string; name?: string }, ack?: (response?: unknown) => void) => this.handleJoin(socket, data, ack)) socket.on('message', (data: Partial & { roomId?: string; content: string | Array>; id?: string; mentionDepth?: number }, ack?: (response?: unknown) => void) => this.handleMessage(socket, data, ack)) socket.on('message_stream_start', (data: { roomId?: string; id?: string; senderId?: string; senderName?: string; timestamp?: number }) => this.handleMessageStreamStart(socket, data)) socket.on('message_stream_delta', (data: { roomId?: string; id?: string; delta?: string }) => this.handleMessageStreamDelta(socket, data)) @@ -1022,6 +1221,33 @@ export class GroupChatServer { } } + private agentSessionIsCurrent(roomId: string, member: Member | undefined, agentSessionId: unknown): boolean { + const sessionId = typeof agentSessionId === 'string' ? agentSessionId.trim() : '' + if (!sessionId || member?.source !== 'agent') return false + const room = typeof this.storage.getRoom === 'function' ? this.storage.getRoom(roomId) : undefined + if (!room) return false + const roomAgent = this.storage.getRoomAgentByAgentId(roomId, member.userId) + if (!roomAgent) return false + const expected = groupBridgeSessionId(roomId, roomAgent.profile, roomAgent.name, String(room.sessionSeed || '0')) + if (sessionId !== expected) return false + return !this.isRoomAgentSessionFenced(roomId, sessionId) + } + + private canPersistAgentMessageForCurrentSession(roomId: string, member: Member | undefined, data: Partial): boolean { + if (member?.source !== 'agent') return true + const role = normalizeMessageRole(data.role) + const isRunTrace = role === 'assistant' || role === 'tool' || Array.isArray(data.tool_calls) || Boolean(data.tool_call_id) + if (!isRunTrace) return true + return this.agentSessionIsCurrent(roomId, member, data.agentSessionId) + } + + private getCurrentAgentEventMember(socket: Socket, roomId: string, agentName: string, agentSessionId?: unknown): Member | null { + const joined = this.getOnlineRoomMember(socket, roomId) + if (!joined || joined.member.source !== 'agent') return null + if (agentName && joined.member.name !== agentName) return null + if (!this.agentSessionIsCurrent(roomId, joined.member, agentSessionId)) return null + return joined.member + } private handleJoin(socket: Socket, data: { roomId?: string; name?: string; description?: string; inviteCode?: string }, ack?: (res: any) => void): void { const socketId = socket.id @@ -1140,6 +1366,10 @@ export class GroupChatServer { } const member = room.getOnlineMemberBySocketId(socketId) + if (!this.canPersistAgentMessageForCurrentSession(roomId, member, data)) { + ack?.({ error: 'Stale room session' }) + return + } const userId = member?.userId || socketId const userName = member?.name || `User-${socketId.slice(0, 6)}` const role = normalizeMessageRole(data.role) @@ -1193,18 +1423,18 @@ export class GroupChatServer { } } - private handleMessageStreamStart(socket: Socket, data: { roomId?: string; id?: string; senderId?: string; senderName?: string; timestamp?: number }): void { + private handleMessageStreamStart(socket: Socket, data: { roomId?: string; id?: string; senderId?: string; senderName?: string; timestamp?: number; agentSessionId?: string }): void { const roomId = data.roomId || 'general' - const joined = this.getOnlineRoomMember(socket, roomId) - if (!joined || joined.member.source !== 'agent') return + const member = this.getCurrentAgentEventMember(socket, roomId, '', data.agentSessionId) + if (!member) return const id = this.normalizeClientMessageId(data.id) if (!id) return this.nsp.to(roomId).emit('message_stream_start', { id, roomId, - senderId: joined.member.userId, - senderName: joined.member.name, + senderId: member.userId, + senderName: member.name, content: '', timestamp: data.timestamp || Date.now(), role: 'assistant', @@ -1212,10 +1442,9 @@ export class GroupChatServer { }) } - private handleMessageStreamDelta(socket: Socket, data: { roomId?: string; id?: string; delta?: string }): void { + private handleMessageStreamDelta(socket: Socket, data: { roomId?: string; id?: string; delta?: string; agentSessionId?: string }): void { const roomId = data.roomId || 'general' - const joined = this.getOnlineRoomMember(socket, roomId) - if (!joined || joined.member.source !== 'agent') return + if (!this.getCurrentAgentEventMember(socket, roomId, '', data.agentSessionId)) return const id = this.normalizeClientMessageId(data.id) if (!id || !data.delta) return this.nsp.to(roomId).emit('message_stream_delta', { @@ -1225,10 +1454,9 @@ export class GroupChatServer { }) } - private handleMessageReasoningDelta(socket: Socket, data: { roomId?: string; id?: string; delta?: string }): void { + private handleMessageReasoningDelta(socket: Socket, data: { roomId?: string; id?: string; delta?: string; agentSessionId?: string }): void { const roomId = data.roomId || 'general' - const joined = this.getOnlineRoomMember(socket, roomId) - if (!joined || joined.member.source !== 'agent') return + if (!this.getCurrentAgentEventMember(socket, roomId, '', data.agentSessionId)) return const id = this.normalizeClientMessageId(data.id) if (!id || !data.delta) return this.nsp.to(roomId).emit('message_reasoning_delta', { @@ -1238,10 +1466,9 @@ export class GroupChatServer { }) } - private handleMessageStreamEnd(socket: Socket, data: { roomId?: string; id?: string }): void { + private handleMessageStreamEnd(socket: Socket, data: { roomId?: string; id?: string; agentSessionId?: string }): void { const roomId = data.roomId || 'general' - const joined = this.getOnlineRoomMember(socket, roomId) - if (!joined || joined.member.source !== 'agent') return + if (!this.getCurrentAgentEventMember(socket, roomId, '', data.agentSessionId)) return const id = this.normalizeClientMessageId(data.id) if (!id) return this.nsp.to(roomId).emit('message_stream_end', { roomId, id }) @@ -1294,12 +1521,13 @@ export class GroupChatServer { }) } - private handleContextStatus(socket: Socket, data: { roomId?: string; agentName?: string; status?: string; totalTokens?: number }): void { + private handleContextStatus(socket: Socket, data: { roomId?: string; agentName?: string; status?: string; totalTokens?: number; agentSessionId?: string }): void { const roomId = data.roomId || 'general' const agentName = data.agentName || '' const status = data.status || '' - if (!agentName || !this.isAgentEventSocket(socket, roomId, agentName)) return + const agentMember = this.getCurrentAgentEventMember(socket, roomId, agentName, data.agentSessionId) + if (!agentName || !agentMember) return let roomStatuses = this.contextStatusState.get(roomId) if (!roomStatuses) { @@ -1353,10 +1581,10 @@ export class GroupChatServer { } } - private handleApprovalRequested(socket: Socket, data: { roomId?: string; agentName?: string; approval_id?: string; command?: string; description?: string; choices?: string[]; allow_permanent?: boolean }): void { + private handleApprovalRequested(socket: Socket, data: { roomId?: string; agentName?: string; approval_id?: string; command?: string; description?: string; choices?: string[]; allow_permanent?: boolean; agentSessionId?: string }): void { const roomId = data.roomId const agentName = data.agentName || '' - if (!roomId || !data.approval_id || !this.isAgentEventSocket(socket, roomId, agentName)) return + if (!roomId || !data.approval_id || !this.getCurrentAgentEventMember(socket, roomId, agentName, data.agentSessionId)) return this.emitToRoomManagers(roomId, 'approval.requested', { event: 'approval.requested', roomId, @@ -1369,10 +1597,10 @@ export class GroupChatServer { }) } - private handleApprovalResolved(socket: Socket, data: { roomId?: string; agentName?: string; approval_id?: string; choice?: string }): void { + private handleApprovalResolved(socket: Socket, data: { roomId?: string; agentName?: string; approval_id?: string; choice?: string; agentSessionId?: string }): void { const roomId = data.roomId const agentName = data.agentName || '' - if (!roomId || !data.approval_id || !this.isAgentEventSocket(socket, roomId, agentName)) return + if (!roomId || !data.approval_id || !this.getCurrentAgentEventMember(socket, roomId, agentName, data.agentSessionId)) return this.emitToRoomManagers(roomId, 'approval.resolved', { event: 'approval.resolved', roomId, diff --git a/packages/server/src/services/hermes/run-chat/workspace-diff-tracker.ts b/packages/server/src/services/hermes/run-chat/workspace-diff-tracker.ts index 414b7bab7a..e990f2bcc4 100644 --- a/packages/server/src/services/hermes/run-chat/workspace-diff-tracker.ts +++ b/packages/server/src/services/hermes/run-chat/workspace-diff-tracker.ts @@ -4,7 +4,7 @@ import { existsSync, mkdtempSync, readdirSync, readFileSync, realpathSync, rmSyn import { tmpdir } from 'os' import { basename, extname, join, relative, resolve, sep } from 'path' import { logger } from '../../logger' -import { saveWorkspaceRunChange, type WorkspaceRunChangeSummary } from '../../../db/hermes/workspace-run-changes-store' +import { saveWorkspaceRunChange, type SaveWorkspaceRunChangeInput, type WorkspaceRunChangeSummary } from '../../../db/hermes/workspace-run-changes-store' const MAX_TRACKED_STATUS_PATHS = 20_000 const MAX_CHANGED_FILES = 80 @@ -616,11 +616,21 @@ export function startWorkspaceRunCheckpoint(args: { }) } -export function completeWorkspaceRunCheckpoint(args: { +export function discardWorkspaceRunCheckpoint(args: { + sessionId: string + runId?: string | null +}): void { + const runId = args.runId || '' + if (!runId) return + const key = checkpointKey(args.sessionId, runId) + checkpoints.delete(key) +} + +export function completeWorkspaceRunCheckpointDraft(args: { sessionId: string runId?: string | null workspace?: string | null -}): WorkspaceRunChangeSummary | null { +}): SaveWorkspaceRunChangeInput | null { const runId = args.runId || '' if (!runId) return null const key = checkpointKey(args.sessionId, runId) @@ -678,7 +688,7 @@ export function completeWorkspaceRunCheckpoint(args: { } if (files.length === 0) return null - return saveWorkspaceRunChange({ + return { change_id: checkpoint.changeId, session_id: checkpoint.sessionId, run_id: runId || checkpoint.runId, @@ -693,5 +703,14 @@ export function completeWorkspaceRunCheckpoint(args: { truncated, total_patch_bytes: totalPatchBytes, files, - }) + } +} + +export function completeWorkspaceRunCheckpoint(args: { + sessionId: string + runId?: string | null + workspace?: string | null +}): WorkspaceRunChangeSummary | null { + const draft = completeWorkspaceRunCheckpointDraft(args) + return draft ? saveWorkspaceRunChange(draft) : null } diff --git a/tests/client/group-chat-store-streaming.test.ts b/tests/client/group-chat-store-streaming.test.ts index c63d942132..00ddb7e2c4 100644 --- a/tests/client/group-chat-store-streaming.test.ts +++ b/tests/client/group-chat-store-streaming.test.ts @@ -64,6 +64,7 @@ const room: RoomInfo = { id: 'room-1', name: 'Test Room', inviteCode: 'ROOM1', + workspace: '', } function assistantMessage(overrides: Partial): ChatMessage { diff --git a/tests/client/group-chat-workspace-diff.test.ts b/tests/client/group-chat-workspace-diff.test.ts new file mode 100644 index 0000000000..b9bf12342f --- /dev/null +++ b/tests/client/group-chat-workspace-diff.test.ts @@ -0,0 +1,206 @@ +// @vitest-environment jsdom +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { mount } from '@vue/test-utils' +import { createPinia, setActivePinia } from 'pinia' +import GroupMessageItem from '@/components/hermes/group-chat/GroupMessageItem.vue' +import GroupMessageList from '@/components/hermes/group-chat/GroupMessageList.vue' +import type { ChatMessage } from '@/api/hermes/group-chat' + +const toolTraceVisibleState = vi.hoisted(() => ({ value: true })) + +const groupChatApiMock = vi.hoisted(() => { + const socket: any = { + id: 'socket-1', + connected: true, + on: vi.fn(() => socket), + once: vi.fn(() => socket), + off: vi.fn(() => socket), + emit: vi.fn((event: string, _payload: unknown, ack?: Function) => { + if (event === 'join') { + ack?.({ + roomId: 'room-1', + roomName: 'Room 1', + messages: [], + agents: [], + members: [], + typingUsers: [], + contextStatuses: [], + }) + } + return socket + }), + } + return { + socket, + connectGroupChat: vi.fn(() => socket), + disconnectGroupChat: vi.fn(), + getSocket: vi.fn(() => socket), + getStoredUserId: vi.fn(() => 'user-1'), + getStoredUserName: vi.fn(() => 'tester'), + createRoom: vi.fn(), + listRooms: vi.fn(), + getRoomDetail: vi.fn(), + joinRoomByCode: vi.fn(), + addAgent: vi.fn(), + listAgents: vi.fn(), + removeAgent: vi.fn(), + cloneRoom: vi.fn(), + deleteRoom: vi.fn(), + clearRoomContext: vi.fn(), + updateRoomWorkspace: vi.fn(), + } +}) + +vi.mock('@/api/hermes/group-chat', () => groupChatApiMock) +vi.mock('@/api/client', () => ({ + getApiKey: vi.fn(() => 'token'), + getActiveProfileName: vi.fn(() => 'default'), + getStoredUsername: vi.fn(() => null), +})) +vi.mock('@/api/auth', () => ({ fetchCurrentUser: vi.fn(async () => { throw new Error('no user') }) })) +vi.mock('@/api/hermes/download', () => ({ getDownloadUrl: vi.fn((path: string) => `/download?path=${path}`) })) +vi.mock('@/composables/useToolTraceVisibility', () => ({ + useToolTraceVisibility: () => ({ toolTraceVisible: toolTraceVisibleState, toggleToolTraceVisible: vi.fn() }), +})) +vi.mock('vue-i18n', () => ({ useI18n: () => ({ t: (key: string) => key }) })) +vi.mock('naive-ui', () => ({ + useMessage: () => ({ error: vi.fn(), success: vi.fn(), warning: vi.fn(), info: vi.fn() }), +})) + +const payload = { + kind: 'workspace_diff', + version: 1, + room_id: 'room-1', + session_id: 'session-1', + run_id: '0123456789abcdef0123456789abcdef', + status: 'completed', + change_id: 'change-1', + workspace_basename: 'repo', + workspace: '/tmp/repo', + files_changed: 2, + additions: 3, + deletions: 1, + truncated: false, + files: [ + { id: 1, path: 'src/a.ts', change_type: 'modified', additions: 2, deletions: 1, patch: 'diff --git a/src/a.ts b/src/a.ts\n-old\n+new\n', binary: false, truncated: false }, + { id: 2, path: 'asset.bin', change_type: 'added', additions: 0, deletions: 0, patch: null, binary: true, truncated: false }, + ], +} + +function workspaceDiffMessage(overrides: Partial = {}): ChatMessage { + return { + id: 'diff-1', + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + content: JSON.stringify(payload), + timestamp: 1, + role: 'tool', + tool_call_id: `workspace_diff:${payload.run_id}`, + tool_name: 'workspace_diff', + ...overrides, + } +} + +describe('group chat workspace diff client rendering', () => { + beforeEach(() => { + setActivePinia(createPinia()) + toolTraceVisibleState.value = true + vi.clearAllMocks() + Object.defineProperty(window, 'speechSynthesis', { + configurable: true, + value: { addEventListener: vi.fn(), removeEventListener: vi.fn(), getVoices: vi.fn(() => []), speak: vi.fn(), cancel: vi.fn(), pause: vi.fn(), resume: vi.fn() }, + }) + }) + + it('maps persisted workspace_diff tool JSON to a structured tool result', async () => { + groupChatApiMock.getRoomDetail.mockResolvedValue({ + room: { id: 'room-1', name: 'Room 1', inviteCode: null, workspace: '/tmp/repo' }, + messages: [workspaceDiffMessage()], + agents: [], + members: [], + }) + const { useGroupChatStore } = await import('@/stores/hermes/group-chat') + const store = useGroupChatStore() + + await store.joinRoom('room-1') + + expect(store.sortedMessages[0]).toMatchObject({ + role: 'tool', + toolName: 'workspace_diff', + toolResult: expect.objectContaining({ kind: 'workspace_diff', files_changed: 2 }), + }) + }) + + it('renders a workspace diff card instead of raw JSON details', () => { + const wrapper = mount(GroupMessageItem, { + props: { + message: { + ...workspaceDiffMessage(), + toolName: 'workspace_diff', + toolResult: payload, + toolStatus: 'done', + }, + agents: [], + members: [], + currentUserId: 'user-1', + }, + global: { stubs: { MarkdownRenderer: true, ProfileAvatar: true } }, + }) + + expect(wrapper.find('.workspace-diff-card').exists()).toBe(true) + expect(wrapper.text()).toContain('chat.workspaceChanges') + expect(wrapper.text()).toContain('src/a.ts') + expect(wrapper.find('.tool-line').exists()).toBe(false) + expect(wrapper.text()).not.toContain('"kind"') + }) + + it('keeps workspace diff audit cards visible when generic tool traces are hidden', async () => { + toolTraceVisibleState.value = false + const { useGroupChatStore } = await import('@/stores/hermes/group-chat') + const store = useGroupChatStore() + store.currentRoomId = 'room-1' + store.messages = [ + { + ...workspaceDiffMessage(), + toolName: 'workspace_diff', + toolResult: payload, + toolStatus: 'done', + }, + { + id: 'tool-1', + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + content: '{}', + timestamp: 2, + role: 'tool', + tool_name: 'shell', + toolName: 'shell', + toolStatus: 'done', + } as ChatMessage, + ] + + const wrapper = mount(GroupMessageList, { + global: { + stubs: { + GroupMessageItem: true, + VirtualMessageList: { + name: 'VirtualMessageList', + props: ['messages'], + methods: { + scrollToBottom() {}, + isNearBottom() { return true }, + captureScrollPosition() { return null }, + restoreScrollPosition() {}, + }, + template: '
', + }, + }, + }, + }) + + const messages = wrapper.getComponent({ name: 'VirtualMessageList' }).props('messages') as ChatMessage[] + expect(messages.map(message => message.id)).toEqual(['diff-1']) + }) +}) diff --git a/tests/server/group-chat-agent-routing-baseline.test.ts b/tests/server/group-chat-agent-routing-baseline.test.ts index c53f348b16..f8b3c2bf5b 100644 --- a/tests/server/group-chat-agent-routing-baseline.test.ts +++ b/tests/server/group-chat-agent-routing-baseline.test.ts @@ -4,7 +4,7 @@ import { createTestGroupChatServer, emitAck, } from './group-chat-test-helpers' -import { GROUP_CHAT_AGENT_SOCKET_SECRET } from '../../packages/server/src/services/hermes/group-chat/agent-clients' +import { GROUP_CHAT_AGENT_SOCKET_SECRET, groupBridgeSessionId } from '../../packages/server/src/services/hermes/group-chat/agent-clients' import { authenticateUserToken, isAuthEnabled } from '../../packages/server/src/middleware/user-auth' import type { GroupChatServer } from '../../packages/server/src/services/hermes/group-chat' @@ -40,6 +40,11 @@ describe('group chat agent routing baseline', () => { return { human, agent } } + function currentAgentSessionId() { + const room = groupServer.getStorage().getRoom('room-1') + return groupBridgeSessionId('room-1', 'default', 'Worker', String(room?.sessionSeed || '0')) + } + it('routes human messages through mention processing', async () => { const { human } = await joinHumanAndAgent() const processMentions = vi.spyOn(groupServer.agentClients, 'processMentions').mockResolvedValue(undefined) @@ -84,6 +89,7 @@ describe('group chat agent routing baseline', () => { content: '@Worker chain handoff', role: 'assistant', mentionDepth: 3, + agentSessionId: currentAgentSessionId(), }) expect(processMentions).toHaveBeenCalledWith('room-1', expect.objectContaining({ @@ -103,6 +109,7 @@ describe('group chat agent routing baseline', () => { content: '@Worker stop looping', role: 'assistant', mentionDepth: 4, + agentSessionId: currentAgentSessionId(), }) expect(processMentions).not.toHaveBeenCalled() diff --git a/tests/server/group-chat-agent-workspace.test.ts b/tests/server/group-chat-agent-workspace.test.ts index 6a84d7331d..f4ab31d844 100644 --- a/tests/server/group-chat-agent-workspace.test.ts +++ b/tests/server/group-chat-agent-workspace.test.ts @@ -1,5 +1,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' +const order = vi.hoisted(() => [] as string[]) + const mockSocket = vi.hoisted(() => ({ id: 'agent-socket-1', connected: true, @@ -16,7 +18,8 @@ const mockSocket = vi.hoisted(() => ({ })) const bridgeMock = vi.hoisted(() => ({ - chat: vi.fn(async (_sessionId: string, _input: unknown, _history: unknown, _instructions: unknown, _profile: unknown, _options: any) => { + chat: vi.fn(async (_sessionId: string) => { + order.push('chat') return { ok: true, run_id: 'bridge-run-id', session_id: _sessionId, status: 'running' } }), streamOutput: vi.fn(async function* (runId: string) { @@ -35,6 +38,13 @@ const bridgeMock = vi.hoisted(() => ({ }), contextEstimate: vi.fn(), interrupt: vi.fn(), + destroy: vi.fn(), +})) + +const trackerMock = vi.hoisted(() => ({ + startWorkspaceRunCheckpoint: vi.fn(() => order.push('checkpoint')), + completeWorkspaceRunCheckpointDraft: vi.fn(() => null), + discardWorkspaceRunCheckpoint: vi.fn(), })) vi.mock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })) @@ -46,11 +56,163 @@ vi.mock('../../packages/server/src/db/hermes/usage-store', () => ({ updateUsage: vi.mock('../../packages/server/src/services/hermes/agent-bridge', () => ({ AgentBridgeClient: vi.fn(() => bridgeMock), })) +vi.mock('../../packages/server/src/services/hermes/run-chat/workspace-diff-tracker', () => trackerMock) describe('group chat agent workspace bridge runs', () => { beforeEach(() => { + order.length = 0 vi.clearAllMocks() - bridgeMock.chat.mockResolvedValue({ ok: true, run_id: 'bridge-run-id', session_id: 'session-1', status: 'running' }) + trackerMock.completeWorkspaceRunCheckpointDraft.mockReset() + trackerMock.completeWorkspaceRunCheckpointDraft.mockReturnValue(null) + bridgeMock.chat.mockImplementation(async (_sessionId: string) => { + order.push('chat') + return { ok: true, run_id: 'bridge-run-id', session_id: _sessionId, status: 'running' } + }) + bridgeMock.streamOutput.mockImplementation(async function* (runId: string) { + yield { + ok: true, + run_id: runId, + session_id: 'session-1', + status: 'complete', + delta: 'done', + cursor: 1, + output: 'done', + done: true, + events: [], + event_cursor: 0, + } + }) + bridgeMock.interrupt.mockResolvedValue(undefined) + }) + + function workspaceDraft(runId: string, sessionId = 'session-1') { + return { + session_id: sessionId, + run_id: runId, + workspace: '/tmp/workspace', + files_changed: 1, + additions: 1, + deletions: 0, + truncated: false, + files: [{ + path: 'file.txt', + change_type: 'modified', + additions: 1, + deletions: 0, + size_before: 3, + size_after: 4, + patch: '+new', + binary: false, + truncated: false, + }], + } + } + + async function workerSessionId(seed = 'seed-1') { + const { groupBridgeSessionId } = await import('../../packages/server/src/services/hermes/group-chat/agent-clients') + return groupBridgeSessionId('room-1', 'default', 'Worker', seed) + } + + it('keeps the session key freshness suffix when long names force bridge session id truncation', async () => { + const { groupBridgeSessionId } = await import('../../packages/server/src/services/hermes/group-chat/agent-clients') + const longAgentName = 'Worker'.repeat(40) + + const first = groupBridgeSessionId('room-1', 'default', longAgentName, 'seed-1') + const second = groupBridgeSessionId('room-1', 'default', longAgentName, 'seed-2') + const roomA = `room-${'a'.repeat(130)}` + const roomB = `room-${'a'.repeat(129)}b` + const collidingPrefixA = groupBridgeSessionId(roomA, 'default', longAgentName, '0') + const collidingPrefixB = groupBridgeSessionId(roomB, 'default', longAgentName, '0') + + const nonAsciiA = groupBridgeSessionId('room-1', 'default', '丫鬟', '0') + const nonAsciiB = groupBridgeSessionId('room-1', 'default', '书童', '0') + + expect(first).toHaveLength(120) + expect(second).toHaveLength(120) + expect(first).not.toBe(second) + expect(first).toMatch(/_h_[0-9a-f]{16}$/) + expect(second).toMatch(/_h_[0-9a-f]{16}$/) + expect(collidingPrefixA).not.toBe(collidingPrefixB) + expect(nonAsciiA).not.toBe(nonAsciiB) + }) + + it('does not block room-wide interrupts for idle agents with no bridge session', async () => { + bridgeMock.interrupt.mockRejectedValueOnce(new Error('unknown session')) + const { AgentClients } = await import('../../packages/server/src/services/hermes/group-chat/agent-clients') + const clients = new AgentClients() + const client = await clients.createAgent({ + agentId: 'agent-1', + profile: 'default', + name: 'Worker', + description: '', + invited: 0, + } as any) as any + const storage = { getRoom: vi.fn(() => ({ sessionSeed: 'seed-1', workspace: '' })) } + client.setStorage(storage as any) + ;(clients as any).rooms.set('room-1', new Map([[client.agentId, client]])) + + const sessionId = await workerSessionId() + + await expect(clients.interruptRoom('room-1')).resolves.toBeUndefined() + expect(bridgeMock.interrupt).toHaveBeenCalledWith(sessionId, 'Interrupted by group chat user', 'default') + }) + + it('does not drain queued mentions while a room interrupt is still pending', async () => { + let finishStream!: () => void + let finishInterrupt!: () => void + bridgeMock.streamOutput.mockImplementation(async function* (runId: string) { + await new Promise(resolve => { finishStream = resolve }) + yield { + ok: true, + run_id: runId, + session_id: 'session-1', + status: 'complete', + delta: 'done', + cursor: 1, + output: 'done', + done: true, + events: [], + event_cursor: 0, + } + }) + bridgeMock.interrupt.mockImplementationOnce(async () => { + await new Promise(resolve => { finishInterrupt = resolve }) + return { ok: true, synced: true } + }) + const client = await createClient('/tmp/workspace') + const clients = client.__testClients + const waitFor = async (predicate: () => boolean) => { + for (let i = 0; i < 30; i += 1) { + if (predicate()) return + await new Promise(resolve => setTimeout(resolve, 0)) + } + throw new Error('timed out waiting for condition') + } + + await clients.processMentions('room-1', { + content: '@Worker first', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + await waitFor(() => bridgeMock.chat.mock.calls.length === 1) + await clients.processMentions('room-1', { + content: '@Worker second', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 2, + }) + + const interruptPromise = clients.interruptRoom('room-1') + await waitFor(() => bridgeMock.interrupt.mock.calls.length === 1) + finishStream() + await new Promise(resolve => setTimeout(resolve, 0)) + + expect(bridgeMock.chat).toHaveBeenCalledTimes(1) + finishInterrupt() + await interruptPromise + await new Promise(resolve => setTimeout(resolve, 0)) + expect(bridgeMock.chat).toHaveBeenCalledTimes(1) }) async function createClient(workspace = '') { @@ -65,12 +227,15 @@ describe('group chat agent workspace bridge runs', () => { } as any) const storage = { getRoom: vi.fn(() => ({ sessionSeed: 'seed-1', workspace })), + saveWorkspaceDiffMessageForRun: vi.fn(), + updateRoomTotalTokens: vi.fn(), getMessagesForContext: vi.fn(() => []), getContextSnapshot: vi.fn(() => null), - saveSessionProfile: vi.fn(), - updateRoomTotalTokens: vi.fn(), } client.setStorage(storage as any) + ;(clients as any).rooms.set('room-1', new Map([[client.agentId, client]])) + ;(client as any).__testStorage = storage + ;(client as any).__testClients = clients return client as any } @@ -84,18 +249,27 @@ describe('group chat agent workspace bridge runs', () => { timestamp: 1, }) + expect(trackerMock.startWorkspaceRunCheckpoint).not.toHaveBeenCalled() expect(bridgeMock.chat).toHaveBeenCalledWith( expect.any(String), expect.anything(), expect.any(Array), expect.any(String), 'default', - expect.not.objectContaining({ workspace: expect.anything() }), + expect.not.objectContaining({ workspace: expect.anything(), run_id: expect.anything() }), ) }) - it('passes the normalized room workspace to the bridge run options', async () => { + it('cancels a pending reply when interrupt arrives before bridge.chat starts', async () => { + bridgeMock.interrupt.mockRejectedValueOnce(new Error('unknown session')) const client = await createClient('/tmp/workspace') + client.__testStorage.getRoomMembers = vi.fn(() => []) + client.setContextEngine({ + buildContext: vi.fn(async () => { + await client.interrupt('room-1') + return { conversationHistory: [], instructions: 'ctx', meta: {} } + }), + }) await client.replyToMention('room-1', { content: '@Worker hi', @@ -104,13 +278,238 @@ describe('group chat agent workspace bridge runs', () => { timestamp: 1, }) - expect(bridgeMock.chat).toHaveBeenCalledWith( - expect.any(String), - expect.anything(), - expect.any(Array), - expect.any(String), + const sessionId = await workerSessionId() + expect(bridgeMock.interrupt).toHaveBeenCalledWith(sessionId, 'Interrupted by group chat user', 'default') + expect(trackerMock.startWorkspaceRunCheckpoint).not.toHaveBeenCalled() + expect(bridgeMock.chat).not.toHaveBeenCalled() + }) + + it('does not start a bridge workspace run after the room generation changes before launch', async () => { + const client = await createClient('/tmp/workspace') + const storage = (client as any).__testStorage + storage.getRoom + .mockReturnValueOnce({ sessionSeed: 'seed-1', workspace: '/tmp/workspace' }) + .mockReturnValue({ sessionSeed: 'seed-2', workspace: '/tmp/workspace' }) + + await client.replyToMention('room-1', { + content: '@Worker hi', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + + expect(trackerMock.startWorkspaceRunCheckpoint).not.toHaveBeenCalled() + expect(bridgeMock.chat).not.toHaveBeenCalled() + }) + + it('does not start a bridge workspace run after the room is deleted before launch', async () => { + const client = await createClient('/tmp/workspace') + const storage = (client as any).__testStorage + storage.getRoom + .mockReturnValueOnce({ sessionSeed: '0', workspace: '/tmp/workspace' }) + .mockReturnValue(undefined) + + await client.replyToMention('room-1', { + content: '@Worker hi', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + + expect(trackerMock.startWorkspaceRunCheckpoint).not.toHaveBeenCalled() + expect(bridgeMock.chat).not.toHaveBeenCalled() + }) + + it('starts a checkpoint with the bridge-assigned run_id after bridge.chat starts', async () => { + const client = await createClient('/tmp/workspace') + + await client.replyToMention('room-1', { + content: '@Worker hi', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + + const options = bridgeMock.chat.mock.calls[0][5] + expect(options.workspace).toBe('/tmp/workspace') + expect(options).not.toHaveProperty('run_id') + expect(trackerMock.startWorkspaceRunCheckpoint).toHaveBeenCalledWith(expect.objectContaining({ + runId: 'bridge-run-id', + workspace: '/tmp/workspace', + })) + expect(order.slice(0, 2)).toEqual(['chat', 'checkpoint']) + }) + + it('uses the bridge-assigned run_id when finalizing the workspace diff', async () => { + const client = await createClient('/tmp/workspace') + + await client.replyToMention('room-1', { + content: '@Worker hi', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + + expect(trackerMock.completeWorkspaceRunCheckpointDraft).toHaveBeenCalledWith(expect.objectContaining({ + runId: 'bridge-run-id', + workspace: '/tmp/workspace', + })) + }) + + it('finalizes an aborted workspace diff on interrupt and ignores a later stream finalizer', async () => { + const client = await createClient('/tmp/workspace') + const sessionId = await workerSessionId() + const runId = '0123456789abcdef0123456789abcdef' + const state = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId, workspace: '/tmp/workspace' }) + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + saveWorkspaceDiffMessageForRun.mockReturnValue({ message: { id: 'diff-1', roomId: 'room-1' }, totalTokens: 0 }) + ;(trackerMock.completeWorkspaceRunCheckpointDraft as any).mockReturnValueOnce(workspaceDraft(runId, sessionId)) + + await client.interrupt('room-1') + + expect(bridgeMock.interrupt).toHaveBeenCalledWith(sessionId, 'Interrupted by group chat user', 'default') + expect(saveWorkspaceDiffMessageForRun).toHaveBeenCalledTimes(1) + expect(saveWorkspaceDiffMessageForRun.mock.calls[0][0]).toMatchObject({ + roomId: 'room-1', + sessionId, + runId, + status: 'aborted', + parentMessageId: null, + }) + + await client.finalizeWorkspaceDiffOnce(state, 'failed', 'late-message-id') + + expect(saveWorkspaceDiffMessageForRun).toHaveBeenCalledTimes(1) + expect(trackerMock.completeWorkspaceRunCheckpointDraft).toHaveBeenCalledTimes(1) + }) + + it('does not fail a synced interrupt when best-effort UI status emits cannot use the socket', async () => { + const client = await createClient('/tmp/workspace') + mockSocket.connected = false + const sessionId = await workerSessionId() + const runId = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' + client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId, workspace: '/tmp/workspace' }) + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + saveWorkspaceDiffMessageForRun.mockReturnValue({ message: { id: 'diff-1', roomId: 'room-1' }, totalTokens: 0 }) + ;(trackerMock.completeWorkspaceRunCheckpointDraft as any).mockReturnValueOnce(workspaceDraft(runId, sessionId)) + + try { + await expect(client.interrupt('room-1')).resolves.toBe(true) + } finally { + mockSocket.connected = true + } + + expect(saveWorkspaceDiffMessageForRun).toHaveBeenCalledTimes(1) + expect(saveWorkspaceDiffMessageForRun.mock.calls[0][0]).toMatchObject({ runId, status: 'aborted' }) + }) + + it('does not mark workspace diff runs aborted when bridge interrupt fails', async () => { + bridgeMock.interrupt.mockRejectedValueOnce(new Error('stale session')) + const client = await createClient('/tmp/workspace') + const sessionId = await workerSessionId() + const runId = 'dddddddddddddddddddddddddddddddd' + const state = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId, workspace: '/tmp/workspace' }) + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + + await expect(client.interrupt('room-1')).rejects.toThrow('stale session') + + expect(state.abortRequested).toBe(false) + expect(client.workspaceDiffRuns.size).toBe(1) + expect(saveWorkspaceDiffMessageForRun).not.toHaveBeenCalled() + expect(mockSocket.emit).not.toHaveBeenCalledWith('context_status', expect.objectContaining({ roomId: 'room-1', status: 'ready' })) + }) + + it('keeps workspace diff finalization pending when bridge interrupt is not synced yet', async () => { + bridgeMock.interrupt.mockResolvedValueOnce({ ok: true, synced: false }) + const client = await createClient('/tmp/workspace') + const sessionId = await workerSessionId() + const runId = 'cccccccccccccccccccccccccccccccc' + const state = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId, workspace: '/tmp/workspace' }) + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + saveWorkspaceDiffMessageForRun.mockReturnValue({ message: { id: 'diff-1', roomId: 'room-1' }, totalTokens: 0 }) + + await expect(client.interrupt('room-1')).resolves.toBe(false) + + expect(saveWorkspaceDiffMessageForRun).not.toHaveBeenCalled() + expect(trackerMock.completeWorkspaceRunCheckpointDraft).not.toHaveBeenCalled() + expect(client.workspaceDiffRuns.size).toBe(1) + expect(mockSocket.emit).not.toHaveBeenCalledWith('context_status', expect.objectContaining({ roomId: 'room-1', status: 'ready' })) + + ;(trackerMock.completeWorkspaceRunCheckpointDraft as any).mockReturnValueOnce(workspaceDraft(runId, sessionId)) + await client.finalizeWorkspaceDiffOnce(state, 'failed', 'terminal-message-id') + + expect(saveWorkspaceDiffMessageForRun).toHaveBeenCalledTimes(1) + expect(saveWorkspaceDiffMessageForRun.mock.calls[0][0]).toMatchObject({ runId, status: 'failed' }) + }) + + it('discards workspace diff finalization when the room session generation changed', async () => { + const client = await createClient('/tmp/workspace') + const staleSessionId = await workerSessionId('old-seed') + const runId = 'eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee' + const state = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId: staleSessionId, runId, workspace: '/tmp/workspace' }) + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + + await client.finalizeWorkspaceDiffOnce(state, 'completed', 'late-message-id') + + expect(saveWorkspaceDiffMessageForRun).not.toHaveBeenCalled() + expect(trackerMock.completeWorkspaceRunCheckpointDraft).not.toHaveBeenCalled() + expect(trackerMock.discardWorkspaceRunCheckpoint).toHaveBeenCalledWith(expect.objectContaining({ runId })) + expect(client.workspaceDiffRuns.size).toBe(0) + }) + + it('drops late assistant output after clear-context rotates the room session generation', async () => { + bridgeMock.interrupt.mockResolvedValueOnce({ ok: true, synced: false }) + const client = await createClient('/tmp/workspace') + client.__testStorage.getRoom + .mockReturnValueOnce({ sessionSeed: 'seed-1', workspace: '/tmp/workspace' }) + .mockReturnValueOnce({ sessionSeed: 'seed-1', workspace: '/tmp/workspace' }) + .mockReturnValue({ sessionSeed: 'seed-2', workspace: '/tmp/workspace' }) + + await client.replyToMention('room-1', { + content: '@Worker hi', + senderName: 'Alice', + senderId: 'user-1', + timestamp: 1, + }) + + const sessionId = await workerSessionId() + expect(bridgeMock.interrupt).toHaveBeenCalledWith( + sessionId, + 'Interrupted because group chat room state changed', 'default', - expect.objectContaining({ workspace: '/tmp/workspace' }), ) + expect(bridgeMock.destroy).toHaveBeenCalledWith(sessionId, 'default') + expect(client.__testStorage.saveWorkspaceDiffMessageForRun).not.toHaveBeenCalled() + expect(mockSocket.emit).not.toHaveBeenCalledWith('message_stream_end', expect.objectContaining({ roomId: 'room-1' })) + expect(mockSocket.emit).not.toHaveBeenCalledWith('message', expect.objectContaining({ role: 'assistant' }), expect.any(Function)) + expect(trackerMock.discardWorkspaceRunCheckpoint).not.toHaveBeenCalled() + expect(client.workspaceDiffRuns.size).toBe(0) + }) + + it('cleans up no-change workspace runs and keeps overlapping runs isolated', async () => { + const client = await createClient('/tmp/workspace') + const saveWorkspaceDiffMessageForRun = client.__testStorage.saveWorkspaceDiffMessageForRun + saveWorkspaceDiffMessageForRun.mockReturnValue({ message: { id: 'diff-1', roomId: 'room-1' }, totalTokens: 0 }) + const runA = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' + const runB = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' + const sessionId = await workerSessionId() + const stateA = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId: runA, workspace: '/tmp/workspace' }) + const stateB = client.beginWorkspaceDiffIfNeeded({ roomId: 'room-1', sessionId, runId: runB, workspace: '/tmp/workspace' }) + + ;(trackerMock.completeWorkspaceRunCheckpointDraft as any) + .mockReturnValueOnce(null) + .mockReturnValueOnce(workspaceDraft(runB, sessionId)) + + await client.finalizeWorkspaceDiffOnce(stateA, 'completed', null) + expect(saveWorkspaceDiffMessageForRun).not.toHaveBeenCalled() + expect(client.workspaceDiffRuns.size).toBe(1) + + await client.finalizeWorkspaceDiffOnce(stateA, 'completed', null) + expect(trackerMock.completeWorkspaceRunCheckpointDraft).toHaveBeenCalledTimes(1) + + await client.finalizeWorkspaceDiffOnce(stateB, 'failed', null) + expect(saveWorkspaceDiffMessageForRun).toHaveBeenCalledTimes(1) + expect(saveWorkspaceDiffMessageForRun.mock.calls[0][0]).toMatchObject({ runId: runB, status: 'failed' }) + expect(client.workspaceDiffRuns.size).toBe(0) }) }) diff --git a/tests/server/group-chat-approval.test.ts b/tests/server/group-chat-approval.test.ts index e1c792a7ef..65713498fa 100644 --- a/tests/server/group-chat-approval.test.ts +++ b/tests/server/group-chat-approval.test.ts @@ -5,7 +5,7 @@ import { emitAck, once, } from './group-chat-test-helpers' -import { GROUP_CHAT_AGENT_SOCKET_SECRET } from '../../packages/server/src/services/hermes/group-chat/agent-clients' +import { GROUP_CHAT_AGENT_SOCKET_SECRET, groupBridgeSessionId } from '../../packages/server/src/services/hermes/group-chat/agent-clients' import type { GroupChatServer } from '../../packages/server/src/services/hermes/group-chat' describe('group chat approval and context baseline', () => { @@ -27,6 +27,12 @@ describe('group chat approval and context baseline', () => { }) async function joinPair() { + const agentSessionId = groupBridgeSessionId( + 'room-1', + 'default', + 'Agent', + String(groupServer.getStorage().getRoom('room-1')?.sessionSeed || '0'), + ) const agent = await connectGroupChatClient(port, 'agent-1', 'Agent', { source: 'agent', agentSocketSecret: GROUP_CHAT_AGENT_SOCKET_SECRET, @@ -35,7 +41,7 @@ describe('group chat approval and context baseline', () => { harness.sockets.push(agent, human) await emitAck(agent, 'join', { roomId: 'room-1' }) await emitAck(human, 'join', { roomId: 'room-1', inviteCode: 'ROOM1' }) - return { agent, human } + return { agent, human, agentSessionId } } function wait(ms = 30) { @@ -43,11 +49,11 @@ describe('group chat approval and context baseline', () => { } it('relays context status and updates room token count', async () => { - const { agent, human } = await joinPair() + const { agent, human, agentSessionId } = await joinPair() const statusEvent = once(human, 'context_status') const roomUpdated = once(human, 'room_updated') - agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'replying', totalTokens: 123 }) + agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'replying', totalTokens: 123, agentSessionId }) expect(await statusEvent).toEqual({ roomId: 'room-1', agentName: 'Agent', status: 'replying' }) expect(await roomUpdated).toEqual({ roomId: 'room-1', totalTokens: 123 }) @@ -68,9 +74,9 @@ describe('group chat approval and context baseline', () => { }) it('clears ready context status from join recovery', async () => { - const { agent } = await joinPair() - agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'replying' }) - agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'ready' }) + const { agent, agentSessionId } = await joinPair() + agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'replying', agentSessionId }) + agent.emit('context_status', { roomId: 'room-1', agentName: 'Agent', status: 'ready', agentSessionId }) const lateJoiner = await connectGroupChatClient(port, 'human-2', 'Late') harness.sockets.push(lateJoiner) @@ -80,12 +86,13 @@ describe('group chat approval and context baseline', () => { }) it('relays approval requested with default choices', async () => { - const { agent, human } = await joinPair() + const { agent, human, agentSessionId } = await joinPair() const requested = once(human, 'approval.requested') agent.emit('approval.requested', { roomId: 'room-1', agentName: 'Agent', + agentSessionId, approval_id: 'approval-1', command: 'touch file', description: 'needs approval', @@ -101,7 +108,7 @@ describe('group chat approval and context baseline', () => { }) it('does not relay approval payloads to read-only invite members', async () => { - const { agent, human } = await joinPair() + const { agent, human, agentSessionId } = await joinPair() const readonly = await connectGroupChatClient(port, 'human-readonly', 'ReadOnly') harness.sockets.push(readonly) groupServer.getIO().of('/group-chat').sockets.get(readonly.id!)!.data.authUser = { id: 7, role: 'user', profiles: [] } @@ -114,6 +121,7 @@ describe('group chat approval and context baseline', () => { agent.emit('approval.requested', { roomId: 'room-1', agentName: 'Agent', + agentSessionId, approval_id: 'approval-private', command: 'cat /private/workspace/secret', description: 'needs approval', @@ -140,10 +148,10 @@ describe('group chat approval and context baseline', () => { }) it('relays approval resolved with normalized choice', async () => { - const { agent, human } = await joinPair() + const { agent, human, agentSessionId } = await joinPair() const resolved = once(human, 'approval.resolved') - agent.emit('approval.resolved', { roomId: 'room-1', agentName: 'Agent', approval_id: 'approval-1', choice: 'deny' }) + agent.emit('approval.resolved', { roomId: 'room-1', agentName: 'Agent', agentSessionId, approval_id: 'approval-1', choice: 'deny' }) expect(await resolved).toEqual({ event: 'approval.resolved', diff --git a/tests/server/group-chat-history-window.test.ts b/tests/server/group-chat-history-window.test.ts index 1cbc67b89b..c615bbed11 100644 --- a/tests/server/group-chat-history-window.test.ts +++ b/tests/server/group-chat-history-window.test.ts @@ -39,7 +39,7 @@ vi.mock('../../packages/server/src/services/auth', () => ({ import { countTokens, SUMMARY_PREFIX } from '../../packages/server/src/lib/context-compressor' import { initAllHermesTables } from '../../packages/server/src/db/hermes/schemas' import { GroupChatServer } from '../../packages/server/src/services/hermes/group-chat' -import { AgentClients, mentionMessageToStoredContextMessage } from '../../packages/server/src/services/hermes/group-chat/agent-clients' +import { AgentClients, groupBridgeSessionId, mentionMessageToStoredContextMessage } from '../../packages/server/src/services/hermes/group-chat/agent-clients' import { sortGroupMessagesCanonical } from '../../packages/server/src/services/hermes/group-chat/group-message-ordering' function makeDb(): DatabaseSync { @@ -199,9 +199,12 @@ describe('group chat history windows', () => { role: 'user', timestamp: index + 1, })) + const sessionId = groupBridgeSessionId('room-1', 'default', 'Worker', 'seed-1') const storage = { getMessagesForContext: vi.fn(() => messages), getRecentMessagesForUI: vi.fn(() => messages.slice(-150)), + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room', sessionSeed: 'seed-1' })), + getRoomAgentByAgentId: vi.fn(() => ({ id: 'row-1', roomId: 'room-1', agentId: 'agent-1', profile: 'default', name: 'Worker' })), updateRoomTotalTokens: vi.fn(), } const bridge = { @@ -222,7 +225,7 @@ describe('group chat history windows', () => { } as any) client.setStorage(storage as any) - await (client as any).refreshRoomFullContextEstimate('room-1', 'session-1', bridge, undefined, { model: '', provider: '' }) + await (client as any).refreshRoomFullContextEstimate('room-1', sessionId, bridge, undefined, { model: '', provider: '' }) expect(storage.getMessagesForContext).toHaveBeenCalledWith('room-1') expect(storage.getRecentMessagesForUI).not.toHaveBeenCalled() diff --git a/tests/server/group-chat-member-sync.test.ts b/tests/server/group-chat-member-sync.test.ts index c96cbd2c8c..1c7e3e7a1a 100644 --- a/tests/server/group-chat-member-sync.test.ts +++ b/tests/server/group-chat-member-sync.test.ts @@ -26,7 +26,7 @@ vi.mock('../../packages/server/src/services/auth', () => ({ getToken: vi.fn(async () => 'test-token'), })) -import { AgentClients } from '../../packages/server/src/services/hermes/group-chat/agent-clients' +import { AgentClients, groupBridgeSessionId } from '../../packages/server/src/services/hermes/group-chat/agent-clients' import { GroupChatServer } from '../../packages/server/src/services/hermes/group-chat' import { groupChatRoutes, setGroupChatServer } from '../../packages/server/src/routes/hermes/group-chat' @@ -270,6 +270,263 @@ describe('Group Chat member/agent identity sync', () => { }) }) + it('interrupts runtime room state before deleting persisted room data', async () => { + const calls: string[] = [] + const storage = { + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room 1', ownerAuthUserId: 7 })), + deleteRoom: vi.fn(() => { calls.push('storage-delete') }), + } + const chatServer = { + getStorage: () => storage, + deleteRoomRuntimeState: vi.fn(async () => { calls.push('runtime-delete') }), + } + setGroupChatServer(chatServer as any) + + const handler = routeHandler('/api/hermes/group-chat/rooms/:roomId', 'DELETE') + const ctx: any = { + params: { roomId: 'room-1' }, + state: { user: { id: 1, username: 'root', role: 'super_admin' } }, + status: 200, + body: undefined, + } + await handler(ctx, async () => {}) + + expect(calls).toEqual(['runtime-delete', 'storage-delete']) + expect(chatServer.deleteRoomRuntimeState).toHaveBeenCalledWith('room-1') + expect(ctx.body).toEqual({ success: true }) + }) + + it('does not delete persisted room data when runtime interrupt does not complete', async () => { + const storage = { + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room 1', ownerAuthUserId: 7 })), + deleteRoom: vi.fn(), + } + const chatServer = { + getStorage: () => storage, + deleteRoomRuntimeState: vi.fn(async () => { throw Object.assign(new Error('still running'), { status: 409 }) }), + } + setGroupChatServer(chatServer as any) + + const handler = routeHandler('/api/hermes/group-chat/rooms/:roomId', 'DELETE') + const ctx: any = { + params: { roomId: 'room-1' }, + state: { user: { id: 1, username: 'root', role: 'super_admin' } }, + status: 200, + body: undefined, + } + await handler(ctx, async () => {}) + + expect(ctx.status).toBe(409) + expect(ctx.body).toEqual({ error: 'still running' }) + expect(storage.deleteRoom).not.toHaveBeenCalled() + }) + + it('interrupts agents before evicting in-memory room sockets so deleted rooms reject late realtime messages', async () => { + const calls: string[] = [] + const socketsLeave = vi.fn(() => { calls.push('sockets-leave') }) + const saveMessageAndRefreshRoom = vi.fn() + const server = Object.create(GroupChatServer.prototype) as any + server.rooms = new Map([['room-1', { hasOnlineMember: vi.fn(() => true) }]]) + server.typingState = new Map([['room-1', new Map([['human-1', { userName: 'Human', timer: setTimeout(() => {}, 1000) }]])]]) + server.contextStatusState = new Map([['room-1', new Map([['Worker', { agentName: 'Worker', status: 'replying' }]])]]) + server.agentClients = { + interruptRoom: vi.fn(async () => { calls.push('interrupt') }), + disconnectRoom: vi.fn(() => { calls.push('disconnect') }), + } + server.nsp = { + in: vi.fn(() => ({ socketsLeave })), + to: vi.fn(() => ({ emit: vi.fn() })), + } + server.storage = { saveMessageAndRefreshRoom } + + await server.deleteRoomRuntimeState('room-1') + const ack = vi.fn() + server.handleMessage({ id: 'socket-1' }, { roomId: 'room-1', content: 'late', role: 'user' }, ack) + + expect(calls).toEqual(['interrupt', 'disconnect', 'sockets-leave']) + expect(server.rooms.has('room-1')).toBe(false) + expect(server.agentClients.disconnectRoom).toHaveBeenCalledWith('room-1') + expect(server.nsp.in).toHaveBeenCalledWith('room-1') + expect(socketsLeave).toHaveBeenCalledWith('room-1') + expect(saveMessageAndRefreshRoom).not.toHaveBeenCalled() + expect(ack).toHaveBeenCalledWith({ error: 'Not in room' }) + }) + + it('rejects stale agent context and stream side-channel events after session rotation', () => { + const broadcastEmit = vi.fn() + const roomEmit = vi.fn() + const updateRoomTotalTokens = vi.fn() + const agentMember = { + id: 'agent-socket-1', + userId: 'agent-stable-1', + name: 'Worker', + description: '', + joinedAt: Date.now(), + online: true, + socketId: 'agent-socket-1', + source: 'agent', + avatar: '', + } + const server = Object.create(GroupChatServer.prototype) as any + server.rooms = new Map([['room-1', { + getOnlineMemberBySocketId: vi.fn(() => agentMember), + }]]) + server.contextStatusState = new Map() + server.storage = { + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room', sessionSeed: 'seed-2' })), + getRoomAgentByAgentId: vi.fn(() => ({ id: 'row-1', roomId: 'room-1', agentId: 'agent-stable-1', profile: 'default', name: 'Worker' })), + updateRoomTotalTokens, + } + server.nsp = { to: vi.fn(() => ({ emit: broadcastEmit })) } + const socket = { id: 'agent-socket-1', to: vi.fn(() => ({ emit: roomEmit })) } + const staleSessionId = groupBridgeSessionId('room-1', 'default', 'Worker', 'seed-1') + const currentSessionId = groupBridgeSessionId('room-1', 'default', 'Worker', 'seed-2') + + server.handleContextStatus(socket, { + roomId: 'room-1', + agentName: 'Worker', + status: 'replying', + totalTokens: 123, + agentSessionId: staleSessionId, + }) + server.handleMessageStreamStart(socket, { + roomId: 'room-1', + id: 'late-stream', + agentSessionId: staleSessionId, + }) + + expect(updateRoomTotalTokens).not.toHaveBeenCalled() + expect(roomEmit).not.toHaveBeenCalled() + expect(broadcastEmit).not.toHaveBeenCalled() + expect(server.contextStatusState.size).toBe(0) + + server.handleContextStatus(socket, { + roomId: 'room-1', + agentName: 'Worker', + status: 'replying', + totalTokens: 456, + agentSessionId: currentSessionId, + }) + server.handleMessageStreamStart(socket, { + roomId: 'room-1', + id: 'current-stream', + agentSessionId: currentSessionId, + }) + + expect(updateRoomTotalTokens).toHaveBeenCalledWith('room-1', 456) + expect(roomEmit).toHaveBeenCalledWith('context_status', expect.objectContaining({ roomId: 'room-1', agentName: 'Worker', status: 'replying' })) + expect(broadcastEmit).toHaveBeenCalledWith('room_updated', { roomId: 'room-1', totalTokens: 456 }) + expect(broadcastEmit).toHaveBeenCalledWith('message_stream_start', expect.objectContaining({ id: 'current-stream', senderName: 'Worker' })) + }) + + it('does not drop queued mentions when room interrupt is not synchronized', async () => { + const clients = new AgentClients() as any + const agent = { name: 'Worker', interrupt: vi.fn(async () => false) } + clients.rooms = new Map([['room-1', new Map([['agent-stable-1', agent]])]]) + clients._mentionQueue = new Map([ + ['room-1', [{ agent, msg: { content: '@Worker one', senderName: 'Han', senderId: 'user-1', timestamp: 1 } }]], + ['room-1:Worker', [{ agent, msg: { content: '@Worker two', senderName: 'Han', senderId: 'user-1', timestamp: 2 } }]], + ]) + + await expect(clients.interruptRoom('room-1')).rejects.toMatchObject({ status: 409 }) + + expect(clients._mentionQueue.has('room-1')).toBe(true) + expect(clients._mentionQueue.has('room-1:Worker')).toBe(true) + }) + + it('rejects stale agent assistant/tool messages at persistence time after session rotation', () => { + const emit = vi.fn() + const saveMessageAndRefreshRoom = vi.fn() + const agentMember = { + id: 'agent-socket-1', + userId: 'agent-stable-1', + name: 'Worker', + description: '', + joinedAt: Date.now(), + online: true, + socketId: 'agent-socket-1', + source: 'agent', + avatar: '', + } + const server = Object.create(GroupChatServer.prototype) as any + server.rooms = new Map([['room-1', { + hasOnlineMember: vi.fn(() => true), + getOnlineMemberBySocketId: vi.fn(() => agentMember), + }]]) + server.storage = { + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room', sessionSeed: 'seed-2' })), + getRoomAgentByAgentId: vi.fn(() => ({ id: 'row-1', roomId: 'room-1', agentId: 'agent-stable-1', profile: 'default', name: 'Worker' })), + saveMessageAndRefreshRoom, + } + server.nsp = { to: vi.fn(() => ({ emit })) } + const ack = vi.fn() + const staleSessionId = groupBridgeSessionId('room-1', 'default', 'Worker', 'seed-1') + + server.handleMessage({ id: 'agent-socket-1' }, { + roomId: 'room-1', + content: 'late', + role: 'assistant', + agentSessionId: staleSessionId, + }, ack) + + expect(ack).toHaveBeenCalledWith({ error: 'Stale room session' }) + expect(saveMessageAndRefreshRoom).not.toHaveBeenCalled() + expect(emit).not.toHaveBeenCalled() + }) + + it('clears runtime state before rotating persisted room context', async () => { + const calls: string[] = [] + const room = { id: 'room-1', name: 'Room 1', inviteCode: 'invite', ownerAuthUserId: 7, workspace: '/tmp/workspace' } + const storage = { + getRoom: vi.fn(() => room), + clearRoomContext: vi.fn(() => { calls.push('storage-clear') }), + } + const chatServer = { + getStorage: () => storage, + clearRoomRuntimeState: vi.fn(async () => { calls.push('runtime-clear') }), + } + setGroupChatServer(chatServer as any) + + const handler = routeHandler('/api/hermes/group-chat/rooms/:roomId/clear-context', 'POST') + const ctx: any = { + params: { roomId: 'room-1' }, + state: { user: { id: 1, username: 'root', role: 'super_admin' } }, + status: 200, + body: undefined, + } + await handler(ctx, async () => {}) + + expect(calls).toEqual(['runtime-clear', 'storage-clear']) + expect(chatServer.clearRoomRuntimeState).toHaveBeenCalledWith('room-1') + expect(ctx.body).toEqual({ success: true, room: expect.objectContaining({ id: 'room-1', workspace: '/tmp/workspace' }) }) + }) + + it('does not clear persisted context when runtime interrupt does not complete', async () => { + const room = { id: 'room-1', name: 'Room 1', inviteCode: 'invite', ownerAuthUserId: 7, workspace: '/tmp/workspace' } + const storage = { + getRoom: vi.fn(() => room), + clearRoomContext: vi.fn(), + } + const chatServer = { + getStorage: () => storage, + clearRoomRuntimeState: vi.fn(async () => { throw Object.assign(new Error('still running'), { status: 409 }) }), + } + setGroupChatServer(chatServer as any) + + const handler = routeHandler('/api/hermes/group-chat/rooms/:roomId/clear-context', 'POST') + const ctx: any = { + params: { roomId: 'room-1' }, + state: { user: { id: 1, username: 'root', role: 'super_admin' } }, + status: 200, + body: undefined, + } + await handler(ctx, async () => {}) + + expect(ctx.status).toBe(409) + expect(ctx.body).toEqual({ error: 'still running' }) + expect(storage.clearRoomContext).not.toHaveBeenCalled() + }) + it('rejects authenticated Socket.IO room joins without invite, membership, owner, or profile scope', () => { const emit = vi.fn() const server = Object.create(GroupChatServer.prototype) as any @@ -544,8 +801,10 @@ describe('Group Chat member/agent identity sync', () => { ['agent-1', { name: '丫鬟', description: '' }], ]) server.agentClients = { processMentions: vi.fn(async () => undefined) } + const agentSessionId = groupBridgeSessionId('room-1', 'default', '丫鬟', 'seed-1') server.storage = { - getRoom: vi.fn(() => ({ id: 'room-1' })), + getRoom: vi.fn(() => ({ id: 'room-1', name: 'Room', sessionSeed: 'seed-1' })), + getRoomAgentByAgentId: vi.fn(() => ({ id: 'row-1', roomId: 'room-1', agentId: 'agent-1', profile: 'default', name: '丫鬟' })), saveMessageAndRefreshRoom: vi.fn((msg: any) => ({ message: msg, totalTokens: 123 })), } server.nsp = { to: vi.fn(() => ({ emit })) } @@ -559,7 +818,7 @@ describe('Group Chat member/agent identity sync', () => { })) server.agentClients.processMentions.mockClear() - server.handleMessage({ id: 'agent-socket' }, { roomId: 'room-1', content: '@all agent says hi', role: 'assistant', mentionDepth: 1 }, vi.fn()) + server.handleMessage({ id: 'agent-socket' }, { roomId: 'room-1', content: '@all agent says hi', role: 'assistant', mentionDepth: 1, agentSessionId }, vi.fn()) expect(server.agentClients.processMentions).toHaveBeenCalledTimes(1) expect(server.agentClients.processMentions).toHaveBeenLastCalledWith('room-1', expect.objectContaining({ content: '@all agent says hi', @@ -568,7 +827,7 @@ describe('Group Chat member/agent identity sync', () => { })) server.agentClients.processMentions.mockClear() - server.handleMessage({ id: 'agent-socket' }, { roomId: 'room-1', content: '@all too deep', role: 'assistant', mentionDepth: 4 }, vi.fn()) + server.handleMessage({ id: 'agent-socket' }, { roomId: 'room-1', content: '@all too deep', role: 'assistant', mentionDepth: 4, agentSessionId }, vi.fn()) expect(server.agentClients.processMentions).not.toHaveBeenCalled() }) }) diff --git a/tests/server/group-chat-streaming.test.ts b/tests/server/group-chat-streaming.test.ts index ce5a0d9bbb..13d454f995 100644 --- a/tests/server/group-chat-streaming.test.ts +++ b/tests/server/group-chat-streaming.test.ts @@ -5,7 +5,7 @@ import { emitAck, once, } from './group-chat-test-helpers' -import { GROUP_CHAT_AGENT_SOCKET_SECRET } from '../../packages/server/src/services/hermes/group-chat/agent-clients' +import { GROUP_CHAT_AGENT_SOCKET_SECRET, groupBridgeSessionId } from '../../packages/server/src/services/hermes/group-chat/agent-clients' import type { GroupChatServer } from '../../packages/server/src/services/hermes/group-chat' describe('group chat streaming baseline', () => { @@ -37,14 +37,20 @@ describe('group chat streaming baseline', () => { await emitAck(alice, 'join', { roomId: 'room-1', inviteCode: 'ROOM1' }) await emitAck(bob, 'join', { roomId: 'room-1', inviteCode: 'ROOM1' }) await emitAck(worker, 'join', { roomId: 'room-1' }) - return { alice, bob, worker } + const agentSessionId = groupBridgeSessionId( + 'room-1', + 'default', + 'Worker', + String(groupServer.getStorage().getRoom('room-1')?.sessionSeed || '0'), + ) + return { alice, bob, worker, agentSessionId } } it('relays stream start, content delta, reasoning delta, and stream end to room members', async () => { - const { worker, bob } = await joinPair() + const { worker, bob, agentSessionId } = await joinPair() const streamStart = once(bob, 'message_stream_start') - worker.emit('message_stream_start', { roomId: 'room-1', id: 'stream-1', senderName: 'Spoofed', timestamp: 10 }) + worker.emit('message_stream_start', { roomId: 'room-1', id: 'stream-1', senderName: 'Spoofed', timestamp: 10, agentSessionId }) expect(await streamStart).toMatchObject({ id: 'stream-1', roomId: 'room-1', @@ -54,15 +60,15 @@ describe('group chat streaming baseline', () => { }) const contentDelta = once(bob, 'message_stream_delta') - worker.emit('message_stream_delta', { roomId: 'room-1', id: 'stream-1', delta: 'hello' }) + worker.emit('message_stream_delta', { roomId: 'room-1', id: 'stream-1', delta: 'hello', agentSessionId }) expect(await contentDelta).toEqual({ roomId: 'room-1', id: 'stream-1', delta: 'hello' }) const reasoningDelta = once(bob, 'message_reasoning_delta') - worker.emit('message_reasoning_delta', { roomId: 'room-1', id: 'stream-1', delta: 'thinking' }) + worker.emit('message_reasoning_delta', { roomId: 'room-1', id: 'stream-1', delta: 'thinking', agentSessionId }) expect(await reasoningDelta).toEqual({ roomId: 'room-1', id: 'stream-1', delta: 'thinking' }) const streamEnd = once(bob, 'message_stream_end') - worker.emit('message_stream_end', { roomId: 'room-1', id: 'stream-1' }) + worker.emit('message_stream_end', { roomId: 'room-1', id: 'stream-1', agentSessionId }) expect(await streamEnd).toEqual({ roomId: 'room-1', id: 'stream-1' }) }) diff --git a/tests/server/group-chat-workspace-diff-context.test.ts b/tests/server/group-chat-workspace-diff-context.test.ts new file mode 100644 index 0000000000..042cbbb0ad --- /dev/null +++ b/tests/server/group-chat-workspace-diff-context.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from 'vitest' +import { + buildProjectedGroupChatHistory, + isWorkspaceDiffToolMessage, + projectGroupChatMessage, +} from '../../packages/server/src/services/hermes/group-chat/context-projection' + +function message(overrides: Record) { + return { + id: 'm1', + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + content: 'hello', + timestamp: 1, + role: 'user', + ...overrides, + } as any +} + +describe('group chat workspace diff context exclusion', () => { + it('identifies workspace_diff tool messages for hard context exclusion', () => { + expect(isWorkspaceDiffToolMessage(message({ role: 'tool', tool_name: 'workspace_diff' }))).toBe(true) + expect(isWorkspaceDiffToolMessage(message({ role: 'tool', tool_name: 'search' }))).toBe(false) + }) + + it('excludes workspace_diff messages from projected model history while keeping regular tools', () => { + const diff = message({ + id: 'diff-1', + role: 'tool', + tool_name: 'workspace_diff', + content: JSON.stringify({ kind: 'workspace_diff', files: [{ patch: '+secret patch' }] }), + }) + const regularTool = message({ + id: 'tool-1', + role: 'tool', + tool_name: 'search', + content: 'docs found', + }) + + const history = buildProjectedGroupChatHistory('', [ + message({ id: 'm1', senderName: 'Alice', senderId: 'user-1', role: 'user', content: '@Worker hello' }), + diff, + regularTool, + ], { agentId: 'agent-1', name: 'Worker' }) + + expect(history.map(item => item.content).join('\n')).not.toContain('secret patch') + expect(history).toContainEqual(projectGroupChatMessage(regularTool, { agentId: 'agent-1', name: 'Worker' })) + }) +}) diff --git a/tests/server/group-chat-workspace-diff.test.ts b/tests/server/group-chat-workspace-diff.test.ts new file mode 100644 index 0000000000..627fc1cd46 --- /dev/null +++ b/tests/server/group-chat-workspace-diff.test.ts @@ -0,0 +1,342 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { DatabaseSync } from 'node:sqlite' +import { createServer, type Server as HttpServer } from 'http' +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'fs' +import { join } from 'path' +import { tmpdir } from 'os' + +const dbState = vi.hoisted(() => ({ + db: null as DatabaseSync | null, +})) + +vi.mock('../../packages/server/src/db/index', () => ({ + getDb: () => dbState.db, + isSqliteAvailable: () => Boolean(dbState.db), +})) + +vi.mock('socket.io-client', () => ({ + io: vi.fn(() => ({ + id: 'agent-socket', + connected: true, + io: { on: vi.fn() }, + on: vi.fn(), + emit: vi.fn(), + disconnect: vi.fn(), + })), +})) + +vi.mock('../../packages/server/src/services/auth', () => ({ + getToken: vi.fn(async () => 'test-token'), +})) + +describe('group chat workspace diff persistence', () => { + let root: string + let workspace: string + let httpServer: HttpServer + + beforeEach(async () => { + vi.resetModules() + root = mkdtempSync(join(tmpdir(), 'hermes-gc-diff-')) + workspace = join(root, 'workspace') + mkdirSync(workspace) + dbState.db = new DatabaseSync(':memory:') + const { initAllHermesTables } = await import('../../packages/server/src/db/hermes/schemas') + initAllHermesTables() + httpServer = createServer() + }) + + afterEach(() => { + httpServer?.close() + dbState.db?.close() + dbState.db = null + rmSync(root, { recursive: true, force: true }) + }) + + async function makeDraft(runId = '0123456789abcdef0123456789abcdef') { + const tracker = await import('../../packages/server/src/services/hermes/run-chat/workspace-diff-tracker') + writeFileSync(join(workspace, 'file.txt'), 'old\n') + tracker.startWorkspaceRunCheckpoint({ sessionId: 'session-1', runId, workspace }) + writeFileSync(join(workspace, 'file.txt'), 'new\n') + return tracker.completeWorkspaceRunCheckpointDraft({ sessionId: 'session-1', runId, workspace }) + } + + async function saveDiff(storage: any, roomId = 'room-1', runId = '0123456789abcdef0123456789abcdef') { + const draft = await makeDraft(runId) + const saved = storage.saveWorkspaceDiffMessageForRun({ + roomId, + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId, + status: 'completed', + workspace, + draft: draft!, + }) + return { saved, payload: JSON.parse(saved!.message.content) } + } + + function countRows(table: string, where = '', ...args: unknown[]): number { + return (dbState.db?.prepare(`SELECT COUNT(*) AS count FROM ${table}${where}`).get(...args) as { count: number }).count + } + + function expectWorkspaceChangeDeleted(changeId: string) { + expect(countRows('workspace_run_changes', ' WHERE change_id = ?', changeId)).toBe(0) + expect(countRows('workspace_run_change_files', ' WHERE change_id = ?', changeId)).toBe(0) + } + + it('persists one workspace_run_change and one durable workspace_diff group message', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const runId = '0123456789abcdef0123456789abcdef' + const draft = await makeDraft(runId) + + const saved = storage.saveWorkspaceDiffMessageForRun({ + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId, + status: 'completed', + workspace, + draft: draft!, + }) + + expect(saved?.message).toMatchObject({ + role: 'tool', + tool_name: 'workspace_diff', + tool_call_id: `workspace_diff:${runId}`, + senderId: 'agent-1', + senderName: 'Worker', + }) + const payload = JSON.parse(saved!.message.content) + expect(payload).toMatchObject({ + kind: 'workspace_diff', + version: 1, + room_id: 'room-1', + session_id: 'session-1', + run_id: runId, + status: 'completed', + files_changed: 1, + }) + expect(payload.files[0].patch).toContain('-old') + expect(payload.files[0].patch).toContain('+new') + expect(payload.workspace).toBeUndefined() + expect(payload.workspace_basename).toBe('workspace') + + const rows = dbState.db?.prepare('SELECT COUNT(*) AS count FROM gc_messages WHERE tool_name = ?').get('workspace_diff') as { count: number } + expect(rows.count).toBe(1) + const changeRow = dbState.db?.prepare('SELECT workspace, room_id, message_id FROM workspace_run_changes WHERE change_id = ?').get(payload.change_id) as { workspace: string; room_id: string; message_id: string } + expect(changeRow.workspace).toBe('workspace') + expect(changeRow.workspace).not.toBe(workspace) + expect(changeRow.room_id).toBe('room-1') + expect(changeRow.message_id).toBe(saved!.message.id) + const changeRows = dbState.db?.prepare('SELECT COUNT(*) AS count FROM workspace_run_changes WHERE change_id = ?').get(payload.change_id) as { count: number } + expect(changeRows.count).toBe(1) + server.getIO().close() + }) + + it('keeps workspace_diff message ids unique when room ids are long', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + const roomId = `room-${'x'.repeat(220)}` + storage.saveRoom(roomId, 'Long Room') + const runA = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' + const runB = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' + + const draftA = await makeDraft(runA) + const savedA = storage.saveWorkspaceDiffMessageForRun({ + roomId, + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId: runA, + status: 'completed', + workspace, + draft: draftA!, + }) + const draftB = await makeDraft(runB) + const savedB = storage.saveWorkspaceDiffMessageForRun({ + roomId, + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId: runB, + status: 'completed', + workspace, + draft: draftB!, + }) + + expect(savedA?.message.id).not.toBe(savedB?.message.id) + expect(savedA?.message.id).toContain(runA) + expect(savedB?.message.id).toContain(runB) + const rows = dbState.db?.prepare('SELECT COUNT(*) AS count FROM gc_messages WHERE roomId = ? AND tool_name = ?').get(roomId, 'workspace_diff') as { count: number } + expect(rows.count).toBe(2) + server.getIO().close() + }) + + it('cleans persisted workspace diffs when room context is cleared', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const { payload } = await saveDiff(storage) + + expect(countRows('workspace_run_changes', ' WHERE change_id = ?', payload.change_id)).toBe(1) + storage.clearRoomContext('room-1') + + expectWorkspaceChangeDeleted(payload.change_id) + expect(countRows('gc_messages', ' WHERE roomId = ?', 'room-1')).toBe(0) + server.getIO().close() + }) + + it('does not delete unrelated workspace changes from spoofed workspace_diff message content', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + storage.saveRoom('room-2', 'Room 2') + const { payload } = await saveDiff(storage, 'room-1', 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa') + + storage.saveMessageAndRefreshRoom({ + id: 'fake-workspace-diff', + roomId: 'room-2', + senderId: 'user-2', + senderName: 'Mallory', + content: JSON.stringify({ kind: 'workspace_diff', change_id: payload.change_id }), + timestamp: 1, + role: 'tool', + tool_name: 'workspace_diff', + tool_call_id: 'workspace_diff:fake', + }) + storage.clearRoomContext('room-2') + + expect(countRows('workspace_run_changes', ' WHERE change_id = ?', payload.change_id)).toBe(1) + expect(countRows('workspace_run_change_files', ' WHERE change_id = ?', payload.change_id)).toBeGreaterThan(0) + server.getIO().close() + }) + + it('does not allow client messages to overwrite server-created workspace diff cards', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const { saved, payload } = await saveDiff(storage, 'room-1', 'cccccccccccccccccccccccccccccccc') + const originalTimestamp = saved!.message.timestamp + + const overwrite = storage.saveMessageAndRefreshRoom({ + id: saved!.message.id, + roomId: 'room-1', + senderId: 'user-1', + senderName: 'Mallory', + content: JSON.stringify({ kind: 'workspace_diff', change_id: 'other-change' }), + timestamp: 1, + role: 'tool', + tool_name: 'workspace_diff', + tool_call_id: 'workspace_diff:fake', + }) + + expect(overwrite.message.timestamp).toBe(originalTimestamp) + expect(JSON.parse(String(overwrite.message.content)).change_id).toBe(payload.change_id) + expect(countRows('workspace_run_changes', ' WHERE change_id = ?', payload.change_id)).toBe(1) + server.getIO().close() + }) + + it('cleans persisted workspace diffs when a room is deleted', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const { payload } = await saveDiff(storage, 'room-1', 'dddddddddddddddddddddddddddddddd') + + storage.deleteRoom('room-1') + + expectWorkspaceChangeDeleted(payload.change_id) + expect(countRows('gc_messages', ' WHERE roomId = ?', 'room-1')).toBe(0) + expect(countRows('gc_rooms', ' WHERE id = ?', 'room-1')).toBe(0) + server.getIO().close() + }) + + it('does not write workspace diff rows after a room is deleted', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const draft = await makeDraft('99999999999999999999999999999999') + + storage.deleteRoom('room-1') + const saved = storage.saveWorkspaceDiffMessageForRun({ + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId: '99999999999999999999999999999999', + status: 'completed', + workspace, + draft: draft!, + }) + + expect(saved).toBeNull() + expect(countRows('gc_messages', ' WHERE roomId = ?', 'room-1')).toBe(0) + expect(countRows('workspace_run_changes')).toBe(0) + expect(countRows('workspace_run_change_files')).toBe(0) + server.getIO().close() + }) + + it('cleans persisted workspace diffs when their group messages are pruned', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const { saved, payload } = await saveDiff(storage, 'room-1', 'eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee') + dbState.db?.prepare('UPDATE gc_messages SET timestamp = ? WHERE id = ?').run(1, saved!.message.id) + storage.saveMessageAndRefreshRoom({ + id: 'keep-message', + roomId: 'room-1', + senderId: 'user-1', + senderName: 'User', + content: 'keep', + timestamp: 100, + role: 'user', + }) + + storage.pruneMessages('room-1', 1) + + expectWorkspaceChangeDeleted(payload.change_id) + expect(countRows('gc_messages', ' WHERE id = ?', saved!.message.id)).toBe(0) + expect(countRows('gc_messages', ' WHERE id = ?', 'keep-message')).toBe(1) + server.getIO().close() + }) + + it('rolls back diff rows when the group message insert fails', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + storage.saveRoom('room-1', 'Room 1') + const draft = await makeDraft('fedcba9876543210fedcba9876543210') + dbState.db?.exec(`CREATE TRIGGER fail_workspace_diff_message + BEFORE INSERT ON gc_messages + WHEN NEW.tool_name = 'workspace_diff' + BEGIN + SELECT RAISE(ABORT, 'message failed'); + END`) + + expect(() => storage.saveWorkspaceDiffMessageForRun({ + roomId: 'room-1', + senderId: 'agent-1', + senderName: 'Worker', + sessionId: 'session-1', + runId: 'fedcba9876543210fedcba9876543210', + status: 'failed', + workspace, + draft: draft!, + })).toThrow('message failed') + + expect((dbState.db?.prepare('SELECT COUNT(*) AS count FROM workspace_run_changes').get() as { count: number }).count).toBe(0) + expect((dbState.db?.prepare('SELECT COUNT(*) AS count FROM workspace_run_change_files').get() as { count: number }).count).toBe(0) + expect((dbState.db?.prepare('SELECT COUNT(*) AS count FROM gc_messages WHERE tool_name = ?').get('workspace_diff') as { count: number }).count).toBe(0) + server.getIO().close() + }) +}) diff --git a/tests/server/group-chat-workspace.test.ts b/tests/server/group-chat-workspace.test.ts index d71609f642..484d8f2fb2 100644 --- a/tests/server/group-chat-workspace.test.ts +++ b/tests/server/group-chat-workspace.test.ts @@ -129,6 +129,71 @@ describe('group chat room workspace', () => { server.getIO().close() }) + it('interrupts active room agents before changing the configured workspace', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const { setGroupChatServer } = await import('../../packages/server/src/routes/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + const workspace = join(root, 'repo') + await mkdir(workspace) + storage.saveRoom('room-1', 'Room 1') + setGroupChatServer(server) + const events: string[] = [] + const fenceCurrentRoomAgentSessions = vi.spyOn(server, 'fenceCurrentRoomAgentSessions').mockImplementation(() => { + events.push('fence') + return vi.fn() + }) + const interruptRoom = vi.spyOn(server.agentClients, 'interruptRoom').mockImplementation(async () => { + events.push('interrupt') + }) + + const handler = await routeHandler('/api/hermes/group-chat/rooms/:roomId/workspace', 'PUT') + const ctx: any = { + params: { roomId: 'room-1' }, + request: { body: { workspace } }, + status: 200, + body: undefined, + } + + await handler(ctx, async () => {}) + + expect(fenceCurrentRoomAgentSessions).toHaveBeenCalledWith('room-1') + expect(interruptRoom).toHaveBeenCalledWith('room-1') + expect(events).toEqual(['fence', 'interrupt']) + expect(storage.getRoom('room-1')?.workspace).toBe(workspace) + server.getIO().close() + }) + + it('does not switch workspace when active room agents do not finish interrupting', async () => { + const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') + const { setGroupChatServer } = await import('../../packages/server/src/routes/hermes/group-chat') + const server = new GroupChatServer(httpServer) + const storage = server.getStorage() + const workspace = join(root, 'repo') + await mkdir(workspace) + storage.saveRoom('room-1', 'Room 1') + setGroupChatServer(server) + const releaseSessionFence = vi.fn() + vi.spyOn(server, 'fenceCurrentRoomAgentSessions').mockReturnValue(releaseSessionFence) + vi.spyOn(server.agentClients, 'interruptRoom').mockRejectedValue(Object.assign(new Error('still running'), { status: 409 })) + + const handler = await routeHandler('/api/hermes/group-chat/rooms/:roomId/workspace', 'PUT') + const ctx: any = { + params: { roomId: 'room-1' }, + request: { body: { workspace } }, + status: 200, + body: undefined, + } + + await handler(ctx, async () => {}) + + expect(ctx.status).toBe(409) + expect(ctx.body).toEqual({ error: 'still running' }) + expect(releaseSessionFence).toHaveBeenCalledTimes(1) + expect(storage.getRoom('room-1')?.workspace).toBe('') + server.getIO().close() + }) + it('ignores unvalidated workspace values hidden in create-room compression config', async () => { const { GroupChatServer } = await import('../../packages/server/src/services/hermes/group-chat') const { setGroupChatServer } = await import('../../packages/server/src/routes/hermes/group-chat') diff --git a/tests/server/sessions-controller.test.ts b/tests/server/sessions-controller.test.ts index 7722294391..d67d7f9230 100644 --- a/tests/server/sessions-controller.test.ts +++ b/tests/server/sessions-controller.test.ts @@ -270,7 +270,7 @@ describe('session conversations controller', () => { } }) - it('lists Windows junction-like workspace folders even when their target realpath leaves WORKSPACE_BASE', async () => { + it('blocks Windows junction-like workspace folders that escape WORKSPACE_BASE', async () => { const originalPlatform = process.platform const originalWorkspaceBase = process.env.WORKSPACE_BASE const workspaceBase = await mkdtemp(join(tmpdir(), 'hermes-workspace-win-picker-')) @@ -291,7 +291,7 @@ describe('session conversations controller', () => { await mod.listWorkspaceFolders(rootCtx) expect(rootCtx.status).toBeUndefined() - expect(rootCtx.body.folders).toContainEqual({ + expect(rootCtx.body.folders).not.toContainEqual({ name: 'DrivesD', path: 'DrivesD', fullPath: outsideLink, @@ -300,10 +300,8 @@ describe('session conversations controller', () => { const nestedCtx: any = { query: { path: 'DrivesD' }, body: null } await mod.listWorkspaceFolders(nestedCtx) - expect(nestedCtx.status).toBeUndefined() - expect(nestedCtx.body.folders).toEqual([ - { name: 'project', path: 'DrivesD/project', fullPath: join(outsideLink, 'project') }, - ]) + expect(nestedCtx.status).toBe(403) + expect(nestedCtx.body).toEqual({ error: 'Access denied' }) } finally { Object.defineProperty(process, 'platform', { value: originalPlatform }) if (originalWorkspaceBase === undefined) delete process.env.WORKSPACE_BASE