feat(qa-ops): 建立独立正式主表并保真导入历史问答
build / build (push) Successful in 1m7s

将可维护总表与可替换2.0镜像分离,稳定来源关联和原始事件键保留;
源端更新/删除不覆盖主表或人工复核。首次收录保留原评估审计,
主表输入/复核独立持久化;历史原标注不冒充新AI或冻结证据。

历史导入绑定源哈希、逐行批准清单与真实恢复备份;三字段精确去重,
批次及原字段留档,重复执行不覆盖后续修改。仅操作本地治理库。

验证:完整make verify退出0;镜像与区域两组隔离PG race退出0;
真实恢复数据隔离导入12522、跳过792、暂缓1,总表15050;
重放保留模拟修改,结构down拒绝丢历史;真实浏览器第41周551条完整,
历史原标注和独立主表复核读回通过,移动布局及JS异常0。
This commit is contained in:
姚章浩
2026-10-19 08:49:18 +08:00
parent 6e67034fbd
commit f5e0f8d20f
22 changed files with 1350 additions and 41 deletions
+17
View File
@@ -76,6 +76,23 @@
> `0.1.12`),由 `scripts/version.py` 从 git **算出来**;开发交付的独立说明写入本文件“开发快照”章节,并打 `snapshot/vX.Y.Z` 注释标签 ⇒ 每次开发交付均可追溯,
> 但"发布版本"仍然只能由人定(CI 不会自动升版本、不会打 tag)。
## 开发快照 snapshot/v0.1.33 - 2026-10-19
### Added
- 问题流水读取治理侧独立正式总表,合并历史 Excel 与已收录的 2.0 问答;历史译文、答案、覆盖/正确标注、分类及人员字段保留原值,缺失不造值,来源可辨。
- 新增受输入哈希、已批准逐行清单和真实恢复备份约束的历史导入命令,固定首表与提问时间/提问人/原始问题组合;保留批次、原字段、行号、重复跳过与缺字段暂缓证据。重复导入只读回收据,不覆盖后续主表修改。
### Changed
- 新增治理侧加表迁移 53。2.0 镜像仍独立同步,正式主表只接收新事件;稳定来源关联与原始事件键不随本地修改丢失,源端更新和删除不覆盖或清除主表。
- 主表评估/复核与可替换的源副本分开持久化;首次收录保留已有评估及审计来源,后续人工复核绑定主表自身的输入版本,不被源同步失效。原 2.0 看板、历史冻结及周报统计范围不因导入被静默混算。
### Security
- 主表列表、详情、复核与历史接口复用固定租户、管理员会话、严格参数、同源写意图、乐观锁和幂等回执;拒绝缓存、稳定 ID 与原始溯源不可漂移。字段计数和质量范围只读标量元数据,不为全范围统计解压长答案。
- 导入仅连接本地治理库,不连接或修改 DobotAI 2.0 生产源、现有隧道与外部表;应用回退保留加表,有历史导入、本地修改或新增复核时拒绝有损结构回退。
## 开发快照 snapshot/v0.1.32 - 2026-10-18
### Added
+458
View File
@@ -207,6 +207,464 @@
}
}
},
"/api/v1/qa-ops/master/records": {
"get": {
"tags": [
"QA operations"
],
"operationId": "listQAOperationsRecordsMaster",
"summary": "正式主表:按半开提问时间范围、原问题关键词、稳定游标列出钉钉问答;租户由后端固定;答案包含暂无已发布则治理侧未覆盖",
"parameters": [
{
"name": "start",
"in": "query",
"required": true,
"schema": {
"type": "string",
"format": "date-time"
}
},
{
"name": "end",
"in": "query",
"required": true,
"schema": {
"type": "string",
"format": "date-time"
}
},
{
"name": "keyword",
"in": "query",
"schema": {
"type": "string",
"maxLength": 200
}
},
{
"name": "cursor",
"in": "query",
"schema": {
"type": "string",
"maxLength": 2048
}
},
{
"name": "limit",
"in": "query",
"schema": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": 20
}
}
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "未获员工问答查看授权"
},
"503": {
"description": "源库未接入、不可用或权限护栏阻断,不返回空列表伪装成功"
},
"500": {
"$ref": "#/components/responses/InternalError"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/master/records/{id}": {
"get": {
"tags": [
"QA operations"
],
"operationId": "getQAOperationsRecordMaster",
"summary": "正式主表:读取授权租户内的完整问答与源端覆盖对照;治理覆盖规则独立,不写回源库",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "string",
"maxLength": 256
}
}
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "未获员工问答查看授权"
},
"404": {
"description": "当前授权范围内不存在此问答"
},
"503": {
"description": "独立只读源不可用或权限校验阻断"
},
"500": {
"$ref": "#/components/responses/InternalError"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/master/quality/query": {
"post": {
"operationId": "queryQAQualityCountsMaster",
"summary": "正式主表:按完整筛选范围读取指纹/规则元数据并留档统计,最多5000条;summary.rates 返回 quality-rates-human-first-v1、原始分子/分母、百分比与分母构成。覆盖率=covered/(covered+uncovered+partial),正确率=covered_correct/(covered_correct+covered_incorrect);失效/不计入/相应待确认与未评估排除,零分母percent=null。原文仅在读取问答/复核时采集",
"parameters": [
{
"name": "X-Governance-Action",
"in": "header",
"required": true,
"schema": {
"type": "string",
"enum": [
"qa-quality"
]
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"additionalProperties": false,
"properties": {
"start": {
"type": "string",
"format": "date-time"
},
"end": {
"type": "string",
"format": "date-time"
},
"keyword": {
"type": "string",
"maxLength": 200
}
},
"required": [
"start",
"end"
]
}
}
}
},
"tags": [
"QA operations"
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "仅管理员访问"
},
"503": {
"description": "独立只读源或治理组织存储不可用,不返回伪造的零值"
},
"500": {
"$ref": "#/components/responses/InternalError"
},
"409": {
"description": "输入或复核版本变化/重复操作载荷不一致,未覆盖原结论"
},
"404": {
"description": "当前授权范围内未找到记录/历史输入"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/master/records/{id}/reviews": {
"post": {
"operationId": "saveQAQualityReviewMaster",
"summary": "正式主表:治理库追加人工复核,登录身份审计,源端只读,输入/复核乐观锁和nonce幂等;正确时根因为空,不计入须说明原因",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "string"
}
},
{
"name": "X-Governance-Action",
"in": "header",
"required": true,
"schema": {
"type": "string",
"enum": [
"qa-quality"
]
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"additionalProperties": false,
"properties": {
"input_hash": {
"type": "string",
"pattern": "^[a-f0-9]{64}$"
},
"input_epoch": {
"type": "integer",
"minimum": 1
},
"base_revision": {
"type": "integer",
"minimum": 0
},
"nonce": {
"type": "string",
"pattern": "^[A-Za-z0-9_-]{16,128}$"
},
"fields": {
"type": "object",
"additionalProperties": false,
"properties": {
"coverage": {
"type": "string",
"enum": [
"covered",
"uncovered",
"partial",
"pending",
"excluded"
]
},
"correctness": {
"type": "string",
"enum": [
"correct",
"incorrect",
"pending"
]
},
"product_category": {
"type": "string",
"maxLength": 128
},
"problem_type": {
"type": "string",
"maxLength": 128
},
"root_cause": {
"type": "string",
"maxLength": 2000
},
"exclusion_reason": {
"type": "string",
"maxLength": 2000
}
},
"required": [
"coverage",
"correctness"
]
},
"reason": {
"type": "string",
"minLength": 1,
"maxLength": 4000
}
},
"required": [
"input_hash",
"input_epoch",
"base_revision",
"nonce",
"fields",
"reason"
]
}
}
}
},
"tags": [
"QA operations"
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "仅管理员访问"
},
"503": {
"description": "独立只读源或治理组织存储不可用,不返回伪造的零值"
},
"500": {
"$ref": "#/components/responses/InternalError"
},
"409": {
"description": "输入或复核版本变化/重复操作载荷不一致,未覆盖原结论"
},
"404": {
"description": "当前授权范围内未找到记录/历史输入"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/master/records/{id}/quality/history": {
"get": {
"operationId": "listQAQualityHistoryMaster",
"summary": "正式主表:按当前固定租户查询不可变复检/人工历史,分别倒序分页,各20条",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "string"
}
},
{
"name": "review_before",
"in": "query",
"schema": {
"type": "integer",
"minimum": 1
}
},
{
"name": "check_before",
"in": "query",
"schema": {
"type": "integer",
"minimum": 1
}
}
],
"tags": [
"QA operations"
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "仅管理员访问"
},
"503": {
"description": "独立只读源或治理组织存储不可用,不返回伪造的零值"
},
"500": {
"$ref": "#/components/responses/InternalError"
},
"409": {
"description": "输入或复核版本变化/重复操作载荷不一致,未覆盖原结论"
},
"404": {
"description": "当前授权范围内未找到记录/历史输入"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/master/records/{id}/quality/inputs/{epoch}": {
"get": {
"operationId": "getQAQualityInputSnapshotMaster",
"summary": "正式主表:管理员读取已采集的历史质量输入快照,不另存人员关联/反馈字段;元数据留档但未采集原文返回409,不编造空快照",
"parameters": [
{
"name": "id",
"in": "path",
"required": true,
"schema": {
"type": "string"
}
},
{
"name": "epoch",
"in": "path",
"required": true,
"schema": {
"type": "integer",
"minimum": 1
}
}
],
"tags": [
"QA operations"
],
"responses": {
"200": {
"$ref": "#/components/responses/Envelope"
},
"400": {
"$ref": "#/components/responses/BadRequest"
},
"401": {
"description": "需要登录"
},
"403": {
"description": "仅管理员访问"
},
"503": {
"description": "独立只读源或治理组织存储不可用,不返回伪造的零值"
},
"500": {
"$ref": "#/components/responses/InternalError"
},
"409": {
"description": "该输入版本仅有指纹/规则元数据、原文未采集"
},
"404": {
"description": "当前授权范围内未找到记录/历史输入"
}
},
"description": "仅读治理正式主表,不回退源库;评估/复核采用独立主表命名空间,不被2.0同步覆盖。历史Excel原标注保留,不冒充新AI评估。"
}
},
"/api/v1/qa-ops/records": {
"get": {
"tags": ["QA operations"],
+2
View File
@@ -54,6 +54,8 @@ type RecordFeedback struct {
}
type Record struct {
RecordOrigin string `json:"record_origin,omitempty"`
MasterFields map[string]string `json:"master_fields,omitempty"`
QualityObservedAt time.Time `json:"-"`
QualityInputHash string `json:"-"`
Quality json.RawMessage `json:"quality,omitempty"`
+60
View File
@@ -0,0 +1,60 @@
package httpapi
import (
"context"
"knowledge-governance/internal/dobotaiops"
"knowledge-governance/internal/qaopsquality"
"net/http"
"strings"
)
func masterNoStore(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.URL.Path, "/api/v1/qa-ops/master/") {
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Pragma", "no-cache")
}
next.ServeHTTP(w, r)
})
}
// Master reads and reviews own their state; no source client is accepted as fallback.
type qaMasterStore interface {
ListMasterRecords(context.Context, dobotaiops.RecordQuery) (dobotaiops.RecordPage, error)
GetMasterRecord(context.Context, string, string) (dobotaiops.Record, error)
ReadMasterQualityScope(context.Context, dobotaiops.RecordQuery) (dobotaiops.QualityScope, error)
MasterQualityStore() *qaopsquality.Store
}
type qaMasterReader struct {
QAOpsReader
master qaMasterStore
}
func (s qaMasterReader) ListRecords(ctx context.Context, q dobotaiops.RecordQuery) (dobotaiops.RecordPage, error) {
return s.master.ListMasterRecords(ctx, q)
}
func (s qaMasterReader) GetRecord(ctx context.Context, tenant, id string) (dobotaiops.Record, error) {
return s.master.GetMasterRecord(ctx, tenant, id)
}
func (s qaMasterReader) ReadQualityScope(ctx context.Context, q dobotaiops.RecordQuery) (dobotaiops.QualityScope, error) {
return s.master.ReadMasterQualityScope(ctx, q)
}
func (s qaMasterReader) ReadStoredQuality(ctx context.Context, keys []dobotaiops.QualityKey) (map[string]qaopsquality.View, error) {
return s.master.MasterQualityStore().ReadViews(ctx, keys)
}
func withQAMaster(w http.ResponseWriter, r *http.Request, c Config, handler func(http.ResponseWriter, *http.Request, Config)) {
if !authorizeQAOps(w, r) {
return
}
master, ok := c.QAOps.(qaMasterStore)
if !ok {
qaOpsError(w, dobotaiops.ErrUnavailable)
return
}
c.QAOps = qaMasterReader{QAOpsReader: c.QAOps, master: master}
c.QAQuality = master.MasterQualityStore()
if master.MasterQualityStore() == nil {
c.QAQuality = nil
}
handler(w, r, c)
}
+78
View File
@@ -0,0 +1,78 @@
package httpapi
import (
"context"
"knowledge-governance/internal/dobotaiops"
"knowledge-governance/internal/qaopsquality"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
type masterFixtureReader struct {
calls int
tenant string
}
func (f *masterFixtureReader) ListRecords(context.Context, dobotaiops.RecordQuery) (dobotaiops.RecordPage, error) {
panic("master must never fall back to replaceable mirror/source reads")
}
func (f *masterFixtureReader) GetRecord(context.Context, string, string) (dobotaiops.Record, error) {
panic("master must never fetch the production source")
}
func (f *masterFixtureReader) Available(context.Context) error { return nil }
func (f *masterFixtureReader) Safety() dobotaiops.SourceSafety { return dobotaiops.SourceSafety{} }
func (f *masterFixtureReader) MasterQualityStore() *qaopsquality.Store { return nil }
func (f *masterFixtureReader) ListMasterRecords(_ context.Context, q dobotaiops.RecordQuery) (dobotaiops.RecordPage, error) {
f.calls++
f.tenant = q.Window.TenantKey
return dobotaiops.RecordPage{Items: []dobotaiops.Record{}, Columns: []string{"question"}, AsOf: time.Date(2026, 10, 19, 0, 0, 0, 0, time.UTC), LocalRevision: 3}, nil
}
func (f *masterFixtureReader) GetMasterRecord(_ context.Context, tenant, id string) (dobotaiops.Record, error) {
f.calls++
f.tenant = tenant
return dobotaiops.Record{ID: id, Question: "fixture master"}, nil
}
func (f *masterFixtureReader) ReadMasterQualityScope(context.Context, dobotaiops.RecordQuery) (dobotaiops.QualityScope, error) {
return dobotaiops.QualityScope{}, nil
}
func TestQAMasterUsesFixedTenantAndNeverSourceFallback(t *testing.T) {
auth := newFakeAuthStore()
auth.sessions["fixture-master-admin"] = "usr-boss"
reader := &masterFixtureReader{}
h, e := NewHandler(Config{Store: &fakeJobStore{}, Auth: &AuthConfig{Store: auth}, QAOps: reader, QAOpsTenantKey: "fixed-master-tenant", Now: func() time.Time { return time.Date(2026, 10, 19, 0, 0, 0, 0, time.UTC) }})
if e != nil {
t.Fatal(e)
}
base := "/api/v1/qa-ops/master/records"
for _, c := range []struct {
path string
cookie bool
status int
}{{base + "?start=2026-10-05T00:00:00Z&end=2026-10-12T00:00:00Z", true, 200}, {base + "/fixture-id", true, 200}, {base + "?tenant=other", true, 400}, {base, false, 401}} {
r := httptest.NewRequest(http.MethodGet, c.path, nil)
if c.cookie {
r.AddCookie(&http.Cookie{Name: sessionCookieName, Value: "fixture-master-admin"})
}
w := httptest.NewRecorder()
h.ServeHTTP(w, r)
if w.Code != c.status {
t.Fatalf("master route %s: got %d, wanted %d", c.path, w.Code, c.status)
}
if w.Header().Get("Cache-Control") != "no-store" {
t.Fatal("master employee data was cacheable")
}
}
if reader.calls != 2 || reader.tenant != "fixed-master-tenant" {
t.Fatal("master fixed-tenant read path missing")
}
r := httptest.NewRequest(http.MethodPost, base+"/fixture-id/reviews", strings.NewReader(`{}`))
r.AddCookie(&http.Cookie{Name: sessionCookieName, Value: "fixture-master-admin"})
w := httptest.NewRecorder()
h.ServeHTTP(w, r)
if w.Code != 403 {
t.Fatal("master review lacked explicit same-origin intent gate", w.Code)
}
}
+8 -2
View File
@@ -195,6 +195,12 @@ func NewHandler(config Config) (http.Handler, error) {
mux.HandleFunc("GET /api/v1/qa-ops/status", func(w http.ResponseWriter, r *http.Request) {
qaOpsStatus(w, r, config)
})
mux.HandleFunc("GET /api/v1/qa-ops/master/records", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, listQAOpsRecords) })
mux.HandleFunc("GET /api/v1/qa-ops/master/records/{id}", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, getQAOpsRecord) })
mux.HandleFunc("POST /api/v1/qa-ops/master/quality/query", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, qaQualityQuery) })
mux.HandleFunc("POST /api/v1/qa-ops/master/records/{id}/reviews", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, qaQualityReview) })
mux.HandleFunc("GET /api/v1/qa-ops/master/records/{id}/quality/history", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, qaQualityHistory) })
mux.HandleFunc("GET /api/v1/qa-ops/master/records/{id}/quality/inputs/{epoch}", func(w http.ResponseWriter, r *http.Request) { withQAMaster(w, r, config, qaQualitySnapshot) })
mux.HandleFunc("GET /api/v1/qa-ops/records", func(w http.ResponseWriter, r *http.Request) {
listQAOpsRecords(w, r, config)
})
@@ -552,9 +558,9 @@ func NewHandler(config Config) (http.Handler, error) {
}
config.authRoutes(mux)
config.userRoutes(mux)
return config.Auth.authenticate(mux), nil
return masterNoStore(config.Auth.authenticate(mux)), nil
}
return mux, nil
return masterNoStore(mux), nil
}
type createJobRequest struct {
@@ -0,0 +1,22 @@
DO $$ BEGIN
IF EXISTS(SELECT 1 FROM qa_ops_quality_reviews r JOIN qa_ops_master_records m ON r.tenant_key=m.tenant_key||'::master' AND r.record_id=m.id LEFT JOIN qa_ops_quality_reviews o ON o.tenant_key=m.tenant_key AND o.record_id=r.record_id AND o.revision=r.revision WHERE o.request_hash IS DISTINCT FROM r.request_hash) OR EXISTS(SELECT 1 FROM qa_ops_master_import_batches) OR EXISTS(SELECT 1 FROM qa_ops_master_records m LEFT JOIN qa_ops_mirror_records s ON s.tenant_key=m.tenant_key AND s.id=m.id WHERE m.origin<>'dobotai_2_0' OR s.id IS NULL OR m.payload IS DISTINCT FROM s.payload OR m.row_revision<>1) THEN
RAISE EXCEPTION 'durable master/import/local edits exist; keep additive schema on application rollback';
END IF;
END $$;
DROP TRIGGER qa_mirror_append_to_master ON qa_ops_mirror_records;
DROP FUNCTION qa_ops_append_source_to_master();
DELETE FROM qa_ops_quality_reviews q USING qa_ops_master_records m WHERE q.tenant_key=m.tenant_key||'::master' AND q.record_id=m.id;
DELETE FROM qa_ops_quality_checks q USING qa_ops_master_records m WHERE q.tenant_key=m.tenant_key||'::master' AND q.record_id=m.id;
DELETE FROM qa_ops_quality_inputs q USING qa_ops_master_records m WHERE q.tenant_key=m.tenant_key||'::master' AND q.record_id=m.id;
DELETE FROM qa_ops_quality_records q USING qa_ops_master_records m WHERE q.tenant_key=m.tenant_key||'::master' AND q.record_id=m.id;
DROP TABLE qa_ops_master_import_rows;
DROP TABLE qa_ops_master_source_links;
DROP TABLE qa_ops_master_records;
DROP TABLE qa_ops_master_import_batches;
DROP TABLE qa_ops_master_state;
DROP FUNCTION qa_ops_master_protect_provenance();
DROP FUNCTION qa_ops_master_revision();
DROP FUNCTION qa_ops_master_observe_quality();
DROP FUNCTION qa_ops_master_copy_source_quality(text,text);
DROP FUNCTION qa_ops_master_event_key(timestamptz,text,text);
DROP FUNCTION qa_ops_master_normalize(text);
@@ -0,0 +1,119 @@
-- A durable, locally-owned master is separate from the replaceable 2.0 mirror.
LOCK TABLE qa_ops_mirror_records,qa_ops_quality_records,qa_ops_quality_inputs,qa_ops_quality_checks,qa_ops_quality_reviews IN SHARE ROW EXCLUSIVE MODE;
CREATE TABLE qa_ops_master_state (
tenant_key TEXT PRIMARY KEY, revision BIGINT NOT NULL DEFAULT 0 CHECK(revision>=0), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE qa_ops_master_import_batches (
tenant_key TEXT NOT NULL, id TEXT NOT NULL, source_sha256 TEXT NOT NULL CHECK(source_sha256~'^[a-f0-9]{64}$'),
source_name TEXT NOT NULL, sheet_name TEXT NOT NULL, total_rows INTEGER NOT NULL CHECK(total_rows>=0),
imported_rows INTEGER NOT NULL CHECK(imported_rows>=0), skipped_rows INTEGER NOT NULL CHECK(skipped_rows>=0), held_rows INTEGER NOT NULL CHECK(held_rows>=0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY(tenant_key,id), UNIQUE(tenant_key,source_sha256,sheet_name),
CHECK(total_rows=imported_rows+skipped_rows+held_rows)
);
CREATE TABLE qa_ops_master_records (
tenant_key TEXT NOT NULL, id TEXT NOT NULL, event_key TEXT NOT NULL CHECK(event_key~'^[a-f0-9]{64}$'),
origin TEXT NOT NULL CHECK(origin IN ('dobotai_2_0','historical_excel')),
asked_at TIMESTAMPTZ NOT NULL, question TEXT NOT NULL, status TEXT NOT NULL, identity JSONB NOT NULL,
quality_hash TEXT NOT NULL, payload JSONB NOT NULL CHECK(jsonb_typeof(payload)='object'),
CHECK(payload?'id' AND jsonb_typeof(payload->'id')='string' AND payload->>'id'=id),
original_fields JSONB NOT NULL DEFAULT '{}'::jsonb CHECK(jsonb_typeof(original_fields)='object'),
batch_id TEXT, source_row INTEGER, row_revision BIGINT NOT NULL DEFAULT 1 CHECK(row_revision>0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY(tenant_key,id), FOREIGN KEY(tenant_key,batch_id) REFERENCES qa_ops_master_import_batches(tenant_key,id),
CHECK((origin='historical_excel' AND batch_id IS NOT NULL AND source_row>1) OR (origin='dobotai_2_0' AND batch_id IS NULL AND source_row IS NULL))
);
CREATE INDEX idx_qa_master_window ON qa_ops_master_records(tenant_key,asked_at DESC,id DESC);
CREATE INDEX idx_qa_master_original_event ON qa_ops_master_records(tenant_key,event_key);
CREATE TABLE qa_ops_master_source_links (
tenant_key TEXT NOT NULL, source_kind TEXT NOT NULL, source_id TEXT NOT NULL, master_id TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY(tenant_key,source_kind,source_id),
FOREIGN KEY(tenant_key,master_id) REFERENCES qa_ops_master_records(tenant_key,id)
);
CREATE TABLE qa_ops_master_import_rows (
tenant_key TEXT NOT NULL, batch_id TEXT NOT NULL, row_number INTEGER NOT NULL CHECK(row_number>1),
disposition TEXT NOT NULL CHECK(disposition IN ('imported','skipped_duplicate','held_missing_key')),
original_fields JSONB NOT NULL CHECK(jsonb_typeof(original_fields)='object'), master_id TEXT,
PRIMARY KEY(tenant_key,batch_id,row_number), FOREIGN KEY(tenant_key,batch_id) REFERENCES qa_ops_master_import_batches(tenant_key,id),
FOREIGN KEY(tenant_key,master_id) REFERENCES qa_ops_master_records(tenant_key,id)
);
CREATE FUNCTION qa_ops_master_normalize(value text) RETURNS text LANGUAGE sql IMMUTABLE AS $$ SELECT btrim(replace(replace(COALESCE(value,''),E'\r\n',E'\n'),E'\r',E'\n'),chr(9)||chr(10)||chr(11)||chr(12)||chr(13)||chr(28)||chr(29)||chr(30)||chr(31)||chr(32)||chr(133)||chr(160)||chr(5760)||chr(8192)||chr(8193)||chr(8194)||chr(8195)||chr(8196)||chr(8197)||chr(8198)||chr(8199)||chr(8200)||chr(8201)||chr(8202)||chr(8232)||chr(8233)||chr(8239)||chr(8287)||chr(12288)) $$;
CREATE FUNCTION qa_ops_master_event_key(at_time timestamptz,person text,original_question text) RETURNS text LANGUAGE sql STABLE AS $$
SELECT encode(sha256(convert_to(jsonb_build_array(to_char(at_time AT TIME ZONE 'Asia/Shanghai','YYYY-MM-DD HH24:MI:SS'),qa_ops_master_normalize(person),qa_ops_master_normalize(original_question))::text,'UTF8')),'hex')
$$;
CREATE FUNCTION qa_ops_master_revision() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
INSERT INTO qa_ops_master_state(tenant_key,revision) VALUES(NEW.tenant_key,1) ON CONFLICT(tenant_key) DO UPDATE SET revision=qa_ops_master_state.revision+1,updated_at=clock_timestamp();
RETURN NEW;
END $$;
CREATE TRIGGER qa_master_revision AFTER INSERT OR UPDATE ON qa_ops_master_records FOR EACH ROW EXECUTE FUNCTION qa_ops_master_revision();
CREATE FUNCTION qa_ops_master_protect_provenance() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
IF ROW(NEW.tenant_key,NEW.id,NEW.event_key,NEW.origin,NEW.original_fields,NEW.batch_id,NEW.source_row) IS DISTINCT FROM ROW(OLD.tenant_key,OLD.id,OLD.event_key,OLD.origin,OLD.original_fields,OLD.batch_id,OLD.source_row) THEN
RAISE EXCEPTION 'master provenance and original deduplication key are immutable' USING ERRCODE='55000';
END IF;
NEW.row_revision=OLD.row_revision+1;NEW.updated_at=clock_timestamp();
IF jsonb_build_array(NEW.payload->'status',NEW.payload->'question',NEW.payload->'answer',NEW.payload->'references',NEW.payload->'media') IS DISTINCT FROM jsonb_build_array(OLD.payload->'status',OLD.payload->'question',OLD.payload->'answer',OLD.payload->'references',OLD.payload->'media') THEN
NEW.quality_hash=encode(sha256(convert_to(jsonb_build_array(NEW.id,NEW.payload->'status',NEW.payload->'question',NEW.payload->'answer',NEW.payload->'references',NEW.payload->'media')::text,'UTF8')),'hex');
NEW.payload=NEW.payload||jsonb_build_object('_quality_hash',NEW.quality_hash,'_quality_observed_at',NEW.updated_at);
END IF;
IF NEW.payload IS DISTINCT FROM OLD.payload THEN
NEW.question=NEW.payload->>'question';NEW.asked_at=(NEW.payload->>'asked_at')::timestamptz;NEW.status=NEW.payload->>'status';
END IF;
RETURN NEW;
END $$;
CREATE TRIGGER qa_master_provenance BEFORE UPDATE ON qa_ops_master_records FOR EACH ROW EXECUTE FUNCTION qa_ops_master_protect_provenance();
-- Keep ALL existing 2.0 IDs, even pre-existing equal-key events; this is not an authorization to delete them.
INSERT INTO qa_ops_master_records(tenant_key,id,event_key,origin,asked_at,question,status,identity,quality_hash,payload)
SELECT tenant_key,id,qa_ops_master_event_key(asked_at,payload->>'asker',question),'dobotai_2_0',asked_at,question,status,identity,quality_hash,payload FROM qa_ops_mirror_records;
INSERT INTO qa_ops_master_source_links(tenant_key,source_kind,source_id,master_id)
SELECT tenant_key,'dobotai_2_0',id,id FROM qa_ops_mirror_records;
-- Preserve copied audit identities/timestamps; the original namespace is never updated.
CREATE FUNCTION qa_ops_master_copy_source_quality(source_tenant text,source_id text) RETURNS boolean LANGUAGE plpgsql AS $$
DECLARE copied bigint;
BEGIN
INSERT INTO qa_ops_quality_records SELECT q.tenant_key||'::master',q.record_id,q.observed_at,q.input_hash,q.input_epoch,q.review_revision FROM qa_ops_quality_records q JOIN qa_ops_master_records m ON m.tenant_key=q.tenant_key AND m.id=q.record_id WHERE q.tenant_key=source_tenant AND m.origin='dobotai_2_0' AND (source_id IS NULL OR q.record_id=source_id) ON CONFLICT DO NOTHING;
GET DIAGNOSTICS copied=ROW_COUNT; IF copied=0 THEN RETURN false; END IF;
INSERT INTO qa_ops_quality_inputs SELECT q.tenant_key||'::master',q.record_id,q.input_epoch,q.input_hash,q.payload,q.created_at FROM qa_ops_quality_inputs q JOIN qa_ops_master_records m ON m.tenant_key=q.tenant_key AND m.id=q.record_id WHERE q.tenant_key=source_tenant AND m.origin='dobotai_2_0' AND (source_id IS NULL OR q.record_id=source_id);
INSERT INTO qa_ops_quality_checks(tenant_key,record_id,input_epoch,rule_version,result,source,created_at) SELECT q.tenant_key||'::master',q.record_id,q.input_epoch,q.rule_version,q.result,q.source,q.created_at FROM qa_ops_quality_checks q JOIN qa_ops_master_records m ON m.tenant_key=q.tenant_key AND m.id=q.record_id WHERE q.tenant_key=source_tenant AND m.origin='dobotai_2_0' AND (source_id IS NULL OR q.record_id=source_id);
INSERT INTO qa_ops_quality_reviews SELECT q.tenant_key||'::master',q.record_id,q.revision,q.input_epoch,q.input_hash,q.fields,q.reason,q.reviewer,q.nonce,q.request_hash,q.created_at FROM qa_ops_quality_reviews q JOIN qa_ops_master_records m ON m.tenant_key=q.tenant_key AND m.id=q.record_id WHERE q.tenant_key=source_tenant AND m.origin='dobotai_2_0' AND (source_id IS NULL OR q.record_id=source_id);
RETURN true;
END $$;
SELECT qa_ops_master_copy_source_quality(tenant_key,NULL) FROM (SELECT DISTINCT tenant_key FROM qa_ops_master_records) tenants;
CREATE FUNCTION qa_ops_master_observe_quality() RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE quality_tenant text=NEW.tenant_key||'::master'; next_epoch bigint; assessment jsonb; evidence jsonb; observed timestamptz;
BEGIN
IF TG_OP='UPDATE' AND NEW.quality_hash=OLD.quality_hash THEN RETURN NEW; END IF;
IF TG_OP='INSERT' AND NEW.origin='dobotai_2_0' THEN
PERFORM qa_ops_master_copy_source_quality(NEW.tenant_key,NEW.id);
IF EXISTS(SELECT 1 FROM qa_ops_quality_records WHERE tenant_key=quality_tenant AND record_id=NEW.id AND input_hash=NEW.quality_hash) THEN RETURN NEW; END IF;
END IF;
observed=COALESCE((NEW.payload->>'_quality_observed_at')::timestamptz,NEW.updated_at);
INSERT INTO qa_ops_quality_records(tenant_key,record_id) VALUES(quality_tenant,NEW.id) ON CONFLICT DO NOTHING;
SELECT input_epoch+1 INTO next_epoch FROM qa_ops_quality_records WHERE tenant_key=quality_tenant AND record_id=NEW.id FOR UPDATE;
evidence=jsonb_build_object('version','governance-master-quality-v1','id',NEW.id,'status',NEW.status,'question',NEW.question,'answer',COALESCE(NEW.payload->>'answer',''),'references',COALESCE(NEW.payload->'references','null'::jsonb),'media',COALESCE(NEW.payload->'media','null'::jsonb));
assessment=jsonb_build_object('state','pending','rule_version','governance-answer-substring-v1');
IF position('暂无已发布' IN COALESCE(NEW.payload->>'answer',''))>0 THEN assessment=assessment||jsonb_build_object('state','checked','verdict','uncovered','matched_text','暂无已发布'); END IF;
INSERT INTO qa_ops_quality_inputs(tenant_key,record_id,input_epoch,input_hash,payload) VALUES(quality_tenant,NEW.id,next_epoch,NEW.quality_hash,evidence);
INSERT INTO qa_ops_quality_checks(tenant_key,record_id,input_epoch,rule_version,result,source) VALUES(quality_tenant,NEW.id,next_epoch,'governance-answer-substring-v1',assessment,COALESCE(NEW.payload->'source_coverage',jsonb_build_object('verdict','','reason_code','','evaluator_version','')));
UPDATE qa_ops_quality_records SET input_epoch=next_epoch,input_hash=NEW.quality_hash,observed_at=observed WHERE tenant_key=quality_tenant AND record_id=NEW.id;
RETURN NEW;
END $$;
CREATE TRIGGER qa_master_quality AFTER INSERT OR UPDATE ON qa_ops_master_records FOR EACH ROW EXECUTE FUNCTION qa_ops_master_observe_quality();
CREATE FUNCTION qa_ops_append_source_to_master() RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE key_value text; matched_id text; matches bigint;
BEGIN
IF EXISTS(SELECT 1 FROM qa_ops_master_source_links WHERE tenant_key=NEW.tenant_key AND source_kind='dobotai_2_0' AND source_id=NEW.id) THEN RETURN NEW; END IF;
key_value=qa_ops_master_event_key(NEW.asked_at,NEW.payload->>'asker',NEW.question);
PERFORM pg_advisory_xact_lock(hashtextextended(NEW.tenant_key||key_value,254));
SELECT count(*),min(id) INTO matches,matched_id FROM qa_ops_master_records WHERE tenant_key=NEW.tenant_key AND event_key=key_value;
IF matches<>1 THEN
matched_id=NEW.id;
INSERT INTO qa_ops_master_records(tenant_key,id,event_key,origin,asked_at,question,status,identity,quality_hash,payload)
VALUES(NEW.tenant_key,NEW.id,key_value,'dobotai_2_0',NEW.asked_at,NEW.question,NEW.status,NEW.identity,NEW.quality_hash,NEW.payload);
END IF;
INSERT INTO qa_ops_master_source_links(tenant_key,source_kind,source_id,master_id) VALUES(NEW.tenant_key,'dobotai_2_0',NEW.id,matched_id);
RETURN NEW;
END $$;
-- UPDATE only admits an unlinked source. A linked row never rewrites the master;
-- DELETE deliberately has NO corresponding master trigger or cascading FK.
CREATE TRIGGER qa_mirror_append_to_master AFTER INSERT OR UPDATE ON qa_ops_mirror_records FOR EACH ROW EXECUTE FUNCTION qa_ops_append_source_to_master();
@@ -11,8 +11,8 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
if err != nil {
t.Fatalf("loadMigrations() error = %v", err)
}
if len(migrations) != 52 {
t.Fatalf("migration count = %d, want 52", len(migrations))
if len(migrations) != 53 {
t.Fatalf("migration count = %d, want 53", len(migrations))
}
if migrations[0].Version != 1 || !strings.Contains(migrations[0].UpSQL, "CREATE TABLE IF NOT EXISTS jobs") {
t.Fatalf("migration 1 = %#v", migrations[0])
@@ -0,0 +1,32 @@
package postgres
import (
"strings"
"testing"
)
func TestDurableMasterMigrationOwnsDataAndRetainsAuditOnRollback(t *testing.T) {
migrations, err := loadMigrations()
if err != nil {
t.Fatal(err)
}
var up, down string
for _, m := range migrations {
if m.Version == 53 {
up, down = m.UpSQL, m.DownSQL
}
}
for _, required := range []string{"CREATE TABLE qa_ops_master_state", "CREATE TABLE qa_ops_master_records", "CREATE TABLE qa_ops_master_source_links", "CREATE TABLE qa_ops_master_import_batches", "CREATE TABLE qa_ops_master_import_rows", "CREATE FUNCTION qa_ops_master_copy_source_quality", "CREATE FUNCTION qa_ops_master_protect_provenance", "qa_ops_master_normalize(person)", "qa_ops_master_normalize(original_question)", "payload->>'id'=id", "AFTER INSERT OR UPDATE ON qa_ops_mirror_records", "source_id=NEW.id"} {
if !strings.Contains(up, required) {
t.Fatalf("migration 53 lacks durable-master contract %q", required)
}
}
if strings.Contains(up, "AFTER DELETE ON qa_ops_mirror_records") || strings.Contains(up, "UPDATE qa_ops_mirror_records") || strings.Contains(up, "DELETE FROM qa_ops_mirror_records") {
t.Fatal("master migration must neither rewrite mirror business rows nor follow source deletions")
}
for _, required := range []string{"qa_ops_master_import_batches", "row_revision<>1", "o.request_hash IS DISTINCT FROM r.request_hash", "keep additive schema on application rollback"} {
if !strings.Contains(down, required) {
t.Fatalf("down migration may lose locally-owned data: missing %q", required)
}
}
}
+32
View File
@@ -0,0 +1,32 @@
package qaopsmirror
import (
"context"
"knowledge-governance/internal/dobotaiops"
"knowledge-governance/internal/qaopsquality"
"strings"
)
func (s *Store) MasterQualityStore() *qaopsquality.Store {
return qaopsquality.NewStore(s.pool, s.tenant+"::master")
}
const masterBaseSQL = `WITH records AS (SELECT id,asked_at,question,status,identity,quality_hash,payload||jsonb_build_object('record_origin',origin) AS payload FROM qa_ops_master_records WHERE tenant_key=$1)`
var masterMetadataBase = strings.NewReplacer("qa_ops_mirror_records", "qa_ops_master_records", "h.tenant_key=m.tenant_key", "h.tenant_key=m.tenant_key||'::master'").Replace(localMetadataBase)
var masterSummarySQL = masterMetadataBase + ` SELECT count(*),'{}'::text,COALESCE(jsonb_agg(DISTINCT identity),'[]'::jsonb)::text,COALESCE(jsonb_agg(jsonb_build_object('id',id,'input_hash',quality_hash,'marker',marker,'source',source_coverage)),'[]'::jsonb)::text FROM records WHERE $5::text IS NOT NULL AND `
var masterQualityScopeSQL = masterMetadataBase + ` SELECT id,quality_hash,marker,source_coverage::text FROM records WHERE $5::text IS NOT NULL AND `
func (s *Store) ListMasterRecords(ctx context.Context, q dobotaiops.RecordQuery) (dobotaiops.RecordPage, error) {
var revision int64
p, e := s.repositoryFor(&revision, true).ListRecords(ctx, q)
p.LocalRevision = revision
p.Columns = []string{"question_chinese", "answer", "coverage", "is_correct", "product_category", "problem_type", "root_cause", "asked_at", "asker", "question", "language", "trigger_person", "department", "region", "feedback"}
return p, e
}
func (s *Store) ReadMasterQualityScope(ctx context.Context, q dobotaiops.RecordQuery) (dobotaiops.QualityScope, error) {
return s.repositoryFor(nil, true).ReadQualityScope(ctx, q)
}
func (s *Store) GetMasterRecord(ctx context.Context, tenant, id string) (dobotaiops.Record, error) {
return s.repositoryFor(nil, true).GetRecord(ctx, tenant, id)
}
@@ -0,0 +1,43 @@
package qaopsmirror
import (
"context"
"knowledge-governance/internal/dobotaiops"
pg "knowledge-governance/internal/platform/postgres"
"knowledge-governance/internal/qaopsquality"
"testing"
)
func TestMasterAdmissionPreservesPreExistingSourceReviewHistory(t *testing.T) {
p := mirrorFixture(t)
ctx := context.Background()
s := NewStore(p, "fixture-pre-master-review")
queue := pg.NewStore(p)
entry := fixtureEntry("preexisting-review", "retained original answer")
record, e := dobotaiops.DecodeMirrorPayload(string(entry.Payload))
if e != nil {
t.Fatal(e)
}
original := qaopsquality.NewStore(p, s.tenant)
views, e := original.Observe(ctx, []dobotaiops.Record{record})
if e != nil {
t.Fatal(e)
}
v := views[record.ID]
receipt, e := original.Save(ctx, record, qaopsquality.SaveRequest{InputHash: v.InputHash, InputEpoch: v.InputEpoch, BaseRevision: v.Revision, Nonce: "fixture_pre_master_review_01", Reason: "existing fixture review", Fields: qaopsquality.Fields{Coverage: "covered", Correctness: "correct"}}, "fixture-original-reviewer")
if e != nil {
t.Fatal(e)
}
job := claimMirrorJob(t, s, queue, "admit-reviewed")
source := &fixtureSource{rows: []dobotaiops.MirrorRecord{entry}}
if e = s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
copy, e := s.MasterQualityStore().ReadViews(ctx, []dobotaiops.QualityKey{{ID: record.ID, InputHash: record.QualityInputHash}})
if e != nil || copy[record.ID].State != "reviewed" || copy[record.ID].Revision != receipt.SavedRevision || copy[record.ID].LatestReview.Reviewer != "fixture-original-reviewer" {
t.Fatal("existing source review/audit provenance was lost when admitted to master", e)
}
if _, e = original.History(ctx, record.ID, 0, 0); e != nil {
t.Fatal(e)
}
}
@@ -0,0 +1,87 @@
package qaopsmirror
import (
"context"
"encoding/json"
"knowledge-governance/internal/dobotaiops"
pg "knowledge-governance/internal/platform/postgres"
"testing"
"time"
)
func TestMasterTableKeepsLocalEditsAcrossSourceRefreshAndDeletion(t *testing.T) {
p := mirrorFixture(t)
ctx := context.Background()
s := NewStore(p, "fixture-master")
queue := pg.NewStore(p)
entry := fixtureEntry("master-source", "source original answer")
var wire map[string]any
if e := json.Unmarshal(entry.Payload, &wire); e != nil {
t.Fatal(e)
}
wire["asked_at"] = "2026-10-06T08:00:00+08:00"
wire["asker"] = "fixture asker"
entry.Payload, _ = json.Marshal(wire)
keeper := fixtureEntry("other-outside-window", "unrelated retained row")
source := &fixtureSource{rows: []dobotaiops.MirrorRecord{entry, keeper}}
job := claimMirrorJob(t, s, queue, "master-first")
if e := s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
start, _ := time.Parse(time.RFC3339, "2026-10-05T00:00:00+08:00")
q := dobotaiops.RecordQuery{Window: dobotaiops.Window{TenantKey: s.tenant, Start: start, End: start.AddDate(0, 0, 7)}, Limit: 100}
if err := queue.Complete(ctx, job.ID, job.LeaseOwner, time.Now()); err != nil {
t.Fatal(err)
}
page, e := s.ListMasterRecords(ctx, q)
if e != nil || page.Total != 1 || len(page.Items) != 1 {
t.Fatal("new source record was not admitted into the durable master", e)
}
if _, e = p.Exec(ctx, `UPDATE qa_ops_master_records SET payload=jsonb_set(payload,'{answer}','"local edited answer"') WHERE tenant_key=$1 AND id='master-source'`, s.tenant); e != nil {
t.Fatal(e)
}
wire["answer"] = "changed at source"
entry.Payload, _ = json.Marshal(wire)
entry.Key.Fingerprint = stringsForMasterFixture("b")
source.rows = []dobotaiops.MirrorRecord{entry, keeper}
job = claimMirrorJob(t, s, queue, "master-source-update")
if e = s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
if err := queue.Complete(ctx, job.ID, job.LeaseOwner, time.Now()); err != nil {
t.Fatal(err)
}
page, e = s.ListMasterRecords(ctx, q)
if e != nil || page.Items[0].Answer != "local edited answer" {
t.Fatal("source refresh overwrote master edits", e)
}
source.rows = []dobotaiops.MirrorRecord{keeper}
job = claimMirrorJob(t, s, queue, "master-source-delete")
if e = s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
if err := queue.Complete(ctx, job.ID, job.LeaseOwner, time.Now()); err != nil {
t.Fatal(err)
}
page, e = s.ListMasterRecords(ctx, q)
if e != nil || page.Total != 1 || page.Items[0].Answer != "local edited answer" {
t.Fatal("source deletion removed local master content", e)
}
if _, e = p.Exec(ctx, `UPDATE qa_ops_mirror_state SET ready=false,blocked=true WHERE tenant_key=$1`, s.tenant); e != nil {
t.Fatal(e)
}
page, e = s.ListMasterRecords(ctx, q)
if e != nil || page.Total != 1 {
t.Fatal("source availability blocked independent master data", e)
}
if _, e = p.Exec(ctx, `UPDATE qa_ops_master_records SET payload=jsonb_set(payload,'{id}','"relabelled-id"') WHERE tenant_key=$1 AND id='master-source'`, s.tenant); e == nil {
t.Fatal("master payload ID was allowed to diverge from the stable master ID")
}
}
func stringsForMasterFixture(s string) string {
out := ""
for i := 0; i < 64; i++ {
out += s
}
return out
}
@@ -0,0 +1,61 @@
package qaopsmirror
import (
"context"
"knowledge-governance/internal/dobotaiops"
pg "knowledge-governance/internal/platform/postgres"
"knowledge-governance/internal/qaopsquality"
"testing"
"time"
)
func TestMasterHumanReviewIsIndependentOfSourceInputUpdates(t *testing.T) {
p := mirrorFixture(t)
ctx := context.Background()
s := NewStore(p, "fixture-master-quality")
queue := pg.NewStore(p)
source := &fixtureSource{rows: []dobotaiops.MirrorRecord{fixtureEntry("master-review", "first")}}
job := claimMirrorJob(t, s, queue, "review-first")
if e := s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
if e := queue.Complete(ctx, job.ID, job.LeaseOwner, time.Now()); e != nil {
t.Fatal(e)
}
rec, e := s.GetMasterRecord(ctx, s.tenant, "master-review")
if e != nil {
t.Fatal(e)
}
quality := s.MasterQualityStore()
key := []dobotaiops.QualityKey{{ID: rec.ID, InputHash: rec.QualityInputHash}}
views, e := quality.ReadViews(ctx, key)
if e != nil {
t.Fatal(e)
}
v := views[rec.ID]
request := qaopsquality.SaveRequest{InputHash: v.InputHash, InputEpoch: v.InputEpoch, BaseRevision: v.Revision, Nonce: "fixture_master_review_001", Reason: "fixture only", Fields: qaopsquality.Fields{Coverage: "covered", Correctness: "correct", ProductCategory: "fixture-local"}}
if _, e = quality.Save(ctx, rec, request, "fixture-admin"); e != nil {
t.Fatal(e)
}
source.rows = []dobotaiops.MirrorRecord{fixtureEntry("master-review", "changed at source")}
job = claimMirrorJob(t, s, queue, "review-source-change")
if e = s.HandleSync(ctx, *job, func(context.Context) (Source, error) { return source, nil }); e != nil {
t.Fatal(e)
}
views, e = quality.ReadViews(ctx, key)
if e != nil || views[rec.ID].State != "reviewed" || views[rec.ID].Effective.ProductCategory != "fixture-local" {
t.Fatal("source sync invalidated a locally-owned master review", e)
}
current, e := qaopsquality.NewStore(p, s.tenant).ReadViews(ctx, []dobotaiops.QualityKey{{ID: rec.ID, InputHash: sourceRecordHash(t, source.rows[0])}})
if e != nil || current[rec.ID].Revision != 0 {
t.Fatal("master review modified original source evaluation", e)
}
}
func sourceRecordHash(t *testing.T, r dobotaiops.MirrorRecord) string {
t.Helper()
v, e := dobotaiops.DecodeMirrorPayload(string(r.Payload))
if e != nil {
t.Fatal(e)
}
return v.QualityInputHash
}
+31 -13
View File
@@ -98,8 +98,15 @@ func (s *Store) Available(ctx context.Context) error {
}
func (s *Store) Safety() dobotaiops.SourceSafety { s.mu.RLock(); defer s.mu.RUnlock(); return s.safety }
func (s *Store) repository(revision *int64) *dobotaiops.RecordsRepository {
return s.repositoryFor(revision, false)
}
func (s *Store) repositoryFor(revision *int64, master bool) *dobotaiops.RecordsRepository {
base, summary, scope := localBaseSQL, localSummarySQL, localQualityScopeSQL
if master {
base, summary, scope = masterBaseSQL, masterSummarySQL, masterQualityScopeSQL
}
return dobotaiops.NewRecordsRepository(dobotaiops.RecordsRepositoryOptions{
BaseSQL: localBaseSQL, SummarySQL: localSummarySQL, QualityScopeSQL: localQualityScopeSQL, TenantKey: s.tenant,
BaseSQL: base, SummarySQL: summary, QualityScopeSQL: scope, TenantKey: s.tenant,
ValidateWindow: func(w dobotaiops.Window) error {
if w.TenantKey != s.tenant {
return dobotaiops.ErrInvalidQuery
@@ -108,6 +115,9 @@ func (s *Store) repository(revision *int64) *dobotaiops.RecordsRepository {
return e
},
ReadAsOf: func(ctx context.Context, q dobotaiops.RecordQuerier, t *time.Time) error {
if master {
return q.QueryRow(ctx, `SELECT updated_at FROM qa_ops_master_state WHERE tenant_key=$1`, s.tenant).Scan(t)
}
return q.QueryRow(ctx, `SELECT source_observed_at FROM qa_ops_mirror_state WHERE tenant_key=$1`, s.tenant).Scan(t)
},
Transaction: func(ctx context.Context, fn func(context.Context, dobotaiops.RecordQuerier) error) error {
@@ -122,18 +132,26 @@ func (s *Store) repository(revision *int64) *dobotaiops.RecordsRepository {
}
defer func() { _ = tx.Rollback(context.Background()) }()
q := txQuery{tx}
state, e := s.readState(ctx, q)
if e != nil {
return e
}
if state.Blocked {
return dobotaiops.ErrNotReadOnly
}
if !state.Ready {
return dobotaiops.ErrUnavailable
}
if revision != nil {
*revision = state.Revision
if master {
if revision != nil {
if e = q.QueryRow(ctx, `SELECT COALESCE((SELECT revision FROM qa_ops_master_state WHERE tenant_key=$1),0)`, s.tenant).Scan(revision); e != nil {
return dobotaiops.ErrUnavailable
}
}
} else {
state, err := s.readState(ctx, q)
if err != nil {
return err
}
if state.Blocked {
return dobotaiops.ErrNotReadOnly
}
if !state.Ready {
return dobotaiops.ErrUnavailable
}
if revision != nil {
*revision = state.Revision
}
}
if e = fn(ctx, q); e != nil {
return e
@@ -26,7 +26,7 @@ func TestLocalMirrorMetadataDoesNotReadLargeAnswerPayload(t *testing.T) {
if _, e = p.Exec(ctx, `INSERT INTO qa_ops_mirror_state(tenant_key,ready,revision,record_count,source_observed_at,synced_at) VALUES($1,true,1,1,now(),now());`, tenant); e != nil {
t.Fatal(e)
}
if _, e = p.Exec(ctx, `INSERT INTO qa_ops_mirror_records(tenant_key,id,fingerprint,asked_at,question,status,identity,quality_hash,payload) VALUES($1,$2,$3,now()-interval '1 hour','fixture','succeeded','{}',$3,'{}')`, tenant, record.ID, hash); e != nil {
if _, e = p.Exec(ctx, `INSERT INTO qa_ops_mirror_records(tenant_key,id,fingerprint,asked_at,question,status,identity,quality_hash,payload) VALUES($1,$2,$3,now()-interval '1 hour','fixture','succeeded','{}',$3,jsonb_build_object('id',$2::text))`, tenant, record.ID, hash); e != nil {
t.Fatal(e)
}
role := fmt.Sprintf("kg_mirror_metadata_%d", time.Now().UnixNano())
@@ -34,7 +34,7 @@ func TestLocalMirrorMetadataDoesNotReadLargeAnswerPayload(t *testing.T) {
t.Fatal(e)
}
defer func() { p.Exec(ctx, "DROP OWNED BY "+role); p.Exec(ctx, "DROP ROLE "+role) }()
if _, e = p.Exec(ctx, "GRANT SELECT(tenant_key,id,fingerprint,asked_at,question,status,identity,quality_hash) ON qa_ops_mirror_records TO "+role+";GRANT SELECT ON qa_ops_mirror_state,qa_ops_quality_records,qa_ops_quality_inputs,qa_ops_quality_checks TO "+role); e != nil {
if _, e = p.Exec(ctx, "GRANT SELECT(tenant_key,id,fingerprint,asked_at,question,status,identity,quality_hash) ON qa_ops_mirror_records TO "+role+";GRANT SELECT(tenant_key,id,asked_at,question,status,identity,quality_hash) ON qa_ops_master_records TO "+role+";GRANT SELECT ON qa_ops_master_state,qa_ops_mirror_state,qa_ops_quality_records,qa_ops_quality_inputs,qa_ops_quality_checks TO "+role); e != nil {
t.Fatal(e)
}
config := p.Config().Copy()
@@ -48,6 +48,10 @@ func TestLocalMirrorMetadataDoesNotReadLargeAnswerPayload(t *testing.T) {
if e != nil || len(scope.Keys) != 1 || !scope.Keys[0].Marker {
t.Fatal("local metadata still requires SELECT on large payload instead of committed quality evidence", e)
}
scope, e = NewStore(limited, tenant).ReadMasterQualityScope(ctx, dobotaiops.RecordQuery{Window: dobotaiops.Window{TenantKey: tenant, Start: time.Now().Add(-2 * time.Hour), End: time.Now()}, Limit: 100})
if e != nil || len(scope.Keys) != 1 || !scope.Keys[0].Marker {
t.Fatal("master metadata requires forbidden payload access or lost copied coverage evidence", e)
}
}
func TestLocalMirrorStoredQualityUsesReadOnlyBatchedProjection(t *testing.T) {
+197
View File
@@ -0,0 +1,197 @@
"""Hash-bound historical import into the local governance master, never the source DB.
Requires an approved row decision ledger and a genuinely restored native backup.
All employee/answer-bearing subprocess I/O is private; stdout contains counts only.
"""
from __future__ import annotations
import argparse
import collections
import datetime as dt
import hashlib
import io
import json
import os
from pathlib import Path
import posixpath
import re
import subprocess
import uuid
import xml.etree.ElementTree as ET
import zipfile
PG = "knowledge-governance-p1-postgres-1"
TZ = dt.timezone(dt.timedelta(hours=8))
NS = {"s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
STATUSES = {"new_import_candidate", "existing_2_0_duplicate", "missing_key_review"}
def sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def text(value: str) -> str:
return value.replace("\r\n", "\n").replace("\r", "\n").strip()
def literal(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def read_rows(path: Path, expected_sha: str) -> tuple[dict[str,str],list[tuple[int,dict[str,str]]]]:
content=path.read_bytes()
if sha(content)!=expected_sha:
raise ValueError("workbook changed before snapshot parsing")
with zipfile.ZipFile(io.BytesIO(content)) as z:
if z.testzip() is not None:
raise ValueError("workbook ZIP integrity failed")
workbook = ET.fromstring(z.read("xl/workbook.xml"))
sheets = workbook.find("s:sheets", NS)
if sheets is None or len(sheets)==0:
raise ValueError("workbook contains no worksheets")
sheet = sheets[0]
if sheet.attrib["name"] != "问题流水表":
raise ValueError("approved first worksheet not found")
rid = sheet.attrib["{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"]
relationships = ET.fromstring(z.read("xl/_rels/workbook.xml.rels"))
target = next(r.attrib["Target"] for r in relationships if r.attrib["Id"] == rid)
target = target.lstrip("/") if target.startswith("/") else posixpath.normpath("xl/"+target)
shared = []
if "xl/sharedStrings.xml" in z.namelist():
for _, item in ET.iterparse(z.open("xl/sharedStrings.xml"),events=("end",)):
if item.tag.endswith("}si"):
shared.append("".join(n.text or "" for n in item.iter() if n.tag.endswith("}t")))
item.clear()
def value(cell):
v=cell.find("s:v",NS)
if cell.attrib.get("t")=="s" and v is not None:
return shared[int(v.text)]
if cell.attrib.get("t")=="inlineStr":
return "".join(n.text or "" for n in cell.iter() if n.tag.endswith("}t"))
return (v.text or "") if v is not None else ""
headers={}; rows=[]
for _, item in ET.iterparse(z.open(target),events=("end",)):
if not item.tag.endswith("}row"):
continue
rowno=int(item.attrib["r"])
cells={re.sub(r"\d","",c.attrib["r"]):value(c) for c in item}
if rowno==1:
headers=cells;item.clear();continue
raw={c:cells.get(c,"") for c in "ABCDEFGHIJKLMNO"}
if any(text(v) for v in raw.values()):
if any("\x00" in v for v in raw.values()):
raise ValueError("NUL in workbook; no lossy conversion allowed")
rows.append((rowno,raw))
item.clear()
if any(headers.get(c)!=h for c,h in {"H":"提问时间","I":"提问人","J":"原始问题"}.items()):
raise ValueError("approved event columns changed")
return headers,rows
def prepare(args,plan,decisions):
_,rows=read_rows(args.workbook,plan["source_sha256"])
by_row={r["excel_row"]:r for r in decisions}
if len(by_row)!=len(decisions) or len(rows)!=plan["data_rows"] or set(by_row)!={r for r,_ in rows}:
raise ValueError("approved decisions do not cover exactly the input rows")
observed=dt.datetime.now(TZ).isoformat()
for number,raw in rows:
decision=by_row[number]
if decision["status"] not in STATUSES:
raise ValueError("unapproved or ambiguous row disposition")
asked=None
if text(raw["H"]):
asked=dt.datetime.fromisoformat(text(raw["H"]).replace("/","-").replace("Z","+00:00"))
asked=asked.replace(tzinfo=TZ) if asked.tzinfo is None else asked.astimezone(TZ)
if asked.microsecond:
raise ValueError("timestamp precision differs from the approved seconds-only input")
key=[asked.strftime("%Y-%m-%d %H:%M:%S") if asked else "",text(raw["I"]),text(raw["J"])]
key_hash=sha(json.dumps(key,ensure_ascii=False).encode())
if key_hash!=decision["key_sha256"]:
raise ValueError("approved event key changed")
record_id="legacy_"+plan["source_sha256"]+"_"+str(number)
fields={k:raw[c] for k,c in {"question_chinese":"A","coverage":"C","is_correct":"D","product_category":"E","problem_type":"F","root_cause":"G","language":"K"}.items()}
payload={"id":record_id,"status":"historical_import","question":raw["J"],"answer":raw["B"],"asked_at":asked.isoformat() if asked else None,"time_source":"historical_excel","asker":raw["I"],"trigger_person":raw["L"],"department":raw["M"],"region":raw["N"],"references":[],"media":[],"source_coverage":{"verdict":"","reason_code":"","evaluator_version":""},"feedback":{"asker_value":"","up":0,"down":0},"master_fields":fields,"_identity":{},"_quality_observed_at":observed}
payload["_quality_hash"]=sha(json.dumps({"id":record_id,"question":raw["J"],"answer":raw["B"],"status":payload["status"],"references":[],"media":[]},ensure_ascii=False,sort_keys=True).encode())
yield {"row_number":number,"status":decision["status"],"event_key":key_hash,"expected_matches":decision["matched_record_sha256"],"asked_at":payload["asked_at"],"raw":raw,"payload":payload}
def build_sql(args,plan,stage,batch):
tenant=literal(args.tenant);digest=literal(plan["source_sha256"]);sheet=literal(plan["sheet"]);batchsql=literal(batch)
counts=(plan["data_rows"],plan["new_import_candidate_rows"],plan["existing_2_0_duplicate_rows"],plan["missing_key_review_rows"])
sql="BEGIN ISOLATION LEVEL SERIALIZABLE; SET LOCAL statement_timeout='120s'; SET LOCAL lock_timeout='15s';\n"
sql+="SELECT pg_advisory_xact_lock(hashtextextended("+tenant+"||':qa-history-import',0));\n"
sql+="LOCK TABLE qa_ops_master_records IN SHARE ROW EXCLUSIVE MODE;\n"
sql+="CREATE TEMP TABLE incoming_json(raw jsonb) ON COMMIT DROP;\nCOPY incoming_json(raw) FROM STDIN;\n"
for item in stage:
line=json.dumps(item,ensure_ascii=False,separators=(",",":"))
sql+=line.replace("\\","\\\\").replace("\n","\\n").replace("\r","\\r").replace("\t","\\t")+"\n"
sql+="\\.\nCREATE TEMP TABLE incoming ON COMMIT DROP AS SELECT (raw->>'row_number')::int row_number,raw->>'status' status,raw->>'event_key' event_key,raw->'expected_matches' expected_matches,(raw->>'asked_at')::timestamptz asked_at,raw->'raw' original_fields,raw->'payload' payload FROM incoming_json; CREATE INDEX ON incoming(event_key); ANALYZE incoming; ANALYZE qa_ops_master_records;\nCREATE TEMP TABLE admission_status(replayed boolean NOT NULL,inserted integer NOT NULL) ON COMMIT DROP;\n"
sql+="DO $import$ DECLARE replay boolean; bad bigint; BEGIN\n"
sql+="IF (SELECT max(version) FROM schema_migrations)<>53 THEN RAISE EXCEPTION 'unverified governance schema'; END IF;\n"
sql+="IF EXISTS(SELECT 1 FROM incoming WHERE status<>'missing_key_review' AND event_key<>qa_ops_master_event_key((asked_at)::timestamptz,original_fields->>'I',original_fields->>'J')) THEN RAISE EXCEPTION 'SQL and approved Python event identities differ'; END IF;\n"
sql+="SELECT EXISTS(SELECT 1 FROM qa_ops_master_import_batches WHERE tenant_key="+tenant+" AND source_sha256="+digest+" AND sheet_name="+sheet+") INTO replay;\nIF replay THEN\n"
sql+="IF NOT EXISTS(SELECT 1 FROM qa_ops_master_import_batches WHERE tenant_key="+tenant+" AND id="+batchsql+" AND ROW(total_rows,imported_rows,skipped_rows,held_rows)=ROW("+",".join(str(n)for n in counts)+")) THEN RAISE EXCEPTION 'existing import manifest differs'; END IF;\n"
sql+="IF (SELECT count(*) FROM qa_ops_master_import_rows WHERE tenant_key="+tenant+" AND batch_id="+batchsql+")<>"+str(counts[0])+" OR EXISTS(SELECT 1 FROM incoming i LEFT JOIN qa_ops_master_import_rows r ON r.tenant_key="+tenant+" AND r.batch_id="+batchsql+" AND r.row_number=(i.row_number)::int WHERE r.original_fields IS DISTINCT FROM i.original_fields OR r.disposition IS DISTINCT FROM CASE i.status WHEN 'new_import_candidate' THEN 'imported' WHEN 'existing_2_0_duplicate' THEN 'skipped_duplicate' ELSE 'held_missing_key' END) THEN RAISE EXCEPTION 'existing row audit differs'; END IF;\n"
sql+="INSERT INTO admission_status VALUES(true,0);\nELSE\n"
sql+="CREATE TEMP TABLE master_before ON COMMIT DROP AS SELECT id,row_to_json(m)::text AS contents FROM qa_ops_master_records m WHERE tenant_key="+tenant+";\n"
sql+="CREATE TEMP TABLE source_quality_before ON COMMIT DROP AS SELECT 'records' AS kind,to_jsonb(q) AS value FROM qa_ops_quality_records q WHERE tenant_key="+tenant+" UNION ALL SELECT 'inputs',to_jsonb(q) FROM qa_ops_quality_inputs q WHERE tenant_key="+tenant+" UNION ALL SELECT 'checks',to_jsonb(q) FROM qa_ops_quality_checks q WHERE tenant_key="+tenant+" UNION ALL SELECT 'reviews',to_jsonb(q) FROM qa_ops_quality_reviews q WHERE tenant_key="+tenant+";\n"
sql+="IF EXISTS(SELECT 1 FROM incoming i WHERE i.status='new_import_candidate' AND EXISTS(SELECT 1 FROM qa_ops_master_records m WHERE m.tenant_key="+tenant+" AND m.event_key=i.event_key)) OR EXISTS(SELECT 1 FROM incoming i WHERE i.status='existing_2_0_duplicate' AND (SELECT COALESCE(jsonb_agg(encode(sha256(convert_to(m.id,'UTF8')),'hex') ORDER BY m.id),'[]'::jsonb) FROM qa_ops_master_records m WHERE m.tenant_key="+tenant+" AND m.event_key=i.event_key) IS DISTINCT FROM i.expected_matches) THEN RAISE EXCEPTION 'comparison changed; refresh the preview before writing'; END IF;\n"
sql+="INSERT INTO qa_ops_master_import_batches(tenant_key,id,source_sha256,source_name,sheet_name,total_rows,imported_rows,skipped_rows,held_rows) VALUES("+tenant+","+batchsql+","+digest+","+literal(args.workbook.name)+","+sheet+","+",".join(str(n)for n in counts)+");\n"
sql+="INSERT INTO qa_ops_master_records(tenant_key,id,event_key,origin,asked_at,question,status,identity,quality_hash,payload,original_fields,batch_id,source_row) SELECT "+tenant+",payload->>'id',event_key,'historical_excel',(asked_at)::timestamptz,original_fields->>'J','historical_import','{}'::jsonb,payload->>'_quality_hash',payload,original_fields,"+batchsql+",(row_number)::int FROM incoming WHERE status='new_import_candidate';\n"
sql+="GET DIAGNOSTICS bad=ROW_COUNT; IF bad<>"+str(counts[1])+" THEN RAISE EXCEPTION 'imported row count differs'; END IF;\n"
sql+="INSERT INTO qa_ops_master_import_rows(tenant_key,batch_id,row_number,disposition,original_fields,master_id) SELECT "+tenant+","+batchsql+",(i.row_number)::int,CASE i.status WHEN 'new_import_candidate' THEN 'imported' WHEN 'existing_2_0_duplicate' THEN 'skipped_duplicate' ELSE 'held_missing_key' END,i.original_fields,CASE WHEN i.status='missing_key_review' THEN NULL ELSE m.id END FROM incoming i LEFT JOIN qa_ops_master_records m ON m.tenant_key="+tenant+" AND m.event_key=i.event_key AND i.status<>'missing_key_review';\n"
sql+="INSERT INTO qa_ops_master_source_links(tenant_key,source_kind,source_id,master_id) SELECT "+tenant+",'historical_excel',"+digest+"||':'||"+sheet+"||':'||row_number,master_id FROM qa_ops_master_import_rows WHERE tenant_key="+tenant+" AND batch_id="+batchsql+" AND master_id IS NOT NULL;\n"
sql+="IF EXISTS(SELECT 1 FROM master_before b LEFT JOIN qa_ops_master_records m ON m.tenant_key="+tenant+" AND m.id=b.id WHERE row_to_json(m)::text IS DISTINCT FROM b.contents) THEN RAISE EXCEPTION 'existing master record changed'; END IF;\n"
sql+="IF EXISTS((SELECT kind,value FROM source_quality_before EXCEPT SELECT kind,value FROM (SELECT 'records' kind,to_jsonb(q) value FROM qa_ops_quality_records q WHERE tenant_key="+tenant+" UNION ALL SELECT 'inputs',to_jsonb(q) FROM qa_ops_quality_inputs q WHERE tenant_key="+tenant+" UNION ALL SELECT 'checks',to_jsonb(q) FROM qa_ops_quality_checks q WHERE tenant_key="+tenant+" UNION ALL SELECT 'reviews',to_jsonb(q) FROM qa_ops_quality_reviews q WHERE tenant_key="+tenant+") current_source)) THEN RAISE EXCEPTION 'original source evaluation changed'; END IF;\n"
sql+="INSERT INTO admission_status VALUES(false,"+str(counts[1])+");\nEND IF;\nEND $import$;\n"
sql+="SELECT jsonb_build_object('state','IMPORT_TRANSACTION_VERIFIED','batch_id',"+batchsql+",'inserted_rows',inserted,'skipped_rows',"+str(counts[2])+",'held_rows',"+str(counts[3])+",'replayed',replayed,'master_total',(SELECT count(*) FROM qa_ops_master_records WHERE tenant_key="+tenant+"),'production_source_connected',false) FROM admission_status;\nCOMMIT;\n"
return sql
def main():
os.umask(0o077)
parser=argparse.ArgumentParser(description=__doc__)
for name in ["workbook","decisions","plan","backup-verification","evidence-directory"]:
parser.add_argument("--"+name,type=Path,required=True)
parser.add_argument("--tenant",required=True)
parser.add_argument("--database",required=True)
parser.add_argument("--apply",action="store_true",required=True)
parser.add_argument("--live-apply",action="store_true")
args=parser.parse_args()
if not (re.fullmatch(r"kg_history_import_test_[a-f0-9]{12}",args.database) or (args.database=="governance" and args.live_apply)):
raise ValueError("only an isolated history test DB or explicitly approved local governance DB is permitted")
if not args.tenant or len(args.tenant)>128 or any(ord(c)<32 for c in args.tenant):
raise ValueError("invalid fixed governance tenant")
plan=json.loads(args.plan.read_text());decisions=json.loads(args.decisions.read_text())
expected=plan["source_sha256"]
if not re.fullmatch(r"[a-f0-9]{64}",expected) or sha(args.workbook.read_bytes())!=expected:
raise ValueError("approved input SHA differs")
if plan["sheet"]!="问题流水表" or plan["dedup_key"]!=["提问时间","提问人","原始问题"] or plan["within_excel_same_key_extra_rows"] or plan["ambiguous_existing_key_rows"]:
raise ValueError("unapproved matching contract")
approved=collections.Counter(d["status"] for d in decisions)
if approved!={"new_import_candidate":plan["new_import_candidate_rows"],"existing_2_0_duplicate":plan["existing_2_0_duplicate_rows"],"missing_key_review":plan["missing_key_review_rows"]}:
raise ValueError("approved row counts differ")
backup=json.loads(args.backup_verification.read_text());dump=args.backup_verification.parent/backup["backup_file"]
if backup["state"]!="BACKUP_RESTORE_VERIFIED" or not backup["real_restore_completed"] or not backup["all_table_row_counts_equal"] or not backup["all_qa_quality_org_frozen_content_hashes_equal"] or backup["source_workbook_sha256"]!=expected or dump.stat().st_size!=backup["backup_bytes"] or sha(dump.read_bytes())!=backup["backup_sha256"]:
raise ValueError("verified native backup gate failed")
args.evidence_directory.mkdir(mode=0o700,parents=True,exist_ok=True)
stage=list(prepare(args,plan,decisions));batch="xlsx_"+expected+"_"+sha(plan["sheet"].encode())[:12]
sql=build_sql(args,plan,stage,batch)
private=args.evidence_directory/("import-"+uuid.uuid4().hex[:12]+"-private.log")
result=subprocess.run(["docker","exec","-i",PG,"psql","-X","-qAt","-v","ON_ERROR_STOP=1","-U","governance","-d",args.database],input=sql,text=True,capture_output=True,timeout=150)
private.write_text(result.stderr)
if result.returncode:
raise RuntimeError("governance import has no verified commit receipt; read the batch before retrying; the transaction is all-or-nothing; inspect the restricted evidence log")
reports=[json.loads(line) for line in result.stdout.splitlines() if line.startswith('{')]
if len(reports)!=1:
raise RuntimeError("import receipt unavailable; inspect batch read-back before any retry")
report=reports[0]
# A committed receipt is checked by an independent connection, not just COMMIT's exit code.
query="BEGIN READ ONLY; SELECT jsonb_build_object('total',count(*),'imported',count(*)FILTER(WHERE disposition='imported'),'skipped',count(*)FILTER(WHERE disposition='skipped_duplicate'),'held',count(*)FILTER(WHERE disposition='held_missing_key')) FROM qa_ops_master_import_rows WHERE tenant_key="+literal(args.tenant)+" AND batch_id="+literal(batch)+"; COMMIT;"
check=subprocess.run(["docker","exec",PG,"psql","-X","-qAt","-v","ON_ERROR_STOP=1","-U","governance","-d",args.database,"-c",query],capture_output=True,text=True,timeout=30)
if check.returncode:
raise RuntimeError("import committed but independent audit read-back failed; do not claim complete")
actual=json.loads(check.stdout.strip())
if actual!={"total":plan["data_rows"],"imported":plan["new_import_candidate_rows"],"skipped":plan["existing_2_0_duplicate_rows"],"held":plan["missing_key_review_rows"]}:
raise RuntimeError("committed audit row counts differ")
if sha(args.workbook.read_bytes())!=expected:
raise RuntimeError("input changed during import")
report.update(independent_read_back_verified=True,source_unchanged=True)
evidence=args.evidence_directory/("import-"+uuid.uuid4().hex[:12]+"-verification.json")
evidence.write_text(json.dumps(report,ensure_ascii=False,indent=2))
print(json.dumps(report,ensure_ascii=False))
if __name__=="__main__":
main()
+14 -2
View File
@@ -238,6 +238,17 @@ export function QAOperationsView(props: ViewProps) {
</dd>
</>
) : null}
{detail.record_origin ? (
<>
<dt>记录来源</dt>
<dd>
{detail.record_origin === 'historical_excel'
? '历史 Excel 导入(原标注保留)'
: 'DobotAI 2.0 收录副本'}
;主表不随源同步覆盖或删除
</dd>
</>
) : null}
<dt>问答状态</dt>
<dd>{statusLabels[detail.status] ?? detail.status}</dd>
<dt>流程触发人员信息</dt>
@@ -264,6 +275,7 @@ export function QAOperationsView(props: ViewProps) {
<QAQualityEditor
key={detail.id}
recordId={detail.id}
dataset={detail.record_origin ? 'master' : undefined}
quality={detail.quality}
onSaved={props.onQualitySaved ?? (() => {})}
onReload={props.onReloadDetail ?? (() => {})}
@@ -355,7 +367,7 @@ export function QAOperations() {
.catch(() => {
if (!controller.signal.aborted) setStatus(null)
})
void fetchAllQARecords(query, controller.signal, (loaded, total) => {
void fetchAllQARecords({ ...query, dataset: 'master' }, controller.signal, (loaded, total) => {
if (!controller.signal.aborted) setProgress({ loaded, total })
})
.then((result) => {
@@ -398,7 +410,7 @@ export function QAOperations() {
setDetailError(null)
setDetailLoading(true)
try {
const result = await fetchQARecord(id, controller.signal)
const result = await fetchQARecord(id, controller.signal, 'master')
if (!controller.signal.aborted) setDetail(result)
} catch (reason) {
if (!controller.signal.aborted) setDetailError(reason instanceof Error ? reason.message : '详情读取失败')
+13 -15
View File
@@ -1,6 +1,6 @@
import { useEffect, useRef, useState } from 'react'
import type { FormEvent } from 'react'
import { getPayload, fetchQARecord } from './qaOps'
import { getPayload, fetchQARecord, qaDatasetPrefix } from './qaOps'
import type { QAQuery, QARecord } from './qaOps'
import { newOperationID } from './operationKey'
import {
@@ -15,12 +15,14 @@ import type { QAQuality, QualityFields, QualityHistory, QualitySnapshot, Quality
const stamp = (v: string) => new Date(v).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai', hour12: false })
const message = (e: unknown) => (e instanceof Error ? e.message : '质量操作失败,请重试')
interface EditorProps {
dataset?: 'master'
recordId: string
quality: QAQuality
onSaved: (record: QARecord) => void
onReload: () => void
}
export function QAQualityEditor({ recordId, quality, onSaved, onReload }: EditorProps) {
export function QAQualityEditor({ recordId, quality, onSaved, onReload, dataset }: EditorProps) {
const apiPrefix = qaDatasetPrefix(dataset)
const initial = quality.state === 'reviewed' ? quality.effective : emptyQualityFields
const [fields, setFields] = useState<QualityFields>({ ...initial }),
[reason, setReason] = useState(quality.state === 'reviewed' ? quality.latest_review?.reason || '' : ''),
@@ -85,15 +87,11 @@ export function QAQualityEditor({ recordId, quality, onSaved, onReload }: Editor
const signature = JSON.stringify(body)
try {
if (nonce.current.signature !== signature) nonce.current = { signature, value: newOperationID() }
const raw = await getPayload(
`/api/v1/qa-ops/records/${encodeURIComponent(recordId)}/reviews`,
controller.signal,
{
method: 'POST',
headers: { 'Content-Type': 'application/json', 'X-Governance-Action': 'qa-quality' },
body: JSON.stringify({ ...body, nonce: nonce.current.value }),
}
)
const raw = await getPayload(`${apiPrefix}/records/${encodeURIComponent(recordId)}/reviews`, controller.signal, {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'X-Governance-Action': 'qa-quality' },
body: JSON.stringify({ ...body, nonce: nonce.current.value }),
})
if (!raw || typeof raw !== 'object') throw Error('复核回执无效')
const receipt = raw as Record<string, unknown>
parseQAQuality(receipt.view)
@@ -102,7 +100,7 @@ export function QAQualityEditor({ recordId, quality, onSaved, onReload }: Editor
const savedRevision = Number(receipt.saved_revision)
const audit = parseQualityHistory(
await getPayload(
`/api/v1/qa-ops/records/${encodeURIComponent(recordId)}/quality/history?review_before=${savedRevision + 1}`,
`${apiPrefix}/records/${encodeURIComponent(recordId)}/quality/history?review_before=${savedRevision + 1}`,
controller.signal
)
)
@@ -117,7 +115,7 @@ export function QAQualityEditor({ recordId, quality, onSaved, onReload }: Editor
)
)
throw Error('保存后的精确复核记录尚未读回,请用原表单重试,重复请求不会重复留档')
const fresh = await fetchQARecord(recordId, controller.signal)
const fresh = await fetchQARecord(recordId, controller.signal, dataset)
if (!fresh.quality || fresh.quality.revision < savedRevision)
throw Error('复核已留档但当前问答尚未读回,请重新读取')
if (!controller.signal.aborted) {
@@ -144,7 +142,7 @@ export function QAQualityEditor({ recordId, quality, onSaved, onReload }: Editor
try {
const next = parseQualityHistory(
await getPayload(
`/api/v1/qa-ops/records/${encodeURIComponent(recordId)}/quality/history${params.size ? '?' + params : ''}`,
`${apiPrefix}/records/${encodeURIComponent(recordId)}/quality/history${params.size ? '?' + params : ''}`,
controller.signal
)
)
@@ -171,7 +169,7 @@ export function QAQualityEditor({ recordId, quality, onSaved, onReload }: Editor
try {
const next = parseQualitySnapshot(
await getPayload(
`/api/v1/qa-ops/records/${encodeURIComponent(recordId)}/quality/inputs/${epoch}`,
`${apiPrefix}/records/${encodeURIComponent(recordId)}/quality/inputs/${epoch}`,
controller.signal
)
)
+2 -2
View File
@@ -1,5 +1,5 @@
import type { QARecord, QAQuery } from './qaOps'
import { getPayload } from './qaOps'
import { qaDatasetPrefix, getPayload } from './qaOps'
import { parseQualitySummary } from './qaQuality'
export type QACountPoint = { label: string; count: number }
const day = 24 * 60 * 60 * 1000,
@@ -34,7 +34,7 @@ export function qaCategorySeries(items: QARecord[], key: 'product_category' | 'p
}
export async function fetchQADashboardQuality(query: QAQuery, signal?: AbortSignal) {
const value = parseQualitySummary(
await getPayload('/api/v1/qa-ops/quality/query', signal, {
await getPayload(`${qaDatasetPrefix(query.dataset)}/quality/query`, signal, {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'X-Governance-Action': 'qa-quality' },
body: JSON.stringify({ start: query.start, end: query.end, keyword: query.keyword }),
+38
View File
@@ -0,0 +1,38 @@
import { afterEach, expect, it, vi } from 'vitest'
import { fetchQARecords, parseQARecord, qaCellValue } from './qaOps'
import { fixtureRecord } from './qaOps.test'
afterEach(() => vi.unstubAllGlobals())
it('keeps imported historical labels and genuine missing values instead of replacing them with new rule labels', () => {
const raw = {
...fixtureRecord,
record_origin: 'historical_excel',
master_fields: {
question_chinese: '',
coverage: '是',
is_correct: '待确认',
product_category: '原分类',
problem_type: '',
root_cause: '',
language: 'en',
},
}
const item = parseQARecord(raw)
expect(qaCellValue(item, 'coverage')).toBe('是')
expect(qaCellValue(item, 'is_correct')).toBe('待确认')
expect(qaCellValue(item, 'question_chinese')).toBeNull()
expect(qaCellValue(item, 'product_category')).toBe('原分类')
for (const bad of [null, [], { coverage: 123 }, { unapproved: 'x' }])
expect(() => parseQARecord({ ...raw, master_fields: bad })).toThrow('响应无效')
})
it('selects the durable master explicitly and leaves original 2.0 callers on their existing dataset', async () => {
const page = { items: [], columns: [], total: 0, has_more: false, next_cursor: '', as_of: '2026-10-19T00:00:00Z' }
const mocked = vi
.fn()
.mockImplementation(() => Promise.resolve(new Response(JSON.stringify({ data: page }), { status: 200 })))
vi.stubGlobal('fetch', mocked)
const query = { start: '2026-10-05T00:00:00Z', end: '2026-10-12T00:00:00Z', keyword: '', cursor: '' }
await fetchQARecords({ ...query, dataset: 'master' } as Parameters<typeof fetchQARecords>[0])
await fetchQARecords(query)
expect(String(mocked.mock.calls[0][0])).toContain('/api/v1/qa-ops/master/records?')
expect(String(mocked.mock.calls[1][0])).toContain('/api/v1/qa-ops/records?')
})
+28 -3
View File
@@ -82,6 +82,8 @@ export function mergeQAPages(current: QARecordPage, next: QARecordPage): QARecor
}
}
export interface QARecord {
record_origin?: 'dobotai_2_0' | 'historical_excel'
master_fields?: Partial<Record<QAColumn, string>>
quality?: QAQuality
organization?: QARecordOrganization
id: string
@@ -117,6 +119,7 @@ export interface QARecordPage {
as_of: string
}
export interface QAQuery {
dataset?: 'master'
start: string
end: string
keyword: string
@@ -190,6 +193,19 @@ export function parseQARecord(value: unknown): QARecord {
if (typeof row.is_correct !== 'boolean' && row.is_correct !== null) return invalid()
item.is_correct = row.is_correct
}
if (row.record_origin !== undefined) {
if (row.record_origin !== 'dobotai_2_0' && row.record_origin !== 'historical_excel') return invalid()
item.record_origin = row.record_origin
}
if (row.master_fields !== undefined) {
const f = object(row.master_fields)
const fields: Partial<Record<QAColumn, string>> = {}
for (const [key, value] of Object.entries(f)) {
if (!columnKeys.has(key)) return invalid()
fields[key as QAColumn] = text(value)
}
item.master_fields = fields
}
if (row.organization !== undefined) {
const o = object(row.organization)
if (
@@ -270,7 +286,7 @@ export async function fetchQARecords(query: QAQuery, signal?: AbortSignal): Prom
cursor: query.cursor,
limit: String(query.limit ?? 20),
})
return parseQARecordPage(await getPayload(`/api/v1/qa-ops/records?${parameters}`, signal))
return parseQARecordPage(await getPayload(`${qaDatasetPrefix(query.dataset)}/records?${parameters}`, signal))
}
export async function fetchAllQARecords(
query: QAQuery,
@@ -301,8 +317,11 @@ export async function fetchAllQARecords(
return page
}
export async function fetchQARecord(id: string, signal?: AbortSignal): Promise<QARecord> {
return parseQARecord(await getPayload(`/api/v1/qa-ops/records/${encodeURIComponent(id)}`, signal))
export function qaDatasetPrefix(dataset?: 'master') {
return dataset === 'master' ? '/api/v1/qa-ops/master' : '/api/v1/qa-ops'
}
export async function fetchQARecord(id: string, signal?: AbortSignal, dataset?: 'master'): Promise<QARecord> {
return parseQARecord(await getPayload(`${qaDatasetPrefix(dataset)}/records/${encodeURIComponent(id)}`, signal))
}
export async function fetchQAStatus(signal?: AbortSignal): Promise<QAStatus> {
const value = object(await getPayload('/api/v1/qa-ops/status', signal))
@@ -325,6 +344,12 @@ export async function fetchQAStatus(signal?: AbortSignal): Promise<QAStatus> {
}
export function qaCellValue(item: QARecord, column: QAColumn): string | null {
const reviewedQuality =
item.quality?.state === 'reviewed' &&
['coverage', 'is_correct', 'product_category', 'problem_type', 'root_cause'].includes(column)
if (!reviewedQuality && item.master_fields && Object.hasOwn(item.master_fields, column))
return item.master_fields[column] || null
if (column === 'question_chinese')
return item.question_chinese || (/\p{Script=Han}/u.test(item.question) ? item.question : '待翻译')
if (column === 'language') return item.language || '未识别'