Wait for completed projections in flaky Backend tests #226
@@ -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 自动检索结果摘录）
|
||||
|
||||
- 相关近期提交：<git log -- <paths> 命中；无则写无>。
|
||||
- 相关经验条目：<检索命中；无则写无>。
|
||||
|
||||
## 同步记录
|
||||
|
||||
| 节点 | sync_base_sha | tested_head_sha | checked_at | 结果 |
|
||||
|---|---|---|---|---|
|
||||
| 启动 | <R0> | <HEAD> | <日期时间> | <一致/已同步> |
|
||||
| 验收前 | <填写> | <填写> | <填写> | <填写> |
|
||||
|
||||
字段没有发生的写"无"，不省略。本卡与 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> | <HEAD> | <exit> | GATE/REPLAY/SELF | <摘要/位置> |
|
||||
|
||||
## 回收（task-finish）
|
||||
|
||||
- 合入核对：<分支已包含于目标分支的核对输出>。
|
||||
- 未跟踪/ignored 文件处置：<归档/可再生/存疑 逐项；无则写无>。
|
||||
- 回收动作：<worktree remove、分支删除、prune；放弃路径注明原因>。
|
||||
|
||||
## 恢复与重新定义单元（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 与实际合入结果在源码外记录，不修改冻结候选。
|
||||
@@ -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
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user