diff --git a/agent/mainagent.go b/agent/mainagent.go index ae20a4a..a2d5336 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) + } } ctx = attachSideCapture(ctx, &opts) s := agentcore.NewSession(opts) diff --git a/db/exploration.go b/db/exploration.go index 479c8d1..51a22fa 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). @@ -987,10 +990,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 } @@ -1202,7 +1205,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 '' @@ -1239,7 +1242,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) @@ -1256,7 +1259,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 { @@ -1306,20 +1309,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: @@ -1336,7 +1352,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...) @@ -1356,7 +1372,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) @@ -1433,6 +1449,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/exploration_tokens_test.go b/db/exploration_tokens_test.go index a0bbf16..9dbbc00 100644 --- a/db/exploration_tokens_test.go +++ b/db/exploration_tokens_test.go @@ -88,7 +88,7 @@ func TestTokenStatsBySessionUsesCompleteIntentHistory(t *testing.T) { if len(got) != 3 { t.Fatalf("sessions = %+v, want only main, plan, and the local intent", sessions) } - assertSessionTokens(t, got["main"], "main", 11, 12, 13, 14) + assertSessionTokens(t, got["main:0"], "main:0", 11, 12, 13, 14) assertSessionTokens(t, got["plan"], "plan", 21, 22, 23, 24) assertSessionTokens(t, got[fmt.Sprintf("intent:%d", intentID)], fmt.Sprintf("intent:%d", intentID), completedRuns, completedRuns*2, completedRuns*3, completedRuns*4) diff --git a/db/schema.sql b/db/schema.sql index 72c8566..e74a1e6 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/orchestration.go b/server/orchestration.go index ee5aa95..37205ca 100644 --- a/server/orchestration.go +++ b/server/orchestration.go @@ -244,6 +244,7 @@ func (s *Server) toolSpawnTask() actool.CoreTool { "description": strParam("任务描述(简短标题)"), "goal": strParam("任务目标(要达成什么)"), "parent_ref": strParam("可选：父任务 id(做父子关联)"), + "source_task_ids": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": fmt.Sprintf("可选：只读继承的来源任务 id 列表(最多 %d 个)。子任务可只读引用这些任务已探明的资产/结论作为起点；与 parent_ref 的纯父子指针不同，这是内容继承。", db.MaxTaskSourceCount)}, "llm_profile_id": map[string]any{"type": "integer", "description": "可选：指定本子任务 planner/worker 用的 LLM 配置 id(见 list_llm_profiles)；留空则继承父任务、再回退全局激活配置"}, "timeout_seconds": map[string]any{"type": "integer", "description": "可选：任务级超时(秒)。到点后触发优雅收尾并进入 timeout 终态；留空或 0 = 不限时"}, "plan_heartbeat_seconds": map[string]any{"type": "integer", "description": "可选：planner 心跳触发间隔(秒)。距上轮规划结束/任务开始满该值且期间无触发 → 触发一轮规划(兜底死锁 + 唤醒去监督飞行中的 worker)。留空或 0 = 默认 600(10min)；"}, @@ -251,11 +252,14 @@ func (s *Server) toolSpawnTask() actool.CoreTool { }, "description", "goal"), func(_ context.Context, in json.RawMessage) (actool.Result, error) { var a struct { - Description, Goal, ParentRef string - LLMProfileID json.RawMessage `json:"llm_profile_id"` - TimeoutSeconds int `json:"timeout_seconds"` - PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` - SeedFirstIntent bool `json:"seed_first_intent"` + Description string `json:"description"` + Goal string `json:"goal"` + ParentRef string `json:"parent_ref"` + SourceTaskIDs []string `json:"source_task_ids"` + LLMProfileID json.RawMessage `json:"llm_profile_id"` + TimeoutSeconds int `json:"timeout_seconds"` + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` + SeedFirstIntent bool `json:"seed_first_intent"` } _ = json.Unmarshal(in, &a) if strings.TrimSpace(a.Description) == "" { @@ -267,6 +271,23 @@ func (s *Server) toolSpawnTask() actool.CoreTool { if a.TimeoutSeconds < 0 { a.TimeoutSeconds = 0 } + // 只读继承来源任务：数量上限 + 每个 id 有效/去重/存在，校验规则与 HTTP 建任务一致。 + if len(a.SourceTaskIDs) > db.MaxTaskSourceCount { + return actool.Errorf(fmt.Sprintf("关联任务最多选择 %d 个", db.MaxTaskSourceCount)), nil + } + sourceIDs := make([]int64, 0, len(a.SourceTaskIDs)) + seenSources := map[int64]bool{} + for _, raw := range a.SourceTaskIDs { + id, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64) + if err != nil || id <= 0 || seenSources[id] { + return actool.Errorf("关联任务 id 无效或重复"), nil + } + if _, ok := s.m.Task(strconv.FormatInt(id, 10)); !ok { + return actool.Errorf(fmt.Sprintf("关联任务 #%d 不存在", id)), nil + } + seenSources[id] = true + sourceIDs = append(sourceIDs, id) + } // LLM profile resolution: explicit id > inherit parent's pin > active(nil). var pin *int64 if id := parseProfileID(a.LLMProfileID); id > 0 { @@ -279,7 +300,16 @@ func (s *Server) toolSpawnTask() actool.CoreTool { pin = pt.LLMProfileID } } - t, err := s.m.CreateTask(a.Description, a.Goal, pin, a.TimeoutSeconds, a.PlanHeartbeatSeconds) + var llmIDs []int64 + if pin != nil { + llmIDs = []int64{*pin} + } + t, err := s.m.CreateTaskWithOptions(a.Description, a.Goal, db.TaskCreateOptions{ + SourceTaskIDs: sourceIDs, + LLMProfileIDs: llmIDs, + TimeoutSeconds: a.TimeoutSeconds, + PlanHeartbeatSeconds: a.PlanHeartbeatSeconds, + }) if err != nil { return actool.Errorf(err.Error()), nil } diff --git a/server/server.go b/server/server.go index 9ebea8d..ba49fa8 100644 --- a/server/server.go +++ b/server/server.go @@ -845,6 +845,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) @@ -2585,7 +2587,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:"): @@ -2648,6 +2657,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) @@ -3393,6 +3410,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 { @@ -3408,6 +3466,7 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { var req struct { Message string `json:"message"` Attachments []chatAttachment `json:"attachments,omitempty"` // 方式1 上传的文件(路径相对任务工作目录) + Seg *int `json:"seg,omitempty"` // 目标主会话分段;缺省=最新段 } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeErr(w, 400, err.Error()) @@ -3432,12 +3491,26 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { s.chatCancel[t.ID] = cancel s.chatMu.Unlock() + // The turn belongs to whichever main-agent segment the user is chatting in (any + // segment is interactive, like the top-level chat conversations). Stamp every + // mainagent row with it so this turn's transcript + activity land in that segment. + // Missing seg (older clients) falls back to the newest segment. + mainSeg := 0 + if req.Seg != nil && *req.Seg >= 0 { + mainSeg = *req.Seg + } else { + 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 @@ -3455,7 +3528,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 打开文件。 @@ -3463,17 +3539,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 eddf142..eb3148b 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, @@ -142,7 +143,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"; @@ -238,14 +239,22 @@ 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, }; // The planner session is, like the main-agent session, a fixed UI affordance with @@ -275,9 +284,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}`; @@ -461,6 +470,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 @@ -558,7 +595,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); @@ -802,12 +839,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. @@ -830,13 +869,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 }, }), ); @@ -845,15 +884,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, @@ -931,10 +985,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; @@ -1100,17 +1155,52 @@ export function SessionsTab({ taskId }: { taskId: string }) { }, [store], ); - const mainLive = sending || (mainChatRunning ?? recentLive("main")); + const currentMainKey = mainSessionKey(currentSeg); const plannerLive = recentLive("plan"); + // Which main segment is streaming RIGHT NOW. A main turn is serialized per task, so at + // most one segment is live. Gate on the real running flag (sending / mainChatRunning) + // rather than "recent activity", and drop it the moment the turn's terminal record + // (kind='result', or an error) lands — otherwise the badge lingers for STREAM_WINDOW_MS + // after the agent already finished. null = nothing running. + const liveMainSeg = React.useMemo(() => { + if (!(sending || mainChatRunning)) return null; + // the streaming segment is the one with the freshest activity (incl. the just-sent turn) + let seg = currentSeg; + let bestTs = -1; + for (const m of mainSegs) { + const raw = store[mainSessionKey(m.seq)]?.lastTs; + const ts = raw ? Date.parse(raw) : -1; + if (ts > bestTs) { + bestTs = ts; + seg = m.seq; + } + } + const items = store[mainSessionKey(seg)]?.items ?? []; + const last = items[items.length - 1]; + if (last && (last.kind === "result" || (last.kind === "text" && last.is_error))) return null; + return seg; + }, [sending, mainChatRunning, mainSegs, store, currentSeg]); + + // Every main-agent segment is an independent, interactive session (like the top-level + // chat conversations) — you can talk in any of them, newest-first. + const mainSessions = React.useMemo( + () => + mainSegs.map((m) => ({ + id: mainSessionId(m.seq), + role: "mainagent", + title: mainSessionTitle(m.seq), + status: "running", + live: m.seq === liveMainSeg, + last_activity: m.created_at, + seg: m.seq, + })), + [mainSegs, liveMainSeg], + ); + 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 = { @@ -1133,6 +1223,15 @@ export function SessionsTab({ taskId }: { taskId: string }) { const isSystem = active.role === "system"; const activeKey = keyForSession(active); const activeState = store[activeKey]; + // A main turn is serialized per task (chat lock), so "busy" is task-wide: while any + // main segment is mid-turn the active composer is disabled. Clear it the moment the + // active session's terminal record (kind='result', or an error) lands, so the input + // re-enables immediately instead of waiting for the next chat-status poll. + const activeItems = activeState?.items ?? []; + const activeLast = activeItems[activeItems.length - 1]; + const activeSettled = + !!activeLast && (activeLast.kind === "result" || (activeLast.kind === "text" && activeLast.is_error)); + const mainBusy = isMain && (sending || (!activeSettled && (mainChatRunning ?? recentLive(activeKey)))); // 折叠态（手机端）标题栏要替代整张列表：显示当前会话名 + 其它会话的未读合计， // 否则收起后既不知道自己在看哪个会话，也看不到别处有新消息。 const activeDisplayTitle = (active.role === "worker" ? sessionMeta.get(active.id)?.title : "") || active.title; @@ -1321,7 +1420,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { const text = input.trim(); const atts = attachments; if (side.handleCommand(text, () => setInput(""))) return; - if ((!text && atts.length === 0) || sending || mainLive) return; + if ((!text && atts.length === 0) || sending || mainBusy) return; // No optimistic echo: the backend persists+broadcasts the human turn before it // returns, so it streams back over SSE (worker="mainagent") with its real DB // seq — the transcript renders it from server data like every other step. Clear @@ -1331,7 +1430,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { setSending(true); chatStatusRequestRef.current++; api - .chat(text, taskId, atts.length > 0 ? atts : undefined) + .chat(text, taskId, atts.length > 0 ? atts : undefined, active.seg ?? 0) .then(({ mode }) => { chatStatusRequestRef.current++; setMainChatRunning(mode === "llm"); @@ -1372,7 +1471,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. @@ -1488,6 +1587,23 @@ export function SessionsTab({ taskId }: { taskId: string }) {
{Meta.label} + {role === "mainagent" && ( + + )}
{items.map((s) => { const meta = s.role === "worker" ? sessionMeta.get(s.id) : undefined; @@ -1736,7 +1852,7 @@ export function SessionsTab({ taskId }: { taskId: string }) { setInput(e.target.value)} onKeyDown={(e) => { @@ -1751,18 +1867,18 @@ export function SessionsTab({ taskId }: { taskId: string }) { size="icon-xs" variant="ghost" onClick={() => fileInputRef.current?.click()} - disabled={mainLive || uploading} + disabled={mainBusy || uploading} title="上传文件" aria-label="上传文件" > {uploading ? : } - {mainLive && isBtwCommand(input) && ( + {mainBusy && isBtwCommand(input) && ( )} - {mainLive ? ( + {mainBusy ? ( + + !open && setConfirmNewMain(false)}> + + + 开启新会话？ + + 当前会话会被归档（可随时切回），主 Agent 将以干净的上下文继续。任务的图谱、资产、目标不受影响。 + + + + 取消 + void createMainSession()}> + {creatingMain ? "开启中…" : "开启新会话"} + + + + ); diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index bc534a4..7b47056 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 = "") => @@ -711,8 +721,8 @@ export const api = { if (!r.ok) throw new Error(`report: ${r.status}`); return r.text(); }, - chat: (message: string, task?: string, attachments?: ChatAttachment[]) => - post<{ reply: string; mode: string }>(`/chat${tq(task)}`, { message, attachments }), + chat: (message: string, task?: string, attachments?: ChatAttachment[], seg?: number) => + post<{ reply: string; mode: string }>(`/chat${tq(task)}`, { message, attachments, seg }), chatStatus: (taskId: string) => get<{ running: boolean }>(`/tasks/${taskId}/chat/status`), // 方式1 文件上传:落到会话/任务工作目录 uploads/，返回可供 agent Read 的相对路径。 chatUpload: async (scope: "task" | "session" | "staging", id: string, files: File[]) => { diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 13ca429..b5bab8a 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,7 @@ export interface Session { intent_id?: string; source_task_id?: string; inherited?: boolean; + seg?: number; // main-agent session: which conversation segment (0 = original) } // ---- Security ----