From 606d87de4daf46da577120428ad9b7219f1d03eb Mon Sep 17 00:00:00 2001 From: test Date: Sat, 1 Aug 2026 00:21:18 +0000 Subject: [PATCH] Daemon sweep driver: run sweep-bearing backfill providers on the tick loop New `src/core/daemon/backfill_sweep.js`. `createBackfillSweepDriver({backfills, backfillMaterializers, env, config, storage})`'s `tick({now})` walks `backfills.list()`, skips any contribution with no `sweep` field or a cron that is not due (`cronMatches`, the sink driver's own due-check), and fires `runBackfillProvider` per due contribution with a `sweep--` dev run id. Runs are fired unblocked: `tick()` resolves once each run has been started, never once one finishes, so a provider's transcript scan cannot stall the sink snapshots, the source-detail refresh, or `persist()` later in the same tick. Both settlements are handled, so a failing run is a logged `backfill.sweep_failed` record (component `openclaw`, operation `backfill.sweep`, `error_kind`) rather than an unhandled rejection that would take the daemon process down. A malformed `sweep.cron` is logged and treated as not due rather than thrown, so one provider's bad metadata cannot skip the rest of the list. `runtime.js`'s `runTick()` calls `await sweepDriver.tick({now})` directly after the existing sink-driver tick, riding the same `DEFAULT_TICK_INTERVAL_MS` 60-second loop: a `*/5 * * * *` schedule only needs a due-check once a minute, so this opens no second timer to start, drain, and account for at shutdown. Also repairs the typecheck this task's branch point already failed: T7's `sweep: { cron: opts.config?.backfill?.sweep_cron ?? ... }` does not compile, because the plugin's config slice is a `JsonObject` and every step below its root is a `JsonValue`. Read through a `resolveSweepCron` mirroring the `resolveQuiesceMs` helper already sitting beside it; behavior is unchanged. Externally blocked for real capture: until PR #552 (issue #543) merges, the LLP 0158 reader still reads OpenClaw v3 fields flat, so a sweep projects nothing from a real transcript. These tests passing is not evidence that it does. Tests: the due-check fires only sweep-bearing, cron-due contributions and builds the narrowed `BackfillRunnerContext` from the daemon's own runtime fields; a rejected run neither throws out of `tick()` nor lands as an unhandled rejection, and a never-settling run does not block the tick. A separate wiring test boots a real daemon with a fixture plugin whose contribution opts into a sweep and proves the tick actually runs it, which no unit test of the driver can show. Task-Id: T9 --- .../openclaw/src/backfill.js | 21 +- src/core/daemon/backfill_sweep.js | 164 +++++++++++++ src/core/daemon/runtime.js | 20 ++ src/core/daemon/types.d.ts | 52 ++++ .../core/daemon-backfill-sweep-wiring.test.js | 138 +++++++++++ test/core/daemon-backfill-sweep.test.js | 222 ++++++++++++++++++ 6 files changed, 616 insertions(+), 1 deletion(-) create mode 100644 src/core/daemon/backfill_sweep.js create mode 100644 test/core/daemon-backfill-sweep-wiring.test.js create mode 100644 test/core/daemon-backfill-sweep.test.js diff --git a/hypaware-core/plugins-workspace/openclaw/src/backfill.js b/hypaware-core/plugins-workspace/openclaw/src/backfill.js index f7c86c59..8940c12f 100644 --- a/hypaware-core/plugins-workspace/openclaw/src/backfill.js +++ b/hypaware-core/plugins-workspace/openclaw/src/backfill.js @@ -189,7 +189,7 @@ export function createOpenclawBackfillProvider(opts) { // @ref LLP 0172#lane-b-sweep [implements]: opt-in Lane B scheduling // metadata, tunable via `backfill.sweep_cron` (R7), defaulting to // every 5 minutes when the config key is absent. - sweep: { cron: opts.config?.backfill?.sweep_cron ?? DEFAULT_SWEEP_CRON }, + sweep: { cron: resolveSweepCron(config) }, /** * @param {BackfillPlanContext} _ctx * @returns {Promise} @@ -724,6 +724,25 @@ function resolveQuiesceMs(config) { return typeof quiesceMs === 'number' ? quiesceMs : DEFAULT_QUIESCE_MS } +/** + * The contribution's `sweep.cron`, resolved from the plugin's own validated + * `config` slice, or {@link DEFAULT_SWEEP_CRON} when `config.backfill.sweep_cron` + * is absent. Read through the same `isPlainObject` narrowing `resolveQuiesceMs` + * uses rather than an optional-property chain: the slice is a `JsonObject`, so + * every step below its root is a `JsonValue` to the checker and a bare + * `config?.backfill?.sweep_cron` does not typecheck. `config.js`'s + * `validateBackfillSection` already rejects a malformed cron string before it + * reaches here. + * + * @param {JsonObject | undefined} config + * @returns {string} + */ +function resolveSweepCron(config) { + const backfill = isPlainObject(config) && isPlainObject(config.backfill) ? config.backfill : undefined + const sweepCron = backfill?.sweep_cron + return typeof sweepCron === 'string' ? sweepCron : DEFAULT_SWEEP_CRON +} + /** * @param {string} dir * @param {'dir' | 'file'} kind diff --git a/src/core/daemon/backfill_sweep.js b/src/core/daemon/backfill_sweep.js new file mode 100644 index 00000000..85ced390 --- /dev/null +++ b/src/core/daemon/backfill_sweep.js @@ -0,0 +1,164 @@ +// @ts-check + +import { Attr, getLogger } from '../observability/index.js' +import { runBackfillProvider } from '../commands/backfill.js' +import { cronMatches } from '../sinks/driver.js' + +// The sweep's telemetry identity, fixed by LLP 0173's implementer note so a +// failing run is greppable by the same pair everywhere it is logged. OpenClaw +// owns the only contribution that opts into a sweep today; every record also +// carries `hyp_plugin` and `provider`, so a second opt-in stays attributable +// without changing what an operator greps for. +const SWEEP_COMPONENT = 'openclaw' +const SWEEP_OPERATION = 'backfill.sweep' + +/** + * @import { BackfillContribution } from '../../../hypaware-plugin-kernel-types.js' + * @import { + * BackfillSweepDriver, + * BackfillSweepDriverOptions, + * BackfillSweepTickOptions, + * BackfillSweepTickReport, + * } from '../../../src/core/daemon/types.js' + */ + +/** + * Build the daemon's backfill sweep driver: the periodic, in-process re-run of + * every registered backfill provider that opted into a schedule. + * + * The driver owns no timer. `tick({ now })` is called from the daemon's + * existing 60-second sink tick, evaluates each contribution's `sweep.cron` + * against `now` with the same `cronMatches` the sink driver uses, and fires a + * run for each due provider. A contribution with no `sweep` field is never + * ticked, which is why adding this driver is zero behavior change for Claude's + * and Codex's contributions. + * + * `tick()` resolves once every due provider's run has been *started*, not once + * any of them finishes. Runs are fired unblocked: `runProvider`'s scan, materialize, + * write and flush pass is unbounded in the size of a user's transcript tree, and + * the tick it rides also refreshes source details and persists `status.json`. + * Blocking on a sweep would stall those behind a provider's disk walk. The + * fired promise is still handled, so a failing run is a logged + * `backfill.sweep_failed` record rather than an unhandled rejection that takes + * the daemon process down. + * + * @ref LLP 0172#lane-b-sweep [implements]: the sweep rides the existing sink-tick cadence with `cronMatches` as its due-check, and fires each due provider without blocking the tick + * @ref LLP 0170#decision [implements]: scheduling an existing job (the backfill provider) on the daemon's existing cron-matched loop, not building a new scheduling primitive + * @param {BackfillSweepDriverOptions} opts + * @returns {BackfillSweepDriver} + */ +export function createBackfillSweepDriver(opts) { + const { backfills, backfillMaterializers, env, config, storage } = 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') + const runBackfill = opts.runBackfill ?? runBackfillProvider + const log = getLogger('backfill-sweep') + + /** + * @param {BackfillSweepTickOptions} [tickOpts] + * @returns {Promise} + */ + async function tick(tickOpts = {}) { + const now = tickOpts.now ?? new Date() + /** @type {string[]} */ + const fired = [] + for (const provider of backfills.list()) { + if (!provider.sweep) continue + if (!isDue(provider, now, tickOpts.force === true)) continue + const devRunId = `sweep-${provider.name}-${now.getTime()}` + fired.push(provider.name) + log.info('backfill.sweep_due', { + [Attr.COMPONENT]: SWEEP_COMPONENT, + [Attr.OPERATION]: SWEEP_OPERATION, + [Attr.PLUGIN]: provider.plugin, + [Attr.DEV_RUN_ID]: devRunId, + provider: provider.name, + hyp_sweep_schedule: provider.sweep.cron, + status: 'ok', + }) + // 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 }, + provider: provider.name, + dryRun: false, + devRunId, + }).then( + (result) => { logSettled(provider, devRunId, result) }, + (err) => { logFailed(provider, devRunId, err) } + ) + } + return { fired } + } + + /** + * Whether a contribution's schedule is due at `now`. A malformed cron + * expression throws out of `cronMatches`; here that is one provider's + * scheduling metadata being wrong, not a reason to skip every later + * provider in the list or to fail the daemon tick this runs inside, so it + * is logged and treated as not due. + * + * @param {BackfillContribution} provider + * @param {Date} now + * @param {boolean} force + * @returns {boolean} + */ + function isDue(provider, now, force) { + if (force) return true + try { + return cronMatches(provider.sweep?.cron ?? '', now) + } catch (err) { + log.warn('backfill.sweep_schedule_invalid', { + [Attr.COMPONENT]: SWEEP_COMPONENT, + [Attr.OPERATION]: SWEEP_OPERATION, + [Attr.ERROR_KIND]: 'invalid_cron', + [Attr.PLUGIN]: provider.plugin, + provider: provider.name, + hyp_sweep_schedule: provider.sweep?.cron, + status: 'failed', + }) + return false + } + } + + /** + * @param {BackfillContribution} provider + * @param {string} devRunId + * @param {{ ok: boolean, scanned: number, rowsWritten: number, skipped: number }} result + */ + function logSettled(provider, devRunId, result) { + log.info('backfill.sweep_finished', { + [Attr.COMPONENT]: SWEEP_COMPONENT, + [Attr.OPERATION]: SWEEP_OPERATION, + [Attr.PLUGIN]: provider.plugin, + [Attr.DEV_RUN_ID]: devRunId, + provider: provider.name, + status: result.ok ? 'ok' : 'failed', + ...(result.ok ? {} : { [Attr.ERROR_KIND]: 'provider_run_failed' }), + items_seen: result.scanned, + rows_written: result.rowsWritten, + rows_skipped: result.skipped, + }) + } + + /** + * @param {BackfillContribution} provider + * @param {string} devRunId + * @param {unknown} err + */ + function logFailed(provider, devRunId, err) { + log.error('backfill.sweep_failed', { + [Attr.COMPONENT]: SWEEP_COMPONENT, + [Attr.OPERATION]: SWEEP_OPERATION, + [Attr.ERROR_KIND]: 'sweep_run_rejected', + [Attr.PLUGIN]: provider.plugin, + [Attr.DEV_RUN_ID]: devRunId, + provider: provider.name, + status: 'failed', + error: err instanceof Error ? err.message : String(err), + }) + } + + return { tick } +} diff --git a/src/core/daemon/runtime.js b/src/core/daemon/runtime.js index c5f96ba8..5bc3c565 100644 --- a/src/core/daemon/runtime.js +++ b/src/core/daemon/runtime.js @@ -19,6 +19,7 @@ import { backfillHandler } from '../config/action_backfill.js' import { bootKernel, resolveLayeredConfigForDaemon } from '../runtime/boot.js' import { createSinkDriver } from '../sinks/driver.js' import { materializeSinks } from '../sinks/materialize.js' +import { createBackfillSweepDriver } from './backfill_sweep.js' import { clearPidFile, pidFilePath, @@ -442,6 +443,18 @@ export async function runDaemon(opts = {}) { config: boot.config ?? undefined, }) + // ----- Backfill sweep driver ----- + // Rides the sink tick below rather than owning a timer of its own: a + // contribution's coarsest useful schedule still only needs a due-check once + // a minute, which is exactly this loop's cadence. + const sweepDriver = createBackfillSweepDriver({ + backfills: boot.runtime.backfills, + backfillMaterializers: boot.runtime.backfillMaterializers, + env, + storage: boot.runtime.storage, + config: boot.config ?? undefined, + }) + status.sinks = collectSinkSnapshots({ runtime: boot.runtime, sinkSnapshots }) persist() // Derive the boot health event from the SAME aggregate written to @@ -599,6 +612,13 @@ export async function runDaemon(opts = {}) { }, async () => { const report = await driver.tick({ now, source: 'daemon' }) + // The scheduled backfill sweep (LLP 0170) rides this same tick. The + // await covers only the cron due-check and the fire: each due + // provider's run is started unblocked inside `tick`, so a slow + // transcript scan never stalls the sink snapshots, the source-detail + // refresh, or `persist()` below. + // @ref LLP 0172#lane-b-sweep [implements]: one sibling call on the existing 60-second loop, not a second timer + await sweepDriver.tick({ now }) for (const sinkReport of report.sinks) { const snap = sinkSnapshots.get(sinkReport.instance) ?? { instance: sinkReport.instance, diff --git a/src/core/daemon/types.d.ts b/src/core/daemon/types.d.ts index a5f6d65e..9aef78cb 100644 --- a/src/core/daemon/types.d.ts +++ b/src/core/daemon/types.d.ts @@ -1,7 +1,12 @@ import type { + BackfillMaterializerRegistry, + BackfillRegistry, CapabilityRegistry, + HypAwareV2Config, QueryRegistry, + QueryStorageService, } from '../../../hypaware-plugin-kernel-types.d.ts' +import type { BackfillRunnerContext } from '../commands/types.d.ts' import type { ActionReconciler, ConfigControlStatus, ConfigLayerDrop, V1Diagnostic } from '../config/types.d.ts' import type { ExtendedSinkRegistry, @@ -561,3 +566,50 @@ export interface PidFileEntry { /** `foreground` (Phase 3) or `detached` (Phase 4 installers). */ mode: string } + +/** + * The runner the sweep driver fires per due contribution: exactly + * `runBackfillProvider`'s (`src/core/commands/backfill.js`) shape, narrowed to + * the arguments a sweep passes. Declared as a type rather than taken from the + * import so the driver can accept an injected fake in a unit test without the + * test having to stand up a real cache, storage service, and materializer set. + */ +export interface BackfillSweepRunner { + (args: { + ctx: BackfillRunnerContext + provider: string + dryRun: boolean + devRunId?: string + }): Promise<{ ok: boolean, scanned: number, rowsWritten: number, skipped: number }> +} + +export interface BackfillSweepDriverOptions { + backfills: BackfillRegistry + backfillMaterializers: BackfillMaterializerRegistry + env: NodeJS.ProcessEnv + storage: QueryStorageService + /** The daemon's effective config; absent on a host with no readable document. */ + config?: HypAwareV2Config + /** Test seam: defaults to `runBackfillProvider`. */ + runBackfill?: BackfillSweepRunner +} + +export interface BackfillSweepTickOptions { + /** Tick instant the cron due-check evaluates against. Defaults to `new Date()`. */ + now?: Date + /** Ignore the cron due-check and fire every sweep-bearing provider (test use). */ + force?: boolean +} + +/** + * What one `tick()` decided, for the caller's telemetry and for tests. Runs are + * fired unblocked, so `fired` names the providers a run was *started* for, never + * the ones that finished. + */ +export interface BackfillSweepTickReport { + fired: string[] +} + +export interface BackfillSweepDriver { + tick(opts?: BackfillSweepTickOptions): Promise +} diff --git a/test/core/daemon-backfill-sweep-wiring.test.js b/test/core/daemon-backfill-sweep-wiring.test.js new file mode 100644 index 00000000..f56b2c60 --- /dev/null +++ b/test/core/daemon-backfill-sweep-wiring.test.js @@ -0,0 +1,138 @@ +// @ts-check + +import test from 'node:test' +import assert from 'node:assert/strict' +import fs from 'node:fs/promises' +import os from 'node:os' +import path from 'node:path' + +import { runDaemon } from '../../src/core/daemon/runtime.js' +import { defaultConfigPath } from '../../src/core/config/schema.js' +import { writeLock } from '../../src/core/plugin_install/lock.js' + +// The unit tests next door prove the driver's due-check and containment. This +// one proves the part no unit test can: that the daemon's own tick actually +// calls it. The sweep rides the sink tick rather than owning a timer, so a +// wiring regression here is silent - the driver still passes every test it has +// and simply never runs. +// @ref LLP 0172#lane-b-sweep [tests]: the sweep is called from `runTick()`, on the existing tick interval, with no timer of its own + +const PLUGIN = '@third-party/sweeping-fixture' + +/** + * Stage a plugin whose backfill contribution opts into a once-a-minute sweep + * and records each run by appending to a file. It yields nothing, so the run + * exercises the runner's full lifecycle without needing a materializer or a + * writable dataset. + * + * @param {string} hypHome + * @param {string} marker + * @returns {Promise} + */ +async function stageSweepingPlugin(hypHome, marker) { + const installDir = path.join(hypHome, 'hypaware', 'plugins', PLUGIN) + await fs.mkdir(installDir, { recursive: true }) + await fs.writeFile(path.join(installDir, 'hypaware.plugin.json'), JSON.stringify({ + schema_version: 1, + name: PLUGIN, + version: '0.1.0', + hypaware_api: '^1.0.0', + runtime: 'node', + entrypoint: './index.js', + })) + await fs.writeFile( + path.join(installDir, 'index.js'), + ` +import fs from 'node:fs' + +export async function activate(ctx) { + ctx.backfills.register({ + name: 'sweeping-fixture', + plugin: '${PLUGIN}', + datasets: ['ai_gateway_messages'], + sweep: { cron: '* * * * *' }, + async *run() { + fs.appendFileSync(${JSON.stringify(marker)}, 'swept\\n') + }, + }) + ctx.sources.register({ + name: 'sweeping-fixture', + plugin: '${PLUGIN}', + async start() { + return { + async status() { return { state: 'ready', details: {} } }, + async stop() {}, + } + }, + }) +} +` + ) + return installDir +} + +/** + * @param {string} hypHome + * @param {string} installDir + */ +async function writeInstall(hypHome, installDir) { + await writeLock(path.join(hypHome, 'hypaware'), { + schema_version: 1, + plugins: { + [PLUGIN]: { + name: PLUGIN, + version: '0.1.0', + source: { kind: 'local-dir', raw: installDir, path: installDir }, + install_dir: installDir, + content_hash: 'a'.repeat(64), + manifest_hash: 'b'.repeat(64), + installed_at: '2026-07-30T00:00:00.000Z', + }, + }, + }) + const configPath = defaultConfigPath(hypHome) + await fs.mkdir(path.dirname(configPath), { recursive: true }) + await fs.writeFile(configPath, JSON.stringify({ + version: 2, + plugins: [{ name: PLUGIN, config: {} }], + })) + return configPath +} + +test('the daemon tick runs a sweep-bearing backfill contribution', async () => { + const hypHome = await fs.mkdtemp(path.join(os.tmpdir(), 'hypaware-sweep-tick-')) + const marker = path.join(hypHome, 'sweeps.log') + let handle + try { + const configPath = await writeInstall(hypHome, await stageSweepingPlugin(hypHome, marker)) + handle = await runDaemon({ + hypHome, + configPath, + env: { ...process.env, HYP_HOME: hypHome }, + runId: 'sweep-tick', + // Fast enough that a tick lands inside the wait below, slow enough that + // the poll sees the first sweep rather than a dozen piled-up ones. + tickIntervalMs: 200, + installSignalHandlers: false, + }) + + const deadline = Date.now() + 20_000 + let swept = false + while (Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 50)) + swept = await fs.readFile(marker, 'utf8').then((t) => t.includes('swept'), () => false) + if (swept) break + } + assert.ok(swept, 'the daemon tick never ran the sweep-bearing provider') + } finally { + if (handle) { + await handle.stop() + await handle.done + } + // Sweeps are fired unblocked, so shutdown does not drain them: a run + // started by the last tick can still be touching the state tree here. + // Retry the teardown rather than racing it. + await new Promise((resolve) => setTimeout(resolve, 100)) + await fs.rm(hypHome, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) + } +}) diff --git a/test/core/daemon-backfill-sweep.test.js b/test/core/daemon-backfill-sweep.test.js new file mode 100644 index 00000000..5e704c02 --- /dev/null +++ b/test/core/daemon-backfill-sweep.test.js @@ -0,0 +1,222 @@ +// @ts-check + +import test from 'node:test' +import assert from 'node:assert/strict' + +import { createBackfillSweepDriver } from '../../src/core/daemon/backfill_sweep.js' + +// Lane B's scheduling seam. The sweep is the only reason a transcript that +// never crossed the live gateway lands at all, and it runs inside the daemon's +// own tick loop, so the two things worth pinning are *which* contributions it +// fires (opt-in only, cron-due only) and that a failing run stays contained: +// an unhandled rejection here would take the daemon process down with it. +// @ref LLP 0172#lane-b-sweep [tests]: the due-check is `sweep`-gated and cron-gated, and the fired run never blocks or breaks the tick it rides +// @ref LLP 0171#requirements [tests]: R7's periodic sweep fires on the contribution's own configured schedule + +/** + * @param {Record} [overrides] + * @returns {any} + */ +function contribution(overrides = {}) { + return { + name: 'openclaw', + plugin: '@hypaware/openclaw', + datasets: ['ai_gateway_messages'], + async *run() {}, + ...overrides, + } +} + +/** + * A `BackfillRegistry` over a fixed contribution list: `list()` is the only + * method the sweep driver calls, and the runner is faked, so nothing here + * needs a real kernel. + * + * @param {any[]} contributions + * @returns {any} + */ +function registry(contributions) { + return { + register() {}, + get: (name) => contributions.find((c) => c.name === name), + list: () => contributions.slice(), + } +} + +/** + * @param {{ contributions: any[], runBackfill: any, config?: any }} args + */ +function driverFor(args) { + return createBackfillSweepDriver({ + backfills: registry(args.contributions), + backfillMaterializers: /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }), + env: /** @type {any} */ ({ HYP_HOME: '/nonexistent-home' }), + storage: /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }), + config: args.config, + runBackfill: args.runBackfill, + }) +} + +/** @param {string} iso */ +function at(iso) { + return new Date(iso) +} + +const OK = { ok: true, scanned: 0, rowsWritten: 0, skipped: 0 } + +test('tick fires only the sweep-bearing contributions that are cron-due', async () => { + /** @type {any[]} */ + const calls = [] + const driver = driverFor({ + contributions: [ + // Opted in, due every five minutes. + contribution({ name: 'openclaw', sweep: { cron: '*/5 * * * *' } }), + // Opted in, but only on the hour: not due at :05. + contribution({ name: 'hourly', plugin: '@hypaware/hourly', sweep: { cron: '0 * * * *' } }), + // Never opted in: the absent-by-default case every provider is in today. + contribution({ name: 'claude', plugin: '@hypaware/claude' }), + ], + runBackfill: async (args) => { calls.push(args); return OK }, + }) + + const report = await driver.tick({ now: at('2026-08-01T10:05:00.000Z') }) + + assert.deepEqual(report.fired, ['openclaw']) + assert.equal(calls.length, 1) + assert.equal(calls[0].provider, 'openclaw') + assert.equal(calls[0].dryRun, false) + assert.equal(calls[0].devRunId, `sweep-openclaw-${at('2026-08-01T10:05:00.000Z').getTime()}`) +}) + +test('tick fires nothing when no contribution is due, and both when both are', async () => { + /** @type {string[]} */ + const fired = [] + const driver = driverFor({ + contributions: [ + contribution({ name: 'openclaw', sweep: { cron: '*/5 * * * *' } }), + contribution({ name: 'hourly', plugin: '@hypaware/hourly', sweep: { cron: '0 * * * *' } }), + ], + runBackfill: async (args) => { fired.push(args.provider); return OK }, + }) + + // :07 is neither a five-minute boundary nor the top of the hour. + assert.deepEqual((await driver.tick({ now: at('2026-08-01T10:07:00.000Z') })).fired, []) + assert.deepEqual(fired, []) + + // :00 satisfies both schedules. + assert.deepEqual((await driver.tick({ now: at('2026-08-01T11:00:00.000Z') })).fired, ['openclaw', 'hourly']) + assert.deepEqual(fired, ['openclaw', 'hourly']) +}) + +test('the fired run gets the narrowed runner context, built from the daemon runtime fields', async () => { + /** @type {any} */ + let seen = null + const contributions = [contribution({ sweep: { cron: '* * * * *' } })] + const config = { version: 2, plugins: [{ name: '@hypaware/openclaw', config: {} }] } + const backfills = registry(contributions) + const backfillMaterializers = /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }) + const env = /** @type {any} */ ({ HYP_HOME: '/nonexistent-home' }) + const storage = /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }) + const driver = createBackfillSweepDriver({ + backfills, + backfillMaterializers, + env, + storage, + config: /** @type {any} */ (config), + runBackfill: async (args) => { seen = args.ctx; return OK }, + }) + + await driver.tick({ now: at('2026-08-01T10:00:00.000Z') }) + + assert.equal(seen.env, env) + assert.equal(seen.storage, storage) + assert.equal(seen.config, config) + assert.equal(seen.backfills, backfills) + assert.equal(seen.backfillMaterializers, backfillMaterializers) +}) + +test('a rejected sweep run neither throws out of tick nor becomes an unhandled rejection', async () => { + /** @type {unknown[]} */ + const unhandled = [] + /** @param {unknown} reason */ + const onUnhandled = (reason) => { unhandled.push(reason) } + process.on('unhandledRejection', onUnhandled) + try { + const driver = driverFor({ + contributions: [ + contribution({ name: 'openclaw', sweep: { cron: '* * * * *' } }), + contribution({ name: 'codex', plugin: '@hypaware/codex', sweep: { cron: '* * * * *' } }), + ], + runBackfill: async (args) => { + if (args.provider === 'openclaw') throw new Error('cache is unwritable') + return OK + }, + }) + + // The rejection is raised by the fired run, not by the due-check, so the + // tick itself resolves normally and the *later* provider still fires: one + // broken run does not cancel the rest of the sweep. + const report = await driver.tick({ now: at('2026-08-01T10:00:00.000Z') }) + assert.deepEqual(report.fired, ['openclaw', 'codex']) + + // Two macrotask turns: enough for the rejected promise's handler to run, + // and for Node to have reported it had there been none. + await new Promise((resolve) => setTimeout(resolve, 0)) + await new Promise((resolve) => setTimeout(resolve, 0)) + assert.deepEqual(unhandled, [], 'the fired run left an unhandled rejection') + } finally { + process.off('unhandledRejection', onUnhandled) + } +}) + +test('tick does not block on a run that never settles', async () => { + let settle = () => {} + const pending = new Promise((resolve) => { settle = () => resolve(OK) }) + const driver = driverFor({ + contributions: [contribution({ sweep: { cron: '* * * * *' } })], + runBackfill: () => /** @type {any} */ (pending), + }) + + // If `tick` awaited the run, this would hang until the test timed out. + const report = await driver.tick({ now: at('2026-08-01T10:00:00.000Z') }) + assert.deepEqual(report.fired, ['openclaw']) + settle() + await pending +}) + +test('a malformed sweep cron is skipped, not thrown, and later providers still fire', async () => { + /** @type {string[]} */ + const fired = [] + const driver = driverFor({ + contributions: [ + contribution({ name: 'broken', plugin: '@hypaware/broken', sweep: { cron: 'not a cron' } }), + contribution({ name: 'openclaw', sweep: { cron: '* * * * *' } }), + ], + runBackfill: async (args) => { fired.push(args.provider); return OK }, + }) + + const report = await driver.tick({ now: at('2026-08-01T10:00:00.000Z') }) + assert.deepEqual(report.fired, ['openclaw']) + assert.deepEqual(fired, ['openclaw']) +}) + +test('createBackfillSweepDriver refuses to build without the registries it fires through', () => { + const ok = { + backfills: registry([]), + backfillMaterializers: /** @type {any} */ ({ register() {}, get: () => undefined, list: () => [] }), + env: /** @type {any} */ ({}), + storage: /** @type {any} */ ({ cacheRoot: '/nonexistent-cache' }), + } + assert.throws( + () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, backfills: undefined })), + /backfills required/ + ) + assert.throws( + () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, backfillMaterializers: undefined })), + /backfillMaterializers required/ + ) + assert.throws( + () => createBackfillSweepDriver(/** @type {any} */ ({ ...ok, storage: undefined })), + /storage required/ + ) +})