graph-db.js
7.56 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
/**
* Graph Database — SQLite for the Graph Agent architecture (P0 + P2)
*
* 表结构：
* system_agents 全局 / 模板 / 公司专属 agent 元数据（TEXT PRIMARY KEY = 目录名）
* org_graph_nodes 公司组织图节点（company 对某个 system_agent 的引用 + 角色名）
* org_graph_edges 公司组织图**执行流**边（P2 起：condition + edge_kind，与 work_graph_edges 同构）
*
* P2（27-p2-org-graph.md §3.2）的三处改造：
* 1. org_graph_nodes 追加 status（软删除，供 fire 用）+ reports_to_node_id（汇报关系降级为节点属性）
* 2. 全量唯一索引 idx_org_graph_nodes_company_role → 部分唯一索引 ..._active
* （fire 成 status='removed' 之后，同角色能重新 hire）
* 3. org_graph_edges 从「汇报关系」重建为「执行流边」
*
* P3（27-p3-p4-work-graph-console.md §四.D.2.2，T-P3）的两处改造：
* 1. org_graph_edges 追加可空 fail_reason —— 一个节点可以有多条 fail 出边，
* 靠边上的 fail_reason 做细分路由（精确匹配优先，无匹配退到 fail_reason IS NULL 的兜底边）。
* 2. 唯一索引 idx_org_graph_edges_dedupe_active 的键必须把 fail_reason 一起算进去。
* （R3-F2）原本以为「多条 fail 边分别指向不同 to_node_id，本来就不撞」——不成立：
* D.2.2 允许同一对 from/to 上并存「精确 fail 边 + 兜底边」（例：verify 失败因
* input_underspecified 退回 plan，其余未归类的 fail 也退回 plan），
* 旧键把这两条视为同一条，insertEdge 的 ON CONFLICT DO NOTHING 会静默吞掉兜底边。
* // 假设(R-16待定案): 归因保持 org node 粒度封装，org 层回退只退到 org node，不跨层寻址具体子节点。
* // 假设(R-16待定案): 归因节点名保留 validator-reflection。
*
* 本文件所有 DDL 都是幂等的，getDb() 可重复调用 / 进程可反复重启。
*/
const Database = require('better-sqlite3');
const path = require('path');
const fs = require('fs');
const DATA_DIR = path.resolve(__dirname, '../../data');
const DB_PATH = process.env.GRAPH_DB_PATH || path.join(DATA_DIR, 'graph.sqlite');
let db = null;
/** 表是否已有某列（SQLite 的 ALTER TABLE ADD COLUMN 对已存在的列会直接报错） */
function hasColumn(database, table, column) {
return database.pragma(`table_info(${table})`).some((c) => c.name === column);
}
/**
* 存量库迁移。新建库走上面的 CREATE TABLE 就已经是目标形态，这里全部是 no-op。
*/
function migrate(database) {
// 1. org_graph_nodes 追加两列
if (!hasColumn(database, 'org_graph_nodes', 'status')) {
database.exec(`ALTER TABLE org_graph_nodes ADD COLUMN status TEXT NOT NULL DEFAULT 'active'`);
}
if (!hasColumn(database, 'org_graph_nodes', 'reports_to_node_id')) {
// 默认值必须是 NULL，SQLite 才允许 ADD COLUMN 带 REFERENCES
database.exec(
`ALTER TABLE org_graph_nodes ADD COLUMN reports_to_node_id INTEGER REFERENCES org_graph_nodes(id)`
);
}
// 2. P1 建的全量唯一索引挡住了「fire 后重新 hire 同角色」，换成只约束 active 行的部分索引
// 注意 graph-service.js 的 ensureOrgGraphNodeIndex() 也必须同步改成建这个部分索引，
// 否则这里 DROP 掉的索引会被它重新建出来。
database.exec(`
DROP INDEX IF EXISTS idx_org_graph_nodes_company_role;
CREATE UNIQUE INDEX IF NOT EXISTS idx_org_graph_nodes_company_role_active
ON org_graph_nodes(company_id, role_name) WHERE status = 'active';
`);
// 3. 老的 org_graph_edges（relation_type 汇报关系）从建表至今零数据，直接 DROP 重建，无迁移负担
if (hasColumn(database, 'org_graph_edges', 'relation_type')) {
database.exec(`DROP TABLE org_graph_edges`);
createEdgesTable(database);
}
// 4. P3 T-P3：condition='fail' 的细分路由键（27-p3 §四.D.2.2）。
// 可空，旧行为 NULL——NULL 正好就是「兜底边」语义，存量数据不受影响。
if (!hasColumn(database, 'org_graph_edges', 'fail_reason')) {
database.exec(`ALTER TABLE org_graph_edges ADD COLUMN fail_reason TEXT`);
}
// 5. P3 R3-F2：去重键补上 fail_reason，否则同一对 from/to 上的「精确 fail 边 + 兜底边」
// 只能活一条（insertEdge 用 ON CONFLICT DO NOTHING，被吞的那条不报错）。
//
// 键里写 COALESCE(fail_reason,'') 而不是裸列：SQLite 的 UNIQUE 索引里多个 NULL 互不相等，
// 裸列会让「两条都是兜底边（fail_reason 都是 NULL）的完全重复边」不再被去重——
// 修一个吞边反而丢掉去重能力。COALESCE 把 NULL 折成 ''，NULL 之间照旧互相冲突。
// org-graph-service.insertEdge 的 ON CONFLICT 目标必须原样带上这个表达式和 WHERE。
//
// 幂等：存量库上是 P2 的四列索引（键更窄 = 约束更严），先 DROP 再按新定义建；
// 新键是旧键的超集，约束只会更松，已有数据一定建得起来。第二遍跑时 sqlite_master
// 里的定义已含 fail_reason，跳过 DROP，CREATE 走 IF NOT EXISTS，不重建索引。
const dedupeIdx = database
.prepare(`SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?`)
.get('idx_org_graph_edges_dedupe_active');
if (dedupeIdx && !/fail_reason/.test(dedupeIdx.sql || '')) {
database.exec(`DROP INDEX idx_org_graph_edges_dedupe_active`);
}
database.exec(`
CREATE UNIQUE INDEX IF NOT EXISTS idx_org_graph_edges_dedupe_active
ON org_graph_edges(company_id, from_node_id, to_node_id, condition, COALESCE(fail_reason, ''))
WHERE status = 'active';
`);
}
function createEdgesTable(database) {
database.exec(`
CREATE TABLE IF NOT EXISTS org_graph_edges (
id INTEGER PRIMARY KEY AUTOINCREMENT,
company_id TEXT NOT NULL,
from_node_id INTEGER NOT NULL REFERENCES org_graph_nodes(id),
to_node_id INTEGER NOT NULL REFERENCES org_graph_nodes(id),
condition TEXT NOT NULL CHECK (condition IN ('success','fail')),
fail_reason TEXT,
edge_kind TEXT NOT NULL DEFAULT 'forward' CHECK (edge_kind IN ('forward','backward')),
status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL
);
`);
}
function getDb() {
if (db) return db;
if (!fs.existsSync(DATA_DIR)) {
fs.mkdirSync(DATA_DIR, { recursive: true });
}
db = new Database(DB_PATH);
db.pragma('journal_mode = WAL');
db.pragma('foreign_keys = ON');
db.exec(`
CREATE TABLE IF NOT EXISTS system_agents (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
scope TEXT NOT NULL CHECK (scope IN ('public','company')),
owner_company_id TEXT,
source_kind TEXT NOT NULL CHECK (source_kind IN ('global_agent','template','company_hired')),
legacy_path TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
benchmark_score REAL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS org_graph_nodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
company_id TEXT NOT NULL,
system_agent_id TEXT NOT NULL REFERENCES system_agents(id),
role_name TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
reports_to_node_id INTEGER REFERENCES org_graph_nodes(id),
created_at TEXT NOT NULL
);
`);
createEdgesTable(db);
migrate(db);
return db;
}
module.exports = {
getDb,
};