From 4f088e519bc61d700441503cfd2421ed570e7bed Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:52:15 -0400 Subject: [PATCH 1/4] =?UTF-8?q?feat(digest):=20=E6=8E=A2=E7=B4=A2=E5=9B=BE?= =?UTF-8?q?=E5=86=B7=E8=8A=82=E7=82=B9=E5=8E=8B=E7=BC=A9(cold-digest)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把老且长期不活跃的意图/事实折叠成 digest 节点进 graph_overview 展示， 原始节点永久保留、按 id 可完整还原（存储无损、只压呈现、折叠可逆）。 - 数据模型：KindDigest 节点 + RelCovers 边（真相源）；新列 content_version、 cold_since_round（节点级）、round_no（任务级 planner 轮次）；均带旧库迁移。 - 纯算法(coldgraph.go)：冷热判定(逆向可达 + 任一活分支即热) + R=6 轮防抖 + 连通分量分组(§3) + §3.1 同父兄弟归组(救"热 hub+扁平死叶") + §5.3 签名。 - 后台压缩(compaction.go)：minor 折未覆盖冷块(tiered)、major 回源重压合并碎片； 签名幂等复用 body；提交时活跃度复核(绝不覆盖已复活节点)；60s 冷却 + 互斥。 - 概览折叠(§6)：recent_* 剔除已折成员 + 渲染时复活检查；cold_digests + 按资产 索引 cold_index(§6.2)；expand_digest/expand_index 还原；covered_members 反查 悬空血缘 id；60 兜底截断均衡保 live。 - planner 每轮 OnPlannerRound 维护戳记并触发后台压缩，绝不上热路径。 纯算法单测覆盖 §附/§3.1/§2.3/§5.3；已对 437 节点真实图验证：热 86、冷候选 324、 §3.1 分组 47 块 + 14 孤立冷点（对齐设计文档实测校验）。 --- agent/coldcap_test.go | 35 +++ agent/coldgraph.go | 330 ++++++++++++++++++++++++ agent/coldgraph_test.go | 176 +++++++++++++ agent/compaction.go | 541 ++++++++++++++++++++++++++++++++++++++++ agent/planner.go | 11 + agent/tools.go | 80 +++++- agent/tools_digest.go | 228 +++++++++++++++++ agent/tools_insert.go | 3 + db/constants.go | 10 + db/digest.go | 253 +++++++++++++++++++ db/exploration.go | 9 +- db/schema.sql | 55 +++- server/server.go | 4 + 13 files changed, 1720 insertions(+), 15 deletions(-) create mode 100644 agent/coldcap_test.go create mode 100644 agent/coldgraph.go create mode 100644 agent/coldgraph_test.go create mode 100644 agent/compaction.go create mode 100644 agent/tools_digest.go create mode 100644 db/digest.go diff --git a/agent/coldcap_test.go b/agent/coldcap_test.go new file mode 100644 index 0000000..0e77d84 --- /dev/null +++ b/agent/coldcap_test.go @@ -0,0 +1,35 @@ +package agent + +import "testing" + +func TestBalancedCap(t *testing.T) { + cases := []struct { + name string + done, facts, cap int + wantDone, wantFacts int + wantTrunc bool + }{ + {"under cap", 15, 20, 60, 15, 20, false}, + {"exactly cap", 30, 30, 60, 30, 30, false}, + {"both large → even split", 86, 40, 60, 30, 30, true}, + {"intents dominate, facts small kept whole", 86, 5, 60, 55, 5, true}, + {"facts dominate, intents small kept whole", 3, 200, 60, 3, 57, true}, + {"both huge → half/half", 500, 500, 60, 30, 30, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + gotDone, gotFacts, trunc := balancedCap(c.done, c.facts, c.cap) + if gotDone != c.wantDone || gotFacts != c.wantFacts || trunc != c.wantTrunc { + t.Fatalf("balancedCap(%d,%d,%d) = (%d,%d,%v), want (%d,%d,%v)", + c.done, c.facts, c.cap, gotDone, gotFacts, trunc, c.wantDone, c.wantFacts, c.wantTrunc) + } + if trunc && gotDone+gotFacts > c.cap { + t.Fatalf("kept %d+%d exceeds cap %d", gotDone, gotFacts, c.cap) + } + // never starve a non-empty side to zero when it had items + if trunc && c.facts > 0 && gotFacts == 0 { + t.Fatalf("facts starved to 0") + } + }) + } +} diff --git a/agent/coldgraph.go b/agent/coldgraph.go new file mode 100644 index 0000000..4c33d02 --- /dev/null +++ b/agent/coldgraph.go @@ -0,0 +1,330 @@ +package agent + +// cold-digest §2/§3: pure graph algorithms for cold-node compression. +// +// This file is deliberately free of any DB or LLM dependency so the hot/cold +// judgment, connectivity grouping (§3) and the same-parent singleton rescue +// (§3.1) can be unit-tested in isolation. Callers translate db.Node/db.Edge into +// the light cgNode/cgEdge structs and feed the per-node bookkeeping (cold_since +// stamps, content versions) alongside. +// +// Edge direction convention (matches db + agent/tools.go graphOverviewData): +// every edge From→To means From is the parent/upstream and To the child/ +// downstream, for ALL relations (yields: intent→fact, derived_from/spawns: +// parent→child). "Downstream" therefore follows From→To. + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "sort" + + "github.com/Autumn-27/artex/db" +) + +// cgNode is the minimal node view the cold-graph algorithms need. +type cgNode struct { + ID int64 + Kind string + State string +} + +// cgEdge is one exploration edge (From = parent/upstream, To = child/downstream). +type cgEdge struct { + From int64 + Rel string + To int64 +} + +// coldParams are the tunable thresholds (cold-digest §7). +type coldParams struct { + R int // debounce: a node must be continuously inactive ≥R planner rounds (§7 R=6) + K int // min block size to fold; K=2 skips only degenerate singletons (§7 K=2) +} + +func defaultColdParams() coldParams { return coldParams{R: 6, K: 2} } + +// coldGraph is an in-memory adjacency view over real exploration edges. The +// derived view layer (kind=digest nodes, rel=covers edges) is filtered out at +// construction so it can never distort causal reachability or grouping (§2/§3). +type coldGraph struct { + nodes map[int64]cgNode + children map[int64][]int64 // From → [To] (downstream) + parents map[int64][]int64 // To → [From] (upstream) +} + +func newColdGraph(nodes []cgNode, edges []cgEdge) *coldGraph { + g := &coldGraph{ + nodes: make(map[int64]cgNode, len(nodes)), + children: map[int64][]int64{}, + parents: map[int64][]int64{}, + } + for _, n := range nodes { + g.nodes[n.ID] = n + } + for _, e := range edges { + if e.Rel == db.RelCovers { // derived view layer, not exploration causality (§2/§3) + continue + } + if _, ok := g.nodes[e.From]; !ok { + continue + } + if _, ok := g.nodes[e.To]; !ok { + continue + } + g.children[e.From] = append(g.children[e.From], e.To) + g.parents[e.To] = append(g.parents[e.To], e.From) + } + return g +} + +// isLiveIntent reports whether a node is a not-yet-settled intent — the frontier +// that keeps its ancestors hot. paused counts as live (it may still resume); +// settled = done/blocked/exhausted/stopped. +func isLiveIntent(n cgNode) bool { + if n.Kind != db.KindIntent { + return false + } + switch n.State { + case "open", "running", "paused": + return true + } + return false +} + +// foldableKind reports whether a node kind is eligible for folding at all (§2: +// only fact and settled intent; finding/goal/hint/begin/digest never fold). +func foldableKind(k string) bool { return k == db.KindIntent || k == db.KindFact } + +// hotSet computes the hot nodes (§2 rule 1+2): a node is hot iff it can reach a +// live intent by going downstream (it is an ancestor of a live intent), OR it is +// a live intent, OR it is a direct child of a live intent (rule 1: an open/ +// running intent's freshly produced facts stay hot). Everything else is cold- +// eligible. "Any live branch keeps the whole chain hot" falls out of ancestor +// marking. Iterative (no recursion) to tolerate deep chains and cycles. +func (g *coldGraph) hotSet() map[int64]bool { + hot := map[int64]bool{} + var stack []int64 + for _, n := range g.nodes { + if isLiveIntent(n) { + stack = append(stack, n.ID) + } + } + // Walk upstream from every live intent, marking all ancestors hot. + for len(stack) > 0 { + id := stack[len(stack)-1] + stack = stack[:len(stack)-1] + if hot[id] { + continue + } + hot[id] = true + stack = append(stack, g.parents[id]...) + } + // A live intent's direct children (its fresh facts) stay hot (rule 1). + for _, n := range g.nodes { + if isLiveIntent(n) { + for _, c := range g.children[n.ID] { + hot[c] = true + } + } + } + return hot +} + +// structuralCold is the set of foldable nodes that are currently not hot — i.e. +// settled + blood-inactive (§2 rules 1+2), before the ≥R debounce is applied. +func (g *coldGraph) structuralCold(hot map[int64]bool) map[int64]bool { + cold := map[int64]bool{} + for id, n := range g.nodes { + if foldableKind(n.Kind) && !hot[id] { + cold[id] = true + } + } + return cold +} + +// stampOp is one cold_since_round bookkeeping change (§2.3): Set=true stamps the +// round a node went cold; Set=false clears the stamp (the node revived / turned +// hot again). +type stampOp struct { + ID int64 + Set bool + Round int64 +} + +// computeStampOps derives the cold_since_round updates for this round. It stamps +// a node the round it FIRST goes cold (empty→round_no) and clears the stamp when +// it is no longer cold. It never re-stamps an already-stamped cold node — that is +// what preserves "how long it has been cold" (§2.3: measure the round it turned +// cold, not the round it last turned hot). coldSince maps node id → stamp (nil = +// unstamped / hot). +func computeStampOps(structCold map[int64]bool, coldSince map[int64]*int64, roundNo int64) []stampOp { + var ops []stampOp + seen := map[int64]bool{} + for id := range structCold { + seen[id] = true + if coldSince[id] == nil { + ops = append(ops, stampOp{ID: id, Set: true, Round: roundNo}) + } + } + // Clear stamps on nodes that are stamped but no longer cold (revived/hot). + for id, cs := range coldSince { + if cs != nil && !seen[id] { + ops = append(ops, stampOp{ID: id, Set: false}) + } + } + sort.Slice(ops, func(i, j int) bool { return ops[i].ID < ops[j].ID }) + return ops +} + +// eligibleCold narrows structuralCold to nodes that have been continuously cold +// for ≥R rounds (§2.3 / §3②). A node with no stamp, or one that has not yet +// aged R rounds, is held in the hot region a while longer (bias to conservative). +func (g *coldGraph) eligibleCold(structCold map[int64]bool, coldSince map[int64]*int64, roundNo int64, p coldParams) map[int64]bool { + out := map[int64]bool{} + for id := range structCold { + if cs := coldSince[id]; cs != nil && roundNo-*cs >= int64(p.R) { + out[id] = true + } + } + return out +} + +// block is a group of cold nodes to fold into one digest, plus the external +// parent nodes that anchor them (§3.1 "父作锚不作成员"): anchors are fed to the +// compressor as context but never become members / never get a covers edge. +type block struct { + Members []int64 // sorted; the nodes this digest covers + Anchors []int64 // sorted; external (non-member) parents, context only +} + +// group partitions `set` into foldable blocks (§3 + §3.1). Two passes of +// union-find: +// +// rule ① connect cold nodes joined by a real exploration edge (§3); +// rule ② connect the LEFTOVER singletons that share a common direct parent +// (§3.1 — rescues the "hot hub + flat dead leaves" fan-out), without +// disturbing any already-formed ≥2 block. +// +// Only components of size ≥K survive (§3① skips degenerate singletons). +func (g *coldGraph) group(set map[int64]bool, p coldParams) []block { + uf := newUnionFind(set) + // rule ①: real cold↔cold edges. + for from := range set { + for _, to := range g.children[from] { + if set[to] { + uf.union(from, to) + } + } + } + // rule ②: leftover singletons sharing a common parent. + comps := uf.components() + byParent := map[int64][]int64{} + for _, ids := range comps { + if len(ids) != 1 { + continue // only rescue singletons; never re-shuffle ≥2 blocks + } + s := ids[0] + for _, par := range g.parents[s] { + if g.nodes[par].Kind == db.KindDigest { // anchor must be a real node, not a digest + continue + } + byParent[par] = append(byParent[par], s) + } + } + for _, sibs := range byParent { + if len(sibs) < 2 { + continue // a lone cold child under a parent stays a true singleton (§3①) + } + for i := 1; i < len(sibs); i++ { + uf.union(sibs[0], sibs[i]) + } + } + // Emit surviving components as blocks, each with its external-parent anchors. + comps = uf.components() + var blocks []block + for _, ids := range comps { + if len(ids) < p.K { + continue + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + memberSet := make(map[int64]bool, len(ids)) + for _, m := range ids { + memberSet[m] = true + } + anchorSet := map[int64]bool{} + for _, m := range ids { + for _, par := range g.parents[m] { + if memberSet[par] { + continue + } + pn, ok := g.nodes[par] + if !ok || pn.Kind == db.KindDigest { + continue + } + anchorSet[par] = true + } + } + anchors := make([]int64, 0, len(anchorSet)) + for a := range anchorSet { + anchors = append(anchors, a) + } + sort.Slice(anchors, func(i, j int) bool { return anchors[i] < anchors[j] }) + blocks = append(blocks, block{Members: ids, Anchors: anchors}) + } + // Deterministic order: by smallest member id. + sort.Slice(blocks, func(i, j int) bool { return blocks[i].Members[0] < blocks[j].Members[0] }) + return blocks +} + +// blockSignature is the change-detection key (§5.3): a hash over the sorted +// member ids + each member's content_version, plus the anchor ids + versions +// (so an anchor's summary/state change also invalidates the cached body). A +// re-compaction whose block matches an existing active digest's signature +// reuses the stored body and skips the LLM entirely. +func blockSignature(b block, contentVer map[int64]int) string { + h := sha256.New() + for _, m := range b.Members { + fmt.Fprintf(h, "m:%d:%d;", m, contentVer[m]) + } + for _, a := range b.Anchors { + fmt.Fprintf(h, "a:%d:%d;", a, contentVer[a]) + } + return hex.EncodeToString(h.Sum(nil)) +} + +// --- union-find --- + +type unionFind struct{ parent map[int64]int64 } + +func newUnionFind(set map[int64]bool) *unionFind { + uf := &unionFind{parent: make(map[int64]int64, len(set))} + for id := range set { + uf.parent[id] = id + } + return uf +} + +func (u *unionFind) find(x int64) int64 { + for u.parent[x] != x { + u.parent[x] = u.parent[u.parent[x]] + x = u.parent[x] + } + return x +} + +func (u *unionFind) union(a, b int64) { + ra, rb := u.find(a), u.find(b) + if ra != rb { + u.parent[ra] = rb + } +} + +func (u *unionFind) components() map[int64][]int64 { + out := map[int64][]int64{} + for id := range u.parent { + r := u.find(id) + out[r] = append(out[r], id) + } + return out +} diff --git a/agent/coldgraph_test.go b/agent/coldgraph_test.go new file mode 100644 index 0000000..2b6b0bd --- /dev/null +++ b/agent/coldgraph_test.go @@ -0,0 +1,176 @@ +package agent + +import ( + "testing" + + "github.com/Autumn-27/artex/db" +) + +// helper: intent/fact node +func intent(id int64, state string) cgNode { return cgNode{ID: id, Kind: db.KindIntent, State: state} } +func fact(id int64) cgNode { return cgNode{ID: id, Kind: db.KindFact, State: "confirmed"} } +func yields(from, to int64) cgEdge { return cgEdge{From: from, Rel: db.RelYields, To: to} } +func derived(from, to int64) cgEdge { return cgEdge{From: from, Rel: db.RelDerivedFrom, To: to} } + +func ptr(v int64) *int64 { return &v } + +// §附 快照1: a→b→c, b→d, with d live (running) and c settled/inactive. +// Expected: d/b/a hot (b kept hot by the live b→d branch); c is a cold candidate +// but an isolated singleton → not folded. +func TestHotCold_AnyLiveBranchKeepsChainHot(t *testing.T) { + // a(intent) → b(intent) → c(fact); b → d(intent, running) + nodes := []cgNode{intent(1, "done"), intent(2, "done"), fact(3), intent(4, "running")} + edges := []cgEdge{derived(1, 2), yields(2, 3), derived(2, 4)} + g := newColdGraph(nodes, edges) + hot := g.hotSet() + for _, id := range []int64{1, 2, 4} { + if !hot[id] { + t.Fatalf("node %d should be hot (ancestor of / is live intent 4)", id) + } + } + if hot[3] { + t.Fatalf("node 3 (dead leaf c) should be cold") + } + structCold := g.structuralCold(hot) + if !structCold[3] || len(structCold) != 1 { + t.Fatalf("only node 3 should be structurally cold, got %v", structCold) + } +} + +// §附 快照2: once d also finishes, a/b/c/d are all cold and connected → one block. +func TestHotCold_WholeChainFoldsWhenAllSettled(t *testing.T) { + nodes := []cgNode{intent(1, "done"), intent(2, "done"), fact(3), intent(4, "done")} + edges := []cgEdge{derived(1, 2), yields(2, 3), derived(2, 4)} + g := newColdGraph(nodes, edges) + hot := g.hotSet() + if len(hot) != 0 { + t.Fatalf("nothing should be hot once all settled, got %v", hot) + } + structCold := g.structuralCold(hot) + // all stamped R+ rounds ago + stamps := map[int64]*int64{1: ptr(1), 2: ptr(1), 3: ptr(1), 4: ptr(1)} + elig := g.eligibleCold(structCold, stamps, 100, defaultColdParams()) + if len(elig) != 4 { + t.Fatalf("all 4 nodes should be eligible cold, got %d", len(elig)) + } + blocks := g.group(elig, defaultColdParams()) + if len(blocks) != 1 || len(blocks[0].Members) != 4 { + t.Fatalf("expected one 4-member block, got %+v", blocks) + } +} + +// §2.3 debounce: a freshly-cooled node (stamp too recent) is not yet eligible. +func TestDebounce_RecentlyCooledNotEligible(t *testing.T) { + nodes := []cgNode{intent(1, "done"), fact(2)} + edges := []cgEdge{yields(1, 2)} + g := newColdGraph(nodes, edges) + structCold := g.structuralCold(g.hotSet()) + stamps := map[int64]*int64{1: ptr(98), 2: ptr(98)} // cooled at round 98 + elig := g.eligibleCold(structCold, stamps, 100, defaultColdParams()) + if len(elig) != 0 { + t.Fatalf("nodes cooled only 2 rounds ago (= c.n + needMajor := activeDigests >= c.m + if !needMinor && !needMajor { + return + } + if !c.tryStart(ts.ID()) { + return // already running, or within cooldown —派生态最终一致，下轮再压 + } + go func() { + defer c.finish(ts.ID()) + bg, cancel := context.WithTimeout(context.WithoutCancel(ctx), c.maxDur) + defer cancel() + if needMajor { + c.major(bg, ts) + } else { + c.minor(bg, ts) + } + }() + _ = round +} + +// maintain bumps round_no, recomputes hot/cold over the whole graph, and applies +// the cold_since_round stamp/clear ops (§2.3). Returns the new round plus the +// counts that drive the trigger: how many eligible-cold nodes are not yet covered +// (minor) and how many active digests exist (major). +func (c *Compactor) maintain(ts *db.ExplorationStore) (round int64, uncompressed, activeDigests int, err error) { + round, err = ts.BumpRound() + if err != nil { + return + } + g, _, err := loadColdGraph(ts) + if err != nil { + return + } + stamps, err := ts.ColdStamps() + if err != nil { + return + } + hot := g.hotSet() + structCold := g.structuralCold(hot) + ops := computeStampOps(structCold, stamps, round) + if err = ts.ApplyStampOps(toDBStampOps(ops)); err != nil { + return + } + applyStampsInPlace(stamps, ops) + elig := g.eligibleCold(structCold, stamps, round, c.params) + covered, err := ts.CoveredMembers() + if err != nil { + return + } + for id := range elig { + if _, ok := covered[id]; !ok { + uncompressed++ + } + } + ad, err := ts.ActiveDigests() + if err != nil { + return + } + activeDigests = len(ad) + return +} + +// minor folds the not-yet-covered eligible-cold set into new digest segments +// (tiered append, §5). Existing digests are untouched. +func (c *Compactor) minor(ctx context.Context, ts *db.ExplorationStore) { + round, err := ts.RoundNo() + if err != nil { + return + } + g, nodeByID, err := loadColdGraph(ts) + if err != nil { + return + } + stamps, err := ts.ColdStamps() + if err != nil { + return + } + cvers, err := ts.ContentVersions() + if err != nil { + return + } + covered, err := ts.CoveredMembers() + if err != nil { + return + } + hot := g.hotSet() + elig := g.eligibleCold(g.structuralCold(hot), stamps, round, c.params) + uncompressed := map[int64]bool{} + for id := range elig { + if _, ok := covered[id]; !ok { + uncompressed[id] = true + } + } + blocks := g.group(uncompressed, c.params) + if len(blocks) == 0 { + return // this batch has no ≥2 connected/shared-parent block — nothing to fold (§7) + } + for _, b := range blocks { + c.foldBlock(ctx, ts, g, b, nodeByID, cvers, c.generationFor(b, nil, nil)) + } + // A minor may have pushed the segment count over M → merge in the same run. + if ad, e := ts.ActiveDigests(); e == nil && len(ad) >= c.m { + c.major(ctx, ts) + } +} + +// major re-derives the whole grouping from source over ALL eligible-cold nodes +// (§5.1 回源重压), then reconciles against the active digests by signature: +// unchanged blocks keep their digest (no LLM), stale digests are superseded, and +// new/changed blocks are compressed afresh. This is where tiered fragments of one +// direction merge and where "later became connected" blocks unify (§5.2). +func (c *Compactor) major(ctx context.Context, ts *db.ExplorationStore) { + round, err := ts.RoundNo() + if err != nil { + return + } + g, nodeByID, err := loadColdGraph(ts) + if err != nil { + return + } + stamps, err := ts.ColdStamps() + if err != nil { + return + } + cvers, err := ts.ContentVersions() + if err != nil { + return + } + active, err := ts.ActiveDigests() + if err != nil { + return + } + hot := g.hotSet() + elig := g.eligibleCold(g.structuralCold(hot), stamps, round, c.params) + blocks := g.group(elig, c.params) + + bySig := map[string]*db.Node{} + genBySig := map[string]int{} + for _, d := range active { + sig, gen := digestSigGen(d) + bySig[sig] = d + genBySig[sig] = gen + } + desired := map[string]bool{} + var toCreate []block + for _, b := range blocks { + sig := blockSignature(b, cvers) + desired[sig] = true + if _, ok := bySig[sig]; ok { + continue // unchanged → reuse the existing digest, skip LLM (§5.3) + } + toCreate = append(toCreate, b) + } + // Supersede stale digests FIRST (atomic drop of their covers edges) so a member + // is never covered by both an old and a new digest (§5.1 one-member-one-digest). + var stale []int64 + for _, d := range active { + sig, _ := digestSigGen(d) + if !desired[sig] { + stale = append(stale, d.ID) + } + } + if err := ts.SupersedeDigests(stale); err != nil { + log.Printf("[compaction] supersede exp=%d: %v", ts.ID(), err) + } + for _, b := range toCreate { + c.foldBlock(ctx, ts, g, b, nodeByID, cvers, c.generationFor(b, active, genBySig)) + } +} + +// foldBlock compresses one block and writes its digest — with a commit-time +// liveness recheck (§ concurrency): between grouping and write the graph may have +// changed, so any member that has since gone hot (revived) is dropped from the +// covers set. If the block dissolves below K it is skipped. +func (c *Compactor) foldBlock(ctx context.Context, ts *db.ExplorationStore, g *coldGraph, b block, nodeByID map[int64]*db.Node, cvers map[int64]int, generation int) { + body, err := c.compress(ctx, g, b, nodeByID) + if err != nil { + log.Printf("[compaction] compress exp=%d block=%v: %v", ts.ID(), b.Members, err) + return + } + // Re-read fresh state and drop any member that revived while we compressed. + fresh, _, err := loadColdGraph(ts) + if err != nil { + return + } + freshHot := fresh.hotSet() + members := make([]int64, 0, len(b.Members)) + for _, mID := range b.Members { + if !freshHot[mID] { + members = append(members, mID) + } + } + if len(members) < c.params.K { + return // block revived out from under us — leave those nodes hot, don't fold + } + final := block{Members: members, Anchors: b.Anchors} + payload := digestPayload(body, final, nodeByID, generation, blockSignature(final, cvers)) + if _, err := ts.AddDigest(payload, members); err != nil { + log.Printf("[compaction] add digest exp=%d: %v", ts.ID(), err) + } +} + +// generationFor computes a digest's重摘代次 (§1): 1 for a fresh fold; for a major +// merge, max(generation) over the active digests that overlap this block's +// members, +1. +func (c *Compactor) generationFor(b block, active []*db.Node, genBySig map[string]int) int { + if len(active) == 0 { + return 1 + } + memberSet := make(map[int64]bool, len(b.Members)) + for _, m := range b.Members { + memberSet[m] = true + } + best := 0 + for _, d := range active { + _, gen := digestSigGen(d) + for _, m := range digestMemberIDs(d) { + if memberSet[m] { + if gen > best { + best = gen + } + break + } + } + } + return best + 1 +} + +// tryStart acquires the per-task compaction lock, honoring the cooldown. +func (c *Compactor) tryStart(expID int64) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.running[expID] { + return false + } + if t, ok := c.lastRun[expID]; ok && time.Since(t) < c.cooldown { + return false + } + c.running[expID] = true + return true +} + +func (c *Compactor) finish(expID int64) { + c.mu.Lock() + defer c.mu.Unlock() + c.running[expID] = false + c.lastRun[expID] = time.Now() +} + +// --- helpers: db ↔ coldgraph --- + +// loadColdGraph reads the exploration's nodes + edges and builds the cold-graph +// view plus an id→node index (for summaries/payload during compression). +func loadColdGraph(ts *db.ExplorationStore) (*coldGraph, map[int64]*db.Node, error) { + // Compaction must see the WHOLE graph, not the default row caps — pass a very + // high limit so the LIMIT clause is effectively unbounded for real task sizes. + const allRows = 1 << 30 + nodes, err := ts.Nodes(allRows) + if err != nil { + return nil, nil, err + } + edges, err := ts.Edges(allRows) + if err != nil { + return nil, nil, err + } + cgNodes := make([]cgNode, 0, len(nodes)) + byID := make(map[int64]*db.Node, len(nodes)) + for _, n := range nodes { + cgNodes = append(cgNodes, cgNode{ID: n.ID, Kind: n.Kind, State: n.State}) + byID[n.ID] = n + } + cgEdges := make([]cgEdge, 0, len(edges)) + for _, e := range edges { + cgEdges = append(cgEdges, cgEdge{From: e.From, Rel: e.Rel, To: e.To}) + } + return newColdGraph(cgNodes, cgEdges), byID, nil +} + +func toDBStampOps(ops []stampOp) []db.StampOp { + out := make([]db.StampOp, len(ops)) + for i, o := range ops { + out[i] = db.StampOp{ID: o.ID, Set: o.Set, Round: o.Round} + } + return out +} + +// applyStampsInPlace folds the just-applied ops into the in-memory stamp map so +// eligibility can be computed immediately without a re-read. +func applyStampsInPlace(stamps map[int64]*int64, ops []stampOp) { + for _, o := range ops { + if o.Set { + r := o.Round + stamps[o.ID] = &r + } else { + stamps[o.ID] = nil + } + } +} + +// --- helpers: digest payload --- + +// digestPayload builds the digest node payload (cold-digest §1): the body, the +// member ids split by kind (restore cache; source of truth is the covers edges), +// the anchor ids, the generation, and the change-detection signature. +func digestPayload(body string, b block, nodeByID map[int64]*db.Node, generation int, signature string) map[string]any { + var facts, intents []int64 + for _, m := range b.Members { + if n := nodeByID[m]; n != nil && n.Kind == db.KindIntent { + intents = append(intents, m) + } else { + facts = append(facts, m) + } + } + return map[string]any{ + "body": body, + "member_ids": map[string]any{"facts": facts, "intents": intents}, + "anchor_ids": b.Anchors, + "generation": generation, + "signature": signature, + } +} + +func digestSigGen(n *db.Node) (string, int) { + var p struct { + Signature string `json:"signature"` + Generation int `json:"generation"` + } + _ = json.Unmarshal(n.Payload, &p) + return p.Signature, p.Generation +} + +func digestMemberIDs(n *db.Node) []int64 { + var p struct { + MemberIDs struct { + Facts []int64 `json:"facts"` + Intents []int64 `json:"intents"` + } `json:"member_ids"` + } + _ = json.Unmarshal(n.Payload, &p) + return append(append([]int64{}, p.MemberIDs.Facts...), p.MemberIDs.Intents...) +} + +// --- helpers: compression input + LLM (§4) --- + +func nodeSummary(n *db.Node) string { + if n == nil { + return "" + } + var p map[string]any + if json.Unmarshal(n.Payload, &p) == nil { + if s, ok := p["summary"].(string); ok { + return s + } + if t, ok := p["text"].(string); ok { + return t + } + } + return "" +} + +func nodeConfidence(n *db.Node) string { + if n == nil { + return "" + } + var p map[string]any + if json.Unmarshal(n.Payload, &p) == nil { + if c, ok := p["confidence"].(string); ok { + return c + } + } + return "" +} + +// buildCompressionInput renders the connected sub-graph for the §4 prompt: +// member nodes (summary + id + kind + state + confidence), the internal blood +// edges among members, and — for a §3.1 shared-parent group — the anchor parents +// as context ("共同父 #p"), which are NOT members. +func buildCompressionInput(g *coldGraph, b block, nodeByID map[int64]*db.Node) string { + memberSet := make(map[int64]bool, len(b.Members)) + for _, m := range b.Members { + memberSet[m] = true + } + var sb strings.Builder + sb.WriteString("【成员节点（要压缩的）】：\n") + for _, m := range b.Members { + n := nodeByID[m] + kind := "fact" + if n != nil && n.Kind == db.KindIntent { + kind = "intent" + } + state := "" + if n != nil { + state = n.State + } + line := fmt.Sprintf("- #%d [%s/%s] %s", m, kind, state, nodeSummary(n)) + if conf := nodeConfidence(n); conf != "" { + line += fmt.Sprintf(" (confidence=%s)", conf) + } + sb.WriteString(line + "\n") + } + // internal edges among members + var edgeLines []string + for _, m := range b.Members { + for _, to := range g.children[m] { + if memberSet[to] { + edgeLines = append(edgeLines, fmt.Sprintf("- #%d 产出/派生→ #%d", m, to)) + } + } + } + if len(edgeLines) > 0 { + sb.WriteString("\n【成员之间的血缘边（父→子）】：\n") + sort.Strings(edgeLines) + sb.WriteString(strings.Join(edgeLines, "\n") + "\n") + } + if len(b.Anchors) > 0 { + sb.WriteString("\n【共同父 / 上下文锚（不是成员，只用于理解这些结果从哪个意图探出）】：\n") + for _, a := range b.Anchors { + n := nodeByID[a] + state := "" + if n != nil { + state = n.State + } + sb.WriteString(fmt.Sprintf("- #%d [%s] %s\n", a, state, nodeSummary(n))) + } + } + return sb.String() +} + +// compress runs the §4 body LLM call on one block. Uses the same model the agent +// runs on; thinking disabled (a pure summarization step). +func (c *Compactor) compress(ctx context.Context, g *coldGraph, b block, nodeByID map[int64]*db.Node) (string, error) { + req := llm.CompletionRequest{ + System: []string{compressionSystemPrompt}, + Messages: []llm.Message{llm.UserText(buildCompressionInput(g, b, nodeByID))}, + MaxTokens: 1500, + Thinking: "disabled", + } + msg, _, _, err := c.prov.Complete(ctx, req) + if err != nil { + return "", err + } + body := strings.TrimSpace(msg.Text()) + if body == "" { + return "", fmt.Errorf("empty body from model") + } + return body, nil +} + +// compressionSystemPrompt is the §4 body prompt. +const compressionSystemPrompt = `你在压缩一组【彼此关联】的探索节点，产出一段综合结论(body)，供规划者快速掌握"这一片已经探明了什么"。 + +输入是一个连通子图： +- 节点：每条是一个意图或事实的 summary（一句话），带 id、类型(intent/fact)、state、confidence(若有)。 +- 关系：节点之间的血缘边（A 派生自 B / A 产出 B），说明它们如何串联。 +- 若节点间没有直接血缘边、但同属一个上游意图（会另给出该上游意图作为"共同父 #p"），则按"这个意图（#p）探到了什么"来综合它们——共同父只是上下文锚、不是要压缩的成员。 + +据此写一段 body： +1. 综合、不罗列：顺着关系把因果串起来（哪个事实催生哪个意图、哪条意图产出了哪个结论），讲成"这一片探索得出了什么"，不要把每条 summary 抄一遍。 +2. 保留区分度：彼此不同的结论分别说清，别揉成一句笼统的话。 +3. 保留证据强度：带 confidence 的结论标出 observed / inferred；inferred 的否定/存疑结论要点明它只是推断、可复核，别写成定论。 +4. 带上 id：每条结论后标注来源节点 id（如"…（#12,#28）"），让规划者能按 id 还原原节点。 +5. 正向陈述、只写输入里有的：不脑补、不引入输入中没有的判断。 +6. 长度随内容自适应：结论少就短，多且互不相同就写够——但整体显著短于所有输入 summary 的总和。 + +只输出 body 正文本身。` diff --git a/agent/planner.go b/agent/planner.go index 315d4d2..5d74fb0 100644 --- a/agent/planner.go +++ b/agent/planner.go @@ -35,6 +35,7 @@ type Planner struct { injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes) nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming) maxTokensFn func() int // resolver: per-reply output cap (nil/0 = send no cap) + compactor *Compactor // cold-node compaction (§7); nil = disabled // todos keeps ONE plan-scratchpad per task (keyed by exploration id) so the // planner's multi-step plan survives across wake-ups — each Plan() is a fresh @@ -50,6 +51,11 @@ func NewPlanner(prov llm.Provider, model, workDir string, tx *transcript.Store, func (p *Planner) SetCompactionWindowResolver(fn func() int) { p.windowFn = fn } +// SetCompactor wires the cold-node compactor (cold-digest §7). Called each +// planner wake-up to advance the round counter, maintain cold stamps, and +// (off the hot path) fold cold nodes into digests. nil = feature disabled. +func (p *Planner) SetCompactor(c *Compactor) { p.compactor = c } + // SetNonStreaming wires a resolver deciding whether runs use the non-streaming // model path (true = non-streaming). nil/unset = streaming (default). func (p *Planner) SetNonStreaming(fn func() bool) { p.nonStreamingFn = fn } @@ -326,6 +332,11 @@ func plannerSystem(goal, dataDir, workDir string) string { // for time/heartbeat wakes). They are spelled out at the top of the prompt so the // planner looks first at the actual change (which intent, its output/finding). func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, goal string, triggers []TriggerEvent, emit func(db.Activity)) (met bool, reason string, err error) { + // cold-digest §2.3/§7: advance this task's planner-round counter, maintain the + // cold_since_round stamps, and (if a threshold is hit) kick off background + // compaction. Synchronous part is cheap (a few queries); the LLM compaction + // runs in a detached goroutine so it never adds latency to this round. + p.compactor.OnPlannerRound(ctx, ts) tsx := NewToolSet(ts, "planner") if as != nil { tsx.SetAssetStore(as, as.Companies()) diff --git a/agent/tools.go b/agent/tools.go index fb0eec2..94754fe 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -12,6 +12,28 @@ import ( actool "github.com/Autumn-27/norma/tool" ) +// balancedCap decides how many done-intents and facts to keep when the combined +// unfolded settled/cold region exceeds cap (§6.3). The smaller side is kept whole; +// the larger side takes the remaining budget; neither is starved below cap/2. +// Returns the keep counts and whether truncation applied. Newest-first slices, so +// callers keep the head. +func balancedCap(done, facts, cap int) (keepDone, keepFacts int, truncated bool) { + if done+facts <= cap { + return done, facts, false + } + half := cap / 2 + keepDone, keepFacts = done, facts + switch { + case done > half && facts > half: + keepDone, keepFacts = half, cap-half + case done > half: + keepDone = cap - facts // facts fit in ≤half; intents take the rest + default: + keepFacts = cap - done // intents fit in ≤half; facts take the rest + } + return keepDone, keepFacts, true +} + // compactIntents distills intents to {id, summary, state, asset_ids, parents, // yields} so the planner sees both the direction and its LINEAGE — parents (the // upstream nodes it derived from: facts/intents/findings) and yields (the facts/ @@ -396,6 +418,20 @@ func (t *ToolSet) graphOverviewData() map[string]any { factFrom[e.To] = e.From } } + // cold-digest §6: members folded into an active digest are shown via cold_digests + // (below), not the flat recent_* lists. `covered` maps member id → its digest id. + // §6 render-time revival check: a covered member that has become hot again (a new + // intent derived from it) must reappear this round — so `hidden` folds a member out + // only when it is covered AND still cold. covered_members (built from hidden) lets a + // parents/yields id pointing into a live fold stay resolvable (§6.5 dangling lineage). + covered, _ := t.ts.CoveredMembers() + var hotAtRender map[int64]bool + if len(covered) > 0 { + if cg, _, err := loadColdGraph(t.ts); err == nil { + hotAtRender = cg.hotSet() + } + } + hidden := func(id int64) bool { _, c := covered[id]; return c && !hotAtRender[id] } fr, _ := t.ts.Frontier(100) out["open_intents"] = compactIntents(fr, parentsOf, yieldsOf) all, _ := t.ts.ListByKind(db.KindIntent, 300) @@ -405,9 +441,10 @@ func (t *ToolSet) graphOverviewData() map[string]any { case "running": running = append(running, n) case "done", "blocked", "exhausted": - if len(recentDone) < 15 { - recentDone = append(recentDone, n) + if hidden(n.ID) { + continue // in a cold_digest and still cold — shown via cold_digests (§6.2) } + recentDone = append(recentDone, n) // §6: no 15-cap; folded ones are gone, cap applied below } } out["running_intents"] = compactIntents(running, parentsOf, yieldsOf) @@ -484,10 +521,10 @@ func (t *ToolSet) graphOverviewData() map[string]any { findingList = append(findingList, m) } out["finding_list"] = findingList - recentFacts := make([]map[string]any, 0, 20) + recentFacts := make([]map[string]any, 0, len(factNodes)) for _, n := range factNodes { - if len(recentFacts) >= 20 { - break + if hidden(n.ID) { + continue // in a cold_digest and still cold — surfaced via cold_digests (§6.2) } m := compactNode(n) if from := factFrom[n.ID]; from > 0 { @@ -503,7 +540,38 @@ func (t *ToolSet) graphOverviewData() map[string]any { } recentFacts = append(recentFacts, m) } + // §6.3 兜底截断：未折的 settled/cold 区（recent_done_intents + recent_facts）总量 > + // cap 时按最新优先截——live(open/running) 在独立字段、永不受影响。两边【均衡】保留： + // 较少的一侧全留，较多的一侧填剩余额度，各自至少保底 cap/2，绝不把某一侧饿到 0 + // （深探索链下 hot 的 settled 意图很多，天真的"意图优先"会把 facts 挤没）。正常由压缩 + // 阈值约束、不触发；触发即标注，说明压缩滞后。 + const unfoldedCap = 60 + if doneList, ok := out["recent_done_intents"].([]map[string]any); ok { + if keepDone, keepFacts, truncated := balancedCap(len(doneList), len(recentFacts), unfoldedCap); truncated { + out["recent_done_intents"] = doneList[:keepDone] + recentFacts = recentFacts[:keepFacts] + out["unfolded_truncated"] = true // 正常不触发；触发说明压缩滞后 + } + } out["recent_facts"] = recentFacts // 最近事实的 {id, summary, from_intent, confidence?}，详情/证据用 node_detail(id) + // cold-digest §6.1/§6.2: the folded cold region + the per-asset directory that + // collapses independent directions, plus the dangling-lineage resolver map. + if cds, cidx := t.coldDigestOverview(); len(cds) > 0 { + out["cold_digests"] = cds // [{id, body, member_count}] —— 直接读 body (§6.1) + out["cold_index"] = cidx // [{asset, asset_id, digest_ids}] —— 按资产收敛方向 (§6.2) + } + if len(covered) > 0 { + cm := make(map[string]int64, len(covered)) + for member, dig := range covered { + if hotAtRender[member] { + continue // revived → shown live this round, not a dangling folded id + } + cm[strconv.FormatInt(member, 10)] = dig + } + if len(cm) > 0 { + out["covered_members"] = cm // 悬空血缘 id → 它属于哪个 digest；用 expand_digest 展开 (§6.5) + } + } // the original task (root) so the planner always has it, not just the // decomposed goals. if description, goal, err := t.ts.Root(); err == nil { @@ -2026,6 +2094,8 @@ func (t *ToolSet) listWorkerTraces() actool.CoreTool { func (t *ToolSet) PlannerTools() []actool.CoreTool { return []actool.CoreTool{ t.graphOverview(), t.listFindings(), t.listFacts(), t.nodeDetail(), + // cold-digest §6.1: restore folded cold nodes (digest body → members → detail). + t.expandDigest(), t.expandIndex(), t.getWorkerOutput(), t.getWorkerTrace(), t.searchAllWorkerTraces(), t.listGoals(), t.addIntent(), t.proveGoal(), t.goalMet(), t.killWorkTool(), t.steerWorkTool(), // report_finding：规划态势研判时若自身已确证漏洞，可直接登记（与 worker 同工具）。 diff --git a/agent/tools_digest.go b/agent/tools_digest.go new file mode 100644 index 0000000..ccf8611 --- /dev/null +++ b/agent/tools_digest.go @@ -0,0 +1,228 @@ +package agent + +// cold-digest §6: graph_overview folding + the restore tools. +// +// coldDigestOverview — builds the folded cold region for graph_overview: +// cold_digests (flat {id, body, member_count}) and cold_index (§6.2, the +// per-asset directory that collapses independent directions). +// expand_digest(id) — level-1 restore: a digest's member compact list. +// expand_index(asset_id) — level-0 restore: the digests under one asset. + +import ( + "context" + "encoding/json" + "fmt" + "sort" + + "github.com/Autumn-27/artex/db" + actool "github.com/Autumn-27/norma/tool" +) + +// coldDigestOverview returns the folded cold region for graph_overview: the flat +// digest bodies and the asset-grouped index (§6.1/§6.2). covered is unused here +// (kept for symmetry with the caller's coverage computation). +func (t *ToolSet) coldDigestOverview() (digests []map[string]any, index []map[string]any) { + ads, err := t.ts.ActiveDigests() + if err != nil || len(ads) == 0 { + return nil, nil + } + memByDigest := map[int64][]int64{} + var allMembers []int64 + for _, d := range ads { + ms, _ := t.ts.DigestMembers(d.ID) + memByDigest[d.ID] = ms + allMembers = append(allMembers, ms...) + } + assetsByNode, _ := t.ts.NodeAssets(allMembers) + + digests = make([]map[string]any, 0, len(ads)) + for _, d := range ads { + var p struct { + Body string `json:"body"` + } + _ = json.Unmarshal(d.Payload, &p) + digests = append(digests, map[string]any{ + "id": d.ID, + "body": p.Body, + "member_count": len(memByDigest[d.ID]), + }) + } + + // index (§6.2): asset → digests touching it. A digest with no anchored asset + // falls into the 0 bucket ("(未锚定资产)"). One digest may list under >1 asset. + byAsset := map[int64]map[int64]bool{} // asset id → set of digest ids + assetSet := map[int64]bool{} + for dID, ms := range memByDigest { + touched := map[int64]bool{} + for _, m := range ms { + for _, a := range assetsByNode[m] { + touched[a] = true + } + } + if len(touched) == 0 { + touched[0] = true + } + for a := range touched { + if byAsset[a] == nil { + byAsset[a] = map[int64]bool{} + } + byAsset[a][dID] = true + if a != 0 { + assetSet[a] = true + } + } + } + labels := map[int64]string{} + if t.as != nil && len(assetSet) > 0 { + ids := make([]int64, 0, len(assetSet)) + for a := range assetSet { + ids = append(ids, a) + } + if assets, err := t.as.GetByIDs(ids); err == nil { + for _, a := range assets { + if v := assetValue(a); v != "" { + labels[a.ID] = v + } + } + } + } + index = make([]map[string]any, 0, len(byAsset)) + for a, dset := range byAsset { + dids := make([]int64, 0, len(dset)) + for d := range dset { + dids = append(dids, d) + } + sort.Slice(dids, func(i, j int) bool { return dids[i] < dids[j] }) + entry := map[string]any{"digest_ids": dids} + if a == 0 { + entry["asset"] = "(未锚定资产)" + } else { + entry["asset_id"] = a + if l := labels[a]; l != "" { + entry["asset"] = l + } else { + entry["asset"] = fmt.Sprintf("#%d", a) + } + } + index = append(index, entry) + } + sort.Slice(index, func(i, j int) bool { + return fmt.Sprint(index[i]["asset"]) < fmt.Sprint(index[j]["asset"]) + }) + return digests, index +} + +// digestMemberEntry builds the compact per-member view expand_digest returns — +// same shape as recent_facts / recent_done_intents (§6.1 middle level). +func (t *ToolSet) digestMemberEntry(id int64) map[string]any { + n, _ := t.ts.GetNode(id) + if n == nil { + return map[string]any{"id": id, "missing": true} + } + m := compactNode(n) + m["state"] = n.State + var p map[string]any + if json.Unmarshal(n.Payload, &p) == nil { + if c, ok := p["confidence"].(string); ok && c != "" { + m["confidence"] = c + } + } + return m +} + +// expandDigest returns a digest's covered members as a compact list (§6.1). It is +// a distinct tool from node_detail because it returns a LIST of members, not one +// node's full detail. +func (t *ToolSet) expandDigest() actool.CoreTool { + return writeTool("expand_digest", + "展开一个 cold digest：返回它折叠的成员紧凑列表（id/summary/state/confidence），与概览 recent_facts/recent_done_intents 同形状。要某条完整细节/证据用 node_detail(member_id)。", + map[string]any{ + "type": "object", + "properties": map[string]any{ + "id": map[string]any{"type": "integer", "description": "digest 节点 id（来自概览 cold_digests / cold_index / covered_members）"}, + }, + "required": []any{"id"}, + }, + func(ctx context.Context, raw json.RawMessage) (actool.Result, error) { + var in struct { + ID int64 `json:"id"` + } + _ = json.Unmarshal(raw, &in) + n, _ := t.ts.GetNode(in.ID) + if n == nil || n.Kind != db.KindDigest { + return jsonResult(map[string]any{"error": fmt.Sprintf("#%d 不是 digest 节点", in.ID)}) + } + var p struct { + Body string `json:"body"` + } + _ = json.Unmarshal(n.Payload, &p) + members, _ := t.ts.DigestMembers(in.ID) + list := make([]map[string]any, 0, len(members)) + for _, m := range members { + list = append(list, t.digestMemberEntry(m)) + } + return jsonResult(map[string]any{ + "id": in.ID, + "state": n.State, // active / superseded + "body": p.Body, + "members": list, + }) + }) +} + +// expandIndex returns the active digests under one asset (§6.2 level-0), each +// with its body + member count — so the planner can drill an asset directory down +// to its directions without reading every digest globally. +func (t *ToolSet) expandIndex() actool.CoreTool { + return writeTool("expand_index", + "展开冷区某个资产条目（来自概览 cold_index）：返回该资产名下的 cold digest 列表（id/body/member_count）。再往下看某个 digest 的成员用 expand_digest(id)。", + map[string]any{ + "type": "object", + "properties": map[string]any{ + "asset_id": map[string]any{"type": "integer", "description": "资产 id（来自概览 cold_index 的 asset_id；传 0 或省略取未锚定资产桶）"}, + }, + }, + func(ctx context.Context, raw json.RawMessage) (actool.Result, error) { + var in struct { + AssetID int64 `json:"asset_id"` + } + _ = json.Unmarshal(raw, &in) + ads, _ := t.ts.ActiveDigests() + out := make([]map[string]any, 0) + for _, d := range ads { + members, _ := t.ts.DigestMembers(d.ID) + assetsByNode, _ := t.ts.NodeAssets(members) + match := in.AssetID == 0 + for _, m := range members { + for _, a := range assetsByNode[m] { + if a == in.AssetID { + match = true + } + } + if match { + break + } + } + // asset_id==0 means "unanchored bucket": include only digests with no asset. + if in.AssetID == 0 { + anchored := false + for _, m := range members { + if len(assetsByNode[m]) > 0 { + anchored = true + break + } + } + match = !anchored + } + if !match { + continue + } + var p struct { + Body string `json:"body"` + } + _ = json.Unmarshal(d.Payload, &p) + out = append(out, map[string]any{"id": d.ID, "body": p.Body, "member_count": len(members)}) + } + return jsonResult(map[string]any{"asset_id": in.AssetID, "digests": out}) + }) +} diff --git a/agent/tools_insert.go b/agent/tools_insert.go index 66d0238..b0e738c 100644 --- a/agent/tools_insert.go +++ b/agent/tools_insert.go @@ -611,6 +611,8 @@ func splitLines(s string) []string { func (t *ToolSet) WorkerTools() []actool.CoreTool { return []actool.CoreTool{ t.listFindings(), t.listFacts(), t.nodeDetail(), + // cold-digest §6.1: a worker may hit a folded id in lineage — let it expand. + t.expandDigest(), t.expandIndex(), t.addFinding(), t.recordFact(), // cross-work retrieval t.searchAllWorkerTraces(), t.listWorkerTraces(), t.getWorkerTrace(), @@ -624,6 +626,7 @@ func (t *ToolSet) WorkerTools() []actool.CoreTool { func (t *ToolSet) MainAgentTools() []actool.CoreTool { return []actool.CoreTool{ t.graphOverview(), t.listFindings(), t.listFacts(), t.nodeDetail(), + t.expandDigest(), t.expandIndex(), // cold-digest §6.1 t.getWorkerOutput(), t.getWorkerTrace(), t.searchAllWorkerTraces(), t.addHint(), t.addIntent(), // steer_work：人可对某条正在运行的意图(work)实时注入纠偏指令（不打断、不丢进展）。 t.steerWorkTool(), diff --git a/db/constants.go b/db/constants.go index c4f7cb4..901cc5e 100644 --- a/db/constants.go +++ b/db/constants.go @@ -8,6 +8,15 @@ const ( KindFact = "fact" // a worker's exploration result/conclusion (incl. negative results), tied to its intent KindFinding = "finding" // a confirmed vulnerability (report_finding), distinct from a fact KindHint = "hint" + KindDigest = "digest" // a compressed fold of cold intents/facts (cold-digest-spec §1); lossless — members kept, restorable by id +) + +// Digest node states (kind='digest'). A digest is 'active' while it renders in +// graph_overview; major compaction retires a merged-away segment to 'superseded' +// (its covers edges repointed to the new digest) — cold-digest-spec §5.1. +const ( + StateDigestActive = "active" + StateDigestSuperseded = "superseded" ) // StateOrigin marks the task root fact (KindFact) seeded at task creation — the @@ -22,4 +31,5 @@ const ( RelDerivedFrom = "derived_from" RelYields = "yields" RelProves = "proves" + RelCovers = "covers" // digest --covers--> member (cold-digest-spec §1); source of truth for "which digest folds node X" ) diff --git a/db/digest.go b/db/digest.go new file mode 100644 index 0000000..ccdc58f --- /dev/null +++ b/db/digest.go @@ -0,0 +1,253 @@ +package db + +// cold-digest §1/§2.3/§5: persistence for cold-node compression — the round +// counter, cold_since_round stamps, content versions, digest nodes and the +// covers edges that are the source of truth for "which digest folds node X". + +import ( + "database/sql" + "encoding/json" + "fmt" + "sort" + "strings" +) + +// BumpRound advances this exploration's planner-round counter and returns the +// new value (§2.3). Called once per planner wake-up. +func (s *ExplorationStore) BumpRound() (int64, error) { + var r int64 + err := s.db.QueryRow(`UPDATE explorations SET round_no = round_no + 1 WHERE id=$1 RETURNING round_no`, s.expID).Scan(&r) + return r, err +} + +// RoundNo returns the current planner-round counter. +func (s *ExplorationStore) RoundNo() (int64, error) { + var r int64 + err := s.db.QueryRow(`SELECT round_no FROM explorations WHERE id=$1`, s.expID).Scan(&r) + return r, err +} + +// ColdStamps returns cold_since_round for every foldable node (intent/fact): +// id → *round (nil when the node is hot / unstamped). Used to compute the ≥R +// debounce and to know which stamps to set/clear this round. +func (s *ExplorationStore) ColdStamps() (map[int64]*int64, error) { + rows, err := s.db.Query(`SELECT id, cold_since_round FROM exploration_nodes + WHERE exploration_id=$1 AND kind IN ('intent','fact')`, s.expID) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]*int64{} + for rows.Next() { + var id int64 + var cs sql.NullInt64 + if err := rows.Scan(&id, &cs); err != nil { + return nil, err + } + if cs.Valid { + v := cs.Int64 + out[id] = &v + } else { + out[id] = nil + } + } + return out, rows.Err() +} + +// StampOp is one cold_since_round change: Set stamps Round; !Set clears it. +type StampOp struct { + ID int64 + Set bool + Round int64 +} + +// ApplyStampOps writes a batch of cold_since_round changes in one transaction. +func (s *ExplorationStore) ApplyStampOps(ops []StampOp) error { + if len(ops) == 0 { + return nil + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + for _, o := range ops { + if o.Set { + if _, err := tx.Exec(`UPDATE exploration_nodes SET cold_since_round=$1 WHERE id=$2 AND exploration_id=$3`, o.Round, o.ID, s.expID); err != nil { + return err + } + } else { + if _, err := tx.Exec(`UPDATE exploration_nodes SET cold_since_round=NULL WHERE id=$1 AND exploration_id=$2`, o.ID, s.expID); err != nil { + return err + } + } + } + return tx.Commit() +} + +// ContentVersions returns id → content_version for all nodes (§5.3 signature). +func (s *ExplorationStore) ContentVersions() (map[int64]int, error) { + rows, err := s.db.Query(`SELECT id, content_version FROM exploration_nodes WHERE exploration_id=$1`, s.expID) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]int{} + for rows.Next() { + var id int64 + var v int + if err := rows.Scan(&id, &v); err != nil { + return nil, err + } + out[id] = v + } + return out, rows.Err() +} + +// ActiveDigests returns the exploration's live digest nodes (state='active'), +// oldest first. +func (s *ExplorationStore) ActiveDigests() ([]*Node, error) { + rows, err := s.db.Query(`SELECT `+nodeCols+` FROM exploration_nodes + WHERE exploration_id=$1 AND kind=$2 AND state=$3 ORDER BY id`, s.expID, KindDigest, StateDigestActive) + if err != nil { + return nil, err + } + defer rows.Close() + return scanNodes(rows) +} + +// AddDigest writes one digest node and its covers edges (digest→member) in a +// single transaction. payload is the digest body + member_ids + generation + +// signature (see cold-digest §1). Returns the new digest id. +func (s *ExplorationStore) AddDigest(payload map[string]any, memberIDs []int64) (int64, error) { + raw, _ := json.Marshal(payload) + tx, err := s.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + var id int64 + if err := tx.QueryRow(` +INSERT INTO exploration_nodes(exploration_id, kind, payload, priority, state, origin) +VALUES ($1, $2, $3, 0, $4, 'compactor') RETURNING id`, + s.expID, KindDigest, string(raw), StateDigestActive).Scan(&id); err != nil { + return 0, err + } + for _, m := range memberIDs { + if m == id { + continue + } + if _, err := tx.Exec(` +INSERT INTO exploration_edges(exploration_id, src_id, rel, dst_id) VALUES ($1,$2,$3,$4) +ON CONFLICT (exploration_id, src_id, rel, dst_id) DO NOTHING`, s.expID, id, RelCovers, m); err != nil { + return 0, err + } + } + return id, tx.Commit() +} + +// CoveredMembers maps member id → covering digest id, for ACTIVE digests only +// (§6.1 point 1). A node with no entry is not currently folded. Should a member +// carry covers edges from two digests (a torn major write), the lowest digest id +// wins deterministically — callers dedupe on this. +func (s *ExplorationStore) CoveredMembers() (map[int64]int64, error) { + rows, err := s.db.Query(` +SELECT e.dst_id, e.src_id +FROM exploration_edges e +JOIN exploration_nodes d ON d.id=e.src_id AND d.exploration_id=e.exploration_id +WHERE e.exploration_id=$1 AND e.rel=$2 AND d.kind=$3 AND d.state=$4 +ORDER BY e.src_id`, s.expID, RelCovers, KindDigest, StateDigestActive) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int64]int64{} + for rows.Next() { + var member, digest int64 + if err := rows.Scan(&member, &digest); err != nil { + return nil, err + } + if _, seen := out[member]; !seen { // first (lowest digest id) wins + out[member] = digest + } + } + return out, rows.Err() +} + +// DigestMembers returns the member ids a digest covers (covers edges), sorted. +func (s *ExplorationStore) DigestMembers(digestID int64) ([]int64, error) { + rows, err := s.db.Query(`SELECT dst_id FROM exploration_edges + WHERE exploration_id=$1 AND src_id=$2 AND rel=$3`, s.expID, digestID, RelCovers) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + return ids, rows.Err() +} + +// SupersedeDigests retires digest nodes (state→superseded) AND removes their +// covers edges, atomically, so the active-coverage set (CoveredMembers) never +// double-counts a member during a major recompaction (§5.1). The digest node +// itself is kept (node_detail can still resolve it). +func (s *ExplorationStore) SupersedeDigests(ids []int64) error { + if len(ids) == 0 { + return nil + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + for _, id := range ids { + if _, err := tx.Exec(`DELETE FROM exploration_edges + WHERE exploration_id=$1 AND src_id=$2 AND rel=$3`, s.expID, id, RelCovers); err != nil { + return err + } + if _, err := tx.Exec(`UPDATE exploration_nodes SET state=$1 WHERE id=$2 AND exploration_id=$3 AND kind=$4`, + StateDigestSuperseded, id, s.expID, KindDigest); err != nil { + return err + } + } + return tx.Commit() +} + +// NodeAssets maps each of the given node ids → the asset ids it is anchored to +// (exploration_anchors). Used to group cold digests by asset (§6.2 index). +func (s *ExplorationStore) NodeAssets(ids []int64) (map[int64][]int64, error) { + out := map[int64][]int64{} + if len(ids) == 0 { + return out, nil + } + placeholders := make([]string, len(ids)) + args := make([]any, 0, len(ids)+1) + args = append(args, s.expID) + for i, id := range ids { + placeholders[i] = fmt.Sprintf("$%d", i+2) + args = append(args, id) + } + rows, err := s.db.Query(`SELECT a.node_id, a.asset_id + FROM exploration_anchors a + JOIN exploration_nodes n ON n.id=a.node_id + WHERE n.exploration_id=$1 AND a.node_id IN (`+strings.Join(placeholders, ",")+`)`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var node, asset int64 + if err := rows.Scan(&node, &asset); err != nil { + return nil, err + } + out[node] = append(out[node], asset) + } + return out, rows.Err() +} diff --git a/db/exploration.go b/db/exploration.go index a23792c..ad85ee5 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -222,9 +222,10 @@ ON CONFLICT (exploration_id, src_id, rel, dst_id) DO NOTHING`, s.expID, from, re return err } -// SetNodeState updates any node's state (never deletes). +// SetNodeState updates any node's state (never deletes). content_version bumps +// so a folded member's state flip invalidates its digest's cached body (§5.3). func (s *ExplorationStore) SetNodeState(id int64, state string) error { - _, err := s.db.Exec(`UPDATE exploration_nodes SET state=$1, blocked_reason=NULL WHERE id=$2 AND exploration_id=$3`, state, id, s.expID) + _, err := s.db.Exec(`UPDATE exploration_nodes SET state=$1, blocked_reason=NULL, content_version=content_version+1 WHERE id=$2 AND exploration_id=$3`, state, id, s.expID) return err } @@ -310,7 +311,7 @@ func (s *ExplorationStore) SetIntentState(id int64, state string) error { // terminal states stamp completed_at; reopening (back to open/running) clears it. terminal := state == "done" || state == "blocked" || state == "exhausted" || state == "stopped" _, err := s.db.Exec(`UPDATE exploration_nodes -SET state=$1, blocked_reason=NULL, completed_at = CASE WHEN $4 THEN now() ELSE NULL END +SET state=$1, blocked_reason=NULL, content_version=content_version+1, completed_at = CASE WHEN $4 THEN now() ELSE NULL END WHERE id=$2 AND exploration_id=$3 AND kind='intent'`, state, id, s.expID, terminal) return err } @@ -321,7 +322,7 @@ WHERE id=$2 AND exploration_id=$3 AND kind='intent'`, state, id, s.expID, termin func (s *ExplorationStore) CompareAndSetIntentState(id int64, expected, state string) (bool, error) { terminal := state == "done" || state == "blocked" || state == "exhausted" || state == "stopped" res, err := s.db.Exec(`UPDATE exploration_nodes -SET state=$1, blocked_reason=NULL, completed_at = CASE WHEN $5 THEN now() ELSE NULL END +SET state=$1, blocked_reason=NULL, content_version=content_version+1, completed_at = CASE WHEN $5 THEN now() ELSE NULL END WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state=$4`, state, id, s.expID, expected, terminal) if err != nil { return false, err diff --git a/db/schema.sql b/db/schema.sql index 1638074..4a25595 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -159,6 +159,10 @@ CREATE TABLE IF NOT EXISTS explorations ( created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +-- cold-digest (§2.3): per-task planner round counter — bumped once each time the +-- planner wakes and processes a round. Drives the ≥R cold-node debounce (measured in +-- this exploration's own rounds, not global node ids or wall-clock). +ALTER TABLE explorations ADD COLUMN IF NOT EXISTS round_no BIGINT NOT NULL DEFAULT 0; DROP TRIGGER IF EXISTS trg_exp_upd ON explorations; CREATE TRIGGER trg_exp_upd BEFORE UPDATE ON explorations FOR EACH ROW EXECUTE FUNCTION set_updated_at(); @@ -176,24 +180,46 @@ CREATE TABLE IF NOT EXISTS exploration_nodes ( created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), completed_at TIMESTAMPTZ, - CONSTRAINT ck_node_kind CHECK (kind IN ('begin','goal','intent','fact','finding','hint')), + CONSTRAINT ck_node_kind CHECK (kind IN ('begin','goal','intent','fact','finding','hint','digest')), CONSTRAINT ck_node_state CHECK ( (kind='begin' AND state IN ('open')) OR (kind='intent' AND state IN ('open','running','paused','done','blocked','exhausted','stopped')) OR (kind='goal' AND state IN ('open','met','abandoned')) OR (kind='fact' AND state IN ('confirmed','dismissed','origin')) OR (kind='finding' AND state IN ('confirmed','dismissed')) OR - (kind='hint' AND state IN ('active','consumed')) + (kind='hint' AND state IN ('active','consumed')) OR + (kind='digest' AND state IN ('active','superseded')) ) ); ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS blocked_reason TEXT; +-- cold-digest (§2.3/§5.3): content_version bumps on any change that could alter a +-- digest body (summary/state/confidence); cold_since_round stamps the planner round +-- a node most recently went from "has a live downstream branch" to none (NULL = hot). +ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS content_version INT NOT NULL DEFAULT 0; +ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS cold_since_round BIGINT; +-- ck_node_kind: existing installs predate the 'digest' kind — recreate to allow it. +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid='exploration_nodes'::regclass + AND conname='ck_node_kind' + AND pg_get_constraintdef(oid) NOT LIKE '%digest%' + ) THEN + ALTER TABLE exploration_nodes DROP CONSTRAINT ck_node_kind; + ALTER TABLE exploration_nodes ADD CONSTRAINT ck_node_kind + CHECK (kind IN ('begin','goal','intent','fact','finding','hint','digest')); + END IF; +END $$; +-- ck_node_state: recreate when it lacks the 'paused' (older) or 'digest' (this rev) branches. DO $$ BEGIN IF EXISTS ( SELECT 1 FROM pg_constraint WHERE conrelid='exploration_nodes'::regclass AND conname='ck_node_state' - AND pg_get_constraintdef(oid) NOT LIKE '%paused%' + AND (pg_get_constraintdef(oid) NOT LIKE '%paused%' + OR pg_get_constraintdef(oid) NOT LIKE '%superseded%') ) THEN ALTER TABLE exploration_nodes DROP CONSTRAINT ck_node_state; ALTER TABLE exploration_nodes ADD CONSTRAINT ck_node_state CHECK ( @@ -202,7 +228,8 @@ BEGIN (kind='goal' AND state IN ('open','met','abandoned')) OR (kind='fact' AND state IN ('confirmed','dismissed','origin')) OR (kind='finding' AND state IN ('confirmed','dismissed')) OR - (kind='hint' AND state IN ('active','consumed')) + (kind='hint' AND state IN ('active','consumed')) OR + (kind='digest' AND state IN ('active','superseded')) ); END IF; END $$; @@ -217,13 +244,29 @@ CREATE TABLE IF NOT EXISTS exploration_edges ( exploration_id BIGINT NOT NULL REFERENCES explorations(id) ON DELETE CASCADE, src_id BIGINT NOT NULL REFERENCES exploration_nodes(id) ON DELETE CASCADE, dst_id BIGINT NOT NULL REFERENCES exploration_nodes(id) ON DELETE CASCADE, - rel TEXT NOT NULL CHECK (rel IN ('spawns','derived_from','yields','proves')), + rel TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (exploration_id, src_id, rel, dst_id), - CONSTRAINT ck_edge_noself CHECK (src_id <> dst_id) + CONSTRAINT ck_edge_noself CHECK (src_id <> dst_id), + CONSTRAINT ck_edge_rel CHECK (rel IN ('spawns','derived_from','yields','proves','covers')) ); CREATE INDEX IF NOT EXISTS idx_expedges_src ON exploration_edges(src_id, rel); CREATE INDEX IF NOT EXISTS idx_expedges_dst ON exploration_edges(dst_id, rel); +-- cold-digest (§1): the 'covers' relation (digest→member) postdates shipped installs, +-- whose rel CHECK is an inline auto-named constraint. Find and recreate it as ck_edge_rel. +DO $$ +DECLARE cname text; +BEGIN + SELECT conname INTO cname FROM pg_constraint + WHERE conrelid='exploration_edges'::regclass AND contype='c' + AND pg_get_constraintdef(oid) LIKE '%rel%' + AND pg_get_constraintdef(oid) NOT LIKE '%covers%'; + IF cname IS NOT NULL THEN + EXECUTE 'ALTER TABLE exploration_edges DROP CONSTRAINT '||quote_ident(cname); + ALTER TABLE exploration_edges ADD CONSTRAINT ck_edge_rel + CHECK (rel IN ('spawns','derived_from','yields','proves','covers')); + END IF; +END $$; CREATE TABLE IF NOT EXISTS exploration_anchors ( node_id BIGINT NOT NULL REFERENCES exploration_nodes(id) ON DELETE CASCADE, diff --git a/server/server.go b/server/server.go index 7abf506..2d103ad 100644 --- a/server/server.go +++ b/server/server.go @@ -414,6 +414,10 @@ func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开;每轮读) pl.SetNonStreaming(nonStreamingResolver(pCfg)) // 该 profile 选非流式时走 Provider.Complete pl.SetMaxTokens(maxTokensResolver(pCfg)) // 单次回复输出上限(0 = 不发) + // cold-digest §7: background cold-node compaction, on the planner's provider/ + // model (§4 uses the running agent's model). Uses Complete (non-streaming) for + // the one-shot body summarization. + pl.SetCompactor(agent.NewCompactor(pProv, pCfg.Model)) return pl, wk } From eefd5ee2f20475c3aa8454307d6d5629b2c13cf3 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:04:17 -0400 Subject: [PATCH 2/4] =?UTF-8?q?fix(digest):=20=E5=A4=9A=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=E8=AF=84=E5=AE=A1=E4=BF=AE=E6=AD=A3=20=E2=80=94=20?= =?UTF-8?q?=E8=B5=84=E4=BA=A7=E7=B4=A2=E5=BC=95=E6=94=B6=E6=95=9B=20+=2060?= =?UTF-8?q?cap=20=E4=BF=9D=20hot=20+=20=E6=96=87=E6=A1=A3=E5=90=8C?= =?UTF-8?q?=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对照 cold-digest-spec 的对抗式评审(20 项发现)逐条核对实现，两处真实代码问题已修： - 资产索引膨胀(评审 §6.2 高危)：早前把一个 digest 落进它成员命中的【每个】asset 桶， 跨资产 digest 会复制进几十个桶、顶层条目不降反升(真实图 asset_ids 约 144 个)。改为 每个 digest 只归一个【代表资产】桶(成员锚定众数、并列取最小 id；无锚定 → 0 号兜底桶)， 条目数 ≤ digest 数。 - 60 兜底截断误砍活跃上下文(评审 §6.3 高危)：hot(live 前沿祖先链 / live 意图下新事实) 是活跃推理上下文，绝不能截。改为 hot 全保、60-cap 只约束【冷但未折】残余；冷区两侧 均衡保留。graph_overview 每轮计算 render 期 hot 集(兼用于复活检查)。 其余 16 项评审发现经核对，本实现已由设计选择规避/处理：全量扫描维护(静默变冷/传递翻转 盖章)、签名含锚节点、minor+major 统一走同一 group()(§3.1 在两处都生效、major 不会拆散 同父组)、提交时活跃度复核(不折入复活成员)、content_version 列+状态变更 bump、渲染期索引 (无 KindIndex 陈旧)、cold_digests 与索引并列(body 仍在顶层)、singleton 只进 recent。 文档 §6.2 同步上述解决口径(body 归属、一 digest 一桶、singleton 归 recent)。 --- agent/tools.go | 57 ++++++++++++++++++++++++++++--------------- agent/tools_digest.go | 34 +++++++++++++++----------- 2 files changed, 57 insertions(+), 34 deletions(-) diff --git a/agent/tools.go b/agent/tools.go index 94754fe..ad80fcb 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -425,11 +425,13 @@ func (t *ToolSet) graphOverviewData() map[string]any { // only when it is covered AND still cold. covered_members (built from hidden) lets a // parents/yields id pointing into a live fold stay resolvable (§6.5 dangling lineage). covered, _ := t.ts.CoveredMembers() + // Hot set at render time serves two §6 needs: (1) a covered member that revived + // (now hot) must reappear this round; (2) the 60-cap must never truncate hot + // (active-context) nodes — only the cold-but-unfolded region is cappable (§6.3). + // Computed every round (cheap for real graph sizes); nil map degrades safely. var hotAtRender map[int64]bool - if len(covered) > 0 { - if cg, _, err := loadColdGraph(t.ts); err == nil { - hotAtRender = cg.hotSet() - } + if cg, _, err := loadColdGraph(t.ts); err == nil { + hotAtRender = cg.hotSet() } hidden := func(id int64) bool { _, c := covered[id]; return c && !hotAtRender[id] } fr, _ := t.ts.Frontier(100) @@ -448,7 +450,8 @@ func (t *ToolSet) graphOverviewData() map[string]any { } } out["running_intents"] = compactIntents(running, parentsOf, yieldsOf) - out["recent_done_intents"] = compactIntents(recentDone, parentsOf, yieldsOf) + // recent_done_intents is finalized further below (after facts are known) so the + // 60-cap can balance the two lists and protect hot nodes (§6.3). // done_intents_total：已结束意图（done/blocked/exhausted）总数，与 recent_done_intents // 平行命名——后者只是它的最新窗口（≤15）截断视图。两键并排即自描述："看到的是 N/总数"， // 让 planner 去重时别把"没显示"当成"没派过"，无需在提示词里另行解释。 @@ -521,7 +524,9 @@ func (t *ToolSet) graphOverviewData() map[string]any { findingList = append(findingList, m) } out["finding_list"] = findingList - recentFacts := make([]map[string]any, 0, len(factNodes)) + // Build facts, split hot (active context — a fact under a live intent) from cold + // (not-yet-folded). Hidden (folded & still cold) ones are surfaced via cold_digests. + var recentFactsHot, recentFactsCold []map[string]any for _, n := range factNodes { if hidden(n.ID) { continue // in a cold_digest and still cold — surfaced via cold_digests (§6.2) @@ -538,22 +543,34 @@ func (t *ToolSet) graphOverviewData() map[string]any { m["confidence"] = c } } - recentFacts = append(recentFacts, m) - } - // §6.3 兜底截断：未折的 settled/cold 区（recent_done_intents + recent_facts）总量 > - // cap 时按最新优先截——live(open/running) 在独立字段、永不受影响。两边【均衡】保留： - // 较少的一侧全留，较多的一侧填剩余额度，各自至少保底 cap/2，绝不把某一侧饿到 0 - // （深探索链下 hot 的 settled 意图很多，天真的"意图优先"会把 facts 挤没）。正常由压缩 - // 阈值约束、不触发；触发即标注，说明压缩滞后。 - const unfoldedCap = 60 - if doneList, ok := out["recent_done_intents"].([]map[string]any); ok { - if keepDone, keepFacts, truncated := balancedCap(len(doneList), len(recentFacts), unfoldedCap); truncated { - out["recent_done_intents"] = doneList[:keepDone] - recentFacts = recentFacts[:keepFacts] - out["unfolded_truncated"] = true // 正常不触发；触发说明压缩滞后 + if hotAtRender[n.ID] { + recentFactsHot = append(recentFactsHot, m) + } else { + recentFactsCold = append(recentFactsCold, m) } } - out["recent_facts"] = recentFacts // 最近事实的 {id, summary, from_intent, confidence?}，详情/证据用 node_detail(id) + // Split the settled intents the same way: hot = ancestor of a live intent (active + // context); cold = not-yet-folded settled. + var doneHot, doneCold []*db.Node + for _, n := range recentDone { + if hotAtRender[n.ID] { + doneHot = append(doneHot, n) + } else { + doneCold = append(doneCold, n) + } + } + // §6.3 兜底截断：hot（活跃探索上下文——live 前沿的祖先链、live 意图下的新事实）【永不】 + // 被截；60-cap 只约束【冷但未折】的残余（压缩滞后时才增长）。live(open/running) 另有字段、 + // 同样不受影响。冷区两侧均衡保留：较少一侧全留、较多一侧填余额，各自保底 cap/2，绝不饿到 0。 + const unfoldedCap = 60 + keepColdDone, keepColdFacts, truncated := balancedCap(len(doneCold), len(recentFactsCold), unfoldedCap) + if truncated { + out["unfolded_truncated"] = true // 正常不触发；触发说明压缩滞后 + } + out["recent_done_intents"] = append( + compactIntents(doneHot, parentsOf, yieldsOf), + compactIntents(doneCold[:keepColdDone], parentsOf, yieldsOf)...) + out["recent_facts"] = append(recentFactsHot, recentFactsCold[:keepColdFacts]...) // {id, summary, from_intent, confidence?}；详情用 node_detail(id) // cold-digest §6.1/§6.2: the folded cold region + the per-asset directory that // collapses independent directions, plus the dangling-lineage resolver map. if cds, cidx := t.coldDigestOverview(); len(cds) > 0 { diff --git a/agent/tools_digest.go b/agent/tools_digest.go index ccf8611..e83e832 100644 --- a/agent/tools_digest.go +++ b/agent/tools_digest.go @@ -48,28 +48,34 @@ func (t *ToolSet) coldDigestOverview() (digests []map[string]any, index []map[st }) } - // index (§6.2): asset → digests touching it. A digest with no anchored asset - // falls into the 0 bucket ("(未锚定资产)"). One digest may list under >1 asset. + // index (§6.2): asset → digests. Each digest lands in EXACTLY ONE bucket — its + // representative asset = the asset anchored on the most of its members (mode; + // tie-break lowest id). This is what makes the index converge: bucketing a digest + // into every asset its members touch would duplicate it across dozens of buckets + // and blow up the top-level count instead of shrinking it (asset_ids are fine- + // grained — a real task has ~144 of them). A digest whose members anchor no asset + // falls into the 0 bucket ("(未锚定资产)"). byAsset := map[int64]map[int64]bool{} // asset id → set of digest ids assetSet := map[int64]bool{} for dID, ms := range memByDigest { - touched := map[int64]bool{} + counts := map[int64]int{} for _, m := range ms { for _, a := range assetsByNode[m] { - touched[a] = true + counts[a]++ } } - if len(touched) == 0 { - touched[0] = true + rep, best := int64(0), 0 + for a, c := range counts { + if c > best || (c == best && (rep == 0 || a < rep)) { + rep, best = a, c + } } - for a := range touched { - if byAsset[a] == nil { - byAsset[a] = map[int64]bool{} - } - byAsset[a][dID] = true - if a != 0 { - assetSet[a] = true - } + if byAsset[rep] == nil { + byAsset[rep] = map[int64]bool{} + } + byAsset[rep][dID] = true + if rep != 0 { + assetSet[rep] = true } } labels := map[int64]string{} From 775189d9160fa483c8687fc92eb0389f4bf9200b Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:09:01 -0400 Subject: [PATCH 3/4] =?UTF-8?q?feat(digest):=20related=5Ftasks=20=E5=8F=AA?= =?UTF-8?q?=E8=AF=BB=E6=8A=98=E5=8F=A0(=C2=A72=20=E8=B7=A8=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 关联/继承任务的概览现在复用其【自身的折叠视图】,与当前任务同逻辑: - relatedTaskOverviews 按 source 任务自己的 covered/hot 折出已折成员(recent_facts / recent_intent_results 跳过已折且仍冷的成员),并把该任务的 cold_digests(body)只读带出, 统一 inherited=true / source_task_id 标记。 - expand_digest 支持跨任务:先查本任务,再查直接关联任务(read-only),返回成员紧凑列表并 标继承;node_detail 已有的 GetNodeWithSources 负责成员完整还原。 - 复活检查同样生效:source 任务里已复活(转热)的被覆盖成员照常显示,不被折出。 至此设计文档 §2 跨任务折叠全部落地;仅 §2.3 墙钟兜底(任务停摆判稳)按既定作为后续。 --- agent/tools.go | 24 +++++++++++- agent/tools_digest.go | 85 ++++++++++++++++++++++++++++++++++++++----- 2 files changed, 98 insertions(+), 11 deletions(-) diff --git a/agent/tools.go b/agent/tools.go index ad80fcb..2e89656 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -810,6 +810,9 @@ func (t *ToolSet) relatedTaskOverviews() []map[string]any { out := make([]map[string]any, 0, len(sources)) for _, source := range sources { ts := source.Store + // §2 cross-task: render the source task's OWN folded view — fold out the + // members it has already folded, and surface its cold_digests read-only. + hidden := hiddenMembersFor(ts) budget := overviewTextBudget{remaining: perSourceTextBudget} item := map[string]any{ "source_task_id": source.Task.TaskID, @@ -884,6 +887,9 @@ func (t *ToolSet) relatedTaskOverviews() []map[string]any { item["recent_findings"] = recentFindings recentFacts := make([]map[string]any, 0, len(facts)) for _, fact := range facts { + if hidden(fact.ID) { + continue // folded into this source's cold_digests — shown there (§2/§6.2) + } m := inheritedMap(compactNode(fact), source.Task.TaskID) m["summary"] = budget.take(m["summary"], 400) if from := factFrom[fact.ID]; from > 0 && terminalIntent[from] { @@ -899,7 +905,14 @@ func (t *ToolSet) relatedTaskOverviews() []map[string]any { } item["recent_facts"] = recentFacts - recentDone := recentTerminalIntents(ts, relatedOverviewMaxIntentsPerTask) + recentDoneRaw := recentTerminalIntents(ts, relatedOverviewMaxIntentsPerTask) + recentDone := recentDoneRaw[:0] // in-place filter: drop this source's folded intents (§2) + for _, intent := range recentDoneRaw { + if hidden(intent.ID) { + continue + } + recentDone = append(recentDone, intent) + } for _, intent := range recentDone { intent.Inherited = true intent.SourceTaskID = source.Task.TaskID @@ -928,6 +941,15 @@ func (t *ToolSet) relatedTaskOverviews() []map[string]any { } } item["recent_intent_results"] = intentResults + // §2 cross-task: the source task's folded cold region, read-only. Members are + // resolvable via expand_digest(id)/node_detail(id), which search source tasks. + if cds := activeDigestBodies(ts); len(cds) > 0 { + for _, cd := range cds { + cd["inherited"] = true + cd["source_task_id"] = source.Task.TaskID + } + item["cold_digests"] = cds + } if statsErr == nil { item["node_stats"] = stats } diff --git a/agent/tools_digest.go b/agent/tools_digest.go index e83e832..1cf0e87 100644 --- a/agent/tools_digest.go +++ b/agent/tools_digest.go @@ -119,9 +119,10 @@ func (t *ToolSet) coldDigestOverview() (digests []map[string]any, index []map[st } // digestMemberEntry builds the compact per-member view expand_digest returns — -// same shape as recent_facts / recent_done_intents (§6.1 middle level). -func (t *ToolSet) digestMemberEntry(id int64) map[string]any { - n, _ := t.ts.GetNode(id) +// same shape as recent_facts / recent_done_intents (§6.1 middle level). store is +// the digest's OWNING store (the current task, or a read-only source task §2). +func (t *ToolSet) digestMemberEntry(store *db.ExplorationStore, id int64) map[string]any { + n, _ := store.GetNode(id) if n == nil { return map[string]any{"id": id, "missing": true} } @@ -136,6 +137,60 @@ func (t *ToolSet) digestMemberEntry(id int64) map[string]any { return m } +// activeDigestBodies returns [{id, body, member_count}] for a store's active +// digests — the folded cold region as flat bodies (§6.1). Shared by the current +// task overview and the read-only related-task overview (§2 cross-task reuse). +func activeDigestBodies(store *db.ExplorationStore) []map[string]any { + ads, err := store.ActiveDigests() + if err != nil || len(ads) == 0 { + return nil + } + out := make([]map[string]any, 0, len(ads)) + for _, d := range ads { + var p struct { + Body string `json:"body"` + } + _ = json.Unmarshal(d.Payload, &p) + ms, _ := store.DigestMembers(d.ID) + out = append(out, map[string]any{"id": d.ID, "body": p.Body, "member_count": len(ms)}) + } + return out +} + +// hiddenMembersFor returns a predicate telling whether a member is hidden (folded +// into an active digest AND still cold) in the given store — so a source task's +// overview folds exactly the way that task folds itself (§2 cross-task: "当前任务 +// 什么展示逻辑，关联任务就什么逻辑"). A revived (now hot) covered member is NOT +// hidden (§6 render-time revival check). Returns a never-hidden predicate when the +// store has no digests. +func hiddenMembersFor(store *db.ExplorationStore) func(int64) bool { + covered, err := store.CoveredMembers() + if err != nil || len(covered) == 0 { + return func(int64) bool { return false } + } + var hot map[int64]bool + if cg, _, err := loadColdGraph(store); err == nil { + hot = cg.hotSet() + } + return func(id int64) bool { _, c := covered[id]; return c && !hot[id] } +} + +// resolveDigest finds a digest node by id in the current task, else in a direct +// source task (read-only, §2). Returns the node, its owning store, and the source +// task id (0 = current task). +func (t *ToolSet) resolveDigest(id int64) (*db.Node, *db.ExplorationStore, int64) { + if n, _ := t.ts.GetNode(id); n != nil && n.Kind == db.KindDigest { + return n, t.ts, 0 + } + srcs, _ := t.ts.DirectSourceStores() + for _, s := range srcs { + if n, _ := s.Store.GetNode(id); n != nil && n.Kind == db.KindDigest { + return n, s.Store, s.Task.TaskID + } + } + return nil, nil, 0 +} + // expandDigest returns a digest's covered members as a compact list (§6.1). It is // a distinct tool from node_detail because it returns a LIST of members, not one // node's full detail. @@ -154,25 +209,35 @@ func (t *ToolSet) expandDigest() actool.CoreTool { ID int64 `json:"id"` } _ = json.Unmarshal(raw, &in) - n, _ := t.ts.GetNode(in.ID) - if n == nil || n.Kind != db.KindDigest { - return jsonResult(map[string]any{"error": fmt.Sprintf("#%d 不是 digest 节点", in.ID)}) + n, store, srcTaskID := t.resolveDigest(in.ID) + if n == nil { + return jsonResult(map[string]any{"error": fmt.Sprintf("#%d 不是 digest 节点（本任务或直接关联任务里都没找到）", in.ID)}) } var p struct { Body string `json:"body"` } _ = json.Unmarshal(n.Payload, &p) - members, _ := t.ts.DigestMembers(in.ID) + members, _ := store.DigestMembers(in.ID) list := make([]map[string]any, 0, len(members)) for _, m := range members { - list = append(list, t.digestMemberEntry(m)) + entry := t.digestMemberEntry(store, m) + if srcTaskID > 0 { // 关联任务的成员：只读，带继承标记（§2） + entry["inherited"] = true + entry["source_task_id"] = srcTaskID + } + list = append(list, entry) } - return jsonResult(map[string]any{ + out := map[string]any{ "id": in.ID, "state": n.State, // active / superseded "body": p.Body, "members": list, - }) + } + if srcTaskID > 0 { + out["inherited"] = true + out["source_task_id"] = srcTaskID + } + return jsonResult(out) }) } From 67787b5be829c873c6ffcde4e01d31f8de92a388 Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:28:39 -0400 Subject: [PATCH 4/4] =?UTF-8?q?chore(digest):=20expand=5Fdigest/expand=5Fi?= =?UTF-8?q?ndex=20=E5=8F=AA=E7=BB=99=20planner=20=E4=B8=8E=20main=20agent?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E7=BB=99=20worker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit worker 不读 graph_overview、不需要按 digest 列成员；命中折叠成员 id 时用 node_detail (直接取原节点、非折叠感知)即可还原。故从 WorkerTools 移除这两个工具，只保留在 PlannerTools 与 MainAgentTools。 --- agent/tools_insert.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/agent/tools_insert.go b/agent/tools_insert.go index b0e738c..4396921 100644 --- a/agent/tools_insert.go +++ b/agent/tools_insert.go @@ -611,8 +611,6 @@ func splitLines(s string) []string { func (t *ToolSet) WorkerTools() []actool.CoreTool { return []actool.CoreTool{ t.listFindings(), t.listFacts(), t.nodeDetail(), - // cold-digest §6.1: a worker may hit a folded id in lineage — let it expand. - t.expandDigest(), t.expandIndex(), t.addFinding(), t.recordFact(), // cross-work retrieval t.searchAllWorkerTraces(), t.listWorkerTraces(), t.getWorkerTrace(),