From b8af4341abe333e5edd35cdc3e5e7b1fead4a02b Mon Sep 17 00:00:00 2001
From: weichen <weichen@home.local>
Date: Wed, 14 Oct 2026 14:07:28 +0000
Subject: [PATCH] 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
---
 internal/server/run_process_events_test.go | 56 ++++++++++++++++++++++
 internal/server/server.go                  | 43 +++++++++++++++++
 internal/store/messages.go                 | 27 +++++++++++
 3 files changed, 126 insertions(+)
 create mode 100644 internal/server/run_process_events_test.go

diff --git a/internal/server/run_process_events_test.go b/internal/server/run_process_events_test.go
new file mode 100644
index 0000000..5e7978f
--- /dev/null
+++ b/internal/server/run_process_events_test.go
@@ -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)
+	}
+}
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..e015d68 100644
--- a/internal/store/messages.go
+++ b/internal/store/messages.go
@@ -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