fix(library): 后台静默必须在关入口之后证明，并在停写者后复核

原顺序先证明后台空、再关入口排空：排空期间还在服务的在途请求会派生新任务，
那一刻的空到停写者时已经失效，而 app 收到 SIGTERM 会 exit(1) 把任务腰斩。

现在固定为：关入口并排空 → 证明后台静默（连续为空 + 稳定窗口 + 任务指纹）→
停写入者 → 停完再复核（队列非空/指纹变化即判 background_truncated）。
scheduled/retry 自行产生任务或任务卡住时 fail closed；旧 journal 的
background_idle 布尔不再当证明；入口关到一半失败时恢复入口并保持维护。

CHANGELOG.md

@@ -1,5 +1,39 @@
 # Changelog
 
+## 开发快照（后台静默必须在关入口之后证明，且要能复核）
+
+### Fixed
+
+- **顺序缺口**：原顺序是「先证明后台静默 → 关入口排空 → 停写者」。排空期间还在服务的
+  在途请求随时会派生新的异步任务，那个"空"到停写者时早就失效；而 app 收到 SIGTERM 会
+  exit(1)，正在跑的任务被腰斩。现在固定为「关入口并排空 → 证明后台静默 → 停写者 →
+  停完再复核」，并且在**入口关闭之前禁止取样**（`awaitBackgroundQuiet` 要求
+  `Gate.Closed && Gate.Drained`）。
+- **单次瞬时 0 不算证明**：新证据 `QuiesceProof` 要求连续取样为空、跨过稳定窗口
+  （默认 1.5s）、任务集合指纹一致；只有 `Idle` 没有指纹的检查器在实例级直接拒绝。
+- **停写者后复核**：`verifyBackgroundAfterStop` 再取一次样，队列非空或指纹变化即判
+  `background_truncated`，按"有任务被截断"失败，绝不继续写目标。
+- **scheduled/retry 自行产生任务**：队列长时间没有进展（默认 30s）就直接失败
+  （`background_busy`），不拿等待时间冒充排空。
+- **旧 journal 的 `background_idle` 布尔不再当证明**：只保留为历史字段，恢复流程一律
+  重新取样（`RecoverWindow` 不读它）；只有布尔、没有证明的旧窗口按"未证明"处理。
+- **恢复路径同样要求先停任务再停写者**：写入者已停但队列还有任务时立即失败并写清
+  外部处理方式（先起服务、入口保持关闭让任务跑完，再重试恢复），不再拿"服务没跑"
+  冒充"后台已静默"。
+- **入口关到一半失败会把入口留在关闭状态**：`restore` 现在对实例级维护总是重新打开
+  入口（不再只看"是否成功关闭过"），放弃窗口时还要用 `ingressReady` 确认入口真的回来了，
+  否则保持维护不释放——避免"维护已释放、服务在跑、入口却是关的"。
+- **zset 队列读取**：对 `scheduled`/`retry` 先 `TYPE` 再 `ZRANGE`，此前直接 `LRANGE`
+  会报 WRONGTYPE，任何自带重试/定时任务的实例都开不了窗口。
+
+### Added
+
+- `QuiesceSnapshot` / `QuiesceProof` 与 `StableQuiescer`：后台静默从"一次布尔"变成
+  可复核证据（样本数、稳定窗口、任务指纹、停写后复核时间）。
+- `SetQuiesceStability(stable, stall)`：调整稳定窗口与"任务无进展"判定。
+- 回归测试：取样必须晚于关入口、排空期间产生的任务要等它结束、队列一直有任务要
+  fail closed、停写后冒出新任务要判截断、旧布尔不算证明、只有 Idle 的检查器被拒。
+
 ## 开发快照（恢复校验按归档身份核对共享表）
 
 ### Fixed

internal/libraryfreeze/freeze.go

@@ -56,6 +56,44 @@ type Quiescer interface {
 	Idle(ctx context.Context) (bool, string, error)
 }
 
+// QuiesceSnapshot 是一次后台任务取样：队列深度 + 任务集合指纹。
+// 指纹的作用是把"这一瞬间恰好是 0"和"一段时间内都没有新任务产生"区分开：
+// scheduled/retry 会自己冒出新任务，只看瞬时 0 无法证明切点。
+type QuiesceSnapshot struct {
+	Depth      int64  `json:"depth"`
+	PendingOps int64  `json:"pending_ops"`
+	Unsettled  int64  `json:"unsettled"`
+	Digest     string `json:"digest"`
+	Detail     string `json:"detail"`
+}
+
+// StableQuiescer 能给出任务集合指纹，才谈得上"证明切点"。
+// 实例级维护必须用它；只会回答 Idle 的检查器只能证明某一瞬间是 0。
+type StableQuiescer interface {
+	Quiescer
+	Snapshot(ctx context.Context) (QuiesceSnapshot, error)
+}
+
+// QuiesceProof 是"入口关掉之后后台真的一直是空的"的可复核证据：
+//
+//	AfterIngress 之后才开始取样（排空期间在途请求还会派生新任务）；
+//	连续 Samples 次取样都为空，且跨过 StableMS 的时间窗；
+//	停写入者之后再取一次样，Depth 仍为 0、指纹不变（否则就是切点不一致=有任务被截断）。
+type QuiesceProof struct {
+	AfterIngress   time.Time `json:"after_ingress"`
+	Samples        int       `json:"samples"`
+	StableMS       int64     `json:"stable_ms"`
+	Depth          int64     `json:"depth"`
+	PendingOps     int64     `json:"pending_ops"`
+	Unsettled      int64     `json:"unsettled"`
+	Digest         string    `json:"digest"`
+	Detail         string    `json:"detail"`
+	ProvenAt       time.Time `json:"proven_at"`
+	PostStopDepth  int64     `json:"post_stop_depth"`
+	PostStopDigest string    `json:"post_stop_digest"`
+	PostStopAt     time.Time `json:"post_stop_at"`
+}
+
 // Inventory must enumerate the complete, independently scoped test writer set,
 // using immutable instance IDs. Never implement it as only the currently running
 // subset. Start must wait for readiness; callers still read back the result.
@@ -97,8 +135,11 @@ type journal struct {
 	BackupSnapshotID string `json:"backup_snapshot_id,omitempty"`
 	// TargetFingerprint 是开窗口时的目标身份：恢复时必须还是同一个目标。
 	TargetFingerprint string `json:"target_fingerprint,omitempty"`
-	// BackgroundIdle 记录"停写入者之前后台任务队列确认为空"。
+	// BackgroundIdle 只记录"这个窗口曾经等到过空队列"；它是历史事实，
+	// 不能当作"再次开入口/重启之后仍然静默"的证明——那种判断只认 Quiesce。
 	BackgroundIdle bool `json:"background_idle,omitempty"`
+	// Quiesce 是入口关闭之后的后台切点证据（稳定窗口 + 停写者后的复核）。
+	Quiesce *QuiesceProof `json:"quiesce,omitempty"`
 	// 恢复证据：只有真正跑完"恢复 + 校验"才会写这三项。
 	RecoveredBackup string          `json:"recovered_backup,omitempty"`
 	RecoveryDigest  string          `json:"recovery_digest,omitempty"`
@@ -122,6 +163,10 @@ type Coordinator struct {
 	quiescer      Quiescer
 	fingerprint   string
 	quiesceWait   time.Duration
+	// quiesceStable 是"连续为空必须跨过的时间窗"：单次瞬时 0 不算证明。
+	quiesceStable time.Duration
+	// quiesceStall 是"同一批任务多久没有进展就判定卡住/被截断"。
+	quiesceStall time.Duration
 }
 
 func Open(root, source string, c Controller) (*Coordinator, error) {
@@ -194,6 +239,15 @@ func (c *Coordinator) SetQuiesceTimeout(d time.Duration) error {
 	return nil
 }
 
+// SetQuiesceStability 调整"稳定切点"的时间窗与任务停滞判定（默认 1.5s / 30s）。
+func (c *Coordinator) SetQuiesceStability(stable, stall time.Duration) error {
+	if stable <= 0 || stall <= 0 {
+		return errors.New("invalid quiesce stability window")
+	}
+	c.quiesceStable, c.quiesceStall = stable, stall
+	return nil
+}
+
 // SetQuiescer 声明后台任务静默检查（HTTP 排空不等于后台任务排空）。
 func (c *Coordinator) SetQuiescer(q Quiescer) error {
 	if q == nil {
@@ -408,6 +462,25 @@ func (c *Coordinator) load(job string) (journal, error) {
 	if j.BackupSnapshotID != "" && !validDigest(j.BackupSnapshotID) {
 		return journal{}, errors.New("invalid backup snapshot id")
 	}
+	// 后台静默只认带稳定窗口与停写后复核的证据。旧的 background_idle 布尔
+	// 只是历史记录：它不能证明"重启/再开入口之后仍然静默"，也不能证明停写者
+	// 那一刻没有任务在跑。这里允许它存在（生存旧 journal），但绝不能当证明用：
+	// RecoverWindow 每次都会重新取样，不读这个布尔。
+	if j.BackgroundIdle && j.Quiesce == nil && !terminal(j.State) {
+		// 非终态窗口没有证明：状态上就是"还没证明静默"，交给恢复流程重新取证。
+		j.BackgroundIdle = false
+	}
+	if q := j.Quiesce; q != nil {
+		if q.Samples < 2 || q.StableMS < 0 || q.Depth != 0 || q.PendingOps != 0 || q.Unsettled != 0 {
+			return journal{}, errors.New("invalid quiesce proof")
+		}
+		if !validDigest(q.Digest) || q.AfterIngress.IsZero() || q.ProvenAt.IsZero() || q.ProvenAt.Before(q.AfterIngress) {
+			return journal{}, errors.New("invalid quiesce proof")
+		}
+		if !q.PostStopAt.IsZero() && (q.PostStopDepth != 0 || q.PostStopDigest != q.Digest || q.PostStopAt.Before(q.ProvenAt)) {
+			return journal{}, errors.New("invalid post-stop quiesce proof")
+		}
+	}
 	// 恢复证据只允许指向"这次窗口自己那份预备份"，且必须带校验摘要。
 	if j.RecoveredAt != nil {
 		if j.RecoveredBackup == "" || j.RecoveredBackup != j.BackupSnapshotID || j.RecoveryDigest == "" {
@@ -547,12 +620,18 @@ func (c *Coordinator) restore(ctx context.Context, j *journal) error {
 		}
 	}
 	// 服务都回来了才开放入口：先开门再启动服务，等于把流量打向没准备好的目标。
-	if j.Gate != nil {
+	//
+	// 注意：入口关到一半失败（闸门已被停、排空超时）时 j.Gate 可能还是 nil，
+	// 但入口确实已经不在了。所以实例级维护一律重新打开入口，而不是只看
+	// "有没有成功关闭的证据"——否则会出现"维护已释放、服务在跑、入口却是关的"。
+	if c.level == "instance" || j.Gate != nil {
 		if c.gate == nil {
 			failures = errors.Join(failures, errors.New("入口闸门缺失：不能确认目标已恢复"))
 		} else if err := c.gate.OpenIngress(ctx); err != nil {
 			failures = errors.Join(failures, fmt.Errorf("目标入口恢复失败：%w", err))
-		} else {
+		} else if j.Gate != nil {
+			// 只有"确实关过"的窗口才记"重新打开"：没有关闭证据时开门是恢复原状，
+			// 不能写成"重开"，否则 journal 校验会误判。
 			j.GateReopened = true
 			if err := c.save(*j); err != nil {
 				failures = errors.Join(failures, err)

internal/libraryfreeze/quiesce_order_test.go

@@ -0,0 +1,340 @@
+package libraryfreeze
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// scriptedQuiescer 按脚本回答取样，并把每次取样记进共享调用序列：
+// 顺序断言必须能看见"取样发生在关入口之后、停写者之前"。
+type scriptedQuiescer struct {
+	calls   *[]string
+	samples int
+	next    func(n int) QuiesceSnapshot
+}
+
+func (q *scriptedQuiescer) Idle(ctx context.Context) (bool, string, error) {
+	snapshot, err := q.Snapshot(ctx)
+	if err != nil {
+		return false, "", err
+	}
+	idle := snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0
+	return idle, snapshot.Detail, nil
+}
+
+func (q *scriptedQuiescer) Snapshot(context.Context) (QuiesceSnapshot, error) {
+	q.samples++
+	*q.calls = append(*q.calls, "quiesce")
+	return q.next(q.samples), nil
+}
+
+func busySample(digest string) QuiesceSnapshot {
+	return QuiesceSnapshot{Depth: 1, Digest: strings.Repeat(digest, 64), Detail: "asynq pending=1"}
+}
+
+func quietSample() QuiesceSnapshot {
+	return QuiesceSnapshot{Depth: 0, Digest: strings.Repeat("e", 64), Detail: "asynq=0 pending_ops=0"}
+}
+
+func firstIndex(calls []string, want string) int {
+	for i, call := range calls {
+		if call == want {
+			return i
+		}
+	}
+	return -1
+}
+
+// 取样必须发生在入口确认关闭之后：关之前拿到的"空"会被排空期间的新任务推翻。
+func TestQuiesceProofStartsOnlyAfterIngressClosed(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	q := &scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot { return quietSample() }}
+	if err := e.SetQuiescer(q); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := e.RunWindow(context.Background(), "job-order", func(context.Context) error { return nil },
+		func(context.Context, *Window) (string, error) { return "", nil }, func(error) bool { return true }); err != nil {
+		t.Fatal(err)
+	}
+	gateAt := firstIndex(c.calls, "gate:close")
+	stopAt := firstIndex(c.calls, "stop:app")
+	if gateAt < 0 || stopAt < 0 {
+		t.Fatalf("calls=%v", c.calls)
+	}
+	for i, call := range c.calls {
+		if call == "quiesce" && i < gateAt {
+			t.Fatalf("quiesce sampled before the ingress was closed: %v", c.calls)
+		}
+	}
+	proofProbes := 0
+	for i, call := range c.calls {
+		if call == "quiesce" && i < stopAt {
+			proofProbes++
+		}
+	}
+	if proofProbes < 2 {
+		t.Fatalf("only %d quiet samples before the writers were stopped: %v", proofProbes, c.calls)
+	}
+	afterStop := 0
+	for i, call := range c.calls {
+		if call == "quiesce" && i > stopAt {
+			afterStop++
+		}
+	}
+	if afterStop == 0 {
+		t.Fatalf("no post-stop verification sample: %v", c.calls)
+	}
+	journalAfter, err := e.load("job-order")
+	if err != nil || journalAfter.Quiesce == nil || journalAfter.Quiesce.Samples < 2 {
+		t.Fatalf("journal=%+v err=%v", journalAfter, err)
+	}
+	if journalAfter.Quiesce.PostStopAt.IsZero() {
+		t.Fatal("停写者之后没有复核后台任务集合")
+	}
+}
+
+// 排空期间通过真实入口派生出来的任务：窗口必须等它结束，不能直接停写者。
+func TestWindowWaitsForTaskEnqueuedDuringDrain(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	q := &scriptedQuiescer{calls: &c.calls, next: func(n int) QuiesceSnapshot {
+		if n <= 3 {
+			if firstIndex(c.calls, "stop:app") >= 0 {
+				t.Fatalf("writer stopped while a task was still running: %v", c.calls)
+			}
+			return busySample("b")
+		}
+		return quietSample()
+	}}
+	if err := e.SetQuiescer(q); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	work := 0
+	if _, err := e.RunWindow(context.Background(), "job-drain-task", func(context.Context) error { return nil },
+		func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true }); err != nil {
+		t.Fatal(err)
+	}
+	if work != 1 {
+		t.Fatalf("work=%d", work)
+	}
+	journalAfter, err := e.load("job-drain-task")
+	if err != nil {
+		t.Fatal(err)
+	}
+	proof := journalAfter.Quiesce
+	if proof == nil || proof.Depth != 0 || proof.PostStopDepth != 0 || proof.PostStopDigest != proof.Digest {
+		t.Fatalf("proof=%+v", proof)
+	}
+	if q.samples < 4 {
+		t.Fatalf("sampled %d times, expected to keep sampling while the task ran", q.samples)
+	}
+}
+
+// 队列一直不空（scheduled/retry 自己冒任务）：宁可不进窗口，也不把任务留给 SIGTERM 腰斩。
+func TestWindowFailsClosedWhenBackgroundKeepsProducing(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	q := &scriptedQuiescer{calls: &c.calls, next: func(n int) QuiesceSnapshot {
+		return busySample([]string{"a", "b", "c"}[n%3])
+	}}
+	if err := e.SetQuiescer(q); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiesceTimeout(400 * time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	work := 0
+	_, err := e.RunWindow(context.Background(), "job-producer", func(context.Context) error { return nil },
+		func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
+	if err == nil || !strings.Contains(err.Error(), "后台任务") {
+		t.Fatalf("err=%v", err)
+	}
+	if work != 0 {
+		t.Fatal("work ran although a producer kept the queue busy")
+	}
+	if firstIndex(c.calls, "stop:app") >= 0 {
+		t.Fatalf("writers were stopped while tasks kept arriving: %v", c.calls)
+	}
+	for _, s := range c.services {
+		if !s.Running {
+			t.Fatalf("service %s stayed stopped after the window was abandoned", s.ID)
+		}
+	}
+	journalAfter, loadErr := e.load("job-producer")
+	if loadErr != nil || journalAfter.State != "failed" || journalAfter.Code != "background_busy" {
+		t.Fatalf("journal=%+v err=%v", journalAfter, loadErr)
+	}
+	if journalAfter.Quiesce != nil {
+		t.Fatalf("failed window carries a fake quiet proof: %+v", journalAfter.Quiesce)
+	}
+}
+
+// 任务在"证明静默"和"停写者"之间冒出来：停完的复核必须发现它，按截断处理。
+func TestWindowFailsClosedWhenTaskAppearsAfterWriterStopped(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	q := &scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
+		if firstIndex(c.calls, "stop:app") >= 0 {
+			return busySample("c")
+		}
+		return quietSample()
+	}}
+	if err := e.SetQuiescer(q); err != nil {
+		t.Fatal(err)
+	}
+	work := 0
+	_, err := e.RunWindow(context.Background(), "job-late-task", func(context.Context) error { return nil },
+		func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
+	if err == nil || !strings.Contains(err.Error(), "切点") {
+		t.Fatalf("err=%v", err)
+	}
+	if work != 0 {
+		t.Fatal("work ran with an inconsistent cut point")
+	}
+	journalAfter, loadErr := e.load("job-late-task")
+	if loadErr != nil || journalAfter.Code != "background_truncated" || journalAfter.State != "failed" {
+		t.Fatalf("journal=%+v err=%v", journalAfter, loadErr)
+	}
+	// 还没写目标：可以把服务恢复，但绝不能再往下写。
+	for _, s := range c.services {
+		if !s.Running {
+			t.Fatalf("service %s stayed stopped after an abandoned window", s.ID)
+		}
+	}
+}
+
+// 旧 journal 里的 background_idle 布尔仍然能读，但绝不当作"后台已静默"的证据。
+func TestLegacyBackgroundIdleIsNotProof(t *testing.T) {
+	root, _, e := instanceFixture(t)
+	j := heldJournal("job-old-idle", strings.Repeat("a", 64), strings.Repeat("b", 64))
+	j.State = "stopping"
+	j.ContentRisk = false
+	j.RestoreBlocked = false
+	j.BackgroundIdle = true
+	if err := e.save(j); err != nil {
+		t.Fatal(err)
+	}
+	reopened, err := Open(root, "target-instance", e.control)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer reopened.Close()
+	loaded, err := reopened.load("job-old-idle")
+	if err != nil {
+		t.Fatalf("legacy journal became unreadable: %v", err)
+	}
+	if loaded.Quiesce != nil || loaded.BackgroundIdle {
+		t.Fatalf("legacy idle flag was promoted to a proof: %+v", loaded)
+	}
+}
+
+// 恢复要自己重新证明后台静默：journal 里的旧证明不能拿来复用。
+func TestRecoverWindowReProvesBackgroundQuiesce(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	stopAllServices(c)
+	if err := e.SetTargetFingerprint(strings.Repeat("a", 64)); err != nil {
+		t.Fatal(err)
+	}
+	j := heldJournal("job-stale-proof", strings.Repeat("a", 64), strings.Repeat("b", 64))
+	now := time.Now().UTC()
+	j.BackgroundIdle = true
+	j.Quiesce = &QuiesceProof{AfterIngress: now.Add(-time.Minute), Samples: 3, StableMS: 1500,
+		Digest: strings.Repeat("e", 64), Detail: "asynq=0", ProvenAt: now.Add(-30 * time.Second)}
+	if err := e.save(j); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := e.load("job-stale-proof"); err != nil {
+		t.Fatalf("fixture journal is invalid: %v", err)
+	}
+	if err := e.SetQuiescer(&scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
+		return busySample("d")
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiesceTimeout(300 * time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiesceStability(5*time.Millisecond, time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	called := false
+	if _, err := e.RecoverWindow(context.Background(), "job-stale-proof",
+		restorerFunc(func(context.Context, string) error { called = true; return nil }),
+		verifierFunc(func(context.Context, string) (string, error) { return "x", nil })); err == nil {
+		t.Fatal("recovery reused the journal's stale quiet proof")
+	}
+	if called {
+		t.Fatal("restore ran although the background was busy")
+	}
+	after, loadErr := e.load("job-stale-proof")
+	if loadErr != nil || terminal(after.State) {
+		t.Fatalf("journal=%+v err=%v", after, loadErr)
+	}
+	for _, s := range c.services {
+		if s.Running {
+			t.Fatal("service was started although recovery could not prove background quiet")
+		}
+	}
+}
+
+// 写入者已停但队列里还有任务：这些任务被截断过，服务不起来就永远排不掉——直接报出来。
+func TestRecoverWindowRefusesWithTruncatedTasks(t *testing.T) {
+	_, c, e := instanceFixture(t)
+	stopAllServices(c)
+	if err := e.SetTargetFingerprint(strings.Repeat("a", 64)); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.save(heldJournal("job-trunc", strings.Repeat("a", 64), strings.Repeat("b", 64))); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.SetQuiescer(&scriptedQuiescer{calls: &c.calls, next: func(int) QuiesceSnapshot {
+		return busySample("f")
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	called := false
+	_, err := e.RecoverWindow(context.Background(), "job-trunc",
+		restorerFunc(func(context.Context, string) error { called = true; return nil }),
+		verifierFunc(func(context.Context, string) (string, error) { return "x", nil }))
+	if err == nil || !strings.Contains(err.Error(), "截断") {
+		t.Fatalf("err=%v", err)
+	}
+	if called {
+		t.Fatal("restore ran with truncated tasks in the queue")
+	}
+	after, loadErr := e.load("job-trunc")
+	if loadErr != nil || after.State != "held" || after.Code != "background_truncated" || !after.RestoreBlocked {
+		t.Fatalf("journal=%+v err=%v", after, loadErr)
+	}
+	for _, s := range c.services {
+		if s.Running {
+			t.Fatal("service was started although truncated tasks remain")
+		}
+	}
+}
+
+// 恢复时拿不到任务指纹（只会回答 Idle 的检查器）就不许收尾：那是"瞬时 0"，不是切点。
+func TestInstanceWindowRejectsQuiescerWithoutSnapshots(t *testing.T) {
+	_, _, e := instanceFixture(t)
+	if err := e.SetQuiescer(idleOnlyQuiescer{}); err != nil {
+		t.Fatal(err)
+	}
+	work := 0
+	_, err := e.RunWindow(context.Background(), "job-idle-only", func(context.Context) error { return nil },
+		func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
+	if err == nil || !strings.Contains(err.Error(), "指纹") {
+		t.Fatalf("err=%v", err)
+	}
+	if work != 0 {
+		t.Fatal("window opened without a provable cut point")
+	}
+}
+
+type idleOnlyQuiescer struct{}
+
+func (idleOnlyQuiescer) Idle(context.Context) (bool, string, error) { return true, "asynq=0", nil }

internal/libraryfreeze/recovery.go

@@ -37,6 +37,9 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
 	}
 	defer unlock(lock)
 
+	if c.level == "instance" && c.gate == nil {
+		return Result{}, errors.New("实例级恢复缺少入口闸门：不能证明入口已关闭")
+	}
 	j, err := c.load(job)
 	if err != nil {
 		return Result{}, err
@@ -71,6 +74,23 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
 			return j.Result, err
 		}
 	}
+	// 写入者已经停住但队列还有任务：这些任务在上一次中断里被截断，服务不起来就永远排不掉。
+	// 直接失败并写清外部处理方式，不拿"服务没跑"冒充"后台已静默"。
+	running, err := c.anyServiceRunning(ctx, j)
+	if err != nil {
+		return j.Result, err
+	}
+	if !running {
+		if stuck := c.checkBackgroundStuck(ctx); stuck != nil {
+			return c.failRecovery(&j, "background_truncated", stuck)
+		}
+	}
+	// 入口已经关了才证明后台静默：恢复期间关闭的入口不会派生新任务，
+	// 但上次中断留下的在跑任务必须等它结束，否则恢复后的服务一起来它就会继续写。
+	proof, quietErr := c.awaitBackgroundQuiet(ctx, &j)
+	if quietErr != nil {
+		return c.failRecovery(&j, "background_busy", fmt.Errorf("恢复前后台任务未静默：%w", quietErr))
+	}
 	if err := c.stopAll(ctx, &j); err != nil {
 		return j.Result, fmt.Errorf("恢复前无法停下写入者：%w", err)
 	}
@@ -80,8 +100,8 @@ func (c *Coordinator) RecoverWindow(ctx context.Context, job string, restorer Re
 	if err := c.verifyDrain(ctx, &j); err != nil {
 		return j.Result, fmt.Errorf("恢复前排空未确认：%w", err)
 	}
-	if err := c.awaitBackgroundIdle(ctx, &j); err != nil {
-		return j.Result, fmt.Errorf("恢复前后台任务未静默：%w", err)
+	if err := c.verifyBackgroundAfterStop(ctx, &j, proof); err != nil {
+		return c.failRecovery(&j, "background_truncated", fmt.Errorf("恢复前切点不一致：%w", err))
 	}
 	j.State = "recovering"
 	if err := c.save(j); err != nil {

internal/libraryfreeze/window.go

@@ -100,6 +100,11 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
 		return Result{}, err
 	}
 	defer unlock(lock)
+	// 实例级维护的入口必须由我们关：没有闸门就没有"入口已关"的证据，
+	// 后面拿什么证明后台静默都是空的。
+	if c.level == "instance" && c.gate == nil {
+		return Result{}, errors.New("实例级维护缺少入口闸门：不能证明入口已关闭")
+	}
 
 	previous, loadErr := c.load(job)
 	j := journal{}
@@ -144,14 +149,9 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
 		}
 	}
 
-	// 后台任务必须先在"还能跑"的时候跑完：入口排空只覆盖 HTTP 连接，
-	// app 内的 asynq worker 会被上游的 exit(1) 直接切断。
-	if idleErr := c.awaitBackgroundIdle(ctx, &j); idleErr != nil {
-		return c.abandonWindow(ctx, &j, idleErr, "background_busy")
-	}
-
-	// 顺序是有原因的：入口先停（新请求被拒、在途请求跑完），写者后停。
-	// 反过来做，正在处理的请求就会被 SIGTERM 切断——那不算排空。
+	// 顺序不能换：先把外部任务的生产入口关掉并排空，再证明后台静默，最后才停写入者。
+	// 先证明静默再关入口是错的——排空期间还在跑的在途请求随时会派生新的异步任务，
+	// 那一刻的"空"到停写者时早就失效了；而 app 一旦 SIGTERM 就 exit(1)，会把任务腰斩。
 	if c.gate != nil && (j.Gate == nil || !j.Gate.Drained) {
 		evidence, gateErr := c.gate.CloseIngress(ctx)
 		if gateErr != nil {
@@ -166,6 +166,12 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
 		}
 	}
 
+	// 入口关了才谈得上"稳定切点"：连续为空 + 跨过稳定窗口 + 拿得到任务指纹。
+	proof, quietErr := c.awaitBackgroundQuiet(ctx, &j)
+	if quietErr != nil {
+		return c.abandonWindow(ctx, &j, quietErr, "background_busy")
+	}
+
 	// 停写入者：任何一步不成立都不许往下写。
 	if stopErr := c.stopAll(ctx, &j); stopErr != nil {
 		return c.abandonWindow(ctx, &j, stopErr, "stop_failed")
@@ -176,6 +182,11 @@ func (c *Coordinator) RunWindow(ctx context.Context, job string, prepare func(co
 	if stopErr := c.verifyDrain(ctx, &j); stopErr != nil {
 		return c.abandonWindow(ctx, &j, stopErr, "drain_unverified")
 	}
+	// 停完再复核一次：入队/派发可能在"证明静默"和"停写者"之间发生。
+	// 队列仍有任务说明有任务被截断——这比同步失败严重，按切点不一致处理，绝不往下写。
+	if stopErr := c.verifyBackgroundAfterStop(ctx, &j, proof); stopErr != nil {
+		return c.abandonWindow(ctx, &j, stopErr, "background_truncated")
+	}
 	if c.ingressClosed != nil {
 		if stopErr := c.ingressClosed(ctx); stopErr != nil {
 			return c.abandonWindow(ctx, &j, fmt.Errorf("目标入口仍然可用：%w", stopErr), "ingress_open")
@@ -228,6 +239,15 @@ func (c *Coordinator) abandonWindow(ctx context.Context, j *journal, cause error
 		_ = c.save(*j)
 		return j.Result, errors.Join(cause, restoreErr)
 	}
+	// 入口必须真的回来了才算"安全放弃"：服务在跑但入口关着，等于把目标静默下线。
+	if c.ingressReady != nil {
+		if readyErr := c.ingressReady(recoveryCtx); readyErr != nil {
+			j.Code = code
+			j.State = "held"
+			_ = c.save(*j)
+			return j.Result, errors.Join(cause, fmt.Errorf("目标入口没有恢复：%w", readyErr))
+		}
+	}
 	ended := time.Now().UTC()
 	j.State = "failed"
 	j.Code = code
@@ -292,42 +312,128 @@ func (c *Coordinator) closeWindow(ctx context.Context, j *journal, cause error,
 	return j.Result, cause
 }
 
-// awaitBackgroundIdle 等到后台任务队列为空再往下走。拿不到结论（查询失败）
-// 也按"没空"处理：不能把"问不到"当成"没任务"。
-func (c *Coordinator) awaitBackgroundIdle(ctx context.Context, j *journal) error {
+// awaitBackgroundQuiet 证明"入口关掉之后后台一直没任务"，返回可复核证据。
+//
+// 关键点：
+//   - 必须在入口关闭且排空之后调用。排空期间在途请求还会派生任务，
+//     在关入口之前取样得到的 0 到停写者时可能已失效。
+//   - 只认"连续为空 + 跨过稳定窗口 + 两次指纹一致"，单次瞬时 0 不算证明。
+//   - 队列一直不空（scheduled/retry 自己冒任务、任务卡住）就失败：宁可不开窗口，
+//     也不能把正在跑的任务留给 SIGTERM 腰斩。
+func (c *Coordinator) awaitBackgroundQuiet(ctx context.Context, j *journal) (*QuiesceProof, error) {
 	if c.quiescer == nil {
 		if c.level == "instance" {
-			return errors.New("没有后台任务静默检查：不能证明停写者时后台任务已结束")
+			return nil, errors.New("没有后台任务静默检查：不能证明停写者时后台任务已结束")
 		}
-		return nil
+		return nil, nil
+	}
+	stable := c.quiesceStable
+	if stable <= 0 {
+		stable = 1500 * time.Millisecond
+	}
+	stall := c.quiesceStall
+	if stall <= 0 {
+		stall = 30 * time.Second
 	}
 	wait := c.quiesceWait
 	if wait <= 0 {
 		wait = 2 * time.Minute
 	}
-	deadline := time.Now().Add(wait)
-	var last string
+	st, stableOK := c.quiescer.(StableQuiescer)
+	if !stableOK {
+		if c.level == "instance" {
+			return nil, errors.New("后台任务检查不给任务集合指纹：无法证明稳定切点（scheduled/retry 可能自行产生）")
+		}
+		return nil, nil
+	}
+	if j.Gate == nil || !j.Gate.Drained {
+		return nil, errors.New("必须先关闭并排空入口，再证明后台静默")
+	}
+
+	started := time.Now().UTC()
+	deadline := started.Add(wait)
+	var anchor QuiesceSnapshot
+	var anchorAt time.Time
+	var lastDetail string
+	samples := 0
+	var busyDigest string
+	var busySince time.Time
 	for {
-		idle, detail, err := c.quiescer.Idle(ctx)
+		snapshot, err := st.Snapshot(ctx)
 		if err != nil {
-			return fmt.Errorf("读不到后台任务状态：%w", err)
+			return nil, fmt.Errorf("读不到后台任务状态：%w", err)
 		}
-		if idle {
-			j.BackgroundIdle = true
-			return nil
+		samples++
+		lastDetail = snapshot.Detail
+		now := time.Now().UTC()
+		if snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0 {
+			if anchorAt.IsZero() {
+				anchor, anchorAt = snapshot, now
+			} else if snapshot.Digest == anchor.Digest && now.Sub(anchorAt) >= stable {
+				proof := &QuiesceProof{
+					AfterIngress: started, Samples: samples, StableMS: now.Sub(anchorAt).Milliseconds(),
+					Depth: 0, PendingOps: 0, Unsettled: 0,
+					Digest: snapshot.Digest, Detail: snapshot.Detail, ProvenAt: now,
+				}
+				j.Quiesce = proof
+				j.BackgroundIdle = true
+				if err := c.save(*j); err != nil {
+					return nil, err
+				}
+				return proof, nil
+			}
+		} else {
+			anchor, anchorAt = QuiesceSnapshot{}, time.Time{}
+			if busyDigest != snapshot.Digest {
+				busyDigest, busySince = snapshot.Digest, now
+			} else if now.Sub(busySince) >= stall {
+				// 同一批任务长时间没有进展：被截断或需要外部处理，直接失败，不猜。
+				return nil, fmt.Errorf("后台任务没有进展（%s）：先让它们结束再开窗口，否则会被 SIGTERM 切断", lastDetail)
+			}
 		}
-		last = detail
-		if time.Now().After(deadline) {
-			return fmt.Errorf("后台任务仍在运行（%s）：等它结束再开窗口，否则会被 SIGTERM 切断", last)
+		if now.After(deadline) {
+			return nil, fmt.Errorf("后台任务仍在运行（%s）：等它结束再开窗口，否则会被 SIGTERM 切断", lastDetail)
 		}
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
-		case <-time.After(500 * time.Millisecond):
+			return nil, ctx.Err()
+		case <-time.After(200 * time.Millisecond):
 		}
 	}
 }
 
+// verifyBackgroundAfterStop 复核"停写者那一刻没有任务在跑"：
+// 队列在停完之后仍为空、指纹不变，才算切点一致；否则说明有任务被截断。
+func (c *Coordinator) verifyBackgroundAfterStop(ctx context.Context, j *journal, proof *QuiesceProof) error {
+	if c.quiescer == nil {
+		return nil
+	}
+	if proof == nil {
+		if c.level == "instance" {
+			return errors.New("没有后台静默证据：不能声称停写者时后台已空")
+		}
+		return nil
+	}
+	st, ok := c.quiescer.(StableQuiescer)
+	if !ok {
+		return nil
+	}
+	snapshot, err := st.Snapshot(ctx)
+	if err != nil {
+		return fmt.Errorf("停写者后读不到后台任务状态：%w", err)
+	}
+	if snapshot.Depth != 0 || snapshot.PendingOps != 0 || snapshot.Unsettled != 0 || snapshot.Digest != proof.Digest {
+		return fmt.Errorf("停写入者后后台任务集合变了（%s）：切点不一致，按任务被截断处理", snapshot.Detail)
+	}
+	proof.PostStopDepth = snapshot.Depth
+	proof.PostStopDigest = snapshot.Digest
+	proof.PostStopAt = time.Now().UTC()
+	if err := c.save(*j); err != nil {
+		return err
+	}
+	return nil
+}
+
 // stopAll 只对"原本在跑"的服务发停止指令，每发一个都先落盘意图：
 // 进程在停止过程中死掉时，磁盘上仍能看出"哪些服务被要求停过"。
 func (c *Coordinator) stopAll(ctx context.Context, j *journal) error {
@@ -415,3 +521,44 @@ func validDigest(id string) bool {
 	decoded, err := hex.DecodeString(id)
 	return err == nil && len(decoded) == 32 && strings.ToLower(id) == id
 }
+
+// anyServiceRunning 判断这个窗口记录的写入者里是否还有在跑的。
+func (c *Coordinator) anyServiceRunning(ctx context.Context, j journal) (bool, error) {
+	items, err := inventory(ctx, c.control)
+	if err != nil {
+		return false, err
+	}
+	running := map[string]bool{}
+	for _, s := range items {
+		running[s.ID] = s.Running
+	}
+	for _, s := range j.Services {
+		if running[s.ID] {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// checkBackgroundStuck 在"写入者已经停住"的前提下判断后台是否还有没跑完的任务：
+// 这时候没人能消费队列，等下去没有意义，只能报出来交给外部处理。
+func (c *Coordinator) checkBackgroundStuck(ctx context.Context) error {
+	if c.quiescer == nil {
+		return nil
+	}
+	st, ok := c.quiescer.(StableQuiescer)
+	if !ok {
+		if c.level == "instance" {
+			return errors.New("后台任务检查不给任务集合指纹：无法判断是否有被截断的任务")
+		}
+		return nil
+	}
+	snapshot, err := st.Snapshot(ctx)
+	if err != nil {
+		return fmt.Errorf("读不到后台任务状态：%w", err)
+	}
+	if snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0 {
+		return nil
+	}
+	return fmt.Errorf("写入者已停但后台仍有任务（%s）：这些任务在上次中断时被截断，需要先把服务起来（入口保持关闭）让它们跑完，再重试恢复", snapshot.Detail)
+}

internal/libraryfreeze/window_test.go

@@ -30,6 +30,7 @@ func (c *exitControl) Inventory(ctx context.Context) ([]Service, error) {
 }
 
 // idleQuiescer：实例级窗口必须能证明后台任务已空，测试里默认给一个空的。
+// 它同时实现 StableQuiescer：只有能给出任务集合指纹的检查器才谈得上证明切点。
 type idleQuiescer struct{ busy bool }
 
 func (q *idleQuiescer) Idle(context.Context) (bool, string, error) {
@@ -39,6 +40,13 @@ func (q *idleQuiescer) Idle(context.Context) (bool, string, error) {
 	return true, "asynq=0 pending_ops=0", nil
 }
 
+func (q *idleQuiescer) Snapshot(context.Context) (QuiesceSnapshot, error) {
+	if q.busy {
+		return QuiesceSnapshot{Depth: 1, Digest: strings.Repeat("b", 64), Detail: "asynq pending=1"}, nil
+	}
+	return QuiesceSnapshot{Depth: 0, Digest: strings.Repeat("e", 64), Detail: "asynq=0 pending_ops=0"}, nil
+}
+
 func instanceFixture(t *testing.T) (string, *exitControl, *Coordinator) {
 	t.Helper()
 	root := t.TempDir()
@@ -58,6 +66,13 @@ func instanceFixture(t *testing.T) (string, *exitControl, *Coordinator) {
 	if err := engine.SetQuiescer(&idleQuiescer{}); err != nil {
 		t.Fatal(err)
 	}
+	// 实例级窗口必须有入口闸门：默认装一个会正常排空的假闸门。
+	if err := engine.SetGate(&fakeGate{calls: &c.calls, drain: true, active: 1}); err != nil {
+		t.Fatal(err)
+	}
+	if err := engine.SetQuiesceStability(5*time.Millisecond, 200*time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
 	return root, c, engine
 }
 
@@ -95,27 +110,31 @@ func TestWindowStopsWritersBeforeWorkAndRestoresAfter(t *testing.T) {
 	if !j.DrainVerified || !j.IngressClosed || !j.ContentRisk || j.Level != "instance" {
 		t.Fatalf("journal evidence missing: %+v", j)
 	}
-	if strings.Join(c.calls, ",") != "stop:app,stop:store,start:app,start:store" {
+	if strings.Join(c.calls, ",") != "gate:close,stop:app,stop:store,start:app,start:store,gate:open" {
 		t.Fatalf("calls=%v", c.calls)
 	}
 }
 
-// 没有优雅退出证据（退出码非 0）时不许开始写目标：在途操作可能被切断。
-func TestWindowRefusesWorkWhenDrainUnverified(t *testing.T) {
+// 实例级窗口必须有入口闸门：没有闸门就无法证明"入口已关"，后台静默也就没有前提。
+// 拒绝发生在动任何容器之前。
+func TestInstanceWindowRequiresGate(t *testing.T) {
 	_, c, e := instanceFixture(t)
-	c.exit["app"] = 143
+	e.gate = nil
 	work := 0
-	_, err := e.RunWindow(context.Background(), "job-drain", func(context.Context) error { return nil },
+	_, err := e.RunWindow(context.Background(), "job-nogate", func(context.Context) error { return nil },
 		func(context.Context, *Window) (string, error) { work++; return "", nil }, func(error) bool { return true })
-	if err == nil || !strings.Contains(err.Error(), "排空") {
+	if err == nil || !strings.Contains(err.Error(), "入口闸门") {
 		t.Fatalf("err=%v", err)
 	}
 	if work != 0 {
-		t.Fatal("work ran without drain evidence")
+		t.Fatal("work ran without an ingress gate")
+	}
+	if len(c.calls) != 0 {
+		t.Fatalf("containers touched without a gate: %v", c.calls)
 	}
 	for _, s := range c.services {
 		if !s.Running {
-			t.Fatal("writers were not restored after drain failure")
+			t.Fatal("writers were stopped without a gate")
 		}
 	}
 }
@@ -212,7 +231,7 @@ func TestWindowCrashDuringWriteKeepsTargetHeld(t *testing.T) {
 	if err = reopened.Recover(context.Background()); err == nil {
 		t.Fatal("recovery released a target with unverified content")
 	}
-	if len(c.calls) != 2 {
+	if len(c.calls) != 3 || c.calls[0] != "gate:close" {
 		t.Fatalf("calls=%v", c.calls)
 	}
 }
@@ -237,7 +256,7 @@ func TestWindowAdoptsStoppingJournal(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if strings.Join(c.calls, ",") != "stop:store,start:app,start:store" {
+	if strings.Join(c.calls, ",") != "gate:close,stop:store,start:app,start:store,gate:open" {
 		t.Fatalf("calls=%v", c.calls)
 	}
 }
@@ -321,9 +340,16 @@ func TestWindowRefusesWhenGateNotDrained(t *testing.T) {
 	if err == nil || work != 0 {
 		t.Fatalf("err=%v work=%d", err, work)
 	}
-	if strings.Join(c.calls, ",") != "gate:close" {
+	// 写入者一个都不许停；但入口已经被关过一次，放弃窗口时必须把它重新打开，
+	// 否则会出现"维护已释放、服务在跑、入口却是关的"。
+	if strings.Join(c.calls, ",") != "gate:close,gate:open" {
 		t.Fatalf("writers were stopped before the ingress was drained: %v", c.calls)
 	}
+	for _, s := range c.services {
+		if !s.Running {
+			t.Fatalf("service %s was stopped although the ingress never drained", s.ID)
+		}
+	}
 }
 
 // 窗口开过之后被人工拉起来的写入者，恢复时必须再停一次：

internal/librarymaintenance/quiesce.go

@@ -3,14 +3,20 @@ package librarymaintenance
 import (
 	"bufio"
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
+
+	"knowledge-governance/internal/libraryfreeze"
 )
 
 // targetQuiescer 回答"后台写入者是不是已经空了"：
@@ -45,55 +51,96 @@ func newTargetQuiescer(ctx context.Context, databaseURL, redisAddr, redisPass st
 
 func (q *targetQuiescer) Close() { q.pool.Close() }
 
-func (q *targetQuiescer) Idle(ctx context.Context) (bool, string, error) {
+// Snapshot 是一次"后台任务集合"取样：既给出深度，也给出任务指纹。
+//
+// 指纹是把"这一瞬间恰好空"和"一段时间内没有新任务产生"分开的唯一手段：
+// scheduled/retry 会自己冒出新任务，只看瞬时 0 无法证明停写者时的切点。
+func (q *targetQuiescer) Snapshot(ctx context.Context) (libraryfreeze.QuiesceSnapshot, error) {
 	var pendingOps, unsettled int64
-	// task_pending_ops / knowledges 是 app 自己的后台账本：不看它们只看 HTTP 会漏。
+	// task_pending_ops / pending_subtasks_count 是 app 自己的"正在做"账本：
+	// 不看它们只看 HTTP 会漏。注意不能拿 parse_status<>'completed' 当"在跑"——
+	// 解析失败/卡住的文档会永远停在那里，那是数据状态，不是会跑动的任务；
+	// 用它当判据会让维护窗口永久开不了（真跑踩过）。
 	if err := q.pool.QueryRow(ctx, `SELECT
  (SELECT count(*) FROM task_pending_ops),
  (SELECT count(*) FROM knowledges WHERE deleted_at IS NULL
-    AND (parse_status IS DISTINCT FROM 'completed' OR pending_subtasks_count IS NULL OR pending_subtasks_count<>0))`).
+    AND coalesce(pending_subtasks_count,0) <> 0)`).
 		Scan(&pendingOps, &unsettled); err != nil {
-		return false, "", fmt.Errorf("读不到后台账本：%w", err)
+		return libraryfreeze.QuiesceSnapshot{}, fmt.Errorf("读不到后台账本：%w", err)
 	}
-	depth, err := q.queueDepth(ctx)
+	tasks, err := q.queueTasks(ctx)
+	if err != nil {
+		return libraryfreeze.QuiesceSnapshot{}, err
+	}
+	sort.Strings(tasks)
+	sum := sha256.Sum256([]byte(strings.Join(tasks, "\n")))
+	return libraryfreeze.QuiesceSnapshot{
+		Depth:      int64(len(tasks)),
+		PendingOps: pendingOps,
+		Unsettled:  unsettled,
+		Digest:     hex.EncodeToString(sum[:]),
+		Detail: fmt.Sprintf("pending_ops=%d unsettled_knowledges=%d asynq_tasks=%d",
+			pendingOps, unsettled, len(tasks)),
+	}, nil
+}
+
+func (q *targetQuiescer) Idle(ctx context.Context) (bool, string, error) {
+	snapshot, err := q.Snapshot(ctx)
 	if err != nil {
 		return false, "", err
 	}
-	detail := fmt.Sprintf("pending_ops=%d unsettled_knowledges=%d asynq_active=%d", pendingOps, unsettled, depth)
-	return pendingOps == 0 && unsettled == 0 && depth == 0, detail, nil
+	idle := snapshot.Depth == 0 && snapshot.PendingOps == 0 && snapshot.Unsettled == 0
+	return idle, snapshot.Detail, nil
 }
 
-// queueDepth 统计 asynq 里"还没结束"的任务数（排队/在跑/待重试）。
+// queueTasks 列出"还没结束"的 asynq 任务（排队/在跑/待重试），每项是 键|任务ID。
 // archived/completed 是历史，不算忙。
-func (q *targetQuiescer) queueDepth(ctx context.Context) (int64, error) {
+func (q *targetQuiescer) queueTasks(ctx context.Context) ([]string, error) {
 	conn, err := dialRedis(q.redisAddr, q.redisPass, q.redisDB, q.timeout)
 	if err != nil {
-		return 0, fmt.Errorf("连不上目标队列：%w", err)
+		return nil, fmt.Errorf("连不上目标队列：%w", err)
 	}
 	defer conn.Close()
 	keys, err := conn.scan(ctx, q.queueMatch)
 	if err != nil {
-		return 0, fmt.Errorf("扫描目标队列失败：%w", err)
+		return nil, fmt.Errorf("扫描目标队列失败：%w", err)
 	}
-	var total int64
+	sort.Strings(keys)
+	tasks := make([]string, 0, 32)
 	for _, key := range keys {
 		suffix := key[strings.LastIndex(key, ":"):]
 		switch suffix {
-		case ":pending", ":active", ":inprogress":
-			n, err := conn.intCmd(ctx, "LLEN", key)
-			if err != nil {
-				return 0, err
-			}
-			total += n
-		case ":scheduled", ":retry":
-			n, err := conn.intCmd(ctx, "ZCARD", key)
-			if err != nil {
-				return 0, err
-			}
-			total += n
+		case ":pending", ":inprogress", ":active", ":scheduled", ":retry":
+		default:
+			continue
+		}
+		items, err := conn.rangeAll(ctx, key)
+		if err != nil {
+			return nil, fmt.Errorf("读取目标队列 %s 失败：%w", key, err)
+		}
+		for _, item := range items {
+			tasks = append(tasks, key+"|"+taskIdentity(item))
 		}
 	}
-	return total, nil
+	return tasks, nil
+}
+
+// taskIdentity 从 asynq 的任务 JSON 里取稳定 ID；取不到就用整条内容摘要，
+// 目的都是"同一批任务 -> 同一个指纹"。
+func taskIdentity(raw string) string {
+	var payload map[string]json.RawMessage
+	if err := json.Unmarshal([]byte(raw), &payload); err == nil {
+		for _, key := range []string{"ID", "id"} {
+			if value, ok := payload[key]; ok {
+				var id string
+				if err := json.Unmarshal(value, &id); err == nil && id != "" {
+					return id
+				}
+			}
+		}
+	}
+	sum := sha256.Sum256([]byte(raw))
+	return hex.EncodeToString(sum[:])
 }
 
 // ---- 最小 RESP 客户端：只用到 AUTH/SELECT/SCAN/LLEN/ZCARD，不引第三方库 ----
@@ -129,6 +176,39 @@ func dialRedis(addr, password string, db int, timeout time.Duration) (*redisConn
 
 func (c *redisConn) Close() error { return c.conn.Close() }
 
+// rangeAll 读整个 list/zset。必须先问 TYPE 再选命令：
+// 对 zset 发 LRANGE 会返回错误，而不是"不是 list"的普通回复。
+func (c *redisConn) rangeAll(ctx context.Context, key string) ([]string, error) {
+	kind, err := c.do("TYPE", key)
+	if err != nil {
+		return nil, err
+	}
+	var command string
+	switch kind {
+	case "list":
+		command = "LRANGE"
+	case "zset":
+		command = "ZRANGE"
+	default:
+		return nil, nil
+	}
+	reply, err := c.do(command, key, "0", "-1")
+	if err != nil {
+		return nil, err
+	}
+	items, ok := reply.([]any)
+	if !ok {
+		return nil, errors.New("unexpected queue reply")
+	}
+	out := make([]string, 0, len(items))
+	for _, item := range items {
+		if text, ok := item.(string); ok {
+			out = append(out, text)
+		}
+	}
+	return out, nil
+}
+
 func (c *redisConn) do(args ...string) (any, error) {
 	_ = c.conn.SetDeadline(time.Now().Add(c.timeout))
 	var builder strings.Builder