Merge pull request 'Make U7 two-instance acceptance failures point at the real cause' (#678) from fix/multi-instance-test-failure-diagnostics into main
This commit was merged in pull request #678.
This commit is contained in:
@@ -0,0 +1,95 @@
|
||||
# fix/multi-instance-test-failure-diagnostics 任务卡
|
||||
|
||||
任务主记录:PR #678 https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Dev/pulls/678;状态:已完成(合入前闭合);owner:weichen
|
||||
|
||||
- 权威仓:ssh://git@192.168.3.2:2222/weichen/Agent-System-Dev.git;目标分支:main;R0(initial_base_sha)= 5360884e9283d962caf945d944df05b5e5059177
|
||||
- 工作区:/home/weichen/projects/Agent-System-Dev-wt-fix-multi-instance-test-failure-diagnostics(worktree);分支:fix/multi-instance-test-failure-diagnostics
|
||||
- 分支类型:standalone;root_target:main;immediate_target:main;accepting_requirements:false
|
||||
- 需求分流:verdict=NEW;selected_initiative=无;requirement_sha256=e2453ebdabc4cc9c61bb25a3535fbe7cf0d0fe687f2ea7bc565b799dcf0fcb6d;triage_receipt=/tmp/triage-mi-test-hygiene-20261021.json;receipt_sha256=4f0a610213a5f6363e303e7129cd09a630ae1a04acbaa5116635c804b2f7f56c
|
||||
|
||||
- 授权与交付终点:修复 `tests/integration/multi-instance` 两处测试卫生缺陷(轮询静默超时、断言早于恢复视图),让失败直指真正原因。不得改 Backend 或任何生产代码(其根因已由 Backend PR #237 修复)、不得弱化断言、不得放大超时预算。
|
||||
- 交付模式:goal-first(目标、范围和交付终点一次确认;范围内连续推进)
|
||||
- 实施执行:agent-per-unit(配置:AGENT-EXECUTION.md)
|
||||
- 合入执行:agent(已授权交付范围内,Review 通过、冲突解决且当前 HEAD 的 CI 通过后合入)
|
||||
- 流程依据:仓库根目录 policy/delivery.md(归档后仍按仓库根目录定位)
|
||||
|
||||
## 目标与非目标
|
||||
|
||||
<预期结果、受影响入口/模块/契约;明确不做什么。>
|
||||
|
||||
## Agent 实施收据
|
||||
|
||||
路由与 fail-stop 规则见根目录 AGENT-EXECUTION.md 与 policy/delivery.md;每次尝试一行,原始日志留在源码外收据路径。
|
||||
|
||||
| 单元 ID | 预期范围 | 配置摘要 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 |
|
||||
|---|---|---|---|---|---|---|---|---|---|---|
|
||||
| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — |
|
||||
|
||||
## 同步记录
|
||||
|
||||
| 节点 | sync_base_sha | tested_head_sha | checked_at |
|
||||
|---|---|---|---|
|
||||
| 启动 | 5360884e9283d962caf945d944df05b5e5059177 | 5360884e9283d962caf945d944df05b5e5059177 | 2026-10-21T12:07:32Z |
|
||||
|
||||
## 证据(Evidence)
|
||||
|
||||
<!-- CMD="<命令>" make evidence 写入;手抄/转写输出视为无效证据(AGENTS.md 证据与验证门禁)。 -->
|
||||
| 时间(UTC) | 命令 | 被测HEAD | exit | 来源 | sha256(16) |
|
||||
|---|---|---|---|---|---|
|
||||
|
||||
## 结果与验证
|
||||
|
||||
- 结果:owner_crash_takeover 的 interrupted 轮询改为有界等待 + 显式失败(点名未观察到的 run 状态序列、轮询实例、实际耗时与预算、实例日志),未收口时不再落到后续 POST 的误导性 409;restart_invariants 在断言消息数之前新增「恢复视图」有界等待(存在 interrupted run 且无 pending/running/waiting_hitl),未恢复时给出点名判据的失败信息;其余 `t.Fatal`→`t.Fatalf` 仅补失败信息,断言条件未变。新增 `runStatuses` 与 `e.instanceOf` 两个小helper 供等待条件与失败信息复用。
|
||||
- 验证:PASS——真实 PostgreSQL(127.0.0.1:54331,fsync=on)下空闲机器 15.1s 通过、96 忙循环压满 32 核 17.4s 通过(新增失败路径未误触发);`gofmt`/`go vet` 干净;独立 Review(pi-glm,PASS 无阻塞项)报告 `.task/reviews/2026-10-21-mi-diagnostics-review.md`,6 项非阻塞发现中第 1 项如实指出恢复视图判据的 interrupted 半边由历史 run 平凡满足(不检出 B 重启对账回归,但改动前同样不检出,覆盖无回退)。
|
||||
- 影响与剩余项:不改生产代码、不弱化任何断言、未放大任何超时预算、包级 `go test -timeout` 未动。Review 其余非阻塞项(busy 白名单对新增非终态 status 不设防、轮询只附 logA、`instanceOf` 守卫不可达、顺序耦合、瞬态 5xx 即失败)均为既有共享弱点或死代码,不阻塞。
|
||||
|
||||
## 合入与清理
|
||||
|
||||
- Review 记录:PASS(独立会话 pi-glm / zai-coding-cn/glm-5.3-flash;base 5360884e9283d962caf945d944df05b5e5059177,head 232477b22f997bc4ab87b5983ae0b350f6ae4c43;报告 `.task/reviews/2026-10-21-mi-diagnostics-review.md`;无阻塞项,6 项非阻塞发现;重放依据:host 用真实 PG 在空闲与 96 负载两种条件下各跑一次多实例验收,均通过)
|
||||
- 冲突记录:NOT_RUN(一次交付前同步,0 提交 behind)
|
||||
- CI 记录:NOT_RUN(当前 HEAD 完整 SHA、全部必需检查、运行链接、执行身份及 GATE 结果)
|
||||
- 合入记录:NOT_RUN(交付授权范围、PR、被核验 HEAD、合入 SHA、目标分支及内容核验)
|
||||
- 清理记录:NOT_RUN(授权依据、写者退出确认、worktree/分支/任务卡处置)
|
||||
|
||||
<!-- Local receipt: SELF. Digests detect corruption; they do not authenticate a reviewer or CI. -->
|
||||
```json
|
||||
{
|
||||
"summary_schema": 1,
|
||||
"source": "SELF",
|
||||
"timestamp": "2026-10-21T12:06:44.036919+00:00",
|
||||
"command": "make precheck",
|
||||
"exit_code": 0,
|
||||
"cwd": "/home/weichen/projects/Agent-System-Dev",
|
||||
"before_head": "5360884e9283d962caf945d944df05b5e5059177",
|
||||
"after_head": "5360884e9283d962caf945d944df05b5e5059177",
|
||||
"branch": "main",
|
||||
"target_branch": "main",
|
||||
"target_head": "5360884e9283d962caf945d944df05b5e5059177",
|
||||
"root_target": "main",
|
||||
"immediate_target": "main",
|
||||
"before_tracked_diff_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
|
||||
"after_tracked_diff_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
|
||||
"output_tail": [
|
||||
"[repo] branch=main head=5360884e92",
|
||||
"[remote] origin/main=5360884e92 ahead=0 behind=0",
|
||||
"[tree] clean submodules_drift=0 uninit=0",
|
||||
"[worktrees] linked=3",
|
||||
"[initiatives] local cards (not authoritative): none",
|
||||
"[hooks] core.hooksPath=scripts/hooks(pre-commit/post-merge 实际接线)",
|
||||
"[facts] policy/facts.md 6 条事实均在有效期内",
|
||||
"[required-reading] core/baseline.md, workflows/delivery.md §1, policy/delivery.md",
|
||||
"[R0] 5360884e92(完整:5360884e9283d962caf945d944df05b5e5059177)",
|
||||
"[worktree] before-sync: recursive component checks passed",
|
||||
"[verdict] READY — 保留本次 precheck receipt 后建任务工作区:PRECHECK_RECEIPT=/tmp/task-precheck.json make task-start NAME=feature/your-branch"
|
||||
],
|
||||
"output_sha256": "6a290a280e9e8ec5b78b7bf430d194dbe3a55fbbd9507b05b6c12899b01ceebb",
|
||||
"receipt_sha256": "577c22f0e49c012c394aeb6dbcc188fbf681ab8315c5864da6da3d0ca62b90f8",
|
||||
"receipt_path": "/home/weichen/.cache/agent-system-dev/evidence/577c22f0e49c012c394aeb6dbcc188fbf681ab8315c5864da6da3d0ca62b90f8.json"
|
||||
}
|
||||
```
|
||||
|
||||
## 闭合记录(合入前折叠)
|
||||
|
||||
- 本卡在 PR #678(https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Dev/pulls/678)合入前闭合并归档,随该 PR 一同合入。
|
||||
- 该 PR 的合并提交即本任务的落地记录。
|
||||
- 合入后无需回填,也无需单独的闭合 PR。
|
||||
@@ -0,0 +1,94 @@
|
||||
# 独立代码审查:fix/multi-instance-test-failure-diagnostics(多实例双进程验收的测试卫生修复)
|
||||
|
||||
- 审查对象:`5360884e9283d962caf945d944df05b5e5059177..2eaa7510848ba1fc13013f5b0a9ff1fd548b2071`(R0..HEAD,分支 fix/multi-instance-test-failure-diagnostics;测试侧变更落在 transport 提交 2eaa751)
|
||||
- 涉及文件(审查范围内):仅 `tests/integration/multi-instance/multi_instance_test.go`(+95/−17);范围内的另一文件 `.task/fix-multi-instance-test-failure-diagnostics.md` 属任务卡豁免范围,不在本审查内
|
||||
- 审查者身份:全新会话独立审查,未参与实现;本会话除本报告外不写入任何文件,全部只读
|
||||
- 背景核对:范围内 `git diff 5360884e92..HEAD -- Makefile .gitea/` 为空(0 行)——包级 `go test -count=1 -timeout 15m .`(Makefile:190)未被改动
|
||||
|
||||
## 结论:PASS(无阻塞项;6 项非阻塞发现)
|
||||
|
||||
两处修复意图均达成:owner_crash_takeover 的轮询从「到点静默退出 → POST 落到误导性 409」改为显式 `reconciled` 硬要求 + 定点失败;restart_invariants 在断言前先等恢复视图。逐条核对未发现断言被削弱,未发现超时预算放大,未发现实际的 e.a/e.b 解引用风险。
|
||||
|
||||
---
|
||||
|
||||
## 一、断言是否被削弱(重点 1)
|
||||
|
||||
逐 hunk 核对新旧文件(旧:5360884e92 版,新:HEAD 版):
|
||||
|
||||
1. **cross_event_sse**:仅 `t.Fatal(...)` → `t.Fatalf(..., lastSeq, instanceOf)`,select 的 8s 预算与事件分支条件(`!ok`、`ev.seq <= lastSeq`)逐字未动。**等价,仅增强诊断。**
|
||||
2. **owner_crash_takeover 轮询**:旧代码 `break` 后落到 POST,超时静默继续;新代码引入 `reconciled` 标志,未观察到即 `t.Fatalf`。判定语义与旧版一致(最后一个 run 的 status == "interrupted",经 `runStatuses` 取末位;类型断言失败时旧为 nil map 比较、新为空串,均判 false,等价)。**这是收紧而非削弱**:旧版存在一个能「未观察 interrupted 仍通过」的窄缝——最后一次 300ms 轮询之后、POST 之前对账恰好收口,POST 得 200 而通过,测试从未显式观察到 interrupted;新版关闭该缝,通过必须以观察到 interrupted 为前提。轮询粒度(300ms)与预算(`httpWait`=15s)不变。
|
||||
3. **restart_invariants**:新增前置等待,其后的既有断言(`msgs >= 5`、`lastSeq != 0`、SSE 全量重放 seq 严格连续、重启后 postMessage)逐字未动。**无削弱。**
|
||||
4. **waitConversationIdle / waitActiveHITL / waitRunStatus / waitHealthy**:成功条件逐字等价(idle 的 busy 白名单 pending/running/waiting_hitl 不变;HITL 的 `ok && len>0 && id!="" && runID!=""` 不变;waitRunStatus 的 `s == status` 不变;waitHealthy 的 200/401 不变);改动仅是新增 statuses/cardCount/observed/last 探针跟踪与更详细消息。`runStatuses` 把 `runs, ok := …` 的 ok 守卫换为 `runs, _ := …`,缺失/类型不符时返回空切片,与旧「ok=false 则不进循环」语义一致。
|
||||
5. **新增等待是否掩盖真实缺陷**:核对了「run 卡在 running」这一类真 bug 的走向——新等待会超时并在 Fatalf 中给出完整 statuses 列表(含卡住的 running),缺陷以测试失败形式暴露且可定位;旧代码同场景会落到后面 postMessage 的 `waitConversationIdle(20s)` 以笼统的 "still busy" 失败。**不存在把失败变成通过或无限等待的路径**(所有等待都有界且超时即 Fatalf)。唯一「弱化形状」见第二节发现 1:恢复视图判据的 interrupted 半边被历史 run 平凡满足,对「B 重启对账」回归不敏感——但该回归在改动前同样不被任何断言覆盖,覆盖无回退。
|
||||
|
||||
## 二、新增失败路径的信息是否足以定位(重点 2)
|
||||
|
||||
逐处核对新增 Fatalf 的字段:
|
||||
|
||||
- owner_crash_takeover:点名被轮询实例(`instanceOf(uc.api)` → "miA (http://addr)")、实际耗时 vs 预算(15s)、完整 statuses 序列、重启后 A 的日志。**足以定位。**
|
||||
- restart_invariants:点名轮询实例(miB)、耗时/预算、statuses、logA+logB 双日志。**足以定位。**
|
||||
- waitConversationIdle / waitActiveHITL / waitRunStatus / waitHealthy / 两处 SSE 超时:均点名实例与地址、给出耗时 vs 预算与最后观测值(statuses / cardCount / observed status / last probe / 期望 seq)。**充分。**
|
||||
|
||||
非阻塞发现:
|
||||
|
||||
- **发现 1(判据语义弱于注释声明)**:restart_invariants 等待注释称「对账收口的 interrupted run 可见」,但此刻会话里的 interrupted run 来自 owner_crash_takeover 中 **A** 重启时对 A 遗留 run 的收口(该子测试已硬性要求其存在并保持终态),而 B 被杀时并未持有任何 in-flight run(自 owner_crash_takeover 起全部活动在 A 上),**B 的重启对账在本场景无事可做**。因此该判据实际语义是「会话无 pending/running/waiting_hitl(且历史上存在 interrupted run)」,不能检出「B 启动对账回归」类缺陷。无害但注释高估了判据的覆盖面;且与 owner_crash_takeover 存在顺序耦合——未来若删改该子测试,本等待将在正常路径必然空转 15s 后失败(见发现 5 的脆弱性)。
|
||||
- **发现 2(非阻塞)**:waitActiveHITL / waitRunStatus 轮询 uc.apiB 时消息正确点名 B,但日志只附 logA;被轮询实例 B 的日志往往更相关。与改动前「只附 logA」的旧风格一致,非回归,建议后续补 logB。
|
||||
- **发现 3(非阻塞)**:restart_invariants 失败路径的 logA 是 owner_crash_takeover 重启后的 A 的日志;原 A(持有崩溃 run 时期)的日志在重启时已丢弃。此处崩溃 run 早已收口,影响有限,仅作知悉。
|
||||
|
||||
## 三、恢复视图判据本身是否正确(重点 3)
|
||||
|
||||
判据:∃ status == "interrupted" 且 ∀ status ∉ {pending, running, waiting_hitl}。
|
||||
|
||||
- **正常路径误判为未恢复的风险**:在本测试流内不存在——interrupted run 由 owner_crash_takeover 硬性保证存在且终态不可逆;唯一可能 busy 的 run 是「A resumes after restart」(mock runner,秒级收口),15s 预算充裕。当前流内**无误判风险**;跨子测试顺序耦合的脆弱性见发现 1/5。
|
||||
- **真实卡死误判为已恢复的风险**:卡死的 run 必处于 pending/running/waiting_hitl 之一 → busy=true → 不会误判。**但该 busy 集是白名单**:若 backend 未来引入新的非终态 status 字符串(如 finalizing/cancelling 之类),白名单外的非终态会被当作空闲而误判已恢复。这是与 waitConversationIdle 完全相同的既有语义(非本次引入),两处共享该潜在弱点。status 词汇表(pending/running/waiting_hitl/completed/failed/interrupted/cancelled)取自测试自身与改动前代码,未对照 backend 源码核验,**[未经证实]**。
|
||||
- 轮询经 `e.do`:非 200(含启动后瞬态 5xx)直接 Fatalf 终止轮询——与改动前该子测试的单次 GET 行为一致,失败是响亮且有信息的,可接受。
|
||||
|
||||
## 四、e.a/e.b 解引用风险(重点 4)
|
||||
|
||||
- `instanceOf`:先 `e.baseFor(c)` 再比较。`baseFor` 在 e.uc 非空且 c 为 apiB/streamB 时即解引用 `e.b.base`——因此 `instanceOf` 里的 `e.b != nil` 守卫实际不可达(e.b 为 nil 时 baseFor 先行 panic)。**实践无风险**:e.b 由 startPair 在任何 client 使用前赋值,两处重启均为非空重赋值;守卫属无害死代码(发现 4,非阻塞)。
|
||||
- base 串等值判定:`freePort` 保证 A/B 地址不同,且重启复用原 addr 字符串(`strings.TrimPrefix(e.b.base, "http://")`),重启后 `instanceOf` 对 apiB/streamB 仍正确映射到**新的** B。无误判面。
|
||||
- restart_invariants 中 `killNine(e.b)` → `startInstance` 之后 **e.a 仍有效**:e.a 未被触碰,指向 owner_crash_takeover 重启后的存活 A 进程;`uc.api` 经 baseFor 取 `e.a.base` 正确;`e.a.log` 是当前 A 实例的 lockedBuffer,非悬垂。无解引用风险。
|
||||
- `runStatuses` 对 nil map / 缺键 / 元素非 map 全部安全(带 ok 断言),失败路径打印不会引入二次 panic。
|
||||
|
||||
## 五、超时预算与包级 timeout(重点 5)
|
||||
|
||||
- `httpWait`(15s)、waitActiveHITL(15s)、waitHealthy(60s)、两处 SSE select(8s/15s)、waitConversationIdle 调用方(20s)全部未动。新增的 restart_invariants 等待复用 `httpWait`(≤15s,正常路径一次性成本),owner_crash_takeover 失败路径与旧版同界(旧:15s 空转 + POST 409;新:≤15s 定点失败)。**无预算悄悄放大。**
|
||||
- 包级 `go test -count=1 -timeout 15m .`(Makefile:190,由 CI test-pg job 调用)不在 diff 内:`git diff 5360884e92..HEAD -- Makefile .gitea/` 为 0 行。**未被改动。**
|
||||
|
||||
## 六、检测器可失败性验证(packet 约束)
|
||||
|
||||
按约束在 /tmp 构造 3 个测试语料未覆盖的早退读取器形态,直接调用 `tests.test_shell_pipe_hazard.hazards()` 并以真实 bash(`set -euo pipefail`)验证危害性:
|
||||
|
||||
| 形态(均不在既有 12 个测试的语料内) | hazards() | bash 实测 |
|
||||
|---|---|---|
|
||||
| `seq 1 200000 \| sed '5q;d'`(数字地址前缀的 sed q) | `[(1, "sed '5q;d'")]` 命中 | rc=141,无 "survived" |
|
||||
| `seq 1 200000 \| grep --max-count=3 1`(长选项 =值 形式) | `[(1, 'grep --max-count=3 1')]` 命中 | rc=141,无 "survived" |
|
||||
| `seq 1 200000 \| nice -5 head -1`(nice 旧式捆绑 -5 包装 head) | `[(1, 'nice -5 head -1')]` 命中 | rc=141,无 "survived" |
|
||||
|
||||
检测器对全部 3 个新形态均报出,且实测确为 SIGPIPE(141)——**检测器真能失败,非阻塞项**。配套只读套件 `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest tests.test_shell_pipe_hazard`:Ran 12 tests, OK。探针脚本仅在 /tmp/mi-review-probe,仓库内无残留。
|
||||
|
||||
## 七、执行过的只读命令与结果(证据)
|
||||
|
||||
| 命令 | 结果 |
|
||||
|---|---|
|
||||
| `git status --porcelain=v2 --branch`、`git log --oneline -8` | detached HEAD 2eaa751;范围含 fd3ccd0(任务卡)、2eaa751(transport,含测试变更) |
|
||||
| `git diff 5360884e92..HEAD --stat` / `--name-only` | 仅 2 文件:任务卡 + multi_instance_test.go(+95/−17) |
|
||||
| `git diff 5360884e92..HEAD -- Makefile .gitea/ \| wc -l` | 0(包级 -timeout 15m 未动) |
|
||||
| `git show 5360884e92:tests/…/multi_instance_test.go`(对照读) | 新旧逐 hunk 等价性核对依据 |
|
||||
| `gofmt -l .`(tests/integration/multi-instance) | 无输出(格式干净) |
|
||||
| `go vet .`(tests/integration/multi-instance) | exit 0,无输出(可编译可 vet) |
|
||||
| `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest tests.test_shell_pipe_hazard` | Ran 12 tests, OK |
|
||||
| /tmp/mi-review-probe/probe.py(hazards() + bash 141 实测) | 3/3 新形态命中,3/3 rc=141 |
|
||||
|
||||
**未验证项**:集成测试本体未运行(需 AGENT_PG_TEST_DSN + AGENT_BIN + 构建 backend,超出本只读审查范围)——两处修复在真实双进程下的运行时行为 NOT_RUN;status 词汇表未对照 backend 源码核验,标注 [未经证实]。
|
||||
|
||||
## 非阻塞发现汇总
|
||||
|
||||
1. restart_invariants 恢复视图判据的 interrupted 半边由历史 run 平凡满足,注释高估其覆盖面(不检出 B 重启对账回归;改动前同样不检出,覆盖无回退),并与 owner_crash_takeover 存在顺序耦合。
|
||||
2. busy 白名单(pending/running/waiting_hitl)对 backend 新增非终态 status 不设防——与 waitConversationIdle 相同的既有共享弱点,非本次引入。
|
||||
3. waitActiveHITL / waitRunStatus 轮询 B 时只附 logA,建议后续补 logB。
|
||||
4. `instanceOf` 的 `e.b != nil` 守卫不可达(baseFor 已先行解引用);无害死代码。
|
||||
5. restart_invariants 对 owner_crash_takeover 的顺序依赖:删改前者会使本等待在正常路径必然空转 15s 后失败(当前流内不触发)。
|
||||
6. 轮询经 `e.do` 对任意非 200 立即 Fatalf(含瞬态 5xx)——与改动前行为一致的响亮失败,可接受。
|
||||
|
||||
以上均不阻塞交付;本次变更可以进入 Review/合入流程(合入门禁按 policy/delivery.md 另行满足)。
|
||||
@@ -153,7 +153,7 @@ func TestTwoInstancesOverSharedPG(t *testing.T) {
|
||||
t.Fatalf("stale event replayed: seq %d <= since %d", ev.seq, lastSeq)
|
||||
}
|
||||
case <-time.After(8 * time.Second):
|
||||
t.Fatal("event written via A not visible on B SSE within 8s")
|
||||
t.Fatalf("event written via A not visible on B SSE within 8s: no event with seq > %d observed on %s", lastSeq, e.instanceOf(uc.streamB))
|
||||
}
|
||||
})
|
||||
|
||||
@@ -212,17 +212,24 @@ func TestTwoInstancesOverSharedPG(t *testing.T) {
|
||||
// (interrupted),随后原实例重新占用会话并推进。
|
||||
e.a = e.startInstance(t, e.a.name, strings.TrimPrefix(e.a.base, "http://"))
|
||||
var detail map[string]any
|
||||
deadline := time.Now().Add(httpWait)
|
||||
var statuses []string
|
||||
reconciled := false
|
||||
start := time.Now()
|
||||
deadline := start.Add(httpWait)
|
||||
for time.Now().Before(deadline) {
|
||||
e.do(t, uc.api, "GET", "/api/conversations/"+convID, nil, &detail)
|
||||
if runs, ok := detail["runs"].([]any); ok && len(runs) > 0 {
|
||||
last, _ := runs[len(runs)-1].(map[string]any)
|
||||
if last["status"] == "interrupted" {
|
||||
break
|
||||
}
|
||||
statuses = runStatuses(detail)
|
||||
if n := len(statuses); n > 0 && statuses[n-1] == "interrupted" {
|
||||
reconciled = true
|
||||
break
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
if !reconciled {
|
||||
// 不能落到下面的 POST:未收口时它会得到看似无关的 409 in-flight。
|
||||
t.Fatalf("restarted %s never reported the crashed run as interrupted: waited %v (budget %v), last run status never became \"interrupted\"; run statuses %v; log:\n%s",
|
||||
e.instanceOf(uc.api), time.Since(start).Round(time.Millisecond), httpWait, statuses, e.a.log.String())
|
||||
}
|
||||
e.do(t, uc.api, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "A resumes after restart"}, nil)
|
||||
})
|
||||
|
||||
@@ -230,6 +237,36 @@ func TestTwoInstancesOverSharedPG(t *testing.T) {
|
||||
e.killNine(t, e.b)
|
||||
e.b = e.startInstance(t, e.b.name, strings.TrimPrefix(e.b.base, "http://"))
|
||||
|
||||
// 断言前先等重启实例给出完整的恢复视图:对账收口的 interrupted run 可见,
|
||||
// 且会话内无进行中 run。断言本身不依赖这次轮询的结果。
|
||||
var statuses []string
|
||||
recovered := false
|
||||
start := time.Now()
|
||||
readyBy := start.Add(httpWait)
|
||||
for time.Now().Before(readyBy) {
|
||||
var view map[string]any
|
||||
e.do(t, uc.apiB, "GET", "/api/conversations/"+convID, nil, &view)
|
||||
statuses = runStatuses(view)
|
||||
interrupted, busy := false, false
|
||||
for _, s := range statuses {
|
||||
switch s {
|
||||
case "interrupted":
|
||||
interrupted = true
|
||||
case "pending", "running", "waiting_hitl":
|
||||
busy = true
|
||||
}
|
||||
}
|
||||
if interrupted && !busy {
|
||||
recovered = true
|
||||
break
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
if !recovered {
|
||||
t.Fatalf("restarted %s never showed the recovered view (an interrupted run and no pending/running/waiting_hitl run): waited %v (budget %v); run statuses %v; logA:\n%s\nlogB:\n%s",
|
||||
e.instanceOf(uc.apiB), time.Since(start).Round(time.Millisecond), httpWait, statuses, e.a.log.String(), e.b.log.String())
|
||||
}
|
||||
|
||||
var detail map[string]any
|
||||
e.do(t, uc.apiB, "GET", "/api/conversations/"+convID, nil, &detail)
|
||||
msgs, _ := detail["messages"].([]any)
|
||||
@@ -255,7 +292,7 @@ func TestTwoInstancesOverSharedPG(t *testing.T) {
|
||||
}
|
||||
seen = ev.seq
|
||||
case <-deadline:
|
||||
t.Fatalf("replay stalled at seq %d / %d", seen, lastSeq)
|
||||
t.Fatalf("replay stalled at seq %d / %d: event seq %d never observed on %s SSE within 15s", seen, lastSeq, seen+1, e.instanceOf(uc.streamB))
|
||||
}
|
||||
}
|
||||
// 重启后写入仍可推进。
|
||||
@@ -410,6 +447,7 @@ func (e *env) uploadArtifact(t *testing.T, c *http.Client, projectID, name strin
|
||||
|
||||
func (e *env) waitHealthy(t *testing.T, inst *instance) {
|
||||
t.Helper()
|
||||
last := "no probe completed"
|
||||
deadline := time.Now().Add(60 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
resp, err := http.Get(inst.base + "/api/health")
|
||||
@@ -419,10 +457,13 @@ func (e *env) waitHealthy(t *testing.T, inst *instance) {
|
||||
if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusUnauthorized {
|
||||
return
|
||||
}
|
||||
last = "status " + strconv.Itoa(resp.StatusCode)
|
||||
} else {
|
||||
last = err.Error()
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("instance %s not healthy in 60s; log:\n%s", inst.name, inst.log.String())
|
||||
t.Fatalf("instance %s (%s) not healthy in 60s: GET /api/health never returned 200/401 (last probe: %s); log:\n%s", inst.name, inst.base, last, inst.log.String())
|
||||
}
|
||||
|
||||
func (e *env) stopInstance(inst *instance) {
|
||||
@@ -642,6 +683,27 @@ func (e *env) baseFor(c *http.Client) string {
|
||||
return e.a.base
|
||||
}
|
||||
|
||||
// instanceOf 给出 client 实际轮询的实例(名 + 地址),供有界等待的失败信息定位。
|
||||
func (e *env) instanceOf(c *http.Client) string {
|
||||
base := e.baseFor(c)
|
||||
if e.b != nil && base == e.b.base {
|
||||
return e.b.name + " (" + base + ")"
|
||||
}
|
||||
return e.a.name + " (" + base + ")"
|
||||
}
|
||||
|
||||
// runStatuses 按顺序取会话详情里各 run 的状态,供等待条件与失败信息复用。
|
||||
func runStatuses(detail map[string]any) []string {
|
||||
runs, _ := detail["runs"].([]any)
|
||||
out := make([]string, 0, len(runs))
|
||||
for _, r := range runs {
|
||||
m, _ := r.(map[string]any)
|
||||
s, _ := m["status"].(string)
|
||||
out = append(out, s)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ---- 场景辅助 ----
|
||||
|
||||
type sseEvent struct {
|
||||
@@ -671,10 +733,13 @@ func (e *env) postMessage(t *testing.T, c *http.Client, convID, text string) {
|
||||
|
||||
func (e *env) waitConversationIdle(t *testing.T, c *http.Client, convID string, within time.Duration) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(within)
|
||||
var statuses []string
|
||||
start := time.Now()
|
||||
deadline := start.Add(within)
|
||||
for time.Now().Before(deadline) {
|
||||
var detail map[string]any
|
||||
e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
|
||||
statuses = runStatuses(detail)
|
||||
busy := false
|
||||
if runs, ok := detail["runs"].([]any); ok {
|
||||
for _, r := range runs {
|
||||
@@ -690,7 +755,8 @@ func (e *env) waitConversationIdle(t *testing.T, c *http.Client, convID string,
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("conversation %s still busy after %v", convID, within)
|
||||
t.Fatalf("conversation %s still busy on %s: waited %v (budget %v) and never observed all runs out of pending/running/waiting_hitl; run statuses %v",
|
||||
convID, e.instanceOf(c), time.Since(start).Round(time.Millisecond), within, statuses)
|
||||
}
|
||||
|
||||
func (e *env) lastSeq(t *testing.T, c *http.Client, convID string) int64 {
|
||||
@@ -755,11 +821,17 @@ func (e *env) subscribe(t *testing.T, c *http.Client, convID string, since int64
|
||||
|
||||
func (e *env) waitActiveHITL(t *testing.T, c *http.Client, convID string) hitlCard {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(15 * time.Second)
|
||||
var statuses []string
|
||||
cardCount := 0
|
||||
start := time.Now()
|
||||
deadline := start.Add(15 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
var detail map[string]any
|
||||
e.do(t, c, "GET", "/api/conversations/"+convID, nil, &detail)
|
||||
if cards, ok := detail["activeHitl"].([]any); ok && len(cards) > 0 {
|
||||
statuses = runStatuses(detail)
|
||||
cards, ok := detail["activeHitl"].([]any)
|
||||
cardCount = len(cards)
|
||||
if ok && len(cards) > 0 {
|
||||
m, _ := cards[0].(map[string]any)
|
||||
id, _ := m["id"].(string)
|
||||
runID, _ := m["runId"].(string)
|
||||
@@ -769,21 +841,27 @@ func (e *env) waitActiveHITL(t *testing.T, c *http.Client, convID string) hitlCa
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("no active HITL card in 15s; logA:\n%s", e.a.log.String())
|
||||
t.Fatalf("no active HITL card with id and runId on %s for conversation %s: waited %v (budget 15s); last activeHitl count %d, run statuses %v; logA:\n%s",
|
||||
e.instanceOf(c), convID, time.Since(start).Round(time.Millisecond), cardCount, statuses, e.a.log.String())
|
||||
return hitlCard{}
|
||||
}
|
||||
|
||||
func (e *env) waitRunStatus(t *testing.T, c *http.Client, runID, status string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(httpWait)
|
||||
observed := "<none>"
|
||||
start := time.Now()
|
||||
deadline := start.Add(httpWait)
|
||||
for time.Now().Before(deadline) {
|
||||
var out map[string]any
|
||||
e.do(t, c, "GET", "/api/runs/"+runID, nil, &out)
|
||||
run, _ := out["run"].(map[string]any)
|
||||
if s, _ := run["status"].(string); s == status {
|
||||
s, _ := run["status"].(string)
|
||||
if s == status {
|
||||
return
|
||||
}
|
||||
observed = s
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("run %s not %s within %v", runID, status, httpWait)
|
||||
t.Fatalf("run %s not %s on %s: waited %v (budget %v), last observed status %q; logA:\n%s",
|
||||
runID, status, e.instanceOf(c), time.Since(start).Round(time.Millisecond), httpWait, observed, e.a.log.String())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user