mirror of
https://github.com/RuoJi6/ARTEX.git
synced 2026-10-27 17:03:49 +08:00
feat(activity): 任务会话历史反向分页 + SSE 可靠性整改
后端:
- ActivityPage(会话过滤 main/plan/intent + before 反向分页)、ActivityMaxID
(任务级快照游标)、ListByKindPage(Intent 列表分页,突破固定 300 上限)
- 新增 GET /api/exploration/activity/history(items/snapshot_cursor/
earliest_cursor/has_more);snapshot_cursor 为任务级游标
- streamActivity:一次 2000 → 循环补偿追平;输出标准 SSE id 行;支持
Last-Event-ID;补偿查询错误记录日志并关闭连接
- intents 支持分页形态(?before/?page → {items,has_more}),无参数保持旧裸数组
- 新增 idx_act_worker(exploration_id, worker, id) 支撑 Main/Plan 反向翻页
(schema.sql 幂等,旧库启动自动补建,无迁移)
前端:
- sessions-tab 单一 allActivity → 分会话缓存;首屏只加载 Main 最新页并据其
快照游标建唯一任务级 SSE;Plan/Worker 点击懒加载
- 统一 mergeBySeq 合并去重;未激活会话只维护 lastTs/unread;向上滚动分页
加载旧历史并保持滚动位置;每会话内存上限裁剪
- 按 session_key 写各自缓存 + 每 key 请求令牌防会话切换串数据;SSE 连接状态
可见、加载失败可重试,异常不再静默为"暂无消息"
- api 新增 activityHistory / intentsPage
测试:db/activity_page_test.go 覆盖多 Agent 过滤、反向翻页无重复无遗漏、
快照游标、Intent 超页分页(已对真实 PostgreSQL 通过)
This commit is contained in:
1 parent
95aba396c2
commit
79493d5a4a
6 files changed
+906
-130
No files matched your search
@@ -0,0 +1,178 @@
|
||||
package db
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestActivityPageSessions covers the reverse-paginated, per-session history added
|
||||
// for the SSE remediation: Main/Plan/Worker filtering, before-cursor paging without
|
||||
// gaps/overlap, hasMore, and the task-level snapshot cursor. Mirrors docs §11.1.
|
||||
func TestActivityPageSessions(t *testing.T) {
|
||||
d, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) — skipping", err)
|
||||
}
|
||||
defer d.Close()
|
||||
|
||||
expID, err := d.CreateExploration("test", "分页历史")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Exec(`DELETE FROM explorations WHERE id=$1`, expID)
|
||||
es := d.Exploration(expID)
|
||||
|
||||
intentA, err := es.AddIntent(map[string]any{"summary": "intent A"}, 5, nil, "planner")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
intentB, err := es.AddIntent(map[string]any{"summary": "intent B"}, 5, nil, "planner")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Interleave a mix of agents so a session filter must actually discriminate.
|
||||
// 25 main, 25 planner (Goal+Planner share worker=planner), 30 workerA, 5 workerB.
|
||||
appendN := func(n int, a Activity) {
|
||||
for range n {
|
||||
if _, err := es.AppendActivity(a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
// Interleaving order matters: emit round-robin-ish so ids of one session are
|
||||
// scattered, proving the WHERE filter (not a contiguous range) is what selects.
|
||||
for range 25 {
|
||||
appendN(1, Activity{Worker: "mainagent", Kind: "text", Summary: "m"})
|
||||
appendN(1, Activity{Worker: "planner", Kind: "text", Summary: "p"})
|
||||
appendN(1, Activity{NodeID: &intentA, Worker: "work#1", Kind: "text", Summary: "a"})
|
||||
}
|
||||
appendN(5, Activity{NodeID: &intentA, Worker: "work#1", Kind: "text", Summary: "a2"}) // workerA → 30 total
|
||||
appendN(5, Activity{NodeID: &intentB, Worker: "work#2", Kind: "text", Summary: "b"})
|
||||
|
||||
// snapshot cursor = max id across the whole task.
|
||||
snap, err := es.ActivityMaxID()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Helper: page through a whole session backward and assert coverage.
|
||||
collect := func(f ActivitySessionFilter, pageSize int) []Activity {
|
||||
var all []Activity
|
||||
before := int64(0)
|
||||
seen := map[int64]bool{}
|
||||
for {
|
||||
items, hasMore, err := es.ActivityPage(f, before, pageSize)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// ascending order within a page
|
||||
for i := 1; i < len(items); i++ {
|
||||
if items[i-1].ID >= items[i].ID {
|
||||
t.Fatalf("page not ascending: %d >= %d", items[i-1].ID, items[i].ID)
|
||||
}
|
||||
}
|
||||
// no overlap across pages
|
||||
for _, a := range items {
|
||||
if seen[a.ID] {
|
||||
t.Fatalf("duplicate id %d across pages", a.ID)
|
||||
}
|
||||
seen[a.ID] = true
|
||||
}
|
||||
all = append([]Activity{}, append(items, all...)...) // prepend older page
|
||||
if !hasMore || len(items) == 0 {
|
||||
break
|
||||
}
|
||||
before = items[0].ID
|
||||
}
|
||||
return all
|
||||
}
|
||||
|
||||
main := collect(ActivitySessionFilter{Worker: "mainagent"}, 10)
|
||||
if len(main) != 25 {
|
||||
t.Fatalf("main count = %d, want 25", len(main))
|
||||
}
|
||||
plan := collect(ActivitySessionFilter{Worker: "planner"}, 7)
|
||||
if len(plan) != 25 {
|
||||
t.Fatalf("plan count = %d, want 25", len(plan))
|
||||
}
|
||||
wa := collect(ActivitySessionFilter{NodeID: &intentA}, 8)
|
||||
if len(wa) != 30 {
|
||||
t.Fatalf("workerA count = %d, want 30", len(wa))
|
||||
}
|
||||
wb := collect(ActivitySessionFilter{NodeID: &intentB}, 8)
|
||||
if len(wb) != 5 {
|
||||
t.Fatalf("workerB count = %d, want 5", len(wb))
|
||||
}
|
||||
// full ascending order across the reconstructed session
|
||||
for i := 1; i < len(wa); i++ {
|
||||
if wa[i-1].ID >= wa[i].ID {
|
||||
t.Fatalf("reconstructed session not ascending at %d", i)
|
||||
}
|
||||
}
|
||||
// latest page (before=0) must include the session's newest record.
|
||||
latest, hasMore, err := es.ActivityPage(ActivitySessionFilter{NodeID: &intentA}, 0, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !hasMore {
|
||||
t.Fatalf("workerA should have more than one page")
|
||||
}
|
||||
if latest[len(latest)-1].ID != wa[len(wa)-1].ID {
|
||||
t.Fatalf("latest page missing newest record")
|
||||
}
|
||||
// snapshot cursor is the whole-task max, ≥ any session's max.
|
||||
if snap < wa[len(wa)-1].ID {
|
||||
t.Fatalf("snapshot %d < workerA max %d", snap, wa[len(wa)-1].ID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListByKindPage covers the paged worker(intent) list that lets the session list
|
||||
// reach past the old fixed 300 cap (docs §8 / §11.1 item 10).
|
||||
func TestListByKindPage(t *testing.T) {
|
||||
d, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) — skipping", err)
|
||||
}
|
||||
defer d.Close()
|
||||
|
||||
expID, err := d.CreateExploration("test", "意图分页")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Exec(`DELETE FROM explorations WHERE id=$1`, expID)
|
||||
es := d.Exploration(expID)
|
||||
|
||||
const total = 25
|
||||
for range total {
|
||||
if _, err := es.AddIntent(map[string]any{"summary": "i"}, 1, nil, "planner"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// Page backward in chunks of 10; expect 10,10,5 and hasMore false on last.
|
||||
seen := map[int64]bool{}
|
||||
before := int64(0)
|
||||
pages := 0
|
||||
for {
|
||||
items, hasMore, err := es.ListByKindPage(KindIntent, before, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pages++
|
||||
for _, n := range items {
|
||||
if seen[n.ID] {
|
||||
t.Fatalf("dup intent %d across pages", n.ID)
|
||||
}
|
||||
seen[n.ID] = true
|
||||
}
|
||||
if len(items) == 0 || !hasMore {
|
||||
break
|
||||
}
|
||||
// newest-first within a page → the oldest (smallest id) is last; page older
|
||||
// history before it next.
|
||||
before = items[len(items)-1].ID
|
||||
}
|
||||
if len(seen) != total {
|
||||
t.Fatalf("paged intents = %d, want %d", len(seen), total)
|
||||
}
|
||||
if pages < 3 {
|
||||
t.Fatalf("expected ≥3 pages for %d items at size 10, got %d", total, pages)
|
||||
}
|
||||
}
|
||||
@@ -267,6 +267,32 @@ WHERE exploration_id=$1 AND kind=$2 ORDER BY id DESC LIMIT $3`, s.expID, kind, l
|
||||
return scanNodes(rows)
|
||||
}
|
||||
|
||||
// ListByKindPage returns one newest-first page of nodes of a kind for reverse
|
||||
// pagination: up to `limit` nodes with id < `before` (before<=0 = the newest page).
|
||||
// hasMore reports whether still-older nodes exist, so a client (e.g. the worker
|
||||
// session list) can page past the old fixed cap instead of losing older intents.
|
||||
func (s *ExplorationStore) ListByKindPage(kind string, before int64, limit int) ([]*Node, bool, error) {
|
||||
if limit <= 0 {
|
||||
limit = 300
|
||||
}
|
||||
rows, err := s.db.Query(`SELECT `+nodeCols+` FROM exploration_nodes
|
||||
WHERE exploration_id=$1 AND kind=$2 AND ($3 <= 0 OR id < $3)
|
||||
ORDER BY id DESC LIMIT $4`, s.expID, kind, before, limit+1)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
defer rows.Close()
|
||||
nodes, err := scanNodes(rows)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
hasMore := len(nodes) > limit
|
||||
if hasMore {
|
||||
nodes = nodes[:limit]
|
||||
}
|
||||
return nodes, hasMore, nil
|
||||
}
|
||||
|
||||
// GetNode returns one node of this exploration by id (nil, nil if not found).
|
||||
func (s *ExplorationStore) GetNode(id int64) (*Node, error) {
|
||||
n, err := scanNode(s.db.QueryRow(`SELECT `+nodeCols+` FROM exploration_nodes WHERE id=$1 AND exploration_id=$2`, id, s.expID))
|
||||
@@ -560,6 +586,90 @@ FROM activity WHERE exploration_id=$1 AND id>$2 ORDER BY id LIMIT $3`, s.expID,
|
||||
return out, cursor, rows.Err()
|
||||
}
|
||||
|
||||
// ActivitySessionFilter selects one UI "session" within a task's activity stream.
|
||||
// Exactly one field is meaningful:
|
||||
// - Worker != "" → filter by worker name (Main = "mainagent", Plan = "planner").
|
||||
// Goal Agent 的第 0 轮拆解也以 worker="planner" 落库,故 Plan 会话完整覆盖 Goal+Planner。
|
||||
// - NodeID != nil → a Worker session, filtered by node_id (= intent id).
|
||||
// A zero value (both empty) matches the whole task (no session filter).
|
||||
type ActivitySessionFilter struct {
|
||||
Worker string
|
||||
NodeID *int64
|
||||
}
|
||||
|
||||
func (f ActivitySessionFilter) cond(argStart int) (string, []any) {
|
||||
switch {
|
||||
case f.NodeID != nil:
|
||||
return fmt.Sprintf(" AND node_id=$%d", argStart), []any{*f.NodeID}
|
||||
case f.Worker != "":
|
||||
return fmt.Sprintf(" AND worker=$%d", argStart), []any{f.Worker}
|
||||
default:
|
||||
return "", nil
|
||||
}
|
||||
}
|
||||
|
||||
// ActivityPage returns one page for reverse (newest-first) pagination of a session:
|
||||
// up to `limit` steps ending before id `before` (exclusive; before<=0 = the latest
|
||||
// page), returned in ASCENDING id order for direct display. hasMore reports whether
|
||||
// still-older steps exist before the returned window (so the client can stop loading
|
||||
// on scroll-up). Summary-only columns — detail is lazy-loaded via ActivityDetail.
|
||||
func (s *ExplorationStore) ActivityPage(f ActivitySessionFilter, before int64, limit int) ([]Activity, bool, error) {
|
||||
if limit <= 0 {
|
||||
limit = 200
|
||||
}
|
||||
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens`
|
||||
args := []any{s.expID}
|
||||
cond, cargs := f.cond(len(args) + 1)
|
||||
args = append(args, cargs...)
|
||||
beforeArg := len(args) + 1
|
||||
args = append(args, before)
|
||||
limitArg := len(args) + 1
|
||||
args = append(args, limit+1) // one extra row to detect older history
|
||||
q := fmt.Sprintf(`SELECT `+cols+`
|
||||
FROM activity WHERE exploration_id=$1%s AND ($%d <= 0 OR id < $%d)
|
||||
ORDER BY id DESC LIMIT $%d`, cond, beforeArg, beforeArg, limitArg)
|
||||
rows, err := s.db.Query(q, args...)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
defer rows.Close()
|
||||
desc := []Activity{}
|
||||
for rows.Next() {
|
||||
var a Activity
|
||||
if err := rows.Scan(&a.ID, &a.NodeID, &a.Worker, &a.Kind, &a.Tool, &a.ToolUseID, &a.IsError, &a.Summary, &a.CreatedAt,
|
||||
&a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
desc = append(desc, a)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
hasMore := len(desc) > limit
|
||||
if hasMore {
|
||||
desc = desc[:limit]
|
||||
}
|
||||
// reverse the newest-first window into ascending id order for display.
|
||||
out := make([]Activity, len(desc))
|
||||
for i, a := range desc {
|
||||
out[len(desc)-1-i] = a
|
||||
}
|
||||
return out, hasMore, nil
|
||||
}
|
||||
|
||||
// ActivityMaxID returns the task's current maximum activity id (0 when empty). It is
|
||||
// the task-level snapshot cursor handed to the SSE stream so history (id<=cursor) and
|
||||
// the live tail (id>cursor) meet with no gap — deliberately task-level, not per
|
||||
// session, since one SSE covers the whole task.
|
||||
func (s *ExplorationStore) ActivityMaxID() (int64, error) {
|
||||
var max sql.NullInt64
|
||||
err := s.db.QueryRow(`SELECT MAX(id) FROM activity WHERE exploration_id=$1`, s.expID).Scan(&max)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return max.Int64, nil
|
||||
}
|
||||
|
||||
// ActivityDetail lazily returns the full detail blob for one step.
|
||||
func (s *ExplorationStore) ActivityDetail(id int64) (string, error) {
|
||||
var d sql.NullString
|
||||
|
||||
@@ -187,6 +187,9 @@ CREATE TABLE IF NOT EXISTS activity (
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_act_node ON activity(exploration_id, node_id, id);
|
||||
CREATE INDEX IF NOT EXISTS idx_act_since ON activity(exploration_id, id);
|
||||
-- Main/Plan history pages filter by worker (both carry NULL node_id, so idx_act_node
|
||||
-- can't distinguish them); this covers reverse pagination of those sessions.
|
||||
CREATE INDEX IF NOT EXISTS idx_act_worker ON activity(exploration_id, worker, id);
|
||||
|
||||
-- =====================================================================
|
||||
-- C. LLM profiles
|
||||
|
||||
+125
-9
@@ -521,6 +521,7 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.HandleFunc("GET /api/exploration/intents", s.intents)
|
||||
mux.HandleFunc("GET /api/exploration/graph", s.explorationGraph)
|
||||
mux.HandleFunc("GET /api/exploration/activity", s.activity)
|
||||
mux.HandleFunc("GET /api/exploration/activity/history", s.activityHistory)
|
||||
mux.HandleFunc("GET /api/exploration/activity/stream", s.streamActivity)
|
||||
mux.HandleFunc("GET /api/exploration/activity/{seq}", s.activityDetail)
|
||||
mux.HandleFunc("GET /api/exploration/tokens", s.tokenStats)
|
||||
@@ -1255,11 +1256,34 @@ func (s *Server) findings(w http.ResponseWriter, r *http.Request) {
|
||||
func (s *Server) intents(w http.ResponseWriter, r *http.Request) {
|
||||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||||
if t == nil {
|
||||
// Back-compat: bare list shape when the task can't be resolved.
|
||||
writeJSON(w, 200, []any{})
|
||||
return
|
||||
}
|
||||
in, _ := t.Store.ListByKind(db.KindIntent, 300)
|
||||
writeJSON(w, 200, taskNodeDTOs(in))
|
||||
q := r.URL.Query()
|
||||
limit := min(atoiDefault(q.Get("limit"), 300), 500)
|
||||
// No paging params → preserve the legacy bare-array response so existing callers
|
||||
// (and the poll) keep working unchanged.
|
||||
if q.Get("before") == "" && q.Get("page") == "" {
|
||||
in, err := t.Store.ListByKind(db.KindIntent, limit)
|
||||
if err != nil {
|
||||
log.Printf("[intents] task=%s limit=%d: %v", t.ID, limit, err)
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, 200, taskNodeDTOs(in))
|
||||
return
|
||||
}
|
||||
// Paged form: ?before=<id> (or ?page as a marker) → {items, has_more} so the
|
||||
// worker session list can reach past the old fixed 300 boundary on scroll.
|
||||
before := int64(atoiDefault(q.Get("before"), 0))
|
||||
in, hasMore, err := t.Store.ListByKindPage(db.KindIntent, before, limit)
|
||||
if err != nil {
|
||||
log.Printf("[intents] task=%s before=%d limit=%d: %v", t.ID, before, limit, err)
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"items": taskNodeDTOs(in), "has_more": hasMore})
|
||||
}
|
||||
|
||||
// explorationGraph returns the whole exploration chain (task graph) as nodes+edges.
|
||||
@@ -1291,12 +1315,76 @@ func (s *Server) activity(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
items, cursor, err := t.Store.ActivityList(intentPtr, since, limit)
|
||||
if err != nil {
|
||||
log.Printf("[activity] task=%s since=%d limit=%d intent=%v: %v", t.ID, since, limit, intentPtr, err)
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"items": activityDTOs(items), "cursor": cursor})
|
||||
}
|
||||
|
||||
// parseActivitySession maps a stable session key (main | plan | intent:<ID>) to a
|
||||
// DB session filter. Goal Agent + Planner both live under worker="planner" (the
|
||||
// single Plan session); a Worker session is one intent, keyed by its node id.
|
||||
func parseActivitySession(sess string) (db.ActivitySessionFilter, bool) {
|
||||
switch {
|
||||
case sess == "" || sess == "main":
|
||||
return db.ActivitySessionFilter{Worker: "mainagent"}, true
|
||||
case sess == "plan":
|
||||
return db.ActivitySessionFilter{Worker: "planner"}, true
|
||||
case strings.HasPrefix(sess, "intent:"):
|
||||
id, err := strconv.ParseInt(strings.TrimPrefix(sess, "intent:"), 10, 64)
|
||||
if err != nil {
|
||||
return db.ActivitySessionFilter{}, false
|
||||
}
|
||||
return db.ActivitySessionFilter{NodeID: &id}, true
|
||||
}
|
||||
return db.ActivitySessionFilter{}, false
|
||||
}
|
||||
|
||||
// activityHistory serves one reverse-paginated page of a session's activity history.
|
||||
// The latest page (no ?before) opens a session; ?before=<id> pulls the older page on
|
||||
// scroll-up. snapshot_cursor is the TASK-level max id at query time — the client uses
|
||||
// it to open the single task SSE at since=snapshot_cursor so history (id<=cursor) and
|
||||
// the live tail (id>cursor) meet with no gap and no overlap.
|
||||
func (s *Server) activityHistory(w http.ResponseWriter, r *http.Request) {
|
||||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||||
if t == nil {
|
||||
writeErr(w, 404, "task not found")
|
||||
return
|
||||
}
|
||||
q := r.URL.Query()
|
||||
sess := q.Get("session")
|
||||
filter, ok := parseActivitySession(sess)
|
||||
if !ok {
|
||||
writeErr(w, 400, "bad session")
|
||||
return
|
||||
}
|
||||
before := int64(atoiDefault(q.Get("before"), 0))
|
||||
limit := min(atoiDefault(q.Get("limit"), 200), 500) // cap so one request can't pull an unbounded slice
|
||||
snapshot, err := t.Store.ActivityMaxID()
|
||||
if err != nil {
|
||||
log.Printf("[activity/history] task=%s session=%s snapshot: %v", t.ID, sess, err)
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
}
|
||||
items, hasMore, err := t.Store.ActivityPage(filter, before, limit)
|
||||
if err != nil {
|
||||
log.Printf("[activity/history] task=%s session=%s before=%d limit=%d: %v", t.ID, sess, before, limit, err)
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
}
|
||||
earliest := before
|
||||
if len(items) > 0 {
|
||||
earliest = items[0].ID
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{
|
||||
"items": activityDTOs(items),
|
||||
"snapshot_cursor": snapshot,
|
||||
"earliest_cursor": earliest,
|
||||
"has_more": hasMore,
|
||||
})
|
||||
}
|
||||
|
||||
// tokenStats returns per-worker token usage (input/output/cache read/write) for a
|
||||
// task — main agent, planner, and each work#N.
|
||||
func (s *Server) tokenStats(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -1471,7 +1559,16 @@ func (s *Server) streamActivity(w http.ResponseWriter, r *http.Request) {
|
||||
intentPtr = &n
|
||||
}
|
||||
}
|
||||
// Cursor precedence: the browser's automatic reconnect sends Last-Event-ID (the
|
||||
// last id it received) — trust it over the query so an auto-reconnect resumes
|
||||
// exactly where it dropped. A fresh/manual connect has no header and passes
|
||||
// since=<snapshot_cursor> from the history page instead.
|
||||
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
|
||||
if le := r.Header.Get("Last-Event-ID"); le != "" {
|
||||
if n, err := strconv.ParseInt(le, 10, 64); err == nil {
|
||||
since = n
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
@@ -1483,18 +1580,37 @@ func (s *Server) streamActivity(w http.ResponseWriter, r *http.Request) {
|
||||
ch, unsub := s.engine.Broadcaster().Subscribe(t.ID)
|
||||
defer unsub()
|
||||
|
||||
// Emit a standard SSE id: line so the browser echoes it as Last-Event-ID on
|
||||
// auto-reconnect (see cursor precedence above).
|
||||
sendSSE := func(a db.Activity) {
|
||||
b, _ := json.Marshal(activityDTO(a))
|
||||
fmt.Fprintf(w, "data: %s\n\n", b)
|
||||
fmt.Fprintf(w, "id: %d\ndata: %s\n\n", a.ID, b)
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
items, cursor, _ := t.Store.ActivityList(intentPtr, since, 2000)
|
||||
for _, a := range items {
|
||||
sendSSE(a)
|
||||
}
|
||||
if cursor > since {
|
||||
since = cursor
|
||||
// Compensate the DB backlog after `since` in batches until caught up. This is the
|
||||
// gap between the history snapshot and the live tail — NOT the first-page history
|
||||
// (that's the /activity/history endpoint). A long task can have far more than one
|
||||
// batch, so loop instead of a single fixed read; on a query error log it and close
|
||||
// so the client reconnects and retries from its last id (broadcast is lossy — the
|
||||
// DB is the source of truth). The Broadcaster keeps buffering live events meanwhile;
|
||||
// the id<=since skip below drops any that this replay already covered.
|
||||
const replayBatch = 500
|
||||
for {
|
||||
items, cursor, err := t.Store.ActivityList(intentPtr, since, replayBatch)
|
||||
if err != nil {
|
||||
log.Printf("[activity/stream] task=%s replay since=%d: %v", t.ID, since, err)
|
||||
return
|
||||
}
|
||||
for _, a := range items {
|
||||
sendSSE(a)
|
||||
}
|
||||
if cursor > since {
|
||||
since = cursor
|
||||
}
|
||||
if len(items) < replayBatch {
|
||||
break
|
||||
}
|
||||
}
|
||||
flusher.Flush()
|
||||
|
||||
|
||||
@@ -14,6 +14,8 @@ import {
|
||||
CircleXIcon,
|
||||
CircleSlashIcon,
|
||||
ShieldAlertIcon,
|
||||
WifiOffIcon,
|
||||
RotateCwIcon,
|
||||
} from "lucide-react";
|
||||
import { cn } from "@/lib/utils";
|
||||
import { Transcript } from "@/components/transcript";
|
||||
@@ -38,6 +40,70 @@ import type {
|
||||
TokenTotal,
|
||||
} from "@/lib/types";
|
||||
|
||||
// ── Reliability model (see docs/task-session-history-sse-remediation.md) ──────────
|
||||
// The task's activity is NO LONGER one unbounded `allActivity` array replayed from
|
||||
// SSE since=0. Instead:
|
||||
// • Each UI session (main | plan | intent:<id>) has its own lazily-loaded, reverse-
|
||||
// paginated cache (SessionState below). Opening a session loads only its latest
|
||||
// page; scrolling up pages older history in.
|
||||
// • A single task-level SSE (opened at since=snapshot_cursor from the first history
|
||||
// page) tails ALL agents' new activity; frames are dispatched by session_key.
|
||||
// • History + SSE meet gap-free at snapshot_cursor and are merged by seq (dedup),
|
||||
// so refresh / tab-switch / sleep / reconnect never drop the newest records.
|
||||
|
||||
const PAGE = 200; // history page size
|
||||
const MAX_KEEP = 4000; // per-session in-memory cap; older pages re-fetched on scroll-up
|
||||
const STREAM_WINDOW_MS = 5000; // "live" = activity seen within this window
|
||||
|
||||
// One session's lazily-loaded, reverse-paginated cache. `lastTs`/`unread` are kept
|
||||
// live even for sessions that were never opened, so the list shows liveness + unread
|
||||
// without holding their full history.
|
||||
type SessionState = {
|
||||
items: Activity[];
|
||||
loaded: boolean;
|
||||
loading: boolean;
|
||||
loadingMore: boolean;
|
||||
hasMore: boolean; // older history remains above the loaded window
|
||||
earliestSeq: number; // earliest loaded id — reverse-pagination anchor
|
||||
unread: number;
|
||||
lastTs: string; // most-recent activity time (drives live badge; updated even when unloaded)
|
||||
error?: string;
|
||||
};
|
||||
type SessionStore = Record<string, SessionState>;
|
||||
|
||||
function emptyState(): SessionState {
|
||||
return {
|
||||
items: [],
|
||||
loaded: false,
|
||||
loading: false,
|
||||
loadingMore: false,
|
||||
hasMore: false,
|
||||
earliestSeq: 0,
|
||||
unread: 0,
|
||||
lastTs: "",
|
||||
};
|
||||
}
|
||||
|
||||
// sessionKeyOf routes an activity to its stable session key. worker="planner" covers
|
||||
// BOTH the Goal Agent's round-0 decomposition and the Planner (single Plan session).
|
||||
function sessionKeyOf(a: Activity): string {
|
||||
if (a.worker === "mainagent") return "main";
|
||||
if (a.worker === "planner") return "plan";
|
||||
if (a.intent_id) return `intent:${a.intent_id}`;
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
// mergeBySeq unions two activity lists by seq (dedup) in ascending seq order. Every
|
||||
// data source — latest page, older page, SSE compensation, live tail — goes through
|
||||
// this, so history responses can never clobber live records received meanwhile.
|
||||
function mergeBySeq(current: Activity[], incoming: Activity[]): Activity[] {
|
||||
if (!incoming.length) return current;
|
||||
const bySeq = new Map<number, Activity>();
|
||||
for (const a of current) bySeq.set(a.seq, a);
|
||||
for (const a of incoming) bySeq.set(a.seq, a);
|
||||
return [...bySeq.values()].sort((p, q) => p.seq - q.seq);
|
||||
}
|
||||
|
||||
// statusIcon maps a session status to its icon. Worker terminal states are
|
||||
// distinct & color-coded: 完成(绿勾圈) / 取消停止(琥珀斜杠圈) / 出错(红叉圈) /
|
||||
// 步数耗尽(紫). running=蓝色转圈, pending(待领取)=灰时钟.
|
||||
@@ -83,7 +149,7 @@ const roleMeta = {
|
||||
|
||||
// The main-agent session is the interactive entry point of this tab and has no
|
||||
// dedicated backend "sessions" endpoint — it is a fixed UI affordance whose
|
||||
// transcript is the full live activity stream for the task.
|
||||
// transcript is the main-agent activity stream (worker="mainagent") for the task.
|
||||
const MAIN_ID = "s-main";
|
||||
const MAIN_SESSION: Session = {
|
||||
id: MAIN_ID,
|
||||
@@ -96,8 +162,8 @@ const MAIN_SESSION: Session = {
|
||||
|
||||
// The planner session is, like the main-agent session, a fixed UI affordance with
|
||||
// no dedicated backend "sessions" endpoint — its transcript is every activity step
|
||||
// the planner emits (worker === "planner", which carries no intent_id since the
|
||||
// planner is what generates intents).
|
||||
// the planner emits (worker === "planner", which also carries the Goal Agent's
|
||||
// round-0 decomposition; the planner carries no intent_id since it generates intents).
|
||||
const PLANNER_ID = "s-planner";
|
||||
const PLANNER_SESSION: Session = {
|
||||
id: PLANNER_ID,
|
||||
@@ -108,6 +174,13 @@ const PLANNER_SESSION: Session = {
|
||||
last_activity: "",
|
||||
};
|
||||
|
||||
// keyForSession maps a UI Session → its stable store key (main | plan | intent:<id>).
|
||||
function keyForSession(s: Session): string {
|
||||
if (s.role === "mainagent") return "main";
|
||||
if (s.role === "planner") return "plan";
|
||||
return `intent:${s.intent_id}`;
|
||||
}
|
||||
|
||||
// Map an exploration intent (TaskNode) state → a session status the UI renders.
|
||||
function intentStatus(state: string): SessionStatus {
|
||||
switch (state) {
|
||||
@@ -147,12 +220,14 @@ function SessionItem({
|
||||
active,
|
||||
displayTitle,
|
||||
hasPending,
|
||||
unread,
|
||||
onClick,
|
||||
}: {
|
||||
s: Session;
|
||||
active: boolean;
|
||||
displayTitle: string;
|
||||
hasPending?: boolean;
|
||||
unread?: number;
|
||||
onClick: () => void;
|
||||
}) {
|
||||
const icon =
|
||||
@@ -180,6 +255,11 @@ function SessionItem({
|
||||
{hasPending && (
|
||||
<ShieldAlertIcon className="size-3.5 shrink-0 text-amber-500" />
|
||||
)}
|
||||
{!active && unread ? (
|
||||
<span className="inline-flex min-w-4 items-center justify-center rounded-full bg-blue-500/15 px-1 text-[10px] font-medium tabular-nums text-blue-600 dark:text-blue-400">
|
||||
{unread > 99 ? "99+" : unread}
|
||||
</span>
|
||||
) : null}
|
||||
{s.live && (
|
||||
<span className="inline-flex items-center gap-1 rounded bg-blue-500/15 px-1.5 py-0.5 text-[10px] font-medium text-blue-600 dark:text-blue-400">
|
||||
<span className="size-1 animate-pulse rounded-full bg-blue-500" />
|
||||
@@ -192,21 +272,121 @@ function SessionItem({
|
||||
|
||||
export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
const [activeId, setActiveId] = React.useState(MAIN_ID);
|
||||
// Live activity stream — the primary data for this tab.
|
||||
const [allActivity, setAllActivity] = React.useState<Activity[]>([]);
|
||||
// Worker sessions derived from exploration intents.
|
||||
// Per-session lazily-loaded caches, keyed by session_key (main | plan | intent:<id>).
|
||||
const [store, setStore] = React.useState<SessionStore>({});
|
||||
// Worker sessions derived from exploration intents (paged past the old 300 cap).
|
||||
const [intents, setIntents] = React.useState<TaskNode[]>([]);
|
||||
const [olderIntents, setOlderIntents] = React.useState<TaskNode[]>([]);
|
||||
const [intentsHasMore, setIntentsHasMore] = React.useState(false);
|
||||
const [loadingOlderIntents, setLoadingOlderIntents] = React.useState(false);
|
||||
// Local-only chat messages overlaid onto the main-agent transcript.
|
||||
const [chatExtra, setChatExtra] = React.useState<Activity[]>([]);
|
||||
const [input, setInput] = React.useState("");
|
||||
const [sending, setSending] = React.useState(false);
|
||||
const [stopping, setStopping] = React.useState(false);
|
||||
const [loaded, setLoaded] = React.useState(false);
|
||||
// SSE connection state — surfaced so a dropped realtime link is visible, never
|
||||
// silently shown as "no messages".
|
||||
const [sseLive, setSseLive] = React.useState(false);
|
||||
// Whole-task token total (all agents), polled from the backend aggregate.
|
||||
const [taskTokens, setTaskTokens] = React.useState<TokenTotal | null>(null);
|
||||
// Pending intercept requests for this task — used to show warning icons on sessions.
|
||||
const [pendingIntercepts, setPendingIntercepts] = React.useState<InterceptApprovalRow[]>([]);
|
||||
|
||||
// Refs backing SSE/loading without re-render churn.
|
||||
const snapshotRef = React.useRef(0); // task-level snapshot cursor → SSE since=
|
||||
const esRef = React.useRef<EventSource | null>(null);
|
||||
const activeKeyRef = React.useRef("main"); // current session key (for SSE dispatch/unread)
|
||||
const atBottomRef = React.useRef(true); // transcript pinned to bottom?
|
||||
// Per-key request token: a stale response for a key is ignored (guards fast
|
||||
// latest/older interleaving). Writes are ALWAYS keyed, so a late response can only
|
||||
// touch its own session cache — never the currently-viewed one (see §7.5).
|
||||
const reqTokenRef = React.useRef<Record<string, number>>({});
|
||||
// Keys with a latest-page load in flight — dedups the double trigger where the
|
||||
// first-load effect and the active-session effect both want "main" on mount (the
|
||||
// former also opens the SSE, so it must not be pre-empted).
|
||||
const loadingKeysRef = React.useRef<Set<string>>(new Set());
|
||||
|
||||
// ── store helpers ──────────────────────────────────────────────────────────────
|
||||
const patchStore = React.useCallback(
|
||||
(key: string, fn: (s: SessionState) => SessionState) => {
|
||||
setStore((prev) => ({ ...prev, [key]: fn(prev[key] ?? emptyState()) }));
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
// Load a session's LATEST page (before=0) on first open. Keyed + request-token
|
||||
// guarded so a switch away can't corrupt the view.
|
||||
const loadSession = React.useCallback(
|
||||
(key: string) => {
|
||||
if (MOCK) return; // MOCK preloads everything up front
|
||||
if (loadingKeysRef.current.has(key)) return; // already in flight (e.g. main on mount)
|
||||
loadingKeysRef.current.add(key);
|
||||
const token = (reqTokenRef.current[key] ?? 0) + 1;
|
||||
reqTokenRef.current[key] = token;
|
||||
patchStore(key, (s) => ({ ...s, loading: true, error: undefined }));
|
||||
api
|
||||
.activityHistory(taskId, key, 0, PAGE)
|
||||
.then((r) => {
|
||||
if (reqTokenRef.current[key] !== token) return; // superseded
|
||||
if (r.snapshotCursor > snapshotRef.current) snapshotRef.current = r.snapshotCursor;
|
||||
patchStore(key, (s) => {
|
||||
const items = mergeBySeq(r.items, s.items); // keep any live frames arrived meanwhile
|
||||
return {
|
||||
...s,
|
||||
items,
|
||||
loaded: true,
|
||||
loading: false,
|
||||
hasMore: r.hasMore,
|
||||
earliestSeq: items.length ? items[0].seq : 0,
|
||||
unread: 0,
|
||||
error: undefined,
|
||||
};
|
||||
});
|
||||
})
|
||||
.catch((e) => {
|
||||
if (reqTokenRef.current[key] !== token) return;
|
||||
patchStore(key, (s) => ({ ...s, loading: false, error: (e as Error).message || "加载失败" }));
|
||||
})
|
||||
.finally(() => loadingKeysRef.current.delete(key));
|
||||
},
|
||||
[taskId, patchStore],
|
||||
);
|
||||
|
||||
// Load one older page (scroll-up) for a session, preserving scroll position.
|
||||
const loadEarlier = React.useCallback(
|
||||
(key: string, viewport: () => HTMLElement | null) => {
|
||||
const st = store[key];
|
||||
if (!st || st.loadingMore || !st.hasMore || !st.earliestSeq) return;
|
||||
const vp = viewport();
|
||||
const prevH = vp?.scrollHeight ?? 0;
|
||||
const prevTop = vp?.scrollTop ?? 0;
|
||||
patchStore(key, (s) => ({ ...s, loadingMore: true }));
|
||||
api
|
||||
.activityHistory(taskId, key, st.earliestSeq, PAGE)
|
||||
.then((r) => {
|
||||
patchStore(key, (s) => {
|
||||
const items = mergeBySeq(r.items, s.items);
|
||||
return {
|
||||
...s,
|
||||
items,
|
||||
loadingMore: false,
|
||||
hasMore: r.hasMore,
|
||||
earliestSeq: items.length ? items[0].seq : s.earliestSeq,
|
||||
};
|
||||
});
|
||||
requestAnimationFrame(() => {
|
||||
const v = viewport();
|
||||
if (v) v.scrollTop = prevTop + (v.scrollHeight - prevH);
|
||||
});
|
||||
})
|
||||
.catch(() => {
|
||||
patchStore(key, (s) => ({ ...s, loadingMore: false }));
|
||||
});
|
||||
},
|
||||
[taskId, store, patchStore],
|
||||
);
|
||||
|
||||
// ── task token total (whole task, all agents) ───────────────────────────────────
|
||||
React.useEffect(() => {
|
||||
let alive = true;
|
||||
const load = () =>
|
||||
@@ -241,67 +421,140 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
const id = setInterval(() => setTick((t) => t + 1), 1500);
|
||||
return () => clearInterval(id);
|
||||
}, []);
|
||||
// A session is "streaming" only if its agent emitted activity in the last few
|
||||
// seconds (i.e. SSE actually in flight) — not a hardcoded always-on flag.
|
||||
const STREAM_WINDOW_MS = 5000;
|
||||
const streamingFor = React.useCallback(
|
||||
(w: string) => {
|
||||
let latest = 0;
|
||||
for (const a of allActivity) {
|
||||
if (a.worker === w) {
|
||||
const t = Date.parse(a.ts);
|
||||
if (t > latest) latest = t;
|
||||
}
|
||||
}
|
||||
return latest > 0 && Date.now() - latest < STREAM_WINDOW_MS;
|
||||
},
|
||||
[allActivity],
|
||||
);
|
||||
const mainLive = streamingFor("mainagent");
|
||||
const plannerLive = streamingFor("planner");
|
||||
|
||||
// Live activity stream via SSE: the endpoint replays history after since=0 then
|
||||
// tails each new step. EventSource auto-reconnects; on reconnect history replays
|
||||
// and the seq dedup below keeps it gap-free without duplicates.
|
||||
// ── first load + single task SSE ────────────────────────────────────────────────
|
||||
// On task open: load Main's latest page, take the task-level snapshot cursor from
|
||||
// it, THEN open ONE task SSE at since=snapshot_cursor. History covers id≤cursor and
|
||||
// the SSE covers id>cursor with no gap. The SSE tails ALL agents; frames are routed
|
||||
// by session_key. EventSource auto-reconnects and (via our `id:` lines → Last-Event-
|
||||
// ID) resumes from the DB, so a dropped realtime link self-heals; seq-merge dedups.
|
||||
React.useEffect(() => {
|
||||
setAllActivity([]);
|
||||
setLoaded(false);
|
||||
// Mock demo:无 SSE 后端,一次性拉取活动快照(执行过程)。
|
||||
setStore({});
|
||||
setChatExtra([]);
|
||||
setSseLive(false);
|
||||
setActiveId(MAIN_ID); // a stale worker id from the previous task must not leak in
|
||||
snapshotRef.current = 0;
|
||||
reqTokenRef.current = {};
|
||||
// Reserve "main" so the active-session effect (which also fires for main on mount)
|
||||
// won't double-load it and pre-empt the SSE opened here.
|
||||
loadingKeysRef.current = new Set(["main"]);
|
||||
let alive = true;
|
||||
|
||||
// MOCK demo: no SSE backend — pull one activity snapshot and bucket by session.
|
||||
if (MOCK) {
|
||||
api
|
||||
.activity(taskId)
|
||||
.then((r) => setAllActivity(r.items))
|
||||
.catch(() => {})
|
||||
.finally(() => setLoaded(true));
|
||||
return;
|
||||
.then((r) => {
|
||||
if (!alive) return;
|
||||
const buckets: SessionStore = {};
|
||||
for (const a of r.items) {
|
||||
const k = sessionKeyOf(a);
|
||||
if (!buckets[k]) buckets[k] = emptyState();
|
||||
buckets[k].items.push(a);
|
||||
}
|
||||
for (const k of Object.keys(buckets)) {
|
||||
const st = buckets[k];
|
||||
st.items.sort((p, q) => p.seq - q.seq);
|
||||
st.loaded = true;
|
||||
st.hasMore = false;
|
||||
st.earliestSeq = st.items.length ? st.items[0].seq : 0;
|
||||
st.lastTs = st.items.length ? st.items[st.items.length - 1].ts : "";
|
||||
}
|
||||
buckets.main ??= { ...emptyState(), loaded: true };
|
||||
setStore(buckets);
|
||||
})
|
||||
.catch(() => setStore({ main: { ...emptyState(), loaded: true } }));
|
||||
return () => {
|
||||
alive = false;
|
||||
};
|
||||
}
|
||||
const es = new EventSource(
|
||||
sseUrl(`/api/exploration/activity/stream?task=${encodeURIComponent(taskId)}&since=0`),
|
||||
);
|
||||
es.onopen = () => setLoaded(true);
|
||||
es.onmessage = (e) => {
|
||||
try {
|
||||
const a = JSON.parse(e.data) as Activity;
|
||||
setAllActivity((prev) =>
|
||||
prev.some((x) => x.seq === a.seq)
|
||||
? prev
|
||||
: [...prev, a].sort((p, q) => p.seq - q.seq),
|
||||
|
||||
const token = (reqTokenRef.current.main ?? 0) + 1;
|
||||
reqTokenRef.current.main = token;
|
||||
patchStore("main", (s) => ({ ...s, loading: true }));
|
||||
api
|
||||
.activityHistory(taskId, "main", 0, PAGE)
|
||||
.then((r) => {
|
||||
if (!alive || reqTokenRef.current.main !== token) return;
|
||||
snapshotRef.current = r.snapshotCursor;
|
||||
patchStore("main", (s) => {
|
||||
const items = mergeBySeq(r.items, s.items);
|
||||
return {
|
||||
...s,
|
||||
items,
|
||||
loaded: true,
|
||||
loading: false,
|
||||
hasMore: r.hasMore,
|
||||
earliestSeq: items.length ? items[0].seq : 0,
|
||||
unread: 0,
|
||||
};
|
||||
});
|
||||
// Open the single task SSE from the snapshot cursor.
|
||||
const es = new EventSource(
|
||||
sseUrl(`/api/exploration/activity/stream?task=${encodeURIComponent(taskId)}&since=${snapshotRef.current}`),
|
||||
);
|
||||
} catch {
|
||||
/* ignore malformed frame */
|
||||
}
|
||||
esRef.current = es;
|
||||
es.onopen = () => setSseLive(true);
|
||||
es.onerror = () => setSseLive(false); // EventSource auto-reconnects; DB compensates the gap
|
||||
es.onmessage = (e) => {
|
||||
let a: Activity;
|
||||
try {
|
||||
a = JSON.parse(e.data) as Activity;
|
||||
} catch {
|
||||
return; // ignore malformed frame
|
||||
}
|
||||
const k = sessionKeyOf(a);
|
||||
setStore((prev) => {
|
||||
const cur = prev[k] ?? emptyState();
|
||||
const activeK = activeKeyRef.current;
|
||||
// Merge into sessions that are loaded, actively loading, or the current
|
||||
// view (so returning is instant + a frame that lands mid-load isn't lost).
|
||||
// A cold, inactive session only tracks liveness + unread until it's opened.
|
||||
if (!cur.loaded && !cur.loading && k !== activeK) {
|
||||
return { ...prev, [k]: { ...cur, lastTs: a.ts, unread: cur.unread + 1 } };
|
||||
}
|
||||
let items = mergeBySeq(cur.items, [a]);
|
||||
// Memory bound: trim oldest when over cap (older re-fetched on scroll-up),
|
||||
// but never while the user is reading this session's history (scrolled up).
|
||||
let hasMore = cur.hasMore;
|
||||
let earliestSeq = cur.earliestSeq;
|
||||
const trimmable = k !== activeK || atBottomRef.current;
|
||||
if (trimmable && items.length > MAX_KEEP) {
|
||||
items = items.slice(items.length - MAX_KEEP);
|
||||
hasMore = true;
|
||||
earliestSeq = items[0].seq;
|
||||
}
|
||||
const unread = k === activeK ? 0 : cur.unread + 1;
|
||||
return { ...prev, [k]: { ...cur, items, lastTs: a.ts, unread, hasMore, earliestSeq } };
|
||||
});
|
||||
};
|
||||
})
|
||||
.catch((err) => {
|
||||
if (!alive || reqTokenRef.current.main !== token) return;
|
||||
patchStore("main", (s) => ({ ...s, loading: false, error: (err as Error).message || "加载失败" }));
|
||||
})
|
||||
.finally(() => loadingKeysRef.current.delete("main"));
|
||||
|
||||
return () => {
|
||||
alive = false;
|
||||
esRef.current?.close();
|
||||
esRef.current = null;
|
||||
};
|
||||
return () => es.close();
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [taskId]);
|
||||
|
||||
// Worker session list derives from exploration intents — poll lightly (changes
|
||||
// far less often than the activity stream).
|
||||
// ── worker (intent) session list — paged, poll first page lightly ───────────────
|
||||
React.useEffect(() => {
|
||||
let active = true;
|
||||
const load = () =>
|
||||
api.intents(taskId).then((nodes) => {
|
||||
if (active) setIntents(nodes);
|
||||
}).catch(() => {});
|
||||
api
|
||||
.intentsPage(taskId, 0, 300)
|
||||
.then((r) => {
|
||||
if (!active) return;
|
||||
setIntents(r.items);
|
||||
setIntentsHasMore(r.hasMore);
|
||||
})
|
||||
.catch(() => {});
|
||||
load();
|
||||
const t = setInterval(load, 5000);
|
||||
return () => {
|
||||
@@ -310,18 +563,43 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
};
|
||||
}, [taskId]);
|
||||
|
||||
const loadOlderIntents = React.useCallback(() => {
|
||||
if (loadingOlderIntents) return;
|
||||
const all = [...intents, ...olderIntents];
|
||||
const minId = all.reduce((m, n) => Math.min(m, Number(n.id)), Infinity);
|
||||
if (!Number.isFinite(minId)) return;
|
||||
setLoadingOlderIntents(true);
|
||||
api
|
||||
.intentsPage(taskId, minId, 300)
|
||||
.then((r) => {
|
||||
setOlderIntents((prev) => {
|
||||
const seen = new Set([...intents, ...prev].map((n) => n.id));
|
||||
return [...prev, ...r.items.filter((n) => !seen.has(n.id))];
|
||||
});
|
||||
setIntentsHasMore(r.hasMore);
|
||||
})
|
||||
.catch(() => {})
|
||||
.finally(() => setLoadingOlderIntents(false));
|
||||
}, [taskId, intents, olderIntents, loadingOlderIntents]);
|
||||
|
||||
// Combined, de-duplicated worker list (newest first page + older loaded pages).
|
||||
const allIntents = React.useMemo(() => {
|
||||
const byId = new Map<string, TaskNode>();
|
||||
for (const n of olderIntents) byId.set(n.id, n);
|
||||
for (const n of intents) byId.set(n.id, n); // fresh poll wins over older snapshot
|
||||
return [...byId.values()].sort((a, b) => Number(b.id) - Number(a.id));
|
||||
}, [intents, olderIntents]);
|
||||
|
||||
const workerSessions = React.useMemo(
|
||||
() => intents.map(intentToSession),
|
||||
[intents],
|
||||
() => allIntents.map(intentToSession),
|
||||
[allIntents],
|
||||
);
|
||||
|
||||
// For each worker session (intent), derive the display title from the latest
|
||||
// activity summary (so the list shows what's actually happening, not the raw
|
||||
// intent payload). Fall back to payload when no activity exists yet.
|
||||
// Also store the full TaskNode for the hover-JSON tooltip.
|
||||
// For each worker session (intent), derive the display title from the intent
|
||||
// payload summary. Also store the full TaskNode for the hover-JSON tooltip.
|
||||
const sessionMeta = React.useMemo(() => {
|
||||
const map = new Map<string, { title: string; json: unknown }>();
|
||||
for (const node of intents) {
|
||||
for (const node of allIntents) {
|
||||
let title = `Intent ${node.id}`;
|
||||
let parsedPayload: unknown = node.payload;
|
||||
if (node.payload) {
|
||||
@@ -333,12 +611,11 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
title = node.payload.trim() || title;
|
||||
}
|
||||
}
|
||||
// Hover JSON: full node with payload replaced by parsed object for readability
|
||||
const json = { ...node, payload: parsedPayload };
|
||||
map.set(node.id, { title, json });
|
||||
}
|
||||
return map;
|
||||
}, [intents]);
|
||||
}, [allIntents]);
|
||||
|
||||
// Returns true if any pending intercept belongs to this session.
|
||||
// Worker agent_name format: "work#N · #intentID". Main/planner match by role key.
|
||||
@@ -356,6 +633,19 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
[pendingIntercepts],
|
||||
);
|
||||
|
||||
// Liveness from each session's last-seen activity time (kept fresh by the 1.5s tick).
|
||||
const recentLive = React.useCallback(
|
||||
(key: string) => {
|
||||
const ts = store[key]?.lastTs;
|
||||
if (!ts) return false;
|
||||
const t = Date.parse(ts);
|
||||
return t > 0 && Date.now() - t < STREAM_WINDOW_MS;
|
||||
},
|
||||
[store],
|
||||
);
|
||||
const mainLive = recentLive("main");
|
||||
const plannerLive = recentLive("plan");
|
||||
|
||||
const sessions = React.useMemo(
|
||||
() => [
|
||||
{ ...MAIN_SESSION, live: mainLive },
|
||||
@@ -374,47 +664,49 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
||||
const active = sessions.find((s) => s.id === activeId) ?? MAIN_SESSION;
|
||||
const isMain = active.role === "mainagent";
|
||||
const isPlanner = active.role === "planner";
|
||||
const activeKey = keyForSession(active);
|
||||
const activeState = store[activeKey];
|
||||
|
||||
// Main agent is the human↔orchestrator CONSOLE: it shows only the conversation
|
||||
// (the user's messages + the main agent's own replies/steps), used to steer the
|
||||
// run — NOT the planner/worker activity firehose (those live in their own
|
||||
// sessions). The planner session shows only planner steps; a worker session
|
||||
// shows only the activity for its intent.
|
||||
// Keep the SSE dispatcher's notion of the active session current, and lazily load
|
||||
// + clear unread whenever the active session changes.
|
||||
React.useEffect(() => {
|
||||
activeKeyRef.current = activeKey;
|
||||
const st = store[activeKey];
|
||||
if (!st || (!st.loaded && !st.loading)) {
|
||||
loadSession(activeKey);
|
||||
} else if (st.unread) {
|
||||
patchStore(activeKey, (s) => ({ ...s, unread: 0 }));
|
||||
}
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [activeKey]);
|
||||
|
||||
// Main agent is the human↔orchestrator CONSOLE: only the conversation (user msgs +
|
||||
// the main agent's own replies/steps). Planner session shows planner steps; a
|
||||
// worker session shows only its intent's activity, led by the intent objective.
|
||||
const activity = React.useMemo(() => {
|
||||
const items = activeState?.items ?? [];
|
||||
if (isMain) {
|
||||
// The conversation is persisted server-side in the activity stream
|
||||
// (worker="mainagent"), so it survives reloads. chatExtra is only an
|
||||
// optimistic local echo — drop any entry the persisted (SSE) copy already
|
||||
// covers, so messages don't show twice once the backend round-trips.
|
||||
const mine = allActivity.filter((a) => a.worker === "mainagent");
|
||||
const seen = new Set(mine.map((a) => `${a.kind} | ||||