diff --git a/internal/server/run_process_events_test.go b/internal/server/run_process_events_test.go new file mode 100644 index 0000000..76d0533 --- /dev/null +++ b/internal/server/run_process_events_test.go @@ -0,0 +1,81 @@ +package server + +import ( + "encoding/json" + "net/http" + "strconv" + "testing" + "time" + + core "agent-system-core" +) + +func TestRunProcessEventsAreOrderedAndBoundToConversation(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + _, convID := setupConversation(t, h) + _, otherConvID := setupConversation(t, h) + runID := postMessage(t, h, convID, "整理材料") + waitRunTerminal(t, h, runID, 5*time.Second) + last, err := st.LastEventSeq(convID) + if err != nil { + t.Fatal(err) + } + events := []core.StreamEvent{ + {RunID: runID, Type: core.EvThinkingDelta, Payload: json.RawMessage(`{"index":0,"text":"先核对日期"}`)}, + {RunID: runID, Type: core.EvToolOutput, Payload: json.RawMessage(`{"callId":"call-1","stream":"stdout","text":"完整工具输出"}`)}, + } + if _, err := st.AppendEvents(convID, events); err != nil { + t.Fatal(err) + } + path := "/api/conversations/" + convID + "/runs/" + runID + "/events?after=" + strconv.FormatInt(last, 10) + "&limit=1" + code, first := doJSON(t, h, http.MethodGet, path, nil) + if code != http.StatusOK { + t.Fatalf("first page: %d %v", code, first) + } + page := first["events"].([]any) + if len(page) != 1 || page[0].(map[string]any)["type"] != string(core.EvThinkingDelta) { + t.Fatalf("first page: %v", page) + } + seq := int64(page[0].(map[string]any)["seq"].(float64)) + code, second := doJSON(t, h, http.MethodGet, "/api/conversations/"+convID+"/runs/"+runID+"/events?after="+strconv.FormatInt(seq, 10)+"&limit=1", nil) + if code != http.StatusOK { + t.Fatalf("second page: %d %v", code, second) + } + page = second["events"].([]any) + if len(page) != 1 || page[0].(map[string]any)["type"] != string(core.EvToolOutput) { + t.Fatalf("second page: %v", page) + } + 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/server/server.go b/internal/server/server.go index d90b3c3..fb905f3 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -300,6 +300,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/projects/{id}/conversations", s.createConversation) mux.HandleFunc("GET /api/conversations", s.listTasks) mux.HandleFunc("GET /api/conversations/{id}", s.conversationDetail) + mux.HandleFunc("GET /api/conversations/{id}/runs/{runId}/events", s.runProcessEvents) mux.HandleFunc("GET /api/conversations/{id}/export", s.exportConversation) mux.HandleFunc("PATCH /api/conversations/{id}", s.patchConversation) mux.HandleFunc("DELETE /api/conversations/{id}", s.deleteConversation) @@ -878,6 +879,48 @@ func (s *Server) conversationDetail(w http.ResponseWriter, r *http.Request) { }) } +// runProcessEvents lets an owner reopen the exact persisted process after the +// transient SSE bubble has been replaced by the settled assistant message. +func (s *Server) runProcessEvents(w http.ResponseWriter, r *http.Request) { + id, runID := r.PathValue("id"), r.PathValue("runId") + owner := s.owner(r) + if _, err := s.st.GetConversation(owner, id); err != nil { + writeStoreErr(w, err) + return + } + run, err := s.st.GetRun(owner, runID) + if err != nil { + writeStoreErr(w, err) + return + } + if string(run.ConversationID) != id { + http.NotFound(w, r) + return + } + after, err := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64) + if r.URL.Query().Get("after") == "" { + after, err = 0, nil + } + if err != nil || after < 0 { + http.Error(w, "invalid after", http.StatusBadRequest) + return + } + limit := 500 + if raw := r.URL.Query().Get("limit"); raw != "" { + limit, err = strconv.Atoi(raw) + if err != nil || limit < 1 || limit > 500 { + http.Error(w, "invalid limit", http.StatusBadRequest) + return + } + } + events, err := s.st.RunEventsAfter(id, runID, after, limit) + if err != nil { + writeStoreErr(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"events": events}) +} + func (s *Server) patchConversation(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") var req struct { diff --git a/internal/store/messages.go b/internal/store/messages.go index 58724f2..e2b7490 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -355,6 +355,38 @@ func (s *Store) EventsAfter(conversationID string, since int64, limit int) ([]co return out, rows.Err() } +// 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 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 + } + defer rows.Close() + out := []core.StreamEvent{} + for rows.Next() { + var e core.StreamEvent + var typ string + var payload []byte + if err := rows.Scan(&e.Seq, &e.RunID, &e.TS, &e.V, &typ, &payload); err != nil { + return nil, err + } + e.ConversationID = conversationID + e.Type = core.StreamEventType(typ) + if len(payload) > 0 { + e.Payload = payload + } + out = append(out, e) + } + return out, rows.Err() +} + // EventAt 读取单个事件(PG 档 NOTIFY 扇出的按序号取回路径)。 func (s *Store) EventAt(conversationID string, seq int64) (core.StreamEvent, error) { var e core.StreamEvent