Add files using upload-large-folder tool

4e23b01 verified about 1 month ago

24.1 kB

import { randomUUID } from 'node:crypto';
import { join } from 'node:path';
import { stat } from 'node:fs/promises';

import {
  createDecorator,
  databaseSearchEnabled,
  IBootstrapService,
  IConfigService,
  ILogService,
  ISessionIndex,
  LifecycleScope,
  ScopeActivation,
  registerScopedService,
  sessionDirOf,
  workspacePersistenceScope,
  type SessionSummary,
} from '@moonshot-ai/agent-core-v2';
import { normalizeLiteral, tokenize } from '@moonshot-ai/minidb';
import type { TranscriptStore } from '@moonshot-ai/transcript';

import {
  GlobalSearchError,
  type GlobalSearchHit,
  type GlobalSearchIndexState,
  type GlobalSearchPage,
  type GlobalSearchQuery,
} from './contract';
import { MAX_DOC_TEXT_CHARS, type MessageDoc, type TitleDoc } from './docs';
import {
  SearchIndexCore,
  type CoreIndexView,
  type CoreLifecycleReport,
  type CoreSearchParams,
  type CoreSearchResult,
  type CoreStatus,
  type CoreSyncOutcome,
  type SyncSessionInput,
} from './indexCore';
import {
  boundaryOf,
  decodePageToken,
  encodePageToken,
  matchDocs,
  paginateRows,
  type MatchedRow,
  type NormalizedQuery,
  type SearchBudgets,
} from './match';
import { makeSnippet } from './snippet';
import { SearchWorkerError, SearchWorkerHost, dropLiveLockToken, noteLiveLockToken } from './worker/host';

export { GlobalSearchError } from './contract';
export type { GlobalSearchErrorReason } from './contract';

const INDEX_DIR_NAME = 'search-index';
const SESSION_PAGE_SIZE = 500;

const MAX_QUERY_TERMS = 32;
const MAX_LITERAL_QUERY_CHARS = 1_024;
const MAX_POSTINGS_VISITS = 250_000;
const QUERY_DEADLINE_MS = 500;
const QUERY_TEXT_BUDGET_CHARS = 16_000_000;
const MAX_TEXT_HITS = 100_000;
const LITERAL_CANDIDATE_CAP = 10_000;

function errorMessage(error: unknown): string {
  return error instanceof Error ? error.message : String(error);
}

async function pathExists(path: string): Promise<boolean> {
  try {
    await stat(path);
    return true;
  } catch {
    return false;
  }
}

const pendingDisposals = new Set<Promise<void>>();

export async function drainGlobalSearchDisposals(): Promise<void> {
  while (pendingDisposals.size > 0) {
    await Promise.all(pendingDisposals);
  }
}

export interface IGlobalSearchService {
  readonly _serviceBrand: undefined;
  search(query: GlobalSearchQuery): Promise<GlobalSearchPage>;
  reindex(): Promise<{ sessions: number; documents: number }>;
  status(): Promise<{
    sessions: number;
    documents: number;
    lastIndexedAt: number | null;
    generation: number;
    degraded?: string;
    lifecycle: CoreLifecycleReport;
  }>;
  lifecycleReport(): CoreLifecycleReport;
  setLiveTranscriptSource(source: LiveTranscriptSource): void;
}

export const IGlobalSearchService = createDecorator<IGlobalSearchService>('globalSearch');

export interface LiveTranscriptSource {
  forSessionLive(sessionId: string): TranscriptStore | undefined;
  whenReady(sessionId: string): Promise<void>;
  ensureAgentHistory(sessionId: string, agentId: string): Promise<void>;
}

function normalizeQuery(input: GlobalSearchQuery, maxQueryTerms: number): NormalizedQuery {
  const mode = input.mode ?? 'terms';
  const query = mode === 'literal' ? input.query : input.query.trim();
  if (query.length === 0) {
    throw new GlobalSearchError('invalid_query', 'query must be a non-empty string');
  }
  const literalQuery = mode === 'literal' ? normalizeLiteral(query) : undefined;
  const termsQuery = mode === 'terms' ? [...new Set(tokenize(query))] : undefined;
  if (termsQuery !== undefined && termsQuery.length > maxQueryTerms) {
    throw new GlobalSearchError(
      'invalid_query',
      `query has too many terms (${termsQuery.length} > ${maxQueryTerms}); narrow it down`,
    );
  }
  const pageSize = input.pageSize ?? 20;
  if (!Number.isInteger(pageSize) || pageSize < 1 || pageSize > 50) {
    throw new GlobalSearchError('invalid_query', 'pageSize must be an integer between 1 and 50');
  }
  return {
    query,
    mode,
    literalQuery,
    termsQuery,
    op: input.op ?? 'AND',
    container: input.container,
    role: input.role,
    startTime: input.startTime,
    endTime: input.endTime,
    sort: input.sort ?? 'score',
    pageSize,
  };
}

export interface SearchBackend {
  beginClose(): void;
  lifecycleSnapshot(): CoreLifecycleReport;
  ensureOpen(): Promise<unknown>;
  search(params: CoreSearchParams): Promise<CoreSearchResult>;
  sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome>;
  refresh(): Promise<unknown>;
  reindex(): Promise<unknown>;
  status(): Promise<CoreStatus>;
  dispose(): Promise<void>;
}

export class InlineSearchBackend implements SearchBackend {
  readonly core: SearchIndexCore;

  constructor(options: { indexDir: string; log: ILogService }) {
    this.core = new SearchIndexCore({
      ...options,
      bootSalt: randomUUID(),
      onLockToken: noteLiveLockToken,
    });
  }

  beginClose(): void {
    this.core.beginClose();
  }

  lifecycleSnapshot(): CoreLifecycleReport {
    return this.core.lifecycleState();
  }

  ensureOpen(): Promise<void> {
    return this.core.ensureOpen();
  }

  search(params: CoreSearchParams): Promise<CoreSearchResult> {
    return this.core.search(params);
  }

  sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome> {
    return this.core.sync(sessions);
  }

  refresh(): Promise<void> {
    return this.core.refresh();
  }

  reindex(): Promise<void> {
    return this.core.reindex();
  }

  status(): Promise<CoreStatus> {
    return this.core.status();
  }

  dispose(): Promise<void> {
    dropLiveLockToken(this.core.lockTokenView);
    return this.core.close();
  }
}

export class GlobalSearchService implements IGlobalSearchService {
  declare readonly _serviceBrand: undefined;

  syncDebounceMs = 2_000;

  literalCandidateCap = LITERAL_CANDIDATE_CAP;

  maxTextHits = MAX_TEXT_HITS;

  postingsVisitBudget = MAX_POSTINGS_VISITS;

  queryDeadlineMs = QUERY_DEADLINE_MS;

  queryTextBudgetChars = QUERY_TEXT_BUDGET_CHARS;

  maxQueryTerms = MAX_QUERY_TERMS;

  private backend: SearchBackend | null = null;
  private backendPromise: Promise<SearchBackend> | null = null;
  private syncPromise: Promise<void> | null = null;
  private refreshPromise: Promise<void> | null = null;
  private lastSyncStartedAt = 0;
  private summaries = new Map<string, SessionSummary>();
  private disposed = false;
  private reindexing = false;
  private liveSource: LiveTranscriptSource | null = null;
  private syncQueued = false;
  private syncTimer: ReturnType<typeof setTimeout> | null = null;
  private lastRefreshError: { at: number; message: string } | null = null;
  private drainSettled = false;

  constructor(
    @ISessionIndex private readonly sessionIndex: ISessionIndex,
    @IBootstrapService private readonly bootstrap: IBootstrapService,
    @ILogService private readonly log: ILogService,
    @IConfigService private readonly config: IConfigService,
  ) {
    this.requestSync();
  }

  setLiveTranscriptSource(source: LiveTranscriptSource): void {
    this.liveSource = source;
  }

  private get indexDir(): string {
    return join(this.bootstrap.homeDir, INDEX_DIR_NAME);
  }

  private ensureBackend(): Promise<SearchBackend> {
    if (this.backend !== null) return Promise.resolve(this.backend);
    this.backendPromise ??= this.config.ready.then(() => {
      if (this.backend === null) {
        this.backend = databaseSearchEnabled(this.config)
          ? new SearchWorkerHost({ dir: this.indexDir, log: this.log })
          : new InlineSearchBackend({ indexDir: this.indexDir, log: this.log });
        if (this.disposed) this.backend.beginClose();
      }
      return this.backend;
    });
    return this.backendPromise;
  }

  private toSyncInput(summary: SessionSummary): SyncSessionInput {
    return {
      id: summary.id,
      workspaceId: summary.workspaceId,
      title: summary.title,
      updatedAt: summary.updatedAt,
      dir: sessionDirOf(
        this.bootstrap.homeDir,
        workspacePersistenceScope(this.bootstrap.scope('sessions'), summary.workspaceId),
        summary.id,
      ),
    };
  }

  dispose(): void {
    this.disposed = true;
    if (this.syncTimer !== null) {
      clearTimeout(this.syncTimer);
      this.syncTimer = null;
    }
    this.backend?.beginClose();
    const pending = (async () => {
      await this.syncPromise?.catch(() => {});
      await this.refreshPromise?.catch(() => {});
      const backend = await this.backendPromise;
      await backend?.dispose();
      this.drainSettled = true;
    })();
    pendingDisposals.add(pending);
    void pending.finally(() => pendingDisposals.delete(pending));
  }

  private requestSync(): void {
    if (this.disposed || this.reindexing) return;
    if (this.syncPromise !== null) {
      this.syncQueued = true;
      return;
    }
    const wait = this.syncDebounceMs - (Date.now() - this.lastSyncStartedAt);
    if (wait > 0) {
      if (this.syncTimer === null) {
        this.syncTimer = setTimeout(() => {
          this.syncTimer = null;
          this.requestSync();
        }, wait);
        this.syncTimer.unref?.();
      }
      return;
    }
    this.startSyncPass();
  }

  private startSyncPass(): void {
    this.syncQueued = false;
    void this.ensureSyncStarted().then(
      () => {
        this.lastRefreshError = null;
        if (this.syncQueued) {
          this.syncQueued = false;
          this.requestSync();
        }
      },
      (error: unknown) => {
        this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
        this.log.warn('global search: background sync failed', { error: errorMessage(error) });
      },
    );
  }

  private ensureSyncStarted(): Promise<void> {
    if (this.syncPromise === null) {
      const p = this.runSync().finally(() => {
        if (this.syncPromise === p) this.syncPromise = null;
      });
      this.syncPromise = p;
    }
    return this.syncPromise;
  }

  private async runSync(): Promise<void> {
    if (this.disposed || this.reindexing) return;
    const backend = await this.ensureBackend();
    const sessions = await this.listAllSessions();
    if (this.disposed) return;
    if (sessions.length === 0 && !(await pathExists(this.indexDir))) {
      this.summaries = new Map();
      this.lastSyncStartedAt = Date.now();
      return;
    }
    this.summaries = new Map(sessions.map((s) => [s.id, s]));
    this.lastSyncStartedAt = Date.now();
    const outcome = await backend.sync(sessions.map((s) => this.toSyncInput(s)));
    if (outcome.truncated || outcome.failures > 0) this.requestSync();
  }

  private async listAllSessions(): Promise<SessionSummary[]> {
    const out: SessionSummary[] = [];
    let cursor: string | undefined;
    do {
      const page = await this.sessionIndex.listRecent({ before: cursor, limit: SESSION_PAGE_SIZE });
      out.push(...page.items);
      cursor = page.nextCursor;
    } while (cursor !== undefined);
    return out;
  }

  private refreshReadonly(): Promise<void> {
    if (this.refreshPromise === null) {
      this.refreshPromise = this.ensureBackend()
        .then((backend) => backend.refresh())
        .then(
          () => {},
          (error: unknown) => {
            this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
            this.log.warn('global search: read-only refresh failed; serving the stale view', {
              error: errorMessage(error),
            });
          },
        );
      void this.refreshPromise.finally(() => {
        this.refreshPromise = null;
      });
    }
    return this.refreshPromise;
  }

  async search(input: GlobalSearchQuery): Promise<GlobalSearchPage> {
    const q = normalizeQuery(input, this.maxQueryTerms);
    const sessionId = q.container?.sessionId;
    const liveStore = sessionId !== undefined ? this.liveSource?.forSessionLive(sessionId) : undefined;
    if (liveStore !== undefined && sessionId !== undefined) {
      return this.searchLive(q, sessionId, liveStore, input.pageToken);
    }
    return this.searchIndex(q, input.pageToken);
  }

  private async searchLive(
    q: NormalizedQuery,
    sessionId: string,
    store: TranscriptStore,
    pageToken: string | undefined,
  ): Promise<GlobalSearchPage> {
    const page = decodePageToken(q, 'live', pageToken, undefined);
    const source = this.liveSource;
    if (source === null) {
      throw new GlobalSearchError('index_unavailable', 'live transcript source is not wired');
    }
    await source.whenReady(sessionId);
    const agentIds =
      q.container?.agentId !== undefined
        ? [q.container.agentId]
        : store.agents().map((agent) => agent.agentId);
    for (const agentId of agentIds) {
      await source.ensureAgentHistory(sessionId, agentId);
    }
    const docs = await this.collectLiveDocs(sessionId, store, agentIds);
    const budget = {
      deadlineAt: Date.now() + this.queryDeadlineMs,
      textCharsLeft: this.queryTextBudgetChars,
    };
    const boundary = page.kind === 'keyset' ? page.boundary : undefined;
    const matched =
      q.mode === 'literal'
        ? matchDocs(
            q,
            docs.map(({ key, value }) => ({ key, value, score: 0 })),
            boundary,
            budget,
          )
        : matchDocs(q, matchLiveTerms(q.termsQuery ?? [], docs), boundary, budget);
    const { pageRows, hasMore } = paginateRows(q, page, matched.rows);
    return {
      items: pageRows.map((row) => this.projectHit(q, row)),
      hasMore,
      pageToken: hasMore
        ? encodePageToken(q, 'live', boundaryOf(q, pageRows[pageRows.length - 1]!), undefined)
        : undefined,
      incomplete: matched.incomplete,
      indexState: {
        state: 'ready',
        indexedSessions: 1,
        totalSessions: 1,
        documents: docs.length,
      },
      source: 'live',
    };
  }

  private async collectLiveDocs(
    sessionId: string,
    store: TranscriptStore,
    agentIds: readonly string[],
  ): Promise<{ key: string; value: MessageDoc | TitleDoc }[]> {
    const summary = await this.sessionIndex.get(sessionId);
    const workspaceId = summary?.workspaceId ?? '';
    const sessionTitle = summary?.title ?? '';
    const fallbackTime = summary?.updatedAt ?? 0;
    const parseTime = (iso: string | undefined): number => {
      if (iso === undefined) return fallbackTime;
      const ms = Date.parse(iso);
      return Number.isNaN(ms) ? fallbackTime : ms;
    };
    const docs: { key: string; value: MessageDoc | TitleDoc }[] = [];
    for (const agentId of agentIds) {
      const transcript = store.getAgent(agentId);
      if (transcript === undefined) continue;
      for (const item of transcript.snapshot().items) {
        if (item.kind !== 'turn') continue;
        const turnTime = parseTime(item.startedAt);
        const prompt = item.prompt?.trim() ?? '';
        if (prompt.length > 0) {
          docs.push({
            key: `${sessionId}/${agentId}/live/u/t${item.ordinal}`,
            value: {
              kind: 'message',
              sessionId,
              workspaceId,
              sessionTitle,
              agentId,
              role: 'user',
              text: prompt.length > MAX_DOC_TEXT_CHARS ? prompt.slice(0, MAX_DOC_TEXT_CHARS) : prompt,
              time: turnTime,
              turn: item.ordinal,
              stepId: undefined,
            },
          });
        }
        for (const step of item.steps) {
          const stepTime = parseTime(step.endedAt ?? step.startedAt ?? item.startedAt);
          for (const frame of step.frames) {
            if (frame.kind !== 'text' || frame.role !== 'assistant') continue;
            const text = frame.text.trim();
            if (text.length === 0) continue;
            docs.push({
              key: `${sessionId}/${agentId}/live/a/${frame.frameId}`,
              value: {
                kind: 'message',
                sessionId,
                workspaceId,
                sessionTitle,
                agentId,
                role: 'assistant',
                text: text.length > MAX_DOC_TEXT_CHARS ? text.slice(0, MAX_DOC_TEXT_CHARS) : text,
                time: stepTime,
                turn: item.ordinal,
                stepId: step.stepId,
              },
            });
          }
        }
      }
    }
    if (sessionTitle.length > 0) {
      docs.push({
        key: `${sessionId}/$title`,
        value: {
          kind: 'title',
          sessionId,
          workspaceId,
          sessionTitle,
          agentId: '',
          role: 'title',
          text: sessionTitle,
          time: fallbackTime,
        },
      });
    }
    return docs;
  }

  private budgets(): SearchBudgets {
    return {
      literalCandidateCap: this.literalCandidateCap,
      maxTextHits: this.maxTextHits,
      postingsVisitBudget: this.postingsVisitBudget,
      queryDeadlineMs: this.queryDeadlineMs,
      queryTextBudgetChars: this.queryTextBudgetChars,
    };
  }

  private async searchIndex(
    q: NormalizedQuery,
    pageToken: string | undefined,
  ): Promise<GlobalSearchPage> {
    if (q.mode === 'literal') {
      const literalLength = Array.from(q.literalQuery ?? '').length;
      if (literalLength < 2) {
        throw new GlobalSearchError(
          'invalid_query',
          'literal queries need at least 2 characters (after Unicode normalization)',
        );
      }
      if (literalLength > MAX_LITERAL_QUERY_CHARS) {
        throw new GlobalSearchError(
          'invalid_query',
          `literal queries are limited to ${MAX_LITERAL_QUERY_CHARS} characters`,
        );
      }
    }

    let result: CoreSearchResult;
    try {
      const backend = await this.ensureBackend();
      result = await backend.search({ q, pageToken, budgets: this.budgets() });
    } catch (error) {
      if (error instanceof GlobalSearchError) {
        if (error.reason === 'index_unavailable') this.requestSync();
        throw error;
      }
      if (error instanceof SearchWorkerError) {
        this.lastRefreshError = { at: Date.now(), message: error.message };
        this.log.warn('global search: search worker unavailable; serving a degraded page', {
          error: error.message,
          code: error.code,
        });
        if (error.code === 'disposed') {
          throw new GlobalSearchError('index_unavailable', 'search service is disposed');
        }
        this.requestSync();
        if (pageToken !== undefined) {
          throw new GlobalSearchError(
            'invalid_page_token',
            'the search index is not ready yet; restart the search',
          );
        }
        return this.buildingPage(null);
      }
      throw error;
    }

    if (!result.index.readOnly) this.requestSync();

    if (result.kind === 'building') return this.buildingPage(result.index);

    return {
      items: result.rows.map((row) => this.projectHit(q, row)),
      hasMore: result.hasMore,
      pageToken: result.hasMore
        ? encodePageToken(
            q,
            'index',
            boundaryOf(q, result.rows[result.rows.length - 1]!),
            result.generation,
          )
        : undefined,
      incomplete: result.incomplete,
      indexState: this.composeIndexState(result.index),
      source: 'index',
    };
  }

  private projectHit(q: NormalizedQuery, row: MatchedRow): GlobalSearchHit {
    const doc = row.value;
    return {
      sessionId: doc.sessionId,
      workspaceId: doc.workspaceId,
      sessionTitle: doc.sessionTitle,
      agentId: doc.agentId,
      role: doc.role,
      snippet:
        doc.kind === 'title'
          ? doc.text
          : row.anchor !== undefined && q.literalQuery !== undefined
            ? makeSnippet(doc.text, q.query, 80, { at: row.anchor, len: q.literalQuery.length })
            : makeSnippet(doc.text, q.query),
      time: doc.time,
      turn: doc.kind === 'message' ? doc.turn : undefined,
      stepId: doc.kind === 'message' ? doc.stepId : undefined,
      score: row.score,
    };
  }

  private composeIndexState(view: CoreIndexView): GlobalSearchIndexState {
    const coordinatorStale = view.readOnly
      ? this.refreshPromise !== null
      : this.syncPromise !== null || this.syncQueued || this.syncTimer !== null;
    return {
      state: view.state,
      indexedSessions: view.indexedSessions,
      totalSessions: view.readOnly
        ? view.indexedSessions
        : Math.max(view.indexedSessions, this.summaries.size),
      documents: view.documents,
      stale: view.freshnessStale || coordinatorStale || undefined,
      degraded: this.lastRefreshError?.message ?? view.degraded,
    };
  }

  private buildingPage(view: CoreIndexView | null): GlobalSearchPage {
    const indexed = view?.indexedSessions ?? 0;
    const readOnly = view?.readOnly === true;
    return {
      items: [],
      hasMore: false,
      pageToken: undefined,
      incomplete: undefined,
      indexState: {
        state: 'building',
        indexedSessions: indexed,
        totalSessions: readOnly ? indexed : Math.max(indexed, this.summaries.size),
        documents: view?.documents ?? 0,
        stale: true,
        degraded: this.lastRefreshError?.message ?? view?.degraded,
      },
      source: 'index',
    };
  }

  async reindex(): Promise<{ sessions: number; documents: number }> {
    const backend = await this.ensureBackend();
    try {
      this.reindexing = true;
      await backend.ensureOpen();
      await this.syncPromise?.catch(() => {});
      await backend.reindex();
      this.reindexing = false;
      await this.ensureSyncStarted();
      this.lastRefreshError = null;
    } catch (error) {
      this.reindexing = false;
      this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
      throw error;
    }
    const stats = await backend.status();
    return { sessions: stats.sessions, documents: stats.documents };
  }

  async status(): Promise<{
    sessions: number;
    documents: number;
    lastIndexedAt: number | null;
    generation: number;
    degraded?: string;
    lifecycle: CoreLifecycleReport;
  }> {
    const empty = { sessions: 0, documents: 0, lastIndexedAt: null, generation: 0 };
    if (this.disposed) {
      return {
        ...empty,
        lifecycle: { state: this.drainSettled ? 'stopped' : 'closing' },
      };
    }
    try {
      const backend = await this.ensureBackend();
      const status = await backend.status();
      if (!status.readOnly) this.requestSync();
      return {
        sessions: status.sessions,
        documents: status.documents,
        lastIndexedAt: status.lastIndexedAt,
        generation: status.generation,
        degraded: this.lastRefreshError?.message ?? status.degraded,
        lifecycle: status.lifecycle,
      };
    } catch (error) {
      const message = errorMessage(error);
      return { ...empty, degraded: message, lifecycle: { state: 'degraded', detail: message } };
    }
  }

  lifecycleReport(): CoreLifecycleReport {
    if (this.disposed) return { state: this.drainSettled ? 'stopped' : 'closing' };
    return this.backend?.lifecycleSnapshot() ?? { state: 'stopped' };
  }
}

function matchLiveTerms(
  terms: readonly string[],
  docs: readonly { key: string; value: MessageDoc | TitleDoc }[],
): { key: string; value: MessageDoc | TitleDoc; score: number }[] {
  if (terms.length === 0) return [];
  const matched: { key: string; value: MessageDoc | TitleDoc; score: number }[] = [];
  for (const { key, value: doc } of docs) {
    const counts = new Map<string, number>();
    for (const token of tokenize(doc.text)) counts.set(token, (counts.get(token) ?? 0) + 1);
    let score = 0;
    let hit = true;
    for (const term of terms) {
      const tf = counts.get(term) ?? 0;
      if (tf === 0) {
        hit = false;
        break;
      }
      score += Math.log(1 + tf);
    }
    if (hit) matched.push({ key, value: doc, score });
  }
  return matched;
}

registerScopedService(
  LifecycleScope.App,
  IGlobalSearchService,
  GlobalSearchService,
  ScopeActivation.OnDemand,
  'globalSearch',
);

	import { randomUUID } from 'node:crypto';
	import { join } from 'node:path';
	import { stat } from 'node:fs/promises';

	import {
	createDecorator,
	databaseSearchEnabled,
	IBootstrapService,
	IConfigService,
	ILogService,
	ISessionIndex,
	LifecycleScope,
	ScopeActivation,
	registerScopedService,
	sessionDirOf,
	workspacePersistenceScope,
	type SessionSummary,
	} from '@moonshot-ai/agent-core-v2';
	import { normalizeLiteral, tokenize } from '@moonshot-ai/minidb';
	import type { TranscriptStore } from '@moonshot-ai/transcript';

	import {
	GlobalSearchError,
	type GlobalSearchHit,
	type GlobalSearchIndexState,
	type GlobalSearchPage,
	type GlobalSearchQuery,
	} from './contract';
	import { MAX_DOC_TEXT_CHARS, type MessageDoc, type TitleDoc } from './docs';
	import {
	SearchIndexCore,
	type CoreIndexView,
	type CoreLifecycleReport,
	type CoreSearchParams,
	type CoreSearchResult,
	type CoreStatus,
	type CoreSyncOutcome,
	type SyncSessionInput,
	} from './indexCore';
	import {
	boundaryOf,
	decodePageToken,
	encodePageToken,
	matchDocs,
	paginateRows,
	type MatchedRow,
	type NormalizedQuery,
	type SearchBudgets,
	} from './match';
	import { makeSnippet } from './snippet';
	import { SearchWorkerError, SearchWorkerHost, dropLiveLockToken, noteLiveLockToken } from './worker/host';

	export { GlobalSearchError } from './contract';
	export type { GlobalSearchErrorReason } from './contract';

	const INDEX_DIR_NAME = 'search-index';
	const SESSION_PAGE_SIZE = 500;

	const MAX_QUERY_TERMS = 32;
	const MAX_LITERAL_QUERY_CHARS = 1_024;
	const MAX_POSTINGS_VISITS = 250_000;
	const QUERY_DEADLINE_MS = 500;
	const QUERY_TEXT_BUDGET_CHARS = 16_000_000;
	const MAX_TEXT_HITS = 100_000;
	const LITERAL_CANDIDATE_CAP = 10_000;

	function errorMessage(error: unknown): string {
	return error instanceof Error ? error.message : String(error);
	}

	async function pathExists(path: string): Promise<boolean> {
	try {
	await stat(path);
	return true;
	} catch {
	return false;
	}
	}

	const pendingDisposals = new Set<Promise<void>>();

	export async function drainGlobalSearchDisposals(): Promise<void> {
	while (pendingDisposals.size > 0) {
	await Promise.all(pendingDisposals);
	}
	}

	export interface IGlobalSearchService {
	readonly _serviceBrand: undefined;
	search(query: GlobalSearchQuery): Promise<GlobalSearchPage>;
	reindex(): Promise<{ sessions: number; documents: number }>;
	status(): Promise<{
	sessions: number;
	documents: number;
	lastIndexedAt: number \| null;
	generation: number;
	degraded?: string;
	lifecycle: CoreLifecycleReport;
	}>;
	lifecycleReport(): CoreLifecycleReport;
	setLiveTranscriptSource(source: LiveTranscriptSource): void;
	}

	export const IGlobalSearchService = createDecorator<IGlobalSearchService>('globalSearch');

	export interface LiveTranscriptSource {
	forSessionLive(sessionId: string): TranscriptStore \| undefined;
	whenReady(sessionId: string): Promise<void>;
	ensureAgentHistory(sessionId: string, agentId: string): Promise<void>;
	}

	function normalizeQuery(input: GlobalSearchQuery, maxQueryTerms: number): NormalizedQuery {
	const mode = input.mode ?? 'terms';
	const query = mode === 'literal' ? input.query : input.query.trim();
	if (query.length === 0) {
	throw new GlobalSearchError('invalid_query', 'query must be a non-empty string');
	}
	const literalQuery = mode === 'literal' ? normalizeLiteral(query) : undefined;
	const termsQuery = mode === 'terms' ? [...new Set(tokenize(query))] : undefined;
	if (termsQuery !== undefined && termsQuery.length > maxQueryTerms) {
	throw new GlobalSearchError(
	'invalid_query',
	`query has too many terms (${termsQuery.length} > ${maxQueryTerms}); narrow it down`,
	);
	}
	const pageSize = input.pageSize ?? 20;
	if (!Number.isInteger(pageSize) \|\| pageSize < 1 \|\| pageSize > 50) {
	throw new GlobalSearchError('invalid_query', 'pageSize must be an integer between 1 and 50');
	}
	return {
	query,
	mode,
	literalQuery,
	termsQuery,
	op: input.op ?? 'AND',
	container: input.container,
	role: input.role,
	startTime: input.startTime,
	endTime: input.endTime,
	sort: input.sort ?? 'score',
	pageSize,
	};
	}

	export interface SearchBackend {
	beginClose(): void;
	lifecycleSnapshot(): CoreLifecycleReport;
	ensureOpen(): Promise<unknown>;
	search(params: CoreSearchParams): Promise<CoreSearchResult>;
	sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome>;
	refresh(): Promise<unknown>;
	reindex(): Promise<unknown>;
	status(): Promise<CoreStatus>;
	dispose(): Promise<void>;
	}

	export class InlineSearchBackend implements SearchBackend {
	readonly core: SearchIndexCore;

	constructor(options: { indexDir: string; log: ILogService }) {
	this.core = new SearchIndexCore({
	...options,
	bootSalt: randomUUID(),
	onLockToken: noteLiveLockToken,
	});
	}

	beginClose(): void {
	this.core.beginClose();
	}

	lifecycleSnapshot(): CoreLifecycleReport {
	return this.core.lifecycleState();
	}

	ensureOpen(): Promise<void> {
	return this.core.ensureOpen();
	}

	search(params: CoreSearchParams): Promise<CoreSearchResult> {
	return this.core.search(params);
	}

	sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome> {
	return this.core.sync(sessions);
	}

	refresh(): Promise<void> {
	return this.core.refresh();
	}

	reindex(): Promise<void> {
	return this.core.reindex();
	}

	status(): Promise<CoreStatus> {
	return this.core.status();
	}

	dispose(): Promise<void> {
	dropLiveLockToken(this.core.lockTokenView);
	return this.core.close();
	}
	}

	export class GlobalSearchService implements IGlobalSearchService {
	declare readonly _serviceBrand: undefined;

	syncDebounceMs = 2_000;

	literalCandidateCap = LITERAL_CANDIDATE_CAP;

	maxTextHits = MAX_TEXT_HITS;

	postingsVisitBudget = MAX_POSTINGS_VISITS;

	queryDeadlineMs = QUERY_DEADLINE_MS;

	queryTextBudgetChars = QUERY_TEXT_BUDGET_CHARS;

	maxQueryTerms = MAX_QUERY_TERMS;

	private backend: SearchBackend \| null = null;
	private backendPromise: Promise<SearchBackend> \| null = null;
	private syncPromise: Promise<void> \| null = null;
	private refreshPromise: Promise<void> \| null = null;
	private lastSyncStartedAt = 0;
	private summaries = new Map<string, SessionSummary>();
	private disposed = false;
	private reindexing = false;
	private liveSource: LiveTranscriptSource \| null = null;
	private syncQueued = false;
	private syncTimer: ReturnType<typeof setTimeout> \| null = null;
	private lastRefreshError: { at: number; message: string } \| null = null;
	private drainSettled = false;

	constructor(
	@ISessionIndex private readonly sessionIndex: ISessionIndex,
	@IBootstrapService private readonly bootstrap: IBootstrapService,
	@ILogService private readonly log: ILogService,
	@IConfigService private readonly config: IConfigService,
	) {
	this.requestSync();
	}

	setLiveTranscriptSource(source: LiveTranscriptSource): void {
	this.liveSource = source;
	}

	private get indexDir(): string {
	return join(this.bootstrap.homeDir, INDEX_DIR_NAME);
	}

	private ensureBackend(): Promise<SearchBackend> {
	if (this.backend !== null) return Promise.resolve(this.backend);
	this.backendPromise ??= this.config.ready.then(() => {
	if (this.backend === null) {
	this.backend = databaseSearchEnabled(this.config)
	? new SearchWorkerHost({ dir: this.indexDir, log: this.log })
	: new InlineSearchBackend({ indexDir: this.indexDir, log: this.log });
	if (this.disposed) this.backend.beginClose();
	}
	return this.backend;
	});
	return this.backendPromise;
	}

	private toSyncInput(summary: SessionSummary): SyncSessionInput {
	return {
	id: summary.id,
	workspaceId: summary.workspaceId,
	title: summary.title,
	updatedAt: summary.updatedAt,
	dir: sessionDirOf(
	this.bootstrap.homeDir,
	workspacePersistenceScope(this.bootstrap.scope('sessions'), summary.workspaceId),
	summary.id,
	),
	};
	}

	dispose(): void {
	this.disposed = true;
	if (this.syncTimer !== null) {
	clearTimeout(this.syncTimer);
	this.syncTimer = null;
	}
	this.backend?.beginClose();
	const pending = (async () => {
	await this.syncPromise?.catch(() => {});
	await this.refreshPromise?.catch(() => {});
	const backend = await this.backendPromise;
	await backend?.dispose();
	this.drainSettled = true;
	})();
	pendingDisposals.add(pending);
	void pending.finally(() => pendingDisposals.delete(pending));
	}

	private requestSync(): void {
	if (this.disposed \|\| this.reindexing) return;
	if (this.syncPromise !== null) {
	this.syncQueued = true;
	return;
	}
	const wait = this.syncDebounceMs - (Date.now() - this.lastSyncStartedAt);
	if (wait > 0) {
	if (this.syncTimer === null) {
	this.syncTimer = setTimeout(() => {
	this.syncTimer = null;
	this.requestSync();
	}, wait);
	this.syncTimer.unref?.();
	}
	return;
	}
	this.startSyncPass();
	}

	private startSyncPass(): void {
	this.syncQueued = false;
	void this.ensureSyncStarted().then(
	() => {
	this.lastRefreshError = null;
	if (this.syncQueued) {
	this.syncQueued = false;
	this.requestSync();
	}
	},
	(error: unknown) => {
	this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
	this.log.warn('global search: background sync failed', { error: errorMessage(error) });
	},
	);
	}

	private ensureSyncStarted(): Promise<void> {
	if (this.syncPromise === null) {
	const p = this.runSync().finally(() => {
	if (this.syncPromise === p) this.syncPromise = null;
	});
	this.syncPromise = p;
	}
	return this.syncPromise;
	}

	private async runSync(): Promise<void> {
	if (this.disposed \|\| this.reindexing) return;
	const backend = await this.ensureBackend();
	const sessions = await this.listAllSessions();
	if (this.disposed) return;
	if (sessions.length === 0 && !(await pathExists(this.indexDir))) {
	this.summaries = new Map();
	this.lastSyncStartedAt = Date.now();
	return;
	}
	this.summaries = new Map(sessions.map((s) => [s.id, s]));
	this.lastSyncStartedAt = Date.now();
	const outcome = await backend.sync(sessions.map((s) => this.toSyncInput(s)));
	if (outcome.truncated \|\| outcome.failures > 0) this.requestSync();
	}

	private async listAllSessions(): Promise<SessionSummary[]> {
	const out: SessionSummary[] = [];
	let cursor: string \| undefined;
	do {
	const page = await this.sessionIndex.listRecent({ before: cursor, limit: SESSION_PAGE_SIZE });
	out.push(...page.items);
	cursor = page.nextCursor;
	} while (cursor !== undefined);
	return out;
	}

	private refreshReadonly(): Promise<void> {
	if (this.refreshPromise === null) {
	this.refreshPromise = this.ensureBackend()
	.then((backend) => backend.refresh())
	.then(
	() => {},
	(error: unknown) => {
	this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
	this.log.warn('global search: read-only refresh failed; serving the stale view', {
	error: errorMessage(error),
	});
	},
	);
	void this.refreshPromise.finally(() => {
	this.refreshPromise = null;
	});
	}
	return this.refreshPromise;
	}

	async search(input: GlobalSearchQuery): Promise<GlobalSearchPage> {
	const q = normalizeQuery(input, this.maxQueryTerms);
	const sessionId = q.container?.sessionId;
	const liveStore = sessionId !== undefined ? this.liveSource?.forSessionLive(sessionId) : undefined;
	if (liveStore !== undefined && sessionId !== undefined) {
	return this.searchLive(q, sessionId, liveStore, input.pageToken);
	}
	return this.searchIndex(q, input.pageToken);
	}

	private async searchLive(
	q: NormalizedQuery,
	sessionId: string,
	store: TranscriptStore,
	pageToken: string \| undefined,
	): Promise<GlobalSearchPage> {
	const page = decodePageToken(q, 'live', pageToken, undefined);
	const source = this.liveSource;
	if (source === null) {
	throw new GlobalSearchError('index_unavailable', 'live transcript source is not wired');
	}
	await source.whenReady(sessionId);
	const agentIds =
	q.container?.agentId !== undefined
	? [q.container.agentId]
	: store.agents().map((agent) => agent.agentId);
	for (const agentId of agentIds) {
	await source.ensureAgentHistory(sessionId, agentId);
	}
	const docs = await this.collectLiveDocs(sessionId, store, agentIds);
	const budget = {
	deadlineAt: Date.now() + this.queryDeadlineMs,
	textCharsLeft: this.queryTextBudgetChars,
	};
	const boundary = page.kind === 'keyset' ? page.boundary : undefined;
	const matched =
	q.mode === 'literal'
	? matchDocs(
	q,
	docs.map(({ key, value }) => ({ key, value, score: 0 })),
	boundary,
	budget,
	)
	: matchDocs(q, matchLiveTerms(q.termsQuery ?? [], docs), boundary, budget);
	const { pageRows, hasMore } = paginateRows(q, page, matched.rows);
	return {
	items: pageRows.map((row) => this.projectHit(q, row)),
	hasMore,
	pageToken: hasMore
	? encodePageToken(q, 'live', boundaryOf(q, pageRows[pageRows.length - 1]!), undefined)
	: undefined,
	incomplete: matched.incomplete,
	indexState: {
	state: 'ready',
	indexedSessions: 1,
	totalSessions: 1,
	documents: docs.length,
	},
	source: 'live',
	};
	}

	private async collectLiveDocs(
	sessionId: string,
	store: TranscriptStore,
	agentIds: readonly string[],
	): Promise<{ key: string; value: MessageDoc \| TitleDoc }[]> {
	const summary = await this.sessionIndex.get(sessionId);
	const workspaceId = summary?.workspaceId ?? '';
	const sessionTitle = summary?.title ?? '';
	const fallbackTime = summary?.updatedAt ?? 0;
	const parseTime = (iso: string \| undefined): number => {
	if (iso === undefined) return fallbackTime;
	const ms = Date.parse(iso);
	return Number.isNaN(ms) ? fallbackTime : ms;
	};
	const docs: { key: string; value: MessageDoc \| TitleDoc }[] = [];
	for (const agentId of agentIds) {
	const transcript = store.getAgent(agentId);
	if (transcript === undefined) continue;
	for (const item of transcript.snapshot().items) {
	if (item.kind !== 'turn') continue;
	const turnTime = parseTime(item.startedAt);
	const prompt = item.prompt?.trim() ?? '';
	if (prompt.length > 0) {
	docs.push({
	key: `${sessionId}/${agentId}/live/u/t${item.ordinal}`,
	value: {
	kind: 'message',
	sessionId,
	workspaceId,
	sessionTitle,
	agentId,
	role: 'user',
	text: prompt.length > MAX_DOC_TEXT_CHARS ? prompt.slice(0, MAX_DOC_TEXT_CHARS) : prompt,
	time: turnTime,
	turn: item.ordinal,
	stepId: undefined,
	},
	});
	}
	for (const step of item.steps) {
	const stepTime = parseTime(step.endedAt ?? step.startedAt ?? item.startedAt);
	for (const frame of step.frames) {
	if (frame.kind !== 'text' \|\| frame.role !== 'assistant') continue;
	const text = frame.text.trim();
	if (text.length === 0) continue;
	docs.push({
	key: `${sessionId}/${agentId}/live/a/${frame.frameId}`,
	value: {
	kind: 'message',
	sessionId,
	workspaceId,
	sessionTitle,
	agentId,
	role: 'assistant',
	text: text.length > MAX_DOC_TEXT_CHARS ? text.slice(0, MAX_DOC_TEXT_CHARS) : text,
	time: stepTime,
	turn: item.ordinal,
	stepId: step.stepId,
	},
	});
	}
	}
	}
	}
	if (sessionTitle.length > 0) {
	docs.push({
	key: `${sessionId}/$title`,
	value: {
	kind: 'title',
	sessionId,
	workspaceId,
	sessionTitle,
	agentId: '',
	role: 'title',
	text: sessionTitle,
	time: fallbackTime,
	},
	});
	}
	return docs;
	}

	private budgets(): SearchBudgets {
	return {
	literalCandidateCap: this.literalCandidateCap,
	maxTextHits: this.maxTextHits,
	postingsVisitBudget: this.postingsVisitBudget,
	queryDeadlineMs: this.queryDeadlineMs,
	queryTextBudgetChars: this.queryTextBudgetChars,
	};
	}

	private async searchIndex(
	q: NormalizedQuery,
	pageToken: string \| undefined,
	): Promise<GlobalSearchPage> {
	if (q.mode === 'literal') {
	const literalLength = Array.from(q.literalQuery ?? '').length;
	if (literalLength < 2) {
	throw new GlobalSearchError(
	'invalid_query',
	'literal queries need at least 2 characters (after Unicode normalization)',
	);
	}
	if (literalLength > MAX_LITERAL_QUERY_CHARS) {
	throw new GlobalSearchError(
	'invalid_query',
	`literal queries are limited to ${MAX_LITERAL_QUERY_CHARS} characters`,
	);
	}
	}

	let result: CoreSearchResult;
	try {
	const backend = await this.ensureBackend();
	result = await backend.search({ q, pageToken, budgets: this.budgets() });
	} catch (error) {
	if (error instanceof GlobalSearchError) {
	if (error.reason === 'index_unavailable') this.requestSync();
	throw error;
	}
	if (error instanceof SearchWorkerError) {
	this.lastRefreshError = { at: Date.now(), message: error.message };
	this.log.warn('global search: search worker unavailable; serving a degraded page', {
	error: error.message,
	code: error.code,
	});
	if (error.code === 'disposed') {
	throw new GlobalSearchError('index_unavailable', 'search service is disposed');
	}
	this.requestSync();
	if (pageToken !== undefined) {
	throw new GlobalSearchError(
	'invalid_page_token',
	'the search index is not ready yet; restart the search',
	);
	}
	return this.buildingPage(null);
	}
	throw error;
	}

	if (!result.index.readOnly) this.requestSync();

	if (result.kind === 'building') return this.buildingPage(result.index);

	return {
	items: result.rows.map((row) => this.projectHit(q, row)),
	hasMore: result.hasMore,
	pageToken: result.hasMore
	? encodePageToken(
	q,
	'index',
	boundaryOf(q, result.rows[result.rows.length - 1]!),
	result.generation,
	)
	: undefined,
	incomplete: result.incomplete,
	indexState: this.composeIndexState(result.index),
	source: 'index',
	};
	}

	private projectHit(q: NormalizedQuery, row: MatchedRow): GlobalSearchHit {
	const doc = row.value;
	return {
	sessionId: doc.sessionId,
	workspaceId: doc.workspaceId,
	sessionTitle: doc.sessionTitle,
	agentId: doc.agentId,
	role: doc.role,
	snippet:
	doc.kind === 'title'
	? doc.text
	: row.anchor !== undefined && q.literalQuery !== undefined
	? makeSnippet(doc.text, q.query, 80, { at: row.anchor, len: q.literalQuery.length })
	: makeSnippet(doc.text, q.query),
	time: doc.time,
	turn: doc.kind === 'message' ? doc.turn : undefined,
	stepId: doc.kind === 'message' ? doc.stepId : undefined,
	score: row.score,
	};
	}

	private composeIndexState(view: CoreIndexView): GlobalSearchIndexState {
	const coordinatorStale = view.readOnly
	? this.refreshPromise !== null
	: this.syncPromise !== null \|\| this.syncQueued \|\| this.syncTimer !== null;
	return {
	state: view.state,
	indexedSessions: view.indexedSessions,
	totalSessions: view.readOnly
	? view.indexedSessions
	: Math.max(view.indexedSessions, this.summaries.size),
	documents: view.documents,
	stale: view.freshnessStale \|\| coordinatorStale \|\| undefined,
	degraded: this.lastRefreshError?.message ?? view.degraded,
	};
	}

	private buildingPage(view: CoreIndexView \| null): GlobalSearchPage {
	const indexed = view?.indexedSessions ?? 0;
	const readOnly = view?.readOnly === true;
	return {
	items: [],
	hasMore: false,
	pageToken: undefined,
	incomplete: undefined,
	indexState: {
	state: 'building',
	indexedSessions: indexed,
	totalSessions: readOnly ? indexed : Math.max(indexed, this.summaries.size),
	documents: view?.documents ?? 0,
	stale: true,
	degraded: this.lastRefreshError?.message ?? view?.degraded,
	},
	source: 'index',
	};
	}

	async reindex(): Promise<{ sessions: number; documents: number }> {
	const backend = await this.ensureBackend();
	try {
	this.reindexing = true;
	await backend.ensureOpen();
	await this.syncPromise?.catch(() => {});
	await backend.reindex();
	this.reindexing = false;
	await this.ensureSyncStarted();
	this.lastRefreshError = null;
	} catch (error) {
	this.reindexing = false;
	this.lastRefreshError = { at: Date.now(), message: errorMessage(error) };
	throw error;
	}
	const stats = await backend.status();
	return { sessions: stats.sessions, documents: stats.documents };
	}

	async status(): Promise<{
	sessions: number;
	documents: number;
	lastIndexedAt: number \| null;
	generation: number;
	degraded?: string;
	lifecycle: CoreLifecycleReport;
	}> {
	const empty = { sessions: 0, documents: 0, lastIndexedAt: null, generation: 0 };
	if (this.disposed) {
	return {
	...empty,
	lifecycle: { state: this.drainSettled ? 'stopped' : 'closing' },
	};
	}
	try {
	const backend = await this.ensureBackend();
	const status = await backend.status();
	if (!status.readOnly) this.requestSync();
	return {
	sessions: status.sessions,
	documents: status.documents,
	lastIndexedAt: status.lastIndexedAt,
	generation: status.generation,
	degraded: this.lastRefreshError?.message ?? status.degraded,
	lifecycle: status.lifecycle,
	};
	} catch (error) {
	const message = errorMessage(error);
	return { ...empty, degraded: message, lifecycle: { state: 'degraded', detail: message } };
	}
	}

	lifecycleReport(): CoreLifecycleReport {
	if (this.disposed) return { state: this.drainSettled ? 'stopped' : 'closing' };
	return this.backend?.lifecycleSnapshot() ?? { state: 'stopped' };
	}
	}

	function matchLiveTerms(
	terms: readonly string[],
	docs: readonly { key: string; value: MessageDoc \| TitleDoc }[],
	): { key: string; value: MessageDoc \| TitleDoc; score: number }[] {
	if (terms.length === 0) return [];
	const matched: { key: string; value: MessageDoc \| TitleDoc; score: number }[] = [];
	for (const { key, value: doc } of docs) {
	const counts = new Map<string, number>();
	for (const token of tokenize(doc.text)) counts.set(token, (counts.get(token) ?? 0) + 1);
	let score = 0;
	let hit = true;
	for (const term of terms) {
	const tf = counts.get(term) ?? 0;
	if (tf === 0) {
	hit = false;
	break;
	}
	score += Math.log(1 + tf);
	}
	if (hit) matched.push({ key, value: doc, score });
	}
	return matched;
	}

	registerScopedService(
	LifecycleScope.App,
	IGlobalSearchService,
	GlobalSearchService,
	ScopeActivation.OnDemand,
	'globalSearch',
	);