feat(monitor): 需下架车道可逐条移除测试库知识 —— 补齐最后一个没有页面入口的流程

背景：来源侧被判「禁止导入」后，已建好的测试库知识不会自动消失；此前只有变更集
删除路径会入队 test.remove，页面既列不出条目也没有按钮，运维无法闭环（车道只有按
库计数）。

新增：
- GET /api/v1/test-publications?removal=required：条目级列表，判据与车道计数同源
  （superseded_at IS NULL / 有 knowledge_id / completed / retrieval_state NOT IN
  (disabling,disabled) / policy_state='blocked'），过滤是叠加而不是替换。
- POST /api/v1/test-publications/{id}/removal：engineer 级写路由，要求 Idempotency-Key，
  按条目幂等（operator-remove:<id>）；置 disabling 并交既有 test.remove 工作流去删除
  远端知识并读回校验。错误映射 404/409(已移除)/409(不可移除)/501(未接线)。
- 界面：需下架车道逐条显示条目 + 「下架（从测试库移除）」，页内二次确认，成功提示，
  完成后计数归零。

修掉过程中踩到的两个真问题：
1. 二次确认的确认态原先经容器 props 传给视图，一次容器改动把三个 props 丢了 ——
   视图永远渲染原按钮、确认步骤静默消失，而所有静态单元测试仍然通过。改为把确认态
   放在视图内部；并把「取消」放在原按钮位置（真实指针连点两下只会落到取消）。
2. jobs 上有部分唯一索引 UNIQUE (business_change_set_id, business_version_id,
   business_target_kb_id, business_action) WHERE business_change_set_id IS NOT NULL。
   本路径给这三列写空串（非 NULL），('','','','remove') 对所有条目是同一个键 ——
   第一条之后每次 INSERT 都撞唯一索引、被 ON CONFLICT DO NOTHING 静默吞掉，留下
   「状态已 disabling、没有任务」的永久卡死条目，而车道恰好隐藏 disabling，运维也
   看不到。改为写 NULL（语义：无变更集身份）+ 冲突目标收敛到 idempotency_key +
   校验 RowsAffected()（0 行回查既有任务；查不到就报错，不提交无任务的 disabling）。
   卡死状态在重试时可自愈。

测试：internal/platform/postgres/test_publication_removal_integration_test.go（两条发布
各自入队、business_* 为 NULL、卡死自愈；把修复退回旧写法该测试即 FAIL，已实测）、
internal/httpapi/test_publication_removal_test.go（权限/幂等键/错误映射/未接线 501/列表
过滤）、前端 testPublications.test.ts + WeKnoraMonitor.test.tsx。

验收（纯界面，两条都走完）：解析监控 → 需下架（2）→ 逐条「下架」+ 二次确认 →
retrieval_state=disabled（PDF 03:57:03Z、md 04:08:55Z），jobs completed（attempt 2，
首次 test_remove_pending＝远端仍在处理，重试通过；工作流以 GetKnowledge 读回 404 判
成功）→ 车道需下架（0）。核验「确认前不执行」：首次点击后仍 unpublished、移除任务数仍 1。

docs/USER_FLOWS_ACCEPTANCE.md

@@ -164,6 +164,23 @@
 - ⚠️ 过程中踩到并修掉 **Bug #6**（见 F.3h）：禁止导入只写 `source_items.policy_state`，而**两处**"期望策略"解析
   （计划 + 建批 delete 就绪判定）都不看它 → 计划不给 delete、建批直接 422
 
+### C5 测试库侧移除（解析监控 → 需下架 → 逐条下架）★2026-10-10 首次实测
+
+- 入口：**解析监控 → 需下架** 车道逐条列出「来源侧被判禁止导入、但知识仍存活在测试库里」的条目
+  （`GET /api/v1/test-publications?removal=required`，口径与车道计数同源）→ 每条 **下架（从测试库移除）**
+  → 页内二次确认（**取消在左、确认在右**）→ 走既有 `test.remove` 工作流（删除测试库知识并读回校验）
+- 期望：条目置 `disabling` → 工作流完成后 `disabled`；车道计数递减；生产知识库不受影响
+- 证据（2026-10-10，两条都经界面走完）：
+  - 点击前车道 `需下架（2）`：`相机线束固定孔位尺寸图.pdf`、`DH-AG95夹爪适配-指令协议.md`（与直接查库的候选一致）
+  - 第 1 条（PDF）：点下架 → 车道回到 1 → `test_publications.retrieval_state=disabled`（`03:57:03Z`）；
+    `jobs` `test.remove` / `completed` / attempt 2（第 1 次 `test_remove_pending`＝远端仍在处理，重试通过）；
+    工作流以 **`GetKnowledge` 读回 404** 判成功
+  - 第 2 条（md）：点下架 → **确认态出现但未执行**（核库：仍 `unpublished`、移除任务数仍 1）→ 点确认
+    → `disabling` → `disabled`（`04:08:55Z`）；job `job-67e7e62c…` `completed`（attempt 2）
+  - 界面终态：`需下架（0）` + 空态文案；页面提示「已受理下架…」
+- ⚠️ 过程中踩到并修掉 **Bug #7**（二次确认按钮出现在原按钮位置，连点即误执行；已把「取消」放左）
+  与 **Bug #8**（`jobs` 部分唯一索引 + 空串 business_* → 第二次移除被静默吞掉、条目永久卡死；已改 NULL + 落行校验）
+
 ### C3 查看结果 → 回滚
 - 入口：**发布管理 → 查看结果** → 需要时 **回滚** → 确认语句 `rollback-production-release`
 - 期望：回滚后目标库该知识停用（**不删知识行**），wiki 页保持
@@ -284,8 +301,9 @@
    唯一约束 `uq_ui_setting_active` 只允许一条 active）恢复，两条 update 各影响 1 行，并复核线上值 ✓；
    根因修复＝给该用例加 per-test schema 隔离（照 `internal/monitoring`），复跑后线上设置不再被改动 ✓。
    **规矩**：任何"创建并激活"的集成测试必须建独立 schema；只判 `TEST_DATABASE_URL` 存在就写共享库的用例一律视为危险。
-3i. **测试库侧的移除仍无页面入口**：生产侧下架已可通过发布通道完成（见 C2b）；但"禁止导入后从**测试库**移除知识"
-   仍只能靠 `test.remove` 工作流（变更集删除路径触发），需下架车道也只有按库计数、没有条目级列表与按钮。
+3i. ~~**测试库侧的移除仍无页面入口**~~ **已闭环（2026-10-10）**：解析监控 → 需下架 现在逐条列出条目，
+   每条可「下架（从测试库移除）」并二次确认（`POST /api/v1/test-publications/{id}/removal`，engineer 权限、
+   写路由需 `Idempotency-Key`、按条目幂等）；实测两条走通（见 C5）。
 3e. **发布/回滚结果页不自动刷新**：Release Worker 完成后，`发布管理` 的执行行仍显示"待执行"（首次与第二遍都遇到），
    需手动 `刷新当前页面`/reload 才看到 `已完成`/`已回滚`。后端状态是对的（DB 已是终态），属展示层轮询缺口。
 3d. **`release_executions.snapshot_verification_id` 为空**：发布前快照门禁确实生效（缺少/过期会 `release_target_snapshot_missing`），但执行行没有记下"用的是哪份快照"，追踪性待补。

internal/governance/test_publication.go

@@ -28,8 +28,11 @@ type TestPublicationFilter struct {
 	ProcessingState string
 	// TargetScope filters the test queue by KB routing target (P4).
 	TargetScope string
-	Limit       int
-	Offset      int
+	// RemovalRequired 只取「被禁止导入、但仍在测试库里存活」的条目：
+	// 解析监控「需下架」车道用它列出待移除的具体条目（此前只有按库计数）。
+	RemovalRequired bool
+	Limit           int
+	Offset          int
 }
 
 type TestPublicationPage struct {

internal/httpapi/server.go

@@ -447,6 +447,11 @@ func NewHandler(config Config) (http.Handler, error) {
 	mux.HandleFunc("GET /api/v1/test-publications", func(w http.ResponseWriter, r *http.Request) {
 		listTestPublications(w, r, config.Store)
 	})
+	// 需下架条目的移除入口：把「被禁止导入但仍存活在测试库」的知识从测试库移除
+	// （置 disabling + 入队既有 test.remove 工作流）。与开始处理同级：engineer 即可。
+	mux.HandleFunc("POST /api/v1/test-publications/{id}/removal", func(w http.ResponseWriter, r *http.Request) {
+		startTestPublicationRemoval(w, r, config)
+	})
 	mux.HandleFunc("POST /api/v1/test-publications/start", func(w http.ResponseWriter, r *http.Request) {
 		startTestPublications(w, r, config)
 	})
@@ -1966,6 +1971,13 @@ func getReleaseBatch(w http.ResponseWriter, r *http.Request, store Store) {
 
 func listTestPublications(w http.ResponseWriter, r *http.Request, store Store) {
 	filter := governance.TestPublicationFilter{ProcessingState: r.URL.Query().Get("processing_state"), TargetScope: r.URL.Query().Get("target_scope"), Limit: 50}
+	if raw := r.URL.Query().Get("removal"); raw != "" {
+		if raw != "required" {
+			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "removal must be required"})
+			return
+		}
+		filter.RemovalRequired = true
+	}
 	if filter.ProcessingState != "" && !validProcessingStateFilter(filter.ProcessingState) {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "processing_state must be a comma-separated list of planned, pending, processing, finalizing, completed, failed, stalled, or cancelled"})
 		return

internal/httpapi/test_publication_removal.go

@@ -0,0 +1,51 @@
+package httpapi
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strings"
+	"time"
+
+	postgresstore "knowledge-governance/internal/platform/postgres"
+)
+
+// testPublicationRemovalStarter 是本路由需要的最小能力面。
+//
+// 真实 Store 实现它；httpapi 的既有 fake 不必实现（断言失败时路由返回 501，
+// 而不是把"没接线"当成成功）。这样避免为一条新路由改动庞大的 Store 接口、连带改掉所有 fake。
+type testPublicationRemovalStarter interface {
+	StartTestPublicationRemoval(context.Context, string, time.Time) (string, error)
+}
+
+// startTestPublicationRemoval 处理「解析监控 → 需下架 → 移除」：
+// 把被禁止导入、但仍在测试库里存活的知识置为 disabling，并交给既有的 test.remove 工作流
+// 去真正删除远端知识并读回校验。语义与变更集删除路径一致，只是由运维在页面上逐个触发。
+func startTestPublicationRemoval(w http.ResponseWriter, r *http.Request, config Config) {
+	if _, ok := requireOperationID(w, r); !ok {
+		return
+	}
+	publicationID := strings.TrimSpace(r.PathValue("id"))
+	if !validCatalogID(publicationID) {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid publication id"})
+		return
+	}
+	starter, ok := config.Store.(testPublicationRemovalStarter)
+	if !ok {
+		writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "test publication removal is not wired"})
+		return
+	}
+	jobID, err := starter.StartTestPublicationRemoval(r.Context(), publicationID, config.Now().UTC())
+	switch {
+	case err == nil:
+		writeJSON(w, http.StatusAccepted, map[string]string{"publication_id": publicationID, "job_id": jobID, "state": "disabling"})
+	case errors.Is(err, postgresstore.ErrPublicationNotFound):
+		writeJSON(w, http.StatusNotFound, map[string]string{"error": "publication not found"})
+	case errors.Is(err, postgresstore.ErrPublicationRemovalDone):
+		writeJSON(w, http.StatusConflict, map[string]string{"error": "该知识的检索已经移除，无需重复下架"})
+	case errors.Is(err, postgresstore.ErrPublicationStateConflict):
+		writeJSON(w, http.StatusConflict, map[string]string{"error": "只有仍在测试库存活的知识才能移除"})
+	default:
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not start test publication removal"})
+	}
+}

internal/httpapi/test_publication_removal_test.go

@@ -0,0 +1,133 @@
+package httpapi
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"knowledge-governance/internal/platform/postgres"
+)
+
+// removalFakeStore 在既有 fakeJobStore 之上补出「移除入口」这一条能力，
+// 用来验证路由的权限、幂等键要求与错误映射。
+type removalFakeStore struct {
+	fakeJobStore
+	capturedPublicationID string
+	jobID                 string
+	err                   error
+}
+
+func (s *removalFakeStore) StartTestPublicationRemoval(_ context.Context, publicationID string, _ time.Time) (string, error) {
+	s.capturedPublicationID = publicationID
+	return s.jobID, s.err
+}
+
+// storeWithoutRemoval 只满足大接口、没有移除能力：路由必须诚实报 501，
+// 而不是把"没接线"当成成功。
+type storeWithoutRemoval struct{ Store }
+
+func TestStartTestPublicationRemovalHandler(t *testing.T) {
+	store := &removalFakeStore{jobID: "job-remove-1"}
+	handler, err := NewHandler(Config{Store: store})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// 缺 Idempotency-Key：写路由一律要求幂等键。
+	missing := httptest.NewRecorder()
+	handler.ServeHTTP(missing, httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/tp-1/removal", nil))
+	if missing.Code != http.StatusBadRequest {
+		t.Fatalf("missing idempotency key status=%d", missing.Code)
+	}
+
+	// 非法 id。
+	badID := httptest.NewRecorder()
+	badRequest := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/bad%20id/removal", nil)
+	badRequest.Header.Set("Idempotency-Key", "op-remove-1")
+	handler.ServeHTTP(badID, badRequest)
+	if badID.Code != http.StatusBadRequest {
+		t.Fatalf("invalid id status=%d", badID.Code)
+	}
+
+	// 正常受理。
+	accepted := httptest.NewRecorder()
+	acceptedRequest := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/tp-1/removal", nil)
+	acceptedRequest.Header.Set("Idempotency-Key", "op-remove-2")
+	handler.ServeHTTP(accepted, acceptedRequest)
+	if accepted.Code != http.StatusAccepted {
+		t.Fatalf("accepted status=%d body=%s", accepted.Code, accepted.Body.String())
+	}
+	var body map[string]string
+	if err := json.Unmarshal(accepted.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if body["publication_id"] != "tp-1" || body["job_id"] != "job-remove-1" || body["state"] != "disabling" {
+		t.Fatalf("accepted body=%v", body)
+	}
+	if store.capturedPublicationID != "tp-1" {
+		t.Fatalf("store saw %q", store.capturedPublicationID)
+	}
+
+	// 错误映射。
+	for _, testCase := range []struct {
+		name   string
+		err    error
+		status int
+	}{
+		{"not found", postgres.ErrPublicationNotFound, http.StatusNotFound},
+		{"already removed", postgres.ErrPublicationRemovalDone, http.StatusConflict},
+		{"state conflict", postgres.ErrPublicationStateConflict, http.StatusConflict},
+		{"unexpected", errors.New("boom"), http.StatusInternalServerError},
+	} {
+		failing := &removalFakeStore{err: testCase.err}
+		failingHandler, err := NewHandler(Config{Store: failing})
+		if err != nil {
+			t.Fatal(err)
+		}
+		recorder := httptest.NewRecorder()
+		request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/tp-1/removal", nil)
+		request.Header.Set("Idempotency-Key", "op-remove-3")
+		failingHandler.ServeHTTP(recorder, request)
+		if recorder.Code != testCase.status {
+			t.Fatalf("%s status=%d want=%d", testCase.name, recorder.Code, testCase.status)
+		}
+	}
+
+	// 未接线：诚实 501。
+	unwired, err := NewHandler(Config{Store: storeWithoutRemoval{Store: store}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	unwiredRecorder := httptest.NewRecorder()
+	unwiredRequest := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/tp-1/removal", nil)
+	unwiredRequest.Header.Set("Idempotency-Key", "op-remove-4")
+	unwired.ServeHTTP(unwiredRecorder, unwiredRequest)
+	if unwiredRecorder.Code != http.StatusNotImplemented {
+		t.Fatalf("unwired status=%d", unwiredRecorder.Code)
+	}
+}
+
+func TestListTestPublicationsRemovalFilter(t *testing.T) {
+	store := &fakeJobStore{}
+	handler, err := NewHandler(Config{Store: store})
+	if err != nil {
+		t.Fatal(err)
+	}
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?removal=required", nil))
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status=%d", recorder.Code)
+	}
+	if !store.publicationFilter.RemovalRequired {
+		t.Fatalf("filter=%#v, want RemovalRequired", store.publicationFilter)
+	}
+	bogus := httptest.NewRecorder()
+	handler.ServeHTTP(bogus, httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?removal=bogus", nil))
+	if bogus.Code != http.StatusBadRequest {
+		t.Fatalf("bogus removal status=%d", bogus.Code)
+	}
+}

internal/platform/postgres/test_publication_removal.go

@@ -0,0 +1,136 @@
+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// ErrPublicationRemovalDone：该发布单元的检索已经移除过（路由映射 409；
+// 条目不存在复用既有的 ErrPublicationNotFound，状态不允许复用 ErrPublicationStateConflict）。
+var ErrPublicationRemovalDone = errors.New("test publication retrieval already removed")
+
+// StartTestPublicationRemoval 让运维把一个「被禁止导入、但仍存活在测试库」的知识从测试库里移除：
+// 把该发布单元置为 disabling 并入队既有的 test.remove 工作流（该工作流会调 WeKnora 删除知识并读回校验，
+// 只有它能把状态推进到 disabled/failed）。
+//
+// 为什么要有这条入口：禁止导入本身只撤销待入库条目并置 policy_state='blocked'，测试库里已经建好的知识
+// 不会自动消失；此前只有"变更集删除路径"会入队 test.remove，界面既列不出条目也没有按钮，运维无法闭环。
+//
+// 幂等：按 publication 固定 idempotency_key（ON CONFLICT DO NOTHING）。重复调用（含并发与用户重试）
+// 只会有一个任务；已在 disabling 时直接返回既有任务 id；已 disabled 时报 ErrPublicationRemovalDone。
+func (s *Store) StartTestPublicationRemoval(ctx context.Context, publicationID string, now time.Time) (string, error) {
+	publicationID = strings.TrimSpace(publicationID)
+	if publicationID == "" || len(publicationID) > 128 || strings.ContainsAny(publicationID, "\x00\r\n") {
+		return "", ErrPublicationNotFound
+	}
+	beginner, ok := s.db.(interface {
+		Begin(context.Context) (pgx.Tx, error)
+	})
+	if !ok {
+		return "", errors.New("test publication removal transactions are unavailable")
+	}
+	tx, err := beginner.Begin(ctx)
+	if err != nil {
+		return "", fmt.Errorf("begin test publication removal: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	var retrievalState, knowledgeID, processingState string
+	var superseded bool
+	err = tx.QueryRow(ctx, `SELECT retrieval_state, knowledge_id, processing_state, superseded_at IS NOT NULL
+FROM test_publications WHERE id=$1 FOR UPDATE`, publicationID).Scan(&retrievalState, &knowledgeID, &processingState, &superseded)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", ErrPublicationNotFound
+	}
+	if err != nil {
+		return "", fmt.Errorf("load test publication for removal: %w", err)
+	}
+	key := "operator-remove:" + publicationID
+	if retrievalState == "disabled" {
+		return "", ErrPublicationRemovalDone
+	}
+	if retrievalState == "disabling" {
+		// 已经在移除中：返回既有任务，不重复入队。
+		var existingJobID string
+		if err := tx.QueryRow(ctx, `SELECT id FROM jobs WHERE idempotency_key=$1`, key).Scan(&existingJobID); err == nil {
+			if err := tx.Commit(ctx); err != nil {
+				return "", fmt.Errorf("commit test publication removal: %w", err)
+			}
+			return existingJobID, nil
+		}
+	}
+	if superseded || knowledgeID == "" || processingState != "completed" {
+		// 只处理"测试库里确实活着"的条目：没上传过的走撤销，不进这里。
+		return "", ErrPublicationStateConflict
+	}
+	tag, err := tx.Exec(ctx, `UPDATE test_publications SET retrieval_state='disabling', last_error='', updated_at=$2 WHERE id=$1`, publicationID, now)
+	if err != nil {
+		return "", fmt.Errorf("mark test publication removal: %w", err)
+	}
+	if tag.RowsAffected() != 1 {
+		return "", ErrPublicationStateConflict
+	}
+	jobID, err := newOpaqueID("job-")
+	if err != nil {
+		return "", err
+	}
+	payload, err := json.Marshal(map[string]string{"publication_id": publicationID})
+	if err != nil {
+		return "", errors.New("encode test publication removal job")
+	}
+	// business_* 三列必须写 NULL，不能写空串：jobs 上有部分唯一索引
+	//   UNIQUE (business_change_set_id, business_version_id, business_target_kb_id, business_action)
+	//   WHERE business_change_set_id IS NOT NULL
+	// 变更集路径写的是真实 id，空串则是"非 NULL"，于是 ('','','','remove') 对**所有**运维移除都是同一个键，
+	// 第一条之后每次插入都撞唯一索引；而 ON CONFLICT DO NOTHING 会把它静默吞掉，留下"状态已 disabling、
+	// 却没有任务"的永久卡死条目（车道还看不见它）。NULL 表达"没有变更集身份"，索引因此不适用，幂等只由
+	// 下面的 idempotency_key 决定。
+	tag, err = tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at,business_change_set_id,business_version_id,business_target_kb_id,business_action)
+VALUES($1,'test.remove',$2,'pending',20,$3,10,$4,$3,$3,NULL,NULL,NULL,'remove')
+ON CONFLICT (idempotency_key) DO NOTHING`, jobID, payload, now, key)
+	if err != nil {
+		return "", fmt.Errorf("enqueue test publication removal: %w", err)
+	}
+	if tag.RowsAffected() != 1 {
+		// 键已存在（并发重复调用）：必须是同一个任务，不能留下没有任务的 disabling。
+		var existingJobID string
+		if err := tx.QueryRow(ctx, `SELECT id FROM jobs WHERE idempotency_key=$1`, key).Scan(&existingJobID); err != nil {
+			return "", fmt.Errorf("test publication removal job was not enqueued: %w", err)
+		}
+		if err := tx.Commit(ctx); err != nil {
+			return "", fmt.Errorf("commit test publication removal: %w", err)
+		}
+		return existingJobID, nil
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return "", fmt.Errorf("commit test publication removal: %w", err)
+	}
+	return jobID, nil
+}
+
+func mustJSONString(value string) string {
+	var builder strings.Builder
+	builder.WriteByte('"')
+	for _, r := range value {
+		switch r {
+		case '"':
+			builder.WriteString(`\"`)
+		case '\\':
+			builder.WriteString(`\\`)
+		default:
+			if r < 0x20 {
+				builder.WriteString(fmt.Sprintf(`\u%04x`, r))
+				continue
+			}
+			builder.WriteRune(r)
+		}
+	}
+	builder.WriteByte('"')
+	return builder.String()
+}

internal/platform/postgres/test_publication_removal_integration_test.go

@@ -0,0 +1,225 @@
+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"knowledge-governance/internal/governance"
+)
+
+// TestStartTestPublicationRemovalIntegration 固化「需下架 → 移除」这条运维闭环的服务端语义：
+// 置 disabling + 入队既有 test.remove 工作流（该工作流负责真正删远端知识并读回校验），
+// 且必须幂等——重复点击/并发只有一个任务；已移除、未上传、已 supersede 的条目要明确拒绝。
+func TestStartTestPublicationRemovalIntegration(t *testing.T) {
+	databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
+	if databaseURL == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pool.Close()
+	if err := ApplyMigrations(ctx, pool); err != nil {
+		t.Fatal(err)
+	}
+	store := NewStore(pool)
+	now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
+	if _, err := pool.Exec(ctx, `INSERT INTO source_connections (id, source_type, name) VALUES ('conn-manual', 'manual', 'Manual')`); err != nil {
+		t.Fatal(err)
+	}
+	insertReviewFixture(t, ctx, pool, 2, "tp-removal", now)
+	insertReviewFixture(t, ctx, pool, 3, "tp-planned", now)
+	// 「未上传」的条目：没有 knowledge id，只能走撤销，不能走移除。
+	if _, err := pool.Exec(ctx, `UPDATE test_publications SET knowledge_id='', ingest_state='pending', processing_state='planned' WHERE id='tp-planned'`); err != nil {
+		t.Fatal(err)
+	}
+
+	jobID, err := store.StartTestPublicationRemoval(ctx, "tp-removal", now)
+	if err != nil || jobID == "" {
+		t.Fatalf("first removal job=%q error=%v", jobID, err)
+	}
+	var retrievalState string
+	if err := pool.QueryRow(ctx, `SELECT retrieval_state FROM test_publications WHERE id='tp-removal'`).Scan(&retrievalState); err != nil {
+		t.Fatal(err)
+	}
+	if retrievalState != "disabling" {
+		t.Fatalf("retrieval_state=%q, want disabling", retrievalState)
+	}
+	var jobType, payload, key string
+	var jobCount int
+	if err := pool.QueryRow(ctx, `SELECT type, payload::text, idempotency_key FROM jobs WHERE id=$1`, jobID).Scan(&jobType, &payload, &key); err != nil {
+		t.Fatal(err)
+	}
+	var decoded struct {
+		PublicationID string `json:"publication_id"`
+	}
+	if err := json.Unmarshal([]byte(payload), &decoded); err != nil || decoded.PublicationID != "tp-removal" {
+		t.Fatalf("job payload=%q error=%v", payload, err)
+	}
+	if jobType != "test.remove" || key != "operator-remove:tp-removal" {
+		t.Fatalf("job type=%q key=%q", jobType, key)
+	}
+
+	// 重复调用（此时已是 disabling）必须返回同一个任务，且不会多入队。
+	again, err := store.StartTestPublicationRemoval(ctx, "tp-removal", now.Add(time.Minute))
+	if err != nil || again != jobID {
+		t.Fatalf("repeat removal job=%q error=%v, want %q", again, err, jobID)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE idempotency_key='operator-remove:tp-removal'`).Scan(&jobCount); err != nil {
+		t.Fatal(err)
+	}
+	if jobCount != 1 {
+		t.Fatalf("removal jobs=%d, want 1", jobCount)
+	}
+
+	// 已移除：明确 409 语义，而不是静默成功。
+	if _, err := pool.Exec(ctx, `UPDATE test_publications SET retrieval_state='disabled' WHERE id='tp-removal'`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.StartTestPublicationRemoval(ctx, "tp-removal", now); !errors.Is(err, ErrPublicationRemovalDone) {
+		t.Fatalf("already removed error=%v, want ErrPublicationRemovalDone", err)
+	}
+	// 未上传：不可移除。
+	if _, err := store.StartTestPublicationRemoval(ctx, "tp-planned", now); !errors.Is(err, ErrPublicationStateConflict) {
+		t.Fatalf("planned removal error=%v, want ErrPublicationStateConflict", err)
+	}
+	// 不存在的条目。
+	if _, err := store.StartTestPublicationRemoval(ctx, "tp-missing", now); !errors.Is(err, ErrPublicationNotFound) {
+		t.Fatalf("missing removal error=%v, want ErrPublicationNotFound", err)
+	}
+}
+
+// TestStartTestPublicationRemovalEnqueuesOneJobPerPublication 固化踩到过的陷阱：
+// jobs 上有部分唯一索引 UNIQUE (business_change_set_id, business_version_id, business_target_kb_id, business_action)
+// WHERE business_change_set_id IS NOT NULL。运维移除若把 business_* 写成空串（非 NULL），
+// (”,”,”,'remove') 对所有条目都是同一个键——第一条之后每次插入都撞唯一索引，被 ON CONFLICT DO NOTHING
+// 静默吞掉，留下"状态 disabling、没有任务"的永久卡死条目。这里断言：两条不同的移除各自都能入队，
+// 且 business_* 必须是 NULL；外加"只有 disabling 没有任务"的半成品状态能被重试自愈。
+func TestStartTestPublicationRemovalEnqueuesOneJobPerPublication(t *testing.T) {
+	databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
+	if databaseURL == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pool.Close()
+	if err := ApplyMigrations(ctx, pool); err != nil {
+		t.Fatal(err)
+	}
+	store := NewStore(pool)
+	now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
+	if _, err := pool.Exec(ctx, `INSERT INTO source_connections (id, source_type, name) VALUES ('conn-manual', 'manual', 'Manual')`); err != nil {
+		t.Fatal(err)
+	}
+	insertReviewFixture(t, ctx, pool, 8, "tp-remove-a", now)
+	insertReviewFixture(t, ctx, pool, 9, "tp-remove-b", now)
+
+	firstJob, err := store.StartTestPublicationRemoval(ctx, "tp-remove-a", now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondJob, err := store.StartTestPublicationRemoval(ctx, "tp-remove-b", now)
+	if err != nil {
+		t.Fatalf("second removal must still be enqueued: %v", err)
+	}
+	if firstJob == secondJob {
+		t.Fatalf("both removals collapsed onto one job %q", firstJob)
+	}
+	var jobCount int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE type='test.remove'`).Scan(&jobCount); err != nil {
+		t.Fatal(err)
+	}
+	if jobCount != 2 {
+		t.Fatalf("test.remove jobs=%d, want 2", jobCount)
+	}
+	var nullBusinessColumns int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE id=$1 AND business_change_set_id IS NULL AND business_version_id IS NULL AND business_target_kb_id IS NULL`, secondJob).Scan(&nullBusinessColumns); err != nil {
+		t.Fatal(err)
+	}
+	if nullBusinessColumns != 1 {
+		t.Fatalf("business_* columns must be NULL so the partial unique index does not apply")
+	}
+
+	// 半成品状态（只有 disabling、没有任务）必须能被重试自愈，而不是永久卡死。
+	if _, err := pool.Exec(ctx, `DELETE FROM jobs WHERE idempotency_key='operator-remove:tp-remove-a'`); err != nil {
+		t.Fatal(err)
+	}
+	healed, err := store.StartTestPublicationRemoval(ctx, "tp-remove-a", now.Add(time.Minute))
+	if err != nil || healed == "" {
+		t.Fatalf("stuck removal must heal on retry: job=%q error=%v", healed, err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE idempotency_key='operator-remove:tp-remove-a'`).Scan(&jobCount); err != nil {
+		t.Fatal(err)
+	}
+	if jobCount != 1 {
+		t.Fatalf("healed removal jobs=%d, want 1", jobCount)
+	}
+}
+
+// TestListTestPublicationsRemovalRequiredIntegration 固化「需下架」车道的条目列表口径：
+// 只取「被禁止导入 + 仍在测试库存活」的条目；未禁止的、已在移除中的、未上传的都不应出现。
+func TestListTestPublicationsRemovalRequiredIntegration(t *testing.T) {
+	databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
+	if databaseURL == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pool.Close()
+	if err := ApplyMigrations(ctx, pool); err != nil {
+		t.Fatal(err)
+	}
+	store := NewStore(pool)
+	now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
+	if _, err := pool.Exec(ctx, `INSERT INTO source_connections (id, source_type, name) VALUES ('conn-manual', 'manual', 'Manual')`); err != nil {
+		t.Fatal(err)
+	}
+	insertReviewFixture(t, ctx, pool, 4, "tp-live-blocked", now)
+	insertReviewFixture(t, ctx, pool, 5, "tp-live-internal", now)
+	insertReviewFixture(t, ctx, pool, 6, "tp-live-disabling", now)
+	insertReviewFixture(t, ctx, pool, 7, "tp-planned-blocked", now)
+	for _, update := range []string{
+		`UPDATE source_items SET policy_state='blocked' WHERE id='si-review-4'`,
+		`UPDATE source_items SET policy_state='blocked' WHERE id='si-review-6'`,
+		`UPDATE source_items SET policy_state='blocked' WHERE id='si-review-7'`,
+		`UPDATE test_publications SET retrieval_state='disabling' WHERE id='tp-live-disabling'`,
+		`UPDATE test_publications SET knowledge_id='', processing_state='planned' WHERE id='tp-planned-blocked'`,
+	} {
+		if _, err := pool.Exec(ctx, update); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	page, err := store.ListTestPublications(ctx, governance.TestPublicationFilter{RemovalRequired: true, Limit: 50})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(page.Items) != 1 || page.Items[0].ID != "tp-live-blocked" {
+		t.Fatalf("removal list=%#v", page.Items)
+	}
+	if page.Total != 1 {
+		t.Fatalf("removal total=%d, want 1", page.Total)
+	}
+	// 不带过滤时四条都在（过滤是叠加的，不是替换）。
+	all, err := store.ListTestPublications(ctx, governance.TestPublicationFilter{Limit: 50})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(all.Items) != 4 {
+		t.Fatalf("unfiltered list=%d, want 4", len(all.Items))
+	}
+}

internal/platform/postgres/test_publications.go

@@ -117,6 +117,15 @@ const countTestPublicationsSQL = `
 SELECT COUNT(*)
 FROM test_publications
 WHERE ($1 = '' OR processing_state = ANY(string_to_array($1, ','))) AND ($2 = '' OR target_scope = $2)
+  AND ($3::boolean IS NOT TRUE OR EXISTS (
+        SELECT 1 FROM source_item_versions siv
+        JOIN source_items si ON si.id = siv.source_item_id
+        WHERE siv.id = test_publications.source_item_version_id
+          AND test_publications.superseded_at IS NULL
+          AND test_publications.knowledge_id <> ''
+          AND test_publications.processing_state = 'completed'
+          AND test_publications.retrieval_state NOT IN ('disabling','disabled')
+          AND si.policy_state = 'blocked'))
 `
 
 const listTestPublicationsSQL = `
@@ -148,6 +157,9 @@ JOIN source_item_versions siv ON siv.id = p.source_item_version_id
 JOIN source_items si ON si.id = siv.source_item_id
 LEFT JOIN content_objects co ON co.id = siv.content_object_id
 WHERE ($1 = '' OR p.processing_state = ANY(string_to_array($1, ','))) AND ($2 = '' OR p.target_scope = $2) AND p.superseded_at IS NULL
+  -- removal=required：被禁止导入但仍在测试库存活（口径与监控「需下架」车道一致）
+  AND ($5::boolean IS NOT TRUE OR (p.knowledge_id <> '' AND p.processing_state = 'completed'
+        AND p.retrieval_state NOT IN ('disabling','disabled') AND si.policy_state = 'blocked'))
 ORDER BY p.updated_at DESC, p.id
 LIMIT $3 OFFSET $4
 `
@@ -466,10 +478,10 @@ func (s *Store) ListTestPublications(ctx context.Context, filter governance.Test
 	if filter.TargetScope != "" && filter.TargetScope != string(governance.VisibilityInternal) && filter.TargetScope != string(governance.VisibilityExternal) {
 		return governance.TestPublicationPage{}, errors.New("invalid test publication target scope")
 	}
-	if err := s.db.QueryRow(ctx, countTestPublicationsSQL, filter.ProcessingState, filter.TargetScope).Scan(&page.Total); err != nil {
+	if err := s.db.QueryRow(ctx, countTestPublicationsSQL, filter.ProcessingState, filter.TargetScope, filter.RemovalRequired).Scan(&page.Total); err != nil {
 		return governance.TestPublicationPage{}, fmt.Errorf("count test publications: %w", err)
 	}
-	rows, err := s.db.Query(ctx, listTestPublicationsSQL, filter.ProcessingState, filter.TargetScope, filter.Limit, filter.Offset)
+	rows, err := s.db.Query(ctx, listTestPublicationsSQL, filter.ProcessingState, filter.TargetScope, filter.Limit, filter.Offset, filter.RemovalRequired)
 	if err != nil {
 		return governance.TestPublicationPage{}, fmt.Errorf("list test publications: %w", err)
 	}

node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json

@@ -0,0 +1 @@
+{"version":"5.0.3","results":[[":web/src/WeKnoraMonitor.test.tsx",{"duration":113.22230200000001,"failed":false}]]}

web/src/WeKnoraMonitor.test.tsx

@@ -34,6 +34,13 @@ describe('WeKnoraMonitorView',()=>{
   const removal=renderToStaticMarkup(<WeKnoraMonitorView {...baseProps} lane="removal" />)
   expect(removal).toContain('需下架（3）')
   expect(removal).toContain('3 份仍在检索中')
+  const withItems = renderToStaticMarkup(<WeKnoraMonitorView {...baseProps} lane="removal" removalItems={[{ id: 'tp-live', title: '相机线束固定孔位尺寸图.pdf', target_scope: 'internal', processing_state: 'completed', ingest_state: 'uploaded', retrieval_state: 'unpublished', source_item_id: 'si-1', source_item_version_id: 'siv-1', knowledge_id: 'knowledge-1', updated_at: '2026-10-10T03:00:00Z', created_at: '2026-10-10T02:00:00Z' } as never]} /> as never)
+  expect(withItems).toContain('相机线束固定孔位尺寸图.pdf')
+  expect(withItems).toContain('下架（从测试库移除）')
+  // 确认态在视图内部（不依赖容器传参）：初始渲染必须是普通按钮，且不得出现确认/取消，
+  // 避免"点一下就直接执行"的降级。
+  expect(withItems).toContain('下架（从测试库移除）')
+  expect(withItems).not.toContain('确认下架（从测试库移除）')
  })
  it('shows inbox lanes for the knowledge-base diff view',()=>{
   const attention=renderToStaticMarkup(<WeKnoraMonitorView {...baseProps} lane='attention' />)

web/src/WeKnoraMonitor.tsx

@@ -3,7 +3,7 @@ import { useContext, useEffect, useMemo, useState } from 'react'
 import { fetchMonitorFiles, fetchMonitorOverview, fetchRunMetrics } from './monitor'
 import type { RunMetrics } from './monitor'
 import type { KnowledgeProgress, MonitorFilePage, MonitorOverview, MonitorStatus } from './monitor'
-import { fetchJobFailures, fetchTestPublications, retryPublication, retryPublicationsBatch, cancelPublicationsBatch, requeueJobFailure, startTestPublications } from './testPublications'
+import { fetchJobFailures, fetchTestPublications, retryPublication, retryPublicationsBatch, cancelPublicationsBatch, requeueJobFailure, startTestPublications, fetchRemovalRequiredPublications, startPublicationRemoval } from './testPublications'
 import type { JobFailure, ProcessingState, TestPublication, TestPublicationPage } from './testPublications'
 import { fetchActiveUISettings, settingsRefreshDelays } from './uiSettings'
 import { WorkspaceContextTabsSetter } from './workspaceContext'
@@ -107,6 +107,11 @@ interface ViewProps {
   onStart?: (ids: string[]) => void
   onLoadMore?: () => void
   onRefresh: () => void
+  /** 需下架车道的条目列表（被禁止导入但仍存活在测试库），以及移除动作。 */
+  removalItems?: TestPublication[]
+  removalBusyIDs?: string[]
+  removalMessage?: string
+  onStartRemoval?: (id: string) => void
 }
 
 export function WeKnoraMonitorView({
@@ -143,9 +148,15 @@ export function WeKnoraMonitorView({
   onStart = () => undefined,
   onLoadMore = () => undefined,
   onRefresh,
+  removalItems = [],
+  removalBusyIDs = [],
+  removalMessage = '',
+  onStartRemoval = () => undefined,
 }: ViewProps) {
   const [selectedStartIDs, setSelectedStartIDs] = useState<string[]>([])
   const [selectedRetryIDs, setSelectedRetryIDs] = useState<string[]>([])
+  // 破坏性动作的二次确认状态放在视图内部：容器漏传参也不会把"确认"这一步静默降级掉。
+  const [confirmRemovalID, setConfirmRemovalID] = useState('')
   const toggleStartSelection = (id: string) => setSelectedStartIDs((current) => current.includes(id) ? current.filter((value) => value !== id) : [...current, id])
   const toggleRetrySelection = (id: string) => setSelectedRetryIDs((current) => current.includes(id) ? current.filter((value) => value !== id) : [...current, id])
   const stages: Array<[MonitorStatus, number]> = selectedKB ? [
@@ -217,8 +228,10 @@ export function WeKnoraMonitorView({
         </section> : null}
         {lane === 'removal' ? <section className="inbox-section" aria-label="需下架">
           <div className="section-head"><h2>需下架（{laneCounts.removal}）</h2><span className="tiny">来源目录里被改为“禁止导入”或撤销标注、但仍在知识库中可检索的文件</span></div>
-          {(overview?.knowledge_bases || []).some((kb) => kb.needs_removal > 0) ? <ul className="dlq-list">{(overview?.knowledge_bases || []).filter((kb) => kb.needs_removal > 0).map((kb) => <li key={kb.id}><span className="monitor-status failed">需下架</span><b>{kb.name}</b><small>{kb.needs_removal} 份仍在检索中</small></li>)}</ul> : emptyLane('没有标注为禁止但仍留在知识库里的文件。')}
-          <p className="tiny">这里只负责把差异暴露出来：来源项被判“禁止导入”后，仍存活在测试库里的知识必须移除。移除动作走 <code>test.remove</code> 工作流（会删除测试库里的知识条目，重新标注后可再次入库）；该入口当前由变更集删除路径触发，页面尚未提供独立按钮。</p>
+          {removalMessage ? <p className="tiny" role="status">{removalMessage}</p> : null}
+          {removalItems.length ? <ul className="dlq-list">{removalItems.map((item) => <li key={item.id}><span className="monitor-status failed">需下架</span><b>「{item.title}」</b><small>{scope(item.target_scope)} · 知识仍在测试库中可检索</small>{confirmRemovalID === item.id ? <><button onClick={() => setConfirmRemovalID('')}>取消</button><button disabled={removalBusyIDs.includes(item.id)} onClick={() => { setConfirmRemovalID(''); onStartRemoval(item.id) }}>{removalBusyIDs.includes(item.id) ? '下架中…' : '确认下架（从测试库移除）'}</button></> : <button disabled={removalBusyIDs.includes(item.id)} onClick={() => setConfirmRemovalID(item.id)}>下架（从测试库移除）</button>}</li>)}</ul> : null}
+          {(overview?.knowledge_bases || []).some((kb) => kb.needs_removal > 0) ? <ul className="dlq-list">{(overview?.knowledge_bases || []).filter((kb) => kb.needs_removal > 0).map((kb) => <li key={kb.id}><span className="monitor-status failed">需下架</span><b>{kb.name}</b><small>{kb.needs_removal} 份仍在检索中</small></li>)}</ul> : removalItems.length ? null : emptyLane('没有标注为禁止但仍留在知识库里的文件。')}
+          <p className="tiny">这里只负责把差异暴露出来：来源项被判“禁止导入”后，仍存活在测试库里的知识必须移除。逐条点“下架（从测试库移除）”并二次确认后，走 <code>test.remove</code> 工作流删除测试库里的知识条目（会读回校验；重新标注后可再次入库，不影响生产知识库）。</p>
         </section> : null}
         {lane === 'inflight' ? <section className="inbox-section" aria-label="处理中">
           <div className="section-head"><h2>处理中（{inflightItems.length}）</h2><span className="tiny">无需操作，仅供查看</span></div>
@@ -276,6 +289,9 @@ export function WeKnoraMonitor() {
   const [batchMessage, setBatchMessage] = useState('')
   const [failures, setFailures] = useState<JobFailure[]>([])
   const [attention, setAttention] = useState<TestPublication[] | null>(null)
+  const [removalItems, setRemovalItems] = useState<TestPublication[]>([])
+  const [removalBusyIDs, setRemovalBusyIDs] = useState<string[]>([])
+  const [removalMessage, setRemovalMessage] = useState('')
   const [loadingMore, setLoadingMore] = useState(false)
   const [metrics, setMetrics] = useState<RunMetrics | null>(null)
   const setContextTabs = useContext(WorkspaceContextTabsSetter)
@@ -306,8 +322,9 @@ export function WeKnoraMonitor() {
       fetchTestPublications({ processing_state: queueStatus, target_scope: queueScope, limit: 50, offset: 0 }, controller.signal),
       fetchTestPublications({ processing_state: 'planned', limit: 100, offset: 0 }, controller.signal),
       fetchTestPublications({ processing_state: ['failed', 'stalled'], limit: 100, offset: 0 }, controller.signal),
-    ]).then(([queuePage, planPage, attentionPage]) => {
-      if (!controller.signal.aborted) { setQueue(queuePage); setPlans(planPage.items); setAttention(attentionPage.items); setError(null) }
+      fetchRemovalRequiredPublications(controller.signal),
+    ]).then(([queuePage, planPage, attentionPage, removalPage]) => {
+      if (!controller.signal.aborted) { setQueue(queuePage); setPlans(planPage.items); setAttention(attentionPage.items); setRemovalItems(removalPage.items); setError(null) }
     }).catch((cause) => { if (!controller.signal.aborted) setError(cause instanceof Error ? cause.message : '加载解析队列失败') }).finally(() => { if (!controller.signal.aborted) setLoading(false) })
     return () => controller.abort()
   }, [queueStatus, queueScope, documentRevision])
@@ -379,6 +396,22 @@ export function WeKnoraMonitor() {
     }
   }
 
+  // 需下架 → 移除：把被禁止导入、但仍存活在测试库的知识交给既有的 test.remove 工作流。
+  const startRemoval = async (id: string) => {
+    if (removalBusyIDs.includes(id)) return
+    setRemovalBusyIDs((current: string[]) => [...current, id])
+    setRemovalMessage('')
+    try {
+      await startPublicationRemoval(id)
+      setRemovalMessage('已受理下架：正在移除测试库中的知识条目，完成后本条会从「需下架」消失。')
+      setDocumentRevision((value) => value + 1)
+    } catch (cause) {
+      setRemovalMessage(cause instanceof Error ? cause.message : '触发下架失败')
+    } finally {
+      setRemovalBusyIDs((current: string[]) => current.filter((value) => value !== id))
+    }
+  }
+
   const loadMoreQueue = async () => {
     if (loadingMore) return
     setLoadingMore(true)
@@ -441,5 +474,5 @@ export function WeKnoraMonitor() {
       setStarting(false)
     }
   }
-  return <WeKnoraMonitorView overview={overview} files={files} queue={queue} plans={plans} selectedKB={selected} status={status} queueStatus={queueStatus} queueScope={queueScope} loading={loading} error={error} retryingIDs={retryingIDs} requeueingIDs={requeueingIDs} cancellingIDs={cancellingIDs} starting={starting} startMessage={startMessage} batchMessage={batchMessage} failures={failures} metrics={metrics} onSelectKB={(kb) => { setSelected(kb); setStatus(''); setFiles(null) }} onStatus={(value) => { setStatus(value); setFiles(null) }} onQueueStatus={setQueueStatus} onQueueScope={setQueueScope} lane={lane} onLaneChange={(value) => setLane(value as typeof lane)} onRetry={(id) => void retryOne(id)} onBatchRetry={(ids) => void retryBatch(ids)} onBatchCancel={(ids) => void cancelBatch(ids)} onRequeueFailure={(id) => void requeueFailure(id)} onStart={(ids) => void startPlans(ids)} attention={attention ?? undefined} loadingMore={loadingMore} onLoadMore={() => void loadMoreQueue()} onRefresh={() => { setDocumentRevision((value) => value + 1); setResourceRevision((value) => value + 1) }} />
+  return <WeKnoraMonitorView overview={overview} files={files} queue={queue} plans={plans} selectedKB={selected} status={status} queueStatus={queueStatus} queueScope={queueScope} loading={loading} error={error} retryingIDs={retryingIDs} requeueingIDs={requeueingIDs} cancellingIDs={cancellingIDs} starting={starting} startMessage={startMessage} batchMessage={batchMessage} failures={failures} metrics={metrics} onSelectKB={(kb) => { setSelected(kb); setStatus(''); setFiles(null) }} onStatus={(value) => { setStatus(value); setFiles(null) }} onQueueStatus={setQueueStatus} onQueueScope={setQueueScope} lane={lane} onLaneChange={(value) => setLane(value as typeof lane)} onRetry={(id) => void retryOne(id)} onBatchRetry={(ids) => void retryBatch(ids)} onBatchCancel={(ids) => void cancelBatch(ids)} onRequeueFailure={(id) => void requeueFailure(id)} onStart={(ids) => void startPlans(ids)} attention={attention ?? undefined} loadingMore={loadingMore} onLoadMore={() => void loadMoreQueue()} onRefresh={() => { setDocumentRevision((value) => value + 1); setResourceRevision((value) => value + 1) }} removalItems={removalItems} removalBusyIDs={removalBusyIDs} removalMessage={removalMessage} onStartRemoval={(id) => void startRemoval(id)} />
 }

web/src/testPublications.test.ts

@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 
-import { fetchAcceptanceResults, fetchTestPublications, parseTestPublicationsResponse, retryPublication, retryPublicationsBatch, startTestPublications, fetchJobFailures } from './testPublications'
+import { fetchAcceptanceResults, fetchTestPublications, parseTestPublicationsResponse, retryPublication, retryPublicationsBatch, startTestPublications, fetchJobFailures, fetchRemovalRequiredPublications, startPublicationRemoval } from './testPublications'
 
 afterEach(() => vi.unstubAllGlobals())
 
@@ -89,6 +89,37 @@ describe('retryPublication', () => {
   })
 })
 
+describe('需下架条目的移除入口', () => {
+  it('只取被禁止但仍存活在测试库的条目', async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true, json: async () => ({ data: [], total: 0, limit: 50, offset: 0 }) })
+    vi.stubGlobal('fetch', fetchMock)
+
+    await fetchRemovalRequiredPublications()
+
+    const url = String(fetchMock.mock.calls[0][0])
+    expect(url).toContain('/api/v1/test-publications?')
+    expect(url).toContain('removal=required')
+  })
+
+  it('带幂等键触发下架并回传任务号', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ publication_id: 'tp-1', job_id: 'job-remove-1', state: 'disabling' }), { status: 202 }))
+    vi.stubGlobal('fetch', fetchMock)
+
+    await expect(startPublicationRemoval('tp-1')).resolves.toEqual({ job_id: 'job-remove-1' })
+    const [url, init] = fetchMock.mock.calls[0]
+    expect(url).toBe('/api/v1/test-publications/tp-1/removal')
+    expect(init.method).toBe('POST')
+    expect(String(init.headers['Idempotency-Key'])).not.toHaveLength(0)
+  })
+
+  it('把服务端拒绝原因原样抛给界面', async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ status: 409, json: async () => ({ error: '该知识的检索已经移除，无需重复下架' }) })
+    vi.stubGlobal('fetch', fetchMock)
+
+    await expect(startPublicationRemoval('tp-1')).rejects.toThrow('该知识的检索已经移除，无需重复下架')
+  })
+})
+
 describe('startTestPublications', () => {
   it('starts only the explicitly confirmed processing plans', async () => {
     const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ data: { selected: 2, started: 1, already_started: 1, unavailable: 0 } }), { status: 202 }))

web/src/testPublications.ts

@@ -34,6 +34,8 @@ export interface AcceptanceRecord {
 }
 
 export interface TestPublicationQuery {
+  /** removal: 'required' 只取「被禁止导入但仍存活在测试库」的条目（解析监控 → 需下架）。 */
+  removal?: 'required' 
   // A single state, or several at once (joined into a comma list server-side).
   processing_state: ProcessingState | '' | ProcessingState[]
   target_scope?: 'internal' | 'external' | ''
@@ -96,6 +98,7 @@ export async function fetchTestPublications(query: TestPublicationQuery, signal?
   const parameters = new URLSearchParams()
   if (query.processing_state) parameters.set('processing_state', Array.isArray(query.processing_state) ? query.processing_state.join(',') : query.processing_state)
   if (query.target_scope) parameters.set('target_scope', query.target_scope)
+  if (query.removal) parameters.set('removal', query.removal)
   parameters.set('limit', String(query.limit))
   parameters.set('offset', String(query.offset))
   const response = await fetch(`/api/v1/test-publications?${parameters.toString()}`, { headers: { Accept: 'application/json' }, signal })
@@ -103,6 +106,32 @@ export async function fetchTestPublications(query: TestPublicationQuery, signal?
   return parseTestPublicationsResponse(await response.json())
 }
 
+/**
+ * 需下架车道用：列出「来源侧被判禁止导入、但知识仍存活在测试库里」的具体条目。
+ * 此前车道只有按库计数，运维看不到条目、也没有处置入口。
+ */
+export async function fetchRemovalRequiredPublications(signal?: AbortSignal): Promise<TestPublicationPage> {
+  return fetchTestPublications({ processing_state: '', removal: 'required', limit: 50, offset: 0 }, signal)
+}
+
+/**
+ * 触发移除：把该条目置为下架中，交服务端既有的 test.remove 工作流去删远端知识并读回校验。
+ * 服务端按条目幂等（重复点击不会重复入队），因此这里每次生成新的幂等键是安全的。
+ */
+export async function startPublicationRemoval(publicationId: string): Promise<{ job_id: string }> {
+  const response = await fetch(`/api/v1/test-publications/${encodeURIComponent(publicationId)}/removal`, {
+    method: 'POST',
+    headers: { Accept: 'application/json', 'Idempotency-Key': newOperationID() },
+  })
+  if (response.status !== 202) {
+    const body = (await response.json().catch(() => null)) as { error?: string } | null
+    throw new Error(body?.error ?? `触发下架失败（HTTP ${response.status}）`)
+  }
+  const value = await response.json() as unknown
+  if (!isRecord(value) || typeof value.job_id !== 'string') throw new Error('下架响应无效')
+  return { job_id: value.job_id }
+}
+
 export async function fetchAcceptanceResults(publicationId: string, signal?: AbortSignal): Promise<AcceptanceRecord[]> {
   const response = await fetch(`/api/v1/test-publications/${encodeURIComponent(publicationId)}/acceptance`, { headers: { Accept: 'application/json' }, signal })
   if (!response.ok) throw new Error(`加载验收结果失败（HTTP ${response.status}）`)