diff --git a/tests/integration/multi-instance/multi_instance_test.go b/tests/integration/multi-instance/multi_instance_test.go index 408765b2..13018213 100644 --- a/tests/integration/multi-instance/multi_instance_test.go +++ b/tests/integration/multi-instance/multi_instance_test.go @@ -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 := "" + 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()) }