From 2ee469011e66b6bedcf6e9d87c8b1c026f7a8cef Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Tue, 20 Oct 2026 09:03:51 +0000 Subject: [PATCH] agent-exec transport 662a926e726431b930433273fc26f39a/e8392e75611f --- internal/runmgr/background_test.go | 13 ++++++-- internal/server/e2e_test.go | 42 ++++++++++++++++++++++---- internal/server/project_events_test.go | 39 +++++++++++++++++++++--- 3 files changed, 82 insertions(+), 12 deletions(-) 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) {