src/server/sessions.ts
Rama main · réplica de lectura del repositorio original.
import { readFileSync, writeFileSync, appendFileSync, existsSync, mkdirSync, readdirSync, unlinkSync, statSync, renameSync } from 'node:fs';
import { readFile } from 'node:fs/promises';
import path from 'node:path';
import { homedir } from 'node:os';
import { DATA_DIR, dataPath } from './paths.ts';
import type { Directives } from './directives.ts';
import type { ClaudeAccountRef } from './claude-account.ts';
import { findClaudeTranscript, type ClaudeResumeState } from './engine/claude-resume.ts';
const SESSIONS_PATH = dataPath('sessions.json');
export interface SessionMeta {
sessionId: string;
title: string;
userEmail: string;
userName: string;
uid: string;
createdAt: number;
lastModified: number;
autoApprove?: boolean;
/** Visibility: 'system' is shared across whitelisted users; 'user' (default) is private to `uid`. */
scope?: 'system' | 'user';
/** 'terminal' = a PTY-only session (created to host a standalone/terminal-first shell); hidden from
* the conversation list until it receives a real prompt (then it graduates to a normal conversation). */
kind?: 'terminal';
scheduleId?: string;
scheduleRunAt?: number;
scheduleRunStatus?: 'running' | 'ok' | 'error' | 'aborted';
slackContext?: { type: 'dm' | 'channel' | 'mention'; channelName?: string; userName?: string };
/** Slack message ts values already merged into this session's context (dedup for thread sync). */
seenSlackTs?: string[];
/** Skill names trigger-matched earlier in this session — re-injected on every subsequent turn. */
triggeredSkills?: string[];
/** Emails allowed to see this session (lowercase). Checked in addition to uid/scope. */
visibleTo?: string[];
runStatus?: 'running' | 'idle';
runOrigin?: 'web' | 'slack' | 'scheduler' | 'gmail';
lastStopReason?: 'max_turns_reached' | 'error' | 'aborted';
runRetryCount?: number;
directives?: Directives;
/** Actual model the engine resolved at runtime (from the SDK init message) — ground truth, unlike directives.model which is the request. */
lastModel?: string;
/** Engine that actually ran the last turn — ground truth beside `lastModel`, so the UI can report
* what executed instead of inferring the engine from the model id's shape. */
lastEngine?: string;
/** Claude login the last turn ran on (claude-code engine, subscription auth only). `personal` = the
* requester's own routed login (claude-account.ts), false = the box's shared login. Never a token. */
lastAccount?: ClaudeAccountRef;
/** claude-code engine SDK-resume mapping (opt-in flag) — see engine/claude-resume.ts. */
claudeResume?: ClaudeResumeState;
forkedFrom?: string;
/** Security taint (SECURITY_ENFORCE): rank of the lowest-ranked principal that ever contributed. Only lowers. */
floorRank?: number;
}
// ── Index cache ──────────────────────────────────────────────────────────────
// The index is ~9 MB / 12.8k records on the Circles box. Every mutation used to readFileSync +
// JSON.parse the whole file and then stringify + write it back, SYNCHRONOUSLY (~260 ms per
// mutation on a 2-vCPU host, and a turn does several), so the event loop stalled on bookkeeping.
//
// Single-writer, deliberately: the HTTP server, its scheduler and the MCP surface all run in ONE
// bun process (`src/server/index.ts` → boot.ts, which mounts the MCP in-process; the only spawned
// children are vendor MCP sidecars and background-job shells, none of which import this module).
// data-sync cannot touch it either — `sessions.json` is in the data repo's canonical .gitignore,
// so its pull/reset/stash never sees the file. The one out-of-band writer is the manual
// `defaults/scripts/backfill-slack-usernames.ts`, and this cache does NOT make that safe:
//
// The mtime guard below only runs when the cache is CLEAN. `loadIndex()` short-circuits on
// `if (dirty) return cache` first, so while a flush is pending the server neither sees an
// external edit nor survives it — the next flushIndex() writes the whole in-memory cache over
// the file, and the backfill's work is gone. The guard buys exactly one thing over the old
// read-every-time behaviour: an external edit made while nothing is pending IS picked up.
// Fixing the pending-flush case needs a merge or a lock, i.e. a multi-process protocol, which
// this file deliberately does not have. flushIndex() detects the clobber and logs it loudly
// instead of losing it silently — run the backfill with the server stopped.
let cache: SessionMeta[] | null = null;
let cacheMtimeMs = -1;
let dirty = false;
let flushTimer: ReturnType<typeof setTimeout> | null = null;
/**
* Coalescing window for the 9 MB write.
*
* Durability trade, stated explicitly: a crash can lose at most this much session BOOKKEEPING —
* runStatus, lastModified, seenSlackTs, triggeredSkills. Conversation content is not in this file
* (it lives in `data/conversations/`), so no user-visible message can be lost here. Every exit
* path we control flushes synchronously (`process.on('exit')` — reached by the SIGTERM/SIGINT
* graceful shutdown, which ends in process.exit — plus setShuttingDown() at the top of the drain),
* so the exposure is a SIGKILL or power loss inside a 250 ms window. Taken over the alternative:
* a ~260 ms sync stall on every single mutation.
*/
const FLUSH_DEBOUNCE_MS = 250;
function loadIndex(): SessionMeta[] {
if (cache) {
if (dirty) return cache; // memory is ahead of disk — we are the authority
try {
if (statSync(SESSIONS_PATH).mtimeMs === cacheMtimeMs) return cache;
} catch {
return cache; // file vanished mid-run; keep serving what we have
}
}
if (!existsSync(SESSIONS_PATH)) return (cache = []);
try {
cache = JSON.parse(readFileSync(SESSIONS_PATH, 'utf-8'));
cacheMtimeMs = statSync(SESSIONS_PATH).mtimeMs;
} catch (err) {
console.error('[sessions] index unreadable, serving an empty list:', (err as Error).message);
cache = [];
}
return cache!;
}
function saveIndex(sessions: SessionMeta[]): void {
cache = sessions;
dirty = true;
if (!flushTimer) scheduleFlush();
}
function scheduleFlush(): void {
const t = setTimeout(flushIndex, FLUSH_DEBOUNCE_MS);
(t as { unref?: () => void }).unref?.(); // never hold the process open for bookkeeping
flushTimer = t;
}
/**
* Write the index to disk now. Atomic (tmp + rename): the old truncate-in-place write left a
* half-written 9 MB file if it was interrupted, and loadIndex's catch turned that into an EMPTY
* session list — a silent total loss of the index.
*/
export function flushIndex(): void {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
if (!dirty || !cache) return;
dirty = false;
const tmp = `${SESSIONS_PATH}.tmp`;
try {
// Someone wrote the file behind us while this flush was pending (see the cache note above).
// We cannot merge without a multi-process protocol, so at least never lose it silently.
try {
const onDisk = statSync(SESSIONS_PATH).mtimeMs;
if (cacheMtimeMs >= 0 && onDisk !== cacheMtimeMs) {
console.error(`[sessions] index changed on disk (mtime ${onDisk} != ${cacheMtimeMs}) while a flush was pending — overwriting that external edit`);
}
} catch { /* file absent: nothing to clobber */ }
mkdirSync(DATA_DIR, { recursive: true });
writeFileSync(tmp, JSON.stringify(cache)); // no pretty-print: 9 MB nobody reads by hand
renameSync(tmp, SESSIONS_PATH);
cacheMtimeMs = statSync(SESSIONS_PATH).mtimeMs;
} catch (err) {
console.error('[sessions] index flush failed, retrying on the next tick:', (err as Error).message);
dirty = true;
scheduleFlush();
}
}
// Covers the graceful-shutdown paths (SIGTERM/SIGINT end in process.exit) and any normal exit,
// without adding signal listeners that would change the process's default signal semantics.
process.on('exit', () => flushIndex());
/**
* The subset of SessionMeta the conversation list actually renders. `triggeredSkills` (381 KB) and
* `seenSlackTs` (174 KB) are server bookkeeping and must never be shipped to a browser; the full
* record is still available per-session from /api/sessions/:id/meta.
*/
export interface SessionListItem {
sessionId: string;
title: string;
userName: string;
lastModified: number;
slackContext?: SessionMeta['slackContext'];
runStatus?: SessionMeta['runStatus'];
lastStopReason?: SessionMeta['lastStopReason'];
scheduleRunStatus?: SessionMeta['scheduleRunStatus'];
/** Owned by (or explicitly shared with) the caller. Stamped per-request: the trimmed row drops
* `uid`/`visibleTo`, so without it the client cannot scope anything by "mine". */
mine?: boolean;
}
/** The `mine` predicate, shared by the list filter and the per-row stamp so they cannot drift. */
export function isOwnSession(s: SessionMeta, uid: string, email: string): boolean {
return s.uid === uid || !!s.visibleTo?.includes(email);
}
export function toListItem(s: SessionMeta, mine?: boolean): SessionListItem {
return {
sessionId: s.sessionId,
title: s.title,
userName: s.userName,
lastModified: s.lastModified,
...(s.slackContext ? { slackContext: s.slackContext } : {}),
...(s.runStatus ? { runStatus: s.runStatus } : {}),
...(s.lastStopReason ? { lastStopReason: s.lastStopReason } : {}),
...(s.scheduleRunStatus ? { scheduleRunStatus: s.scheduleRunStatus } : {}),
...(mine === undefined ? {} : { mine }),
};
}
function summarize(prompt: string): string {
const cleaned = prompt.replace(/\s+/g, ' ').trim();
if (cleaned.length <= 80) return cleaned;
return cleaned.slice(0, 77) + '…';
}
export function upsertSession(sessionId: string, firstPrompt: string, user: { uid: string; email: string; name?: string }, scope?: 'system' | 'user', kind?: 'terminal'): void {
const sessions = loadIndex();
const existing = sessions.find((s) => s.sessionId === sessionId);
const now = Date.now();
if (existing) {
existing.lastModified = now;
if (!existing.title || existing.title === 'New session') {
existing.title = summarize(firstPrompt);
}
// A terminal-only session that now receives a real prompt graduates to a normal conversation.
if (existing.kind === 'terminal' && kind !== 'terminal') delete existing.kind;
} else {
sessions.unshift({
sessionId,
title: summarize(firstPrompt),
userEmail: user.email,
userName: user.name || user.email.split('@')[0],
uid: user.uid,
createdAt: now,
lastModified: now,
...(scope ? { scope } : {}),
...(kind ? { kind } : {}),
...floorField(sessionId),
});
}
saveIndex(sessions);
}
export function setSlackContext(sessionId: string, ctx: NonNullable<SessionMeta['slackContext']>): void {
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (s) {
s.slackContext = ctx;
saveIndex(sessions);
}
}
/** Record Slack ts values as merged into this session's context (dedup for thread sync). */
export function recordSeenSlackTs(sessionId: string, ts: string[]): void {
if (!ts.length) return;
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (!s) return;
const set = new Set(s.seenSlackTs ?? []);
for (const t of ts) set.add(t);
s.seenSlackTs = [...set];
saveIndex(sessions);
}
/** Persist trigger-matched skill names so they stay injected for the rest of the session. */
export function addTriggeredSkills(sessionId: string, names: string[]): void {
if (!names.length) return;
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (!s) return;
const merged = [...new Set([...(s.triggeredSkills ?? []), ...names])];
if (merged.length === (s.triggeredSkills?.length ?? 0)) return;
s.triggeredSkills = merged;
saveIndex(sessions);
}
/**
* Stable pagination cursor: lastModified alone is not unique across 12k sessions, and a paging
* cursor that only carries the timestamp drops or repeats every record in a tie.
*/
export function sessionCursor(s: SessionMeta | SessionListItem): string {
return `${String(s.lastModified).padStart(16, '0')}.${s.sessionId}`;
}
/**
* One page of an already-sorted (getAllSessions order) list, newest first. `before` is the cursor
* from the previous page's `nextCursor`; it is composite (lastModified + sessionId) because
* lastModified alone is not unique and a timestamp-only cursor drops or repeats records on a tie.
*/
export function pageSessions(list: SessionMeta[], opts: { limit: number; before?: string; mine?: (s: SessionMeta) => boolean }): { sessions: SessionListItem[]; nextCursor: string | null } {
const rest = opts.before ? list.filter((s) => sessionCursor(s) < opts.before!) : list;
const page = rest.slice(0, opts.limit);
return {
sessions: page.map((s) => toListItem(s, opts.mine?.(s))),
nextCursor: rest.length > page.length ? sessionCursor(page[page.length - 1]) : null,
};
}
export function getAllSessions(): SessionMeta[] {
// Copy before sorting: loadIndex() now returns the live cache array, and an in-place sort here
// would silently reorder the store for every other caller. Sorted by the same composite key the
// pagination cursor uses, so `cursor < before` is exactly consistent with this order.
return [...loadIndex()].sort((a, b) => b.lastModified - a.lastModified || (a.sessionId < b.sessionId ? 1 : a.sessionId > b.sessionId ? -1 : 0));
}
export function isSessionVisibleTo(s: SessionMeta, uid: string, isOwner = false, email?: string): boolean {
if (isOwner) return true;
if (s.scope === 'system') return true;
if (s.visibleTo && email && s.visibleTo.includes(email.toLowerCase())) return true;
return s.uid === uid;
}
export function getSessionsVisibleTo(uid: string, isOwner = false, email?: string): SessionMeta[] {
return getAllSessions().filter((s) => isSessionVisibleTo(s, uid, isOwner, email));
}
export function setVisibleTo(sessionId: string, emails: string[]): void {
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (s) {
s.visibleTo = emails.map((e) => e.toLowerCase());
saveIndex(sessions);
}
}
export function getSession(sessionId: string): SessionMeta | undefined {
return loadIndex().find((s) => s.sessionId === sessionId);
}
export function updateSessionTitle(sessionId: string, title: string): void {
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (s) {
s.title = title;
saveIndex(sessions);
}
}
export async function generateSessionTitle(sessionId: string, userPrompt: string, assistantText: string): Promise<string | null> {
try {
const { runTextQuery } = await import('./sdk-utils.ts');
const title = await runTextQuery({
prompt: `Generate a very short title (3-6 words, no quotes) for this conversation:\n\nUser: ${userPrompt.slice(0, 300)}\nAssistant: ${assistantText.slice(0, 500)}`,
systemPrompt: 'You generate concise conversation titles. Reply with ONLY the title, nothing else. No quotes, no punctuation at the end.',
});
const cleaned = title.trim().replace(/^["']|["']$/g, '').replace(/\.+$/, '');
if (!cleaned) return null;
updateSessionTitle(sessionId, cleaned);
return cleaned;
} catch (err: any) {
console.error(`[sessions] Title generation failed for ${sessionId.slice(0, 8)}:`, err.message);
return null;
}
}
export function setSessionDirectives(sessionId: string, directives: NonNullable<SessionMeta['directives']>): void {
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
if (s) {
// COPY, don't alias. streamChat() stores its working `directives` object here and then keeps
// mutating it for the rest of the turn (the per-skill turn budget is gap-filled afterwards).
// While the index was re-parsed from disk on every read that leak was invisible; against an
// in-memory store it would silently PIN a skill's 250-turn budget onto the session forever.
s.directives = { ...directives };
saveIndex(sessions);
}
}
// Resolved model per running session (set by the engine from the SDK init message).
// appendMessage stamps assistant messages from this map so every channel records it.
const liveModels = new Map<string, string>();
const liveEngines = new Map<string, string>();
const liveAccounts = new Map<string, string>();
/** Last resolved model for a session (live map first, then persisted lastModel). */
export function getSessionModel(sessionId: string): string | undefined {
return liveModels.get(sessionId) ?? loadIndex().find((s) => s.sessionId === sessionId)?.lastModel;
}
/** Record the runtime ground truth for a turn. `engine` is the engine that actually ran it — the
* pair is what the header reports, so neither half may be inferred from the other. */
export function setSessionModel(sessionId: string, model: string, engine?: string, account?: ClaudeAccountRef): void {
// An engine-tagged record describes a whole turn, so its account is part of it: a turn that ran
// on no Claude login (API key, another engine) CLEARS the previous turn's account.
const accountKey = account ? JSON.stringify(account) : '';
if (liveModels.get(sessionId) === model && (!engine || (liveEngines.get(sessionId) === engine && (liveAccounts.get(sessionId) ?? '') === accountKey))) return;
liveModels.set(sessionId, model);
if (engine) { liveEngines.set(sessionId, engine); liveAccounts.set(sessionId, accountKey); }
const sessions = loadIndex();
const s = sessions.find((s) => s.sessionId === sessionId);
const accountChanged = !!engine && JSON.stringify(s?.lastAccount ?? '') !== JSON.stringify(account ?? '');
if (s && (s.lastModel !== model || (engine && s.lastEngine !== engine) || accountChanged)) {
s.lastModel = model;
if (engine) {
s.lastEngine = engine;
if (account) s.lastAccount = account; else delete s.lastAccount;
}
saveIndex(sessions);
}
}
/** Store (or clear, with undefined) the claude-code resume mapping. `patch` merges into the existing one. */
export function setClaudeResume(sessionId: string, state: ClaudeResumeState | undefined, patch?: Partial<ClaudeResumeState>): void {
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (!s) return;
if (patch) { if (!s.claudeResume) return; s.claudeResume = { ...s.claudeResume, ...patch }; }
else if (state) s.claudeResume = state;
else if (s.claudeResume) delete s.claudeResume;
else return;
saveIndex(sessions);
}
// ── Security taint floor (SECURITY_ENFORCE) ─────────────────────────────────
// The engine's tool gate re-reads the floor on EVERY tool call, so it's served from this Map (write-through to
// `floorRank` in the index; single-writer process, see the index cache note). A floor only ever lowers.
const floors = new Map<string, number>();
/** The session's taint floor, or undefined when none was recorded. */
export function getSessionFloor(sessionId: string): number | undefined {
const cached = floors.get(sessionId);
if (cached !== undefined) return cached;
const f = loadIndex().find((s) => s.sessionId === sessionId)?.floorRank;
if (f !== undefined) floors.set(sessionId, f);
return f;
}
/** Lower the floor to min(floor, rank) — the first contributor sets it. Returns the new floor. Never raises it.
* Works before the index record exists (some channels create it after streaming): creation picks it up. */
export function lowerSessionFloor(sessionId: string, rank: number): number {
const prev = getSessionFloor(sessionId);
const next = prev === undefined ? rank : Math.min(prev, rank);
floors.set(sessionId, next);
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (s && s.floorRank !== next) { s.floorRank = next; saveIndex(sessions); }
return next;
}
/** A pending floor for a record being created now. */
const floorField = (sessionId: string) => (floors.has(sessionId) ? { floorRank: floors.get(sessionId)! } : {});
export function getSessionsByScheduleId(scheduleId: string): SessionMeta[] {
return loadIndex()
.filter((s) => s.scheduleId === scheduleId)
.sort((a, b) => b.lastModified - a.lastModified);
}
export function createScheduledSession(sessionId: string, scheduleId: string, title: string, createdBy: { uid: string; email: string }, scope: 'system' | 'user' = 'user'): void {
const sessions = loadIndex();
if (sessions.find((s) => s.sessionId === sessionId)) return;
const now = Date.now();
sessions.unshift({
sessionId,
title,
userEmail: createdBy.email,
userName: createdBy.email.split('@')[0],
uid: createdBy.uid,
createdAt: now,
lastModified: now,
scope,
scheduleId,
scheduleRunAt: now,
scheduleRunStatus: 'running',
...floorField(sessionId),
});
saveIndex(sessions);
}
export function updateScheduledSessionStatus(sessionId: string, status: 'ok' | 'error' | 'aborted'): void {
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (!s) return;
s.scheduleRunStatus = status;
s.lastModified = Date.now();
saveIndex(sessions);
}
export function getAutoApprove(uid: string): boolean {
const sessions = loadIndex();
const userSession = sessions.find((s) => s.uid === uid);
return userSession?.autoApprove ?? false;
}
export function setAutoApprove(uid: string, value: boolean): void {
const sessions = loadIndex();
// Set on all sessions for this user
let found = false;
for (const s of sessions) {
if (s.uid === uid) {
s.autoApprove = value;
found = true;
}
}
if (found) saveIndex(sessions);
}
let _shuttingDown = false;
export function setShuttingDown(): void { _shuttingDown = true; flushIndex(); }
export function setRunStatus(sessionId: string, status: 'running' | 'idle', origin?: 'web' | 'slack' | 'scheduler' | 'gmail', stopReason?: SessionMeta['lastStopReason']): void {
if (_shuttingDown && status === 'idle') return;
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (!s) return;
s.runStatus = status;
if (origin) s.runOrigin = origin;
if (status === 'idle') {
s.lastStopReason = stopReason;
// runRetryCount caps crash-resume loops (interrupt → resume → interrupt). Reaching 'idle' at all
// proves this turn was NOT killed by a restart — the guard above drops idle writes during a
// drain, and a hard kill never gets here — so the loop is broken and the count starts over.
// It used to be reset only on the web path, so Slack and scheduler sessions accumulated it
// forever: after two interruptions in their whole life they permanently lost auto-resume and
// every later restart ended at "please send a message to continue".
s.runRetryCount = 0;
}
s.lastModified = Date.now();
saveIndex(sessions);
}
export function incrementRetryCount(sessionId: string): void {
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (!s) return;
s.runRetryCount = (s.runRetryCount ?? 0) + 1;
saveIndex(sessions);
}
export function resetRetryCount(sessionId: string): void {
const sessions = loadIndex();
const s = sessions.find((x) => x.sessionId === sessionId);
if (!s || !s.runRetryCount) return;
s.runRetryCount = 0;
saveIndex(sessions);
}
export function getRunningSessions(): SessionMeta[] {
return loadIndex().filter((s) => s.runStatus === 'running');
}
// ── Global session lock ─────────────────────────────────────────────────────
// Single source of truth for "is a session currently running a query?"
// All paths (WebSocket, Slack, REST API, scheduler) must acquire before streaming.
interface SessionLock {
origin: 'web' | 'slack' | 'scheduler' | 'api';
abortController: AbortController;
startedAt: number;
}
const globalSessionLocks = new Map<string, SessionLock>();
export function acquireSessionLock(sessionId: string, origin: SessionLock['origin'], abortController: AbortController): boolean {
if (globalSessionLocks.has(sessionId)) return false;
globalSessionLocks.set(sessionId, { origin, abortController, startedAt: Date.now() });
return true;
}
export function releaseSessionLock(sessionId: string, owner?: AbortController): boolean {
if (owner) {
const lock = globalSessionLocks.get(sessionId);
if (lock && lock.abortController !== owner) return false;
}
globalSessionLocks.delete(sessionId);
return true;
}
export function isSessionLocked(sessionId: string): boolean {
return globalSessionLocks.has(sessionId);
}
/** Count of queries streaming in THIS process right now. The drain loop must use this, not the
* persisted index: setRunStatus freezes `running` at shutdown as the crash-recovery marker, so
* index entries can't go idle mid-drain (and stale entries from a prior crash would stall the
* drain for sessions that aren't streaming at all). */
export function getActiveLockCount(): number {
return globalSessionLocks.size;
}
export function replaceSessionLock(sessionId: string, origin: SessionLock['origin'], abortController: AbortController): void {
globalSessionLocks.set(sessionId, { origin, abortController, startedAt: Date.now() });
}
export function getSessionAbortController(sessionId: string): AbortController | undefined {
return globalSessionLocks.get(sessionId)?.abortController;
}
// Deleting a conversation is deliberately not exposed by the app (no route, no helper): the audit log is
// append-only and an owner who really means to remove one does it by hand on the box.
function findSessionFile(sessionId: string): string | null {
return findClaudeTranscript(sessionId, path.join(homedir(), '.claude'));
}
// ── Our own conversation store ───────────────────────────────────────────────
export type ConvBlock =
| { type: 'text'; text: string }
| { type: 'image'; src: string }
| { type: 'file'; src: string; name: string; mimeType: string }
| { type: 'context'; label: string; text: string }
| { type: 'tool_use'; tool: string; toolUseId: string; input: unknown }
| { type: 'tool_result'; toolUseId: string; output: string }
| { type: 'thinking'; text: string }
// A run that failed at the engine/adapter level. Persisted so the transcript records the failure
// durably — otherwise a dead session (esp. an unattended scheduled one) is indistinguishable from
// one still thinking, with the cause only in stderr.
| { type: 'error'; text: string }
// Persisted block written by an add-on engine's background worker (e.g. EE's duplex voice brain).
// The core stores/renders it but owns none of its semantics; name kept for stored-history back-compat.
| { type: 'duplex_result'; label?: string; tier?: string; text?: string }
| { type: 'summary'; text: string; compactedCount: number }
| { type: 'compact_marker'; summary: string; compactedCount: number };
export interface ConvMessage {
id: string;
// 'system' is written for out-of-band notices persisted into the thread (e.g. a playback
// interruption). History builders treat any non-'user' role as assistant-side.
role: 'user' | 'assistant' | 'system';
blocks: ConvBlock[];
channel?: string;
ts?: number;
senderName?: string;
/** Model that produced this assistant message (stamped from the engine's resolved model). */
model?: string;
}
const CONV_DIR = dataPath('conversations');
export function loadConversation(sessionId: string): ConvMessage[] {
const file = path.join(CONV_DIR, `${sessionId}.jsonl`);
if (!existsSync(file)) return [];
const messages: ConvMessage[] = [];
for (const line of readFileSync(file, 'utf-8').split('\n')) {
if (!line) continue;
try { messages.push(JSON.parse(line)); } catch {
console.warn(`[sessions] skipping bad JSONL line in ${sessionId.slice(0, 8)}`);
}
}
return messages;
}
export function saveConversation(sessionId: string, messages: ConvMessage[]): void {
mkdirSync(CONV_DIR, { recursive: true });
const content = messages.length > 0
? messages.map((m) => JSON.stringify(m)).join('\n') + '\n'
: '';
writeFileSync(path.join(CONV_DIR, `${sessionId}.jsonl`), content);
}
export function appendMessage(sessionId: string, message: ConvMessage): void {
if (!message.ts) message.ts = Date.now();
if (message.role === 'assistant' && !message.model) {
const model = liveModels.get(sessionId);
if (model) message.model = model;
}
mkdirSync(CONV_DIR, { recursive: true });
const file = path.join(CONV_DIR, `${sessionId}.jsonl`);
const line = JSON.stringify(message) + '\n';
appendFileSync(file, line);
}
// ── Live partial registry (in-memory, for real-time reads) ───────────────────
const livePartialCollectors = new Map<string, () => ConvBlock[]>();
export function registerLivePartial(sessionId: string, collector: () => ConvBlock[]): void {
livePartialCollectors.set(sessionId, collector);
}
export function unregisterLivePartial(sessionId: string): void {
livePartialCollectors.delete(sessionId);
}
export function readLivePartial(sessionId: string): ConvBlock[] | null {
const collector = livePartialCollectors.get(sessionId);
if (!collector) return null;
const blocks = collector();
return blocks.length ? blocks : null;
}
// ── Partial flush (survives crashes) ─────────────────────────────────────────
export function writePartial(sessionId: string, blocks: ConvBlock[]): void {
if (!blocks.length) return;
mkdirSync(CONV_DIR, { recursive: true });
writeFileSync(path.join(CONV_DIR, `${sessionId}.partial.json`), JSON.stringify({ blocks, ts: Date.now() }));
}
export function readPartial(sessionId: string): ConvBlock[] | null {
const file = path.join(CONV_DIR, `${sessionId}.partial.json`);
if (!existsSync(file)) return null;
try {
const data = JSON.parse(readFileSync(file, 'utf-8'));
return data.blocks ?? null;
} catch {
return null;
}
}
export function clearPartial(sessionId: string): void {
const file = path.join(CONV_DIR, `${sessionId}.partial.json`);
try { unlinkSync(file); } catch {}
}
// ── Claude JSONL fallback ────────────────────────────────────────────────────
export async function getSessionHistory(sessionId: string): Promise<any[]> {
const file = findSessionFile(sessionId);
if (!file) {
console.log(`[sessions] no JSONL file found for ${sessionId.slice(0, 8)}`);
return [];
}
try {
const content = await readFile(file, 'utf-8');
const messages: any[] = [];
for (const line of content.split('\n')) {
if (!line) continue;
const obj = JSON.parse(line);
if (obj.type === 'user' || obj.type === 'assistant') {
messages.push(obj);
}
}
console.log(`[sessions] loaded ${messages.length} messages for ${sessionId.slice(0, 8)} from ${path.basename(path.dirname(file))}`);
return messages;
} catch (err: any) {
console.warn(`[sessions] failed to read ${sessionId.slice(0, 8)}:`, err.message);
return [];
}
}
// ── One-time backfill: set visibleTo/scope on legacy bot sessions ───────────
export function forkSession(sourceId: string, user: { uid: string; email: string; name?: string }, truncateAtIndex?: number): string | null {
const source = getSession(sourceId);
if (!source) return null;
const messages = loadConversation(sourceId);
if (!messages.length) return null;
const newId = crypto.randomUUID();
const forkedMessages = truncateAtIndex != null ? messages.slice(0, truncateAtIndex + 1) : messages;
saveConversation(newId, forkedMessages);
const sessions = loadIndex();
const now = Date.now();
sessions.unshift({
sessionId: newId,
title: `Fork: ${source.title}`,
userEmail: user.email,
userName: user.name || user.email.split('@')[0],
uid: user.uid,
createdAt: now,
lastModified: now,
forkedFrom: sourceId,
...(source.directives ? { directives: source.directives } : {}),
// The fork copies the source's (possibly tainted) content, so it keeps its floor.
...(getSessionFloor(sourceId) !== undefined ? { floorRank: getSessionFloor(sourceId)! } : {}),
});
saveIndex(sessions);
return newId;
}
export function backfillSessionVisibility(findContact: (opts: { name?: string }) => { emails: string[] } | null): void {
const sessions = loadIndex();
let patched = 0;
for (const s of sessions) {
if (s.visibleTo) continue;
if (s.uid === 'slack-bot') {
if (s.slackContext?.type === 'dm') {
s.scope = 'user';
if (s.slackContext.userName) {
const contact = findContact({ name: s.slackContext.userName });
if (contact?.emails.length) {
s.visibleTo = contact.emails.map((e) => e.toLowerCase());
patched++;
}
}
} else if (!s.scope || s.scope !== 'system') {
s.scope = 'system';
patched++;
}
}
if (s.uid === 'gmail-bot') {
const conv = loadConversation(s.sessionId);
const ctxBlock = conv[0]?.blocks?.find((b: any) => b.type === 'context');
if (ctxBlock && 'text' in ctxBlock) {
const emails: string[] = [];
for (const line of (ctxBlock as any).text.split('\n')) {
if (/^(From|To|Cc):/i.test(line)) {
for (const m of line.matchAll(/[\w.+-]+@[\w.-]+\.\w+/g)) emails.push(m[0].toLowerCase());
}
}
const botEmail = (process.env.GMAIL_USER_EMAIL || '').toLowerCase();
const filtered = [...new Set(emails)].filter((e) => e !== botEmail);
if (filtered.length) {
s.visibleTo = filtered;
patched++;
}
}
}
}
if (patched > 0) {
saveIndex(sessions);
console.log(`[sessions] backfill: patched visibility on ${patched} sessions`);
}
}