diff --git a/.task/fix-ci-pg-flaky-tests.md b/.task/fix-ci-pg-flaky-tests.md new file mode 100644 index 0000000..9a5df6b --- /dev/null +++ b/.task/fix-ci-pg-flaky-tests.md @@ -0,0 +1,85 @@ +# fix/ci-pg-flaky-tests 任务卡 + +任务主记录：本文件随 Backend PR 描述维护；阻塞来源为 Dev PR #651。 +状态：已完成（合入前闭合）；owner：weichen + +## 基线与授权 + +- 权威仓：ssh://git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Backend.git；root_target：main；immediate_target：main +- branch_kind：standalone；initiative_id： +- R0（initial_main_sha）：b5eb89c557e69d532c8d0a6e2e2cbe0258a7f086；工作区：/home/weichen/projects/Agent-System-Backend-wt-fix-ci-pg-flaky-tests（worktree）；分支：fix/ci-pg-flaky-tests +- 父任务信息（仅子任务填写）：无 +- accepting_requirements：false +- 需求分流：verdict=NEW；selected_initiative=无；候选=无；理由=no accepting initiative；triage 收据=/tmp/backend-ci-flakes-triage-20261020.json；receipt_sha256=21ec3e6337883b983cc00d6b798c30fff859a94b77e0be197744832acd9f35e1；requirement_sha256=1a3a7a2160a5c543f0ab9f7f766034de5b90a8984d99989454b340966758d50e +- 授权与交付终点：用户授权排查和修复阻塞 Dev PR #651 合入的 Backend 测试失败，按项目门禁交付 Backend 修复；不部署、不改 CI、路由或其他业务。 +- 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md；外部 Agent 一单元一新进程/会话/工作区） + +## 目标与非目标 + +修复 CI 已观察到的门面卡锚点、后台完成等待、项目事件私有历史分页测试失败。保留断言与隐私边界，以真实异步完成条件消除竞态；不修改与这三个失败无关的功能、依赖或执行器。 + +## Agent 实施收据 + +按根目录 AGENT-EXECUTION.md 路由；外部 Agent 一单元一新进程/会话/隔离工作区。实施、修复与测试必须通过登记的候选执行；未配置或候选安全耗尽停在 unavailable/blocked，禁止 host 隐式代写。 + +| 单元 ID | 预期范围 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 | +|---|---|---|---|---|---|---|---|---|---| +| stabilize-ci-pg-server-tests | 原工作包文件清单与误用的 ** 模式 | repair | cursor, antigravity, pi | 1 | cursor / auto | scope-expansion | 无 | false | /tmp/backend-ci-flakes-implementation-20261020.json | + +| ci-pg-async-completion-r2 | e2e_test.go / project_events_test.go / background_test.go | repair | cursor, antigravity, pi | 1 | cursor / auto (effective not-observable) | completed (controller success) | 2ee469011e66b6bedcf6e9d87c8b1c026f7a8cef | false | /tmp/pr651-repair-r2.json | + +## 复用材料（task-start 自动检索结果摘录） + +- 相关近期提交： 命中；无则写无>。 +- 相关经验条目：<检索命中；无则写无>。 + +## 同步记录 + +| 节点 | sync_base_sha | tested_head_sha | checked_at | 结果 | +|---|---|---|---|---| +| 启动 | | | <日期时间> | <一致/已同步> | +| 验收前 | <填写> | <填写> | <填写> | <填写> | + +字段没有发生的写"无"，不省略。本卡与 PR 描述同源：开 PR 时将"目标与非目标、结果与验证、影响与剩余项"复制进 PR 描述，不另写第二份。 + +## 交付调度（有多个就绪任务时） + +- 调度记录 / entry_id / 当前 attempt_id：<平台或调度服务链接及标识；未入队写无>。 +- 入队源 HEAD / owner / 依赖：<完整 SHA、身份、依赖或无>。 +- 固定候选引用：<调度尝试中的 base/head、组件与配置；未固定写无>。 +- 窗口、最终 CI、Review 与真实合入结果：<引用调度记录及原始证据，不复制全队列>。 + +定义见 [合入调度与 CI 分配](../standards/merge-queue.md)。本段在候选固定前整理；固定后先使用源码之外的记录保存运行结果，再按项目闭合流程回填。 + +## 结果与验证（完成后填写） + +- 结果：仅修复三个测试文件的异步完成等待；门面卡等待 run.completed 持久事件，项目分页读到共享帧即结束，后台 idle 使用独立有界等待。生产代码、CI 与断言未修改。 +- 验证：worker red/green 对照与 focused count=10 通过；host 独立相关用例 count=3、go vet ./...、go test -count=1 ./... 通过（SELF）。完整原始收据由 Dev make evidence 捕获：/tmp/pr651-backend-host-full-20261020.json；源码对象 2ee469011e66b6bedcf6e9d87c8b1c026f7a8cef。当前最终 HEAD GATE 待运行，不以 SELF 放行。 +- 影响与剩余项：只影响测试等待；30s/20s 是基于磁盘实测及停顿注入的卡死上限，实际 CI 停顿时长仍不确定。AuthOwner 每帧持久写入的性能观察不在本次范围。旧 fail-stop 隔离现场保留。 + +| 时间(UTC) | 命令 | 被测 HEAD | exit | 来源 | 摘要/位置 | +|---|---|---|---|---|---| +| <时间> | <命令；由项目 evidence 工具写入，无工具时附命令+对象 SHA+输出 digest 并标 SELF> | | | GATE/REPLAY/SELF | <摘要/位置> | + +## 回收（task-finish） + +- 合入核对：<分支已包含于目标分支的核对输出>。 +- 未跟踪/ignored 文件处置：<归档/可再生/存疑 逐项；无则写无>。 +- 回收动作：。 + +## 恢复与重新定义单元（2026-10-20） + +- 用户在交接后明确“继续”；旧 run 290c8b13f4a4a9e5d67df066754ab3fd 已停止，不自动重试或接纳其诊断文件。 +- 现场已对账：唯一源文件差异为原隔离目录的 zz_diag_tmp_test.go；实际事件中仅有读源码、局部构建与本机测试/缓存操作，没有发现提交/推送、正式工作树修改或外部请求。原进程与现场写者已退出；隔离目录与本机诊断产物均保留。 +- 原执行器 allowed_paths 使用精确路径/目录前缀匹配，工作包的 ** 不起通配作用。新单元采用明确文件清单；诊断只能在这些现有测试中或仓外生成，禁止继续使用原会话。 +- 任务 worktree 已完整移至本卡登记位置，Git 与 development-manager 子模块路径已修复；没有删除文件。 +- Dev 启动 SELF 收据：/tmp/pr651-resume-20261020-a1.json（发现嵌套 worktree）、a2 至 a5（后续恢复检查，均未宣称通过）。原始完整文件在 /tmp。 +- 权威 Backend main 已由 API 核验并合入 c16495006dbf1cf6ee581b48d99bba1e7f9a377c；继承的 O2 改动仅为必需目标同步，不属本次修复。固定 Guidelines 与依赖已物化，配置与治理准入均通过。 +- 新单元目标：分别确认三项失败根因，在现有测试的精确路径内建立 red/green 证据；若确需修改生产逻辑，先形成能复现真实行为缺陷的失败测试。禁止弱化断言、仅拉长时间或不受控压力测试。 + +## Review 与交付 + +- 主协调者独立 Review：base c16495006dbf1cf6ee581b48d99bba1e7f9a377c，实现对象 2ee469011e66b6bedcf6e9d87c8b1c026f7a8cef；已核对真实完成事件的顺序、共享帧位于所有私有帧之后、等待仍有界、所有隐私与锚点断言保留；无阻塞发现。 +- 实施包与终态原始事件分开保存：/tmp/pr651-repair-unit-r2.json、/tmp/pr651-repair-r2.json、/tmp/agent-exec-1000/662a926e726431b930433273fc26f39a/raw-e8392e75611f.log；有效 completed 已在实际 result envelope 中核验。 +- 原始实验输出：/tmp/pr651-r2-red-anchor.txt、/tmp/pr651-r2-green-anchor-experiment.txt、/tmp/pr651-r2-stall-redgreen.txt。临时实验代码已撤除。 +- 交付：当前最终 HEAD 仍需自身 CI（ci/vet-and-test 与 governance），权威目标同步核验后经 PR 合入；GATE 与实际合入结果在源码外记录，不修改冻结候选。 diff --git a/internal/runmgr/background_test.go b/internal/runmgr/background_test.go index 44e17f7..aa328fe 100644 --- a/internal/runmgr/background_test.go +++ b/internal/runmgr/background_test.go @@ -49,7 +49,12 @@ func (noopRunner) Execute(rc runtime.RunCtx) error { func waitUntil(t *testing.T, cond func() bool) { t.Helper() - deadline := time.Now().Add(2 * time.Second) + waitUntilWithin(t, 2*time.Second, cond) +} + +func waitUntilWithin(t *testing.T, bound time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(bound) for time.Now().Before(deadline) { if cond() { return @@ -79,9 +84,13 @@ func systemTexts(t *testing.T, c *runCtx) []string { return out } +// backgroundIdleBound only bounds a hang: each child settles through several +// fsync'd SQLite commits, and CI disk stalls have exceeded a 2s window. +const backgroundIdleBound = 20 * time.Second + func waitBackgroundIdle(t *testing.T, c *runCtx) { t.Helper() - waitUntil(t, func() bool { + waitUntilWithin(t, backgroundIdleBound, func() bool { c.mgr.noticeMu.Lock() defer c.mgr.noticeMu.Unlock() return c.mgr.bgInFlight[c.run.ConversationID] == 0 diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index c2b1af9..b1b1426 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -216,14 +216,44 @@ func waitRunTerminal(t *testing.T, h http.Handler, runID string, timeout time.Du return nil } +// waitRunCompleted also waits for the persisted run.completed event. The run row +// turns terminal in the settle commit, before finishRun backfills card anchors; +// run.completed is emitted only after that backfill. +func waitRunCompleted(t *testing.T, h http.Handler, st *store.Store, convID, runID string, timeout time.Duration) map[string]any { + t.Helper() + terminal := waitRunTerminal(t, h, runID, timeout) + if terminal["status"] != "completed" { + return terminal + } + deadline := time.Now().Add(timeout) + var after int64 + for time.Now().Before(deadline) { + events, err := st.EventsAfter(convID, after, 1000) + if err != nil { + t.Fatal(err) + } + for _, e := range events { + if e.RunID == runID && e.Type == core.EvRunCompleted { + return terminal + } + after = e.Seq + } + if len(events) < 1000 { + time.Sleep(10 * time.Millisecond) + } + } + t.Fatalf("run %s completed without run.completed event after %v", runID, timeout) + return nil +} + // TestMockCardflowFullLoop 卡流全链路（D39/D35/D32）：mock #cardflow → result 卡入收件箱 // → 助手消息含 cardId 引用部件 → finishRun 回填锚点（卡→消息互定位）。 func TestMockCardflowFullLoop(t *testing.T) { - srv, _ := newTestServer(t) + srv, st := newTestServer(t) h := srv.Handler() _, convID := setupConversation(t, h) runID := postMessage(t, h, convID, "#cardflow") - terminal := waitRunTerminal(t, h, runID, 5*time.Second) + terminal := waitRunCompleted(t, h, st, convID, runID, 5*time.Second) if terminal["status"] != "completed" { t.Fatalf("run status: %v", terminal["status"]) } @@ -272,11 +302,11 @@ func TestMockCardflowFullLoop(t *testing.T) { // card_begin（active 卡入收件箱）→ card_append×N（card.updated 增量）→ // card_commit（done + cardId 引用部件 + finishRun 锚点回填），且正文逐段合并。 func TestMockCardflowStreamLoop(t *testing.T) { - srv, _ := newTestServer(t) + srv, st := newTestServer(t) h := srv.Handler() _, convID := setupConversation(t, h) runID := postMessage(t, h, convID, "#cardstream") - terminal := waitRunTerminal(t, h, runID, 5*time.Second) + terminal := waitRunCompleted(t, h, st, convID, runID, 5*time.Second) if terminal["status"] != "completed" { t.Fatalf("run status: %v", terminal["status"]) } @@ -617,11 +647,11 @@ func TestArtifactFlow(t *testing.T) { // artifact_ref 部件 → 门面卡入收件箱（摘要 kv + doc-ref）→ 消息含 card 引用 → // finishRun 锚点回填。 func TestArtifactFacadeCard(t *testing.T) { - srv, _ := newTestServer(t) + srv, st := newTestServer(t) h := srv.Handler() _, convID := setupConversation(t, h) runID := postMessage(t, h, convID, "#doc") - terminal := waitRunTerminal(t, h, runID, 5*time.Second) + terminal := waitRunCompleted(t, h, st, convID, runID, 5*time.Second) if terminal["status"] != "completed" { t.Fatalf("run status: %v", terminal["status"]) } diff --git a/internal/server/project_events_test.go b/internal/server/project_events_test.go index 7bff0cd..ca250b0 100644 --- a/internal/server/project_events_test.go +++ b/internal/server/project_events_test.go @@ -28,6 +28,35 @@ func projectStreamReplay(t *testing.T, s *Server, key, project string) *httptest s.Handler().ServeHTTP(w, r.WithContext(ctx)) return w } + +// projectStreamReplayUntil reads a replay until marker arrives or the stream +// ends. Every replayed frame rechecks the API key with a durable write, so a +// thousand-event replay has taken over 11s on a disk-backed SQLite; the +// deadline only bounds a hang. +func projectStreamReplayUntil(t *testing.T, s *Server, key, project, marker string) (int, []byte) { + t.Helper() + httpServer := httptest.NewServer(s.Handler()) + defer httpServer.Close() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + r, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+project+"&since=0", nil) + r.Header.Set("Authorization", "Bearer "+key) + response, err := httpServer.Client().Do(r) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + reader := bufio.NewReader(response.Body) + var body bytes.Buffer + for !bytes.Contains(body.Bytes(), []byte(marker)) { + line, err := reader.ReadString('\n') + body.WriteString(line) + if err != nil { + break + } + } + return response.StatusCode, body.Bytes() +} func projectEventFixture(t *testing.T) (*Server, string, string, string, *atomic.Bool) { t.Helper() s, st := newAuthTestServer(t) @@ -78,10 +107,12 @@ func TestProjectEventsSharedProofAndPrivatePagination(t *testing.T) { if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "EXPLICIT SHARED"}, ""); err != nil { t.Fatal(err) } - w := projectStreamReplay(t, s, bob, p) - if w.Code != 200 || bytes.Contains(w.Body.Bytes(), []byte("PRIVATE HISTORY")) || !bytes.Contains(w.Body.Bytes(), []byte("EXPLICIT SHARED")) { - t.Fatalf("private pagination/spoof proof failed: code=%d private=%t shared=%t bytes=%d", w.Code, - bytes.Contains(w.Body.Bytes(), []byte("PRIVATE HISTORY")), bytes.Contains(w.Body.Bytes(), []byte("EXPLICIT SHARED")), w.Body.Len()) + // The shared frame follows every private event in seq order, so reading up + // to it observes the whole private replay. + code, body := projectStreamReplayUntil(t, s, bob, p, "EXPLICIT SHARED") + if code != 200 || bytes.Contains(body, []byte("PRIVATE HISTORY")) || !bytes.Contains(body, []byte("EXPLICIT SHARED")) { + t.Fatalf("private pagination/spoof proof failed: code=%d private=%t shared=%t bytes=%d", code, + bytes.Contains(body, []byte("PRIVATE HISTORY")), bytes.Contains(body, []byte("EXPLICIT SHARED")), len(body)) } } func TestProjectEventsLivePrivacyAndRevocation(t *testing.T) {