work-graph-db.js
47.9 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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
/**
* Work Graph DB — P3 图执行引擎的数据层(T1 + T14)
*
* 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md
* §四.A.1 分层(定义层 ≠ 运行时层)
* §四.A.2 五张表(本文件照抄那段 SQL,含两个唯一索引与 graph_runs.backward_count)
* §四.A.4 两张状态转换表
* §四.A.6 幂等(UPDATE ... WHERE status = ? 的条件更新,靠受影响行数判定)
* §三.7 + §四.D.2.2 写入前五项校验
*
* 分层要点:
* 定义层 work_graphs / work_graph_nodes / work_graph_edges ← 本文件新建
* org_graph_nodes / org_graph_edges ← P2 已建,本文件只读
* 运行时层 graph_runs / graph_run_nodes / graph_run_loop_records ← 两层共用,引擎只认这个
*
* DB 文件复用 P0/P2 的 graph.sqlite:直接用 graph-db.js 的 getDb(),同一个连接、同一个进程内单例。
*
* 本文件只管「存」,不管「跑」。advanceOne / executor / 熔断计数逻辑在 graph-engine.js(T2/T7/T8)。
*/
const crypto = require('crypto');
const { getDb: getGraphDb } = require('./graph-db');
// §三.7 明确要求 work graph 的写入校验**复用** P2 这一套,不另写一份
// R-19:normalizeContracts 也从那边拿 —— 边上的 produces / needs 两层通用,
// 两处各写一份归一逻辑就是「差一个字母静默失效」的温床
const { validateGraph, normalizeContracts } = require('./org-graph-service');
// ────────────────────────────────────────────────────────────────
// ID 生成
// ────────────────────────────────────────────────────────────────
/** 前缀 + 12 位 hex。不引入 nanoid 依赖,crypto 够用(碰撞概率 2^-48 量级)。 */
function newId(prefix) {
return `${prefix}_${crypto.randomBytes(6).toString('hex')}`;
}
// ────────────────────────────────────────────────────────────────
// 建表(幂等,可反复调用)
// ────────────────────────────────────────────────────────────────
let schemaEnsured = false;
/** 表是否已有某列(org_graph_edges.fail_reason 由 T-P3 加,本文件读时要能容忍它还没加) */
function hasColumn(db, table, column) {
return db.pragma(`table_info(${table})`).some((c) => c.name === column);
}
function getDb() {
const db = getGraphDb(); // 复用 graph.sqlite 的同一连接(WAL + foreign_keys=ON 已由它设好)
if (schemaEnsured) return db;
db.exec(`
-- ── 定义层:work graph ──────────────────────────────────────
CREATE TABLE IF NOT EXISTS work_graphs (
id TEXT PRIMARY KEY,
company_id TEXT NOT NULL,
owner_org_node_id INTEGER REFERENCES org_graph_nodes(id),
planned_by TEXT NOT NULL DEFAULT 'workgraph-planning',
goal_id TEXT,
task_id TEXT,
okr_node_id TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS work_graph_nodes (
id TEXT PRIMARY KEY,
work_graph_id TEXT NOT NULL REFERENCES work_graphs(id),
node_name TEXT NOT NULL,
system_agent_id TEXT NOT NULL REFERENCES system_agents(id),
node_kind TEXT NOT NULL DEFAULT 'agent_call'
CHECK (node_kind IN ('agent_call','subgraph')),
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS work_graph_edges (
id TEXT PRIMARY KEY,
work_graph_id TEXT NOT NULL REFERENCES work_graphs(id),
from_node_id TEXT NOT NULL REFERENCES work_graph_nodes(id),
to_node_id TEXT NOT NULL REFERENCES work_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')),
created_at TEXT NOT NULL
);
-- ── 运行时层:org / work 两层共用 ───────────────────────────
CREATE TABLE IF NOT EXISTS graph_runs (
id TEXT PRIMARY KEY,
graph_kind TEXT NOT NULL CHECK (graph_kind IN ('org','work')),
graph_id TEXT NOT NULL,
company_id TEXT NOT NULL,
root_run_id TEXT NOT NULL,
parent_run_id TEXT REFERENCES graph_runs(id),
parent_node_run_id TEXT REFERENCES graph_run_nodes(id),
goal_id TEXT,
task_id TEXT,
okr_node_id TEXT,
execution_mode TEXT NOT NULL DEFAULT 'manual'
CHECK (execution_mode IN ('manual','auto')),
run_status TEXT NOT NULL DEFAULT 'planning',
pause_requested INTEGER NOT NULL DEFAULT 0,
stop_requested INTEGER NOT NULL DEFAULT 0,
backward_count INTEGER NOT NULL DEFAULT 0,
budget_json TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
-- v1 约束:同一 company 至多一个 run_status ∈ {ready,running,paused} 的 root run
CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_runs_one_active_root
ON graph_runs(company_id)
WHERE parent_run_id IS NULL AND run_status IN ('ready','running','paused');
CREATE TABLE IF NOT EXISTS graph_run_nodes (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL REFERENCES graph_runs(id),
def_node_id TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempt INTEGER NOT NULL DEFAULT 0,
agent_run_id TEXT,
child_run_id TEXT REFERENCES graph_runs(id),
output_ref TEXT,
fail_reason TEXT,
-- R-17.1 跨层节点地址:归因**指向**的那个 work graph node,形如
-- {"orgNode":"org-coder","workNode":"impl-a"}
-- 语义是「这一行的失败,根因在 orgNode 那张 work graph 的 workNode 上」。
-- 与 fail_reason 同源同轨:同一份 attempt-{n}.json 产出、同一次条件更新落库、
-- 同一条 deriveChildOutcome 透传上去。**不复用 fail_reason 承载**(它是选边的匹配键,
-- 掺进地址会让所有精确 fail 边失配),也**不放在边上**(边是定义层,
-- 而 work graph 是运行时才由 workgraph-planning 规划出来的,定义层根本写不出这个 id)。
fail_node_ref TEXT,
-- child_run_id 那张子 run 是为**本节点的哪一次 attempt** 起/复用的(R-17.1 幂等判据)。
-- 判据从「子 run 是不是还活着」改成「它服务的是不是本轮」:R-17.1 要求复用旧子 run,
-- 只看终态就会让「上一轮留下的终态」和「这一轮刚重置过」长得一样——那正是第二轮
-- 那个永久死锁的成因。存代次是唯一能把两者分开的信息。
child_run_attempt INTEGER,
-- 「留给这个节点下一轮的修复建议」。载体必须是节点行而不是 graph_run_loop_records:
-- 后者只在命中 backward 边时才有记录,走 forward fail 边 / 死路 blocked / infra_flake
-- 原地重跑这三条路径上归因节点写下的 remediation_note 就地蒸发(R4-F1)。
-- 且 loop record 是熔断账本(loop_count 按 (run_id, edge_id) 记录数算),
-- 借它存判词会直接污染熔断计数。
remediation_note TEXT,
started_at TEXT,
completed_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_run_nodes_unique
ON graph_run_nodes(run_id, def_node_id);
CREATE TABLE IF NOT EXISTS graph_run_loop_records (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL REFERENCES graph_runs(id),
edge_id TEXT NOT NULL,
from_node_run_id TEXT NOT NULL REFERENCES graph_run_nodes(id),
to_node_run_id TEXT NOT NULL REFERENCES graph_run_nodes(id),
loop_count INTEGER NOT NULL,
verdict TEXT NOT NULL
CHECK (verdict IN ('registered','fixed','not_fixed','force_terminated')),
remediation_note TEXT,
-- R-17.1:这次回退**实际落到**的跨层地址(解析后的结果,含 childRunId / nodeRunId)。
-- 与 graph_run_nodes.fail_node_ref 的分工:那边是归因的**主张**,这边是引擎的**动作**。
-- 主张可能解析不了(图换过、节点名不存在、与选中的边不是同一个 org node),
-- 那时这一列为空 = 「退化成整张 work graph 重跑」,控制台据此显示回退粒度。
target_node_ref TEXT,
created_at TEXT NOT NULL
);
-- ── R-19:产物索引(27-P4 §4.6)────────────────────────────────────
-- 补的是「同一张 work graph 内部上游到下游没有数据通路」这个断点:边一直只承载
-- 控制流(condition / edge_kind / fail_reason),而「边的输入输出」此前只是
-- 从 output_ref **推断**出来的 UI 展示、没有实体,推断的那一端是节点私有目录,
-- 下游读不到。第四次真跑就死在这:prd 节点的归因原话是
-- 「不是上游没产出,是它的产出结构上到不了我手上」。
-- 产物**实例**:运行时由**引擎**追加。不靠 agent 自觉写 —— 否则索引与实际文件必然漂移
-- (本轮反复验证:凡「靠 agent 自觉」的环节几乎全部落空过)。
-- rel_path 是**相对公司根**的路径,根由代码推导不入库:存绝对路径会烂,
-- W13 清掉的那 10 条 legacy_path 悬空行就是这个病。
CREATE TABLE IF NOT EXISTS graph_files (
id TEXT PRIMARY KEY,
company_id TEXT NOT NULL,
task_id TEXT,
rel_path TEXT NOT NULL,
kind TEXT NOT NULL,
produced_by_run_id TEXT,
produced_by_node_run_id TEXT,
created_at TEXT NOT NULL,
superseded_by TEXT
);
-- **部分**唯一:只约束「当前版本」(superseded_by IS NULL),历史行可以同路径共存。
-- 与 idx_org_graph_nodes_company_role_active 同一个模式 —— 想要「同一键只有一个活的」
-- 又要留历史,就得把条件写进索引,而不是靠删行。
CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_files_current
ON graph_files(company_id, COALESCE(task_id, ''), rel_path) WHERE superseded_by IS NULL;
CREATE INDEX IF NOT EXISTS idx_graph_files_lookup
ON graph_files(company_id, COALESCE(task_id, ''), kind);
-- 边上声明的数据**契约**:规划时写,只有 kind 没有实例。
-- edge_layer 区分两层边表 —— org_graph_edges.id 是 INTEGER、work_graph_edges.id 是 TEXT,
-- 与 graph_run_nodes.def_node_id 同一处类型不整齐(见 27-P3/P4 §四.A.2 的注),
-- 所以这里存字符串形式、不建外键,引用完整性靠服务层。
CREATE TABLE IF NOT EXISTS graph_edge_contracts (
id TEXT PRIMARY KEY,
edge_layer TEXT NOT NULL CHECK (edge_layer IN ('org','work')),
edge_id TEXT NOT NULL,
kind TEXT NOT NULL,
required INTEGER NOT NULL DEFAULT 1,
-- produces = 产出方在自己的出边上声明「我产这个 kind 交给下游」
-- needs = 消费方在自己的入边上声明「我要吃这个 kind」,产出方不必是紧邻上游
-- 这个非对称必须落库:规则 6(data-contract)靠「有人 needs 但没人 produces」判错,
-- 只存 kind 的话读回来就分不清谁该产、谁只是要吃,校验在库这一侧永远算不出来。
role TEXT NOT NULL DEFAULT 'produces' CHECK (role IN ('produces','needs')),
created_at TEXT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_edge_contracts_dedupe
ON graph_edge_contracts(edge_layer, edge_id, kind);
`);
// 存量库迁移(写法照抄 graph-db.js migrate():hasColumn 判断 + ALTER TABLE ADD COLUMN,
// 新建库走上面的 CREATE TABLE 就已经是目标形态,这里是 no-op)。
// R4-F1:graph_run_nodes 追加 remediation_note,可空,旧行 NULL = 「没有留给下一轮的话」。
if (!hasColumn(db, 'graph_run_nodes', 'remediation_note')) {
db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN remediation_note TEXT`);
}
// R-19.A2:graph_edge_contracts 追加 role。旧行一律回填 'produces' 而不是留空:
// 规则 6 里 NULL 既不算产出也不算需求,一条本该是 produces 的旧行会让下游的 needs
// 突然找不到产出方 —— 那是**误拒一张合法的图**。校验闸门宁可漏判不可错拦,
// 且 produces 是压倒性多数(forward 边基本都写它),所以回填它是更接近事实的一侧。
if (!hasColumn(db, 'graph_edge_contracts', 'role')) {
db.exec(`ALTER TABLE graph_edge_contracts ADD COLUMN role TEXT`);
db.exec(`UPDATE graph_edge_contracts SET role = 'produces' WHERE role IS NULL`);
}
// R-17.1:跨层地址 + 子 run 代次。旧行两列都是 NULL——
// fail_node_ref = NULL → 没有跨层归因,回退退化成整张图重跑(与返工前同语义)
// child_run_attempt = NULL → 代次未知,executeSubgraph 按「不是本轮的」处理(复用并重置),
// 最坏情况多重跑一轮,绝不会返回 pending 卡死。
if (!hasColumn(db, 'graph_run_nodes', 'fail_node_ref')) {
db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN fail_node_ref TEXT`);
}
if (!hasColumn(db, 'graph_run_nodes', 'child_run_attempt')) {
db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN child_run_attempt INTEGER`);
}
if (!hasColumn(db, 'graph_run_loop_records', 'target_node_ref')) {
db.exec(`ALTER TABLE graph_run_loop_records ADD COLUMN target_node_ref TEXT`);
}
schemaEnsured = true;
return db;
}
// ────────────────────────────────────────────────────────────────
// §四.A.4 状态机(模式照抄 devloop/unit-registry.js:显式转换表 + 非法转换抛错)
// ────────────────────────────────────────────────────────────────
class InvalidTransitionError extends Error {
constructor(scope, from, to, id) {
super(`Invalid ${scope} transition for ${id}: ${from} → ${to}`);
this.name = 'InvalidTransitionError';
this.scope = scope; // 'node' | 'run'
this.from = from;
this.to = to;
this.id = id;
}
}
/** graph_run_nodes.status 转换表 */
const NODE_TRANSITIONS = {
pending: ['ready', 'skipped'],
ready: ['running', 'skipped'],
running: ['success', 'fail'],
// fail → ready 是命中 backward 边或人工「重试」;fail → blocked 是死路 / 硬顶耗尽
fail: ['ready', 'blocked'],
// success 是终态,唯一出路是人工「重试」强制打回 ready
success: ['ready'],
// blocked 唯一走出方式就是人工「重试」
blocked: ['ready'],
skipped: [],
};
/** graph_runs.run_status 转换表 */
const RUN_TRANSITIONS = {
planning: ['ready'],
ready: ['running', 'stopped'],
// running → failed:§四.A.4 的转换列表里没写这一条,但 run_status 枚举含 failed,
// 且 §四.D.0 明确「子 run blocked/failed → 父节点 fail」,说明 failed 是可达态,故补上。
running: ['paused', 'blocked', 'completed', 'failed', 'stopped'],
paused: ['running', 'stopped'],
blocked: ['running', 'stopped'],
// R-17.1:completed / failed 不再是绝对终态。跨层寻址允许**上一层**的回退边点名一个
// 已经收敛的子 run 里的节点(org-validator 归因到 org-coder 的某个 work graph node,
// 而 org-coder 那张子 run 早就 completed 了),复用旧子 run 就必须能把它重新拉起来。
//
// **出口只开到 ready,不开到 running**:`ready` 的语义正是「图还在、已就绪、等启动」,
// 重开一张收敛过的 run 就是这个状态;随后走既有的 `ready → running` 启动转换,
// 于是「怎么起跑」全仓库仍然只有一条路径。给终态直开一条到 running 的捷径既多一条
// 语义重复的边,也会让「run 是被谁拉起来的」在 timeline 上分不清。
completed: ['ready'],
failed: ['ready'],
// stopped 仍是绝对终态:它是**人工主动终止**的结果,自动流程不许把它复活(见 reopenChildRun)
stopped: [],
};
function canTransition(table, from, to) {
return Boolean(table[from]) && table[from].includes(to);
}
const canTransitionNode = (from, to) => canTransition(NODE_TRANSITIONS, from, to);
const canTransitionRun = (from, to) => canTransition(RUN_TRANSITIONS, from, to);
// ────────────────────────────────────────────────────────────────
// 写入前校验:§三.7 四项(复用)+ 第五项 fail-fallback-edge(T14)
// ────────────────────────────────────────────────────────────────
/**
* 第五项校验(§四.D.2.2 末尾):任何有 fail 出边的节点,必须至少留一条
* fail_reason IS NULL 的兜底边。否则归因产出一个图上没有对应边的取值时会落到死路,
* 表现为「莫名其妙 blocked」。
*
* ⚠️ 现状:org 层**没有**复用本函数,而是在 org-graph-service.validateGraph 的「规则 5」里
* 内联实现了同一条规则——本文件 require org-graph-service(validateGraph),反向 require 会成环。
* 于是 validateWorkGraph 走的是「validateGraph 的规则 5 + 本函数」两份实现,同一条违规会报两行;
* 两份逻辑目前语义一致,但改一处漏一处就会漂移。要收敛的话把规则抽到一个无依赖的小模块里,
* 两边都 require 它(见 27-P3P4 §三.7 的风险记录)。
*
* @param {Array<{from, to, condition, failReason?}>} edges from/to 用什么标识都行(只做分组)
* @returns {Array<{rule: string, detail: string}>}
*/
function validateFailFallback(edges) {
const errors = [];
const byFrom = new Map(); // fromKey -> { hasFail: bool, hasFallback: bool }
for (const e of edges || []) {
if ((e.condition || 'success') !== 'fail') continue;
const reason = e.failReason !== undefined ? e.failReason : e.fail_reason;
const slot = byFrom.get(e.from) || { hasFail: false, hasFallback: false };
slot.hasFail = true;
if (reason === null || reason === undefined || reason === '') slot.hasFallback = true;
byFrom.set(e.from, slot);
}
for (const [from, slot] of byFrom) {
if (slot.hasFail && !slot.hasFallback) {
errors.push({
rule: 'fail-fallback-edge',
detail: `节点 ${from} 有 fail 出边,但没有一条 fail_reason 为空的兜底边,归因结果不匹配任何边时会直接死路`,
});
}
}
return errors;
}
/**
* work graph 写入前的全部校验 = §三.7 四项(acyclic / single-entry / reachable /
* backward-labeled,直接调 org-graph-service.validateGraph)+ 第五项 fail-fallback-edge。
*
* @param {Array} nodes [{nodeName, systemAgentId?, nodeKind?}] 或 ['nodeName', ...]
* @param {Array} edges [{from, to, condition?, failReason?, edgeKind?}],from/to 是 nodeName
* @returns {{ok: boolean, errors: Array<{rule: string, detail: string}>}}
*/
function validateWorkGraph(nodes, edges) {
const ns = normalizeNodes(nodes);
const es = normalizeEdges(edges);
// 四项复用:org 侧用 roleName 表达节点,work 侧叫 nodeName,此处做一次名字映射
const base = validateGraph(
ns.map((n) => ({ roleName: n.nodeName, systemAgentId: n.systemAgentId })),
es
);
const errors = base.errors.concat(validateFailFallback(es));
return { ok: errors.length === 0, errors };
}
const NODE_KINDS = ['agent_call', 'subgraph'];
function normalizeNodes(nodes) {
return (nodes || []).map((n) => {
const nodeName = typeof n === 'string' ? n : n && (n.nodeName || n.node_name);
const systemAgentId =
(typeof n === 'object' && n && (n.systemAgentId || n.system_agent_id)) || nodeName;
const nodeKind = (typeof n === 'object' && n && (n.nodeKind || n.node_kind)) || 'agent_call';
return { nodeName, systemAgentId, nodeKind };
});
}
function normalizeEdges(edges) {
return (edges || []).map((e) => ({
from: e.from,
to: e.to,
condition: e.condition || 'success',
failReason: e.failReason !== undefined ? e.failReason : e.fail_reason ?? null,
edgeKind: e.edgeKind || e.edge_kind || 'forward',
// R-19:边上声明的数据契约(createWorkGraph 插边后写 graph_edge_contracts)
contracts: normalizeContracts(e),
}));
}
// ────────────────────────────────────────────────────────────────
// 定义层 CRUD
// ────────────────────────────────────────────────────────────────
/**
* 建一张 work graph(图定义一次性写完,之后不再改——work graph 生命周期是一次性的,
* 要改就重新规划一张,所以这里没有 patch)。
*
* @param {object} p
* @param {string} p.companyId
* @param {number} [p.ownerOrgNodeId] 这张图属于哪个 org node
* @param {string} [p.plannedBy='workgraph-planning']
* @param {string} [p.goalId] @param {string} [p.taskId] @param {string} [p.okrNodeId]
* 三者基数不同不可混用,见 27-NAMING §四
* @param {Array} p.nodes 见 validateWorkGraph
* @param {Array} p.edges 见 validateWorkGraph,from/to 是 nodeName
* @returns {{ok: true, id, nodeIds: Map<string,string>} | {ok: false, errors: Array}}
*/
function createWorkGraph({
companyId,
ownerOrgNodeId = null,
plannedBy = 'workgraph-planning',
goalId = null,
taskId = null,
okrNodeId = null,
nodes = [],
edges = [],
} = {}) {
const ns = normalizeNodes(nodes);
const es = normalizeEdges(edges);
const verdict = validateWorkGraph(ns, es);
if (!verdict.ok) return { ok: false, errors: verdict.errors };
const bad = ns.find((n) => !NODE_KINDS.includes(n.nodeKind));
if (bad) {
return {
ok: false,
errors: [{ rule: 'schema', detail: `节点 ${bad.nodeName} 的 nodeKind 非法: ${bad.nodeKind}` }],
};
}
const db = getDb();
const now = new Date().toISOString();
const graphId = newId('wg');
const nodeIds = new Map();
db.transaction(() => {
db.prepare(`
INSERT INTO work_graphs (id, company_id, owner_org_node_id, planned_by, goal_id, task_id, okr_node_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`).run(graphId, companyId, ownerOrgNodeId, plannedBy, goalId, taskId, okrNodeId, now);
const insNode = db.prepare(`
INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at)
VALUES (?, ?, ?, ?, ?, ?)
`);
for (const n of ns) {
const id = newId('wgn');
nodeIds.set(n.nodeName, id);
insNode.run(id, graphId, n.nodeName, n.systemAgentId, n.nodeKind, now);
}
const insEdge = db.prepare(`
INSERT INTO work_graph_edges (id, work_graph_id, from_node_id, to_node_id, condition, fail_reason, edge_kind, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
for (const e of es) {
const edgeId = newId('wge');
insEdge.run(
edgeId,
graphId,
nodeIds.get(e.from),
nodeIds.get(e.to),
e.condition,
e.failReason,
e.edgeKind,
now
);
// R-19 落库侧(work 层):declareEdgeContract 在这一层的**唯一调用方**。
// 没有它,inboundContracts 恒返回空 → describeInputs 拼不出任何上游路径 →
// 下游节点重演第四次真跑那句「不是上游没产出,是它的产出结构上到不了我手上」。
for (const c of e.contracts) {
declareEdgeContract({
edgeLayer: 'work', edgeId, kind: c.kind, required: c.required, role: c.role,
});
}
}
})();
return { ok: true, id: graphId, nodeIds };
}
/** 读一张 work graph 定义;不存在返回 null */
function getWorkGraph(workGraphId) {
const db = getDb();
const graph = db.prepare(`SELECT * FROM work_graphs WHERE id = ?`).get(workGraphId);
if (!graph) return null;
return {
graph,
nodes: db
.prepare(`SELECT * FROM work_graph_nodes WHERE work_graph_id = ? ORDER BY created_at, id`)
.all(workGraphId),
edges: db
.prepare(`SELECT * FROM work_graph_edges WHERE work_graph_id = ? ORDER BY created_at, id`)
.all(workGraphId),
};
}
// ────────────────────────────────────────────────────────────────
// 定义层的统一读取(org / work 两层同构,运行时层只认 graph_kind + graph_id)
// ────────────────────────────────────────────────────────────────
/**
* 按 graph_kind 读定义层的节点 + 边,归一成同一形状:
* node: {defNodeId(TEXT), name, systemAgentId, nodeKind}
* edge: {id, from, to, condition, failReason, edgeKind} from/to 是 defNodeId
*
* org 侧 id 是 INTEGER,这里统一 String() 成 TEXT(§四.A.2 那条 ⚠️ 的落地处)。
*/
function readDefGraph(kind, graphId) {
const db = getDb();
if (kind === 'work') {
const g = getWorkGraph(graphId);
if (!g) return { nodes: [], edges: [] };
return {
nodes: g.nodes.map((n) => ({
defNodeId: n.id,
name: n.node_name,
systemAgentId: n.system_agent_id,
nodeKind: n.node_kind,
})),
edges: g.edges.map((e) => ({
id: e.id,
from: e.from_node_id,
to: e.to_node_id,
condition: e.condition,
failReason: e.fail_reason,
edgeKind: e.edge_kind,
})),
};
}
// kind === 'org':graphId 就是 company_id(§四.A.2)
const nodes = db
.prepare(`
SELECT id, role_name, system_agent_id FROM org_graph_nodes
WHERE company_id = ? AND status = 'active' ORDER BY id
`)
.all(graphId);
// fail_reason 列由 T-P3 追加,未加时按 NULL 读,避免本文件被 T-P3 的落地顺序卡住
const failCol = hasColumn(db, 'org_graph_edges', 'fail_reason') ? 'fail_reason' : 'NULL AS fail_reason';
const edges = db
.prepare(`
SELECT id, from_node_id, to_node_id, condition, ${failCol}, edge_kind FROM org_graph_edges
WHERE company_id = ? AND status = 'active' ORDER BY id
`)
.all(graphId);
const alive = new Set(nodes.map((n) => n.id));
return {
nodes: nodes.map((n) => ({
defNodeId: String(n.id),
name: n.role_name,
systemAgentId: n.system_agent_id,
// org graph 的节点全部是 subgraph executor(§四.D.0)
nodeKind: 'subgraph',
})),
edges: edges
.filter((e) => alive.has(e.from_node_id) && alive.has(e.to_node_id))
.map((e) => ({
id: String(e.id),
from: String(e.from_node_id),
to: String(e.to_node_id),
condition: e.condition,
failReason: e.fail_reason,
edgeKind: e.edge_kind,
})),
};
}
// ────────────────────────────────────────────────────────────────
// 运行时层 CRUD
// ────────────────────────────────────────────────────────────────
/**
* 建一次执行实例,并按定义层的节点批量铺出 graph_run_nodes(status 全 'pending')。
*
* root run:不传 parentRunId,root_run_id = 自身 id,budget_json 只在这里非空(§四.A.3)。
* 子 run: 传 parentRunId + parentNodeRunId,root_run_id 从父 run 继承(冗余存储避免上溯)。
*
* 同一 company 只允许一个 active root run(idx_graph_runs_one_active_root);
* 冲突时抛 SqliteError,调用方自己决定是拒绝还是复用已有 run。
*
* @returns {{ok: true, id, nodeRunIds: Map<defNodeId,string>}}
*/
function createRun({
graphKind,
graphId,
companyId,
parentRunId = null,
parentNodeRunId = null,
goalId = null,
taskId = null,
okrNodeId = null,
executionMode = 'manual',
runStatus = 'planning',
budget = null,
} = {}) {
if (graphKind !== 'org' && graphKind !== 'work') {
throw new Error(`graphKind 必须是 'org' 或 'work',收到: ${graphKind}`);
}
const db = getDb();
const now = new Date().toISOString();
const runId = newId('run');
const nodeRunIds = new Map();
db.transaction(() => {
let rootRunId = runId;
if (parentRunId) {
const parent = db.prepare(`SELECT root_run_id FROM graph_runs WHERE id = ?`).get(parentRunId);
if (!parent) throw new Error(`parentRunId 不存在: ${parentRunId}`);
rootRunId = parent.root_run_id;
}
db.prepare(`
INSERT INTO graph_runs (
id, graph_kind, graph_id, company_id, root_run_id, parent_run_id, parent_node_run_id,
goal_id, task_id, okr_node_id, execution_mode, run_status, budget_json, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
runId, graphKind, graphId, companyId, rootRunId, parentRunId, parentNodeRunId,
goalId, taskId, okrNodeId, executionMode, runStatus,
// budget 只挂 root run:子 run 通过 root_run_id 扣同一个池
parentRunId ? null : budget && JSON.stringify(budget),
now, now
);
const ins = db.prepare(`
INSERT INTO graph_run_nodes (id, run_id, def_node_id, status, created_at, updated_at)
VALUES (?, ?, ?, 'pending', ?, ?)
`);
for (const n of readDefGraph(graphKind, graphId).nodes) {
const id = newId('rn');
nodeRunIds.set(n.defNodeId, id);
ins.run(id, runId, n.defNodeId, now, now);
}
})();
return { ok: true, id: runId, nodeRunIds };
}
function getRun(runId) {
return getDb().prepare(`SELECT * FROM graph_runs WHERE id = ?`).get(runId) || null;
}
/**
* 取一次 run 的全量状态:run 元信息 + 每个节点(定义 + 运行时合并)+ 边(定义层原样)。
* §四.B / T6 的 `GET /api/graph-runs/:id` 直接吐这个形状。
*
* @returns {{run, nodes: Array, edges: Array} | null}
*/
function getRunState(runId) {
const db = getDb();
const run = getRun(runId);
if (!run) return null;
const def = readDefGraph(run.graph_kind, run.graph_id);
const defById = new Map(def.nodes.map((n) => [n.defNodeId, n]));
const rows = db
.prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? ORDER BY created_at, id`)
.all(runId);
return {
run,
nodes: rows.map((r) => {
const d = defById.get(r.def_node_id) || {};
return {
id: r.id,
defNodeId: r.def_node_id,
name: d.name || null,
systemAgentId: d.systemAgentId || null,
nodeKind: d.nodeKind || null,
status: r.status,
attempt: r.attempt,
agentRunId: r.agent_run_id,
childRunId: r.child_run_id,
outputRef: r.output_ref,
failReason: r.fail_reason,
// R-17.1 的两列同样要读回来(控制台 §四.B.3 用它们表达层级语义):
// failNodeRef = 归因指向的跨层地址;childRunAttempt = 这张子 run 服务的是哪一轮
failNodeRef: r.fail_node_ref,
childRunAttempt: r.child_run_attempt,
// 落库了就必须读回来:字段在「产生 → 传递 → 落库 → 读回」四环里漏掉任何一环
// 都等于没有这个字段(P3 前几轮反复踩的就是这个)
remediationNote: r.remediation_note,
startedAt: r.started_at,
completedAt: r.completed_at,
};
}),
edges: def.edges,
};
}
/** 某次 run 的运行时节点(按定义节点 id 索引),供引擎选边时用 */
function getRunNodeByDefId(runId, defNodeId) {
return (
getDb()
.prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND def_node_id = ?`)
.get(runId, String(defNodeId)) || null
);
}
function getRunNode(nodeRunId) {
return getDb().prepare(`SELECT * FROM graph_run_nodes WHERE id = ?`).get(nodeRunId) || null;
}
/** 允许被条件更新一并写入的运行时列(白名单,避免拼 SQL 时把任意键塞进来) */
const NODE_PATCH_COLUMNS = {
attempt: 'attempt',
agentRunId: 'agent_run_id',
childRunId: 'child_run_id',
outputRef: 'output_ref',
failReason: 'fail_reason',
// 跟 fail_reason 走**同一次**条件更新:地址与根因是同一份归因产出的两半,
// 分成两次写就会出现「有 fail_reason 没地址」的中间态,回退恰好读在那一刻就退化了
failNodeRef: 'fail_node_ref',
childRunAttempt: 'child_run_attempt',
// 跟着状态转换一起落库:写在这里而不是另开一个 setter,是为了让「改状态 + 写判词」
// 共用 updateNodeStatus 那一次条件更新,不出现「状态改了但判词没落」的第二条路径
remediationNote: 'remediation_note',
startedAt: 'started_at',
completedAt: 'completed_at',
};
/**
* 条件更新节点状态(§四.A.6 幂等的落地点)。
*
* UPDATE ... WHERE id = ? AND status = <from>,返回受影响行数:
* 1 → 本次转换是我做成的
* 0 → 已经被别人(自动驱动器 / 人工点击)拿走了,调用方原样返回当前状态即可,不报错
* 不加分布式锁——SQLite 行级事务 + 条件更新就够。
*
* from 缺省时从当前行读,此时不具备幂等语义(仅用于确定单线程的场景)。
* 转换本身先过 NODE_TRANSITIONS,非法直接抛 InvalidTransitionError。
*
* @param {string} nodeRunId
* @param {string|null} from 期望的当前状态
* @param {string} to
* @param {object} [patch] NODE_PATCH_COLUMNS 里的键
* @returns {number} 受影响行数(0 或 1)
*/
function updateNodeStatus(nodeRunId, from, to, patch = {}) {
const db = getDb();
const cur = getRunNode(nodeRunId);
if (!cur) throw new Error(`graph_run_nodes 不存在: ${nodeRunId}`);
const expected = from === null || from === undefined ? cur.status : from;
if (!canTransitionNode(expected, to)) {
throw new InvalidTransitionError('node', expected, to, nodeRunId);
}
const sets = ['status = @to', 'updated_at = @now'];
const params = { id: nodeRunId, to, from: expected, now: new Date().toISOString() };
for (const [key, col] of Object.entries(NODE_PATCH_COLUMNS)) {
if (patch[key] !== undefined) {
sets.push(`${col} = @${key}`);
params[key] = patch[key];
}
}
return db
.prepare(`UPDATE graph_run_nodes SET ${sets.join(', ')} WHERE id = @id AND status = @from`)
.run(params).changes;
}
/** attempt + 1(回退时用,§四.D.2.1 第 3 步);返回受影响行数 */
function bumpNodeAttempt(nodeRunId) {
return getDb()
.prepare(`UPDATE graph_run_nodes SET attempt = attempt + 1, updated_at = ? WHERE id = ?`)
.run(new Date().toISOString(), nodeRunId).changes;
}
const RUN_PATCH_COLUMNS = {
pauseRequested: 'pause_requested',
stopRequested: 'stop_requested',
executionMode: 'execution_mode',
budgetJson: 'budget_json',
};
/**
* 条件更新 run 状态,语义与 updateNodeStatus 完全一致(同样返回受影响行数供幂等判定)。
* @returns {number}
*/
function updateRunStatus(runId, from, to, patch = {}) {
const db = getDb();
const cur = getRun(runId);
if (!cur) throw new Error(`graph_runs 不存在: ${runId}`);
const expected = from === null || from === undefined ? cur.run_status : from;
if (!canTransitionRun(expected, to)) {
throw new InvalidTransitionError('run', expected, to, runId);
}
const sets = ['run_status = @to', 'updated_at = @now'];
const params = { id: runId, to, from: expected, now: new Date().toISOString() };
for (const [key, col] of Object.entries(RUN_PATCH_COLUMNS)) {
if (patch[key] !== undefined) {
sets.push(`${col} = @${key}`);
params[key] = patch[key];
}
}
return db
.prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id AND run_status = @from`)
.run(params).changes;
}
/** 只改软信号,不动 run_status(「暂停」/「停止」是下一个节点边界才生效的软信号) */
function setRunFlags(runId, { pauseRequested, stopRequested } = {}) {
const sets = ['updated_at = @now'];
const params = { id: runId, now: new Date().toISOString() };
if (pauseRequested !== undefined) {
sets.push('pause_requested = @pause');
params.pause = pauseRequested ? 1 : 0;
}
if (stopRequested !== undefined) {
sets.push('stop_requested = @stop');
params.stop = stopRequested ? 1 : 0;
}
return getDb()
.prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id`)
.run(params).changes;
}
/** run 级回退总数 +1(§四.D.2.1 双层硬顶的外层),返回加完之后的值 */
function bumpBackwardCount(runId) {
const db = getDb();
db.prepare(`UPDATE graph_runs SET backward_count = backward_count + 1, updated_at = ? WHERE id = ?`)
.run(new Date().toISOString(), runId);
const row = db.prepare(`SELECT backward_count FROM graph_runs WHERE id = ?`).get(runId);
return row ? row.backward_count : null;
}
/** 某个 run 的直接子 run(级联暂停/停止用) */
function listChildRuns(runId) {
return getDb().prepare(`SELECT * FROM graph_runs WHERE parent_run_id = ? ORDER BY created_at`).all(runId);
}
/** 同一棵 run 树的全部 run(按 root_run_id,预算扣减与三层级联都按这个口径) */
function listRunTree(rootRunId) {
return getDb()
.prepare(`SELECT * FROM graph_runs WHERE root_run_id = ? ORDER BY created_at`)
.all(rootRunId);
}
/** 一家公司的 run 列表(T6 的 GET /api/company/:cid/graph-runs) */
function listCompanyRuns(companyId, { rootOnly = false } = {}) {
return getDb()
.prepare(`
SELECT * FROM graph_runs
WHERE company_id = ? ${rootOnly ? 'AND parent_run_id IS NULL' : ''}
ORDER BY created_at DESC
`)
.all(companyId);
}
// ────────────────────────────────────────────────────────────────
// 回退记录(§四.D.2.1 的备案账本,判词由 T7 写)
// ────────────────────────────────────────────────────────────────
/** 这条边在本 run 上已经退过几次(loop_count = 该值 + 1) */
function countLoopRecords(runId, edgeId) {
return getDb()
.prepare(`SELECT COUNT(*) AS n FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ?`)
.get(runId, String(edgeId)).n;
}
/**
* 备案一次回退(§四.D.2.1 第 1 步)。loop_count 缺省按已有记录数 + 1 自动算。
* @returns {{id, loopCount}}
*/
function insertLoopRecord({
runId,
edgeId,
fromNodeRunId,
toNodeRunId,
loopCount,
verdict = 'registered',
remediationNote = null,
targetNodeRef = null,
} = {}) {
const db = getDb();
const id = newId('lr');
const n = loopCount === undefined || loopCount === null ? countLoopRecords(runId, edgeId) + 1 : loopCount;
db.prepare(`
INSERT INTO graph_run_loop_records
(id, run_id, edge_id, from_node_run_id, to_node_run_id, loop_count, verdict,
remediation_note, target_node_ref, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(id, runId, String(edgeId), fromNodeRunId, toNodeRunId, n, verdict, remediationNote,
targetNodeRef, new Date().toISOString());
return { id, loopCount: n };
}
/** 改判词(registered → fixed / not_fixed / force_terminated),返回受影响行数 */
function updateLoopVerdict(recordId, verdict, remediationNote) {
const sets = ['verdict = @verdict'];
const params = { id: recordId, verdict };
if (remediationNote !== undefined) {
sets.push('remediation_note = @note');
params.note = remediationNote;
}
return getDb()
.prepare(`UPDATE graph_run_loop_records SET ${sets.join(', ')} WHERE id = @id`)
.run(params).changes;
}
/** 本 run 的回退记录,最新在前(控制台「循环」tab + 下一轮读 remediation_note 用) */
function listLoopRecords(runId, { edgeId = null } = {}) {
const db = getDb();
return edgeId === null
? db.prepare(`SELECT * FROM graph_run_loop_records WHERE run_id = ? ORDER BY created_at DESC`).all(runId)
: db
.prepare(`
SELECT * FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ?
ORDER BY created_at DESC
`)
.all(runId, String(edgeId));
}
// ────────────────────────────────────────────────────────────────
// R-19 产物索引(27-P4 §4.6)
//
// 分工与 output_ref 不重复:
// output_ref 节点级指针 —— 「我这次跑出了什么」,一对一挂在 graph_run_nodes 行上
// graph_files 公司级索引 —— 「这家公司有哪些产物、谁能用」,聚合视图,由前者派生
// 所以 graph_files 必须可从节点行重建(rebuildFilesFromNodes),索引坏了不至于丢产物。
// ────────────────────────────────────────────────────────────────
/**
* 记一条产物实例。**只由引擎调用**,不给 agent 自己写 ——
* 否则索引与实际文件必然漂移(本轮反复验证过这一点)。
*
* 同一 (company, task, relPath) 重复产出时:把旧行标 superseded_by 指向新行,
* 而不是堆一堆同路径的行。这样「最新的那份」是唯一 superseded_by IS NULL 的行,
* 历史仍可追溯。
*/
function recordFile({ companyId, taskId = null, relPath, kind, runId = null, nodeRunId = null }) {
if (!companyId || !relPath || !kind) throw new Error('recordFile 需要 companyId / relPath / kind');
const db = getDb();
const now = new Date().toISOString();
const id = newId('gf');
const tx = db.transaction(() => {
const prev = db
.prepare(`
SELECT id FROM graph_files
WHERE company_id = ? AND COALESCE(task_id, '') = ? AND rel_path = ?
AND superseded_by IS NULL
`)
.get(companyId, taskId || '', relPath);
if (prev) {
// 同路径再产出:先把旧行指向新行(腾出部分唯一索引的位置),再插新行。
// 顺序不能反 —— 反了会撞 idx_graph_files_current。
db.prepare(`UPDATE graph_files SET superseded_by = ? WHERE id = ?`).run(id, prev.id);
}
db.prepare(`
INSERT INTO graph_files
(id, company_id, task_id, rel_path, kind,
produced_by_run_id, produced_by_node_run_id, created_at, superseded_by)
VALUES (@id, @company_id, @task_id, @rel_path, @kind, @run_id, @node_run_id, @now, NULL)
`).run({
id, company_id: companyId, task_id: taskId, rel_path: relPath, kind,
run_id: runId, node_run_id: nodeRunId, now,
});
});
tx();
return id;
}
/** 按 kind 查这家公司/这个 task 下最新的产物实例(superseded_by IS NULL) */
function findFiles({ companyId, taskId = null, kinds = null }) {
const db = getDb();
const rows = db
.prepare(`
SELECT * FROM graph_files
WHERE company_id = ? AND COALESCE(task_id, '') = ? AND superseded_by IS NULL
ORDER BY created_at
`)
.all(companyId, taskId || '');
if (!kinds || !kinds.length) return rows;
const want = new Set(kinds);
return rows.filter((r) => want.has(r.kind));
}
/** 在边上声明「这条边要传什么 kind」。规划时写,只有 kind 没有实例。 */
function declareEdgeContract({ edgeLayer, edgeId, kind, required = true, role = 'produces' }) {
if (!['org', 'work'].includes(edgeLayer)) throw new Error(`edgeLayer 非法: ${edgeLayer}`);
if (!edgeId || !kind) throw new Error('declareEdgeContract 需要 edgeId / kind');
// role 默认 produces:出边声明产出是压倒性多数,needs 必须显式写。
// 去重键仍是 (layer, edge, kind) 而不含 role —— 同一条边对同一个 kind 既声明 produces
// 又声明 needs 是自相矛盾的输入(「我产它」和「我要别人产它」),
// normalizeContracts 的遍历顺序让 produces 先落库、矛盾的那条被 DO NOTHING 吞掉,
// 即保留更强的那个断言。
getDb()
.prepare(`
INSERT INTO graph_edge_contracts (id, edge_layer, edge_id, kind, required, role, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(edge_layer, edge_id, kind) DO NOTHING
`)
.run(
newId('gec'), edgeLayer, String(edgeId), kind, required ? 1 : 0,
role === 'needs' ? 'needs' : 'produces', new Date().toISOString()
);
}
/** 某条边上声明的契约 */
function edgeContracts(edgeLayer, edgeId) {
return getDb()
.prepare(`SELECT * FROM graph_edge_contracts WHERE edge_layer = ? AND edge_id = ?`)
.all(edgeLayer, String(edgeId));
}
/**
* 某个节点的**入边**上声明了哪些 kind —— 这是「这个节点需要什么输入」的答案。
* 只看 forward 入边:backward 入边是回退路径,不是数据来源。
*/
function inboundContracts(edgeLayer, edges, defNodeId) {
const inbound = (edges || []).filter(
(e) => String(e.to) === String(defNodeId) && (e.edgeKind || 'forward') !== 'backward'
);
const out = [];
for (const e of inbound) {
for (const c of edgeContracts(edgeLayer, e.id)) out.push({ ...c, fromNodeId: e.from });
}
return out;
}
/**
* 从 graph_run_nodes 重建索引(索引损坏时的兜底,类比 okr 的 _rebuildIndex)。
* 只能恢复「节点产出过什么目录」这一层,恢复不了 kind —— 所以 kind 统一记 'artifact',
* 这是有损重建,够用来定位文件,不够还原语义。
*/
function rebuildFilesFromNodes(companyId) {
const db = getDb();
const rows = db
.prepare(`
SELECT n.id AS node_run_id, n.run_id, n.output_ref, r.task_id
FROM graph_run_nodes n JOIN graph_runs r ON r.id = n.run_id
WHERE r.company_id = ? AND n.output_ref IS NOT NULL
`)
.all(companyId);
let n = 0;
for (const r of rows) {
recordFile({
companyId, taskId: r.task_id, relPath: r.output_ref, kind: 'artifact',
runId: r.run_id, nodeRunId: r.node_run_id,
});
n += 1;
}
return n;
}
module.exports = {
getDb,
newId,
// 校验
validateWorkGraph,
validateFailFallback,
// 定义层
createWorkGraph,
getWorkGraph,
readDefGraph,
// 运行时层
createRun,
getRun,
getRunState,
getRunNode,
getRunNodeByDefId,
updateNodeStatus,
bumpNodeAttempt,
updateRunStatus,
setRunFlags,
bumpBackwardCount,
listChildRuns,
listRunTree,
listCompanyRuns,
// R-19 产物索引(27-P4 §4.6)
recordFile,
findFiles,
declareEdgeContract,
edgeContracts,
inboundContracts,
rebuildFilesFromNodes,
// 回退记录
countLoopRecords,
insertLoopRecord,
updateLoopVerdict,
listLoopRecords,
// 状态机
NODE_TRANSITIONS,
RUN_TRANSITIONS,
canTransitionNode,
canTransitionRun,
InvalidTransitionError,
};