diff --git a/CHANGELOG.md b/CHANGELOG.md index f604033..a3fbdeb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,55 @@ ## [Unreleased] +## [0.3.4] - 2026-08-24 + +### UI + +#### 新增的功能 + +- 任务列表勾选多个任务后可批量修改分类，目标分类支持「未分类」以将任务移出当前分类；整批写入在一个事务内完成，失败项只会是勾选后已被删除的任务。 + +#### 修改的功能 + +- 新建任务的「任务分类」由下拉选择改为可搜索输入框：输入即过滤已有分类，库里没有的名称回车（或点下拉里的「创建」）即时新建并选中，选中项以可移除的 tag 展示；仍限单个分类。 + +### 对话 + +#### 修复的问题 + +- 修复会话已选定 LLM 配置、但全局无激活配置时，发送消息被「LLM 未配置，无法对话」拦截的问题。发送前的检查原本只看全局激活配置，而实际执行会优先用会话所选配置，两处解析逻辑不一致；现统一为同一套（会话/Agent 绑定优先，全局兜底）。 +- 无法对话时的提示按状态细分：完全没有配置提示「添加配置」，有配置但未激活提示「激活或为本会话指定一个配置」，不再一律显示「未配置」，便于定位。 + +### Agent + +#### 新增的功能 + +- 目标已全部达成的任务，主 Agent 经 `add_intent` 下发的意图可被 Worker 直接领取执行、跑完即止：此时 Planner 不再运行（任务无 open 目标即不进规划），避免其重判目标达成而取消刚下发的意图；frontier 抽干后任务回到已完成。主 Agent 下发意图前会依据意图内容反问用户是否将其登记为正式目标，登记则任务恢复常规自主规划。 + +#### 修改的功能 + +- `get_worker_trace` / `get_task_worker_trace` 的 `step_ids` 超过一次上限（5 个）时不再直接报错，改为返回前 5 步的完整内容，并在结果中通过 `returned_step_ids` / `omitted_step_ids` / `notice` 告知本次取了哪些、还有哪些未取，并提示「若已足够则无需再取」。传入的重复与非法 id 会先去重、剔除后再计数。 + +#### 修复的问题 + +- 任务列表改为按任务 id 降序（最新创建在最前），替换原先的按创建时间排序。同一时刻创建的多个任务时间戳相同、排序不稳定，配合每 10 秒轮询和内存 map 的随机遍历，会导致它们在列表里频繁互换位置。 + +### 资产 + +#### 修复的问题 + +- 修复企业归属重算被单行脏数据打挂：`assets.ip` 存了主机名（Agent 或资产 API 写入）时，`a.ip::inet` 会抛 `22P02`，导致新增/修改企业资产范围和删除企业全部失败并回滚。改用安全转换 `try_inet()`，非法值跳过而不是中断整条语句。 +- 无法解析的 `ip` 不再静默跳过：企业范围保存后会明确提示有多少条资产的 `ip` 不是合法 IP（含具体 id 和取值），提醒这些资产不会被 IP/CIDR 规则归属；同一信息也写入服务端日志，覆盖没有前端响应的路径（删除企业、scopesentry 同步、Agent 写入）。 +- 任务测试范围的 IP/CIDR 匹配同步改用 `try_inet()`，替换原先只检查字符集的正则守卫（`abc.def` 这类全 hex 字母的主机名此前仍会漏过并触发同样的报错）。 + +#### 修改的功能 + +- `ip`、`service`、`endpoint` 资产的 `ip` 字段不再接受主机名：`insert_assets` 和资产 API 会逐条返回带 `index` 的错误，并直接给出改正方法（改用 `type=subdomain` 填 `domain`，或先解析 A/AAAA 记录），Agent 可据此自行修正后重插。同批次中的其他资产不受影响，照常入库。 + +### 贡献者 + +- [@neouks](https://github.com/neouks) + ## [0.3.3] - 2026-08-23 ### Worker @@ -194,6 +243,7 @@ - [@Autumn-27](https://github.com/Autumn-27) - [@neouks](https://github.com/neouks) -[Unreleased]: https://github.com/Autumn-27/ARTEX/compare/v0.3.3...HEAD +[Unreleased]: https://github.com/Autumn-27/ARTEX/compare/v0.3.4...HEAD +[0.3.4]: https://github.com/Autumn-27/ARTEX/compare/v0.3.3...v0.3.4 [0.3.3]: https://github.com/Autumn-27/ARTEX/compare/v0.3.2...v0.3.3 [0.3.2]: https://github.com/Autumn-27/ARTEX/compare/v0.3.1...v0.3.2 diff --git a/agent/blackboard_inheritance_test.go b/agent/blackboard_inheritance_test.go index 343d422..fcbcb6f 100644 --- a/agent/blackboard_inheritance_test.go +++ b/agent/blackboard_inheritance_test.go @@ -272,3 +272,75 @@ func TestBlackboardToolsReadDirectSources(t *testing.T) { t.Fatalf("inherited trace search: %#v", hits) } } + +// TestGetWorkerTraceStepIDsDegradeGracefully pins the over-cap behaviour: instead +// of erroring, get_worker_trace returns the first 5 requested steps and tells the +// model which ids it deferred, after de-duplicating and dropping invalid ids. +func TestGetWorkerTraceStepIDsDegradeGracefully(t *testing.T) { + d := testDB(t) + defer d.Close() + + task, err := d.CreateTask("trace-cap", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(task.ID) }) + store := d.Exploration(task.ExplorationID) + intent, err := store.AddIntent(map[string]any{"summary": "cap work"}, 1, nil, "planner") + if err != nil { + t.Fatal(err) + } + + var stepIDs []int64 + for i := range 6 { + sid, err := store.AppendActivity(db.Activity{ + NodeID: &intent, Worker: "w", Kind: "result", Tool: "HTTP", + Summary: fmt.Sprintf("step %d", i), Detail: fmt.Sprintf("detail %d", i), + }) + if err != nil { + t.Fatal(err) + } + stepIDs = append(stepIDs, sid) + } + + tools := NewToolSet(store, "worker") + tools.SetAssetStore(d.Assets(), d.Companies()) + + // Request 7 ids: a duplicate of the first, an invalid 0, then all 6 real ids. + // After dedup/cleanup that is 6 valid ids — one over the cap. + requested := []int64{stepIDs[0], stepIDs[0], 0, stepIDs[1], stepIDs[2], stepIDs[3], stepIDs[4], stepIDs[5]} + input, _ := json.Marshal(map[string]any{"intent_id": intent, "step_ids": requested}) + res := callReadJSON(t, tools.getWorkerTrace(), string(input)).(map[string]any) + + returned, _ := res["returned_step_ids"].([]any) + if len(returned) != 5 { + t.Fatalf("returned_step_ids=%v, want the first 5", res["returned_step_ids"]) + } + // First 5 distinct valid ids, in request order. + wantReturned := []int64{stepIDs[0], stepIDs[1], stepIDs[2], stepIDs[3], stepIDs[4]} + for i, raw := range returned { + if int64(raw.(float64)) != wantReturned[i] { + t.Fatalf("returned[%d]=%v, want %d", i, raw, wantReturned[i]) + } + } + omitted, _ := res["omitted_step_ids"].([]any) + if len(omitted) != 1 || int64(omitted[0].(float64)) != stepIDs[5] { + t.Fatalf("omitted_step_ids=%v, want [%d]", res["omitted_step_ids"], stepIDs[5]) + } + if notice, _ := res["notice"].(string); notice == "" { + t.Fatalf("notice missing — model would not know a step was deferred") + } + if steps, _ := res["steps"].([]any); len(steps) != 5 { + t.Fatalf("steps=%d, want 5 detail rows", len(steps)) + } + + // At or under the cap: no notice, no omitted list. + okInput, _ := json.Marshal(map[string]any{"intent_id": intent, "step_ids": stepIDs[:3]}) + okRes := callReadJSON(t, tools.getWorkerTrace(), string(okInput)).(map[string]any) + if _, hasNotice := okRes["notice"]; hasNotice { + t.Fatalf("notice present for an in-cap request: %#v", okRes["notice"]) + } + if _, hasOmitted := okRes["omitted_step_ids"]; hasOmitted { + t.Fatalf("omitted_step_ids present for an in-cap request: %#v", okRes["omitted_step_ids"]) + } +} diff --git a/agent/insert_assets_test.go b/agent/insert_assets_test.go index 1a4a195..573562d 100644 --- a/agent/insert_assets_test.go +++ b/agent/insert_assets_test.go @@ -3,6 +3,7 @@ package agent import ( "context" "encoding/json" + "strings" "testing" "github.com/Autumn-27/artex/db" @@ -394,3 +395,56 @@ func TestInsertAssetsDedup(t *testing.T) { t.Errorf("dedup: want same ID on double insert, got %v vs %v", id1, id2) } } + +// ===================================================================== +// TestInsertAssetsRejectsHostnameIPPerItem +// 一批里混入 ip 填了主机名的一条 → 只有那条失败，其余照常入库， +// 且错误里带得上 index 和改正方法，Agent 下一轮能自己修好。 +// ===================================================================== +func TestInsertAssetsRejectsHostnameIPPerItem(t *testing.T) { + d := testDB(t) + defer d.Close() + + ts := NewToolSet(nil, "") + ts.SetAssetStore(d.Assets(), d.Companies()) + defer d.Exec(`DELETE FROM assets WHERE ip IN ('198.51.100.23','cdn.badip-test.com') OR domain='badip-test.com'`) + + out := callInsertAssets(t, ts, map[string]any{ + "assets": []any{ + map[string]any{"type": "root_domain", "domain": "badip-test.com"}, + map[string]any{"type": "ip", "ip": "cdn.badip-test.com"}, + map[string]any{"type": "ip", "ip": "198.51.100.23"}, + }, + }) + + // The two valid entries must survive the bad one — a whole-batch failure + // would make the agent re-send assets that were already fine. + results, _ := out["results"].([]any) + if len(results) != 2 { + t.Fatalf("results=%v, want the 2 valid assets", out["results"]) + } + + errsRaw, _ := out["errors"].([]any) + if len(errsRaw) != 1 { + t.Fatalf("errors=%v, want exactly the invalid entry", out["errors"]) + } + entry, _ := errsRaw[0].(map[string]any) + if index, _ := entry["index"].(float64); int(index) != 1 { + t.Fatalf("error index=%v, want 1", entry["index"]) + } + message, _ := entry["error"].(string) + for _, want := range []string{"cdn.badip-test.com", "type=subdomain", "A/AAAA"} { + if !strings.Contains(message, want) { + t.Fatalf("error message %q lacks %q — agent cannot act on it", message, want) + } + } + + // The rejected value must not have reached the table. + var stored int + if err := d.QueryRow(`SELECT count(*) FROM assets WHERE ip='cdn.badip-test.com'`).Scan(&stored); err != nil { + t.Fatal(err) + } + if stored != 0 { + t.Fatalf("rejected hostname still stored in assets.ip (%d rows)", stored) + } +} diff --git a/agent/mainagent.go b/agent/mainagent.go index be9e573..99531df 100644 --- a/agent/mainagent.go +++ b/agent/mainagent.go @@ -62,7 +62,8 @@ const mainAgentDefaultTmpl = `你是一个授权渗透测试系统的"主 agent" 1. 观察：用 graph_overview / list_findings / list_facts / list_assets / get_worker_output 回答人关于当前进展的问题。 2. 操舵（把人的意图落到系统）： - 人想"改方向/强调某类漏洞/重点某区域" → 用 add_hint 写提示（规划者下次会读到）。 - - 人想"立刻测某个具体目标" → 用 add_intent 直接注入一条高优先级意图（priority 8-10）。 + - 人想"立刻测某个具体目标" → 用 add_intent 直接注入一条高优先级意图（priority 8-10）。系统会自动把已完成的任务拉回运行态、让 worker 领这条意图执行，跑完即回到已完成状态。 + **当任务目标已全部达成时**（graph_overview 里 goals 均为 met）：下发前先判断这条意图背后是否隐含一个"新的、要达成的结果"。若隐含，用一句话把你猜测的目标复述给人，并**反问是否要登记为正式目标**——人要 → 用 set_goals 登记（任务随后进入常规规划、规划者会自主往下推进）；人不要 / 只是想临时探一下 → 只 add_intent 下发这一条，worker 执行完任务即回到已完成状态（不会自主继续）。若这条意图明显只是一次性查证、不隐含新目标，直接 add_intent 即可，不必每次都问。 - 人想"对某条正在运行的意图(work)实时纠偏（别再走 X、聚焦 Y）" → 用 steer_work（不打断、不丢已有进展，worker 下一步动作前生效）；先用 get_worker_output 看它在干嘛。方向整个错了则改用 add_intent 另下新意图。 - 人想"新增一个要达成的最终目标" → 用 set_goals 增补目标。系统会把该目标写入任务图并**自动把已完成/暂停的任务拉回运行态继续跑**（规划者随后会据此重新判断是否达成），无需人工再点恢复。 - 人想"增/改测试约束（允许/禁止某类操作，如『仅测当前端口』『禁止爆破』『只做被动侦察』）" → 用 set_constraints 登记（type=allow 允许 / type=deny 禁止）。约束会在下一轮规划时注入 planner/worker 的提示词以框定探索边界；也可在总览「约束管理」里增删改。 diff --git a/agent/tools.go b/agent/tools.go index a1830f2..f8493ab 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -1016,6 +1016,7 @@ func (t *ToolSet) addIntent() actool.CoreTool { ids := make([]int64, len(items)) errs := map[string]string{} + createdAny := false for i, it := range items { id, err := t.addOneIntent(it) if err != nil { @@ -1023,6 +1024,15 @@ func (t *ToolSet) addIntent() actool.CoreTool { continue } ids[i] = id + createdAny = true + } + + // 人经主 agent 直投意图 → 若任务已 done（无 open 目标的 goalless 分支），把它 + // 拉回 running，worker 才能领这条意图执行。resumeTask 仅由主 agent 的 Chat 接入 + // (SetResumeTask)；planner 的 ToolSet 为 nil，故 planner 自己调 add_intent 时此段 + // no-op，不影响其正常产意图。意图节点已在上面建好(open)，复活时不会被误判抽干。 + if createdAny && t.resumeTask != nil { + t.resumeTask() } if !batch { // 单条：保持原返回 @@ -1676,12 +1686,12 @@ func (t *ToolSet) getWorkerTrace() actool.CoreTool { "查看某条意图(work)的【执行过程】（区别于 get_worker_output 只给最终结论）。三种用法：\n"+ "① 只传 intent_id → 返回该 work 每一步的摘要流（summary≤100字，含 step_id；只是动作轮廓，不含完整输出）；\n"+ "② intent_id + q → 只返回命中关键字的步骤摘要（在摘要和完整输出里都搜；仍只给 summary，要看内容用③）；\n"+ - "③ intent_id + step_ids（最多5个）→ 返回这些步骤的完整内容(detail)。\n"+ + "③ intent_id + step_ids → 返回这些步骤的完整内容(detail)；一次最多取 5 个，超出只返回前 5 个并在 notice/omitted_step_ids 里告知未取的。\n"+ "典型流程：先①/②定位可疑步骤的 step_id，再用③取其完整输出。不含思考(thinking)步骤。支持直接关联任务的历史 trace；其结果带 source_task_id/inherited=true 且只读。", obj(map[string]any{ "intent_id": idp("意图 id（= work 句柄）"), "q": str("关键字：只返回摘要/完整输出命中它的步骤（可选；与 step_ids 互斥）"), - "step_ids": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}, "description": "要取完整内容的 step_id（来自①/②返回，一次最多 5 个）"}, + "step_ids": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}, "description": "要取完整内容的 step_id（来自①/②返回；一次最多取 5 个，多传只返回前 5 个，其余在 omitted_step_ids 里列出）"}, "limit": intp("摘要流/检索的返回上限（可选）"), }, "intent_id"), func(_ context.Context, in json.RawMessage) (actool.Result, error) { @@ -1703,16 +1713,26 @@ func (t *ToolSet) getWorkerTrace() actool.CoreTool { if intentNode == nil || intentNode.Kind != db.KindIntent { return actool.Errorf("intent_id 不属于本任务或其直接关联任务"), nil } - // ③ detail drill-down by step ids (max 5), thinking excluded by the store. + // ③ detail drill-down by step ids, thinking excluded by the store. if len(a.StepIDs) > 0 { + // Dedup + drop invalid ids first so garbage/duplicates don't eat into + // the per-call cap. detail is returned in full (untruncated), so the + // cap bounds one tool result; over the cap we serve the first N and + // tell the model exactly which ids were deferred, instead of erroring + // and forcing it to re-plan the call. + const maxStepIDs = 5 var ids []int64 + seen := make(map[int64]bool) for _, raw := range a.StepIDs { - if v := pid(raw); v > 0 { + if v := pid(raw); v > 0 && !seen[v] { + seen[v] = true ids = append(ids, v) } } - if len(ids) > 5 { - return actool.Errorf("step_ids 一次最多 5 个（先用不带 step_ids 的调用定位，再分批取）"), nil + var omitted []int64 + if len(ids) > maxStepIDs { + omitted = append(omitted, ids[maxStepIDs:]...) + ids = ids[:maxStepIDs] } acts, err := t.ts.ActivityByIDsWithSources(ids) if err != nil { @@ -1733,7 +1753,16 @@ func (t *ToolSet) getWorkerTrace() actool.CoreTool { } steps = append(steps, step) } - result := map[string]any{"intent_id": id, "steps": steps} + result := map[string]any{"intent_id": id, "steps": steps, "returned_step_ids": ids} + if len(omitted) > 0 { + // returned_step_ids/omitted_step_ids let the model decide programmatically + // whether another call is worth it; the notice states the same in prose. + result["omitted_step_ids"] = omitted + result["notice"] = fmt.Sprintf( + "每次最多取 %d 个步骤的完整内容，本次已返回前 %d 个（%v），未取的 %d 个为 %v。"+ + "若这些内容已足够定位，则无需再取剩余步骤；确需继续时，用这些 step_id 再调一次。", + maxStepIDs, len(ids), ids, len(omitted), omitted) + } if intentNode.Inherited { inheritedMap(result, intentNode.SourceTaskID) } diff --git a/agent/tools_insert.go b/agent/tools_insert.go index 5b4d58c..8c6fb5b 100644 --- a/agent/tools_insert.go +++ b/agent/tools_insert.go @@ -79,7 +79,7 @@ func (t *ToolSet) insertAssets() actool.CoreTool { "发现新资产时立即批量插入资产到资产库。支持一次提交多种资产类型：root_domain/ip/subdomain/app/service/endpoint。\n"+ "每条资产的字段说明：\n"+ "• root_domain：domain(根域名, 必填)、icp(备案, 可选)\n"+ - "• ip：ip(IP地址, 必填)、bound_domains([域名])、open_ports([{port,service}])\n"+ + "• ip：ip(IP地址, 必填, 必须是 IPv4/IPv6 地址而非主机名)、bound_domains([域名])、open_ports([{port,service}])\n"+ "• subdomain：domain(子域名, 必填)、record_type(A/AAAA/CNAME等)、record_value(记录值)、icp(可选)\n"+ "• app：app_name(应用名, 必填)、bundle_id(bundle id, 可选)、category、description、app_icp、company_id(归属企业 id, 可选)\n"+ "• service(http)：url(必填)、technologies([指纹])、status_code、content_length、page_title、favicon_mmh3、auth([{type,username,password,...}])、service_ip\n"+ @@ -108,7 +108,7 @@ func (t *ToolSet) insertAssets() actool.CoreTool { "description": "DNS 解析值列表（subdomain 可选，如 [\"1.2.3.4\",\"2.3.4.5\"]）", }, // ip - "ip": str("IP 地址（ip 类型必填；service/endpoint 类型可填，用于关联 IP）"), + "ip": str("IP 地址，必须是 IPv4/IPv6 地址，不能填主机名（主机名请用 type=subdomain 的 domain 字段）；ip 类型必填；service/endpoint 类型可填，用于关联 IP"), "bound_domains": map[string]any{ "type": "array", "items": map[string]any{"type": "string"}, diff --git a/db/assets.go b/db/assets.go index ec3a7cb..ebb2534 100644 --- a/db/assets.go +++ b/db/assets.go @@ -3,6 +3,7 @@ package db import ( "database/sql" "encoding/json" + "errors" "fmt" "net" "net/url" @@ -300,6 +301,27 @@ RETURNING id`, domain, icpVal, companyID, taskIDs).Scan(&id) return id, err } +// ErrAssetIPInvalid marks a non-address value in an asset's ip field. Both +// insert_assets and the asset API report it per item with the item index, so one +// bad entry never costs the rest of the batch. +var ErrAssetIPInvalid = errors.New("invalid asset ip") + +// ValidateAssetIP keeps hostnames out of assets.ip. Network attribution casts +// that column to inet (see try_inet in schema.sql), so a hostname stored here is +// silently invisible to every IP/CIDR scope rule — the asset simply never gets +// attributed and nobody can tell why. The message states the fix rather than +// just the fault, so an agent can correct the item on its next turn. An empty +// value is accepted: the ip field is optional for service and endpoint assets. +func ValidateAssetIP(value string) error { + if value == "" || net.ParseIP(value) != nil { + return nil + } + return fmt.Errorf( + "%w: ip 必须是 IPv4/IPv6 地址，收到 %q。若这是主机名，请改用 type=subdomain 并填 domain 字段；"+ + "若确实要登记地址，请先解析出 A/AAAA 记录，再用解析出的地址填 ip", + ErrAssetIPInvalid, value) +} + // ===================================================================== // UpsertIP // ===================================================================== @@ -317,6 +339,9 @@ func (s *AssetStore) UpsertIP(req UpsertIPReq) (int64, error) { if req.IP == "" { return 0, fmt.Errorf("ip is required") } + if err := ValidateAssetIP(req.IP); err != nil { + return 0, err + } if s.tx == nil { return s.withCompanyScopeMutation(func(scoped *AssetStore) (int64, error) { return scoped.UpsertIP(req) @@ -659,6 +684,9 @@ func (s *AssetStore) UpsertHTTPService(req UpsertHTTPServiceReq) (int64, error) if req.URL == "" { return 0, fmt.Errorf("url is required") } + if err := ValidateAssetIP(req.IP); err != nil { + return 0, err + } if s.tx == nil { return s.withCompanyScopeMutation(func(scoped *AssetStore) (int64, error) { return scoped.UpsertHTTPService(req) @@ -793,6 +821,9 @@ func (s *AssetStore) UpsertOtherService(req UpsertOtherServiceReq) (int64, error if req.Domain == "" && req.IP == "" { return 0, fmt.Errorf("domain or ip is required") } + if err := ValidateAssetIP(req.IP); err != nil { + return 0, err + } if req.Port == 0 { return 0, fmt.Errorf("port is required") } @@ -936,6 +967,9 @@ func (s *AssetStore) UpsertEndpoint(req UpsertEndpointReq) (int64, error) { if req.Method == "" { return 0, fmt.Errorf("method is required") } + if err := ValidateAssetIP(req.IP); err != nil { + return 0, err + } if s.tx == nil { return s.withCompanyScopeMutation(func(scoped *AssetStore) (int64, error) { return scoped.UpsertEndpoint(req) diff --git a/db/assets_test.go b/db/assets_test.go index c2ae2e5..9bebb97 100644 --- a/db/assets_test.go +++ b/db/assets_test.go @@ -1,8 +1,10 @@ package db import ( + "errors" "fmt" "slices" + "strings" "testing" "time" ) @@ -946,3 +948,54 @@ func TestAssetPaginationUsesStableIDTieBreaker(t *testing.T) { t.Fatalf("DSL pagination order=%v want=%v", got, want) } } + +// TestAssetIPRejectsHostname pins the write-side guard added after a hostname in +// assets.ip broke company attribution: every asset type that persists an ip must +// refuse a non-address, and the message must tell the caller how to fix it. +func TestAssetIPRejectsHostname(t *testing.T) { + d, av2, _ := testSetup(t) + defer d.Close() + + hostname := "d63cd476.cdn.ucloud.com.cn" + cases := []struct { + name string + call func() (int64, error) + }{ + {"ip", func() (int64, error) { + return av2.UpsertIP(UpsertIPReq{IP: hostname}) + }}, + {"http service", func() (int64, error) { + return av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: "https://example.com/", IP: hostname}) + }}, + {"other service", func() (int64, error) { + return av2.UpsertOtherService(UpsertOtherServiceReq{IP: hostname, Port: 22, ServiceName: "ssh"}) + }}, + {"endpoint", func() (int64, error) { + return av2.UpsertEndpoint(UpsertEndpointReq{URL: "https://example.com/a", Method: "GET", IP: hostname}) + }}, + } + for _, tc := range cases { + id, err := tc.call() + if !errors.Is(err, ErrAssetIPInvalid) { + deleteAsset(d, id) + t.Fatalf("%s: err=%v, want %v", tc.name, err, ErrAssetIPInvalid) + } + // The caller has to learn the remedy, not just that something was wrong. + if !strings.Contains(err.Error(), hostname) || !strings.Contains(err.Error(), "type=subdomain") { + t.Fatalf("%s: message is not actionable: %v", tc.name, err) + } + } + + // An empty ip stays legal for the types where it is optional, and a real + // address is still accepted. + id, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: "https://ip-guard.example.com/"}) + if err != nil { + t.Fatalf("empty ip rejected: %v", err) + } + deleteAsset(d, id) + id, err = av2.UpsertIP(UpsertIPReq{IP: "203.0.113.7"}) + if err != nil { + t.Fatalf("valid ip rejected: %v", err) + } + deleteAsset(d, id) +} diff --git a/db/companies.go b/db/companies.go index e360002..d2d5745 100644 --- a/db/companies.go +++ b/db/companies.go @@ -4,6 +4,7 @@ import ( "database/sql" "errors" "fmt" + "log" "net" "strings" "unicode/utf8" @@ -161,9 +162,11 @@ RETURNING id`, name, nkey, logoVal).Scan(&id); err != nil { return 0, 0, 0, invalid, validationErrors, err } if needsAttribution { - if err := recomputeAttributionTx(tx); err != nil { + warning, err := recomputeAttributionTx(tx) + if err != nil { return 0, 0, 0, invalid, validationErrors, err } + logAttributionWarning(warning) } if err := tx.Commit(); err != nil { return 0, 0, 0, invalid, validationErrors, err @@ -244,9 +247,13 @@ func (s *CompanyStore) DeleteCompanyWithAssets(id int64, deleteAssets bool) (ass if deleted == 0 { return 0, ErrCompanyNotFound } - if err := recomputeAttributionTx(tx); err != nil { + // This path has no per-request warning channel, so the log is the only place + // the operator can learn about unparseable ip rows here. + warning, err := recomputeAttributionTx(tx) + if err != nil { return 0, err } + logAttributionWarning(warning) if err := tx.Commit(); err != nil { return 0, err } @@ -370,9 +377,11 @@ func (s *CompanyStore) AddScopeInputsChecked(companyID int64, inputs []ScopeInpu return 0, 0, invalid, errors, err } if needsAttribution { - if err := recomputeAttributionTx(tx); err != nil { + warning, err := recomputeAttributionTx(tx) + if err != nil { return 0, 0, invalid, errors, fmt.Errorf("重新计算企业归属失败: %w", err) } + logAttributionWarning(warning) } if err := tx.Commit(); err != nil { return 0, 0, invalid, errors, err @@ -502,20 +511,26 @@ func (s *CompanyStore) RecomputeAttribution() error { if err := lockCompanyScopeMutation(tx); err != nil { return err } - if err := recomputeAttributionTx(tx); err != nil { + warning, err := recomputeAttributionTx(tx) + if err != nil { return err } + logAttributionWarning(warning) return tx.Commit() } -func recomputeAttributionTx(tx *sql.Tx) error { +// recomputeAttributionTx rebuilds scope-derived ownership. It returns a warning +// for assets whose ip column cannot be parsed: try_inet skips them instead of +// aborting the statement, so without this they would silently never receive a +// network-based company. Callers surface the warning and it is always logged. +func recomputeAttributionTx(tx *sql.Tx) (string, error) { // Only derived rows are cleared. Historical rows migrated without provenance // are marked explicit by schema.sql, which is the non-destructive default. if _, err := tx.Exec(` UPDATE assets SET company_id = NULL, company_source = 'scope' WHERE company_source = 'scope'`); err != nil { - return err + return "", err } // Domain-based attribution (root_domain exact match). @@ -533,7 +548,7 @@ UPDATE assets a SET company_id = matched.company_id, company_source = 'scope' FROM matched WHERE a.id = matched.asset_id`); err != nil { - return err + return "", err } // IP/CIDR attribution for still-unowned assets. @@ -541,7 +556,7 @@ WHERE a.id = matched.asset_id`); err != nil { WITH matched AS ( SELECT DISTINCT ON (a.id) a.id AS asset_id, cs.company_id FROM assets a - JOIN company_scope cs ON cs.kind IN ('ip','cidr') AND cs.net >>= a.ip::inet + JOIN company_scope cs ON cs.kind IN ('ip','cidr') AND cs.net >>= try_inet(a.ip) WHERE a.company_id IS NULL AND a.type IN ('ip','subdomain','service','endpoint') AND a.ip IS NOT NULL @@ -551,7 +566,7 @@ UPDATE assets a SET company_id = matched.company_id, company_source = 'scope' FROM matched WHERE a.id = matched.asset_id`); err != nil { - return err + return "", err } // Exact normalized ICP attribution after domain/network precedence. @@ -572,9 +587,78 @@ UPDATE assets a SET company_id = matched.company_id, company_source = 'scope' FROM matched WHERE a.id = matched.asset_id`); err != nil { - return err + return "", err } - return nil + return malformedIPAssetWarning(tx) +} + +// malformedIPAssetsSampled bounds how many offending ids one warning names, so a +// large batch of bad rows stays readable in a toast and in the log. +const malformedIPAssetsSampled = 5 + +// logAttributionWarning records a recompute warning in the server log. Every +// recompute path calls it, so the warning is reported even for triggers with no +// per-request response (company deletion, scopesentry sync, agent asset writes). +func logAttributionWarning(warning string) { + if warning != "" { + log.Printf("[assets] %s", warning) + } +} + +// malformedIPAssetQueryer is satisfied by both *sql.Tx and *DB so the warning +// can be produced inside a recompute transaction or read standalone by the API. +type malformedIPAssetQueryer interface { + Query(query string, args ...any) (*sql.Rows, error) +} + +// MalformedIPAssetWarning reports assets with an unparseable ip outside of any +// mutation, letting the API attach the warning to a scope response without +// widening the mutation signatures — an unrelated data problem is not one of +// this request's validation errors. +func (s *CompanyStore) MalformedIPAssetWarning() (string, error) { + return malformedIPAssetWarning(s.db) +} + +// malformedIPAssetWarning describes assets whose ip column is not a valid +// address. They are invisible to network attribution, so the operator has to be +// told which rows to fix — silently skipping them would look like scope rules +// that simply do not work. +func malformedIPAssetWarning(q malformedIPAssetQueryer) (string, error) { + rows, err := q.Query(` +SELECT id, ip, count(*) OVER () AS total +FROM assets +WHERE ip IS NOT NULL AND ip <> '' AND try_inet(ip) IS NULL + AND type IN ('ip','subdomain','service','endpoint') +ORDER BY id +LIMIT $1`, malformedIPAssetsSampled) + if err != nil { + return "", err + } + defer rows.Close() + var total int + samples := make([]string, 0, malformedIPAssetsSampled) + for rows.Next() { + var id int64 + var ip string + if err := rows.Scan(&id, &ip, &total); err != nil { + return "", err + } + samples = append(samples, fmt.Sprintf("#%d %s", id, ip)) + } + if err := rows.Err(); err != nil { + return "", err + } + if total == 0 { + return "", nil + } + warning := fmt.Sprintf( + "%d 条资产的 ip 字段不是合法 IP，已跳过 IP/CIDR 范围匹配（这些资产不会被网段规则归属到企业）：%s", + total, strings.Join(samples, "、"), + ) + if total > len(samples) { + warning += fmt.Sprintf(" 等 %d 条", total) + } + return warning, nil } // UpdateScope replaces all scope rules for a company and reattributes. @@ -635,9 +719,11 @@ func (s *CompanyStore) UpdateScopeInputsChecked(companyID int64, inputs []ScopeI } // Rebuild even for an empty replacement because removing the old rules may // detach scope-derived assets or expose a lower-precedence company match. - if err := recomputeAttributionTx(tx); err != nil { + warning, err := recomputeAttributionTx(tx) + if err != nil { return 0, invalid, errs, fmt.Errorf("重新计算企业归属失败: %w", err) } + logAttributionWarning(warning) if err := tx.Commit(); err != nil { return 0, invalid, errs, err } diff --git a/db/exploration.go b/db/exploration.go index 8747fa6..3a8c563 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -857,6 +857,16 @@ WHERE exploration_id=$1 AND kind='intent' AND state IN ('open','running'))`, s.e return exists, err } +// HasOpenGoal reports whether this exploration still has any goal in state 'open'. +// false ⇒ 所有目标已 met/abandoned（或本任务无目标）⇒ 进入 goalless（人工直投）分支： +// planner 停跑，任务是否结束改由 frontier 是否抽干决定。met 与 abandoned 都算"已了结"。 +func (s *ExplorationStore) HasOpenGoal() (bool, error) { + var exists bool + err := s.db.QueryRow(`SELECT EXISTS(SELECT 1 FROM exploration_nodes +WHERE exploration_id=$1 AND kind='goal' AND state='open')`, s.expID).Scan(&exists) + return exists, err +} + // ClaimIntent atomically moves an open intent to running. Returns true if claimed. func (s *ExplorationStore) ClaimIntent(id int64, owner string) (bool, error) { res, err := s.db.Exec(`UPDATE exploration_nodes SET state='running', owner=$1 diff --git a/db/schema.sql b/db/schema.sql index 6738ea7..28e4d0f 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -8,6 +8,18 @@ CREATE OR REPLACE FUNCTION set_updated_at() RETURNS trigger AS $$ BEGIN NEW.updated_at = now(); RETURN NEW; END; $$ LANGUAGE plpgsql; +-- 安全的 text→inet 转换：非法值返回 NULL 而不是抛 22P02。assets.ip 是自由文本 +-- (Agent / 资产 API 可能写进主机名)，裸转 a.ip::inet 会让单独一行脏数据把整条 +-- 企业归属重算语句打挂。调用方用 try_inet(...) IS NULL 找出这些行并告警。 +-- 不用 pg_input_is_valid 是因为那要 PG16+，这里要兼容更老的存量库。 +CREATE OR REPLACE FUNCTION try_inet(value text) RETURNS inet AS $$ +BEGIN + RETURN value::inet; +EXCEPTION WHEN others THEN + RETURN NULL; +END; +$$ LANGUAGE plpgsql IMMUTABLE STRICT; + -- ===================================================================== -- A. 资产层：companies / assets / company_scope -- ===================================================================== diff --git a/db/task_assets.go b/db/task_assets.go index db87fef..cd8823c 100644 --- a/db/task_assets.go +++ b/db/task_assets.go @@ -288,6 +288,10 @@ FROM assets WHERE id=ANY($2::bigint[])`, taskID, assetIDs).Scan(&found, &existin if found != len(assetIDs) { return mutation, ErrTaskAssetAssetNotFound } + // Order matters: this UPDATE fires trg_assets_task_links, which creates the + // link rows with the generic source='system'. The INSERT below must stay + // after it so the operator-authored 'manual' provenance wins; swapping the + // two statements silently degrades every manual attach back to 'system'. if _, err := tx.Exec(` UPDATE assets SET task_ids=CASE WHEN $1=ANY(task_ids) THEN task_ids ELSE array_append(task_ids,$1) END diff --git a/db/task_assets_context.go b/db/task_assets_context.go index b14c79a..9d2c066 100644 --- a/db/task_assets_context.go +++ b/db/task_assets_context.go @@ -29,12 +29,12 @@ target AS ( JOIN task_scope ts ON ( (ts.kind='company' AND a.company_id = ts.company_id) OR (ts.kind='root_domain' AND a.root_domain = ts.domain) - OR (ts.kind='subdomain' AND a.domain = ts.domain) - OR (ts.kind IN ('ip','cidr') AND a.ip IS NOT NULL AND a.ip !~ '[^0-9a-fA-F:.]' AND ts.net >>= a.ip::inet) - OR (ts.kind='icp' AND ( - lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value - OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value - )) + OR (ts.kind='subdomain' AND a.domain = ts.domain) + OR (ts.kind IN ('ip','cidr') AND ts.net >>= try_inet(a.ip)) + OR (ts.kind='icp' AND ( + lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value + OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value + )) ) JOIN context_tasks ctx ON ctx.task_id=ts.task_id UNION @@ -179,12 +179,12 @@ context_assets AS ( JOIN task_scope ts ON ( (ts.kind='company' AND a.company_id=ts.company_id) OR (ts.kind='root_domain' AND a.root_domain=ts.domain) - OR (ts.kind='subdomain' AND a.domain=ts.domain) - OR (ts.kind IN ('ip','cidr') AND a.ip IS NOT NULL AND a.ip !~ '[^0-9a-fA-F:.]' AND ts.net >>= a.ip::inet) - OR (ts.kind='icp' AND ( - lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value - OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value - )) + OR (ts.kind='subdomain' AND a.domain=ts.domain) + OR (ts.kind IN ('ip','cidr') AND ts.net >>= try_inet(a.ip)) + OR (ts.kind='icp' AND ( + lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value + OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value + )) ) JOIN context_tasks ctx ON ctx.task_id=ts.task_id ) diff --git a/db/task_categories.go b/db/task_categories.go index 5a4b30a..e06185c 100644 --- a/db/task_categories.go +++ b/db/task_categories.go @@ -13,6 +13,10 @@ import ( const MaxTaskCategoryNameRunes = 80 +// MaxTaskCategoryBatchSize bounds one batch move so a single request cannot lock +// an unbounded number of task rows. +const MaxTaskCategoryBatchSize = 100 + var ( ErrTaskCategoryInvalid = errors.New("invalid task category") ErrTaskCategoryNameConflict = errors.New("task category name already exists") @@ -199,3 +203,80 @@ FROM selected, updated`, taskID, *categoryID)) } return &category, nil } + +// SetTasksCategory moves several tasks into one category (nil = uncategorized) +// inside a single transaction, so a half-applied batch is never observable. +// It returns the ids that were actually updated — ids missing from that slice +// were deleted between selection and submit — plus the refreshed category row +// whose task_count already reflects this move. +func (d *DB) SetTasksCategory(taskIDs []int64, categoryID *int64) ([]int64, *TaskCategory, error) { + if len(taskIDs) == 0 { + return nil, nil, fmt.Errorf("%w: task ids are required", ErrTaskCategoryInvalid) + } + if len(taskIDs) > MaxTaskCategoryBatchSize { + return nil, nil, fmt.Errorf("%w: at most %d tasks per request", ErrTaskCategoryInvalid, MaxTaskCategoryBatchSize) + } + for _, id := range taskIDs { + if id <= 0 { + return nil, nil, fmt.Errorf("%w: task id must be positive", ErrTaskCategoryInvalid) + } + } + if categoryID != nil && *categoryID <= 0 { + return nil, nil, fmt.Errorf("%w: category id must be positive", ErrTaskCategoryInvalid) + } + tx, err := d.Begin() + if err != nil { + return nil, nil, err + } + defer tx.Rollback() //nolint:errcheck + + // Checking the category inside the transaction keeps a concurrent delete from + // turning the UPDATE below into a foreign key violation. + if categoryID != nil { + var exists bool + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM task_categories WHERE id=$1)`, *categoryID).Scan(&exists); err != nil { + return nil, nil, err + } + if !exists { + return nil, nil, ErrTaskCategoryNotFound + } + } + rows, err := tx.Query(` +UPDATE tasks SET category_id=$2 +WHERE id=ANY($1::bigint[]) AND deleted_at IS NULL +RETURNING id`, taskIDs, categoryID) + if err != nil { + return nil, nil, err + } + updated := make([]int64, 0, len(taskIDs)) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, nil, err + } + updated = append(updated, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, nil, err + } + + var category *TaskCategory + if categoryID != nil { + fetched, err := scanTaskCategory(tx.QueryRow(` +SELECT `+taskCategoryCols+` +FROM task_categories category +LEFT JOIN tasks task ON task.category_id=category.id +WHERE category.id=$1 +GROUP BY category.id`, *categoryID)) + if err != nil { + return nil, nil, err + } + category = &fetched + } + if err := tx.Commit(); err != nil { + return nil, nil, err + } + return updated, category, nil +} diff --git a/db/task_categories_test.go b/db/task_categories_test.go index e1d5d60..e409a69 100644 --- a/db/task_categories_test.go +++ b/db/task_categories_test.go @@ -89,3 +89,95 @@ func TestCreateTaskRejectsMissingCategoryAtomically(t *testing.T) { t.Fatalf("failed create leaked %d exploration rows", count) } } + +func TestSetTasksCategoryBatch(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) - skipping", err) + } + defer d.Close() + + stamp := time.Now().UnixNano() + category, err := d.CreateTaskCategory(fmt.Sprintf("Batch %d", stamp)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = d.DeleteTaskCategory(category.ID) }) + + ids := make([]int64, 0, 3) + for i := range 3 { + task, err := d.CreateTaskWithOptions(fmt.Sprintf("batch-move-%d-%d", stamp, i), "goal", TaskCreateOptions{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(task.ID) }) + ids = append(ids, task.ID) + } + + // A deleted id must not abort the move for the surviving tasks; it is simply + // absent from the returned slice so the caller can report it. + missing := int64(1 << 62) + updated, moved, err := d.SetTasksCategory(append(append([]int64{}, ids...), missing), &category.ID) + if err != nil { + t.Fatal(err) + } + if len(updated) != len(ids) { + t.Fatalf("updated=%v, want the %d live ids only", updated, len(ids)) + } + if moved == nil || moved.TaskCount != len(ids) { + t.Fatalf("category=%+v, want task_count=%d", moved, len(ids)) + } + for _, id := range ids { + loaded, err := d.GetTask(id) + if err != nil || loaded.CategoryID == nil || *loaded.CategoryID != category.ID { + t.Fatalf("task %d not moved: %+v err=%v", id, loaded, err) + } + } + + // A nil category clears the assignment for the whole batch. + updated, moved, err = d.SetTasksCategory(ids, nil) + if err != nil || len(updated) != len(ids) || moved != nil { + t.Fatalf("clear updated=%v category=%+v err=%v", updated, moved, err) + } + for _, id := range ids { + loaded, _ := d.GetTask(id) + if loaded.CategoryID != nil || loaded.CategoryName != "" { + t.Fatalf("task %d still categorized: %+v", id, loaded) + } + } +} + +func TestSetTasksCategoryRejectsBadInputAtomically(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) - skipping", err) + } + defer d.Close() + + stamp := time.Now().UnixNano() + task, err := d.CreateTaskWithOptions(fmt.Sprintf("batch-reject-%d", stamp), "goal", TaskCreateOptions{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(task.ID) }) + + missingCategory := int64(1 << 62) + if _, _, err := d.SetTasksCategory([]int64{task.ID}, &missingCategory); !errors.Is(err, ErrTaskCategoryNotFound) { + t.Fatalf("missing category error=%v, want %v", err, ErrTaskCategoryNotFound) + } + if _, _, err := d.SetTasksCategory(nil, nil); !errors.Is(err, ErrTaskCategoryInvalid) { + t.Fatalf("empty ids error=%v, want %v", err, ErrTaskCategoryInvalid) + } + oversized := make([]int64, MaxTaskCategoryBatchSize+1) + for i := range oversized { + oversized[i] = task.ID + } + if _, _, err := d.SetTasksCategory(oversized, nil); !errors.Is(err, ErrTaskCategoryInvalid) { + t.Fatalf("oversized error=%v, want %v", err, ErrTaskCategoryInvalid) + } + // The rejected calls must leave the task untouched. + loaded, _ := d.GetTask(task.ID) + if loaded.CategoryID != nil { + t.Fatalf("rejected batch mutated task: %+v", loaded) + } +} diff --git a/db/task_scope.go b/db/task_scope.go index f584b5d..e93f79d 100644 --- a/db/task_scope.go +++ b/db/task_scope.go @@ -308,13 +308,13 @@ target AS ( JOIN task_scope ts ON ts.task_id = $1 AND ( (ts.kind='company' AND a.company_id = ts.company_id) OR (ts.kind='root_domain' AND a.root_domain = ts.domain) - OR (ts.kind='subdomain' AND a.domain = ts.domain) - OR (ts.kind IN ('ip','cidr') AND a.ip IS NOT NULL AND a.ip !~ '[^0-9a-fA-F:.]' AND ts.net >>= a.ip::inet) - OR (ts.kind='icp' AND ( - lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value - OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value - )) - ) + OR (ts.kind='subdomain' AND a.domain = ts.domain) + OR (ts.kind IN ('ip','cidr') AND ts.net >>= try_inet(a.ip)) + OR (ts.kind='icp' AND ( + lower(regexp_replace(COALESCE(a.icp,''), '[[:space:]]+', '', 'g')) = ts.value + OR lower(regexp_replace(COALESCE(a.app_icp,''), '[[:space:]]+', '', 'g')) = ts.value + )) + ) ), tested AS ( SELECT DISTINCT ea.asset_id diff --git a/db/tasks.go b/db/tasks.go index bc0a57b..2e0d1a4 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -322,7 +322,9 @@ func (d *DB) SetParentRef(id int64, parentRef string) error { // ListTasks returns alive tasks, newest first. func (d *DB) ListTasks() ([]*Task, error) { - rows, err := d.Query(`SELECT ` + taskCols + ` FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC`) + // 按 id 降序:id 是 BIGSERIAL,单调递增即插入顺序,最新的在最前。不用 created_at + // 是因为同一时刻创建的多个任务时间戳相同,排序不稳定,列表会频繁换位(见 List)。 + rows, err := d.Query(`SELECT ` + taskCols + ` FROM tasks WHERE deleted_at IS NULL ORDER BY id DESC`) if err != nil { return nil, err } diff --git a/db/tasks_test.go b/db/tasks_test.go index 0f622e3..ef5dc26 100644 --- a/db/tasks_test.go +++ b/db/tasks_test.go @@ -563,3 +563,45 @@ func TestCreateTaskWithCompanyScopes(t *testing.T) { t.Fatalf("failed company association leaked %d exploration rows", leaked) } } + +// TestListTasksOrderByIDDesc pins list ordering to id-descending. created_at is +// deliberately not the sort key: tasks created in the same instant share a +// timestamp and would reorder between polls; id is unique and monotonic. +func TestListTasksOrderByIDDesc(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + stamp := time.Now().UnixNano() + var ids []int64 + for i := range 3 { + tk, err := d.CreateTask(fmt.Sprintf("order-%d-%d", stamp, i), "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(tk.ID) }) + ids = append(ids, tk.ID) + } + + list, err := d.ListTasks() + if err != nil { + t.Fatal(err) + } + // Reduce to just the ids created here; other rows may exist in the shared DB. + mine := map[int64]bool{ids[0]: true, ids[1]: true, ids[2]: true} + var seen []int64 + for _, task := range list { + if mine[task.ID] { + seen = append(seen, task.ID) + } + } + if len(seen) != 3 { + t.Fatalf("found %d of the created tasks in the list, want 3", len(seen)) + } + // Newest (largest id) first. + if seen[0] != ids[2] || seen[1] != ids[1] || seen[2] != ids[0] { + t.Fatalf("order=%v, want descending %v", seen, []int64{ids[2], ids[1], ids[0]}) + } +} diff --git a/server/assets.go b/server/assets.go index dc854b2..17d2b5a 100644 --- a/server/assets.go +++ b/server/assets.go @@ -230,6 +230,13 @@ func (s *Server) addCompanyScope(w http.ResponseWriter, r *http.Request) { if len(errs) > 0 { out["errors"] = errs } + // Reported separately from errors: this is pre-existing bad data, not a fault + // in the submitted rules, but the operator still has to see it or those assets + // look like the IP/CIDR rules simply never match. A failure to build the + // warning must not fail the scope write that already committed. + if warning, err := cs.MalformedIPAssetWarning(); err == nil && warning != "" { + out["warnings"] = []string{warning} + } writeJSON(w, 200, out) } diff --git a/server/chat_llm_resolve_test.go b/server/chat_llm_resolve_test.go new file mode 100644 index 0000000..5108988 --- /dev/null +++ b/server/chat_llm_resolve_test.go @@ -0,0 +1,97 @@ +package server + +import ( + "context" + "strings" + "testing" + + "github.com/Autumn-27/artex/db" +) + +// TestChatUnavailableReasonDistinguishesStates pins the operator-facing message +// to the actual configuration state: nothing configured vs. configured-but-not- +// activated must read differently so the user knows what to do. +func TestChatUnavailableReasonDistinguishesStates(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer m.Close() + td := t.TempDir() + s := New(context.Background(), m, td, td, td) + + // Start from a clean profile table; other tests in the shared DB may have left rows. + existing, _ := m.pg.ListProfiles() + for _, p := range existing { + if p.IsDefault { + // The active profile cannot be deleted directly; clear the flag first. + _, _ = m.pg.Exec(`UPDATE llm_profiles SET is_default=false WHERE id=$1`, p.ID) + } + _ = m.pg.DeleteProfile(p.ID) + } + + if reason := s.chatUnavailableReason(); !strings.Contains(reason, "尚未配置") { + t.Fatalf("no-profile reason=%q, want 尚未配置", reason) + } + + id, err := m.pg.SaveProfile(&db.LLMProfile{Name: "p1", Format: "anthropic", Model: "claude-x", APIKey: "sk-test"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, _ = m.pg.Exec(`UPDATE llm_profiles SET is_default=false WHERE id=$1`, id) + _ = m.pg.DeleteProfile(id) + }) + + // Profile exists but is not activated. + if reason := s.chatUnavailableReason(); !strings.Contains(reason, "没有已激活") { + t.Fatalf("inactive reason=%q, want 没有已激活", reason) + } + + // Once activated, the reason no longer claims a missing/inactive config. + if err := m.pg.SetActiveProfile(id); err != nil { + t.Fatal(err) + } + if reason := s.chatUnavailableReason(); strings.Contains(reason, "尚未配置") || strings.Contains(reason, "没有已激活") { + t.Fatalf("active reason=%q should not report missing/inactive", reason) + } +} + +// TestResolveChatAgentHonoursConversationProfile is the regression guard for the +// reported bug: a conversation that picked a valid profile must resolve a chat +// agent even when NO global profile is active, so the send precheck stops +// rejecting it with "LLM 未配置". +func TestResolveChatAgentHonoursConversationProfile(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer m.Close() + td := t.TempDir() + s := New(context.Background(), m, td, td, td) + + // Force the global fallback to nil so a non-nil result can ONLY come from the + // conversation's own profile — this is exactly the situation the user hit + // (no usable global config) and isolates the fix from the test host's env. + s.cfgMu.Lock() + s.chatAgent = nil + s.cfgMu.Unlock() + + id, err := m.pg.SaveProfile(&db.LLMProfile{Name: "conv-pick", Format: "anthropic", Model: "claude-x", APIKey: "sk-test"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = m.pg.DeleteProfile(id) }) + + // A conversation that pinned this profile resolves a chat agent despite the + // nil global fallback; without the fix the send precheck rejected it outright. + conv := &db.Conversation{AgentKey: "mainagent", LLMProfileID: &id} + if s.resolveChatAgent(conv) == nil { + t.Fatalf("resolveChatAgent(conversation with valid profile) = nil; global fallback wrongly gates the pick") + } + + // With neither a pick nor a global fallback there is genuinely nothing to run. + if s.resolveChatAgent(&db.Conversation{AgentKey: "mainagent"}) != nil { + t.Fatalf("resolveChatAgent with neither pick nor global config should be nil") + } +} diff --git a/server/conversations.go b/server/conversations.go index de81e4c..e11b78c 100644 --- a/server/conversations.go +++ b/server/conversations.go @@ -315,9 +315,12 @@ func (s *Server) pgSendConversationMessage(w http.ResponseWriter, r *http.Reques writeErr(w, 400, "消息不能为空") return } - ca := s.chatAgentRef() - if ca == nil { - writeErr(w, 503, "LLM 未配置，无法对话") + // Same resolution as the background runner: honour the agent binding / this + // conversation's chosen profile before falling back to the global config, so a + // conversation that picked a valid LLM is not rejected just because no global + // config is active. + if s.resolveChatAgent(c) == nil { + writeErr(w, 503, s.chatUnavailableReason()) return } @@ -390,12 +393,7 @@ func (s *Server) runConversationSync(c *db.Conversation, msg, busyKey string) { s.chatMu.Unlock() }() // precedence: the conversation agent's own binding → this conversation's pin → global. - ca := s.chatAgentRef() - if eff := s.effectiveProfileForAgent(c.AgentKey, c.LLMProfileID); eff != nil { - if pa := s.chatAgentForProfile(*eff); pa != nil { - ca = pa - } - } + ca := s.resolveChatAgent(c) if ca == nil { return } diff --git a/server/engine.go b/server/engine.go index 551aa0a..f3c7074 100644 --- a/server/engine.go +++ b/server/engine.go @@ -792,6 +792,24 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) { if e.isSettling(t.ID) { return } + // goalless（人工直投）分支：任务已无 open 目标时 planner 不跑——跑了会重判 + // met→cancelExec 杀掉用户经主 agent 直投的意图。是否结束改由 frontier 决定： + // 还有 open/running 意图 → 保持 running、静默等待；意图已全部跑干 → 落 done。 + // 整段纯 Go、不触发任何 LLM 调用，也不打规划轮 marker。 + if open, err := t.Store.HasOpenGoal(); err == nil && !open { + t.drainTriggers() // 丢弃累积的 done/finding 触发，避免 goalless 长会话里无界增长 + if active, err := t.Store.HasActiveIntent(); err == nil && !active { + // frontier 抽干且无在跑意图 → 收尾。用 Guarded 版做 CAS，避免踩到并发的 + // pause/delete/超时收尾的状态转换。 + if won, err := e.m.SetTaskStatusGuarded(t.ID, "done"); err != nil { + log.Printf("[goalless] task %s 收尾落 done 失败: %v", t.ID, err) + } else if won { + e.emitActivity(t, db.Activity{Worker: "system", Kind: "text", + Summary: "目标已全部达成，直投意图已执行完毕，任务结束"}) + } + } + return // goalless 分支永不进入 planner.Plan + } if !e.beginTaskOperation(t.ID) { return } diff --git a/server/manager.go b/server/manager.go index 46cf066..b6ba14a 100644 --- a/server/manager.go +++ b/server/manager.go @@ -812,6 +812,49 @@ func (m *Manager) SetTaskCategory(taskID string, categoryID *int64) (*pgdb.TaskC return category, nil } +// SetTasksCategory applies one category change to several tasks at once. The +// database write and the in-memory refresh share taskStateMu, so a concurrent +// single-task update cannot interleave and leave a live DTO stale. The returned +// set holds the ids that were actually moved; callers report the rest as gone. +func (m *Manager) SetTasksCategory(taskIDs []string, categoryID *int64) (map[string]bool, *pgdb.TaskCategory, error) { + m.taskStateMu.Lock() + defer m.taskStateMu.Unlock() + numeric := make([]int64, 0, len(taskIDs)) + for _, taskID := range taskIDs { + id, err := strconv.ParseInt(taskID, 10, 64) + if err != nil || id <= 0 { + return nil, nil, pgdb.ErrTaskCategoryTaskNotFound + } + numeric = append(numeric, id) + } + updatedIDs, category, err := m.pg.SetTasksCategory(numeric, categoryID) + if err != nil { + return nil, nil, err + } + categoryName := "" + if category != nil { + categoryName = category.Name + } + updated := make(map[string]bool, len(updatedIDs)) + m.mu.RLock() + tasks := make([]*Task, 0, len(updatedIDs)) + for _, id := range updatedIDs { + taskID := strconv.FormatInt(id, 10) + updated[taskID] = true + if task := m.tasks[taskID]; task != nil { + tasks = append(tasks, task) + } + } + m.mu.RUnlock() + for _, task := range tasks { + task.updateLifecycle(func(state *taskLifecycleState) { + state.CategoryID = cloneInt64Ptr(categoryID) + state.CategoryName = categoryName + }) + } + return updated, category, nil +} + // DeleteCompanyWithAssets keeps the database cascade and live task handles in // one manager-level critical section. This closes the gap where a task could // commit its company scope immediately before registration and miss the @@ -1515,7 +1558,14 @@ func (m *Manager) List() []*Task { for _, t := range m.tasks { out = append(out, t) } - sort.Slice(out, func(i, j int) bool { return out[i].CreatedAt > out[j].CreatedAt }) + // 按 id 降序(最新插入在最前)。m.tasks 是 map,遍历顺序随机,每次轮询都要重排; + // 若按 CreatedAt 排,同一秒创建的任务时间戳相同,排序不稳定,会在列表里频繁换 + // 位置。id 由 BIGSERIAL 单调递增分配,唯一且稳定,顺序恒定。 + sort.Slice(out, func(i, j int) bool { + ai, _ := strconv.ParseInt(out[i].ID, 10, 64) + aj, _ := strconv.ParseInt(out[j].ID, 10, 64) + return ai > aj + }) return out } diff --git a/server/orchestration.go b/server/orchestration.go index 3d4b104..9057ea7 100644 --- a/server/orchestration.go +++ b/server/orchestration.go @@ -345,11 +345,11 @@ func (s *Server) toolAddHint() actool.CoreTool { func (s *Server) toolGetWorkerTrace() actool.CoreTool { return roTool("get_task_worker_trace", - "看指定任务里某个 work(意图)的执行过程：get_task_worker_trace(task_id, intent_id) 看步骤摘要；再带 step_ids=[...] 取那几步完整内容(一次≤5)。", + "看指定任务里某个 work(意图)的执行过程：get_task_worker_trace(task_id, intent_id) 看步骤摘要；再带 step_ids=[...] 取那几步完整内容(一次最多 5 个,多传只返回前 5 个)。", objSchema(map[string]any{ "task_id": strParam("任务 id"), "intent_id": map[string]any{"type": "integer", "description": "意图 id(该任务里的 work)"}, - "step_ids": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}, "description": "可选：要取完整内容的步骤 id(≤5)"}, + "step_ids": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}, "description": "可选：要取完整内容的步骤 id(一次最多 5 个,多传只返回前 5 个,其余在 omitted_step_ids 里列出)"}, }, "task_id", "intent_id"), func(ctx context.Context, in json.RawMessage) (actool.Result, error) { return s.delegateToTask(ctx, in, (*agent.ToolSet).GetWorkerTraceTool) @@ -451,6 +451,7 @@ func (s *Server) seedOrchestrationTools() { s.seedAutoReportFindingBinding() s.unbindGoalMetDefault() s.reseedGoalsPrompt() // goals 提示词加入「抽操作约束」步 → 旧库追加一版新默认(一次性) + s.reseedMainAgentPrompt() // mainagent 提示词加入「目标达成后 add_intent 反问是否建目标」(一次性) s.seedReporterAgent() // 预置「报告撰写」agent + 工具绑定 + finding 触发器(一次性) // 注：pentest 的默认工具绑定无需迁移——BuiltinToolSeeds 在全新初始化时就把 // list_assets/insert_assets/report_finding/list_findings/list_companies 连同 @@ -540,6 +541,36 @@ func (s *Server) reseedGoalsPrompt() { log.Printf("[prompts] goals 提示词已追加新默认版本(加入抽操作约束步,一次性)") } +// reseedMainAgentPrompt 把 mainagent 提示词刷成【当前代码默认】——默认正文新增了「目标全部 +// 达成后 add_intent 直投意图时,反问人是否登记为正式目标」这段引导,而 SeedPromptIfEmpty 首插入 +// only,旧库已有版本收不到。用版本管理【追加一个新版本】并切过去(ResetPromptToDefault),旧版本仍 +// 保留在历史里,用户若自定义过可从版本记录找回。settings flag 守卫 → 只做一次。全新库无需处理 +// (SeedPromptIfEmpty 已 seed 最新默认)。与 reseedGoalsPrompt 完全同构。 +func (s *Server) reseedMainAgentPrompt() { + const flag = "mainagent_prompt_goalless_intent_v1" + if v, _, _ := s.m.pg.GetSetting(flag); v == "true" { + return + } + defer func() { _ = s.m.pg.SetSetting(flag, "true") }() // 无论成功与否只尝试一次 + a, err := s.m.pg.GetAgentByKey("mainagent") + if err != nil || a == nil { + return // 全新库尚未建 agent 行时,seedPrompts 会直接 seed 最新默认,无需此迁移 + } + tmpl := agent.BuiltinPromptSeeds()["mainagent"] + if tmpl == "" { + return + } + // 全新库 seedPrompts 已 seed 最新默认 → 当前版本已等于代码默认,不必再追加重复版本。 + if cur, err := s.m.pg.CurrentPrompt(a.ID); err == nil && cur == tmpl { + return + } + if _, err := s.m.pg.ResetPromptToDefault(a.ID, tmpl); err != nil { + log.Printf("[prompts] mainagent 提示词重刷为新默认失败: %v", err) + return + } + log.Printf("[prompts] mainagent 提示词已追加新默认版本(加入目标达成后反问建目标,一次性)") +} + // seedReporterAgent 预置一个「报告撰写」自定义 agent(builtin=false，可在 UI 编辑/删除)： // 绑定 update_finding_report + 任务查询工具，并挂一个「report_finding 被调用即触发」的 // 触发器 —— 每登记一个漏洞就唤起它写详细报告。一次性(settings flag 守卫)：用户删掉后不再重建。 diff --git a/server/server.go b/server/server.go index 219ab50..0e5ffb7 100644 --- a/server/server.go +++ b/server/server.go @@ -492,6 +492,37 @@ func (s *Server) effectiveProfileForAgent(agentKey string, pinID *int64) *int64 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 "尚未配置 LLM：请到 系统 → LLM 配置 添加一个配置" + } + if active, err := s.m.pg.ActiveProfile(); err == nil && active == nil { + return "没有已激活的 LLM 配置：请到 系统 → LLM 配置 激活一个，或在本对话为该会话指定一个配置" + } + } + return "LLM 未就绪，无法对话：请检查 系统 → LLM 配置是否有可用且已激活的配置" +} + // 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. @@ -650,6 +681,7 @@ func (s *Server) Handler() http.Handler { 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) diff --git a/server/task_categories.go b/server/task_categories.go index 9cca440..e176312 100644 --- a/server/task_categories.go +++ b/server/task_categories.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/json" "errors" + "fmt" "net/http" "strings" "unicode/utf8" @@ -119,6 +120,26 @@ func (s *Server) pgDeleteTaskCategory(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"deleted": id}) } +// parseCategoryIDField reads the category_id field shared by the single-task and +// batch endpoints. The field is required but may be null — that clears the +// category — so it arrives as RawMessage instead of *int64, which cannot tell +// an explicit null from an omitted key. +func parseCategoryIDField(w http.ResponseWriter, raw json.RawMessage) (*int64, bool) { + if len(raw) == 0 { + writeErr(w, http.StatusBadRequest, "category_id is required") + return nil, false + } + if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) { + return nil, true + } + var id int64 + if err := json.Unmarshal(raw, &id); err != nil || id <= 0 { + writeErr(w, http.StatusBadRequest, "任务分类 id 无效") + return nil, false + } + return &id, true +} + func (s *Server) updateTaskCategory(w http.ResponseWriter, r *http.Request) { taskID := r.PathValue("id") if _, ok := s.m.Task(taskID); !ok { @@ -133,21 +154,8 @@ func (s *Server) updateTaskCategory(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusBadRequest, err.Error()) return } - if len(request.CategoryID) == 0 { - writeErr(w, http.StatusBadRequest, "category_id is required") - return - } - var categoryID *int64 - if !bytes.Equal(bytes.TrimSpace(request.CategoryID), []byte("null")) { - var id int64 - if err := json.Unmarshal(request.CategoryID, &id); err != nil { - writeErr(w, http.StatusBadRequest, "任务分类 id 无效") - return - } - categoryID = &id - } - if categoryID != nil && *categoryID <= 0 { - writeErr(w, http.StatusBadRequest, "任务分类 id 无效") + categoryID, ok := parseCategoryIDField(w, request.CategoryID) + if !ok { return } if _, err := s.m.SetTaskCategory(taskID, categoryID); err != nil { @@ -157,3 +165,77 @@ func (s *Server) updateTaskCategory(w http.ResponseWriter, r *http.Request) { task, _ := s.m.Task(taskID) writeJSON(w, http.StatusOK, taskDTO(task, s.resolvedTaskStatus(task))) } + +// batchCategoryItem reports one requested task. Unlike batch pause/resume there +// is no per-task runtime outcome, so only the failure reason is carried. +type batchCategoryItem struct { + ID string `json:"id"` + OK bool `json:"ok"` + Error string `json:"error,omitempty"` +} + +// updateTasksCategoryBatch moves every selected task into one category, or +// clears it when category_id is null. The database write is atomic; per-task +// entries only distinguish ids that never resolved to a live task, so a partial +// success here means those tasks were deleted, not that the move half-applied. +func (s *Server) updateTasksCategoryBatch(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, maxTaskCategoryRequestBytes) + var request struct { + TaskIDs []string `json:"task_ids"` + CategoryID json.RawMessage `json:"category_id"` + } + if err := decode(r, &request); err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + writeErr(w, http.StatusRequestEntityTooLarge, "请求正文过大") + } else { + writeErr(w, http.StatusBadRequest, err.Error()) + } + return + } + categoryID, ok := parseCategoryIDField(w, request.CategoryID) + if !ok { + return + } + taskIDs := normalizeBatchTaskIDs(request.TaskIDs) + if len(taskIDs) == 0 || len(taskIDs) > db.MaxTaskCategoryBatchSize { + writeErr(w, http.StatusBadRequest, fmt.Sprintf("task_ids 数量必须为 1-%d", db.MaxTaskCategoryBatchSize)) + return + } + items := make([]batchCategoryItem, 0, len(taskIDs)) + pending := make([]string, 0, len(taskIDs)) + for _, parsed := range taskIDs { + item := batchCategoryItem{ID: parsed.id} + switch { + case !parsed.valid: + item.Error = "bad task id" + default: + if _, live := s.m.Task(parsed.id); live { + pending = append(pending, parsed.id) + } else { + item.Error = "task not found" + } + } + items = append(items, item) + } + var category *db.TaskCategory + if len(pending) > 0 { + updated, moved, err := s.m.SetTasksCategory(pending, categoryID) + if err != nil { + writeTaskCategoryError(w, err) + return + } + category = moved + for i := range items { + if items[i].Error != "" { + continue + } + if updated[items[i].ID] { + items[i].OK = true + } else { + items[i].Error = "task not found" + } + } + } + writeJSON(w, http.StatusOK, map[string]any{"items": items, "category": category}) +} diff --git a/server/task_categories_test.go b/server/task_categories_test.go new file mode 100644 index 0000000..64f721a --- /dev/null +++ b/server/task_categories_test.go @@ -0,0 +1,84 @@ +package server + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// TestTaskCategoryBatchRoute pins the contract of the batch move endpoint: it is +// registered ahead of the /api/tasks/{id}/... patterns, it validates the request +// before touching the database, and it reports unknown ids per task instead of +// failing the whole batch. +func TestTaskCategoryBatchRoute(t *testing.T) { + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Skipf("postgres unavailable (%v) - skipping", err) + } + defer m.Close() + + td := t.TempDir() + s := New(context.Background(), m, td, td, td) + h := s.Handler() + token, err := signJWT(s.jwtKey) + if err != nil { + t.Fatalf("sign jwt: %v", err) + } + + post := func(body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/api/tasks/category/batch", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec + } + + // category_id is required, and omitting it must not be read as "uncategorize". + if rec := post(`{"task_ids":["1"]}`); rec.Code != http.StatusBadRequest || + !strings.Contains(rec.Body.String(), "category_id is required") { + t.Fatalf("missing category_id: status=%d body=%s", rec.Code, rec.Body.String()) + } + + // An empty selection is rejected before any database work. + if rec := post(`{"task_ids":[],"category_id":null}`); rec.Code != http.StatusBadRequest || + !strings.Contains(rec.Body.String(), "task_ids 数量必须为") { + t.Fatalf("empty task_ids: status=%d body=%s", rec.Code, rec.Body.String()) + } + + // Unknown and malformed ids come back as per-task failures on a 200, so the + // caller can tell exactly which selections went stale. + rec := post(`{"task_ids":["999999999","abc","999999999"],"category_id":null}`) + if rec.Code != http.StatusOK { + t.Fatalf("batch status=%d body=%s", rec.Code, rec.Body.String()) + } + var response struct { + Items []struct { + ID string `json:"id"` + OK bool `json:"ok"` + Error string `json:"error"` + } `json:"items"` + Category *struct { + ID int64 `json:"id"` + } `json:"category"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatalf("decode: %v body=%s", err, rec.Body.String()) + } + // The duplicate id is collapsed, so two entries remain. + if len(response.Items) != 2 { + t.Fatalf("items=%+v, want 2 after de-duplication", response.Items) + } + if response.Items[0].ID != "999999999" || response.Items[0].OK || response.Items[0].Error != "task not found" { + t.Fatalf("unknown id entry=%+v", response.Items[0]) + } + if response.Items[1].ID != "abc" || response.Items[1].OK || response.Items[1].Error != "bad task id" { + t.Fatalf("malformed id entry=%+v", response.Items[1]) + } + if response.Category != nil { + t.Fatalf("category=%+v, want null when clearing", response.Category) + } +} diff --git a/server/task_control.go b/server/task_control.go index 4e61f57..3f09087 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -27,6 +27,40 @@ type intentControlResult struct { Deleted *db.IntentCleanup `json:"deleted,omitempty"` } +// parsedTaskID carries one requested batch id together with whether it parsed. +// Invalid ids are kept rather than dropped so the response can name them. +type parsedTaskID struct { + id string + valid bool +} + +// normalizeBatchTaskIDs trims, canonicalizes and de-duplicates the ids of one +// batch request while preserving the caller's order. Shared by every batch +// endpoint so they agree on what counts as a duplicate. +func normalizeBatchTaskIDs(raw []string) []parsedTaskID { + seen := map[string]bool{} + seenInvalid := map[string]bool{} + taskIDs := make([]parsedTaskID, 0, len(raw)) + for _, item := range raw { + trimmed := strings.TrimSpace(item) + id, valid := canonicalTaskID(trimmed) + if !valid { + if seenInvalid[trimmed] { + continue + } + seenInvalid[trimmed] = true + taskIDs = append(taskIDs, parsedTaskID{id: trimmed}) + continue + } + if seen[id] { + continue + } + seen[id] = true + taskIDs = append(taskIDs, parsedTaskID{id: id, valid: true}) + } + return taskIDs +} + type batchControlItem struct { ID string `json:"id"` OK bool `json:"ok"` @@ -192,30 +226,7 @@ func (s *Server) controlTasksBatch(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, "action must be pause|resume") return } - type parsedTaskID struct { - id string - valid bool - } - seen := map[string]bool{} - seenInvalid := map[string]bool{} - taskIDs := make([]parsedTaskID, 0, len(req.TaskIDs)) - for _, raw := range req.TaskIDs { - trimmed := strings.TrimSpace(raw) - id, valid := canonicalTaskID(trimmed) - if !valid { - if seenInvalid[trimmed] { - continue - } - seenInvalid[trimmed] = true - taskIDs = append(taskIDs, parsedTaskID{id: trimmed}) - continue - } - if seen[id] { - continue - } - seen[id] = true - taskIDs = append(taskIDs, parsedTaskID{id: id, valid: true}) - } + taskIDs := normalizeBatchTaskIDs(req.TaskIDs) if len(taskIDs) == 0 || len(taskIDs) > maxBatchControlIDs { writeErr(w, 400, fmt.Sprintf("task_ids 数量必须为 1-%d", maxBatchControlIDs)) return diff --git a/web/src/app/(main)/function/assets/page.tsx b/web/src/app/(main)/function/assets/page.tsx index b9b1b76..bcc4438 100644 --- a/web/src/app/(main)/function/assets/page.tsx +++ b/web/src/app/(main)/function/assets/page.tsx @@ -1321,6 +1321,14 @@ function CompanyAvatar({ name, logo }: { name: string; logo?: string }) { ); } +// 后端返回的 warnings 说的是既有数据问题（不是本次提交的行有错），保存本身已经 +// 成功。给更长的停留时间，因为它需要用户去处理具体的资产，扫一眼标题不够。 +function showScopeWarnings(warnings?: string[]) { + for (const warning of warnings ?? []) { + toast.warning(warning, { duration: 15000 }); + } +} + function savedScopeRules(company: Company): CompanyScopeRule[] { return (company.scope ?? []) .map((scope) => ({ kind: scope.kind, value: scope.raw.trim() })) @@ -1442,6 +1450,7 @@ function EditScopeDialog({ company, onSaved }: { company: Company; onSaved: () = const errCount = res.invalid ?? 0; if (errCount > 0) toast.warning(`已保存；${errCount} 行无效`); else toast.success(`范围已更新，共 ${res.added} 条`); + showScopeWarnings(res.warnings); setOpen(false); onSaved(); } catch (e) { @@ -1528,6 +1537,7 @@ function AppendScopeDialog({ company, onSaved }: { company: Company; onSaved: () const errCount = res.invalid ?? 0; if (errCount > 0) toast.warning(`已保存；${errCount} 行无效`); else toast.success(`已追加 ${res.added} 条范围`); + showScopeWarnings(res.warnings); setOpen(false); onSaved(); } catch (e) { diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx index e26497b..8952667 100644 --- a/web/src/app/(main)/function/tasks/page.tsx +++ b/web/src/app/(main)/function/tasks/page.tsx @@ -19,6 +19,7 @@ import { import { ChevronRightIcon, EyeIcon, + FolderInputIcon, GripVerticalIcon, Loader2Icon, PaperclipIcon, @@ -219,6 +220,10 @@ const STATUS_OPTIONS: { value: TaskStatus; label: string }[] = [ { value: "timeout", label: "已超时" }, ]; +// Select 不接受空字符串 value,所以「无分类」在筛选器、新建表单和批量移动里 +// 统一用这个哨兵值,提交时再翻译成后端的 null。 +const UNCATEGORIZED_VALUE = "uncategorized"; + // 可暂停 = 非终态且未暂停,与后端 applyTaskControlWithCause 的门控一致 // (done/failed/timeout 为终态);paused 才可继续。行内按钮与批量控制共用此判断, // 两边不会出现一个可点、另一个不可点的分歧。 @@ -240,13 +245,18 @@ export default function TasksPage() { const [pageSize, setPageSize] = React.useState(20); const [nowSec, setNowSec] = React.useState(() => Math.floor(Date.now() / 1000)); const [batchControlling, setBatchControlling] = React.useState<"pause" | "resume" | null>(null); + const [movingCategory, setMovingCategory] = React.useState(false); const filtered = React.useMemo(() => { const q = query.trim().toLowerCase(); return tasks.filter((t) => { if (statusFilter !== "all" && t.status !== statusFilter) return false; - if (categoryFilter === "uncategorized" && t.category_id != null) return false; - if (categoryFilter !== "all" && categoryFilter !== "uncategorized" && String(t.category_id) !== categoryFilter) + if (categoryFilter === UNCATEGORIZED_VALUE && t.category_id != null) return false; + if ( + categoryFilter !== "all" && + categoryFilter !== UNCATEGORIZED_VALUE && + String(t.category_id) !== categoryFilter + ) return false; if (!q) return true; return ( @@ -374,7 +384,7 @@ export default function TasksPage() { ); React.useEffect(() => { - if (categoryFilter === "all" || categoryFilter === "uncategorized") return; + if (categoryFilter === "all" || categoryFilter === UNCATEGORIZED_VALUE) return; if (!categories.some((category) => String(category.id) === categoryFilter)) setCategoryFilter("all"); }, [categories, categoryFilter]); @@ -535,6 +545,42 @@ export default function TasksPage() { [batchControlling, load], ); + // 后端把整批分类写入放在一个事务里，所以失败项只可能是勾选后又被删掉的任务。 + const moveSelectedTasksCategory = React.useCallback( + async (categoryID?: number) => { + const ids = [...selectedIds]; + if (ids.length === 0 || movingCategory) return; + if (ids.length > 100) { + toast.error("一次最多修改 100 个任务的分类"); + return; + } + setMovingCategory(true); + try { + const result = await api.updateTasksCategory(ids, categoryID); + const succeeded = result.items.filter((item) => item.ok); + const failed = result.items.filter((item) => !item.ok); + const target = result.category?.name ?? "未分类"; + if (succeeded.length > 0) { + toast.success(`已将 ${succeeded.length} 个任务移动到「${target}」`); + setSelectedIds(new Set()); + } + if (failed.length > 0) { + const details = failed + .slice(0, 3) + .map((item) => `#${item.id}（${item.error || "任务已不存在"}）`) + .join("；"); + toast.error(`${failed.length} 个任务未能移动：${details}${failed.length > 3 ? " 等" : ""}`); + } + refreshCategoriesAndTasks(); + } catch (error) { + toast.error(`修改分类失败：${(error as Error).message}`); + } finally { + setMovingCategory(false); + } + }, + [movingCategory, refreshCategoriesAndTasks, selectedIds], + ); + return ( @@ -580,7 +626,7 @@ export default function TasksPage() { 全部分类 - 未分类 + 未分类 {categories.map((category) => ( {category.name} @@ -628,6 +674,12 @@ export default function TasksPage() { 继续 {resumableTaskIDs.length} )} + )} @@ -638,7 +690,12 @@ export default function TasksPage() { onChanged={refreshCategoriesAndTasks} onTaskMoved={applyTaskCategoryMove} /> - + {tasks.length === 0 ? ( @@ -1147,6 +1204,81 @@ function DeleteTaskDialog({ // BulkDeleteTasksDialog deletes every checked task with one shared set of cleanup options. // The backend has no batch endpoint, so deletion runs one task at a time (see deleteTasks) // and the button shows live progress. +// MoveTasksCategoryDialog confirms a target before applying, so a mis-click on a +// large selection cannot silently re-file every task. UNCATEGORIZED_VALUE stands +// in for "no category" because Select rejects an empty string value. +function MoveTasksCategoryDialog({ + categories, + count, + moving, + onMove, +}: { + categories: TaskCategory[]; + count: number; + moving: boolean; + onMove: (categoryID?: number) => Promise; +}) { + const [open, setOpen] = React.useState(false); + const [target, setTarget] = React.useState(UNCATEGORIZED_VALUE); + + const handleOpenChange = (next: boolean) => { + if (moving) return; + setOpen(next); + if (next) setTarget(UNCATEGORIZED_VALUE); + }; + + return ( + + + + + + + 修改所选 {count} 个任务的分类 + 目标分类对所选任务统一生效；选「未分类」会把它们移出当前分类。 + + + 目标分类 + + {categories.length === 0 && 还没有任何分类，先用「分类管理」创建一个。} + + + + + + + + + + ); +} + function BulkDeleteTasksDialog({ ids, onDelete, @@ -1292,6 +1424,130 @@ function SourceTaskPicker({ ); } +// CategoryPicker 是新建任务里的单选分类选择器：可搜索已有分类；输入库里没有的名称后 +// 回车（或点下拉里的「创建」）即时新建分类并选中，选中项以可移除的 tag 展示。分类是 +// 全局资源，这里即时创建与「分类管理」里手动新建等价。只允许一个分类。 +function CategoryPicker({ + categories, + value, + onValueChange, + onCategoryCreated, + portalContainer, +}: { + categories: TaskCategory[]; + value?: number; + onValueChange: (categoryID?: number) => void; + onCategoryCreated: () => void; + portalContainer?: React.RefObject; +}) { + const [inputValue, setInputValue] = React.useState(""); + const [creating, setCreating] = React.useState(false); + // 新建的分类要等父层重新拉取才回流到 categories，先本地留一份，避免选中的 chip 和 + // 下拉在这段窗口里显示成「未知分类」。 + const [localExtra, setLocalExtra] = React.useState([]); + + const allCategories = React.useMemo(() => { + const byID = new Map(); + for (const category of categories) byID.set(category.id, category); + for (const category of localExtra) if (!byID.has(category.id)) byID.set(category.id, category); + return [...byID.values()]; + }, [categories, localExtra]); + + const byID = React.useMemo(() => new Map(allCategories.map((c) => [String(c.id), c])), [allCategories]); + const categoryIDs = React.useMemo(() => allCategories.map((c) => String(c.id)), [allCategories]); + const selectedIDs = value != null ? [String(value)] : []; + + const trimmed = inputValue.trim(); + const lower = trimmed.toLowerCase(); + // 与 base-ui 默认子串过滤保持一致，用来判断「有没有相关分类」。 + const matchCount = trimmed + ? allCategories.filter((c) => c.name.toLowerCase().includes(lower)).length + : allCategories.length; + + const createAndSelect = async () => { + if (!trimmed || creating) return; + // 精确同名已存在则直接选中，不重复创建。 + const existing = allCategories.find((c) => c.name.toLowerCase() === lower); + if (existing) { + onValueChange(existing.id); + setInputValue(""); + return; + } + setCreating(true); + try { + const created = await api.createTaskCategory(trimmed); + setLocalExtra((prev) => [...prev, created]); + onValueChange(created.id); + setInputValue(""); + onCategoryCreated(); + toast.success(`已创建分类「${created.name}」`); + } catch (e) { + toast.error(`创建分类失败：${(e as Error).message}`); + } finally { + setCreating(false); + } + }; + + return ( + byID.get(id)?.name ?? id} + multiple + value={selectedIDs} + onValueChange={(next: string[]) => { + // 单选：取最新选中的一个；移除 chip（清空）则回到未分类。 + const last = next[next.length - 1]; + onValueChange(last ? Number(last) : undefined); + setInputValue(""); + }} + inputValue={inputValue} + onInputValueChange={setInputValue} + > + + + {selectedIDs.map((id) => ( + {byID.get(id)?.name ?? "未知分类"} + ))} + + { + // 完全无匹配时回车 = 创建；有匹配项时保留 base-ui 的「回车选中高亮项」。 + if (e.key === "Enter" && matchCount === 0 && trimmed) { + e.preventDefault(); + void createAndSelect(); + } + }} + /> + + + + {(id: string) => ( + + {byID.get(id)?.name ?? id} + + )} + + {matchCount === 0 && + (trimmed ? ( + + ) : ( +
输入名称以搜索或创建分类
+ ))} + + + ); +} + const COMPANY_SCOPE_LABELS: Record = { domain: "域名", ip: "IP", @@ -1799,14 +2055,16 @@ function CreateTaskSheet({ tasks, categories, onCreated, + onCategoriesChanged, }: { tasks: Task[]; categories: TaskCategory[]; onCreated: () => void; + onCategoriesChanged: () => void; }) { const [open, setOpen] = React.useState(false); const [name, setName] = React.useState(""); - const [categoryID, setCategoryID] = React.useState("uncategorized"); + const [categoryID, setCategoryID] = React.useState(undefined); const [description, setDescription] = React.useState(""); const [goal, setGoal] = React.useState(""); const [selectedTemplateID, setSelectedTemplateID] = React.useState(null); @@ -1876,7 +2134,7 @@ function CreateTaskSheet({ const heartbeatSec = Math.max(10, Math.floor(Number(heartbeatMin) || 10)) * 60; // 下限 10min，与后端归一一致 await api.createTask({ name: name.trim(), - categoryId: categoryID === "uncategorized" ? undefined : Number(categoryID), + categoryId: categoryID, description: description.trim(), goal: goal.trim(), llmProfileIds: llmProfileIDs.map(Number), @@ -1889,7 +2147,7 @@ function CreateTaskSheet({ }); toast.success("任务已创建"); setName(""); - setCategoryID("uncategorized"); + setCategoryID(undefined); setDescription(""); setGoal(""); setSelectedTemplateID(null); @@ -1955,22 +2213,14 @@ function CreateTaskSheet({ 任务分类 - - 用于任务列表筛选和归档，不影响 Agent 执行。 + + 可选，单个分类；用于任务列表筛选和归档，不影响 Agent 执行。
diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 927dce4..706efb0 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -12,10 +12,12 @@ import type { AgentTrigger, Asset, Audit, + BatchCategoryItem, BatchControlItem, ChatAttachment, CommandRecord, Company, + CompanyScopeMutation, CompanyScopeRule, Conversation, ConvTokenSummary, @@ -218,6 +220,12 @@ export const api = { deleteTaskCategory: (id: number) => del<{ deleted: number }>(`/task-categories/${id}`), updateTaskCategory: (taskId: string, categoryId?: number) => patch(`/tasks/${taskId}/category`, { category_id: categoryId ?? null }), + // categoryId 省略/undefined = 移出分类（后端收到 null） + updateTasksCategory: (taskIds: string[], categoryId?: number) => + post<{ items: BatchCategoryItem[]; category: TaskCategory | null }>("/tasks/category/batch", { + task_ids: taskIds, + category_id: categoryId ?? null, + }), taskTemplates: () => get<{ templates: TaskTemplate[] }>("/task-templates").then((r) => arr(r.templates)), createTaskTemplate: (input: Pick) => post("/task-templates", input), @@ -388,12 +396,12 @@ export const api = { }, ), addCompanyScope: (id: number, scope: CompanyScopeRule[], reason = "") => - post<{ added: number; skipped: number; invalid: number; errors?: string[] }>(`/companies/${id}/scope`, { + post(`/companies/${id}/scope`, { scope, reason, }), updateCompanyScope: (id: number, scope: CompanyScopeRule[], reason = "") => - post<{ added: number; skipped: number; invalid: number; errors?: string[] }>(`/companies/${id}/scope`, { + post(`/companies/${id}/scope`, { scope, reason, reset: true, diff --git a/web/src/lib/mock/handler.ts b/web/src/lib/mock/handler.ts index cb3e260..8dfb02d 100644 --- a/web/src/lib/mock/handler.ts +++ b/web/src/lib/mock/handler.ts @@ -356,6 +356,25 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re } return { deleted: categoryID }; } + if (path === "/tasks/category/batch" && m === "POST") { + const requested = Array.isArray(b.task_ids) ? b.task_ids.map(String) : []; + const taskIDs = [...new Set(requested)]; + if (taskIDs.length === 0 || taskIDs.length > 100) throw new Error("task_ids 数量必须为 1-100"); + const categoryID = typeof b.category_id === "number" ? b.category_id : undefined; + const category = categoryID === undefined ? undefined : mockTaskCategories.find((item) => item.id === categoryID); + if (categoryID !== undefined && !category) throw new Error("任务分类不存在"); + const items = taskIDs.map((id) => { + const task = mockTasks.find((item) => item.id === id); + if (!task) return { id, ok: false, error: "task not found" }; + task.category_id = category?.id; + task.category_name = category?.name; + return { id, ok: true }; + }); + return { + items, + category: category ? mockTaskCategorySnapshot().find((item) => item.id === category.id) : null, + }; + } if (seg[0] === "tasks" && seg[2] === "category" && seg.length === 3 && m === "PATCH") { const task = mockTasks.find((item) => item.id === seg[1]); if (!task) throw new Error("任务不存在"); diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 6377d97..3a07d3c 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -298,6 +298,16 @@ export interface CompanyScopeRule { value: string; } +// 资产范围写入的结果。errors 是本次提交里不合法的行；warnings 是与本次提交无关、 +// 但会让归属结果不符合预期的既有数据问题（如 ip 字段存了主机名的资产）。 +export interface CompanyScopeMutation { + added: number; + skipped: number; + invalid: number; + errors?: string[]; + warnings?: string[]; +} + // 公司资产范围规则的一条（归属唯一真值来源）。 export interface ScopeRow { id: number; @@ -624,6 +634,13 @@ export interface BatchControlItem { error?: string; } +// 批量改分类的逐任务结果。失败只可能是任务已被删除，分类本身的写入是原子的。 +export interface BatchCategoryItem { + id: string; + ok: boolean; + error?: string; +} + // Whole-task (all agents) token aggregate. export interface TokenTotal { input_tokens: number;