fix(ui): 接通取件后自动解析与一次性定位
build / build (push) Successful in 1m42s

F17:MarkArtifactFetched 与已验证内容复用都在同一事务内经过策略、来源启用、
过滤、superseded 和当前绑定的复核后,自动把 planned 转 pending 并入队唯一
test.upload;确认与作业幂等,失败整笔回滚。356 已卡记录可对同一文件再次保存
当前策略,经同一路径恢复,不做全库补扫或手工补队列。来源页提示改为自动接续。

F14:来源→监控定位清空未结束/动作/阶段/时间筛选;来源目录定位改为一次性请求
并由父消费,revision/sources 重载不再重放旧请求,手工换目录保持。

F13:抽出 readiness.Resolver 作为端点/绑定/凭据唯一来源;worker 上传/轮询/
校验/移除改用动态客户端,每次调用重读有效配置,不再固化启动 env 客户端。

验证:隔离 PG 的 F17 正负例与恢复用例、前端挂载定位回归、全量 Go/前端测试
及 make verify 17 项通过。未部署、未做 8088 UI 验收。
This commit is contained in:
姚章浩
2026-11-05 11:46:42 +08:00
parent 1c80b07f2c
commit 5bf4aae30e
17 changed files with 873 additions and 46 deletions
+27
View File
@@ -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
+1 -1
View File
@@ -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,
+16 -19
View File
@@ -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 让「系统设置」的知识库下拉与保存校验同样跟随当前端点/凭据,
+22 -11
View File
@@ -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,
@@ -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)
}
})
}
}
@@ -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 {
@@ -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
}
+102 -3
View File
@@ -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)
+15 -2
View File
@@ -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 &copy
}
// WithRestoreDispatchEvidence 绑定维护 journal 的受信根目录(部署配置
// LIBRARY_TARGET_JOURNAL_ROOT)。没有它时激活门一律失败关闭。
func (s *Store) WithRestoreDispatchEvidence(root string) *Store {
+37
View File
@@ -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
+92
View File
@@ -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)
})
}
+53
View File
@@ -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)
}
}
+2
View File
@@ -2974,6 +2974,7 @@ export default function App() {
{workspace.activeTab === 'dashboard' ? (
<SourceCatalogBrowser
focus={sourceFocus}
onFocusApplied={() => 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' ? (
<WeKnoraMonitor
focus={monitorFocus}
onFocusApplied={() => setMonitorFocus(null)}
onLocateSource={(sourceID, parentNodeID) => {
setSourceFocus({ connectionID: sourceID, nodeID: parentNodeID })
setWorkspace((current) => openModule(current, 'dashboard'))
+155
View File
@@ -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(<SourceCatalogBrowser focus={focus} onFocusApplied={applied} />))
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(<SourceCatalogBrowser focus={focus} onFocusApplied={applied} />))
await settle()
expect(requested.at(-1)).toBe('folder-2')
// 新的显式定位请求必须仍然有效。
const next = { connectionID: 'conn-atom', nodeID: 'folder-1' }
await act(async () => root.render(<SourceCatalogBrowser focus={next} onFocusApplied={applied} />))
await settle()
expect(requested.at(-1)).toBe('folder-1')
expect(applied).toHaveBeenCalledTimes(2)
})
})
+10 -4
View File
@@ -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<CatalogSource[]>([])
const [selected, setSelected] = useState<CatalogSource | null>(null)
@@ -656,16 +658,20 @@ export function SourceCatalogBrowser({
return () => controller.abort()
}, [selected, parentID, offset, revision])
// 从解析监控“在来源与目录定位”进入:选中该来源并直接打开文件所在目录。
// 从解析监控“在来源与目录定位”进入:这是一次性导航请求。应用后由父消费;
// 同时用 ref 防同一请求在 sources 刷新/revision 重载时重放,手工换目录不会被拽回。
const appliedFocus = useRef<typeof focus>(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
+95
View File
@@ -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(<WeKnoraMonitor focus={{ source: 'conn-parent', q: '已解析文件.pdf' }} onFocusApplied={applied} />)
)
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()
})
})
+6 -2
View File
@@ -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()