diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a7a65d..029f621 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,32 @@ # Changelog +## 开发快照 0.1.357(取件后自动解析、定位一次性请求、worker 有效配置) + +### Fixed + +- **F17 取件成功没有自动上传解析**:`MarkArtifactFetched` 与“本地已验证内容复用”两条路径在 + 同一事务内调用 `autoStartVerifiedPublicationsTx`:仅当来源仍启用、保存策略与 + `catalog_import_intents.visibility` 仍匹配、文件名过滤未命中、publication 未 superseded、 + 目标仍是当前 `routing_kb_bindings`(env 只作未绑定兜底)时,才把 planned→pending 并入队 + 唯一 `test.upload`;确认记录与作业幂等防双入队。356 已卡 planned/verified 的记录可通过对 + 同一文件再次保存当前策略,经同一自动接续路径恢复,不手工补队列、不启动全库补扫。 + 来源页成功提示同步改为“取件校验完成后自动上传解析”,不再指向已删除的“待入库开始处理”。 +- **F14 查看处理被默认筛选清空**:来源→监控的一次性定位改为套用干净筛选(清空 + status/action/stage/since/until),已完成文件可直接出现;定位请求应用后即消费。 +- **F14 旧定位请求重放**:来源目录定位增加一次性请求身份与父级消费语义;同一旧请求不会在 + sources/revision 重载或重挂载时把用户手工切换的目录拽回,新显式定位仍生效。 +- **F13 worker 有效配置与 API 分叉**:抽出 `readiness.Resolver` 作为端点/当前绑定/凭据唯一 + 解析来源;worker 的 upload/poll/validate/remove 改用 `weknoraaccess` 动态客户端,每次调用 + 重读网页端点、持久化绑定与同一 key 文件,不再在启动时固化 env 端点/客户端。 + +### Tests + +- 真实隔离 PG:策略→路由→取件 verified→唯一上传任务、重复完成幂等、复用已验证内容、 + 356 卡住记录重存恢复;禁用来源/阻止策略/文件名过滤/superseded/绑定改变均不入队。 +- 前端挂载回归:已完成文件定位不被 `status=active` 遮蔽;定位→手工换目录→保存策略/刷新后 + 位置保持,新请求仍能定位。 +- worker 动态客户端同进程端点、凭据、绑定改变均跟随;真实 CI 与本地门禁保持不变。 + ## 开发快照 0.1.356(三模块正常入口与追溯修复) ### Fixed diff --git a/cmd/api/main.go b/cmd/api/main.go index 44c3380..06bda53 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -206,7 +206,7 @@ func run(ctx context.Context, getenv func(string) string) error { QAOpsTenantKey: qaSettings.TenantKey, QAOpsEnabled: qaSettings.Enabled, QAOpsProblem: qaSettings.Problem, - Store: postgresstore.NewStore(pool).WithLibraryTarget(config.LibraryTargetIdentity()), + Store: postgresstore.NewStore(pool).WithLibraryTarget(config.LibraryTargetIdentity()).WithRoutingDefaults(config.InternalRoutingKnowledgeBaseID, config.ExternalRoutingKnowledgeBaseID), Monitor: monitorProvider, InternalRoutingKnowledgeBaseID: config.InternalRoutingKnowledgeBaseID, ExternalRoutingKnowledgeBaseID: config.ExternalRoutingKnowledgeBaseID, diff --git a/cmd/api/readiness.go b/cmd/api/readiness.go index a430943..bff2c7c 100644 --- a/cmd/api/readiness.go +++ b/cmd/api/readiness.go @@ -21,34 +21,31 @@ func newWeKnoraReadinessResolver( 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) + resolver := readiness.Resolver{ + Endpoint: endpoint, + DefaultKB: envKnowledgeBaseID, + LoadBinding: func(resolveCtx context.Context) (string, error) { + if loadBinding == nil { + return "", nil + } + stored, err := loadBinding(resolveCtx) if err != nil { - // 诊断细节进服务日志;探测结果只暴露固定文案,避免把内部地址带进页面。 log.Printf("readiness: load routing bindings: %v", err) - return readiness.Target{}, fmt.Errorf("读取当前知识库绑定失败") } - if stored = strings.TrimSpace(stored); stored != "" { - knowledgeBaseID = stored + return stored, err + }, + ReadKey: func() (string, error) { + if readKey == nil { + return "", nil } - } - 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 + return value, err + }, } + return resolver.Resolve } // weKnoraLister 让「系统设置」的知识库下拉与保存校验同样跟随当前端点/凭据, diff --git a/cmd/worker/main.go b/cmd/worker/main.go index eb86462..022d715 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -9,6 +9,7 @@ import ( "os" "os/signal" "path/filepath" + "strings" "syscall" "time" @@ -23,7 +24,9 @@ import ( "knowledge-governance/internal/librarysnapshot" postgresstore "knowledge-governance/internal/platform/postgres" "knowledge-governance/internal/platform/weknora" + "knowledge-governance/internal/readiness" "knowledge-governance/internal/versionhistory" + "knowledge-governance/internal/weknoraaccess" artifactfetch "knowledge-governance/internal/workflows/artifactfetch" catalogrefresh "knowledge-governance/internal/workflows/catalogrefresh" "knowledge-governance/internal/workflows/sourcescan" @@ -146,7 +149,7 @@ func run(ctx context.Context, getenv func(string) string) error { } else if len(pending) > 0 { return fmt.Errorf("治理库 schema 落后(待应用迁移 %v):本角色不拥有 DDL,先由控制面进程完成迁移", pending) } - store := postgresstore.NewStore(pool).WithLibraryTarget(config.LibraryTargetIdentity()). + store := postgresstore.NewStore(pool).WithLibraryTarget(config.LibraryTargetIdentity()).WithRoutingDefaults(config.InternalRoutingKnowledgeBaseID, config.ExternalRoutingKnowledgeBaseID). WithRestoreDispatchEvidence(config.LibraryTargetJournalRoot) log.Printf("library target evidence identity=%s", config.LibraryTargetIdentity()) // 启动先接管被中断的写作业(复用已持久化的预备份),再把剩下的孤儿判失败: @@ -471,19 +474,27 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st if !config.TestWeKnoraEnabled || !config.LibraryWorkerRole.ExecutesSourceWork() { return handlers, nil } - apiKey, err := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile) - if err != nil { + // 启动只验证凭据“存在”(缺失像以前一样拒绝启动);具体值仍由 resolver 每次调用重读。 + if _, err := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile); err != nil { return nil, fmt.Errorf("read test WeKnora API key: %w", err) } - client, err := weknora.NewClient(weknora.ClientConfig{ - BaseURL: config.TestWeKnoraBaseURL, - APIKey: apiKey, - Timeout: config.TestWeKnoraHTTPTimeout, - AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP, - }) - if err != nil { - return nil, fmt.Errorf("create test WeKnora client: %w", err) + // 与 API 探测共用同一有效目标:网页保存的端点优先、env 只作未配置兜底; + // 当前持久化绑定优先、env 兜底;凭据每次调用重读同一 key 文件。 + remoteResolver := readiness.Resolver{ + Endpoint: func(resolveCtx context.Context) string { + if settings, settingsErr := store.GetActiveUISettings(resolveCtx); settingsErr == nil && strings.TrimSpace(settings.TestWeKnoraBaseURL) != "" { + return settings.TestWeKnoraBaseURL + } + return config.TestWeKnoraBaseURL + }, + LoadBinding: func(resolveCtx context.Context) (string, error) { + internal, _, _, bindingErr := store.LoadRoutingKBBindings(resolveCtx) + return internal, bindingErr + }, + ReadKey: func() (string, error) { return weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile) }, + DefaultKB: config.InternalRoutingKnowledgeBaseID, } + client := &weknoraaccess.Client{Resolver: remoteResolver, Timeout: config.TestWeKnoraHTTPTimeout, AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP} upload, err := testweknora.NewUploadHandler(testweknora.UploadHandlerConfig{ Store: store, Client: client, diff --git a/internal/platform/postgres/auto_ingest_continuation_integration_test.go b/internal/platform/postgres/auto_ingest_continuation_integration_test.go new file mode 100644 index 0000000..1cda464 --- /dev/null +++ b/internal/platform/postgres/auto_ingest_continuation_integration_test.go @@ -0,0 +1,236 @@ +package postgres + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/governance" +) + +// F17:取件 verified 后必须沿同一策略/当前绑定自动接续唯一 test.upload。 +// 这里用真实隔离 PG 跑 RouteCatalogFiles + MarkArtifactFetched 的真实 SQL 路径。 +type autoIngestFixture struct { + t *testing.T + ctx context.Context + pool *pgxpool.Pool + store *Store + now time.Time + connID string + kbID string +} + +func newAutoIngestFixture(t *testing.T) *autoIngestFixture { + t.Helper() + 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) + } + t.Cleanup(pool.Close) + if err = ApplyMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + now := time.Date(2026, 11, 5, 4, 0, 0, 0, time.UTC) + exec := func(q string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, q, args...); err != nil { + t.Fatalf("%v: %s", err, q) + } + } + exec(`INSERT INTO source_connections (id,source_type,name,enabled) VALUES ('conn-auto','dingtalk','ATOM自动衔接',TRUE)`) + exec(`INSERT INTO source_catalog_snapshots (id,source_connection_id,snapshot_sha256,source_label,node_count,folder_count,file_count,observed_at) + VALUES ('snap-auto','conn-auto',repeat('a',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-auto','node-dir',NULL,'folder','常见问题汇总','常见问题汇总',repeat('b',64))`) + exec(`INSERT INTO source_catalog_nodes (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash) + VALUES ('snap-auto','node-file','node-dir','file','人形常见问题.adoc','常见问题汇总/人形常见问题.adoc',repeat('c',64))`) + exec(`INSERT INTO source_catalog_heads (source_connection_id,snapshot_id,activated_at) VALUES ('conn-auto','snap-auto',$1)`, now) + exec(`INSERT INTO routing_kb_bindings (id,internal_kb_id,external_kb_id,updated_by) VALUES ('active','kb-current','kb-ext', 'fixture') + ON CONFLICT (id) DO UPDATE SET internal_kb_id='kb-current',external_kb_id='kb-ext'`) + return &autoIngestFixture{t: t, ctx: ctx, pool: pool, store: NewStore(pool).WithRoutingDefaults("kb-env", "kb-ext"), now: now, connID: "conn-auto", kbID: "kb-current"} +} + +func (f *autoIngestFixture) route() (versionID, publicationID string) { + f.t.Helper() + result, err := f.store.RouteCatalogFiles(f.ctx, governance.CatalogRouteRequest{ + ConnectionID: f.connID, + NodeIDs: []string{"node-file"}, + Visibility: governance.VisibilityInternal, + TargetKnowledgeBaseID: f.kbID, + OperatorRef: "fixture", + Reason: "fixture", + Now: f.now, + }) + if err != nil { + f.t.Fatal(err) + } + if result.SelectedFiles != 1 || result.PlannedPublications != 1 { + f.t.Fatalf("route result = %#v", result) + } + if err := f.pool.QueryRow(f.ctx, `SELECT publication.source_item_version_id,publication.id +FROM test_publications publication +JOIN catalog_import_intents intent ON intent.source_item_id=(SELECT source_item_id FROM source_item_versions WHERE id=publication.source_item_version_id) +WHERE intent.node_id='node-file' AND publication.test_kb_id=$1`, f.kbID).Scan(&versionID, &publicationID); err != nil { + f.t.Fatal(err) + } + return versionID, publicationID +} + +func (f *autoIngestFixture) markFetched(versionID string) { + f.t.Helper() + if err := f.store.MarkArtifactFetched(f.ctx, versionID, "32222102cafd6e5cdaeee415aad51e7df52629136d5ef2c3eae6e0e30ff7153d", 6991, "spool/auto.docx", "docx", true, f.now.Add(time.Minute)); err != nil { + f.t.Fatal(err) + } +} + +func (f *autoIngestFixture) uploadJobs(publicationID string) int { + f.t.Helper() + var count int + if err := f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM jobs WHERE type='test.upload' AND payload->>'publication_id'=$1`, publicationID).Scan(&count); err != nil { + f.t.Fatal(err) + } + return count +} + +func (f *autoIngestFixture) publicationState(publicationID string) string { + f.t.Helper() + var state string + if err := f.pool.QueryRow(f.ctx, `SELECT processing_state FROM test_publications WHERE id=$1`, publicationID).Scan(&state); err != nil { + f.t.Fatal(err) + } + return state +} + +func TestFetchedVerifiedArtifactAutoStartsExactlyOneUpload(t *testing.T) { + f := newAutoIngestFixture(t) + versionID, publicationID := f.route() + if jobs := f.uploadJobs(publicationID); jobs != 0 { + t.Fatalf("upload job before verification = %d", jobs) + } + f.markFetched(versionID) + if jobs := f.uploadJobs(publicationID); jobs != 1 { + t.Fatalf("upload jobs after verified fetch = %d, want 1", jobs) + } + if state := f.publicationState(publicationID); state != "pending" { + t.Fatalf("publication state = %q, want pending", state) + } + // 幂等:重复完成不能造第二个上传任务。 + f.markFetched(versionID) + if jobs := f.uploadJobs(publicationID); jobs != 1 { + t.Fatalf("duplicate verified fetch created %d upload jobs", jobs) + } +} + +func TestRouteReusingVerifiedContentAlsoAutoStarts(t *testing.T) { + f := newAutoIngestFixture(t) + versionID, publicationID := f.route() + // 模拟"本地已有先比对":内容已 verified,删除尚未消费的取件作业后再路由。 + if _, err := f.pool.Exec(f.ctx, `UPDATE source_item_versions SET content_object_id=$2,content_state='verified' +WHERE id=$1`, versionID, f.insertVerifiedContent()); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(f.ctx, `DELETE FROM jobs WHERE type='artifact.fetch' AND payload->>'version_id'=$1`, versionID); err != nil { + t.Fatal(err) + } + if _, err := f.store.RouteCatalogFiles(f.ctx, governance.CatalogRouteRequest{ + ConnectionID: f.connID, NodeIDs: []string{"node-file"}, Visibility: governance.VisibilityInternal, + TargetKnowledgeBaseID: f.kbID, OperatorRef: "fixture", Reason: "reuse", Now: f.now.Add(time.Minute), + }); err != nil { + t.Fatal(err) + } + if jobs := f.uploadJobs(publicationID); jobs != 1 { + t.Fatalf("reused verified content upload jobs = %d, want 1", jobs) + } +} + +func (f *autoIngestFixture) insertVerifiedContent() string { + f.t.Helper() + if _, err := f.pool.Exec(f.ctx, `INSERT INTO content_objects(id,sha256,size_bytes,mime,validated_format,local_spool_ref,content_state) +VALUES('co-auto',repeat('d',64),6991,'','docx','spool/auto.docx','verified') +ON CONFLICT(sha256) DO UPDATE SET local_spool_ref=EXCLUDED.local_spool_ref,content_state='verified'`); err != nil { + f.t.Fatal(err) + } + var id string + if err := f.pool.QueryRow(f.ctx, `SELECT id FROM content_objects WHERE sha256=repeat('d',64)`).Scan(&id); err != nil { + f.t.Fatal(err) + } + return id +} + +func TestStuckPlannedVerifiedPublicationRecoversViaReRoute(t *testing.T) { + f := newAutoIngestFixture(t) + versionID, publicationID := f.route() + f.markFetched(versionID) // 新代码本会入队;先清掉,复现 356 已卡记录 + if _, err := f.pool.Exec(f.ctx, `DELETE FROM jobs WHERE type='test.upload' AND payload->>'publication_id'=$1`, publicationID); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(f.ctx, `UPDATE test_publications SET processing_state='planned' WHERE id=$1`, publicationID); err != nil { + t.Fatal(err) + } + // 父在 UI 对同一文件再次保存当前策略:只这一条、只当前绑定,幂等接续。 + if _, err := f.store.RouteCatalogFiles(f.ctx, governance.CatalogRouteRequest{ + ConnectionID: f.connID, NodeIDs: []string{"node-file"}, Visibility: governance.VisibilityInternal, + TargetKnowledgeBaseID: f.kbID, OperatorRef: "fixture", Reason: "resume", Now: f.now.Add(2 * time.Minute), + }); err != nil { + t.Fatal(err) + } + if jobs := f.uploadJobs(publicationID); jobs != 1 { + t.Fatalf("stuck record recovered upload jobs = %d, want 1", jobs) + } + if state := f.publicationState(publicationID); state != "pending" { + t.Fatalf("stuck record state = %q", state) + } +} + +func TestAutoStartRefusesStaleOrIneligiblePolicies(t *testing.T) { + cases := []struct { + name string + mutate func(f *autoIngestFixture, publicationID string) + }{ + {"disabled source", func(f *autoIngestFixture, _ string) { + if _, err := f.pool.Exec(f.ctx, `UPDATE source_connections SET enabled=FALSE WHERE id='conn-auto'`); err != nil { + f.t.Fatal(err) + } + }}, + {"blocked policy", func(f *autoIngestFixture, _ string) { + if _, err := f.pool.Exec(f.ctx, `UPDATE source_items SET policy_state='blocked' WHERE source_connection_id='conn-auto'`); err != nil { + f.t.Fatal(err) + } + }}, + {"filtered filename", func(f *autoIngestFixture, _ string) { + if _, err := f.pool.Exec(f.ctx, `INSERT INTO filename_filter_rules(source_connection_id,directory_path,excluded_extensions,updated_by,mode) +VALUES('conn-auto','',ARRAY['adoc'],'fixture','exclude')`); err != nil { + f.t.Fatal(err) + } + }}, + {"superseded publication", func(f *autoIngestFixture, publicationID string) { + if _, err := f.pool.Exec(f.ctx, `UPDATE test_publications SET superseded_at=$2 WHERE id=$1`, publicationID, f.now); err != nil { + f.t.Fatal(err) + } + }}, + {"binding changed", func(f *autoIngestFixture, _ string) { + if _, err := f.pool.Exec(f.ctx, `UPDATE routing_kb_bindings SET internal_kb_id='kb-other' WHERE id='active'`); err != nil { + f.t.Fatal(err) + } + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newAutoIngestFixture(t) + versionID, publicationID := f.route() + tc.mutate(f, publicationID) + f.markFetched(versionID) + if jobs := f.uploadJobs(publicationID); jobs != 0 { + t.Fatalf("%s auto-started %d upload jobs, want 0", tc.name, jobs) + } + }) + } +} diff --git a/internal/platform/postgres/catalog_routing.go b/internal/platform/postgres/catalog_routing.go index 213fb60..9530ca9 100644 --- a/internal/platform/postgres/catalog_routing.go +++ b/internal/platform/postgres/catalog_routing.go @@ -27,7 +27,7 @@ func (s *Store) RouteCatalogFiles(ctx context.Context, request governance.Catalo return governance.CatalogRouteResult{}, fmt.Errorf("begin catalog route: %w", err) } defer func() { _ = tx.Rollback(ctx) }() - result, err := routeCatalogFilesTx(ctx, tx, request) + result, err := routeCatalogFilesTx(ctx, tx, s, request) if err != nil { return governance.CatalogRouteResult{}, err } @@ -57,7 +57,7 @@ func validateCatalogRouteRequest(request governance.CatalogRouteRequest) error { return nil } -func routeCatalogFilesTx(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest) (governance.CatalogRouteResult, error) { +func routeCatalogFilesTx(ctx context.Context, tx pgx.Tx, store *Store, request governance.CatalogRouteRequest) (governance.CatalogRouteResult, error) { targetKnowledgeBaseIDs := request.TargetKnowledgeBaseIDs if len(targetKnowledgeBaseIDs) == 0 && request.TargetKnowledgeBaseID != "" { targetKnowledgeBaseIDs = []string{request.TargetKnowledgeBaseID} @@ -142,7 +142,7 @@ ORDER BY node.node_id`, request.ConnectionID, nodeIDs) return governance.CatalogRouteResult{}, fmt.Errorf("iterate catalog route files: %w", err) } rows.Close() - return fulfillCatalogIntentsTx(ctx, tx, request, files, request.Now) + return fulfillCatalogIntentsTx(ctx, tx, store, request, files, request.Now) } func setCatalogDirectoryPolicy(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest, directoryPath string) error { diff --git a/internal/platform/postgres/operation_requests.go b/internal/platform/postgres/operation_requests.go index 2562ffe..133661d 100644 --- a/internal/platform/postgres/operation_requests.go +++ b/internal/platform/postgres/operation_requests.go @@ -35,7 +35,7 @@ func (s *Store) RouteCatalogFilesWithOperation(ctx context.Context, request gove return governance.CatalogRouteResult{}, false, err } resultJSON, replayed, err := s.executeOperation(ctx, "catalog.route", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) { - result, err := routeCatalogFilesTx(ctx, tx, request) + result, err := routeCatalogFilesTx(ctx, tx, s, request) if err != nil { return nil, err } diff --git a/internal/platform/postgres/policy_intents.go b/internal/platform/postgres/policy_intents.go index e73aca9..eca58e0 100644 --- a/internal/platform/postgres/policy_intents.go +++ b/internal/platform/postgres/policy_intents.go @@ -23,14 +23,14 @@ import ( // 开始处理 promotes them within the configured concurrency budget. Blocked // intents retire any planned publication and flip the item policy so the // knowledge-base diff demands removal of anything still live. -func fulfillCatalogIntentsTx(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest, files []governance.CatalogRouteFile, now time.Time) (governance.CatalogRouteResult, error) { +func fulfillCatalogIntentsTx(ctx context.Context, tx pgx.Tx, store *Store, request governance.CatalogRouteRequest, files []governance.CatalogRouteFile, now time.Time) (governance.CatalogRouteResult, error) { targetKnowledgeBaseIDs := request.TargetKnowledgeBaseIDs if len(targetKnowledgeBaseIDs) == 0 && request.TargetKnowledgeBaseID != "" { targetKnowledgeBaseIDs = []string{request.TargetKnowledgeBaseID} } result := governance.CatalogRouteResult{SelectedFiles: len(files)} for _, file := range files { - recorded, fetchQueued, planned, blocked, err := fulfillCatalogIntent(ctx, tx, request, file, targetKnowledgeBaseIDs, now) + recorded, fetchQueued, planned, blocked, err := fulfillCatalogIntent(ctx, tx, store, request, file, targetKnowledgeBaseIDs, now) if err != nil { return governance.CatalogRouteResult{}, err } @@ -54,7 +54,7 @@ func fulfillCatalogIntentsTx(ctx context.Context, tx pgx.Tx, request governance. // fulfillCatalogIntent applies one file decision. It returns recorded=false // only when the node vanished mid-transaction; fetchQueued reports a newly // enqueued artifact.fetch job and planned the number of publications created. -func fulfillCatalogIntent(ctx context.Context, tx pgx.Tx, request governance.CatalogRouteRequest, file governance.CatalogRouteFile, targetKnowledgeBaseIDs []string, now time.Time) (bool, bool, int, bool, error) { +func fulfillCatalogIntent(ctx context.Context, tx pgx.Tx, store *Store, request governance.CatalogRouteRequest, file governance.CatalogRouteFile, targetKnowledgeBaseIDs []string, now time.Time) (bool, bool, int, bool, error) { policy := string(request.Visibility) if _, err := tx.Exec(ctx, `INSERT INTO catalog_import_intents( source_connection_id,node_id,visibility,decided_by,reason,relative_path,name,size_bytes,modified_at,content_sha256, @@ -240,6 +240,11 @@ ON CONFLICT (source_item_version_id,test_kb_id) DO NOTHING`, } planned += int(tag.RowsAffected()) } + if contentState == string(governance.ContentVerified) { + if err := store.autoStartVerifiedPublicationsTx(ctx, tx, versionID, now); err != nil { + return false, false, 0, false, err + } + } fetchQueued := false if contentState != string(governance.ContentVerified) { @@ -295,6 +300,97 @@ func sha256Hex(value string) string { return hex.EncodeToString(sum[:]) } +// autoStartVerifiedPublicationsTx continues the normal ingestion chain in the +// SAME transaction that made the bytes verified. It deliberately re-checks the +// saved policy, the current binding and the live source state: a file routed +// before a policy/binding change must not be uploaded to a stale target. +func (s *Store) autoStartVerifiedPublicationsTx(ctx context.Context, tx pgx.Tx, versionID string, now time.Time) error { + if versionID == "" { + return errors.New("auto-start requires a version") + } + rows, err := tx.Query(ctx, ` +WITH binding AS ( + SELECT COALESCE(NULLIF(internal_kb_id,''), $3::text) AS internal_kb_id, + COALESCE(NULLIF(external_kb_id,''), $4::text) AS external_kb_id + FROM routing_kb_bindings WHERE id='active' +), eligible AS ( + SELECT publication.id + FROM test_publications publication + JOIN source_item_versions version ON version.id=publication.source_item_version_id + JOIN content_objects content ON content.id=version.content_object_id + JOIN source_items item ON item.id=version.source_item_id + JOIN source_connections connection ON connection.id=item.source_connection_id + LEFT JOIN LATERAL ( + SELECT intent.visibility FROM catalog_import_intents intent + WHERE intent.source_item_id=item.id AND intent.source_connection_id=item.source_connection_id + ORDER BY intent.updated_at DESC,intent.node_id LIMIT 1 + ) intent ON TRUE + CROSS JOIN binding + WHERE version.id=$1 + AND publication.superseded_at IS NULL + AND publication.processing_state='planned' + AND publication.ingest_state='pending' + AND publication.knowledge_id='' + AND publication.retrieval_state='unpublished' + AND version.content_state='verified' + AND content.content_state='verified' + AND content.local_spool_ref<>'' + AND content.size_bytes>0 + AND connection.enabled + AND item.policy_state IN (publication.target_scope,'both') + AND intent.visibility IN (publication.target_scope,'both') + AND NOT filename_is_filtered(item.source_connection_id,item.normalized_path,item.title) + AND NOT EXISTS (SELECT 1 FROM jobs job WHERE job.type='test.upload' AND job.payload->>'publication_id'=publication.id) + AND ((publication.target_scope='internal' AND binding.internal_kb_id<>'' AND publication.test_kb_id=binding.internal_kb_id) + OR (publication.target_scope='external' AND binding.external_kb_id<>'' AND publication.test_kb_id=binding.external_kb_id)) +) +UPDATE test_publications publication SET processing_state='pending',updated_at=$2 +FROM eligible WHERE publication.id=eligible.id +RETURNING publication.id`, versionID, now, s.defaultInternalKBID, s.defaultExternalKBID) + if err != nil { + return fmt.Errorf("auto-start verified publications: %w", err) + } + ids := make([]string, 0, 2) + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return fmt.Errorf("scan auto-started publication: %w", err) + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + rows.Close() + return fmt.Errorf("iterate auto-started publications: %w", err) + } + rows.Close() + for _, publicationID := range ids { + confirmationID, err := newOpaqueID("pc-") + if err != nil { + return err + } + 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 auto upload job") + } + if _, err := tx.Exec(ctx, `INSERT INTO processing_confirmations(id,publication_id,operator_ref,reason,confirmed_at,created_at) +VALUES($1,$2,'system:auto','取件 verified 自动接续',$3,$3) +ON CONFLICT (publication_id) DO NOTHING`, confirmationID, publicationID, now); err != nil { + return fmt.Errorf("record auto processing confirmation: %w", err) + } + if _, err := tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at) +VALUES($1,'test.upload',$2::jsonb,'pending',10,$3,5,$4,$3,$3) +ON CONFLICT (idempotency_key) DO NOTHING`, jobID, payload, now, "auto-test-upload:"+publicationID); err != nil { + return fmt.Errorf("enqueue auto upload: %w", err) + } + } + return nil +} + // MarkArtifactFetched finalizes one artifact.fetch success: it (re)uses the // content_objects row for the verified hash (content-level dedupe), attaches it // to the pending version and flips the intent to ready. matchedLocal records @@ -358,6 +454,9 @@ WHERE id=$1`, WHERE source_connection_id=$1 AND node_id=$2`, connectionID, nodeID); err != nil { return fmt.Errorf("mark intent fetched: %w", err) } + if err := s.autoStartVerifiedPublicationsTx(ctx, tx, versionID, now); err != nil { + return fmt.Errorf("continue fetched artifact into upload: %w", err) + } if ownsTransaction { if err := tx.Commit(ctx); err != nil { return fmt.Errorf("commit artifact fetch: %w", err) diff --git a/internal/platform/postgres/store.go b/internal/platform/postgres/store.go index bed0a36..939fba1 100644 --- a/internal/platform/postgres/store.go +++ b/internal/platform/postgres/store.go @@ -23,8 +23,10 @@ type database interface { } type Store struct { - db database - libraryTargetID string + db database + libraryTargetID string + defaultInternalKBID string + defaultExternalKBID string // restoreDispatchEvidenceRoot 是激活门唯一允许读取维护 journal 的根目录。 // 空值表示没有配置受信证据源:激活必须失败关闭,而不是退回"只看字符串"。 restoreDispatchEvidenceRoot string @@ -43,6 +45,17 @@ func NewStore(db database) *Store { return &Store{db: db} } +// WithRoutingDefaults supplies the deployment's bootstrap bindings. The UI +// binding row still wins; defaults are only the unbound fallback, exactly like +// the route/readiness resolvers. Auto-continuation uses them only to prove the +// publication still targets the CURRENT effective knowledge base. +func (s *Store) WithRoutingDefaults(internal, external string) *Store { + copy := *s + copy.defaultInternalKBID = strings.TrimSpace(internal) + copy.defaultExternalKBID = strings.TrimSpace(external) + return © +} + // WithRestoreDispatchEvidence 绑定维护 journal 的受信根目录(部署配置 // LIBRARY_TARGET_JOURNAL_ROOT)。没有它时激活门一律失败关闭。 func (s *Store) WithRestoreDispatchEvidence(root string) *Store { diff --git a/internal/readiness/weknora.go b/internal/readiness/weknora.go index 5de01bd..6e30105 100644 --- a/internal/readiness/weknora.go +++ b/internal/readiness/weknora.go @@ -36,6 +36,43 @@ type Result struct { // Prober 每次 Check 都重新解析目标并新建客户端:端点、凭据、绑定都允许 // 在进程运行期间变化,不能缓存成启动快照。 +// Resolver is the single source for the effective WeKnora target: endpoint +// saved in the console (env fallback), current persisted binding (env fallback) +// and the same key file. API probes and worker clients both use it, so a +// runtime settings/binding change cannot leave one side on the old value. +type Resolver struct { + Endpoint func(context.Context) string + LoadBinding func(context.Context) (string, error) + ReadKey func() (string, error) + DefaultKB string +} + +func (resolver Resolver) Resolve(ctx context.Context) (Target, error) { + knowledgeBaseID := strings.TrimSpace(resolver.DefaultKB) + if resolver.LoadBinding != nil { + stored, err := resolver.LoadBinding(ctx) + if err != nil { + return Target{}, fmt.Errorf("读取当前知识库绑定失败") + } + if stored = strings.TrimSpace(stored); stored != "" { + knowledgeBaseID = stored + } + } + apiKey := "" + if resolver.ReadKey != nil { + value, err := resolver.ReadKey() + if err != nil { + return Target{}, fmt.Errorf("读取 WeKnora API 凭据失败") + } + apiKey = strings.TrimSpace(value) + } + baseURL := "" + if resolver.Endpoint != nil { + baseURL = strings.TrimSpace(resolver.Endpoint(ctx)) + } + return Target{BaseURL: baseURL, KnowledgeBaseID: knowledgeBaseID, APIKey: apiKey}, nil +} + type Prober struct { Resolve func(context.Context) (Target, error) Timeout time.Duration diff --git a/internal/weknoraaccess/client.go b/internal/weknoraaccess/client.go new file mode 100644 index 0000000..d7524d4 --- /dev/null +++ b/internal/weknoraaccess/client.go @@ -0,0 +1,92 @@ +package weknoraaccess + +import ( + "context" + "errors" + "time" + + "knowledge-governance/internal/platform/weknora" + "knowledge-governance/internal/readiness" +) + +var ErrNotConfigured = errors.New("WeKnora 有效端点/凭据未配置") + +// Client resolves the effective WeKnora target on every call. It deliberately +// owns no cached *weknora.Client: endpoint, key and binding are console-managed. +type Client struct { + Resolver readiness.Resolver + Timeout time.Duration + AllowDockerHTTP bool +} + +func (client *Client) with(ctx context.Context, call func(*weknora.Client) error) error { + target, err := client.Resolver.Resolve(ctx) + if err != nil { + return err + } + if target.BaseURL == "" || target.APIKey == "" { + return ErrNotConfigured + } + timeout := client.Timeout + if timeout <= 0 || timeout > 10*time.Second { + timeout = 10 * time.Second + } + remote, err := weknora.NewClient(weknora.ClientConfig{ + BaseURL: target.BaseURL, + APIKey: target.APIKey, + Timeout: timeout, + AllowDockerHTTP: client.AllowDockerHTTP, + }) + if err != nil { + return err + } + return call(remote) +} + +func (client *Client) UploadFile(ctx context.Context, request weknora.UploadRequest) (weknora.Knowledge, error) { + var knowledge weknora.Knowledge + err := client.with(ctx, func(remote *weknora.Client) error { + value, err := remote.UploadFile(ctx, request) + knowledge = value + return err + }) + return knowledge, err +} + +func (client *Client) ListKnowledge(ctx context.Context, knowledgeBaseID string, page, pageSize int) ([]weknora.Knowledge, int64, error) { + var items []weknora.Knowledge + var total int64 + err := client.with(ctx, func(remote *weknora.Client) error { + value, count, err := remote.ListKnowledge(ctx, knowledgeBaseID, page, pageSize) + items, total = value, count + return err + }) + return items, total, err +} + +func (client *Client) GetKnowledge(ctx context.Context, knowledgeID string) (weknora.Knowledge, error) { + var knowledge weknora.Knowledge + err := client.with(ctx, func(remote *weknora.Client) error { + value, err := remote.GetKnowledge(ctx, knowledgeID) + knowledge = value + return err + }) + return knowledge, err +} + +func (client *Client) ListChunks(ctx context.Context, knowledgeID string, page, pageSize int, chunkTypes []string) ([]weknora.Chunk, int64, error) { + var chunks []weknora.Chunk + var total int64 + err := client.with(ctx, func(remote *weknora.Client) error { + value, count, err := remote.ListChunks(ctx, knowledgeID, page, pageSize, chunkTypes) + chunks, total = value, count + return err + }) + return chunks, total, err +} + +func (client *Client) DeleteKnowledge(ctx context.Context, knowledgeID string) error { + return client.with(ctx, func(remote *weknora.Client) error { + return remote.DeleteKnowledge(ctx, knowledgeID) + }) +} diff --git a/internal/weknoraaccess/client_test.go b/internal/weknoraaccess/client_test.go new file mode 100644 index 0000000..54be5bf --- /dev/null +++ b/internal/weknoraaccess/client_test.go @@ -0,0 +1,53 @@ +package weknoraaccess + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "knowledge-governance/internal/readiness" +) + +// F13:worker 的 upload/poll/validate/remove 必须与 API 探测一样,每次调用重新读 +// 当前端点、凭据和绑定;启动时捕获 env 会让页面改配置后两套身份分叉。 +func TestDynamicClientReResolvesEndpointKeyAndBinding(t *testing.T) { + server := func(wantKey, id string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-API-Key") != wantKey { + w.WriteHeader(http.StatusUnauthorized) + return + } + _, _ = w.Write([]byte(`{"success":true,"data":{"id":"` + id + `"}}`)) + })) + } + first, second := server("key-1", "kn-1"), server("key-2", "kn-2") + defer first.Close() + defer second.Close() + var endpoint, key, binding = first.URL, "key-1", "kb-old" + client := &Client{ + Resolver: readiness.Resolver{ + Endpoint: func(context.Context) string { return endpoint }, + ReadKey: func() (string, error) { return key, nil }, + LoadBinding: func(context.Context) (string, error) { return binding, nil }, + DefaultKB: "kb-env", + }, + Timeout: time.Second, + } + if knowledge, err := client.GetKnowledge(context.Background(), "kn-1"); err != nil || knowledge.ID != "kn-1" { + t.Fatalf("first target = %#v, %v", knowledge, err) + } + target, err := client.Resolver.Resolve(context.Background()) + if err != nil || target.KnowledgeBaseID != "kb-old" { + t.Fatalf("binding target = %#v, %v", target, err) + } + endpoint, key, binding = second.URL, "key-2", "kb-new" + target, err = client.Resolver.Resolve(context.Background()) + if err != nil || target.KnowledgeBaseID != "kb-new" { + t.Fatalf("changed binding target = %#v, %v", target, err) + } + if knowledge, err := client.GetKnowledge(context.Background(), "kn-2"); err != nil || knowledge.ID != "kn-2" { + t.Fatalf("rotated endpoint/key = %#v, %v", knowledge, err) + } +} diff --git a/web/src/App.tsx b/web/src/App.tsx index 7241e4b..2320ad0 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -2974,6 +2974,7 @@ export default function App() { {workspace.activeTab === 'dashboard' ? ( setSourceFocus(null)} onViewProcessing={(source, node) => { setMonitorFocus({ source: source.connection_id, q: node.name }) setWorkspace((current) => openModule(current, 'tasks')) @@ -2982,6 +2983,7 @@ export default function App() { ) : workspace.activeTab === 'tasks' ? ( setMonitorFocus(null)} onLocateSource={(sourceID, parentNodeID) => { setSourceFocus({ connectionID: sourceID, nodeID: parentNodeID }) setWorkspace((current) => openModule(current, 'dashboard')) diff --git a/web/src/SourceCatalogBrowser.focus.test.tsx b/web/src/SourceCatalogBrowser.focus.test.tsx new file mode 100644 index 0000000..ccd7cbe --- /dev/null +++ b/web/src/SourceCatalogBrowser.focus.test.tsx @@ -0,0 +1,155 @@ +// @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 { SourceCatalogBrowser } from './SourceCatalogBrowser' +import type { CatalogNode, CatalogSource } from './sourceCatalog' + +// F14:定位是“一次性导航请求”。旧请求不得在目录刷新/来源重载时重放, +// 手工换目录后必须留在原地;新的显式请求仍要能定位。 +const source: CatalogSource = { + connection_id: 'conn-atom', + source_type: 'dingtalk', + name: 'ATOM知识库', + root_node_id: 'root-1', + folder_count: 2, + file_count: 2, + observed_at: '2026-11-05T00:00:00Z', + snapshot_only: true, + snapshot_status: 'ready', + last_refresh_duration_ms: 0, +} +const node = (id: string, name: string, type: 'folder' | 'file', parent: string): CatalogNode => ({ + node_id: id, + parent_node_id: parent, + node_type: type, + name, + direct_child_count: type === 'folder' ? 1 : 0, + subtree_file_count: 1, + subtree_bytes: 1024, + size_bytes: type === 'file' ? 1024 : undefined, + visibility: 'internal', + imported_count: type === 'file' ? 1 : 0, + parsed_count: type === 'file' ? 1 : 0, + active_count: 0, + failed_count: 0, + routable_count: 1, +}) +const page = (parent: string) => { + const crumbs = + parent === 'root-1' + ? [{ node_id: 'root-1', name: 'ATOM知识库' }] + : [ + { node_id: 'root-1', name: 'ATOM知识库' }, + { node_id: parent, name: parent === 'folder-1' ? '售后机器出厂问题整改' : '常见问题汇总' }, + ] + return { + current: node(parent, crumbs[crumbs.length - 1].name, 'folder', parent === 'root-1' ? '' : 'root-1'), + breadcrumbs: crumbs, + items: + parent === 'root-1' + ? [ + node('folder-1', '售后机器出厂问题整改', 'folder', 'root-1'), + node('folder-2', '常见问题汇总', 'folder', 'root-1'), + ] + : [node(parent === 'folder-1' ? 'file-1' : 'file-2', '人形常见问题.adoc', 'file', parent)], + total: 1, + limit: 50, + offset: 0, + } +} + +let container: HTMLDivElement +let root: Root +let requested: string[] +const json = (body: unknown) => new Response(JSON.stringify(body), { status: 200 }) + +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) + requested = [] + vi.stubGlobal( + 'fetch', + vi.fn(async (input: RequestInfo | URL) => { + const url = new URL(String(input), 'http://localhost') + if (url.pathname === '/api/v1/source-catalog/sources') return json({ data: [source] }) + if (url.pathname === '/api/v1/source-catalog/routing-capabilities') + return json({ data: { internal_enabled: true, external_enabled: true } }) + if (url.pathname === '/api/v1/source-catalog/nodes') { + const parent = url.searchParams.get('parent_id') || '' + requested.push(parent) + return json(page(parent)) + } + if (url.pathname === '/api/v1/jobs') return json({ data: [] }) + if (url.pathname === '/api/v1/source-catalog/route') + return json({ + data: { + selected_files: 1, + intents_recorded: 1, + fetch_queued: 1, + planned_publications: 1, + blocked_items: 0, + unbound_files: 0, + }, + }) + return new Response('{}', { status: 404 }) + }) + ) +}) + +afterEach(() => { + act(() => root.unmount()) + container.remove() + vi.unstubAllGlobals() +}) + +const settle = async () => { + await act(async () => { + await Promise.resolve() + await Promise.resolve() + }) +} + +describe('one-shot source location', () => { + it('keeps the manually chosen directory when the old request is re-rendered', async () => { + const focus = { connectionID: 'conn-atom', nodeID: 'folder-1' } + const applied = vi.fn() + await act(async () => root.render()) + await settle() + expect(applied).toHaveBeenCalledTimes(1) + expect(requested.at(-1)).toBe('folder-1') + + // 用户手工返回根目录并进入另一个目录(不经过新旧定位请求)。 + const back = [...container.querySelectorAll('button')].find((b) => b.getAttribute('aria-label') === '返回上一级') + await act(async () => back?.click()) + await settle() + const other = [...container.querySelectorAll('button')].find((b) => b.textContent?.includes('常见问题汇总')) + await act(async () => other?.click()) + await settle() + expect(requested.at(-1)).toBe('folder-2') + + // 在当前目录保存一次策略:revision 会让 sources 重新加载,旧 focus 不得重放。 + const checkbox = container.querySelector('input[aria-label="选择人形常见问题.adoc"]') as HTMLInputElement | null + await act(async () => checkbox?.click()) + const route = [...container.querySelectorAll('button')].find((b) => b.textContent === '对内') + await act(async () => route?.click()) + await settle() + expect(requested.at(-1)).toBe('folder-2') + + // 同一旧请求对象因父重渲染再次传入(模拟 revision/sources 刷新):不得拽回 folder-1。 + await act(async () => root.render()) + await settle() + expect(requested.at(-1)).toBe('folder-2') + + // 新的显式定位请求必须仍然有效。 + const next = { connectionID: 'conn-atom', nodeID: 'folder-1' } + await act(async () => root.render()) + await settle() + expect(requested.at(-1)).toBe('folder-1') + expect(applied).toHaveBeenCalledTimes(2) + }) +}) diff --git a/web/src/SourceCatalogBrowser.tsx b/web/src/SourceCatalogBrowser.tsx index 5de4be0..87463e6 100644 --- a/web/src/SourceCatalogBrowser.tsx +++ b/web/src/SourceCatalogBrowser.tsx @@ -1,4 +1,4 @@ -import { useContext, useEffect, useState } from 'react' +import { useContext, useEffect, useRef, useState } from 'react' import { fetchCatalogNodes, fetchCatalogRoutingCapabilities, @@ -529,9 +529,11 @@ export function SourceCatalogBrowserView({ export function SourceCatalogBrowser({ focus, onViewProcessing, + onFocusApplied, }: { focus?: { connectionID: string; nodeID: string } | null onViewProcessing?: (source: CatalogSource, node: CatalogNode) => void + onFocusApplied?: () => void } = {}) { const [sources, setSources] = useState([]) const [selected, setSelected] = useState(null) @@ -656,16 +658,20 @@ export function SourceCatalogBrowser({ return () => controller.abort() }, [selected, parentID, offset, revision]) - // 从解析监控“在来源与目录定位”进入:选中该来源并直接打开文件所在目录。 + // 从解析监控“在来源与目录定位”进入:这是一次性导航请求。应用后由父消费; + // 同时用 ref 防同一请求在 sources 刷新/revision 重载时重放,手工换目录不会被拽回。 + const appliedFocus = useRef(null) useEffect(() => { - if (!focus || !sources.length) return + if (!focus || !sources.length || appliedFocus.current === focus) return const target = sources.find((source) => source.connection_id === focus.connectionID) if (!target) return + appliedFocus.current = focus setSelected(target) setParentID(focus.nodeID || target.root_node_id) setOffset(0) setPage(null) closeHistory() + onFocusApplied?.() }, [focus, sources]) // 有来源正在首次扫描时自动刷新列表:百分比会自己往前走,扫完那一行自动变成正式来源。 useEffect(() => { @@ -759,7 +765,7 @@ export function SourceCatalogBrowser({ reason: `目录页策略:${scope(policy)}`, }) setRoutingMessage( - `策略已保存 ${result.selected_files} 份;已记入意图 ${result.intents_recorded} 份;排队从源头取件 ${result.fetch_queued} 份;列入待入库 ${result.planned_publications} 份;标记禁止 ${result.blocked_items} 份;目录外无文件可标注 ${result.unbound_files} 份。到“解析监控 → 待入库”点开始处理即可。` + `策略已保存 ${result.selected_files} 份;已记入意图 ${result.intents_recorded} 份;排队从源头取件 ${result.fetch_queued} 份;列入处理 ${result.planned_publications} 份;标记禁止 ${result.blocked_items} 份;目录外无文件可标注 ${result.unbound_files} 份。取件校验完成后会自动上传解析,可在“解析监控”跟踪进度。` ) setRevision((value) => value + 1) return true diff --git a/web/src/WeKnoraMonitor.focus.test.tsx b/web/src/WeKnoraMonitor.focus.test.tsx new file mode 100644 index 0000000..73b2ade --- /dev/null +++ b/web/src/WeKnoraMonitor.focus.test.tsx @@ -0,0 +1,95 @@ +// @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 { WeKnoraMonitor } from './WeKnoraMonitor' + +// F14:来源“查看处理”进入监控时,已完成文件不能被默认 status=active 遮蔽。 +let container: HTMLDivElement +let root: Root +const requests: URL[] = [] + +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) + requests.length = 0 + const completed = { + id: 'tp-parent', + title: '已解析文件.pdf', + path: '售后/已解析文件.pdf', + node_id: 'node-file', + parent_node_id: 'node-folder', + source_id: 'conn-parent', + source_name: 'ATOM', + target_scope: 'internal', + target_kb: 'kb-parent', + action: 'add', + stage: 'finished', + status: 'completed', + detail: '处理完成', + updated_at: '2026-11-05T02:00:00Z', + retry_kind: '', + failure_id: '', + retry_count: 0, + version_id: 'sv-parent', + history: [], + } + vi.stubGlobal( + 'fetch', + vi.fn(async (input: RequestInfo | URL) => { + const url = new URL(String(input), 'http://localhost') + if (url.pathname === '/api/v1/monitor/tasks') { + requests.push(url) + const active = url.searchParams.get('status') === 'active' + const items = active ? [] : [completed] + return new Response( + JSON.stringify({ + data: { + items, + total: items.length, + limit: 20, + offset: 0, + sources: [{ id: 'conn-parent', name: 'ATOM' }], + targets: [{ id: 'kb-parent', name: '测试库', scope: 'internal' }], + bindings: { internal: 'kb-parent', external: '' }, + retryable: 0, + }, + }), + { status: 200 } + ) + } + return new Response('{}', { status: 503 }) + }) + ) +}) + +afterEach(() => { + act(() => root.unmount()) + container.remove() + vi.unstubAllGlobals() +}) + +describe('monitor focus from source directory', () => { + it('shows an already completed imported file without manual filter clearing', async () => { + vi.useFakeTimers() + const applied = vi.fn() + await act(async () => + root.render() + ) + await act(async () => { + await vi.advanceTimersByTimeAsync(250) + }) + expect(requests.length).toBeGreaterThan(0) + const last = requests[requests.length - 1] + expect(last.searchParams.get('source')).toBe('conn-parent') + expect(last.searchParams.get('q')).toBe('已解析文件.pdf') + expect(last.searchParams.get('status'), 'locator must not retain the default active filter').not.toBe('active') + expect(container.querySelector('tbody')?.textContent).toContain('已解析文件.pdf') + expect(applied).toHaveBeenCalledTimes(1) + vi.useRealTimers() + }) +}) diff --git a/web/src/WeKnoraMonitor.tsx b/web/src/WeKnoraMonitor.tsx index e2ddfaf..6b4c1e0 100644 --- a/web/src/WeKnoraMonitor.tsx +++ b/web/src/WeKnoraMonitor.tsx @@ -574,9 +574,11 @@ function TaskDetail({ export function WeKnoraMonitor({ focus, onLocateSource, + onFocusApplied, }: { focus?: { source: string; q: string } | null onLocateSource?: (sourceID: string, parentNodeID: string) => void + onFocusApplied?: () => void } = {}) { const auth = useOptionalAuth() const canOperate = auth?.identity?.role === 'admin' || auth?.identity?.role === 'engineer' @@ -599,12 +601,14 @@ export function WeKnoraMonitor({ setTabs([]) return () => setTabs([]) }, [setTabs]) - // 从来源目录“查看处理”进入:一次性套用来源+文件名定位,之后由用户自行改筛选。 + // 从来源目录“查看处理”进入:定位是一次性请求,必须清掉未结束/动作/阶段/时间等 + // 会遮蔽已完成文件的筛选;应用后由父消费掉,避免重挂载时旧请求再次覆盖手工筛选。 useEffect(() => { if (!focus) return - setFilters((current) => ({ ...current, q: focus.q, source: focus.source, target: '', scope: '' })) + setFilters({ ...defaultMonitorFilters, q: focus.q, source: focus.source, status: '' }) setOffset(0) setSelected([]) + onFocusApplied?.() }, [focus]) useEffect(() => { const c = new AbortController()