beaver-gateway/ui/src/lib/panel/index.svelte.ts

import type { ApiClient } from "$lib/api/client";
import { LiveStream } from "$lib/api/live.svelte";
import type { BusEvent, ConversationSummary } from "$lib/api/types";
import type { PanelCache } from "./cache";
import { cacheableHistory, historyKey, infoKey } from "./history-cache";

const RELOAD_DEBOUNCE_MS = 1500;
const PAGE = 500;
const CACHED_ROWS = 200;
const INDEX_KEY = "index";
const WARM_TTL_MS = 60_000;
const WARM_GAP_MS = 120;
const WARM_AHEAD = 8;
const QUIET_KINDS = new Set(["job", "fork"]);

// The conversation index kept fresh by ``/api/events``: rows land as
// ``conversation.*`` events arrive, turn markers flip ``running_turn``
// without a refetch, and anything that moves the queue reloads once.
// Shared by the admin's rail and the panel's picker.
export class ConversationIndex {
  conversations = $state<ConversationSummary[]>([]);
  loaded = $state(false);
  readonly live: LiveStream;
  protected readonly client: () => ApiClient | null;
  private readonly cache: PanelCache | null;
  private reloadTimer: ReturnType<typeof setTimeout> | null = null;
  private readonly warmed = new Map<string, number>();
  private warming: Promise<void> | null = null;

  constructor(client: () => ApiClient | null, cache: PanelCache | null = null) {
    this.client = client;
    this.cache = cache;
    const cached = cache?.get<ConversationSummary[]>(INDEX_KEY);
    if (cached && cached.length > 0) {
      this.conversations = cached;
      this.loaded = true;
    }
    this.live = new LiveStream(client, "/api/events", {
      onEvent: (event) => this.apply(event),
      prepare: () => this.load(),
    });
  }

  get running(): ConversationSummary[] {
    return this.conversations.filter((row) => row.running_turn);
  }

  get open(): ConversationSummary[] {
    return this.conversations.filter((row) => row.status === "open");
  }

  byId(id: string): ConversationSummary | undefined {
    return this.conversations.find((row) => row.id === id);
  }

  async load(): Promise<void> {
    const client = this.client();
    if (!client) {
      return;
    }
    const list = await client.conversations({ limit: PAGE });
    this.conversations = list.conversations;
    this.loaded = true;
    this.cache?.set(INDEX_KEY, list.conversations.slice(0, CACHED_ROWS));
    this.prefetch(this.likely());
  }

  // What the operator opens next: anything alive, then the freshest strips.
  private likely(): string[] {
    const rows = this.conversations.filter(
      (row) => !QUIET_KINDS.has(row.kind) && row.status === "open"
    );
    const alive = rows.filter(
      (row) => row.running_turn || row.pending_question
    );
    const rest = rows
      .filter((row) => !(row.running_turn || row.pending_question))
      .sort((a, b) =>
        (b.last_activity_at ?? b.created_at ?? "").localeCompare(
          a.last_activity_at ?? a.created_at ?? ""
        )
      );
    return [...alive, ...rest].slice(0, WARM_AHEAD).map((row) => row.id);
  }

  // Warm the cache for these threads, one at a time, in the background.
  prefetch(ids: string[]): void {
    if (!this.cache) {
      return;
    }
    const now = Date.now();
    const fresh = ids.filter(
      (id) => now - (this.warmed.get(id) ?? 0) > WARM_TTL_MS
    );
    if (fresh.length === 0) {
      return;
    }
    for (const id of fresh) {
      this.warmed.set(id, now);
    }
    const run = async () => {
      for (const id of fresh) {
        const client = this.client();
        if (!client) {
          return;
        }
        try {
          // biome-ignore lint/performance/noAwaitInLoops: one thread at a time, on purpose - background warmth, not a race
          const [history, info] = await Promise.all([
            client.history(id),
            client.conversation(id),
          ]);
          this.cache?.set(historyKey(id), cacheableHistory(history.messages));
          this.cache?.set(infoKey(id), info);
        } catch {
          this.warmed.delete(id);
        }
        await new Promise((resolve) => setTimeout(resolve, WARM_GAP_MS));
      }
    };
    this.warming = (this.warming ?? Promise.resolve()).then(run, run);
  }

  apply(event: BusEvent): void {
    switch (event.type) {
      case "conversation.created":
      case "conversation.updated": {
        if (typeof event.id === "string") {
          this.upsert(event as unknown as ConversationSummary);
        }
        return;
      }
      case "turn.start":
      case "turn.end": {
        const row = event.conversation_id
          ? this.byId(event.conversation_id)
          : undefined;
        if (row) {
          row.running_turn =
            event.type === "turn.start" ? (event.turn_id ?? null) : null;
          row.last_activity_at = event.ts;
        }
        return;
      }
      case "message.queued":
      case "inject.queued":
      case "reply": {
        this.reloadSoon();
        return;
      }
      default:
    }
  }

  start(): void {
    this.live.start();
  }

  stop(): void {
    this.live.stop();
  }

  private upsert(row: ConversationSummary): void {
    const index = this.conversations.findIndex((c) => c.id === row.id);
    if (index >= 0) {
      this.conversations[index] = { ...this.conversations[index], ...row };
    } else {
      this.conversations.unshift(row);
    }
  }

  private reloadSoon(): void {
    if (this.reloadTimer) {
      return;
    }
    this.reloadTimer = setTimeout(() => {
      this.reloadTimer = null;
      this.load().catch(() => undefined);
    }, RELOAD_DEBOUNCE_MS);
  }
}