wist-contracts 0.9.2

Versioned contract and schema objects shared by the wist edge and center components
Documentation
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
//! **工作授权快照**:网关授权(grant)、agentd 拉取(`WorkGrant`)并确认。
//!
//! ## 为什么是「快照」而不是「指令流」
//!
//! 常驻工作(`StandingWork`)表达的是**期望状态**:某个采集面应当持续执行什么。
//! 期望状态天然是幂等的(同一份重复拉取不产生副作用),断了网、重启了进程,
//! 再拉一次就回到期望 —— 不必让网关记住「上次推到哪条」。
//! 这跟 `PollControlCommands` 的长轮询指令流是两种东西:那条要求**不重放**,这条要求**可重拉**。
//!
//! 一次性工作(`OneShotWork`)是命令式的,但它也搭这份快照回来:
//! 决定「现在还该做吗」的是它的状态(未了结才出现在快照里),不是「推没推过」。
//!
//! ## 为什么在契约 crate
//!
//! 这是**两侧都要解析**的字节:网关写授权、agentd 读并执行。各写一份结构体,
//! 迟早出现「网关发了 `plan_version`、agentd 读的是 `version`」这种只能在真机联调时才发现的分叉。
//! 与 `DiscoveryAspectPolicySet` 同一个理由。
//!
//! ## 与本地保护暂停的区别
//!
//! agentd 自己也会暂停(如 `spool over limit`),那是**本地保护**:自动、临时、本机可见。
//! 这里下发的 `paused` 是**授权层状态**:人工、持久、跨重启。两者在观测里必须能分辨,
//! 不要合并成一个 paused —— 合并之后「谁把采集停了」就再也说不清。

use serde::{Deserialize, Serialize};

/// 常驻工作的状态取值(对应模型 `StandingWork.status`)。
///
/// `superseded` 与 `revoked` 都不出现在 `WorkGrant.standing` 里:前者是「被新版本取代」,
/// 后者是「授权被撤」。两者都要留痕,所以还是要有状态而不是删行。
pub const STANDING_WORK_STATUSES: [&str; 4] = ["active", "paused", "superseded", "revoked"];

/// 一次性工作的状态取值(对应模型 `OneShotWork.status`)。
pub const ONE_SHOT_WORK_STATUSES: [&str; 9] = [
    "dispatched",
    "accepted",
    "running",
    "paused",
    "succeeded",
    "failed",
    "timed_out",
    "canceled",
    "expired",
];

/// Agent **允许上报**的一次性工作状态([`ONE_SHOT_WORK_STATUSES`] 的真子集)。
///
/// 为什么不是全集:`dispatched` 是网关自己写的(派下去那一刻),`paused` / `canceled` / `expired`
/// 归**控制面与期限**管(运维暂停撤回、或过了截止)—— agent 无权把它们写回去。
/// 剩下的问题只有一种:「这件活做完了没有」,答案就这三种。
pub const AGENT_REPORTABLE_WORK_STATUSES: [&str; 3] = ["running", "succeeded", "failed"];

/// 一次性工作的**终态**:到了这几个状态就了结了,不再出现在快照里。
pub const ONE_SHOT_TERMINAL_STATUSES: [&str; 5] =
    ["succeeded", "failed", "timed_out", "canceled", "expired"];

/// 工作类型:常驻(持续到被替换或撤回)与一次性(有期限与终态)。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
#[jumo(kind = "state", domain = "Control", module = "Control.Agent.Work")]
#[serde(rename_all = "PascalCase")]
pub enum WorkKind {
    /// 常驻工作:按**采集面**授权,一个面一份。
    Standing,
    /// 一次性工作:按**动作**授权。
    OneShot,
}

/// 常驻工作:网关声明该 Agent 应当持续执行的工作。
///
/// 粒度是**采集面**(一个面 = 一份工作),不是 capability:一份 `collect_logs` 会裹住十几个面,
/// 那样「按面暂停 / 限流 / 审计」全都无从下手。面由模板的 `family_scope` 展开而来,
/// **不携带 capability** —— 该面由哪个采集器承接,由 `spec` 里的单元各自决定。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
#[serde(deny_unknown_fields)]
pub struct StandingWork {
    pub work_id: String,
    pub agent_id: String,
    /// 采集面(`CollectionFamily`)。
    pub family: String,
    /// 工作参数:由采集目录的条目组合而成(**不是自由文本**),随 `plan_version` 整体替换。
    pub spec: String,
    /// 本工作按哪一版目录展开:目录换版**不追改**已授权工作(要跟新版得走新提案 + 审定)。
    pub catalog_version: i64,
    /// 生效依据:指向已批准的提案(人工直填 spec 时为空)。
    #[serde(default)]
    pub proposal_id: Option<String>,
    /// 期望版本:网关每次改动 +1;agentd 回报实际版本,与它比对即得漂移。
    pub plan_version: i64,
    pub effective_from: String,
    /// 见 [`STANDING_WORK_STATUSES`]。
    pub status: String,
    pub updated_by: String,
    pub updated_at: String,
}

/// 一次性工作:有计划开始时间与期限,有明确终态。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
#[serde(deny_unknown_fields)]
pub struct OneShotWork {
    pub work_id: String,
    pub agent_id: String,
    /// 动作面:upgrade / snapshot / exec / ...(执行载体是 Reporting 域的 ActionPlan)。
    pub action: String,
    pub spec: String,
    /// 计划开始时间:到点前不应执行(与「立即派发」区分开)。
    pub scheduled_at: String,
    /// 绝对截止:业务要求的时间点,**暂停也照走**(不由暂停顺延)。
    pub deadline_at: String,
    /// 执行预算(秒):**只在实际执行时消耗**,暂停期间不计。
    pub timeout_seconds: i64,
    /// 可中断性:只有可中断的动作才允许运行中暂停。
    pub interruptible: bool,
    /// 见 [`ONE_SHOT_WORK_STATUSES`]。
    pub status: String,
    /// 当前暂停的起点(运行中暂停才有;恢复后清空)。
    #[serde(default)]
    pub paused_at: Option<String>,
    /// 累计暂停时长(秒):用于审计与「预算未被暂停消耗」的核对。
    pub paused_total_seconds: i64,
    /// 步级断点:已完成步骤保留、`current_step` 在恢复时重做。
    #[serde(default)]
    pub current_step: Option<String>,
    #[serde(default)]
    pub completed_steps: Vec<String>,
    /// 已尝试次数(含恢复后的重做)。
    pub attempt: i64,
    pub issued_by: String,
    pub issued_at: String,
}

impl OneShotWork {
    /// 是否**未了结**(快照里只带未了结的活)。
    pub fn is_outstanding(&self) -> bool {
        !ONE_SHOT_TERMINAL_STATUSES.contains(&self.status.as_str())
    }

    /// 是否允许在运行期间暂停。
    ///
    /// 两条都要满足:动作声明了 `interruptible`,且此刻确实在做(`accepted`/`running`)。
    /// 不可中断的动作**拒绝**而不是「尽力暂停」—— 挂起半个升级进程比不暂停更危险。
    pub fn can_pause(&self) -> bool {
        self.interruptible && matches!(self.status.as_str(), "accepted" | "running")
    }
}

/// 管理面授权或撤回工作的回执:一份工作一次。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
#[serde(deny_unknown_fields)]
pub struct WorkReceipt {
    pub work_id: String,
    pub agent_id: String,
    pub work_kind: WorkKind,
    /// accepted | paused | resumed | revoked | rejected。
    pub status: String,
    pub plan_version: i64,
    pub created_at: String,
}

// ─────────────────────────────────────────────────────────────────────────────
// 工作参数(`StandingWork.spec` 的内容)
// ─────────────────────────────────────────────────────────────────────────────
//
// 模型里 `StandingWork.spec` 是一个 `String`,语义是「工作参数:由采集目录的条目
// 组合而成,**不是自由文本**」。这里的三个类型就是它的**编码**:一串已物化的采集单元。
//
// 为什么必须带上来源与规则标识,而不是只给一串 `unit_id`:agentd 拿到工作要能
// **直接照做**。单元的采集来源(`sources`)与数据面规则标识(`rule_ref`)本来都只
// 存在于网关的采集目录里,只发 id 等于发了一张自己去不了的地址 —— 于是要么再去网关
// 拉一次目录(多一条必须鉴权的路径),要么两边各维护一份目录(必然漂移)。
//
// 为什么不把它们塞成 `StandingWork` 的字段:那会把「工作」与「内容目录」的边界糊掉;
// 模型里这个字段就是**不透明的工作参数**,保持它不透明是两侧能独立演进的前提。

/// 一份常驻工作的工作参数。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkSpec {
    /// 本工作包含的采集单元(已按该机器的事实裁剪过)。
    #[serde(default)]
    pub units: Vec<WorkSpecUnit>,
}

/// 一个已物化的采集单元:只说「采什么、怎么落地」,不复述策展元信息。
///
/// 刻意不带 `status` / `match` / `catalog_version`:那些是网关策展与裁剪的输入,
/// 工作一旦发出去就已经裁剪完了。带上它们会让 agentd 有「再判断一次」的空间,
/// 而 agentd 不是第二个策展器。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkSpecUnit {
    pub unit_id: String,
    /// `collect_logs` | `collect_metrics`:该单元由哪类采集器承接。
    pub capability: String,
    /// 数据面 rule/oml 标识(空 = 该单元还没接上规则,理论不会入工作)。
    #[serde(default)]
    pub rule_ref: String,
    /// `none` | `root` | `fda`:要采到这东西得有什么权限 —— 缺权限时应**说清缺什么**,
    /// 而不是安静地采不到。
    #[serde(default)]
    pub requires_privilege: String,
    #[serde(default)]
    pub sources: Vec<WorkSpecSource>,
}

/// 采集来源:`kind` ∈ `FileGlob` | `Exporter` | `UnifiedLogPredicate` | `MetricInterval`。
///
/// 与模型 `CollectionSource` 同形:`target` 的含义由 `kind` 决定
/// (路径通配 / 导出器标识 / 谓词 / 周期)。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkSpecSource {
    pub kind: String,
    pub target: String,
    /// 怎么读这条来源:`none`(一行一条)| `indented`(行首缩进是上一条的续行)。
    ///
    /// 只对可 tail 的 `FileGlob` 有意义,其余 kind 恒为 `none`。**不是**"要不要多行"这个偏好,
    /// 而是这条日志的**固有格式**:说错了要么把多行记录拆散、要么把独立记录粘成一条。
    #[serde(default = "default_multiline")]
    pub multiline: String,
    /// 从 `target` 匹配集合里剔掉的**绝对路径 glob**(默认空)。
    ///
    /// 只对 `FileGlob` 有意义,按**展开后的完整路径**匹配(不是文件名)。为什么需要它:
    /// 容器日志的选源根是 `/var/log/containers/*.log`,而目录里同时有不该采的容器
    /// (sidecar / init / 平台自身的 pod);没有排除项就只能把业务服务名写进 pattern,
    /// 而服务名会随集群漂移。
    ///
    /// 空数组**不序列化**:旧 agent 的 `WorkSpecSource` 是 `deny_unknown_fields` 的,
    /// 发一个空字段过去只会给旧机队添一种没必要的解析失败。
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub exclude_globs: Vec<String>,
    /// 容器选源的结构化过滤器(默认空)。只对 `FileGlob` 有意义。
    ///
    /// 空(两个列表都空)**不序列化**:与 `exclude_globs` 同一个理由 —— 旧 agent 的
    /// `WorkSpecSource` 是 `deny_unknown_fields` 的,发一个空字段过去只会给旧机队
    /// 添一种没必要的解析失败。
    ///
    /// ⚠ 非空时**要求 agent 支持**(与 glob 单元同一条纪律):能力上报(D7)未落地前,
    /// 目录里带 filters 的单元必须保持 `draft`,机队升级验证后再单独翻 `active`。
    #[serde(default, skip_serializing_if = "ContainerFilters::is_empty")]
    pub filters: ContainerFilters,
}

/// 容器日志的选源根(写死,D4)。
///
/// `/var/log/pods` 下有 `0.log` / `1.log` 与重启序号歧义;`/var/log/containers` 里
/// 每个活容器恰好一条 symlink,文件名自带 pod / ns / container。
pub const CONTAINER_LOG_ROOT: &str = "/var/log/containers";

/// 容器选源的结构化过滤器(默认全空 = 不过滤)。
///
/// 为什么不是「再写几条 pattern」:容器日志文件名本身就是
/// `<pod>_<namespace>_<container>-<id>.log`,选源要用的三个维度全在里面;把 namespace
/// 白名单散成 N 条 path pattern 会随集群漂移,而过滤器是**结构化**的、与文件名格式解耦。
///
/// 求值位置:**发现阶段**(agentd 展开 glob、剔完 `exclude_globs` 之后,读之前),
/// 读取器不参与 —— 所以它不影响 tail / checkpoint / 轮转语义。
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContainerFilters {
    /// 只采这些 namespace(空 = 不限)。
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub namespace_in: Vec<String>,
    /// 这些 pod 不采(空 = 不排除)。
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pod_not_in: Vec<String>,
}

impl ContainerFilters {
    /// 两个列表都空 = 不过滤(空过滤器不上线,见 [`WorkSpecSource::filters`])。
    pub fn is_empty(&self) -> bool {
        self.namespace_in.is_empty() && self.pod_not_in.is_empty()
    }

    /// 形状校验:过滤器里的值是**精确名字**,不是 pattern。
    ///
    /// 网关、web、agentd 三处共用这一份(否则会出现「页面能填、agentd 不认」)。
    /// 不强制 RFC1123:这里只拦“一定对不上”的形态(空串 / 空白 / 通配符 / 超长),
    /// 名字是否真存在由集群决定,不是配置校验能回答的。
    pub fn validate(&self) -> Result<(), String> {
        for (field, values) in [
            ("namespace_in", &self.namespace_in),
            ("pod_not_in", &self.pod_not_in),
        ] {
            for value in values {
                if value.trim().is_empty() {
                    return Err(format!("{field} 里有空值"));
                }
                if value.trim() != value {
                    return Err(format!("{field} 里 {value:?} 带首尾空白"));
                }
                if value.contains(['*', '?', '[', ']']) {
                    return Err(format!(
                        "{field} 里 {value:?} 含通配符(这里是精确名字,不是 pattern)"
                    ));
                }
                if value.len() > 253 {
                    return Err(format!("{field} 里 {value:?} 过长(>253)"));
                }
            }
        }
        Ok(())
    }
}

fn default_multiline() -> String {
    "none".to_string()
}

impl WorkSpec {
    /// 解析 `StandingWork.spec`。
    ///
    /// 解析失败**不当作空工作**:空工作会让「参数坏了」退化成「没事可做」,
    /// 两种情形的运维动作完全不同。
    pub fn parse(spec: &str) -> Result<Self, serde_json::Error> {
        serde_json::from_str(spec)
    }

    /// 序列化(网关写 `spec` 用)。字段顺序固定 → 同一份工作每次编码都一样,
    /// `plan_version` 之外的字节也稳定,方便对账。
    pub fn encode(&self) -> Result<String, serde_json::Error> {
        serde_json::to_string(self)
    }

    /// 周期类来源声明的间隔(秒):取所有 `MetricInterval` 里最密的那一个。
    ///
    /// 取最密而不是取第一个:一份工作里若有两个指标单元,采慢的那个会
    /// 静默压掉采密的要求,而“要得最急”的需求才是上送频率的下界。
    pub fn metric_interval_seconds(&self) -> Option<i64> {
        self.units
            .iter()
            .flat_map(|unit| unit.sources.iter())
            .filter(|source| source.kind == "MetricInterval")
            .filter_map(|source| parse_interval_seconds(&source.target))
            .min()
    }

    /// 是否要采日志(有 `collect_logs` 单元)。
    pub fn collects_logs(&self) -> bool {
        self.units
            .iter()
            .any(|unit| unit.capability == "collect_logs")
    }
}

/// agentd **有可能真的执行**的来源 kind(其余 kind 一律 `unsupported`,不当没看见)。
///
/// 注意:kind 对了**还不够** —— 目标形态也有限制,见 [`is_executable_source`]:
/// `FileGlob` 要是**绝对路径**(显式路径或受支持的 glob)、`Exporter` 要是**已知导出器 ID**。
/// 只按 kind 判会把「`**` 递归」「`~` 与相对路径」「写错的导出器」当成可采,而那在采集端落不了地。
pub const EXECUTABLE_SOURCE_KINDS: &[&str] = &["FileGlob", "MetricInterval", "Exporter"];

/// 目标里有没有通配元字符 —— 有就说明它不是**显式单路径**。
pub fn has_glob_meta(target: &str) -> bool {
    target.contains(['*', '?', '['])
}

/// 这个目标是不是采集端今天**真能打开**的路径:**绝对路径、无通配**。
///
/// 采集端(agentd 第一版)只实现显式单路径输入:**不做 glob 展开、也不展开 `~`**
/// (见 `wist/wist-agentd/docs/design/log-file-input-spec.md` §2 的能力边界)。
/// 单层 glob 从 W1 起可采,但那走 [`is_glob_path`];`~` 与相对路径仍永远不可采。
///
/// 从 W1 起,带通配的目标由 [`is_glob_path`] / [`is_file_target`] 判定 —— 它们才是
/// 「采集端接不接得了」的口径;本函数只留给**单文件**场景(例如「一个显式路径能否参与
/// 存在性裁剪」:那一步只能对显式路径做,探测不到的路径不能当不存在)。
pub fn is_explicit_path(target: &str) -> bool {
    target.starts_with('/') && !has_glob_meta(target)
}

/// 这个目标是不是**受支持的文件族 glob**:**绝对路径、有通配、不含 `**`**。
///
/// `**` 被拒不是保守,而是口径:采集端只扫 pattern 的固定前缀目录、`*` 不跨目录
/// (`literal_separator`),递归展开既没实现也无法给出上限 —— 让它在目录装载期就被挡住,
/// 比让 agent 每轮报一次 IO 错误好。
pub fn is_glob_path(target: &str) -> bool {
    target.starts_with('/') && has_glob_meta(target) && !target.contains("**")
}

/// 这个目标是不是采集端**能承接的文件目标**:显式绝对路径,或受支持的文件族 glob。
///
/// 不展开 `~`、不接受相对路径:相对路径的结果取决于进程 cwd,同一台机器换一种启动方式
/// 就换一个采集对象,那不是「配置」,是运气。
pub fn is_file_target(target: &str) -> bool {
    is_explicit_path(target) || is_glob_path(target)
}

/// 定时导出器(`Exporter`)的**已知 ID**:协议层词表,网关 / agentd / web 三侧共用。
///
/// 为什么词表放这里:`is_executable_source` 要判「这条 `Exporter` 目标今天真能不能采」,
/// 而网关的采集就绪度与 agentd 的「接不接得了」必须是**同一个判据**(见 [`is_executable_source`])。
/// 词表只收**已实现**的导出器 —— 收录一个 ID 就等于承诺 agentd 能跑它;未知 ID(旧网关发新 ID、
/// 或写错)一律当不可采,不要「写了就算」。新增能力 = 四侧(本 crate + 网关 + agentd + web)
/// 一起改 + 知识库把对应单元开 `active`。
pub const EXPORTER_IDS: &[&str] = &[
    "journalctl-unit",
    "journalctl-shutdown",
    "last-reboot",
    "nft-ruleset",
    "iptables-save",
    "smartctl",
    "dmesg",
    "auditd-execve",
];

/// 把一个 `Exporter` 目标拆成 `(id, arg)`:`"dmesg:panic"` → `("dmesg", Some("panic"))`。
///
/// 无 `:` 时 `arg` 为 `None`;空 id 或含空白的 id 视为无效(返回 `None`)。`arg` 是**声明式的窄选项**
/// (如 `dmesg` 的过滤键),不是命令行片段 —— 实现侧各自解释,绝不拼进 shell。
pub fn parse_exporter_target(target: &str) -> Option<(&str, Option<&str>)> {
    let (id, arg) = match target.split_once(':') {
        Some((id, arg)) => (id, Some(arg)),
        None => (target, None),
    };
    let id = id.trim();
    if id.is_empty() || id.contains(char::is_whitespace) {
        return None;
    }
    Some((id, arg))
}

/// 这条 `Exporter` 目标的 ID 是不是**已知且已实现**的导出器。
pub fn is_known_exporter(target: &str) -> bool {
    match parse_exporter_target(target) {
        Some((id, _)) => EXPORTER_IDS.contains(&id),
        None => false,
    }
}

/// 一条来源今天**能不能被采**:`kind` 要 agentd 能执行,`target` 还要是它接得住的形态。
///
/// 为什么把两件事放在一个函数里:网关的采集就绪度(「这个面能不能派下去」)与 agentd 的
/// 「这条来源接不接得了」必须是**同一个判据**,否则会出现「网关说可采、agent 拿到后报
/// unsupported」这种没人能发现的矛盾。加一种 kind 支持时**只改这里**。
pub fn is_executable_source(kind: &str, target: &str) -> bool {
    match kind {
        // 本地文件采集:显式绝对路径,或受支持的文件族 glob(无 `**`、无 `~`)。
        "FileGlob" => is_file_target(target),
        // 指标:`target` 是采集周期(如 `15s`、`60s`),不是路径。
        "MetricInterval" => true,
        // 定时导出器:`target` 要是**已知导出器 ID**(可带 `:arg`)。未知 ID 不可执行。
        "Exporter" => is_known_exporter(target),
        _ => false,
    }
}

impl WorkSpecUnit {
    /// 该单元里可以交给本地文件采集器的来源(`FileGlob`)。
    ///
    /// 其它来源(导出器 / 统一日志谓词)不是本地 tail 一个文件能承接的:
    /// 调用方应当把它们当成**还没支持的单元**如实报出来,而不是当没看见。
    pub fn file_sources(&self) -> Vec<&WorkSpecSource> {
        self.sources
            .iter()
            .filter(|source| source.kind == "FileGlob")
            .collect()
    }

    /// 本单元有没有 agentd 目前接不了的来源(kind 不认识,或目标不是可采的绝对路径)。
    pub fn unsupported_sources(&self) -> Vec<&WorkSpecSource> {
        self.sources
            .iter()
            .filter(|source| !is_executable_source(&source.kind, &source.target))
            .collect()
    }

    /// 本单元至少有一条来源是 agentd **今天真能采**的。
    ///
    /// 「能不能采」只看这个:与解析规则(`rule_ref`)无关 —— 采原文不需要规则,
    /// 规则只决定采下来的东西能不能被归类、抽字段。
    pub fn has_executable_source(&self) -> bool {
        self.sources
            .iter()
            .any(|source| is_executable_source(&source.kind, &source.target))
    }
}

/// 解析 `15s` / `60s` / `5m` 这类周期写法(与目录里 `MetricInterval` 的写法一致)。
///
/// 只认秒与分两种后缀:目录里的值是人写的,多一个单位就多一种笔误的可能,
/// 而解析不了时返回 `None` 会让调用方回退到自己的默认值(不静默取 0)。
pub fn parse_interval_seconds(raw: &str) -> Option<i64> {
    let raw = raw.trim();
    if let Some(value) = raw.strip_suffix('s') {
        return value.trim().parse::<i64>().ok().filter(|value| *value > 0);
    }
    if let Some(value) = raw.strip_suffix('m') {
        let minutes = value.trim().parse::<i64>().ok()?;
        // `checked_mul`:`123456789012345678m` 这类输入会让 `minutes * 60` 溢出 i64。
        // 契约是「解析不了返回 None」(调用方回退到自己的默认值),不能让它 panic(debug)
        // 或悄悄回绕成一个错的间隔(release)。
        return minutes.checked_mul(60).filter(|value| *value > 0);
    }
    None
}

#[cfg(test)]
mod tests {
    use super::*;

    fn one_shot(status: &str, interruptible: bool) -> OneShotWork {
        OneShotWork {
            work_id: "work-1".to_string(),
            agent_id: "agent-1".to_string(),
            action: "upgrade".to_string(),
            spec: "0.1.4".to_string(),
            scheduled_at: "2026-09-23T00:00:00Z".to_string(),
            deadline_at: "2026-09-24T00:00:00Z".to_string(),
            timeout_seconds: 600,
            interruptible,
            status: status.to_string(),
            paused_at: None,
            paused_total_seconds: 0,
            current_step: None,
            completed_steps: vec![],
            attempt: 0,
            issued_by: "admin".to_string(),
            issued_at: "2026-09-23T00:00:00Z".to_string(),
        }
    }

    #[test]
    fn terminal_one_shot_work_is_not_outstanding() {
        // 未了结的活才进快照:终态留在库里供审计,但不必再发给 agent。
        for status in ONE_SHOT_TERMINAL_STATUSES {
            assert!(!one_shot(status, true).is_outstanding(), "{status}");
        }
        for status in ["dispatched", "accepted", "running", "paused"] {
            assert!(one_shot(status, true).is_outstanding(), "{status}");
        }
    }

    #[test]
    fn only_interruptible_running_work_can_pause() {
        // 不可中断:拒绝,而不是「尽力暂停」。
        assert!(!one_shot("running", false).can_pause());
        // 可中断但没在做(还没到点 / 已经了结):也没什么可暂停的。
        assert!(!one_shot("dispatched", true).can_pause());
        assert!(!one_shot("succeeded", true).can_pause());
        assert!(one_shot("accepted", true).can_pause());
        assert!(one_shot("running", true).can_pause());
    }

    #[test]
    fn work_kind_serializes_as_the_model_names() {
        assert_eq!(
            serde_json::to_string(&WorkKind::Standing).expect("serialize"),
            "\"Standing\""
        );
        assert_eq!(
            serde_json::to_string(&WorkKind::OneShot).expect("serialize"),
            "\"OneShot\""
        );
    }

    #[test]
    fn only_explicit_absolute_paths_are_collectable_file_targets() {
        // `is_explicit_path` 现在只服务「单文件」语义(例如存在性裁剪只能对显式路径做)。
        // 这个判据被两侧共用(网关的采集就绪度 + agentd 的接受逻辑),所以在这里钉死。
        assert!(is_explicit_path("/var/log/install.log"));
        assert!(!is_explicit_path("/var/log/wifi.log*"));
        assert!(!is_explicit_path("/Library/Logs/DiagnosticReports/*.ips"));
        assert!(!is_explicit_path("~/Library/Logs/Homebrew/*"));
        assert!(!is_explicit_path("relative/app.log"));
    }

    #[test]
    fn file_targets_accept_single_level_globs_and_reject_recursion() {
        // W1:`FileGlob` 判据从「显式路径」放宽为「绝对路径,允许通配」。
        assert!(is_glob_path("/var/log/containers/*.log"));
        assert!(is_glob_path("/var/log/pods/*/*/*.log"));
        assert!(is_glob_path("/var/lib/postgresql/*/data/log/*.csv"));
        assert!(is_file_target("/var/log/containers/*.log"));
        // 显式路径仍是合法文件目标(单文件 input 走它)。
        assert!(is_file_target("/var/log/auth.log"));
        // `**` 不展开:采集端只扫固定前缀目录,接受它只会在采集端变成每轮 IO 错误。
        assert!(!is_glob_path("/var/log/**/*.log"));
        assert!(!is_file_target("/var/log/**/*.log"));
        // `~` / 相对路径不展开。
        assert!(!is_glob_path("~/Library/Logs/*.log"));
        assert!(!is_file_target("relative/*.log"));
        // 空目标不是文件目标。
        assert!(!is_file_target(""));
    }

    #[test]
    fn executability_needs_both_a_supported_kind_and_a_supported_target() {
        // 指标:`target` 是周期不是路径,不管显式路径那一条。
        assert!(is_executable_source("MetricInterval", "15s"));
        // 文件:显式路径与单层 glob 都可执行;`**` / `~` 不可执行(网关不能把它们报成“可采”)。
        assert!(is_executable_source("FileGlob", "/var/log/app.log"));
        assert!(is_executable_source(
            "FileGlob",
            "/var/log/containers/*.log"
        ));
        assert!(!is_executable_source("FileGlob", "/var/log/**/*.log"));
        assert!(!is_executable_source("FileGlob", "~/logs/*.log"));
        // 导出器:kind 对了,ID 还得是**已知的**(`last,lastb` 是 macOS 侧、尚未实现)。
        assert!(is_executable_source("Exporter", "smartctl"));
        assert!(is_executable_source("Exporter", "dmesg:panic"));
        assert!(!is_executable_source("Exporter", "last,lastb"));
        // 还没实现的 kind:一律不可执行。
        assert!(!is_executable_source("UnifiedLogPredicate", "syspolicyd"));
    }

    #[test]
    fn exporter_targets_split_into_a_known_id_and_an_optional_arg() {
        assert_eq!(parse_exporter_target("smartctl"), Some(("smartctl", None)));
        assert_eq!(
            parse_exporter_target("dmesg:panic"),
            Some(("dmesg", Some("panic")))
        );
        // 空 id / 含空白的 id 不是合法 ID。
        assert_eq!(parse_exporter_target(":panic"), None);
        assert_eq!(parse_exporter_target(" dm esg"), None);
        // 判定只看冒号前的 ID,`arg` 不参与。
        assert!(is_known_exporter("dmesg:nvidia-xid"));
        assert!(!is_known_exporter("nope"));
    }

    // ── 工作参数(`spec`)的编码 ──

    fn spec() -> WorkSpec {
        WorkSpec {
            units: vec![
                WorkSpecUnit {
                    unit_id: "mac-host-metrics".to_string(),
                    capability: "collect_metrics".to_string(),
                    rule_ref: "agent_uplink".to_string(),
                    requires_privilege: "none".to_string(),
                    sources: vec![WorkSpecSource {
                        kind: "MetricInterval".to_string(),
                        target: "15s".to_string(),
                        multiline: "none".to_string(),
                        exclude_globs: Vec::new(),
                        filters: ContainerFilters::default(),
                    }],
                },
                WorkSpecUnit {
                    unit_id: "mac-privacy-tcc".to_string(),
                    capability: "collect_logs".to_string(),
                    rule_ref: "macos/tcc".to_string(),
                    requires_privilege: "fda".to_string(),
                    sources: vec![
                        WorkSpecSource {
                            kind: "Exporter".to_string(),
                            target: "sqlite-snapshot(TCC.db)".to_string(),
                            multiline: "none".to_string(),
                            exclude_globs: Vec::new(),
                            filters: ContainerFilters::default(),
                        },
                        WorkSpecSource {
                            kind: "FileGlob".to_string(),
                            // 显式单路径:单文件采集走 `startup_position` / 检查点的单文件语义。
                            target: "/var/log/tccd.log".to_string(),
                            // 多行格式是这条日志的固有属性,跟着来源走。
                            multiline: "indented".to_string(),
                            exclude_globs: Vec::new(),
                            filters: ContainerFilters::default(),
                        },
                    ],
                },
            ],
        }
    }

    #[test]
    fn spec_round_trips_and_reports_what_agentd_can_do() {
        let encoded = spec().encode().expect("encode");
        let decoded = WorkSpec::parse(&encoded).expect("parse");
        assert_eq!(decoded, spec());

        assert!(decoded.collects_logs());
        // 取了最密的周期:要得最急的才是上送频率的下界。
        assert_eq!(decoded.metric_interval_seconds(), Some(15));
        // 可本地 tail 的来源挑得出来,接不了的来源也报得出来(而不是当没看见)。
        let files = decoded.units[1].file_sources();
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].target, "/var/log/tccd.log");
        assert_eq!(files[0].multiline, "indented");
        assert_eq!(decoded.units[1].unsupported_sources().len(), 1);
        assert_eq!(decoded.units[1].unsupported_sources()[0].kind, "Exporter");
    }

    #[test]
    fn a_glob_file_source_is_collectable_but_recursive_globs_are_not() {
        // W1 起:单层 glob 是采集端能执行的来源(容器日志就靠它)。
        let mut unit = WorkSpecUnit {
            unit_id: "linux-container-stdout".to_string(),
            capability: "collect_logs".to_string(),
            rule_ref: String::new(),
            requires_privilege: "root".to_string(),
            sources: vec![WorkSpecSource {
                kind: "FileGlob".to_string(),
                target: "/var/log/containers/*.log".to_string(),
                multiline: "none".to_string(),
                exclude_globs: vec!["/var/log/containers/*_kube-system_*.log".to_string()],
                filters: ContainerFilters::default(),
            }],
        };
        assert_eq!(unit.file_sources().len(), 1, "它是一条文件来源");
        assert!(unit.unsupported_sources().is_empty());
        assert!(unit.has_executable_source());

        // `**` 仍不可执行:递归展开没实现,接受它就等于承诺一个采不到的面。
        unit.sources[0].target = "/var/log/**/*.log".to_string();
        assert_eq!(unit.unsupported_sources().len(), 1);
        assert!(!unit.has_executable_source());
    }

    #[test]
    fn exclude_globs_round_trip_and_stay_out_of_the_wire_when_empty() {
        let mut source = WorkSpecSource {
            kind: "FileGlob".to_string(),
            target: "/var/log/containers/*.log".to_string(),
            multiline: "none".to_string(),
            exclude_globs: Vec::new(),
            filters: ContainerFilters::default(),
        };
        // 空排除项不进 JSON:旧 agent 的 `WorkSpecSource` 是 `deny_unknown_fields` 的。
        assert_eq!(
            serde_json::to_string(&source).expect("serialize"),
            r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","multiline":"none"}"#
        );

        source.exclude_globs = vec!["/var/log/containers/*_kube-system_*.log".to_string()];
        let json = serde_json::to_string(&source).expect("serialize");
        assert_eq!(
            serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
            source
        );
        // 老网关发来的来源没有这两个字段:默认必须能解出来。
        let old = r#"{"kind":"FileGlob","target":"/a/*"}"#;
        let parsed: WorkSpecSource = serde_json::from_str(old).expect("parse legacy source");
        assert!(parsed.exclude_globs.is_empty());
        assert!(parsed.filters.is_empty());
        assert_eq!(parsed.multiline, "none");
    }

    #[test]
    fn container_filters_round_trip_and_stay_out_of_the_wire_when_empty() {
        let mut source = WorkSpecSource {
            kind: "FileGlob".to_string(),
            target: "/var/log/containers/*.log".to_string(),
            multiline: "none".to_string(),
            exclude_globs: Vec::new(),
            filters: ContainerFilters::default(),
        };
        // 空过滤器不进 JSON(与 exclude_globs 同一条纪律:旧 agent 是 deny_unknown_fields)。
        let json = serde_json::to_string(&source).expect("serialize");
        assert!(!json.contains("filters"), "{json}");

        source.filters = ContainerFilters {
            namespace_in: vec!["prod".to_string()],
            pod_not_in: vec!["debug-pod".to_string()],
        };
        let json = serde_json::to_string(&source).expect("serialize");
        assert!(
            json.contains(r#""filters":{"namespace_in":["prod"],"pod_not_in":["debug-pod"]}"#),
            "{json}"
        );
        assert_eq!(
            serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
            source
        );

        // 只给一半也要能解:另一侧默认空。
        let half = r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","filters":{"namespace_in":["prod"]}}"#;
        let parsed: WorkSpecSource = serde_json::from_str(half).expect("parse half filters");
        assert_eq!(parsed.filters.namespace_in, vec!["prod".to_string()]);
        assert!(parsed.filters.pod_not_in.is_empty());
    }

    #[test]
    fn an_explicitly_empty_filters_object_is_not_serialized() {
        // 策展/页面可能显式写一个空对象(`filters = {}`):它仍然等于“不过滤”,
        // 不能在线上多出一个键 —— 旧 agent 是 `deny_unknown_fields` 的。
        let source = WorkSpecSource {
            kind: "FileGlob".to_string(),
            target: "/var/log/containers/*.log".to_string(),
            multiline: "none".to_string(),
            exclude_globs: Vec::new(),
            filters: ContainerFilters {
                namespace_in: Vec::new(),
                pod_not_in: Vec::new(),
            },
        };
        let json = serde_json::to_string(&source).expect("serialize");
        assert!(!json.contains("filters"), "{json}");
    }

    #[test]
    fn container_filters_keep_values_verbatim() {
        // 名字**不做归一化**(不折叠大小写、不去前后缀):k8s 名字永远小写,
        // 归一化反而会把写错的值当成能用。逐字保存、逐字比较(比较在 agentd)。
        let filters = ContainerFilters {
            namespace_in: vec!["Prod".to_string()],
            pod_not_in: vec!["debug-pod".to_string()],
        };
        let json = serde_json::to_string(&filters).expect("serialize");
        let parsed: ContainerFilters = serde_json::from_str(&json).expect("parse");
        assert_eq!(parsed.namespace_in, vec!["Prod".to_string()]);
        assert_eq!(parsed.pod_not_in, vec!["debug-pod".to_string()]);
    }

    #[test]
    fn container_filters_reject_unknown_keys() {
        // `deny_unknown_fields`:拼错一个键(`namespaces`)不能静默当空过滤器 ——
        // 那会让策展人以为「只采 prod」,实际全采。
        assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespaces":["prod"]}"#).is_err());
        assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespace_in":["prod"]}"#).is_ok());
    }

    #[test]
    fn container_filters_reject_values_that_can_never_match() {
        let ok = |namespace: &[&str], pods: &[&str]| ContainerFilters {
            namespace_in: namespace.iter().map(|v| v.to_string()).collect(),
            pod_not_in: pods.iter().map(|v| v.to_string()).collect(),
        };

        assert!(
            ok(&["prod", "kube-system"], &["debug-pod"])
                .validate()
                .is_ok()
        );
        assert!(ok(&[], &[]).is_empty());
        // 这些一定对不上(过滤器存的是**精确名字**,不是 pattern):错在最早的入口就说。
        assert!(ok(&[""], &[]).validate().is_err());
        assert!(ok(&[" prod"], &[]).validate().is_err());
        assert!(ok(&["prod*"], &[]).validate().is_err());
        assert!(ok(&[], &["debug?pod"]).validate().is_err());
        assert!(ok(&[&"x".repeat(254)], &[]).validate().is_err());
    }

    #[test]
    fn a_source_without_multiline_decodes_as_single_line() {
        // 老网关发来的 spec 没有 `multiline`:默认必须是一行一条。
        // 取错默认值会把多行日志**粘**成一条 —— 那是无声的内容损坏,不是格式问题。
        let json = r#"{"units":[{"unit_id":"u","capability":"collect_logs","sources":[{"kind":"FileGlob","target":"/a/*"}]}]}"#;
        let spec = WorkSpec::parse(json).expect("parse");
        assert_eq!(spec.units[0].sources[0].multiline, "none");
        // 再编码出去会把默认值写实:新网关发的 spec 字段总是齐的。
        assert_eq!(
            spec.encode().expect("encode"),
            r#"{"units":[{"unit_id":"u","capability":"collect_logs","rule_ref":"","requires_privilege":"","sources":[{"kind":"FileGlob","target":"/a/*","multiline":"none"}]}]}"#
        );
    }

    #[test]
    fn a_broken_spec_is_an_error_not_an_empty_work() {
        // 参数坏了与「没事可做」是两回事:前者要人去看,后者什么都不用做。
        assert!(WorkSpec::parse("mac-host-metrics").is_err());
        assert!(WorkSpec::parse("{").is_err());
        // 空工作本身是合法的(能表示「这个面暂时没东西可采」)。
        assert!(WorkSpec::parse(r#"{"units":[]}"#).unwrap().units.is_empty());
        assert_eq!(
            WorkSpec::parse(r#"{"units":[]}"#)
                .unwrap()
                .metric_interval_seconds(),
            None
        );
    }

    #[test]
    fn interval_parsing_accepts_seconds_and_minutes_only() {
        assert_eq!(parse_interval_seconds("15s"), Some(15));
        assert_eq!(parse_interval_seconds(" 5s "), Some(5));
        assert_eq!(parse_interval_seconds("5m"), Some(300));
        // 认不出来的回 None(调用方回退到自己的默认值),而不是静默当 0。
        assert_eq!(parse_interval_seconds("15"), None);
        assert_eq!(parse_interval_seconds("0s"), None);
        assert_eq!(parse_interval_seconds(""), None);
    }

    #[test]
    fn interval_parsing_never_overflows_on_huge_values() {
        // 契约是「解析不了返回 None」,不是 panic:溢出在 debug 下 panic、release 下回绕成一个
        // 错误的间隔 —— 两个后果都不能接受。这是两侧都要解析的字节,输入可能来自远端。
        assert_eq!(parse_interval_seconds("922337203685477580m"), None);
        assert_eq!(parse_interval_seconds("153722867280912931m"), None);
        // 秒分支本来就靠 parse 失败兜底。
        assert_eq!(parse_interval_seconds("99999999999999999999s"), None);
        // 合法但很大的分钟值仍要算得出来(不误伤)。
        assert_eq!(parse_interval_seconds("600m"), Some(36_000));
    }

    #[test]
    fn metric_interval_ignores_unparseable_targets_instead_of_treating_them_as_zero() {
        // 认不出的周期不能静默当 0(0 会被当成「无限密」),而要如实跳过,只取能解析的最密值。
        let unit = |target: &str| WorkSpecUnit {
            unit_id: "u".to_string(),
            capability: "collect_metrics".to_string(),
            rule_ref: String::new(),
            requires_privilege: String::new(),
            sources: vec![WorkSpecSource {
                kind: "MetricInterval".to_string(),
                target: target.to_string(),
                multiline: "none".to_string(),
                exclude_globs: Vec::new(),
                filters: ContainerFilters::default(),
            }],
        };
        let mixed = WorkSpec {
            units: vec![unit("garbage"), unit("30s")],
        };
        assert_eq!(mixed.metric_interval_seconds(), Some(30));

        let all_bad = WorkSpec {
            units: vec![unit("nope")],
        };
        assert_eq!(all_bad.metric_interval_seconds(), None);
    }
}