agent-exec transport 662a926e726431b930433273fc26f39a/e8392e75611f

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

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"])
 	}

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) {