/** * Sync Engine for HAPI Telegram Bot (Direct Connect) * * In the direct-connect architecture: * - hapi-hub is the hub (Socket.IO + REST) * - hapi CLI connects directly to the hub (no relay) * - No E2E encryption; data is stored as JSON in SQLite */ import { isKnownFlavor, type LocalResumeTarget, type ResumableSession } from '@hapi/protocol' import type { SlashCommandsResponse } from '@hapi/protocol/apiTypes' import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types' import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages' import type { Server } from 'socket.io' import type { Store, CancelQueuedMessageResult } from '../store' import type { HapiSessionExportResult } from '@hapi/protocol/sessionExport' import type { RpcRegistry } from '../socket/rpcRegistry' import type { SSEManager } from '../sse/sseManager' import { EventPublisher, type SyncEventListener } from './eventPublisher' import { MachineCache, type Machine } from './machineCache' import { MessageService } from './messageService' import { RpcGateway, type RpcCodexModel, type RpcCommandResponse, type RpcDeleteUploadResponse, type RpcGeneratedImageResponse, type RpcListDirectoryResponse, type RpcListCodexModelsResponse, type RpcListCursorModelsResponse, type RpcListOpencodeModelsResponse, type RpcCursorModel, type RpcOpencodeModel, type RpcPathExistsResponse, type RpcReadFileResponse, type RpcUploadFileResponse } from './rpcGateway' import { SessionCache } from './sessionCache' export type { Session, SyncEvent } from '@hapi/protocol/types' export type { Machine } from './machineCache' export type { SyncEventListener } from './eventPublisher' export type { RpcCodexModel, RpcCommandResponse, RpcDeleteUploadResponse, RpcGeneratedImageResponse, RpcListDirectoryResponse, RpcListCodexModelsResponse, RpcListCursorModelsResponse, RpcListOpencodeModelsResponse, RpcCursorModel, RpcOpencodeModel, RpcPathExistsResponse, RpcReadFileResponse, RpcUploadFileResponse } from './rpcGateway' export type ResumeSessionResult = | { type: 'success'; sessionId: string } | { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'no_machine_online' | 'resume_unavailable' | 'resume_failed' } export type ReopenSessionResult = | { type: 'success'; sessionId: string; resumed: boolean; cursorSessionProtocol?: 'acp' | 'stream-json' } | { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'no_machine_online' | 'resume_unavailable' | 'resume_failed' | 'metadata_conflict' } | { type: 'incomplete'; message: string; missing: [string, ...string[]] } export type LocalResumeTargetResult = | { type: 'success'; target: LocalResumeTarget } | { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'resume_unavailable' } export type LocalHandoffResult = | { type: 'success' } | { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'already_local' | 'handoff_failed' } function asRecord(value: unknown): Record | null { return value !== null && typeof value === 'object' && !Array.isArray(value) ? value as Record : null } function normalizeUserMessageText(value: string): string | undefined { const text = value.trim().replace(/\s+/g, ' ') return text.length > 0 ? text : undefined } function extractUserMessageText(content: unknown): string | undefined { if (typeof content === 'string') { return normalizeUserMessageText(content) } if (Array.isArray(content)) { const parts = content .map((block) => { const record = asRecord(block) return record?.type === 'text' && typeof record.text === 'string' ? record.text : null }) .filter((text): text is string => text !== null) return normalizeUserMessageText(parts.join(' ')) } const record = asRecord(content) if (record?.type === 'text' && typeof record.text === 'string') { return normalizeUserMessageText(record.text) } return undefined } function extractClaudeUserMessageTextFromAgentOutput(content: unknown): string | undefined { const record = asRecord(content) if (record?.type !== 'output') return undefined const data = asRecord(record.data) if (data?.type !== 'user') return undefined const message = asRecord(data.message) if (message?.role !== 'user') return undefined return extractUserMessageText(message.content) } export class SyncEngine { private readonly eventPublisher: EventPublisher private readonly sessionCache: SessionCache private readonly machineCache: MachineCache private readonly messageService: MessageService private readonly rpcGateway: RpcGateway private inactivityTimer: NodeJS.Timeout | null = null constructor( private readonly store: Store, io: Server, rpcRegistry: RpcRegistry, sseManager: SSEManager ) { this.eventPublisher = new EventPublisher(sseManager, (event) => this.resolveNamespace(event)) this.sessionCache = new SessionCache(store, this.eventPublisher) this.machineCache = new MachineCache(store, this.eventPublisher) this.messageService = new MessageService( store, io, this.eventPublisher, (sessionId, updatedAt) => this.recordSessionActivity(sessionId, updatedAt) ) this.rpcGateway = new RpcGateway(io, rpcRegistry) this.reloadAll() this.inactivityTimer = setInterval(() => this.expireInactive(), 5_000) } stop(): void { if (this.inactivityTimer) { clearInterval(this.inactivityTimer) this.inactivityTimer = null } } subscribe(listener: SyncEventListener): () => void { return this.eventPublisher.subscribe(listener) } private resolveNamespace(event: SyncEvent): string | undefined { if (event.namespace) { return event.namespace } if ('sessionId' in event) { return this.getSession(event.sessionId)?.namespace } if ('machineId' in event) { return this.machineCache.getMachine(event.machineId)?.namespace } return undefined } getSessions(): Session[] { return this.sessionCache.getSessions() } getSessionsByNamespace(namespace: string): Session[] { return this.sessionCache.getSessionsByNamespace(namespace) } getFutureScheduledMessageCounts(sessionIds: string[], now: number = Date.now()): Map { return this.store.messages.countFutureScheduledBySessionIds(sessionIds, now) } getSession(sessionId: string): Session | undefined { return this.sessionCache.getSession(sessionId) ?? this.sessionCache.refreshSession(sessionId) ?? undefined } getSessionByNamespace(sessionId: string, namespace: string): Session | undefined { const session = this.sessionCache.getSessionByNamespace(sessionId, namespace) ?? this.sessionCache.refreshSession(sessionId) if (!session || session.namespace !== namespace) { return undefined } return session } resolveSessionAccess( sessionId: string, namespace: string ): { ok: true; sessionId: string; session: Session } | { ok: false; reason: 'not-found' | 'access-denied' } { return this.sessionCache.resolveSessionAccess(sessionId, namespace) } getActiveSessions(): Session[] { return this.sessionCache.getActiveSessions() } getMachines(): Machine[] { return this.machineCache.getMachines() } getMachinesByNamespace(namespace: string): Machine[] { return this.machineCache.getMachinesByNamespace(namespace) } getMachine(machineId: string): Machine | undefined { return this.machineCache.getMachine(machineId) } getMachineByNamespace(machineId: string, namespace: string): Machine | undefined { return this.machineCache.getMachineByNamespace(machineId, namespace) } getOnlineMachines(): Machine[] { return this.machineCache.getOnlineMachines() } getOnlineMachinesByNamespace(namespace: string): Machine[] { return this.machineCache.getOnlineMachinesByNamespace(namespace) } getMessagesPage( sessionId: string, options: { limit: number; before?: { at: number; seq: number } | null } ): { messages: DecryptedMessage[] page: { limit: number nextBeforeSeq: number | null nextBeforeAt: number | null hasMore: boolean } } { return this.messageService.getMessagesPage(sessionId, options) } getSessionExport(sessionId: string, session: Session): HapiSessionExportResult { return this.messageService.getSessionExport(sessionId, session) } getDeliverableMessagesAfter(sessionId: string, options: { afterSeq: number; limit: number; now: number }): DecryptedMessage[] { return this.messageService.getDeliverableMessagesAfter(sessionId, options) } handleRealtimeEvent(event: SyncEvent): void { if (event.type === 'session-updated' && event.sessionId) { // Snapshot agent session IDs before refresh — safe because JS is single-threaded // and refreshSession replaces the Map entry with a new object. const before = this.sessionCache.getSession(event.sessionId) this.sessionCache.refreshSession(event.sessionId) const after = this.sessionCache.getSession(event.sessionId) if (after?.metadata && !this.hasSameAgentSessionIds(before?.metadata ?? null, after.metadata)) { void this.sessionCache.deduplicateByAgentSessionId(event.sessionId).catch(() => { // best-effort: dedup failure is harmless, web-side safety net hides remaining duplicates }) } return } if (event.type === 'machine-updated' && event.machineId) { this.machineCache.refreshMachine(event.machineId) return } if (event.type === 'message-received' && event.sessionId) { if (!this.getSession(event.sessionId)) { this.sessionCache.refreshSession(event.sessionId) } } this.eventPublisher.emit(event) } handleSessionAlive(payload: { sid: string time: number thinking?: boolean mode?: 'local' | 'remote' permissionMode?: PermissionMode model?: string | null modelReasoningEffort?: string | null effort?: string | null collaborationMode?: CodexCollaborationMode }): void { this.sessionCache.handleSessionAlive(payload) this.triggerDedupIfNeeded(payload.sid) } clearQueuedThinkingGrace(sessionId: string): void { this.sessionCache.clearQueuedThinkingGrace(sessionId) } handleSessionEnd(payload: { sid: string; time: number; reason?: 'completed' | 'terminated' | 'error' }): void { this.sessionCache.handleSessionEnd(payload) this.eventPublisher.emit({ type: 'session-ended', sessionId: payload.sid, reason: payload.reason }) // Retry dedup now that this session is inactive — a prior dedup may have // skipped it because it was still active at the time. this.triggerDedupIfNeeded(payload.sid) } handleBackgroundTaskDelta(sessionId: string, delta: { started: number; completed: number }): void { this.sessionCache.applyBackgroundTaskDelta(sessionId, delta) } recordSessionActivity(sessionId: string, updatedAt: number): void { this.sessionCache.recordSessionActivity(sessionId, updatedAt) } handleMachineAlive(payload: { machineId: string; time: number }): void { this.machineCache.handleMachineAlive(payload) } private expireInactive(): void { const expired = this.sessionCache.expireInactive() // Sort by most recent first so dedup keeps the newest session when multiple // duplicates for the same agent thread expire in the same sweep. const sorted = expired .map((id) => this.sessionCache.getSession(id)) .filter((s): s is NonNullable => s != null) .sort((a, b) => (b.activeAt - a.activeAt) || (b.updatedAt - a.updatedAt)) for (const session of sorted) { this.triggerDedupIfNeeded(session.id) } this.machineCache.expireInactive() // Piggybacked on the inactivity tick; not a logical part of expireInactive // but shares its 5s cadence (avoids a second timer). this.messageService.releaseMatureScheduledMessages(Date.now()) } private reloadAll(): void { this.sessionCache.reloadAll() this.machineCache.reloadAll() } getOrCreateSession( tag: string, metadata: unknown, agentState: unknown, namespace: string, model?: string, effort?: string, modelReasoningEffort?: string ): Session { return this.sessionCache.getOrCreateSession(tag, metadata, agentState, namespace, model, effort, modelReasoningEffort) } getOrCreateMachine(id: string, metadata: unknown, runnerState: unknown, namespace: string): Machine { return this.machineCache.getOrCreateMachine(id, metadata, runnerState, namespace) } async sendMessage( sessionId: string, payload: { text: string localId?: string | null attachments?: Array<{ id: string filename: string mimeType: string size: number path: string previewUrl?: string }> sentFrom?: 'telegram-bot' | 'webapp' scheduledAt?: number | null } ): Promise { await this.messageService.sendMessage(sessionId, payload) this.sessionCache.markMessageQueued(sessionId) this.sessionCache.recordSessionActivity(sessionId, Date.now()) } async cancelQueuedMessage( sessionId: string, messageId: string ): Promise { return this.messageService.cancelQueuedMessage(sessionId, messageId) } sweepImmediateQueuedOnSessionEnd(sessionId: string, invokedAt: number): void { this.messageService.sweepImmediateQueuedOnSessionEnd(sessionId, invokedAt) } async approvePermission( sessionId: string, requestId: string, mode?: PermissionMode, allowTools?: string[], decision?: 'approved' | 'approved_for_session' | 'denied' | 'abort', answers?: Record | Record ): Promise { await this.rpcGateway.approvePermission(sessionId, requestId, mode, allowTools, decision, answers) } async denyPermission( sessionId: string, requestId: string, decision?: 'approved' | 'approved_for_session' | 'denied' | 'abort' ): Promise { await this.rpcGateway.denyPermission(sessionId, requestId, decision) } async abortSession(sessionId: string): Promise { await this.rpcGateway.abortSession(sessionId) } async archiveSession(sessionId: string): Promise { await this.rpcGateway.killSession(sessionId) this.handleSessionEnd({ sid: sessionId, time: Date.now() }) } async switchSession(sessionId: string, to: 'remote' | 'local'): Promise { await this.rpcGateway.switchSession(sessionId, to) } async renameSession(sessionId: string, name: string): Promise { await this.sessionCache.renameSession(sessionId, name) } async deleteSession(sessionId: string): Promise { await this.sessionCache.deleteSession(sessionId) } async applySessionConfig( sessionId: string, config: { permissionMode?: PermissionMode model?: string | null modelReasoningEffort?: string | null effort?: string | null collaborationMode?: CodexCollaborationMode } ): Promise { const session = this.sessionCache.getSession(sessionId) if (!session?.active) { // For inactive sessions, update the in-memory cache directly without // an RPC call — the CLI is not running yet. The updated value will be // passed to the spawned process when the session is resumed. this.sessionCache.applySessionConfig(sessionId, config) return } const result = await this.rpcGateway.requestSessionConfig(sessionId, config) if (!result || typeof result !== 'object') { throw new Error('Invalid response from session config RPC') } const obj = result as { error?: string applied?: { permissionMode?: Session['permissionMode'] model?: Session['model'] modelReasoningEffort?: Session['modelReasoningEffort'] effort?: Session['effort'] collaborationMode?: Session['collaborationMode'] } } if (typeof obj.error === 'string' && obj.error.trim().length > 0) { throw new Error(obj.error) } const applied = obj.applied if (!applied || typeof applied !== 'object') { throw new Error('Missing applied session config') } const requestedKeys = Object.keys(config) as Array for (const key of requestedKeys) { if (!(key in applied)) { throw new Error(`Session did not apply ${key}`) } } this.sessionCache.applySessionConfig(sessionId, applied) } async spawnSession( machineId: string, directory: string, agent: AgentFlavor = 'claude', model?: string, modelReasoningEffort?: string, yolo?: boolean, sessionType?: 'simple' | 'worktree', worktreeName?: string, resumeSessionId?: string, effort?: string, permissionMode?: PermissionMode ): Promise<{ type: 'success'; sessionId: string } | { type: 'error'; message: string }> { return await this.rpcGateway.spawnSession( machineId, directory, agent, model, modelReasoningEffort, yolo, sessionType, worktreeName, resumeSessionId, effort, permissionMode ) } private resolveFlavor(session: Session): AgentFlavor { const flavor = session.metadata?.flavor return isKnownFlavor(flavor) ? flavor : 'claude' } private resolveAgentResumeId(session: Session, namespace: string): string | null { const metadata = session.metadata if (!metadata) { return null } const flavor = this.resolveFlavor(session) if (flavor === 'codex') return metadata.codexSessionId ?? null if (flavor === 'gemini') return metadata.geminiSessionId ?? null if (flavor === 'opencode') return metadata.opencodeSessionId ?? null if (flavor === 'cursor') return metadata.cursorSessionId ?? null if (flavor === 'kimi') return metadata.kimiSessionId ?? null return metadata.claudeSessionId ?? this.recoverClaudeSessionIdFromMessages(session.id, namespace) } resolveLocalResumeTarget(sessionId: string, namespace: string): LocalResumeTargetResult { const access = this.sessionCache.resolveSessionAccess(sessionId, namespace) if (!access.ok) { return { type: 'error', message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' } } const session = access.session const metadata = session.metadata if (!metadata || typeof metadata.path !== 'string' || metadata.path.length === 0) { return { type: 'error', message: 'Session metadata missing path', code: 'resume_unavailable' } } const agentSessionId = this.resolveAgentResumeId(session, namespace) if (!agentSessionId) { return { type: 'error', message: 'Resume session ID unavailable. Start a new session in this directory, or retry after the agent has initialized.', code: 'resume_unavailable' } } return { type: 'success', target: { sessionId: access.sessionId, flavor: this.resolveFlavor(session), directory: metadata.path, machineId: metadata.machineId, host: metadata.host, active: session.active, thinking: session.thinking, controlledByUser: session.agentState?.controlledByUser === true, agentSessionId, model: session.model ?? null, effort: session.effort ?? null, modelReasoningEffort: session.modelReasoningEffort ?? null, permissionMode: session.permissionMode, collaborationMode: session.collaborationMode } } } listLocalResumableSessions(namespace: string, opts?: { machineId?: string }): ResumableSession[] { return this.getSessionsByNamespace(namespace) .map((session) => this.resolveLocalResumeTarget(session.id, namespace)) .filter((result): result is { type: 'success'; target: LocalResumeTarget } => result.type === 'success') .map(({ target }) => { const session = this.getSessionByNamespace(target.sessionId, namespace) return { sessionId: target.sessionId, flavor: target.flavor, directory: target.directory, machineId: target.machineId, host: target.host, active: target.active, thinking: target.thinking, controlledByUser: target.controlledByUser, agentSessionId: target.agentSessionId, model: target.model, effort: target.effort, modelReasoningEffort: target.modelReasoningEffort, permissionMode: target.permissionMode, collaborationMode: target.collaborationMode, updatedAt: session?.updatedAt ?? 0, name: session?.metadata?.name, summary: session?.metadata?.summary?.text, firstUserMessage: this.resolveFirstUserMessage(target.sessionId) } }) .filter((session) => !opts?.machineId || session.machineId === opts.machineId) .sort((a, b) => b.updatedAt - a.updatedAt) } private resolveFirstUserMessage(sessionId: string): string | undefined { for (const message of this.store.messages.getFirstMessages(sessionId, 50)) { const roleWrapped = unwrapRoleWrappedRecordEnvelope(message.content) const text = roleWrapped?.role === 'user' ? extractUserMessageText(roleWrapped.content) : roleWrapped?.role === 'agent' ? extractClaudeUserMessageTextFromAgentOutput(roleWrapped.content) : undefined if (text) return text } return undefined } /** Inactive session with directory path but no agent thread and no prior user turn. */ private canFreshSpawnNeverStartedSession(session: Session, sessionId: string, namespace: string): boolean { const metadata = session.metadata if (!metadata || typeof metadata.path !== 'string' || metadata.path.length === 0) { return false } if (this.resolveAgentResumeId(session, namespace)) { return false } return this.store.messages.getFirstMessages(sessionId, 1).length === 0 } async resumeSession(sessionId: string, namespace: string, opts?: { permissionMode?: PermissionMode }): Promise { const access = this.sessionCache.resolveSessionAccess(sessionId, namespace) if (!access.ok) { return { type: 'error', message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' } } const session = access.session if (session.active) { return { type: 'success', sessionId: access.sessionId } } const targetResult = this.resolveLocalResumeTarget(access.sessionId, namespace) let flavor: AgentFlavor let resumeToken: string | undefined let directory: string if (targetResult.type === 'success') { flavor = targetResult.target.flavor resumeToken = targetResult.target.agentSessionId directory = targetResult.target.directory } else if ( targetResult.code === 'resume_unavailable' && this.canFreshSpawnNeverStartedSession(session, access.sessionId, namespace) ) { const metadata = session.metadata! flavor = this.resolveFlavor(session) resumeToken = undefined directory = metadata.path } else { return targetResult } const metadata = session.metadata! const onlineMachines = this.machineCache.getOnlineMachinesByNamespace(namespace) if (onlineMachines.length === 0) { return { type: 'error', message: 'No machine online', code: 'no_machine_online' } } const targetMachine = (() => { if (metadata.machineId) { const exact = onlineMachines.find((machine) => machine.id === metadata.machineId) if (exact) return exact } if (metadata.host) { const hostMatch = onlineMachines.find((machine) => machine.metadata?.host === metadata.host) if (hostMatch) return hostMatch } return null })() if (!targetMachine) { return { type: 'error', message: 'No machine online', code: 'no_machine_online' } } const preferredPermissionMode = opts?.permissionMode ?? session.permissionMode ?? session.metadata?.preferredPermissionMode const spawnResult = await this.rpcGateway.spawnSession( targetMachine.id, directory, flavor, session.model ?? undefined, session.modelReasoningEffort ?? undefined, undefined, undefined, undefined, resumeToken, session.effort ?? undefined, preferredPermissionMode ) if (spawnResult.type !== 'success') { return { type: 'error', message: spawnResult.message, code: 'resume_failed' } } const becameActive = await this.waitForSessionActive(spawnResult.sessionId) if (!becameActive) { return { type: 'error', message: 'Session failed to become active', code: 'resume_failed' } } // permissionMode is passed to spawnSession above; do not call set-session-config here. // session-alive can arrive before the CLI registers that RPC handler, which caused resume_failed. if (spawnResult.sessionId !== access.sessionId) { // The old session may have already been merged by the automatic dedup path // (triggered when the spawned CLI sets its agent session ID in metadata). // Only attempt the explicit merge if the old session still exists. const oldSession = this.sessionCache.getSessionByNamespace(access.sessionId, namespace) if (oldSession) { try { await this.sessionCache.mergeSessions(access.sessionId, spawnResult.sessionId, namespace) } catch (error) { const message = error instanceof Error ? error.message : 'Failed to merge resumed session' return { type: 'error', message, code: 'resume_failed' } } } } return { type: 'success', sessionId: spawnResult.sessionId } } /** * Revive an archived session so the web UI can reach it again. * * Behaviour: * - Active session: idempotent no-op (`resumed: false`). * - Non-archived inactive session: forwards to `resumeSession` without touching metadata. * - Archived session: validates that the agent has enough metadata to resume (Cursor * sessions require a `cursorSessionId` once they have any messages), clears the * archive metadata (`lifecycleState`, `archivedBy`, `archiveReason`), defaults the * Cursor protocol to `stream-json` for pre-#799 sessions, then forwards to * `resumeSession`. The CLI's `sessionFactory` will re-stamp `lifecycleState='running'` * when it boots, so we do not pre-write that here. * * Failure rollback: if `resumeSession` fails (no machine online, spawn timeout, etc.) * the archive snapshot is restored so the operator can retry without losing * `archiveReason`/`archivedBy`/`lifecycleState` and the UI still shows the row as * archived rather than a dangling inactive non-archived ghost. * * Returns `incomplete` (HTTP 422 from the route layer) when the agent metadata * needed to resume is missing. */ async reopenSession(sessionId: string, namespace: string): Promise { const access = this.sessionCache.resolveSessionAccess(sessionId, namespace) if (!access.ok) { return { type: 'error', message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' } } const session = access.session const metadata = session.metadata if (session.active) { return { type: 'success', sessionId: access.sessionId, resumed: false } } const isArchived = metadata?.lifecycleState === 'archived' if (isArchived && metadata) { if (metadata.flavor === 'cursor' && !metadata.cursorSessionId) { const hasMessages = this.store.messages.getFirstMessages(access.sessionId, 1).length > 0 if (hasMessages) { return { type: 'incomplete', message: 'Cursor session id is missing from metadata; reopen requires the original cursor chat id', missing: ['cursorSessionId'] } } } const archiveSnapshot = { lifecycleState: metadata.lifecycleState, archivedBy: metadata.archivedBy, archiveReason: metadata.archiveReason, lifecycleStateSince: metadata.lifecycleStateSince } let applied: { cursorSessionProtocol?: 'acp' | 'stream-json' } try { applied = await this.sessionCache.clearSessionArchiveMetadata(access.sessionId) } catch (error) { const message = error instanceof Error ? error.message : 'Failed to clear archive metadata' return { type: 'error', message, code: 'metadata_conflict' } } const resumeResult = await this.resumeSession(access.sessionId, namespace) if (resumeResult.type === 'error') { // Resume failed - put the archive flags back so the row stays archived in the UI // and the operator can retry. Best-effort: a concurrent metadata write that // succeeded between clear and restore (e.g. an unrelated rename) wins, in // which case we surface the original resume error rather than masking it. try { await this.sessionCache.restoreSessionArchiveMetadata(access.sessionId, archiveSnapshot) } catch { // Swallow restore failures - the resume error is the more important signal. } return resumeResult } return { type: 'success', sessionId: resumeResult.sessionId, resumed: true, ...(applied.cursorSessionProtocol ? { cursorSessionProtocol: applied.cursorSessionProtocol } : {}) } } // Not active and not archived (e.g. brand-new session that has not yet connected, // or one that ended without writing archive metadata). Forward to resume so the // operator still gets one-click revival. const resumeResult = await this.resumeSession(access.sessionId, namespace) if (resumeResult.type === 'error') { return resumeResult } return { type: 'success', sessionId: resumeResult.sessionId, resumed: true } } async handoffSessionToLocal(sessionId: string, namespace: string): Promise { const access = this.sessionCache.resolveSessionAccess(sessionId, namespace) if (!access.ok) { return { type: 'error', message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' } } if (!access.session.active) { return { type: 'success' } } if (access.session.agentState?.controlledByUser === true) { return { type: 'error', message: 'Session is already controlled by a local terminal', code: 'already_local' } } try { await this.rpcGateway.handoffSessionToLocal(access.sessionId) } catch (error) { return { type: 'error', message: error instanceof Error ? error.message : String(error), code: 'handoff_failed' } } const inactive = await this.waitForSessionInactive(access.sessionId) if (!inactive) { return { type: 'error', message: 'Timed out waiting for remote session to hand off', code: 'handoff_failed' } } return { type: 'success' } } private recoverClaudeSessionIdFromMessages(sessionId: string, namespace: string): string | null { const messages = this.messageService.getMessages(sessionId, 200) for (let i = messages.length - 1; i >= 0; i -= 1) { const found = this.extractClaudeSessionId(messages[i].content) if (!found) continue this.persistRecoveredClaudeSessionId(sessionId, namespace, found) return found } return null } private extractClaudeSessionId(value: unknown): string | null { if (!value || typeof value !== 'object') { return null } const obj = value as Record const direct = this.normalizeClaudeSessionId(obj.session_id) ?? this.normalizeClaudeSessionId(obj.sessionId) if (direct) { return direct } const content = obj.content if (content && typeof content === 'object') { const found = this.extractClaudeSessionId(content) if (found) return found } const data = obj.data if (data && typeof data === 'object') { const found = this.extractClaudeSessionId(data) if (found) return found } return null } private normalizeClaudeSessionId(value: unknown): string | null { if (typeof value !== 'string') { return null } const trimmed = value.trim() return /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(trimmed) ? trimmed : null } private persistRecoveredClaudeSessionId(sessionId: string, namespace: string, claudeSessionId: string): void { for (let attempt = 0; attempt < 2; attempt += 1) { const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace) ?? this.sessionCache.refreshSession(sessionId) if (!latest?.metadata) return if (latest.metadata.claudeSessionId === claudeSessionId) return const result = this.store.sessions.updateSessionMetadata( sessionId, { ...latest.metadata, claudeSessionId }, latest.metadataVersion, namespace, { touchUpdatedAt: false } ) if (result.result === 'success') { this.sessionCache.refreshSession(sessionId) return } if (result.result !== 'version-mismatch') { return } } } private hasSameAgentSessionIds( prev: Session['metadata'] | null, next: NonNullable ): boolean { return (prev?.codexSessionId ?? null) === (next.codexSessionId ?? null) && (prev?.claudeSessionId ?? null) === (next.claudeSessionId ?? null) && (prev?.geminiSessionId ?? null) === (next.geminiSessionId ?? null) && (prev?.opencodeSessionId ?? null) === (next.opencodeSessionId ?? null) && (prev?.cursorSessionId ?? null) === (next.cursorSessionId ?? null) } private triggerDedupIfNeeded(sessionId: string): void { const session = this.sessionCache.getSession(sessionId) if (session?.metadata) { void this.sessionCache.deduplicateByAgentSessionId(sessionId).catch(() => { // best-effort: web-side safety net hides remaining duplicates }) } } async waitForSessionActive(sessionId: string, timeoutMs: number = 15_000): Promise { const start = Date.now() while (Date.now() - start < timeoutMs) { const session = this.getSession(sessionId) if (session?.active) { return true } await new Promise((resolve) => setTimeout(resolve, 250)) } return false } async waitForSessionInactive(sessionId: string, timeoutMs: number = 15_000): Promise { const start = Date.now() while (Date.now() - start < timeoutMs) { const session = this.getSession(sessionId) if (!session?.active) { return true } await new Promise((resolve) => setTimeout(resolve, 250)) } return false } async checkPathsExist(machineId: string, paths: string[]): Promise> { return await this.rpcGateway.checkPathsExist(machineId, paths) } async listMachineDirectory(machineId: string, path: string): Promise { return await this.rpcGateway.listMachineDirectory(machineId, path) } async getGitStatus(sessionId: string, cwd?: string): Promise { return await this.rpcGateway.getGitStatus(sessionId, cwd) } async getGitDiffNumstat(sessionId: string, options: { cwd?: string; staged?: boolean }): Promise { return await this.rpcGateway.getGitDiffNumstat(sessionId, options) } async getGitDiffFile(sessionId: string, options: { cwd?: string; filePath: string; staged?: boolean }): Promise { return await this.rpcGateway.getGitDiffFile(sessionId, options) } async readSessionFile(sessionId: string, path: string): Promise { return await this.rpcGateway.readSessionFile(sessionId, path) } async readGeneratedImage(sessionId: string, imageId: string): Promise { return await this.rpcGateway.readGeneratedImage(sessionId, imageId) } async listDirectory(sessionId: string, path: string): Promise { return await this.rpcGateway.listDirectory(sessionId, path) } async uploadFile(sessionId: string, filename: string, content: string, mimeType: string): Promise { return await this.rpcGateway.uploadFile(sessionId, filename, content, mimeType) } async deleteUploadFile(sessionId: string, path: string): Promise { return await this.rpcGateway.deleteUploadFile(sessionId, path) } async runRipgrep(sessionId: string, args: string[], cwd?: string): Promise { return await this.rpcGateway.runRipgrep(sessionId, args, cwd) } async listSlashCommands(sessionId: string, agent: string): Promise { return await this.rpcGateway.listSlashCommands(sessionId, agent) } async listSkills(sessionId: string, flavor?: string): Promise<{ success: boolean skills?: Array<{ name: string; description?: string }> error?: string }> { return await this.rpcGateway.listSkills(sessionId, flavor) } async listCodexModelsForSession(sessionId: string): Promise { return await this.rpcGateway.listCodexModelsForSession(sessionId) } async listCodexModelsForMachine(machineId: string): Promise { return await this.rpcGateway.listCodexModelsForMachine(machineId) } async listCursorModelsForSession(sessionId: string): Promise { return await this.rpcGateway.listCursorModelsForSession(sessionId) } async listCursorModelsForMachine(machineId: string): Promise { return await this.rpcGateway.listCursorModelsForMachine(machineId) } async listOpencodeModelsForSession(sessionId: string): Promise { return await this.rpcGateway.listOpencodeModelsForSession(sessionId) } async listOpencodeModelsForCwd(machineId: string, cwd: string): Promise { return await this.rpcGateway.listOpencodeModelsForCwd(machineId, cwd) } }