Skip to main content

wist_contracts/
work.rs

1//! **工作授权快照**:网关授权(grant)、agentd 拉取(`WorkGrant`)并确认。
2//!
3//! ## 为什么是「快照」而不是「指令流」
4//!
5//! 常驻工作(`StandingWork`)表达的是**期望状态**:某个采集面应当持续执行什么。
6//! 期望状态天然是幂等的(同一份重复拉取不产生副作用),断了网、重启了进程,
7//! 再拉一次就回到期望 —— 不必让网关记住「上次推到哪条」。
8//! 这跟 `PollControlCommands` 的长轮询指令流是两种东西:那条要求**不重放**,这条要求**可重拉**。
9//!
10//! 一次性工作(`OneShotWork`)是命令式的,但它也搭这份快照回来:
11//! 决定「现在还该做吗」的是它的状态(未了结才出现在快照里),不是「推没推过」。
12//!
13//! ## 为什么在契约 crate
14//!
15//! 这是**两侧都要解析**的字节:网关写授权、agentd 读并执行。各写一份结构体,
16//! 迟早出现「网关发了 `plan_version`、agentd 读的是 `version`」这种只能在真机联调时才发现的分叉。
17//! 与 `DiscoveryAspectPolicySet` 同一个理由。
18//!
19//! ## 与本地保护暂停的区别
20//!
21//! agentd 自己也会暂停(如 `spool over limit`),那是**本地保护**:自动、临时、本机可见。
22//! 这里下发的 `paused` 是**授权层状态**:人工、持久、跨重启。两者在观测里必须能分辨,
23//! 不要合并成一个 paused —— 合并之后「谁把采集停了」就再也说不清。
24
25use serde::{Deserialize, Serialize};
26
27use crate::capability_report::LogsCapabilities;
28
29/// 常驻工作的状态取值(对应模型 `StandingWork.status`)。
30///
31/// `superseded` 与 `revoked` 都不出现在 `WorkGrant.standing` 里:前者是「被新版本取代」,
32/// 后者是「授权被撤」。两者都要留痕,所以还是要有状态而不是删行。
33pub const STANDING_WORK_STATUSES: [&str; 4] = ["active", "paused", "superseded", "revoked"];
34
35/// 一次性工作的状态取值(对应模型 `OneShotWork.status`)。
36pub const ONE_SHOT_WORK_STATUSES: [&str; 9] = [
37    "dispatched",
38    "accepted",
39    "running",
40    "paused",
41    "succeeded",
42    "failed",
43    "timed_out",
44    "canceled",
45    "expired",
46];
47
48/// Agent **允许上报**的一次性工作状态([`ONE_SHOT_WORK_STATUSES`] 的真子集)。
49///
50/// 为什么不是全集:`dispatched` 是网关自己写的(派下去那一刻),`paused` / `canceled` / `expired`
51/// 归**控制面与期限**管(运维暂停撤回、或过了截止)—— agent 无权把它们写回去。
52/// 剩下的问题只有一种:「这件活做完了没有」,答案就这三种。
53pub const AGENT_REPORTABLE_WORK_STATUSES: [&str; 3] = ["running", "succeeded", "failed"];
54
55/// 一次性工作的**终态**:到了这几个状态就了结了,不再出现在快照里。
56pub const ONE_SHOT_TERMINAL_STATUSES: [&str; 5] =
57    ["succeeded", "failed", "timed_out", "canceled", "expired"];
58
59/// 工作类型:常驻(持续到被替换或撤回)与一次性(有期限与终态)。
60#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
61#[jumo(kind = "state", domain = "Control", module = "Control.Agent.Work")]
62#[serde(rename_all = "PascalCase")]
63pub enum WorkKind {
64    /// 常驻工作:按**采集面**授权,一个面一份。
65    Standing,
66    /// 一次性工作:按**动作**授权。
67    OneShot,
68}
69
70/// 常驻工作:网关声明该 Agent 应当持续执行的工作。
71///
72/// 粒度是**采集面**(一个面 = 一份工作),不是 capability:一份 `collect_logs` 会裹住十几个面,
73/// 那样「按面暂停 / 限流 / 审计」全都无从下手。面由模板的 `family_scope` 展开而来,
74/// **不携带 capability** —— 该面由哪个采集器承接,由 `spec` 里的单元各自决定。
75#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
76#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
77#[serde(deny_unknown_fields)]
78pub struct StandingWork {
79    pub work_id: String,
80    pub agent_id: String,
81    /// 采集面(`CollectionFamily`)。
82    pub family: String,
83    /// 工作参数:由采集目录的条目组合而成(**不是自由文本**),随 `plan_version` 整体替换。
84    pub spec: String,
85    /// 本工作按哪一版目录展开:目录换版**不追改**已授权工作(要跟新版得走新提案 + 审定)。
86    pub catalog_version: i64,
87    /// 生效依据:指向已批准的提案(人工直填 spec 时为空)。
88    #[serde(default)]
89    pub proposal_id: Option<String>,
90    /// 期望版本:网关每次改动 +1;agentd 回报实际版本,与它比对即得漂移。
91    pub plan_version: i64,
92    pub effective_from: String,
93    /// 见 [`STANDING_WORK_STATUSES`]。
94    pub status: String,
95    pub updated_by: String,
96    pub updated_at: String,
97}
98
99/// 一次性工作:有计划开始时间与期限,有明确终态。
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
101#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
102#[serde(deny_unknown_fields)]
103pub struct OneShotWork {
104    pub work_id: String,
105    pub agent_id: String,
106    /// 动作面:upgrade / snapshot / exec / ...(执行载体是 Reporting 域的 ActionPlan)。
107    pub action: String,
108    pub spec: String,
109    /// 计划开始时间:到点前不应执行(与「立即派发」区分开)。
110    pub scheduled_at: String,
111    /// 绝对截止:业务要求的时间点,**暂停也照走**(不由暂停顺延)。
112    pub deadline_at: String,
113    /// 执行预算(秒):**只在实际执行时消耗**,暂停期间不计。
114    pub timeout_seconds: i64,
115    /// 可中断性:只有可中断的动作才允许运行中暂停。
116    pub interruptible: bool,
117    /// 见 [`ONE_SHOT_WORK_STATUSES`]。
118    pub status: String,
119    /// 当前暂停的起点(运行中暂停才有;恢复后清空)。
120    #[serde(default)]
121    pub paused_at: Option<String>,
122    /// 累计暂停时长(秒):用于审计与「预算未被暂停消耗」的核对。
123    pub paused_total_seconds: i64,
124    /// 步级断点:已完成步骤保留、`current_step` 在恢复时重做。
125    #[serde(default)]
126    pub current_step: Option<String>,
127    #[serde(default)]
128    pub completed_steps: Vec<String>,
129    /// 已尝试次数(含恢复后的重做)。
130    pub attempt: i64,
131    pub issued_by: String,
132    pub issued_at: String,
133}
134
135impl OneShotWork {
136    /// 是否**未了结**(快照里只带未了结的活)。
137    pub fn is_outstanding(&self) -> bool {
138        !ONE_SHOT_TERMINAL_STATUSES.contains(&self.status.as_str())
139    }
140
141    /// 是否允许在运行期间暂停。
142    ///
143    /// 两条都要满足:动作声明了 `interruptible`,且此刻确实在做(`accepted`/`running`)。
144    /// 不可中断的动作**拒绝**而不是「尽力暂停」—— 挂起半个升级进程比不暂停更危险。
145    pub fn can_pause(&self) -> bool {
146        self.interruptible && matches!(self.status.as_str(), "accepted" | "running")
147    }
148}
149
150/// 管理面授权或撤回工作的回执:一份工作一次。
151#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
152#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
153#[serde(deny_unknown_fields)]
154pub struct WorkReceipt {
155    pub work_id: String,
156    pub agent_id: String,
157    pub work_kind: WorkKind,
158    /// accepted | paused | resumed | revoked | rejected。
159    pub status: String,
160    pub plan_version: i64,
161    pub created_at: String,
162}
163
164// ─────────────────────────────────────────────────────────────────────────────
165// 工作参数(`StandingWork.spec` 的内容)
166// ─────────────────────────────────────────────────────────────────────────────
167//
168// 模型里 `StandingWork.spec` 是一个 `String`,语义是「工作参数:由采集目录的条目
169// 组合而成,**不是自由文本**」。这里的三个类型就是它的**编码**:一串已物化的采集单元。
170//
171// 为什么必须带上来源与规则标识,而不是只给一串 `unit_id`:agentd 拿到工作要能
172// **直接照做**。单元的采集来源(`sources`)与数据面规则标识(`rule_ref`)本来都只
173// 存在于网关的采集目录里,只发 id 等于发了一张自己去不了的地址 —— 于是要么再去网关
174// 拉一次目录(多一条必须鉴权的路径),要么两边各维护一份目录(必然漂移)。
175//
176// 为什么不把它们塞成 `StandingWork` 的字段:那会把「工作」与「内容目录」的边界糊掉;
177// 模型里这个字段就是**不透明的工作参数**,保持它不透明是两侧能独立演进的前提。
178
179/// 一份常驻工作的工作参数。
180#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
181#[serde(deny_unknown_fields)]
182pub struct WorkSpec {
183    /// 本工作包含的采集单元(已按该机器的事实裁剪过)。
184    #[serde(default)]
185    pub units: Vec<WorkSpecUnit>,
186}
187
188/// 一个已物化的采集单元:只说「采什么、怎么落地」,不复述策展元信息。
189///
190/// 刻意不带 `status` / `match` / `catalog_version`:那些是网关策展与裁剪的输入,
191/// 工作一旦发出去就已经裁剪完了。带上它们会让 agentd 有「再判断一次」的空间,
192/// 而 agentd 不是第二个策展器。
193#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(deny_unknown_fields)]
195pub struct WorkSpecUnit {
196    pub unit_id: String,
197    /// `collect_logs` | `collect_metrics`:该单元由哪类采集器承接。
198    pub capability: String,
199    /// 数据面 rule/oml 标识(空 = 该单元还没接上规则,理论不会入工作)。
200    #[serde(default)]
201    pub rule_ref: String,
202    /// `none` | `root` | `fda`:要采到这东西得有什么权限 —— 缺权限时应**说清缺什么**,
203    /// 而不是安静地采不到。
204    #[serde(default)]
205    pub requires_privilege: String,
206    #[serde(default)]
207    pub sources: Vec<WorkSpecSource>,
208}
209
210/// 采集来源:`kind` ∈ `FileGlob` | `Exporter` | `UnifiedLogPredicate` | `MetricInterval`。
211///
212/// 与模型 `CollectionSource` 同形:`target` 的含义由 `kind` 决定
213/// (路径通配 / 导出器标识 / 谓词 / 周期)。
214#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215#[serde(deny_unknown_fields)]
216pub struct WorkSpecSource {
217    pub kind: String,
218    pub target: String,
219    /// 怎么读这条来源:`none`(一行一条)| `indented`(行首缩进是上一条的续行)|
220    /// `mysql_slow`(`# Time:` 起块)| `postgres_csv`(**引号外**的换行才切记录)。
221    ///
222    /// 只对可 tail 的 `FileGlob` 有意义,其余 kind 恒为 `none`。**不是**"要不要多行"这个偏好,
223    /// 而是这条日志的**固有格式**:说错了要么把多行记录拆散、要么把独立记录粘成一条。
224    #[serde(default = "default_multiline")]
225    pub multiline: String,
226    /// 从 `target` 匹配集合里剔掉的**绝对路径 glob**(默认空)。
227    ///
228    /// 只对 `FileGlob` 有意义,按**展开后的完整路径**匹配(不是文件名)。为什么需要它:
229    /// 容器日志的选源根是 `/var/log/containers/*.log`,而目录里同时有不该采的容器
230    /// (sidecar / init / 平台自身的 pod);没有排除项就只能把业务服务名写进 pattern,
231    /// 而服务名会随集群漂移。
232    ///
233    /// 空数组**不序列化**:旧 agent 的 `WorkSpecSource` 是 `deny_unknown_fields` 的,
234    /// 发一个空字段过去只会给旧机队添一种没必要的解析失败。
235    #[serde(default, skip_serializing_if = "Vec::is_empty")]
236    pub exclude_globs: Vec<String>,
237    /// 容器选源的结构化过滤器(默认空)。只对 `FileGlob` 有意义。
238    ///
239    /// 空(两个列表都空)**不序列化**:与 `exclude_globs` 同一个理由 —— 旧 agent 的
240    /// `WorkSpecSource` 是 `deny_unknown_fields` 的,发一个空字段过去只会给旧机队
241    /// 添一种没必要的解析失败。
242    ///
243    /// ⚠ 非空时**要求 agent 支持**(与 glob 单元同一条纪律):能力轴是
244    /// `LogsCapabilities.filters`(`namespace_in` / `pod_not_in`),网关在派活时裁掉
245    /// 不支持的来源;旧 agent 收到非空 `filters` 会拒收整份 spec。
246    #[serde(default, skip_serializing_if = "ContainerFilters::is_empty")]
247    pub filters: ContainerFilters,
248    /// 读进来的记录里,含这些**子串**的整条丢掉(默认空 = 不丢)。只对 `FileGlob` 有意义。
249    ///
250    /// 为什么是子串、不是正则:这是**噪音行**的白名单(`pam_unix(cron:session)` 这类
251    /// 每分钟一条、信息量为零的行),不是解析。正则意味着在采集端维护一套可执行表达式
252    /// (回溯、超时、误匹配都是新故障面),而子串能做到的事不需要它。
253    ///
254    /// 语义是“整条记录”:子串命中就丢整条,**不会只删命中那一行**而留下半条多行记录。
255    /// 丢弃数进 agentd 自观测(`telemetry noise dropped …`),不然“怎么少了”无处可查。
256    ///
257    /// 空数组**不序列化**:旧 agent 的 `WorkSpecSource` 是 `deny_unknown_fields` 的。
258    /// 非空时要求 agent 支持 `LogsCapabilities.drop_if_contains`(`substring`),网关据此裁源。
259    #[serde(default, skip_serializing_if = "Vec::is_empty")]
260    pub drop_if_contains: Vec<String>,
261}
262
263/// 容器日志的选源根(写死,D4)。
264///
265/// `/var/log/pods` 下有 `0.log` / `1.log` 与重启序号歧义;`/var/log/containers` 里
266/// 每个活容器恰好一条 symlink,文件名自带 pod / ns / container。
267pub const CONTAINER_LOG_ROOT: &str = "/var/log/containers";
268
269/// 容器选源的结构化过滤器(默认全空 = 不过滤)。
270///
271/// 为什么不是「再写几条 pattern」:容器日志文件名本身就是
272/// `<pod>_<namespace>_<container>-<id>.log`,选源要用的三个维度全在里面;把 namespace
273/// 白名单散成 N 条 path pattern 会随集群漂移,而过滤器是**结构化**的、与文件名格式解耦。
274///
275/// 求值位置:**发现阶段**(agentd 展开 glob、剔完 `exclude_globs` 之后,读之前),
276/// 读取器不参与 —— 所以它不影响 tail / checkpoint / 轮转语义。
277#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
278#[serde(deny_unknown_fields)]
279pub struct ContainerFilters {
280    /// 只采这些 namespace(空 = 不限)。
281    #[serde(default, skip_serializing_if = "Vec::is_empty")]
282    pub namespace_in: Vec<String>,
283    /// 这些 pod 不采(空 = 不排除)。
284    #[serde(default, skip_serializing_if = "Vec::is_empty")]
285    pub pod_not_in: Vec<String>,
286}
287
288/// 容器过滤器的维度名(**能力词表**的值):与 [`ContainerFilters`] 的字段同名。
289///
290/// 为什么能力名 = 字段名:网关的报错文案(`filters:namespace_in`)与目录里的写法
291/// (`filters = { namespace_in = [...] }`)能一一对上,排查时不用再查一份映射表。
292pub const FILTER_NAMESPACE_IN: &str = "namespace_in";
293/// 见 [`FILTER_NAMESPACE_IN`]。
294pub const FILTER_POD_NOT_IN: &str = "pod_not_in";
295
296/// agentd 能承接的**全部**容器过滤维度(与 [`ContainerFilters::used_dimensions`] 同源)。
297///
298/// 新增一个过滤维度时:先在 [`ContainerFilters`] 里加字段,再往这里加名字 ——
299/// 两侧对不上的漂移由 `a_fully_declared_agent_is_missing_nothing_the_contract_calls_executable` 钉住。
300pub const CONTAINER_FILTERS: &[&str] = &[FILTER_NAMESPACE_IN, FILTER_POD_NOT_IN];
301
302/// 噪音行丢弃的**求值方式**(能力词表的值)。
303///
304/// 目前只有子串一种 —— 它跟正则不是同一个东西(见 [`WorkSpecSource::drop_if_contains`])。
305/// 以后真要加正则,那是新的一项能力:旧 agent 不该被当成“也懂正则”。
306pub const DROP_IF_CONTAINS_SUBSTRING: &str = "substring";
307/// agentd 能承接的全部噪音行丢弃方式(目前只有子串)。
308pub const DROP_IF_CONTAINS_MODES: &[&str] = &[DROP_IF_CONTAINS_SUBSTRING];
309
310/// `drop_if_contains` 的条数上限(超过就是“在采集端写解析规则”,那条路不该走)。
311pub const DROP_IF_CONTAINS_MAX_ITEMS: usize = 8;
312/// 单个子串的长度上限(字节)。
313pub const DROP_IF_CONTAINS_MAX_BYTES: usize = 200;
314
315/// 校验 `drop_if_contains` 的形状:精确子串,不是 pattern。
316///
317/// 网关装载期、agentd 配置校验、web 表单三处共用同一口径(否则会出现“页面能填、
318/// agentd 不认”)。只拦“一定有意义不明”的形态:空串 / 首尾空白 / 控制字符 / 过长 / 太多 / 重复。
319pub fn validate_drop_if_contains(needles: &[String]) -> Result<(), String> {
320    if needles.len() > DROP_IF_CONTAINS_MAX_ITEMS {
321        return Err(format!(
322            "drop_if_contains 条数过多({},上限 {DROP_IF_CONTAINS_MAX_ITEMS})",
323            needles.len()
324        ));
325    }
326    for (index, needle) in needles.iter().enumerate() {
327        if needle.trim().is_empty() {
328            return Err(format!("drop_if_contains[{index}] 是空串"));
329        }
330        if needle.trim() != needle {
331            return Err(format!("drop_if_contains[{index}] {needle:?} 带首尾空白"));
332        }
333        if needle.chars().any(char::is_control) {
334            return Err(format!("drop_if_contains[{index}] {needle:?} 含控制字符"));
335        }
336        if needle.len() > DROP_IF_CONTAINS_MAX_BYTES {
337            return Err(format!(
338                "drop_if_contains[{index}] 过长({} 字节,上限 {DROP_IF_CONTAINS_MAX_BYTES})",
339                needle.len()
340            ));
341        }
342        if needles[..index].contains(needle) {
343            return Err(format!("drop_if_contains[{index}] {needle:?} 重复"));
344        }
345    }
346    Ok(())
347}
348
349impl ContainerFilters {
350    /// 两个列表都空 = 不过滤(空过滤器不上线,见 [`WorkSpecSource::filters`])。
351    pub fn is_empty(&self) -> bool {
352        self.namespace_in.is_empty() && self.pod_not_in.is_empty()
353    }
354
355    /// 这条来源**实际用到**的过滤器维度(空 = 不过滤)。
356    ///
357    /// 能力判定([`missing_log_capabilities`])与 agentd 的能力上报都从这里取:
358    /// 只写一遍字段 → 能力名的对应关系,少一处「新增一个过滤字段、忘了加进能力词表」的漏判。
359    pub fn used_dimensions(&self) -> Vec<&'static str> {
360        let mut used = Vec::new();
361        if !self.namespace_in.is_empty() {
362            used.push(FILTER_NAMESPACE_IN);
363        }
364        if !self.pod_not_in.is_empty() {
365            used.push(FILTER_POD_NOT_IN);
366        }
367        used
368    }
369
370    /// 形状校验:过滤器里的值是**精确名字**,不是 pattern。
371    ///
372    /// 网关、web、agentd 三处共用这一份(否则会出现「页面能填、agentd 不认」)。
373    /// 不强制 RFC1123:这里只拦“一定对不上”的形态(空串 / 空白 / 通配符 / 超长),
374    /// 名字是否真存在由集群决定,不是配置校验能回答的。
375    pub fn validate(&self) -> Result<(), String> {
376        for (field, values) in [
377            ("namespace_in", &self.namespace_in),
378            ("pod_not_in", &self.pod_not_in),
379        ] {
380            for value in values {
381                if value.trim().is_empty() {
382                    return Err(format!("{field} 里有空值"));
383                }
384                if value.trim() != value {
385                    return Err(format!("{field} 里 {value:?} 带首尾空白"));
386                }
387                if value.contains(['*', '?', '[', ']']) {
388                    return Err(format!(
389                        "{field} 里 {value:?} 含通配符(这里是精确名字,不是 pattern)"
390                    ));
391                }
392                if value.len() > 253 {
393                    return Err(format!("{field} 里 {value:?} 过长(>253)"));
394                }
395            }
396        }
397        Ok(())
398    }
399}
400
401fn default_multiline() -> String {
402    "none".to_string()
403}
404
405impl WorkSpec {
406    /// 解析 `StandingWork.spec`。
407    ///
408    /// 解析失败**不当作空工作**:空工作会让「参数坏了」退化成「没事可做」,
409    /// 两种情形的运维动作完全不同。
410    pub fn parse(spec: &str) -> Result<Self, serde_json::Error> {
411        serde_json::from_str(spec)
412    }
413
414    /// 序列化(网关写 `spec` 用)。字段顺序固定 → 同一份工作每次编码都一样,
415    /// `plan_version` 之外的字节也稳定,方便对账。
416    pub fn encode(&self) -> Result<String, serde_json::Error> {
417        serde_json::to_string(self)
418    }
419
420    /// 周期类来源声明的间隔(秒):取所有 `MetricInterval` 里最密的那一个。
421    ///
422    /// 取最密而不是取第一个:一份工作里若有两个指标单元,采慢的那个会
423    /// 静默压掉采密的要求,而“要得最急”的需求才是上送频率的下界。
424    pub fn metric_interval_seconds(&self) -> Option<i64> {
425        self.units
426            .iter()
427            .flat_map(|unit| unit.sources.iter())
428            .filter(|source| source.kind == "MetricInterval")
429            .filter_map(|source| parse_interval_seconds(&source.target))
430            .min()
431    }
432
433    /// 是否要采日志(有 `collect_logs` 单元)。
434    pub fn collects_logs(&self) -> bool {
435        self.units
436            .iter()
437            .any(|unit| unit.capability == "collect_logs")
438    }
439}
440
441/// agentd **有可能真的执行**的来源 kind(其余 kind 一律 `unsupported`,不当没看见)。
442///
443/// 注意:kind 对了**还不够** —— 目标形态也有限制,见 [`is_executable_source`]:
444/// `FileGlob` 要是**绝对路径**(显式路径或受支持的 glob)、`Exporter` 要是**已知导出器 ID**。
445/// 只按 kind 判会把「`**` 递归」「`~` 与相对路径」「写错的导出器」当成可采,而那在采集端落不了地。
446pub const EXECUTABLE_SOURCE_KINDS: &[&str] = &["FileGlob", "MetricInterval", "Exporter"];
447
448/// 目标里有没有通配元字符 —— 有就说明它不是**显式单路径**。
449pub fn has_glob_meta(target: &str) -> bool {
450    target.contains(['*', '?', '['])
451}
452
453/// 这个目标是不是采集端今天**真能打开**的路径:**绝对路径、无通配**。
454///
455/// 采集端(agentd 第一版)只实现显式单路径输入:**不做 glob 展开、也不展开 `~`**
456/// (见 `wist/wist-agentd/docs/design/log-file-input-spec.md` §2 的能力边界)。
457/// 单层 glob 从 W1 起可采,但那走 [`is_glob_path`];`~` 与相对路径仍永远不可采。
458///
459/// 从 W1 起,带通配的目标由 [`is_glob_path`] / [`is_file_target`] 判定 —— 它们才是
460/// 「采集端接不接得了」的口径;本函数只留给**单文件**场景(例如「一个显式路径能否参与
461/// 存在性裁剪」:那一步只能对显式路径做,探测不到的路径不能当不存在)。
462pub fn is_explicit_path(target: &str) -> bool {
463    target.starts_with('/') && !has_glob_meta(target)
464}
465
466/// 这个目标是不是**受支持的文件族 glob**:**绝对路径、有通配、不含 `**`**。
467///
468/// `**` 被拒不是保守,而是口径:采集端只扫 pattern 的固定前缀目录、`*` 不跨目录
469/// (`literal_separator`),递归展开既没实现也无法给出上限 —— 让它在目录装载期就被挡住,
470/// 比让 agent 每轮报一次 IO 错误好。
471pub fn is_glob_path(target: &str) -> bool {
472    target.starts_with('/') && has_glob_meta(target) && !target.contains("**")
473}
474
475/// 这个目标是不是采集端**能承接的文件目标**:显式绝对路径,或受支持的文件族 glob。
476///
477/// 不展开 `~`、不接受相对路径:相对路径的结果取决于进程 cwd,同一台机器换一种启动方式
478/// 就换一个采集对象,那不是「配置」,是运气。
479pub fn is_file_target(target: &str) -> bool {
480    is_explicit_path(target) || is_glob_path(target)
481}
482
483/// 内核日志导出器(W3):`journalctl -k -p 4 -o json`,**靠游标续读**。
484///
485/// 与其余导出器不同,它是**有状态**的:上一轮读到的游标落盘,下一轮从那里接着读。
486/// 契约只承诺 ID 与「可执行」;游标怎么存、失效(被 vacuum)怎么回退是采集端的事
487/// (见 `linux-server-log-coverage-plan.md` §6 W3 / D6)。
488///
489/// 为什么单独给个常量:agentd 的 `argv` 与状态处理要**按 ID 分支**,字面量散在两侧
490/// 迟早会漂(一处改名、另一处还在按旧名特判,表现就是「游标不生效、每轮重读」)。
491pub const KERNEL_LOG_EXPORTER: &str = "journalctl-kernel";
492
493/// 这条来源缺哪些日志能力(空 = 齐了)。
494///
495/// `None` = 这个 kind 不归 logs 能力管(`MetricInterval` 是指标;未知/不可执行的形态
496/// 已经被 [`is_executable_source`] 拦下,不需要第二套说法)。
497///
498/// 返回值是「桶:值」(`file_inputs:glob` / `multiline_modes:indented` /
499/// `filters:namespace_in` / `drop_if_contains:substring` / `exporters:journalctl-kernel`),
500/// 可以直接用在拒绝文案与自观测里。
501///
502/// 为什么收整条来源而不是拆成 `kind` / `target` / `multiline`:每加一个**选源/读法字段**
503/// (W2 的 `filters`、W4 的 `drop_if_contains`)都要重新讨论「要不要传」,而“忘了传”
504/// 在这里的后果是**整份 spec 被旧 agentd 拒收**(`deny_unknown_fields`),不是少采一条。
505/// 收整条来源,新增字段时编译器会直接把所有调用点指出来。
506///
507/// 为什么把判定放契约:网关(派活时)与 agentd(上报时)必须用**同一份**词表。
508/// 两边各写一份的结果是「agent 说自己支持 glob、网关按一套旧名单判它不支持」——
509/// 而且这种矛盾只能靠人去两台机器上对日志才能发现。
510pub fn missing_log_capabilities(
511    source: &WorkSpecSource,
512    declared: &LogsCapabilities,
513) -> Option<Vec<String>> {
514    let declared_has = |bucket: &[String], value: &str| bucket.iter().any(|item| item == value);
515    let kind = source.kind.as_str();
516    let target = source.target.as_str();
517
518    match kind {
519        "FileGlob" => {
520            // 不可执行的形态交给 `is_executable_source` 说,不在这里扣一顶“能力不够”的帽子。
521            if !is_file_target(target) {
522                return None;
523            }
524            let mut missing = Vec::new();
525            let need = if is_glob_path(target) {
526                FILE_INPUT_GLOB
527            } else {
528                FILE_INPUT_ABSOLUTE_PATH
529            };
530            if !declared_has(&declared.file_inputs, need) {
531                missing.push(format!("file_inputs:{need}"));
532            }
533            // `none` 是老读法(一行一条),不算一项需要协商的能力。
534            if source.multiline != "none"
535                && !declared_has(&declared.multiline_modes, &source.multiline)
536            {
537                missing.push(format!("multiline_modes:{}", source.multiline));
538            }
539            // 过滤器是**选源层**的能力(与读法正交):旧 agentd 收到 `filters` 会拒收整份
540            // spec,所以逐维度点名。
541            for dimension in source.filters.used_dimensions() {
542                if !declared_has(&declared.filters, dimension) {
543                    missing.push(format!("filters:{dimension}"));
544                }
545            }
546            // 噪音行丢弃(W4):同样是新字段,旧 agentd 会把整份 spec 当未知字段拒收。
547            // 注意:网关对这项**不裁源**,而是去掉字段降级(它是减法,少采不如多采);
548            // 这里仍如实报告“缺这项能力”,供其它消费方与诊断使用。
549            if !source.drop_if_contains.is_empty()
550                && !declared_has(&declared.drop_if_contains, DROP_IF_CONTAINS_SUBSTRING)
551            {
552                missing.push(format!("drop_if_contains:{DROP_IF_CONTAINS_SUBSTRING}"));
553            }
554            Some(missing)
555        }
556        "Exporter" => {
557            let (id, _) = parse_exporter_target(target)?;
558            Some(if declared_has(&declared.exporters, id) {
559                Vec::new()
560            } else {
561                vec![format!("exporters:{id}")]
562            })
563        }
564        _ => None,
565    }
566}
567
568/// agentd 能承接的**文件输入形态**(能力名)。
569///
570/// 与 [`is_file_target`] 同源:判定能采的形态,就是这里要声明的形态 —— 两边各写一份必然漂。
571pub const FILE_INPUT_ABSOLUTE_PATH: &str = "absolute-path";
572/// 见 [`FILE_INPUT_ABSOLUTE_PATH`]。
573pub const FILE_INPUT_GLOB: &str = "glob";
574
575/// agentd 支持的 `multiline` 模式(与 [`WorkSpecSource::multiline`] 的取值同源)。
576///
577/// * `none` / `indented`:W1;
578/// * `mysql_slow`:MySQL slow log,`# Time:` 起块(W4);
579/// * `postgres_csv`:PostgreSQL csvlog,**引号外**的换行才切记录(W4)。
580pub const MULTILINE_MODES: &[&str] = &["none", "indented", "mysql_slow", "postgres_csv"];
581
582/// 定时导出器(`Exporter`)的**已知 ID**:协议层词表,网关 / agentd / web 三侧共用。
583///
584/// 为什么词表放这里:`is_executable_source` 要判「这条 `Exporter` 目标今天真能不能采」,
585/// 而网关的采集就绪度与 agentd 的「接不接得了」必须是**同一个判据**(见 [`is_executable_source`])。
586/// 词表只收**已实现**的导出器 —— 收录一个 ID 就等于承诺 agentd 能跑它;未知 ID(旧网关发新 ID、
587/// 或写错)一律当不可采,不要「写了就算」。新增能力 = 四侧(本 crate + 网关 + agentd + web)
588/// 一起改 + 知识库把对应单元开 `active`。
589pub const EXPORTER_IDS: &[&str] = &[
590    "journalctl-unit",
591    "journalctl-shutdown",
592    "last-reboot",
593    "nft-ruleset",
594    "iptables-save",
595    "smartctl",
596    "dmesg",
597    "auditd-execve",
598    KERNEL_LOG_EXPORTER,
599];
600
601/// 把一个 `Exporter` 目标拆成 `(id, arg)`:`"dmesg:panic"` → `("dmesg", Some("panic"))`。
602///
603/// 无 `:` 时 `arg` 为 `None`;空 id 或含空白的 id 视为无效(返回 `None`)。`arg` 是**声明式的窄选项**
604/// (如 `dmesg` 的过滤键),不是命令行片段 —— 实现侧各自解释,绝不拼进 shell。
605pub fn parse_exporter_target(target: &str) -> Option<(&str, Option<&str>)> {
606    let (id, arg) = match target.split_once(':') {
607        Some((id, arg)) => (id, Some(arg)),
608        None => (target, None),
609    };
610    let id = id.trim();
611    if id.is_empty() || id.contains(char::is_whitespace) {
612        return None;
613    }
614    Some((id, arg))
615}
616
617/// 这条 `Exporter` 目标的 ID 是不是**已知且已实现**的导出器。
618pub fn is_known_exporter(target: &str) -> bool {
619    match parse_exporter_target(target) {
620        Some((id, _)) => EXPORTER_IDS.contains(&id),
621        None => false,
622    }
623}
624
625/// 一条来源今天**能不能被采**:`kind` 要 agentd 能执行,`target` 还要是它接得住的形态。
626///
627/// 为什么把两件事放在一个函数里:网关的采集就绪度(「这个面能不能派下去」)与 agentd 的
628/// 「这条来源接不接得了」必须是**同一个判据**,否则会出现「网关说可采、agent 拿到后报
629/// unsupported」这种没人能发现的矛盾。加一种 kind 支持时**只改这里**。
630pub fn is_executable_source(kind: &str, target: &str) -> bool {
631    match kind {
632        // 本地文件采集:显式绝对路径,或受支持的文件族 glob(无 `**`、无 `~`)。
633        "FileGlob" => is_file_target(target),
634        // 指标:`target` 是采集周期(如 `15s`、`60s`),不是路径。
635        "MetricInterval" => true,
636        // 定时导出器:`target` 要是**已知导出器 ID**(可带 `:arg`)。未知 ID 不可执行。
637        "Exporter" => is_known_exporter(target),
638        _ => false,
639    }
640}
641
642impl WorkSpecUnit {
643    /// 该单元里可以交给本地文件采集器的来源(`FileGlob`)。
644    ///
645    /// 其它来源(导出器 / 统一日志谓词)不是本地 tail 一个文件能承接的:
646    /// 调用方应当把它们当成**还没支持的单元**如实报出来,而不是当没看见。
647    pub fn file_sources(&self) -> Vec<&WorkSpecSource> {
648        self.sources
649            .iter()
650            .filter(|source| source.kind == "FileGlob")
651            .collect()
652    }
653
654    /// 本单元有没有 agentd 目前接不了的来源(kind 不认识,或目标不是可采的绝对路径)。
655    pub fn unsupported_sources(&self) -> Vec<&WorkSpecSource> {
656        self.sources
657            .iter()
658            .filter(|source| !is_executable_source(&source.kind, &source.target))
659            .collect()
660    }
661
662    /// 本单元至少有一条来源是 agentd **今天真能采**的。
663    ///
664    /// 「能不能采」只看这个:与解析规则(`rule_ref`)无关 —— 采原文不需要规则,
665    /// 规则只决定采下来的东西能不能被归类、抽字段。
666    pub fn has_executable_source(&self) -> bool {
667        self.sources
668            .iter()
669            .any(|source| is_executable_source(&source.kind, &source.target))
670    }
671}
672
673/// 解析 `15s` / `60s` / `5m` 这类周期写法(与目录里 `MetricInterval` 的写法一致)。
674///
675/// 只认秒与分两种后缀:目录里的值是人写的,多一个单位就多一种笔误的可能,
676/// 而解析不了时返回 `None` 会让调用方回退到自己的默认值(不静默取 0)。
677pub fn parse_interval_seconds(raw: &str) -> Option<i64> {
678    let raw = raw.trim();
679    if let Some(value) = raw.strip_suffix('s') {
680        return value.trim().parse::<i64>().ok().filter(|value| *value > 0);
681    }
682    if let Some(value) = raw.strip_suffix('m') {
683        let minutes = value.trim().parse::<i64>().ok()?;
684        // `checked_mul`:`123456789012345678m` 这类输入会让 `minutes * 60` 溢出 i64。
685        // 契约是「解析不了返回 None」(调用方回退到自己的默认值),不能让它 panic(debug)
686        // 或悄悄回绕成一个错的间隔(release)。
687        return minutes.checked_mul(60).filter(|value| *value > 0);
688    }
689    None
690}
691
692#[cfg(test)]
693mod tests {
694    use super::*;
695    use crate::capability_report::LogsCapabilities;
696
697    fn one_shot(status: &str, interruptible: bool) -> OneShotWork {
698        OneShotWork {
699            work_id: "work-1".to_string(),
700            agent_id: "agent-1".to_string(),
701            action: "upgrade".to_string(),
702            spec: "0.1.4".to_string(),
703            scheduled_at: "2026-09-23T00:00:00Z".to_string(),
704            deadline_at: "2026-09-24T00:00:00Z".to_string(),
705            timeout_seconds: 600,
706            interruptible,
707            status: status.to_string(),
708            paused_at: None,
709            paused_total_seconds: 0,
710            current_step: None,
711            completed_steps: vec![],
712            attempt: 0,
713            issued_by: "admin".to_string(),
714            issued_at: "2026-09-23T00:00:00Z".to_string(),
715        }
716    }
717
718    #[test]
719    fn terminal_one_shot_work_is_not_outstanding() {
720        // 未了结的活才进快照:终态留在库里供审计,但不必再发给 agent。
721        for status in ONE_SHOT_TERMINAL_STATUSES {
722            assert!(!one_shot(status, true).is_outstanding(), "{status}");
723        }
724        for status in ["dispatched", "accepted", "running", "paused"] {
725            assert!(one_shot(status, true).is_outstanding(), "{status}");
726        }
727    }
728
729    #[test]
730    fn only_interruptible_running_work_can_pause() {
731        // 不可中断:拒绝,而不是「尽力暂停」。
732        assert!(!one_shot("running", false).can_pause());
733        // 可中断但没在做(还没到点 / 已经了结):也没什么可暂停的。
734        assert!(!one_shot("dispatched", true).can_pause());
735        assert!(!one_shot("succeeded", true).can_pause());
736        assert!(one_shot("accepted", true).can_pause());
737        assert!(one_shot("running", true).can_pause());
738    }
739
740    #[test]
741    fn work_kind_serializes_as_the_model_names() {
742        assert_eq!(
743            serde_json::to_string(&WorkKind::Standing).expect("serialize"),
744            "\"Standing\""
745        );
746        assert_eq!(
747            serde_json::to_string(&WorkKind::OneShot).expect("serialize"),
748            "\"OneShot\""
749        );
750    }
751
752    #[test]
753    fn only_explicit_absolute_paths_are_collectable_file_targets() {
754        // `is_explicit_path` 现在只服务「单文件」语义(例如存在性裁剪只能对显式路径做)。
755        // 这个判据被两侧共用(网关的采集就绪度 + agentd 的接受逻辑),所以在这里钉死。
756        assert!(is_explicit_path("/var/log/install.log"));
757        assert!(!is_explicit_path("/var/log/wifi.log*"));
758        assert!(!is_explicit_path("/Library/Logs/DiagnosticReports/*.ips"));
759        assert!(!is_explicit_path("~/Library/Logs/Homebrew/*"));
760        assert!(!is_explicit_path("relative/app.log"));
761    }
762
763    #[test]
764    fn file_targets_accept_single_level_globs_and_reject_recursion() {
765        // W1:`FileGlob` 判据从「显式路径」放宽为「绝对路径,允许通配」。
766        assert!(is_glob_path("/var/log/containers/*.log"));
767        assert!(is_glob_path("/var/log/pods/*/*/*.log"));
768        assert!(is_glob_path("/var/lib/postgresql/*/data/log/*.csv"));
769        assert!(is_file_target("/var/log/containers/*.log"));
770        // 显式路径仍是合法文件目标(单文件 input 走它)。
771        assert!(is_file_target("/var/log/auth.log"));
772        // `**` 不展开:采集端只扫固定前缀目录,接受它只会在采集端变成每轮 IO 错误。
773        assert!(!is_glob_path("/var/log/**/*.log"));
774        assert!(!is_file_target("/var/log/**/*.log"));
775        // `~` / 相对路径不展开。
776        assert!(!is_glob_path("~/Library/Logs/*.log"));
777        assert!(!is_file_target("relative/*.log"));
778        // 空目标不是文件目标。
779        assert!(!is_file_target(""));
780    }
781
782    #[test]
783    fn executability_needs_both_a_supported_kind_and_a_supported_target() {
784        // 指标:`target` 是周期不是路径,不管显式路径那一条。
785        assert!(is_executable_source("MetricInterval", "15s"));
786        // 文件:显式路径与单层 glob 都可执行;`**` / `~` 不可执行(网关不能把它们报成“可采”)。
787        assert!(is_executable_source("FileGlob", "/var/log/app.log"));
788        assert!(is_executable_source(
789            "FileGlob",
790            "/var/log/containers/*.log"
791        ));
792        assert!(!is_executable_source("FileGlob", "/var/log/**/*.log"));
793        assert!(!is_executable_source("FileGlob", "~/logs/*.log"));
794        // 导出器:kind 对了,ID 还得是**已知的**(`last,lastb` 是 macOS 侧、尚未实现)。
795        assert!(is_executable_source("Exporter", "smartctl"));
796        assert!(is_executable_source("Exporter", "dmesg:panic"));
797        assert!(!is_executable_source("Exporter", "last,lastb"));
798        // 还没实现的 kind:一律不可执行。
799        assert!(!is_executable_source("UnifiedLogPredicate", "syspolicyd"));
800    }
801
802    #[test]
803    fn exporter_targets_split_into_a_known_id_and_an_optional_arg() {
804        assert_eq!(parse_exporter_target("smartctl"), Some(("smartctl", None)));
805        assert_eq!(
806            parse_exporter_target("dmesg:panic"),
807            Some(("dmesg", Some("panic")))
808        );
809        // 空 id / 含空白的 id 不是合法 ID。
810        assert_eq!(parse_exporter_target(":panic"), None);
811        assert_eq!(parse_exporter_target(" dm esg"), None);
812        // 判定只看冒号前的 ID,`arg` 不参与。
813        assert!(is_known_exporter("dmesg:nvidia-xid"));
814        assert!(!is_known_exporter("nope"));
815    }
816
817    #[test]
818    fn the_kernel_log_exporter_is_in_the_contract_vocabulary() {
819        // 词表只增不改名:这一条丢了,网关会把内核单元判成「不可采」(`active` 直接被拦),
820        // agentd 也会把它报成 unsupported。
821        assert!(EXPORTER_IDS.contains(&KERNEL_LOG_EXPORTER));
822        assert!(is_known_exporter(KERNEL_LOG_EXPORTER));
823        assert!(is_executable_source("Exporter", KERNEL_LOG_EXPORTER));
824        // 前缀相似但不是同一个 ID:不能靠 starts_with 之类的模糊匹配混过去。
825        assert!(!is_known_exporter("journalctl-kernelx"));
826    }
827
828    // ── 能力协商(D7):词表与判定同源 ──
829
830    /// 一条测试来源:只关心 kind / target / multiline 时用它(不带 filters / exclude)。
831    fn source(kind: &str, target: &str, multiline: &str) -> WorkSpecSource {
832        WorkSpecSource {
833            kind: kind.to_string(),
834            target: target.to_string(),
835            multiline: multiline.to_string(),
836            exclude_globs: Vec::new(),
837            filters: ContainerFilters::default(),
838            drop_if_contains: Vec::new(),
839        }
840    }
841
842    /// 一条容器来源(`FileGlob` + 给定过滤器)。
843    fn container_source(target: &str, filters: ContainerFilters) -> WorkSpecSource {
844        WorkSpecSource {
845            filters,
846            ..source("FileGlob", target, "none")
847        }
848    }
849
850    /// [`missing_log_capabilities`] 的简写:不带筛选条件的来源。
851    fn missing(
852        kind: &str,
853        target: &str,
854        multiline: &str,
855        declared: &LogsCapabilities,
856    ) -> Option<Vec<String>> {
857        missing_log_capabilities(&source(kind, target, multiline), declared)
858    }
859
860    /// 一个“什么都支持”的 agent 声明(agentd 上报的就是这一份的全集)。
861    fn full_logs() -> LogsCapabilities {
862        LogsCapabilities {
863            file_inputs: vec![
864                FILE_INPUT_ABSOLUTE_PATH.to_string(),
865                FILE_INPUT_GLOB.to_string(),
866            ],
867            multiline_modes: MULTILINE_MODES
868                .iter()
869                .map(|mode| mode.to_string())
870                .collect(),
871            exporters: EXPORTER_IDS.iter().map(|id| id.to_string()).collect(),
872            filters: CONTAINER_FILTERS.iter().map(|di| di.to_string()).collect(),
873            drop_if_contains: DROP_IF_CONTAINS_MODES
874                .iter()
875                .map(|m| m.to_string())
876                .collect(),
877            ..LogsCapabilities::default()
878        }
879    }
880
881    #[test]
882    fn capability_gaps_name_the_bucket_and_value() {
883        let old_agent = LogsCapabilities {
884            // W1 之前的 agentd:只会显式路径,不认识导出器能力。
885            file_inputs: vec![FILE_INPUT_ABSOLUTE_PATH.to_string()],
886            ..LogsCapabilities::default()
887        };
888        assert_eq!(
889            missing("FileGlob", "/var/log/app.log", "none", &old_agent),
890            Some(Vec::new())
891        );
892        assert_eq!(
893            missing("FileGlob", "/var/log/app/*.log", "none", &old_agent),
894            Some(vec!["file_inputs:glob".to_string()])
895        );
896        // 多行模式也是能力:说错了会把记录拼碎。
897        assert_eq!(
898            missing("FileGlob", "/var/log/a.log", "indented", &old_agent),
899            Some(vec!["multiline_modes:indented".to_string()])
900        );
901        // 导出器是具名能力:新 ID(如 W3 的内核日志)旧 agent 一定不认识。
902        assert_eq!(
903            missing("Exporter", KERNEL_LOG_EXPORTER, "none", &old_agent),
904            Some(vec![format!("exporters:{KERNEL_LOG_EXPORTER}")])
905        );
906        // `arg` 不参与判定(与 `is_known_exporter` 同口径)。
907        assert_eq!(
908            missing("Exporter", "dmesg:panic", "none", &full_logs()),
909            Some(Vec::new())
910        );
911        // 空的过滤器不要求任何能力:它不上线(`skip_serializing_if`)。
912        assert_eq!(
913            missing("FileGlob", "/var/log/containers/*.log", "none", &old_agent),
914            Some(vec!["file_inputs:glob".to_string()])
915        );
916    }
917
918    #[test]
919    fn container_filter_dimensions_are_negotiated_separately() {
920        // 0.9.4 的 agentd:已经有 glob / multiline / exporters,但**不认识 `filters`**。
921        // 带 filters 的来源必须被点名 —— 否则那份 spec 会被 `deny_unknown_fields` 整份拒收。
922        let before_w2 = LogsCapabilities {
923            filters: Vec::new(),
924            ..full_logs()
925        };
926        let namespace_only = ContainerFilters {
927            namespace_in: vec!["prod".to_string()],
928            pod_not_in: Vec::new(),
929        };
930        let target = "/var/log/containers/*.log";
931        assert_eq!(
932            missing_log_capabilities(&container_source(target, namespace_only), &before_w2),
933            Some(vec!["filters:namespace_in".to_string()])
934        );
935        // 两个维度都用到 → 两个都点名(按字段顺序,文案可对账)。
936        let both = ContainerFilters {
937            namespace_in: vec!["prod".to_string()],
938            pod_not_in: vec!["debug-pod".to_string()],
939        };
940        assert_eq!(
941            missing_log_capabilities(&container_source(target, both.clone()), &before_w2),
942            Some(vec![
943                "filters:namespace_in".to_string(),
944                "filters:pod_not_in".to_string(),
945            ])
946        );
947        // 声明了这两个维度就齐了。
948        assert_eq!(
949            missing_log_capabilities(&container_source(target, both), &full_logs()),
950            Some(Vec::new())
951        );
952        // 能力名就是过滤器的字段名(网关文案与目录写法能一一对上)。
953        assert_eq!(
954            ContainerFilters {
955                namespace_in: vec!["a".to_string()],
956                pod_not_in: vec!["b".to_string()],
957            }
958            .used_dimensions(),
959            vec![FILTER_NAMESPACE_IN, FILTER_POD_NOT_IN]
960        );
961        assert!(ContainerFilters::default().used_dimensions().is_empty());
962    }
963
964    #[test]
965    fn noise_drop_needles_are_a_capability_of_their_own() {
966        // W4 之前的 agentd:有 glob / filters,但**不认识 `drop_if_contains`**。
967        // 不点名就会把整份 spec 送给一个 `deny_unknown_fields` 的旧版本。
968        let before_w4 = LogsCapabilities {
969            drop_if_contains: Vec::new(),
970            ..full_logs()
971        };
972        let mut source = source("FileGlob", "/var/log/auth.log", "none");
973        source.drop_if_contains = vec!["pam_unix(cron:session)".to_string()];
974        assert_eq!(
975            missing_log_capabilities(&source, &before_w4),
976            Some(vec!["drop_if_contains:substring".to_string()])
977        );
978        // 声明了(本版只有子串一种)就齐了。
979        assert_eq!(
980            missing_log_capabilities(&source, &full_logs()),
981            Some(Vec::new())
982        );
983    }
984
985    #[test]
986    fn drop_if_contains_is_substring_shaped_and_stays_out_of_the_wire_when_empty() {
987        // 空列表不上线(与 `exclude_globs` / `filters` 同一条纪律)。
988        let mut source = source("FileGlob", "/var/log/auth.log", "none");
989        assert!(
990            !serde_json::to_string(&source)
991                .expect("serialize")
992                .contains("drop_if_contains")
993        );
994
995        source.drop_if_contains = vec!["pam_unix(cron:session)".to_string()];
996        let json = serde_json::to_string(&source).expect("serialize");
997        assert!(
998            json.contains(r#""drop_if_contains":["pam_unix(cron:session)"]"#),
999            "{json}"
1000        );
1001        assert_eq!(
1002            serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1003            source
1004        );
1005
1006        // 形状校验:空串 / 首尾空白 / 控制字符 / 过长 / 太多 / 重复都要拒。
1007        assert!(validate_drop_if_contains(&["ok".to_string()]).is_ok());
1008        assert!(validate_drop_if_contains(&[]).is_ok());
1009        for bad in ["", " x", "x ", "line\nbreak", " nul\0"] {
1010            assert!(
1011                validate_drop_if_contains(&[bad.to_string()]).is_err(),
1012                "{bad:?} should be rejected"
1013            );
1014        }
1015        assert!(validate_drop_if_contains(&["x".repeat(DROP_IF_CONTAINS_MAX_BYTES + 1)]).is_err());
1016        let too_many: Vec<String> = (0..=DROP_IF_CONTAINS_MAX_ITEMS)
1017            .map(|i| format!("needle-{i}"))
1018            .collect();
1019        assert!(validate_drop_if_contains(&too_many).is_err());
1020        assert!(validate_drop_if_contains(&["dup".to_string(), "dup".to_string()]).is_err());
1021    }
1022
1023    #[test]
1024    fn the_new_multiline_modes_are_in_the_contract_vocabulary() {
1025        // 网关装载期与 agentd 配置校验都看这份词表:漏一项,目录里写了也会被当成非法。
1026        for mode in ["mysql_slow", "postgres_csv"] {
1027            assert!(MULTILINE_MODES.contains(&mode), "{mode}");
1028        }
1029        // 新读法一律要求能力声明(`none` 除外)。
1030        let old_agent = LogsCapabilities {
1031            multiline_modes: vec!["none".to_string(), "indented".to_string()],
1032            ..full_logs()
1033        };
1034        assert_eq!(
1035            missing(
1036                "FileGlob",
1037                "/var/log/postgresql/x.csv",
1038                "postgres_csv",
1039                &old_agent
1040            ),
1041            Some(vec!["multiline_modes:postgres_csv".to_string()])
1042        );
1043        assert_eq!(
1044            missing(
1045                "FileGlob",
1046                "/var/log/mysql/slow.log",
1047                "mysql_slow",
1048                &old_agent
1049            ),
1050            Some(vec!["multiline_modes:mysql_slow".to_string()])
1051        );
1052    }
1053
1054    #[test]
1055    fn an_exporter_id_outside_the_vocabulary_is_reported_as_missing() {
1056        // 写错 ID / 未来 ID 遇到旧 agent:都是“没声明这个能力”,不能当成“不归我管”放过 ——
1057        // 放过的后果是派下去、agent 回 unsupported。
1058        assert_eq!(
1059            missing("Exporter", "journalctl-kernelx", "none", &full_logs()),
1060            Some(vec!["exporters:journalctl-kernelx".to_string()])
1061        );
1062        // 带 `arg` 也一样:只按 ID 判。
1063        assert_eq!(
1064            missing("Exporter", "dmesg:panic", "none", &old_style_logs()),
1065            Some(vec!["exporters:dmesg".to_string()])
1066        );
1067    }
1068
1069    /// 只声明了“显式路径 + 任何导出器”这类旧能力的 agent。
1070    fn old_style_logs() -> LogsCapabilities {
1071        LogsCapabilities {
1072            file_inputs: vec![FILE_INPUT_ABSOLUTE_PATH.to_string()],
1073            exporters: vec!["smartctl".to_string()],
1074            ..LogsCapabilities::default()
1075        }
1076    }
1077
1078    #[test]
1079    fn capability_checks_say_nothing_about_things_they_do_not_own() {
1080        // 指标不走 logs 能力;不可执行的形态由 `is_executable_source` 说,不重复扣帽子。
1081        assert_eq!(missing("MetricInterval", "15s", "none", &full_logs()), None);
1082        assert_eq!(missing("FileGlob", "~/a/*.log", "none", &full_logs()), None);
1083        assert_eq!(
1084            missing("UnifiedLogPredicate", "syspolicyd", "none", &full_logs()),
1085            None
1086        );
1087    }
1088
1089    #[test]
1090    fn a_fully_declared_agent_is_missing_nothing_the_contract_calls_executable() {
1091        // 防漂:契约说“能执行”的每一种形态,声明了全集就必须不报缺。
1092        // 新增一个导出器 ID / 一种 multiline 模式 / 一个过滤维度而忘了加到常量里,这条会红。
1093        let full = full_logs();
1094        for target in ["/var/log/a.log", "/var/log/containers/*.log"] {
1095            assert_eq!(
1096                missing("FileGlob", target, "none", &full),
1097                Some(Vec::new()),
1098                "{target}"
1099            );
1100        }
1101        for mode in MULTILINE_MODES {
1102            assert_eq!(
1103                missing("FileGlob", "/var/log/a.log", mode, &full),
1104                Some(Vec::new()),
1105                "{mode}"
1106            );
1107        }
1108        for id in EXPORTER_IDS {
1109            assert_eq!(
1110                missing("Exporter", id, "none", &full),
1111                Some(Vec::new()),
1112                "{id}"
1113            );
1114        }
1115        // 每一个过滤维度都真的能“声明全了就不报缺”。
1116        let filters = ContainerFilters {
1117            namespace_in: vec!["prod".to_string()],
1118            pod_not_in: vec!["debug-pod".to_string()],
1119        };
1120        assert_eq!(
1121            filters.used_dimensions(),
1122            CONTAINER_FILTERS,
1123            "`used_dimensions` 与 `CONTAINER_FILTERS` 漂了"
1124        );
1125        assert_eq!(
1126            missing_log_capabilities(
1127                &container_source("/var/log/containers/*.log", filters),
1128                &full
1129            ),
1130            Some(Vec::new())
1131        );
1132    }
1133
1134    // ── 工作参数(`spec`)的编码 ──
1135
1136    fn spec() -> WorkSpec {
1137        WorkSpec {
1138            units: vec![
1139                WorkSpecUnit {
1140                    unit_id: "mac-host-metrics".to_string(),
1141                    capability: "collect_metrics".to_string(),
1142                    rule_ref: "agent_uplink".to_string(),
1143                    requires_privilege: "none".to_string(),
1144                    sources: vec![WorkSpecSource {
1145                        kind: "MetricInterval".to_string(),
1146                        target: "15s".to_string(),
1147                        multiline: "none".to_string(),
1148                        exclude_globs: Vec::new(),
1149                        filters: ContainerFilters::default(),
1150                        drop_if_contains: Vec::new(),
1151                    }],
1152                },
1153                WorkSpecUnit {
1154                    unit_id: "mac-privacy-tcc".to_string(),
1155                    capability: "collect_logs".to_string(),
1156                    rule_ref: "macos/tcc".to_string(),
1157                    requires_privilege: "fda".to_string(),
1158                    sources: vec![
1159                        WorkSpecSource {
1160                            kind: "Exporter".to_string(),
1161                            target: "sqlite-snapshot(TCC.db)".to_string(),
1162                            multiline: "none".to_string(),
1163                            exclude_globs: Vec::new(),
1164                            filters: ContainerFilters::default(),
1165                            drop_if_contains: Vec::new(),
1166                        },
1167                        WorkSpecSource {
1168                            kind: "FileGlob".to_string(),
1169                            // 显式单路径:单文件采集走 `startup_position` / 检查点的单文件语义。
1170                            target: "/var/log/tccd.log".to_string(),
1171                            // 多行格式是这条日志的固有属性,跟着来源走。
1172                            multiline: "indented".to_string(),
1173                            exclude_globs: Vec::new(),
1174                            filters: ContainerFilters::default(),
1175                            drop_if_contains: Vec::new(),
1176                        },
1177                    ],
1178                },
1179            ],
1180        }
1181    }
1182
1183    #[test]
1184    fn spec_round_trips_and_reports_what_agentd_can_do() {
1185        let encoded = spec().encode().expect("encode");
1186        let decoded = WorkSpec::parse(&encoded).expect("parse");
1187        assert_eq!(decoded, spec());
1188
1189        assert!(decoded.collects_logs());
1190        // 取了最密的周期:要得最急的才是上送频率的下界。
1191        assert_eq!(decoded.metric_interval_seconds(), Some(15));
1192        // 可本地 tail 的来源挑得出来,接不了的来源也报得出来(而不是当没看见)。
1193        let files = decoded.units[1].file_sources();
1194        assert_eq!(files.len(), 1);
1195        assert_eq!(files[0].target, "/var/log/tccd.log");
1196        assert_eq!(files[0].multiline, "indented");
1197        assert_eq!(decoded.units[1].unsupported_sources().len(), 1);
1198        assert_eq!(decoded.units[1].unsupported_sources()[0].kind, "Exporter");
1199    }
1200
1201    #[test]
1202    fn a_glob_file_source_is_collectable_but_recursive_globs_are_not() {
1203        // W1 起:单层 glob 是采集端能执行的来源(容器日志就靠它)。
1204        let mut unit = WorkSpecUnit {
1205            unit_id: "linux-container-stdout".to_string(),
1206            capability: "collect_logs".to_string(),
1207            rule_ref: String::new(),
1208            requires_privilege: "root".to_string(),
1209            sources: vec![WorkSpecSource {
1210                kind: "FileGlob".to_string(),
1211                target: "/var/log/containers/*.log".to_string(),
1212                multiline: "none".to_string(),
1213                exclude_globs: vec!["/var/log/containers/*_kube-system_*.log".to_string()],
1214                filters: ContainerFilters::default(),
1215                drop_if_contains: Vec::new(),
1216            }],
1217        };
1218        assert_eq!(unit.file_sources().len(), 1, "它是一条文件来源");
1219        assert!(unit.unsupported_sources().is_empty());
1220        assert!(unit.has_executable_source());
1221
1222        // `**` 仍不可执行:递归展开没实现,接受它就等于承诺一个采不到的面。
1223        unit.sources[0].target = "/var/log/**/*.log".to_string();
1224        assert_eq!(unit.unsupported_sources().len(), 1);
1225        assert!(!unit.has_executable_source());
1226    }
1227
1228    #[test]
1229    fn exclude_globs_round_trip_and_stay_out_of_the_wire_when_empty() {
1230        let mut source = WorkSpecSource {
1231            kind: "FileGlob".to_string(),
1232            target: "/var/log/containers/*.log".to_string(),
1233            multiline: "none".to_string(),
1234            exclude_globs: Vec::new(),
1235            filters: ContainerFilters::default(),
1236            drop_if_contains: Vec::new(),
1237        };
1238        // 空排除项不进 JSON:旧 agent 的 `WorkSpecSource` 是 `deny_unknown_fields` 的。
1239        assert_eq!(
1240            serde_json::to_string(&source).expect("serialize"),
1241            r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","multiline":"none"}"#
1242        );
1243
1244        source.exclude_globs = vec!["/var/log/containers/*_kube-system_*.log".to_string()];
1245        let json = serde_json::to_string(&source).expect("serialize");
1246        assert_eq!(
1247            serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1248            source
1249        );
1250        // 老网关发来的来源没有这两个字段:默认必须能解出来。
1251        let old = r#"{"kind":"FileGlob","target":"/a/*"}"#;
1252        let parsed: WorkSpecSource = serde_json::from_str(old).expect("parse legacy source");
1253        assert!(parsed.exclude_globs.is_empty());
1254        assert!(parsed.filters.is_empty());
1255        assert_eq!(parsed.multiline, "none");
1256    }
1257
1258    #[test]
1259    fn container_filters_round_trip_and_stay_out_of_the_wire_when_empty() {
1260        let mut source = WorkSpecSource {
1261            kind: "FileGlob".to_string(),
1262            target: "/var/log/containers/*.log".to_string(),
1263            multiline: "none".to_string(),
1264            exclude_globs: Vec::new(),
1265            filters: ContainerFilters::default(),
1266            drop_if_contains: Vec::new(),
1267        };
1268        // 空过滤器不进 JSON(与 exclude_globs 同一条纪律:旧 agent 是 deny_unknown_fields)。
1269        let json = serde_json::to_string(&source).expect("serialize");
1270        assert!(!json.contains("filters"), "{json}");
1271
1272        source.filters = ContainerFilters {
1273            namespace_in: vec!["prod".to_string()],
1274            pod_not_in: vec!["debug-pod".to_string()],
1275        };
1276        let json = serde_json::to_string(&source).expect("serialize");
1277        assert!(
1278            json.contains(r#""filters":{"namespace_in":["prod"],"pod_not_in":["debug-pod"]}"#),
1279            "{json}"
1280        );
1281        assert_eq!(
1282            serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1283            source
1284        );
1285
1286        // 只给一半也要能解:另一侧默认空。
1287        let half = r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","filters":{"namespace_in":["prod"]}}"#;
1288        let parsed: WorkSpecSource = serde_json::from_str(half).expect("parse half filters");
1289        assert_eq!(parsed.filters.namespace_in, vec!["prod".to_string()]);
1290        assert!(parsed.filters.pod_not_in.is_empty());
1291    }
1292
1293    #[test]
1294    fn an_explicitly_empty_filters_object_is_not_serialized() {
1295        // 策展/页面可能显式写一个空对象(`filters = {}`):它仍然等于“不过滤”,
1296        // 不能在线上多出一个键 —— 旧 agent 是 `deny_unknown_fields` 的。
1297        let source = WorkSpecSource {
1298            kind: "FileGlob".to_string(),
1299            target: "/var/log/containers/*.log".to_string(),
1300            multiline: "none".to_string(),
1301            exclude_globs: Vec::new(),
1302            filters: ContainerFilters {
1303                namespace_in: Vec::new(),
1304                pod_not_in: Vec::new(),
1305            },
1306            drop_if_contains: Vec::new(),
1307        };
1308        let json = serde_json::to_string(&source).expect("serialize");
1309        assert!(!json.contains("filters"), "{json}");
1310    }
1311
1312    #[test]
1313    fn container_filters_keep_values_verbatim() {
1314        // 名字**不做归一化**(不折叠大小写、不去前后缀):k8s 名字永远小写,
1315        // 归一化反而会把写错的值当成能用。逐字保存、逐字比较(比较在 agentd)。
1316        let filters = ContainerFilters {
1317            namespace_in: vec!["Prod".to_string()],
1318            pod_not_in: vec!["debug-pod".to_string()],
1319        };
1320        let json = serde_json::to_string(&filters).expect("serialize");
1321        let parsed: ContainerFilters = serde_json::from_str(&json).expect("parse");
1322        assert_eq!(parsed.namespace_in, vec!["Prod".to_string()]);
1323        assert_eq!(parsed.pod_not_in, vec!["debug-pod".to_string()]);
1324    }
1325
1326    #[test]
1327    fn container_filters_reject_unknown_keys() {
1328        // `deny_unknown_fields`:拼错一个键(`namespaces`)不能静默当空过滤器 ——
1329        // 那会让策展人以为「只采 prod」,实际全采。
1330        assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespaces":["prod"]}"#).is_err());
1331        assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespace_in":["prod"]}"#).is_ok());
1332    }
1333
1334    #[test]
1335    fn container_filters_reject_values_that_can_never_match() {
1336        let ok = |namespace: &[&str], pods: &[&str]| ContainerFilters {
1337            namespace_in: namespace.iter().map(|v| v.to_string()).collect(),
1338            pod_not_in: pods.iter().map(|v| v.to_string()).collect(),
1339        };
1340
1341        assert!(
1342            ok(&["prod", "kube-system"], &["debug-pod"])
1343                .validate()
1344                .is_ok()
1345        );
1346        assert!(ok(&[], &[]).is_empty());
1347        // 这些一定对不上(过滤器存的是**精确名字**,不是 pattern):错在最早的入口就说。
1348        assert!(ok(&[""], &[]).validate().is_err());
1349        assert!(ok(&[" prod"], &[]).validate().is_err());
1350        assert!(ok(&["prod*"], &[]).validate().is_err());
1351        assert!(ok(&[], &["debug?pod"]).validate().is_err());
1352        assert!(ok(&[&"x".repeat(254)], &[]).validate().is_err());
1353    }
1354
1355    #[test]
1356    fn a_source_without_multiline_decodes_as_single_line() {
1357        // 老网关发来的 spec 没有 `multiline`:默认必须是一行一条。
1358        // 取错默认值会把多行日志**粘**成一条 —— 那是无声的内容损坏,不是格式问题。
1359        let json = r#"{"units":[{"unit_id":"u","capability":"collect_logs","sources":[{"kind":"FileGlob","target":"/a/*"}]}]}"#;
1360        let spec = WorkSpec::parse(json).expect("parse");
1361        assert_eq!(spec.units[0].sources[0].multiline, "none");
1362        // 再编码出去会把默认值写实:新网关发的 spec 字段总是齐的。
1363        assert_eq!(
1364            spec.encode().expect("encode"),
1365            r#"{"units":[{"unit_id":"u","capability":"collect_logs","rule_ref":"","requires_privilege":"","sources":[{"kind":"FileGlob","target":"/a/*","multiline":"none"}]}]}"#
1366        );
1367    }
1368
1369    #[test]
1370    fn a_broken_spec_is_an_error_not_an_empty_work() {
1371        // 参数坏了与「没事可做」是两回事:前者要人去看,后者什么都不用做。
1372        assert!(WorkSpec::parse("mac-host-metrics").is_err());
1373        assert!(WorkSpec::parse("{").is_err());
1374        // 空工作本身是合法的(能表示「这个面暂时没东西可采」)。
1375        assert!(WorkSpec::parse(r#"{"units":[]}"#).unwrap().units.is_empty());
1376        assert_eq!(
1377            WorkSpec::parse(r#"{"units":[]}"#)
1378                .unwrap()
1379                .metric_interval_seconds(),
1380            None
1381        );
1382    }
1383
1384    #[test]
1385    fn interval_parsing_accepts_seconds_and_minutes_only() {
1386        assert_eq!(parse_interval_seconds("15s"), Some(15));
1387        assert_eq!(parse_interval_seconds(" 5s "), Some(5));
1388        assert_eq!(parse_interval_seconds("5m"), Some(300));
1389        // 认不出来的回 None(调用方回退到自己的默认值),而不是静默当 0。
1390        assert_eq!(parse_interval_seconds("15"), None);
1391        assert_eq!(parse_interval_seconds("0s"), None);
1392        assert_eq!(parse_interval_seconds(""), None);
1393    }
1394
1395    #[test]
1396    fn interval_parsing_never_overflows_on_huge_values() {
1397        // 契约是「解析不了返回 None」,不是 panic:溢出在 debug 下 panic、release 下回绕成一个
1398        // 错误的间隔 —— 两个后果都不能接受。这是两侧都要解析的字节,输入可能来自远端。
1399        assert_eq!(parse_interval_seconds("922337203685477580m"), None);
1400        assert_eq!(parse_interval_seconds("153722867280912931m"), None);
1401        // 秒分支本来就靠 parse 失败兜底。
1402        assert_eq!(parse_interval_seconds("99999999999999999999s"), None);
1403        // 合法但很大的分钟值仍要算得出来(不误伤)。
1404        assert_eq!(parse_interval_seconds("600m"), Some(36_000));
1405    }
1406
1407    #[test]
1408    fn metric_interval_ignores_unparseable_targets_instead_of_treating_them_as_zero() {
1409        // 认不出的周期不能静默当 0(0 会被当成「无限密」),而要如实跳过,只取能解析的最密值。
1410        let unit = |target: &str| WorkSpecUnit {
1411            unit_id: "u".to_string(),
1412            capability: "collect_metrics".to_string(),
1413            rule_ref: String::new(),
1414            requires_privilege: String::new(),
1415            sources: vec![WorkSpecSource {
1416                kind: "MetricInterval".to_string(),
1417                target: target.to_string(),
1418                multiline: "none".to_string(),
1419                exclude_globs: Vec::new(),
1420                filters: ContainerFilters::default(),
1421                drop_if_contains: Vec::new(),
1422            }],
1423        };
1424        let mixed = WorkSpec {
1425            units: vec![unit("garbage"), unit("30s")],
1426        };
1427        assert_eq!(mixed.metric_interval_seconds(), Some(30));
1428
1429        let all_bad = WorkSpec {
1430            units: vec![unit("nope")],
1431        };
1432        assert_eq!(all_bad.metric_interval_seconds(), None);
1433    }
1434}