feat(history): semantic_search over messages + summaries via a shared embed-service

changelog.d/history-semantic-search.added.md

@@ -0,0 +1,9 @@
+- `HistoryModule` gains an optional `semantic_search` tool: meaning-based search
+  over the agent's raw messages (text plus its own think/journal/skip_reply
+  notes) and every compression summary, backed by a shared remote
+  embed-service whose vector index lives server-side, one namespace per store
+  (`new HistoryModule({ semantic: { url, token, namespace } })`). The module
+  keeps the index in sync itself — a background tick every 60 s and a bounded
+  catch-up before each search, messages watermarked by timestamp with an
+  overlap re-scan and summaries by `createdMs` — and backs off cleanly when
+  the service is unreachable, so the other four history tools are unaffected.

src/modules/history/index.ts

@@ -49,6 +49,7 @@ import type { ToolDefinition, ToolCall, ToolResult, ProcessEvent } from '../../t
 import type { EventResponse, ProcessState } from '../../types/module.js';
 import type { SearchWorkerMessage, SearchWorkerMatch } from './search-regex-worker.js';
 import type { ChannelRegistry } from '../../mcpl/channel-registry.js';
+import { SemanticIndexClient, SemanticIndexer, type SemanticIndexConfig, type SyncReport } from './semantic.js';
 
 // ============================================================================
 // Tool input shapes
@@ -88,6 +89,22 @@ interface OverviewInput {
   limit?: number;
 }
 
+interface SemanticSearchInput {
+  query: string;
+  limit?: number;
+  from?: string;
+  to?: string;
+  channelId?: string;
+  kinds?: 'messages' | 'summaries' | 'both';
+  level?: number;
+  minScore?: number;
+}
+
+export interface HistoryModuleOptions {
+  /** Enable `semantic_search` against a shared embed-service (see ./semantic.ts). Absent = tool not offered. */
+  semantic?: SemanticIndexConfig;
+}
+
 // ============================================================================
 // Limits
 // ============================================================================
@@ -126,6 +143,10 @@ const SEARCH_MAX_MAX_SCAN = 50000;
 const OVERVIEW_DEFAULT_LIMIT = 50;
 const OVERVIEW_MAX_LIMIT = 200;
 
+const SEMANTIC_DEFAULT_LIMIT = 10;
+const SEMANTIC_MAX_LIMIT = 50;
+const SEMANTIC_SNIPPET_CHARS = 400;
+
 /**
  * A `<@id>`/`<@!id>` Discord-style mention — same shape `ChannelRegistry`
  * uses for its own DM mention-form matching. A string matching this can
@@ -163,6 +184,16 @@ export class HistoryModule implements Module {
   private ctx: ModuleContext | null = null;
   private cm: ContextManager | null = null;
   private channelRegistry: ChannelRegistry | null = null;
+  private readonly semanticCfg: SemanticIndexConfig | null;
+  private semanticClient: SemanticIndexClient | null = null;
+  private indexer: SemanticIndexer | null = null;
+  private syncTimer: ReturnType<typeof setInterval> | null = null;
+  private firstSyncTimer: ReturnType<typeof setTimeout> | null = null;
+
+  constructor(options: HistoryModuleOptions = {}) {
+    this.semanticCfg = options.semantic ?? null;
+    if (this.semanticCfg) this.semanticClient = new SemanticIndexClient(this.semanticCfg);
+  }
 
   /**
    * Wire the context-manager instance, and optionally the host's
@@ -180,6 +211,10 @@ export class HistoryModule implements Module {
   bind(contextManager: ContextManager, channelRegistry?: ChannelRegistry): void {
     this.cm = contextManager;
     this.channelRegistry = channelRegistry ?? null;
+    if (this.semanticCfg && this.semanticClient) {
+      this.indexer = new SemanticIndexer(contextManager, this.semanticClient, this.semanticCfg, (m) => console.warn(m));
+      this.startSyncTimer();
+    }
   }
 
   /**
@@ -260,10 +295,36 @@ export class HistoryModule implements Module {
 
   async start(ctx: ModuleContext): Promise<void> {
     this.ctx = ctx;
+    this.startSyncTimer();
   }
 
   async stop(): Promise<void> {
     this.ctx = null;
+    if (this.syncTimer) { clearInterval(this.syncTimer); this.syncTimer = null; }
+    if (this.firstSyncTimer) { clearTimeout(this.firstSyncTimer); this.firstSyncTimer = null; }
+  }
+
+  /**
+   * Background sync into the semantic index: every `syncIntervalMs` (default
+   * 60 s) push up to `maxSyncPerTick` new items. Idempotent to call — starts
+   * once we have both a context-manager (bind) and a config; `unref`'d so it
+   * never keeps a shutting-down process alive. A first tick runs after 5 s so
+   * a fresh store starts backfilling immediately rather than a minute later.
+   */
+  private startSyncTimer(): void {
+    if (this.syncTimer || !this.indexer || !this.semanticCfg) return;
+    const interval = this.semanticCfg.syncIntervalMs ?? 60_000;
+    if (interval <= 0) return;
+    const perTick = this.semanticCfg.maxSyncPerTick ?? 1024;
+    const tick = (): void => { void this.indexer?.catchUp(perTick); };
+    this.firstSyncTimer = setTimeout(() => { this.firstSyncTimer = null; tick(); }, 5_000);
+    this.firstSyncTimer.unref?.();
+    this.syncTimer = setInterval(tick, interval); this.syncTimer.unref?.();
+  }
+
+  /** Exposed for hosts/tests: one sync pass now. */
+  syncSemanticIndex(maxItems = 1024): Promise<SyncReport> | null {
+    return this.indexer ? this.indexer.catchUp(maxItems) : null;
   }
 
   getTools(): ToolDefinition[] {
@@ -373,9 +434,41 @@ export class HistoryModule implements Module {
           },
         },
       },
+      ...(this.semanticCfg ? [this.semanticSearchTool()] : []),
     ];
   }
 
+  private semanticSearchTool(): ToolDefinition {
+    return {
+      name: 'semantic_search',
+      description:
+        'Search your own history by MEANING, not exact words: an embedding index over your raw messages ' +
+        '(text plus your own think/journal/skip_reply notes) and every compression summary. Use it when you ' +
+        'remember roughly what something was about but not the words — "the night the fluid sim was read ' +
+        'back to me as art" — then narrow with `from`/`to`/`channelId` and drill into the exact span with ' +
+        '`extract` or `overview`. Results are ranked by cosine similarity (score ~0.6+ is a strong match, ' +
+        '~0.3 is thematic, below ~0.2 is noise); each hit carries its id (`msg:<id>` or `sum:<id>`), ' +
+        'timestamp, channel, kind/level and a snippet. The index catches up with recent messages before ' +
+        'searching (bounded, so a huge backlog is reported as `index.behind` rather than blocking; if a ' +
+        'background sync is already running, the search waits for that run to finish instead). ' +
+        'Purely a read: nothing is written to your history.',
+      inputSchema: {
+        type: 'object' as const,
+        properties: {
+          query: { type: 'string', description: 'What you are looking for, in natural language. A sentence works better than keywords.' },
+          limit: { type: 'number', description: `Max hits (default ${SEMANTIC_DEFAULT_LIMIT}, cap ${SEMANTIC_MAX_LIMIT}).` },
+          from: { type: 'string', description: 'ISO 8601 inclusive lower bound on the message timestamp / summary span start. Omit for open-ended.' },
+          to: { type: 'string', description: 'ISO 8601 inclusive upper bound. Omit for open-ended.' },
+          channelId: { type: 'string', description: 'Only raw messages from this channel (label like "#general" or the raw internal id). Summaries are not per-channel and are excluded when this is set; combining it with kinds="summaries" or level is an error.' },
+          kinds: { type: 'string', enum: ['messages', 'summaries', 'both'], description: 'What to search: raw messages, compression summaries, or both (default).' },
+          level: { type: 'number', description: 'Only summaries of this exact level (implies kinds=summaries).' },
+          minScore: { type: 'number', description: 'Drop hits below this cosine score (0..1). Default none.' },
+        },
+        required: ['query'],
+      },
+    };
+  }
+
   async handleToolCall(call: ToolCall): Promise<ToolResult> {
     try {
       if (!this.cm) {
@@ -390,6 +483,8 @@ export class HistoryModule implements Module {
           return await this.handleSearch((call.input ?? {}) as SearchInput);
         case 'overview':
           return this.handleOverview((call.input ?? {}) as OverviewInput);
+        case 'semantic_search':
+          return await this.handleSemanticSearch((call.input ?? {}) as SemanticSearchInput);
         default:
           return { success: false, isError: true, error: `Unknown tool: ${call.name}` };
       }
@@ -649,6 +744,73 @@ export class HistoryModule implements Module {
   // overview
   // ==========================================================================
 
+  private async handleSemanticSearch(input: SemanticSearchInput): Promise<ToolResult> {
+    if (!this.semanticCfg || !this.semanticClient) {
+      throw new Error('semantic_search is not configured for this resident (no embed-service in the recipe).');
+    }
+    if (typeof input.query !== 'string' || !input.query.trim()) throw new Error('query must be a non-empty string');
+    const limit = clampCount(input.limit, SEMANTIC_DEFAULT_LIMIT, SEMANTIC_MAX_LIMIT, 'limit');
+    if (limit < 1) throw new Error('limit must be at least 1');
+    const fromMs = parseIsoDate(input.from, 'from');
+    const toMs = parseIsoDate(input.to, 'to');
+    if (fromMs !== undefined && toMs !== undefined && fromMs > toMs) throw new Error('`from` must not be after `to`');
+    const channelId = this.resolveChannel(input.channelId);
+    // Summaries carry no channel, so a channel filter on a summaries-only
+    // search can only ever match nothing. Say so instead of returning [].
+    const wantsSummariesOnly = input.level !== undefined || input.kinds === 'summaries';
+    if (channelId && wantsSummariesOnly) {
+      throw new Error('channelId cannot be combined with kinds="summaries" or level: summaries are not per-channel. Drop channelId, or search kinds="messages".');
+    }
+    if (input.level !== undefined && input.kinds === 'messages') {
+      throw new Error('level applies to summaries only and cannot be combined with kinds="messages".');
+    }
+    let kinds: string[] | undefined;
+    if (wantsSummariesOnly) kinds = ['summary'];
+    else if (input.kinds === 'messages' || channelId) kinds = ['message'];
+    if (input.level !== undefined && (!Number.isInteger(input.level) || input.level < 0)) throw new Error('level must be a non-negative integer');
+    if (input.minScore !== undefined && (typeof input.minScore !== 'number' || input.minScore < -1 || input.minScore > 1)) {
+      throw new Error('minScore must be a number in -1..1');
+    }
+
+    // Bounded catch-up so the newest messages are searchable; never block on a backlog.
+    let sync: SyncReport | null = null;
+    if (this.indexer) {
+      try { sync = await this.indexer.catchUp(this.semanticCfg.maxSyncBeforeSearch ?? 256); } catch { /* reported via indexer.lastError */ }
+    }
+
+    const res = await this.semanticClient.search({
+      query: input.query, k: limit,
+      ts_from: fromMs === undefined ? undefined : fromMs / 1000,
+      ts_to: toMs === undefined ? undefined : toMs / 1000,
+      channel: channelId, kinds, level: input.level, min_score: input.minScore, snippet: SEMANTIC_SNIPPET_CHARS,
+    });
+    const hits = res.hits.map((h) => ({
+      id: h.id,
+      kind: h.kind,
+      level: h.level,
+      score: h.score,
+      timestamp: h.ts === null ? null : new Date(h.ts * 1000).toISOString(),
+      channelId: h.channel,
+      participant: (h.meta as { participant?: unknown }).participant ?? null,
+      author: (h.meta as { author?: unknown }).author ?? null,
+      snippet: h.text ?? '',
+      chars: h.chars,
+    }));
+    return {
+      success: true,
+      data: {
+        hits,
+        index: {
+          indexed: res.count_indexed,
+          behind: sync ? sync.more : this.indexer === null ? null : true,
+          syncedThisCall: sync ? sync.pushed : 0,
+          lastError: this.indexer?.lastError ?? null,
+        },
+        timingMs: res.timing_ms,
+      },
+    };
+  }
+
   private handleOverview(input: OverviewInput): ToolResult {
     const channelId = this.resolveChannel(input.channelId);
     const fromMs = parseIsoDate(input.from, 'from');