diff --git a/CHANGELOG.md b/CHANGELOG.md index a54eb3c..1a7a65d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,34 @@ # Changelog +## 开发快照 0.1.356（三模块正常入口与追溯修复） + +### Fixed + +- **F12 实时刷新镜像缺扫描模块**：`Dockerfile` 的运行阶段补齐 + `scripts/dws_dingtalk_scan.py`（`live_catalog_refresh.py` 的导入闭包）。真实刷新作业 + `job-6e18ea86dce10d5472978573a4022a1a` 在源读取前即 `ModuleNotFoundError`，旧快照保留； + 最终镜像以 `--network none`、无挂载/凭据、entrypoint override 的纯导入与 CLI 依赖探针验证。 +- **F13 就绪探测绑定旧库**：`cmd/api/main.go` 不再在启动时捕获 env 知识库 ID，新增 + `internal/readiness` 每次探测重新解析「网页保存的端点 + 当前持久化绑定 + 同一份 API Key」； + 404/未配置/不可达分别归类，健康行带脱敏探测对象（实例 + 短 ID），存活与业务就绪分开显示。 +- **F14 来源目录在处理记录中丢失**：路由/取件时把 `relative_path`/父目录写入 `source_items`； + 历史空值只在读时按「同连接 + 同 item」的 `catalog_import_intents` 补齐，不按文件名猜配、 + 不覆盖已有路径。解析监控的列表/详情/目录搜索同一口径，并提供来源与监控之间的往返定位。 +- **F15 同名目标难辨**：目标选项与行内显示「名称 + 短 ID + 当前绑定/历史」，随页返回当前 + 生效绑定；新增「仅当前绑定」快捷筛选，计数仍是后端全量口径，历史文档保留。 +- **F16 刷新失败仍显示后台运行**：终态移除「后台运行」，失败显示失败时间、旧快照时间、 + 脱敏原因与作业 ID，并提供重试入口；运行中禁止重复提交，完成后不再显示重试。 + +### Added + +- 新增 `runtime-scripts` 门禁：静态比对运行镜像脚本依赖闭包，缺模块的正负例都必须可见， + 避免「源码能导入、镜像少 COPY 一个文件」再次只在真实刷新时暴露。 + +### Tests + +- 新增就绪解析/探测的真实 HTTP 用例、来源目录读时与写入侧隔离用例（一次性 PG schema）、 + 同名目标标签用例，以及刷新卡 running→failed/completed 的挂载级状态转换用例。 + ## 开发快照 0.1.355（同号版本说明归属修正） ### Docs diff --git a/Dockerfile b/Dockerfile index 0cd3aa1..1cfc521 100644 --- a/Dockerfile +++ b/Dockerfile @@ -50,7 +50,7 @@ RUN set -eux; \ dws version WORKDIR /app COPY --from=build /out/ /app/ -COPY scripts/source_catalog_import.py scripts/live_catalog_refresh.py scripts/artifact_fetch.py \ - scripts/axls_to_xlsx.py scripts/dws_error_map.py /app/scripts/ +COPY scripts/source_catalog_import.py scripts/live_catalog_refresh.py scripts/dws_dingtalk_scan.py \ + scripts/artifact_fetch.py scripts/axls_to_xlsx.py scripts/dws_error_map.py /app/scripts/ USER 65532:65532 CMD ["/app/api"] diff --git a/cmd/api/main.go b/cmd/api/main.go index faac31e..44c3380 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -8,7 +8,6 @@ import ( "net/http" "os" "os/signal" - "strings" "syscall" "time" @@ -26,6 +25,7 @@ import ( "knowledge-governance/internal/qaopsorg" "knowledge-governance/internal/qaopsquality" "knowledge-governance/internal/qawemail" + "knowledge-governance/internal/readiness" "knowledge-governance/internal/serviceverify" "knowledge-governance/internal/version" ) @@ -90,26 +90,6 @@ func run(ctx context.Context, getenv func(string) string) error { return errors.New("read production release authorization token") } } - // P10: with a mounted API key the probe verifies a real knowledge-base read - // (readiness); without one it degrades to the liveness-only HTTP check. - var weknoraReadiness func(context.Context) error - var knowledgeBaseLister httpapi.KnowledgeBaseLister - if strings.TrimSpace(config.TestWeKnoraBaseURL) != "" { - if apiKey, keyErr := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile); keyErr == nil { - probeTimeout := config.TestWeKnoraHTTPTimeout - if probeTimeout > 10*time.Second || probeTimeout <= 0 { - probeTimeout = 10 * time.Second - } - if probeClient, clientErr := weknora.NewClient(weknora.ClientConfig{BaseURL: config.TestWeKnoraBaseURL, APIKey: apiKey, Timeout: probeTimeout, AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP}); clientErr == nil { - knowledgeBaseLister = probeClient - routingKB := config.InternalRoutingKnowledgeBaseID - weknoraReadiness = func(probeCtx context.Context) error { - _, _, err := probeClient.ListKnowledge(probeCtx, routingKB, 1, 1) - return err - } - } - } - } // Production WeKnora is probed for liveness only: the console must show the // real state of that endpoint, while credentials and the publish target stay // in the release runtime config (RELEASE_CONFIG_FILE). @@ -131,7 +111,43 @@ func run(ctx context.Context, getenv func(string) string) error { } return settings.TestWeKnoraBaseURL } - monitorProvider, err := monitoring.NewRealProvider(monitoring.RealProviderConfig{DB: pool, WeKnoraBaseURL: config.TestWeKnoraBaseURL, TestWeKnoraResolver: testEndpointResolver, ProductionWeKnoraBaseURL: config.ProductionWeKnoraBaseURL, ProductionWeKnoraResolver: productionEndpointResolver, MinerUBaseURL: config.MinerUBaseURL, WeKnoraReadiness: weknoraReadiness, WorkerStaleAfter: config.WorkerLivenessStaleAfter}) + // P10 readiness 与执行链看同一目标：网页保存的端点、当前持久化绑定、同一份 + // API Key。启动时只捕获 resolver，不捕获任何一项的值（F13：旧 env 绑定 404）。 + readinessResolver := newWeKnoraReadinessResolver( + func(resolveCtx context.Context) string { + if saved := testEndpointResolver(resolveCtx); saved != "" { + return saved + } + return config.TestWeKnoraBaseURL + }, + config.InternalRoutingKnowledgeBaseID, + func(resolveCtx context.Context) (string, error) { + internal, _, _, err := actionStore.LoadRoutingKBBindings(resolveCtx) + return internal, err + }, + func() (string, error) { return weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile) }, + ) + readinessProber := readiness.Prober{ + Resolve: readinessResolver, + Timeout: config.TestWeKnoraHTTPTimeout, + AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP, + } + weknoraProbe := func(probeCtx context.Context) (string, error) { + result := readinessProber.Check(probeCtx) + return result.Target, result.Err + } + knowledgeBaseLister := &weKnoraLister{ + endpoint: func(resolveCtx context.Context) string { + if saved := testEndpointResolver(resolveCtx); saved != "" { + return saved + } + return config.TestWeKnoraBaseURL + }, + readKey: func() (string, error) { return weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile) }, + timeout: config.TestWeKnoraHTTPTimeout, + allowDockerHTTP: config.TestWeKnoraAllowDockerHTTP, + } + monitorProvider, err := monitoring.NewRealProvider(monitoring.RealProviderConfig{DB: pool, WeKnoraBaseURL: config.TestWeKnoraBaseURL, TestWeKnoraResolver: testEndpointResolver, ProductionWeKnoraBaseURL: config.ProductionWeKnoraBaseURL, ProductionWeKnoraResolver: productionEndpointResolver, MinerUBaseURL: config.MinerUBaseURL, WeKnoraReadiness: weknoraProbe, WorkerStaleAfter: config.WorkerLivenessStaleAfter}) if err != nil { return err } diff --git a/cmd/api/readiness.go b/cmd/api/readiness.go new file mode 100644 index 0000000..a430943 --- /dev/null +++ b/cmd/api/readiness.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "fmt" + "log" + "strings" + "time" + + "knowledge-governance/internal/platform/weknora" + "knowledge-governance/internal/readiness" +) + +// newWeKnoraReadinessResolver 把「探测目标」收敛到与业务同一组真源： +// 端点来自网页保存的服务配置（endpoint 内部已做 env 兜底），知识库来自 +// 当前持久化绑定（未绑定才用 env 引导值），凭据每次重新读同一份 key 文件。 +// 任何一项读取失败都必须显式报错——静默退回旧值正是 F13 的成因。 +func newWeKnoraReadinessResolver( + endpoint func(context.Context) string, + envKnowledgeBaseID string, + loadBinding func(context.Context) (string, error), + readKey func() (string, error), +) func(context.Context) (readiness.Target, error) { + return func(ctx context.Context) (readiness.Target, error) { + knowledgeBaseID := strings.TrimSpace(envKnowledgeBaseID) + if loadBinding != nil { + stored, err := loadBinding(ctx) + if err != nil { + // 诊断细节进服务日志；探测结果只暴露固定文案，避免把内部地址带进页面。 + log.Printf("readiness: load routing bindings: %v", err) + return readiness.Target{}, fmt.Errorf("读取当前知识库绑定失败") + } + if stored = strings.TrimSpace(stored); stored != "" { + knowledgeBaseID = stored + } + } + apiKey := "" + if readKey != nil { + value, err := readKey() + if err != nil { + log.Printf("readiness: read WeKnora API key: %v", err) + return readiness.Target{}, fmt.Errorf("读取 WeKnora API 凭据失败") + } + apiKey = strings.TrimSpace(value) + } + baseURL := "" + if endpoint != nil { + baseURL = strings.TrimSpace(endpoint(ctx)) + } + return readiness.Target{BaseURL: baseURL, KnowledgeBaseID: knowledgeBaseID, APIKey: apiKey}, nil + } +} + +// weKnoraLister 让「系统设置」的知识库下拉与保存校验同样跟随当前端点/凭据， +// 而不是复用启动时的客户端（否则改端点后选项仍来自旧实例）。 +type weKnoraLister struct { + endpoint func(context.Context) string + readKey func() (string, error) + timeout time.Duration + allowDockerHTTP bool +} + +func (lister *weKnoraLister) ListKnowledgeBases(ctx context.Context) ([]weknora.KnowledgeBase, error) { + baseURL := "" + if lister.endpoint != nil { + baseURL = strings.TrimSpace(lister.endpoint(ctx)) + } + if baseURL == "" { + return nil, fmt.Errorf("WeKnora 端点未配置") + } + apiKey, err := lister.readKey() + if err != nil { + return nil, err + } + timeout := lister.timeout + if timeout <= 0 || timeout > 10*time.Second { + timeout = 10 * time.Second + } + client, err := weknora.NewClient(weknora.ClientConfig{BaseURL: baseURL, APIKey: apiKey, Timeout: timeout, AllowDockerHTTP: lister.allowDockerHTTP}) + if err != nil { + return nil, err + } + return client.ListKnowledgeBases(ctx) +} diff --git a/cmd/api/readiness_test.go b/cmd/api/readiness_test.go new file mode 100644 index 0000000..8a9f1c4 --- /dev/null +++ b/cmd/api/readiness_test.go @@ -0,0 +1,112 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "knowledge-governance/internal/readiness" +) + +// F13 的接线回归：探测目标必须来自「网页保存的端点 + 当前持久化绑定」， +// env 只能是未绑定时的引导兜底；任何一项都不许在启动时固化。 +func TestReadinessResolverUsesPersistedBindingOverEnv(t *testing.T) { + const envKB = "1b52c140-2e18-4676-afc5-4c7d8d190a8c" + const storedKB = "0c84a098-d3ed-4b82-a165-e6de9691d6a8" + resolve := newWeKnoraReadinessResolver( + func(context.Context) string { return "https://weknora.test" }, + envKB, + func(context.Context) (string, error) { return storedKB, nil }, + func() (string, error) { return "key-1", nil }, + ) + target, err := resolve(context.Background()) + if err != nil { + t.Fatal(err) + } + if target.KnowledgeBaseID != storedKB || target.BaseURL != "https://weknora.test" || target.APIKey != "key-1" { + t.Fatalf("resolved target = %#v", target) + } +} + +func TestReadinessResolverFallsBackToEnvOnlyWhenUnbound(t *testing.T) { + const envKB = "1b52c140-2e18-4676-afc5-4c7d8d190a8c" + resolve := newWeKnoraReadinessResolver( + func(context.Context) string { return "https://weknora.test" }, + envKB, + func(context.Context) (string, error) { return "", nil }, + func() (string, error) { return "key-1", nil }, + ) + target, err := resolve(context.Background()) + if err != nil || target.KnowledgeBaseID != envKB { + t.Fatalf("unbound fallback = %#v, %v", target, err) + } +} + +func TestReadinessResolverReportsBindingAndCredentialFailures(t *testing.T) { + bindingDown := newWeKnoraReadinessResolver( + func(context.Context) string { return "https://weknora.test" }, + "env-kb", + func(context.Context) (string, error) { return "", errors.New("db unavailable") }, + func() (string, error) { return "key-1", nil }, + ) + target, err := bindingDown(context.Background()) + if err == nil { + t.Fatalf("binding read failure must not silently probe env: %#v", target) + } + keyDown := newWeKnoraReadinessResolver( + func(context.Context) string { return "https://weknora.test" }, + "env-kb", + func(context.Context) (string, error) { return "", nil }, + func() (string, error) { return "", errors.New("key file missing") }, + ) + if _, err := keyDown(context.Background()); err == nil { + t.Fatal("missing API key must surface, not degrade to liveness-only") + } +} + +func TestReadinessResolverMarksUnconfiguredEndpoint(t *testing.T) { + resolve := newWeKnoraReadinessResolver( + func(context.Context) string { return "" }, + "env-kb", + func(context.Context) (string, error) { return "", nil }, + func() (string, error) { return "key-1", nil }, + ) + // 端点缺失时归类为「未配置」，不是「服务不可用」。 + result := readiness.Prober{Resolve: resolve}.Check(context.Background()) + if !errors.Is(result.Err, readiness.ErrNotConfigured) { + t.Fatalf("blank endpoint probe = %v, want ErrNotConfigured", result.Err) + } +} + +func TestKnowledgeBaseListerFollowsTheCurrentEndpoint(t *testing.T) { + base := func(name string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/api/v1/knowledge-bases" { + writer.WriteHeader(http.StatusNotFound) + return + } + _, _ = writer.Write([]byte(`{"success":true,"data":[{"id":"` + name + `","name":"` + name + `"}]}`)) + })) + } + first, second := base("kb-a"), base("kb-b") + defer first.Close() + defer second.Close() + endpoint := first.URL + lister := &weKnoraLister{ + endpoint: func(context.Context) string { return endpoint }, + readKey: func() (string, error) { return "key-1", nil }, + timeout: time.Second, + } + bases, err := lister.ListKnowledgeBases(context.Background()) + if err != nil || len(bases) != 1 || bases[0].ID != "kb-a" { + t.Fatalf("initial endpoint bases = %#v, %v", bases, err) + } + endpoint = second.URL + bases, err = lister.ListKnowledgeBases(context.Background()) + if err != nil || len(bases) != 1 || bases[0].ID != "kb-b" { + t.Fatalf("rotated endpoint bases = %#v, %v", bases, err) + } +} diff --git a/internal/governance/monitor.go b/internal/governance/monitor.go index 940d3e1..3d1c6aa 100644 --- a/internal/governance/monitor.go +++ b/internal/governance/monitor.go @@ -36,8 +36,11 @@ type ServiceHealth struct { ObservedAt time.Time `json:"observed_at"` LastSuccessAt *time.Time `json:"last_success_at"` ConsecutiveFailures int `json:"consecutive_failures"` - Configured bool `json:"configured"` - Applicable bool `json:"applicable"` + // ProbeTarget names the object being probed (instance + short binding id), + // never credentials or full UUIDs, so a red light can be attributed. + ProbeTarget string `json:"probe_target"` + Configured bool `json:"configured"` + Applicable bool `json:"applicable"` } type KnowledgeBaseProgress struct { diff --git a/internal/governance/source_catalog.go b/internal/governance/source_catalog.go index 123c113..1157505 100644 --- a/internal/governance/source_catalog.go +++ b/internal/governance/source_catalog.go @@ -116,6 +116,7 @@ type CatalogRouteRequest struct { // fulfillment path needs to materialize and fetch it later. type CatalogRouteFile struct { NodeID string + ParentNodeID string Name string RelativePath string SizeBytes *int64 diff --git a/internal/httpapi/monitor_tasks.go b/internal/httpapi/monitor_tasks.go index 99adad7..4ad5a0f 100644 --- a/internal/httpapi/monitor_tasks.go +++ b/internal/httpapi/monitor_tasks.go @@ -97,6 +97,10 @@ func listMonitorTasks(w http.ResponseWriter, r *http.Request, config Config) { writeJSON(w, 500, map[string]string{"error": "无法读取文件处理状态"}) return } + // 目标同名不罕见（两侧都叫“对内知识库”）：把当前生效绑定随页返回， + // 前端才能在同一口径下标注“当前绑定/历史”，而不是拿本页条目反推全局。 + internal, external := config.effectiveRoutingTargets(r.Context()) + page.Bindings = postgresstore.MonitorBindings{Internal: internal, External: external} w.Header().Set("Cache-Control", "no-store") writeJSON(w, 200, map[string]any{"data": page}) } diff --git a/internal/httpapi/monitor_tasks_test.go b/internal/httpapi/monitor_tasks_test.go index 38623a0..00a558c 100644 --- a/internal/httpapi/monitor_tasks_test.go +++ b/internal/httpapi/monitor_tasks_test.go @@ -2,6 +2,7 @@ package httpapi import ( "context" + "encoding/json" postgresstore "knowledge-governance/internal/platform/postgres" "net/http/httptest" "testing" @@ -51,3 +52,39 @@ func TestMonitorTasksRoute(t *testing.T) { t.Fatalf("viewer removal %d", removal.Code) } } + +// F15：筛选里的“当前绑定”必须来自持久化绑定（与业务同一真源），不是 env 旧值， +// 也不能让前端拿本页数据猜全局。 +func TestMonitorTasksReportsEffectiveTargetBindings(t *testing.T) { + store := &monitorTasksTestStore{} + handler, err := NewHandler(Config{ + Store: store, + InternalRoutingKnowledgeBaseID: "env-old-internal", + ExternalRoutingKnowledgeBaseID: "env-old-external", + RoutingBindings: &bindingStoreFake{internal: "current-internal", external: "current-external"}, + }) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest("GET", "/api/v1/monitor/tasks?limit=20", nil) + req = req.WithContext(context.WithValue(req.Context(), sessionKey{}, sessionIdentity{Username: "tester", Role: roleViewer})) + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, req) + if recorder.Code != 200 { + t.Fatalf("status %d: %s", recorder.Code, recorder.Body.String()) + } + var payload struct { + Data struct { + Bindings struct { + Internal string `json:"internal"` + External string `json:"external"` + } `json:"bindings"` + } `json:"data"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &payload); err != nil { + t.Fatal(err) + } + if payload.Data.Bindings.Internal != "current-internal" || payload.Data.Bindings.External != "current-external" { + t.Fatalf("bindings = %#v, want the persisted current bindings", payload.Data.Bindings) + } +} diff --git a/internal/monitoring/health.go b/internal/monitoring/health.go index 14cce75..2fde222 100644 --- a/internal/monitoring/health.go +++ b/internal/monitoring/health.go @@ -177,10 +177,13 @@ func (provider *RealProvider) weKnoraProbe(ctx context.Context, now time.Time) g return provider.track(classify("weknora", "WeKnora API", false, true, ProbeOutcome{ObservedAt: now})) } outcome := provider.httpProbeOutcome(ctx, baseURL, now) + probeTarget := "" if outcome.Reachability == nil && provider.weknoraReadiness != nil { - outcome.WorkReadiness = provider.weknoraReadiness(ctx) + probeTarget, outcome.WorkReadiness = provider.weknoraReadiness(ctx) } - return provider.track(classify("weknora", "WeKnora API", true, true, outcome)) + health := provider.track(classify("weknora", "WeKnora API", true, true, outcome)) + health.ProbeTarget = probeTarget + return health } // productionWeKnoraProbe reports the real state of the production WeKnora diff --git a/internal/monitoring/health_test.go b/internal/monitoring/health_test.go index b6783f5..bb5f036 100644 --- a/internal/monitoring/health_test.go +++ b/internal/monitoring/health_test.go @@ -12,6 +12,7 @@ import ( "time" "knowledge-governance/internal/governance" + "knowledge-governance/internal/readiness" ) func testNow() time.Time { return time.Date(2026, 9, 12, 9, 0, 0, 0, time.UTC) } @@ -141,3 +142,31 @@ func TestOverallReadinessTreatsProductionEndpointAsAdvisory(t *testing.T) { t.Fatalf("real dependency failure must still block: ok=%v blocked=%v", ok, blocked) } } + +// F13: 旧 env 绑定 404 是「业务就绪被阻断」，不是「服务不可达」； +// 同时必须把脱敏后的探测对象带出来，否则红灯无法归因。 +func TestWeKnoraReadinessReportsBindingBlockWithoutLosingLiveness(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.WriteHeader(http.StatusOK) + })) + defer server.Close() + provider := &RealProvider{ + httpClient: server.Client(), + weknoraBaseURL: server.URL, + health: newHealthTracker(), + weknoraReadiness: func(context.Context) (string, error) { + target := "实例 weknora.test · 知识库 1b52c140" + return target, fmt.Errorf("%w（%s）", readiness.ErrKnowledgeBaseMissing, target) + }, + } + health := provider.weKnoraProbe(context.Background(), testNow()) + if health.Liveness != LiveUp || health.Readiness != ReadBlocked { + t.Fatalf("404 binding = %+v, want up/blocked", health) + } + if health.ProbeTarget != "实例 weknora.test · 知识库 1b52c140" { + t.Fatalf("probe target = %q", health.ProbeTarget) + } + if !strings.Contains(health.Reason, "1b52c140") || strings.Contains(health.Reason, "1b52c140-2e18-4676-afc5-4c7d8d190a8c") { + t.Fatalf("reason = %q, want short target only", health.Reason) + } +} diff --git a/internal/monitoring/real.go b/internal/monitoring/real.go index 9b9fd26..6de783b 100644 --- a/internal/monitoring/real.go +++ b/internal/monitoring/real.go @@ -44,9 +44,9 @@ type RealProviderConfig struct { // place operators configure it — and the env value above stays the // bootstrap fallback for containers that were wired before the UI existed. ProductionWeKnoraResolver func(ctx context.Context) string - // WeKnoraReadiness optionally exercises a real read against a routing - // knowledge base (nil = API key not configured; liveness-only). - WeKnoraReadiness func(ctx context.Context) error + // WeKnoraReadiness optionally exercises a real read against the CURRENT + // routing binding and reports the sanitized probe target alongside the error. + WeKnoraReadiness func(ctx context.Context) (target string, err error) // WorkerStaleAfter bounds how long a missing worker heartbeat still counts // as alive (P10). WorkerStaleAfter time.Duration @@ -61,7 +61,7 @@ type RealProvider struct { testWeKnoraResolver func(ctx context.Context) string productionWeKnoraBaseURL string productionWeKnoraResolver func(ctx context.Context) string - weknoraReadiness func(ctx context.Context) error + weknoraReadiness func(ctx context.Context) (string, error) workerStaleAfter time.Duration health *healthTracker } diff --git a/internal/platform/postgres/catalog_routing.go b/internal/platform/postgres/catalog_routing.go index e1ce0a9..213fb60 100644 --- a/internal/platform/postgres/catalog_routing.go +++ b/internal/platform/postgres/catalog_routing.go @@ -114,7 +114,7 @@ FOR UPDATE OF node`, request.ConnectionID, nodeIDs) JOIN source_catalog_nodes child ON child.snapshot_id=parent.snapshot_id AND child.parent_node_id=parent.node_id WHERE parent.node_type='folder' ) -SELECT DISTINCT node.node_id,node.name,node.relative_path,node.size_bytes,node.modified_at,node.content_sha256, +SELECT DISTINCT node.node_id,node.parent_node_id,node.name,node.relative_path,node.size_bytes,node.modified_at,node.content_sha256, node.source_locator_hash,(node.node_id=ANY($2::text[])) AS directly_selected FROM descendants JOIN source_catalog_nodes node ON node.snapshot_id=descendants.snapshot_id AND node.node_id=descendants.node_id @@ -130,7 +130,7 @@ ORDER BY node.node_id`, request.ConnectionID, nodeIDs) files := make([]governance.CatalogRouteFile, 0, len(nodeIDs)) for rows.Next() { var file governance.CatalogRouteFile - if err := rows.Scan(&file.NodeID, &file.Name, &file.RelativePath, &file.SizeBytes, &file.ModifiedAt, + if err := rows.Scan(&file.NodeID, &file.ParentNodeID, &file.Name, &file.RelativePath, &file.SizeBytes, &file.ModifiedAt, &file.ContentSHA256, &file.LocatorHash, &file.Direct); err != nil { rows.Close() return governance.CatalogRouteResult{}, fmt.Errorf("scan catalog route file: %w", err) diff --git a/internal/platform/postgres/monitor_catalog_path_integration_test.go b/internal/platform/postgres/monitor_catalog_path_integration_test.go new file mode 100644 index 0000000..1bfc407 --- /dev/null +++ b/internal/platform/postgres/monitor_catalog_path_integration_test.go @@ -0,0 +1,217 @@ +package postgres + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/governance" +) + +// F14：来源目录在取件/路由时写入 source_items；对历史行（当时没写）只在读时 +// 按「同一连接 + 同一 item」的 catalog intent 补齐，绝不按文件名猜配。 +func TestMonitorTaskCatalogDirectoryIdentityIntegration(t *testing.T) { + url := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL")) + if url == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err = ApplyMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + exec := func(query string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, query, args...); err != nil { + t.Fatalf("%v: %s", err, query) + } + } + now := time.Date(2026, 11, 5, 2, 0, 0, 0, time.UTC) + exec(`INSERT INTO source_connections (id,source_type,name) VALUES + ('conn-atom','dingtalk','ATOM知识库'),('conn-pp','pp','PP资料下载中心')`) + for index, connection := range []struct{ id, snapshot, folder, file string }{ + {"conn-atom", "snap-atom", "node-atom-folder", "node-atom-file"}, + {"conn-pp", "snap-pp", "node-pp-folder", "node-pp-file"}, + } { + exec(`INSERT INTO source_catalog_snapshots (id,source_connection_id,snapshot_sha256,source_label,node_count,folder_count,file_count,observed_at) + VALUES ($1,$2,repeat($3,64),'fixture',2,1,1,$4)`, + connection.snapshot, connection.id, string(rune('a'+index)), now) + exec(`INSERT INTO source_catalog_nodes (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash) + VALUES ($1,$2,NULL,'folder','售后机器出厂问题整改','售后机器出厂问题整改',repeat($3,64))`, + connection.snapshot, connection.folder, "b") + exec(`INSERT INTO source_catalog_nodes (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash) + VALUES ($1,$2,$3,'file','末端航插线序问题.adoc.docx',$4,repeat($5,64))`, + connection.snapshot, connection.file, connection.folder, + map[string]string{"conn-atom": "售后机器出厂问题整改/末端航插线序问题.adoc.docx", "conn-pp": "PP公开资料/末端航插线序问题.adoc.docx"}[connection.id], + "c") + exec(`INSERT INTO source_catalog_heads (source_connection_id,snapshot_id,activated_at) VALUES ($1,$2,$3)`, + connection.id, connection.snapshot, now) + } + // 两条来源的标题完全相同、来源不同；路径必须各自来自自己的 intent。 + insertItem := func(id, connection, title, storedPath, parent string) { + exec(`INSERT INTO source_items (id,source_connection_id,stable_external_id,canonical_uri_hash,normalized_path,parent_external_id,title,item_type,revision,visibility,visibility_origin,source_change,policy_state,updated_at) + VALUES ($1,$2,$7,repeat('d',64),$3,$4,$5,'file','sha:1','internal','configured_scope','added','internal',$6)`, + id, connection, storedPath, parent, title, now, id) + exec(`INSERT INTO source_item_versions (id,source_item_id,revision,content_state,raw_record_summary,updated_at) + VALUES ('sv-'||$1,$1,'sha:1','verified','{}'::jsonb,$2)`, id, now) + } + insertItem("si-atom", "conn-atom", "末端航插线序问题.adoc.docx", "", "") + insertItem("si-pp", "conn-pp", "末端航插线序问题.adoc.docx", "", "") + insertItem("si-unbound", "conn-pp", "未登记目录文件.adoc.docx", "", "") + insertItem("si-stored", "conn-atom", "已登记目录文件.adoc.docx", "既有登记/已登记目录文件.adoc.docx", "node-atom-folder") + exec(`INSERT INTO catalog_import_intents (source_connection_id,node_id,visibility,source_item_id,decided_by,relative_path,name,locator_hash,updated_at) + VALUES ('conn-atom','node-atom-file','internal','si-atom','fixture','售后机器出厂问题整改/末端航插线序问题.adoc.docx','末端航插线序问题.adoc.docx',repeat('e',64),$1)`, now) + exec(`INSERT INTO catalog_import_intents (source_connection_id,node_id,visibility,source_item_id,decided_by,relative_path,name,locator_hash,updated_at) + VALUES ('conn-pp','node-pp-file','internal','si-pp','fixture','PP公开资料/末端航插线序问题.adoc.docx','末端航插线序问题.adoc.docx',repeat('f',64),$1)`, now) + exec(`INSERT INTO catalog_import_intents (source_connection_id,node_id,visibility,source_item_id,decided_by,relative_path,name,locator_hash,updated_at) + VALUES ('conn-atom','node-atom-folder','internal','si-stored','fixture','售后机器出厂问题整改/已登记目录文件.adoc.docx','已登记目录文件.adoc.docx',repeat('1',64),$1)`, now) + for index, publication := range []struct{ id, item, kb, scope string }{ + {"tp-atom", "si-atom", "kb-internal-a", "internal"}, + {"tp-pp", "si-pp", "kb-internal-b", "internal"}, + {"tp-unbound", "si-unbound", "kb-internal-b", "internal"}, + {"tp-stored", "si-stored", "kb-internal-a", "internal"}, + } { + exec(`INSERT INTO test_publications (id,source_item_version_id,test_kb_id,ingest_state,processing_state,retrieval_state,target_scope,updated_at) + VALUES ($1,'sv-'||$2,$3,'uploaded','completed','unpublished',$4,$5)`, + publication.id, publication.item, publication.kb, publication.scope, now.Add(time.Duration(index)*time.Minute)) + } + store := NewStore(pool) + page, err := store.ListMonitorTasks(ctx, MonitorTaskFilter{Limit: 20}) + if err != nil { + t.Fatal(err) + } + byID := map[string]MonitorTask{} + for _, item := range page.Items { + byID[item.ID] = item + } + atom := byID["tp-atom"] + if atom.Path != "售后机器出厂问题整改/末端航插线序问题.adoc.docx" || atom.ParentNodeID != "node-atom-folder" || atom.NodeID != "node-atom-file" { + t.Fatalf("historical item lost its catalog directory: %#v", atom) + } + pp := byID["tp-pp"] + if pp.Path != "PP公开资料/末端航插线序问题.adoc.docx" || pp.ParentNodeID != "node-pp-folder" { + t.Fatalf("same-title item took another source's path: %#v", pp) + } + if byID["tp-unbound"].Path != "" || byID["tp-unbound"].ParentNodeID != "" { + t.Fatalf("missing intent must not invent a directory: %#v", byID["tp-unbound"]) + } + if stored := byID["tp-stored"]; stored.Path != "既有登记/已登记目录文件.adoc.docx" { + t.Fatalf("stored path must win over the intent snapshot: %#v", stored) + } + searched, err := store.ListMonitorTasks(ctx, MonitorTaskFilter{Search: "售后机器出厂问题整改", Limit: 20}) + if err != nil { + t.Fatal(err) + } + if searched.Total != 1 || len(searched.Items) != 1 || searched.Items[0].SourceID != "conn-atom" { + t.Fatalf("directory search total=%d items=%#v, want exactly the ATOM row", searched.Total, searched.Items) + } + // 同名文件在另一来源下不应被这条目录搜索带出来（同名异源隔离）。 + other, err := store.ListMonitorTasks(ctx, MonitorTaskFilter{Search: "PP公开资料", Limit: 20}) + if err != nil { + t.Fatal(err) + } + if other.Total != 1 || len(other.Items) != 1 || other.Items[0].SourceID != "conn-pp" { + t.Fatalf("cross-source search leaked: %#v", other.Items) + } + // F15：目标选项带 scope（同名知识库靠短 ID + scope 区分），筛选总数由后端全量算。 + if len(page.Targets) != 2 { + t.Fatalf("target options = %#v", page.Targets) + } + for _, option := range page.Targets { + if option.Scope != "internal" || len(option.ID) < 8 { + t.Fatalf("target option missing identity: %#v", option) + } + } + filtered, err := store.ListMonitorTasks(ctx, MonitorTaskFilter{Target: "kb-internal-b", Limit: 20}) + if err != nil { + t.Fatal(err) + } + if filtered.Total != 2 || len(filtered.Items) != 2 { + t.Fatalf("target filter total=%d items=%d, want 2", filtered.Total, len(filtered.Items)) + } +} + +// F14 写入侧：路由/取件时就把来源路径与父目录写进 source_items； +// 对已存在但当时没写的行，只在同一 intent 绑定的 item 上补空值，绝不覆盖既有值。 +func TestCatalogRoutePersistsSourceDirectoryIntegration(t *testing.T) { + url := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL")) + if url == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err = ApplyMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + exec := func(query string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, query, args...); err != nil { + t.Fatalf("%v: %s", err, query) + } + } + now := time.Date(2026, 11, 5, 3, 0, 0, 0, time.UTC) + exec(`INSERT INTO source_connections (id,source_type,name) VALUES ('conn-route','dingtalk','路由来源')`) + exec(`INSERT INTO source_catalog_snapshots (id,source_connection_id,snapshot_sha256,source_label,node_count,folder_count,file_count,observed_at) + VALUES ('snap-route','conn-route',repeat('9',64),'fixture',2,1,1,$1)`, now) + exec(`INSERT INTO source_catalog_nodes (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash) + VALUES ('snap-route','node-route-folder',NULL,'folder','售后机器出厂问题整改','售后机器出厂问题整改',repeat('8',64))`) + exec(`INSERT INTO source_catalog_nodes (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash) + VALUES ('snap-route','node-route-file','node-route-folder','file','末端航插线序问题.adoc.docx','售后机器出厂问题整改/末端航插线序问题.adoc.docx',repeat('7',64))`) + exec(`INSERT INTO source_catalog_heads (source_connection_id,snapshot_id,activated_at) VALUES ('conn-route','snap-route',$1)`, now) + // 历史行：当时没写路径，但 intent 已把它绑定到本 item。 + exec(`INSERT INTO source_items (id,source_connection_id,stable_external_id,canonical_uri_hash,title,item_type,revision,visibility,visibility_origin,source_change,policy_state,updated_at) + VALUES ('si-route-old','conn-route','legacy-item',repeat('6',64),'末端航插线序问题.adoc.docx','file','','internal','configured_scope','unchanged','internal',$1)`, now) + exec(`INSERT INTO source_item_versions (id,source_item_id,revision,content_state,raw_record_summary,updated_at) + VALUES ('sv-route-old','si-route-old','','unknown','{}'::jsonb,$1)`, now) + exec(`INSERT INTO catalog_import_intents (source_connection_id,node_id,visibility,source_item_id,decided_by,relative_path,name,locator_hash,updated_at) + VALUES ('conn-route','node-route-file','internal','si-route-old','fixture','售后机器出厂问题整改/末端航插线序问题.adoc.docx','末端航插线序问题.adoc.docx',repeat('5',64),$1)`, now) + store := NewStore(pool) + if _, err := store.RouteCatalogFiles(ctx, governance.CatalogRouteRequest{ + ConnectionID: "conn-route", + NodeIDs: []string{"node-route-file"}, + Visibility: governance.VisibilityInternal, + TargetKnowledgeBaseID: "kb-route", + OperatorRef: "fixture", + Reason: "fixture", + Now: now, + }); err != nil { + t.Fatal(err) + } + var path, parent string + if err := pool.QueryRow(ctx, `SELECT normalized_path,parent_external_id FROM source_items WHERE id='si-route-old'`).Scan(&path, &parent); err != nil { + t.Fatal(err) + } + if path != "售后机器出厂问题整改/末端航插线序问题.adoc.docx" || parent != "node-route-folder" { + t.Fatalf("route did not persist the source directory: path=%q parent=%q", path, parent) + } + // 已有值不被后续路由覆盖（现有目录精确匹配）。 + exec(`UPDATE source_items SET normalized_path='既有登记/保留路径',parent_external_id='node-existing' WHERE id='si-route-old'`) + if _, err := store.RouteCatalogFiles(ctx, governance.CatalogRouteRequest{ + ConnectionID: "conn-route", + NodeIDs: []string{"node-route-file"}, + Visibility: governance.VisibilityInternal, + TargetKnowledgeBaseID: "kb-route", + OperatorRef: "fixture", + Reason: "fixture", + Now: now.Add(time.Minute), + }); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT normalized_path,parent_external_id FROM source_items WHERE id='si-route-old'`).Scan(&path, &parent); err != nil { + t.Fatal(err) + } + if path != "既有登记/保留路径" || parent != "node-existing" { + t.Fatalf("existing directory was overwritten: path=%q parent=%q", path, parent) + } +} diff --git a/internal/platform/postgres/monitor_tasks.go b/internal/platform/postgres/monitor_tasks.go index 7885c68..3d88794 100644 --- a/internal/platform/postgres/monitor_tasks.go +++ b/internal/platform/postgres/monitor_tasks.go @@ -16,23 +16,27 @@ type MonitorTaskEvent struct { Detail string `json:"detail"` } type MonitorTask struct { - History []MonitorTaskEvent `json:"history"` - ID string `json:"id"` - Title string `json:"title"` - Path string `json:"path"` - SourceID string `json:"source_id"` - SourceName string `json:"source_name"` - TargetScope string `json:"target_scope"` - TargetKB string `json:"target_kb"` - Action string `json:"action"` - Stage string `json:"stage"` - Status string `json:"status"` - Detail string `json:"detail"` - UpdatedAt string `json:"updated_at"` - RetryKind string `json:"retry_kind"` - FailureID string `json:"failure_id"` - RetryCount int `json:"retry_count"` - VersionID string `json:"version_id"` + History []MonitorTaskEvent `json:"history"` + ID string `json:"id"` + Title string `json:"title"` + Path string `json:"path"` + // NodeID/ParentNodeID let the UI return to the exact source directory; + // both come from the catalog snapshot bound to THIS source item. + NodeID string `json:"node_id"` + ParentNodeID string `json:"parent_node_id"` + SourceID string `json:"source_id"` + SourceName string `json:"source_name"` + TargetScope string `json:"target_scope"` + TargetKB string `json:"target_kb"` + Action string `json:"action"` + Stage string `json:"stage"` + Status string `json:"status"` + Detail string `json:"detail"` + UpdatedAt string `json:"updated_at"` + RetryKind string `json:"retry_kind"` + FailureID string `json:"failure_id"` + RetryCount int `json:"retry_count"` + VersionID string `json:"version_id"` } type MonitorTaskPage struct { Items []MonitorTask `json:"items"` @@ -42,10 +46,18 @@ type MonitorTaskPage struct { Sources []MonitorOption `json:"sources"` Targets []MonitorOption `json:"targets"` Retryable int `json:"retryable"` + // Bindings is the effective (persisted, env-fallback aware) target for each + // scope. The UI marks rows/options against it; counts stay this page's totals. + Bindings MonitorBindings `json:"bindings"` } type MonitorOption struct { - ID string `json:"id"` - Name string `json:"name"` + ID string `json:"id"` + Name string `json:"name"` + Scope string `json:"scope,omitempty"` +} +type MonitorBindings struct { + Internal string `json:"internal"` + External string `json:"external"` } // Reduce publication versions before filtering; totals and pages share one snapshot. @@ -62,7 +74,11 @@ const monitorTasksCTE = `WITH history_publications AS ( WHERE p.superseded_at IS NULL ORDER BY v.source_item_id,p.test_kb_id,p.created_at DESC,p.id DESC ), base AS ( - SELECT p.id,si.title,si.normalized_path AS path,c.id AS source_id,c.name AS source_name, + SELECT p.id,si.title, + COALESCE(NULLIF(si.normalized_path,''),ci.relative_path,'') AS path, + COALESCE(NULLIF(si.parent_external_id,''),cn.parent_node_id,'') AS parent_node_id, + COALESCE(ci.node_id,'') AS node_id, + c.id AS source_id,c.name AS source_name, p.target_scope,p.test_kb_id AS target_kb,p.source_item_version_id AS version_id,p.retry_count, GREATEST(p.updated_at,j.updated_at,f.created_at) AS updated_at, p.processing_state,p.ingest_state,p.retrieval_state,p.last_error,p.knowledge_id, @@ -72,6 +88,13 @@ const monitorTasksCTE = `WITH history_publications AS ( COALESCE(j.type,'') AS job_type,COALESCE(j.state,'') AS job_state,COALESCE(j.attempt,0) AS job_attempt,j.available_at, COALESCE(f.id,'') AS failure_id,COALESCE(f.job_type,'') AS failure_type,COALESCE(NULLIF(f.last_error,''),f.error_code,'') AS failure_error FROM latest p JOIN source_items si ON si.id=p.source_item_id JOIN source_connections c ON c.id=si.source_connection_id + -- 历史行路径为空时，只用「同一连接 + 同一 item」的取件意图补齐：不按文件名猜配。 + LEFT JOIN LATERAL(SELECT intent.relative_path,intent.node_id FROM catalog_import_intents intent + WHERE intent.source_item_id=si.id AND intent.source_connection_id=si.source_connection_id + ORDER BY intent.updated_at DESC,intent.node_id LIMIT 1) ci ON TRUE + LEFT JOIN LATERAL(SELECT node.parent_node_id FROM source_catalog_heads head + JOIN source_catalog_nodes node ON node.snapshot_id=head.snapshot_id + WHERE head.source_connection_id=si.source_connection_id AND node.node_id=ci.node_id LIMIT 1) cn ON TRUE LEFT JOIN content_objects co ON co.id=p.content_object_id LEFT JOIN LATERAL(SELECT j.type,j.state,j.attempt,j.available_at,j.updated_at FROM jobs j WHERE (j.payload->>'publication_id'=p.id OR (j.type='artifact.fetch' AND j.payload->>'version_id'=p.source_item_version_id)) @@ -101,7 +124,7 @@ const monitorTasksCTE = `WITH history_publications AS ( ELSE 'waiting' END AS stage FROM base ), projected AS ( - SELECT id,title,path,source_id,source_name,target_scope,target_kb,version_id,retry_count,updated_at,action,stage,status,failure_id, + SELECT id,title,path,parent_node_id,node_id,source_id,source_name,target_scope,target_kb,version_id,retry_count,updated_at,action,stage,status,failure_id, CASE WHEN status<>'attention' THEN '' WHEN action='remove' THEN CASE WHEN job_type='test.remove' AND job_state='failed' AND retrieval_state IN ('failed','disabling') AND policy_state='blocked' THEN 'remove' ELSE '' END WHEN failure_type='artifact.fetch' THEN 'fetch' @@ -130,7 +153,7 @@ func (s *Store) ListMonitorTasks(ctx context.Context, f MonitorTaskFilter) (Moni return MonitorTaskPage{}, errors.New("invalid monitor pagination") } var raw []byte - err := s.db.QueryRow(ctx, monitorTasksCTE+`SELECT jsonb_build_object('items',COALESCE((SELECT jsonb_agg(to_jsonb(x)||jsonb_build_object('history',COALESCE((SELECT jsonb_agg(e ORDER BY e.at DESC,e.kind,e.detail) FROM (SELECT * FROM history_events ORDER BY at DESC,kind,detail LIMIT 50)e),'[]'::jsonb)) ORDER BY x.updated_at DESC,x.id) FROM (SELECT * FROM filtered ORDER BY updated_at DESC,id LIMIT $8 OFFSET $9)x),'[]'::jsonb),'total',(SELECT count(*) FROM filtered),'limit',$8::int,'offset',$9::int,'retryable',(SELECT count(*) FROM filtered WHERE retry_kind<>''),'sources',COALESCE((SELECT jsonb_agg(x ORDER BY x.name) FROM (SELECT DISTINCT source_id AS id,source_name AS name FROM projected)x),'[]'::jsonb),'targets',COALESCE((SELECT jsonb_agg(x ORDER BY x.name) FROM (SELECT DISTINCT target_kb AS id,target_kb AS name FROM projected)x),'[]'::jsonb))`, f.Search, f.Source, f.Scope, f.Action, f.Status, f.Stage, f.ID, f.Limit, f.Offset, f.Target, f.Since, f.Until).Scan(&raw) + err := s.db.QueryRow(ctx, monitorTasksCTE+`SELECT jsonb_build_object('items',COALESCE((SELECT jsonb_agg(to_jsonb(x)||jsonb_build_object('history',COALESCE((SELECT jsonb_agg(e ORDER BY e.at DESC,e.kind,e.detail) FROM (SELECT * FROM history_events ORDER BY at DESC,kind,detail LIMIT 50)e),'[]'::jsonb)) ORDER BY x.updated_at DESC,x.id) FROM (SELECT * FROM filtered ORDER BY updated_at DESC,id LIMIT $8 OFFSET $9)x),'[]'::jsonb),'total',(SELECT count(*) FROM filtered),'limit',$8::int,'offset',$9::int,'retryable',(SELECT count(*) FROM filtered WHERE retry_kind<>''),'sources',COALESCE((SELECT jsonb_agg(x ORDER BY x.name) FROM (SELECT DISTINCT source_id AS id,source_name AS name FROM projected)x),'[]'::jsonb),'targets',COALESCE((SELECT jsonb_agg(x ORDER BY x.scope,x.id) FROM (SELECT target_kb AS id,MIN(target_scope) AS scope FROM projected GROUP BY target_kb)x),'[]'::jsonb))`, f.Search, f.Source, f.Scope, f.Action, f.Status, f.Stage, f.ID, f.Limit, f.Offset, f.Target, f.Since, f.Until).Scan(&raw) if err != nil { return MonitorTaskPage{}, err } diff --git a/internal/platform/postgres/policy_intents.go b/internal/platform/postgres/policy_intents.go index e0e6e92..e73aca9 100644 --- a/internal/platform/postgres/policy_intents.go +++ b/internal/platform/postgres/policy_intents.go @@ -118,11 +118,11 @@ ORDER BY head.activated_at DESC NULLS LAST LIMIT 1 FOR UPDATE OF node`, request. if _, err := tx.Exec(ctx, `INSERT INTO source_items( id,source_connection_id,stable_external_id,canonical_uri_hash,title,item_type,revision, visibility,visibility_origin,source_change,absence_streak,identity_confidence, - policy_state,created_at,updated_at) -VALUES($1,$2,$3,$4,$5,'file','',$7,'configured_scope','unchanged',0,'object_id',$6,NOW(),NOW())`, + policy_state,normalized_path,parent_external_id,created_at,updated_at) +VALUES($1,$2,$3,$4,$5,'file','',$7,'configured_scope','unchanged',0,'object_id',$6,$8,$9,NOW(),NOW())`, newItemID, request.ConnectionID, "cat:"+file.NodeID, sha256Hex("policy:"+request.ConnectionID+"\x00"+file.NodeID), - file.Name, policy, itemVisibilityFor(request.Visibility)); err != nil { + file.Name, policy, itemVisibilityFor(request.Visibility), file.RelativePath, file.ParentNodeID); err != nil { return false, false, 0, false, fmt.Errorf("materialize policy source item: %w", err) } if _, err := tx.Exec(ctx, `UPDATE catalog_import_intents SET source_item_id=$3 WHERE source_connection_id=$1 AND node_id=$2`, @@ -131,6 +131,15 @@ VALUES($1,$2,$3,$4,$5,'file','',$7,'configured_scope','unchanged',0,'object_id', } itemID = newItemID } + // 历史行（当时没写来源路径）只补空值：意图已把 item 绑定到本连接的这个 node， + // 身份可证；已有路径绝不被覆盖，避免把下架连接重绑成当前来源。 + if _, err := tx.Exec(ctx, `UPDATE source_items SET + normalized_path=CASE WHEN normalized_path='' THEN $2 ELSE normalized_path END, + parent_external_id=CASE WHEN parent_external_id='' THEN $3 ELSE parent_external_id END, + updated_at=CASE WHEN normalized_path='' OR parent_external_id='' THEN NOW() ELSE updated_at END +WHERE id=$1`, itemID, file.RelativePath, file.ParentNodeID); err != nil { + return false, false, 0, false, fmt.Errorf("persist policy source path: %w", err) + } // Bind the snapshot nodes too (new or reused item): the browser's // 已导入/已解析 counters join source_catalog_nodes.source_item_id; without // this write every catalog-only source keeps showing "尚未导入" even after diff --git a/internal/readiness/weknora.go b/internal/readiness/weknora.go new file mode 100644 index 0000000..5de01bd --- /dev/null +++ b/internal/readiness/weknora.go @@ -0,0 +1,111 @@ +package readiness + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "knowledge-governance/internal/platform/weknora" +) + +// WeKnora 的 readiness 探测必须与业务执行链看同一个目标：网页保存的端点、 +// 当前生效的知识库绑定、同一份 API Key。任何一项在启动时被捕获成常量， +// 都会在管理员改绑之后继续探测旧库（F13：旧库 404 被当成整个服务不可用）。 + +var ( + ErrNotConfigured = errors.New("WeKnora 就绪探测未配置") + ErrKnowledgeBaseMissing = errors.New("WeKnora 探测目标知识库不存在") + ErrNoPermission = errors.New("WeKnora 探测目标知识库无读取权限") +) + +type Target struct { + BaseURL string + KnowledgeBaseID string + APIKey string +} + +type Result struct { + // Target 是脱敏后的探测对象（实例 + 短 ID），用于界面说明“探的是谁”。 + Target string + Err error +} + +// Prober 每次 Check 都重新解析目标并新建客户端：端点、凭据、绑定都允许 +// 在进程运行期间变化，不能缓存成启动快照。 +type Prober struct { + Resolve func(context.Context) (Target, error) + Timeout time.Duration + AllowDockerHTTP bool +} + +func (prober Prober) Check(ctx context.Context) Result { + if prober.Resolve == nil { + return Result{Err: ErrNotConfigured} + } + target, err := prober.Resolve(ctx) + if err != nil { + return Result{Err: err} + } + described := describe(target) + if strings.TrimSpace(target.BaseURL) == "" || strings.TrimSpace(target.KnowledgeBaseID) == "" || strings.TrimSpace(target.APIKey) == "" { + return Result{Target: described, Err: ErrNotConfigured} + } + timeout := prober.Timeout + if timeout <= 0 || timeout > 10*time.Second { + timeout = 10 * time.Second + } + client, err := weknora.NewClient(weknora.ClientConfig{ + BaseURL: target.BaseURL, + APIKey: target.APIKey, + Timeout: timeout, + AllowDockerHTTP: prober.AllowDockerHTTP, + }) + if err != nil { + return Result{Target: described, Err: err} + } + if _, _, err := client.ListKnowledge(ctx, target.KnowledgeBaseID, 1, 1); err != nil { + return Result{Target: described, Err: classify(target, described, err)} + } + return Result{Target: described} +} + +func classify(target Target, described string, err error) error { + var httpErr *weknora.HTTPError + if errors.As(err, &httpErr) { + switch httpErr.StatusCode { + case http.StatusNotFound: + return fmt.Errorf("%w（%s）", ErrKnowledgeBaseMissing, described) + case http.StatusUnauthorized, http.StatusForbidden: + return fmt.Errorf("%w（%s）", ErrNoPermission, described) + case http.StatusTooManyRequests: + return fmt.Errorf("目标知识库读取被限流（HTTP 429，%s）", described) + } + } + return err +} + +// describe 只保留实例主机与知识库短 ID：诊断够用，不输出完整 UUID、路径或凭据。 +func describe(target Target) string { + host := "" + if parsed, err := url.Parse(strings.TrimSpace(target.BaseURL)); err == nil && parsed.Host != "" { + host = parsed.Host + } + id := strings.TrimSpace(target.KnowledgeBaseID) + if len(id) > 8 { + id = id[:8] + } + switch { + case host == "" && id == "": + return "" + case host == "": + return "知识库 " + id + case id == "": + return "实例 " + host + default: + return "实例 " + host + " · 知识库 " + id + } +} diff --git a/internal/readiness/weknora_test.go b/internal/readiness/weknora_test.go new file mode 100644 index 0000000..4efd1c1 --- /dev/null +++ b/internal/readiness/weknora_test.go @@ -0,0 +1,138 @@ +package readiness + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +// fakeWeKnora answers /health and /api/v1/knowledge-bases/{kb}/knowledge. +// 404 is how WeKnora says "this knowledge base does not exist" (F13 的旧 env 绑定). +func fakeWeKnora(t *testing.T, known map[string]int, wantKey string) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("X-API-Key"); wantKey != "" && got != wantKey { + w.WriteHeader(http.StatusUnauthorized) + return + } + switch { + case r.URL.Path == "/health": + _, _ = w.Write([]byte(`ok`)) + case strings.HasPrefix(r.URL.Path, "/api/v1/knowledge-bases/"): + parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/api/v1/knowledge-bases/"), "/") + total, ok := known[parts[0]] + if !ok || len(parts) != 2 || parts[1] != "knowledge" { + w.WriteHeader(http.StatusNotFound) + return + } + _, _ = fmt.Fprintf(w, `{"success":true,"total":%d,"data":[]}`, total) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(server.Close) + return server +} + +func TestProbeFollowsTheCurrentBindingWithinOneProcess(t *testing.T) { + oldKB := "1b52c140-2e18-4676-afc5-4c7d8d190a8c" + currentKB := "0c84a098-d3ed-4b82-a165-e6de9691d6a8" + server := fakeWeKnora(t, map[string]int{currentKB: 4}, "key-1") + var mu sync.Mutex + binding := oldKB + prober := Prober{ + Timeout: 2 * time.Second, + Resolve: func(context.Context) (Target, error) { + mu.Lock() + defer mu.Unlock() + return Target{BaseURL: server.URL, KnowledgeBaseID: binding, APIKey: "key-1"}, nil + }, + } + first := prober.Check(context.Background()) + if !errors.Is(first.Err, ErrKnowledgeBaseMissing) { + t.Fatalf("old binding probe = %v, want ErrKnowledgeBaseMissing", first.Err) + } + if !strings.Contains(first.Target, "1b52c140") || strings.Contains(first.Target, oldKB) { + t.Fatalf("probe target must name the probed KB in short form only: %q", first.Target) + } + mu.Lock() + binding = currentKB + mu.Unlock() + second := prober.Check(context.Background()) + if second.Err != nil { + t.Fatalf("current binding probe = %v, want ready", second.Err) + } + if !strings.Contains(second.Target, "0c84a098") { + t.Fatalf("probe target did not follow the new binding: %q", second.Target) + } +} + +func TestProbeReResolvesEndpointAndCredentialOnEveryCheck(t *testing.T) { + kb := "0c84a098-d3ed-4b82-a165-e6de9691d6a8" + denied := fakeWeKnora(t, map[string]int{kb: 4}, "new-key") + allowed := fakeWeKnora(t, map[string]int{kb: 4}, "new-key") + var mu sync.Mutex + endpoint, key := denied.URL, "old-key" + prober := Prober{ + Timeout: 2 * time.Second, + Resolve: func(context.Context) (Target, error) { + mu.Lock() + defer mu.Unlock() + return Target{BaseURL: endpoint, KnowledgeBaseID: kb, APIKey: key}, nil + }, + } + if err := prober.Check(context.Background()).Err; !errors.Is(err, ErrNoPermission) { + t.Fatalf("stale credential probe = %v, want ErrNoPermission", err) + } + mu.Lock() + endpoint, key = allowed.URL, "new-key" + mu.Unlock() + if err := prober.Check(context.Background()).Err; err != nil { + t.Fatalf("rotated endpoint/credential probe = %v, want ready", err) + } +} + +func TestProbeDistinguishesMissingConfigurationFromUnreachable(t *testing.T) { + missing := Prober{Resolve: func(context.Context) (Target, error) { + return Target{BaseURL: "http://127.0.0.1:1", KnowledgeBaseID: ""}, nil + }} + if !errors.Is(missing.Check(context.Background()).Err, ErrNotConfigured) { + t.Fatalf("blank binding must be ErrNotConfigured, got %v", missing.Check(context.Background()).Err) + } + unreachable := Prober{Timeout: 200 * time.Millisecond, Resolve: func(context.Context) (Target, error) { + return Target{BaseURL: "http://127.0.0.1:1", KnowledgeBaseID: "kb-1", APIKey: "key"}, nil + }} + err := unreachable.Check(context.Background()).Err + if err == nil || errors.Is(err, ErrNotConfigured) || errors.Is(err, ErrKnowledgeBaseMissing) { + t.Fatalf("connection refused must stay a transport failure, got %v", err) + } +} + +func TestProbeSurfaces404AsMissingKnowledgeBaseNotServiceDown(t *testing.T) { + server := fakeWeKnora(t, map[string]int{}, "key") + prober := Prober{Timeout: 2 * time.Second, Resolve: func(context.Context) (Target, error) { + return Target{BaseURL: server.URL, KnowledgeBaseID: "gone-kb", APIKey: "key"}, nil + }} + err := prober.Check(context.Background()).Err + if !errors.Is(err, ErrKnowledgeBaseMissing) { + t.Fatalf("404 = %v, want ErrKnowledgeBaseMissing", err) + } + if strings.Contains(err.Error(), "key") { + t.Fatalf("probe error leaked the credential: %v", err) + } +} + +func TestProbeRejectsAnUnusableResolverResult(t *testing.T) { + prober := Prober{Resolve: func(context.Context) (Target, error) { + return Target{}, errors.New("settings unavailable") + }} + if err := prober.Check(context.Background()).Err; err == nil || errors.Is(err, ErrNotConfigured) { + t.Fatalf("resolver failure must be reported, got %v", err) + } +} diff --git a/scripts/ci_checks.py b/scripts/ci_checks.py index 1e7cf18..2a7937b 100644 --- a/scripts/ci_checks.py +++ b/scripts/ci_checks.py @@ -37,7 +37,7 @@ from typing import Callable ROOT = Path(__file__).resolve().parent.parent WEB = ROOT / "web" REPORT_DEFAULT = ROOT / "ci-report.json" -POLICY_REVISION = "ci-checks/4" # 判据变更时改这里，报告里跟着变 +POLICY_REVISION = "ci-checks/5" # 判据变更时改这里，报告里跟着变 MANIFESTS = ["go.mod", "go.sum", "web/package.json", "web/package-lock.json"] SUCCESS, FAILURE, SKIPPED, NOT_RUN = "success", "failure", "skipped", "not_run" @@ -255,6 +255,9 @@ def checks() -> list[Check]: # 2026-10-29 那次演练 exit 5 就是"脚本要的老文案只有一条代码路径会打印"。 Check("takeover-log-contract", "接管日志契约（认领/决策行判据 + 正负例自检）", [py3, "scripts/takeover_log_contract.py", "--self-test"]), + # F12：运行镜像少 COPY 一个模块只会在真实刷新时炸；静态闭包比对进日常门禁。 + Check("runtime-scripts", "运行镜像脚本依赖闭包（缺模块负例必须红）", + [py3, "scripts/test_runtime_script_packaging.py"]), Check("version", "CHANGELOG 与版本记录（四类小节无空栏目 + 正式 tag 与提交内版本一致）", [py3, "scripts/check_changelog.py", "--tags"]), Check("security-secrets", "密钥泄漏扫描（含扫描器自检）", @@ -524,6 +527,7 @@ CHECK_MANIFEST: dict[str, tuple[str, ...]] = { "tests-web": ("local", "ci"), "contracts": ("local", "ci"), "takeover-log-contract": ("local", "ci"), + "runtime-scripts": ("local", "ci"), "version": ("local", "ci"), "security-secrets": ("local", "ci"), "security-deps": ("local", "ci"), diff --git a/scripts/test_runtime_script_packaging.py b/scripts/test_runtime_script_packaging.py new file mode 100644 index 0000000..3b952fa --- /dev/null +++ b/scripts/test_runtime_script_packaging.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""运行镜像必须带齐脚本闭包：Dockerfile COPY 漏一个模块 ⇒ 线上入口必败。 + +F12 的真实现象是 live_catalog_refresh 能导入、psql 在、根身份正常， +但 `from dws_dingtalk_scan import walk` 抛 ModuleNotFoundError。只检查主脚本 +自己的 import 不够：被依赖的模块也可能再 import 兄弟模块。这里按源码算出 +闭包，再与 Dockerfile 实际 COPY 进 /app/scripts 的文件比对。 + +纯静态检查：不起容器、不连库、不需要凭据。 +""" +from __future__ import annotations + +import ast +import re +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +SCRIPTS = ROOT / "scripts" +APP_SCRIPTS = "/app/scripts" + + +def copied_scripts(dockerfile_text: str) -> set[str]: + """Dockerfile 里 COPY 到 /app/scripts 的源文件名（含续行）。""" + logical = re.sub(r"\\\r?\n\s*", " ", dockerfile_text) + copied: set[str] = set() + for line in logical.splitlines(): + parts = line.split() + targets = [part.rstrip("/") for part in parts] + if not parts or parts[0].upper() != "COPY" or APP_SCRIPTS not in targets: + continue + sources = parts[1:targets.index(APP_SCRIPTS)] + for source in sources: + name = Path(source).name + if name.endswith(".py"): + copied.add(name) + return copied + + +def local_imports(path: Path, siblings: set[str]) -> set[str]: + """该脚本 import 的同目录模块（只认 scripts/ 下真实存在的兄弟文件）。""" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + found: set[str] = set() + for node in ast.walk(tree): + names: list[str] = [] + if isinstance(node, ast.Import): + names = [alias.name for alias in node.names] + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + names = [node.module] + for name in names: + top = name.split(".")[0] + if f"{top}.py" in siblings: + found.add(f"{top}.py") + return found + + +def dependency_closure(scripts_dir: Path, copied: set[str]) -> set[str]: + siblings = {entry.name for entry in scripts_dir.glob("*.py")} + needed = {name for name in copied if (scripts_dir / name).is_file()} + pending = list(needed) + while pending: + current = pending.pop() + for dependency in local_imports(scripts_dir / current, siblings) - needed: + needed.add(dependency) + pending.append(dependency) + return needed + + +def missing_from_image(scripts_dir: Path, dockerfile: Path) -> list[str]: + copied = copied_scripts(dockerfile.read_text(encoding="utf-8")) + return sorted(dependency_closure(scripts_dir, copied) - copied) + + +class RuntimeScriptPackagingTests(unittest.TestCase): + def test_worker_image_ships_every_local_dependency(self): + missing = missing_from_image(SCRIPTS, ROOT / "Dockerfile") + # RED 证据：F12 就是这一行缺失（dws_dingtalk_scan.py），不是宿主环境问题。 + self.assertEqual(missing, [], f"运行镜像缺少脚本依赖：{missing}") + + def test_missing_module_negative_fixture_is_rejected(self): + """去掉扫描模块的 COPY 行必须被判红——证明这个断言真的会因为漏拷而失败。""" + dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8") + broken = dockerfile.replace(" scripts/dws_dingtalk_scan.py", "") + self.assertIn("scripts/live_catalog_refresh.py", broken) + self.assertNotIn("dws_dingtalk_scan.py", broken) + with TemporaryDockerfile(broken) as path: + self.assertEqual(missing_from_image(SCRIPTS, path), ["dws_dingtalk_scan.py"]) + + def test_third_party_and_runtime_paths_are_not_counted_as_shipped(self): + siblings = {entry.name for entry in SCRIPTS.glob("*.py")} + imported = local_imports(SCRIPTS / "artifact_fetch.py", siblings) + self.assertTrue({"axls_to_xlsx.py", "dws_error_map.py"} <= imported) + self.assertNotIn("requests.py", siblings) + + +class TemporaryDockerfile: + def __init__(self, content: str): + self.content = content + self.path = SCRIPTS / ".test-dockerfile-missing-module" + + def __enter__(self) -> Path: + self.path.write_text(self.content, encoding="utf-8") + return self.path + + def __exit__(self, *_exc: object) -> None: + self.path.unlink(missing_ok=True) + + +if __name__ == "__main__": + unittest.main(verbosity=2) \ No newline at end of file diff --git a/web/src/App.tsx b/web/src/App.tsx index ae91167..7241e4b 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -2504,6 +2504,10 @@ export default function App() { return initialWorkspace } }) + // F14 往返定位：两个模块之间只传「可证明的身份」（来源连接 + 节点/文件名）， + // 不共享业务状态，也不在 URL 或全局写入路径。 + const [monitorFocus, setMonitorFocus] = useState<{ source: string; q: string } | null>(null) + const [sourceFocus, setSourceFocus] = useState<{ connectionID: string; nodeID: string } | null>(null) const load = useCallback(async (signal?: AbortSignal) => { setLoading(true) @@ -2968,9 +2972,21 @@ export default function App() { onOpen={(moduleId) => setWorkspace((current) => openModule(current, moduleId))} > {workspace.activeTab === 'dashboard' ? ( - + { + setMonitorFocus({ source: source.connection_id, q: node.name }) + setWorkspace((current) => openModule(current, 'tasks')) + }} + /> ) : workspace.activeTab === 'tasks' ? ( - + { + setSourceFocus({ connectionID: sourceID, nodeID: parentNodeID }) + setWorkspace((current) => openModule(current, 'dashboard')) + }} + /> ) : workspace.activeTab === 'sources' ? ( ) : workspace.activeTab === 'settings' ? ( diff --git a/web/src/SourceCatalogBrowser.locator.test.tsx b/web/src/SourceCatalogBrowser.locator.test.tsx new file mode 100644 index 0000000..4d53023 --- /dev/null +++ b/web/src/SourceCatalogBrowser.locator.test.tsx @@ -0,0 +1,96 @@ +// @vitest-environment jsdom +import { act } from 'react' +import { createRoot } from 'react-dom/client' +import type { Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { SourceCatalogBrowserView } from './SourceCatalogBrowser' +import type { CatalogNode, CatalogNodePage, CatalogSource } from './sourceCatalog' + +// F14 的界面入口：已导入的文件能从这里直接跳到解析监控（不再靠手抄文件名）。 +const source: CatalogSource = { + connection_id: 'conn-atom', + source_type: 'dingtalk', + name: 'ATOM知识库', + root_node_id: 'root-1', + folder_count: 1, + file_count: 1, + observed_at: '2026-10-26T10:41:01Z', + snapshot_only: true, + snapshot_status: 'ready', + last_refresh_duration_ms: 0, +} +const file: CatalogNode = { + node_id: 'node-atom-file', + parent_node_id: 'node-atom-folder', + node_type: 'file', + name: '末端航插线序问题.adoc.docx', + direct_child_count: 0, + subtree_file_count: 1, + subtree_bytes: 1024, + size_bytes: 1024, + visibility: 'internal', + imported_count: 1, + parsed_count: 1, + active_count: 0, + failed_count: 0, + routable_count: 1, +} +const page: CatalogNodePage = { + current: { + ...file, + node_id: 'node-atom-folder', + node_type: 'folder', + name: '售后机器出厂问题整改', + parent_node_id: undefined, + }, + breadcrumbs: [ + { node_id: 'root-1', name: 'ATOM知识库' }, + { node_id: 'node-atom-folder', name: '售后机器出厂问题整改' }, + ], + items: [file], + total: 1, + limit: 50, + offset: 0, +} + +let container: HTMLDivElement +let root: Root +beforeEach(() => { + ;(globalThis as unknown as { IS_REACT_ACT_ENVIRONMENT: boolean }).IS_REACT_ACT_ENVIRONMENT = true + container = document.createElement('div') + document.body.appendChild(container) + root = createRoot(container) +}) +afterEach(() => { + act(() => root.unmount()) + container.remove() +}) + +describe('source directory to processing handoff', () => { + it('offers 查看处理 for an imported file and passes the exact node identity', () => { + const onViewProcessing = vi.fn() + act(() => + root.render( + + ) + ) + const button = [...container.querySelectorAll('button')].find((item) => item.textContent === '查看处理') + expect(button).toBeTruthy() + act(() => button?.click()) + expect(onViewProcessing).toHaveBeenCalledWith(file) + }) +}) diff --git a/web/src/SourceCatalogBrowser.refreshState.test.tsx b/web/src/SourceCatalogBrowser.refreshState.test.tsx new file mode 100644 index 0000000..5087af2 --- /dev/null +++ b/web/src/SourceCatalogBrowser.refreshState.test.tsx @@ -0,0 +1,100 @@ +// @vitest-environment jsdom +import { act } from 'react' +import { createRoot } from 'react-dom/client' +import type { Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { SourceCatalogBrowserView } from './SourceCatalogBrowser' +import type { CatalogSource } from './sourceCatalog' +import type { Job } from './jobs' + +// F16 的挂载级回归：同一个组件从 running 走到 failed/completed， +// 终态不能残留“后台运行”，失败要能看到时间/旧快照/原因/作业号并可重试。 +const source: CatalogSource = { + connection_id: 'conn-atom', + source_type: 'dingtalk', + name: 'ATOM知识库', + root_node_id: 'root-1', + folder_count: 1, + file_count: 191, + observed_at: '2026-10-26T10:41:01Z', + snapshot_only: true, + snapshot_status: 'ready', + last_refresh_duration_ms: 12000, +} + +const job = (state: Job['state'], extra: Partial = {}): Job => ({ + id: 'job-6e18ea86dce10d5472978573a4022a1a', + type: 'source.catalog.refresh', + payload: { connection_id: 'conn-atom', progress: 0, stage: 'queued', message: '等待后台刷新' }, + state, + priority: 0, + available_at: '2026-11-04T14:00:00Z', + attempt: 3, + max_attempts: 3, + last_error: 'ModuleNotFoundError: dws_dingtalk_scan', + idempotency_key: 'source.catalog.refresh:conn-atom', + created_at: '2026-11-04T14:00:00Z', + updated_at: '2026-11-04T14:03:30Z', + ...extra, +}) + +let container: HTMLDivElement +let root: Root +const render = (refreshJob: Job) => + act(() => + root.render( + + ) + ) + +beforeEach(() => { + ;(globalThis as unknown as { IS_REACT_ACT_ENVIRONMENT: boolean }).IS_REACT_ACT_ENVIRONMENT = true + container = document.createElement('div') + document.body.appendChild(container) + root = createRoot(container) +}) + +afterEach(() => { + act(() => root.unmount()) + container.remove() +}) + +describe('SourceCatalogBrowser refresh terminal states', () => { + it('drops the background-running text when the same card turns failed', () => { + render(job('running')) + expect(container.textContent).toContain('后台运行') + render(job('failed')) + expect(container.textContent).not.toContain('后台运行') + expect(container.textContent).toContain('实时刷新失败') + expect(container.textContent).toContain('job-6e18ea86dce10d5472978573a4022a1a') + expect(container.textContent).toContain('旧快照') + expect(container.textContent).toContain('dws_dingtalk_scan') + const retry = [...container.querySelectorAll('button')].find((button) => button.textContent === '重试刷新') + expect(retry).toBeTruthy() + }) + + it('keeps the retry entry disabled while running and hides it when completed', () => { + render(job('running')) + const retryRunning = [...container.querySelectorAll('button')].find((button) => button.textContent === '重试刷新') + expect(retryRunning?.disabled).toBe(true) + render(job('completed', { last_error: '' })) + expect(container.textContent).not.toContain('后台运行') + expect([...container.querySelectorAll('button')].some((button) => button.textContent === '重试刷新')).toBe(false) + expect(container.textContent).toContain('实时快照已完成') + }) +}) diff --git a/web/src/SourceCatalogBrowser.tsx b/web/src/SourceCatalogBrowser.tsx index e6961f5..5de4be0 100644 --- a/web/src/SourceCatalogBrowser.tsx +++ b/web/src/SourceCatalogBrowser.tsx @@ -46,6 +46,14 @@ const scope = (value: string) => : '未选择' const progress = (node: CatalogNode) => node.imported_count ? `已解析 ${node.parsed_count} / 已导入 ${node.imported_count}` : '尚未导入' +const time = (value?: string) => (value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '未记录') +// 失败原因可能带来源地址或凭据路径；界面只展示脱敏摘要，详细值留在作业表。 +const sanitizeRefreshReason = (value?: string) => + (value || '') + .replace(/https?:\/\/\S+/g, '[地址已隐藏]') + .replace(/\/run\/source-secrets\/\S+/g, '[凭据路径已隐藏]') + .replace(/\b(?:token|api[-_]?key|password|authorization|cookie|secret)=\S+/gi, '[凭据已隐藏]') + .slice(0, 160) interface ViewProps { sources: CatalogSource[] @@ -79,6 +87,7 @@ interface ViewProps { refreshHidden?: boolean onToggleRefreshHidden?: () => void onFiltersChanged?: () => void + onViewProcessing?: (node: CatalogNode) => void } export function SourceCatalogBrowserView({ sources, @@ -101,6 +110,7 @@ export function SourceCatalogBrowserView({ refreshHidden = false, onToggleRefreshHidden, onFiltersChanged, + onViewProcessing, }: ViewProps) { const [selectedIDs, setSelectedIDs] = useState([]) const [appliedPolicies, setAppliedPolicies] = useState>({}) @@ -204,13 +214,22 @@ export function SourceCatalogBrowserView({ - {refreshProgress}%{refreshEstimate ? '（估算）' : ''} · 后台运行，可以切换页面 + {refreshActive + ? `${refreshProgress}%${refreshEstimate ? '（估算）' : ''} · 后台运行，可以切换页面` + : refreshJob.state === 'completed' + ? `完成于 ${time(refreshJob.updated_at)}` + : `失败于 ${time(refreshJob.updated_at)} · 旧快照 ${time(selectedSource?.observed_at)}`} {refreshActive && onToggleRefreshHidden ? ( ) : null} + {refreshJob.state !== 'completed' ? ( + + ) : null} {refreshActive && (refreshDirectory || refreshFiles) ? (
@@ -220,7 +239,8 @@ export function SourceCatalogBrowserView({ ) : null} {refreshJob.state === 'failed' ? (

- 刷新失败；旧快照仍然保留，目录浏览与标注不受影响。 + 刷新失败；旧快照仍然保留，目录浏览与标注不受影响。原因： + {sanitizeRefreshReason(refreshJob.last_error) || '未提供'} · 作业 {refreshJob.id}

) : null} @@ -427,6 +447,15 @@ export function SourceCatalogBrowserView({ {node.failed_count ? ( 失败 {node.failed_count} ) : null} + {onViewProcessing && node.node_type === 'file' && node.imported_count > 0 ? ( + + ) : null} {node.modified_at @@ -497,7 +526,13 @@ export function SourceCatalogBrowserView({ ) } -export function SourceCatalogBrowser() { +export function SourceCatalogBrowser({ + focus, + onViewProcessing, +}: { + focus?: { connectionID: string; nodeID: string } | null + onViewProcessing?: (source: CatalogSource, node: CatalogNode) => void +} = {}) { const [sources, setSources] = useState([]) const [selected, setSelected] = useState(null) const [parentID, setParentID] = useState('') @@ -621,6 +656,17 @@ export function SourceCatalogBrowser() { return () => controller.abort() }, [selected, parentID, offset, revision]) + // 从解析监控“在来源与目录定位”进入：选中该来源并直接打开文件所在目录。 + useEffect(() => { + if (!focus || !sources.length) return + const target = sources.find((source) => source.connection_id === focus.connectionID) + if (!target) return + setSelected(target) + setParentID(focus.nodeID || target.root_node_id) + setOffset(0) + setPage(null) + closeHistory() + }, [focus, sources]) // 有来源正在首次扫描时自动刷新列表：百分比会自己往前走，扫完那一行自动变成正式来源。 useEffect(() => { if (!sources.some((source) => source.snapshot_status === 'scanning')) return @@ -777,6 +823,7 @@ export function SourceCatalogBrowser() { refreshHidden={refreshHidden} onToggleRefreshHidden={() => setRefreshHidden((value) => !value)} onFiltersChanged={() => setRevision((value) => value + 1)} + onViewProcessing={onViewProcessing && selected ? (node) => onViewProcessing(selected, node) : undefined} /> ) } diff --git a/web/src/WeKnoraMonitor.test.tsx b/web/src/WeKnoraMonitor.test.tsx index bb5dd46..10398c1 100644 --- a/web/src/WeKnoraMonitor.test.tsx +++ b/web/src/WeKnoraMonitor.test.tsx @@ -9,6 +9,8 @@ const page: MonitorTaskPage = { id: 'tp-1', title: '维护手册.pdf', path: '产品/维护', + node_id: 'node-1', + parent_node_id: 'folder-1', source_id: 's1', source_name: '技术资料', target_scope: 'internal', @@ -27,6 +29,8 @@ const page: MonitorTaskPage = { id: 'tp-2', title: '过期资料.pdf', path: '历史', + node_id: 'node-2', + parent_node_id: 'folder-2', source_id: 's1', source_name: '技术资料', target_scope: 'external', @@ -47,6 +51,7 @@ const page: MonitorTaskPage = { offset: 0, sources: [{ id: 's1', name: '技术资料' }], targets: [], + bindings: { internal: '', external: '' }, retryable: 1, } const props = { @@ -103,3 +108,65 @@ describe('unified monitor', () => { expect(html).not.toContain('aria-label="选择过期资料.pdf"') }) }) +describe('target identity and probe attribution', () => { + it('labels same-named targets with short id and current/history state', () => { + const current = '0c84a098-d3ed-4b82-a165-e6de9691d6a8' + const historical = 'sc-ed1f812c4e5fabe817fc6351104938c4' + const withTargets = { + ...page, + items: [ + { ...page.items[0], target_kb: current }, + { ...page.items[1], target_kb: historical }, + ], + targets: [ + { id: current, name: current, scope: 'internal' }, + { id: historical, name: historical, scope: 'internal' }, + ], + bindings: { internal: current, external: '' }, + } + const html = renderToStaticMarkup() + expect(html).toContain('0c84a098') + expect(html).toContain('当前绑定') + expect(html).toContain('sc-ed1f8') + expect(html).toContain('历史') + expect(html).toContain('产品/维护') + }) + it('shows liveness and the sanitized probe target separately from readiness', () => { + const overview = { + mode: 'real', + simulated: false, + observed_at: '2026-11-05T02:00:00Z', + host: { + cpu_percent: 1, + memory_used_bytes: 1, + memory_total_bytes: 2, + disk_used_bytes: 1, + disk_total_bytes: 2, + gpu_state: 'none', + gpu_detail: '', + simulated: false, + }, + services: [], + knowledge_bases: [], + health: [ + { + key: 'weknora', + name: 'WeKnora API', + liveness: 'up', + readiness: 'blocked', + reason: 'WeKnora 探测目标知识库不存在（实例 weknora.test · 知识库 1b52c140）', + latency_ms: 3, + observed_at: '2026-11-05T02:00:00Z', + last_success_at: null, + consecutive_failures: 1, + probe_target: '实例 weknora.test · 知识库 1b52c140', + configured: true, + applicable: true, + }, + ], + } + const html = renderToStaticMarkup() + expect(html).toContain('存活 up') + expect(html).toContain('探测对象 实例 weknora.test · 知识库 1b52c140') + }) +}) diff --git a/web/src/WeKnoraMonitor.tsx b/web/src/WeKnoraMonitor.tsx index 1ad0479..e2ddfaf 100644 --- a/web/src/WeKnoraMonitor.tsx +++ b/web/src/WeKnoraMonitor.tsx @@ -8,6 +8,7 @@ import { actionLabels, stageLabels, taskStatusLabels, + targetLabel, } from './monitorTasks' import type { MonitorFilters, MonitorTask, MonitorTaskPage } from './monitorTasks' import { requeueJobFailure, retryPublicationsBatch } from './testPublications' @@ -16,7 +17,16 @@ import { useOptionalAuth } from './auth' import { WorkspaceContextTabsSetter } from './workspaceContext' import './monitorWorkbench.css' -const emptyPage: MonitorTaskPage = { items: [], total: 0, limit: 20, offset: 0, sources: [], targets: [], retryable: 0 } +const emptyPage: MonitorTaskPage = { + items: [], + total: 0, + limit: 20, + offset: 0, + sources: [], + targets: [], + bindings: { internal: '', external: '' }, + retryable: 0, +} const time = (value: string) => (value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—') const percent = (value: number | null | undefined) => (value == null ? '—' : `${value}%`) const seconds = (value: number | null | undefined) => (value == null ? '—' : `${Number(value.toFixed(1))} 秒`) @@ -51,6 +61,7 @@ interface ViewProps { onRetry: (rows: MonitorTask[]) => void onDetail: (row: MonitorTask) => void onRefresh: () => void + onLocateSource?: (sourceID: string, parentNodeID: string) => void } export function WeKnoraMonitorView(p: ViewProps) { const { overview, metrics, page, filters, selected, busy, loading } = p @@ -63,6 +74,8 @@ export function WeKnoraMonitorView(p: ViewProps) { const selectedIDs = new Set(selected.map((r) => r.id)) const allPage = eligible.length > 0 && eligible.every((r) => selectedIDs.has(r.id)) const kbName = (id: string) => overview?.knowledge_bases.find((k) => k.id === id)?.name || id + const bindings = page.bindings + const scopeBinding = filters.scope === 'external' ? bindings.external : bindings.internal return (
@@ -180,7 +193,8 @@ export function WeKnoraMonitorView(p: ViewProps) {

{s.reason || '未提供状态说明'}

- 检测：{time(s.observed_at)} · {s.latency_ms} ms + 检测：{time(s.observed_at)} · {s.latency_ms} ms · 存活 {s.liveness || '未确认'} + {s.probe_target ? ` · 探测对象 ${s.probe_target}` : ''} )) @@ -234,10 +248,23 @@ export function WeKnoraMonitorView(p: ViewProps) { {page.targets.map((t) => ( ))} +