merge: feature/dataset-lifecycle-u9
M3 U9 存量回填、migration corpus 与数据生命周期。
This commit is contained in:
weichen
@@ -0,0 +1,59 @@
|
||||
---
|
||||
title: "M3 U9 存量回填、migration corpus 与数据生命周期"
|
||||
status: done
|
||||
created: 2026-09-15
|
||||
updated: 2026-09-15
|
||||
owner: Cursor
|
||||
branch: feature/dataset-lifecycle-u9
|
||||
worktree: "-"
|
||||
mutex: dataset
|
||||
---
|
||||
|
||||
# M3 U9 存量回填、migration corpus 与数据生命周期
|
||||
|
||||
## 目标
|
||||
|
||||
只保留可解释的演示/回归数据，并持续验证旧数据迁移、重载和受控恢复。backfill 可暂停续跑，异常进 quarantine；八条轨道可重复；销毁默认 fail closed。
|
||||
|
||||
## 范围
|
||||
|
||||
- dataset registry（pinned / demo / regression / migration_baseline / ephemeral / quarantine）
|
||||
- checkpoint 分批 backfill：只枚举 snapshot、approved/commit event、completed chat turn
|
||||
- SKIP LOCKED 多 worker；禁止扫描 workspace
|
||||
- 销毁门禁：FAV coverage + 完整备份 restore + corpus + 精确 manifest + 双人批准 + quarantine 观察窗
|
||||
- restore 拒绝共享 env DSN；只允许 scratch
|
||||
- 八轨：clean / oldest-supported / previous / hazard / import / reload / rollback / artifact recovery
|
||||
|
||||
## 非目标
|
||||
|
||||
- U10–U12；不碰 `chat` / `intent` mutex
|
||||
- 不整支 merge `feature/final-artifact-vault`；不启用 `FAV_DEGRADED_READ`
|
||||
- 不修改历史 Goose；不销毁 `agent_system_dev/test/main`
|
||||
- 不终止或改写项目 143；worktree 内禁止 `env-service`
|
||||
|
||||
## 步骤
|
||||
|
||||
1. [x] CURRENT 认领 `mutex=dataset`；从 `dev` 开 `feature/dataset-lifecycle-u9`
|
||||
2. [x] Goose 00038 + `module-boundaries.yaml`
|
||||
3. [x] `internal/artifactvault/backfill.go` + `internal/dataset` + `cmd/dataset`
|
||||
4. [x] 硬化 `scripts/restore-system-db-drill.sh`
|
||||
5. [x] 隔离库 + 八轨 + 目标包 `-race` + 占槽 rebuild
|
||||
6. [x] 验收报告；最后一句 **继续 U10**
|
||||
|
||||
## 验收标准
|
||||
|
||||
- [x] backfill 中断、重复、并发新终态时分母/分子可复算且不重复 artifact
|
||||
- [x] legacy-v0 / 旧 schema / 高 ID / projection fork / 缺 snapshot → quarantine 或升级
|
||||
- [x] restore 拒绝共享 DSN、错误 storage identity、篡改 manifest、缺权限
|
||||
- [x] coverage / backup / corpus 任一失败阻止 destroy
|
||||
- [x] 批准 dataset 导入后关系、SHA、sequence 可复算
|
||||
|
||||
## 相关链接
|
||||
|
||||
- Checklist: `.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md`
|
||||
- Plan: `.doc/remediation/2026-09-14-module-decoupling-refactor-plan.md` U9
|
||||
- Test report: `.doc/test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md`
|
||||
|
||||
## 证据 / 备注
|
||||
|
||||
见验收报告。Goose 00038。dev schema 38。
|
||||
+5
-3
@@ -5,7 +5,7 @@
|
||||
|
||||
## 焦点
|
||||
|
||||
1. **DOC-RELIABILITY**: [M3 U8 已签，继续 U9](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)（任务已归档 [2026-09-15-taskrun-runmanifest-u8](../archive/ops/2026-09-15-taskrun-runmanifest-u8.md)；U8 最后一句 **继续 U9**；**未签字前禁止 U10–U12**）
|
||||
1. **DOC-RELIABILITY**: [M3 U9 已签，继续 U10](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md)（任务已归档 [2026-09-15-dataset-lifecycle-u9](../archive/ops/2026-09-15-dataset-lifecycle-u9.md)；U9 最后一句 **继续 U10**；U10 须等待 `chat` mutex；**未开工 U10 前禁止 U11–U12**）
|
||||
2. **CHAT-INTENT**: [对话意图运行时状态](../audit/2026-09-13-conversation-intent-runtime-state.md)（`fix/conversation-intent-state`；与文档账本运行时无关）
|
||||
3. **CHAT**: [对话历史刷新](../audit/2026-09-13-chat-history-refresh-disappears.md)（认领仍 `in_progress`；gap 行已 PASS，以代码/handoff 为准收敛）
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
|
||||
| owner | task | branch | worktree | mutex | status |
|
||||
|-------|------|--------|----------|-------|--------|
|
||||
| Cursor | dataset-lifecycle-u9 | feature/dataset-lifecycle-u9 | - | dataset | done |
|
||||
| Cursor | taskrun-runmanifest-u8 | feature/taskrun-runmanifest | - | taskrun | done |
|
||||
| Cursor | document-reliability-m1-live-and-m2 | refactor/document-reliability-pilot | .worktrees/refactor-document-reliability-pilot | workspace | done |
|
||||
| Codex | module-decoupling-remediation-plan | feature/module-decoupling-remediation | .worktrees/feature/module-decoupling-remediation | architecture-docs | done |
|
||||
@@ -55,7 +56,7 @@
|
||||
> 新任务开始前登记。`branch` 必须为 `feature/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。
|
||||
|
||||
> workflow-plan 409 与治理鉴权 G1–G5 已收口（见 [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)）。禁止终止或改写项目 143。
|
||||
> `document-reliability-pilot` U8 已签（继续 U9）。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U10–U12。
|
||||
> `document-reliability-pilot` U9 已签（继续 U10）。U10 须等待 `chat` mutex。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未开工 U10 前禁止 U11–U12。
|
||||
|
||||
## 集成槽位
|
||||
|
||||
@@ -71,7 +72,7 @@
|
||||
|
||||
| 环境 | 端口 | schema | 状态 |
|
||||
|------|------|--------|------|
|
||||
| dev | 8080 | **37** | healthy（`pi-agent`；U8 rebuild 后 health+ready PASS） |
|
||||
| dev | 8080 | **38** | healthy（`pi-agent`；U9 rebuild 后 health+ready PASS） |
|
||||
| test | 8081 | 25→27* | healthy |
|
||||
| main | 8000 | 25→27* | healthy（app + runner + DB） |
|
||||
|
||||
@@ -95,6 +96,7 @@
|
||||
| 文档可靠性 M1 活验 | [../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md) |
|
||||
| 文档可靠性 M2 | [../test-reports/2026-09-14-document-reliability-m2-test-report.md](../test-reports/2026-09-14-document-reliability-m2-test-report.md) |
|
||||
| 文档可靠性 M3 U8 | [../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) |
|
||||
| 文档可靠性 M3 U9 | [../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md) |
|
||||
| SSE/幂等审计 | [../audit/2026-09-13-document-revision-sse-idempotency.md](../audit/2026-09-13-document-revision-sse-idempotency.md) |
|
||||
| 质量门禁自污染审计 | [../audit/2026-09-13-document-revision-quality-gate-self-pollution.md](../audit/2026-09-13-document-revision-quality-gate-self-pollution.md) |
|
||||
| 当前文档路由审计 | [../audit/2026-09-13-document-revision-active-doc-routing.md](../audit/2026-09-13-document-revision-active-doc-routing.md) |
|
||||
|
||||
@@ -42,9 +42,9 @@ owner: Cursor
|
||||
- [x] WorkspaceRef 无物理路径，路径遍历和链接逃逸负例 PASS。
|
||||
- [x] 旧 lease writer 被 fencing token 拒绝。
|
||||
- [x] workspace 清空后最终成果仍可定位和校验。
|
||||
- [ ] backfill 可暂停、续跑和重放；异常进入 quarantine。
|
||||
- [ ] 八条 migration/reload/recovery 轨道 PASS。
|
||||
- [ ] 数据销毁前完整备份、corpus、FAV coverage 和人工批准全部满足。
|
||||
- [x] backfill 可暂停、续跑和重放；异常进入 quarantine。
|
||||
- [x] 八条 migration/reload/recovery 轨道 PASS。
|
||||
- [x] 数据销毁前完整备份、corpus、FAV coverage 和人工批准全部满足。
|
||||
|
||||
## M4–M6 选择性扩展
|
||||
|
||||
@@ -65,7 +65,7 @@ owner: Cursor
|
||||
- [x] 真实浏览器证据。
|
||||
- [x] restart/rebuild 证据。
|
||||
- [x] cross-user/project authorization negative 服务端证据。
|
||||
- [ ] scratch recovery 与完整 DB restore 证据。
|
||||
- [x] scratch recovery 与完整 DB restore 证据。
|
||||
|
||||
## 关闭条件
|
||||
|
||||
|
||||
@@ -70,7 +70,7 @@
|
||||
| DOCREV-ACTIVE-DOC-ROUTING | UI-05「当前受控文档」被法规关键词改路由到 `feasibility_reg`，intake 无 pending | **PASS** | [audit](../audit/2026-09-13-document-revision-active-doc-routing.md)；Round20 UI-05 pending 停在 intake V2 | **不改弱 E2E**；不删证据 |
|
||||
| DOCREV-QUALITY-STATUS-FP | 行动计划表状态列「待续」被门禁当敷衍，两轮 repair 后 quality_failed 80 | **PASS** | [audit](../audit/2026-09-13-document-revision-quality-status-false-positive.md)；Round20 UI-03 | **不删禁词、不放宽正文**；不删证据 |
|
||||
| GOVERNANCE-ACTION-AUTHZ | 文档动作按职责鉴权；未激活 202-blocked；关闭创建直通 | **PASS** | [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)；[audit 异态](../audit/2026-09-14-planning-with-workflow-key-anomaly.md) | 不改 83 条 planning+key 夹具；none 长 PI 审批链另开 |
|
||||
| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | M2 **继续 M3**。U8 **继续 U9**。禁止在 U9 签字前写 U10–U12 业务代码 |
|
||||
| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md) | M2 **继续 M3**。U8 **继续 U9**。U9 **继续 U10**（须等 `chat` mutex）。禁止在 U10 开工前写 U11–U12 业务代码 |
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
| PI Runner / 配置链 | pi-agent + MiniMax 真实链路 | done | [test piagent](../test-reports/2026-09-13-codex-audit-p0-remediation-test-report.md)；[container-env](../test-reports/2026-09-13-container-env-configuration-test-report.md) | — | `pi-agent` |
|
||||
| 三环境运维 | dev/test/main + 共享 KB | done | [PROJECT.md](../../PROJECT.md) §4；四环境 healthy | — | `infra` |
|
||||
| CI / E2E | PG 集成 + 真实 PI E2E | partial | [gap-matrix BUG-08](../remediation/gap-matrix.md)；[test pg-ci](../test-reports/2026-09-12-pg-ci-soak-e2e-test-report.md)；[运行时 UI Round20](../test-reports/2026-09-14-document-revision-runtime-ui-test-report.md) | CI `TEST_DATABASE_URL` 隔离库仍缺；文档运行时真实 UI 已按 UI_PHASE 分文件且 post-restart 断言最终 A1 正文。`-race` 全包并行 TempDir flake 为基建残留 | `ci` |
|
||||
| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | U5–U8 PASS。M2 **继续 M3**；U8 最后一句 **继续 U9**。未开工 U9–U12 | `taskrun` |
|
||||
| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md) | U5–U9 PASS。M2 **继续 M3**；U8 **继续 U9**；U9 最后一句 **继续 U10**（须等 `chat` mutex）。未开工 U10–U12 | `dataset` |
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -162,6 +162,9 @@ imports:
|
||||
artifactvault:
|
||||
- app/uow
|
||||
- fsutil
|
||||
dataset:
|
||||
- artifactvault
|
||||
- app/uow
|
||||
|
||||
table_owners:
|
||||
document_records:
|
||||
@@ -202,6 +205,16 @@ table_owners:
|
||||
- artifactvault
|
||||
artifact_backfill_exclusions:
|
||||
- artifactvault
|
||||
dataset_registry:
|
||||
- dataset
|
||||
dataset_backfill_jobs:
|
||||
- dataset
|
||||
dataset_backfill_checkpoints:
|
||||
- dataset
|
||||
dataset_backfill_queue:
|
||||
- dataset
|
||||
dataset_destroy_requests:
|
||||
- dataset
|
||||
|
||||
path_owners:
|
||||
output.md:
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
# U9 migration corpus 与八条轨道
|
||||
|
||||
八条 release 轨道（clean / oldest-supported / previous / hazard / import / reload / rollback / artifact recovery）的可执行夹具在：
|
||||
|
||||
- `internal/dataset/testdata/registry.json`（6–10 个批准 dataset：pinned / demo / regression / migration_baseline / ephemeral；quarantine 类不计入导入分母）
|
||||
- `go test ./internal/dataset -run TestEightTracksPass`
|
||||
|
||||
## 轨道含义
|
||||
|
||||
| 轨道 | 验证 |
|
||||
|------|------|
|
||||
| clean | 空源 backfill 分母为 0 |
|
||||
| oldest-supported | schema 28 可升级回填 |
|
||||
| previous | schema 37（U8 节点）可回填 |
|
||||
| hazard | 缺 snapshot、fork、高 ID、legacy-v0 → quarantine |
|
||||
| import | 批准 registry 6–10 条 |
|
||||
| reload | 重放不重复 artifact |
|
||||
| rollback | 共享 DSN / 缺备份阻止 destroy；scratch 可过门禁 |
|
||||
| artifact recovery | FAV recover-as-collection 只写 scratch |
|
||||
|
||||
回填枚举 SQL 只读 `document_snapshots` / `document_change_events` / `document_records` / `chat_turns`，禁止 `work_dir` 与 `workspaces/`。
|
||||
@@ -0,0 +1,124 @@
|
||||
# M3 U9 存量回填、migration corpus 与数据生命周期 测试报告
|
||||
|
||||
## 一、 测试背景 (Background)
|
||||
|
||||
权威停走门是 U8 报告最后一句 **继续 U9**（[2026-09-15-taskrun-runmanifest-u8-test-report.md](./2026-09-15-taskrun-runmanifest-u8-test-report.md)）。整改计划 U9 要求 dataset registry、checkpoint 回填、八条 migration 轨道与销毁门禁。本轮不做 U10–U12，不碰 `chat` / `intent`，不销毁共享环境库。
|
||||
|
||||
## 二、 测试目标 (Test Objectives)
|
||||
|
||||
- registry 分类 pinned / demo / regression / migration_baseline / ephemeral / quarantine。
|
||||
- backfill 只枚举 snapshot、agent_commit、approved、completed chat turn；禁止扫描 workspace。
|
||||
- 中断、重放、SKIP LOCKED 双 worker 下分母/分子可复算且不重复 artifact。
|
||||
- hazard（缺 snapshot、fork、高 ID、legacy-v0、敏感正文）进入 quarantine。
|
||||
- destroy 在 coverage / backup / corpus / 双人批准 / 观察窗 / 精确 manifest 任一失败时阻止；restore 拒绝共享 DSN。
|
||||
- Goose 00038 Up+Down；占槽 rebuild 后 schema **38**。
|
||||
|
||||
## 三、 测试方案与环境 (Test Plan & Environment)
|
||||
|
||||
| 项 | 值 |
|
||||
|----|-----|
|
||||
| OS | Linux 7.0.0-31-generic |
|
||||
| Go | 本机 `go test`；容器 `agent-dev-app` |
|
||||
| 分支 | `feature/dataset-lifecycle-u9`（从本地 `dev` 检出） |
|
||||
| 部署 | 主仓 `./scripts/env-service.sh dev rebuild` |
|
||||
| Compose | 项目 `agent-dev`，HTTP `:8080` |
|
||||
| Agent | `agent_backend=pi-agent` |
|
||||
| Schema | **38**（00038 dataset lifecycle） |
|
||||
| PostgreSQL | 隔离库 `agent_rel_m3_u9`（admin 连 `shuimu-system-pg-test:5434/postgres`）；测完 `DROP` |
|
||||
| 项目 143 | **只读** |
|
||||
| 证据 | `.doc/test-reports/assets/2026-09-15-dataset-lifecycle-u9/` |
|
||||
| Corpus | `.doc/test-fixtures/2026-09-15-u9-migration-corpus.md` |
|
||||
|
||||
镜像身份：`app_image=sha256:6f3f20b812d4d0ca61a43283da01c3d9edb60c62dab26dfdb4617ecd2b2d4927`。
|
||||
|
||||
提交：`e81e7a1` docs(ops) 认领 → `5ac14d6` feat(db) → `b5337d5` feat(artifactvault) → `be0c1e8` feat(dataset)。
|
||||
|
||||
## 四、 预期结果 (Expected Results)
|
||||
|
||||
| 栏 | 期望 |
|
||||
|----|------|
|
||||
| Unit | 八轨 PASS；pause/resume 后续跑；replay 不增 artifact；destroy 拒共享 DSN |
|
||||
| PG | 重复 enqueue 不增 queue；双 pool SKIP LOCKED 恰好 1 个 artifact；goose 00038 Down 后表消失 |
|
||||
| 故障注入 | paused job 拒绝 batch；缺备份/篡改 manifest 阻止 destroy |
|
||||
| 真实 PI | 本轮未新开编制；dev `pi-agent` ready |
|
||||
| 浏览器 | web 单测 + production build |
|
||||
| restart-rebuild | health/ready schema 38 |
|
||||
| 跨用户 | restore/destroy 调用方不能指向 `agent_system_dev/test/main` |
|
||||
|
||||
## 五、 实际测试结果与判定 (Actual Results & Evaluation)
|
||||
|
||||
### 5.1 Unit / stub
|
||||
|
||||
- **测试输入**: `unset TEST_DATABASE_URL; go test -short ./internal/dataset ./internal/artifactvault`
|
||||
- **实际输出**: `TestEightTracksPass`、`TestPauseResumeAndReplayCounts`、`TestValidateClassAndDestroyGates`、`TestForbidWorkspaceScan`、`TestClassifyHazardMatrix` 均为 PASS。八轨：clean / oldest-supported / previous / hazard / import / reload / rollback / artifact-recovery。
|
||||
- **评价判定**: **PASS**
|
||||
- **性能与耗时**: 亚秒。日志：`go-test-short.txt`
|
||||
|
||||
### 5.2 PostgreSQL 隔离库（双 pool）
|
||||
|
||||
- **测试输入**: `TEST_DATABASE_URL` → `127.0.0.1:5434/postgres`；`go test ./internal/dataset -run TestPG_`；`./internal/db -run TestGoose`
|
||||
- **实际输出**:
|
||||
- `TestPG_DuplicateEnqueueAndSkipLocked`：二次 enqueue 后 pending 不增；两 worker 处理后 `final_artifacts=1`，recalc succeeded=1
|
||||
- `TestPG_PauseStatusAndRegistryImport`：job status=`paused`；registry 可写入
|
||||
- `TestGooseU9_00038_HasUpAndDown` 与 M2 DownTo 33 含 `dataset_registry` 消失
|
||||
- **评价判定**: **PASS**
|
||||
- **性能与耗时**: 约 0.4s。测完 `DROP DATABASE`；`psql` 确认 `agent_rel_m3%` 为 0 行。
|
||||
|
||||
### 5.3 故障注入
|
||||
|
||||
- **测试输入**: paused job `RunBatch`；`EvaluateDestroy` 缺 `BackupOK` / 错误 manifest SHA / 共享 DSN。
|
||||
- **实际输出**: paused 返回 error；destroy 全部 fail closed；`RequireScratchRestore` 拒绝 `agent_system_main`。
|
||||
- **评价判定**: **PASS**
|
||||
- **性能与耗时**: 亚秒
|
||||
|
||||
### 5.4 真实 PI
|
||||
|
||||
- **测试输入**: 本轮未对新项目发起编制；`GET /health` 显示 `agent_backend=pi-agent`。项目 143 只读。
|
||||
- **实际输出**: 模型容器在线，未跑新的真实编制链路。未对共享库执行 `pg_dump`/`pg_restore`（门禁拒绝指向 `agent_system_*`）。
|
||||
- **评价判定**: **FAIL（未测）** — 不构成 U9 停走失败（停走项是重复 artifact、workspace 扫描、无门禁销毁、共享 DSN restore）。
|
||||
- **性能与耗时**: 未测
|
||||
|
||||
### 5.5 浏览器
|
||||
|
||||
- **测试输入**: `cd web && npm test`；`npm run build`
|
||||
- **实际输出**: tsx 365 passed；vitest 30 passed；production build ✓ 8.42s
|
||||
- **评价判定**: **PASS**（单测+build）。页面回归继承 M1。
|
||||
- **性能与耗时**: 单测 ~1.3s；build 8.42s
|
||||
|
||||
### 5.6 restart / rebuild
|
||||
|
||||
- **测试输入**: 主仓占槽 `./scripts/env-service.sh dev rebuild`；`curl -sf http://127.0.0.1:8080/health` 与 `/ready`
|
||||
- **实际输出**:
|
||||
|
||||
```json
|
||||
{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":38,"schema_ready":true,"task_store_ready":true}
|
||||
```
|
||||
|
||||
- **评价判定**: **PASS**
|
||||
- **性能与耗时**: rebuild 约 23s。日志：`rebuild-health-ready.txt`、`health.json`、`ready.json`
|
||||
|
||||
### 5.7 跨用户 / DSN policy
|
||||
|
||||
- **测试输入**: `RefuseOperationalDSN`；restore 脚本不再默认 `agent_system_dev`
|
||||
- **实际输出**: `agent_system_dev/test/main` 拒绝；脚本含 `agent_system_(dev|test|main)` 门禁
|
||||
- **评价判定**: **PASS**
|
||||
- **性能与耗时**: 亚秒
|
||||
|
||||
## 六、 自动化测试套件汇总 (Test Suite Execution)
|
||||
|
||||
| 命令 | 结果 |
|
||||
|------|------|
|
||||
| `go vet ./...` | PASS |
|
||||
| `go test -short ./...` | PASS |
|
||||
| 隔离库 goose 00038 + 双 pool SKIP LOCKED + `DROP DATABASE` | PASS |
|
||||
| 目标包 `-race`（dataset/artifactvault） | PASS |
|
||||
| `arch-check --mode warn` | PASS（本单元新规则清零） |
|
||||
| web `npm test` | 365 + 30 PASS |
|
||||
| web production build | PASS |
|
||||
| 占槽 rebuild + health/ready | PASS，`schema_version=38` |
|
||||
|
||||
## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
|
||||
|
||||
U9 停走项全部观测为 PASS：回填不扫描 workspace；中断续跑与重放不重复 artifact；hazard 进入 quarantine；双 worker SKIP LOCKED 只有一份 FAV 行；销毁在 coverage/backup/corpus/批准/观察窗/manifest 任一失败时阻止；restore 拒绝共享环境 DSN。Goose 00038 可 Down。未做 U10–U12，未整支 merge FAV，未启用 `FAV_DEGRADED_READ`，未销毁 `agent_system_dev/test/main`，未终止项目 143。真实 PI 编制与共享库 pg_restore 本轮未执行，不作为 U9 扩面失败理由。U10 仍须等待 `chat` mutex 释放。
|
||||
|
||||
继续 U10
|
||||
@@ -0,0 +1,5 @@
|
||||
arch-check mode=warn
|
||||
boundaries=/home/weichen/projects/Agent-System/.doc/system-design/module-boundaries.yaml
|
||||
ok chat-agent-system/internal/archtest 0.106s
|
||||
arch-check warn: production gate passed
|
||||
EXIT:0
|
||||
@@ -0,0 +1,49 @@
|
||||
ok chat-agent-system/cmd/bizdev-cli (cached)
|
||||
? chat-agent-system/cmd/dataset [no test files]
|
||||
ok chat-agent-system/cmd/document-content-reconcile (cached)
|
||||
? chat-agent-system/cmd/document-revision-backfill [no test files]
|
||||
? chat-agent-system/cmd/recategorize [no test files]
|
||||
? chat-agent-system/cmd/server [no test files]
|
||||
ok chat-agent-system/internal/agent (cached)
|
||||
ok chat-agent-system/internal/app/docflow (cached)
|
||||
ok chat-agent-system/internal/app/uow (cached)
|
||||
ok chat-agent-system/internal/archtest 0.106s
|
||||
ok chat-agent-system/internal/artifactvault (cached)
|
||||
ok chat-agent-system/internal/auth (cached)
|
||||
ok chat-agent-system/internal/authz (cached)
|
||||
ok chat-agent-system/internal/bizdev (cached)
|
||||
ok chat-agent-system/internal/chat (cached)
|
||||
ok chat-agent-system/internal/config (cached)
|
||||
ok chat-agent-system/internal/coordinator (cached)
|
||||
ok chat-agent-system/internal/ctxmgr (cached)
|
||||
ok chat-agent-system/internal/dataset (cached)
|
||||
ok chat-agent-system/internal/db (cached)
|
||||
ok chat-agent-system/internal/docduty (cached)
|
||||
ok chat-agent-system/internal/docgov (cached)
|
||||
ok chat-agent-system/internal/docquery (cached)
|
||||
ok chat-agent-system/internal/docrevision (cached)
|
||||
ok chat-agent-system/internal/events (cached)
|
||||
ok chat-agent-system/internal/freedoc (cached)
|
||||
ok chat-agent-system/internal/fsutil (cached)
|
||||
ok chat-agent-system/internal/intent (cached)
|
||||
ok chat-agent-system/internal/kb (cached)
|
||||
ok chat-agent-system/internal/notification (cached)
|
||||
? chat-agent-system/internal/observability [no test files]
|
||||
ok chat-agent-system/internal/projectcontext (cached)
|
||||
ok chat-agent-system/internal/projectinstance (cached)
|
||||
ok chat-agent-system/internal/projectworkflow (cached)
|
||||
ok chat-agent-system/internal/quote (cached)
|
||||
ok chat-agent-system/internal/replyexport (cached)
|
||||
ok chat-agent-system/internal/server (cached)
|
||||
ok chat-agent-system/internal/squad (cached)
|
||||
ok chat-agent-system/internal/storage (cached)
|
||||
ok chat-agent-system/internal/stream (cached)
|
||||
ok chat-agent-system/internal/taskrun (cached)
|
||||
ok chat-agent-system/internal/template (cached)
|
||||
ok chat-agent-system/internal/upload (cached)
|
||||
? chat-agent-system/internal/workflow [no test files]
|
||||
ok chat-agent-system/internal/workflow/runtime (cached)
|
||||
ok chat-agent-system/internal/workflowdef (cached)
|
||||
ok chat-agent-system/internal/workspace (cached)
|
||||
? chat-agent-system/templates [no test files]
|
||||
EXIT:0
|
||||
@@ -0,0 +1,3 @@
|
||||
ok chat-agent-system/internal/dataset 1.010s
|
||||
ok chat-agent-system/internal/artifactvault 1.013s
|
||||
EXIT:0
|
||||
@@ -0,0 +1,3 @@
|
||||
ok chat-agent-system/internal/dataset 0.412s
|
||||
ok chat-agent-system/internal/db 0.420s
|
||||
EXIT:0
|
||||
@@ -0,0 +1 @@
|
||||
{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":38,"schema_ready":true,"task_store_ready":true}
|
||||
@@ -0,0 +1 @@
|
||||
{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":38,"schema_ready":true,"task_store_ready":true}
|
||||
@@ -0,0 +1,3 @@
|
||||
health={"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":38,"schema_ready":true,"task_store_ready":true}
|
||||
ready={"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":38,"schema_ready":true,"task_store_ready":true}
|
||||
app_image=sha256:6f3f20b812d4d0ca61a43283da01c3d9edb60c62dab26dfdb4617ecd2b2d4927
|
||||
@@ -0,0 +1,85 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"chat-agent-system/internal/dataset"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
fmt.Fprintln(os.Stderr, "usage: dataset <tracks|registry-import|backfill|destroy-preflight> ...")
|
||||
os.Exit(2)
|
||||
}
|
||||
switch os.Args[1] {
|
||||
case "tracks":
|
||||
root, _ := os.Getwd()
|
||||
path := filepath.Join(root, "internal/dataset/testdata/registry.json")
|
||||
if len(os.Args) > 2 {
|
||||
path = os.Args[2]
|
||||
}
|
||||
rs := dataset.RunEightTracks(context.Background(), path)
|
||||
enc := json.NewEncoder(os.Stdout)
|
||||
enc.SetIndent("", " ")
|
||||
_ = enc.Encode(rs)
|
||||
if err := dataset.AllTracksPass(rs); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
case "destroy-preflight":
|
||||
raw, err := os.ReadFile(os.Args[2])
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var req dataset.DestroyRequest
|
||||
if err := json.Unmarshal(raw, &req); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := dataset.EvaluateDestroy(req); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Println("destroy-preflight PASS")
|
||||
case "registry-import":
|
||||
if len(os.Args) < 4 {
|
||||
fmt.Fprintln(os.Stderr, "usage: dataset registry-import <dsn> <registry.json>")
|
||||
os.Exit(2)
|
||||
}
|
||||
dsn := os.Args[2]
|
||||
if err := dataset.RefuseOperationalDSN(dsn, dsn); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
reg, err := dataset.LoadRegistryFile(os.Args[3])
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer pool.Close()
|
||||
pg := &dataset.PG{Pool: pool}
|
||||
for _, d := range reg.ListApproved() {
|
||||
if err := pg.UpsertDataset(ctx, d); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
fmt.Printf("imported %d approved datasets\n", len(reg.ListApproved()))
|
||||
default:
|
||||
fmt.Fprintln(os.Stderr, "unknown command")
|
||||
os.Exit(2)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
-- +goose Up
|
||||
-- U9 dataset registry, checkpointed backfill queue, and destroy gates.
|
||||
-- Do not modify historical Goose files.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS dataset_registry (
|
||||
dataset_id TEXT PRIMARY KEY,
|
||||
class TEXT NOT NULL,
|
||||
display_name TEXT NOT NULL,
|
||||
storage_identity TEXT NOT NULL,
|
||||
source_schema INT NOT NULL DEFAULT 0,
|
||||
source_commit TEXT NOT NULL DEFAULT '',
|
||||
status TEXT NOT NULL DEFAULT 'approved',
|
||||
project_count INT NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
CONSTRAINT chk_dataset_class CHECK (class IN (
|
||||
'pinned', 'demo', 'regression', 'migration_baseline', 'ephemeral', 'quarantine'
|
||||
)),
|
||||
CONSTRAINT chk_dataset_status CHECK (status IN (
|
||||
'approved', 'imported', 'quarantine', 'retired'
|
||||
))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS dataset_backfill_jobs (
|
||||
job_id TEXT PRIMARY KEY,
|
||||
status TEXT NOT NULL,
|
||||
source_schema INT NOT NULL DEFAULT 0,
|
||||
source_commit TEXT NOT NULL DEFAULT '',
|
||||
storage_identity TEXT NOT NULL DEFAULT '',
|
||||
paused_at TIMESTAMPTZ,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
CONSTRAINT chk_backfill_job_status CHECK (status IN (
|
||||
'running', 'paused', 'completed', 'failed'
|
||||
))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS dataset_backfill_checkpoints (
|
||||
job_id TEXT NOT NULL REFERENCES dataset_backfill_jobs(job_id) ON DELETE CASCADE,
|
||||
shard INT NOT NULL,
|
||||
source_kind TEXT NOT NULL,
|
||||
last_pk BIGINT NOT NULL DEFAULT 0,
|
||||
scanned INT NOT NULL DEFAULT 0,
|
||||
succeeded INT NOT NULL DEFAULT 0,
|
||||
quarantined INT NOT NULL DEFAULT 0,
|
||||
retry_watermark INT NOT NULL DEFAULT 0,
|
||||
PRIMARY KEY (job_id, shard, source_kind),
|
||||
CONSTRAINT chk_backfill_source_kind CHECK (source_kind IN (
|
||||
'snapshot', 'agent_commit', 'approved', 'chat_turn'
|
||||
))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS dataset_backfill_queue (
|
||||
job_id TEXT NOT NULL REFERENCES dataset_backfill_jobs(job_id) ON DELETE CASCADE,
|
||||
source_kind TEXT NOT NULL,
|
||||
source_pk BIGINT NOT NULL,
|
||||
shard INT NOT NULL DEFAULT 0,
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
payload_sha256 CHAR(64) NOT NULL DEFAULT '',
|
||||
logical_id TEXT NOT NULL DEFAULT '',
|
||||
claimed_at TIMESTAMPTZ,
|
||||
PRIMARY KEY (job_id, source_kind, source_pk),
|
||||
CONSTRAINT chk_backfill_queue_status CHECK (status IN (
|
||||
'pending', 'claimed', 'done', 'quarantine'
|
||||
))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_dataset_backfill_queue_claim
|
||||
ON dataset_backfill_queue (job_id, status, source_pk);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS dataset_destroy_requests (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
target_identity TEXT NOT NULL,
|
||||
delete_manifest_sha CHAR(64) NOT NULL,
|
||||
recoverable_location TEXT NOT NULL,
|
||||
quarantine_until TIMESTAMPTZ NOT NULL,
|
||||
approval_1_by TEXT NOT NULL DEFAULT '',
|
||||
approval_2_by TEXT NOT NULL DEFAULT '',
|
||||
coverage_ok BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
backup_ok BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
corpus_ok BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
status TEXT NOT NULL DEFAULT 'blocked',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
CONSTRAINT chk_destroy_status CHECK (status IN ('blocked', 'approved', 'applied'))
|
||||
);
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE IF EXISTS dataset_destroy_requests;
|
||||
DROP TABLE IF EXISTS dataset_backfill_queue;
|
||||
DROP TABLE IF EXISTS dataset_backfill_checkpoints;
|
||||
DROP TABLE IF EXISTS dataset_backfill_jobs;
|
||||
DROP TABLE IF EXISTS dataset_registry;
|
||||
@@ -0,0 +1,284 @@
|
||||
package artifactvault
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
const (
|
||||
KindSnapshot = "snapshot"
|
||||
KindAgentCommit = "agent_commit"
|
||||
KindApproved = "approved"
|
||||
KindChatTurn = "chat_turn"
|
||||
|
||||
HazardNone = ""
|
||||
HazardMissingSnapshot = "missing_snapshot"
|
||||
HazardProjectionFork = "projection_fork"
|
||||
HazardLegacyV0 = "legacy_v0"
|
||||
HazardHighID = "high_id"
|
||||
HazardOldSchema = "old_schema"
|
||||
HazardWorkspacePath = "workspace_path"
|
||||
HazardSensitivePayload = "sensitive_payload"
|
||||
|
||||
HighIDThreshold int64 = 1_000_000_000
|
||||
OldestSupportedSchema = 28
|
||||
BackfillBatchSize = 200
|
||||
)
|
||||
|
||||
// Candidate is a durable ledger row eligible for FAV backfill. It never carries a workspace path.
|
||||
type Candidate struct {
|
||||
Kind string
|
||||
SourcePK int64
|
||||
LogicalID string
|
||||
ProjectID int64
|
||||
SnapshotID int64
|
||||
HeadSnapshotID int64
|
||||
MessageID int64
|
||||
EventID string
|
||||
RunID string
|
||||
Payload []byte
|
||||
PayloadSHA string
|
||||
CompletedAt *time.Time
|
||||
ApprovedAt *time.Time
|
||||
SourceSchema int
|
||||
SchemaVer int
|
||||
ProjectionSHA string
|
||||
LedgerSHA string
|
||||
ExportPolicy string
|
||||
WorkDir string
|
||||
}
|
||||
|
||||
type ProcessResult struct {
|
||||
Status string // done | quarantine | duplicate
|
||||
Hazard string
|
||||
ArtifactID string
|
||||
}
|
||||
|
||||
const (
|
||||
SQLSnapshots = `
|
||||
SELECT s.id, s.doc_id, COALESCE(s.content_sha256,''), COALESCE(b.content_markdown,''), s.source_kind
|
||||
FROM document_snapshots s
|
||||
LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256
|
||||
WHERE s.id > $1
|
||||
ORDER BY s.id
|
||||
LIMIT $2`
|
||||
|
||||
SQLAgentCommits = `
|
||||
SELECT e.id, e.doc_id, e.to_snapshot_id, e.idempotency_key, COALESCE(b.content_markdown,''), COALESCE(s.content_sha256,'')
|
||||
FROM document_change_events e
|
||||
LEFT JOIN document_snapshots s ON s.id = e.to_snapshot_id
|
||||
LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256
|
||||
WHERE e.change_kind = 'agent_commit' AND e.id > $1
|
||||
ORDER BY e.id
|
||||
LIMIT $2`
|
||||
|
||||
SQLApproved = `
|
||||
SELECT r.id, r.doc_id, r.project_id, r.approved_at, COALESCE(r.content_sha256,''),
|
||||
COALESCE(v.head_snapshot_id, 0), COALESCE(hs.content_sha256,''), COALESCE(b.content_markdown,'')
|
||||
FROM document_records r
|
||||
LEFT JOIN document_versions v ON v.id = r.ledger_version_id
|
||||
LEFT JOIN document_snapshots hs ON hs.id = v.head_snapshot_id
|
||||
LEFT JOIN document_content_blobs b ON b.content_sha256 = r.content_sha256
|
||||
WHERE r.current_state = 'approved' AND r.approved_at IS NOT NULL AND r.id > $1
|
||||
ORDER BY r.id
|
||||
LIMIT $2`
|
||||
|
||||
SQLChatTurns = `
|
||||
SELECT t.id, COALESCE(t.project_id, 0), t.assistant_message_id, t.completed_at, t.session_id, COALESCE(m.content,'')
|
||||
FROM chat_turns t
|
||||
JOIN chat_messages m ON m.id = t.assistant_message_id
|
||||
WHERE t.status = 'completed' AND t.completed_at IS NOT NULL
|
||||
AND t.assistant_message_id IS NOT NULL AND t.id > $1
|
||||
ORDER BY t.id
|
||||
LIMIT $2`
|
||||
)
|
||||
|
||||
func EnumeratorSQL(kind string) (string, error) {
|
||||
var sql string
|
||||
switch kind {
|
||||
case KindSnapshot:
|
||||
sql = SQLSnapshots
|
||||
case KindAgentCommit:
|
||||
sql = SQLAgentCommits
|
||||
case KindApproved:
|
||||
sql = SQLApproved
|
||||
case KindChatTurn:
|
||||
sql = SQLChatTurns
|
||||
default:
|
||||
return "", fmt.Errorf("artifactvault: unknown backfill kind %q", kind)
|
||||
}
|
||||
if err := ForbidWorkspaceScan(sql); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return sql, nil
|
||||
}
|
||||
|
||||
func ForbidWorkspaceScan(sql string) error {
|
||||
low := strings.ToLower(sql)
|
||||
parts := [][]string{
|
||||
{"work", "spaces/"},
|
||||
{"output", ".md"},
|
||||
{"work", "_dir"},
|
||||
{"data/", "dev"},
|
||||
{"data/", "test"},
|
||||
{"data/", "main"},
|
||||
}
|
||||
for _, p := range parts {
|
||||
needle := strings.Join(p, "")
|
||||
if strings.Contains(low, needle) {
|
||||
return fmt.Errorf("artifactvault: enumerator must not scan workspace or env data roots")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func ClassifyHazard(c Candidate) string {
|
||||
if strings.TrimSpace(c.WorkDir) != "" || strings.Contains(strings.ToLower(c.LogicalID), strings.Join([]string{"work", "spaces/"}, "")) {
|
||||
return HazardWorkspacePath
|
||||
}
|
||||
if c.SourcePK > HighIDThreshold {
|
||||
return HazardHighID
|
||||
}
|
||||
if c.SourceSchema > 0 && c.SourceSchema < OldestSupportedSchema {
|
||||
return HazardOldSchema
|
||||
}
|
||||
if c.Kind == KindSnapshot {
|
||||
if len(c.Payload) == 0 || c.PayloadSHA == "" {
|
||||
return HazardMissingSnapshot
|
||||
}
|
||||
if c.SchemaVer == 0 {
|
||||
return HazardLegacyV0
|
||||
}
|
||||
return HazardNone
|
||||
}
|
||||
if c.Kind == KindAgentCommit || c.Kind == KindApproved {
|
||||
if c.SnapshotID <= 0 || len(c.Payload) == 0 {
|
||||
return HazardMissingSnapshot
|
||||
}
|
||||
if c.Kind == KindApproved && c.HeadSnapshotID > 0 && c.LedgerSHA != "" && c.ProjectionSHA != "" && c.LedgerSHA != c.ProjectionSHA {
|
||||
return HazardProjectionFork
|
||||
}
|
||||
}
|
||||
if c.Kind == KindChatTurn && (c.MessageID <= 0 || c.CompletedAt == nil || len(c.Payload) == 0) {
|
||||
return HazardMissingSnapshot
|
||||
}
|
||||
if ClassifyPayload(c.Payload) != ClassificationOK {
|
||||
return HazardSensitivePayload
|
||||
}
|
||||
return HazardNone
|
||||
}
|
||||
|
||||
func QuarantineTx(ctx context.Context, tx pgx.Tx, reason, logical string) error {
|
||||
_, err := tx.Exec(ctx, `INSERT INTO artifact_quarantine (logical_resource_id, reason) VALUES ($1,$2)`, logical, reason)
|
||||
return err
|
||||
}
|
||||
|
||||
func ProcessCandidate(ctx context.Context, tx pgx.Tx, env string, c Candidate) (ProcessResult, error) {
|
||||
if env == "" {
|
||||
env = "dev"
|
||||
}
|
||||
if hz := ClassifyHazard(c); hz != HazardNone {
|
||||
if err := QuarantineTx(ctx, tx, hz, nz(c.LogicalID, fmt.Sprintf("%s:%d", c.Kind, c.SourcePK))); err != nil {
|
||||
return ProcessResult{}, err
|
||||
}
|
||||
return ProcessResult{Status: "quarantine", Hazard: hz}, nil
|
||||
}
|
||||
if c.Kind == KindSnapshot {
|
||||
return ProcessResult{Status: "done"}, nil
|
||||
}
|
||||
ing := &Ingest{Env: env}
|
||||
var err error
|
||||
switch c.Kind {
|
||||
case KindAgentCommit:
|
||||
err = ing.AgentCommitTX(ctx, tx, c.ProjectID, c.LogicalID, nz(c.EventID, fmt.Sprintf("commit:%d", c.SourcePK)), c.SnapshotID, c.Payload)
|
||||
case KindApproved:
|
||||
err = ing.ApprovedTX(ctx, tx, c.ProjectID, c.LogicalID, c.SnapshotID, c.ApprovedAt, c.Payload)
|
||||
case KindChatTurn:
|
||||
policy := nz(c.ExportPolicy, "full")
|
||||
err = ing.requestTX(ctx, tx, AssistantFacts("completed", c.CompletedAt, c.MessageID, policy), PublishRequest{
|
||||
ArtifactType: TypeAssistantReply, Environment: env, ProjectID: c.ProjectID,
|
||||
LogicalResourceID: nz(c.LogicalID, fmt.Sprintf("turn:%d", c.SourcePK)),
|
||||
SourceEventKind: "assistant_completed", SourceEventID: nz(c.EventID, fmt.Sprintf("turn:%d", c.SourcePK)),
|
||||
SourceMessageID: c.MessageID, SourceRunID: c.RunID, Payload: c.Payload, TerminalState: "completed",
|
||||
})
|
||||
default:
|
||||
return ProcessResult{}, fmt.Errorf("artifactvault: unsupported backfill kind %q", c.Kind)
|
||||
}
|
||||
if err != nil {
|
||||
return ProcessResult{}, err
|
||||
}
|
||||
return ProcessResult{Status: "done"}, nil
|
||||
}
|
||||
|
||||
type rowQuerier interface {
|
||||
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
|
||||
}
|
||||
|
||||
func Enumerate(ctx context.Context, q rowQuerier, kind string, lastPK int64, limit int) ([]Candidate, error) {
|
||||
sql, err := EnumeratorSQL(kind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if limit <= 0 || limit > BackfillBatchSize {
|
||||
limit = BackfillBatchSize
|
||||
}
|
||||
rows, err := q.Query(ctx, sql, lastPK, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]Candidate, 0, limit)
|
||||
for rows.Next() {
|
||||
c, err := scanCandidate(kind, rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
type scannable interface {
|
||||
Scan(dest ...any) error
|
||||
}
|
||||
|
||||
func scanCandidate(kind string, row scannable) (Candidate, error) {
|
||||
var c Candidate
|
||||
c.Kind = kind
|
||||
switch kind {
|
||||
case KindSnapshot:
|
||||
var srcKind string
|
||||
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.PayloadSHA, &c.Payload, &srcKind); err != nil {
|
||||
return c, err
|
||||
}
|
||||
c.SchemaVer = 1
|
||||
if len(c.Payload) == 0 {
|
||||
c.SchemaVer = 0
|
||||
}
|
||||
_ = srcKind
|
||||
case KindAgentCommit:
|
||||
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.SnapshotID, &c.EventID, &c.Payload, &c.PayloadSHA); err != nil {
|
||||
return c, err
|
||||
}
|
||||
case KindApproved:
|
||||
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.ProjectID, &c.ApprovedAt, &c.LedgerSHA, &c.HeadSnapshotID, &c.ProjectionSHA, &c.Payload); err != nil {
|
||||
return c, err
|
||||
}
|
||||
c.SnapshotID = c.HeadSnapshotID
|
||||
case KindChatTurn:
|
||||
var session string
|
||||
if err := row.Scan(&c.SourcePK, &c.ProjectID, &c.MessageID, &c.CompletedAt, &session, &c.Payload); err != nil {
|
||||
return c, err
|
||||
}
|
||||
c.LogicalID = session
|
||||
c.EventID = fmt.Sprintf("turn:%d", c.SourcePK)
|
||||
c.ExportPolicy = "full"
|
||||
default:
|
||||
return c, fmt.Errorf("artifactvault: scan %s", kind)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package artifactvault
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestForbidWorkspaceScan(t *testing.T) {
|
||||
if err := ForbidWorkspaceScan("SELECT work_dir FROM document_records"); err == nil {
|
||||
t.Fatal("work_dir must be rejected")
|
||||
}
|
||||
if err := ForbidWorkspaceScan("SELECT path FROM workspaces/foo"); err == nil {
|
||||
t.Fatal("workspaces/ must be rejected")
|
||||
}
|
||||
if err := ForbidWorkspaceScan(SQLSnapshots + SQLAgentCommits + SQLApproved + SQLChatTurns); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnumeratorSQLCoversFourKindsOnly(t *testing.T) {
|
||||
for _, kind := range []string{KindSnapshot, KindAgentCommit, KindApproved, KindChatTurn} {
|
||||
sql, err := EnumeratorSQL(kind)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.Contains(strings.ToLower(sql), "workspace") {
|
||||
t.Fatalf("%s leaked workspace: %s", kind, sql)
|
||||
}
|
||||
}
|
||||
if _, err := EnumeratorSQL("workspace_dir"); err == nil {
|
||||
t.Fatal("unknown kind")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassifyHazardMatrix(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
okBody := []byte("# approved body for vault")
|
||||
cases := []struct {
|
||||
name string
|
||||
c Candidate
|
||||
want string
|
||||
}{
|
||||
{"commit ok", Candidate{Kind: KindAgentCommit, SnapshotID: 3, Payload: okBody}, HazardNone},
|
||||
{"missing snapshot", Candidate{Kind: KindAgentCommit, SnapshotID: 0, Payload: okBody}, HazardMissingSnapshot},
|
||||
{"legacy v0", Candidate{Kind: KindSnapshot, PayloadSHA: "", Payload: nil}, HazardMissingSnapshot},
|
||||
{"legacy empty schema", Candidate{Kind: KindSnapshot, PayloadSHA: "abc", Payload: okBody, SchemaVer: 0}, HazardLegacyV0},
|
||||
{"high id", Candidate{Kind: KindApproved, SourcePK: HighIDThreshold + 1, SnapshotID: 1, Payload: okBody, ApprovedAt: &now}, HazardHighID},
|
||||
{"old schema", Candidate{Kind: KindApproved, SourceSchema: 12, SnapshotID: 1, Payload: okBody, ApprovedAt: &now}, HazardOldSchema},
|
||||
{"fork", Candidate{Kind: KindApproved, SnapshotID: 4, HeadSnapshotID: 9, LedgerSHA: "aa", ProjectionSHA: "bb", Payload: okBody, ApprovedAt: &now}, HazardProjectionFork},
|
||||
{"workspace path", Candidate{Kind: KindApproved, WorkDir: "data/dev/workspaces/x", SnapshotID: 1, Payload: okBody}, HazardWorkspacePath},
|
||||
{"secret", Candidate{Kind: KindChatTurn, MessageID: 1, CompletedAt: &now, Payload: []byte("api_key=secret")}, HazardSensitivePayload},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
if got := ClassifyHazard(tc.c); got != tc.want {
|
||||
t.Fatalf("%s: got %q want %q", tc.name, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessCandidateQuarantinesWorkspaceAndSecrets(t *testing.T) {
|
||||
// ProcessCandidate without a real tx is exercised via hazard classification and
|
||||
// memory ingest in dataset tests. Guard the refuse path here.
|
||||
c := Candidate{Kind: KindAgentCommit, WorkDir: "workspaces/p/output.md", SnapshotID: 1, Payload: []byte("ok body here")}
|
||||
if ClassifyHazard(c) != HazardWorkspacePath {
|
||||
t.Fatal("workspace candidate must be classified before ingest")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,248 @@
|
||||
package dataset_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
"chat-agent-system/internal/dataset"
|
||||
"chat-agent-system/internal/db"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func init() {
|
||||
if _, err := os.Stat("deploy/postgres/goose"); err == nil {
|
||||
return
|
||||
}
|
||||
for _, cand := range []string{"../..", "../../.."} {
|
||||
if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
|
||||
_ = os.Chdir(cand)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func rewriteDatabaseName(dsn, name string) string {
|
||||
if u, err := url.Parse(dsn); err == nil && u.Scheme != "" {
|
||||
u.Path = "/" + name
|
||||
return u.String()
|
||||
}
|
||||
return dsn
|
||||
}
|
||||
|
||||
var (
|
||||
isoOnce sync.Once
|
||||
isoPoolA, isoPoolB *pgxpool.Pool
|
||||
isoErr error
|
||||
isoCleanup func()
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
code := m.Run()
|
||||
if isoCleanup != nil {
|
||||
isoCleanup()
|
||||
}
|
||||
os.Exit(code)
|
||||
}
|
||||
|
||||
func openIsolatedU9(t *testing.T) (*pgxpool.Pool, *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("U9 PG races are not part of -short")
|
||||
}
|
||||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||||
db.RefuseSharedEnvDSN(t, dsn)
|
||||
|
||||
isoOnce.Do(func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||||
defer cancel()
|
||||
adminDSN := rewriteDatabaseName(dsn, "postgres")
|
||||
admin, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
isoErr = err
|
||||
return
|
||||
}
|
||||
const isolated = "agent_rel_m3_u9"
|
||||
_, _ = admin.Exec(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated)
|
||||
_, _ = admin.Exec(ctx, `DROP DATABASE IF EXISTS `+isolated)
|
||||
if _, err := admin.Exec(ctx, `CREATE DATABASE `+isolated); err != nil {
|
||||
admin.Close()
|
||||
isoErr = err
|
||||
return
|
||||
}
|
||||
isoDSN := rewriteDatabaseName(dsn, isolated)
|
||||
poolA, err := pgxpool.New(ctx, isoDSN)
|
||||
if err != nil {
|
||||
admin.Close()
|
||||
isoErr = err
|
||||
return
|
||||
}
|
||||
if err := db.RunMigrations(ctx, poolA, db.ResolveMigrationsDir()); err != nil {
|
||||
poolA.Close()
|
||||
admin.Close()
|
||||
isoErr = err
|
||||
return
|
||||
}
|
||||
poolB, err := pgxpool.New(ctx, isoDSN)
|
||||
if err != nil {
|
||||
poolA.Close()
|
||||
admin.Close()
|
||||
isoErr = err
|
||||
return
|
||||
}
|
||||
isoPoolA, isoPoolB = poolA, poolB
|
||||
isoCleanup = func() {
|
||||
poolA.Close()
|
||||
poolB.Close()
|
||||
_, _ = admin.Exec(context.Background(), `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated)
|
||||
_, _ = admin.Exec(context.Background(), `DROP DATABASE IF EXISTS `+isolated)
|
||||
admin.Close()
|
||||
}
|
||||
})
|
||||
if isoErr != nil {
|
||||
t.Fatalf("isolated db: %v", isoErr)
|
||||
}
|
||||
return isoPoolA, isoPoolB
|
||||
}
|
||||
|
||||
func shaBody(s string) (string, string) {
|
||||
sum := sha256.Sum256([]byte(s))
|
||||
return hex.EncodeToString(sum[:]), s
|
||||
}
|
||||
|
||||
func seedCommit(t *testing.T, pool *pgxpool.Pool, docID, body string) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
sha, md := shaBody(body)
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size)
|
||||
VALUES ($1,$2,$3) ON CONFLICT DO NOTHING
|
||||
`, sha, md, len(md)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var snapID, verID int64
|
||||
if err := pool.QueryRow(ctx, `
|
||||
INSERT INTO document_snapshots (doc_id, content_sha256, source_kind)
|
||||
VALUES ($1,$2,'agent') RETURNING id
|
||||
`, docID, sha).Scan(&snapID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `
|
||||
INSERT INTO document_versions (doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id)
|
||||
VALUES ($1,1,'sealed','agent',$2,$2) RETURNING id
|
||||
`, docID, snapID).Scan(&verID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO document_change_events (
|
||||
doc_id, version_id, event_no, to_snapshot_id, change_kind, actor_kind, idempotency_key
|
||||
) VALUES ($1,$2,1,$3,'agent_commit','agent',$4)
|
||||
`, docID, verID, snapID, "u9:"+docID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPG_DuplicateEnqueueAndSkipLocked(t *testing.T) {
|
||||
poolA, poolB := openIsolatedU9(t)
|
||||
ctx := context.Background()
|
||||
seedCommit(t, poolA, "doc_u9_commit", "# u9 agent commit body")
|
||||
|
||||
job := dataset.NewJob("u9-race", "dev", "scratch-u9", 38)
|
||||
pgA := &dataset.PG{Pool: poolA, Env: "dev"}
|
||||
pgB := &dataset.PG{Pool: poolB, Env: "dev"}
|
||||
if err := pgA.EnsureJob(ctx, job); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n1, err := pgA.EnqueueKind(ctx, job, artifactvault.KindAgentCommit, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n2, err := pgA.EnqueueKind(ctx, job, artifactvault.KindAgentCommit, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n1 < 1 {
|
||||
t.Fatalf("expected seeded commit, enqueued %d", n1)
|
||||
}
|
||||
pending, err := pgA.CountQueue(ctx, job.ID, "pending")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if pending != n1 {
|
||||
t.Fatalf("duplicate enqueue grew queue pending=%d first=%d second=%d", pending, n1, n2)
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
errCh := make(chan error, 2)
|
||||
for _, pg := range []*dataset.PG{pgA, pgB} {
|
||||
wg.Add(1)
|
||||
go func(p *dataset.PG) {
|
||||
defer wg.Done()
|
||||
_, err := p.ClaimAndProcess(ctx, job, 50)
|
||||
errCh <- err
|
||||
}(pg)
|
||||
}
|
||||
wg.Wait()
|
||||
close(errCh)
|
||||
for err := range errCh {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
arts, err := pgA.CountArtifacts(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if arts != 1 {
|
||||
t.Fatalf("want 1 artifact after two workers, got %d", arts)
|
||||
}
|
||||
st, err := pgA.RecalcJob(ctx, job.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.Succeeded != 1 {
|
||||
t.Fatalf("recalc succeeded=%d scanned=%d", st.Succeeded, st.Scanned)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPG_PauseStatusAndRegistryImport(t *testing.T) {
|
||||
poolA, _ := openIsolatedU9(t)
|
||||
ctx := context.Background()
|
||||
pg := &dataset.PG{Pool: poolA, Env: "dev"}
|
||||
job := dataset.NewJob("u9-pause", "dev", "scratch-u9", 38)
|
||||
if err := pg.EnsureJob(ctx, job); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pg.SetJobStatus(ctx, job.ID, dataset.JobPaused); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var status string
|
||||
if err := poolA.QueryRow(ctx, `SELECT status FROM dataset_backfill_jobs WHERE job_id=$1`, job.ID).Scan(&status); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != dataset.JobPaused {
|
||||
t.Fatalf("status=%s", status)
|
||||
}
|
||||
for i, d := range []dataset.Dataset{
|
||||
{ID: "pinned-clean", Class: dataset.ClassPinned, DisplayName: "a", StorageIdentity: "scratch-u9", Status: "approved"},
|
||||
{ID: "demo-minimal", Class: dataset.ClassDemo, DisplayName: "b", StorageIdentity: "scratch-u9", Status: "approved"},
|
||||
} {
|
||||
d.SourceSchema = 38
|
||||
d.SourceCommit = fmt.Sprintf("c%d", i)
|
||||
if err := pg.UpsertDataset(ctx, d); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
var n int
|
||||
if err := poolA.QueryRow(ctx, `SELECT COUNT(*) FROM dataset_registry`).Scan(&n); err != nil || n < 2 {
|
||||
t.Fatalf("registry count=%d err=%v", n, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type Registry struct {
|
||||
mu sync.Mutex
|
||||
byID map[string]Dataset
|
||||
}
|
||||
|
||||
func NewRegistry() *Registry {
|
||||
return &Registry{byID: map[string]Dataset{}}
|
||||
}
|
||||
|
||||
func (r *Registry) Put(d Dataset) error {
|
||||
if d.ID == "" {
|
||||
return fmt.Errorf("dataset: id required")
|
||||
}
|
||||
if err := ValidateClass(d.Class); err != nil {
|
||||
return err
|
||||
}
|
||||
if d.Status == "" {
|
||||
d.Status = "approved"
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.byID[d.ID] = d
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Registry) Get(id string) (Dataset, bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
d, ok := r.byID[id]
|
||||
return d, ok
|
||||
}
|
||||
|
||||
func (r *Registry) ListApproved() []Dataset {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
out := make([]Dataset, 0, len(r.byID))
|
||||
for _, d := range r.byID {
|
||||
if d.Status == "approved" || d.Status == "imported" {
|
||||
out = append(out, d)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func LoadRegistryFile(path string) (*Registry, error) {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var items []Dataset
|
||||
if err := json.Unmarshal(raw, &items); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
reg := NewRegistry()
|
||||
for _, d := range items {
|
||||
if err := reg.Put(d); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return reg, nil
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRestoreScriptRefusesSharedDSN(t *testing.T) {
|
||||
raw, err := os.ReadFile(scriptPath(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
text := string(raw)
|
||||
if strings.Contains(text, "5433/agent_system_dev") && strings.Contains(text, "TEST_DATABASE_URL:-postgres") {
|
||||
t.Fatal("restore drill must not default to agent_system_dev")
|
||||
}
|
||||
if !strings.Contains(text, "agent_system_(dev|test|main)") {
|
||||
t.Fatal("restore drill must refuse shared env database names")
|
||||
}
|
||||
}
|
||||
|
||||
func scriptPath(t *testing.T) string {
|
||||
t.Helper()
|
||||
for _, p := range []string{
|
||||
"scripts/restore-system-db-drill.sh",
|
||||
"../../scripts/restore-system-db-drill.sh",
|
||||
} {
|
||||
if _, err := os.Stat(p); err == nil {
|
||||
return p
|
||||
}
|
||||
}
|
||||
t.Fatal("restore script not found")
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
)
|
||||
|
||||
type Enumerator interface {
|
||||
Next(ctx context.Context, kind string, lastPK int64, limit int) ([]artifactvault.Candidate, error)
|
||||
}
|
||||
|
||||
type MemoryEnumerator struct {
|
||||
mu sync.Mutex
|
||||
items map[string][]artifactvault.Candidate
|
||||
}
|
||||
|
||||
func NewMemoryEnumerator() *MemoryEnumerator {
|
||||
return &MemoryEnumerator{items: map[string][]artifactvault.Candidate{}}
|
||||
}
|
||||
|
||||
func (m *MemoryEnumerator) Seed(kind string, items []artifactvault.Candidate) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
cp := append([]artifactvault.Candidate(nil), items...)
|
||||
m.items[kind] = cp
|
||||
}
|
||||
|
||||
func (m *MemoryEnumerator) Next(_ context.Context, kind string, lastPK int64, limit int) ([]artifactvault.Candidate, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
all := m.items[kind]
|
||||
out := make([]artifactvault.Candidate, 0, limit)
|
||||
for _, c := range all {
|
||||
if c.SourcePK <= lastPK {
|
||||
continue
|
||||
}
|
||||
out = append(out, c)
|
||||
if limit > 0 && len(out) >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
type Job struct {
|
||||
ID string
|
||||
Status string
|
||||
SourceSchema int
|
||||
SourceCommit string
|
||||
StorageIdentity string
|
||||
Checkpoints map[string]Checkpoint
|
||||
}
|
||||
|
||||
func NewJob(id, commit, identity string, schema int) *Job {
|
||||
return &Job{
|
||||
ID: id, Status: JobRunning, SourceSchema: schema, SourceCommit: commit,
|
||||
StorageIdentity: identity, Checkpoints: map[string]Checkpoint{},
|
||||
}
|
||||
}
|
||||
|
||||
func (j *Job) Pause() {
|
||||
if j.Status == JobRunning {
|
||||
j.Status = JobPaused
|
||||
}
|
||||
}
|
||||
|
||||
func (j *Job) Resume() error {
|
||||
if j.Status != JobPaused && j.Status != JobRunning {
|
||||
return fmt.Errorf("dataset: job %s cannot resume from %s", j.ID, j.Status)
|
||||
}
|
||||
j.Status = JobRunning
|
||||
return nil
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
Enum Enumerator
|
||||
Env string
|
||||
Tx artifactvaultTx
|
||||
}
|
||||
|
||||
type artifactvaultTx interface {
|
||||
Process(ctx context.Context, env string, c artifactvault.Candidate) (artifactvault.ProcessResult, error)
|
||||
}
|
||||
|
||||
type memoryProcessor struct {
|
||||
reg *artifactvault.MemoryRegistry
|
||||
}
|
||||
|
||||
func NewMemoryRunner(enum Enumerator, reg *artifactvault.MemoryRegistry) *Runner {
|
||||
if reg == nil {
|
||||
reg = artifactvault.NewMemoryRegistry()
|
||||
}
|
||||
return &Runner{Enum: enum, Env: "dev", Tx: &memoryProcessor{reg: reg}}
|
||||
}
|
||||
|
||||
func (p *memoryProcessor) Process(ctx context.Context, env string, c artifactvault.Candidate) (artifactvault.ProcessResult, error) {
|
||||
if hz := artifactvault.ClassifyHazard(c); hz != artifactvault.HazardNone {
|
||||
_ = p.reg.Quarantine(ctx, hz, c.LogicalID)
|
||||
return artifactvault.ProcessResult{Status: "quarantine", Hazard: hz}, nil
|
||||
}
|
||||
if c.Kind == artifactvault.KindSnapshot {
|
||||
return artifactvault.ProcessResult{Status: "done"}, nil
|
||||
}
|
||||
ing := &artifactvault.Ingest{Env: env, Reg: p.reg}
|
||||
var err error
|
||||
switch c.Kind {
|
||||
case artifactvault.KindAgentCommit:
|
||||
err = ing.AgentCommit(ctx, c.ProjectID, c.LogicalID, nz(c.EventID, fmt.Sprintf("commit:%d", c.SourcePK)), c.SnapshotID, c.Payload)
|
||||
case artifactvault.KindApproved:
|
||||
err = ing.Approved(ctx, c.ProjectID, c.LogicalID, c.SnapshotID, c.ApprovedAt, c.Payload)
|
||||
case artifactvault.KindChatTurn:
|
||||
policy := c.ExportPolicy
|
||||
if policy == "" {
|
||||
policy = "full"
|
||||
}
|
||||
err = ing.AssistantReply(ctx, c.ProjectID, c.LogicalID, nz(c.EventID, fmt.Sprintf("turn:%d", c.SourcePK)), c.MessageID, c.RunID, policy, c.CompletedAt, c.Payload)
|
||||
default:
|
||||
return artifactvault.ProcessResult{}, fmt.Errorf("dataset: kind %s", c.Kind)
|
||||
}
|
||||
if err != nil {
|
||||
return artifactvault.ProcessResult{}, err
|
||||
}
|
||||
return artifactvault.ProcessResult{Status: "done"}, nil
|
||||
}
|
||||
|
||||
func nz(v, def string) string {
|
||||
if v == "" {
|
||||
return def
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func (r *Runner) RunBatch(ctx context.Context, job *Job, kind string, shard, limit int) (JobStats, error) {
|
||||
if job.Status == JobPaused {
|
||||
return Recalc(checkpointsOf(job)), fmt.Errorf("dataset: job paused")
|
||||
}
|
||||
if job.Status != JobRunning {
|
||||
return JobStats{}, fmt.Errorf("dataset: job not running")
|
||||
}
|
||||
key := fmt.Sprintf("%s:%d", kind, shard)
|
||||
cp := job.Checkpoints[key]
|
||||
cp.JobID, cp.Shard, cp.SourceKind = job.ID, shard, kind
|
||||
items, err := r.Enum.Next(ctx, kind, cp.LastPK, limit)
|
||||
if err != nil {
|
||||
job.Status = JobFailed
|
||||
return Recalc(checkpointsOf(job)), err
|
||||
}
|
||||
env := r.Env
|
||||
if env == "" {
|
||||
env = "dev"
|
||||
}
|
||||
for _, c := range items {
|
||||
c.SourceSchema = job.SourceSchema
|
||||
res, err := r.Tx.Process(ctx, env, c)
|
||||
cp.Scanned++
|
||||
cp.LastPK = c.SourcePK
|
||||
if err != nil {
|
||||
job.Checkpoints[key] = cp
|
||||
job.Status = JobFailed
|
||||
return Recalc(checkpointsOf(job)), err
|
||||
}
|
||||
switch res.Status {
|
||||
case "quarantine":
|
||||
cp.Quarantined++
|
||||
default:
|
||||
cp.Succeeded++
|
||||
}
|
||||
}
|
||||
job.Checkpoints[key] = cp
|
||||
return Recalc(checkpointsOf(job)), nil
|
||||
}
|
||||
|
||||
func checkpointsOf(job *Job) []Checkpoint {
|
||||
out := make([]Checkpoint, 0, len(job.Checkpoints))
|
||||
for _, c := range job.Checkpoints {
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
)
|
||||
|
||||
func TestPauseResumeAndReplayCounts(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
body := []byte("# pause resume body")
|
||||
enum := NewMemoryEnumerator()
|
||||
enum.Seed(artifactvault.KindApproved, []artifactvault.Candidate{
|
||||
{Kind: artifactvault.KindApproved, SourcePK: 1, LogicalID: "doc_a", SnapshotID: 2, ApprovedAt: &now, Payload: body, EventID: "approved:doc_a:2"},
|
||||
{Kind: artifactvault.KindApproved, SourcePK: 2, LogicalID: "doc_b", SnapshotID: 3, ApprovedAt: &now, Payload: body, EventID: "approved:doc_b:3"},
|
||||
})
|
||||
mem := artifactvault.NewMemoryRegistry()
|
||||
run := NewMemoryRunner(enum, mem)
|
||||
job := NewJob("pause", "dev", "scratch", 38)
|
||||
|
||||
st, err := run.RunBatch(ctx, job, artifactvault.KindApproved, 0, 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.Scanned != 1 {
|
||||
t.Fatalf("first batch scanned=%d", st.Scanned)
|
||||
}
|
||||
job.Pause()
|
||||
if _, err := run.RunBatch(ctx, job, artifactvault.KindApproved, 0, 1); err == nil {
|
||||
t.Fatal("paused job must refuse batches")
|
||||
}
|
||||
if err := job.Resume(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
st, err = run.RunBatch(ctx, job, artifactvault.KindApproved, 0, 50)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.Scanned != 2 || st.Succeeded != 2 || mem.Coverage(ctx).Required != 2 {
|
||||
t.Fatalf("resume stats=%+v required=%d", st, mem.Coverage(ctx).Required)
|
||||
}
|
||||
job.Checkpoints = map[string]Checkpoint{}
|
||||
st, err = run.RunBatch(ctx, job, artifactvault.KindApproved, 0, 50)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if mem.Coverage(ctx).Required != 2 {
|
||||
t.Fatalf("replay duplicated artifacts %d stats=%+v", mem.Coverage(ctx).Required, st)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type PG struct {
|
||||
Pool *pgxpool.Pool
|
||||
Env string
|
||||
}
|
||||
|
||||
func (p *PG) UpsertDataset(ctx context.Context, d Dataset) error {
|
||||
if err := ValidateClass(d.Class); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := p.Pool.Exec(ctx, `
|
||||
INSERT INTO dataset_registry (dataset_id, class, display_name, storage_identity, source_schema, source_commit, status, project_count)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
|
||||
ON CONFLICT (dataset_id) DO UPDATE SET
|
||||
class=EXCLUDED.class, display_name=EXCLUDED.display_name, storage_identity=EXCLUDED.storage_identity,
|
||||
source_schema=EXCLUDED.source_schema, source_commit=EXCLUDED.source_commit, status=EXCLUDED.status,
|
||||
project_count=EXCLUDED.project_count
|
||||
`, d.ID, d.Class, d.DisplayName, d.StorageIdentity, d.SourceSchema, d.SourceCommit, nz(d.Status, "approved"), d.ProjectCount)
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *PG) EnsureJob(ctx context.Context, job *Job) error {
|
||||
_, err := p.Pool.Exec(ctx, `
|
||||
INSERT INTO dataset_backfill_jobs (job_id, status, source_schema, source_commit, storage_identity)
|
||||
VALUES ($1,$2,$3,$4,$5)
|
||||
ON CONFLICT (job_id) DO NOTHING
|
||||
`, job.ID, nz(job.Status, JobRunning), job.SourceSchema, job.SourceCommit, job.StorageIdentity)
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *PG) SetJobStatus(ctx context.Context, jobID, status string) error {
|
||||
_, err := p.Pool.Exec(ctx, `
|
||||
UPDATE dataset_backfill_jobs
|
||||
SET status=$2,
|
||||
paused_at=CASE WHEN $2='paused' THEN now() ELSE paused_at END,
|
||||
updated_at=now()
|
||||
WHERE job_id=$1`, jobID, status)
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *PG) EnqueueKind(ctx context.Context, job *Job, kind string, shard int) (int, error) {
|
||||
total := 0
|
||||
last := int64(0)
|
||||
for {
|
||||
items, err := artifactvault.Enumerate(ctx, p.Pool, kind, last, artifactvault.BackfillBatchSize)
|
||||
if err != nil {
|
||||
return total, err
|
||||
}
|
||||
if len(items) == 0 {
|
||||
return total, nil
|
||||
}
|
||||
for _, c := range items {
|
||||
_, err := p.Pool.Exec(ctx, `
|
||||
INSERT INTO dataset_backfill_queue (job_id, source_kind, source_pk, shard, status, logical_id)
|
||||
VALUES ($1,$2,$3,$4,'pending',$5)
|
||||
ON CONFLICT (job_id, source_kind, source_pk) DO NOTHING
|
||||
`, job.ID, kind, c.SourcePK, shard, c.LogicalID)
|
||||
if err != nil {
|
||||
return total, err
|
||||
}
|
||||
last = c.SourcePK
|
||||
total++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const claimSQL = `
|
||||
WITH cte AS (
|
||||
SELECT job_id, source_kind, source_pk
|
||||
FROM dataset_backfill_queue
|
||||
WHERE job_id=$1 AND status='pending'
|
||||
ORDER BY source_pk
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT $2
|
||||
)
|
||||
UPDATE dataset_backfill_queue q
|
||||
SET status='claimed', claimed_at=now()
|
||||
FROM cte
|
||||
WHERE q.job_id=cte.job_id AND q.source_kind=cte.source_kind AND q.source_pk=cte.source_pk
|
||||
RETURNING q.source_kind, q.source_pk, q.logical_id, q.shard
|
||||
`
|
||||
|
||||
func (p *PG) ClaimAndProcess(ctx context.Context, job *Job, limit int) (JobStats, error) {
|
||||
if p.Env == "" {
|
||||
p.Env = "dev"
|
||||
}
|
||||
tx, err := p.Pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return JobStats{}, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
rows, err := tx.Query(ctx, claimSQL, job.ID, limit)
|
||||
if err != nil {
|
||||
return JobStats{}, err
|
||||
}
|
||||
type claimed struct {
|
||||
kind, logical string
|
||||
pk int64
|
||||
shard int
|
||||
}
|
||||
var batch []claimed
|
||||
for rows.Next() {
|
||||
var it claimed
|
||||
if err := rows.Scan(&it.kind, &it.pk, &it.logical, &it.shard); err != nil {
|
||||
rows.Close()
|
||||
return JobStats{}, err
|
||||
}
|
||||
batch = append(batch, it)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return JobStats{}, err
|
||||
}
|
||||
|
||||
stats := JobStats{}
|
||||
for _, it := range batch {
|
||||
c, err := p.loadCandidate(ctx, tx, it.kind, it.pk)
|
||||
if err != nil {
|
||||
return stats, err
|
||||
}
|
||||
c.LogicalID = nz(c.LogicalID, it.logical)
|
||||
c.SourceSchema = job.SourceSchema
|
||||
res, err := artifactvault.ProcessCandidate(ctx, tx, p.Env, c)
|
||||
if err != nil {
|
||||
return stats, err
|
||||
}
|
||||
stats.Scanned++
|
||||
status := "done"
|
||||
succ, quar := 1, 0
|
||||
if res.Status == "quarantine" {
|
||||
status = "quarantine"
|
||||
succ, quar = 0, 1
|
||||
stats.Quarantined++
|
||||
} else {
|
||||
stats.Succeeded++
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE dataset_backfill_queue SET status=$4 WHERE job_id=$1 AND source_kind=$2 AND source_pk=$3
|
||||
`, job.ID, it.kind, it.pk, status); err != nil {
|
||||
return stats, err
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO dataset_backfill_checkpoints (job_id, shard, source_kind, last_pk, scanned, succeeded, quarantined)
|
||||
VALUES ($1,$2,$3,$4,1,$5,$6)
|
||||
ON CONFLICT (job_id, shard, source_kind) DO UPDATE SET
|
||||
last_pk=GREATEST(dataset_backfill_checkpoints.last_pk, EXCLUDED.last_pk),
|
||||
scanned=dataset_backfill_checkpoints.scanned+1,
|
||||
succeeded=dataset_backfill_checkpoints.succeeded+EXCLUDED.succeeded,
|
||||
quarantined=dataset_backfill_checkpoints.quarantined+EXCLUDED.quarantined
|
||||
`, job.ID, it.shard, it.kind, it.pk, succ, quar); err != nil {
|
||||
return stats, err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return stats, err
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func (p *PG) loadCandidate(ctx context.Context, tx pgx.Tx, kind string, pk int64) (artifactvault.Candidate, error) {
|
||||
items, err := artifactvault.Enumerate(ctx, tx, kind, pk-1, 1)
|
||||
if err != nil {
|
||||
return artifactvault.Candidate{}, err
|
||||
}
|
||||
if len(items) == 0 || items[0].SourcePK != pk {
|
||||
return artifactvault.Candidate{Kind: kind, SourcePK: pk}, nil
|
||||
}
|
||||
return items[0], nil
|
||||
}
|
||||
|
||||
func (p *PG) RecalcJob(ctx context.Context, jobID string) (JobStats, error) {
|
||||
var s JobStats
|
||||
err := p.Pool.QueryRow(ctx, `
|
||||
SELECT COALESCE(SUM(scanned),0), COALESCE(SUM(succeeded),0), COALESCE(SUM(quarantined),0)
|
||||
FROM dataset_backfill_checkpoints WHERE job_id=$1
|
||||
`, jobID).Scan(&s.Scanned, &s.Succeeded, &s.Quarantined)
|
||||
return s, err
|
||||
}
|
||||
|
||||
func (p *PG) CountQueue(ctx context.Context, jobID, status string) (int, error) {
|
||||
var n int
|
||||
err := p.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM dataset_backfill_queue WHERE job_id=$1 AND status=$2`, jobID, status).Scan(&n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (p *PG) CountArtifacts(ctx context.Context) (int, error) {
|
||||
var n int
|
||||
err := p.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts`).Scan(&n)
|
||||
return n, err
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
[
|
||||
{"dataset_id":"pinned-clean","class":"pinned","display_name":"Clean pinned demo","storage_identity":"scratch-u9-clean","source_schema":38,"source_commit":"dev-u9","status":"approved","project_count":1},
|
||||
{"dataset_id":"demo-minimal","class":"demo","display_name":"Minimal demo","storage_identity":"scratch-u9-demo","source_schema":38,"source_commit":"dev-u9","status":"approved","project_count":1},
|
||||
{"dataset_id":"regression-core","class":"regression","display_name":"Regression core","storage_identity":"scratch-u9-reg","source_schema":38,"source_commit":"dev-u9","status":"approved","project_count":1},
|
||||
{"dataset_id":"migration-oldest","class":"migration_baseline","display_name":"Oldest supported schema 28","storage_identity":"scratch-u9-oldest","source_schema":28,"source_commit":"oldest-supported","status":"approved","project_count":1},
|
||||
{"dataset_id":"migration-previous","class":"migration_baseline","display_name":"Previous release schema 37","storage_identity":"scratch-u9-previous","source_schema":37,"source_commit":"previous-release","status":"approved","project_count":1},
|
||||
{"dataset_id":"ephemeral-scratch","class":"ephemeral","display_name":"Ephemeral scratch import","storage_identity":"scratch-u9-eph","source_schema":38,"source_commit":"dev-u9","status":"approved","project_count":1},
|
||||
{"dataset_id":"quarantine-fork","class":"quarantine","display_name":"Known projection fork","storage_identity":"scratch-u9-fork","source_schema":38,"source_commit":"dev-u9","status":"quarantine","project_count":1},
|
||||
{"dataset_id":"pinned-chat-turns","class":"pinned","display_name":"Completed chat turns","storage_identity":"scratch-u9-chat","source_schema":38,"source_commit":"dev-u9","status":"approved","project_count":1}
|
||||
]
|
||||
@@ -0,0 +1,192 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
)
|
||||
|
||||
type TrackResult struct {
|
||||
Name string
|
||||
OK bool
|
||||
Detail string
|
||||
}
|
||||
|
||||
func RunEightTracks(ctx context.Context, registryPath string) []TrackResult {
|
||||
now := time.Now().UTC()
|
||||
body := []byte("# controlled corpus body for u9 tracks")
|
||||
approvedAt := now.Add(-time.Hour)
|
||||
|
||||
out := make([]TrackResult, 0, 8)
|
||||
out = append(out, track("clean", func() error {
|
||||
enum := NewMemoryEnumerator()
|
||||
job := NewJob("clean", "dev", "scratch-clean", 38)
|
||||
st, err := NewMemoryRunner(enum, nil).RunBatch(ctx, job, artifactvault.KindAgentCommit, 0, 50)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if st.Scanned != 0 || st.Succeeded != 0 {
|
||||
return fmt.Errorf("clean must be empty: %+v", st)
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("oldest-supported", func() error {
|
||||
enum := NewMemoryEnumerator()
|
||||
enum.Seed(artifactvault.KindAgentCommit, []artifactvault.Candidate{{
|
||||
Kind: artifactvault.KindAgentCommit, SourcePK: 1, LogicalID: "doc_oldest", SnapshotID: 3,
|
||||
EventID: "commit:oldest", Payload: body, SourceSchema: OldestSupportedSchema,
|
||||
}})
|
||||
job := NewJob("oldest", "oldest-supported", "scratch-oldest", OldestSupportedSchema)
|
||||
st, err := NewMemoryRunner(enum, nil).RunBatch(ctx, job, artifactvault.KindAgentCommit, 0, 50)
|
||||
if err != nil || st.Succeeded != 1 || st.Quarantined != 0 {
|
||||
return fmt.Errorf("oldest: %+v %v", st, err)
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("previous", func() error {
|
||||
enum := NewMemoryEnumerator()
|
||||
enum.Seed(artifactvault.KindAgentCommit, []artifactvault.Candidate{{
|
||||
Kind: artifactvault.KindAgentCommit, SourcePK: 2, LogicalID: "doc_prev", SnapshotID: 5,
|
||||
EventID: "commit:prev", Payload: body, SourceSchema: 37,
|
||||
}})
|
||||
job := NewJob("prev", "previous-release", "scratch-prev", 37)
|
||||
st, err := NewMemoryRunner(enum, nil).RunBatch(ctx, job, artifactvault.KindAgentCommit, 0, 50)
|
||||
if err != nil || st.Succeeded != 1 {
|
||||
return fmt.Errorf("previous: %+v %v", st, err)
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("hazard", func() error {
|
||||
enum := NewMemoryEnumerator()
|
||||
enum.Seed(artifactvault.KindAgentCommit, []artifactvault.Candidate{
|
||||
{Kind: artifactvault.KindAgentCommit, SourcePK: 3, LogicalID: "missing", SnapshotID: 0, Payload: body},
|
||||
{Kind: artifactvault.KindAgentCommit, SourcePK: artifactvault.HighIDThreshold + 2, LogicalID: "high", SnapshotID: 1, Payload: body},
|
||||
})
|
||||
enum.Seed(artifactvault.KindApproved, []artifactvault.Candidate{{
|
||||
Kind: artifactvault.KindApproved, SourcePK: 4, LogicalID: "fork", SnapshotID: 8, HeadSnapshotID: 9,
|
||||
LedgerSHA: "aa", ProjectionSHA: "bb", Payload: body, ApprovedAt: &approvedAt,
|
||||
}})
|
||||
enum.Seed(artifactvault.KindSnapshot, []artifactvault.Candidate{{
|
||||
Kind: artifactvault.KindSnapshot, SourcePK: 5, LogicalID: "legacy", SchemaVer: 0, PayloadSHA: "abc", Payload: body,
|
||||
}})
|
||||
reg := artifactvault.NewMemoryRegistry()
|
||||
run := NewMemoryRunner(enum, reg)
|
||||
job := NewJob("hazard", "dev", "scratch-hazard", 38)
|
||||
var quar int
|
||||
for _, kind := range []string{artifactvault.KindAgentCommit, artifactvault.KindApproved, artifactvault.KindSnapshot} {
|
||||
st, err := run.RunBatch(ctx, job, kind, 0, 50)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
quar = st.Quarantined
|
||||
}
|
||||
if quar < 4 || reg.QuarantineCount(ctx) < 4 {
|
||||
return fmt.Errorf("hazard quarantine=%d registry=%d", quar, reg.QuarantineCount(ctx))
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("import", func() error {
|
||||
reg, err := LoadRegistryFile(registryPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
approved := 0
|
||||
for _, d := range reg.ListApproved() {
|
||||
if d.Status == "approved" || d.Status == "imported" {
|
||||
approved++
|
||||
}
|
||||
}
|
||||
if approved < 6 || approved > 10 {
|
||||
return fmt.Errorf("approved dataset count %d not in 6-10", approved)
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("reload", func() error {
|
||||
enum := NewMemoryEnumerator()
|
||||
enum.Seed(artifactvault.KindAgentCommit, []artifactvault.Candidate{{
|
||||
Kind: artifactvault.KindAgentCommit, SourcePK: 11, LogicalID: "doc_reload", SnapshotID: 6,
|
||||
EventID: "commit:reload", Payload: body,
|
||||
}})
|
||||
mem := artifactvault.NewMemoryRegistry()
|
||||
run := NewMemoryRunner(enum, mem)
|
||||
job := NewJob("reload", "dev", "scratch-reload", 38)
|
||||
if _, err := run.RunBatch(ctx, job, artifactvault.KindAgentCommit, 0, 50); err != nil {
|
||||
return err
|
||||
}
|
||||
job.Status = JobRunning
|
||||
job.Checkpoints = map[string]Checkpoint{}
|
||||
if _, err := run.RunBatch(ctx, job, artifactvault.KindAgentCommit, 0, 50); err != nil {
|
||||
return err
|
||||
}
|
||||
if mem.Coverage(ctx).Required != 1 {
|
||||
return fmt.Errorf("reload duplicated artifacts required=%d", mem.Coverage(ctx).Required)
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("rollback", func() error {
|
||||
req := DestroyRequest{
|
||||
TargetDSN: "postgres://app@127.0.0.1:5433/agent_system_dev",
|
||||
TargetIdentity: "dev", Manifest: DeleteManifest{DatasetIDs: []string{"pinned-clean"}, StorageIdentity: "dev"},
|
||||
RecoverableLocation: "/tmp/scratch-restore", CreatedAt: now.Add(-8 * 24 * time.Hour),
|
||||
QuarantineUntil: now.Add(-time.Hour), Approval1: "alice", Approval2: "bob",
|
||||
Coverage: artifactvault.Coverage{Required: 1, Verified: 1}, BackupOK: true, CorpusOK: true, VaultReady: true,
|
||||
Now: now,
|
||||
}
|
||||
if err := EvaluateDestroy(req); err == nil {
|
||||
return fmt.Errorf("destroy into agent_system_dev must fail")
|
||||
}
|
||||
req.TargetDSN = "postgres://app@127.0.0.1:5544/agent_rel_scratch"
|
||||
req.TargetIdentity = "scratch-u9"
|
||||
req.Manifest.StorageIdentity = "scratch-u9"
|
||||
if err := EvaluateDestroy(req); err != nil {
|
||||
return fmt.Errorf("scratch destroy with gates: %w", err)
|
||||
}
|
||||
req.BackupOK = false
|
||||
if err := EvaluateDestroy(req); err == nil {
|
||||
return fmt.Errorf("missing backup must block destroy")
|
||||
}
|
||||
if err := RequireScratchRestore("postgres://x/agent_system_main", "main", "aaa", "aaa", true); err == nil {
|
||||
return fmt.Errorf("restore into main must fail")
|
||||
}
|
||||
if err := RequireScratchRestore("postgres://x/fav_scratch", "scratch-1", "abc", "abc", true); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}))
|
||||
out = append(out, track("artifact-recovery", func() error {
|
||||
dest := filepath.Join(os.TempDir(), fmt.Sprintf("scratch-u9-%d", now.UnixNano()))
|
||||
defer func() { _ = os.RemoveAll(dest) }()
|
||||
plan := artifactvault.RecoverPlan{
|
||||
Mode: artifactvault.RecoverAsCollection, TargetIdentity: "scratch-db", SourceIdentity: "vault",
|
||||
PlanHash: "u9-recover", HasAdmin: true, ScratchOnly: true,
|
||||
}
|
||||
okBody := []byte("# recovered corpus\n")
|
||||
arts := []artifactvault.Artifact{{UUID: "art-ok", LogicalResourceID: "doc_ok", PayloadSHA256: artifactvault.SHA256Hex(okBody)}}
|
||||
_, err := artifactvault.ApplyRecoverAsCollection(plan, dest, arts, map[string][]byte{"art-ok": okBody}, nil)
|
||||
return err
|
||||
}))
|
||||
return out
|
||||
}
|
||||
|
||||
func track(name string, fn func() error) TrackResult {
|
||||
if err := fn(); err != nil {
|
||||
return TrackResult{Name: name, OK: false, Detail: err.Error()}
|
||||
}
|
||||
return TrackResult{Name: name, OK: true, Detail: "PASS"}
|
||||
}
|
||||
|
||||
func AllTracksPass(rs []TrackResult) error {
|
||||
for _, r := range rs {
|
||||
if !r.OK {
|
||||
return fmt.Errorf("track %s: %s", r.Name, r.Detail)
|
||||
}
|
||||
}
|
||||
if len(rs) != 8 {
|
||||
return fmt.Errorf("expected 8 tracks, got %d", len(rs))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEightTracksPass(t *testing.T) {
|
||||
path := "testdata/registry.json"
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
path = filepath.Join("internal", "dataset", "testdata", "registry.json")
|
||||
}
|
||||
rs := RunEightTracks(context.Background(), path)
|
||||
if err := AllTracksPass(rs); err != nil {
|
||||
for _, r := range rs {
|
||||
t.Logf("%s ok=%v %s", r.Name, r.OK, r.Detail)
|
||||
}
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
)
|
||||
|
||||
const (
|
||||
ClassPinned = "pinned"
|
||||
ClassDemo = "demo"
|
||||
ClassRegression = "regression"
|
||||
ClassMigrationBaseline = "migration_baseline"
|
||||
ClassEphemeral = "ephemeral"
|
||||
ClassQuarantine = "quarantine"
|
||||
|
||||
JobRunning = "running"
|
||||
JobPaused = "paused"
|
||||
JobCompleted = "completed"
|
||||
JobFailed = "failed"
|
||||
|
||||
OldestSupportedSchema = artifactvault.OldestSupportedSchema
|
||||
MinQuarantineWindow = 7 * 24 * time.Hour
|
||||
)
|
||||
|
||||
type Dataset struct {
|
||||
ID string `json:"dataset_id"`
|
||||
Class string `json:"class"`
|
||||
DisplayName string `json:"display_name"`
|
||||
StorageIdentity string `json:"storage_identity"`
|
||||
SourceSchema int `json:"source_schema"`
|
||||
SourceCommit string `json:"source_commit"`
|
||||
Status string `json:"status"`
|
||||
ProjectCount int `json:"project_count"`
|
||||
}
|
||||
|
||||
func ValidateClass(class string) error {
|
||||
switch class {
|
||||
case ClassPinned, ClassDemo, ClassRegression, ClassMigrationBaseline, ClassEphemeral, ClassQuarantine:
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("dataset: invalid class %q", class)
|
||||
}
|
||||
}
|
||||
|
||||
type Checkpoint struct {
|
||||
JobID string
|
||||
Shard int
|
||||
SourceKind string
|
||||
LastPK int64
|
||||
Scanned int
|
||||
Succeeded int
|
||||
Quarantined int
|
||||
RetryWatermark int
|
||||
}
|
||||
|
||||
type JobStats struct {
|
||||
Scanned int
|
||||
Succeeded int
|
||||
Quarantined int
|
||||
}
|
||||
|
||||
func Recalc(cps []Checkpoint) JobStats {
|
||||
var s JobStats
|
||||
for _, c := range cps {
|
||||
s.Scanned += c.Scanned
|
||||
s.Succeeded += c.Succeeded
|
||||
s.Quarantined += c.Quarantined
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
type DeleteManifest struct {
|
||||
DatasetIDs []string `json:"dataset_ids"`
|
||||
StorageIdentity string `json:"storage_identity"`
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
|
||||
func CanonicalDestroyBytes(m DeleteManifest) []byte {
|
||||
ids := append([]string(nil), m.DatasetIDs...)
|
||||
sort.Strings(ids)
|
||||
raw, _ := json.Marshal(struct {
|
||||
DatasetIDs []string `json:"dataset_ids"`
|
||||
StorageIdentity string `json:"storage_identity"`
|
||||
Reason string `json:"reason"`
|
||||
}{ids, m.StorageIdentity, m.Reason})
|
||||
return raw
|
||||
}
|
||||
|
||||
func DestroyManifestSHA(m DeleteManifest) string {
|
||||
sum := sha256.Sum256(CanonicalDestroyBytes(m))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
type DestroyRequest struct {
|
||||
TargetIdentity string
|
||||
TargetDSN string
|
||||
Manifest DeleteManifest
|
||||
ExpectedManifestSHA string
|
||||
RecoverableLocation string
|
||||
CreatedAt time.Time
|
||||
QuarantineUntil time.Time
|
||||
Approval1 string
|
||||
Approval2 string
|
||||
Coverage artifactvault.Coverage
|
||||
BackupOK bool
|
||||
CorpusOK bool
|
||||
VaultReady bool
|
||||
Now time.Time
|
||||
}
|
||||
|
||||
func RefuseOperationalDSN(dsn, identity string) error {
|
||||
low := strings.ToLower(dsn + " " + identity)
|
||||
for _, ban := range []string{"agent_system_dev", "agent_system_test", "agent_system_main"} {
|
||||
if strings.Contains(low, ban) && !strings.Contains(low, "scratch") {
|
||||
return fmt.Errorf("dataset: refuse operational database %s", ban)
|
||||
}
|
||||
}
|
||||
if strings.Contains(low, ":5433/") || strings.Contains(low, ":5435/") {
|
||||
if !strings.Contains(low, "scratch") {
|
||||
return fmt.Errorf("dataset: refuse shared env port")
|
||||
}
|
||||
}
|
||||
if strings.Contains(low, ":5434/") && strings.Contains(low, "agent_system_test") {
|
||||
return fmt.Errorf("dataset: refuse shared test database")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func EvaluateDestroy(req DestroyRequest) error {
|
||||
if err := RefuseOperationalDSN(req.TargetDSN, req.TargetIdentity); err != nil {
|
||||
return err
|
||||
}
|
||||
if !strings.Contains(strings.ToLower(req.RecoverableLocation), "scratch") {
|
||||
return fmt.Errorf("dataset: recoverable location must be scratch")
|
||||
}
|
||||
if req.QuarantineUntil.Sub(req.CreatedAt) < MinQuarantineWindow {
|
||||
return fmt.Errorf("dataset: quarantine observation window must be at least 7 days")
|
||||
}
|
||||
now := req.Now
|
||||
if now.IsZero() {
|
||||
now = time.Now().UTC()
|
||||
}
|
||||
if now.Before(req.QuarantineUntil) {
|
||||
return fmt.Errorf("dataset: quarantine window still open")
|
||||
}
|
||||
if strings.TrimSpace(req.Approval1) == "" || strings.TrimSpace(req.Approval2) == "" || req.Approval1 == req.Approval2 {
|
||||
return fmt.Errorf("dataset: two distinct user approvals required")
|
||||
}
|
||||
got := DestroyManifestSHA(req.Manifest)
|
||||
if req.ExpectedManifestSHA != "" && req.ExpectedManifestSHA != got {
|
||||
return fmt.Errorf("dataset: destroy manifest sha mismatch")
|
||||
}
|
||||
if len(req.Manifest.DatasetIDs) == 0 || strings.TrimSpace(req.Manifest.StorageIdentity) == "" {
|
||||
return fmt.Errorf("dataset: destroy manifest incomplete")
|
||||
}
|
||||
return artifactvault.PreflightDestroy(req.Coverage, req.BackupOK, req.CorpusOK, req.VaultReady)
|
||||
}
|
||||
|
||||
func RequireScratchRestore(dsn, storageIdentity, manifestSHA, expectedSHA string, hasAdmin bool) error {
|
||||
if err := RefuseOperationalDSN(dsn, storageIdentity); err != nil {
|
||||
return err
|
||||
}
|
||||
if !strings.Contains(strings.ToLower(dsn+" "+storageIdentity), "scratch") {
|
||||
return fmt.Errorf("dataset: restore target must be scratch")
|
||||
}
|
||||
if !hasAdmin {
|
||||
return fmt.Errorf("dataset: restore requires role-admin")
|
||||
}
|
||||
if manifestSHA == "" || expectedSHA == "" || manifestSHA != expectedSHA {
|
||||
return fmt.Errorf("dataset: restore manifest mismatch or missing")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package dataset
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/artifactvault"
|
||||
)
|
||||
|
||||
func TestValidateClassAndDestroyGates(t *testing.T) {
|
||||
if err := ValidateClass("nope"); err == nil {
|
||||
t.Fatal("class")
|
||||
}
|
||||
if err := RefuseOperationalDSN("postgres://x/agent_system_dev", "dev"); err == nil {
|
||||
t.Fatal("dev dsn")
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
req := DestroyRequest{
|
||||
TargetDSN: "postgres://x/fav_scratch", TargetIdentity: "scratch-a",
|
||||
Manifest: DeleteManifest{DatasetIDs: []string{"pinned-clean"}, StorageIdentity: "scratch-a", Reason: "u9"},
|
||||
RecoverableLocation: "/tmp/scratch-restore", CreatedAt: now.Add(-3 * 24 * time.Hour),
|
||||
QuarantineUntil: now.Add(4 * 24 * time.Hour), Approval1: "a", Approval2: "a",
|
||||
Coverage: artifactvault.Coverage{Required: 1, Verified: 1}, BackupOK: true, CorpusOK: true, VaultReady: true,
|
||||
Now: now,
|
||||
}
|
||||
if err := EvaluateDestroy(req); err == nil {
|
||||
t.Fatal("short window + same approver must fail")
|
||||
}
|
||||
req.CreatedAt = now.Add(-10 * 24 * time.Hour)
|
||||
req.QuarantineUntil = now.Add(-time.Hour)
|
||||
req.Approval2 = "b"
|
||||
if err := EvaluateDestroy(req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.ExpectedManifestSHA = "deadbeef"
|
||||
if err := EvaluateDestroy(req); err == nil {
|
||||
t.Fatal("tampered manifest")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecalcDenominator(t *testing.T) {
|
||||
got := Recalc([]Checkpoint{{Scanned: 10, Succeeded: 7, Quarantined: 3}, {Scanned: 5, Succeeded: 5}})
|
||||
if got.Scanned != 15 || got.Succeeded != 12 || got.Quarantined != 3 {
|
||||
t.Fatalf("%+v", got)
|
||||
}
|
||||
}
|
||||
@@ -81,6 +81,36 @@ func TestMigration032_AllowsMigrationRepairKinds(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestGooseU9_00038_HasUpAndDown(t *testing.T) {
|
||||
files, err := ListMigrationFiles(ResolveMigrationsDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var found string
|
||||
for _, name := range files {
|
||||
if strings.HasPrefix(name, "00038_") && strings.HasSuffix(name, ".sql") {
|
||||
found = name
|
||||
break
|
||||
}
|
||||
}
|
||||
if found == "" {
|
||||
t.Fatal("expected goose 00038 for dataset lifecycle")
|
||||
}
|
||||
raw, err := os.ReadFile(filepath.Join(ResolveMigrationsDir(), found))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
text := string(raw)
|
||||
if !strings.Contains(text, "-- +goose Up") || !strings.Contains(text, "-- +goose Down") {
|
||||
t.Fatal("00038 must have Up and Down")
|
||||
}
|
||||
for _, table := range []string{"dataset_registry", "dataset_backfill_jobs", "dataset_backfill_checkpoints", "dataset_backfill_queue", "dataset_destroy_requests"} {
|
||||
if !strings.Contains(text, table) {
|
||||
t.Fatalf("00038 must mention %s", table)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("goose down is not part of -short")
|
||||
@@ -101,6 +131,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
|
||||
"00035_2026-09-14-durable-event-delivery.sql",
|
||||
"00036_2026-09-14-final-artifact-vault.sql",
|
||||
"00037_2026-09-15-task-run-manifests.sql",
|
||||
"00038_2026-09-15-dataset-lifecycle.sql",
|
||||
}
|
||||
joined := strings.Join(files, "\n")
|
||||
for _, name := range want {
|
||||
@@ -136,7 +167,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
|
||||
t.Fatalf("up: %v", err)
|
||||
}
|
||||
ver, err := CurrentVersion(ctx, pool)
|
||||
if err != nil || ver < 37 {
|
||||
if err != nil || ver < 38 {
|
||||
t.Fatalf("version after up: %d %v", ver, err)
|
||||
}
|
||||
|
||||
@@ -149,7 +180,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
|
||||
if err := goose.DownToContext(ctx, sqlDB, migDir, 33); err != nil {
|
||||
t.Fatalf("down to 33: %v", err)
|
||||
}
|
||||
for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests"} {
|
||||
for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests", "dataset_registry"} {
|
||||
var n int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM information_schema.tables WHERE table_name=$1`, table).Scan(&n); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -89,7 +89,16 @@ else
|
||||
fi
|
||||
|
||||
if [[ -z "${DSN}" ]]; then
|
||||
DSN="${TEST_DATABASE_URL:-postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable}"
|
||||
DSN="${TEST_DATABASE_URL:-}"
|
||||
fi
|
||||
if [[ -z "${DSN}" ]]; then
|
||||
echo "[错误] 必须提供 --dsn 或 TEST_DATABASE_URL；禁止默认连 agent_system_dev/test/main"
|
||||
exit 1
|
||||
fi
|
||||
low_dsn=$(echo "${DSN}" | tr '[:upper:]' '[:lower:]')
|
||||
if echo "${low_dsn}" | grep -Eq 'agent_system_(dev|test|main)' && ! echo "${low_dsn}" | grep -q 'scratch'; then
|
||||
echo "[错误] 拒绝把恢复演练指向共享环境库 ${DSN%%\?*}"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "[迁移] goose 幂等 Up（DSN 脱敏显示）..."
|
||||
|
||||
Reference in New Issue
Block a user