T9: daemon backfill sweep driver on the existing tick loop

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<BackfillPlan | undefined>}
@@ -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

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<BackfillSweepTickReport>}
+   */
+  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 }
+}

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,

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<BackfillSweepTickReport>
+}