dela
0335d572de
ci / go (push) Waiting to run
ci / go-db (agent) (push) Waiting to run
ci / go-db (config) (push) Waiting to run
ci / go-db (db) (push) Waiting to run
ci / go-db (evidence) (push) Waiting to run
ci / go-db (llmrec) (push) Waiting to run
ci / go-db (server) (push) Waiting to run
detections / detections (push) Waiting to run
web / web (push) Waiting to run
docs / links (push) Canceled after 0s
3994 lines
150 KiB
Go
3994 lines
150 KiB
Go
package server
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net"
|
||
"net/http"
|
||
"net/url"
|
||
"os"
|
||
"path/filepath"
|
||
"regexp"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/Autumn-27/artex/agent"
|
||
"github.com/Autumn-27/artex/db"
|
||
"github.com/Autumn-27/artex/intercept"
|
||
"github.com/Autumn-27/artex/llmpool"
|
||
"github.com/Autumn-27/artex/llmrec"
|
||
"github.com/Autumn-27/artex/report"
|
||
"github.com/Autumn-27/artex/traffic"
|
||
"github.com/Autumn-27/norma/llm"
|
||
actool "github.com/Autumn-27/norma/tool"
|
||
"github.com/Autumn-27/norma/transcript"
|
||
)
|
||
|
||
// BuildVersion is the backend application version, injected from cmd/artex at
|
||
// startup (which in turn gets it from -ldflags "-X main.version=<tag>").
|
||
// Defaults to "dev" for local builds. Exposed to the frontend via GET /api/health.
|
||
var BuildVersion = "dev"
|
||
|
||
// 사용자에게 노출되는 writeErr 응답 문구(한국어). 에이전트에 전달되는 프롬프트·도구
|
||
// 설명·payload·로그는 원문을 보존하고, HTTP 핸들러가 writeErr/fmt.Errorf 로 내보내는
|
||
// 최종 사용자 응답만 한국어로 둔다. (F3b, 번역어는 완료 파일·web i18n 과 동일 표기)
|
||
const (
|
||
// 의도 제어·재실행 (controlIntent / rerunIntent / rerunBlocked)
|
||
errTaskDeletingIntentControl = "작업을 삭제하는 중이라 의도를 제어할 수 없습니다"
|
||
errTaskDeletingIntentRerun = "작업을 삭제하는 중이라 의도를 재실행할 수 없습니다"
|
||
errIntentNotRerunnable = "재실행할 수 있는 상태가 아닙니다(blocked·exhausted·stopped 만 재실행할 수 있습니다)"
|
||
|
||
// 작업 생성 (createTask) — 소스 작업·기업·분류·작업 수준 가로채기 규칙 검증
|
||
errCreateTaskSourceLimit = "소스 작업은 최대 %d개까지 연결할 수 있습니다"
|
||
errCreateTaskSourceInvalid = "소스 작업 id가 올바르지 않거나 중복되었습니다"
|
||
errCreateTaskSourceNotFound = "소스 작업 #%d를 찾을 수 없습니다"
|
||
errCreateTaskCompanyLimit = "연결 기업이 올바르지 않습니다: 유효한 기업은 최대 %d개까지 선택할 수 있습니다"
|
||
errCreateTaskInterceptRules = "작업 수준 가로채기 규칙이 올바르지 않습니다: "
|
||
errCreateTaskCategoryInvalid = "작업 분류를 찾을 수 없거나 올바르지 않습니다"
|
||
errCreateTaskCompanyInvalid = "연결 기업을 찾을 수 없거나 올바르지 않습니다"
|
||
|
||
// LLM 설정 검증 (validateTaskProfileIDs → createTask/updateTaskLLMProfiles writeErr)
|
||
errLLMProfileInvalid = "LLM 설정 id가 올바르지 않거나 중복되었습니다"
|
||
errLLMProfileNotFound = "LLM 설정 #%d를 찾을 수 없거나 API Key가 설정되어 있지 않습니다"
|
||
|
||
// 자산 저장소·커버리지 (taskCoverage / taskCoverageGraph / taskAssetRefs …)
|
||
errAssetStoreDisabled = "자산 저장소가 활성화되어 있지 않습니다"
|
||
errAssetIDRequired = "asset_id가 필요합니다"
|
||
|
||
// 파이썬 인터프리터 탐지 (pgDetectPython)
|
||
errPythonNotDetected = "python을 찾을 수 없습니다(python3·python 모두 PATH에 없습니다)"
|
||
|
||
// 알림 전역 설정 (링크 기준 주소·요약 주기 — web notify.global 라벨과 동일 표기)
|
||
errNotifyBaseURLScheme = "링크 기준 주소는 http:// 또는 https://로 시작해야 합니다"
|
||
errNotifyDigestRange = "요약 주기는 1~1440분 사이여야 합니다"
|
||
|
||
// 메인 에이전트 대화·세션 (newMainSession / chat)
|
||
errTaskDeletingNewSession = "작업을 삭제하는 중이라 새 세션을 만들 수 없습니다"
|
||
errTaskDeletingNewMessage = "작업을 삭제하는 중이라 새 메시지를 보낼 수 없습니다"
|
||
errMainAgentBusy = "메인 에이전트가 이전 메시지를 처리하는 중입니다. 잠시 후 다시 시도해 주세요"
|
||
)
|
||
|
||
// 사용자에게 노출되지만 writeErr 를 거치지 않는 응답 문구(한국어). writeJSON 의 error
|
||
// 필드나 메서드 return 으로 화면에 그대로 뜨므로 위 블록과 성격은 같되 경로가 다르다.
|
||
// nav 경로는 web i18n 라벨(nav.system="시스템", llmConfig="LLM 설정")과 동일 표기. (F34)
|
||
const (
|
||
// 채팅 LLM 부재 사유 (chatUnavailableReason → writeErr 503 / 대화 종료 사유)
|
||
errChatNoLLMProfile = "아직 LLM 설정이 없습니다: 시스템 → LLM 설정에서 설정을 하나 추가해 주세요"
|
||
errChatNoActiveLLMProfile = "활성화된 LLM 설정이 없습니다: 시스템 → LLM 설정에서 하나를 활성화하거나, 이 대화에서 사용할 설정을 지정해 주세요"
|
||
errChatLLMNotReady = "LLM이 준비되지 않아 대화할 수 없습니다: 시스템 → LLM 설정에 사용할 수 있고 활성화된 설정이 있는지 확인해 주세요"
|
||
|
||
// LLM 연결 테스트·웹 검색 프로브 (writeJSON {ok:false, error:…}, 설정 화면 점검 결과)
|
||
errLLMTestNoAPIKey = "API Key가 제공되지 않았습니다"
|
||
errWebSearchProbeNoResults = "검색 결과가 0건입니다(요청이 제한되었거나 프록시가 연결되지 않았을 수 있습니다)"
|
||
)
|
||
|
||
// 규칙 모드 채팅(LLM 미설정 시 사람이 직접 조종하는 fallbackChat) 사용자 응답(한국어).
|
||
// reply 는 writeJSON 과 대화 로그 양쪽으로 사용자에게 그대로 노출된다. 입력 트리거 키워드는
|
||
// 기존 영어 별칭(intent/hint)과 같은 방식으로 한국어 별칭(의도/힌트)을 추가로 받는다 —
|
||
// 순수 입력 파싱이고 와이어 포맷이 아니므로 중국어·영어 트리거를 유지한 채 더하기만 한다.
|
||
// 용어는 web i18n·GLOSSARY 와 정합: intent→의도, hint→힌트, finding→취약점, planner→플래너. (F34 d)
|
||
const (
|
||
fallbackIntentInjected = "우선순위 높은 의도를 하나 주입했습니다: "
|
||
fallbackHintRecorded = "힌트를 기록했습니다. 플래너가 다음에 읽습니다: "
|
||
fallbackChatStatus = "(규칙 모드, LLM 미설정) 현재 현황: 자산 %d개, 대기 의도 %d개, 확인된 취약점 %d개.\n사용 가능한 명령: \"의도 ...\"로 의도를 주입하고, \"힌트 ...\"로 플래너에게 힌트를 전달합니다."
|
||
)
|
||
|
||
// Server exposes the ARTEX backend over a JSON HTTP API for the shadcn/ui
|
||
// frontend.
|
||
type Server struct {
|
||
m *Manager
|
||
engine *Engine
|
||
ctx context.Context
|
||
|
||
skillDir string // root directory for skill subdirectories
|
||
jwtKey []byte // HS256 signing key loaded from / generated into dataDir/jwt.key
|
||
|
||
// concMu serializes concurrency-cap decisions (admission + reconcile) so a
|
||
// scheduler tick and an HTTP settings change / task creation can't both count
|
||
// free slots off the same snapshot and over-promote past the limit.
|
||
concMu sync.Mutex
|
||
|
||
cfgMu sync.Mutex
|
||
chatAgent *agent.ChatAgent // conversational runner for the chat page; nil w/o LLM
|
||
// llmProv is the fully-decorated global provider (recorder + failover chain)
|
||
// installed by applyLLM. Task routers use it after an explicit chain is cleared.
|
||
llmProv llm.Provider
|
||
llmDirect llm.Provider // concrete global provider, before any failover pool
|
||
llmCfg agent.Config // current LLM config (key not exposed)
|
||
llmOn bool
|
||
llmProf string // active LLM profile name (for llmrec tagging)
|
||
|
||
// chatBusy guards the per-task main-agent run: the chat handler launches the
|
||
// agent on the server's background ctx (not the request ctx) and returns
|
||
// immediately, so a page reload / proxy timeout can't abort a live run. This
|
||
// map serializes turns per task — one main-agent run at a time per task, so
|
||
// concurrent messages don't corrupt the shared exp<id>-main transcript.
|
||
chatMu sync.Mutex
|
||
chatBusy map[string]bool
|
||
// chatCancel holds the cancel func for each in-flight conversation run (keyed by
|
||
// convBusyKey), so a manual stop can abort JUST that session's agent run. Set/
|
||
// cleared alongside chatBusy under chatMu. Aborting a run does NOT touch the P3
|
||
// trigger queue — the drain goroutine simply proceeds to the next queued fire.
|
||
chatCancel map[string]context.CancelCauseFunc
|
||
|
||
// triggerQ buffers P3 trigger fires PER AGENT. A per-agent "pump" launches runs up
|
||
// to a concurrency limit derived from the agent's策略: serial → limit 1 (+ optional
|
||
// merge); parallel → limit = trigger_max_parallel (0=∞), no merge. triggerActive
|
||
// counts in-flight runs per agent (replaces a boolean drain flag); a run's
|
||
// completion decrements it and re-pumps to fill the freed slot. triggerCfg caches
|
||
// the agent's last-read策略 so the pump never queries the DB while holding queueMu.
|
||
// Distinct agents always run concurrently. Queue is in-memory (matches chatBusy); a
|
||
// restart drops pending fires — the scheduler re-fires from watermarks next tick.
|
||
queueMu sync.Mutex
|
||
triggerQ map[string][]triggeredRun
|
||
triggerActive map[string]int
|
||
triggerCfg map[string]triggerBehavior
|
||
|
||
// profChatAgents caches a per-profile ChatAgent (chat page), keyed by profile id.
|
||
// Built lazily on first use; invalidated when any profile is saved/activated/deleted
|
||
// so edits take effect.
|
||
profMu sync.Mutex
|
||
profChatAgents map[int64]*agent.ChatAgent // per-profile ChatAgent cache (chat page)
|
||
|
||
// provByProfile caches ONE provider per LLM profile id so every agent bound or
|
||
// pinned to the same profile shares a single provider instance — hence one rate
|
||
// limiter. applyLLM also resolves the persisted global-active profile through this
|
||
// cache, so global fallback and task chains do not accidentally double the
|
||
// configured request rate. Cleared on profile edits, then repopulated by active reapply.
|
||
provCacheMu sync.Mutex
|
||
provByProfile map[int64]*provEntry
|
||
provCacheGen uint64
|
||
|
||
// llmHealth is the process-wide circuit-breaker state for LLM failover (轮询).
|
||
// It deliberately lives OUTSIDE the provider caches: rebuilding the chain
|
||
// (saving an unrelated profile, flipping a setting) must not erase what we
|
||
// learned about which backends are out of credit / rate-limited.
|
||
llmHealth *llmpool.Registry
|
||
|
||
// Explicit task chains use one task-scoped dynamic provider shared by goals,
|
||
// planner, workers, and the main agent. Bundles are stable; their runtime reads
|
||
// the persisted current profile at every new LLM call.
|
||
taskAgentMu sync.Mutex
|
||
taskAgents map[string]*taskAgentBundle
|
||
|
||
// Cold task archives run through one persistent FIFO worker. The buffered wake
|
||
// channel coalesces enqueue bursts; the database remains the source of truth.
|
||
archiveWake chan struct{}
|
||
archiveWG sync.WaitGroup
|
||
side *sideQuestionState
|
||
}
|
||
|
||
// provEntry is a cached provider + its config for one LLM profile id.
|
||
type provEntry struct {
|
||
prov llm.Provider
|
||
cfg agent.Config
|
||
}
|
||
|
||
// triggeredRun is one queued P3 trigger fire awaiting its turn for an agent.
|
||
// taskID + mergeable let the drainer coalesce several event triggers (finding/goal)
|
||
// from the SAME task into one conversation before it starts (interval fires don't merge).
|
||
type triggeredRun struct {
|
||
agentKey string
|
||
title string
|
||
message string // 事件正文(触发语 + 工具/入参/返回等);不含任务描述/目标头
|
||
taskID int64 // source task for finding/goal triggers; 0 for interval/none
|
||
taskDesc string // 任务描述(任务级,同任务相同);合并时只渲染一次
|
||
taskGoal string // 任务目标(任务级,同任务相同);合并时只渲染一次
|
||
mergeable bool // true for finding/goal event triggers (merge by taskID)
|
||
}
|
||
|
||
func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDir string) *Server {
|
||
key, err := loadOrCreateJWTKey(keyDir, dataDir)
|
||
if err != nil {
|
||
log.Fatalf("[auth] JWT key: %v", err)
|
||
}
|
||
s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, jwtKey: key, chatBusy: map[string]bool{},
|
||
chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{},
|
||
triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{},
|
||
profChatAgents: map[int64]*agent.ChatAgent{},
|
||
provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg),
|
||
taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1)}
|
||
s.initSideQuestions()
|
||
// 熔断阈值/冷却是失败路径上的热参数，启动时把全局重试策略推给 Registry 一次；
|
||
// 之后每次保存策略再推一次（saveLLMRetryPolicy）。
|
||
s.applyRetryPolicy()
|
||
// Every task uses a stable task router. An empty explicit chain is resolved by
|
||
// that router through Agent bindings and then the global provider, so adding a
|
||
// first chain to a running task takes effect on its very next LLM call.
|
||
s.engine.SetAuthoritativeAgentResolver(func(t *Task) (*agent.Planner, *agent.Worker) {
|
||
if !s.taskRuntimeAvailable(t, "planner", "worker") {
|
||
return nil, nil
|
||
}
|
||
b := s.agentsForTask(t)
|
||
return b.pl, b.wk
|
||
})
|
||
// Global readiness (Ready()/llm_configured): a global active LLM provider is
|
||
// installed. Task-level runnability is separate (ReadyFor → the resolver above).
|
||
s.engine.SetReadiness(func() bool {
|
||
s.cfgMu.Lock()
|
||
defer s.cfgMu.Unlock()
|
||
return s.llmOn
|
||
})
|
||
// Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no
|
||
// override row, agents keep their built-in defaults — behavior is unchanged.
|
||
if m.pg != nil {
|
||
agent.PromptOverride = func(key string) (string, bool) {
|
||
a, err := m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil {
|
||
return "", false
|
||
}
|
||
t, err := m.pg.CurrentPrompt(a.ID)
|
||
if err != nil || t == "" {
|
||
return "", false
|
||
}
|
||
return t, true
|
||
}
|
||
// Wire DB-stored wrap-up (settlement) prompts. Empty column → built-in default.
|
||
agent.WrapupOverride = func(key string) (string, bool) {
|
||
a, err := m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil || a.WrapupPrompt == "" {
|
||
return "", false
|
||
}
|
||
return a.WrapupPrompt, true
|
||
}
|
||
// Wire DB-stored wrap-up turn budgets. 0 / missing → built-in per-agent default.
|
||
agent.WrapupMaxTurnsOverride = func(key string) (int, bool) {
|
||
a, err := m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil || a.WrapupMaxTurns <= 0 {
|
||
return 0, false
|
||
}
|
||
return a.WrapupMaxTurns, true
|
||
}
|
||
// Wire DB-stored TASK-TIMEOUT wrap-up prompt/turns (worker/planner). Empty/0 → default.
|
||
agent.WrapupTaskTimeoutOverride = func(key string) (string, bool) {
|
||
a, err := m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil || a.TaskTimeoutWrapupPrompt == "" {
|
||
return "", false
|
||
}
|
||
return a.TaskTimeoutWrapupPrompt, true
|
||
}
|
||
agent.WrapupTaskTimeoutTurnsOverride = func(key string) (int, bool) {
|
||
a, err := m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil || a.TaskTimeoutWrapupMaxTurns <= 0 {
|
||
return 0, false
|
||
}
|
||
return a.TaskTimeoutWrapupMaxTurns, true
|
||
}
|
||
wireAgentAugment(m.pg, s.skillDir, s.hostTools) // 可见 skills/MCP + 流量/编排 host 工具装配进 agent 工具集
|
||
domainReg := buildDomainReg(m.Assets())
|
||
wireTools(m.pg, domainReg) // 内置工具表：按 agent 过滤 + 覆盖描述/schema + 注入默认值
|
||
seedPrompts(m.pg) // 内置 agent 默认提示词正文播种进 agent_prompts(仅空时)
|
||
s.seedOrchestrationTools() // P2 跨任务编排工具 seed 进 tools 表(可按 agent 绑定)
|
||
if err := s.seedFindingRetester(); err != nil {
|
||
log.Printf("[retester] seed: %v", err)
|
||
}
|
||
go s.evidenceStore().RunGC(s.ctx)
|
||
s.seedPythonInterpreter() // 自定义脚本工具:开机检测 python 解释器入库(仅空时)
|
||
go newScheduler(s).Run(s.ctx) // P3 触发器调度(定时/finding/目标事件),仅自定义 agent
|
||
// 漏洞 IM 推送投递引擎。与 Scheduler 并列但独立：推送的实时性要求(3s)
|
||
// 与触发器的业务节奏不同，且两者失败互不牵连——推送卡住不该影响 agent 触发。
|
||
// 통합 테스트는 stepRealtime·stepDigest 를 직접 호출해 분배를 결정론적으로
|
||
// 검증하므로, 이 환경 변수로 백그라운드 루프만 끌 수 있다(미설정이 기본).
|
||
if os.Getenv(notifyBackgroundDisabledEnv) == "" {
|
||
go newNotifier(s).Run(s.ctx)
|
||
} else {
|
||
// 핵심 기능(취약점 IM 알림 전송)을 끄는 분기라 시작 로그를 남긴다. 이 변수는
|
||
// 통합 테스트 전용이므로, 운영에서 켜져 있으면 실수나 환경 상속을 의심할 단서가 된다.
|
||
log.Printf("[notify] 백그라운드 알림 전송 루프가 %s 로 꺼졌습니다(테스트 전용) — 취약점 IM 알림이 전송되지 않습니다", notifyBackgroundDisabledEnv)
|
||
}
|
||
// Fill the tool cache for any enabled MCP that has none yet (notably the
|
||
// seeded browser MCP on first run). Async so it never blocks startup.
|
||
go s.discoverEmptyMCPsOnStartup()
|
||
logSink.SetDB(ctx, m.pg) // restore last 100 log rows and enable async persistence
|
||
}
|
||
// precedence: persisted DB config > env.
|
||
if cfg, ok := s.loadLLMConfig(); ok {
|
||
if err := s.applyLLM(cfg); err != nil {
|
||
log.Printf("[engine] saved LLM config init failed — engine idle: %v", err)
|
||
} else {
|
||
log.Printf("[engine] LLM configured from DB: %s / %s", cfg.Provider(), cfg.Model)
|
||
}
|
||
} else if cfg, ok := agent.FromEnv(); ok {
|
||
if err := s.applyLLM(cfg); err != nil {
|
||
log.Printf("[engine] env provider init failed — engine idle: %v", err)
|
||
} else {
|
||
log.Printf("[engine] LLM configured from env: %s / %s", cfg.Provider(), cfg.Model)
|
||
}
|
||
} else {
|
||
log.Printf("[engine] no LLM provider configured — engine idle until set via /api/llm or env")
|
||
}
|
||
s.restoreTaskRuntimes()
|
||
go s.reconcileConcurrency()
|
||
s.startTaskArchiveWorker()
|
||
s.wireInterceptReviewer() // LLM 兜底审批:未命中拦截规则的命令交给模型判定
|
||
return s
|
||
}
|
||
|
||
// Restored deadline and worker loops must inherit the same context as new tasks,
|
||
// including the side-question checkpoint publisher installed during startup.
|
||
func (s *Server) restoreTaskRuntimes() {
|
||
m := s.m
|
||
// reload tasks persisted on disk so the task list survives a restart, and
|
||
// restore persisted paused state (so a task paused before restart stays paused).
|
||
for _, t := range m.LoadExisting() {
|
||
lifecycle := t.lifecycleSnapshot()
|
||
// clear stale 'running' intents from a prior crash/restart (no live worker
|
||
// owns them) so they re-claim instead of spinning forever in the UI.
|
||
if n, _ := t.Store.ResetRunningIntents(); n > 0 {
|
||
log.Printf("[engine] task %s 重置 %d 个残留 running 意图为 open", t.ID, n)
|
||
}
|
||
if lifecycle.Paused {
|
||
s.engine.Pause(t.ID, agent.AbortPausedOnReload)
|
||
}
|
||
// 任务级超时:为每个未终态、带 timeout 的任务起 deadline 协调器,独立于 planner/worker
|
||
// loop——非活跃任务重启后也能在到点后被收尾(deadline 已过则立即走收尾时序)。
|
||
if !isTerminalStatus(lifecycle.Status) {
|
||
s.engine.startDeadlineCoordinator(s.ctx, t)
|
||
}
|
||
}
|
||
// Restore every task that had already been admitted before shutdown. Starting
|
||
// only the active UI task left other non-queued tasks counted as concurrency
|
||
// occupants without live loops, which could permanently block the persistent
|
||
// FIFO. Paused loops remain idle; queued tasks are admitted below as slots allow.
|
||
for _, t := range m.List() {
|
||
lifecycle := t.lifecycleSnapshot()
|
||
if !lifecycle.Queued && !isTerminalStatus(lifecycle.Status) {
|
||
s.engine.Run(s.ctx, t)
|
||
}
|
||
}
|
||
}
|
||
|
||
// agentMaxTurns returns the configured max_turns for an agent key (0 = unlimited,
|
||
// also the fallback when no DB or no row).
|
||
func (s *Server) agentMaxTurns(key string) int {
|
||
if s.m.pg == nil {
|
||
return 0
|
||
}
|
||
a, err := s.m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil {
|
||
return 0
|
||
}
|
||
return a.MaxTurns
|
||
}
|
||
|
||
// agentRunSeconds returns the configured wall-clock run budget (seconds) for an
|
||
// agent key (0 = unlimited; 1200 fallback when no DB or no row, matching schema).
|
||
func (s *Server) agentRunSeconds(key string) int {
|
||
if s.m.pg == nil {
|
||
return 1200
|
||
}
|
||
a, err := s.m.pg.GetAgentByKey(key)
|
||
if err != nil || a == nil {
|
||
return 1200
|
||
}
|
||
return a.RunSecs
|
||
}
|
||
|
||
// loadLLMConfig reads the active LLM profile from PG (llm_profiles).
|
||
func (s *Server) loadLLMConfig() (agent.Config, bool) {
|
||
p, err := s.m.pg.ActiveProfile()
|
||
if err != nil || p == nil {
|
||
return agent.Config{}, false
|
||
}
|
||
cfg := agent.ConfigFrom(p.Format, p.Model, p.BaseURL, p.APIKey, p.Proxy)
|
||
cfg.RatePerSecond, cfg.RatePerMinute = p.RatePerSecond, p.RatePerMinute
|
||
cfg.ContextWindowK = p.ContextWindowK
|
||
cfg.ThinkingType = p.ThinkingType
|
||
cfg.ReasoningEffort = p.ReasoningEffort
|
||
cfg.Stream = p.Streaming
|
||
cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField
|
||
cfg.SessionHeaderKey = p.SessionHeaderKey
|
||
s.applyProfileRetry(&cfg, p)
|
||
if cfg.APIKey == "" {
|
||
return cfg, false
|
||
}
|
||
s.cfgMu.Lock()
|
||
s.llmProf = p.Name
|
||
s.cfgMu.Unlock()
|
||
return cfg, true
|
||
}
|
||
|
||
// saveLLMConfig persists the LLM config as the active "default" profile in PG.
|
||
func (s *Server) saveLLMConfig(cfg agent.Config) error {
|
||
// cfg.Provider() already returns one of the three valid format strings
|
||
// (anthropic / openai / openai-responses), matching the DB CHECK constraint.
|
||
format := cfg.Provider()
|
||
var id int64
|
||
// 这个 legacy 端点的请求体不含轮询/收发/输出上限参数,故把库里已存的值原样带回 —
|
||
// 否则每次保存都会把 profile 的 priority、pool_exclude、streaming 以及输出上限
|
||
// (max_tokens / max_tokens_field)悄悄重置成零值。
|
||
var priority int
|
||
var poolExclude bool
|
||
streaming := true // 旧库/新建默认流式
|
||
var maxTokens int
|
||
var maxTokensField string
|
||
var sessionHeaderKey string
|
||
var retry db.RetryOverride
|
||
if profs, _ := s.m.pg.ListProfiles(); profs != nil {
|
||
for _, p := range profs {
|
||
if p.Name == "default" {
|
||
id, priority, poolExclude, streaming = p.ID, p.Priority, p.PoolExclude, p.Streaming
|
||
maxTokens, maxTokensField = p.MaxTokens, p.MaxTokensField
|
||
sessionHeaderKey = p.SessionHeaderKey
|
||
retry = p.Retry
|
||
break
|
||
}
|
||
}
|
||
}
|
||
newID, err := s.m.pg.SaveProfile(&db.LLMProfile{
|
||
ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy,
|
||
APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute,
|
||
ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true,
|
||
Priority: priority, PoolExclude: poolExclude, Streaming: streaming,
|
||
MaxTokens: maxTokens, MaxTokensField: maxTokensField, SessionHeaderKey: sessionHeaderKey,
|
||
Retry: retry,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return s.m.pg.SetActiveProfile(newID)
|
||
}
|
||
|
||
// reapplyActiveProfile hot-reloads the engine from the active DB profile so that
|
||
// saving or activating a profile takes effect without a restart. Best-effort:
|
||
// logs on failure and leaves the running engine untouched.
|
||
func (s *Server) reapplyActiveProfile() {
|
||
cfg, ok := s.loadLLMConfig()
|
||
if !ok {
|
||
return
|
||
}
|
||
if err := s.applyLLM(cfg); err != nil {
|
||
log.Printf("[engine] reapply active profile failed: %v", err)
|
||
return
|
||
}
|
||
log.Printf("[engine] LLM reapplied from active profile: %s / %s", cfg.Provider(), cfg.Model)
|
||
}
|
||
|
||
// webSearchFor gates the global web-search opts (backend/key) by an agent's own
|
||
// web_search flag: the backend/key come from the global config, each agent decides on/off.
|
||
func (s *Server) webSearchFor(key string) agent.WebSearchOpts {
|
||
o := s.m.WebSearchOpts()
|
||
if a, err := s.m.pg.GetAgentByKey(key); err != nil || a == nil || !a.WebSearch {
|
||
o.Enabled = false
|
||
}
|
||
return o
|
||
}
|
||
|
||
// nonStreamingResolver returns a resolver capturing a profile's streaming choice.
|
||
// The agents read it per run; a profile change rebuilds the agents (applyLLM),
|
||
// so the captured value is always the one in effect for this build.
|
||
func nonStreamingResolver(cfg agent.Config) func() bool {
|
||
nonStreaming := !cfg.Stream
|
||
return func() bool { return nonStreaming }
|
||
}
|
||
|
||
// maxTokensResolver mirrors nonStreamingResolver for the per-reply output cap.
|
||
func maxTokensResolver(cfg agent.Config) func() int {
|
||
maxTokens := cfg.MaxTokens
|
||
return func() int { return maxTokens }
|
||
}
|
||
|
||
// applyLLM (re)builds the planner/worker/main-agent from cfg and installs them on
|
||
// the running engine as the GLOBAL active pair. Safe to call at runtime (UI configures LLM).
|
||
func (s *Server) applyLLM(cfg agent.Config) error {
|
||
var prov llm.Provider
|
||
// A persisted active profile must use the same cached provider as task chains
|
||
// and Agent bindings, otherwise each path owns a separate rate limiter.
|
||
if active, _ := s.m.pg.ActiveProfile(); active != nil {
|
||
if activeCfg, ok := s.loadProfileConfig(active.ID); ok && activeCfg == cfg {
|
||
prov, _, _ = s.providerForProfile(active.ID)
|
||
}
|
||
}
|
||
if prov == nil {
|
||
var err error
|
||
prov, err = cfg.NewProvider()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// Non-persisted/env configs do not have a profile cache key.
|
||
s.cfgMu.Lock()
|
||
profName := s.llmProf
|
||
s.cfgMu.Unlock()
|
||
prov = llmrec.Wrap(prov, s.m.PG(), cfg.Model, profName, cfg.ThinkingType, cfg.ReasoningEffort, s.m.LLMRecordEnabled)
|
||
prov = bindSideProvider(prov, cfg, 0, profName)
|
||
}
|
||
s.cfgMu.Lock()
|
||
s.llmDirect = prov
|
||
s.cfgMu.Unlock()
|
||
// LLM 轮询(默认关):把激活配置包进故障转移链,当前配置不可用时自动切下一个。
|
||
// 只影响「走全局激活配置」的这条路径——agent 绑定 / 任务 pin 的走 providerForProfile,
|
||
// 默认仍然独占该配置(见 poolForBinding)。关闭或无备选时返回原 provider,行为不变。
|
||
if act, err := s.m.pg.ActiveProfile(); err == nil && act != nil {
|
||
prov = s.poolForActive(act.ID, prov, cfg)
|
||
}
|
||
// No global planner/worker pair: every task runs on its own task-routed pair
|
||
// (agentsForTask), resolved through the engine's authoritative resolver. Global
|
||
// readiness (Ready()/llm_configured) is reported from s.llmOn, set below.
|
||
|
||
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
|
||
win := cfg.CompactionWindow()
|
||
// chat stays the GLOBAL fallback since one ChatAgent serves many agent keys — its
|
||
// per-agent binding is resolved at Chat time (runConversationSync →
|
||
// chatAgentForProfile). The main-agent has no global instance: each task builds its
|
||
// own via agentsForTask (task-routed), so nothing is constructed for it here.
|
||
s.cfgMu.Lock()
|
||
// chat agent serves MANY custom agents by key → it holds the GLOBAL opts
|
||
// (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。
|
||
s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
|
||
s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
|
||
s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
|
||
s.chatAgent.SetGuard(s.chatGuard())
|
||
s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
|
||
s.chatAgent.SetMaxTokens(maxTokensResolver(cfg))
|
||
s.chatAgent.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读)
|
||
s.llmProv = prov
|
||
s.llmCfg = cfg
|
||
s.llmOn = true
|
||
s.cfgMu.Unlock()
|
||
s.invalidateTaskAgents()
|
||
|
||
// wake the active task so a task created while idle starts exploring.
|
||
if t := s.m.ActiveTask(); t != nil {
|
||
t.Notify()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// loadProfileConfig builds an agent.Config from a specific profile id (with its key).
|
||
// ok=false when the profile is missing or has no api key.
|
||
func (s *Server) loadProfileConfig(id int64) (agent.Config, bool) {
|
||
p, err := s.m.pg.ProfileByID(id)
|
||
if err != nil || p == nil {
|
||
return agent.Config{}, false
|
||
}
|
||
cfg := agent.ConfigFrom(p.Format, p.Model, p.BaseURL, p.APIKey, p.Proxy)
|
||
cfg.RatePerSecond, cfg.RatePerMinute = p.RatePerSecond, p.RatePerMinute
|
||
cfg.ContextWindowK = p.ContextWindowK
|
||
cfg.ThinkingType = p.ThinkingType
|
||
cfg.ReasoningEffort = p.ReasoningEffort
|
||
cfg.Stream = p.Streaming
|
||
cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField
|
||
cfg.SessionHeaderKey = p.SessionHeaderKey
|
||
s.applyProfileRetry(&cfg, p)
|
||
if cfg.APIKey == "" {
|
||
return cfg, false
|
||
}
|
||
return cfg, true
|
||
}
|
||
|
||
// effectiveProfileForAgent resolves the LLM profile id an agent should run on, by
|
||
// precedence: agent binding (agents.llm_profile_id) → pin (task/conversation) → nil
|
||
// (caller falls back to the global active profile). A binding to a deleted profile
|
||
// can't happen (FK ON DELETE SET NULL); an otherwise-invalid one is dropped downstream
|
||
// by loadProfileConfig, letting the caller fall back.
|
||
func (s *Server) effectiveProfileForAgent(agentKey string, pinID *int64) *int64 {
|
||
if s.m.pg != nil && agentKey != "" {
|
||
if a, _ := s.m.pg.GetAgentByKey(agentKey); a != nil && a.LLMProfileID != nil {
|
||
return a.LLMProfileID
|
||
}
|
||
}
|
||
return pinID
|
||
}
|
||
|
||
// resolveChatAgent picks the ChatAgent for one conversation: the agent's own
|
||
// binding or this conversation's chosen profile first, the global active config
|
||
// only as a fallback. Both the send precheck and the background runner MUST use
|
||
// this — resolving differently in the two paths is how a conversation that had
|
||
// picked a valid profile still got rejected with "LLM 未配置" when no global
|
||
// config was active.
|
||
func (s *Server) resolveChatAgent(c *db.Conversation) *agent.ChatAgent {
|
||
ca := s.chatAgentRef()
|
||
if eff := s.effectiveProfileForAgent(c.AgentKey, c.LLMProfileID); eff != nil {
|
||
if pa := s.chatAgentForProfile(*eff); pa != nil {
|
||
ca = pa
|
||
}
|
||
}
|
||
return ca
|
||
}
|
||
|
||
// chatUnavailableReason explains why no ChatAgent could be resolved, so the user
|
||
// knows whether to add a config, activate one, or pick one for this conversation
|
||
// — rather than a flat "not configured" that hides which of those it is.
|
||
func (s *Server) chatUnavailableReason() string {
|
||
if s.m.pg != nil {
|
||
if profiles, err := s.m.pg.ListProfiles(); err == nil && len(profiles) == 0 {
|
||
return errChatNoLLMProfile
|
||
}
|
||
if active, err := s.m.pg.ActiveProfile(); err == nil && active == nil {
|
||
return errChatNoActiveLLMProfile
|
||
}
|
||
}
|
||
return errChatLLMNotReady
|
||
}
|
||
|
||
// providerForProfile returns a cached provider+cfg for a profile id, so every agent
|
||
// bound/pinned to the same profile shares one provider instance (one rate limiter).
|
||
// ok=false when the profile is missing/invalid → caller falls back to the global pair.
|
||
func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool) {
|
||
s.provCacheMu.Lock()
|
||
if e := s.provByProfile[id]; e != nil {
|
||
s.provCacheMu.Unlock()
|
||
return e.prov, e.cfg, true
|
||
}
|
||
generation := s.provCacheGen
|
||
s.provCacheMu.Unlock()
|
||
cfg, ok := s.loadProfileConfig(id)
|
||
if !ok {
|
||
return nil, agent.Config{}, false
|
||
}
|
||
prov, err := cfg.NewProvider()
|
||
if err != nil {
|
||
log.Printf("[engine] build provider for LLM profile %d failed: %v", id, err)
|
||
return nil, agent.Config{}, false
|
||
}
|
||
// Wrap with recorder, tagged with this profile's name.
|
||
if p, _ := s.m.pg.ProfileByID(id); p != nil {
|
||
prov = llmrec.Wrap(prov, s.m.PG(), cfg.Model, p.Name, cfg.ThinkingType, cfg.ReasoningEffort, s.m.LLMRecordEnabled)
|
||
prov = bindSideProvider(prov, cfg, id, p.Name)
|
||
}
|
||
s.provCacheMu.Lock()
|
||
if generation != s.provCacheGen {
|
||
s.provCacheMu.Unlock()
|
||
return s.providerForProfile(id)
|
||
}
|
||
if e := s.provByProfile[id]; e != nil { // lost the race → keep the winner
|
||
prov, cfg = e.prov, e.cfg
|
||
} else {
|
||
s.provByProfile[id] = &provEntry{prov: prov, cfg: cfg}
|
||
log.Printf("[engine] built provider for LLM profile %d (%s / %s)", id, cfg.Provider(), cfg.Model)
|
||
}
|
||
s.provCacheMu.Unlock()
|
||
return prov, cfg, true
|
||
}
|
||
|
||
// chatAgentForProfile returns a ChatAgent built from a specific LLM profile, cached
|
||
// per profile id. Returns nil if the profile is missing or has no API key.
|
||
func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
|
||
s.profMu.Lock()
|
||
cached := s.profChatAgents[id]
|
||
s.profMu.Unlock()
|
||
if cached != nil {
|
||
return cached
|
||
}
|
||
prov, cfg, ok := s.providerForProfile(id)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
|
||
ca := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow())
|
||
ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
|
||
ca.SetWebSearch(s.m.WebSearchOpts())
|
||
ca.SetGuard(s.chatGuard())
|
||
ca.SetNonStreaming(nonStreamingResolver(cfg))
|
||
ca.SetMaxTokens(maxTokensResolver(cfg))
|
||
ca.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读)
|
||
s.profMu.Lock()
|
||
if ex := s.profChatAgents[id]; ex != nil { // lost the race → keep the winner
|
||
ca = ex
|
||
} else {
|
||
s.profChatAgents[id] = ca
|
||
}
|
||
s.profMu.Unlock()
|
||
return ca
|
||
}
|
||
|
||
// invalidateProfileAgents drops the per-profile ChatAgent + provider caches so a profile
|
||
// save/activate/delete — or an agent's binding change — rebuilds pinned tasks' agents
|
||
// (and re-resolves each agent's bound model) on their next round.
|
||
func (s *Server) invalidateProfileAgents() {
|
||
s.profMu.Lock()
|
||
s.profChatAgents = map[int64]*agent.ChatAgent{}
|
||
s.profMu.Unlock()
|
||
s.provCacheMu.Lock()
|
||
s.provCacheGen++
|
||
s.provByProfile = map[int64]*provEntry{}
|
||
s.provCacheMu.Unlock()
|
||
s.invalidateTaskAgents()
|
||
}
|
||
|
||
func (s *Server) chatAgentRef() *agent.ChatAgent {
|
||
s.cfgMu.Lock()
|
||
defer s.cfgMu.Unlock()
|
||
return s.chatAgent
|
||
}
|
||
|
||
func (s *Server) Handler() http.Handler {
|
||
mux := http.NewServeMux()
|
||
s.registerSideRoutes(mux)
|
||
|
||
// Auth routes — exempt from JWT check (handled in requireAuth)
|
||
mux.HandleFunc("GET /api/auth/status", s.authStatus)
|
||
mux.HandleFunc("POST /api/auth/init", s.authInit)
|
||
mux.HandleFunc("POST /api/auth/login", s.authLogin)
|
||
mux.HandleFunc("POST /api/auth/change-password", s.authChangePassword)
|
||
|
||
mux.HandleFunc("GET /api/health", s.health)
|
||
mux.HandleFunc("GET /api/stats", s.stats)
|
||
mux.HandleFunc("GET /api/logs", s.getLogs)
|
||
mux.HandleFunc("GET /api/logs/history", s.getLogsHistory)
|
||
mux.HandleFunc("GET /api/logs/stream", s.streamLogs)
|
||
|
||
// 页面一键更新。走的是默认的 JWT 鉴权（auth.go 只放行 /api/auth/* 和
|
||
// /api/health），所以这几个改动程序自身的接口天然需要登录。
|
||
mux.HandleFunc("GET /api/update/check", s.updateCheck)
|
||
mux.HandleFunc("POST /api/update/apply", s.updateApply)
|
||
mux.HandleFunc("POST /api/update/rollback", s.updateRollback)
|
||
mux.HandleFunc("GET /api/update/stream", s.updateStream)
|
||
|
||
mux.HandleFunc("GET /api/tasks", s.listTasks)
|
||
mux.HandleFunc("POST /api/tasks", s.createTask)
|
||
mux.HandleFunc("GET /api/task-categories", s.pgListTaskCategories)
|
||
mux.HandleFunc("POST /api/task-categories", s.pgCreateTaskCategory)
|
||
mux.HandleFunc("PATCH /api/task-categories/{id}", s.pgRenameTaskCategory)
|
||
mux.HandleFunc("DELETE /api/task-categories/{id}", s.pgDeleteTaskCategory)
|
||
mux.HandleFunc("POST /api/tasks/category/batch", s.updateTasksCategoryBatch)
|
||
mux.HandleFunc("GET /api/task-templates", s.pgListTaskTemplates)
|
||
mux.HandleFunc("POST /api/task-templates", s.pgCreateTaskTemplate)
|
||
mux.HandleFunc("PATCH /api/task-templates/{id}", s.pgUpdateTaskTemplate)
|
||
mux.HandleFunc("DELETE /api/task-templates/{id}", s.pgDeleteTaskTemplate)
|
||
mux.HandleFunc("GET /api/tasks/{id}", s.getTask)
|
||
mux.HandleFunc("PATCH /api/tasks/{id}", s.updateTaskMetadata)
|
||
mux.HandleFunc("PATCH /api/tasks/{id}/category", s.updateTaskCategory)
|
||
// 任务级资产拦截/允许规则
|
||
mux.HandleFunc("GET /api/tasks/{id}/intercept-rules", s.taskInterceptListRules)
|
||
mux.HandleFunc("POST /api/tasks/{id}/intercept-rules", s.taskInterceptCreateRule)
|
||
mux.HandleFunc("PUT /api/tasks/{id}/intercept-rules/{rid}", s.taskInterceptUpdateRule)
|
||
mux.HandleFunc("DELETE /api/tasks/{id}/intercept-rules/{rid}", s.taskInterceptDeleteRule)
|
||
mux.HandleFunc("POST /api/tasks/{id}/intercept-rules/{rid}/toggle", s.taskInterceptToggleRule)
|
||
mux.HandleFunc("POST /api/tasks/control/batch", s.controlTasksBatch)
|
||
mux.HandleFunc("GET /api/task-archives", s.listTaskArchives)
|
||
mux.HandleFunc("GET /api/task-archives/{id}", s.getTaskArchive)
|
||
mux.HandleFunc("POST /api/tasks/{id}/archive", s.queueTaskArchive)
|
||
mux.HandleFunc("POST /api/tasks/archive/batch", s.queueTaskArchivesBatch)
|
||
mux.HandleFunc("POST /api/task-archives/{id}/restore", s.queueTaskArchiveRestore)
|
||
mux.HandleFunc("POST /api/task-archives/restore/batch", s.restoreTaskArchivesBatch)
|
||
mux.HandleFunc("DELETE /api/task-archives/{id}", s.queueTaskArchiveDelete)
|
||
mux.HandleFunc("POST /api/task-archives/delete/batch", s.deleteTaskArchivesBatch)
|
||
mux.HandleFunc("GET /api/tasks/{id}/coverage", s.taskCoverage)
|
||
mux.HandleFunc("GET /api/tasks/{id}/coverage-graph", s.taskCoverageGraph)
|
||
mux.HandleFunc("GET /api/tasks/{id}/asset-refs", s.taskAssetRefs)
|
||
mux.HandleFunc("POST /api/tasks/{id}/assets", s.attachTaskAssets)
|
||
mux.HandleFunc("DELETE /api/tasks/{id}/assets/{assetID}", s.detachTaskAsset)
|
||
mux.HandleFunc("GET /api/tasks/{id}/intent-assets", s.taskIntentAssets)
|
||
|
||
// 工作空间文件管理器（针对 workDir）
|
||
mux.HandleFunc("GET /api/workspace/list", s.wsList)
|
||
mux.HandleFunc("GET /api/workspace/read", s.wsRead)
|
||
mux.HandleFunc("POST /api/workspace/write", s.wsWrite)
|
||
mux.HandleFunc("POST /api/workspace/mkdir", s.wsMkdir)
|
||
mux.HandleFunc("DELETE /api/workspace/delete", s.wsDelete)
|
||
mux.HandleFunc("GET /api/workspace/download", s.wsDownload)
|
||
mux.HandleFunc("POST /api/workspace/upload", s.wsUpload)
|
||
mux.HandleFunc("GET /api/tasks/{id}/scope", s.taskScopeList)
|
||
mux.HandleFunc("POST /api/tasks/{id}/scope", s.taskScopeAdd)
|
||
mux.HandleFunc("DELETE /api/tasks/{id}/scope/{sid}", s.taskScopeDelete)
|
||
mux.HandleFunc("GET /api/tasks/{id}/goals", s.listGoals) // 目标管理:列出本任务全部目标
|
||
mux.HandleFunc("POST /api/tasks/{id}/goals", s.addGoal) // 目标管理:人工新增目标(复活任务)
|
||
mux.HandleFunc("PATCH /api/tasks/{id}/goals/{gid}", s.editGoal) // 目标管理:修改目标(复活任务)
|
||
mux.HandleFunc("DELETE /api/tasks/{id}/goals/{gid}", s.deleteGoal) // 目标管理:硬删除目标(不复活)
|
||
mux.HandleFunc("GET /api/tasks/{id}/constraints", s.listConstraints) // 约束管理:列出本任务操作约束
|
||
mux.HandleFunc("POST /api/tasks/{id}/constraints", s.addConstraint) // 约束管理:新增约束(不通知 planner)
|
||
mux.HandleFunc("PATCH /api/tasks/{id}/constraints/{cid}", s.editConstraint) // 约束管理:修改约束
|
||
mux.HandleFunc("DELETE /api/tasks/{id}/constraints/{cid}", s.deleteConstraint) // 约束管理:删除约束
|
||
mux.HandleFunc("POST /api/tasks/{id}/control", s.control)
|
||
mux.HandleFunc("PUT /api/tasks/{id}/llm", s.updateTaskLLMProfiles)
|
||
mux.HandleFunc("GET /api/tasks/{id}/llm/resolution", s.taskLLMResolutionHandler)
|
||
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/control", s.controlIntent)
|
||
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/messages", s.sendWorkerMessage)
|
||
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/rerun", s.rerunIntent) // 重跑单条 blocked/exhausted/stopped 意图
|
||
mux.HandleFunc("POST /api/tasks/{id}/intents/rerun-blocked", s.rerunBlocked) // 批量重跑本任务全部 blocked 意图
|
||
mux.HandleFunc("POST /api/active", s.setActive)
|
||
|
||
mux.HandleFunc("GET /api/llm", s.getLLM)
|
||
mux.HandleFunc("POST /api/llm", s.setLLM)
|
||
mux.HandleFunc("POST /api/llm/test", s.testLLM)
|
||
|
||
// asset system
|
||
mux.HandleFunc("GET /api/assets", s.listAssets)
|
||
mux.HandleFunc("GET /api/assets/counts", s.assetCounts)
|
||
mux.HandleFunc("POST /api/assets", s.insertAssets)
|
||
mux.HandleFunc("DELETE /api/assets", s.deleteAssets)
|
||
|
||
// company system
|
||
mux.HandleFunc("GET /api/companies", s.listCompanies)
|
||
mux.HandleFunc("POST /api/companies", s.createCompany)
|
||
mux.HandleFunc("GET /api/companies/{id}", s.getCompany)
|
||
mux.HandleFunc("DELETE /api/companies/{id}", s.deleteCompany)
|
||
mux.HandleFunc("POST /api/companies/{id}/scope", s.addCompanyScope)
|
||
mux.HandleFunc("POST /api/companies/reattribute", s.reattribute)
|
||
|
||
mux.HandleFunc("GET /api/exploration/frontier", s.frontier)
|
||
mux.HandleFunc("GET /api/exploration/findings", s.findings)
|
||
mux.HandleFunc("GET /api/exploration/findings/groups", s.findingGroups)
|
||
mux.HandleFunc("GET /api/exploration/findings/asset-tree", s.findingAssetTree)
|
||
mux.HandleFunc("GET /api/exploration/findings/stats", s.findingStats)
|
||
mux.HandleFunc("GET /api/exploration/findings/export", s.findingsExport)
|
||
s.registerFindingTraffic(mux)
|
||
mux.HandleFunc("GET /api/exploration/findings/{id}", s.getFinding)
|
||
mux.HandleFunc("GET /api/exploration/findings/{id}/lineage", s.findingLineage)
|
||
mux.HandleFunc("POST /api/exploration/findings/{id}/deepen", s.deepenFinding)
|
||
mux.HandleFunc("GET /api/exploration/findings/{id}/retests", s.listFindingRetests)
|
||
mux.HandleFunc("GET /api/exploration/findings/retests/active", s.listActiveFindingRetests)
|
||
mux.HandleFunc("POST /api/exploration/findings/{id}/retests", s.startFindingRetest)
|
||
mux.HandleFunc("PATCH /api/exploration/findings/{id}", s.patchFinding)
|
||
mux.HandleFunc("DELETE /api/exploration/findings/{id}", s.deleteFinding)
|
||
mux.HandleFunc("GET /api/exploration/intents", s.intents)
|
||
mux.HandleFunc("GET /api/exploration/graph", s.explorationGraph)
|
||
mux.HandleFunc("GET /api/exploration/nodes", s.explorationNodes)
|
||
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)
|
||
mux.HandleFunc("GET /api/tokens/daily", s.tokenDailyStats)
|
||
mux.HandleFunc("GET /api/tokens/conversations", s.conversationTokens)
|
||
mux.HandleFunc("GET /api/tokens/usage", s.pgUsageStats) // 全局 llm_usage 聚合（仪表盘新版视图）
|
||
|
||
mux.HandleFunc("GET /api/audit", s.getAudit)
|
||
mux.HandleFunc("POST /api/gc", s.gc)
|
||
mux.HandleFunc("GET /api/traffic", s.getTraffic)
|
||
mux.HandleFunc("GET /api/traffic/hosts", s.getTrafficHosts)
|
||
mux.HandleFunc("DELETE /api/traffic", s.deleteTraffic)
|
||
mux.HandleFunc("DELETE /api/traffic/hosts", s.deleteTrafficHosts)
|
||
mux.HandleFunc("DELETE /api/traffic/all", s.deleteAllTraffic)
|
||
mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange)
|
||
mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob)
|
||
mux.HandleFunc("GET /api/commands", s.pgListCommands)
|
||
mux.HandleFunc("GET /api/commands/stats", s.pgToolStats) // 按工具聚合调用次数
|
||
mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords)
|
||
mux.HandleFunc("DELETE /api/llm/records", s.pgDeleteLLMRecords)
|
||
mux.HandleFunc("GET /api/llm/records/tasks", s.pgLLMTasks)
|
||
mux.HandleFunc("GET /api/llm/records/by-model", s.pgTokenByModel) // 按模型聚合本任务 token 用量
|
||
mux.HandleFunc("GET /api/llm/records/{id}", s.pgGetLLMRecord)
|
||
mux.HandleFunc("GET /api/settings", s.getSettings)
|
||
mux.HandleFunc("PUT /api/settings", s.putSettings)
|
||
// 漏洞 IM 推送。渠道是「多实例 + 各自过滤规则」的资源，因此独立成一组
|
||
// REST 接口，而不是塞进扁平的 /api/settings 键值里。
|
||
mux.HandleFunc("GET /api/notify/meta", s.notifyMeta)
|
||
mux.HandleFunc("GET /api/notify/channels", s.notifyListChannels)
|
||
mux.HandleFunc("POST /api/notify/channels", s.notifyCreateChannel)
|
||
mux.HandleFunc("PATCH /api/notify/channels/{id}", s.notifyUpdateChannel)
|
||
mux.HandleFunc("DELETE /api/notify/channels/{id}", s.notifyDeleteChannel)
|
||
mux.HandleFunc("POST /api/notify/channels/{id}/test", s.notifyTestChannel)
|
||
mux.HandleFunc("GET /api/notify/deliveries", s.notifyListDeliveries)
|
||
mux.HandleFunc("POST /api/notify/deliveries/{id}/retry", s.notifyRetryDelivery)
|
||
mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch)
|
||
mux.HandleFunc("GET /api/report", s.getReport)
|
||
mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions)
|
||
mux.HandleFunc("POST /api/chat", s.chat)
|
||
mux.HandleFunc("POST /api/chat/upload", s.chatUpload) // 方式1 文件上传:落到会话/任务工作目录 uploads/
|
||
mux.HandleFunc("GET /api/tasks/{id}/chat/status", s.taskChatStatus)
|
||
mux.HandleFunc("POST /api/tasks/{id}/chat/stop", s.stopChat)
|
||
|
||
// --- 管理后台 API (PostgreSQL 数据源; 新版数据库与管理后台方案) ---
|
||
mux.HandleFunc("DELETE /api/tasks/{id}", s.pgDeleteTask)
|
||
// Agents
|
||
// conversations (chat page)
|
||
mux.HandleFunc("GET /api/conversations", s.pgListConversations)
|
||
mux.HandleFunc("POST /api/conversations", s.pgCreateConversation)
|
||
mux.HandleFunc("POST /api/conversations/delete/batch", s.pgDeleteConversationsBatch)
|
||
mux.HandleFunc("PATCH /api/conversations/{id}", s.pgRenameConversation)
|
||
mux.HandleFunc("PATCH /api/conversations/{id}/profile", s.pgUpdateConversation)
|
||
mux.HandleFunc("DELETE /api/conversations/{id}", s.pgDeleteConversation)
|
||
mux.HandleFunc("GET /api/conversations/{id}/messages", s.pgConversationMessages)
|
||
mux.HandleFunc("POST /api/conversations/{id}/messages", s.pgSendConversationMessage)
|
||
mux.HandleFunc("POST /api/conversations/{id}/stop", s.pgStopConversation)
|
||
mux.HandleFunc("GET /api/conversations/{id}/messages/{seq}", s.pgConversationMsgDetail)
|
||
|
||
mux.HandleFunc("GET /api/agents", s.pgListAgents)
|
||
mux.HandleFunc("POST /api/agents", s.pgCreateAgent)
|
||
mux.HandleFunc("GET /api/agents/{key}", s.pgGetAgent)
|
||
mux.HandleFunc("PATCH /api/agents/{key}", s.pgUpdateAgent)
|
||
mux.HandleFunc("DELETE /api/agents/{key}", s.pgDeleteAgent)
|
||
mux.HandleFunc("PUT /api/agents/{key}/config", s.pgSaveAgentConfig)
|
||
mux.HandleFunc("PUT /api/agents/{key}/prompt", s.pgSavePrompt)
|
||
mux.HandleFunc("POST /api/agents/{key}/prompt/reset", s.pgResetPrompt)
|
||
mux.HandleFunc("PUT /api/agents/{key}/wrapup", s.pgSaveWrapup)
|
||
mux.HandleFunc("POST /api/agents/{key}/wrapup/reset", s.pgResetWrapup)
|
||
mux.HandleFunc("PUT /api/agents/{key}/wrapup/task-timeout", s.pgSaveTaskTimeoutWrapup)
|
||
mux.HandleFunc("POST /api/agents/{key}/wrapup/task-timeout/reset", s.pgResetTaskTimeoutWrapup)
|
||
mux.HandleFunc("GET /api/agents/{key}/triggers", s.pgListTriggers)
|
||
mux.HandleFunc("POST /api/agents/{key}/triggers", s.pgCreateTrigger)
|
||
mux.HandleFunc("PATCH /api/triggers/{id}", s.pgUpdateTrigger)
|
||
mux.HandleFunc("DELETE /api/triggers/{id}", s.pgDeleteTrigger)
|
||
mux.HandleFunc("GET /api/agents/{key}/prompts", s.pgListPromptVersions)
|
||
mux.HandleFunc("GET /api/agents/{key}/variables", s.pgPromptVars)
|
||
mux.HandleFunc("POST /api/agents/{key}/prompt/preview", s.pgPreviewPrompt)
|
||
mux.HandleFunc("GET /api/agents/{key}/visibility", s.pgGetAgentVisibility)
|
||
mux.HandleFunc("PUT /api/agents/{key}/visibility", s.pgSetAgentVisibility)
|
||
// 内置工具目录（描述/参数默认值可改、按 agent 绑定；key 与 handler 在代码层）
|
||
mux.HandleFunc("GET /api/tools", s.pgListTools)
|
||
mux.HandleFunc("PUT /api/tools/{key}", s.pgUpdateTool)
|
||
mux.HandleFunc("POST /api/tools/custom", s.pgCreateCustomTool)
|
||
mux.HandleFunc("POST /api/tools/custom/test", s.pgTestCustomTool)
|
||
mux.HandleFunc("PUT /api/tools/custom/{key}", s.pgUpdateCustomTool)
|
||
mux.HandleFunc("DELETE /api/tools/custom/{key}", s.pgDeleteCustomTool)
|
||
mux.HandleFunc("POST /api/settings/python/detect", s.pgDetectPython)
|
||
mux.HandleFunc("POST /api/tools/{key}/reset", s.pgResetTool)
|
||
// MCP CRUD
|
||
mux.HandleFunc("GET /api/mcp", s.pgListMCP)
|
||
mux.HandleFunc("POST /api/mcp", s.pgSaveMCP)
|
||
mux.HandleFunc("DELETE /api/mcp/{id}", s.pgDeleteMCP)
|
||
mux.HandleFunc("GET /api/mcp/{id}/tools", s.pgMCPTools)
|
||
mux.HandleFunc("POST /api/mcp/{id}/refresh", s.pgRefreshMCP)
|
||
// 资产同步 — ScopeSentry 数据源
|
||
mux.HandleFunc("GET /api/sync/scopesentry/status", s.syncSSStatus)
|
||
mux.HandleFunc("POST /api/sync/scopesentry/datasource", s.syncSSDatasource)
|
||
mux.HandleFunc("GET /api/sync/scopesentry/projects", s.syncSSProjects)
|
||
mux.HandleFunc("GET /api/sync/scopesentry/tasks", s.syncSSTasks)
|
||
mux.HandleFunc("POST /api/sync/scopesentry/sync", s.syncSSRun)
|
||
// Skill CRUD (文件系统)
|
||
mux.HandleFunc("GET /api/skills", s.fsListSkills)
|
||
mux.HandleFunc("POST /api/skills", s.fsCreateSkill)
|
||
mux.HandleFunc("POST /api/skills/upload", s.fsUploadSkill)
|
||
mux.HandleFunc("DELETE /api/skills/{name}", s.fsDeleteSkill)
|
||
mux.HandleFunc("GET /api/skills/missing", s.fsMissingSkills) // 未命中(想调但不存在)的 skill 名
|
||
mux.HandleFunc("GET /api/skills/{name}/usage", s.fsSkillUsage) // 单个 skill 的最近调用
|
||
mux.HandleFunc("PUT /api/skills/{name}/meta", s.fsUpdateSkillMeta)
|
||
mux.HandleFunc("POST /api/skills/{name}/dirs", s.fsCreateDir)
|
||
mux.HandleFunc("GET /api/skills/{name}/files", s.fsListFiles)
|
||
// {file...} captures path segments including slashes (e.g. scripts/extract.py)
|
||
mux.HandleFunc("GET /api/skills/{name}/files/{file...}", s.fsReadFile)
|
||
mux.HandleFunc("PUT /api/skills/{name}/files/{file...}", s.fsWriteFile)
|
||
mux.HandleFunc("DELETE /api/skills/{name}/files/{file...}", s.fsDeletePath)
|
||
// MCP 资源侧可见性（更具体的 skill 路由会优先匹配）
|
||
mux.HandleFunc("GET /api/visibility/{kind}/{id}", s.pgResourceVisibility)
|
||
mux.HandleFunc("POST /api/visibility/toggle", s.pgToggleVisibility)
|
||
// Skill 可见性（按名称，更具体，优先于上面的通配路由）
|
||
mux.HandleFunc("GET /api/visibility/skill/{name}", s.pgSkillVisibility)
|
||
mux.HandleFunc("POST /api/visibility/skill/toggle", s.pgToggleSkillVisibility)
|
||
// LLM 多 profile
|
||
mux.HandleFunc("GET /api/llm/profiles", s.pgListProfiles)
|
||
mux.HandleFunc("POST /api/llm/profiles", s.pgSaveProfile)
|
||
mux.HandleFunc("DELETE /api/llm/profiles/{id}", s.pgDeleteProfile)
|
||
mux.HandleFunc("POST /api/llm/profiles/active", s.pgActivateProfile)
|
||
mux.HandleFunc("GET /api/llm/retry-policy", s.pgGetLLMRetryPolicy)
|
||
mux.HandleFunc("POST /api/llm/retry-policy", s.pgSaveLLMRetryPolicy)
|
||
mux.HandleFunc("GET /api/llm/pool", s.pgLLMPoolStatus)
|
||
mux.HandleFunc("POST /api/llm/pool/reset", s.pgLLMPoolReset)
|
||
mux.HandleFunc("POST /api/llm/models", s.pgListModels)
|
||
|
||
// 拦截规则管理
|
||
mux.HandleFunc("GET /api/intercept/rules", s.interceptListRules)
|
||
mux.HandleFunc("POST /api/intercept/rules", s.interceptCreateRule)
|
||
mux.HandleFunc("PUT /api/intercept/rules/{id}", s.interceptUpdateRule)
|
||
mux.HandleFunc("DELETE /api/intercept/rules/{id}", s.interceptDeleteRule)
|
||
mux.HandleFunc("POST /api/intercept/rules/{id}/toggle", s.interceptToggleRule)
|
||
|
||
// 资产拦截规则管理（全局黑名单：域名/IP/URL/CIDR）
|
||
mux.HandleFunc("GET /api/asset-intercept/rules", s.assetInterceptListRules)
|
||
mux.HandleFunc("POST /api/asset-intercept/rules", s.assetInterceptCreateRule)
|
||
mux.HandleFunc("PUT /api/asset-intercept/rules/{id}", s.assetInterceptUpdateRule)
|
||
mux.HandleFunc("DELETE /api/asset-intercept/rules/{id}", s.assetInterceptDeleteRule)
|
||
mux.HandleFunc("POST /api/asset-intercept/rules/{id}/toggle", s.assetInterceptToggleRule)
|
||
|
||
mux.HandleFunc("GET /api/intercept/pending", s.interceptListPending)
|
||
mux.HandleFunc("GET /api/intercept/pending/{id}", s.interceptGetOne)
|
||
mux.HandleFunc("POST /api/intercept/pending/{id}/decide", s.interceptDecide)
|
||
mux.HandleFunc("GET /api/intercept/history", s.interceptHistory)
|
||
mux.HandleFunc("GET /api/intercept/history/{id}", s.interceptDetail)
|
||
mux.HandleFunc("GET /api/intercept/history/{id}/execution", s.interceptExecution)
|
||
mux.HandleFunc("GET /api/intercept/task/{taskID}", s.interceptListTaskItems)
|
||
mux.HandleFunc("GET /api/intercept/tool-config", s.interceptGetToolConfig)
|
||
mux.HandleFunc("PUT /api/intercept/tool-config", s.interceptSetToolConfig)
|
||
mux.HandleFunc("GET /api/intercept/judge", s.interceptGetJudgeConfig)
|
||
mux.HandleFunc("PUT /api/intercept/judge", s.interceptSetJudgeConfig)
|
||
|
||
// /api/* goes through CORS + JWT; everything else is served by the embedded
|
||
// frontend (public — auth is enforced client-side and on the API). With the
|
||
// no-embed build the webui handler just 404s (run `next dev` separately).
|
||
api := cors(s.requireAuth(mux))
|
||
root := http.NewServeMux()
|
||
root.Handle("/api/", api)
|
||
root.Handle("/", s.webuiHandler())
|
||
return root
|
||
}
|
||
|
||
// --- handlers ---
|
||
|
||
func (s *Server) health(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, 200, map[string]any{"ok": true, "service": "artex", "version": BuildVersion})
|
||
}
|
||
|
||
func (s *Server) stats(w http.ResponseWriter, r *http.Request) {
|
||
out := map[string]any{}
|
||
out["engine_mode"] = "idle" // spec enum; overridden below when a task is active
|
||
out["llm_configured"] = s.engine.Ready() // is an LLM provider installed at all
|
||
if tr := s.m.Traffic(); tr != nil {
|
||
c, _ := tr.Count()
|
||
out["traffic"] = c
|
||
out["traffic_enabled"] = true
|
||
}
|
||
if counts, err := s.m.Assets().CountsByType(); err == nil {
|
||
total := 0
|
||
for _, n := range counts {
|
||
total += n
|
||
}
|
||
out["assets"] = total
|
||
out["asset_counts"] = counts
|
||
}
|
||
|
||
// resolve the task: explicit ?task=<id> binds to that task (so a detail view
|
||
// never silently follows a globally-changed active task); empty = active.
|
||
taskParam := r.URL.Query().Get("task")
|
||
t := s.m.ResolveTask(taskParam)
|
||
if t == nil {
|
||
if taskParam != "" && taskParam != "active" {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
writeJSON(w, 200, out) // no task selected: only global fields
|
||
return
|
||
}
|
||
out["llm_configured"] = s.engine.ReadyFor(t)
|
||
|
||
st, _ := t.Store.Stats()
|
||
out["exploration"] = st
|
||
|
||
// per-task running state + heartbeat (distinct from "LLM configured").
|
||
intents, _ := t.Store.ListByKind(db.KindIntent, 100000)
|
||
inFlight := 0
|
||
for _, in := range intents {
|
||
if in.State == "running" {
|
||
inFlight++
|
||
}
|
||
}
|
||
goals, _ := t.Store.ListByKind(db.KindGoal, 10000)
|
||
goalsMet := 0
|
||
for _, g := range goals {
|
||
if g.State == "met" {
|
||
goalsMet++
|
||
}
|
||
}
|
||
last := s.engine.LastActivity(t.ID)
|
||
paused := s.engine.IsPaused(t.ID)
|
||
activeCalls := s.engine.ActiveLLMCalls(t.ID)
|
||
running := s.engine.ReadyFor(t) && s.engine.Started(t.ID) && !paused
|
||
// A live event-driven engine with no work is healthy idle. Only a persisted
|
||
// running intent without any corresponding LLM call can be considered stalled.
|
||
stalled := running && activeCalls == 0 && inFlight > 0 && last > 0 && time.Now().Unix()-last > 120
|
||
|
||
// engine_mode follows the spec enum (exploring|paused|stalled|idle).
|
||
engineMode := "idle"
|
||
switch {
|
||
case paused:
|
||
engineMode = "paused"
|
||
case stalled:
|
||
engineMode = "stalled"
|
||
case running && activeCalls > 0:
|
||
engineMode = "exploring"
|
||
}
|
||
out["engine_mode"] = engineMode
|
||
|
||
out["active_task"] = map[string]any{
|
||
"id": t.ID, "description": t.Description, "goal": t.Goal,
|
||
"running": running,
|
||
"paused": paused,
|
||
"engine_mode": engineMode,
|
||
"in_flight": inFlight,
|
||
"llm_in_flight": activeCalls,
|
||
"last_activity": last,
|
||
"stalled": stalled,
|
||
"goals_total": len(goals),
|
||
"goals_met": goalsMet,
|
||
}
|
||
writeJSON(w, 200, out)
|
||
}
|
||
|
||
func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
|
||
active := ""
|
||
if t := s.m.ActiveTask(); t != nil {
|
||
active = t.ID
|
||
}
|
||
list := s.m.List()
|
||
metrics, _ := s.m.PG().TaskListMetricsAll()
|
||
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
|
||
dtos := make([]TaskDTO, 0, len(list))
|
||
for _, t := range list {
|
||
dto := taskDTO(t, s.resolvedTaskStatus(t))
|
||
applyTaskArchiveBlocker(&dto, archiveBlockers)
|
||
metric := metrics[t.ExpID]
|
||
dto.Tokens = tokenTotalDTO(metric.Tokens)
|
||
// prefer the live in-memory heartbeat (fresher) and fall back to the
|
||
// persisted max activity time (survives restarts) for run-duration display.
|
||
dto.LastActivity = metric.LastActivity
|
||
if live := s.engine.LastActivity(t.ID); live > dto.LastActivity {
|
||
dto.LastActivity = live
|
||
}
|
||
dto.GoalsTotal = metric.Goals.Total
|
||
dto.GoalsMet = metric.Goals.Met
|
||
dto.InFlight = metric.RunningIntents
|
||
dto.Findings = FindingSeverityDTO{
|
||
Critical: metric.Findings.Critical,
|
||
High: metric.Findings.High,
|
||
Medium: metric.Findings.Medium,
|
||
Low: metric.Findings.Low,
|
||
}
|
||
dtos = append(dtos, dto)
|
||
}
|
||
writeJSON(w, 200, map[string]any{"tasks": dtos, "active": active})
|
||
}
|
||
|
||
func (s *Server) resolvedTaskStatus(t *Task) string {
|
||
if t == nil {
|
||
return "created"
|
||
}
|
||
lifecycle := t.lifecycleSnapshot()
|
||
switch {
|
||
case isTerminalStatus(lifecycle.Status):
|
||
return lifecycle.Status
|
||
case lifecycle.Queued:
|
||
return "queued"
|
||
case lifecycle.Paused || s.engine.IsPaused(t.ID):
|
||
return "paused"
|
||
case t.llmStateSnapshot().FailoverState == "chain_exhausted" && s.engine.Started(t.ID):
|
||
// LLM readiness is an execution dependency, not lifecycle state. Keeping
|
||
// an exhausted task running lets the user edit/reset its chain directly.
|
||
return "running"
|
||
case s.engine.ReadyFor(t) && s.engine.Started(t.ID):
|
||
return "running"
|
||
default:
|
||
return "created"
|
||
}
|
||
}
|
||
|
||
func (s *Server) setActive(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
ID string `json:"id"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
if !s.m.SetActive(req.ID) {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
// resume the engine for the opened task (idempotent — no-op if already running).
|
||
// 排队中的任务:仅设为活跃可查看,不启动引擎(维持并发上限,由 reconcile 补位)。
|
||
if t, ok := s.m.Task(req.ID); ok && !t.lifecycleSnapshot().Queued {
|
||
s.engine.Run(s.ctx, t)
|
||
}
|
||
writeJSON(w, 200, map[string]any{"active": req.ID})
|
||
}
|
||
|
||
// control pauses/resumes a task's autonomous execution (planner + workers).
|
||
func (s *Server) control(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
var req struct {
|
||
Action string `json:"action"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
if req.Action != "pause" && req.Action != "resume" {
|
||
writeErr(w, 400, "action must be pause|resume")
|
||
return
|
||
}
|
||
result, err := s.applyTaskControl(t, req.Action)
|
||
if err != nil {
|
||
writeErr(w, 409, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, result)
|
||
}
|
||
|
||
// controlIntent pauses, resumes, or cancels one local worker intent. A running
|
||
// worker is always stopped before state mutation/cleanup, preventing tool output
|
||
// that arrives after the user action from recreating deleted blackboard records.
|
||
func (s *Server) controlIntent(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
if !s.engine.beginTaskOperation(t.ID) {
|
||
writeErr(w, 409, errTaskDeletingIntentControl)
|
||
return
|
||
}
|
||
defer s.engine.decInflight(t.ID)
|
||
|
||
iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64)
|
||
if err != nil || iid <= 0 {
|
||
writeErr(w, 400, "bad intent id")
|
||
return
|
||
}
|
||
var req struct {
|
||
Action string `json:"action"`
|
||
Reason string `json:"reason"` // cancel(删除)时必填:删除原因
|
||
Mode string `json:"mode"` // cancel 专用:soft(默认,假删除)| hard(真删除,级联移除独占子孙)
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, "bad json: "+err.Error())
|
||
return
|
||
}
|
||
if req.Action != "pause" && req.Action != "resume" && req.Action != "cancel" {
|
||
writeErr(w, 400, "action must be pause|resume|cancel")
|
||
return
|
||
}
|
||
|
||
result, err := s.applyIntentControl(r.Context(), t, iid, req.Action, req.Reason, req.Mode)
|
||
if err != nil {
|
||
writeErr(w, 409, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, result)
|
||
}
|
||
|
||
// rerunIntent 重跑一条没跑成功的意图(blocked/exhausted/stopped):把它置回 open,worker
|
||
// 会重新认领、从头再跑(已写回图谱的 fact/finding/asset 保留);若任务已终态/暂停则顺带复活。
|
||
// 用于「出错的 work 点击继续运行」——网络/LLM 抖动导致 blocked 后可一键重试。
|
||
func restoreRerunIntent(t *Task, before *db.Node) error {
|
||
if t == nil || before == nil {
|
||
return fmt.Errorf("missing intent rollback snapshot")
|
||
}
|
||
if before.State == "blocked" && before.BlockedReason != "" {
|
||
return t.Store.SetIntentBlockedReason(before.ID, before.BlockedReason)
|
||
}
|
||
return t.Store.SetIntentState(before.ID, before.State)
|
||
}
|
||
|
||
func (s *Server) rerunIntent(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64)
|
||
if err != nil {
|
||
writeErr(w, 400, "bad intent id")
|
||
return
|
||
}
|
||
if !s.engine.beginTaskOperation(t.ID) {
|
||
writeErr(w, 409, errTaskDeletingIntentRerun)
|
||
return
|
||
}
|
||
defer s.engine.decInflight(t.ID)
|
||
before, err := t.Store.GetNode(iid)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
reopened, err := t.Store.ReopenIntent(iid)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if !reopened {
|
||
writeErr(w, 409, errIntentNotRerunnable)
|
||
return
|
||
}
|
||
queued, err := s.admitTask(t, "resume")
|
||
if err != nil {
|
||
if rollbackErr := restoreRerunIntent(t, before); rollbackErr != nil {
|
||
err = fmt.Errorf("%w; restore intent %d after admission failure: %v", err, iid, rollbackErr)
|
||
}
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
log.Printf("[task] #%s 意图 #%d 已重开(重跑)", t.ID, iid)
|
||
writeJSON(w, 200, map[string]any{"id": t.ID, "reopened": iid, "queued": queued})
|
||
}
|
||
|
||
// rerunBlocked 批量重跑本任务全部 blocked 意图(适合一次网络/LLM 断连导致多条 blocked 后
|
||
// 一键全部重试),置回 open 并复活任务;返回重开的条数。
|
||
func (s *Server) rerunBlocked(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
if !s.engine.beginTaskOperation(t.ID) {
|
||
writeErr(w, 409, errTaskDeletingIntentRerun)
|
||
return
|
||
}
|
||
defer s.engine.decInflight(t.ID)
|
||
intents, err := t.Store.ListByKind(db.KindIntent, 1000000)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
before := make([]*db.Node, 0)
|
||
for _, intent := range intents {
|
||
if intent.State == "blocked" {
|
||
copy := *intent
|
||
before = append(before, ©)
|
||
}
|
||
}
|
||
n, err := t.Store.ReopenBlockedIntents()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
queued := false
|
||
if n > 0 {
|
||
queued, err = s.admitTask(t, "resume")
|
||
if err != nil {
|
||
rollbackErrors := make([]string, 0)
|
||
for _, intent := range before {
|
||
if rollbackErr := restoreRerunIntent(t, intent); rollbackErr != nil {
|
||
rollbackErrors = append(rollbackErrors, fmt.Sprintf("intent %d: %v", intent.ID, rollbackErr))
|
||
}
|
||
}
|
||
if len(rollbackErrors) > 0 {
|
||
err = fmt.Errorf("%w; restore blocked intents after admission failure: %s", err, strings.Join(rollbackErrors, "; "))
|
||
}
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
log.Printf("[task] #%s 批量重开 %d 条 blocked 意图", t.ID, n)
|
||
}
|
||
writeJSON(w, 200, map[string]any{"id": t.ID, "reopened": n, "queued": queued})
|
||
}
|
||
|
||
// getLLM returns the current LLM config (key never exposed).
|
||
func (s *Server) getLLM(w http.ResponseWriter, r *http.Request) {
|
||
s.cfgMu.Lock()
|
||
defer s.cfgMu.Unlock()
|
||
writeJSON(w, 200, map[string]any{
|
||
"configured": s.llmOn,
|
||
"provider": s.llmCfg.Provider(),
|
||
"model": s.llmCfg.Model,
|
||
"base_url": s.llmCfg.BaseURL,
|
||
"proxy": s.llmCfg.Proxy,
|
||
"key_set": s.llmCfg.APIKey != "",
|
||
"rate_per_second": s.llmCfg.RatePerSecond,
|
||
"rate_per_minute": s.llmCfg.RatePerMinute,
|
||
"context_window_k": s.llmCfg.ContextWindowK,
|
||
"thinking_type": s.llmCfg.ThinkingType,
|
||
"reasoning_effort": s.llmCfg.ReasoningEffort,
|
||
})
|
||
}
|
||
|
||
// setLLM configures the LLM at runtime. A blank api_key keeps the existing key.
|
||
func (s *Server) setLLM(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
Provider string `json:"provider"`
|
||
Model string `json:"model"`
|
||
BaseURL string `json:"base_url"`
|
||
Proxy string `json:"proxy"`
|
||
APIKey string `json:"api_key"`
|
||
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"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
cfg := agent.ConfigFrom(req.Provider, req.Model, req.BaseURL, req.APIKey, req.Proxy)
|
||
cfg.RatePerSecond, cfg.RatePerMinute = req.RatePerSecond, req.RatePerMinute
|
||
cfg.ThinkingType = req.ThinkingType
|
||
cfg.ReasoningEffort = req.ReasoningEffort
|
||
if k := req.ContextWindowK; k > 0 { // 0 = keep default (200K); cap at 1M
|
||
if k > 1000 {
|
||
k = 1000
|
||
}
|
||
cfg.ContextWindowK = k
|
||
}
|
||
if cfg.APIKey == "" {
|
||
s.cfgMu.Lock()
|
||
cfg.APIKey = s.llmCfg.APIKey // keep existing key if not re-entered
|
||
s.cfgMu.Unlock()
|
||
}
|
||
if cfg.APIKey == "" {
|
||
writeErr(w, 400, "api_key required")
|
||
return
|
||
}
|
||
// Validate provider construction before persisting it as the active profile.
|
||
if _, err := cfg.NewProvider(); err != nil {
|
||
writeErr(w, 400, "provider init failed: "+err.Error())
|
||
return
|
||
}
|
||
if err := s.saveLLMConfig(cfg); err != nil {
|
||
writeErr(w, 500, "persist provider failed: "+err.Error())
|
||
return
|
||
}
|
||
s.invalidateProfileAgents()
|
||
if _, ok := s.loadLLMConfig(); !ok {
|
||
writeErr(w, 500, "saved provider is unavailable")
|
||
return
|
||
}
|
||
if err := s.applyLLM(cfg); err != nil {
|
||
writeErr(w, 400, "provider init failed: "+err.Error())
|
||
return
|
||
}
|
||
log.Printf("[engine] LLM configured via UI: %s / %s", cfg.Provider(), cfg.Model)
|
||
s.getLLM(w, r)
|
||
}
|
||
|
||
// testLLM makes a real minimal completion to verify the config works.
|
||
func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
Provider string `json:"provider"`
|
||
Model string `json:"model"`
|
||
BaseURL string `json:"base_url"`
|
||
Proxy string `json:"proxy"`
|
||
APIKey string `json:"api_key"`
|
||
ThinkingType string `json:"thinking_type"`
|
||
ReasoningEffort string `json:"reasoning_effort"`
|
||
ProfileID *int64 `json:"profile_id"` // 测已存 profile 时传入：api_key 为空则用它存的 key
|
||
Streaming *bool `json:"streaming"` // 省略=流式，与保存 profile 时同一套默认
|
||
SessionHeaderKey string `json:"session_header_key"` // 非空=测试时也带该自定义会话头，值为一次性随机 session id
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
cfg := agent.ConfigFrom(req.Provider, req.Model, req.BaseURL, req.APIKey, req.Proxy)
|
||
// mirror production: send the SAME thinking params so a provider that rejects the
|
||
// reasoning_effort/thinking field fails the test too (no false "test ok, run 400").
|
||
cfg.ThinkingType = req.ThinkingType
|
||
cfg.ReasoningEffort = req.ReasoningEffort
|
||
// 同理，收发模式也照该 profile 的选择来：只支持其中一种通道的端点必须在这里就
|
||
// 暴露，而不是等会话里才发现"测试通过的配置根本跑不动"。
|
||
if req.Streaming != nil {
|
||
cfg.Stream = *req.Streaming
|
||
}
|
||
// 自定义会话头名照该配置来：非空则测试请求也发这个头(值为一次性随机 session id，
|
||
// 见 TestConnection)。opencode zen 等强制要求 x-opencode-session 的端点，缺了它
|
||
// 直接 400，必须在测试路径上也带上，否则"对话通、测试 400"。
|
||
cfg.SessionHeaderKey = req.SessionHeaderKey
|
||
// API Key 解析优先级：表单输入 > 指定 profile 存的 key > 全局配置的 key。
|
||
// 已存 profile 的 key 不回传浏览器，所以测试已存配置时表单为空，需从 DB 取。
|
||
// 会话头名同理：表单未带时用已存 profile 的值兜底。
|
||
if req.ProfileID != nil && (cfg.APIKey == "" || cfg.SessionHeaderKey == "") {
|
||
if p, err := s.m.pg.ProfileByID(*req.ProfileID); err == nil && p != nil {
|
||
if cfg.APIKey == "" {
|
||
cfg.APIKey = p.APIKey
|
||
}
|
||
if cfg.SessionHeaderKey == "" {
|
||
cfg.SessionHeaderKey = p.SessionHeaderKey
|
||
}
|
||
}
|
||
}
|
||
if cfg.APIKey == "" {
|
||
s.cfgMu.Lock()
|
||
cfg.APIKey = s.llmCfg.APIKey
|
||
s.cfgMu.Unlock()
|
||
}
|
||
if cfg.APIKey == "" {
|
||
writeJSON(w, 200, map[string]any{"ok": false, "error": errLLMTestNoAPIKey})
|
||
return
|
||
}
|
||
// 重试参数【不】带进连接测试:测试有 30s 硬超时,把配置的重试次数/长间隔叠上去
|
||
// 只会让一个本来能用的端点测成"超时失败"。测试看的是"这个端点通不通",重试节奏
|
||
// 是跑起来之后的事。
|
||
lat, reply, err := agent.TestConnection(r.Context(), cfg)
|
||
if err != nil {
|
||
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()})
|
||
return
|
||
}
|
||
// 回传模型实际回复，让"测试通过"有据可查：看得见它确实说了话，而不只是 HTTP 200。
|
||
writeJSON(w, 200, map[string]any{
|
||
"ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model, "reply": truncateReply(reply),
|
||
})
|
||
}
|
||
|
||
// truncateReply clips a connection-test reply for display. A model told to answer
|
||
// "OK" can still ramble (or think out loud); the UI only needs enough to show it
|
||
// really said something. Rune-based so multibyte text never splits mid-character.
|
||
func truncateReply(s string) string {
|
||
const max = 200
|
||
r := []rune(s)
|
||
if len(r) <= max {
|
||
return s
|
||
}
|
||
return string(r[:max]) + "…"
|
||
}
|
||
|
||
type createTaskReq struct {
|
||
Name string `json:"name,omitempty"` // 可选任务名称;省略/空=未命名
|
||
CategoryID *int64 `json:"category_id,omitempty"`
|
||
Description string `json:"description"`
|
||
Goal string `json:"goal"`
|
||
LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务的 LLM 配置;省略/null=用激活配置
|
||
LLMProfileIDs []int64 `json:"llm_profile_ids,omitempty"` // 有序任务级配置链;第一项初始生效
|
||
SourceTaskIDs []string `json:"source_task_ids,omitempty"` // 仅直接、只读继承的来源任务
|
||
CompanyIDs []int64 `json:"company_ids,omitempty"` // 关联企业范围并快照关联当前企业资产;不复制资产或强制生成意图
|
||
TimeoutSeconds int `json:"timeout_seconds"` // 任务级超时(秒);0/省略=不限时
|
||
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒);0/省略=默认600(10min);下限=默认=600,低于自动抬到600
|
||
SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // 创建时直接下发一条种子意图(内容=描述+目标),让 worker 免等首轮 planner 直接开跑;省略/null=默认关闭,走标准先规划再执行。显式传 true 才开(CTF 常一 work 解决时可省掉开跑前的 planner 轮)。
|
||
CoverageEnabled *bool `json:"coverage_enabled,omitempty"` // 资产覆盖度功能;省略/null=默认开(true)。false=关闭覆盖度计算/展示/自动累积范围+隐藏 add_task_scope/list_untested_assets。company 关联不受影响。
|
||
// InterceptRules 任务级资产拦截/允许规则(创建时录入,存 task_intercept_rules,不进全局表)。
|
||
InterceptRules []taskInterceptRuleReq `json:"intercept_rules,omitempty"`
|
||
}
|
||
|
||
func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
|
||
var req createTaskReq
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, "bad json: "+err.Error())
|
||
return
|
||
}
|
||
if strings.TrimSpace(req.Description) == "" {
|
||
req.Description = "未命名任务"
|
||
}
|
||
if len(req.LLMProfileIDs) == 0 && req.LLMProfileID != nil {
|
||
req.LLMProfileIDs = []int64{*req.LLMProfileID}
|
||
}
|
||
if err := s.validateTaskProfileIDs(req.LLMProfileIDs); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
if req.TimeoutSeconds < 0 {
|
||
req.TimeoutSeconds = 0
|
||
}
|
||
if len(req.SourceTaskIDs) > db.MaxTaskSourceCount {
|
||
writeErr(w, 400, fmt.Sprintf(errCreateTaskSourceLimit, db.MaxTaskSourceCount))
|
||
return
|
||
}
|
||
sourceIDs := make([]int64, 0, len(req.SourceTaskIDs))
|
||
seenSources := map[int64]bool{}
|
||
for _, raw := range req.SourceTaskIDs {
|
||
id, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
|
||
if err != nil || id <= 0 || seenSources[id] {
|
||
writeErr(w, 400, errCreateTaskSourceInvalid)
|
||
return
|
||
}
|
||
if _, ok := s.m.Task(strconv.FormatInt(id, 10)); !ok {
|
||
writeErr(w, 400, fmt.Sprintf(errCreateTaskSourceNotFound, id))
|
||
return
|
||
}
|
||
seenSources[id] = true
|
||
sourceIDs = append(sourceIDs, id)
|
||
}
|
||
companyIDs, err := db.NormalizeTaskCompanyIDs(req.CompanyIDs)
|
||
if err != nil {
|
||
writeErr(w, 400, fmt.Sprintf(errCreateTaskCompanyLimit, db.MaxTaskCompanyCount))
|
||
return
|
||
}
|
||
req.CompanyIDs = companyIDs
|
||
interceptRules, err := buildTaskInterceptRules(req.InterceptRules)
|
||
if err != nil {
|
||
writeErr(w, 400, errCreateTaskInterceptRules+err.Error())
|
||
return
|
||
}
|
||
t, err := s.m.CreateTaskWithOptions(req.Description, req.Goal, db.TaskCreateOptions{
|
||
Name: strings.TrimSpace(req.Name), CategoryID: req.CategoryID,
|
||
SourceTaskIDs: sourceIDs, CompanyIDs: req.CompanyIDs, LLMProfileIDs: req.LLMProfileIDs,
|
||
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds,
|
||
CoverageEnabled: req.CoverageEnabled,
|
||
InterceptRules: interceptRules,
|
||
})
|
||
if err != nil {
|
||
if errors.Is(err, db.ErrTaskCategoryInvalid) || errors.Is(err, db.ErrTaskCategoryNotFound) {
|
||
writeErr(w, 400, errCreateTaskCategoryInvalid)
|
||
return
|
||
}
|
||
if errors.Is(err, db.ErrTaskCompanyIDsInvalid) || errors.Is(err, db.ErrTaskCompanyNotFound) {
|
||
writeErr(w, 400, errCreateTaskCompanyInvalid)
|
||
return
|
||
}
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
log.Printf("[task] 新建任务 #%s «%s» 目标: %s", t.ID, req.Description, req.Goal)
|
||
// 共享的建后流程(seed + 种子意图 + 后台目标分解 + engine.Run),与 spawn_task 复用同一段。
|
||
// launchTask 内部异步,不阻塞 UI —— 目标分解在后台可见地进行。
|
||
s.launchTask(t, req.Description+" "+req.Goal, req.SeedFirstIntent != nil && *req.SeedFirstIntent)
|
||
writeJSON(w, 201, taskDTO(t, s.resolvedTaskStatus(t)))
|
||
}
|
||
|
||
func (s *Server) validateTaskProfileIDs(ids []int64) error {
|
||
seen := map[int64]bool{}
|
||
for _, id := range ids {
|
||
if id <= 0 || seen[id] {
|
||
return errors.New(errLLMProfileInvalid)
|
||
}
|
||
seen[id] = true
|
||
if _, ok := s.loadProfileConfig(id); !ok {
|
||
return fmt.Errorf(errLLMProfileNotFound, id)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Server) updateTaskLLMProfiles(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
// 任何生命周期状态(含终态)都可以改链:任务结束后主 Agent 对话仍走这条链,
|
||
// 模型不可用时不换链就等于把已完成任务的交互一起锁死。
|
||
before := t.llmStateSnapshot()
|
||
var req struct {
|
||
LLMProfileIDs []int64 `json:"llm_profile_ids"`
|
||
ActiveLLMProfileID *int64 `json:"active_llm_profile_id"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, "bad json: "+err.Error())
|
||
return
|
||
}
|
||
if err := s.validateTaskProfileIDs(req.LLMProfileIDs); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
active := int64(0)
|
||
if req.ActiveLLMProfileID != nil {
|
||
active = *req.ActiveLLMProfileID
|
||
}
|
||
reopened, err := s.m.ReplaceTaskLLMProfiles(t.ID, req.LLMProfileIDs, active)
|
||
if err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
// Profile edits affect only subsequent LLM calls. Existing in-flight calls
|
||
// retain their provider. Concurrency reconciliation treats ActiveLLMCalls as a
|
||
// live slot and postpones an unavailable task's pause/queue transition until
|
||
// the call returns; a nudge lets an idle running task resume promptly.
|
||
t.Notify()
|
||
go s.reconcileConcurrency()
|
||
llmState := t.llmStateSnapshot()
|
||
result := map[string]any{
|
||
"id": t.ID, "llm_profile_ids": llmState.ProfileIDs,
|
||
"active_llm_profile_id": llmState.ActiveID,
|
||
"llm_failover_state": llmState.FailoverState, "reopened_intents": reopened,
|
||
}
|
||
if !sameOptionalID(before.ActiveID, llmState.ActiveID) {
|
||
event := s.emitManualTaskLLMSwitch(t, before.ActiveID, llmState.ActiveID)
|
||
result["switch_event"] = activityDTO(event)
|
||
}
|
||
writeJSON(w, 200, result)
|
||
}
|
||
|
||
var (
|
||
reURL = regexp.MustCompile(`https?://[^\s'"]+`)
|
||
reIPPort = regexp.MustCompile(`\b((?:\d{1,3}\.){3}\d{1,3})(?::(\d{1,5}))?`)
|
||
reDomain = regexp.MustCompile(`\b((?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,})(?::(\d{1,5}))?`)
|
||
)
|
||
|
||
// parseTarget extracts a target (scheme, host, port) from free text, supporting
|
||
// full URLs, IP[:port] and domain[:port]. ok=false when nothing parseable.
|
||
func parseTarget(text string) (scheme, host string, port int, ok bool) {
|
||
text = strings.TrimSpace(text)
|
||
if m := reURL.FindString(text); m != "" {
|
||
if u, err := url.Parse(m); err == nil && u.Hostname() != "" {
|
||
scheme = strings.ToLower(u.Scheme)
|
||
host = strings.ToLower(u.Hostname())
|
||
port = portOr(u.Port(), defaultPort(scheme))
|
||
return scheme, host, port, true
|
||
}
|
||
}
|
||
if m := reIPPort.FindStringSubmatch(text); m != nil {
|
||
host = m[1]
|
||
port = portOr(m[2], 80)
|
||
return schemeForPort(port), host, port, true
|
||
}
|
||
if m := reDomain.FindStringSubmatch(text); m != nil {
|
||
host = strings.ToLower(m[1])
|
||
if m[2] == "" {
|
||
return "https", host, 443, true
|
||
}
|
||
port = portOr(m[2], 443)
|
||
return schemeForPort(port), host, port, true
|
||
}
|
||
return "", "", 0, false
|
||
}
|
||
|
||
func portOr(s string, d int) int {
|
||
if n, err := strconv.Atoi(s); err == nil && n > 0 {
|
||
return n
|
||
}
|
||
return d
|
||
}
|
||
func defaultPort(scheme string) int {
|
||
if scheme == "http" {
|
||
return 80
|
||
}
|
||
return 443
|
||
}
|
||
func schemeForPort(p int) string {
|
||
if p == 443 || p == 8443 {
|
||
return "https"
|
||
}
|
||
return "http"
|
||
}
|
||
|
||
// llmHost returns the host of the configured LLM endpoint (to keep it out of scope).
|
||
func (s *Server) llmHost() string {
|
||
s.cfgMu.Lock()
|
||
base := s.llmCfg.BaseURL
|
||
s.cfgMu.Unlock()
|
||
if base == "" {
|
||
return ""
|
||
}
|
||
if u, err := url.Parse(base); err == nil {
|
||
return strings.ToLower(u.Hostname())
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func (s *Server) seed(t *Task, text string) {
|
||
scheme, host, port, ok := parseTarget(text)
|
||
if !ok {
|
||
log.Printf("[seed] task %s: 未能从 %q 解析出目标 host/IP，不创建站点（请手动配置 scope）", t.ID, text)
|
||
return
|
||
}
|
||
// P0-1 guard: never treat the configured LLM gateway as a target.
|
||
if gw := s.llmHost(); gw != "" && host == gw {
|
||
log.Printf("[seed] task %s: 目标 %q 是 LLM 网关，拒绝作为渗透目标", t.ID, host)
|
||
return
|
||
}
|
||
|
||
u := scheme + "://" + host
|
||
if !(scheme == "https" && port == 443) && !(scheme == "http" && port == 80) {
|
||
u += ":" + strconv.Itoa(port)
|
||
}
|
||
var rootID int64
|
||
if as := s.m.Assets(); as != nil {
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
if net.ParseIP(host) != nil {
|
||
rootID, _ = as.UpsertIP(db.UpsertIPReq{IP: host, TaskID: taskID})
|
||
} else if scheme == "https" || scheme == "http" {
|
||
rootID, _ = as.UpsertHTTPService(db.UpsertHTTPServiceReq{URL: u, TaskID: taskID})
|
||
} else {
|
||
rootID, _ = as.UpsertRootDomain(db.UpsertRootDomainReq{Domain: host, TaskID: taskID})
|
||
}
|
||
if rootID > 0 {
|
||
_ = as.SetTaskAssetSource(taskID, rootID, "task", taskAssetSourceTaskSummary, nil)
|
||
}
|
||
}
|
||
// anchor the seeded assets to this task's begin root as lineage/provenance
|
||
// (the asset graph is global and shared; anchoring no longer gates reads).
|
||
if rootID > 0 {
|
||
if begin, _ := t.Store.OriginFactID(); begin > 0 {
|
||
_ = t.Store.Anchor(begin, rootID)
|
||
}
|
||
}
|
||
log.Printf("[seed] task %s: 目标站点 %s", t.ID, u)
|
||
// 不在这里 Notify:首轮是否触发统一由 engine.Run 的 HasActiveIntent 决定(种子意图任务
|
||
// 跳过首轮)。seed 早于 Run 执行,若在此 Notify 会 buffered 到通道、被 plannerLoop 启动时
|
||
// 消费掉而绕过 Run 的门控 → 种子任务仍误触发首轮。
|
||
}
|
||
|
||
// seedFirstIntent writes ONE open intent (summary = 描述+目标) into the task's
|
||
// frontier at creation, so a worker can claim and run it immediately without first
|
||
// waiting a planner round. Mirrors a planner top-level intent: it links from the
|
||
// origin fact (RelDerivedFrom) so it still traces back to a fact node. Best-effort —
|
||
// a failure just falls back to the normal planner-driven flow.
|
||
func (s *Server) seedFirstIntent(t *Task) {
|
||
summary := fmt.Sprintf("完成任务目标：%s（任务：%s）", t.Goal, t.Description)
|
||
id, err := t.Store.AddIntent(map[string]any{"summary": summary}, 8, nil, "seed")
|
||
if err != nil {
|
||
log.Printf("[seed] task %s: 下发种子意图失败: %v", t.ID, err)
|
||
return
|
||
}
|
||
if origin, _ := t.Store.OriginFactID(); origin > 0 {
|
||
_ = t.Store.Link(origin, db.RelDerivedFrom, id)
|
||
}
|
||
log.Printf("[seed] task %s: 已下发种子意图 #%d", t.ID, id)
|
||
}
|
||
|
||
func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
dto := taskDTO(t, s.resolvedTaskStatus(t))
|
||
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
|
||
applyTaskArchiveBlocker(&dto, archiveBlockers)
|
||
writeJSON(w, 200, dto)
|
||
}
|
||
|
||
// taskCoverage returns a task's rough asset test coverage (denominator/tested/backlog).
|
||
func (s *Server) taskCoverage(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
as := s.m.Assets()
|
||
if as == nil {
|
||
writeErr(w, 503, errAssetStoreDisabled)
|
||
return
|
||
}
|
||
// 资产覆盖度功能关闭 → 短路返回 {enabled:false}，前端据此隐藏覆盖度卡片/进度。
|
||
if !t.CoverageEnabled {
|
||
writeJSON(w, 200, &db.Coverage{Enabled: false, ByType: []db.CoverageByType{}})
|
||
return
|
||
}
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
cov, err := as.TaskCoverageWithSources(taskID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
cov.Enabled = true
|
||
writeJSON(w, 200, cov)
|
||
}
|
||
|
||
// taskCoverageGraph returns the force-directed asset coverage graph for a task:
|
||
// all in-scope assets (每种类型) + 连接用的根域名/公司节点, each carrying tested/in_scope.
|
||
func (s *Server) taskCoverageGraph(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
as := s.m.Assets()
|
||
if as == nil {
|
||
writeErr(w, 503, errAssetStoreDisabled)
|
||
return
|
||
}
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
g, err := as.BuildCoverageGraph(taskID, t.ExpID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, g)
|
||
}
|
||
|
||
// taskAssetRefs returns the intents / facts / findings in this task anchored to a
|
||
// given asset id — powers the coverage-graph node drawer's「关联意图 / 关联事实」。
|
||
func (s *Server) taskAssetRefs(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
assetID, _ := strconv.ParseInt(r.URL.Query().Get("asset_id"), 10, 64)
|
||
if assetID <= 0 {
|
||
writeErr(w, 400, errAssetIDRequired)
|
||
return
|
||
}
|
||
refs, err := t.Store.AssetRefsWithSources(assetID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
intents := []CoverageAssetRefDTO{}
|
||
facts := []CoverageAssetRefDTO{}
|
||
findings := []CoverageAssetRefDTO{}
|
||
for _, ref := range refs {
|
||
dto := coverageAssetRefDTO(ref)
|
||
switch ref.Kind {
|
||
case "intent":
|
||
intents = append(intents, dto)
|
||
case "fact":
|
||
facts = append(facts, dto)
|
||
case "finding":
|
||
findings = append(findings, dto)
|
||
}
|
||
}
|
||
writeJSON(w, 200, map[string]any{"intents": intents, "facts": facts, "findings": findings})
|
||
}
|
||
|
||
// taskScopeList returns a task's scope rows (coverage denominator sources).
|
||
func (s *Server) taskScopeList(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
as := s.m.Assets()
|
||
if as == nil {
|
||
writeErr(w, 503, errAssetStoreDisabled)
|
||
return
|
||
}
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
rows, err := as.ListTaskScopeWithSources(taskID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"scope": rows})
|
||
}
|
||
|
||
func (s *Server) taskScopeAdd(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
as := s.m.Assets()
|
||
if as == nil {
|
||
writeErr(w, 503, errAssetStoreDisabled)
|
||
return
|
||
}
|
||
var body struct {
|
||
Kind string `json:"kind"`
|
||
Value string `json:"value"`
|
||
Reason string `json:"reason"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeErr(w, 400, "invalid JSON")
|
||
return
|
||
}
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
ts, err := as.AddAgentScope(taskID, body.Kind, body.Value, body.Reason, "manual")
|
||
if err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, ts)
|
||
}
|
||
|
||
func (s *Server) taskScopeDelete(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := s.m.Task(r.PathValue("id"))
|
||
if !ok {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
as := s.m.Assets()
|
||
if as == nil {
|
||
writeErr(w, 503, errAssetStoreDisabled)
|
||
return
|
||
}
|
||
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
scopeID, err := strconv.ParseInt(r.PathValue("sid"), 10, 64)
|
||
if err != nil {
|
||
writeErr(w, 400, "invalid scope id")
|
||
return
|
||
}
|
||
deleted, err := as.DeleteTaskScope(taskID, scopeID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if !deleted {
|
||
writeErr(w, 404, "scope row not found")
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]bool{"ok": true})
|
||
}
|
||
|
||
func (s *Server) frontier(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
fr, _ := t.Store.Frontier(atoiDefault(r.URL.Query().Get("limit"), 100))
|
||
writeJSON(w, 200, taskNodeDTOs(fr))
|
||
}
|
||
|
||
func (s *Server) findings(w http.ResponseWriter, r *http.Request) {
|
||
// 无 task 参数 → 全局「发现」页：从独立 findings 表读取（任务删除后 finding 依然保留）。
|
||
// 带 task 参数 → 仅该任务（任务概览/发现 Tab 用），从 exploration_nodes 读（任务在则节点在）。
|
||
q := r.URL.Query()
|
||
taskParam := q.Get("task")
|
||
if taskParam == "" {
|
||
// 带 page/limit → 服务端分页 {items,total,...}；不带 → 裸数组（dashboard 汇总用，
|
||
// 与 intents 端点的兼容策略一致）。筛选/排序统一下推到 SQL。
|
||
if q.Get("page") == "" && q.Get("limit") == "" {
|
||
fs, _ := s.m.pg.ListFindings(500)
|
||
assets := s.resolveFindingAssets(fs)
|
||
out := make([]FindingDTO, 0, len(fs))
|
||
for _, f := range fs {
|
||
out = append(out, findingFromDB(f, assets))
|
||
}
|
||
writeJSON(w, 200, out)
|
||
return
|
||
}
|
||
page := findingPaginationParam(q.Get("page"), 1, 0)
|
||
limit := findingPaginationParam(q.Get("limit"), 20, 200)
|
||
fs, total, err := s.m.pg.ListFindingsPage(findingFilterFromQuery(q), page, limit)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
assets := s.resolveFindingAssets(fs)
|
||
out := make([]FindingDTO, 0, len(fs))
|
||
for _, f := range fs {
|
||
out = append(out, findingFromDB(f, assets))
|
||
}
|
||
writeJSON(w, 200, map[string]any{"items": out, "total": total, "page": page, "page_size": limit})
|
||
return
|
||
}
|
||
t := s.m.ResolveTask(taskParam)
|
||
if t == nil {
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
f, err := t.Store.ListByKind(db.KindFinding, 200)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
tid, _ := strconv.ParseInt(t.ID, 10, 64)
|
||
meta, err := s.m.pg.FindingMetaByNodeID(tid)
|
||
if err != nil {
|
||
log.Printf("[findings] task=%s meta: %v", t.ID, err)
|
||
}
|
||
var aidSet []int64
|
||
for _, m := range meta {
|
||
aidSet = append(aidSet, m.AssetIDs...)
|
||
}
|
||
assets := s.resolveAssetIDs(aidSet)
|
||
out := findingDTOsForTask(t, f, meta, assets)
|
||
|
||
// Related-task findings are a live, read-only view. Provenance metadata lets
|
||
// the task UI suppress mutation affordances while retaining stable ids.
|
||
sources, err := t.Store.DirectSourceStores()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
for _, source := range sources {
|
||
nodes, listErr := source.Store.ListByKind(db.KindFinding, 200)
|
||
if listErr != nil {
|
||
writeErr(w, 500, listErr.Error())
|
||
return
|
||
}
|
||
for _, node := range nodes {
|
||
node.SourceTaskID = source.Task.TaskID
|
||
node.Inherited = true
|
||
}
|
||
sourceMeta, metaErr := s.m.pg.FindingMetaByNodeID(source.Task.TaskID)
|
||
if metaErr != nil {
|
||
log.Printf("[findings] source_task=%d meta: %v", source.Task.TaskID, metaErr)
|
||
}
|
||
var sourceAssetIDs []int64
|
||
for _, item := range sourceMeta {
|
||
sourceAssetIDs = append(sourceAssetIDs, item.AssetIDs...)
|
||
}
|
||
out = append(out, findingDTOsForOwner(
|
||
i64s(source.Task.TaskID),
|
||
source.Task.Description,
|
||
nodes,
|
||
sourceMeta,
|
||
s.resolveAssetIDs(sourceAssetIDs),
|
||
)...)
|
||
}
|
||
writeJSON(w, 200, out)
|
||
}
|
||
|
||
// normFilter maps the frontend's "all" sentinel (and empty) to "" so the DB layer
|
||
// treats it as no filter.
|
||
func normFilter(v string) string {
|
||
if v == "all" {
|
||
return ""
|
||
}
|
||
return v
|
||
}
|
||
|
||
// resolveFindingAssets loads every asset anchored by the given findings, keyed by
|
||
// id, so each finding DTO can render its assets' labels.
|
||
func (s *Server) resolveFindingAssets(fs []*db.DBFinding) map[int64]*db.Asset {
|
||
var ids []int64
|
||
for _, f := range fs {
|
||
ids = append(ids, f.AssetIDs...)
|
||
}
|
||
return s.resolveAssetIDs(ids)
|
||
}
|
||
|
||
// resolveAssetIDs de-dupes ids and loads their asset rows into an id→asset map.
|
||
func (s *Server) resolveAssetIDs(ids []int64) map[int64]*db.Asset {
|
||
assets := map[int64]*db.Asset{}
|
||
seen := map[int64]bool{}
|
||
var uniq []int64
|
||
for _, id := range ids {
|
||
if id > 0 && !seen[id] {
|
||
seen[id] = true
|
||
uniq = append(uniq, id)
|
||
}
|
||
}
|
||
if len(uniq) == 0 {
|
||
return assets
|
||
}
|
||
list, err := s.m.pg.Assets().GetByIDs(uniq)
|
||
if err != nil {
|
||
log.Printf("[findings] resolve assets: %v", err)
|
||
}
|
||
for _, a := range list {
|
||
assets[a.ID] = a
|
||
}
|
||
return assets
|
||
}
|
||
|
||
// findingStats serves the whole-table aggregates (stat cards + vuln-class filter)
|
||
// for the paginated 发现 page.
|
||
func (s *Server) findingStats(w http.ResponseWriter, r *http.Request) {
|
||
st, err := s.m.pg.FindingStats()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, st)
|
||
}
|
||
|
||
// getFinding returns one finding by its standalone-table id (DTO finding_id),
|
||
// with anchored assets resolved for display.
|
||
func (s *Server) getFinding(w http.ResponseWriter, r *http.Request) {
|
||
id := int64(atoiDefault(r.PathValue("id"), 0))
|
||
if id <= 0 {
|
||
writeErr(w, 400, "bad finding id")
|
||
return
|
||
}
|
||
f, err := s.m.pg.GetFinding(id)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if f == nil {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
assets := s.resolveAssetIDs(f.AssetIDs)
|
||
dto := findingFromDB(f, assets)
|
||
if contextTaskID := strings.TrimSpace(r.URL.Query().Get("context_task")); contextTaskID != "" {
|
||
contextTask := s.m.ResolveTask(contextTaskID)
|
||
if contextTask == nil {
|
||
writeErr(w, 404, "context task not found")
|
||
return
|
||
}
|
||
sourceTaskID, inherited, allowed := findingProvenanceInTask(contextTask, f.TaskID)
|
||
if !allowed {
|
||
writeErr(w, 404, "finding not available in task context")
|
||
return
|
||
}
|
||
dto.SourceTaskID = sourceTaskID
|
||
dto.Inherited = inherited
|
||
}
|
||
writeJSON(w, 200, dto)
|
||
}
|
||
|
||
// findingsExport 导出发现页的漏洞。
|
||
//
|
||
// scope = filtered（沿用页面筛选）| all（全部）| selected（勾选的 ids）
|
||
// format = md-single（整合一份 .md）| md-zip（一漏洞一 .md,打包 zip）
|
||
// | csv | json
|
||
// ids = 逗号分隔的 finding id（scope=selected 时必填）
|
||
// 筛选参数 severity/status/vulnclass/task_id/q/sort 与列表接口一致（scope=filtered 用）。
|
||
func (s *Server) findingsExport(w http.ResponseWriter, r *http.Request) {
|
||
q := r.URL.Query()
|
||
scope := q.Get("scope")
|
||
format := q.Get("format")
|
||
|
||
var ids []int64
|
||
var filter db.FindingFilter
|
||
switch scope {
|
||
case "selected":
|
||
for _, part := range strings.Split(q.Get("ids"), ",") {
|
||
part = strings.TrimSpace(part)
|
||
if part == "" {
|
||
continue
|
||
}
|
||
id, err := strconv.ParseInt(part, 10, 64)
|
||
if err != nil || id <= 0 {
|
||
writeErr(w, 400, "bad finding id: "+part)
|
||
return
|
||
}
|
||
ids = append(ids, id)
|
||
}
|
||
if len(ids) == 0 {
|
||
writeErr(w, 400, "no findings selected")
|
||
return
|
||
}
|
||
case "all":
|
||
// 空 filter = 不加任何条件。
|
||
case "filtered", "":
|
||
filter = findingFilterFromQuery(q)
|
||
default:
|
||
writeErr(w, 400, "bad scope: "+scope)
|
||
return
|
||
}
|
||
|
||
fs, err := s.m.pg.ListFindingsForExport(filter, ids)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
|
||
stage, err := os.MkdirTemp("", "artex-finding-export-")
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
defer os.RemoveAll(stage)
|
||
if err = s.evidenceStore().StageFindingsExport(r.Context(), fs, stage, format == "md-zip"); err != nil {
|
||
evidenceError(w, err)
|
||
return
|
||
}
|
||
now := time.Now()
|
||
stamp := now.Format("20060102-150405")
|
||
setDownload := func(contentType, filename string) {
|
||
w.Header().Set("Content-Type", contentType)
|
||
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
|
||
}
|
||
|
||
switch format {
|
||
case "md-single":
|
||
setDownload("text/markdown; charset=utf-8", "findings-"+stamp+".md")
|
||
_, _ = w.Write([]byte(report.FindingsMarkdown(fs, now)))
|
||
case "md-zip":
|
||
path := filepath.Join(stage, "findings.zip")
|
||
if err := buildFindingsEvidenceZip(path, fs, stage, now); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
file, err := os.Open(path)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
defer file.Close()
|
||
setDownload("application/zip", "findings-"+stamp+".zip")
|
||
http.ServeContent(w, r, "findings.zip", now, file)
|
||
case "csv":
|
||
setDownload("text/csv; charset=utf-8", "findings-"+stamp+".csv")
|
||
_, _ = w.Write(report.FindingsCSV(fs))
|
||
case "json":
|
||
assets := s.resolveFindingAssets(fs)
|
||
out := make([]FindingDTO, 0, len(fs))
|
||
for _, f := range fs {
|
||
for i := range f.TrafficBindings {
|
||
f.TrafficBindings[i].Snapshot.ReqHead = ""
|
||
f.TrafficBindings[i].Snapshot.RespHead = ""
|
||
}
|
||
out = append(out, findingFromDB(f, assets))
|
||
}
|
||
setDownload("application/json; charset=utf-8", "findings-"+stamp+".json")
|
||
enc := json.NewEncoder(w)
|
||
enc.SetIndent("", " ")
|
||
_ = enc.Encode(out)
|
||
default:
|
||
writeErr(w, 400, "bad format: "+format)
|
||
}
|
||
}
|
||
|
||
func findingProvenanceInTask(contextTask *Task, findingTaskID *int64) (sourceTaskID string, inherited, allowed bool) {
|
||
if contextTask == nil || findingTaskID == nil {
|
||
return "", false, false
|
||
}
|
||
contextID, err := strconv.ParseInt(contextTask.ID, 10, 64)
|
||
if err != nil {
|
||
return "", false, false
|
||
}
|
||
if *findingTaskID == contextID {
|
||
return "", false, true
|
||
}
|
||
for _, sourceID := range contextTask.lifecycleSnapshot().SourceTaskIDs {
|
||
if sourceID == *findingTaskID {
|
||
return i64s(sourceID), true, true
|
||
}
|
||
}
|
||
return "", false, false
|
||
}
|
||
|
||
// findingLineage returns the exploration sub-DAG from the task root to this
|
||
// finding's node — the finding node + all its ancestors + edges among them — so
|
||
// the detail page can show "how this finding was reached". Empty {nodes,edges}
|
||
// when the finding has no node/task (e.g. the originating task was deleted).
|
||
func (s *Server) findingLineage(w http.ResponseWriter, r *http.Request) {
|
||
id := int64(atoiDefault(r.PathValue("id"), 0))
|
||
if id <= 0 {
|
||
writeErr(w, 400, "bad finding id")
|
||
return
|
||
}
|
||
f, err := s.m.pg.GetFinding(id)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if f == nil {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
empty := map[string]any{"nodes": []any{}, "edges": []any{}}
|
||
if f.NodeID == nil || f.TaskID == nil {
|
||
writeJSON(w, 200, empty)
|
||
return
|
||
}
|
||
t := s.m.ResolveTask(i64s(*f.TaskID))
|
||
if t == nil {
|
||
writeJSON(w, 200, empty)
|
||
return
|
||
}
|
||
nodes, edges, err := t.Store.FindingLineage(*f.NodeID)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"nodes": taskNodeDTOs(nodes), "edges": edgeDTOs(edges)})
|
||
}
|
||
|
||
// patchFinding partially updates a finding: any subset of {status, severity}.
|
||
// The id is the standalone findings-table id (DTO finding_id). Severity edits are
|
||
// mirrored onto the originating exploration node so the per-task view stays in
|
||
// sync. Returns the updated finding DTO.
|
||
func (s *Server) patchFinding(w http.ResponseWriter, r *http.Request) {
|
||
id := int64(atoiDefault(r.PathValue("id"), 0))
|
||
if id <= 0 {
|
||
writeErr(w, 400, "bad finding id")
|
||
return
|
||
}
|
||
var body struct {
|
||
Status *string `json:"status"`
|
||
Severity *string `json:"severity"`
|
||
Name *string `json:"name"`
|
||
VulnClass *string `json:"vulnclass"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeErr(w, 400, "bad json: "+err.Error())
|
||
return
|
||
}
|
||
if body.Status == nil && body.Severity == nil && body.Name == nil && body.VulnClass == nil {
|
||
writeErr(w, 400, "nothing to update: provide status/severity/name/vulnclass")
|
||
return
|
||
}
|
||
if body.Status != nil {
|
||
if !db.ValidFindingStatus(*body.Status) {
|
||
writeErr(w, 400, "bad status: "+*body.Status)
|
||
return
|
||
}
|
||
// 走带通知的版本：状态更新与「状态变更推送事件」在同一事务里落库，
|
||
// 避免出现状态已改而推送事件丢失的窗口。事件登记失败不影响状态更新，
|
||
// 所以只记日志、不向调用方报错。
|
||
from, found, notified, err := s.m.pg.SetFindingStatusWithNotify(r.Context(), id, *body.Status)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if !found {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
if !notified && from != *body.Status {
|
||
log.Printf("[notify] 状态变更事件未登记 finding=%d %s→%s（状态已更新）", id, from, *body.Status)
|
||
}
|
||
}
|
||
if body.Severity != nil {
|
||
if !db.ValidSeverity(*body.Severity) {
|
||
writeErr(w, 400, "bad severity: "+*body.Severity)
|
||
return
|
||
}
|
||
n, err := s.m.pg.SetFindingSeverity(id, *body.Severity)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if n == 0 {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
}
|
||
if body.Name != nil {
|
||
n, err := s.m.pg.SetFindingName(id, strings.TrimSpace(*body.Name))
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if n == 0 {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
}
|
||
if body.VulnClass != nil {
|
||
n, err := s.m.pg.SetFindingVulnClass(id, strings.TrimSpace(*body.VulnClass))
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if n == 0 {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
}
|
||
f, err := s.m.pg.GetFinding(id)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if f == nil {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
writeJSON(w, 200, findingFromDB(f, s.resolveAssetIDs(f.AssetIDs)))
|
||
}
|
||
|
||
// deleteFinding removes a finding (findings row + originating exploration node).
|
||
// id is the standalone findings-table id (DTO finding_id).
|
||
func (s *Server) deleteFinding(w http.ResponseWriter, r *http.Request) {
|
||
id := int64(atoiDefault(r.PathValue("id"), 0))
|
||
if id <= 0 {
|
||
writeErr(w, 400, "bad finding id")
|
||
return
|
||
}
|
||
n, err := s.m.pg.DeleteFinding(id)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if n == 0 {
|
||
writeErr(w, 404, "finding not found")
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"deleted": true, "id": id})
|
||
}
|
||
|
||
func (s *Server) intents(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
// Back-compat: bare list shape when the task can't be resolved.
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
q := r.URL.Query()
|
||
limit := min(atoiDefault(q.Get("limit"), 300), 500)
|
||
// No paging params → preserve the legacy bare-array response so existing callers
|
||
// (and the poll) keep working unchanged.
|
||
if q.Get("before") == "" && q.Get("page") == "" {
|
||
in, err := t.Store.ListByKind(db.KindIntent, limit)
|
||
if err != nil {
|
||
log.Printf("[intents] task=%s limit=%d: %v", t.ID, limit, err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, taskNodeDTOs(in))
|
||
return
|
||
}
|
||
// Paged form: ?before=<id> (or ?page as a marker) → {items, has_more} so the
|
||
// worker session list can reach past the old fixed 300 boundary on scroll. It
|
||
// also includes immutable intent results from directly related tasks; those
|
||
// rows never participate in this task's frontier or claim path.
|
||
before := int64(atoiDefault(q.Get("before"), 0))
|
||
in, hasMore, err := taskIntentHistoryPage(t.Store, before, limit, false, 0)
|
||
if err != nil {
|
||
log.Printf("[intents] task=%s before=%d limit=%d: %v", t.ID, before, limit, err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
sources, err := t.Store.DirectSourceStores()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
for _, source := range sources {
|
||
items, sourceMore, sourceErr := taskIntentHistoryPage(source.Store, before, limit, true, source.Task.TaskID)
|
||
if sourceErr != nil {
|
||
writeErr(w, 500, sourceErr.Error())
|
||
return
|
||
}
|
||
in = append(in, items...)
|
||
hasMore = hasMore || sourceMore
|
||
}
|
||
sort.Slice(in, func(i, j int) bool { return in[i].ID > in[j].ID })
|
||
if len(in) > limit {
|
||
in = in[:limit]
|
||
hasMore = true
|
||
}
|
||
writeJSON(w, 200, map[string]any{"items": taskNodeDTOs(in), "has_more": hasMore})
|
||
}
|
||
|
||
func inheritedIntentResult(state string) bool {
|
||
switch state {
|
||
case "done", "blocked", "exhausted", "stopped":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// inheritedGraphSnapshot exposes immutable source-task history without leaking
|
||
// work that is still open or running. Edges are filtered with the nodes so the
|
||
// response cannot contain dangling ids that reveal hidden live intent topology.
|
||
func inheritedGraphSnapshot(nodes []*db.Node, edges []db.Edge, sourceTaskID int64) ([]*db.Node, []db.Edge) {
|
||
visible := make(map[int64]struct{}, len(nodes))
|
||
filteredNodes := make([]*db.Node, 0, len(nodes))
|
||
for _, node := range nodes {
|
||
if node == nil || (node.Kind == db.KindIntent && !inheritedIntentResult(node.State)) {
|
||
continue
|
||
}
|
||
node.SourceTaskID = sourceTaskID
|
||
node.Inherited = true
|
||
visible[node.ID] = struct{}{}
|
||
filteredNodes = append(filteredNodes, node)
|
||
}
|
||
filteredEdges := make([]db.Edge, 0, len(edges))
|
||
for _, edge := range edges {
|
||
if _, ok := visible[edge.From]; !ok {
|
||
continue
|
||
}
|
||
if _, ok := visible[edge.To]; !ok {
|
||
continue
|
||
}
|
||
filteredEdges = append(filteredEdges, edge)
|
||
}
|
||
return filteredNodes, filteredEdges
|
||
}
|
||
|
||
// taskIntentHistoryPage returns one newest-first page for an exploration.
|
||
// A source may have many open/running intents, so keep paging until enough
|
||
// historical results are collected instead of leaking its frontier into the UI.
|
||
func taskIntentHistoryPage(store *db.ExplorationStore, before int64, limit int, inherited bool, sourceTaskID int64) ([]*db.Node, bool, error) {
|
||
if !inherited {
|
||
return store.ListByKindPage(db.KindIntent, before, limit)
|
||
}
|
||
batch := max(limit, 300)
|
||
cursor := before
|
||
out := make([]*db.Node, 0, limit+1)
|
||
for {
|
||
page, more, err := store.ListByKindPage(db.KindIntent, cursor, batch)
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
for _, node := range page {
|
||
if !inheritedIntentResult(node.State) {
|
||
continue
|
||
}
|
||
node.SourceTaskID = sourceTaskID
|
||
node.Inherited = true
|
||
out = append(out, node)
|
||
if len(out) > limit {
|
||
return out[:limit], true, nil
|
||
}
|
||
}
|
||
if !more || len(page) == 0 {
|
||
return out, false, nil
|
||
}
|
||
cursor = page[len(page)-1].ID
|
||
}
|
||
}
|
||
|
||
// explorationGraph returns the whole exploration chain (task graph) as nodes+edges.
|
||
func (s *Server) explorationGraph(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, map[string]any{"nodes": []any{}, "edges": []any{}})
|
||
return
|
||
}
|
||
nodes, err := t.Store.Nodes(2000)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
edges, err := t.Store.Edges(5000)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
sources, err := t.Store.DirectSourceStores()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
for _, source := range sources {
|
||
sourceNodes, nodeErr := source.Store.Nodes(2000)
|
||
if nodeErr != nil {
|
||
writeErr(w, 500, nodeErr.Error())
|
||
return
|
||
}
|
||
sourceEdges, edgeErr := source.Store.Edges(5000)
|
||
if edgeErr != nil {
|
||
writeErr(w, 500, edgeErr.Error())
|
||
return
|
||
}
|
||
sourceNodes, sourceEdges = inheritedGraphSnapshot(sourceNodes, sourceEdges, source.Task.TaskID)
|
||
nodes = append(nodes, sourceNodes...)
|
||
edges = append(edges, sourceEdges...)
|
||
}
|
||
writeJSON(w, 200, map[string]any{"nodes": taskNodeDTOs(nodes), "edges": edgeDTOs(edges)})
|
||
}
|
||
|
||
// explorationNodes serves the 播报板: this task's own exploration nodes as a
|
||
// paged time series (newest first unless ?order=asc), filterable by kind/state
|
||
// and a payload substring. Inherited nodes are deliberately out of scope — the
|
||
// board reports what this task is doing right now, and paging across the source
|
||
// tasks' stores would make the cursor meaningless.
|
||
// Response also carries the edges touching the page plus the neighbour nodes
|
||
// they point at, so each row can state where it came from and what it produced.
|
||
func (s *Server) explorationNodes(w http.ResponseWriter, r *http.Request) {
|
||
q := r.URL.Query()
|
||
page := atoiDefault(q.Get("page"), 1)
|
||
size := atoiDefault(q.Get("size"), 20)
|
||
t := s.m.ResolveTask(q.Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, map[string]any{
|
||
"items": []any{}, "total": 0, "page": page, "size": size,
|
||
"edges": []any{}, "refs": map[string]any{},
|
||
})
|
||
return
|
||
}
|
||
filter := db.NodeFilter{
|
||
Kinds: csvValues(q.Get("kind")),
|
||
States: csvValues(q.Get("state")),
|
||
Query: q.Get("q"),
|
||
Asc: q.Get("order") == "asc",
|
||
}
|
||
nodes, total, err := t.Store.NodesPage(filter, page, size)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
ids := make([]int64, 0, len(nodes))
|
||
onPage := make(map[int64]bool, len(nodes))
|
||
for _, n := range nodes {
|
||
ids = append(ids, n.ID)
|
||
onPage[n.ID] = true
|
||
}
|
||
edges, err := t.Store.EdgesTouching(ids)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
neighbourSet := map[int64]bool{}
|
||
for _, e := range edges {
|
||
if !onPage[e.From] {
|
||
neighbourSet[e.From] = true
|
||
}
|
||
if !onPage[e.To] {
|
||
neighbourSet[e.To] = true
|
||
}
|
||
}
|
||
neighbourIDs := make([]int64, 0, len(neighbourSet))
|
||
for id := range neighbourSet {
|
||
neighbourIDs = append(neighbourIDs, id)
|
||
}
|
||
neighbours, err := t.Store.NodesByIDs(neighbourIDs)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
refs := make(map[string]TaskNodeDTO, len(neighbours))
|
||
for _, n := range neighbours {
|
||
refs[i64s(n.ID)] = taskNodeDTO(n)
|
||
}
|
||
writeJSON(w, 200, map[string]any{
|
||
"items": taskNodeDTOs(nodes),
|
||
"total": total,
|
||
"page": page,
|
||
"size": size,
|
||
"edges": edgeDTOs(edges),
|
||
"refs": refs,
|
||
"assets": s.nodeAnchoredAssets(t, append(ids, neighbourIDs...)),
|
||
})
|
||
}
|
||
|
||
// nodeAnchoredAssets resolves the exploration_anchors of the given nodes into
|
||
// display-ready asset labels, keyed by node id. Anchors are provenance decoration
|
||
// for the 播报板 — a failure here must not cost the caller its page, so errors are
|
||
// logged and degrade to "no assets".
|
||
func (s *Server) nodeAnchoredAssets(t *Task, nodeIDs []int64) map[string][]FindingAssetDTO {
|
||
out := map[string][]FindingAssetDTO{}
|
||
anchors, err := t.Store.NodeAssets(nodeIDs)
|
||
if err != nil {
|
||
log.Printf("[broadcast] node assets: %v", err)
|
||
return out
|
||
}
|
||
var flat []int64
|
||
for _, ids := range anchors {
|
||
flat = append(flat, ids...)
|
||
}
|
||
assets := s.resolveAssetIDs(flat)
|
||
for nodeID, ids := range anchors {
|
||
if dtos := findingAssetDTOs(ids, assets); len(dtos) > 0 {
|
||
out[i64s(nodeID)] = dtos
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// csvValues splits a comma-separated query parameter, dropping empty entries.
|
||
func csvValues(s string) []string {
|
||
out := []string{}
|
||
for _, part := range strings.Split(s, ",") {
|
||
if v := strings.TrimSpace(part); v != "" {
|
||
out = append(out, v)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// activity returns the worker execution step log (incremental via ?since=seq).
|
||
func (s *Server) activity(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, map[string]any{"items": []any{}, "cursor": 0})
|
||
return
|
||
}
|
||
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
|
||
limit := atoiDefault(r.URL.Query().Get("limit"), 300)
|
||
var intentPtr *int64
|
||
if iv := r.URL.Query().Get("intent"); iv != "" {
|
||
if n, err := strconv.ParseInt(iv, 10, 64); err == nil {
|
||
intentPtr = &n
|
||
}
|
||
}
|
||
var (
|
||
items []db.Activity
|
||
cursor int64
|
||
err error
|
||
)
|
||
if intentPtr != nil {
|
||
items, cursor, err = t.Store.ActivityListWithSources(*intentPtr, since, limit)
|
||
} else {
|
||
items, cursor, err = t.Store.ActivityList(nil, since, limit)
|
||
}
|
||
if err != nil {
|
||
log.Printf("[activity] task=%s since=%d limit=%d intent=%v: %v", t.ID, since, limit, intentPtr, err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"items": activityDTOs(items), "cursor": cursor})
|
||
}
|
||
|
||
// parseActivitySession maps a stable session key (main | plan | intent:<ID>) to a
|
||
// DB session filter. Goal Agent + Planner both live under worker="planner" (the
|
||
// single Plan session); a Worker session is one intent, keyed by its node id.
|
||
func parseActivitySession(sess string) (db.ActivitySessionFilter, bool) {
|
||
switch {
|
||
case sess == "" || sess == "main":
|
||
// 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:"):
|
||
id, err := strconv.ParseInt(strings.TrimPrefix(sess, "intent:"), 10, 64)
|
||
if err != nil {
|
||
return db.ActivitySessionFilter{}, false
|
||
}
|
||
return db.ActivitySessionFilter{NodeID: &id}, true
|
||
}
|
||
return db.ActivitySessionFilter{}, false
|
||
}
|
||
|
||
// activitySessionStore resolves a worker session against the current task and
|
||
// its direct sources. Planner/main always stay local; inherited intent sessions
|
||
// are immutable history and use the source exploration only for reads.
|
||
func activitySessionStore(t *Task, filter db.ActivitySessionFilter) (*db.ExplorationStore, int64, error) {
|
||
if filter.NodeID == nil {
|
||
return t.Store, 0, nil
|
||
}
|
||
node, err := t.Store.GetNodeWithSources(*filter.NodeID)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
if node == nil || node.Kind != db.KindIntent {
|
||
return nil, 0, nil
|
||
}
|
||
if !node.Inherited {
|
||
return t.Store, 0, nil
|
||
}
|
||
if !inheritedIntentResult(node.State) {
|
||
return nil, 0, nil
|
||
}
|
||
sources, err := t.Store.DirectSourceStores()
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for _, source := range sources {
|
||
if source.Task.TaskID == node.SourceTaskID {
|
||
return source.Store, source.Task.TaskID, nil
|
||
}
|
||
}
|
||
return nil, 0, nil
|
||
}
|
||
|
||
// activityHistory serves one reverse-paginated page of a session's activity history.
|
||
// The latest page (no ?before) opens a session; ?before=<id> pulls the older page on
|
||
// scroll-up. snapshot_cursor is the TASK-level max id at query time — the client uses
|
||
// it to open the single task SSE at since=snapshot_cursor so history (id<=cursor) and
|
||
// the live tail (id>cursor) meet with no gap and no overlap.
|
||
func (s *Server) activityHistory(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
|
||
}
|
||
q := r.URL.Query()
|
||
sess := q.Get("session")
|
||
filter, ok := parseActivitySession(sess)
|
||
if !ok {
|
||
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)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if store == nil {
|
||
writeErr(w, 404, "session not found")
|
||
return
|
||
}
|
||
// The browser tails only this task's broadcaster. Keep the SSE join cursor
|
||
// local even when the opened worker transcript comes from a related task.
|
||
snapshot, err := t.Store.ActivityMaxID()
|
||
if err != nil {
|
||
log.Printf("[activity/history] task=%s session=%s snapshot: %v", t.ID, sess, err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
var items []db.Activity
|
||
var hasMore bool
|
||
if sourceTaskID > 0 {
|
||
// Source sessions are readable only while the intent remains terminal. The
|
||
// DB query also removes model reasoning/accounting rows from inherited data.
|
||
items, hasMore, err = store.ActivityPageForTerminalIntent(*filter.NodeID, before, limit)
|
||
} else {
|
||
items, hasMore, err = store.ActivityPage(filter, before, limit)
|
||
}
|
||
if err != nil {
|
||
log.Printf("[activity/history] task=%s session=%s before=%d limit=%d: %v", t.ID, sess, before, limit, err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if sourceTaskID > 0 {
|
||
for i := range items {
|
||
items[i].SourceTaskID = sourceTaskID
|
||
items[i].Inherited = true
|
||
}
|
||
}
|
||
earliest := before
|
||
if len(items) > 0 {
|
||
earliest = items[0].ID
|
||
}
|
||
writeJSON(w, 200, map[string]any{
|
||
"items": activityDTOs(items),
|
||
"snapshot_cursor": snapshot,
|
||
"earliest_cursor": earliest,
|
||
"has_more": hasMore,
|
||
})
|
||
}
|
||
|
||
// tokenStats returns per-worker token usage (input/output/cache read/write) for a
|
||
// task — main agent, planner, and each work#N.
|
||
func (s *Server) tokenStats(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, map[string]any{
|
||
"workers": []db.TokenUsage{},
|
||
"sessions": []db.SessionTokenUsage{},
|
||
"total": tokenTotalDTO(db.TokenUsage{}),
|
||
})
|
||
return
|
||
}
|
||
stats, err := t.Store.TokenStatsByWorker()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
sessions, err := t.Store.TokenStatsBySession()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
total, err := t.Store.TokenTotal() // whole-task total (all agents)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"workers": stats, "sessions": sessions, "total": tokenTotalDTO(total)})
|
||
}
|
||
|
||
// tokenDailyStats returns global token consumption aggregated by calendar day
|
||
// (UTC) across all tasks for the past ?days=N days (default 30).
|
||
func (s *Server) tokenDailyStats(w http.ResponseWriter, r *http.Request) {
|
||
days := atoiDefault(r.URL.Query().Get("days"), 30)
|
||
if s.m.pg == nil {
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
buckets, err := s.m.pg.TokenDailyAll(days)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
if buckets == nil {
|
||
buckets = []db.DailyTokenBucket{}
|
||
}
|
||
writeJSON(w, 200, buckets)
|
||
}
|
||
|
||
// conversationTokens returns per-conversation token summaries so the dashboard can
|
||
// merge chat (conversation) usage into its per-profile / daily token stats — which
|
||
// otherwise count only task (exploration) usage.
|
||
func (s *Server) conversationTokens(w http.ResponseWriter, r *http.Request) {
|
||
if s.m.pg == nil {
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
rows, err := s.m.pg.ConversationTokenSummaries()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, rows)
|
||
}
|
||
|
||
// streamActivity is the live SSE tail: it replays history after ?since=<seq>, then
|
||
// pushes each newly appended activity for the task. The seq cursor makes history +
|
||
// live join gap-free; on reconnect the client passes its last seq to catch any
|
||
// dropped events. Optional ?intent=<id> scopes the stream to one worker session.
|
||
// getLogs returns recent backend log lines (Seq > since), newest-last.
|
||
func (s *Server) getLogs(w http.ResponseWriter, r *http.Request) {
|
||
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
|
||
limit := atoiDefault(r.URL.Query().Get("limit"), 500)
|
||
lines, cursor := logSink.recent(since, limit)
|
||
writeJSON(w, 200, map[string]any{"items": lines, "cursor": cursor})
|
||
}
|
||
|
||
// getLogsHistory returns older log lines from the DB (before a given db_id).
|
||
// GET /api/logs/history?before=<db_id>&limit=200
|
||
// Returns {items:[LogLine], has_more: bool}.
|
||
func (s *Server) getLogsHistory(w http.ResponseWriter, r *http.Request) {
|
||
if s.m.pg == nil {
|
||
writeJSON(w, 200, map[string]any{"items": []any{}, "has_more": false})
|
||
return
|
||
}
|
||
before := int64(atoiDefault(r.URL.Query().Get("before"), 0))
|
||
limit := atoiDefault(r.URL.Query().Get("limit"), 200)
|
||
if limit > 500 {
|
||
limit = 500
|
||
}
|
||
// If no before given, return the most recent DB rows (mirrors ring restore).
|
||
var (
|
||
rows []*db.DBLog
|
||
err error
|
||
)
|
||
if before <= 0 {
|
||
rows, err = s.m.pg.RecentLogs(limit)
|
||
} else {
|
||
rows, err = s.m.pg.ListLogsBefore(before, limit)
|
||
}
|
||
if err != nil {
|
||
writeErr(w, 500, "db: "+err.Error())
|
||
return
|
||
}
|
||
items := make([]LogLine, 0, len(rows))
|
||
for _, r := range rows {
|
||
items = append(items, LogLine{
|
||
DBID: r.ID,
|
||
TS: r.CreatedAt.Format(time.RFC3339),
|
||
Level: r.Level,
|
||
Tag: r.Tag,
|
||
Text: r.Text,
|
||
})
|
||
}
|
||
writeJSON(w, 200, map[string]any{"items": items, "has_more": len(rows) == limit})
|
||
}
|
||
|
||
// streamLogs is the live SSE tail of the backend log: replays history after
|
||
// ?since=<seq>, then pushes each new line.
|
||
func (s *Server) streamLogs(w http.ResponseWriter, r *http.Request) {
|
||
flusher, ok := w.(http.Flusher)
|
||
if !ok {
|
||
writeErr(w, 500, "streaming unsupported")
|
||
return
|
||
}
|
||
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
|
||
w.Header().Set("Content-Type", "text/event-stream")
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
w.Header().Set("Connection", "keep-alive")
|
||
w.Header().Set("X-Accel-Buffering", "no")
|
||
|
||
ch, unsub := logSink.subscribe()
|
||
defer unsub()
|
||
send := func(l LogLine) {
|
||
b, _ := json.Marshal(l)
|
||
fmt.Fprintf(w, "data: %s\n\n", b)
|
||
flusher.Flush()
|
||
}
|
||
lines, cursor := logSink.recent(since, 1000)
|
||
for _, l := range lines {
|
||
send(l)
|
||
}
|
||
if cursor > since {
|
||
since = cursor
|
||
}
|
||
flusher.Flush()
|
||
|
||
ctx := r.Context()
|
||
ping := time.NewTicker(20 * time.Second)
|
||
defer ping.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case l, ok := <-ch:
|
||
if !ok {
|
||
return
|
||
}
|
||
if l.Seq <= since {
|
||
continue
|
||
}
|
||
since = l.Seq
|
||
send(l)
|
||
case <-ping.C:
|
||
fmt.Fprint(w, ": ping\n\n")
|
||
flusher.Flush()
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Server) streamActivity(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
|
||
}
|
||
flusher, ok := w.(http.Flusher)
|
||
if !ok {
|
||
writeErr(w, 500, "streaming unsupported")
|
||
return
|
||
}
|
||
var intentPtr *int64
|
||
if iv := r.URL.Query().Get("intent"); iv != "" {
|
||
if n, err := strconv.ParseInt(iv, 10, 64); err == nil {
|
||
intentPtr = &n
|
||
}
|
||
}
|
||
// Cursor precedence: the browser's automatic reconnect sends Last-Event-ID (the
|
||
// last id it received) — trust it over the query so an auto-reconnect resumes
|
||
// exactly where it dropped. A fresh/manual connect has no header and passes
|
||
// since=<snapshot_cursor> from the history page instead.
|
||
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
|
||
if le := r.Header.Get("Last-Event-ID"); le != "" {
|
||
if n, err := strconv.ParseInt(le, 10, 64); err == nil {
|
||
since = n
|
||
}
|
||
}
|
||
|
||
w.Header().Set("Content-Type", "text/event-stream")
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
w.Header().Set("Connection", "keep-alive")
|
||
w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering
|
||
|
||
// Subscribe BEFORE replaying history so events in between aren't lost; dedup the
|
||
// overlap by skipping channel events whose id was already replayed.
|
||
ch, unsub := s.engine.Broadcaster().Subscribe(t.ID)
|
||
defer unsub()
|
||
|
||
// Emit a standard SSE id: line so the browser echoes it as Last-Event-ID on
|
||
// auto-reconnect (see cursor precedence above).
|
||
sendSSE := func(a db.Activity) {
|
||
b, _ := json.Marshal(activityDTO(a))
|
||
fmt.Fprintf(w, "id: %d\ndata: %s\n\n", a.ID, b)
|
||
flusher.Flush()
|
||
}
|
||
|
||
// Compensate the DB backlog after `since` in batches until caught up. This is the
|
||
// gap between the history snapshot and the live tail — NOT the first-page history
|
||
// (that's the /activity/history endpoint). A long task can have far more than one
|
||
// batch, so loop instead of a single fixed read; on a query error log it and close
|
||
// so the client reconnects and retries from its last id (broadcast is lossy — the
|
||
// DB is the source of truth). The Broadcaster keeps buffering live events meanwhile;
|
||
// the id<=since skip below drops any that this replay already covered.
|
||
const replayBatch = 500
|
||
for {
|
||
items, cursor, err := t.Store.ActivityList(intentPtr, since, replayBatch)
|
||
if err != nil {
|
||
log.Printf("[activity/stream] task=%s replay since=%d: %v", t.ID, since, err)
|
||
return
|
||
}
|
||
for _, a := range items {
|
||
sendSSE(a)
|
||
}
|
||
if cursor > since {
|
||
since = cursor
|
||
}
|
||
if len(items) < replayBatch {
|
||
break
|
||
}
|
||
}
|
||
flusher.Flush()
|
||
|
||
ctx := r.Context()
|
||
ping := time.NewTicker(20 * time.Second)
|
||
defer ping.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case a, ok := <-ch:
|
||
if !ok {
|
||
return
|
||
}
|
||
if a.ID <= since {
|
||
continue // already replayed
|
||
}
|
||
if intentPtr != nil && (a.NodeID == nil || *a.NodeID != *intentPtr) {
|
||
continue // scoped session: only this intent's steps
|
||
}
|
||
since = a.ID
|
||
sendSSE(a)
|
||
case <-ping.C:
|
||
fmt.Fprint(w, ": ping\n\n")
|
||
flusher.Flush()
|
||
}
|
||
}
|
||
}
|
||
|
||
// activityDetail lazily returns the full detail blob for one step.
|
||
func (s *Server) activityDetail(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeErr(w, 404, "no task")
|
||
return
|
||
}
|
||
seq, _ := strconv.ParseInt(r.PathValue("seq"), 10, 64)
|
||
d, err := taskActivityDetail(t, seq)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"detail": d})
|
||
}
|
||
|
||
// taskActivityDetail keeps the legacy local-task behavior (including thinking
|
||
// rows), while inherited details are restricted to terminal worker intents.
|
||
// This prevents a guessed global activity id from exposing source planner/main
|
||
// or in-flight transcripts.
|
||
func taskActivityDetail(t *Task, seq int64) (string, error) {
|
||
return t.Store.ActivityDetailWithSources(seq)
|
||
}
|
||
|
||
func (s *Server) getTraffic(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeJSON(w, 200, map[string]any{"enabled": false, "exchanges": []any{}})
|
||
return
|
||
}
|
||
q := r.URL.Query()
|
||
page := atoiDefault(q.Get("page"), 0)
|
||
size := atoiDefault(q.Get("size"), 100)
|
||
ex, matched, _ := tr.Page(traffic.PageQuery{
|
||
Host: q.Get("host"),
|
||
Method: q.Get("method"),
|
||
Query: q.Get("q"),
|
||
Body: q.Get("body"),
|
||
Path: q.Get("path"),
|
||
Status: q.Get("status"),
|
||
RespMin: int64(atoiDefault(q.Get("resp_min"), -1)),
|
||
RespMax: int64(atoiDefault(q.Get("resp_max"), -1)),
|
||
Sort: q.Get("sort"),
|
||
Order: q.Get("order"),
|
||
}, page, size)
|
||
count, _ := tr.Count() // global total, for the stat card
|
||
writeJSON(w, 200, map[string]any{
|
||
"enabled": s.m.TrafficEnabled(), // reflect the capture toggle
|
||
"proxy": s.m.ProxyAddr(),
|
||
"count": count, // total recorded (unfiltered)
|
||
"total": matched, // rows matching the current filter (for pagination)
|
||
"page": page,
|
||
"size": size,
|
||
"exchanges": trafficDTOs(ex),
|
||
})
|
||
}
|
||
|
||
// getTrafficHosts returns distinct recorded hosts with counts, for the page's
|
||
// target picker (pick a host → filter the list, then delete it).
|
||
func (s *Server) getTrafficHosts(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeJSON(w, 200, map[string]any{"hosts": []any{}})
|
||
return
|
||
}
|
||
hosts, err := tr.Hosts()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"hosts": hosts})
|
||
}
|
||
|
||
// deleteTraffic removes recorded traffic for every host containing the query's
|
||
// host substring (the page's host filter is substring-based, so what you
|
||
// filtered is what gets deleted): index rows + each host's file tree, then
|
||
// garbage-collects blobs no remaining exchange references. Empty host → 400.
|
||
// Returns the number of exchanges deleted.
|
||
func (s *Server) deleteTraffic(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeErr(w, 404, "traffic disabled")
|
||
return
|
||
}
|
||
host := strings.TrimSpace(r.URL.Query().Get("host"))
|
||
if host == "" {
|
||
writeErr(w, 400, "missing host")
|
||
return
|
||
}
|
||
n, err := tr.DeleteHost(host)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"deleted": n})
|
||
}
|
||
|
||
// deleteTrafficHosts removes traffic for a set of EXACT hosts (JSON body
|
||
// {"hosts": [...]}) — the batch path for the page's multi-select delete. Exact
|
||
// match, so picking "api.example.com" never sweeps "api.example.com.cn".
|
||
// Returns the number of exchanges deleted.
|
||
func (s *Server) deleteTrafficHosts(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeErr(w, 404, "traffic disabled")
|
||
return
|
||
}
|
||
var req struct {
|
||
Hosts []string `json:"hosts"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, "invalid body")
|
||
return
|
||
}
|
||
hosts := make([]string, 0, len(req.Hosts))
|
||
for _, h := range req.Hosts {
|
||
if h = strings.TrimSpace(h); h != "" {
|
||
hosts = append(hosts, h)
|
||
}
|
||
}
|
||
if len(hosts) == 0 {
|
||
writeErr(w, 400, "missing hosts")
|
||
return
|
||
}
|
||
n, err := tr.DeleteHostsExact(hosts)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"deleted": n})
|
||
}
|
||
|
||
// deleteAllTraffic purges every recorded exchange, then compacts the index so
|
||
// the space is actually returned to the filesystem — an emptied index is the one
|
||
// moment a full rewrite is cheap. Evidence already bound to a finding lives in
|
||
// the evidence store and is deliberately left alone. Returns the number of
|
||
// exchanges deleted and the bytes of index reclaimed.
|
||
func (s *Server) deleteAllTraffic(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeErr(w, 404, "traffic disabled")
|
||
return
|
||
}
|
||
n, reclaimed, err := tr.DeleteAll()
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"deleted": n, "reclaimed": reclaimed})
|
||
}
|
||
|
||
// getTrafficExchange returns the full raw request/response of one exchange,
|
||
// read on demand from the traffic tree (bodies are not in the paged list).
|
||
func (s *Server) getTrafficExchange(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeErr(w, 404, "traffic disabled")
|
||
return
|
||
}
|
||
id := r.URL.Query().Get("id")
|
||
if id == "" {
|
||
writeErr(w, 400, "missing id")
|
||
return
|
||
}
|
||
req, resp, err := tr.Get(id)
|
||
if err != nil {
|
||
writeErr(w, 404, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"req": req, "resp": resp})
|
||
}
|
||
|
||
// getTrafficBlob streams one oversized body by its sha256. Bodies past the inline
|
||
// threshold are not carried by the exchange endpoint — it returns a preview plus
|
||
// an "@blob sha256:<hash>" pointer — so this is how the UI fetches them whole.
|
||
// Streamed rather than buffered: these are the bodies too large to hold in memory.
|
||
func (s *Server) getTrafficBlob(w http.ResponseWriter, r *http.Request) {
|
||
tr := s.m.Traffic()
|
||
if tr == nil {
|
||
writeErr(w, 404, "traffic disabled")
|
||
return
|
||
}
|
||
hash := strings.TrimSpace(r.URL.Query().Get("hash"))
|
||
if hash == "" {
|
||
writeErr(w, 400, "missing hash")
|
||
return
|
||
}
|
||
f, size, err := tr.Blob(hash)
|
||
if err != nil {
|
||
writeErr(w, 404, err.Error())
|
||
return
|
||
}
|
||
defer f.Close()
|
||
w.Header().Set("Content-Type", "application/octet-stream")
|
||
w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
|
||
w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", hash+".bin"))
|
||
if _, err := io.Copy(w, f); err != nil {
|
||
log.Printf("[traffic] 下载 blob %s 中断：%v", hash, err)
|
||
}
|
||
}
|
||
|
||
// getSettings returns the runtime app settings the UI toggles. The Brave API key
|
||
// is returned as a boolean presence flag (brave_key_set), never the value itself,
|
||
// so the UI can show "configured" without echoing the secret back.
|
||
func (s *Server) getSettings(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, 200, s.settingsPayload())
|
||
}
|
||
|
||
func (s *Server) settingsPayload() map[string]any {
|
||
on, backend, braveKey, tavilyKey, proxy := s.m.WebSearch()
|
||
pyStored, _, _ := s.m.pg.GetSetting(settingPythonInterp)
|
||
concOn, concLimit := s.m.ConcurrencyLimit()
|
||
if concLimit == 0 {
|
||
concLimit = defaultConcurrencyLimit // 关闭时也回显一个合理默认值给 UI
|
||
}
|
||
return map[string]any{
|
||
"traffic_capture": s.m.TrafficEnabled(),
|
||
"agent_traffic_binding": s.m.pg.GetBool(settingAgentTrafficBinding, false),
|
||
"llm_record": s.m.LLMRecordEnabled(),
|
||
"web_search_enabled": on,
|
||
"web_search_backend": backend,
|
||
"brave_key_set": strings.TrimSpace(braveKey) != "",
|
||
"tavily_key_set": strings.TrimSpace(tavilyKey) != "",
|
||
"web_search_proxy": proxy, // 独立出口代理(http/https/socks5)，空=直连
|
||
"global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连
|
||
"python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测)
|
||
"workers": s.m.Workers(), // 并发工作 agent 数(默认3)；对之后启动的任务生效
|
||
"task_concurrency_enabled": concOn, // 任务并发上限开关(默认关)
|
||
"task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5)
|
||
// LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时
|
||
// 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。
|
||
"llm_pool_enabled": s.m.LLMPoolEnabled(),
|
||
"llm_pool_bind_fallback": s.m.LLMPoolBindFallback(),
|
||
// 操作约束注入范围(默认都开):把本任务的 allow/deny 约束拼进对应 agent 的系统提示。
|
||
"constraints_inject_planner": s.constraintInjectPlanner(),
|
||
"constraints_inject_worker": s.constraintInjectWorker(),
|
||
// 实验功能:noa 模型驱动上下文压缩(默认关)。开启后平台接入的四类 agent 由 noa
|
||
// 接管上下文压缩,取代内置 compaction;每 run 读一次,对之后启动的 run 生效。
|
||
"noa_compaction": s.m.NoaCompactionEnabled(),
|
||
// 漏洞 IM 推送的全局项。渠道本身是独立资源，走 /api/notify/* 管理；
|
||
// 这里只放「作用于全部渠道」的三项。
|
||
"notify_enabled": s.m.pg.GetBool(settingNotifyEnabled, true),
|
||
"notify_public_base_url": notifyPublicBaseURL(s.m.pg),
|
||
"notify_digest_interval_min": notifyDigestIntervalMin(s.m.pg),
|
||
}
|
||
}
|
||
|
||
// notifyPublicBaseURL 读推送回链用的外部地址。
|
||
func notifyPublicBaseURL(pg *db.DB) string {
|
||
v, _, _ := pg.GetSetting(settingNotifyPublicBaseURL)
|
||
return v
|
||
}
|
||
|
||
// notifyDigestIntervalMin 读汇总周期（分钟），非法或未配置时回落到默认值。
|
||
// 回显默认值而不是空串，UI 才能把当前生效值填进输入框。
|
||
func notifyDigestIntervalMin(pg *db.DB) int {
|
||
v, ok, _ := pg.GetSetting(settingNotifyDigestMinutes)
|
||
if !ok {
|
||
return notifyDefaultDigestMinutes
|
||
}
|
||
n, err := strconv.Atoi(strings.TrimSpace(v))
|
||
if err != nil || n <= 0 {
|
||
return notifyDefaultDigestMinutes
|
||
}
|
||
return n
|
||
}
|
||
|
||
// pgDetectPython re-runs interpreter detection, stores + returns it.
|
||
func (s *Server) pgDetectPython(w http.ResponseWriter, r *http.Request) {
|
||
p := detectPython()
|
||
if p == "" {
|
||
writeErr(w, 404, errPythonNotDetected)
|
||
return
|
||
}
|
||
if err := s.m.pg.SetSetting(settingPythonInterp, p); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"python_interpreter": p})
|
||
}
|
||
|
||
// putSettings applies a settings change. Toggling traffic_capture rebuilds the
|
||
// agents (applyLLM) so the new proxy/traffic-tools/prompt state takes hold — when
|
||
// off, agents get no proxy config, no traffic tools, and no proxy prompt content.
|
||
func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
TrafficCapture *bool `json:"traffic_capture"`
|
||
AgentTrafficBinding *bool `json:"agent_traffic_binding"`
|
||
LLMRecord *bool `json:"llm_record"` // LLM 录制开关（默认关）；即时生效，无需重建 agent
|
||
// Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey
|
||
// are optional — omit (null) to leave a stored key untouched, send "" to clear.
|
||
WebSearchEnabled *bool `json:"web_search_enabled"`
|
||
WebSearchBackend *string `json:"web_search_backend"`
|
||
BraveKey *string `json:"brave_search_api_key"`
|
||
TavilyKey *string `json:"tavily_search_api_key"`
|
||
WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空
|
||
GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连)
|
||
PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径
|
||
Workers *int `json:"workers"` // 并发工作 agent 数(>0)；对之后启动的任务生效
|
||
// 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。
|
||
ConcurrencyEnabled *bool `json:"task_concurrency_enabled"`
|
||
ConcurrencyLimit *int `json:"task_concurrency_limit"`
|
||
// LLM 轮询(故障转移)开关 + 「绑定配置失败也兜底回轮询链」开关。两者都需要
|
||
// 重建 provider 链才生效，走下面的 changed → applyLLM 路径。
|
||
LLMPoolEnabled *bool `json:"llm_pool_enabled"`
|
||
LLMPoolBindFallback *bool `json:"llm_pool_bind_fallback"`
|
||
// 操作约束注入范围开关(默认都开);即时生效(planner/worker 每轮读),无需重建 agent。
|
||
ConstraintsInjectPlanner *bool `json:"constraints_inject_planner"`
|
||
ConstraintsInjectWorker *bool `json:"constraints_inject_worker"`
|
||
// 实验功能:noa 上下文压缩开关(默认关);每 run 读,对之后启动的 run 生效,无需重建 agent。
|
||
NoaCompaction *bool `json:"noa_compaction"`
|
||
// 漏洞 IM 推送的全局项。三者都由投递引擎每轮读一次，改完即时生效，
|
||
// 不需要重建 agent 或重启。
|
||
NotifyEnabled *bool `json:"notify_enabled"`
|
||
NotifyBaseURL *string `json:"notify_public_base_url"`
|
||
NotifyDigestMins *int `json:"notify_digest_interval_min"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
if req.ConstraintsInjectPlanner != nil {
|
||
if err := s.m.pg.SetBool(settingConstraintsInjectPlanner, *req.ConstraintsInjectPlanner); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.ConstraintsInjectWorker != nil {
|
||
if err := s.m.pg.SetBool(settingConstraintsInjectWorker, *req.ConstraintsInjectWorker); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.NoaCompaction != nil {
|
||
// 每 run 读的解析器,切换即时对之后启动的 run 生效,无需 applyLLM 重建。
|
||
if err := s.m.SetNoaCompaction(*req.NoaCompaction); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
// 推送全局项:投递引擎每轮重新读取,所以即时生效、无需重启。
|
||
if req.NotifyEnabled != nil {
|
||
if err := s.m.pg.SetBool(settingNotifyEnabled, *req.NotifyEnabled); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.NotifyBaseURL != nil {
|
||
// 统一裁掉尾部斜杠:回链拼接用的是 fmt.Sprintf("%s/function/..."),
|
||
// 留着尾部斜杠会产出 "//function/..." 这种双斜杠路径。
|
||
base := trimTrailingSlash(strings.TrimSpace(*req.NotifyBaseURL))
|
||
if base != "" && !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
|
||
writeErr(w, 400, errNotifyBaseURLScheme)
|
||
return
|
||
}
|
||
if err := s.m.pg.SetSetting(settingNotifyPublicBaseURL, base); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.NotifyDigestMins != nil {
|
||
// 下限 1 分钟:更短的周期等于实时推送,那样应该直接把渠道改成 realtime 模式。
|
||
if *req.NotifyDigestMins < 1 || *req.NotifyDigestMins > 24*60 {
|
||
writeErr(w, 400, errNotifyDigestRange)
|
||
return
|
||
}
|
||
if err := s.m.pg.SetSetting(settingNotifyDigestMinutes, strconv.Itoa(*req.NotifyDigestMins)); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.Workers != nil {
|
||
if err := s.m.SetWorkers(*req.Workers); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil {
|
||
// 部分 PUT:未给的字段用当前值兜底,避免只改一个把另一个重置。
|
||
curOn, curLimit := s.m.ConcurrencyLimit()
|
||
if curLimit == 0 {
|
||
curLimit = defaultConcurrencyLimit
|
||
}
|
||
on, limit := curOn, curLimit
|
||
if req.ConcurrencyEnabled != nil {
|
||
on = *req.ConcurrencyEnabled
|
||
}
|
||
if req.ConcurrencyLimit != nil {
|
||
limit = *req.ConcurrencyLimit
|
||
}
|
||
if err := s.m.SetConcurrency(on, limit); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
// 立即协调一次:关闭时放行全部排队,调高上限时补位启动,不必等下一个 tick。
|
||
go s.reconcileConcurrency()
|
||
}
|
||
if req.PythonInterp != nil {
|
||
if err := s.m.pg.SetSetting(settingPythonInterp, strings.TrimSpace(*req.PythonInterp)); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.LLMRecord != nil {
|
||
// 录制器每次调用读取该标志，切换即时生效，无需 applyLLM 重建。
|
||
if err := s.m.SetLLMRecordEnabled(*req.LLMRecord); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
changed := false
|
||
if req.LLMPoolEnabled != nil {
|
||
if err := s.m.SetLLMPoolEnabled(*req.LLMPoolEnabled); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
changed = true // the provider chain itself changes shape → rebuild
|
||
}
|
||
if req.LLMPoolBindFallback != nil {
|
||
if err := s.m.SetLLMPoolBindFallback(*req.LLMPoolBindFallback); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
changed = true
|
||
}
|
||
if req.LLMPoolEnabled != nil || req.LLMPoolBindFallback != nil {
|
||
// Pinned tasks' planner/worker and per-profile chat agents hold providers
|
||
// built under the OLD switch state — drop them so they pick up the new one.
|
||
s.invalidateProfileAgents()
|
||
}
|
||
if req.AgentTrafficBinding != nil {
|
||
if err := s.m.pg.SetBool(settingAgentTrafficBinding, *req.AgentTrafficBinding); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
if req.TrafficCapture != nil {
|
||
if err := s.m.SetTrafficEnabled(*req.TrafficCapture); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
changed = true
|
||
}
|
||
if req.GlobalProxy != nil {
|
||
// Validation failure (bad scheme/host) is a client error, not a 500.
|
||
if err := s.m.SetGlobalProxy(*req.GlobalProxy); err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
changed = true // capture-off egress is baked into agents at build time → rebuild
|
||
}
|
||
if req.WebSearchEnabled != nil || req.WebSearchBackend != nil || req.BraveKey != nil || req.TavilyKey != nil || req.WebSearchProxy != nil {
|
||
// Fill unspecified fields from current state so a partial PUT doesn't reset them.
|
||
on, backend, _, _, _ := s.m.WebSearch()
|
||
if req.WebSearchEnabled != nil {
|
||
on = *req.WebSearchEnabled
|
||
}
|
||
if req.WebSearchBackend != nil {
|
||
backend = *req.WebSearchBackend
|
||
}
|
||
if err := s.m.SetWebSearch(on, backend, req.BraveKey, req.TavilyKey, req.WebSearchProxy); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
changed = true
|
||
}
|
||
if changed {
|
||
// rebuild agents so the new proxy/tools/prompt/web-search take hold (only if LLM configured).
|
||
s.cfgMu.Lock()
|
||
cfg, on := s.llmCfg, s.llmOn
|
||
s.cfgMu.Unlock()
|
||
if on {
|
||
if err := s.applyLLM(cfg); err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
}
|
||
}
|
||
writeJSON(w, 200, s.settingsPayload())
|
||
}
|
||
|
||
// testWebSearch runs a real "test" search with the given (or currently saved)
|
||
// backend/proxy/key to verify the config can actually reach a search backend —
|
||
// mirroring testLLM. Backend/proxy come from the request (so the form's unsaved
|
||
// edits are tested); empty API keys fall back to stored values so the user need
|
||
// not retype them. Always 200 with {ok, error?, count?, backend?}.
|
||
func (s *Server) testWebSearch(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
Backend string `json:"web_search_backend"`
|
||
Proxy string `json:"web_search_proxy"`
|
||
BraveKey string `json:"brave_search_api_key"`
|
||
TavilyKey string `json:"tavily_search_api_key"`
|
||
}
|
||
// Empty body is fine — fall back entirely to the saved config below.
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
_, backend, storedBraveKey, storedTavilyKey, _ := s.m.WebSearch()
|
||
if strings.TrimSpace(req.Backend) != "" {
|
||
backend = req.Backend
|
||
}
|
||
// Proxy is taken from the form as-is (empty = direct), so testing reflects exactly
|
||
// what's shown — including an intentional "clear proxy to test direct" before saving.
|
||
proxy := strings.TrimSpace(req.Proxy)
|
||
// API keys are secrets the form omits when already saved, so fall back to stored.
|
||
braveKey := storedBraveKey
|
||
if strings.TrimSpace(req.BraveKey) != "" {
|
||
braveKey = req.BraveKey
|
||
}
|
||
tavilyKey := storedTavilyKey
|
||
if strings.TrimSpace(req.TavilyKey) != "" {
|
||
tavilyKey = req.TavilyKey
|
||
}
|
||
cfg := actool.WebSearchConfig{Backend: backend, BraveAPIKey: braveKey, TavilyAPIKey: tavilyKey, Proxy: proxy}
|
||
// Hard cap so a slow/blocked proxy can't hang the request.
|
||
wall := 30 * time.Second
|
||
// deepseek 的凭据不在表单里，来自当前激活的 LLM 配置；同时它每次搜索都跑一次
|
||
// 模型推理，30s 的通用上限偏紧，单独放宽。这里不预判配置能不能用——测这一下
|
||
// 本来就是给用户自己确认的手段，真跑不通时下面的报错比预判更有信息量。
|
||
probeQuery := "test"
|
||
if strings.TrimSpace(backend) == deepSeekWebSearchBackend {
|
||
cfg.DeepSeekBaseURL, cfg.DeepSeekAPIKey, cfg.DeepSeekModel = s.m.deepSeekSearchCreds()
|
||
wall = 120 * time.Second
|
||
// 搜索词由 DeepSeek 端的模型自行决定，"test" 太空泛会让它跳过搜索直接作答。
|
||
probeQuery = "DeepSeek company official website"
|
||
}
|
||
ctx, cancel := context.WithTimeout(r.Context(), wall)
|
||
defer cancel()
|
||
results, err := actool.WebSearchProbe(ctx, cfg, probeQuery, 3)
|
||
if err != nil {
|
||
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error(), "backend": backend})
|
||
return
|
||
}
|
||
if len(results) == 0 {
|
||
writeJSON(w, 200, map[string]any{"ok": false, "error": errWebSearchProbeNoResults, "backend": backend})
|
||
return
|
||
}
|
||
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, errTaskDeletingNewSession)
|
||
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 {
|
||
writeErr(w, 404, "no active task")
|
||
return
|
||
}
|
||
if s.engine.IsDeleting(t.ID) {
|
||
writeErr(w, 409, errTaskDeletingNewMessage)
|
||
return
|
||
}
|
||
// 注意:任务暂停(paused)不拦截主 Agent 对话。主 Agent 编排会话独立于 planner/
|
||
// worker 的暂停,暂停中仍可继续对话(暂停只终止其正在进行的那一轮,见 control())。
|
||
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())
|
||
return
|
||
}
|
||
agentMessage, ok := s.prepareChatMentionMessage(w, req.Message)
|
||
if !ok {
|
||
return
|
||
}
|
||
// Admission is serialized with deletion's chat cancellation. Mark every chat
|
||
// turn busy before its first activity/file write, including rule-mode turns.
|
||
// If deletion wins the race, the second barrier check rejects this request.
|
||
s.chatMu.Lock()
|
||
if s.engine.IsDeleting(t.ID) {
|
||
s.chatMu.Unlock()
|
||
writeErr(w, 409, errTaskDeletingNewMessage)
|
||
return
|
||
}
|
||
if s.chatBusy[t.ID] {
|
||
s.chatMu.Unlock()
|
||
writeErr(w, 409, errMainAgentBusy)
|
||
return
|
||
}
|
||
ctx, cancel := context.WithCancelCause(s.ctx)
|
||
ctx = intercept.WithReviewContext(ctx, "", intercept.ReviewBackground{Source: intercept.BackgroundUserMessage, Text: req.Message})
|
||
s.chatBusy[t.ID] = true
|
||
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.
|
||
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
|
||
}
|
||
if ma != nil {
|
||
// Run the agent on a per-turn cancellable ctx (NOT r.Context()): the turn can
|
||
// take minutes (multi-tool loop), and binding it to the request lifecycle meant
|
||
// a page reload / proxy timeout cancelled it mid-run ("context canceled"). The
|
||
// steps + final answer stream back live via SSE (worker="mainagent"), so the
|
||
// handler returns immediately and the browser never needs to hold the request.
|
||
go func() {
|
||
defer func() {
|
||
s.finishTaskChat(t.ID, cancel)
|
||
}()
|
||
// 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) {
|
||
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 打开文件。
|
||
// taskDir = agent 的工作目录(CWD),与 chatUpload 落盘、ensureRunDir 一致。
|
||
taskDir := filepath.Join(s.m.dir, "tasks", t.ID)
|
||
agentMsg := composeAgentMessage(agentMessage, req.Attachments, taskDir)
|
||
s.engine.BeginLLMCall(t.ID)
|
||
_, err := ma.Chat(ctx, maTaskID, mainSeg, s.m.Assets(), t.Store, t.Goal, agentMsg, emit, t.Notify, resume, t.NotifyGoal, t.NotifyHint)
|
||
s.engine.EndLLMCall(t.ID)
|
||
if err != nil && ctx.Err() == nil {
|
||
s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", IsError: true, Summary: transcriptErrorSummary("메인 에이전트", 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, MainSeg: segPtr})
|
||
s.finishTaskChat(t.ID, cancel)
|
||
writeJSON(w, 200, map[string]any{"reply": reply, "mode": "rule"})
|
||
}
|
||
|
||
func (s *Server) cancelTaskChat(taskID string, cause error) bool {
|
||
s.chatMu.Lock()
|
||
cancel := s.chatCancel[taskID]
|
||
if cancel != nil {
|
||
cancel(cause)
|
||
}
|
||
s.chatMu.Unlock()
|
||
return cancel != nil
|
||
}
|
||
|
||
func (s *Server) finishTaskChat(taskID string, cancel context.CancelCauseFunc) {
|
||
cancel(agent.AbortChatTurnFinished)
|
||
s.chatMu.Lock()
|
||
delete(s.chatBusy, taskID)
|
||
delete(s.chatCancel, taskID)
|
||
s.chatMu.Unlock()
|
||
}
|
||
|
||
// taskChatStatus reports the authoritative state of the task's main-agent turn.
|
||
// Activity timestamps are not a reliable proxy because a tool or LLM call may run
|
||
// for minutes without emitting an intermediate frame.
|
||
func (s *Server) taskChatStatus(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.PathValue("id"))
|
||
if t == nil {
|
||
writeErr(w, http.StatusNotFound, "task not found")
|
||
return
|
||
}
|
||
s.chatMu.Lock()
|
||
running := s.chatBusy[t.ID]
|
||
s.chatMu.Unlock()
|
||
writeJSON(w, http.StatusOK, map[string]bool{"running": running})
|
||
}
|
||
|
||
// stopChat aborts the in-flight main-agent turn for a task (manual stop button).
|
||
// Mirrors pgStopConversation: cancels only the current turn; planner/workers are
|
||
// unaffected and continue running. The user can send a new message immediately.
|
||
func (s *Server) stopChat(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.PathValue("id"))
|
||
if t == nil {
|
||
writeErr(w, 404, "task not found")
|
||
return
|
||
}
|
||
if !s.cancelTaskChat(t.ID, agent.AbortChatStoppedByUser) {
|
||
writeJSON(w, 200, map[string]any{"status": "idle"})
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"status": "stopping"})
|
||
}
|
||
|
||
// fallbackCommand classifies a rule-mode chat message into an intent/hint command
|
||
// plus the trimmed argument. The trigger keywords accept Chinese (意图/提示), English
|
||
// (intent/hint) and Korean (의도/힌트) prefixes — pure input parsing, not a wire
|
||
// format, so the Korean aliases are additive and backward compatible. cmd is
|
||
// "intent", "hint" or "" (status summary). (F34 d)
|
||
func fallbackCommand(msg string) (cmd, text string) {
|
||
m := strings.TrimSpace(msg)
|
||
lower := strings.ToLower(m)
|
||
switch {
|
||
case strings.HasPrefix(m, "意图") || strings.HasPrefix(m, "의도") || strings.HasPrefix(lower, "intent"):
|
||
return "intent", strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(strings.TrimPrefix(m, "意图"), "의도"), "intent"))
|
||
case strings.HasPrefix(m, "提示") || strings.HasPrefix(m, "힌트") || strings.HasPrefix(lower, "hint"):
|
||
return "hint", strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(strings.TrimPrefix(m, "提示"), "힌트"), "hint"))
|
||
default:
|
||
return "", m
|
||
}
|
||
}
|
||
|
||
// fallbackChat is the no-LLM human-steering handler: simple命令 + 态势摘要.
|
||
func (s *Server) fallbackChat(t *Task, msg string) string {
|
||
switch cmd, text := fallbackCommand(msg); cmd {
|
||
case "intent":
|
||
_, _ = t.Store.AddIntent(map[string]any{"summary": text}, 9, nil, "human")
|
||
return fallbackIntentInjected + text
|
||
case "hint":
|
||
_, _ = t.Store.AddNode(db.KindHint, map[string]any{"text": text}, 0, "active", "human", nil)
|
||
return fallbackHintRecorded + text
|
||
default:
|
||
assetCounts, _ := s.m.Assets().CountsByType()
|
||
assets := 0
|
||
for _, c := range assetCounts {
|
||
assets += c
|
||
}
|
||
fnd, _ := t.Store.ListByKind(db.KindFinding, 1000)
|
||
fr, _ := t.Store.Frontier(1000)
|
||
return fmt.Sprintf(fallbackChatStatus, assets, len(fr), len(fnd))
|
||
}
|
||
}
|
||
|
||
func (s *Server) getReport(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeErr(w, 404, "no active task")
|
||
return
|
||
}
|
||
findings, _ := t.Store.ListByKind(db.KindFinding, 1000) // 纯漏洞（事实是独立的 KindFact，不进报告）
|
||
counts := map[string]int{}
|
||
for _, ty := range []string{"root_domain", "ip", "subdomain", "app", "service", "endpoint"} {
|
||
ns, _ := s.m.Assets().QueryByType(ty, 100000, 0)
|
||
if len(ns) > 0 {
|
||
counts[ty] = len(ns)
|
||
}
|
||
}
|
||
md := report.Markdown(report.Input{
|
||
Title: t.Description, Goal: t.Goal, GeneratedAt: time.Now(),
|
||
AssetCounts: counts, Findings: findings,
|
||
})
|
||
w.Header().Set("Content-Type", "text/markdown; charset=utf-8")
|
||
w.WriteHeader(200)
|
||
_, _ = w.Write([]byte(md))
|
||
}
|
||
|
||
func (s *Server) getAudit(w http.ResponseWriter, r *http.Request) {
|
||
t := s.m.ResolveTask(r.URL.Query().Get("task"))
|
||
if t == nil {
|
||
writeJSON(w, 200, []any{})
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"entries": t.Guard.Audit(), "attributions": t.Guard.Attributions()})
|
||
}
|
||
|
||
// gc is a no-op stub (GC not yet implemented in the new asset store).
|
||
func (s *Server) gc(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, 200, map[string]any{"removed": 0})
|
||
}
|
||
|
||
// --- utils ---
|
||
|
||
func cors(next http.Handler) http.Handler {
|
||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||
w.Header().Set("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
|
||
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
|
||
if r.Method == http.MethodOptions {
|
||
w.WriteHeader(204)
|
||
return
|
||
}
|
||
next.ServeHTTP(w, r)
|
||
})
|
||
}
|
||
|
||
func writeJSON(w http.ResponseWriter, code int, v any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(code)
|
||
_ = json.NewEncoder(w).Encode(v)
|
||
}
|
||
|
||
func writeErr(w http.ResponseWriter, code int, msg string) {
|
||
writeJSON(w, code, map[string]any{"error": msg})
|
||
}
|
||
|
||
func atoiDefault(s string, d int) int {
|
||
if n, err := strconv.Atoi(s); err == nil {
|
||
return n
|
||
}
|
||
return d
|
||
}
|