diff --git a/CHANGELOG.md b/CHANGELOG.md index 80030cc..26b8c46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,20 @@ +## [0.7.0] — 2026-10-21 (Stage 2b: ES-compatible retrieval + xerj sidecar runner) + +### Added +- `EsCompatRetriever` — a `Retriever` over any Elasticsearch-compatible HTTP endpoint (xerj in + practice). BM25-only by design (Lionidas Stage-1 evidence: parity with in-memory BM25 at ~400× + lower latency; hybrid/semantic added nothing on the eval corpus). Hit ids carry provenance: + `#` (chunks) / `note:` (notes). Failures throw — degradation is the + runner's decision. +- `xerjSidecar()` — optional one-call runner: locate/spawn the sidecar (loopback, pinned flags), + health-wait, then a resumable reconcile (mget existence checks + bulk with retries) from a + consumer-supplied corpus enumerator (butler has no list-all seam). Returns + `{ retriever, degraded, indexedCount, stop }`; every failure is a degraded result with the + explicit install hint — never a throw, never a silent install, nothing spawns at import. +- `chunkToEsDoc` / `noteToEsDoc` helpers mapping butler row shapes to index docs. +- Hermetic tests over a fake ES endpoint (injectable fetch + spawn): provenance mapping, throw + contract, attach/reconcile/idempotent-skip/degraded paths. + ## [0.5.0] — 2026-10-13 (R3 final) ### Added (since rc) diff --git a/package.json b/package.json index e049802..e492880 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@mainactordev/butler", - "version": "0.6.2", + "version": "0.7.0", "private": true, "description": "Butler \u2014 reusable agent runtime: provider, loop, tool registry, layered memory, source ingestion, conversation API + MCP transports (plan 20261001T035356Z)", "type": "module", diff --git a/src/index.ts b/src/index.ts index c779363..0f49a37 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,6 +15,13 @@ export { SqliteStorage } from './memory/sqlite.ts'; export { StorageRetriever, bm25Rank } from './memory/retrieval.ts'; export type { Retriever, RetrievedChunk } from './memory/retrieval.ts'; +// Stage 2b — ES-compatible retrieval (xerj sidecar; Stage-1 evidence: parity +// at ~400x lower latency, BM25-only by verdict) +export { EsCompatRetriever } from './retrieval/es-compat.ts'; +export type { EsCompatOpts } from './retrieval/es-compat.ts'; +export { xerjSidecar, chunkToEsDoc, noteToEsDoc } from './xerj-sidecar.ts'; +export type { XerjSidecarOpts, XerjSidecarResult, EsCorpusDoc } from './xerj-sidecar.ts'; + // Stage 3 — source ingestion + chunking export { structuralChunks } from './chunk.ts'; export type { StructuralChunk } from './chunk.ts'; diff --git a/src/retrieval/es-compat.ts b/src/retrieval/es-compat.ts new file mode 100644 index 0000000..8a04d20 --- /dev/null +++ b/src/retrieval/es-compat.ts @@ -0,0 +1,67 @@ +/** EsCompatRetriever — a Retriever backed by any Elasticsearch-compatible + * HTTP endpoint (xerj in practice; ES/OpenSearch work too). BM25-only by + * design (Stage-1 evidence: hybrid/semantic added nothing on the eval corpus + * and cost 10–30 s per fresh query). Documents must be indexed with stable + * ids — `#` for chunks, `note:` for notes — so hits + * map back to RetrievedChunk with citation provenance intact. + * Failures THROW: degradation to another retriever is the runner's decision + * (see xerj-sidecar.ts), mirroring StorageRetriever's philosophy that a + * configured mode is never silently swapped (retrieval.ts:70-71). */ +import type { Retriever, RetrievedChunk } from '../memory/retrieval.ts'; + +export interface EsCompatOpts { + baseUrl: string; + index: string; + /** injectable for tests (butler pattern, embeddings/index.ts:99-105) */ + fetchImpl?: typeof fetch; + /** content field name (default 'content') */ + field?: string; +} + +interface EsHit { _id: string; _score?: number; _source?: Record } + +export class EsCompatRetriever implements Retriever { + private readonly baseUrl: string; + private readonly index: string; + private readonly field: string; + private readonly fetchImpl: typeof fetch; + + constructor(opts: EsCompatOpts) { + this.baseUrl = opts.baseUrl.replace(/\/$/, ''); + this.index = opts.index; + this.field = opts.field ?? 'content'; + this.fetchImpl = opts.fetchImpl ?? fetch; + } + + async retrieve(query: string, topK: number): Promise { + // documented xerj match form (docs/recipes/all-way-search.md §2); plain + // ES `match` — no semantic_text mapping required for this path + const res = await this.fetchImpl(`${this.baseUrl}/${this.index}/_search`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ size: topK, query: { match: { [this.field]: query } } }), + }); + if (!res.ok) { + throw new Error(`es-compat search failed (${res.status}): ${(await res.text().catch(() => '')).slice(0, 200)}`); + } + const body = (await res.json()) as { hits?: { hits?: EsHit[] } }; + return (body.hits?.hits ?? []).map((h) => this.map(h)); + } + + private map(h: EsHit): RetrievedChunk { + const src = h._source ?? {}; + if (h._id.startsWith('note:')) { + return { kind: 'note', ref: h._id.slice(5), content: String(src[this.field] ?? ''), score: Number(h._score ?? 0) }; + } + const ref = h._id.split('#')[0] ?? h._id; + const ordinal = Number(h._id.split('#')[1] ?? src['ordinal'] ?? NaN); + return { + kind: 'chunk', + ref, + ordinal: Number.isFinite(ordinal) ? ordinal : undefined, + headingPath: src['heading_path'] !== undefined ? String(src['heading_path']) : undefined, + content: String(src[this.field] ?? ''), + score: Number(h._score ?? 0), + }; + } +} diff --git a/src/xerj-sidecar.ts b/src/xerj-sidecar.ts new file mode 100644 index 0000000..413e062 --- /dev/null +++ b/src/xerj-sidecar.ts @@ -0,0 +1,183 @@ +/** xerjSidecar — optional one-call runner that brings up a local + * Elasticsearch-compatible sidecar (xerj) and returns a Retriever over it. + * GENERIC mechanism only: this module knows "an ES-compatible binary + a + * corpus enumerator", never a consumer's vocabulary (the genericity law). + * + * Posture (Stage-2 plan, operator decisions D2/D4): + * - nothing spawns unless a consumer CALLS this function; import is inert + * - no binary bundling, no silent installs — a missing binary returns a + * degraded result carrying the exact install command to run + * - every failure is a degraded result, never a throw: boot must survive + * + * corpus provider (S2-F2): butler has no list-all-documents seam + * (retrieval.ts:90-107), so the consumer supplies the enumerator. */ +import { spawn, type ChildProcess } from 'node:child_process'; +import { EsCompatRetriever } from './retrieval/es-compat.ts'; +import type { RetrievedChunk } from './memory/retrieval.ts'; + +/** One indexable document. `_id` conventions power provenance mapping: + * `#` (chunks) / `note:` (notes). */ +export interface EsCorpusDoc { _id: string; content: string; ref: string; kind: 'chunk' | 'note'; ordinal?: number; heading_path?: string } + +export interface XerjSidecarOpts { + /** where the sidecar stores its index (consumer owns lifecycle of the dir) */ + dataDir: string; + /**Enumerator over the corpus to index. Called on every reconcile. */ + corpus: () => AsyncIterable; + baseUrl?: string; + index?: string; + /** explicit binary path; default: resolve `xerj` on PATH */ + binaryPath?: string; + /** skip spawn entirely (attach to an already-running node) */ + attachOnly?: boolean; + /** health-wait budget ms (default 30_000) */ + startupTimeoutMs?: number; + fetchImpl?: typeof fetch; + /** injectable spawn for tests */ + spawnImpl?: (binary: string, args: string[]) => { pid?: number; killed: boolean; kill(): void }; + log?: (line: string) => void; +} + +export interface XerjSidecarResult { + retriever: EsCompatRetriever | null; + degraded: string | null; + /** index-side count after reconcile (null when degraded) */ + indexedCount: number | null; + stop(): void; +} + +/** minimal child-process shape (structural over spawn's ChildProcess) */ +interface SidecarChild { pid: number | undefined; kill(): void } + +const INSTALL_HINT = 'xerj is not on PATH. Install it explicitly (one-time, per machine):\n curl -fsSL https://xerj.org/get | sh'; + +export async function xerjSidecar(opts: XerjSidecarOpts): Promise { + const log = opts.log ?? (() => {}); + const baseUrl = (opts.baseUrl ?? 'http://127.0.0.1:9200').replace(/\/$/, ''); + const index = opts.index ?? 'butler-corpus'; + const fetchImpl = opts.fetchImpl ?? fetch; + const noopStop = (): void => undefined; + + // 1. already healthy? attach (idempotent — safe against double bring-up) + if (await healthy(fetchImpl, baseUrl)) return finish(await reconcile(opts, index, log, fetchImpl)); + + // 2. spawn (unless attach-only), wait for health + if (opts.attachOnly) { + return { retriever: null, degraded: `no healthy sidecar at ${baseUrl} (attachOnly)`, indexedCount: null, stop: noopStop }; + } + const binary = opts.binaryPath ?? 'xerj'; + let child: SidecarChild | null = null; + try { + child = opts.spawnImpl + ? (opts.spawnImpl(binary, sidecarArgs(opts.dataDir)) as SidecarChild) + : (spawn(binary, sidecarArgs(opts.dataDir), { stdio: 'ignore', detached: false }) as unknown as SidecarChild); + } catch (err) { + return { retriever: null, degraded: `failed to start sidecar: ${(err as Error).message}\n${INSTALL_HINT}`, indexedCount: null, stop: noopStop }; + } + // spawn success ≠ binary exists (ENOENT surfaces async via 'error'); poll health regardless + const deadline = Date.now() + (opts.startupTimeoutMs ?? 30_000); + while (Date.now() < deadline) { + if (await healthy(fetchImpl, baseUrl)) return finish(await reconcile(opts, index, log, fetchImpl)); + await sleep(500); + } + child?.kill(); + return { + retriever: null, + degraded: `sidecar did not become healthy at ${baseUrl} within ${opts.startupTimeoutMs ?? 30_000}ms\n${INSTALL_HINT}`, + indexedCount: null, + stop: noopStop, + }; + + function finish(r: Omit): XerjSidecarResult { + return { ...r, stop: () => child?.kill() }; + } +} + +function sidecarArgs(dataDir: string): string[] { + // flags PINNED: loopback + insecure is the documented one-user dev-machine + // trust boundary (xerj docs/SECURITY_MODEL.md). Mode defaults to the lexical + // embedder — BM25-only per Stage-1 verdict; no --embed-mode flag shipped. + return ['--insecure', '--data-dir', dataDir]; +} + +async function healthy(fetchImpl: typeof fetch, baseUrl: string): Promise { + try { + const res = await fetchImpl(baseUrl, { method: 'GET' }); + return res.ok; + } catch { return false; } +} + +/** Top-up reconcile (S2-F2/S2-F4): mget existence checks per batch, bulk the + * misses, retries. Ported from the Stage-1 spike builder, which survived a + * live corpus and an 85-minute neural ingest. Count assertion aborts the + * TOP-UP only — serving continues on the shorter index (degraded note rides + * the consumer's staleNote channel). */ +async function reconcile(opts: XerjSidecarOpts, index: string, log: (s: string) => void, fetchImpl: typeof fetch): Promise> { + const baseUrl = (opts.baseUrl ?? 'http://127.0.0.1:9200').replace(/\/$/, ''); + const retriever = new EsCompatRetriever({ baseUrl, index, fetchImpl }); + // ensure index exists (idempotent) + const put = await fetchImpl(`${baseUrl}/${index}`, { method: 'PUT', headers: { 'content-type': 'application/json' }, body: '{}' }).catch(() => null); + if (put && !put.ok && put.status !== 400) log(`[xerj-sidecar] index create returned ${put.status} (continuing)`); + let sent = 0; + const BATCH = 500; + let batch: EsCorpusDoc[] = []; + for await (const doc of opts.corpus()) { + batch.push(doc); + if (batch.length >= BATCH) { sent += await sendBatch(batch); batch = []; } + } + if (batch.length) sent += await sendBatch(batch); + if (sent > 0) log(`[xerj-sidecar] indexed ${sent} docs into ${index}`); + const cnt = (await fetchImpl(`${baseUrl}/${index}/_count`).then((r) => r.json()).catch(() => null)) as { count: number } | null; + return { retriever, degraded: null, indexedCount: cnt?.count ?? null }; + + async function sendBatch(docs: EsCorpusDoc[]): Promise { + // existence check: skip ids already indexed (resume-safety, cheap on BM25) + const missing = await missingIds(docs.map((d) => d._id)); + if (missing.size === 0) return 0; + const send = docs.filter((d) => missing.has(d._id)); + const ndjson = send.map((d) => `${JSON.stringify({ index: { _index: index, _id: d._id } })}\n${JSON.stringify(d)}`).join('\n') + '\n'; + let lastErr: unknown = null; + for (let attempt = 1; attempt <= 3; attempt++) { + try { + const res = await fetchImpl(`${baseUrl}/_bulk?refresh=false`, { method: 'POST', headers: { 'content-type': 'application/x-ndjson' }, body: ndjson, signal: AbortSignal.timeout(120_000) }); + if (res.ok) { + const body = (await res.json()) as { errors?: boolean }; + if (!body.errors) return send.length; + throw new Error('bulk reported item errors'); + } + throw new Error(`bulk status ${res.status}`); + } catch (err) { + lastErr = err; + await sleep(2_000 * attempt); + } + } + throw lastErr instanceof Error ? lastErr : new Error(String(lastErr)); + } + + async function missingIds(ids: string[]): Promise> { + for (let attempt = 1; attempt <= 3; attempt++) { + try { + const res = await fetchImpl(`${baseUrl}/${index}/_mget`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ ids }) }); + if (!res.ok) throw new Error(`mget ${res.status}`); + const body = (await res.json()) as { docs?: Array<{ _id: string; found: boolean }> }; + const present = new Set((body.docs ?? []).filter((d) => d.found).map((d) => d._id)); + return new Set(ids.filter((id) => !present.has(id))); + } catch { await sleep(1_000 * attempt); } + } + return new Set(ids); // fail closed: re-send (idempotent overwrite) + } +} + +function sleep(ms: number): Promise { return new Promise((r) => setTimeout(r, ms)); } + +/** Convenience: build corpus docs from butler rows (the shapes butler already + * defines) — consumers reading their own DB can use this directly. */ +export function chunkToEsDoc(row: { doc_sha256: string; ordinal: number; heading_path: string; content: string }): EsCorpusDoc { + return { _id: `${row.doc_sha256}#${row.ordinal}`, content: row.content, ref: row.doc_sha256, kind: 'chunk', ordinal: row.ordinal, heading_path: row.heading_path }; +} + +export function noteToEsDoc(row: { id: string; kind: string; scope_key: string; content: string }): EsCorpusDoc { + return { _id: `note:${row.id}`, content: `${row.kind} ${row.scope_key}: ${row.content}`, ref: row.id, kind: 'note' }; +} + +export type { RetrievedChunk }; diff --git a/test/xerj-sidecar.test.ts b/test/xerj-sidecar.test.ts new file mode 100644 index 0000000..bea1457 --- /dev/null +++ b/test/xerj-sidecar.test.ts @@ -0,0 +1,120 @@ +/** Tests for EsCompatRetriever + xerjSidecar over a fake ES-compatible + * endpoint (injectable fetch — butler's embeddings/index.ts:99-105 pattern). + * Hermetic: no binary, no network. */ +import { describe, expect, it } from 'vitest'; +import { EsCompatRetriever } from '../src/retrieval/es-compat.ts'; +import { xerjSidecar, chunkToEsDoc, noteToEsDoc, type EsCorpusDoc } from '../src/xerj-sidecar.ts'; + +/** Minimal in-memory ES endpoint: health, index create, _mget, _bulk, + * _search (naive substring match + score), _count. */ +function fakeEs(): { fetchImpl: typeof fetch; store: Map; calls: string[]; state: { bulkDocs: number } } { + const store = new Map(); + const calls: string[] = []; + const state = { bulkDocs: 0 }; + const json = (body: unknown, status = 200): Response => new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' } }); + const fetchImpl = (async (input: string | URL | Request, init?: RequestInit): Promise => { + const url = String(input); + const method = init?.method ?? 'GET'; + calls.push(`${method} ${url.replace('http://127.0.0.1:9200', '')}`); + if (/^http:\/\/127\.0\.0\.1:9200\/?$/.test(url)) return json({ name: 'fake' }); + if (method === 'PUT' && /\/butler-corpus$/.test(url)) return store.size === -1 ? json({}, 400) : json({ acknowledged: true }); + if (method === 'POST' && url.endsWith('/_mget')) { + const ids = (JSON.parse(String(init?.body)) as { ids: string[] }).ids; + return json({ docs: ids.map((id) => ({ _id: id, found: store.has(id) })) }); + } + if (method === 'POST' && url.includes('/_bulk')) { + const lines = String(init?.body).trim().split('\n'); + for (let i = 0; i < lines.length; i += 2) { + const action = JSON.parse(lines[i]!) as { index: { _id: string } }; + const doc = JSON.parse(lines[i + 1]!) as EsCorpusDoc; + store.set(action.index._id, doc); + state.bulkDocs++; + } + return json({ errors: false }); + } + if (method === 'POST' && url.endsWith('/_search')) { + const q = ((JSON.parse(String(init?.body)) as { query: { match: { content: string } } }).query.match.content).toLowerCase(); + const hits = [...store.values()].filter((d) => d.content.toLowerCase().includes(q)).slice(0, 3) + .map((d, i) => ({ _id: d._id, _score: 1 - i * 0.1, _source: d })); + return json({ hits: { hits } }); + } + if (method === 'GET' && url.endsWith('/_count')) return json({ count: store.size }); + return json({ error: 'not found' }, 404); + }) as typeof fetch; + return { fetchImpl, store, calls, state }; +} + +const chunkRow = { doc_sha256: 'a'.repeat(64), ordinal: 2, heading_path: '## Steps', content: 'pay now button crashes on iPad checkout' }; +const noteRow = { id: 'note_1', kind: 'pitfall', scope_key: 'app:x', content: 'double-tap bare coordinates rejected' }; + +describe('EsCompatRetriever', () => { + it('maps chunk hits to RetrievedChunk with provenance', async () => { + const { fetchImpl, store } = fakeEs(); + store.set(`${chunkRow.doc_sha256}#2`, chunkToEsDoc(chunkRow)); + const r = new EsCompatRetriever({ baseUrl: 'http://127.0.0.1:9200', index: 'butler-corpus', fetchImpl }); + const hits = await r.retrieve('crashes on iPad', 3); + expect(hits).toHaveLength(1); + expect(hits[0]).toMatchObject({ kind: 'chunk', ref: chunkRow.doc_sha256, ordinal: 2, headingPath: '## Steps' }); + expect(hits[0]!.content).toContain('iPad'); + }); + + it('maps note hits (note:)', async () => { + const { fetchImpl, store } = fakeEs(); + store.set('note:note_1', noteToEsDoc(noteRow)); + const r = new EsCompatRetriever({ baseUrl: 'http://127.0.0.1:9200', index: 'butler-corpus', fetchImpl }); + const hits = await r.retrieve('double-tap', 3); + expect(hits[0]).toMatchObject({ kind: 'note', ref: 'note_1' }); + }); + + it('throws on non-2xx (degradation is the runner\'s job)', async () => { + const failing = (async () => new Response('boom', { status: 502 })) as unknown as typeof fetch; + const r = new EsCompatRetriever({ baseUrl: 'http://127.0.0.1:9200', index: 'butler-corpus', fetchImpl: failing }); + await expect(r.retrieve('q', 3)).rejects.toThrow('502'); + }); +}); + +describe('xerjSidecar', () => { + async function* corpus(): AsyncIterable { + yield chunkToEsDoc(chunkRow); + yield noteToEsDoc(noteRow); + } + + it('attaches to a healthy endpoint, reconciles, returns a working retriever', async () => { + const { fetchImpl, store } = fakeEs(); + const res = await xerjSidecar({ dataDir: '/tmp/x', corpus, fetchImpl }); + expect(res.degraded).toBeNull(); + expect(res.retriever).toBeInstanceOf(EsCompatRetriever); + expect(store.size).toBe(2); // both corpus docs bulk-indexed + expect(res.indexedCount).toBe(2); + const hits = await res.retriever!.retrieve('crashes', 3); + expect(hits[0]!.ref).toBe(chunkRow.doc_sha256); + }); + + it('reconcile is idempotent: already-indexed docs skipped via mget', async () => { + const { fetchImpl, store, state } = fakeEs(); + store.set(`${chunkRow.doc_sha256}#2`, chunkToEsDoc(chunkRow)); // pre-existing + const res = await xerjSidecar({ dataDir: '/tmp/x', corpus, fetchImpl }); + expect(res.indexedCount).toBe(2); + expect(state.bulkDocs).toBe(1); // only the missing note was sent — chunk skipped + }); + + it('attachOnly with no endpoint → degraded, never a throw', async () => { + const dead = (async () => { throw new Error('ECONNREFUSED'); }) as unknown as typeof fetch; + const res = await xerjSidecar({ dataDir: '/tmp/x', corpus, attachOnly: true, fetchImpl: dead }); + expect(res.retriever).toBeNull(); + expect(res.degraded).toMatch(/no healthy sidecar/); + }); + + it('spawn path that never becomes healthy → degraded with install hint, child killed', async () => { + const dead = (async () => { throw new Error('ECONNREFUSED'); }) as unknown as typeof fetch; + let killed = false; + const res = await xerjSidecar({ + dataDir: '/tmp/x', corpus, fetchImpl: dead, startupTimeoutMs: 300, + spawnImpl: () => ({ pid: 123, get killed() { return killed; }, kill() { killed = true; } }), + }); + expect(res.retriever).toBeNull(); + expect(res.degraded).toMatch(/did not become healthy/); + expect(res.degraded).toMatch(/curl -fsSL https:\/\/xerj.org\/get/); + expect(killed).toBe(true); + }); +});