diff --git a/docs/16-AI-Session-Harness.md b/docs/16-AI-Session-Harness.md index e4a6789..00d4f8d 100644 --- a/docs/16-AI-Session-Harness.md +++ b/docs/16-AI-Session-Harness.md @@ -264,12 +264,19 @@ confirm_bind = sha256(connection_id + "\n" + normalize(sql_or_command)) | ✅ | MCP 工具面、Ask\|Agent、SSE、P1 确认写、`run_id` / confirmBind、触顶 incomplete | | ✅ | 设置 → **MCP** 子页（安装检测、只读/危险 SQL、白名单、多客户端片段） | | ✅ | 官网 AI/MCP 块 + 文档 `/docs/mcp` | -| ⬜ | 会话落盘 + HTTP 续跑（`/api/ai/sessions`）+ UI「加预算继续」 | -| ⬜ | NeedConfirm 结束 run 为 `awaiting_confirm` outcome（语义对齐） | -| ⬜ | 同参循环拒 / 中段检查点接入主循环 | -| ⬜ | 忙时单槽排队 UI | -| ⬜ | Ask 真 token 流式（非假打字） | +| ✅ | 会话落盘 + HTTP 续跑（`GET/DELETE /api/ai/sessions`、`POST …/continue`）+ UI「加预算继续」 | +| ✅ | NeedConfirm 结束 run 为 `awaiting_confirm` outcome | +| ✅ | 同参循环拒 + 中段检查点（第 4 / 6 轮）接入主循环 | +| ✅ | 忙时单槽排队 UI（覆盖 / 撤销 / 立即 / 结束后自动续发） | +| ⬜ | Ask 真 token 流式（非假打字）— 成本高：需 openai-chat / anthropic / responses 三套 SSE；本轮记缺口，Ask 仍一次性 Complete | | ⬜ | MCP「断开」结束外部 Cursor 会话（当前仅 UI 提示） | | ⬜ | 设置页一键 `npm install -g`（现为复制命令；避免桌面代跑需授权） | -*本节随交付勾选；大 UI 续跑优先于 Evolver / 审批流。* +### 10.1 本轮验收要点 + +- `go test ./internal/ai/ ./internal/aitools/ ./internal/server/` +- Agent 触顶 → `run_outcome=incomplete` + 气泡「加预算继续」→ `continue=true` 注入落盘 prior +- 写工具无确认 → `awaiting_confirm`；确认条仍走 `/api/ai/agent/confirm` + confirmBind +- Agent 忙碌时再发送 → 单槽排队条；结束后自动发或点「立即」 + +*本节随交付勾选；Ask SSE 与 MCP 杀外部进程后置。* diff --git a/server/internal/ai/agent.go b/server/internal/ai/agent.go index dc45c90..5b487a8 100644 --- a/server/internal/ai/agent.go +++ b/server/internal/ai/agent.go @@ -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 diff --git a/server/internal/ai/agent_stream.go b/server/internal/ai/agent_stream.go index a7e733b..afceb5a 100644 --- a/server/internal/ai/agent_stream.go +++ b/server/internal/ai/agent_stream.go @@ -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"` diff --git a/server/internal/ai/harness.go b/server/internal/ai/harness.go index 0c329e5..75d7a5b 100644 --- a/server/internal/ai/harness.go +++ b/server/internal/ai/harness.go @@ -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 "" + } +} diff --git a/server/internal/ai/session.go b/server/internal/ai/session.go index 7ba1ed1..3d4d7ee 100644 --- a/server/internal/ai/session.go +++ b/server/internal/ai/session.go @@ -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/.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 +} diff --git a/server/internal/ai/session_test.go b/server/internal/ai/session_test.go new file mode 100644 index 0000000..aaaea99 --- /dev/null +++ b/server/internal/ai/session_test.go @@ -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") + } +} diff --git a/server/internal/server/router.go b/server/internal/server/router.go index cd65046..c28e2e3 100644 --- a/server/internal/server/router.go +++ b/server/internal/server/router.go @@ -474,6 +474,80 @@ func NewRouter(token string) *gin.Engine { c.JSON(http.StatusOK, gin.H{"ok": true, "count": len(req.Connections)}) }) + + // AI 会话落盘（真·续跑）：~/.opsxdb/ai-sessions/ + api.GET("/ai/sessions", func(c *gin.Context) { + list, err := ai.ListSessions(50) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"sessions": list}) + }) + + api.GET("/ai/sessions/:id", func(c *gin.Context) { + s, err := ai.LoadSession(c.Param("id")) + if err != nil { + if os.IsNotExist(err) { + c.JSON(http.StatusNotFound, gin.H{"error": "session not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, s) + }) + + api.DELETE("/ai/sessions/:id", func(c *gin.Context) { + if err := ai.DeleteSession(c.Param("id")); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true}) + }) + + // 续跑：从落盘轨迹注入 Prior；可选提高 MaxRounds（加预算继续）。 + api.POST("/ai/sessions/:id/continue", func(c *gin.Context) { + var body struct { + Prompt string `json:"prompt"` + ConnectionID string `json:"connectionId"` + Driver string `json:"driver"` + Database string `json:"database"` + Model string `json:"model"` + MaxRounds int `json:"maxRounds"` + Stream bool `json:"stream"` + } + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + sess, err := ai.LoadSession(c.Param("id")) + if err != nil { + if os.IsNotExist(err) { + c.JSON(http.StatusNotFound, gin.H{"error": "session not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + req := ai.AgentReq{ + Prompt: body.Prompt, + ConnectionID: body.ConnectionID, + Driver: body.Driver, + Database: body.Database, + Model: body.Model, + MaxRounds: body.MaxRounds, + Stream: body.Stream, + SessionID: sess.ID, + Continue: true, + Prior: sess.Messages, + } + if req.MaxRounds <= 0 { + req.MaxRounds = 12 // 加预算默认 + } + runAgentHTTP(c, req) + }) + // 侧栏 Agent：多轮 tool call → 同一 aitools.Runtime（进程内，不绕 stdio）。Ask 仍走 /ai/chat。 // stream=true 或 Accept: text/event-stream 时走 SSE（步骤边推 + 正文分块）；否则 JSON fallback。 api.POST("/ai/agent", func(c *gin.Context) { @@ -482,46 +556,13 @@ func NewRouter(token string) *gin.Engine { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - llm, err := ai.DefaultToolChat() - if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) - return - } - rt, err := aitools.DefaultRuntime() - if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) - return - } - rt.Source = "agent" - rt.RejectCallerConfirm = true - - wantStream := req.Stream || strings.Contains(c.GetHeader("Accept"), "text/event-stream") - if wantStream { - flusher, ok := c.Writer.(http.Flusher) - if !ok { - c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming unsupported"}) - return + // 真·续跑：Continue 且未带 Prior 时从落盘会话注入 tool 轨迹 + if req.Continue && len(req.Prior) == 0 && strings.TrimSpace(req.SessionID) != "" { + if sess, err := ai.LoadSession(req.SessionID); err == nil && len(sess.Messages) > 0 { + req.Prior = sess.Messages } - c.Header("Content-Type", "text/event-stream") - c.Header("Cache-Control", "no-cache") - c.Header("Connection", "keep-alive") - c.Status(http.StatusOK) - flusher.Flush() - _, err := ai.RunAgentStream(c.Request.Context(), llm, rt, req, func(ev ai.AgentEvent) { - _ = ai.WriteAgentSSE(c.Writer, flusher, ev) - }) - if err != nil && c.Request.Context().Err() == nil { - _ = ai.WriteAgentSSE(c.Writer, flusher, ai.AgentEvent{Type: ai.AgentEventError, Error: err.Error()}) - } - return } - - res, err := ai.RunAgent(c.Request.Context(), llm, rt, req) - if err != nil { - c.JSON(http.StatusBadGateway, gin.H{"error": err.Error(), "steps": res.Steps, "rounds": res.Rounds}) - return - } - c.JSON(http.StatusOK, res) + runAgentHTTP(c, req) }) // 用户在气泡确认条点「执行」后：单次写操作（AllowConfirmedWrite），绝不由模型静默触发。 @@ -713,6 +754,99 @@ func sanitizeFileToken(s string) string { return b.String() } + +// runAgentHTTP 共用 Agent JSON / SSE；结束后落盘轨迹（真·续跑）。 +func runAgentHTTP(c *gin.Context, req ai.AgentReq) { + llm, err := ai.DefaultToolChat() + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + rt, err := aitools.DefaultRuntime() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + rt.Source = "agent" + rt.RejectCallerConfirm = true + + persist := func(res ai.AgentRes) ai.AgentRes { + title := strings.TrimSpace(req.Prompt) + if title == "" { + for _, m := range req.Prior { + if m.Role == "user" && strings.TrimSpace(m.Content) != "" { + title = m.Content + break + } + } + } + traj := res.Trajectory + if sess, err := ai.PersistAgentRun(res.SessionID, title, traj, res); err == nil { + res.SessionID = sess.ID + } + res.Trajectory = nil + return res + } + + wantStream := req.Stream || strings.Contains(c.GetHeader("Accept"), "text/event-stream") + if wantStream { + flusher, ok := c.Writer.(http.Flusher) + if !ok { + c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming unsupported"}) + return + } + c.Header("Content-Type", "text/event-stream") + c.Header("Cache-Control", "no-cache") + c.Header("Connection", "keep-alive") + c.Status(http.StatusOK) + flusher.Flush() + + var final ai.AgentRes + res, err := ai.RunAgentStream(c.Request.Context(), llm, rt, req, func(ev ai.AgentEvent) { + // 补上 sessionId（run 内已分配） + if ev.Type == ai.AgentEventDone || ev.Type == ai.AgentEventIncomplete || ev.Type == ai.AgentEventError { + if ev.RunID == "" { + ev.RunID = req.RunID + } + } + _ = ai.WriteAgentSSE(c.Writer, flusher, ev) + }) + final = persist(res) + if err != nil && c.Request.Context().Err() == nil { + _ = ai.WriteAgentSSE(c.Writer, flusher, ai.AgentEvent{ + Type: ai.AgentEventError, Error: err.Error(), + RunID: final.RunID, Outcome: ai.OutcomeFailed, + }) + return + } + // 补发 session 元数据（轻量），避免改动 loop 内已发的 done + _ = ai.WriteAgentSSE(c.Writer, flusher, ai.AgentEvent{ + Type: "session", + Content: final.SessionID, + RunID: final.RunID, + Outcome: final.Outcome, + IncompleteReason: final.IncompleteReason, + Resumable: final.Resumable, + StepsBudget: final.StepsBudget, + Rounds: final.Rounds, + }) + return + } + + res, err := ai.RunAgent(c.Request.Context(), llm, rt, req) + if err != nil { + res = persist(res) + c.JSON(http.StatusBadGateway, gin.H{ + "error": err.Error(), "steps": res.Steps, "rounds": res.Rounds, + "sessionId": res.SessionID, "runId": res.RunID, "runOutcome": res.Outcome, + }) + return + } + res = persist(res) + c.JSON(http.StatusOK, res) +} + + func health(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"ok": true, "service": "opsxdb", "version": Version}) } diff --git a/src/AiPanel.tsx b/src/AiPanel.tsx index 8f66fc7..7002909 100644 --- a/src/AiPanel.tsx +++ b/src/AiPanel.tsx @@ -12,7 +12,8 @@ import { type McpStatus, } from "./api"; import { AI_PRESETS, findAiPreset, type AiAuthStyle, type AiProtocol } from "./aiPresets"; -import { useWorkspaceStore, type AiAgentStepView, type AiChatMode, type QueryResult } from "./store"; +import { useWorkspaceStore, type AiAgentStepView, type AiChatMode, + type AiMessage, type QueryResult } from "./store"; import { editorDatabase } from "./redisContext"; const ACTIONS: { id: AiAction; label: string }[] = [ @@ -409,6 +410,8 @@ export function AiPanel() { const model = useWorkspaceStore((s) => s.aiModel); const busy = useWorkspaceStore((s) => s.aiBusy); const aiMode = useWorkspaceStore((s) => s.aiMode); + const aiSessionId = useWorkspaceStore((s) => s.aiSessionId); + const aiQueuedPrompt = useWorkspaceStore((s) => s.aiQueuedPrompt); const connections = useWorkspaceStore((s) => s.connections); const activeConnectionId = useWorkspaceStore((s) => s.activeConnectionId); const tabs = useWorkspaceStore((s) => s.tabs); @@ -503,15 +506,25 @@ export function AiPanel() { } }; - const runAgent = async () => { - const text = ask.trim(); - if (busy || !text) return; + const runAgent = async (opts?: { + prompt?: string; + continue?: boolean; + maxRounds?: number; + skipUserBubble?: boolean; + }) => { + const text = (opts?.prompt ?? ask).trim(); + const isContinue = !!opts?.continue; + if (busy) return; + if (!isContinue && !text) return; set("aiBusy", true); - pushAi("user", text); - set("aiAsk", ""); + if (!opts?.skipUserBubble) { + pushAi("user", isContinue && !text ? "加预算继续" : text); + } + if (!opts?.prompt) set("aiAsk", ""); const asstId = pushAi("assistant", "", [], { streaming: true }); const ac = new AbortController(); abortRef.current = ac; + const sessionId = useWorkspaceStore.getState().aiSessionId || undefined; try { const res = await callAiAgentStream( { @@ -520,15 +533,18 @@ export function AiPanel() { driver: conn?.driver, database: database || undefined, model: model && model !== "opsxdb-stub" ? model : undefined, + sessionId, + continue: isContinue || undefined, + maxRounds: opts?.maxRounds, }, { onStepStart: (step) => { - patchAiMessage(asstId, (m) => ({ - ...m, - steps: [...(m.steps || []).filter((s) => !s.pending), { ...step, pending: true, ok: false, summary: "执行中…", ms: 0 }], - })); + patchAiMessage(asstId, (m) => { + const withoutPending = (m.steps || []).filter((s) => !s.pending); + return { ...m, steps: [...withoutPending, { ...step, pending: true, ok: false, summary: "执行中…", ms: 0 }] }; + }); }, - onStep: (step: AiAgentStepExt) => { + onStep: (step) => { patchAiMessage(asstId, (m) => { const withoutPending = (m.steps || []).filter((s) => !s.pending); return { ...m, steps: [...withoutPending, step] }; @@ -543,14 +559,45 @@ export function AiPanel() { steps: final.steps || [], streaming: false, runId: final.runId, + sessionId: final.sessionId, + runOutcome: final.runOutcome || "completed", + resumable: final.resumable, + incompleteReason: final.incompleteReason, + stepsBudget: final.stepsBudget, }); + if (final.sessionId) set("aiSessionId", final.sessionId); if (final.sql && tab) updateTabSqlIfUnchanged(tab.id, sql, tab.connectionId, tab.database, final.sql); }, + onIncomplete: (final) => { + patchAiMessage(asstId, { + content: final.content, + steps: final.steps || [], + streaming: false, + runId: final.runId, + sessionId: final.sessionId, + runOutcome: "incomplete", + resumable: true, + incompleteReason: final.incompleteReason || "max_steps", + stepsBudget: final.stepsBudget, + }); + if (final.sessionId) set("aiSessionId", final.sessionId); + }, + onSession: (meta) => { + if (meta.sessionId) set("aiSessionId", meta.sessionId); + patchAiMessage(asstId, (m) => ({ + ...m, + sessionId: meta.sessionId || m.sessionId, + runId: meta.runId || m.runId, + runOutcome: meta.runOutcome || m.runOutcome, + resumable: meta.resumable ?? m.resumable, + })); + }, onError: (err, partial) => { patchAiMessage(asstId, { content: partial?.content || `失败：${err}`, steps: (partial?.steps as AiAgentStepView[]) || [], streaming: false, + runOutcome: "failed", }); }, }, @@ -559,25 +606,68 @@ export function AiPanel() { void res; } catch (e) { if ((e as Error)?.name === "AbortError") { - patchAiMessage(asstId, (m) => ({ ...m, content: m.content || "已停止", streaming: false })); + patchAiMessage(asstId, (m) => ({ ...m, content: m.content || "已停止", streaming: false, runOutcome: "cancelled" })); } else { patchAiMessage(asstId, { content: `失败：${e instanceof Error ? e.message : String(e)}`, streaming: false, + runOutcome: "failed", }); } } finally { abortRef.current = null; set("aiBusy", false); patchAiMessage(asstId, (m) => ({ ...m, streaming: false })); + // 忙时单槽：结束后自动续发排队 + const queued = useWorkspaceStore.getState().aiQueuedPrompt; + if (queued && queued.trim()) { + set("aiQueuedPrompt", null); + void runAgent({ prompt: queued.trim() }); + } } }; const send = () => { - if (aiMode === "agent") void runAgent(); - else void runAsk(action); + if (aiMode === "agent") { + const text = ask.trim(); + if (!text) return; + if (busy) { + // 单槽排队：再提交覆盖 + set("aiQueuedPrompt", text); + set("aiAsk", ""); + return; + } + void runAgent(); + return; + } + void runAsk(action); }; + const continueIncomplete = (m: AiMessage) => { + if (busy) return; + const sid = m.sessionId || aiSessionId; + if (sid) set("aiSessionId", sid); + void runAgent({ + prompt: "", + continue: true, + maxRounds: Math.min(16, (m.stepsBudget || 8) + 4), + skipUserBubble: false, + }); + }; + + const flushQueueNow = () => { + const q = useWorkspaceStore.getState().aiQueuedPrompt; + if (!q?.trim()) return; + set("aiQueuedPrompt", null); + abortRef.current?.abort(); + // 下一 tick 跑排队（等 abort finally 清 busy） + window.setTimeout(() => { + void runAgent({ prompt: q.trim() }); + }, 50); + }; + + const discardQueue = () => set("aiQueuedPrompt", null); + const pendingConfirm = (() => { for (let i = messages.length - 1; i >= 0; i--) { const m = messages[i]; @@ -683,6 +773,22 @@ export function AiPanel() { {m.role === "assistant" && m.steps && m.steps.length > 0 ? : null} + {m.role === "assistant" && m.runOutcome === "incomplete" && m.resumable && !m.streaming ? ( +
+ + + {m.incompleteReason === "max_steps" ? "已达轮次上限" : m.incompleteReason || "未完成"} + {m.stepsBudget ? ` · 预算 ${m.stepsBudget}` : ""} + +
+ ) : null} {pendingConfirm && pendingConfirm.messageId === m.id && pendingConfirm.step.confirmSql ? ( + {aiQueuedPrompt ? ( +
+ + 已排队：{aiQueuedPrompt} + + + +
+ ) : null} +
) : (
- Agent 与 MCP 同源工具；默认只读。写操作需 mcp.json mode=read_write 且在气泡内确认。 + Agent 与 MCP 同源工具；默认只读。写操作需确认。忙时可再输入排队（单槽）；触顶可「加预算继续」。
)}
diff --git a/src/api.ts b/src/api.ts index 2c85dd6..4bc85cf 100644 --- a/src/api.ts +++ b/src/api.ts @@ -548,16 +548,49 @@ export type AiAgentRes = { stepsBudget?: number; }; -export async function callAiAgent(payload: { - prompt: string; - connectionId?: string; - driver?: string; - database?: string; - model?: string; -}) { +export async function callAiAgent(payload: AiAgentStreamPayload) { return post("/api/ai/agent", payload); } +export type AiSessionSummary = { + id: string; + title?: string; + status: string; + createdAt: string; + updatedAt: string; + msgCount: number; + lastRunId?: string; +}; + +export async function listAiSessions() { + return get<{ sessions: AiSessionSummary[] }>("/api/ai/sessions"); +} + +export async function getAiSession(id: string) { + return get<{ + id: string; + title?: string; + status: string; + messages: unknown[]; + runs?: unknown[]; + }>(`/api/ai/sessions/${encodeURIComponent(id)}`); +} + +export async function continueAiSession( + id: string, + payload: { + prompt?: string; + connectionId?: string; + driver?: string; + database?: string; + model?: string; + maxRounds?: number; + stream?: boolean; + }, +) { + return post(`/api/ai/sessions/${encodeURIComponent(id)}/continue`, payload); +} + export type AiAgentQueryResult = { columns: string[]; rows: Record[]; @@ -579,18 +612,26 @@ export type AiAgentStreamHandlers = { onStep?: (step: AiAgentStepExt) => void; onDelta?: (delta: string) => void; onDone?: (res: AiAgentRes) => void; + onIncomplete?: (res: AiAgentRes) => void; + onSession?: (meta: { sessionId?: string; runId?: string; runOutcome?: string; resumable?: boolean }) => void; onError?: (error: string, partial?: Partial) => void; }; /** Agent SSE（stream=true）；不可用时回落非流式。signal 供「停止」。 */ +export type AiAgentStreamPayload = { + prompt: string; + connectionId?: string; + driver?: string; + database?: string; + model?: string; + sessionId?: string; + continue?: boolean; + maxRounds?: number; + prior?: unknown[]; +}; + export async function callAiAgentStream( - payload: { - prompt: string; - connectionId?: string; - driver?: string; - database?: string; - model?: string; - }, + payload: AiAgentStreamPayload, handlers: AiAgentStreamHandlers, signal?: AbortSignal, ): Promise { @@ -636,15 +677,38 @@ export async function callAiAgentStream( const d = String(ev.delta || ""); content += d; handlers.onDelta?.(d); - } else if (typ === "done") { + } else if (typ === "done" || typ === "incomplete") { finalRes = { content: String(ev.content ?? content), sql: ev.sql ? String(ev.sql) : undefined, steps: (ev.steps as AiAgentStep[]) || steps, rounds: Number(ev.rounds ?? rounds) || 0, + runId: ev.runId ? String(ev.runId) : undefined, + sessionId: ev.sessionId ? String(ev.sessionId) : undefined, + runOutcome: ev.runOutcome ? String(ev.runOutcome) : typ === "incomplete" ? "incomplete" : undefined, + incompleteReason: ev.incompleteReason ? String(ev.incompleteReason) : undefined, + resumable: !!ev.resumable, + stepsBudget: ev.stepsBudget != null ? Number(ev.stepsBudget) : undefined, }; rounds = finalRes.rounds; - handlers.onDone?.(finalRes); + if (typ === "incomplete") handlers.onIncomplete?.(finalRes); + else handlers.onDone?.(finalRes); + } else if (typ === "session") { + handlers.onSession?.({ + sessionId: ev.content ? String(ev.content) : ev.sessionId ? String(ev.sessionId) : undefined, + runId: ev.runId ? String(ev.runId) : undefined, + runOutcome: ev.runOutcome ? String(ev.runOutcome) : undefined, + resumable: !!ev.resumable, + }); + if (finalRes) { + finalRes = { + ...finalRes, + sessionId: (ev.content ? String(ev.content) : finalRes.sessionId) || finalRes.sessionId, + runId: ev.runId ? String(ev.runId) : finalRes.runId, + runOutcome: ev.runOutcome ? String(ev.runOutcome) : finalRes.runOutcome, + resumable: ev.resumable != null ? !!ev.resumable : finalRes.resumable, + }; + } } else if (typ === "error") { handlers.onError?.(String(ev.error || "agent error"), { content, steps, rounds: Number(ev.rounds) || 0 }); } diff --git a/src/store.ts b/src/store.ts index 496ccaa..ce6fc7d 100644 --- a/src/store.ts +++ b/src/store.ts @@ -217,6 +217,11 @@ export type AiMessage = { content: string; at: string; runId?: string; + sessionId?: string; + runOutcome?: string; + incompleteReason?: string; + resumable?: boolean; + stepsBudget?: number; /** Agent 工具步骤时间线（仅 assistant） */ steps?: AiAgentStepView[]; streaming?: boolean; @@ -266,6 +271,10 @@ type State = { aiBusy: boolean; /** Ask=不调工具；Agent=多轮 aitools */ aiMode: AiChatMode; + /** 当前 Agent 会话 id（落盘续跑） */ + aiSessionId: string | null; + /** 忙时单槽排队文案；再提交覆盖 */ + aiQueuedPrompt: string | null; status: string; notifyModal: NotifyPayload | null; notifyToasts: NotifyPayload[]; @@ -501,6 +510,8 @@ export const useWorkspaceStore = create((set, get) => ({ aiModel: "opsxdb-stub", aiBusy: false, aiMode: "ask", + aiSessionId: null, + aiQueuedPrompt: null, status: "就绪", notifyModal: null, notifyToasts: [],