feat(ai): session persist + continue HTTP/UI, checkpoints, queue

落地 docs/16 P1:会话落盘与真续跑(prior tool 轨迹)、incomplete「加预算继续」、
awaiting_confirm outcome、同参拒+第4/6轮检查点、侧栏忙时单槽排队。
Ask 真 token 流式记缺口(三协议 SSE 成本高)。
This commit is contained in:
logan committed 2026-10-15 14:18:18 +08:00
1 parent 1513f4bc2d
commit d543f3b924
10 files changed
+783 -104

No files matched your search

+23 -12
View File
@@ -59,6 +59,8 @@ type AgentRes struct {
IncompleteReason string `json:"incompleteReason,omitempty"`
Resumable bool `json:"resumable,omitempty"`
StepsBudget int `json:"stepsBudget,omitempty"`
// Trajectory 含 system/user/assistant.tool_calls/tool 的完整对话(供落盘续跑;JSON 默认省略大字段可选)
Trajectory []Message `json:"trajectory,omitempty"`
}
// RunAgent 多轮:LLM tool_call → aitools.Runtime.Call → 回灌。
@@ -90,6 +92,11 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
if runID == "" {
runID = NewRunID()
}
sessionID := strings.TrimSpace(req.SessionID)
if sessionID == "" {
sessionID = NewSessionID()
req.SessionID = sessionID
}
rt.RunID = runID
// Agent 路径禁止模型自行带确认令牌
rt.RejectCallerConfirm = true
@@ -119,13 +126,14 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
}
}
baseMeta := func(steps []AgentStep, rounds int) AgentRes {
finish := func(steps []AgentStep, rounds int, msgs []Message) AgentRes {
return AgentRes{
Steps: steps,
Rounds: rounds,
RunID: runID,
SessionID: req.SessionID,
SessionID: sessionID,
StepsBudget: maxRounds,
Trajectory: append([]Message{}, msgs...),
}
}
@@ -134,19 +142,22 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
rounds := 0
for round := 0; round < maxRounds; round++ {
if err := ctx.Err(); err != nil {
res := baseMeta(steps, rounds)
res := finish(steps, rounds, msgs)
res.Content = "已取消"
res.Outcome = OutcomeCancelled
res.Resumable = len(steps) > 0
emitEvent(emit, AgentEvent{Type: AgentEventError, Error: "cancelled", Steps: steps, Rounds: rounds, RunID: runID, Outcome: OutcomeCancelled})
emitEvent(emit, AgentEvent{Type: AgentEventError, Error: "cancelled", Steps: steps, Rounds: rounds, RunID: runID, SessionID: sessionID, Outcome: OutcomeCancelled})
return res, err
}
rounds = round + 1
if tip := CheckpointHint(rounds); tip != "" {
msgs = append(msgs, Message{Role: "system", Content: tip})
}
turn, err := llm.ChatWithTools(ctx, req.Model, msgs, tools)
if err != nil {
res := baseMeta(steps, rounds)
res := finish(steps, rounds, msgs)
res.Outcome = OutcomeFailed
emitEvent(emit, AgentEvent{Type: AgentEventError, Error: err.Error(), Steps: steps, Rounds: rounds, RunID: runID, Outcome: OutcomeFailed})
emitEvent(emit, AgentEvent{Type: AgentEventError, Error: err.Error(), Steps: steps, Rounds: rounds, RunID: runID, SessionID: sessionID, Outcome: OutcomeFailed})
return res, err
}
if len(turn.ToolCalls) == 0 {
@@ -157,13 +168,13 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
for _, part := range chunkDelta(content, 36) {
emitEvent(emit, AgentEvent{Type: AgentEventDelta, Delta: part})
}
res := baseMeta(steps, rounds)
res := finish(steps, rounds, msgs)
res.Content = content
res.SQL = extractSQL(content)
res.Outcome = OutcomeCompleted
emitEvent(emit, AgentEvent{
Type: AgentEventDone, Content: res.Content, SQL: res.SQL, Steps: steps, Rounds: rounds,
RunID: runID, Outcome: OutcomeCompleted, StepsBudget: maxRounds,
RunID: runID, SessionID: sessionID, Outcome: OutcomeCompleted, StepsBudget: maxRounds,
})
return res, nil
}
@@ -217,13 +228,13 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
}
if awaiting {
content := "需要确认写操作后才能继续。"
res := baseMeta(steps, rounds)
res := finish(steps, rounds, msgs)
res.Content = content
res.Outcome = OutcomeAwaitingConfirm
res.Resumable = true
emitEvent(emit, AgentEvent{
Type: AgentEventDone, Content: content, Steps: steps, Rounds: rounds,
RunID: runID, Outcome: OutcomeAwaitingConfirm, Resumable: true, StepsBudget: maxRounds,
RunID: runID, SessionID: sessionID, Outcome: OutcomeAwaitingConfirm, Resumable: true, StepsBudget: maxRounds,
})
return res, nil
}
@@ -234,14 +245,14 @@ func runAgent(ctx context.Context, llm ToolChat, rt *aitools.Runtime, req AgentR
for _, part := range chunkDelta(content, 36) {
emitEvent(emit, AgentEvent{Type: AgentEventDelta, Delta: part})
}
res := baseMeta(steps, rounds)
res := finish(steps, rounds, msgs)
res.Content = content
res.Outcome = OutcomeIncomplete
res.IncompleteReason = ReasonMaxSteps
res.Resumable = true
emitEvent(emit, AgentEvent{
Type: AgentEventIncomplete, Content: content, Steps: steps, Rounds: rounds,
RunID: runID, Outcome: OutcomeIncomplete, IncompleteReason: ReasonMaxSteps,
RunID: runID, SessionID: sessionID, Outcome: OutcomeIncomplete, IncompleteReason: ReasonMaxSteps,
Resumable: true, StepsBudget: maxRounds,
})
return res, nil
+1
View File
@@ -37,6 +37,7 @@ type AgentEvent struct {
Error string `json:"error,omitempty"`
Result *AgentQueryResult `json:"result,omitempty"`
RunID string `json:"runId,omitempty"`
SessionID string `json:"sessionId,omitempty"`
Outcome string `json:"runOutcome,omitempty"`
IncompleteReason string `json:"incompleteReason,omitempty"`
Resumable bool `json:"resumable,omitempty"`
+12
View File
@@ -163,3 +163,15 @@ func DetectRepeatedToolCall(history []string, fp string, limit int) bool {
}
return n >= limit
}
// CheckpointHint 中段检查点(第 4 / 6 轮,1-based);空串表示本轮无注入。
func CheckpointHint(round1Based int) string {
switch round1Based {
case 4:
return "[检查点] 若已有足够信息请直接作答;避免重复相同参数的工具调用。"
case 6:
return "[检查点] 请尽快收尾给出结论,勿再扩大探索范围。"
default:
return ""
}
}
+156 -15
View File
@@ -1,33 +1,48 @@
package ai
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"time"
)
// Session 桌面本地会话(目标态落盘;HTTP 路由可后挂)。
// Session 桌面本地会话(落盘 ~/.opsxdb/ai-sessions/<id>.json)。
type Session struct {
ID string `json:"id"`
Title string `json:"title,omitempty"`
Status string `json:"status"` // idle|running|incomplete|completed|failed|cancelled
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
Messages []Message `json:"messages"`
Runs []SessionRun `json:"runs,omitempty"`
ID string `json:"id"`
Title string `json:"title,omitempty"`
Status string `json:"status"` // idle|running|incomplete|completed|failed|cancelled|awaiting_confirm
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
Messages []Message `json:"messages"`
Runs []SessionRun `json:"runs,omitempty"`
}
// SessionRun 一次运行摘要。
type SessionRun struct {
RunID string `json:"runId"`
Outcome string `json:"outcome"`
IncompleteReason string `json:"incompleteReason,omitempty"`
Resumable bool `json:"resumable"`
StepsUsed int `json:"stepsUsed"`
StepsBudget int `json:"stepsBudget"`
FinishedAt string `json:"finishedAt"`
RunID string `json:"runId"`
Outcome string `json:"outcome"`
IncompleteReason string `json:"incompleteReason,omitempty"`
Resumable bool `json:"resumable"`
StepsUsed int `json:"stepsUsed"`
StepsBudget int `json:"stepsBudget"`
FinishedAt string `json:"finishedAt"`
}
// SessionSummary 列表项(不含 messages)。
type SessionSummary struct {
ID string `json:"id"`
Title string `json:"title,omitempty"`
Status string `json:"status"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
MsgCount int `json:"msgCount"`
LastRunID string `json:"lastRunId,omitempty"`
}
func sessionsDir() (string, error) {
@@ -44,9 +59,20 @@ func SessionPath(id string) (string, error) {
if err != nil {
return "", err
}
id = strings.TrimSpace(id)
if id == "" || strings.ContainsAny(id, `/\`) {
return "", fmt.Errorf("invalid session id")
}
return filepath.Join(dir, id+".json"), nil
}
// NewSessionID 生成 sess_… 短 id。
func NewSessionID() string {
var b [8]byte
_, _ = rand.Read(b[:])
return "sess_" + hex.EncodeToString(b[:])
}
// SaveSession 脱敏后写入磁盘。
func SaveSession(s Session) error {
if s.ID == "" {
@@ -87,3 +113,118 @@ func LoadSession(id string) (Session, error) {
}
return s, nil
}
// ListSessions 按 UpdatedAt 倒序;limit<=0 表示全部。
func ListSessions(limit int) ([]SessionSummary, error) {
dir, err := sessionsDir()
if err != nil {
return nil, err
}
entries, err := os.ReadDir(dir)
if err != nil {
if os.IsNotExist(err) {
return []SessionSummary{}, nil
}
return nil, err
}
var out []SessionSummary
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".json") {
continue
}
id := strings.TrimSuffix(e.Name(), ".json")
s, err := LoadSession(id)
if err != nil {
continue
}
sum := SessionSummary{
ID: s.ID, Title: s.Title, Status: s.Status,
CreatedAt: s.CreatedAt, UpdatedAt: s.UpdatedAt,
MsgCount: len(s.Messages),
}
if n := len(s.Runs); n > 0 {
sum.LastRunID = s.Runs[n-1].RunID
}
out = append(out, sum)
}
sort.Slice(out, func(i, j int) bool { return out[i].UpdatedAt > out[j].UpdatedAt })
if limit > 0 && len(out) > limit {
out = out[:limit]
}
return out, nil
}
// DeleteSession 删除会话文件。
func DeleteSession(id string) error {
path, err := SessionPath(id)
if err != nil {
return err
}
err = os.Remove(path)
if os.IsNotExist(err) {
return nil
}
return err
}
// OutcomeToSessionStatus 映射 RunOutcome → Session.Status。
func OutcomeToSessionStatus(outcome string) string {
switch outcome {
case OutcomeIncomplete:
return "incomplete"
case OutcomeAwaitingConfirm:
return "awaiting_confirm"
case OutcomeFailed:
return "failed"
case OutcomeCancelled:
return "cancelled"
case OutcomeCompleted:
return "completed"
default:
if outcome == "" {
return "idle"
}
return outcome
}
}
// PersistAgentRun 把本趟轨迹与 outcome 写入会话(真·续跑用)。
func PersistAgentRun(sessionID, title string, trajectory []Message, res AgentRes) (Session, error) {
sessionID = strings.TrimSpace(sessionID)
if sessionID == "" {
sessionID = NewSessionID()
}
s, err := LoadSession(sessionID)
if err != nil {
if !os.IsNotExist(err) {
return Session{}, err
}
s = Session{ID: sessionID, Status: "idle"}
}
if title != "" && s.Title == "" {
// 标题取首条 user 或短截断
t := strings.TrimSpace(title)
r := []rune(t)
if len(r) > 48 {
t = string(r[:48]) + "…"
}
s.Title = t
}
if len(trajectory) > 0 {
s.Messages = trajectory
}
s.Status = OutcomeToSessionStatus(res.Outcome)
s.Runs = append(s.Runs, SessionRun{
RunID: res.RunID,
Outcome: res.Outcome,
IncompleteReason: res.IncompleteReason,
Resumable: res.Resumable,
StepsUsed: res.Rounds,
StepsBudget: res.StepsBudget,
FinishedAt: time.Now().Format(time.RFC3339),
})
if err := SaveSession(s); err != nil {
return Session{}, err
}
return s, nil
}
+175
View File
@@ -0,0 +1,175 @@
package ai
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"opsxdb/server/internal/aitools"
"opsxdb/server/internal/db"
)
func TestSessionPersistAndLoadTrajectory(t *testing.T) {
dir := t.TempDir()
t.Setenv("HOME", dir)
id := NewSessionID()
traj := []Message{
{Role: "system", Content: "sys"},
{Role: "user", Content: "list tables"},
{Role: "assistant", Content: "", ToolCalls: []ToolCall{{ID: "c1", Type: "function", Function: ToolCallFunc{Name: "opsxdb_list_tables", Arguments: "{}"}}}},
{Role: "tool", ToolCallID: "c1", Name: "opsxdb_list_tables", Content: `{"tables":["users"]}`},
}
res := AgentRes{RunID: "run_x", Outcome: OutcomeIncomplete, IncompleteReason: ReasonMaxSteps, Resumable: true, Rounds: 8, StepsBudget: 8, SessionID: id}
s, err := PersistAgentRun(id, "list tables", traj, res)
if err != nil {
t.Fatal(err)
}
if s.Status != "incomplete" {
t.Fatalf("status=%s", s.Status)
}
loaded, err := LoadSession(id)
if err != nil {
t.Fatal(err)
}
if len(loaded.Messages) != 4 {
t.Fatalf("msgs=%d", len(loaded.Messages))
}
if loaded.Messages[3].ToolCallID != "c1" {
t.Fatal("tool trajectory lost")
}
list, err := ListSessions(10)
if err != nil || len(list) < 1 {
t.Fatalf("list: %v %#v", err, list)
}
path := filepath.Join(dir, ".opsxdb", "ai-sessions", id+".json")
if _, err := os.Stat(path); err != nil {
t.Fatal(err)
}
}
func TestRunAgentContinueInjectsPrior(t *testing.T) {
rt := aitools.NewRuntime(aitools.Policy{Mode: aitools.ModeReadOnly, AllowedConnectionIDs: []string{"c1"}, DefaultLimit: 100, MaxLimit: 500})
rt.LoadConns = func() ([]aitools.SavedConnection, error) {
return []aitools.SavedConnection{{ID: "c1", Name: "d", Driver: "mysql"}}, nil
}
rt.Audit = func(aitools.AuditEntry) {}
prior := []Message{
{Role: "system", Content: "sys"},
{Role: "user", Content: "prev"},
{Role: "assistant", Content: "partial", ToolCalls: []ToolCall{{ID: "c1", Type: "function", Function: ToolCallFunc{Name: aitools.ToolListConnections, Arguments: "{}"}}}},
{Role: "tool", ToolCallID: "c1", Name: aitools.ToolListConnections, Content: `{"connections":[]}`},
}
llm := &scriptedLLM{turns: []ToolsTurn{{Content: "continued answer"}}}
res, err := RunAgent(context.Background(), llm, rt, AgentReq{
Prompt: "加预算继续", Continue: true, Prior: prior, MaxRounds: 4, SessionID: "sess_test",
})
if err != nil {
t.Fatal(err)
}
if res.Outcome != OutcomeCompleted {
t.Fatalf("outcome=%s", res.Outcome)
}
if len(llm.seen) < 1 {
t.Fatal("no llm call")
}
sawTool := false
for _, m := range llm.seen[0] {
if m.Role == "tool" && m.ToolCallID == "c1" {
sawTool = true
}
}
if !sawTool {
t.Fatalf("prior tool not injected: %+v", llm.seen[0])
}
if res.SessionID == "" || len(res.Trajectory) < 4 {
t.Fatalf("session/traj: sid=%s traj=%d", res.SessionID, len(res.Trajectory))
}
}
func TestRunAgentAwaitingConfirmOutcome(t *testing.T) {
rt := aitools.NewRuntime(aitools.Policy{
Mode: aitools.ModeReadWrite,
AllowedConnectionIDs: []string{"c1"},
DefaultLimit: 100,
MaxLimit: 500,
})
rt.FindConn = func(id string) (*aitools.SavedConnection, error) {
return &aitools.SavedConnection{ID: id, Name: "d", Driver: "mysql", Host: "127.0.0.1", Port: 3306, Database: "demo"}, nil
}
rt.Audit = func(aitools.AuditEntry) {}
execCalls := 0
rt.ExecSQL = func(_ context.Context, _ db.ConnReq) db.ExecResult {
execCalls++
return db.ExecResult{RowCount: 1}
}
args := `{"connection_id":"c1","database":"demo","sql":"INSERT INTO users(name) VALUES('x')"}`
llm := &scriptedLLM{turns: []ToolsTurn{{
ToolCalls: []ToolCall{{ID: "w1", Type: "function", Function: ToolCallFunc{Name: aitools.ToolExecuteQuery, Arguments: args}}},
}}}
res, err := RunAgent(context.Background(), llm, rt, AgentReq{Prompt: "insert one", MaxRounds: 4})
if err != nil {
t.Fatal(err)
}
if res.Outcome != OutcomeAwaitingConfirm {
t.Fatalf("outcome=%s want awaiting_confirm", res.Outcome)
}
if !res.Resumable {
t.Fatal("expected resumable")
}
if execCalls != 0 {
t.Fatal("must not execute write without confirm")
}
found := false
for _, s := range res.Steps {
if s.NeedConfirm {
found = true
}
}
if !found {
t.Fatalf("steps missing needConfirm: %+v", res.Steps)
}
}
func TestCheckpointHint(t *testing.T) {
if CheckpointHint(4) == "" || CheckpointHint(6) == "" {
t.Fatal("expected hints")
}
if CheckpointHint(1) != "" {
t.Fatal("no hint round 1")
}
}
func TestRunAgentInjectsCheckpoint(t *testing.T) {
rt := aitools.NewRuntime(aitools.Policy{Mode: aitools.ModeReadOnly, AllowedConnectionIDs: []string{"c1"}, DefaultLimit: 100, MaxLimit: 500})
rt.LoadConns = func() ([]aitools.SavedConnection, error) {
return []aitools.SavedConnection{{ID: "c1", Name: "d", Driver: "mysql"}}, nil
}
rt.Audit = func(aitools.AuditEntry) {}
var turns []ToolsTurn
for i := 0; i < 4; i++ {
turns = append(turns, ToolsTurn{ToolCalls: []ToolCall{{
ID: "c" + string(rune('a'+i)), Type: "function",
Function: ToolCallFunc{Name: aitools.ToolListConnections, Arguments: `{"n":` + string(rune('0'+i)) + `}`},
}}})
}
turns = append(turns, ToolsTurn{Content: "ok"})
llm := &scriptedLLM{turns: turns}
_, err := RunAgent(context.Background(), llm, rt, AgentReq{Prompt: "x", MaxRounds: 8})
if err != nil {
t.Fatal(err)
}
if len(llm.seen) < 4 {
t.Fatalf("seen=%d", len(llm.seen))
}
found := false
for _, m := range llm.seen[3] {
if m.Role == "system" && strings.Contains(m.Content, "检查点") {
found = true
}
}
if !found {
t.Fatalf("checkpoint not in round4 msgs")
}
}