feat(jobs): expose job heartbeat and scan backlog

新增 JobActivity 持有作业心跳与扫描积压，JobRunner 每轮 tick 上报（失败只计数、不刷新心跳），PipelineMetrics 注册 msgx.pipeline.job.heartbeat_age_seconds / ticks.total / failures.total / last_sweep_selected，新增 JobRunnerHealthIndicator（未启动 UP 并标 started=false；已启动但心跳缺失或超过 3×扫描周期 DOWN）。对应 reference 作业健康段与 [G-JOB-HEARTBEAT]。

验证：./gradlew test 145 tests / 0 fail / 0 skipped（新增 JobRunnerTest 2、JobRunnerHealthIndicatorTest 4、PipelineMetrics 指标注册 1）。

src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt

@@ -0,0 +1,57 @@
+package com.gzzn.omms.msgexchange.infra.health
+
+import com.gzzn.omms.msgexchange.infra.metrics.JobActivity
+import io.micronaut.core.async.publisher.Publishers
+import io.micronaut.health.HealthStatus
+import io.micronaut.management.health.indicator.HealthIndicator
+import io.micronaut.management.health.indicator.HealthResult
+import jakarta.inject.Singleton
+import org.reactivestreams.Publisher
+import java.time.Clock
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * 维护作业的存活判定：作业线程是回填的唯一驱动，停摆必须可见，而不是只留一条日志。
+ *
+ * - 从未启动（例如 `msgx.pipeline.autostart=false`）→ `UP`，详情标 `started=false`，避免误报；
+ * - 已启动但心跳缺失或超过 `3 × 扫描周期`（周期是 `JobRunner` 的代码常量，见 reference）→ `DOWN`；
+ * - 其余 → `UP`，并给出心跳年龄、上一轮耗时、tick/失败数与扫描积压。
+ */
+@Singleton
+class JobRunnerHealthIndicator(
+    private val activity: JobActivity,
+    private val clock: Clock,
+) : HealthIndicator {
+
+    override fun getResult(): Publisher<HealthResult> =
+        Publishers.just(jobRunnerHealth(activity.snapshot(), clock.instant()))
+}
+
+/** 纯判定，便于单测；`now` 由调用方传入注入时钟的时刻。 */
+internal fun jobRunnerHealth(snapshot: JobActivity.Snapshot, now: Instant): HealthResult {
+    val ageSeconds = snapshot.lastTickAt?.let { Duration.between(it, now).seconds }
+    val stale = snapshot.started && (ageSeconds == null || ageSeconds > STALE_AFTER_SECONDS)
+    val message = when {
+        !snapshot.started -> "job runner not started (started=false)"
+        stale -> "job heartbeat stale: age=${ageSeconds ?: "never"}s > ${STALE_AFTER_SECONDS}s"
+        else -> "job heartbeat ok: age=${ageSeconds}s ticks=${snapshot.ticks}"
+    }
+    return HealthResult.builder("job-runner")
+        .status(if (stale) HealthStatus.DOWN else HealthStatus.UP)
+        .details(
+            mapOf(
+                "message" to message,
+                "started" to snapshot.started,
+                "lastTickAgeSeconds" to (ageSeconds ?: -1L),
+                "lastTickDurationMs" to snapshot.lastTickMillis,
+                "ticks" to snapshot.ticks,
+                "failures" to snapshot.failures,
+                "lastSweepSelected" to snapshot.lastSweepSelected,
+            ),
+        )
+        .build()
+}
+
+/** 3 × 30 秒扫描周期：连续两轮都没跑完才算停摆。 */
+internal const val STALE_AFTER_SECONDS = 90L

src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt

@@ -0,0 +1,65 @@
+package com.gzzn.omms.msgexchange.infra.metrics
+
+import jakarta.inject.Singleton
+import java.time.Instant
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * 维护作业的进程内活动事实（作业心跳、回填扫描积压），供 `/metrics` 与 `/health` 取用。
+ *
+ * 放在 `infra` 是为了让 `jobs`（写）与指标/健康（读）都零依赖：作业线程只更新这个持有者，
+ * 指标名与暴露方式由 [PipelineMetrics] 决定，存活判定由 `JobRunnerHealthIndicator` 决定。
+ * 计数在重启后归零；跨重启的累计值由指标后端聚合，这里不做持久化。
+ */
+@Singleton
+class JobActivity {
+    @Volatile
+    private var started = false
+
+    @Volatile
+    private var lastTickAt: Instant? = null
+
+    @Volatile
+    private var lastTickMillis = 0L
+
+    /** 上一轮回填扫描选中的待办条数（扫描积压的即时值）。 */
+    @Volatile
+    var lastSweepSelected = 0
+        private set
+
+    private val ticks = AtomicLong(0)
+    private val failures = AtomicLong(0)
+
+    fun started() {
+        started = true
+    }
+
+    fun stopped() {
+        started = false
+    }
+
+    /** 一次作业 tick 正常跑完：记录心跳时刻、耗时与本次扫描积压。 */
+    fun tickFinished(now: Instant, durationMillis: Long, sweepSelected: Int) {
+        lastTickAt = now
+        lastTickMillis = durationMillis
+        lastSweepSelected = sweepSelected
+        ticks.incrementAndGet()
+    }
+
+    /** 一次作业 tick 抛错：只累计失败，**不更新心跳时刻**——停摆由心跳年龄暴露。 */
+    fun tickFailed() {
+        failures.incrementAndGet()
+    }
+
+    fun snapshot(): Snapshot =
+        Snapshot(started, lastTickAt, lastTickMillis, lastSweepSelected, ticks.get(), failures.get())
+
+    data class Snapshot(
+        val started: Boolean,
+        val lastTickAt: Instant?,
+        val lastTickMillis: Long,
+        val lastSweepSelected: Int,
+        val ticks: Long,
+        val failures: Long,
+    )
+}

src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt

@@ -23,6 +23,9 @@ import java.time.Duration
  * - `msgx.pipeline.backfill.oldest_unmarked_seconds`：最老一条仍待自动回填的年龄
  * - `msgx.pipeline.watermark.lag`：水位落后信箱最新 ID 的距离
  * - `msgx.pipeline.hole.aged_out.total`：永久空洞放行次数
+ * - `msgx.pipeline.job.heartbeat_age_seconds`：距上一次作业 tick 完成的秒数（未跑过为 -1）
+ * - `msgx.pipeline.job.ticks.total` / `msgx.pipeline.job.failures.total`：作业 tick 完成/抛错次数
+ * - `msgx.pipeline.job.last_sweep_selected`：上一轮回填扫描选中的待办条数（扫描积压）
  *
  * 取数统一走 [BacklogSnapshotProvider]（30 秒 TTL），因此指标抓取不会打穿数据库。
  * 无法取数时以 `NaN` 上报（Micrometer 的惯例表示"本次无值"），而不是伪造 0。
@@ -41,6 +44,7 @@ class PipelineMetrics(
     private val cursor: BeanProvider<InboxCursorRepository>,
     private val mailbox: BeanProvider<CminmsgInboxRepository>,
     private val counters: PipelineCounters,
+    private val activity: JobActivity,
     private val clock: Clock,
 ) {
 
@@ -74,6 +78,23 @@ class PipelineMetrics(
         Gauge.builder("msgx.pipeline.late_arrival.detected.total", counters) { it.lateArrivalDetectedCount().toDouble() }
             .strongReference(true)
             .register(registry)
+
+        // 作业心跳与扫描积压：作业线程是回填的唯一驱动，停摆必须能被 /metrics 与 /health 看见。
+        Gauge.builder("msgx.pipeline.job.heartbeat_age_seconds", activity) { a ->
+            a.snapshot().lastTickAt?.let { Duration.between(it, clock.instant()).seconds.toDouble() } ?: -1.0
+        }.strongReference(true).register(registry)
+
+        Gauge.builder("msgx.pipeline.job.ticks.total", activity) { it.snapshot().ticks.toDouble() }
+            .strongReference(true)
+            .register(registry)
+
+        Gauge.builder("msgx.pipeline.job.failures.total", activity) { it.snapshot().failures.toDouble() }
+            .strongReference(true)
+            .register(registry)
+
+        Gauge.builder("msgx.pipeline.job.last_sweep_selected", activity) { it.snapshot().lastSweepSelected.toDouble() }
+            .strongReference(true)
+            .register(registry)
     }
 
     private fun backlogGauge(name: String, value: (com.gzzn.omms.msgexchange.infra.persistence.Backlog) -> Double) {

src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt

@@ -1,6 +1,7 @@
 package com.gzzn.omms.msgexchange.jobs
 
 import com.gzzn.omms.msgexchange.config.OperationDayProps
+import com.gzzn.omms.msgexchange.infra.metrics.JobActivity
 import com.gzzn.omms.msgexchange.processing.BackfillService
 import jakarta.inject.Singleton
 import java.time.Clock
@@ -23,6 +24,7 @@ class JobRunner(
     private val backfill: BackfillService,
     private val historySweep: HistorySweepJob,
     private val clock: Clock,
+    private val activity: JobActivity,
     operationDayProps: OperationDayProps,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(JobRunner::class.java)
@@ -39,28 +41,46 @@ class JobRunner(
     fun start() {
         if (thread != null) return
         running = true
+        activity.started()
         thread = Thread.ofPlatform().name("msgx-jobs").daemon(true).start { loop() }
-        log.info("job runner started (backfill 30s, history daily 03:30 {})", zone)
+        log.info("job runner started (backfill {}s, history daily 03:30 {})", TICK_PERIOD.seconds, zone)
     }
 
     fun stop() {
         running = false
         thread?.interrupt()
         thread = null
+        activity.stopped()
     }
 
     internal fun loop() {
         while (running) {
             try {
-                backfill.sweep()
-                maybeHistorySweep()
+                tickOnce()
             } catch (e: InterruptedException) {
                 Thread.currentThread().interrupt()
                 return
-            } catch (e: Exception) {
-                log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName)
             }
-            if (running) runCatching { Thread.sleep(Duration.ofSeconds(30).toMillis()) }
+            if (running) runCatching { Thread.sleep(TICK_PERIOD.toMillis()) }
+        }
+    }
+
+    /**
+     * 跑一轮作业并上报心跳（拉出来单独可测，避免测试里 sleep）。
+     * 回填扫描的选中条数即"扫描积压"的即时值；失败只累计计数，不刷新心跳时刻。
+     */
+    internal fun tickOnce() {
+        val startedAt = clock.instant()
+        val startedNanos = System.nanoTime()
+        try {
+            val selected = backfill.sweep(startedAt)
+            maybeHistorySweep()
+            activity.tickFinished(startedAt, (System.nanoTime() - startedNanos) / 1_000_000, selected)
+        } catch (e: InterruptedException) {
+            throw e
+        } catch (e: Exception) {
+            activity.tickFailed()
+            log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName)
         }
     }
 
@@ -78,4 +98,9 @@ class JobRunner(
             }
         }
     }
+
+    private companion object {
+        /** 作业循环周期：回填扫描与历史作业共用的代码常量（reference 记为 `backfill-scan-period`）。 */
+        val TICK_PERIOD: Duration = Duration.ofSeconds(30)
+    }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt

@@ -0,0 +1,70 @@
+package com.gzzn.omms.msgexchange.infra.health
+
+import com.gzzn.omms.msgexchange.MutableClock
+import com.gzzn.omms.msgexchange.infra.metrics.JobActivity
+import io.micronaut.health.HealthStatus
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+/**
+ * 作业存活判定：没启动不误报；启动了但心跳缺失或超过 3 个扫描周期必须 DOWN；
+ * 正常心跳保持 UP 并把扫描积压等事实带进详情。
+ */
+class JobRunnerHealthIndicatorTest {
+
+    private val now = MutableClock.BASE
+
+    private fun details(result: io.micronaut.management.health.indicator.HealthResult): Map<String, Any> {
+        @Suppress("UNCHECKED_CAST")
+        return result.details as Map<String, Any>
+    }
+
+    @Test
+    fun `a job runner that never started is up but flagged`() {
+        val result = jobRunnerHealth(JobActivity().snapshot(), now)
+
+        assertEquals(HealthStatus.UP, result.status)
+        assertEquals(false, details(result)["started"])
+        assertEquals(-1L, details(result)["lastTickAgeSeconds"])
+    }
+
+    @Test
+    fun `a started job with a fresh heartbeat is up`() {
+        val activity = JobActivity().apply {
+            started()
+            tickFinished(now.minusSeconds(30), durationMillis = 12, sweepSelected = 3)
+        }
+
+        val result = jobRunnerHealth(activity.snapshot(), now)
+
+        assertEquals(HealthStatus.UP, result.status)
+        assertEquals(1L, details(result)["ticks"])
+        assertEquals(3, details(result)["lastSweepSelected"])
+        assertEquals(12L, details(result)["lastTickDurationMs"])
+    }
+
+    @Test
+    fun `a started job that never ticked or went stale is down`() {
+        val neverTicked = JobActivity().apply { started() }
+        assertEquals(HealthStatus.DOWN, jobRunnerHealth(neverTicked.snapshot(), now).status)
+
+        val stale = JobActivity().apply {
+            started()
+            tickFinished(now.minusSeconds(STALE_AFTER_SECONDS + 1), durationMillis = 5, sweepSelected = 0)
+        }
+        assertEquals(HealthStatus.DOWN, jobRunnerHealth(stale.snapshot(), now).status)
+    }
+
+    @Test
+    fun `a failed tick counts up without refreshing the heartbeat`() {
+        val activity = JobActivity().apply { started() }
+
+        activity.tickFailed()
+        activity.tickFailed()
+
+        val snapshot = activity.snapshot()
+        assertEquals(2L, snapshot.failures)
+        assertEquals(0L, snapshot.ticks)
+        assertEquals(null, snapshot.lastTickAt)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt

@@ -58,6 +58,20 @@ class PipelineMetricsTest {
         assertEquals(before + 2.0, gauge("msgx.pipeline.hole.aged_out.total"), 0.001)
     }
 
+    @Test
+    fun `job heartbeat gauges are registered and reflect job activity`() {
+        val activity = ctx.getBean(JobActivity::class.java)
+        val ticksBefore = gauge("msgx.pipeline.job.ticks.total")
+        val failuresBefore = gauge("msgx.pipeline.job.failures.total")
+
+        activity.tickFinished(Instant.now(), durationMillis = 12, sweepSelected = 7)
+
+        assertEquals(ticksBefore + 1.0, gauge("msgx.pipeline.job.ticks.total"), 0.001)
+        assertEquals(failuresBefore, gauge("msgx.pipeline.job.failures.total"), 0.001)
+        assertEquals(7.0, gauge("msgx.pipeline.job.last_sweep_selected"), 0.001)
+        org.junit.jupiter.api.Assertions.assertTrue(gauge("msgx.pipeline.job.heartbeat_age_seconds") >= 0.0)
+    }
+
     private fun gauge(name: String): Double =
         requireNotNull(registry.find(name).gauge()) { "gauge not registered: $name" }.value()
 }

src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt

@@ -0,0 +1,75 @@
+package com.gzzn.omms.msgexchange.jobs
+
+import com.gzzn.omms.msgexchange.MutableClock
+import com.gzzn.omms.msgexchange.config.HistoryProps
+import com.gzzn.omms.msgexchange.config.MailboxProps
+import com.gzzn.omms.msgexchange.config.OperationDayProps
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.infra.metrics.JobActivity
+import com.gzzn.omms.msgexchange.infra.persistence.BackfillDue
+import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightState
+import com.gzzn.omms.msgexchange.infra.stub.StubInbox
+import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
+import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock
+import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx
+import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.processing.BackfillService
+import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Test
+import java.time.Instant
+
+/**
+ * 作业心跳的接线：一轮 tick 跑完必须上报心跳与扫描积压；tick 抛错只累计失败、
+ * **不刷新心跳时刻**（停摆由心跳年龄暴露）。用 `TestClocks` 与内存适配器，不 sleep。
+ */
+class JobRunnerTest {
+
+    private val clock = MutableClock(MutableClock.BASE)
+    private val props = PipelineProps()
+
+    private fun historySweep() = HistorySweepJob(
+        StubFlightState(), StubMsgEvents(), HistoryProps().apply { historyStoreEnabled = false },
+        OperationDayProps(), StubPipelineTx(), StubPipelineLock(),
+    )
+
+    private fun runner(proc: ProcStateRepository, activity: JobActivity) = JobRunner(
+        BackfillService(proc, StubInbox(), MailboxProps(), props, clock, MessageLifecycleGate()),
+        historySweep(), clock, activity, OperationDayProps(),
+    )
+
+    @Test
+    fun `a completed tick reports the heartbeat and the backfill scan backlog`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(1L, MutableClock.BASE.minusSeconds(30L * 86400))
+        proc.markTerminal(1L, ProcStatus.SUCCEEDED, now = MutableClock.BASE)   // 终态未打标 → 本轮到期待办
+        val activity = JobActivity()
+
+        runner(proc, activity).tickOnce()
+
+        val snapshot = activity.snapshot()
+        assertEquals(1, snapshot.lastSweepSelected)
+        assertEquals(1L, snapshot.ticks)
+        assertEquals(0L, snapshot.failures)
+        assertEquals(MutableClock.BASE, snapshot.lastTickAt)
+    }
+
+    @Test
+    fun `a failing tick counts the failure and leaves the heartbeat untouched`() {
+        val broken = object : ProcStateRepository by StubProcState() {
+            override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
+                throw IllegalStateException("db-down")
+        }
+        val activity = JobActivity()
+
+        runner(broken, activity).tickOnce()
+
+        val snapshot = activity.snapshot()
+        assertNull(snapshot.lastTickAt)
+        assertEquals(0L, snapshot.ticks)
+        assertEquals(1L, snapshot.failures)
+    }
+}