feat(api): integrate task workflows and mock contracts

This commit is contained in:
pphua committed 2026-08-24 19:31:16 +08:00
1 parent ef57f3f196
commit ec971c63d2
9 files changed
+5867 -836

No files matched your search

+75 -3
View File
@@ -6,6 +6,8 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strconv"
"testing"
"time"
@@ -92,15 +94,85 @@ func TestCoreTaskLifecyclePG(t *testing.T) {
t.Fatalf("list tasks: %d", code)
}
// delete → cascade removes the exploration subgraph
code, _ = doRetry("DELETE", "/api/tasks/"+id, nil)
// The task detail header consumes the top-level engine mode while some clients
// read the active-task snapshot. Keep both representations in sync.
code, out = doRetry("GET", "/api/stats?task="+id, nil)
if code != 200 {
t.Fatalf("task stats: %d (%v)", code, out)
}
activeTask, ok := out["active_task"].(map[string]any)
if !ok || activeTask["engine_mode"] != out["engine_mode"] {
t.Fatalf("engine mode mismatch: top=%v active_task=%v", out["engine_mode"], activeTask)
}
// delete → cascade removes the exploration subgraph and selected related data
taskDir := filepath.Join(m.dir, "tasks", id)
if err := os.MkdirAll(taskDir, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(taskDir, "artifact.txt"), []byte("test"), 0o644); err != nil {
t.Fatal(err)
}
transcriptBase := "exp" + strconv.FormatInt(expID, 10) + "-main"
transcriptPath := filepath.Join(m.dir, "transcripts", transcriptBase+".jsonl")
sidechainPath := filepath.Join(m.dir, "transcripts", transcriptBase)
if err := os.MkdirAll(filepath.Join(sidechainPath, "subagents"), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(transcriptPath, []byte("{}\n"), 0o644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(sidechainPath, "subagents", "child.jsonl"), []byte("{}\n"), 0o644); err != nil {
t.Fatal(err)
}
nid, _ := strconv.ParseInt(id, 10, 64)
findingID, err := m.pg.AddFinding(nid, 0, "__core_task_delete__", "", db.SeverityHigh, "summary", "evidence", "tester", nil)
if err != nil {
t.Fatal(err)
}
if err := m.pg.EnsureLLMRecordsTable(); err != nil {
t.Fatal(err)
}
if err := m.pg.InsertLLMRecord(&db.LLMRecord{TaskID: id, SessionID: "delete-test", Status: "ok", RequestBody: "secret"}); err != nil {
t.Fatal(err)
}
// A non-canonical path still resolves to the one canonical task key for the
// barrier, workspace, in-memory registry and DB delete.
code, out = doRetry("DELETE", "/api/tasks/000"+id, map[string]bool{
"delete_files": true,
"delete_findings": true,
"delete_llm_records": true,
})
if code != 200 {
t.Fatalf("delete task: %d", code)
}
nid, _ := strconv.ParseInt(id, 10, 64)
if deleted, _ := out["files_deleted"].(bool); !deleted {
t.Fatalf("expected files_deleted response, got %v", out)
}
if deleted, _ := out["findings_deleted"].(float64); deleted != 1 {
t.Fatalf("expected one deleted finding, got %v", out)
}
if deleted, _ := out["llm_records_deleted"].(float64); deleted != 1 {
t.Fatalf("expected one deleted LLM record, got %v", out)
}
if _, err := os.Stat(taskDir); !os.IsNotExist(err) {
t.Fatalf("task directory should be deleted, stat err=%v", err)
}
for _, path := range []string{transcriptPath, sidechainPath} {
if _, err := os.Stat(path); !os.IsNotExist(err) {
t.Fatalf("task transcript should be deleted (%s), stat err=%v", path, err)
}
}
if got, _ := m.pg.GetTask(nid); got != nil {
t.Fatalf("task should be deleted from PG")
}
if finding, err := m.pg.GetFinding(findingID); err != nil || finding != nil {
t.Fatalf("finding should be deleted, got finding=%+v err=%v", finding, err)
}
var llmRecords int
if err := m.pg.QueryRow(`SELECT count(*) FROM llm_records WHERE task_id=$1`, id).Scan(&llmRecords); err != nil || llmRecords != 0 {
t.Fatalf("task LLM records should be deleted, count=%d err=%v", llmRecords, err)
}
var n int
m.pg.QueryRow(`SELECT count(*) FROM exploration_nodes WHERE exploration_id=$1`, expID).Scan(&n)
if n != 0 {
+89 -54
View File
@@ -30,26 +30,11 @@ func rawString(raw json.RawMessage) string {
// ---- Task (frontend "Task") ---- created_at as RFC3339, plus a derived status.
type TaskDTO struct {
<<<<<<< Updated upstream
ID string `json:"id"`
ExplorationID int64 `json:"exploration_id"`
Description string `json:"description"`
Goal string `json:"goal"`
Status string `json:"status"` // created | running | paused | done | failed
CreatedAt string `json:"created_at"`
CreatedUnix int64 `json:"created_unix"` // created_at as unix seconds (for run-duration calc)
CompletedAt string `json:"completed_at"` // RFC3339 finish time (done/failed); "" if unfinished
CompletedUnix int64 `json:"completed_unix"` // completed_at as unix seconds (0 if unfinished)
LastActivity int64 `json:"last_activity_unix"` // unix seconds of the last activity (0 if none)
Paused bool `json:"paused"`
Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption
GoalsTotal int `json:"goals_total"`
GoalsMet int `json:"goals_met"`
LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // LLM profile used for this task; nil = default
=======
ID string `json:"id"`
ExplorationID int64 `json:"exploration_id"`
Name string `json:"name"` // 可选任务名称;空=未命名
CategoryID *int64 `json:"category_id,omitempty"`
CategoryName string `json:"category_name,omitempty"`
Description string `json:"description"`
Goal string `json:"goal"`
Status string `json:"status"` // created | running | paused | done | failed
@@ -71,7 +56,6 @@ type TaskDTO struct {
SourceTaskIDs []string `json:"source_task_ids"`
CompanyIDs []int64 `json:"company_ids"`
CoverageEnabled bool `json:"coverage_enabled"` // 资产覆盖度功能开关(创建时定)
>>>>>>> Stashed changes
}
// TokenTotalDTO is a whole-task (all agents) token aggregate.
@@ -92,23 +76,19 @@ func tokenTotalDTO(u db.TokenUsage) TokenTotalDTO {
}
func taskDTO(t *Task, status string) TaskDTO {
lifecycle := t.lifecycleSnapshot()
llmState := t.llmStateSnapshot()
sourceIDs := make([]string, 0, len(lifecycle.SourceTaskIDs))
for _, id := range lifecycle.SourceTaskIDs {
sourceIDs = append(sourceIDs, i64s(id))
}
profileIDs := append(make([]int64, 0, len(llmState.ProfileIDs)), llmState.ProfileIDs...)
return TaskDTO{
<<<<<<< Updated upstream
ID: t.ID,
ExplorationID: t.ExpID,
Description: t.Description,
Goal: t.Goal,
Status: status,
CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)),
CreatedUnix: t.CreatedAt,
CompletedAt: completedRFC(t.CompletedAt),
CompletedUnix: t.CompletedAt,
Paused: t.Paused,
LLMProfileID: t.LLMProfileID,
=======
ID: t.ID,
ExplorationID: t.ExpID,
Name: t.Name,
CategoryID: lifecycle.CategoryID,
CategoryName: lifecycle.CategoryName,
Description: t.Description,
Goal: t.Goal,
Status: status,
@@ -126,7 +106,6 @@ func taskDTO(t *Task, status string) TaskDTO {
SourceTaskIDs: sourceIDs,
CompanyIDs: lifecycle.CompanyIDs,
CoverageEnabled: t.CoverageEnabled,
>>>>>>> Stashed changes
}
}
@@ -170,17 +149,19 @@ func trafficDTOs(ex []traffic.ExchangeMeta) []TrafficExchangeDTO {
// ---- TaskNode (frontend "TaskNode") — frontier, intents, exploration graph nodes ----
type TaskNodeDTO struct {
ID string `json:"id"`
Type string `json:"type"` // db Node.Kind
Payload string `json:"payload,omitempty"`
Priority int `json:"priority"`
State string `json:"state"`
Origin string `json:"origin"`
TS string `json:"ts"`
ID string `json:"id"`
Type string `json:"type"` // db Node.Kind
Payload string `json:"payload,omitempty"`
Priority int `json:"priority"`
State string `json:"state"`
Origin string `json:"origin"`
TS string `json:"ts"`
SourceTaskID string `json:"source_task_id,omitempty"`
Inherited bool `json:"inherited,omitempty"`
}
func taskNodeDTO(n *db.Node) TaskNodeDTO {
return TaskNodeDTO{
d := TaskNodeDTO{
ID: i64s(n.ID),
Type: n.Kind,
Payload: rawString(n.Payload),
@@ -189,6 +170,11 @@ func taskNodeDTO(n *db.Node) TaskNodeDTO {
Origin: n.Origin,
TS: rfc3339(n.CreatedAt),
}
if n.SourceTaskID > 0 {
d.SourceTaskID = i64s(n.SourceTaskID)
}
d.Inherited = n.Inherited
return d
}
// GoalDTO is a goal node with its payload unpacked into text/vulnclass — the shape
@@ -263,6 +249,27 @@ func edgeDTOs(in []db.Edge) []EdgeDTO {
return out
}
// ---- Coverage asset references ----
type CoverageAssetRefDTO struct {
ID int64 `json:"id"`
Kind string `json:"kind"`
State string `json:"state"`
Summary string `json:"summary"`
SourceTaskID string `json:"source_task_id,omitempty"`
Inherited bool `json:"inherited,omitempty"`
}
func coverageAssetRefDTO(ref db.AssetRef) CoverageAssetRefDTO {
out := CoverageAssetRefDTO{
ID: ref.ID, Kind: ref.Kind, State: ref.State, Summary: ref.Summary, Inherited: ref.Inherited,
}
if ref.SourceTaskID > 0 {
out.SourceTaskID = i64s(ref.SourceTaskID)
}
return out
}
// ---- Finding (frontend "Finding") ----
type FindingDTO struct {
@@ -280,6 +287,8 @@ type FindingDTO struct {
ParamID string `json:"param_id,omitempty"`
TaskID string `json:"task_id,omitempty"`
TaskDescription string `json:"task_description,omitempty"`
SourceTaskID string `json:"source_task_id,omitempty"`
Inherited bool `json:"inherited,omitempty"`
Assets []FindingAssetDTO `json:"assets,omitempty"`
TS string `json:"ts"`
}
@@ -335,7 +344,7 @@ type findingPayload struct {
func findingDTO(n *db.Node) FindingDTO {
var p findingPayload
_ = json.Unmarshal(n.Payload, &p)
return FindingDTO{
d := FindingDTO{
ID: i64s(n.ID),
VulnClass: p.VulnClass,
Name: p.Name,
@@ -345,6 +354,11 @@ func findingDTO(n *db.Node) FindingDTO {
Evidence: rawString(p.Evidence),
TS: rfc3339(n.CreatedAt),
}
if n.SourceTaskID > 0 {
d.SourceTaskID = i64s(n.SourceTaskID)
}
d.Inherited = n.Inherited
return d
}
// findingDTOsForTask converts a task's finding nodes to DTOs, stamping each with
@@ -354,11 +368,15 @@ func findingDTO(n *db.Node) FindingDTO {
// nodes with no row keep the 'pending' default and no finding_id (not editable).
// assets pre-resolves the anchored asset rows for label rendering.
func findingDTOsForTask(t *Task, in []*db.Node, meta map[int64]db.FindingMeta, assets map[int64]*db.Asset) []FindingDTO {
return findingDTOsForOwner(t.ID, t.Description, in, meta, assets)
}
func findingDTOsForOwner(taskID, description string, in []*db.Node, meta map[int64]db.FindingMeta, assets map[int64]*db.Asset) []FindingDTO {
out := make([]FindingDTO, 0, len(in))
for _, n := range in {
d := findingDTO(n)
d.TaskID = t.ID
d.TaskDescription = t.Description
d.TaskID = taskID
d.TaskDescription = description
if m, ok := meta[n.ID]; ok {
d.FindingID = i64s(m.ID)
d.Status = m.Status
@@ -414,16 +432,19 @@ func findingFromDB(f *db.DBFinding, assets map[int64]*db.Asset) FindingDTO {
// ---- Activity (frontend "Activity") ----
type ActivityDTO struct {
Seq int64 `json:"seq"` // db Activity.ID
IntentID string `json:"intent_id,omitempty"` // db NodeID
Worker string `json:"worker"`
TS string `json:"ts"` // db CreatedAt
Kind string `json:"kind"`
Tool string `json:"tool,omitempty"`
ToolUseID string `json:"tool_use_id,omitempty"`
IsError bool `json:"is_error"`
Summary string `json:"summary"`
Detail string `json:"detail,omitempty"`
Seq int64 `json:"seq"` // db Activity.ID
IntentID string `json:"intent_id,omitempty"` // db NodeID
Worker string `json:"worker"`
TS string `json:"ts"` // db CreatedAt
Kind string `json:"kind"`
Tool string `json:"tool,omitempty"`
ToolUseID string `json:"tool_use_id,omitempty"`
IsError bool `json:"is_error"`
Summary string `json:"summary"`
Detail string `json:"detail,omitempty"`
Metadata json.RawMessage `json:"metadata,omitempty"`
SourceTaskID string `json:"source_task_id,omitempty"`
Inherited bool `json:"inherited,omitempty"`
// 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"`
@@ -436,7 +457,7 @@ func activityDTO(a db.Activity) ActivityDTO {
if a.NodeID != nil {
intent = i64s(*a.NodeID)
}
return ActivityDTO{
d := ActivityDTO{
Seq: a.ID,
IntentID: intent,
Worker: a.Worker,
@@ -447,11 +468,17 @@ func activityDTO(a db.Activity) ActivityDTO {
IsError: a.IsError,
Summary: a.Summary,
Detail: a.Detail, // list endpoint leaves this empty (lazy)
Metadata: a.Metadata,
InputTokens: a.InputTokens,
OutputTokens: a.OutputTokens,
CacheReadTokens: a.CacheReadTokens,
CacheWriteTokens: a.CacheWriteTokens,
}
if a.SourceTaskID > 0 {
d.SourceTaskID = i64s(a.SourceTaskID)
}
d.Inherited = a.Inherited
return d
}
func activityDTOs(in []db.Activity) []ActivityDTO {
@@ -528,8 +555,13 @@ type LLMProfileDTO struct {
RatePerSecond float64 `json:"rate_per_second"`
RatePerMinute float64 `json:"rate_per_minute"`
ContextWindowK int `json:"context_window_k"`
ThinkingType string `json:"thinking_type"`
ReasoningEffort string `json:"reasoning_effort"`
IsDefault bool `json:"is_default"`
// 轮询(故障转移)参数:priority 越大越先被选中(激活配置恒为链首);
// pool_exclude=true 则不作为故障转移目标,但仍可被 agent/任务显式绑定。
Priority int `json:"priority"`
PoolExclude bool `json:"pool_exclude"`
}
func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO {
@@ -544,8 +576,11 @@ func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO {
RatePerSecond: p.RatePerSecond,
RatePerMinute: p.RatePerMinute,
ContextWindowK: p.ContextWindowK,
ThinkingType: p.ThinkingType,
ReasoningEffort: p.ReasoningEffort,
IsDefault: p.IsDefault,
Priority: p.Priority,
PoolExclude: p.PoolExclude,
}
}
+986 -190
View File
File diff suppressed because it is too large. Load diff
+268 -14
View File
@@ -5,6 +5,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
@@ -66,14 +67,135 @@ func decode(r *http.Request, v any) error { return json.NewDecoder(r.Body).Decod
// ---------- tasks (delete) ----------
const taskDeleteDrainTimeout = 10 * time.Second
func canonicalTaskID(raw string) (string, bool) {
n, err := strconv.ParseInt(raw, 10, 64)
if err != nil || n <= 0 {
return "", false
}
return strconv.FormatInt(n, 10), true
}
// beginTaskDelete serializes the delete barrier with pause, resume, admission,
// and FIFO reconciliation. Every lifecycle path rechecks deleting after taking
// concMu, so none can commit a contradictory state once this returns.
func (s *Server) beginTaskDelete(taskID string) bool {
s.concMu.Lock()
defer s.concMu.Unlock()
if s.engine.IsDeleting(taskID) {
return false
}
return s.engine.BeginDelete(taskID)
}
// abortTaskDelete restores the execution barrier from the committed task row,
// not from the in-memory state observed before deletion began. Keeping the task
// paused on an unreadable row is conservative: a later explicit resume can
// safely recover it without allowing work to escape an uncertain delete.
func (s *Server) abortTaskDelete(taskID string) {
s.concMu.Lock()
defer s.concMu.Unlock()
keepPaused := true
if id, err := strconv.ParseInt(taskID, 10, 64); err == nil && s.m != nil && s.m.pg != nil {
if persisted, getErr := s.m.pg.GetTask(id); getErr == nil && persisted != nil {
keepPaused = persisted.Paused || persisted.Queued
} else if task, ok := s.m.Task(taskID); ok {
state := task.lifecycleSnapshot()
keepPaused = state.Paused || state.Queued
if getErr != nil {
log.Printf("[task-delete] task %s 读取持久状态失败,使用内存状态恢复屏障: %v", taskID, getErr)
}
} else if getErr == nil {
// The request targeted a task that does not exist. Do not retain a
// synthetic pause entry after releasing its temporary delete barrier.
keepPaused = false
}
}
s.engine.AbortDelete(taskID, keepPaused)
}
func (s *Server) pgDeleteTask(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
s.engine.Pause(id) // cancel running loops before removing
if err := s.m.DeleteTask(id); err != nil {
id, ok := canonicalTaskID(r.PathValue("id"))
if !ok {
writeErr(w, http.StatusBadRequest, "任务 id 无效")
return
}
var opts DeleteTaskOptions
if err := decode(r, &opts); err != nil && err != io.EOF {
writeErr(w, 400, "invalid JSON: "+err.Error())
return
}
if !s.beginTaskDelete(id) {
writeErr(w, http.StatusConflict, "任务正在删除")
return
}
deleted := false
defer func() {
if !deleted {
s.abortTaskDelete(id)
}
}()
// Main-agent runs use a separate context from planner/workers. Cancel it, then
// wait until both execution domains have returned before removing transcripts.
s.cancelTaskChat(id, agent.AbortTaskDeleted)
drainCtx, cancelDrain := context.WithTimeout(r.Context(), taskDeleteDrainTimeout)
defer cancelDrain()
if err := s.waitTaskQuiescent(drainCtx, id); err != nil {
writeErr(w, http.StatusConflict, "任务仍有运行中的 Agent,删除已取消")
return
}
result, err := s.m.DeleteTask(id, opts)
if err != nil {
var committed *taskDeleteCommittedError
if errors.As(err, &committed) {
// PostgreSQL is already gone. Complete runtime teardown and return the
// auditable counts together with the post-commit cleanup warning.
s.engine.StopTask(id)
s.taskAgentMu.Lock()
delete(s.taskAgents, id)
s.taskAgentMu.Unlock()
deleted = true
writeCommittedTaskDelete(w, result, err)
return
}
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"deleted": id})
// Manager has removed the task from the registry, so no new API operation can
// resolve it. Now stop and join every task-owned Engine goroutine and clear its
// lifecycle maps before releasing the delete barrier.
s.engine.StopTask(id)
s.taskAgentMu.Lock()
delete(s.taskAgents, id)
s.taskAgentMu.Unlock()
deleted = true
writeJSON(w, 200, result)
}
func writeCommittedTaskDelete(w http.ResponseWriter, result DeleteTaskResult, err error) {
result.CleanupWarning = err.Error()
writeJSON(w, http.StatusOK, result)
}
func (s *Server) waitTaskQuiescent(ctx context.Context, taskID string) error {
ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop()
for {
s.chatMu.Lock()
chatBusy := s.chatBusy[taskID]
s.chatMu.Unlock()
if s.engine.inflightCount(taskID) == 0 && !chatBusy {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
}
}
// ---------- agents ----------
@@ -305,6 +427,9 @@ func (s *Server) pgSaveAgentConfig(w http.ResponseWriter, r *http.Request) {
if profileChanged {
s.invalidateProfileAgents()
}
// Task bundles capture operational Agent settings (turn/time budgets, tools,
// web search). Rebuild them even when no global profile is configured.
s.invalidateTaskAgents()
// rebuild the live agents so the new max_turns/run_seconds/binding apply without a restart.
s.cfgMu.Lock()
cfg, on := s.llmCfg, s.llmOn
@@ -349,8 +474,8 @@ func (s *Server) pgGetAgent(w http.ResponseWriter, r *http.Request) {
}
writeJSON(w, 200, map[string]any{
"agent": agentDTO(a), "prompt": cur, "variables": vars, "versions": vers,
"visibility": map[string]any{"mcp": mcp, "skill": sk},
"llm_profiles": llmProfiles, // 可绑定的 LLM 配置候选
"visibility": map[string]any{"mcp": mcp, "skill": sk},
"llm_profiles": llmProfiles, // 可绑定的 LLM 配置候选
"wrapup_prompt": a.WrapupPrompt, // 已保存的收尾提示词(空=用内置默认)
"wrapup_default": agent.WrapupDefault(a.Key), // 内置默认(供占位/恢复默认)
@@ -810,6 +935,12 @@ type skillFileNode struct {
Compatibility string `json:"compatibility,omitempty"`
MCPs []string `json:"mcps,omitempty"`
Files []string `json:"files"`
// Usage ledger (db/skill_usage.go). Calls is 0 and LastUsed nil for a skill that
// was never invoked — the ledger only has rows for skills agents actually loaded.
Calls int `json:"calls"`
Tasks int `json:"tasks"`
Agents []string `json:"usage_agents"`
LastUsed *time.Time `json:"last_used,omitempty"`
}
func (s *Server) fsListSkills(w http.ResponseWriter, r *http.Request) {
@@ -828,13 +959,28 @@ func (s *Server) fsListSkills(w http.ResponseWriter, r *http.Request) {
writeErr(w, 500, err.Error())
return
}
// Usage is best-effort decoration: a ledger read failure leaves the counts at
// zero rather than failing the skill list itself.
statBySkill := map[string]db.SkillStat{}
if s.m.pg != nil {
stats, err := s.m.pg.SkillStats()
if err != nil {
log.Printf("[skills] 读取调用统计失败: %v", err)
}
for _, st := range stats {
statBySkill[st.Skill] = st
}
}
nodes := []skillFileNode{}
for _, e := range entries {
if !e.IsDir() {
continue
}
dirName := e.Name()
node := skillFileNode{Name: dirName}
node := skillFileNode{Name: dirName, Agents: []string{}}
if st, ok := statBySkill[dirName]; ok {
node.Calls, node.Tasks, node.Agents, node.LastUsed = st.Calls, st.Tasks, st.Agents, st.LastUsed
}
if meta, ok := metaByDir[dirName]; ok {
node.Description = meta.Description
node.License = meta.License
@@ -850,6 +996,42 @@ func (s *Server) fsListSkills(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"skills": nodes})
}
// fsSkillUsage returns one skill's recent invocations, newest first (detail panel).
func (s *Server) fsSkillUsage(w http.ResponseWriter, r *http.Request) {
pg := s.pg(w)
if pg == nil {
return
}
name := r.PathValue("name")
if !validSkillName(name) {
writeErr(w, 400, "非法 skill 名")
return
}
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
calls, err := pg.RecentSkillCalls(name, limit)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"calls": calls})
}
// fsMissingSkills lists skill names agents asked for that do not exist — the gap
// list, i.e. which procedures are worth writing next.
func (s *Server) fsMissingSkills(w http.ResponseWriter, r *http.Request) {
pg := s.pg(w)
if pg == nil {
return
}
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
missing, err := pg.MissingSkillStats(limit)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"missing": missing})
}
// cleanStrs trims each string and drops empties.
func cleanStrs(in []string) []string {
out := make([]string, 0, len(in))
@@ -1560,12 +1742,14 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) {
writeErr(w, 500, err.Error())
return
}
// Editing a profile rebuilds any task pinned to it on its next round.
// Editing a profile rebuilds any task pinned to it on its next round. Reapply
// the active profile too: the global fallback and explicit task chains must
// repopulate the same provider-cache entry and therefore share one limiter.
s.invalidateProfileAgents()
// If we just edited the active profile, hot-apply so changes take effect now.
if act, _ := pg.ActiveProfile(); act != nil && act.ID == id {
s.reapplyActiveProfile()
}
// Hot-apply. Not just when the ACTIVE profile changed: with failover on, any
// profile's key/model/priority/pool_exclude edit reshapes the chain, so the
// engine's provider has to be rebuilt either way.
s.reapplyActiveProfile()
writeJSON(w, 200, map[string]any{"id": id})
}
@@ -1575,14 +1759,50 @@ func (s *Server) pgDeleteProfile(w http.ResponseWriter, r *http.Request) {
return
}
id, _ := pathInt(r, "id")
if err := pg.DeleteProfile(id); err != nil {
writeErr(w, 500, err.Error())
if err := pg.DeleteProfileContext(r.Context(), id); err != nil {
switch {
case errors.Is(err, db.ErrActiveLLMProfileDelete):
writeErr(w, 409, "当前激活的 LLM 配置不能删除,请先激活其他配置")
case errors.Is(err, db.ErrLLMProfileReferencesChanged):
writeErr(w, 409, "LLM 配置正在被任务或会话修改,请重试")
case errors.Is(err, context.DeadlineExceeded):
writeErr(w, 409, "等待 LLM 配置引用释放超时,请重试")
case errors.Is(err, db.ErrLLMProfileNotFound):
writeErr(w, 404, err.Error())
default:
writeErr(w, 500, err.Error())
}
return
}
s.invalidateProfileAgents() // drop cached agents for the removed profile
s.llmHealth.Reset(id) // its breaker state is meaningless now (row is FK-cascaded away)
s.restoreTasksAfterProfileDelete(pg)
// Cache invalidation also removed the active profile's shared provider entry.
// Reapply after task-state sync so the wake-up observes the post-delete chain.
s.reapplyActiveProfile()
writeJSON(w, 200, map[string]any{"deleted": id})
}
func (s *Server) restoreTasksAfterProfileDelete(pg *db.DB) {
for _, task := range s.m.List() {
if taskID, err := strconv.ParseInt(task.ID, 10, 64); err == nil {
if pt, err := pg.GetTask(taskID); err == nil && pt != nil {
s.syncTaskLLMState(pt)
// Deleting the active entry may advance the task to a ready successor,
// or clear the explicit chain and restore its Agent/global fallback.
// Resume only intents that were blocked by the exhausted chain; a
// paused task remains paused and terminal tasks remain immutable.
if !isTerminalStatus(task.lifecycleSnapshot().Status) && s.taskRuntimeAvailable(task, "planner", "worker") {
if _, reopenErr := task.Store.ReopenIntentsByBlockedReason(db.IntentBlockedLLMQuota); reopenErr != nil {
log.Printf("[llm-profile] task %s reopen quota-blocked intents after profile delete: %v", task.ID, reopenErr)
}
}
task.Notify()
}
}
}
}
func (s *Server) pgActivateProfile(w http.ResponseWriter, r *http.Request) {
pg := s.pg(w)
if pg == nil {
@@ -1604,6 +1824,40 @@ func (s *Server) pgActivateProfile(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"ok": true})
}
// pgLLMPoolStatus reports the failover ("轮询") switches, the resolved chain order
// and every profile's circuit-breaker state — what the LLM page renders as the
// "轮询顺序" strip and the per-card health badges.
func (s *Server) pgLLMPoolStatus(w http.ResponseWriter, r *http.Request) {
if s.pg(w) == nil {
return
}
writeJSON(w, 200, s.llmPoolStatus())
}
// pgLLMPoolReset clears a tripped profile's circuit breaker so the next call
// tries it again immediately ("立即恢复"). id=0 clears every profile.
func (s *Server) pgLLMPoolReset(w http.ResponseWriter, r *http.Request) {
pg := s.pg(w)
if pg == nil {
return
}
var body struct {
ID int64 `json:"id"` // 0 / omitted = all
}
if err := decode(r, &body); err != nil {
writeErr(w, 400, err.Error())
return
}
if body.ID > 0 {
s.llmHealth.Reset(body.ID)
} else {
for id := range s.llmHealth.Snapshot() {
s.llmHealth.Reset(id)
}
}
writeJSON(w, 200, s.llmPoolStatus())
}
// pgListModels fetches available models from the provider's API endpoint.
// Supports OpenAI-format (GET /models) and Anthropic-format (GET /v1/models); for
// Anthropic-compatible third parties (e.g. DeepSeek) whose model list lives only on