knowledge-governance/cmd/worker/main.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"knowledge-governance/internal/appconfig"
	"knowledge-governance/internal/governance"
	"knowledge-governance/internal/jobs"
	"knowledge-governance/internal/libraryagent"
	"knowledge-governance/internal/libraryhandoff"
	"knowledge-governance/internal/libraryrelease"
	"knowledge-governance/internal/librarysnapshot"
	postgresstore "knowledge-governance/internal/platform/postgres"
	"knowledge-governance/internal/platform/weknora"
	"knowledge-governance/internal/versionhistory"
	artifactfetch "knowledge-governance/internal/workflows/artifactfetch"
	catalogrefresh "knowledge-governance/internal/workflows/catalogrefresh"
	"knowledge-governance/internal/workflows/sourcescan"
	"knowledge-governance/internal/workflows/testweknora"
)

// libraryStageRecorder 把执行阶段写进持久作业，页面刷新/worker 重启后仍可复现。
// libraryStageStore 是阶段/证据记录用到的存储面：由 libraryAttemptWriter 实现，
// 所以阶段进度与版本登记天然带执行代次（失租代次写不进去）。
type libraryStageStore interface {
	UpdateLibraryStage(ctx context.Context, jobID, stage, state string, total, completed, bytesTotal, bytesDone *int64, detail string, now time.Time) error
	AdvanceLibraryJob(ctx context.Context, jobID string, state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error
	LibraryVersionByRef(ctx context.Context, scope string, kind governance.LibraryVersionKind, sourceJobID, snapshotID string) (governance.LibraryVersion, bool, error)
	RecordLibraryVersion(ctx context.Context, version governance.LibraryVersion) error
}

type libraryStageRecorder struct {
	store libraryStageStore
	jobID string
	scope string
	// operationID 让版本事件 ID 带执行意图：不同 operation 引用同一快照是两条事件，
	// 不会互相覆盖证据（F02）。为空表示历史/测试路径，退回旧 ID 形状。
	operationID string
	totals      map[string]stageSize
	// mappingDigest 是本次读到的生产索引映射，随预备份一起登记成版本证据。
	mappingDigest string
	// onBackup 在预备份完成后、写目标之前调用：把"预备份身份 + 内容风险"落盘。
	onBackup func(context.Context, libraryrelease.BackupRef) error
}

func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail string) {
	// 真实总量来自快照 manifest：已知就写真实数字，未知就保持 NULL（不写 0、不编百分比）。
	var total, completed, bytesTotal, bytesDone *int64
	if size, ok := r.totals[stage]; ok {
		total, bytesTotal = int64Ptr(size.count), int64Ptr(size.bytes)
		if state == "succeeded" {
			completed, bytesDone = total, bytesTotal
		}
	}
	if err := r.store.UpdateLibraryStage(ctx, r.jobID, stage, state, total, completed, bytesTotal, bytesDone,
		detail, time.Now().UTC()); err != nil {
		log.Printf("library stage %s/%s update failed: %v", r.jobID, stage, err)
	}
}

func int64Ptr(value int64) *int64 { return &value }

// stageSize 是某一类数据在快照里的真实总量（行数/字节）。
type stageSize struct {
	count int64
	bytes int64
}

// stageSizesFrom 从 manifest 汇总四类进度的真实总量：
// database/resources/index/wiki 各自的行数与字节，全部来自归档部件清单。
func stageSizesFrom(manifest librarysnapshot.Manifest) map[string]stageSize {
	sizes := map[string]stageSize{}
	kindFor := map[string]string{"database": "database", "resource": "resources", "index": "index", "wiki": "wiki"}
	for _, part := range manifest.Parts {
		stage, ok := kindFor[part.Kind]
		if !ok {
			continue
		}
		current := sizes[stage]
		current.count += part.Count
		current.bytes += part.Bytes
		sizes[stage] = current
	}
	return sizes
}

// libraryCaptureJobType is shared with the API which enqueues it.
const libraryCaptureJobType = "library.capture"

// librarySyncJobType 是生产同步作业的队列类型；与封存严格分开，状态不共用。
const librarySyncJobType = "library.sync"

// libraryRecoverJobType 是"按预备份恢复并校验"的队列类型；与 API 侧常量同值。
const libraryRecoverJobType = "library.recover"

// libraryRestoreJobType 是按指定整库版本恢复的队列类型；与同步分开、授权也分开。
const libraryRestoreJobType = "library.restore"

// libraryBaselineJobType 是"建立目标权威基线"的队列类型（与 API 侧常量同值）。
const libraryBaselineJobType = "library.baseline"

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv); err != nil {
		log.Fatal(err)
	}
}

func run(ctx context.Context, getenv func(string) string) error {
	config, err := appconfig.Load(getenv)
	if err != nil {
		return err
	}
	pool, err := pgxpool.New(ctx, config.DatabaseURL)
	if err != nil {
		return err
	}
	defer pool.Close()
	if err := pool.Ping(ctx); err != nil {
		return err
	}
	// 迁移也是"后台副作用"：只有控制面那个进程拥有 DDL。云端发布执行器只读校验
	// schema 版本，落后即拒绝启动——它不该成为治理库的第二个结构决策者。
	if config.LibraryWorkerRole.ExecutesMigrations() {
		if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
			return err
		}
	} else if pending, pendingErr := postgresstore.PendingMigrations(ctx, pool); pendingErr != nil {
		return fmt.Errorf("治理库 schema 版本检查失败: %w", pendingErr)
	} else if len(pending) > 0 {
		return fmt.Errorf("治理库 schema 落后（待应用迁移 %v）：本角色不拥有 DDL，先由控制面进程完成迁移", pending)
	}
	store := postgresstore.NewStore(pool).WithLibraryTarget(config.LibraryTargetIdentity())
	log.Printf("library target evidence identity=%s", config.LibraryTargetIdentity())
	// 启动先接管被中断的写作业（复用已持久化的预备份），再把剩下的孤儿判失败：
	// 顺序反了会把可接管的作业先标失败，丢掉已完成的备份。
	//
	// 只由"发布执行器"角色接管：release 作业的目标来自本进程配置，capture 角色的
	// 进程按本机维护配置去接管云端目标的作业，等于把快照写到另一套目标上。
	if config.LibraryWorkerRole.ExecutesRelease() {
		if resumed, failed, resumeErr := resumeInterruptedLibraryJobs(context.Background(), store, config); resumeErr != nil {
			log.Printf("library job resume failed: %v", resumeErr)
		} else if resumed > 0 || failed > 0 {
			log.Printf("library job resume: %d resumed, %d failed without backup", resumed, failed)
		}
		// 运行期也要持续接管：只在启动扫一次会漏掉"worker 在两次重启之间被杀"的情况。
		go func() {
			ticker := time.NewTicker(30 * time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					resumed, failed, err := resumeInterruptedLibraryJobs(ctx, store, config)
					if err != nil {
						log.Printf("library job takeover sweep: %v", err)
					} else if resumed > 0 || failed > 0 {
						log.Printf("library job takeover sweep: %d resumed, %d failed without backup", resumed, failed)
					}
				}
			}
		}()
	}
	// 只收拾本角色负责的 kind：跨域"顺手收拾"会把另一台执行器的在飞作业标失败。
	if repaired, reconcileErr := store.ReconcileLibraryJobs(context.Background(), time.Now().UTC(),
		config.LibraryWorkerRole.LibraryJobKinds()); reconcileErr != nil {
		log.Printf("library job reconcile failed: %v", reconcileErr)
	} else if repaired > 0 {
		log.Printf("library job reconcile: %d orphan job(s) marked failed", repaired)
	}
	handlers, err := workerHandlers(store, config)
	if err != nil {
		return err
	}
	if config.LibraryWorkerRole.ExecutesSourceWork() {
		// 快照搬运（本机 → 云端执行器）与云端版本回传都只有 capture 角色做：
		// 它持有源归档与 SSH 凭据；release-executor 在云端没有回本机的通道。
		if handoffConfig, configured, handoffErr := libraryhandoff.FromEnv(getenv); handoffErr != nil {
			return fmt.Errorf("快照搬运接线无效: %w", handoffErr)
		} else if configured {
			if config.LibraryWorkerRole.ExecutesRelease() {
				return errors.New("快照搬运只能接在 capture 角色上：full 角色同时消费发布作业，会与云端执行器抢同一批作业")
			}
			handoff, buildErr := libraryhandoff.New(handoffConfig, nil)
			if buildErr != nil {
				return buildErr
			}
			registerLibraryHandoffHandlers(handlers, store, config, handoff)
			go runLibraryArchivePullLoop(ctx, store, handoff, config.LibraryArchiveRoot, handoffConfig.PullInterval,
				func(err error) { log.Printf("library archive pull: %v", err) })
			log.Printf("library handoff enabled: executor=%s archive=%s pull_every=%s",
				handoffConfig.ExecutorContainer, handoffConfig.RemoteArchive, handoffConfig.PullInterval)
		}
		if syncErr := configureDirectorySync(pool, getenv, handlers); syncErr != nil {
			log.Print("QA directory synchronization unavailable; handler was not registered")
		}
		mirrorStore, mirrorErr := configureRecordsSync(pool, getenv, config.DatabaseURL, handlers)
		if mirrorErr != nil {
			log.Print("QA local mirror configuration invalid; handler was not registered")
		}
		if mirrorStore != nil {
			go runRecordsSyncScheduler(ctx, mirrorStore, func(error) { log.Print("QA local mirror scheduler unavailable; last committed data retained") })
		}
	}
	// 生产同步/恢复的执行器不依赖主机侧封存 agent：没有 agent socket 也要能跑。
	//
	// 执行域准入（R4）在这里统一判定：目标实况、旧 release 路由绑定、未释放维护窗口
	// 任何一项未知或重叠 ⇒ 这些 handler 直接拒绝执行（明确失败，无任何维护动作，
	// 也不停止任何既有服务）。
	//
	// 只有"发布执行器"角色注册它们：capture 角色的进程按本机目标配置执行一条
	// 指向别处目标的作业，后果是跨目标写入，不是失败重试能纠正的事。
	if config.LibraryWorkerRole.ExecutesRelease() {
		admissionErr := verifyLibraryExecutionDomain(context.Background(), config, store)
		if admissionErr != nil {
			log.Printf("library execution domain NOT admitted: %v", admissionErr)
		}
		registerLibraryReleaseHandlers(handlers, store, config, admissionErr)
	}

	worker, err := jobs.NewWorker(jobs.WorkerConfig{
		Queue:         store,
		WorkerID:      config.WorkerID,
		LeaseDuration: config.LeaseDuration,
		RetryDelay:    config.RetryDelay,
		Handlers:      handlers,
		OnError:       func(err error) { log.Printf("worker error: %v", err) },
	})
	if err != nil {
		return err
	}
	if config.LibraryWorkerRole.ExecutesSourceWork() {
		if ai := configureQAProcessing(pool, getenv, config.DatabaseURL); ai != nil {
			go ai.Run(ctx, func(error) { log.Print("QA AI worker unavailable; durable work and original labels retained") })
		}
		go weeklyMailLoop(ctx, pool, getenv)
		// 封存执行器状态：worker 是唯一能连 agent socket 的一方，它写观测、页面读观测。
		go runLibraryExecutorProbe(ctx, store, config, libraryExecutorProbeInterval)
		go runSourceRefreshScheduler(ctx, store, config.SourceRefreshSchedulerInterval, func(err error) {
			log.Printf("worker %s source refresh scheduler error: %v", config.WorkerID, err)
		})
	}
	log.Printf("worker %s started role=%s release=%t source=%t (source refresh scheduler every %s)",
		config.WorkerID, config.LibraryWorkerRole, config.LibraryWorkerRole.ExecutesRelease(),
		config.LibraryWorkerRole.ExecutesSourceWork(), config.SourceRefreshSchedulerInterval)
	return worker.Run(ctx, config.PollInterval)
}

func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[string]jobs.Handler, error) {
	handlers := map[string]jobs.Handler{
		"demo": func(_ context.Context, job jobs.Job) error {
			log.Printf("completed demo job id=%s", job.ID)
			return nil
		},
		"demo.sleep": jobs.DemoSleepHandler,
	}
	// 目录刷新与附件抓取属于"源侧"作业：它们读本机 spool、调本机脚本。
	// 发布执行器角色在云端没有这些挂载，注册它们只会抢走本机队列里的作业。
	if config.LibraryWorkerRole.ExecutesSourceWork() {
		catalogRunner, err := catalogrefresh.NewCommandRunner("/usr/bin/python3", "/app/scripts/live_catalog_refresh.py")
		if err != nil {
			return nil, fmt.Errorf("create catalog refresh runner: %w", err)
		}
		catalogHandler, err := catalogrefresh.NewHandler(catalogrefresh.Config{Store: store, Runner: catalogRunner})
		if err != nil {
			return nil, fmt.Errorf("create catalog refresh handler: %w", err)
		}
		handlers["source.catalog.refresh"] = catalogHandler.Handle
		ingestionRoot := config.IngestionSpoolRoot
		if ingestionRoot == "" {
			ingestionRoot = config.GovernanceSpoolRoot
		}
		if !filepath.IsAbs(ingestionRoot) || ingestionRoot == "" {
			ingestionRoot = "/var/lib/governance/ingestion"
		}
		fetchMaxBytes := config.ArtifactFetchMaxBytes
		if config.TestWeKnoraEnabled && config.TestWeKnoraMaxUploadBytes > 0 && fetchMaxBytes > config.TestWeKnoraMaxUploadBytes {
			fetchMaxBytes = config.TestWeKnoraMaxUploadBytes
		}
		if fetchMaxBytes <= 0 {
			fetchMaxBytes = 50 * 1024 * 1024
		}
		artifactRunner, err := artifactfetch.NewCommandRunner("/usr/bin/python3", "/app/scripts/artifact_fetch.py",
			ingestionRoot, "ingestion", fetchMaxBytes)
		if err != nil {
			return nil, fmt.Errorf("create artifact fetch runner: %w", err)
		}
		artifactHandler, err := artifactfetch.NewHandler(artifactfetch.Config{Store: store, Runner: artifactRunner})
		if err != nil {
			return nil, fmt.Errorf("create artifact fetch handler: %w", err)
		}
		handlers["artifact.fetch"] = artifactHandler.Handle
	}
	if config.LibraryAgentSocket != "" && config.LibraryWorkerRole.ExecutesSourceWork() {
		// 整库封存：worker 只投递请求并记录结果，真正的停机/导出/恢复发生在主机侧
		// agent（同 uid unix socket）。容器里既没有 Docker socket，也没有来源凭据。
		agentConfig := libraryagent.ClientConfig{Socket: config.LibraryAgentSocket, Timeout: 15 * time.Minute}
		handlers[libraryCaptureJobType] = func(ctx context.Context, job jobs.Job) error {
			var payload struct {
				Scope       string `json:"scope"`
				CaptureJob  string `json:"capture_job"`
				RequestedBy string `json:"requested_by"`
			}
			if err := json.Unmarshal(job.Payload, &payload); err != nil || (payload.Scope != "internal" && payload.Scope != "external") || payload.CaptureJob == "" {
				return fmt.Errorf("invalid library capture payload")
			}
			// 持久作业：页面读的是这里写的阶段与结果，worker 重启后仍能如实展示。
			libraryJobID := "libseal-" + payload.CaptureJob
			startedAt := time.Now().UTC()
			if advanceErr := store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobRunning, "export", "", "", "", "", startedAt); advanceErr != nil {
				log.Printf("library seal job %s cannot be advanced: %v", libraryJobID, advanceErr)
			}
			if stageErr := store.UpdateLibraryStage(ctx, libraryJobID, "export", "running", nil, nil, nil, nil,
				"主机侧正在暂停写入者、导出并恢复", startedAt); stageErr != nil {
				log.Printf("library seal job %s stage update failed: %v", libraryJobID, stageErr)
			}
			// 平台审计过的精确死信解除指纹：读取失败必须阻断（fail closed），
			// 不能因为存储不可用就静默按"没有解除"或"全部解除"处理。
			released, releaseErr := store.DeadLetterReleaseFingerprints(ctx, config.SourceInstance())
			if releaseErr != nil {
				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, releaseErr.Error(), time.Now().UTC())
				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", "dead letter release storage unavailable", "", "", "", time.Now().UTC())
				return fmt.Errorf("load dead letter releases: %w", releaseErr)
			}
			// 期望执行器身份随请求一起送进 agent，由接收方在操作互斥内、任何源操作之前
			// 比对：这样即使 socket 在这之前被换掉，错的执行器也一次 run 都不会跑。
			expectation, expectErr := captureExpectation(config)
			if expectErr != nil {
				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, expectErr.Error(), time.Now().UTC())
				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", expectErr.Error(),
					"未停任何写入者；补齐来源身份配置后重试", "", "", time.Now().UTC())
				return expectErr
			}
			source := configuredCaptureSource(config, payload.Scope)
			sourceID, sourceErr := source.ID()
			if sourceErr != nil {
				return sourceErr
			}
			result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released, expectation)
			finishedAt := time.Now().UTC()
			if err != nil {
				// 导出失败时阶段与作业都要留痕；恢复责任在 agent，不会因为作业失败而假装来源已恢复。
				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, err.Error(), finishedAt)
				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", err.Error(),
					"agent 已执行恢复检查；确认来源 healthy 后可重试同一作业号", "", "", finishedAt)
				log.Printf("library capture failed scope=%s capture_job=%s: %v", payload.Scope, payload.CaptureJob, err)
				return err
			}
			// 执行后复核：结果里 agent 自报的身份必须还是期望实例，否则这次封存
			// 可能来自别的栈，绝不能登记成"本实例的测试封存"。
			if identityErr := captureExecutorPostcheck(result.Payload, expectation); identityErr != nil {
				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), finishedAt)
				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
					"快照已产生但来源身份不符，未登记为测试封存；核对执行器后重新封存", "", "", finishedAt)
				return identityErr
			}
			if err := verifyCapturedSource(config.LibraryArchiveRoot, result.SnapshotID, source); err != nil {
				return err
			}
			// 封存产物成版本记录：只登记"测试封存"，绝不写成生产生效版本。
			versionErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
				ID:    governance.LibraryVersionID(governance.LibraryVersionTestSealed, payload.Scope, result.SnapshotID),
				Scope: payload.Scope,
				Kind:  governance.LibraryVersionTestSealed, SnapshotID: result.SnapshotID, SourceJobID: libraryJobID,
				Evidence: map[string]any{"source_id": sourceID, "source_identity": source, "capture_job": payload.CaptureJob, "replay": result.Replay,
					"requested_by": payload.RequestedBy, "expectation_fingerprint": expectationFingerprint(expectation),
					"agent_result": json.RawMessage(result.Payload)},
				CreatedBy: payload.RequestedBy, CreatedAt: finishedAt,
			})
			if versionErr != nil {
				return fmt.Errorf("library seal job %s cannot record version: %w", libraryJobID, versionErr)
			}
			_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "succeeded", nil, nil, nil, nil,
				"已封存快照 "+result.SnapshotID, finishedAt)
			_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobSucceeded, "export", "", "",
				result.SnapshotID, "", finishedAt)
			log.Printf("library capture completed scope=%s capture_job=%s snapshot=%s replay=%v requested_by=%s",
				payload.Scope, payload.CaptureJob, result.SnapshotID, result.Replay, payload.RequestedBy)
			return nil
		}
		// A crash can leave the source writer stopped: recover before taking work,
		// and keep the failure visible instead of pretending the worker is ready.
		go func() {
			recoveryCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
			defer cancel()
			if _, err := libraryagent.Recover(recoveryCtx, agentConfig); err != nil {
				log.Printf("library agent recovery check failed: %v", err)
				return
			}
			log.Printf("library agent reachable; freeze journals recovered")
		}()
	}
	if config.VersionHistoryGiteaBaseURL != "" && config.LibraryWorkerRole.ExecutesSourceWork() {
		// 「版本记录」同步：匿名只读远端仓库的提交/tag，写本库快照。未配置来源就不注册，
		// 界面上的同步按钮也会显示"未接线"，不会入队一个没人处理的作业。
		client, err := versionhistory.NewGiteaClient(versionhistory.GiteaConfig{
			BaseURL:         config.VersionHistoryGiteaBaseURL,
			AllowedBaseURLs: config.VersionHistoryAllowedOrigins,
			LinkOrigin:      config.VersionHistoryLinkOrigin,
			Repo:            config.VersionHistoryRepo,
			Timeout:         config.VersionHistoryHTTPTimeout,
		})
		if err != nil {
			return nil, fmt.Errorf("create version history client: %w", err)
		}
		syncer, err := versionhistory.NewSyncer(versionhistory.SyncerConfig{Source: client, Store: store,
			RepoKey: config.VersionHistoryRepo, MaxPages: config.VersionHistoryMaxPages})
		if err != nil {
			return nil, fmt.Errorf("create version history syncer: %w", err)
		}
		handlers[versionhistory.JobType] = func(ctx context.Context, job jobs.Job) error {
			triggeredBy := ""
			var payload struct {
				TriggeredBy string `json:"triggered_by"`
				TargetSHA   string `json:"target_sha"`
			}
			if err := json.Unmarshal(job.Payload, &payload); err == nil {
				triggeredBy = payload.TriggeredBy
			}
			activeSyncer := syncer
			if payload.TargetSHA != "" {
				var e error
				activeSyncer, e = versionhistory.NewSyncer(versionhistory.SyncerConfig{Source: client, Store: store, RepoKey: config.VersionHistoryRepo, MaxPages: config.VersionHistoryMaxPages, TargetSHAs: []string{payload.TargetSHA}})
				if e != nil {
					return e
				}
			}
			run, err := activeSyncer.Sync(ctx, job.ID, triggeredBy)
			if err != nil {
				log.Printf("version history sync failed job=%s run=%d code=%s: %v", job.ID, run.ID, run.ErrorCode, err)
				return err
			}
			log.Printf("version history sync completed job=%s run=%d fetched=%d expected=%v complete=%v",
				job.ID, run.ID, run.FetchedCount, run.ExpectedCount, run.Complete)
			return nil
		}
	}
	if config.PPScanEnabled && config.LibraryWorkerRole.ExecutesSourceWork() {
		scanner, err := sourcescan.NewFilePPScanner(sourcescan.FilePPScannerConfig{
			AllowedRoot:  config.PPScanAllowedRoot,
			LedgerPath:   config.PPScanLedgerPath,
			ManifestPath: config.PPScanManifestPath,
		})
		if err != nil {
			return nil, fmt.Errorf("create PP source scanner: %w", err)
		}
		handler, err := sourcescan.NewPPHandler(sourcescan.PPHandlerConfig{Store: store, Scanner: scanner})
		if err != nil {
			return nil, fmt.Errorf("create PP source scan handler: %w", err)
		}
		handlers["source.scan.pp"] = handler.Handle
	}
	if !config.TestWeKnoraEnabled || !config.LibraryWorkerRole.ExecutesSourceWork() {
		return handlers, nil
	}
	apiKey, err := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile)
	if 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)
	}
	upload, err := testweknora.NewUploadHandler(testweknora.UploadHandlerConfig{
		Store:          store,
		Client:         client,
		SpoolRoot:      config.GovernanceSpoolRoot,
		MaxUploadBytes: config.TestWeKnoraMaxUploadBytes,
		FirstPollDelay: config.TestWeKnoraPollInitial,
	})
	if err != nil {
		return nil, fmt.Errorf("create test upload handler: %w", err)
	}
	poll, err := testweknora.NewPollHandler(testweknora.PollHandlerConfig{
		Store:      store,
		Client:     client,
		PollDelay:  config.TestWeKnoraPollInitial,
		StallAfter: config.TestWeKnoraStalledAfter,
	})
	if err != nil {
		return nil, fmt.Errorf("create test poll handler: %w", err)
	}
	policy, err := governance.ParseAutoReviewPolicy(config.TestAutoReviewPolicy)
	if err != nil {
		return nil, fmt.Errorf("resolve test auto review policy: %w", err)
	}
	validate, err := testweknora.NewValidateHandler(testweknora.ValidateHandlerConfig{
		Store:            store,
		Client:           client,
		PolicyVersion:    config.TestAcceptancePolicyVersion,
		AutoReviewPolicy: policy,
		Report:           func(err error) { log.Printf("auto review error: %v", err) },
	})
	if err != nil {
		return nil, fmt.Errorf("create test validate handler: %w", err)
	}
	remove, err := testweknora.NewRemoveHandler(testweknora.RemoveHandlerConfig{
		Store:  store,
		Client: client,
	})
	if err != nil {
		return nil, fmt.Errorf("create test removal handler: %w", err)
	}
	handlers["test.upload"] = upload.Handle
	handlers["test.poll"] = poll.Handle
	handlers["test.validate"] = validate.Handle
	handlers["test.remove"] = remove.Handle
	return handlers, nil
}