feat(digest): 探索图冷节点压缩(cold-digest)

把老且长期不活跃的意图/事实折叠成 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 孤立冷点(对齐设计文档实测校验)。
This commit is contained in:
Autumn-27 committed 2026-09-09 11:52:15 -04:00
1 parent 801160930b
commit 4f088e519b
13 files changed
+1720 -15

No files matched your search

+10
View File
@@ -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"
)
+253
View File
@@ -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()
}
+5 -4
View File
@@ -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
+49 -6
View File
@@ -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,