diff --git a/internal/server/run_process_events_test.go b/internal/server/run_process_events_test.go index 5e7978f..76d0533 100644 --- a/internal/server/run_process_events_test.go +++ b/internal/server/run_process_events_test.go @@ -49,6 +49,31 @@ func TestRunProcessEventsAreOrderedAndBoundToConversation(t *testing.T) { if payload := page[0].(map[string]any)["payload"].(map[string]any); payload["text"] != "完整工具输出" { t.Fatalf("payload: %v", payload) } + parent, err := st.GetRunByID(runID) + if err != nil { + t.Fatal(err) + } + child := core.Run{ID: "run_child_process", ParentRunID: parent.ID, ConversationID: parent.ConversationID, + ProjectID: parent.ProjectID, Owner: parent.Owner, Kind: core.RunSubagent, Status: core.RunCompleted} + if err := st.CreateRun(&child); err != nil { + t.Fatal(err) + } + childEvents := []core.StreamEvent{ + {RunID: child.ID, Type: core.EvAgentProgress, Payload: json.RawMessage(`{"agentRun":"run_child_process","note":"正在核对附录"}`)}, + {RunID: child.ID, Type: core.EvToolOutput, Payload: json.RawMessage(`{"callId":"hidden","text":"child raw output"}`)}, + {RunID: child.ID, Type: core.EvAgentResult, Payload: json.RawMessage(`{"agentRun":"run_child_process","status":"completed","summary":"附录已核对"}`)}, + } + if _, err := st.AppendEvents(convID, childEvents); err != nil { + t.Fatal(err) + } + code, childPage := doJSON(t, h, http.MethodGet, "/api/conversations/"+convID+"/runs/"+runID+"/events?after="+strconv.FormatInt(events[1].Seq, 10), nil) + if code != http.StatusOK { + t.Fatalf("child page: %d %v", code, childPage) + } + childRows := childPage["events"].([]any) + if len(childRows) != 2 || childRows[0].(map[string]any)["type"] != string(core.EvAgentProgress) || childRows[1].(map[string]any)["type"] != string(core.EvAgentResult) { + t.Fatalf("child progress/result not retained or raw child output leaked: %v", childRows) + } code, _ = doJSON(t, h, http.MethodGet, "/api/conversations/"+otherConvID+"/runs/"+runID+"/events", nil) if code != http.StatusNotFound { t.Fatalf("mismatched conversation exposed run: %d", code) diff --git a/internal/store/messages.go b/internal/store/messages.go index e015d68..e2b7490 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -355,11 +355,16 @@ func (s *Store) EventsAfter(conversationID string, since int64, limit int) ([]co return out, rows.Err() } -// RunEventsAfter reads a single run's persisted event stream in conversation seq order. +// RunEventsAfter reads a run's persisted stream plus its direct children's +// collaboration status events in conversation seq order. Raw child output is +// intentionally outside the parent's previously visible process stream. // The API checks conversation and run ownership before calling this method. func (s *Store) RunEventsAfter(conversationID, runID string, since int64, limit int) ([]core.StreamEvent, error) { - rows, err := s.db.Query(s.q(`SELECT seq,run_id,ts,v,type,payload FROM events WHERE conversation_id=? AND run_id=? AND seq>? ORDER BY seq LIMIT ?`), - conversationID, runID, since, limit) + rows, err := s.db.Query(s.q(`SELECT seq,run_id,ts,v,type,payload FROM events + WHERE conversation_id=? AND seq>? AND ( + run_id=? OR (run_id IN (SELECT id FROM runs WHERE parent_run_id=? AND conversation_id=?) + AND type IN ('agent.progress','agent.result')) + ) ORDER BY seq LIMIT ?`), conversationID, since, runID, runID, conversationID, limit) if err != nil { return nil, err }