Repository Public
1/**2 * Terminal session state. Every transition goes through `transition()`, which enforces3 * `canTransition` from @cr/shared under a row lock, persists a `terminal_events` row and4 * emits a `session.updated` ServerEvent. The reconciler rules live here too:5 *6 * - runtime unreachable ⇒ nothing changes (never guess); health is reported OFFLINE7 * - runtime healthy, session gone ⇒ RECOVERABLE (STARTING ⇒ FAILED, STOPPING ⇒ STOPPED)8 * - pane dead / `session.exit` ⇒ STOPPED (exit 0 or unknown) / FAILED (non-zero), exit_code stored9 * - attached = 0 ⇒ DETACHED, > 0 ⇒ RUNNING — unless WAITING_FOR_INPUT. "attached" is the larger10 * of what the runtime last reported (tmux clients) and the terminal streams this server is11 * proxying right now: a browser that attaches makes the session RUNNING at once12 * (`streamOpened`), without waiting for the runtime to notice; after a stream closes the13 * runtime is asked for the truth (`streamClosed`), so a client attached elsewhere keeps it14 * RUNNING. Attach/detach bookkeeping never touches WAITING_FOR_INPUT.15 * - hook Notification/Stop ⇒ WAITING_FOR_INPUT; UserPromptSubmit ⇒ RUNNING; any hook16 * carrying a conversation id valid for the agent profile stores `claude_conversation_id`17 * - agent profiles WITHOUT a prompt-submit hook (Codex: its `notify` program only reports the18 * end of a turn): the first terminal input from a browser after WAITING_FOR_INPUT ⇒ RUNNING19 * (`inputReceived`, at most one transition per waiting period). Never done for Claude.20 * WAITING_FOR_INPUT is only ever set from a hook; terminal output is never parsed.21 */22import type { FastifyBaseLogger } from 'fastify';23import { AGENT_PROFILES, LIVE_SESSION_STATES, canTransition, isConversationId, type AgentProfile, type RtEvent, type RtSessionInfo, type SessionState, type TerminalSession } from '@cr/shared';24import type { Db, Queryable, SqlParam } from '../../db.js';25import type { EventHub } from '../../events-hub.js';26import { RuntimeError, type RuntimeClient } from '../../runtime-client.js';27import type { Notifier } from '../notifications/service.js';28import { getSettings } from '../settings/store.js';2930export interface SessionRow {31 id: string;32 owner_id: string;33 project_id: string | null;34 project_name: string | null;35 kind: 'claude' | 'shell';36 title: string;37 state: SessionState;38 cwd_abs: string;39 cwd_rel: string;40 pid: number | null;41 claude_conversation_id: string | null;42 launch_args: string[];43 pinned: boolean;44 archived: boolean;45 exit_code: number | null;46 waiting_reason: string | null;47 created_at: Date;48 last_activity_at: Date | null;49 ended_at: Date | null;50}51export const SESSION_SELECT = `SELECT s.*, p.name AS project_name FROM terminal_sessions s LEFT JOIN projects p ON p.id = s.project_id`;52export const isLive = (s: SessionState): boolean => LIVE_SESSION_STATES.includes(s);53const ENDED: readonly SessionState[] = ['STOPPED', 'FAILED'];5455export interface Patch {56 pid?: number | null;57 exitCode?: number | null;58 waitingReason?: string | null;59 conversationId?: string;60 lastActivityMs?: number | null;61}6263/** After a stream closes: when to ask the runtime how many clients are left (tmux drops the client asynchronously). */64const DETACH_CHECK_DELAYS_MS = [250, 1500];6566export class SessionService {67 /** tmux clients attached per session, as last reported by the runtime. */68 readonly attached = new Map<string, number>();69 /** Terminal streams this server is proxying right now, per session (first-hand knowledge). */70 private readonly streams = new Map<string, number>();71 private readonly detachChecks = new Map<string, NodeJS.Timeout>();72 private stopped = false;73 /**74 * Sessions known to be in WAITING_FOR_INPUT (maintained by `transition`, refreshed by the75 * reconciler after a restart). Lets `inputReceived` ignore keystrokes without a database query.76 */77 private readonly waiting = new Set<string>();78 private readonly agent: AgentProfile;79 /** Last observed runtime reachability (null = not checked yet). */80 runtimeUp: boolean | null = null;81 runtimeCheckedAt: Date | null = null;8283 constructor(84 private readonly d: { db: Db; rt: RuntimeClient; hub: EventHub; notifier: Notifier; log: FastifyBaseLogger; agent?: AgentProfile },85 ) {86 this.agent = d.agent ?? AGENT_PROFILES.claude;87 }8889 /** True when terminal input must end WAITING_FOR_INPUT (the agent CLI has no prompt-submit hook). */90 get inputEndsWaiting(): boolean {91 return !this.agent.capabilities.promptSubmitHook;92 }9394 /**95 * A browser sent terminal input (binary frame) to this session. Only meaningful for profiles96 * without a prompt-submit hook: the first input after WAITING_FOR_INPUT returns the session to97 * RUNNING. Cheap for every other keystroke (one Set lookup, no query).98 */99 async inputReceived(id: string): Promise<void> {100 if (!this.inputEndsWaiting || this.stopped || !this.waiting.delete(id)) return;101 const row = await this.byId(id);102 if (!row || row.state !== 'WAITING_FOR_INPUT') return;103 await this.transition(id, 'RUNNING', {}, { source: 'input' });104 }105106 dto(r: SessionRow): TerminalSession {107 return {108 id: r.id,109 projectId: r.project_id,110 projectName: r.project_name,111 kind: r.kind,112 title: r.title,113 state: r.state,114 cwd: r.cwd_rel,115 pid: r.pid,116 claudeConversationId: r.claude_conversation_id,117 pinned: r.pinned,118 archived: r.archived,119 attachedClients: isLive(r.state) ? this.attachedCount(r.id) : 0,120 exitCode: r.exit_code,121 createdAt: r.created_at.toISOString(),122 lastActivityAt: r.last_activity_at?.toISOString() ?? null,123 endedAt: r.ended_at?.toISOString() ?? null,124 waitingReason: r.waiting_reason,125 };126 }127128 /** Clients attached to a session: never fewer than the streams proxied by this server. */129 attachedCount(id: string): number {130 return Math.max(this.attached.get(id) ?? 0, this.streams.get(id) ?? 0);131 }132133 /** RUNNING ⇄ DETACHED from the current client count; every other state is left alone. */134 private async settleAttachState(row: SessionRow, source: string): Promise<void> {135 const target: SessionState = this.attachedCount(row.id) > 0 ? 'RUNNING' : 'DETACHED';136 if ((row.state === 'RUNNING' || row.state === 'DETACHED') && row.state !== target) await this.transition(row.id, target, {}, { source });137 else this.emit(row); // same state (e.g. WAITING_FOR_INPUT), new `attachedClients`138 }139140 /** A browser terminal stream of this server reached the runtime. */141 async streamOpened(id: string): Promise<void> {142 this.streams.set(id, (this.streams.get(id) ?? 0) + 1);143 if (this.stopped) return;144 const row = await this.byId(id);145 if (row && isLive(row.state)) await this.settleAttachState(row, 'attach');146 }147148 /** A browser terminal stream ended. The runtime decides whether anyone is still attached. */149 async streamClosed(id: string): Promise<void> {150 const left = (this.streams.get(id) ?? 1) - 1;151 if (left > 0) this.streams.set(id, left);152 else this.streams.delete(id);153 // The tmux client of that stream is going away: stop counting it until the runtime reports.154 const known = this.attached.get(id);155 if (known !== undefined && known > left) this.attached.set(id, known - 1);156 if (this.stopped) return;157 const row = await this.byId(id);158 if (!row || !isLive(row.state)) return;159 if (left > 0) return void (await this.settleAttachState(row, 'detach'));160 this.emit(row);161 this.scheduleDetachCheck(id, 0);162 }163164 private scheduleDetachCheck(id: string, step: number): void {165 const delay = DETACH_CHECK_DELAYS_MS[step];166 if (delay === undefined || this.stopped) return; // the runtime's events and the periodic reconciler take over167 const old = this.detachChecks.get(id);168 if (old) clearTimeout(old);169 const timer = setTimeout(() => {170 this.detachChecks.delete(id);171 void this.detachCheck(id, step).catch((err: unknown) => this.d.log.error({ err, sessionId: id }, 'detach check failed'));172 }, delay);173 timer.unref();174 this.detachChecks.set(id, timer);175 }176177 private async detachCheck(id: string, step: number): Promise<void> {178 if (this.stopped || (this.streams.get(id) ?? 0) > 0) return; // someone attached again meanwhile179 const row = await this.byId(id);180 if (!row || !isLive(row.state)) return;181 try {182 const info = await this.d.rt.json<RtSessionInfo>('GET', `/sessions/${id}`, undefined, { timeoutMs: 5000 });183 await this.applyInfo(row, info);184 // Still counted by tmux (client not gone yet, or attached outside this server): look once more.185 if (info.alive && info.attached > 0) this.scheduleDetachCheck(id, step + 1);186 } catch (err) {187 if (!(err instanceof RuntimeError)) throw err;188 // 404 / unreachable: `session.exit` and the reconciler own those cases.189 }190 }191192 /** Stops pending timers (server shutdown). */193 stop(): void {194 this.stopped = true;195 for (const t of this.detachChecks.values()) clearTimeout(t);196 this.detachChecks.clear();197 }198199 byId(id: string, q: Queryable = this.d.db): Promise<SessionRow | null> {200 return q.one<SessionRow>(`${SESSION_SELECT} WHERE s.id = $1`, [id]);201 }202 owned(id: string, ownerId: string): Promise<SessionRow | null> {203 return this.d.db.one<SessionRow>(`${SESSION_SELECT} WHERE s.id = $1 AND s.owner_id = $2`, [id, ownerId]);204 }205 emit(row: SessionRow): void {206 this.d.hub.publish(row.owner_id, { type: 'session.updated', session: this.dto(row) });207 }208209 /**210 * Moves a session to `to` (plus optional column patch). Returns the updated row, or null when211 * the transition is not allowed from the current state (nothing is written in that case).212 */213 async transition(id: string, to: SessionState, patch: Patch = {}, detail: Record<string, unknown> = {}): Promise<SessionRow | null> {214 const result = await this.d.db.withTx(async (tx) => {215 const cur = await tx.one<{ state: SessionState }>('SELECT state FROM terminal_sessions WHERE id = $1 FOR UPDATE', [id]);216 if (!cur) return null;217 if (!canTransition(cur.state, to)) return { refused: cur.state };218 const sets: string[] = ['state = $2'];219 const params: SqlParam[] = [id, to];220 const set = (col: string, v: SqlParam): void => {221 params.push(v);222 sets.push(`${col} = $${params.length}`);223 };224 if (patch.pid !== undefined) set('pid', patch.pid);225 if (patch.exitCode !== undefined) set('exit_code', patch.exitCode);226 if (patch.conversationId !== undefined) set('claude_conversation_id', patch.conversationId);227 if (patch.lastActivityMs) set('last_activity_at', new Date(patch.lastActivityMs));228 if (patch.waitingReason !== undefined) set('waiting_reason', patch.waitingReason);229 else if (to !== 'WAITING_FOR_INPUT' && cur.state === 'WAITING_FOR_INPUT') sets.push('waiting_reason = NULL');230 if (ENDED.includes(to) && cur.state !== to) sets.push('ended_at = now()', 'pid = NULL');231 if (to === 'STARTING') sets.push('ended_at = NULL', 'exit_code = NULL');232 await tx.q(`UPDATE terminal_sessions SET ${sets.join(', ')} WHERE id = $1`, params);233 if (cur.state !== to) {234 await tx.q("INSERT INTO terminal_events (session_id, kind, detail) VALUES ($1, 'state', $2)", [235 id, JSON.stringify({ from: cur.state, to, ...detail }),236 ]);237 }238 return { changed: cur.state !== to, row: await this.byId(id, tx) };239 });240 if (!result) return null;241 if ('refused' in result) {242 this.d.log.warn({ sessionId: id, from: result.refused, to }, 'session transition refused');243 return null;244 }245 if (!result.row) return null;246 if (ENDED.includes(to)) this.attached.delete(id);247 if (to === 'WAITING_FOR_INPUT') this.waiting.add(id);248 else this.waiting.delete(id);249 this.emit(result.row);250 return result.row;251 }252253 /** Applies what the runtime reports for one session it knows about. */254 async applyInfo(row: SessionRow, info: RtSessionInfo): Promise<SessionRow | null> {255 const before = this.attachedCount(row.id);256 this.attached.set(row.id, info.attached);257 if (!info.alive) return this.applyExit(row, info.exitCode, 'reconcile');258 if (row.state === 'STOPPING' || !isLive(row.state)) return row;259 const target: SessionState = row.state === 'WAITING_FOR_INPUT' ? 'WAITING_FOR_INPUT' : this.attachedCount(row.id) > 0 ? 'RUNNING' : 'DETACHED';260 const activityChanged = info.lastActivityMs !== null && Math.abs(info.lastActivityMs - (row.last_activity_at?.getTime() ?? 0)) > 30_000;261 if (target === row.state && info.pid === row.pid && !activityChanged) {262 if (before !== this.attachedCount(row.id)) this.emit(row);263 return row;264 }265 return this.transition(row.id, target, { pid: info.pid, lastActivityMs: info.lastActivityMs }, { source: 'reconcile' });266 }267268 async applyExit(row: SessionRow, exitCode: number | null, source: string): Promise<SessionRow | null> {269 if (!isLive(row.state)) return row;270 const userStop = row.state === 'STOPPING';271 const to: SessionState = userStop || exitCode === 0 || exitCode === null ? 'STOPPED' : 'FAILED';272 const next = await this.transition(row.id, to, { exitCode }, { source, exitCode });273 // The dead pane is kept by tmux (remain-on-exit) only so the code can be read: drop it now.274 void this.d.rt.json('DELETE', `/sessions/${row.id}`, undefined, { query: { force: 1 } }).catch(() => undefined);275 if (next && !userStop) {276 await this.d.notifier.notify(row.owner_id, 'session.exited', 'A terminal session ended', row.title, `/terminal/?s=${row.id}`);277 }278 return next;279 }280281 /** The runtime is healthy and does not know this session. */282 async applyMissing(row: SessionRow): Promise<SessionRow | null> {283 if (!isLive(row.state)) return row;284 this.attached.delete(row.id);285 const to: SessionState = row.state === 'STARTING' ? 'FAILED' : row.state === 'STOPPING' ? 'STOPPED' : 'RECOVERABLE';286 return this.transition(row.id, to, { pid: null }, { source: 'reconcile', reason: 'missing_in_runtime' });287 }288289 /** Full pass against runtime truth. Returns false (and changes nothing) when it is unreachable. */290 async reconcile(): Promise<boolean> {291 let infos: RtSessionInfo[];292 try {293 infos = await this.d.rt.json<RtSessionInfo[]>('GET', '/sessions', undefined, { timeoutMs: 5000 });294 if (!Array.isArray(infos)) throw new RuntimeError(502, 'runtime_bad_response', 'bad /sessions payload');295 } catch (err) {296 this.markRuntime(false);297 if (!(err instanceof RuntimeError)) throw err;298 return false;299 }300 this.markRuntime(true);301 const byId = new Map(infos.map((i) => [i.id, i]));302 const rows = await this.d.db.q<SessionRow>(`${SESSION_SELECT} WHERE s.state = ANY($1::text[])`, [[...LIVE_SESSION_STATES]]);303 for (const row of rows) {304 if (row.state === 'WAITING_FOR_INPUT') this.waiting.add(row.id); // e.g. after a server restart305 const info = byId.get(row.id);306 // A session being created right now may not be listed yet: leave it to its creator.307 if (!info && row.state === 'STARTING' && Date.now() - (row.last_activity_at ?? row.created_at).getTime() < 20_000) continue;308 try {309 if (info) await this.applyInfo(row, info);310 else await this.applyMissing(row);311 } catch (err) {312 this.d.log.error({ err, sessionId: row.id }, 'session reconcile failed');313 }314 }315 return true;316 }317 markRuntime(up: boolean): void {318 this.runtimeUp = up;319 this.runtimeCheckedAt = new Date();320 }321322 async onRuntimeEvent(ev: RtEvent): Promise<void> {323 if (ev.type === 'task.updated') return;324 const row = await this.byId(ev.id);325 if (!row) return;326 if (ev.type === 'session.exit') {327 if (ev.exitCode !== null) return void (await this.applyExit(row, ev.exitCode, 'event'));328 // No exit code: the runtime also reports this when the tmux session simply vanished329 // (killed outside the runtime, tmux server lost after a reboot). Nothing proves the process330 // ended on its own, so ask the runtime for the truth and apply the reconciler rules.331 if (!isLive(row.state)) return;332 try {333 const info = await this.d.rt.json<RtSessionInfo>('GET', `/sessions/${row.id}`, undefined, { timeoutMs: 5000 });334 await this.applyInfo(row, info);335 } catch (err) {336 if (!(err instanceof RuntimeError)) throw err;337 if (err.status === 404) await this.applyMissing(row);338 // unreachable / other error: change nothing, the periodic reconciler settles it339 }340 } else if (ev.type === 'session.activity') {341 const before = this.attachedCount(row.id);342 this.attached.set(row.id, ev.attached);343 if (!isLive(row.state) || row.state === 'STOPPING' || row.state === 'STARTING') return;344 const target: SessionState = row.state === 'WAITING_FOR_INPUT' ? row.state : this.attachedCount(row.id) > 0 ? 'RUNNING' : 'DETACHED';345 const stale = ev.atMs - (row.last_activity_at?.getTime() ?? 0) > 30_000;346 if (target !== row.state || stale) await this.transition(row.id, target, { lastActivityMs: ev.atMs }, { source: 'activity' });347 else if (before !== this.attachedCount(row.id)) this.emit(row);348 } else {349 await this.onHook(row, ev.event, ev.conversationId, ev.message);350 }351 }352353 private async onHook(row: SessionRow, event: string, conversationId: string | null, message: string | null): Promise<void> {354 const conv = conversationId && isConversationId(this.agent.id, conversationId) && conversationId !== row.claude_conversation_id ? conversationId : undefined;355 const live = isLive(row.state) && row.state !== 'STOPPING' && row.state !== 'STARTING';356 let to: SessionState = row.state;357 let waitingReason: string | null | undefined;358 if (live && (event === 'Notification' || event === 'Stop')) {359 to = 'WAITING_FOR_INPUT';360 const name = this.agent.shortName;361 waitingReason = event === 'Stop' ? `${name} finished and is waiting for your next message` : `${name} needs your attention`;362 const settings = await getSettings(this.d.db, row.owner_id);363 if (settings.notifications.detail === 'titles' && message) waitingReason += `: ${message.replace(/[\u0000-\u001f\u007f]/g, ' ').slice(0, 200)}`;364 } else if (live && event === 'UserPromptSubmit') {365 to = 'RUNNING';366 }367 if (to === row.state && conv === undefined && waitingReason === undefined) return;368 const wasWaiting = row.state === 'WAITING_FOR_INPUT';369 const next = await this.transition(row.id, to, { conversationId: conv, waitingReason }, { source: 'hook', event });370 if (next && to === 'WAITING_FOR_INPUT' && !wasWaiting) {371 await this.d.notifier.notify(row.owner_id, 'session.waiting', 'A session is waiting for input', row.title, `/terminal/?s=${row.id}`);372 }373 }374}375