This commit is contained in:
pphua committed 2026-08-21 11:16:19 +08:00
1 parent 9793b522d9
commit b7e0b93023
46 files changed
+3632 -899

No files matched your search

+12
View File
@@ -209,6 +209,18 @@ func (d *DB) ListLLMRecords(model, session, task string, page, size int) ([]LLMR
return out, total, rows.Err()
}
// ModelTokenStat is one model's aggregated token usage for a task, summed from the
// llm_usage metering ledger (see db/llm_usage.go). Calls is the number of LLM calls
// that hit this model.
type ModelTokenStat struct {
Model string `json:"model"`
Calls int `json:"calls"`
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheReadTokens int `json:"cache_read_tokens"`
CacheWriteTokens int `json:"cache_write_tokens"`
}
// LLMTask is one distinct task with its LLM-record count.
type LLMTask struct {
TaskID string `json:"task_id"`
+15 -13
View File
@@ -25,9 +25,11 @@ type LLMProfile struct {
// ContextWindowK is the model's context window in K tokens, used to size
// compaction thresholds. 0 = use a 200K default; capped at 1000 (1M).
ContextWindowK int `json:"context_window_k"`
// ReasoningEffort is the thinking-mode selector. "" = 默认(不发送思考参数);
// "off" = 显式关闭(thinking.type=disabled); "low"/"medium"/"high"/"max" =
// 开启思考并设强度. Mapped to the provider request in agent.Config.NewProvider.
// ThinkingType 独立控制思考「开关」(thinking.type):"" = 不发送(默认);
// "disabled" = 显式关闭; "enabled" = 开启. 与 ReasoningEffort 解耦.
ThinkingType string `json:"thinking_type"`
// ReasoningEffort 独立控制思考「强度」:"" = 不发送(默认);
// "low"/"medium"/"high"/"xhigh"/"max" = 对应强度. 见 agent.Config.NewProvider.
ReasoningEffort string `json:"reasoning_effort"`
IsDefault bool `json:"is_default"`
// Priority orders the failover chain: higher goes first. The ACTIVE profile
@@ -41,14 +43,14 @@ type LLMProfile struct {
// profileCols is the read column list (hint variant, no api key) shared by the
// list query; profileColsKey is the same with api_key for the single-row loads.
const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude`
const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude`
const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')`
const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')`
// scanProfile reads one row in the profileCols / profileColsKey column order. The
// 7th column lands in APIKeyHint or APIKey depending on which list the caller used.
func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile) error {
return sc.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, into,
&p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude)
&p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude, &p.ThinkingType)
}
func (d *DB) ListProfiles() ([]*LLMProfile, error) {
@@ -121,18 +123,18 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) {
}
if p.ID == 0 {
var id int64
err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude)
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13) RETURNING id`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude).Scan(&id)
err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude,thinking_type)
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14) RETURNING id`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType).Scan(&id)
return id, err
}
if p.APIKey == "" {
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11 WHERE id=$12`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ID)
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12 WHERE id=$13`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID)
return p.ID, err
}
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13 WHERE id=$14`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ID)
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14 WHERE id=$15`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID)
return p.ID, err
}
+2 -2
View File
@@ -103,7 +103,7 @@ type builtinAgent struct {
key, name, role, desc string
vars []promptVar
interactiveShell bool // 建行时的默认交互式 shell 开关;ON CONFLICT 不覆盖用户后续手动开关
runSeconds *int // 建行时的单次 run 墙钟上限(秒);nil=用表默认(600),0=不限时
runSeconds *int // 建行时的单次 run 墙钟上限(秒);nil=用种子默认(1200),0=不限时
}
type promptVar struct{ name, desc, example, source string }
@@ -144,7 +144,7 @@ func (d *DB) seedBuiltins() error {
var agentID int64
err := d.QueryRow(`
INSERT INTO agents(key, name, description, role, builtin, enabled, interactive_shell, run_seconds)
VALUES ($1, $2, NULLIF($3,''), $4, true, true, $5, COALESCE($6, 600))
VALUES ($1, $2, NULLIF($3,''), $4, true, true, $5, COALESCE($6, 1200))
ON CONFLICT (key) DO UPDATE SET name = EXCLUDED.name, description = EXCLUDED.description
RETURNING id`, a.key, a.name, a.desc, a.role, a.interactiveShell, a.runSeconds).Scan(&agentID)
if err != nil {
+5 -5
View File
@@ -215,7 +215,7 @@ func (s *ExplorationStore) SetNodeState(id int64, state string) error {
// ResetRunningIntents returns any intent left in 'running' back to 'open' so it
// is re-claimed. Called on startup: a 'running' intent with no live worker (a
// backend restart or crashed worker goroutine left it stuck) would otherwise spin
// forever in the UI. Workers re-run an intent from scratch, so reopening is safe.
// forever in the UI. Workers resume from their transcript, so reopening is safe.
func (s *ExplorationStore) ResetRunningIntents() (int64, error) {
res, err := s.db.Exec(`UPDATE exploration_nodes SET state='open', completed_at=NULL, blocked_reason=NULL
WHERE exploration_id=$1 AND kind='intent' AND state='running'`, s.expID)
@@ -227,9 +227,9 @@ WHERE exploration_id=$1 AND kind='intent' AND state='running'`, s.expID)
}
// ReopenIntent flips ONE not-successfully-finished intent (blocked/exhausted/stopped)
// back to 'open' so a worker re-claims and re-runs it from scratch (graph writes it
// already produced stay). done/open/running are left untouched. Returns whether a row
// changed (false = intent absent or not in a rerunnable state). Used by the "重跑" button.
// back to 'open' so a worker re-claims it (graph writes it already produced stay).
// The worker resumes from its prior LLM transcript instead of restarting from scratch.
// done/open/running are left untouched. Returns whether a row changed. Used by "重跑".
func (s *ExplorationStore) ReopenIntent(id int64) (bool, error) {
res, err := s.db.Exec(`UPDATE exploration_nodes
SET state='open', completed_at=NULL, blocked_reason=NULL
@@ -244,7 +244,7 @@ WHERE id=$1 AND exploration_id=$2 AND kind='intent'
// ReopenBlockedIntents flips EVERY 'blocked' intent in this exploration back to 'open'
// (batch rerun after e.g. an LLM/network outage that blocked many at once). Returns the
// number reopened. Workers re-run each from scratch; kept graph writes remain.
// number reopened. Workers resume from their transcripts; kept graph writes remain.
func (s *ExplorationStore) ReopenBlockedIntents() (int64, error) {
res, err := s.db.Exec(`UPDATE exploration_nodes SET state='open', completed_at=NULL, blocked_reason=NULL
WHERE exploration_id=$1 AND kind='intent' AND state='blocked'`, s.expID)
+3 -3
View File
@@ -356,13 +356,13 @@ func TestTaskAssetContextUsesDirectSourceScopeAndAnchors(t *testing.T) {
_ = d.DeleteTask(grand.ID)
_, _ = assets.DeleteByIDs([]int64{sourceTestedID, sourceUntestedID, sourceAnchoredOnlyID, grandID})
})
if _, err := assets.AddAgentScope(source.ID, "root_domain", sourceTestedDomain, "test"); err != nil {
if _, err := assets.AddAgentScope(source.ID, "root_domain", sourceTestedDomain, "test", "agent"); err != nil {
t.Fatal(err)
}
if _, err := assets.AddAgentScope(source.ID, "root_domain", sourceUntestedDomain, "test"); err != nil {
if _, err := assets.AddAgentScope(source.ID, "root_domain", sourceUntestedDomain, "test", "agent"); err != nil {
t.Fatal(err)
}
if _, err := assets.AddAgentScope(grand.ID, "root_domain", grandDomain, "test"); err != nil {
if _, err := assets.AddAgentScope(grand.ID, "root_domain", grandDomain, "test", "agent"); err != nil {
t.Fatal(err)
}
if _, err := d.Exploration(source.ExplorationID).AddNode(KindFact, map[string]any{"summary": "tested source asset"}, 0, "confirmed", "worker", []int64{sourceTestedID}); err != nil {
+51
View File
@@ -210,6 +210,57 @@ func (d *DB) ListFindingsPage(f FindingFilter, page, pageSize int) ([]*DBFinding
return out, total, err
}
// ListFindingsForExport returns findings for the 发现 page 导出功能,携带完整
// report 字段、不分页。ids 非空时按这批 finding id 精确导出(勾选导出),忽略
// filter;ids 为空时按 filter 导出(导出当前筛选/全部)。结果按严重等级降序、
// 再按时间倒序,与「导出汇总报告」的分组顺序一致。
func (d *DB) ListFindingsForExport(f FindingFilter, ids []int64) ([]*DBFinding, error) {
const order = `ORDER BY CASE f.severity WHEN 'critical' THEN 4 WHEN 'high' THEN 3 WHEN 'medium' THEN 2 WHEN 'low' THEN 1 ELSE 0 END DESC, f.created_at DESC`
cols := findingSelectCols + `, COALESCE(f.report, '')`
var q string
var args []any
if len(ids) > 0 {
ph := make([]string, len(ids))
for i, id := range ids {
ph[i] = fmt.Sprintf("$%d", i+1)
args = append(args, id)
}
q = `SELECT ` + cols + `
FROM findings f
LEFT JOIN tasks t ON f.task_id = t.id
WHERE f.id IN (` + strings.Join(ph, ",") + `)
` + order
} else {
where, wargs := f.where()
q = `SELECT ` + cols + `
FROM findings f
LEFT JOIN tasks t ON f.task_id = t.id` + where + `
` + order
args = wargs
}
rows, err := d.Query(q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []*DBFinding
for rows.Next() {
f := &DBFinding{}
var aidsJSON string
if err := rows.Scan(&f.ID, &f.TaskID, &f.NodeID, &f.VulnClass, &f.Name, &f.Severity,
&f.Summary, &f.Evidence, &f.Worker, &aidsJSON, &f.Status, &f.CreatedAt,
&f.TaskDescription, &f.Report); err != nil {
return nil, err
}
_ = json.Unmarshal([]byte(aidsJSON), &f.AssetIDs)
out = append(out, f)
}
return out, rows.Err()
}
// FindingStats is the whole-table aggregate powering the 发现 page's stat cards
// and vuln-class filter — computed server-side so it stays exact regardless of
// pagination.
+183
View File
@@ -0,0 +1,183 @@
package db
import "strconv"
// LLMUsage is one lightweight LLM-call metering row — the always-on usage ledger,
// distinct from llm_records (which stores full request/response bodies and is a
// gated debug feature). One row per completion call, written on both success and
// error, so token accounting is complete even for interrupted/failed runs. Carries
// only the dimensions needed to slice token spend (model / profile / task / agent),
// never any prompt or response content.
type LLMUsage struct {
TaskID string `json:"task_id"` // task registry id (matches llm_records.task_id)
ExplorationID int64 `json:"exploration_id"` // exploration id parsed from the session (0 = unknown/non-task)
Worker string `json:"worker"` // agent lane: worker / planner / mainagent / goals
Model string `json:"model"`
ProfileName string `json:"profile_name"`
LatencyMs int `json:"latency_ms"`
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheRead int `json:"cache_read"`
CacheWrite int `json:"cache_write"`
Status string `json:"status"` // ok | error
}
const llmUsageSchema = `
CREATE TABLE IF NOT EXISTS llm_usage (
id BIGSERIAL PRIMARY KEY,
ts TIMESTAMPTZ NOT NULL DEFAULT now(),
task_id TEXT,
exploration_id BIGINT,
worker TEXT,
model TEXT,
profile_name TEXT,
latency_ms INTEGER,
input_tokens INTEGER NOT NULL DEFAULT 0,
output_tokens INTEGER NOT NULL DEFAULT 0,
cache_read INTEGER NOT NULL DEFAULT 0,
cache_write INTEGER NOT NULL DEFAULT 0,
status TEXT
);
CREATE INDEX IF NOT EXISTS idx_llm_usage_task ON llm_usage(task_id);
CREATE INDEX IF NOT EXISTS idx_llm_usage_model ON llm_usage(task_id, model);
CREATE INDEX IF NOT EXISTS idx_llm_usage_exp ON llm_usage(exploration_id);
`
// EnsureLLMUsageTable creates the llm_usage metering table if it does not exist.
func (d *DB) EnsureLLMUsageTable() error {
_, err := d.Exec(llmUsageSchema)
return err
}
// InsertLLMUsage appends one metering row. Best-effort: callers log and continue on
// error (a lost metering row must never break the LLM call).
func (d *DB) InsertLLMUsage(u *LLMUsage) error {
var expID any
if u.ExplorationID > 0 {
expID = u.ExplorationID
}
_, err := d.Exec(`
INSERT INTO llm_usage(task_id, exploration_id, worker, model, profile_name, latency_ms, input_tokens, output_tokens, cache_read, cache_write, status)
VALUES (NULLIF($1,''),$2,NULLIF($3,''),NULLIF($4,''),NULLIF($5,''),$6,$7,$8,$9,$10,$11)`,
u.TaskID, expID, u.Worker, u.Model, u.ProfileName,
u.LatencyMs, u.InputTokens, u.OutputTokens, u.CacheRead, u.CacheWrite, u.Status)
return err
}
// TokenByModel aggregates a task's LLM token usage grouped by model, most-used
// first, from the always-on llm_usage ledger. taskID is the task registry id.
// Accurate even with per-agent model bindings, pool rotation/failover, and
// interrupted runs, since every call (success or error) is metered.
func (d *DB) TokenByModel(taskID string) ([]ModelTokenStat, error) {
rows, err := d.Query(`
SELECT COALESCE(NULLIF(model,''),'(unknown)') AS model, COUNT(*) AS calls,
COALESCE(SUM(input_tokens),0), COALESCE(SUM(output_tokens),0),
COALESCE(SUM(cache_read),0), COALESCE(SUM(cache_write),0)
FROM llm_usage
WHERE COALESCE(task_id,'') = $1
GROUP BY model
ORDER BY SUM(input_tokens) + SUM(output_tokens) DESC, model`, taskID)
if err != nil {
return nil, err
}
defer rows.Close()
out := []ModelTokenStat{}
for rows.Next() {
var m ModelTokenStat
if err := rows.Scan(&m.Model, &m.Calls, &m.InputTokens, &m.OutputTokens,
&m.CacheReadTokens, &m.CacheWriteTokens); err != nil {
return nil, err
}
out = append(out, m)
}
return out, rows.Err()
}
// ProfileUsage aggregates the whole ledger's token spend for one LLM profile
// (global, all tasks). Powers the dashboard's per-profile token card (new source).
type ProfileUsage struct {
ProfileName string `json:"profile_name"`
Calls int `json:"calls"`
Tasks int `json:"tasks"`
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheReadTokens int `json:"cache_read_tokens"`
CacheWriteTokens int `json:"cache_write_tokens"`
}
// UsageByProfile returns global token spend grouped by profile name, most-used
// first. profile_name may be empty for calls made on env/non-persisted configs.
func (d *DB) UsageByProfile() ([]ProfileUsage, error) {
rows, err := d.Query(`
SELECT COALESCE(profile_name,'') AS profile_name, COUNT(*) AS calls,
COUNT(DISTINCT task_id) AS tasks,
COALESCE(SUM(input_tokens),0), COALESCE(SUM(output_tokens),0),
COALESCE(SUM(cache_read),0), COALESCE(SUM(cache_write),0)
FROM llm_usage
GROUP BY profile_name
ORDER BY SUM(input_tokens) + SUM(output_tokens) DESC`)
if err != nil {
return nil, err
}
defer rows.Close()
out := []ProfileUsage{}
for rows.Next() {
var p ProfileUsage
if err := rows.Scan(&p.ProfileName, &p.Calls, &p.Tasks,
&p.InputTokens, &p.OutputTokens, &p.CacheReadTokens, &p.CacheWriteTokens); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
// ProfileDayUsage is one (profile, UTC calendar day) token bucket for the daily
// chart. Unlike the activity-based chart, ts is the real call time, so this is
// actual per-day consumption rather than tokens bucketed by task creation date.
type ProfileDayUsage struct {
ProfileName string `json:"profile_name"`
Date string `json:"date"` // YYYY-MM-DD (UTC)
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
CacheReadTokens int `json:"cache_read_tokens"`
}
// UsageDaily returns per-(profile, day) token buckets for the past `days` days
// (default 365 when days<=0), so the dashboard can slice by profile + range.
func (d *DB) UsageDaily(days int) ([]ProfileDayUsage, error) {
if days <= 0 {
days = 365
}
rows, err := d.Query(`
SELECT COALESCE(profile_name,'') AS profile_name,
to_char(ts AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day,
COALESCE(SUM(input_tokens),0), COALESCE(SUM(output_tokens),0), COALESCE(SUM(cache_read),0)
FROM llm_usage
WHERE ts >= now() - ($1 * interval '1 day')
GROUP BY profile_name, day
ORDER BY day`, days)
if err != nil {
return nil, err
}
defer rows.Close()
out := []ProfileDayUsage{}
for rows.Next() {
var p ProfileDayUsage
if err := rows.Scan(&p.ProfileName, &p.Date, &p.InputTokens, &p.OutputTokens, &p.CacheReadTokens); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
// ParseExpID turns the exploration-id segment parsed from a session string into an
// int64 (0 when empty/non-numeric, e.g. chat sessions keyed by conversation id).
func ParseExpID(s string) int64 {
n, err := strconv.ParseInt(s, 10, 64)
if err != nil || n < 0 {
return 0
}
return n
}
+46 -1
View File
@@ -235,7 +235,10 @@ CREATE TABLE IF NOT EXISTS llm_profiles (
rate_per_second DOUBLE PRECISION NOT NULL DEFAULT 0,
rate_per_minute DOUBLE PRECISION NOT NULL DEFAULT 0,
context_window_k INTEGER NOT NULL DEFAULT 0,
-- 思考参数拆成两个独立字段:thinking_type=思考开关(''/disabled/enabled),
-- reasoning_effort=思考强度(''/low/medium/high/xhigh/max),互不牵连。
reasoning_effort TEXT NOT NULL DEFAULT '',
thinking_type TEXT NOT NULL DEFAULT '',
is_default BOOLEAN NOT NULL DEFAULT false,
-- 轮询(故障转移)参数,见 docs/LLM轮询设计.md:
-- priority 顺位,越大越先被选中;激活配置(is_default)永远排链首,与本值无关。
@@ -253,6 +256,26 @@ CREATE TRIGGER trg_llm_upd BEFORE UPDATE ON llm_profiles
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0;
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS pool_exclude BOOLEAN NOT NULL DEFAULT false;
-- 思考开关字段 thinking_type,从旧的单一 reasoning_effort 语义一次性拆分而来。
-- schema.sql 每次启动都执行,故迁移必须只跑一次:仅当该列尚不存在时才回填,
-- 否则每次启动都会把用户后来手动设的组合覆盖回去。旧 reasoning_effort 语义:
-- 'off' → 显式关闭 → thinking_type='disabled',强度清空
-- 'low/medium/high/max' → 开启+强度 → thinking_type='enabled',强度保留
-- '' → 不发送 → 两者皆空(默认)
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'llm_profiles' AND column_name = 'thinking_type'
) THEN
ALTER TABLE llm_profiles ADD COLUMN thinking_type TEXT NOT NULL DEFAULT '';
UPDATE llm_profiles SET thinking_type = 'enabled'
WHERE reasoning_effort IN ('low','medium','high','max');
UPDATE llm_profiles SET thinking_type = 'disabled', reasoning_effort = ''
WHERE reasoning_effort = 'off';
END IF;
END $$;
-- LLM 轮询熔断状态:某个配置连续失败(余额不足/key 失效/限流)后进入冷却,冷却期内
-- 轮询直接跳过它。内存态为准,这里落库只为重启后不丢冷却窗口——加载时只取尚未
-- 到期的行(open_until > now),已到期的自然回到"正常",等下一次调用半开试探。
@@ -391,7 +414,7 @@ CREATE TABLE IF NOT EXISTS agents (
llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL,
current_prompt_id BIGINT,
max_turns INTEGER NOT NULL DEFAULT 0,
run_seconds INTEGER NOT NULL DEFAULT 600,
run_seconds INTEGER NOT NULL DEFAULT 1200,
web_search BOOLEAN NOT NULL DEFAULT false,
interactive_shell BOOLEAN NOT NULL DEFAULT false,
wrapup_prompt TEXT NOT NULL DEFAULT '',
@@ -411,6 +434,8 @@ ALTER TABLE agents ADD COLUMN IF NOT EXISTS trigger_merge_mode TEXT NOT NUL
ALTER TABLE agents ADD COLUMN IF NOT EXISTS trigger_max_parallel INTEGER NOT NULL DEFAULT 5;
-- per-agent LLM 绑定(agent 级默认模型):列自初版即在上方 CREATE 中,此 ALTER 仅为极旧库兜底(幂等)。
ALTER TABLE agents ADD COLUMN IF NOT EXISTS llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL;
-- run_seconds 单次 run 墙钟默认 600→1200:只改列默认(影响将来新插入的行),不动旧库存量行。
ALTER TABLE agents ALTER COLUMN run_seconds SET DEFAULT 1200;
CREATE INDEX IF NOT EXISTS idx_agents_llm_profile ON agents(llm_profile_id) WHERE llm_profile_id IS NOT NULL;
DROP TRIGGER IF EXISTS trg_agents_upd ON agents;
CREATE TRIGGER trg_agents_upd BEFORE UPDATE ON agents
@@ -501,6 +526,26 @@ CREATE TABLE IF NOT EXISTS agent_skill_visibility (
);
CREATE INDEX IF NOT EXISTS idx_askv_skill ON agent_skill_visibility(skill_name);
-- Skill 调用账本(见 db/skill_usage.go)。一次 Skill() 调用一行,只记维度不记正文。
-- 刻意不设外键:任务/会话删除后统计仍要保留(与 llm_usage 同理),skill 本身也只是
-- 文件系统上的目录名,没有对应的表。
CREATE TABLE IF NOT EXISTS skill_usage (
id BIGSERIAL PRIMARY KEY,
ts TIMESTAMPTZ NOT NULL DEFAULT now(),
skill TEXT NOT NULL,
agent_key TEXT,
task_id BIGINT,
exploration_id BIGINT,
intent_id BIGINT,
session_id TEXT,
args_len INTEGER NOT NULL DEFAULT 0,
-- false = 模型点名了一个不存在的 skill(未命中)。这类行同样保留:它反映"想用但没有"
-- 的缺口,是补 skill 的依据。
found BOOLEAN NOT NULL DEFAULT true
);
CREATE INDEX IF NOT EXISTS idx_skill_usage_skill ON skill_usage(skill, ts DESC);
CREATE INDEX IF NOT EXISTS idx_skill_usage_task ON skill_usage(task_id);
-- =====================================================================
-- H. 内置工具目录
-- =====================================================================
+200
View File
@@ -0,0 +1,200 @@
package db
import (
"database/sql"
"strings"
"time"
)
// SkillUsage is one Skill() invocation — the always-on skill call ledger. Written
// from the Skill meta-tool's OnInvoke hook (server/assembly.go), one row per load.
// Carries only dimensions (which skill, which agent, which task/session), never the
// caller's args text — args_len is kept so an "empty vs substantial context" split
// is still possible without storing prompt content, mirroring llm_usage.
//
// Rows deliberately outlive their task: skill_usage has no foreign keys, so deleting
// a task keeps its skill statistics intact (same rationale as llm_usage).
type SkillUsage struct {
Skill string `json:"skill"` // skill directory name (matches agent_skill_visibility.skill_name)
AgentKey string `json:"agent_key"` // worker / planner / mainagent / custom agent key
TaskID int64 `json:"task_id"` // 0 for non-task runs (chat sessions)
ExplorationID int64 `json:"exploration_id"` // 0 when unknown
IntentID int64 `json:"intent_id"` // worker's intent node; 0 for planner/mainagent/chat
SessionID string `json:"session_id"` // chat conversation id; empty for task runs
ArgsLen int `json:"args_len"`
Found bool `json:"found"` // false = the model named a skill that does not exist
}
// InsertSkillUsage appends one ledger row. Best-effort: callers log and continue on
// error (a lost metering row must never break a skill invocation).
func (d *DB) InsertSkillUsage(u *SkillUsage) error {
_, err := d.Exec(`
INSERT INTO skill_usage(skill, agent_key, task_id, exploration_id, intent_id, session_id, args_len, found)
VALUES ($1, NULLIF($2,''), $3, $4, $5, NULLIF($6,''), $7, $8)`,
u.Skill, u.AgentKey, nullIfZero(u.TaskID), nullIfZero(u.ExplorationID),
nullIfZero(u.IntentID), u.SessionID, u.ArgsLen, u.Found)
return err
}
func nullIfZero(v int64) any {
if v > 0 {
return v
}
return nil
}
// SkillStat is one skill's aggregate usage, for the skills page.
type SkillStat struct {
Skill string `json:"skill"`
Calls int `json:"calls"`
Tasks int `json:"tasks"` // distinct tasks that loaded it (chat runs excluded)
Agents []string `json:"agents"` // agent keys that loaded it, most-used first
LastUsed *time.Time `json:"last_used"` // nil when never called
}
// SkillStats aggregates the whole ledger grouped by skill, most-used first. Skills
// that were never invoked are absent — callers merge against the skill list on disk.
// Only resolved calls count; misses are reported separately by MissingSkillStats.
func (d *DB) SkillStats() ([]SkillStat, error) {
// agent keys come back as one comma-joined string rather than text[]: the pgx
// stdlib driver has no database/sql Scan target for arrays, and agent keys are
// [a-z0-9_-] so a comma join is unambiguous.
rows, err := d.Query(`
SELECT skill, COUNT(*) AS calls,
COUNT(DISTINCT task_id) AS tasks,
COALESCE(STRING_AGG(DISTINCT agent_key, ','), '') AS agents,
MAX(ts) AS last_used
FROM skill_usage
WHERE found
GROUP BY skill
ORDER BY COUNT(*) DESC, skill`)
if err != nil {
return nil, err
}
defer rows.Close()
out := []SkillStat{}
for rows.Next() {
var s SkillStat
var agents string
var lastUsed sql.NullTime
if err := rows.Scan(&s.Skill, &s.Calls, &s.Tasks, &agents, &lastUsed); err != nil {
return nil, err
}
s.Agents = []string{}
if agents != "" {
s.Agents = strings.Split(agents, ",")
}
if lastUsed.Valid {
t := lastUsed.Time
s.LastUsed = &t
}
out = append(out, s)
}
return out, rows.Err()
}
// MissingSkillStats returns the skill names agents asked for that do not exist,
// most-requested first — the "wished it existed" gap list. Names come from the model
// so they are shown as-is (already length-capped at insert time).
func (d *DB) MissingSkillStats(limit int) ([]SkillStat, error) {
if limit <= 0 || limit > 200 {
limit = 20
}
rows, err := d.Query(`
SELECT skill, COUNT(*) AS calls,
COALESCE(STRING_AGG(DISTINCT agent_key, ','), '') AS agents,
MAX(ts) AS last_used
FROM skill_usage
WHERE NOT found
GROUP BY skill
ORDER BY COUNT(*) DESC, skill
LIMIT $1`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := []SkillStat{}
for rows.Next() {
var s SkillStat
var agents string
var lastUsed sql.NullTime
if err := rows.Scan(&s.Skill, &s.Calls, &agents, &lastUsed); err != nil {
return nil, err
}
s.Agents = []string{}
if agents != "" {
s.Agents = strings.Split(agents, ",")
}
if lastUsed.Valid {
t := lastUsed.Time
s.LastUsed = &t
}
out = append(out, s)
}
return out, rows.Err()
}
// SkillCall is one row of a skill's recent-call list (detail panel).
type SkillCall struct {
TS time.Time `json:"ts"`
AgentKey string `json:"agent_key"`
TaskID int64 `json:"task_id"`
SessionID string `json:"session_id"`
ArgsLen int `json:"args_len"`
}
// RecentSkillCalls returns the most recent invocations of one skill, newest first.
func (d *DB) RecentSkillCalls(skill string, limit int) ([]SkillCall, error) {
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := d.Query(`
SELECT ts, COALESCE(agent_key,''), COALESCE(task_id,0), COALESCE(session_id,''), args_len
FROM skill_usage
WHERE skill = $1 AND found
ORDER BY ts DESC
LIMIT $2`, skill, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := []SkillCall{}
for rows.Next() {
var c SkillCall
if err := rows.Scan(&c.TS, &c.AgentKey, &c.TaskID, &c.SessionID, &c.ArgsLen); err != nil {
return nil, err
}
out = append(out, c)
}
return out, rows.Err()
}
// SkillCallsByTask counts a task's skill loads, most-used first. Powers a per-task
// view of which procedures its agents actually reached for.
func (d *DB) SkillCallsByTask(taskID int64) ([]SkillStat, error) {
rows, err := d.Query(`
SELECT skill, COUNT(*) AS calls, MAX(ts) AS last_used
FROM skill_usage
WHERE task_id = $1 AND found
GROUP BY skill
ORDER BY COUNT(*) DESC, skill`, taskID)
if err != nil {
return nil, err
}
defer rows.Close()
out := []SkillStat{}
for rows.Next() {
var s SkillStat
var lastUsed sql.NullTime
if err := rows.Scan(&s.Skill, &s.Calls, &lastUsed); err != nil {
return nil, err
}
s.Agents = []string{}
if lastUsed.Valid {
t := lastUsed.Time
s.LastUsed = &t
}
out = append(out, s)
}
return out, rows.Err()
}
+116
View File
@@ -0,0 +1,116 @@
package db
import (
"testing"
)
// TestSkillUsageLedger writes a few rows and checks the three aggregates the
// skills page relies on: per-skill totals (resolved calls only), the miss list,
// and the recent-call list. Rows are namespaced by a unique skill name so the
// shared dev DB stays usable and the test cleans up after itself.
func TestSkillUsageLedger(t *testing.T) {
d, err := Open(testDSN(t))
if err != nil {
t.Skipf("postgres unavailable (%v) — skipping", err)
}
defer d.Close()
const (
skillA = "zz-test-skill-a"
missing = "zz-test-skill-missing"
)
cleanup := func() {
_, _ = d.Exec(`DELETE FROM skill_usage WHERE skill IN ($1,$2)`, skillA, missing)
}
cleanup()
defer cleanup()
rows := []*SkillUsage{
{Skill: skillA, AgentKey: "worker", TaskID: 991, ExplorationID: 5, IntentID: 7, ArgsLen: 12, Found: true},
{Skill: skillA, AgentKey: "worker", TaskID: 991, ExplorationID: 5, ArgsLen: 0, Found: true},
{Skill: skillA, AgentKey: "planner", TaskID: 992, ExplorationID: 6, Found: true},
{Skill: skillA, AgentKey: "chatbot", SessionID: "conv-1", Found: true},
{Skill: missing, AgentKey: "worker", TaskID: 991, Found: false},
{Skill: missing, AgentKey: "worker", TaskID: 991, Found: false},
}
for _, r := range rows {
if err := d.InsertSkillUsage(r); err != nil {
t.Fatalf("insert %s: %v", r.Skill, err)
}
}
stats, err := d.SkillStats()
if err != nil {
t.Fatal(err)
}
var got *SkillStat
for i := range stats {
if stats[i].Skill == skillA {
got = &stats[i]
}
if stats[i].Skill == missing {
t.Fatalf("misses must not appear in SkillStats: %+v", stats[i])
}
}
if got == nil {
t.Fatalf("skill %s absent from SkillStats", skillA)
}
if got.Calls != 4 {
t.Errorf("calls: want 4, got %d", got.Calls)
}
// 991 + 992; the chat row has a NULL task_id and COUNT(DISTINCT) skips NULLs.
if got.Tasks != 2 {
t.Errorf("tasks: want 2, got %d", got.Tasks)
}
if len(got.Agents) != 3 {
t.Errorf("agents: want 3 distinct, got %v", got.Agents)
}
if got.LastUsed == nil {
t.Error("last_used must be set")
}
miss, err := d.MissingSkillStats(10)
if err != nil {
t.Fatal(err)
}
var missCalls int
for _, m := range miss {
if m.Skill == missing {
missCalls = m.Calls
}
}
if missCalls != 2 {
t.Errorf("missing calls: want 2, got %d", missCalls)
}
calls, err := d.RecentSkillCalls(skillA, 10)
if err != nil {
t.Fatal(err)
}
if len(calls) != 4 {
t.Fatalf("recent calls: want 4, got %d", len(calls))
}
// newest first
for i := 1; i < len(calls); i++ {
if calls[i].TS.After(calls[i-1].TS) {
t.Errorf("recent calls not newest-first at %d", i)
}
}
byTask, err := d.SkillCallsByTask(991)
if err != nil {
t.Fatal(err)
}
var taskCalls int
for _, s := range byTask {
if s.Skill == skillA {
taskCalls = s.Calls
}
if s.Skill == missing {
t.Errorf("misses must not appear in SkillCallsByTask: %+v", s)
}
}
if taskCalls != 2 {
t.Errorf("task 991 calls: want 2, got %d", taskCalls)
}
}
+18 -3
View File
@@ -113,11 +113,15 @@ func (s *AssetStore) AddAutoScope(taskID int64, assetType, domain, rawURL, ip st
return nil
}
// AddAgentScope parses an agent-provided (kind,value) and records it (source='agent').
// AddAgentScope parses a (kind,value) pair and records it as task scope.
// source is "agent" when called from an LLM tool, "manual" from the UI.
// company: value = company name or id (must already exist). root_domain/subdomain:
// value = a domain. ip/cidr: value = an IP or CIDR (bare IP → /32,/128).
func (s *AssetStore) AddAgentScope(taskID int64, kind, value, reason string) (TaskScope, error) {
ts := TaskScope{TaskID: taskID, Kind: kind, Source: "agent", Reason: reason}
func (s *AssetStore) AddAgentScope(taskID int64, kind, value, reason, source string) (TaskScope, error) {
if source == "" {
source = "agent"
}
ts := TaskScope{TaskID: taskID, Kind: kind, Source: source, Reason: reason}
if taskID <= 0 {
return ts, fmt.Errorf("需要 task_id")
}
@@ -184,6 +188,17 @@ func (s *AssetStore) AddAgentScope(taskID int64, kind, value, reason string) (Ta
return ts, nil
}
// DeleteTaskScope removes a single scope row by id, scoped to the given task.
// Returns whether a row was actually deleted.
func (s *AssetStore) DeleteTaskScope(taskID, scopeID int64) (bool, error) {
res, err := s.db.Exec(`DELETE FROM task_scope WHERE id=$1 AND task_id=$2`, scopeID, taskID)
if err != nil {
return false, err
}
n, _ := res.RowsAffected()
return n > 0, nil
}
// ListTaskScope returns all scope rows for a task.
func (s *AssetStore) ListTaskScope(taskID int64) ([]TaskScope, error) {
rows, err := s.db.Query(`
+2
View File
@@ -437,6 +437,8 @@ DELETE FROM assets a USING deletable d WHERE a.id=d.id`, id, expID)
}
result.LLMRecordsDeleted, _ = res.RowsAffected()
}
// llm_usage (the token metering ledger) is intentionally NOT deleted with the
// task — it is kept as historical accounting even after the task is gone.
if _, err := tx.Exec(`DELETE FROM tasks WHERE id=$1`, id); err != nil {
return result, err
}