quanttide_work/
task.rs

1//! 任务聚合：工作流的一次执行实例。
2//!
3//! 指令是跑哪条工作流（`workflow_name`，**按名字**引用）与从哪开工（`start`）；
4//! 状态是流水（只增不改）、闸门项、产物落点；另带这次执行的运行上下文。
5//! 任务是运行数据，不是产物——程序只维护它，不往产物里写字。
6//!
7//! 不可变：`recorded` / `with_gates` 都返回新的任务，改动由调用方落盘。
8
9use std::collections::BTreeMap;
10
11use crate::workflow::Workflow;
12use crate::workflow::text_of;
13use serde_yaml::{Mapping, Value as Yaml};
14
15/// 流水里的一条：什么时候、哪一步、一句话、过没过。
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct JournalEvent {
18    pub at: String,
19    pub step: String,
20    pub detail: String,
21    pub ok: bool,
22}
23
24impl JournalEvent {
25    pub fn of(value: &Yaml) -> JournalEvent {
26        JournalEvent {
27            at: text_of(value, "at"),
28            step: text_of(value, "step"),
29            detail: text_of(value, "detail"),
30            ok: value.get("ok").and_then(|v| v.as_bool()).unwrap_or(false),
31        }
32    }
33
34    pub fn to_yaml(&self) -> Yaml {
35        let mut map = Mapping::new();
36        for (key, value) in [
37            ("at", &self.at),
38            ("step", &self.step),
39            ("detail", &self.detail),
40        ] {
41            map.insert(
42                Yaml::String(key.to_string()),
43                Yaml::String(value.to_string()),
44            );
45        }
46        map.insert(Yaml::String("ok".into()), Yaml::Bool(self.ok));
47        Yaml::Mapping(map)
48    }
49}
50
51/// 这次执行自带的运行上下文：工作区根、数据仓、工作流目录。
52///
53/// 三个字段怎么读、怎么写由各自的包定（工具箱只管托着它们）。
54#[derive(Debug, Clone, Default, PartialEq, Eq)]
55pub struct RunContext {
56    pub root: String,
57    pub data: String,
58    pub workflows: String,
59}
60
61impl RunContext {
62    pub fn of(value: &Yaml) -> RunContext {
63        RunContext {
64            root: text_of(value, "root"),
65            data: text_of(value, "data"),
66            workflows: text_of(value, "workflows"),
67        }
68    }
69
70    pub fn to_yaml(&self) -> Yaml {
71        let mut map = Mapping::new();
72        for (key, value) in [
73            ("root", &self.root),
74            ("data", &self.data),
75            ("workflows", &self.workflows),
76        ] {
77            map.insert(
78                Yaml::String(key.to_string()),
79                Yaml::String(value.to_string()),
80            );
81        }
82        Yaml::Mapping(map)
83    }
84}
85
86/// 去掉尾巴上的斜杠，拼路径不出双斜杠。
87fn trim(path: &str) -> &str {
88    path.strip_suffix('/').unwrap_or(path)
89}
90
91/// 任务聚合。
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct Task {
94    pub name: String,
95    pub workflow_name: String,
96    pub start: String,
97    pub context: RunContext,
98    pub journal: Vec<JournalEvent>,
99    /// 等人拍板的事项。
100    pub gates: Vec<String>,
101    /// 这次执行往哪写产物（声明写成什么就是什么，相对工作区根）。
102    pub artifacts: BTreeMap<String, String>,
103}
104
105impl Task {
106    /// 从任务文件里的字段读出；`name` 由调用方给（文件名即任务名）。
107    pub fn of(name: &str, payload: &Yaml) -> Task {
108        let journal = payload
109            .get("log")
110            .and_then(|v| v.as_sequence())
111            .map(|items| items.iter().map(JournalEvent::of).collect())
112            .unwrap_or_default();
113        let gates = payload
114            .get("gates")
115            .and_then(|v| v.as_sequence())
116            .map(|items| {
117                items
118                    .iter()
119                    .filter_map(|item| item.as_str().map(|text| text.to_string()))
120                    .collect()
121            })
122            .unwrap_or_default();
123        let artifacts = payload
124            .get("artifacts")
125            .and_then(|v| v.as_mapping())
126            .map(|mapping| {
127                mapping
128                    .iter()
129                    .filter_map(|(key, value)| {
130                        Some((key.as_str()?.to_string(), value.as_str()?.to_string()))
131                    })
132                    .collect()
133            })
134            .unwrap_or_default();
135        Task {
136            name: name.to_string(),
137            workflow_name: text_of(payload, "workflow"),
138            start: text_of(payload, "start"),
139            context: RunContext::of(payload),
140            journal,
141            gates,
142            artifacts,
143        }
144    }
145
146    /// 这种产物声明了往哪写；没声明给 `None`。
147    pub fn declared(&self, kind: &str) -> Option<String> {
148        let written = self.artifacts.get(kind)?.trim();
149        if written.is_empty() {
150            None
151        } else {
152            Some(written.to_string())
153        }
154    }
155
156    /// 这次执行往哪写这种产物（规范「任务 / 语法」里的落点）。
157    ///
158    /// 声明了按声明的（相对工作区根）；没声明落数据仓的
159    /// `artifacts/<种类>/<任务名>.md`；流水是任务文件本身。
160    pub fn artifact(&self, kind: &str, context: &RunContext) -> String {
161        let data = trim(&context.data);
162        if kind == "log" {
163            return format!("{data}/tasks/{}.yaml", self.name);
164        }
165        if let Some(written) = self.declared(kind) {
166            return if written.starts_with('/') {
167                written
168            } else {
169                format!("{}/{written}", trim(&context.root))
170            };
171        }
172        format!("{data}/artifacts/{kind}/{}.md", self.name)
173    }
174
175    /// 走过哪几步。
176    ///
177    /// 带后缀的流水（`·审` / `·判`）给这一步**投票**，不带后缀的（重走一遍）
178    /// **把结论从头算**；工作流上没有的步骤名不算数。
179    pub fn done_steps(&self, workflow: &Workflow) -> Vec<String> {
180        let names = workflow.step_names();
181        let mut verdict: Vec<(String, bool)> = Vec::new();
182        for event in &self.journal {
183            let (step, extra) = match event.step.split_once('·') {
184                Some((step, rest)) => (step.to_string(), Some(rest.to_string())),
185                None => (event.step.clone(), None),
186            };
187            if !names.contains(&step) {
188                continue;
189            }
190            match verdict.iter_mut().find(|(name, _)| *name == step) {
191                Some((_, last)) => {
192                    if extra.is_some() {
193                        *last = *last && event.ok;
194                    } else {
195                        *last = event.ok;
196                    }
197                }
198                None => verdict.push((step, event.ok)),
199            }
200        }
201        verdict
202            .into_iter()
203            .filter(|(_, ok)| *ok)
204            .map(|(name, _)| name)
205            .collect()
206    }
207
208    /// 第一个没走到的步骤（按工作流里的顺序）。
209    pub fn next_step(&self, workflow: &Workflow) -> Option<String> {
210        let finished = self.done_steps(workflow);
211        workflow
212            .step_names()
213            .into_iter()
214            .find(|name| !finished.contains(name))
215    }
216
217    /// 状态行：下一步是谁，或者都走过了。
218    pub fn state_line(&self, workflow: &Workflow) -> String {
219        let names = workflow.step_names();
220        if names.is_empty() {
221            return format!(
222                "这条工作流没有步骤——在 workflows/{}.yaml 的 steps 里写步骤",
223                self.workflow_name
224            );
225        }
226        match self.next_step(workflow) {
227            Some(step) => format!("下一步：{step}"),
228            None => format!("{} 个步骤都走过了", names.len()),
229        }
230    }
231
232    /// 记一笔流水：流水只增不改，所以返回新的任务。
233    pub fn recorded(&self, at: &str, step: &str, detail: &str, ok: bool) -> Task {
234        let mut task = self.clone();
235        task.journal.push(JournalEvent {
236            at: at.to_string(),
237            step: step.to_string(),
238            detail: detail.to_string(),
239            ok,
240        });
241        task
242    }
243
244    /// 记下闸门项；已有的不重复记。同样返回新的任务。
245    pub fn with_gates(&self, notes: &[String]) -> Task {
246        let mut task = self.clone();
247        for note in notes {
248            if !task.gates.contains(note) {
249                task.gates.push(note.clone());
250            }
251        }
252        task
253    }
254
255    /// 写回任务文件的字段形状。
256    pub fn to_yaml(&self) -> Yaml {
257        let mut map = Mapping::new();
258        map.insert(Yaml::String("name".into()), Yaml::String(self.name.clone()));
259        map.insert(
260            Yaml::String("start".into()),
261            Yaml::String(self.start.clone()),
262        );
263        map.insert(
264            Yaml::String("workflow".into()),
265            Yaml::String(self.workflow_name.clone()),
266        );
267        map.insert(
268            Yaml::String("log".into()),
269            Yaml::Sequence(self.journal.iter().map(JournalEvent::to_yaml).collect()),
270        );
271        map.insert(
272            Yaml::String("gates".into()),
273            Yaml::Sequence(
274                self.gates
275                    .iter()
276                    .map(|note| Yaml::String(note.clone()))
277                    .collect(),
278            ),
279        );
280        let mut artifacts = Mapping::new();
281        for (key, value) in &self.artifacts {
282            artifacts.insert(Yaml::String(key.clone()), Yaml::String(value.clone()));
283        }
284        map.insert(Yaml::String("artifacts".into()), Yaml::Mapping(artifacts));
285        if let Yaml::Mapping(context) = self.context.to_yaml() {
286            for (key, value) in context {
287                map.insert(key, value);
288            }
289        }
290        Yaml::Mapping(map)
291    }
292}
quanttide_work/task.rs

quanttide_work/
task.rs
