diff --git a/src/cli.ts b/src/cli.ts index e7abdfe..976ef3c 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -63,6 +63,7 @@ import { activeEmbedModelId, resolveEmbeddingsConfig, validateEmbedDimension } f import { resolveEmbedProvider, type EmbedProvider } from "./embeddings/provider.js"; import { stderrDimRejectionSink } from "./embeddings/guard.js"; import { DeepLakeEnricherStore } from "./enricher/store-adapter.js"; +import { createRefreshSignal } from "./enricher/refresh-signal.js"; import { createPriorContentCache } from "./enricher/content-cache.js"; import type { ContentReader } from "./enricher/index.js"; import type { PipelineMetricsSink } from "./telemetry/index.js"; @@ -870,6 +871,12 @@ function buildLiveDurableWiring( async function runDaemon(): Promise { const ctx = safeResolveContext(); + const config = resolveConfig(); + // Scale-to-zero: the registration leg marks this dirty on every durable write, + // and the enricher consumes it to decide whether a cycle is worth a Deep Lake + // refresh. An idle repo never marks dirty, so the enricher issues zero Deep Lake + // reads and the Activeloop pod scales to zero (parity with honeycomb PRD-062b). + const refreshSignal = createRefreshSignal(); const portkey = resolvePortkeyConfig(); const embeddingsConfig = resolveEmbeddingsConfig({}); // AC-018i.7: catch a wrong NECTAR_EMBEDDINGS_OUTPUT_DIMENSION at config @@ -907,13 +914,23 @@ async function runDaemon(): Promise { const options: AssembleOptions = { broodPrereqs, + // Kill switch: NECTAR_ENRICHER_ENABLED=false runs registration + on-demand + // brood but no steady-state enricher poll (zero idle Deep Lake reads). + enricherEnabled: config.enricherEnabled, ...(bootProjection !== undefined ? { bootProjection } : {}), // PRD-018b: when Deep Lake creds resolve, start the update-on-change watch // pipeline against the durable store (the watch leg needs no LLM - it appends // pending version rows the enricher describes later). Absent creds -> the // watch leg stays dormant and /health says so (AC-018b.7). ...(ctx !== undefined - ? { tenancy: ctx.tenancy, projectRoot: ctx.projectRoot, registrationStore: ctx.store } + ? { + tenancy: ctx.tenancy, + projectRoot: ctx.projectRoot, + registrationStore: ctx.store, + // Producer half of the scale-to-zero refresh gate: a durable registration + // write means Deep Lake has a row the enricher's mirror needs to pull. + onRegistrationDurableWrite: () => refreshSignal.markDirty(), + } : {}), ...(live !== undefined && ctx !== undefined ? { @@ -933,8 +950,12 @@ async function runDaemon(): Promise { : {}), // AC-018i.1: stamp embed_model on enricher-embedded rows. embedModel, - // AC-018g.6: re-seed the working set each cycle from the durable store. + // AC-018g.6: re-seed the working set from the durable store, but only + // when the registration leg has flagged new durable rows. Consumer half + // of the scale-to-zero refresh gate: an idle repo never marks dirty, so + // the enricher issues zero Deep Lake reads and the pod scales to zero. refreshWorkingSet: () => live.enricher.refresh(ctx.tenancy), + shouldRefresh: () => refreshSignal.consume(), // AC-018g.9: prior-content cache backs the cosmetic-change gate. priorContentCache: createPriorContentCache(), // AC-018g.11/.12: trigger #2 - a debounced projection write after a diff --git a/src/config.ts b/src/config.ts index 10fea9f..6862bd0 100644 --- a/src/config.ts +++ b/src/config.ts @@ -34,6 +34,17 @@ export function isLoopbackHost(host: string): boolean { /** Worker poll floor: the enricher's 30s cadence (ai/enricher-and-llm-model.md). */ export const DEFAULT_POLL_INTERVAL_MS = 30_000; +/** + * Whether the steady-state enricher loop runs. `NECTAR_ENRICHER_ENABLED=false` + * keeps registration (fs-watch) and on-demand `brood` fully working but disables + * the background enricher poll, so an idle daemon issues ZERO Deep Lake reads. + * This is the operator kill switch for the scale-to-zero regression (an idle + * enricher otherwise refreshes Deep Lake every cycle): run nectar with the + * enricher off until the dirty-gated refresh is on the version you have deployed. + * Default: enabled. + */ +export const DEFAULT_ENRICHER_ENABLED = true; + /** Distinct from honeycomb's `daemon.pid`/`daemon.lock` so both daemons coexist in ~/.honeycomb (PRD-002d). */ export const DEFAULT_PID_FILE_NAME = "nectar.pid"; export const DEFAULT_LOCK_FILE_NAME = "nectar.lock"; @@ -57,6 +68,7 @@ export interface RuntimeConfig { readonly pidFilePath: string; readonly lockFilePath: string; readonly pollIntervalMs: number; + readonly enricherEnabled: boolean; } export interface RuntimeConfigOverrides { @@ -66,6 +78,7 @@ export interface RuntimeConfigOverrides { readonly pidFileName?: string; readonly lockFileName?: string; readonly pollIntervalMs?: number; + readonly enricherEnabled?: boolean; } /** @@ -102,6 +115,21 @@ function envStr(name: string): string | undefined { return raw; } +/** + * Read a boolean env var. Accepts 1/true/yes/on and 0/false/no/off (case + * insensitive). Unset, blank, or unrecognized returns `undefined` so the caller + * falls back to a default (an unrecognized value is treated as absent rather than + * silently coerced, matching the strict posture of `envInt`). + */ +function envBool(name: string): boolean | undefined { + const raw = envStr(name); + if (raw === undefined) return undefined; + const v = raw.trim().toLowerCase(); + if (v === "1" || v === "true" || v === "yes" || v === "on") return true; + if (v === "0" || v === "false" || v === "no" || v === "off") return false; + return undefined; +} + /** * Resolve the runtime config. Overrides win, then env, then defaults. The env * layer keeps the daemon operable without a config file; overrides exist so @@ -120,8 +148,13 @@ export function resolveConfig(overrides: RuntimeConfigOverrides = {}): RuntimeCo envInt("NECTAR_POLL_INTERVAL_MS", { min: MIN_POLL_INTERVAL_MS }) ?? DEFAULT_POLL_INTERVAL_MS; + const enricherEnabled = + overrides.enricherEnabled ?? + envBool("NECTAR_ENRICHER_ENABLED") ?? + DEFAULT_ENRICHER_ENABLED; + const pidFilePath = join(runtimeDir, overrides.pidFileName ?? DEFAULT_PID_FILE_NAME); const lockFilePath = join(runtimeDir, overrides.lockFileName ?? DEFAULT_LOCK_FILE_NAME); - return { host, port, runtimeDir, pidFilePath, lockFilePath, pollIntervalMs }; + return { host, port, runtimeDir, pidFilePath, lockFilePath, pollIntervalMs, enricherEnabled }; } diff --git a/src/daemon.ts b/src/daemon.ts index b0850c4..23d7d23 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -71,7 +71,7 @@ import type { AsyncHiveGraphStore, HiveGraphStore } from "./hive-graph/store.js" import { createDiskRegistrationFs } from "./registration/disk-fs.js"; import { RegistrationService, type RegistrationFs } from "./registration/service.js"; import type { WatcherState } from "./registration/fs-watch.js"; -import { StoreBridge } from "./registration/store-bridge.js"; +import { StoreBridge, type DurableWriteOp } from "./registration/store-bridge.js"; import { createSharedIgnore, type IgnorePredicate } from "./registration/ignore.js"; import { createTlshFuzzyStep, DEFAULT_TUNABLE_FUZZY_CONFIG, type FuzzyConfig } from "./registration/tlsh.js"; import { FilePendingReviewStore, type PendingReviewStore } from "./registration/review-store.js"; @@ -239,6 +239,13 @@ export interface AssembleOptions extends RuntimeConfigOverrides { readonly registrationTimer?: Timer; /** Debounce window for the intake (default: the intake's own `DEFAULT_DEBOUNCE_MS`). */ readonly registrationDebounceMs?: number; + /** + * Notified after each SUCCESSFUL durable registration write (the point new rows + * land in Deep Lake). Production wires it to the enricher refresh-signal so the + * enricher only re-reads Deep Lake after real file activity - an idle repo never + * triggers a refresh and the serverless pod scales to zero. Default: no-op. + */ + readonly onRegistrationDurableWrite?: (op: DurableWriteOp) => void; // ── Wave D daemon API surface (PRD-008 / PRD-012b) ────────────────────────── /** @@ -525,7 +532,12 @@ export function assembleDaemon(options: AssembleOptions = {}): AssembledDaemon { // `enricherStore` (see enricher/store-adapter.ts); the default is an empty // in-memory working set. let enricherLoop: EnricherLoop | null = null; - if (options.enricherEnabled ?? true) { + // Gate on the RESOLVED config (resolveConfig(options), line ~454), not the raw + // option: config.enricherEnabled honors an explicit `enricherEnabled` override + // AND the NECTAR_ENRICHER_ENABLED env, and keeps daemon.config in sync with the + // actual runtime behavior. Gating on options.enricherEnabled alone would ignore + // the env for any caller that does not re-forward the resolved value. + if (config.enricherEnabled) { const enricherStore: EnricherStore = options.enricherStore ?? new EnricherInMemoryStore(); const nowIso = options.enricherCycle?.nowIso ?? (() => new Date().toISOString()); const enricherHealthSink: EnricherCycleLogSink = { @@ -679,6 +691,9 @@ export function assembleDaemon(options: AssembleOptions = {}): AssembledDaemon { health.recordWatchFlushFailure(new Date().toISOString()); log({ level: "error", scope: "registration.bridge", op, err: String(err) }); }, + ...(options.onRegistrationDurableWrite !== undefined + ? { onDurableWrite: options.onRegistrationDurableWrite } + : {}), }); const reviews: PendingReviewStore = options.registrationReviews ?? new FilePendingReviewStore(join(config.runtimeDir, "pending-reviews.json")); diff --git a/src/enricher/cycle.ts b/src/enricher/cycle.ts index f255e76..114c20b 100644 --- a/src/enricher/cycle.ts +++ b/src/enricher/cycle.ts @@ -83,6 +83,15 @@ export interface EnricherCycleDeps { * NEC-016 AC-018g.6) so post-boot pending rows are picked up without a restart. */ readonly refreshWorkingSet?: () => Promise; + /** + * Gate for {@link refreshWorkingSet}: returns true when a durable refresh is + * worth a Deep Lake round trip. Production wires it to the registration + * dirty-signal's `consume()`, so an idle repo (no file changes) refreshes zero + * times and the Activeloop serverless pod idles to zero (scale-to-zero parity + * with honeycomb PRD-062b). Absent OR returning true => refresh every cycle + * (the pre-existing behavior every current caller and test relies on). + */ + readonly shouldRefresh?: () => boolean; /** Prior-content cache for the cosmetic-change gate (PRD-018g / NEC-026 AC-018g.9). */ readonly priorContentCache?: PriorContentCache; /** Active embedding model id stamped on rows carrying an embedding (PRD-018i / NEC-018 AC-018i.1). */ @@ -337,8 +346,13 @@ export async function runEnricherCycle( const now = deps.nowIso?.() ?? new Date().toISOString(); const logSink = deps.logSink ?? consoleCycleLogSink; - // Working-set freshness (AC-018g.6): re-seed from the durable store first. - if (deps.refreshWorkingSet !== undefined) { + // Working-set freshness (AC-018g.6): re-seed from the durable store first, but + // ONLY when a refresh could surface new rows. `shouldRefresh` (wired to the + // registration dirty-signal) is false on an idle repo, so an idle enricher + // issues NO Deep Lake read and the serverless pod can scale to zero + // (scale-to-zero parity with honeycomb PRD-062b). Absent `shouldRefresh` => + // refresh every cycle (the legacy behavior). + if (deps.refreshWorkingSet !== undefined && (deps.shouldRefresh?.() ?? true)) { try { await deps.refreshWorkingSet(); } catch { diff --git a/src/enricher/refresh-signal.ts b/src/enricher/refresh-signal.ts new file mode 100644 index 0000000..d711a02 --- /dev/null +++ b/src/enricher/refresh-signal.ts @@ -0,0 +1,54 @@ +/** + * The enricher working-set refresh signal (scale-to-zero). + * + * The steady-state enricher cycle re-seeds its in-memory mirror from Deep Lake + * (`refreshWorkingSet` -> `DeepLakeEnricherStore.refresh` -> a + * `SELECT ... FROM hive_graph_versions`) so version rows the registration watcher + * appended after boot become visible to the enricher's SEPARATE mirror. Doing + * that unconditionally on every ~30s tick means an idle repo (no file changes, + * nothing to enrich) still issues a Deep Lake read every 30s forever, which keeps + * the Activeloop serverless pod warm and defeats its scale-to-zero. That is the + * same idle-burn class honeycomb removed in PRD-062b; nectar re-introduced it by + * refreshing on every cycle. + * + * This one-bit signal gates the refresh. The registration leg is the ONLY + * producer of new durable rows this daemon writes, so it `markDirty()`s after + * each successful durable write; the enricher `consume()`s the flag to decide + * whether a refresh is worth a network round trip. When no file has changed the + * flag stays clear, the enricher issues NO Deep Lake read, and the pod idles to + * zero. The enricher's own describe write-backs go through a different store and + * never mark this signal, so there is no self-triggering refresh loop. + */ + +/** A read-and-clear "durable rows may have changed" flag shared by producer and consumer. */ +export interface RefreshSignal { + /** Producer (registration): a durable write happened; the next enricher tick should refresh. */ + markDirty(): void; + /** Consumer (enricher): read-and-clear. Returns true (and resets to clean) when a refresh is warranted. */ + consume(): boolean; + /** Non-destructive read, for tests and observability. Does NOT clear the flag. */ + readonly dirty: boolean; +} + +/** + * Create a {@link RefreshSignal}. Defaults to dirty so the FIRST enricher cycle + * always refreshes once (picking up any rows persisted before this boot, e.g. a + * prior session that appended `pending` rows but never described them); every + * subsequent refresh is gated on real registration activity. + */ +export function createRefreshSignal(initiallyDirty = true): RefreshSignal { + let dirty = initiallyDirty; + return { + markDirty(): void { + dirty = true; + }, + consume(): boolean { + const was = dirty; + dirty = false; + return was; + }, + get dirty(): boolean { + return dirty; + }, + }; +} diff --git a/src/registration/store-bridge.ts b/src/registration/store-bridge.ts index 2f7f33d..36d32db 100644 --- a/src/registration/store-bridge.ts +++ b/src/registration/store-bridge.ts @@ -51,6 +51,16 @@ export interface StoreBridgeOptions { * `/health` and the log; a failed flush is never silently dropped. */ onFlushError?(err: unknown, op: DurableWriteOp): void; + /** + * Fired after a durable write SUCCESSFULLY settles, signalling that Deep Lake + * now holds rows the enricher's separate mirror has not seen. The daemon wires + * this to the enricher refresh-signal's `markDirty`, so the enricher refreshes + * on its next tick ONLY after real registration activity - an idle repo never + * marks dirty and the enricher never reads Deep Lake (scale-to-zero). Fail-soft: + * a throwing callback is swallowed so it can never break the serialized write + * queue. Not fired for a parked/failed write (nothing durable landed). + */ + onDurableWrite?(op: DurableWriteOp): void; } /** @@ -64,6 +74,7 @@ export class StoreBridge implements HiveGraphStore { private readonly mirror = new InMemoryHiveGraphStore(); private readonly durable: AsyncHiveGraphStore; private readonly onFlushError: (err: unknown, op: DurableWriteOp) => void; + private readonly onDurableWrite: (op: DurableWriteOp) => void; /** The serialized durable-write queue. Every enqueue chains onto this so writes flush in order. */ private tail: Promise = Promise.resolve(); @@ -87,6 +98,7 @@ export class StoreBridge implements HiveGraphStore { constructor(opts: StoreBridgeOptions) { this.durable = opts.durable; this.onFlushError = opts.onFlushError ?? (() => {}); + this.onDurableWrite = opts.onDurableWrite ?? (() => {}); } /** @@ -163,6 +175,13 @@ export class StoreBridge implements HiveGraphStore { () => { this.pending -= 1; if (op === "deleteNectar") this.failedIdentityNectars.delete(nectar); // the nectar is gone; nothing left to orphan + // Deep Lake now holds a row the enricher's mirror has not seen: signal a + // refresh (scale-to-zero: fired only on real activity, never when idle). + try { + this.onDurableWrite(op); + } catch { + // fail-soft: a throwing signal must never poison the write queue. + } }, (err: unknown) => { this.pending -= 1; diff --git a/test/config.test.ts b/test/config.test.ts index a6ff330..c0b2b74 100644 --- a/test/config.test.ts +++ b/test/config.test.ts @@ -46,6 +46,31 @@ test("EX-2 a valid NECTAR_PORT is parsed exactly", () => { }); }); +test("NECTAR_ENRICHER_ENABLED defaults to true (enricher on)", () => { + withEnv("NECTAR_ENRICHER_ENABLED", undefined, () => { + assert.equal(resolveConfig().enricherEnabled, true); + }); +}); + +test("NECTAR_ENRICHER_ENABLED kill switch: false/0/off/no disable the enricher", () => { + for (const off of ["false", "0", "off", "no", "FALSE", "Off"]) { + withEnv("NECTAR_ENRICHER_ENABLED", off, () => { + assert.equal(resolveConfig().enricherEnabled, false, `"${off}" should disable the enricher`); + }); + } +}); + +test("NECTAR_ENRICHER_ENABLED true/1/on/yes enable the enricher; garbage falls back to default", () => { + for (const on of ["true", "1", "on", "yes"]) { + withEnv("NECTAR_ENRICHER_ENABLED", on, () => { + assert.equal(resolveConfig().enricherEnabled, true); + }); + } + withEnv("NECTAR_ENRICHER_ENABLED", "maybe", () => { + assert.equal(resolveConfig().enricherEnabled, true, "unrecognized value falls back to the default"); + }); +}); + test("EX-2 NECTAR_POLL_INTERVAL_MS below the safe floor throws (no 1 ms tight poll)", () => { for (const bad of ["0", "1", "500"]) { withEnv("NECTAR_POLL_INTERVAL_MS", bad, () => { diff --git a/test/enricher-refresh-gate.test.ts b/test/enricher-refresh-gate.test.ts new file mode 100644 index 0000000..a0193c6 --- /dev/null +++ b/test/enricher-refresh-gate.test.ts @@ -0,0 +1,61 @@ +/** + * Scale-to-zero: `runEnricherCycle` must gate its durable working-set refresh on + * `shouldRefresh`. When it returns false (an idle repo, per the registration + * dirty-signal), the cycle issues NO Deep Lake refresh; when true (or absent), + * it refreshes as before. Runs against the compiled module from `dist/`. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import type { Tenancy } from "../dist/hive-graph/model.js"; +import { createOffProvider } from "../dist/embeddings/provider.js"; +import { EnricherInMemoryStore, runEnricherCycle } from "../dist/enricher/index.js"; + +const TEN: Tenancy = { orgId: "legion", workspaceId: "eng", projectId: "nectar" }; + +/** Minimal idle-repo cycle deps: empty store, no LLM, a counting refresh spy. */ +function idleDeps(refreshCalls: { n: number }, shouldRefresh?: () => boolean) { + return { + store: new EnricherInMemoryStore(), + tenancy: TEN, + readContent: { read: () => null }, + portkey: null, + embedProvider: createOffProvider(), + refreshWorkingSet: async () => { + refreshCalls.n += 1; + }, + ...(shouldRefresh !== undefined ? { shouldRefresh } : {}), + }; +} + +test("shouldRefresh=false skips the Deep Lake refresh on an idle cycle", async () => { + const calls = { n: 0 }; + await runEnricherCycle(idleDeps(calls, () => false)); + assert.equal(calls.n, 0, "an idle cycle must not read Deep Lake"); +}); + +test("shouldRefresh=true performs the refresh", async () => { + const calls = { n: 0 }; + await runEnricherCycle(idleDeps(calls, () => true)); + assert.equal(calls.n, 1); +}); + +test("absent shouldRefresh preserves the legacy always-refresh behavior", async () => { + const calls = { n: 0 }; + await runEnricherCycle(idleDeps(calls)); // no shouldRefresh wired + assert.equal(calls.n, 1); +}); + +test("consume-style gate refreshes once then goes quiet across cycles", async () => { + const calls = { n: 0 }; + let armed = true; // one pending refresh, then idle forever + const shouldRefresh = () => { + const was = armed; + armed = false; + return was; + }; + const deps = idleDeps(calls, shouldRefresh); + await runEnricherCycle(deps); + await runEnricherCycle(deps); + await runEnricherCycle(deps); + assert.equal(calls.n, 1, "only the armed cycle refreshes; later idle cycles read nothing"); +}); diff --git a/test/refresh-signal.test.ts b/test/refresh-signal.test.ts new file mode 100644 index 0000000..2fc21d7 --- /dev/null +++ b/test/refresh-signal.test.ts @@ -0,0 +1,47 @@ +/** + * The enricher refresh signal (scale-to-zero): the one-bit dirty flag that gates + * the enricher's Deep Lake working-set refresh. Producer (registration) + * `markDirty`s on a durable write; consumer (enricher) `consume`s read-and-clear. + * Runs against the compiled module from `dist/`. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { createRefreshSignal } from "../dist/enricher/refresh-signal.js"; + +test("starts dirty by default so the FIRST enricher cycle refreshes once", () => { + const sig = createRefreshSignal(); + assert.equal(sig.dirty, true); + assert.equal(sig.consume(), true); +}); + +test("consume is read-and-clear: a second consume with no markDirty is clean", () => { + const sig = createRefreshSignal(); + assert.equal(sig.consume(), true); // the initial dirty + assert.equal(sig.consume(), false); // idle: no refresh warranted + assert.equal(sig.consume(), false); +}); + +test("markDirty re-arms exactly one refresh (idle-then-activity-then-idle)", () => { + const sig = createRefreshSignal(false); // start clean to model a settled idle daemon + assert.equal(sig.consume(), false); // idle tick: no Deep Lake read + sig.markDirty(); // a file changed -> registration wrote durably + assert.equal(sig.consume(), true); // next tick refreshes once + assert.equal(sig.consume(), false); // then goes quiet again +}); + +test("multiple markDirty between consumes collapse to a single refresh", () => { + const sig = createRefreshSignal(false); + sig.markDirty(); + sig.markDirty(); + sig.markDirty(); + assert.equal(sig.consume(), true); + assert.equal(sig.consume(), false); +}); + +test("dirty getter is non-destructive (observability, not consumption)", () => { + const sig = createRefreshSignal(false); + sig.markDirty(); + assert.equal(sig.dirty, true); + assert.equal(sig.dirty, true); // reading it did not clear it + assert.equal(sig.consume(), true); // consume still sees it +});