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 }