From 7c18fe1c36f4ae2e50f3dcdd31edf998701ea20d Mon Sep 17 00:00:00 2001 From: test Date: Sun, 2 Aug 2026 02:33:15 +0000 Subject: [PATCH] Add backfill_openclaw_fixture hermetic smoke for Lane B sweep New backfill_openclaw_fixture.js under hypaware-core/smoke/flows, mirroring backfill_claude_fixture.js / backfill_codex_fixture.js: writes a minimal OpenClaw v3 session JSONL in the nested-message-envelope shape (PR #552's reader) under a temp agents//sessions/ tree with a controllable mtime, drives the real createBackfillSweepDriver (T9) through a cron-due tick, and asserts (a) a file inside the default 180000ms quiesce window is skipped, (b) a file backdated past it is captured with native message identity, and (c) rerunning the sweep on a later cron-due tick (forcing a fresh devRunId, so the ai-gateway materializer's dedupe genuinely re-scans committed partitions) nets zero new rows for the already-written part_ids. Driving a real, non-dry-run sweep write for the first time (T9's own tests only ever exercised a mocked runBackfill seam) surfaced a latent bug: writeRows/flushDataset read ctx.query, which BackfillRunnerContext never carried and the daemon's createBackfillSweepDriver(...) call never supplied, so any real sweep write actually crashed on "Cannot read properties of undefined (reading 'getDataset')" in both the smoke and the real daemon path. Threaded query through BackfillRunnerContext, BackfillSweepDriverOptions, createBackfillSweepDriver, and the daemon's sweepDriver construction, and updated LLP 0172's field enumeration and ctx samples (Sections 4.3/4.4) to match. Extended the existing T9 unit tests (test/core/daemon-backfill-sweep.test.js) to cover the new required field and its passthrough. Task-Id: T12 --- .../smoke/flows/backfill_openclaw_fixture.js | 456 ++++++++++++++++++ llp/0172-openclaw-two-lane-capture.design.md | 58 ++- src/core/commands/backfill.js | 22 +- src/core/commands/types.d.ts | 12 +- src/core/daemon/backfill_sweep.js | 5 +- src/core/daemon/runtime.js | 1 + src/core/daemon/types.d.ts | 8 + test/core/daemon-backfill-sweep.test.js | 9 + 8 files changed, 529 insertions(+), 42 deletions(-) create mode 100644 hypaware-core/smoke/flows/backfill_openclaw_fixture.js diff --git a/hypaware-core/smoke/flows/backfill_openclaw_fixture.js b/hypaware-core/smoke/flows/backfill_openclaw_fixture.js new file mode 100644 index 00000000..77d5af3e --- /dev/null +++ b/hypaware-core/smoke/flows/backfill_openclaw_fixture.js @@ -0,0 +1,456 @@ +// @ts-check + +import fs from 'node:fs/promises' +import path from 'node:path' +import process from 'node:process' + +import { Attr, installObservability, runRoot } from '../../../src/core/observability/index.js' +import { dispatch } from '../../../src/core/cli/dispatch.js' +import { createCommandRegistry } from '../../../src/core/registry/commands.js' +import { registerCoreCommands } from '../../../src/core/cli/core_commands.js' +import { createKernelRuntime } from '../../../src/core/runtime/activation.js' +import { activatePlugins } from '../../../src/core/runtime/loader.js' +import { loadManifests } from '../../../src/core/manifest.js' +import { resolveDependencies } from '../../../src/core/dep_graph.js' +import { createBackfillSweepDriver } from '../../../src/core/daemon/backfill_sweep.js' +import { runBackfillProvider } from '../../../src/core/commands/backfill.js' + +/** + * LLP 0173 T12 smoke: OpenClaw Lane B sweep -> quiesce filter -> dedupe. + * + * Boots `@hypaware/ai-gateway` + `@hypaware/openclaw` against a tmp + * `HYP_HOME` with two staged OpenClaw v3 session fixtures under the fake + * HOME's `.openclaw/agents/main/sessions/`, both in the nested-`message`- + * envelope shape PR #552's reader (`session_file.js`) projects, and drives + * `src/core/daemon/backfill_sweep.js`'s real `createBackfillSweepDriver` + * (the same driver `runTick()` wires into the daemon's sink-tick cadence, + * LLP 0172#lane-b-sweep) directly against this boot's own + * `kernel.backfills` / `kernel.backfillMaterializers` / `kernel.storage`, + * so a sweep-written row and a `hyp query`-read row land in and come from + * the exact same tables `hyp backfill openclaw` would use. + * + * Asserts the three properties LLP 0173's T12 brief names: + * + * - **(a) quiesce skip**: a session file whose mtime is inside the + * default 180000ms quiesce window (LLP 0172#45-the-quiesce-window) is + * absent from the sweep's first tick. + * - **(b) quiesce capture**: a session file backdated past the window is + * captured by that same tick, with native message identity. + * - **(c) cross-write dedupe**: a second sweep tick (a fresh `now`, so the + * ai-gateway materializer's dedupe gets its own `devRunId` and is + * forced to re-scan committed partitions rather than reuse an + * in-memory seen set) finds the first tick's part_ids already + * committed and writes ZERO new rows. R11's identity-convergence + * argument (`openclaw/src/backfill.js`'s own module doc) is exactly + * that Lane A (live) and Lane B (backfill) land on the same + * `part_id` for the same turn, so the dedupe this proves for two + * sweep ticks is indistinguishable, at the write layer, from "a + * live-lane row already wrote it before the sweep ran." + * + * The sweep's own `now` is chosen to land on OpenClaw's default + * `sweep.cron` (every 5th minute) so this exercises the real `cronMatches` + * due-check (`src/core/sinks/driver.js`, imported by the sweep driver), + * not a `force: true` bypass: this is the only automated coverage, of any + * tier, for the sweep driver's `cronMatches` wiring and the quiesce + * filter's composition with it before the human acceptance run + * (LLP 0173's "hermetic-smoke decision" section, LLP 0172 Section 9). + * + * @ref LLP 0172#45-the-quiesce-window [tests]: a file inside the default + * quiesce window is skipped, one backdated past it is captured + * @ref LLP 0172#lane-b-sweep [tests]: the sweep driver fires the due, + * sweep-bearing provider through the real `cronMatches` due-check + * @ref LLP 0161#backfill-provider [tests]: native message identity makes a + * sweep-then-rerun (standing in for Lane A already having written the same + * part_id) net zero new rows + * + * @param {{ harness: any, expect: any }} args + */ +export async function run({ harness, expect }) { + const obs = installObservability() + if (!obs.tracer.provider) { + throw new Error( + 'backfill_openclaw_fixture: tracer provider not installed - expected HYP_DEV_TELEMETRY=1' + ) + } + + const cacheRoot = path.join(harness.stateDir, 'cache') + const registry = createCommandRegistry() + registerCoreCommands(registry) + const kernel = createKernelRuntime({ commandRegistry: registry, cacheRoot }) + + const pluginsRoot = path.resolve(import.meta.dirname, '..', '..', 'plugins-workspace') + const pluginDirs = [ + path.join(pluginsRoot, 'ai-gateway'), + path.join(pluginsRoot, 'openclaw'), + ] + + // The OpenClaw provider captures its `agents/` root from `ctx.env.HOME` + // (-> `/.openclaw/agents`, `session_file.js`'s + // `defaultOpenclawAgentsDir`) at activation, so stage both session + // fixtures and point HOME at the fake home BEFORE activating plugins. + const fakeHome = path.join(harness.tmpDir, 'home') + const agentsDir = path.join(fakeHome, '.openclaw', 'agents') + const agentId = 'main' + + const freshSessionId = `oc-fresh-${harness.devRunId}` + const oldSessionId = `oc-old-${harness.devRunId}` + + // Inside the quiesce window: a freshly-written file, left untouched, sits + // well inside the default 180000ms window for the whole duration of this + // smoke. + await writeOpenclawSession({ agentsDir, agentId, sessionId: freshSessionId }) + // Outside the quiesce window: back-dated 4 minutes, mirroring + // `test/plugins/openclaw-backfill.test.js`'s own default-quiesce-window + // precedent (`ageFile`, 4 * 60 * 1000 against the real 180000ms default). + const oldFilePath = await writeOpenclawSession({ agentsDir, agentId, sessionId: oldSessionId }) + await ageFile(oldFilePath, 4 * 60 * 1000) + + const previousHome = process.env.HOME + process.env.HOME = fakeHome + + try { + await runRoot( + 'kernel.boot', + { + [Attr.COMPONENT]: 'kernel', + [Attr.OPERATION]: 'boot', + [Attr.SMOKE_NAME]: harness.smokeName, + [Attr.SMOKE_STEP]: 'sweep_activate', + [Attr.DEV_RUN_ID]: harness.devRunId, + status: 'ok', + }, + async () => { + const { loaded } = await loadManifests(pluginDirs) + if (loaded.length !== pluginDirs.length) { + throw new Error(`backfill_openclaw_fixture: expected ${pluginDirs.length} manifests, got ${loaded.length}`) + } + const resolution = await resolveDependencies(loaded.map((l) => l.manifest)) + if (resolution.unsatisfied.length > 0) { + throw new Error( + `backfill_openclaw_fixture: unsatisfied requirements: ${ + resolution.unsatisfied.map((u) => `${u.plugin}:${u.errorKind}`).join(', ') + }` + ) + } + const byName = new Map(loaded.map((l) => [l.manifest.name, l])) + const entries = resolution.order + .map((name) => byName.get(name)) + .filter((l) => l !== undefined) + .map((l) => ({ manifest: l.manifest, rootDir: l.rootDir, config: {} })) + return activatePlugins({ + plugins: entries, + stateRoot: harness.stateDir, + runId: harness.devRunId, + runtime: kernel, + tmpRoot: path.join(harness.tmpDir, 'plugin-temp'), + }) + } + ) + + const env = { ...process.env, HYP_HOME: harness.hypHome } + + // The sweep driver itself, wired exactly the way the daemon wires it + // (`src/core/daemon/backfill_sweep.js`'s own doc), reusing this boot's + // `kernel.backfills` / `kernel.backfillMaterializers` / `kernel.storage`. + /** @type {Array>} */ + const pendingRuns = [] + const sweep = createBackfillSweepDriver({ + backfills: kernel.backfills, + backfillMaterializers: kernel.backfillMaterializers, + storage: kernel.storage, + query: kernel.query, + env, + config: { version: 2 }, + // Test seam (`src/core/daemon/types.d.ts`'s `BackfillSweepRunner`): + // `tick()` fires this fire-and-forget internally and resolves once + // runs are STARTED, not finished, so the smoke needs its own handle + // on the underlying promise to await completion before it queries or + // reruns. Still the real `runBackfillProvider`, just with its + // promise captured on the way out. + runBackfill: (/** @type {any} */ args) => { + const p = runBackfillProvider(args) + pendingRuns.push(p) + return p + }, + }) + + /** + * Run one sweep tick and await every run it fired. + * + * @param {Date} now + */ + async function tickAndAwait(now) { + pendingRuns.length = 0 + const report = await sweep.tick({ now }) + const results = await Promise.all(pendingRuns) + return { report, result: results[0] } + } + + // A UTC-minute-0 instant is due against OpenClaw's default `sweep.cron` + // (every 5th minute): real `cronMatches`, not a `force: true` bypass. + const tick1Now = new Date(Date.UTC(2026, 0, 1, 0, 0, 0)) + // A later due instant, still on the 5-minute grid, so the second tick + // gets its own `devRunId` and the ai-gateway materializer's + // `createBackfillDedupe` (memoized per `devRunId`) is forced to + // re-scan committed partitions rather than reuse tick 1's in-memory + // seen set. + const tick2Now = new Date(tick1Now.getTime() + 5 * 60 * 1000) + + // ----- 1. First sweep tick: quiesce skip + quiesce capture ((a)/(b)) ----- + const tick1 = await tickAndAwait(tick1Now) + expect.that( + 'tick 1: the openclaw provider fired', + tick1.report.fired, + (v) => Array.isArray(v) && v.includes('openclaw'), + ) + expect.that( + 'tick 1: exactly one session file scanned (only the one outside the quiesce window)', + tick1.result, + (v) => v !== undefined && v.ok === true && v.scanned === 1, + ) + expect.that( + 'tick 1: both rows of the outside-window session were written', + tick1.result, + (v) => v !== undefined && v.rowsWritten === 2, + ) + + /** @param {string} sessionId */ + const sqlFor = (sessionId) => ` + select role, content_text, message_id, part_id, provider, conversation_source, client_name + from ai_gateway_messages + where session_id = '${sessionId}' + order by message_index, part_index + `.trim().replace(/\s+/g, ' ') + + const freshRowsAfterTick1 = await queryRows({ + dispatch, sql: sqlFor(freshSessionId), kernel, registry, env, expect, label: 'fresh session after tick 1', + }) + expect.that( + '(a) a file with mtime inside the quiesce window is skipped by the sweep run', + freshRowsAfterTick1, + (v) => Array.isArray(v) && v.length === 0, + ) + + const oldRowsAfterTick1 = await queryRows({ + dispatch, sql: sqlFor(oldSessionId), kernel, registry, env, expect, label: 'old session after tick 1', + }) + expect.that( + '(b) a file with mtime outside the quiesce window is captured by the sweep run', + oldRowsAfterTick1, + (v) => Array.isArray(v) && v.length === 2, + ) + expect.that( + '(b) every captured row carries native identity and the right client/source', + oldRowsAfterTick1, + (v) => Array.isArray(v) && v.every( + (/** @type {any} */ r) => r.conversation_source === 'openclaw' && r.client_name === 'openclaw' && + r.provider === 'anthropic' && typeof r.message_id === 'string' && r.message_id.length > 0, + ), + ) + + // ----- 2. Second sweep tick: cross-write dedupe (c) ----- + const tick2 = await tickAndAwait(tick2Now) + expect.that( + 'tick 2: the openclaw provider fired again', + tick2.report.fired, + (v) => Array.isArray(v) && v.includes('openclaw'), + ) + expect.that( + '(c) rerunning the sweep after the part_id was already written nets zero new rows', + tick2.result, + (v) => v !== undefined && v.ok === true && v.rowsWritten === 0, + ) + + const oldRowsAfterTick2 = await queryRows({ + dispatch, sql: sqlFor(oldSessionId), kernel, registry, env, expect, label: 'old session after tick 2', + }) + expect.that( + '(c) the rerun did not duplicate rows (still exactly two)', + oldRowsAfterTick2, + (v) => Array.isArray(v) && v.length === 2, + ) + expect.that( + '(c) the rerun\'s row set is byte-identical to tick 1\'s (same part_ids, no drift)', + oldRowsAfterTick2, + (v) => Array.isArray(v) && + JSON.stringify(v.map((/** @type {any} */ r) => r.part_id).sort()) === + JSON.stringify(oldRowsAfterTick1.map((/** @type {any} */ r) => r.part_id).sort()), + ) + + const freshRowsAfterTick2 = await queryRows({ + dispatch, sql: sqlFor(freshSessionId), kernel, registry, env, expect, label: 'fresh session after tick 2', + }) + expect.that( + '(c) the still-quiesced session remains untouched by the rerun', + freshRowsAfterTick2, + (v) => Array.isArray(v) && v.length === 0, + ) + + // ----- 3. Internal telemetry: the sweep driver's own log lines, distinct + // from `hyp backfill`'s CLI-path logs, prove this ran through + // the daemon-facing driver (T9's cronMatches wiring), not just + // the provider underneath it. ----- + await obs.shutdown() + const logs = await expect.logs() + + const dueLogs = logs.filter( + (/** @type {any} */ l) => l.body === 'backfill.sweep_due' && l.attributes?.provider === 'openclaw', + ) + expect.that( + 'logs: backfill.sweep_due fired once per due tick (twice total)', + dueLogs, + (v) => Array.isArray(v) && v.length === 2, + ) + + const finishedLogs = logs.filter( + (/** @type {any} */ l) => l.body === 'backfill.sweep_finished' && l.attributes?.provider === 'openclaw', + ) + expect.that( + 'logs: backfill.sweep_finished (tick 1) reports rows_written=2', + finishedLogs.find((/** @type {any} */ l) => l.attributes?.rows_written === 2), + (v) => v !== undefined, + ) + expect.that( + 'logs: backfill.sweep_finished (tick 2) reports rows_written=0', + finishedLogs.find((/** @type {any} */ l) => l.attributes?.rows_written === 0), + (v) => v !== undefined, + ) + + const scanCompleteLogs = logs.filter( + (/** @type {any} */ l) => l.body === 'openclaw.backfill.scan_complete', + ) + expect.that( + 'logs: openclaw.backfill.scan_complete (tick 1) saw one file, past the quiesce filter', + scanCompleteLogs[0], + (v) => v !== undefined && v.attributes?.files_seen === 1 && v.attributes?.sessions_projected === 1, + ) + } finally { + if (previousHome === undefined) delete process.env.HOME + else process.env.HOME = previousHome + } +} + +/** + * Write one minimal OpenClaw v3 session JSONL under + * `//sessions/.jsonl`: a `type: "session"` + * header line and one user/assistant turn in the nested-`message`-envelope + * shape PR #552's reader (`session_file.js`'s `parseOpenclawSessionMessage` + * / `openclawMessageEnvelope`) actually projects - `role`/`content` and, + * on the assistant turn, `model`/`provider`/`api`/`stopReason`/`usage` + * nested under the record's own `message` object, never flat on the + * record line (a flat fixture would test the reader's now-fixed #543 bug, + * not its fix). No `cwd` on the header: an absent `cwd` reads as "not + * usable" (`openclawSessionCwd`) and the session is simply not + * usage-policy gated, which keeps this fixture independent of the host's + * real filesystem beyond the temp tree it writes. + * + * @param {{ agentsDir: string, agentId: string, sessionId: string }} args + * @returns {Promise} + */ +async function writeOpenclawSession(args) { + const { agentsDir, agentId, sessionId } = args + const dir = path.join(agentsDir, agentId, 'sessions') + await fs.mkdir(dir, { recursive: true }) + const filePath = path.join(dir, `${sessionId}.jsonl`) + const startedAt = new Date().toISOString() + const lines = [ + JSON.stringify({ type: 'session', version: 3, id: sessionId, timestamp: startedAt }), + JSON.stringify(messageLine({ + id: `${sessionId}-user`, + timestamp: startedAt, + role: 'user', + content: [{ type: 'text', text: 'list the files' }], + })), + JSON.stringify(messageLine({ + id: `${sessionId}-asst`, + timestamp: startedAt, + parentId: `${sessionId}-user`, + role: 'assistant', + content: [{ type: 'text', text: 'here they are' }], + model: 'claude-sonnet-4-5', + provider: 'anthropic', + api: 'anthropic-messages', + stopReason: 'end_turn', + usage: { input: 11, output: 7, cacheRead: 3, cacheWrite: 2 }, + })), + ] + await fs.writeFile(filePath, lines.join('\n') + '\n', 'utf8') + return filePath +} + +/** + * One `type: "message"` line in the shape OpenClaw actually appends: `id`, + * `parentId`, and `timestamp` on the record line, and every message field + * nested under `message`. Mirrors `test/plugins/openclaw-backfill.test.js`'s + * own `messageLine` helper, verified there against a live install (record + * keys `['id', 'message', 'parentId', 'timestamp', 'type']`). + * + * @param {Record} fields + * @returns {Record} + */ +function messageLine(fields) { + const { id, timestamp, parentId, ...message } = fields + return { + type: 'message', + ...(id !== undefined ? { id } : {}), + ...(timestamp !== undefined ? { timestamp } : {}), + parentId: parentId ?? null, + message: { ...message, ...(timestamp !== undefined ? { timestamp } : {}) }, + } +} + +/** + * Back-date `filePath`'s mtime by `msAgo` milliseconds, so a quiesce-window + * scenario can control file recency without waiting on the wall clock. + * + * @param {string} filePath + * @param {number} msAgo + */ +async function ageFile(filePath, msAgo) { + const past = new Date(Date.now() - msAgo) + await fs.utimes(filePath, past, past) +} + +/** + * Run a `query sql ... --format json` dispatch and return the parsed rows, + * asserting a clean exit and parseable output. + * + * @param {{ dispatch: any, sql: string, kernel: any, registry: any, env: any, expect: any, label: string }} args + * @returns {Promise} + */ +async function queryRows(args) { + const { dispatch: doDispatch, sql, kernel, registry, env, expect, label } = args + const out = makeBuf() + const err = makeBuf() + const code = await doDispatch( + ['query', 'sql', sql, '--refresh', 'always', '--format', 'json'], + { stdout: out, stderr: err, kernel, registry, env } + ) + expect.that(`dispatch: query (${label}) exited 0`, code, (/** @type {number} */ v) => v === 0) + expect.that(`stderr: query (${label}) had no errors`, err.text(), (/** @type {string} */ v) => typeof v === 'string' && v.length === 0) + try { + return JSON.parse(out.text()) + } catch (e) { + expect.that( + `stdout: query (${label}) was valid JSON (${e instanceof Error ? e.message : String(e)})`, + false, + (/** @type {boolean} */ v) => v === true, + ) + return [] + } +} + +function makeBuf() { + /** @type {string[]} */ + const chunks = [] + return { + /** @param {unknown} chunk */ + write(chunk) { + chunks.push(typeof chunk === 'string' ? chunk : String(chunk)) + return true + }, + text() { + return chunks.join('') + }, + } +} diff --git a/llp/0172-openclaw-two-lane-capture.design.md b/llp/0172-openclaw-two-lane-capture.design.md index 32390a36..7f479207 100644 --- a/llp/0172-openclaw-two-lane-capture.design.md +++ b/llp/0172-openclaw-two-lane-capture.design.md @@ -402,16 +402,19 @@ already enforces. ### 4.3 Narrowing `runProvider`'s context type, not widening the daemon's -`runProvider()` and `resolveOwnersForRun()` (both in -`src/core/commands/backfill.js`) only ever read `ctx.backfills`, -`ctx.backfillMaterializers`, `ctx.env`, `ctx.storage`, and (via -`resolveOwnersForRun`) `ctx.config` for plugin-configured resolution. None -of `CommandRunContext`'s other fields (`stdout`, `commands`, `verbs`, -`skills`, `agents`, `sources`, `sinks`, `initPresets`, `capabilities`, -`plugins`, `cwd`) are touched anywhere in this call path. Rather than force -the daemon to assemble a full, mostly-unused `CommandRunContext` just to -call `runBackfillProvider`, this design narrows the type both functions -declare their `ctx` parameter as, to a new, smaller type: +`runProvider()`, `resolveOwnersForRun()`, and the materialize/write/flush +helpers they call (all in `src/core/commands/backfill.js`) only ever read +`ctx.backfills`, `ctx.backfillMaterializers`, `ctx.env`, `ctx.storage`, +`ctx.query` (`writeRows`/`flushDataset` resolve a dataset's registered +table path through it before a row can be committed or a partition +flushed), and (via `resolveOwnersForRun`) `ctx.config` for +plugin-configured resolution. None of `CommandRunContext`'s other fields +(`stdout`, `commands`, `verbs`, `skills`, `agents`, `sources`, `sinks`, +`initPresets`, `capabilities`, `plugins`, `cwd`) are touched anywhere in +this call path. Rather than force the daemon to assemble a full, +mostly-unused `CommandRunContext` just to call `runBackfillProvider`, this +design narrows the type both functions declare their `ctx` parameter as, +to a new, smaller type: ```ts // A structural subset of CommandRunContext; every existing @@ -422,26 +425,39 @@ interface BackfillRunnerContext { env: NodeJS.ProcessEnv config: HypAwareV2Config storage: QueryStorageService + query: QueryRegistry backfills: BackfillRegistry backfillMaterializers: BackfillMaterializerRegistry } ``` -`runBackfillProvider`, `runProvider`, and `resolveOwnersForRun`'s `ctx` -parameters change from `CommandRunContext` to `BackfillRunnerContext`. This -is a pure narrowing: `CommandRunContext` is structurally a superset, so no -existing caller's argument stops satisfying the (now smaller) parameter -type. The daemon can now build a `BackfillRunnerContext` object out of -fields `boot.runtime` already carries (`env`, `config`, `storage`, -`backfills`, `backfillMaterializers`, the latter two already referenced at -`src/core/runtime/activation.js` lines 96 and 152) without touching -`CommandRunContext` or constructing stub versions of fields it doesn't need. +`runBackfillProvider`, `runProvider`, `resolveOwnersForRun`, and the +materialize/write/flush helpers' `ctx` parameters change from +`CommandRunContext` to `BackfillRunnerContext`. This is a pure narrowing: +`CommandRunContext` is structurally a superset, so no existing caller's +argument stops satisfying the (now smaller) parameter type. The daemon can +now build a `BackfillRunnerContext` object out of fields `boot.runtime` +already carries (`env`, `config`, `storage`, `query`, `backfills`, +`backfillMaterializers`, all already referenced at +`src/core/runtime/activation.js`) without touching `CommandRunContext` or +constructing stub versions of fields it doesn't need. + +`query` was not part of this list until LLP 0173 T12's hermetic smoke (the +first caller to drive a real, non-dry-run write through the sweep +driver rather than a mocked `runBackfill` seam) found `writeRows` and +`flushDataset` crash on `ctx.query.getDataset` when the daemon-built +`BackfillRunnerContext` reached them: the field really is on this call +path, this design's original field enumeration just missed it because +T9's own tests never exercised a real write. `BackfillSweepDriverOptions` +(Section 4.4) and the daemon's `createBackfillSweepDriver(...)` call +(`src/core/daemon/runtime.js`) both require `query` now for the same +reason. ### 4.4 Wiring the tick New file: `src/core/daemon/backfill_sweep.js`, exporting `createBackfillSweepDriver({backfills, backfillMaterializers, env, config, -storage})` with one method, `tick({now})`: +storage, query})` with one method, `tick({now})`: ```js function tick({ now }) { @@ -449,7 +465,7 @@ function tick({ now }) { if (!provider.sweep) continue if (!cronMatches(provider.sweep.cron, now)) continue void runBackfillProvider({ - ctx: { env, config, storage, backfills, backfillMaterializers }, + ctx: { env, config, storage, query, backfills, backfillMaterializers }, provider: provider.name, dryRun: false, devRunId: `sweep-${provider.name}-${now.getTime()}`, diff --git a/src/core/commands/backfill.js b/src/core/commands/backfill.js index 9bb13769..e44b20fc 100644 --- a/src/core/commands/backfill.js +++ b/src/core/commands/backfill.js @@ -476,17 +476,7 @@ async function runProvider(args) { const rows = await materializeItem({ materializer, item: yielded, - // materializeItem/writeRows/flushDataset are outside T7's - // narrowing (they read ctx.query, which BackfillRunnerContext - // does not carry); today's only real callers (hyp backfill's - // CLI path, the onboarding finale) still hand runProvider a - // full CommandRunContext, so this is a type-level widening - // back to the shape those three helpers already require, not - // a behavior change. - // @ref LLP 0172#lane-b-sweep [constrained-by]: runProvider's - // own ctx param narrows to BackfillRunnerContext; its - // still-CommandRunContext-typed callees need this back-cast - ctx: /** @type {CommandRunContext} */ (ctx), + ctx, devRunId, provider: provider.name, log, @@ -501,7 +491,7 @@ async function runProvider(args) { dataset: yielded.dataset, provider: provider.name, devRunId, - ctx: /** @type {CommandRunContext} */ (ctx), + ctx, log, }) result.rows_written += written.rowsWritten @@ -516,7 +506,7 @@ async function runProvider(args) { dataset, provider: provider.name, devRunId, - ctx: /** @type {CommandRunContext} */ (ctx), + ctx, log, }) } @@ -562,7 +552,7 @@ async function runProvider(args) { * @param {{ * materializer: BackfillMaterializerContribution, * item: BackfillItem, - * ctx: CommandRunContext, + * ctx: BackfillRunnerContext, * devRunId: string, * provider: string, * log: PluginLogger, @@ -607,7 +597,7 @@ async function materializeItem(args) { * dataset: string, * provider: string, * devRunId: string, - * ctx: CommandRunContext, + * ctx: BackfillRunnerContext, * log: PluginLogger, * }} args * @returns {Promise<{ rowsWritten: number, status: 'ok' | 'failed', error?: string }>} @@ -662,7 +652,7 @@ async function writeRows(args) { * dataset: string, * provider: string, * devRunId: string, - * ctx: CommandRunContext, + * ctx: BackfillRunnerContext, * log: PluginLogger, * }} args */ diff --git a/src/core/commands/types.d.ts b/src/core/commands/types.d.ts index 9c77e244..454dcc9a 100644 --- a/src/core/commands/types.d.ts +++ b/src/core/commands/types.d.ts @@ -3,6 +3,7 @@ import type { BackfillMaterializerRegistry, BackfillRegistry, HypAwareV2Config, + QueryRegistry, QueryStorageService, } from '../../../hypaware-plugin-kernel-types.js' @@ -33,18 +34,23 @@ export interface PolicyHumanVocabulary { } // A structural subset of `CommandRunContext`: exactly the fields -// `runBackfillProvider`, `runProvider`, and `resolveOwnersForRun` +// `runBackfillProvider`, `runProvider`, `resolveOwnersForRun`, and the +// materialize/write/flush helpers they call // (`src/core/commands/backfill.js`) read off `ctx`. Every existing // `CommandRunContext` already satisfies it, so `hyp backfill`'s CLI path // and the onboarding finale's call keep typechecking unchanged; the // daemon sweep driver (LLP 0173 T9) can build one directly out of // `boot.runtime` fields without assembling a full, mostly-unused -// `CommandRunContext`. -// @ref LLP 0172#lane-b-sweep [implements]: the narrowed context type `runBackfillProvider`, `runProvider`, and `resolveOwnersForRun` declare, so the daemon sweep driver can build one without a full `CommandRunContext` +// `CommandRunContext`. `query` was missing from this list until LLP 0173 +// T12's smoke (the first caller to drive a real, non-mocked write through +// the sweep driver) found `writeRows`/`flushDataset` crash on +// `ctx.query.getDataset` when a sweep-built `ctx` reached them. +// @ref LLP 0172#lane-b-sweep [implements]: the narrowed context type `runBackfillProvider`, `runProvider`, `resolveOwnersForRun`, and the materialize/write/flush helpers declare, so the daemon sweep driver can build one without a full `CommandRunContext` export interface BackfillRunnerContext { env: NodeJS.ProcessEnv config: HypAwareV2Config storage: QueryStorageService + query: QueryRegistry backfills: BackfillRegistry backfillMaterializers: BackfillMaterializerRegistry } diff --git a/src/core/daemon/backfill_sweep.js b/src/core/daemon/backfill_sweep.js index 85ced390..0b4c728c 100644 --- a/src/core/daemon/backfill_sweep.js +++ b/src/core/daemon/backfill_sweep.js @@ -48,10 +48,11 @@ const SWEEP_OPERATION = 'backfill.sweep' * @returns {BackfillSweepDriver} */ export function createBackfillSweepDriver(opts) { - const { backfills, backfillMaterializers, env, config, storage } = opts + const { backfills, backfillMaterializers, env, config, storage, query } = opts if (!backfills) throw new Error('createBackfillSweepDriver: backfills required') if (!backfillMaterializers) throw new Error('createBackfillSweepDriver: backfillMaterializers required') if (!storage) throw new Error('createBackfillSweepDriver: storage required') + if (!query) throw new Error('createBackfillSweepDriver: query required') const runBackfill = opts.runBackfill ?? runBackfillProvider const log = getLogger('backfill-sweep') @@ -80,7 +81,7 @@ export function createBackfillSweepDriver(opts) { // Fire-and-forget, with both settlements handled: `void` here means "not // awaited", never "not observed". void runBackfill({ - ctx: { env, config: config ?? { version: 2 }, storage, backfills, backfillMaterializers }, + ctx: { env, config: config ?? { version: 2 }, storage, query, backfills, backfillMaterializers }, provider: provider.name, dryRun: false, devRunId, diff --git a/src/core/daemon/runtime.js b/src/core/daemon/runtime.js index 5bc3c565..7ba0e756 100644 --- a/src/core/daemon/runtime.js +++ b/src/core/daemon/runtime.js @@ -452,6 +452,7 @@ export async function runDaemon(opts = {}) { backfillMaterializers: boot.runtime.backfillMaterializers, env, storage: boot.runtime.storage, + query: boot.runtime.query, config: boot.config ?? undefined, }) diff --git a/src/core/daemon/types.d.ts b/src/core/daemon/types.d.ts index 9aef78cb..f354ce9e 100644 --- a/src/core/daemon/types.d.ts +++ b/src/core/daemon/types.d.ts @@ -588,6 +588,14 @@ export interface BackfillSweepDriverOptions { backfillMaterializers: BackfillMaterializerRegistry env: NodeJS.ProcessEnv storage: QueryStorageService + /** + * Dataset registry. `runBackfillProvider`'s write/flush path + * (`writeRows`/`flushDataset` in `src/core/commands/backfill.js`) resolves + * a dataset's registered table path through this before it can commit a + * row, so a fired sweep run needs it on `BackfillRunnerContext` exactly + * like `hyp backfill`'s CLI path already gets it from `CommandRunContext`. + */ + query: QueryRegistry /** The daemon's effective config; absent on a host with no readable document. */ config?: HypAwareV2Config /** Test seam: defaults to `runBackfillProvider`. */ diff --git a/test/core/daemon-backfill-sweep.test.js b/test/core/daemon-backfill-sweep.test.js index 5e704c02..b73ecc24 100644 --- a/test/core/daemon-backfill-sweep.test.js +++ b/test/core/daemon-backfill-sweep.test.js @@ -52,6 +52,7 @@ function driverFor(args) { backfillMaterializers: /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }), env: /** @type {any} */ ({ HYP_HOME: '/nonexistent-home' }), storage: /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }), + query: /** @type {any} */ ({ getDataset: () => undefined }), config: args.config, runBackfill: args.runBackfill, }) @@ -117,11 +118,13 @@ test('the fired run gets the narrowed runner context, built from the daemon runt const backfillMaterializers = /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }) const env = /** @type {any} */ ({ HYP_HOME: '/nonexistent-home' }) const storage = /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }) + const query = /** @type {any} */ ({ getDataset: () => undefined }) const driver = createBackfillSweepDriver({ backfills, backfillMaterializers, env, storage, + query, config: /** @type {any} */ (config), runBackfill: async (args) => { seen = args.ctx; return OK }, }) @@ -130,6 +133,7 @@ test('the fired run gets the narrowed runner context, built from the daemon runt assert.equal(seen.env, env) assert.equal(seen.storage, storage) + assert.equal(seen.query, query) assert.equal(seen.config, config) assert.equal(seen.backfills, backfills) assert.equal(seen.backfillMaterializers, backfillMaterializers) @@ -206,6 +210,7 @@ test('createBackfillSweepDriver refuses to build without the registries it fires backfillMaterializers: /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }), env: /** @type {any} */ ({}), storage: /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }), + query: /** @type {any} */ ({ getDataset: () => undefined }), } assert.throws( () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, backfills: undefined })), @@ -219,4 +224,8 @@ test('createBackfillSweepDriver refuses to build without the registries it fires () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, storage: undefined })), /storage required/ ) + assert.throws( + () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, query: undefined })), + /query required/ + ) })