From 79493d5a4ac4085c610aa6eb8e14fb88c071aff2 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Tue, 11 Aug 2026 09:13:12 -0400 Subject: [PATCH] =?UTF-8?q?feat(activity):=20=E4=BB=BB=E5=8A=A1=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E5=8E=86=E5=8F=B2=E5=8F=8D=E5=90=91=E5=88=86=E9=A1=B5?= =?UTF-8?q?=20+=20SSE=20=E5=8F=AF=E9=9D=A0=E6=80=A7=E6=95=B4=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端: - 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 通过) --- db/activity_page_test.go | 178 ++++++ db/exploration.go | 110 ++++ db/schema.sql | 3 + server/server.go | 134 +++- .../tasks/detail/_tabs/sessions-tab.tsx | 581 ++++++++++++++---- web/src/lib/api.ts | 30 + 6 files changed, 906 insertions(+), 130 deletions(-) create mode 100644 db/activity_page_test.go diff --git a/db/activity_page_test.go b/db/activity_page_test.go new file mode 100644 index 0000000..d98bbf8 --- /dev/null +++ b/db/activity_page_test.go @@ -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) + } +} diff --git a/db/exploration.go b/db/exploration.go index 5b0f79c..8c66cdc 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -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 diff --git a/db/schema.sql b/db/schema.sql index 6944971..d079854 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -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 diff --git a/server/server.go b/server/server.go index d7e68a8..2295a4c 100644 --- a/server/server.go +++ b/server/server.go @@ -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= (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:) 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= 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= 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() diff --git a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx index 5ec6a28..7bd645e 100644 --- a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx +++ b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx @@ -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:) 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; + +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(); + 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:). +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 && ( )} + {!active && unread ? ( + + {unread > 99 ? "99+" : unread} + + ) : null} {s.live && ( @@ -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([]); - // Worker sessions derived from exploration intents. + // Per-session lazily-loaded caches, keyed by session_key (main | plan | intent:). + const [store, setStore] = React.useState({}); + // Worker sessions derived from exploration intents (paged past the old 300 cap). const [intents, setIntents] = React.useState([]); + const [olderIntents, setOlderIntents] = React.useState([]); + 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([]); 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(null); // Pending intercept requests for this task — used to show warning icons on sessions. const [pendingIntercepts, setPendingIntercepts] = React.useState([]); + // Refs backing SSE/loading without re-render churn. + const snapshotRef = React.useRef(0); // task-level snapshot cursor → SSE since= + const esRef = React.useRef(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>({}); + // 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>(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(); + 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(); - 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}${a.summary}`)); - const pending = chatExtra.filter( - (a) => !seen.has(`${a.kind}${a.summary}`), - ); - return [...mine, ...pending].sort((a, b) => a.seq - b.seq); + // chatExtra is an optimistic local echo — drop any entry the persisted (SSE) + // copy already covers so messages don't show twice after the backend round-trips. + const seen = new Set(items.map((a) => `${a.kind} ${a.summary}`)); + const pending = chatExtra.filter((a) => !seen.has(`${a.kind} ${a.summary}`)); + return [...items, ...pending].sort((a, b) => a.seq - b.seq); } - if (isPlanner) { - return allActivity.filter((a) => a.worker === "planner"); - } - // Worker session: the intent is NOT shown as the header title — instead it - // leads the transcript as a right-aligned "user"-style message (the task - // handed to this worker), followed by its execution steps. - const steps = allActivity.filter( - (a) => a.intent_id === active.intent_id || a.worker === active.id, - ); + if (isPlanner) return items; + // Worker session: the intent leads the transcript as a right-aligned "user"-style + // message (the task handed to this worker), followed by its execution steps. const intentTitle = sessionMeta.get(active.id)?.title ?? active.title; const intentMsg: Activity = { seq: -1, // sorts/leads before any real step (real seq ≥ 0) - worker: steps[0]?.worker ?? active.id, // reuse the lane so no worker chips appear + worker: items[0]?.worker ?? active.id, // reuse the lane so no worker chips appear ts: active.last_activity || "", kind: "intent", // LLM-generated objective — rendered as a distinct (non-human) bubble summary: intentTitle, }; - return [intentMsg, ...steps]; - }, [isMain, isPlanner, allActivity, chatExtra, active.intent_id, active.id, active.title, active.last_activity, sessionMeta]); + return [intentMsg, ...items]; + }, [isMain, isPlanner, activeState, chatExtra, active.id, active.title, active.last_activity, sessionMeta]); - // seq of this session's most-recent TodoWrite call — for the Todo popover shown in - // the worker/planner footer (in place of the old read-only replay notice). + // seq of this session's most-recent TodoWrite call — for the Todo popover. const latestTodoSeq = React.useMemo(() => { for (let i = activity.length - 1; i >= 0; i--) { const a = activity[i]; @@ -423,21 +715,12 @@ export function SessionsTab({ taskId }: { taskId: string }) { return null; }, [activity]); - // Per-session token total, live. Each run (one captureRunSession) reports - // cumulative usage per model turn via kind='usage', then a final kind='result'. - // Total = sum of COMPLETED runs' results + the current in-progress run's latest - // usage. A 'result' adds to the sum and clears the live partial; a 'usage' - // overwrites the live partial. So a running work counts up live; a finished one - // shows its final result. (activity is seq-sorted.) + // Per-session token total, live. Sum COMPLETED runs' results + the current run's + // latest usage. Computed over the loaded pages (the running/just-finished run whose + // totals matter sits on the latest page). (activity is seq-sorted.) const tokenTotal = React.useMemo(() => { - let i = 0, - o = 0, - cr = 0, - cw = 0; // completed - let li = 0, - lo = 0, - lcr = 0, - lcw = 0; // live in-progress + let i = 0, o = 0, cr = 0, cw = 0; // completed + let li = 0, lo = 0, lcr = 0, lcw = 0; // live in-progress for (const a of activity) { if (a.kind === "result") { i += a.input_tokens ?? 0; @@ -452,19 +735,14 @@ export function SessionsTab({ taskId }: { taskId: string }) { lcw = a.cache_write_tokens ?? 0; } } - const I = i + li, - O = o + lo, - CR = cr + lcr, - CW = cw + lcw; + const I = i + li, O = o + lo, CR = cr + lcr, CW = cw + lcw; return { i: I, o: O, cr: CR, cw: CW, any: I + O + CR + CW > 0 }; }, [activity]); // Run duration = span from this session's first step to its last (for a live - // session, "now" so it ticks up — the 1.5s setTick above re-renders it). Derived - // purely from activity timestamps (Activity.ts), the only per-step time we have. + // session, "now" so it ticks up — the 1.5s setTick above re-renders it). const runDuration = React.useMemo(() => { - let min = Infinity, - max = 0; + let min = Infinity, max = 0; for (const a of activity) { const t = Date.parse(a.ts); if (!Number.isFinite(t)) continue; @@ -478,7 +756,6 @@ export function SessionsTab({ taskId }: { taskId: string }) { // ---- transcript auto-scroll (open → bottom; stick to bottom unless scrolled up) ---- const contentRef = React.useRef(null); - const atBottomRef = React.useRef(true); const viewport = React.useCallback( () => (contentRef.current?.closest('[data-slot="scroll-area-viewport"]') as HTMLElement | null) ?? @@ -490,10 +767,11 @@ export function SessionsTab({ taskId }: { taskId: string }) { if (!vp) return; const onScroll = () => { atBottomRef.current = vp.scrollTop + vp.clientHeight >= vp.scrollHeight - 60; + if (vp.scrollTop <= 80) loadEarlier(activeKeyRef.current, viewport); // near top → older page }; vp.addEventListener("scroll", onScroll, { passive: true }); return () => vp.removeEventListener("scroll", onScroll); - }, [viewport, activeId]); + }, [viewport, activeId, loadEarlier]); // open/switch a session → jump to the latest (bottom) React.useLayoutEffect(() => { const vp = viewport(); @@ -518,7 +796,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { function send() { const text = input.trim(); if (!text || sending) return; - const base = [...allActivity, ...chatExtra]; + const base = [...(store.main?.items ?? []), ...chatExtra]; const nextSeq = Math.max(0, ...base.map((a) => a.seq)) + 1; const userMsg: Activity = { seq: nextSeq, @@ -549,13 +827,40 @@ export function SessionsTab({ taskId }: { taskId: string }) { .finally(() => setSending(false)); } + const mainLoaded = !!store.main?.loaded; + // What the transcript pane should show for the active session. + const showLoader = !activeState || (activeState.loading && !activeState.loaded); + return (
{/* Left: session list */}
-
会话列表
+
+
会话列表
+ {!MOCK && ( + + {sseLive ? ( + <> + + 实时 + + ) : ( + <> + + 重连中 + + )} + + )} +
{taskTokens && (
setActiveId(s.id)} /> ); })} + {role === "worker" && intentsHasMore && ( + + )}
); })} - {loaded && !workerSessions.length && ( + {mainLoaded && !workerSessions.length && (
暂无运行中的 Worker 会话。
@@ -630,6 +950,11 @@ export function SessionsTab({ taskId }: { taskId: string }) { 实时 )} + {activeState?.hasMore && ( + + ↑ 更早历史 + + )}
{tokenTotal.any && ( @@ -657,11 +982,25 @@ export function SessionsTab({ taskId }: { taskId: string }) { className="min-h-0 min-w-0 flex-1 [&_[data-slot=scroll-area-viewport]>div]:block!" >
- {!loaded ? ( + {activeState?.loadingMore && ( +
+ + 加载更早历史… +
+ )} + {showLoader ? (
加载活动流…
+ ) : activeState?.error ? ( +
+ + 加载失败：{activeState.error} + +
) : activity.length ? ( ) : ( diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index d745726..e8ff447 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -247,6 +247,36 @@ export const api = { return get<{ items: Activity[]; cursor: number }>(`/exploration/activity?${q.toString()}`).then((r) => ({ items: arr(r.items), cursor: r.cursor ?? 0 })); }, activityDetail: (id: number, task?: string) => get<{ detail: string }>(`/exploration/activity/${id}${tq(task)}`), + // Reverse-paginated session history. session = "main" | "plan" | "intent:". + // before=0 → latest page; before= → the older page ending before that id. + // snapshotCursor is the TASK-level max id at query time — open the task SSE at + // since=snapshotCursor so history (id≤cursor) and the live tail (id>cursor) meet + // gap-free. hasMore = still-older steps exist (drives scroll-up loading). + activityHistory: (task: string, session: string, before = 0, limit = 200) => { + const q = new URLSearchParams({ session, limit: String(limit) }); + if (task) q.set("task", task); + if (before > 0) q.set("before", String(before)); + return get<{ items: Activity[]; snapshot_cursor: number; earliest_cursor: number; has_more: boolean }>( + `/exploration/activity/history?${q.toString()}`, + ).then((r) => ({ + items: arr(r.items), + snapshotCursor: r.snapshot_cursor ?? 0, + earliestCursor: r.earliest_cursor ?? 0, + hasMore: !!r.has_more, + })); + }, + // Paged worker (intent) session list — reaches past the legacy fixed 300 cap. + // before=0 → newest page; before= → older page. has_more = older intents exist. + intentsPage: (task: string, before = 0, limit = 300) => { + const q = new URLSearchParams({ limit: String(limit) }); + if (task) q.set("task", task); + q.set("before", String(before > 0 ? before : 0)); + q.set("page", "1"); // marker so the backend returns the paged {items,has_more} shape + return get<{ items: TaskNode[]; has_more: boolean }>(`/exploration/intents?${q.toString()}`).then((r) => ({ + items: arr(r.items), + hasMore: !!r.has_more, + })); + }, // ---- traffic / audit / report / chat ---- audit: (task?: string) => get(`/audit${tq(task)}`),