Dev

.github/workflows/ci.yml

@@ -242,6 +242,8 @@ jobs:
       - name: Prime tmux server (ensures socket dir exists)
         run: tmux new-session -d -s init && tmux kill-session -t init
       - run: npm ci
+      - name: Install web deps (active timeline refresh e2e wrapper invokes web vitest)
+        run: ./scripts/ci-npm-ci.sh web
       - name: Run pipe-pane e2e tests
         run: npx vitest run test/e2e/pipe-pane-stream.test.ts
       - name: Run other e2e tests

server/src/db/migrations/043_session_text_tail_cache.sql

@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS session_text_tail_cache (
+  server_id TEXT NOT NULL,
+  session_name TEXT NOT NULL,
+  events JSONB NOT NULL DEFAULT '[]'::jsonb,
+  latest_ts BIGINT,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (server_id, session_name)
+);
+
+CREATE INDEX IF NOT EXISTS idx_session_text_tail_cache_updated_at
+  ON session_text_tail_cache (updated_at DESC);

server/src/db/queries.ts

@@ -96,6 +96,143 @@ export interface QuickData {
   phrases: string[];
 }
 
+export const SESSION_TEXT_TAIL_CACHE_LIMIT = 50;
+
+export interface SessionTextTailCacheItem {
+  eventId: string;
+  ts: number;
+  type: 'user.message' | 'assistant.text';
+  text: string;
+  source?: string;
+  confidence?: string;
+}
+
+interface DbSessionTextTailCacheRow {
+  server_id: string;
+  session_name: string;
+  events: SessionTextTailCacheItem[] | string | null;
+  latest_ts: number | null;
+  updated_at: Date;
+}
+
+interface ClassifiedSessionTextTailEvent {
+  sessionName: string;
+  item: SessionTextTailCacheItem;
+}
+
+function normalizeSessionTextTailText(text: unknown): string | null {
+  if (typeof text !== 'string') return null;
+  const trimmed = text.trim();
+  return trimmed || null;
+}
+
+function isSessionTextTailType(type: unknown): type is SessionTextTailCacheItem['type'] {
+  return type === 'user.message' || type === 'assistant.text';
+}
+
+function parseSessionTextTailCacheEvents(
+  raw: SessionTextTailCacheItem[] | string | null | undefined,
+): SessionTextTailCacheItem[] {
+  let parsed: unknown = raw;
+  if (typeof parsed === 'string') {
+    try {
+      parsed = JSON.parse(parsed);
+    } catch {
+      return [];
+    }
+  }
+  if (!Array.isArray(parsed)) return [];
+  const items: SessionTextTailCacheItem[] = [];
+  for (const entry of parsed) {
+    if (!entry || typeof entry !== 'object') return [];
+    const row = entry as Record<string, unknown>;
+    if (
+      typeof row.eventId !== 'string'
+      || typeof row.ts !== 'number'
+      || !isSessionTextTailType(row.type)
+      || typeof row.text !== 'string'
+    ) {
+      return [];
+    }
+    const text = normalizeSessionTextTailText(row.text);
+    if (!text) return [];
+    const item: SessionTextTailCacheItem = {
+      eventId: row.eventId,
+      ts: row.ts,
+      type: row.type,
+      text,
+    };
+    if (typeof row.source === 'string' && row.source.trim()) item.source = row.source.trim();
+    if (typeof row.confidence === 'string' && row.confidence.trim()) item.confidence = row.confidence.trim();
+    items.push(item);
+  }
+  return items;
+}
+
+function mergeSessionTextTailCacheEvents(
+  existing: SessionTextTailCacheItem[],
+  incoming: SessionTextTailCacheItem,
+): SessionTextTailCacheItem[] {
+  const deduped = new Map<string, SessionTextTailCacheItem>();
+  for (const item of existing) deduped.set(item.eventId, item);
+  deduped.set(incoming.eventId, incoming);
+  const merged = [...deduped.values()].sort((a, b) => {
+    if (a.ts !== b.ts) return a.ts - b.ts;
+    return a.eventId.localeCompare(b.eventId);
+  });
+  return merged.length > SESSION_TEXT_TAIL_CACHE_LIMIT
+    ? merged.slice(merged.length - SESSION_TEXT_TAIL_CACHE_LIMIT)
+    : merged;
+}
+
+export function mergeSessionTextTailCacheItems(
+  existing: SessionTextTailCacheItem[],
+  incoming: SessionTextTailCacheItem[],
+): SessionTextTailCacheItem[] {
+  const deduped = new Map<string, SessionTextTailCacheItem>();
+  for (const item of existing) deduped.set(item.eventId, item);
+  for (const item of incoming) deduped.set(item.eventId, item);
+  const merged = [...deduped.values()].sort((a, b) => {
+    if (a.ts !== b.ts) return a.ts - b.ts;
+    return a.eventId.localeCompare(b.eventId);
+  });
+  return merged.length > SESSION_TEXT_TAIL_CACHE_LIMIT
+    ? merged.slice(merged.length - SESSION_TEXT_TAIL_CACHE_LIMIT)
+    : merged;
+}
+
+export function classifySessionTextTailEvent(rawEvent: Record<string, unknown>): ClassifiedSessionTextTailEvent | null {
+  const sessionName = typeof rawEvent.sessionId === 'string' ? rawEvent.sessionId : null;
+  const eventId = typeof rawEvent.eventId === 'string' ? rawEvent.eventId : null;
+  const ts = typeof rawEvent.ts === 'number' ? rawEvent.ts : null;
+  const type = isSessionTextTailType(rawEvent.type) ? rawEvent.type : null;
+  const payload = rawEvent.payload && typeof rawEvent.payload === 'object'
+    ? rawEvent.payload as Record<string, unknown>
+    : null;
+  if (!sessionName || !eventId || ts === null || !type || !payload) return null;
+  if (type === 'assistant.text' && payload.streaming === true) return null;
+  const text = normalizeSessionTextTailText(payload.text);
+  if (!text) return null;
+  const item: SessionTextTailCacheItem = { eventId, ts, type, text };
+  if (typeof rawEvent.source === 'string' && rawEvent.source.trim()) item.source = rawEvent.source.trim();
+  if (typeof rawEvent.confidence === 'string' && rawEvent.confidence.trim()) item.confidence = rawEvent.confidence.trim();
+  return { sessionName, item };
+}
+
+export function collectSessionTextTailCacheItems(
+  sessionName: string,
+  rawEvents: unknown[],
+): SessionTextTailCacheItem[] {
+  const items: SessionTextTailCacheItem[] = [];
+  for (const raw of rawEvents) {
+    if (!raw || typeof raw !== 'object') continue;
+    const classified = classifySessionTextTailEvent(raw as Record<string, unknown>);
+    if (!classified || classified.sessionName !== sessionName) continue;
+    items.push(classified.item);
+  }
+  return mergeSessionTextTailCacheItems([], items);
+}
+
 // ── Users ─────────────────────────────────────────────────────────────────
 
 export async function createUser(db: Database, id: string): Promise<DbUser> {
@@ -542,6 +679,68 @@ export async function updateSession(
   );
 }
 
+export async function upsertSessionTextTailCacheEvent(
+  db: Database,
+  serverId: string,
+  rawEvent: Record<string, unknown>,
+): Promise<void> {
+  const classified = classifySessionTextTailEvent(rawEvent);
+  if (!classified) return;
+  await db.transaction(async (tx) => {
+    const row = await tx.queryOne<Pick<DbSessionTextTailCacheRow, 'events'>>(
+      `SELECT events
+         FROM session_text_tail_cache
+        WHERE server_id = $1 AND session_name = $2
+        FOR UPDATE`,
+      [serverId, classified.sessionName],
+    );
+    const existing = parseSessionTextTailCacheEvents(row?.events ?? null);
+    const events = mergeSessionTextTailCacheEvents(existing, classified.item);
+    const latestTs = events.length > 0 ? events[events.length - 1]!.ts : null;
+    await tx.execute(
+      `INSERT INTO session_text_tail_cache (server_id, session_name, events, latest_ts, updated_at)
+       VALUES ($1, $2, $3::jsonb, $4, NOW())
+       ON CONFLICT (server_id, session_name)
+       DO UPDATE SET events = EXCLUDED.events, latest_ts = EXCLUDED.latest_ts, updated_at = NOW()`,
+      [serverId, classified.sessionName, JSON.stringify(events), latestTs],
+    );
+  });
+}
+
+export async function getSessionTextTailCache(
+  db: Database,
+  serverId: string,
+  sessionName: string,
+): Promise<SessionTextTailCacheItem[]> {
+  const row = await db.queryOne<Pick<DbSessionTextTailCacheRow, 'events'>>(
+    `SELECT events
+       FROM session_text_tail_cache
+      WHERE server_id = $1 AND session_name = $2`,
+    [serverId, sessionName],
+  );
+  const events = parseSessionTextTailCacheEvents(row?.events ?? null);
+  return events.length > SESSION_TEXT_TAIL_CACHE_LIMIT
+    ? events.slice(events.length - SESSION_TEXT_TAIL_CACHE_LIMIT)
+    : events;
+}
+
+export async function replaceSessionTextTailCache(
+  db: Database,
+  serverId: string,
+  sessionName: string,
+  events: SessionTextTailCacheItem[],
+): Promise<void> {
+  const bounded = mergeSessionTextTailCacheItems([], events);
+  const latestTs = bounded.length > 0 ? bounded[bounded.length - 1]!.ts : null;
+  await db.execute(
+    `INSERT INTO session_text_tail_cache (server_id, session_name, events, latest_ts, updated_at)
+     VALUES ($1, $2, $3::jsonb, $4, NOW())
+     ON CONFLICT (server_id, session_name)
+     DO UPDATE SET events = EXCLUDED.events, latest_ts = EXCLUDED.latest_ts, updated_at = NOW()`,
+    [serverId, sessionName, JSON.stringify(bounded), latestTs],
+  );
+}
+
 // ── Quick data ────────────────────────────────────────────────────────────
 
 const EMPTY_QUICK_DATA: QuickData = { history: [], sessionHistory: {}, commands: [], phrases: [] };

server/src/routes/watch.ts

@@ -1,13 +1,26 @@
 import { Hono } from 'hono';
 import type { Env } from '../env.js';
-import { getServersByUserId, getDbSessionsByServer, getSubSessionsByServer, getUserPref } from '../db/queries.js';
+import {
+  getServersByUserId,
+  getDbSessionsByServer,
+  getSubSessionsByServer,
+  getUserPref,
+  getSessionTextTailCache,
+  collectSessionTextTailCacheItems,
+  mergeSessionTextTailCacheItems,
+  replaceSessionTextTailCache,
+  SESSION_TEXT_TAIL_CACHE_LIMIT,
+} from '../db/queries.js';
 import { requireAuth, resolveServerRole } from '../security/authorization.js';
 import { WsBridge } from '../ws/bridge.js';
 import { IMCODES_POD_HEADER } from '../../../shared/http-header-names.js';
 import { getPodIdentity } from '../util/pod-identity.js';
 import logger from '../util/logger.js';
 
 export const watchRoutes = new Hono<{ Bindings: Env; Variables: { userId: string; role: string } }>();
+const TEXT_TAIL_HISTORY_PAGE_LIMIT = 500;
+const TEXT_TAIL_HISTORY_MAX_PAGES = 6;
+const TEXT_TAIL_HISTORY_TIMEOUT_MS = 1500;
 
 type WatchSessionState = 'working' | 'idle' | 'error' | 'stopped';
 
@@ -52,6 +65,61 @@ function titleForSubSession(sub: { id: string; label: string | null; type: strin
   return sub.type || sub.id;
 }
 
+async function backfillSessionTextTailFromDaemon(
+  serverId: string,
+  sessionName: string,
+  cached: Awaited<ReturnType<typeof getSessionTextTailCache>>,
+): Promise<Awaited<ReturnType<typeof getSessionTextTailCache>>> {
+  let events = cached;
+  let beforeTs: number | undefined;
+  const seenPages = new Set<string>();
+
+  for (let page = 0; page < TEXT_TAIL_HISTORY_MAX_PAGES; page++) {
+    if (events.length >= SESSION_TEXT_TAIL_CACHE_LIMIT) break;
+
+    const response = await WsBridge.get(serverId).requestTimelineHistory({
+      sessionName,
+      limit: TEXT_TAIL_HISTORY_PAGE_LIMIT,
+      timeoutMs: TEXT_TAIL_HISTORY_TIMEOUT_MS,
+      ...(beforeTs !== undefined ? { beforeTs } : {}),
+    });
+    const rawEvents = Array.isArray(response.events)
+      ? response.events.filter((event): event is Record<string, unknown> => !!event && typeof event === 'object')
+      : [];
+    if (rawEvents.length === 0) break;
+
+    const fingerprint = JSON.stringify([
+      rawEvents.length,
+      rawEvents[0]?.eventId,
+      rawEvents[0]?.ts,
+      rawEvents.at(-1)?.eventId,
+      rawEvents.at(-1)?.ts,
+    ]);
+    if (seenPages.has(fingerprint)) break;
+    seenPages.add(fingerprint);
+
+    const live = collectSessionTextTailCacheItems(sessionName, rawEvents);
+    if (live.length > 0) {
+      events = mergeSessionTextTailCacheItems(events, live);
+    }
+
+    if (rawEvents.length < TEXT_TAIL_HISTORY_PAGE_LIMIT) break;
+
+    let oldestTs: number | undefined;
+    for (const event of rawEvents) {
+      if (typeof event.ts !== 'number' || !Number.isFinite(event.ts)) continue;
+      oldestTs = oldestTs === undefined ? event.ts : Math.min(oldestTs, event.ts);
+    }
+    if (oldestTs === undefined) break;
+
+    // Keep a 1ms overlap on the page boundary so same-ts events are not
+    // skipped when the next page is requested.
+    beforeTs = oldestTs + 1;
+  }
+
+  return events;
+}
+
 function sanitizeWatchTimelineEvent(raw: unknown): {
   eventId: string;
   sessionId: string;
@@ -356,3 +424,39 @@ watchRoutes.get('/server/:id/timeline/history/full', requireAuth(), async (c) =>
     return c.json({ error: 'relay_failed' }, 502);
   }
 });
+
+watchRoutes.get('/server/:id/timeline/text-tail', requireAuth(), async (c) => {
+  const userId = c.get('userId' as never) as string;
+  const serverId = c.req.param('id')!;
+  const role = await resolveServerRole(c.env.DB, serverId, userId);
+  if (role === 'none') return c.json({ error: 'forbidden' }, 403);
+
+  const sessionName = c.req.query('sessionName')?.trim();
+  if (!sessionName) return c.json({ error: 'session_name_required' }, 400);
+
+  try {
+    const cached = await getSessionTextTailCache(c.env.DB, serverId, sessionName);
+    let events = cached;
+    try {
+      events = await backfillSessionTextTailFromDaemon(serverId, sessionName, cached);
+      if (JSON.stringify(events) !== JSON.stringify(cached)) {
+        await replaceSessionTextTailCache(c.env.DB, serverId, sessionName, events);
+      }
+    } catch (err) {
+      logger.info({
+        serverId,
+        sessionName,
+        err: err instanceof Error ? err.message : String(err),
+      }, 'timeline.text-tail backfill skipped');
+    }
+    c.header(IMCODES_POD_HEADER, getPodIdentity());
+    return c.json({ sessionName, events });
+  } catch (err) {
+    logger.warn({
+      serverId,
+      sessionName,
+      err: err instanceof Error ? err.message : String(err),
+    }, 'timeline.text-tail failed');
+    return c.json({ error: 'cache_read_failed' }, 500);
+  }
+});