Expose persisted run events for process review

Why: Completed replies currently drop access to streamed thinking and tool output.

Evidence: go test ./...; isolated run-event pagination and conversation binding test.

Impact: Add owner-scoped paginated event reading for a single conversation run.

Refs: Agent-System-Dev feature/work-companion-visibility UI1
This commit is contained in:
2026-10-14 14:07:28 +00:00
parent 607c38d7e6
commit b8af4341ab
3 changed files with 126 additions and 0 deletions
@@ -0,0 +1,56 @@
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)
}
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)
}
}
+43
View File
@@ -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 {
+27
View File
@@ -355,6 +355,33 @@ 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.
// 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)
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