diff --git a/agent/mainagent.go b/agent/mainagent.go index 3f6cbf2..674f437 100644 --- a/agent/mainagent.go +++ b/agent/mainagent.go @@ -102,7 +102,7 @@ func mainAgentSystem(goal, dataDir, workDir string) string { // non-nil, receives each execution step (thinking / tool_use / tool_result / // text / result) so the main-agent session shows its work — exactly like the // worker/planner sessions — not just the final answer. -func (m *MainAgent) Chat(ctx context.Context, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, goal, message string, emit func(db.Activity), notify, resume func(), notifyGoal func([]string)) (string, error) { +func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.AssetStore, ts *db.ExplorationStore, goal, message string, emit func(db.Activity), notify, resume func(), notifyGoal func([]string)) (string, error) { tsx := NewToolSet(ts, "human") tsx.SetFindingRecorder(m.findingRecorder) if as != nil { @@ -153,9 +153,14 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, as *db.AssetStore, t NonStreaming: m.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete MaxTokens: m.maxTokens(), // 0 = 不发上限,由服务端默认值决定 } - if m.tx != nil { // persist raw human↔AI conversation; one accumulating file per task + if m.tx != nil { // persist raw human↔AI conversation; one accumulating file per segment opts.Transcript = m.tx + // Segment 0 keeps the legacy "exp%d-main" name so existing transcripts still + // load; each new session (seg>=1) gets its own file for a clean context. opts.SessionID = fmt.Sprintf("exp%d-main", ts.ID()) + if mainSeg > 0 { + opts.SessionID = fmt.Sprintf("exp%d-main-s%d", ts.ID(), mainSeg) + } } s := agentcore.NewSession(opts) defer s.Close() diff --git a/db/exploration.go b/db/exploration.go index cd844e4..2c2aa89 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -68,6 +68,9 @@ type Activity struct { CacheWriteTokens *int `json:"cache_write_tokens,omitempty"` SourceTaskID int64 `json:"source_task_id,omitempty"` Inherited bool `json:"inherited,omitempty"` + // MainSeg is the main-agent conversation segment (nil for non-mainagent rows; + // nil/0 == the original session). Lets the UI route a mainagent row to its segment. + MainSeg *int `json:"main_seg,omitempty"` } // TokenUsage is a per-worker token aggregate (TokenStatsByWorker). @@ -982,10 +985,10 @@ func (s *ExplorationStore) AppendActivity(a Activity) (int64, error) { } var id int64 err := s.db.QueryRow(` -INSERT INTO activity(exploration_id, node_id, worker, kind, tool, tool_use_id, is_error, summary, detail, metadata, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens) -VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),NULLIF($5,''),NULLIF($6,''),$7,NULLIF($8,''),NULLIF($9,''),$10,$11,$12,$13,$14) +INSERT INTO activity(exploration_id, node_id, worker, kind, tool, tool_use_id, is_error, summary, detail, metadata, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg) +VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),NULLIF($5,''),NULLIF($6,''),$7,NULLIF($8,''),NULLIF($9,''),$10,$11,$12,$13,$14,$15) RETURNING id`, s.expID, a.NodeID, utf8Clean(a.Worker), utf8Clean(a.Kind), utf8Clean(a.Tool), utf8Clean(a.ToolUseID), a.IsError, utf8Clean(a.Summary), utf8Clean(a.Detail), - metadata, a.InputTokens, a.OutputTokens, a.CacheReadTokens, a.CacheWriteTokens).Scan(&id) + metadata, a.InputTokens, a.OutputTokens, a.CacheReadTokens, a.CacheWriteTokens, a.MainSeg).Scan(&id) return id, err } @@ -1197,7 +1200,7 @@ func (s *ExplorationStore) TokenStatsByWorker() ([]TokenUsage, error) { func (s *ExplorationStore) TokenStatsBySession() ([]SessionTokenUsage, error) { rows, err := s.db.Query(`SELECT CASE - WHEN COALESCE(a.worker,'')='mainagent' THEN 'main' + WHEN COALESCE(a.worker,'')='mainagent' THEN 'main:' || COALESCE(a.main_seg,0)::text WHEN COALESCE(a.worker,'')='planner' THEN 'plan' WHEN n.id IS NOT NULL THEN 'intent:' || n.id::text ELSE '' @@ -1234,7 +1237,7 @@ func (s *ExplorationStore) ActivityList(nodeID *int64, sinceID int64, limit int) } var rows *sql.Rows var err error - const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens` + const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg` if nodeID != nil { rows, err = s.db.Query(`SELECT `+cols+` FROM activity WHERE exploration_id=$1 AND node_id=$2 AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit) @@ -1251,7 +1254,7 @@ FROM activity WHERE exploration_id=$1 AND id>$2 ORDER BY id LIMIT $3`, s.expID, 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.Metadata, &a.CreatedAt, - &a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens); err != nil { + &a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens, &a.MainSeg); err != nil { return nil, sinceID, err } if a.ID > cursor { @@ -1301,20 +1304,33 @@ func (s *ExplorationStore) ActivityListForTerminalIntent(nodeID, sinceID int64, // 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"). +// - Main == true → the main-agent session (worker="mainagent") for one segment +// (MainSeg; nil/0 = the original segment, which also matches legacy NULL rows). +// - Worker != "" → filter by worker name (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). +// A zero value (all empty) matches the whole task (no session filter). type ActivitySessionFilter struct { - Worker string - NodeID *int64 + Worker string + NodeID *int64 + Main bool // main-agent session + MainSeg *int // segment for the main session (nil == current, resolved by caller; 0 == original) } 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.Main: + seg := 0 + if f.MainSeg != nil { + seg = *f.MainSeg + } + if seg == 0 { // original segment also owns legacy rows with NULL main_seg + return " AND worker='mainagent' AND COALESCE(main_seg,0)=0", nil + } + return fmt.Sprintf(" AND worker='mainagent' AND main_seg=$%d", argStart), []any{seg} case f.Worker != "": return fmt.Sprintf(" AND worker=$%d", argStart), []any{f.Worker} default: @@ -1331,7 +1347,7 @@ func (s *ExplorationStore) ActivityPage(f ActivitySessionFilter, before int64, l if limit <= 0 { limit = 200 } - const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens` + const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, main_seg` args := []any{s.expID} cond, cargs := f.cond(len(args) + 1) args = append(args, cargs...) @@ -1351,7 +1367,7 @@ ORDER BY id DESC LIMIT $%d`, cond, beforeArg, beforeArg, limitArg) 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.Metadata, &a.CreatedAt, - &a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens); err != nil { + &a.InputTokens, &a.OutputTokens, &a.CacheReadTokens, &a.CacheWriteTokens, &a.MainSeg); err != nil { return nil, false, err } desc = append(desc, a) @@ -1428,6 +1444,57 @@ func (s *ExplorationStore) ActivityMaxID() (int64, error) { return max.Int64, nil } +// MainSession is one resettable main-agent conversation segment of a task. Segment 0 +// is the original session (implicit, never stored); further segments are created by +// "新建会话" to start the main agent on a clean transcript while the task's graph, +// assets and goal stay shared. +type MainSession struct { + Seq int `json:"seq"` + CreatedAt time.Time `json:"created_at"` +} + +// CurrentMainSeg returns the highest main-session segment (0 when none created yet). +func (s *ExplorationStore) CurrentMainSeg() (int, error) { + var seg int + err := s.db.QueryRow(`SELECT COALESCE(MAX(seq),0) FROM main_sessions WHERE exploration_id=$1`, s.expID).Scan(&seg) + return seg, err +} + +// ListMainSessions returns every main-session segment newest-first, always including +// the implicit original segment 0. Segment 0 carries no stored timestamp. +func (s *ExplorationStore) ListMainSessions() ([]MainSession, error) { + rows, err := s.db.Query(`SELECT seq, created_at FROM main_sessions WHERE exploration_id=$1 ORDER BY seq DESC`, s.expID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []MainSession{} + for rows.Next() { + var m MainSession + if err := rows.Scan(&m.Seq, &m.CreatedAt); err != nil { + return nil, err + } + out = append(out, m) + } + if err := rows.Err(); err != nil { + return nil, err + } + out = append(out, MainSession{Seq: 0}) // implicit original session (oldest) + return out, nil +} + +// NewMainSession creates the next main-session segment (seq = current+1) and returns +// it. It touches only main_sessions — the task's exploration graph/assets/goal are +// untouched, so the new session starts with a clean transcript over the same task. +func (s *ExplorationStore) NewMainSession() (MainSession, error) { + var m MainSession + err := s.db.QueryRow(` +INSERT INTO main_sessions(exploration_id, seq) +VALUES ($1, COALESCE((SELECT MAX(seq) FROM main_sessions WHERE exploration_id=$1),0)+1) +RETURNING seq, created_at`, s.expID).Scan(&m.Seq, &m.CreatedAt) + return m, err +} + // 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 4ff5483..003808e 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -309,11 +309,20 @@ CREATE TABLE IF NOT EXISTS activity ( created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); ALTER TABLE activity ADD COLUMN IF NOT EXISTS metadata JSONB NOT NULL DEFAULT '{}'; +-- main_seg segments the main-agent session into resettable conversations: a new +-- main session bumps the segment so its transcript + activity start clean while the +-- task's graph/assets/goal are untouched. NULL == legacy rows == segment 0 (the +-- original session). Only worker='mainagent' rows carry it. +ALTER TABLE activity ADD COLUMN IF NOT EXISTS main_seg INTEGER; 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); +-- Main-session pages filter by segment on top of worker='mainagent'; this partial +-- index covers reverse pagination within one segment. +CREATE INDEX IF NOT EXISTS idx_act_main_seg ON activity(exploration_id, main_seg, id) + WHERE worker='mainagent'; -- Task-list polls aggregate result usage and find the latest event repeatedly. -- Cover the token columns for index-only aggregation and the timestamp order for -- per-exploration latest-activity lookups. @@ -322,6 +331,18 @@ CREATE INDEX IF NOT EXISTS idx_act_result_usage ON activity(exploration_id) WHERE kind='result'; CREATE INDEX IF NOT EXISTS idx_act_latest ON activity(exploration_id, created_at DESC); +-- main_sessions records the resettable main-agent conversation segments of a task. +-- Segment 0 (the original session) is implicit and never stored; this table holds +-- only the extra segments created by "新建会话" (seq >= 1). The current segment is +-- MAX(seq) or 0. Each segment gets its own transcript file + activity slice; the +-- task's exploration graph/assets/goal are shared and never reset. +CREATE TABLE IF NOT EXISTS main_sessions ( + exploration_id BIGINT NOT NULL REFERENCES explorations(id) ON DELETE CASCADE, + seq INTEGER NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (exploration_id, seq) +); + -- ===================================================================== -- C. LLM profiles -- ===================================================================== diff --git a/server/dto.go b/server/dto.go index d57e344..f9bc1c1 100644 --- a/server/dto.go +++ b/server/dto.go @@ -473,6 +473,7 @@ type ActivityDTO struct { Metadata json.RawMessage `json:"metadata,omitempty"` SourceTaskID string `json:"source_task_id,omitempty"` Inherited bool `json:"inherited,omitempty"` + MainSeg *int `json:"main_seg,omitempty"` // main-agent conversation segment (nil for non-mainagent rows) // token usage (set only on kind='result'); used for per-session token totals. InputTokens *int `json:"input_tokens,omitempty"` OutputTokens *int `json:"output_tokens,omitempty"` @@ -501,6 +502,7 @@ func activityDTO(a db.Activity) ActivityDTO { OutputTokens: a.OutputTokens, CacheReadTokens: a.CacheReadTokens, CacheWriteTokens: a.CacheWriteTokens, + MainSeg: a.MainSeg, } if a.SourceTaskID > 0 { d.SourceTaskID = i64s(a.SourceTaskID) diff --git a/server/server.go b/server/server.go index 6e0f2bf..057d20f 100644 --- a/server/server.go +++ b/server/server.go @@ -829,6 +829,8 @@ func (s *Server) Handler() http.Handler { 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/main-sessions", s.mainSessions) + mux.HandleFunc("POST /api/exploration/main-session/new", s.newMainSession) 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) @@ -2569,7 +2571,14 @@ func (s *Server) activity(w http.ResponseWriter, r *http.Request) { func parseActivitySession(sess string) (db.ActivitySessionFilter, bool) { switch { case sess == "" || sess == "main": - return db.ActivitySessionFilter{Worker: "mainagent"}, true + // bare "main" = the current segment; caller resolves MainSeg via the store. + return db.ActivitySessionFilter{Main: true}, true + case strings.HasPrefix(sess, "main:"): + seg, err := strconv.Atoi(strings.TrimPrefix(sess, "main:")) + if err != nil || seg < 0 { + return db.ActivitySessionFilter{}, false + } + return db.ActivitySessionFilter{Main: true, MainSeg: &seg}, true case sess == "plan": return db.ActivitySessionFilter{Worker: "planner"}, true case strings.HasPrefix(sess, "intent:"): @@ -2632,6 +2641,14 @@ func (s *Server) activityHistory(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, "bad session") return } + if filter.Main && filter.MainSeg == nil { // bare "main" → the current segment + seg, err := t.Store.CurrentMainSeg() + if err != nil { + writeErr(w, 500, err.Error()) + return + } + filter.MainSeg = &seg + } 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 store, sourceTaskID, err := activitySessionStore(t, filter) @@ -3377,6 +3394,47 @@ func (s *Server) testWebSearch(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"ok": true, "count": len(results), "backend": backend}) } +// mainSessions lists the task's main-agent conversation segments (newest-first) and +// the current one. The frontend renders these as switchable sessions under 主 Agent. +func (s *Server) mainSessions(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 + } + list, err := t.Store.ListMainSessions() + if err != nil { + writeErr(w, 500, err.Error()) + return + } + current := 0 + if len(list) > 0 { + current = list[0].Seq // newest-first + } + writeJSON(w, 200, map[string]any{"sessions": list, "current": current}) +} + +// newMainSession starts a fresh main-agent conversation segment. Only the segment +// counter advances — the task's exploration graph, assets and goal are untouched, so +// the main agent continues over the same task with a clean transcript/context. +func (s *Server) newMainSession(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 + } + if s.engine.IsDeleting(t.ID) { + writeErr(w, 409, "任务正在删除，无法新建会话") + return + } + m, err := t.Store.NewMainSession() + if err != nil { + writeErr(w, 500, err.Error()) + return + } + writeJSON(w, 200, map[string]any{"seq": m.Seq, "created_at": rfc3339(m.CreatedAt), "current": m.Seq}) +} + func (s *Server) chat(w http.ResponseWriter, r *http.Request) { t := s.m.ResolveTask(r.URL.Query().Get("task")) if t == nil { @@ -3416,12 +3474,20 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { s.chatCancel[t.ID] = cancel s.chatMu.Unlock() + // The whole turn belongs to the current main-agent segment. Read it once (chatBusy + // serializes turns, so it can't change under us) and stamp every mainagent row with + // it, so this turn's transcript + activity land in the segment the user is viewing. + mainSeg, _ := t.Store.CurrentMainSeg() + segPtr := &mainSeg + // Persist + broadcast the human turn so the 主 Agent 编排会话 survives page // reloads and updates live: the conversation lives in the activity stream as // worker="mainagent" (the per-task activity table, replayed via SSE). With // attachments, the activity's Detail carries {text, attachments} so the transcript // renders attachment cards. - s.engine.emitActivity(t, userActivityWithAttachments("mainagent", req.Message, req.Attachments)) + humanTurn := userActivityWithAttachments("mainagent", req.Message, req.Attachments) + humanTurn.MainSeg = segPtr + s.engine.emitActivity(t, humanTurn) var ma *agent.MainAgent if s.taskRuntimeAvailable(t, "mainagent") { ma = s.agentsForTask(t).main @@ -3439,7 +3505,10 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { // emit every step (thinking/tool_use/tool_result/text/result) so the main // agent session shows its work live, like worker/planner. The final answer // is the captured "result" step — no separate reply emit (would duplicate). - emit := func(rec db.Activity) { s.engine.emitActivity(t, rec) } + emit := func(rec db.Activity) { + rec.MainSeg = segPtr + s.engine.emitActivity(t, rec) + } maTaskID, _ := strconv.ParseInt(t.ID, 10, 64) resume := func() { s.reviveTask(t) } // set_goals 新增目标 → 把任务拉回 running // 把上传附件的【绝对路径】清单拼进发给 agent 的消息,它据此用 Read/Bash 打开文件。 @@ -3447,17 +3516,17 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { taskDir := filepath.Join(s.m.dir, "tasks", t.ID) agentMsg := composeAgentMessage(req.Message, req.Attachments, taskDir) s.engine.BeginLLMCall(t.ID) - _, err := ma.Chat(ctx, maTaskID, s.m.Assets(), t.Store, t.Goal, agentMsg, emit, t.Notify, resume, t.NotifyGoal) + _, err := ma.Chat(ctx, maTaskID, mainSeg, s.m.Assets(), t.Store, t.Goal, agentMsg, emit, t.Notify, resume, t.NotifyGoal) s.engine.EndLLMCall(t.ID) if err != nil && ctx.Err() == nil { - s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", IsError: true, Summary: "（主 Agent 出错：" + err.Error() + "）"}) + s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", IsError: true, Summary: "（主 Agent 出错：" + err.Error() + "）", MainSeg: segPtr}) } }() writeJSON(w, 202, map[string]any{"status": "accepted", "mode": "llm"}) return } reply := s.fallbackChat(t, req.Message) - s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", Summary: reply}) + s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", Summary: reply, MainSeg: segPtr}) s.finishTaskChat(t.ID, cancel) writeJSON(w, 200, map[string]any{"reply": reply, "mode": "rule"}) } 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 3ea1ffb..832ebac 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,7 @@ import { Loader2Icon, PaperclipIcon, PauseIcon, + PlusIcon, RadioIcon, RotateCwIcon, ShieldAlertIcon, @@ -139,7 +140,7 @@ function emptyState(): SessionState { // BOTH the Goal Agent's round-0 decomposition and the Planner (single Plan session). function sessionKeyOf(a: Activity): string { if (a.worker === "system" || a.kind === "llm_switch" || a.kind === "llm_failover") return "system"; - if (a.worker === "mainagent") return "main"; + if (a.worker === "mainagent") return `main:${a.main_seg ?? 0}`; // one key per conversation segment if (a.worker === "planner") return "plan"; if (a.intent_id) return `intent:${a.intent_id}`; return "unknown"; @@ -235,14 +236,23 @@ 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 main-agent activity stream (worker="mainagent") for the task. -const MAIN_ID = "s-main"; +// A main-agent session is one resettable conversation segment. Segment 0 is the +// original session; "新建会话" creates further segments (seq 1,2,…) so the agent +// starts on a clean transcript while the task's graph/assets/goal stay shared. Each +// segment is a switchable UI session; only the current (highest) one is writable. +const mainSessionId = (seg: number) => `s-main-${seg}`; +const mainSessionKey = (seg: number) => `main:${seg}`; +const mainSessionTitle = (seg: number) => `主 Agent · 会话 #${seg + 1}`; +const MAIN_ID = mainSessionId(0); const MAIN_SESSION: Session = { id: MAIN_ID, role: "mainagent", - title: "主 Agent · 编排会话", + title: mainSessionTitle(0), status: "running", live: true, last_activity: "", + seg: 0, + isCurrentMain: true, }; // The planner session is, like the main-agent session, a fixed UI affordance with @@ -272,9 +282,9 @@ const SYSTEM_SESSION: Session = { last_activity: "", }; -// keyForSession maps a UI Session → its stable store key (main | plan | intent:). +// 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 === "mainagent") return `main:${s.seg ?? 0}`; if (s.role === "planner") return "plan"; if (s.role === "system") return "system"; return `intent:${s.intent_id}`; @@ -388,6 +398,16 @@ function SessionItem({ {unread > 99 ? "99+" : unread} ) : null} + {s.role === "mainagent" && s.isCurrentMain && !s.live && ( + + 当前 + + )} + {s.role === "mainagent" && !s.isCurrentMain && ( + + 历史 + + )} {s.live && ( @@ -458,6 +478,11 @@ function WorkerAssetBadge({ assets }: { assets: IntentAsset[] }) { export function SessionsTab({ taskId }: { taskId: string }) { const [activeId, setActiveId] = React.useState(MAIN_ID); + // Main-agent conversation segments (newest-first); currentSeg is the writable one. + const [mainSegs, setMainSegs] = React.useState<{ seq: number; created_at: string }[]>([{ seq: 0, created_at: "" }]); + const [currentSeg, setCurrentSeg] = React.useState(0); + const [creatingMain, setCreatingMain] = React.useState(false); + const [confirmNewMain, setConfirmNewMain] = React.useState(false); // 手机端（ [{ seq: r.seq, created_at: r.created_at }, ...prev.filter((m) => m.seq !== r.seq)]); + setCurrentSeg(r.current ?? r.seq); + // Seed an empty, loaded state so the new (empty) session renders immediately. + setStore((prev) => ({ ...prev, [mainSessionKey(r.seq)]: { ...emptyState(), loaded: true } })); + setActiveId(mainSessionId(r.seq)); + setListOpen(false); + setInput(""); + } catch (e) { + toast.error(`新建会话失败：${(e as Error).message}`); + } finally { + setCreatingMain(false); + setConfirmNewMain(false); + } + } + const patchIntentState = React.useCallback((intentId: string, state?: string) => { const patch = (rows: TaskNode[]) => state @@ -555,7 +603,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { // 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 activeKeyRef = React.useRef(mainSessionKey(0)); // current session key (for SSE dispatch/unread) const atBottomRef = React.useRef(true); // transcript pinned to bottom? const llmToastSeqRef = React.useRef>(new Set()); const chatStatusRequestRef = React.useRef(0); @@ -799,12 +847,14 @@ export function SessionsTab({ taskId }: { taskId: string }) { setLLMResolutions(null); setSseLive(false); setActiveId(MAIN_ID); // a stale worker id from the previous task must not leak in + setMainSegs([{ seq: 0, created_at: "" }]); + setCurrentSeg(0); snapshotRef.current = 0; llmToastSeqRef.current = new Set(); 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"]); + // Reserve the initial main key so the active-session effect (which fires for main on + // mount) won't double-load it and pre-empt the SSE opened here. + loadingKeysRef.current = new Set([mainSessionKey(0)]); let alive = true; // MOCK demo: no SSE backend — pull one activity snapshot and bucket by session. @@ -827,13 +877,13 @@ export function SessionsTab({ taskId }: { taskId: string }) { 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 }; + buckets[mainSessionKey(0)] ??= { ...emptyState(), loaded: true }; buckets.system ??= { ...emptyState(), loaded: true }; setStore(buckets); }) .catch(() => setStore({ - main: { ...emptyState(), loaded: true }, + [mainSessionKey(0)]: { ...emptyState(), loaded: true }, system: { ...emptyState(), loaded: true }, }), ); @@ -842,15 +892,30 @@ export function SessionsTab({ taskId }: { taskId: string }) { }; } - const token = (reqTokenRef.current.main ?? 0) + 1; - reqTokenRef.current.main = token; - patchStore("main", (s) => ({ ...s, loading: true })); + const token = (reqTokenRef.current.mainboot ?? 0) + 1; + reqTokenRef.current.mainboot = token; + let bootKey = mainSessionKey(0); + // Resolve the main-agent segments first, then load the CURRENT segment's history and + // open the SSE from its snapshot cursor. The SSE tails all segments and routes each + // frame by session_key (main:), so switching segments needs no new stream. api - .activityHistory(taskId, "main", 0, PAGE) + .mainSessions(taskId) + .then((ms) => { + if (!alive || reqTokenRef.current.mainboot !== token) throw new Error("superseded"); + const segs = ms.sessions.length ? ms.sessions : [{ seq: 0, created_at: "" }]; + setMainSegs(segs); + setCurrentSeg(ms.current); + bootKey = mainSessionKey(ms.current); + if (bootKey !== mainSessionKey(0)) loadingKeysRef.current.delete(mainSessionKey(0)); + loadingKeysRef.current.add(bootKey); + setActiveId(mainSessionId(ms.current)); + patchStore(bootKey, (s) => ({ ...s, loading: true })); + return api.activityHistory(taskId, bootKey, 0, PAGE); + }) .then((r) => { - if (!alive || reqTokenRef.current.main !== token) return; + if (!alive || reqTokenRef.current.mainboot !== token) return; snapshotRef.current = r.snapshotCursor; - patchStore("main", (s) => { + patchStore(bootKey, (s) => { const items = mergeBySeq(r.items, s.items); return { ...s, @@ -928,10 +993,11 @@ export function SessionsTab({ taskId }: { taskId: string }) { }; }) .catch((err) => { - if (!alive || reqTokenRef.current.main !== token) return; - patchStore("main", (s) => ({ ...s, loading: false, error: (err as Error).message || "加载失败" })); + if (!alive || reqTokenRef.current.mainboot !== token) return; + if ((err as Error).message === "superseded") return; + patchStore(bootKey, (s) => ({ ...s, loading: false, error: (err as Error).message || "加载失败" })); }) - .finally(() => loadingKeysRef.current.delete("main")); + .finally(() => loadingKeysRef.current.delete(bootKey)); return () => { alive = false; @@ -1097,17 +1163,30 @@ export function SessionsTab({ taskId }: { taskId: string }) { }, [store], ); - const mainLive = sending || (mainChatRunning ?? recentLive("main")); + const currentMainKey = mainSessionKey(currentSeg); + const mainLive = sending || (mainChatRunning ?? recentLive(currentMainKey)); const plannerLive = recentLive("plan"); + // One switchable session per main-agent segment (newest-first). Only the current + // (highest) segment is live/writable; older ones are read-only history. + const mainSessions = React.useMemo( + () => + mainSegs.map((m) => ({ + id: mainSessionId(m.seq), + role: "mainagent", + title: mainSessionTitle(m.seq), + status: "running", + live: m.seq === currentSeg && mainLive, + last_activity: m.created_at, + seg: m.seq, + isCurrentMain: m.seq === currentSeg, + })), + [mainSegs, currentSeg, mainLive], + ); + const sessions = React.useMemo( - () => [ - { ...MAIN_SESSION, live: mainLive }, - { ...PLANNER_SESSION, live: plannerLive }, - ...workerSessions, - SYSTEM_SESSION, - ], - [workerSessions, mainLive, plannerLive], + () => [...mainSessions, { ...PLANNER_SESSION, live: plannerLive }, ...workerSessions, SYSTEM_SESSION], + [mainSessions, workerSessions, plannerLive], ); const grouped = { @@ -1119,6 +1198,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { const active = sessions.find((s) => s.id === activeId) ?? MAIN_SESSION; const isMain = active.role === "mainagent"; + const isMainCurrent = isMain && (active.seg ?? 0) === currentSeg; // only the current segment is writable const isPlanner = active.role === "planner"; const isSystem = active.role === "system"; const activeKey = keyForSession(active); @@ -1360,7 +1440,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { .finally(() => setWorkerMessageSending(false)); } - const mainLoaded = !!store.main?.loaded; + const mainLoaded = !!store[currentMainKey]?.loaded; // 发送键位由系统设置决定（localStorage），默认 Enter 发送。 const sendMode = useChatSendMode(); // What the transcript pane should show for the active session. @@ -1476,6 +1556,23 @@ export function SessionsTab({ taskId }: { taskId: string }) {
{Meta.label} + {role === "mainagent" && ( + + )}
{items.map((s) => { const meta = s.role === "worker" ? sessionMeta.get(s.id) : undefined; @@ -1673,12 +1770,30 @@ export function SessionsTab({ taskId }: { taskId: string }) { ) : (
- {isMain ? "还没有对话。在下方给主 Agent 发消息，引导探索方向或介入流程。" : "暂无活动记录。"} + {isMainCurrent ? "还没有对话。在下方给主 Agent 发消息，引导探索方向或介入流程。" : "暂无活动记录。"}
)} - {isMain ? ( + {isMain && !isMainCurrent ? ( +
+ 这是历史会话（只读）。切回当前会话或新建会话以继续对话。 +
+ + +
+
+ ) : isMain ? (
{attachments.length > 0 && (
@@ -1897,6 +2012,23 @@ export function SessionsTab({ taskId }: { taskId: string }) { + + !open && setConfirmNewMain(false)}> + + + 开启新会话？ + + 当前会话会被归档（可随时切回），主 Agent 将以干净的上下文继续。任务的图谱、资产、目标不受影响。 + + + + 取消 + void createMainSession()}> + {creatingMain ? "开启中…" : "开启新会话"} + + + +
); diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index ef32d8d..0aa8e72 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -678,6 +678,16 @@ export const api = { })); }, + // Main-agent conversation segments of a task. Each segment is a resettable session + // (clean transcript/context) over the same task; `current` is the writable one. + mainSessions: (task: string) => + get<{ sessions: { seq: number; created_at: string }[]; current: number }>( + `/exploration/main-sessions${tq(task)}`, + ).then((r) => ({ sessions: arr(r.sessions), current: r.current ?? 0 })), + // Start a fresh main-agent session segment (does not touch the task's graph/assets/goal). + newMainSession: (task: string) => + post<{ seq: number; created_at: string; current: number }>(`/exploration/main-session/new${tq(task)}`), + // ---- traffic / audit / report / chat ---- audit: (task?: string) => get(`/audit${tq(task)}`), traffic: (page = 0, size = 100, host = "", method = "", q = "") => diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 13ca429..2df07e0 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -627,6 +627,7 @@ export interface Activity { }; source_task_id?: string; inherited?: boolean; + main_seg?: number; // main-agent conversation segment (present only on worker="mainagent" rows) // token usage (present only on kind='result') input_tokens?: number; output_tokens?: number; @@ -829,6 +830,8 @@ export interface Session { intent_id?: string; source_task_id?: string; inherited?: boolean; + seg?: number; // main-agent session: which conversation segment (0 = original) + isCurrentMain?: boolean; // main-agent session: the writable current segment } // ---- Security ----