package controlplane import ( "errors" "fmt" "strings" "time" ) // Multi-turn conversations keep a single Agent's follow-up work in context, so a // user can say "把它复制到某个目录" without repeating where the previous step put it. const ( defaultSessionTurnLimit = 50 maxSessionTurnLimit = 200 // One turn is bounded, and the assembled history has its own budget: the // controller always sends plain user/assistant text, never tool transcripts. maxSessionTurnChars = 4000 maxSessionHistory = 60000 ) // createTaskSession starts an empty conversation with the default turn limit. func (c *ControlPlane) createTaskSession(agentID string) (string, error) { sessionID, err := randomHex(16) if err != nil { return "", err } now := time.Now().UTC().Format(time.RFC3339Nano) if _, err := c.db.Exec( "INSERT INTO task_sessions(session_id, agent_id, turn_limit, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", sessionID, agentID, defaultSessionTurnLimit, now, now, ); err != nil { return "", fmt.Errorf("无法创建会话: %w", err) } return sessionID, nil } func normaliseTurnLimit(turnLimit int) int { if turnLimit < 1 { return defaultSessionTurnLimit } if turnLimit > maxSessionTurnLimit { return maxSessionTurnLimit } return turnLimit } func (c *ControlPlane) sessionTurnCount(sessionID string) (int, error) { var turns int if err := c.db.QueryRow("SELECT COUNT(*) FROM task_session_messages WHERE session_id = ? AND role = 'user'", sessionID).Scan(&turns); err != nil { return 0, fmt.Errorf("无法统计会话轮数: %w", err) } return turns, nil } // ContinueTaskSession dispatches a follow-up turn based on one completed task. // The one-shot task that starts a thread becomes its first turn, so 单独指派 keeps // sending context-free tasks while only an explicit follow-up carries context. func (c *ControlPlane) ContinueTaskSession(taskID, instruction string, timeoutSeconds int, inputs []TaskImageInput) (CreateTaskSessionResult, error) { images, err := normalizeTaskImages(inputs) if err != nil { return CreateTaskSessionResult{}, err } var agentID, sessionID, previousInstruction, previousOutput, status string if err := c.db.QueryRow( "SELECT agent_id, COALESCE(session_id, ''), instruction, COALESCE(output, ''), status FROM task_runs WHERE task_id = ?", taskID, ).Scan(&agentID, &sessionID, &previousInstruction, &previousOutput, &status); err != nil { return CreateTaskSessionResult{}, errors.New("未找到要继续的任务") } if status != "completed" { return CreateTaskSessionResult{}, errors.New("只有已完成的任务才能继续对话") } if sessionID == "" { created, createErr := c.createTaskSession(agentID) if createErr != nil { return CreateTaskSessionResult{}, createErr } sessionID = created c.appendSessionTurn(sessionID, previousInstruction, previousOutput) _, _ = c.db.Exec("UPDATE task_runs SET session_id = ? WHERE task_id = ?", sessionID, taskID) c.recordAudit("task.session_started", "controller", "agent:"+agentID, "completed", "已基于一条已完成任务开始多轮对话。") } followUpID, err := c.createTaskWithImages(agentID, instruction, timeoutSeconds, 1, "", images, sessionID) if err != nil { return CreateTaskSessionResult{}, err } turns, countErr := c.sessionTurnCount(sessionID) if countErr != nil { return CreateTaskSessionResult{TaskID: followUpID, SessionID: sessionID, Turns: 1}, nil } return CreateTaskSessionResult{TaskID: followUpID, SessionID: sessionID, Turns: turns + 1}, nil } // TaskSessionTurn is one stored turn of a conversation: exactly the plain // question/answer text the Agent receives as context. type TaskSessionTurn struct { Instruction string `json:"instruction"` Answer string `json:"answer"` CreatedAt string `json:"createdAt"` } // TaskSessionTurns returns the stored turns of one conversation, oldest first, so // the task detail dialog can show what the follow-up request will carry. func (c *ControlPlane) TaskSessionTurns(sessionID string) ([]TaskSessionTurn, error) { turns := make([]TaskSessionTurn, 0) sessionID = strings.TrimSpace(sessionID) if sessionID == "" { return turns, nil } rows, err := c.db.Query("SELECT role, content, created_at FROM task_session_messages WHERE session_id = ? ORDER BY position", sessionID) if err != nil { return nil, fmt.Errorf("无法读取会话内容: %w", err) } defer rows.Close() for rows.Next() { var role, content, createdAt string if err := rows.Scan(&role, &content, &createdAt); err != nil { return nil, err } switch { case role == "user": turns = append(turns, TaskSessionTurn{Instruction: content, CreatedAt: createdAt}) case len(turns) > 0 && turns[len(turns)-1].Answer == "": turns[len(turns)-1].Answer = content } } return turns, rows.Err() } // sessionHistory returns the newest turns that fit the budget, oldest first, so a // follow-up request carries the context the model needs without unbounded growth. func (c *ControlPlane) sessionHistory(sessionID string, turnLimit int) ([]map[string]string, error) { rows, err := c.db.Query( "SELECT role, content FROM task_session_messages WHERE session_id = ? ORDER BY position DESC LIMIT ?", sessionID, normaliseTurnLimit(turnLimit)*2, ) if err != nil { return nil, fmt.Errorf("无法读取会话历史: %w", err) } defer rows.Close() newestFirst := make([]map[string]string, 0) for rows.Next() { var role, content string if err := rows.Scan(&role, &content); err != nil { return nil, err } if strings.TrimSpace(content) == "" { continue } newestFirst = append(newestFirst, map[string]string{"role": role, "content": content}) } if err := rows.Err(); err != nil { return nil, err } // Keep the newest messages that fit the character budget. budget, kept := maxSessionHistory, make([]map[string]string, 0, len(newestFirst)) for _, message := range newestFirst { budget -= len([]rune(message["content"])) if budget < 0 && len(kept) > 0 { break } kept = append(kept, message) } history := make([]map[string]string, 0, len(kept)) for index := len(kept) - 1; index >= 0; index-- { history = append(history, kept[index]) } return history, nil } // appendSessionTurn records one completed turn and trims the oldest turns once the // conversation exceeds its limit. Failed or cancelled turns never enter history. func (c *ControlPlane) appendSessionTurn(sessionID, instruction, output string) { instruction, output = strings.TrimSpace(instruction), strings.TrimSpace(output) if instruction == "" || output == "" { return } var limitValue, messageCount int if err := c.db.QueryRow("SELECT turn_limit FROM task_sessions WHERE session_id = ?", sessionID).Scan(&limitValue); err != nil { return } if err := c.db.QueryRow("SELECT COUNT(*) FROM task_session_messages WHERE session_id = ?", sessionID).Scan(&messageCount); err != nil { return } now := time.Now().UTC().Format(time.RFC3339Nano) if _, err := c.db.Exec( "INSERT INTO task_session_messages(session_id, position, role, content, created_at) VALUES (?, ?, 'user', ?, ?), (?, ?, 'assistant', ?, ?)", sessionID, messageCount+1, limit(instruction, maxSessionTurnChars), now, sessionID, messageCount+2, limit(output, maxSessionTurnChars), now, ); err != nil { return } // Drop the oldest turns beyond the configured limit. if excess := messageCount + 2 - normaliseTurnLimit(limitValue)*2; excess > 0 { _, _ = c.db.Exec( "DELETE FROM task_session_messages WHERE session_id = ? AND position <= ?", sessionID, excess, ) } _, _ = c.db.Exec("UPDATE task_sessions SET updated_at = ? WHERE session_id = ?", now, sessionID) }