piw/state/
types.rs

1//! Serde types mirroring the SQLite workflow state documents specified in
2//! `docs/run-database runs.md`. Unknown fields are tolerated everywhere so database runs
3//! written by newer writers within the same schema version stay readable.
4
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use std::collections::BTreeMap;
8
9pub const RUN_BUNDLE_SCHEMA: &str = "pi-workflows.run-run.v1";
10pub const RUN_STATE_SCHEMA: &str = "pi-workflows.run-state.v1";
11pub const DEFINITION_SNAPSHOT_SCHEMA: &str = "pi-workflows.definition-snapshot.v1";
12pub const SESSION_BINDING_SCHEMA: &str = "pi-workflows.session-binding.v1";
13pub const SESSION_EVENT_SCHEMA: &str = "pi-workflows.session-event.v1";
14pub const SESSION_CAPTURE_SCHEMA: &str = "pi-workflows.session-capture.v1";
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum RunStatus {
19    Queued,
20    Running,
21    Waiting,
22    Completed,
23    Failed,
24    TimedOut,
25    Cancelled,
26}
27
28impl RunStatus {
29    pub fn is_terminal(self) -> bool {
30        matches!(
31            self,
32            RunStatus::Completed | RunStatus::Failed | RunStatus::TimedOut | RunStatus::Cancelled
33        )
34    }
35
36    pub fn label(self) -> &'static str {
37        match self {
38            RunStatus::Queued => "queued",
39            RunStatus::Running => "running",
40            RunStatus::Waiting => "waiting",
41            RunStatus::Completed => "completed",
42            RunStatus::Failed => "failed",
43            RunStatus::TimedOut => "timed_out",
44            RunStatus::Cancelled => "cancelled",
45        }
46    }
47}
48
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum NodeOutcome {
52    Ok,
53    TimedOut,
54    Failed,
55    Cancelled,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(tag = "kind", rename_all = "snake_case")]
60pub enum WorkflowSource {
61    Builtin { id: String, revision: String },
62    File { path: String, hash: String },
63}
64
65impl WorkflowSource {
66    pub fn display(&self) -> String {
67        match self {
68            WorkflowSource::Builtin { id, revision } => format!("builtin:{id}@{revision}"),
69            WorkflowSource::File { path, .. } => path.clone(),
70        }
71    }
72}
73
74#[cfg(test)]
75mod workflow_source_tests {
76    use super::WorkflowSource;
77
78    #[test]
79    fn parses_and_displays_each_source_kind() {
80        let builtin: WorkflowSource =
81            serde_json::from_str(r#"{"kind":"builtin","id":"monitor","revision":"1"}"#)
82                .expect("built-in source should parse");
83        let file: WorkflowSource =
84            serde_json::from_str(r#"{"kind":"file","path":"/tmp/demo.workflow.ts","hash":"abc"}"#)
85                .expect("file source should parse");
86
87        assert_eq!(builtin.display(), "builtin:monitor@1");
88        assert_eq!(file.display(), "/tmp/demo.workflow.ts");
89    }
90}
91
92#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
93pub struct Manifest {
94    pub schema: String,
95    #[serde(rename = "runId")]
96    pub run_id: String,
97    #[serde(rename = "workflowName")]
98    pub workflow_name: String,
99    #[serde(rename = "runTitle", skip_serializing_if = "Option::is_none")]
100    pub run_title: Option<String>,
101    #[serde(rename = "workflowSource", skip_serializing_if = "Option::is_none")]
102    pub workflow_source: Option<WorkflowSource>,
103    #[serde(rename = "startedAt")]
104    pub started_at: String,
105    #[serde(rename = "finishedAt", skip_serializing_if = "Option::is_none")]
106    pub finished_at: Option<String>,
107    pub status: RunStatus,
108    #[serde(rename = "traceSchema")]
109    pub trace_schema: String,
110    pub paths: ManifestPaths,
111}
112
113#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
114pub struct ManifestPaths {
115    pub workflow: String,
116    pub state: String,
117    pub trace: String,
118    #[serde(skip_serializing_if = "Option::is_none")]
119    pub session: Option<String>,
120    #[serde(skip_serializing_if = "Option::is_none")]
121    pub artifacts: Option<String>,
122}
123
124#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
125pub struct RunState {
126    pub schema: String,
127    #[serde(rename = "traceSeq")]
128    pub trace_seq: u64,
129    #[serde(rename = "runId")]
130    pub run_id: String,
131    #[serde(rename = "workflowName")]
132    pub workflow_name: String,
133    #[serde(rename = "runTitle", skip_serializing_if = "Option::is_none")]
134    pub run_title: Option<String>,
135    #[serde(rename = "workflowSource", skip_serializing_if = "Option::is_none")]
136    pub workflow_source: Option<WorkflowSource>,
137    #[serde(rename = "parentRunId", skip_serializing_if = "Option::is_none")]
138    pub parent_run_id: Option<String>,
139    #[serde(rename = "carriedStepCount", skip_serializing_if = "Option::is_none")]
140    pub carried_step_count: Option<u64>,
141    #[serde(rename = "workflowSources", skip_serializing_if = "Option::is_none")]
142    pub workflow_sources: Option<Vec<Value>>,
143    #[serde(rename = "definitionDigest", skip_serializing_if = "Option::is_none")]
144    pub definition_digest: Option<String>,
145    #[serde(rename = "startedAt")]
146    pub started_at: String,
147    #[serde(rename = "finishedAt", skip_serializing_if = "Option::is_none")]
148    pub finished_at: Option<String>,
149    #[serde(rename = "updatedAt")]
150    pub updated_at: String,
151    pub status: RunStatus,
152    pub input: Value,
153    pub outputs: BTreeMap<String, Value>,
154    pub results: BTreeMap<String, NodeResult>,
155    pub steps: Vec<StepRecord>,
156    #[serde(skip_serializing_if = "Option::is_none")]
157    pub updates: Option<Vec<Value>>,
158    #[serde(rename = "currentNode", skip_serializing_if = "Option::is_none")]
159    pub current_node: Option<String>,
160    #[serde(rename = "currentAttemptId", skip_serializing_if = "Option::is_none")]
161    pub current_attempt_id: Option<String>,
162    #[serde(
163        rename = "currentNodeStartedAt",
164        skip_serializing_if = "Option::is_none"
165    )]
166    pub current_node_started_at: Option<String>,
167    #[serde(
168        rename = "currentSettingsScopeId",
169        skip_serializing_if = "Option::is_none"
170    )]
171    pub current_settings_scope_id: Option<String>,
172    #[serde(
173        rename = "currentSettingsChangeNumber",
174        skip_serializing_if = "Option::is_none"
175    )]
176    pub current_settings_change_number: Option<u64>,
177    #[serde(
178        rename = "currentSettingsHash",
179        skip_serializing_if = "Option::is_none"
180    )]
181    pub current_settings_hash: Option<String>,
182    #[serde(rename = "statusDetail", skip_serializing_if = "Option::is_none")]
183    pub status_detail: Option<String>,
184    #[serde(rename = "humanDecision", skip_serializing_if = "Option::is_none")]
185    pub human_decision: Option<Value>,
186    #[serde(skip_serializing_if = "Option::is_none")]
187    pub paused: Option<bool>,
188    #[serde(rename = "waitingOn", skip_serializing_if = "Option::is_none")]
189    pub waiting_on: Option<String>,
190    #[serde(rename = "finalOutput", skip_serializing_if = "Option::is_none")]
191    pub final_output: Option<Value>,
192    #[serde(skip_serializing_if = "Option::is_none")]
193    pub error: Option<String>,
194}
195
196#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
197pub struct NodeResult {
198    #[serde(rename = "attemptId")]
199    pub attempt_id: String,
200    #[serde(rename = "nodeId")]
201    pub node_id: String,
202    #[serde(rename = "nodeType")]
203    pub node_type: String,
204    pub outcome: NodeOutcome,
205    #[serde(rename = "startedAt")]
206    pub started_at: String,
207    #[serde(rename = "finishedAt")]
208    pub finished_at: String,
209    #[serde(rename = "durationMs")]
210    pub duration_ms: f64,
211    #[serde(skip_serializing_if = "Option::is_none")]
212    pub output: Option<Value>,
213    #[serde(skip_serializing_if = "Option::is_none")]
214    pub error: Option<String>,
215}
216
217#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
218pub struct StepRecord {
219    #[serde(rename = "attemptId")]
220    pub attempt_id: String,
221    #[serde(rename = "nodeId")]
222    pub node_id: String,
223    #[serde(rename = "nodeType")]
224    pub node_type: String,
225    pub outcome: NodeOutcome,
226    #[serde(rename = "startedAt")]
227    pub started_at: String,
228    #[serde(rename = "finishedAt")]
229    pub finished_at: String,
230    /// Full prompt for agent steps (`null` otherwise); may be an
231    /// externalized `$artifact` object in persisted form.
232    pub prompt: Value,
233    pub output: Value,
234    #[serde(skip_serializing_if = "Option::is_none")]
235    pub error: Option<String>,
236    #[serde(skip_serializing_if = "Option::is_none")]
237    pub action: Option<ActionReceipt>,
238    #[serde(rename = "assistantMessage", skip_serializing_if = "Option::is_none")]
239    pub assistant_message: Option<Value>,
240    #[serde(skip_serializing_if = "Option::is_none")]
241    pub conversation: Option<ConversationRange>,
242    #[serde(rename = "settingsScopeId", skip_serializing_if = "Option::is_none")]
243    pub settings_scope_id: Option<String>,
244    #[serde(
245        rename = "settingsChangeNumber",
246        skip_serializing_if = "Option::is_none"
247    )]
248    pub settings_change_number: Option<u64>,
249    #[serde(rename = "settingsHash", skip_serializing_if = "Option::is_none")]
250    pub settings_hash: Option<String>,
251}
252
253#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
254pub struct ActionReceipt {
255    #[serde(rename = "actionType")]
256    pub action_type: String,
257    #[serde(skip_serializing_if = "Option::is_none")]
258    pub command: Option<String>,
259    #[serde(skip_serializing_if = "Option::is_none")]
260    pub args: Option<Vec<String>>,
261    #[serde(skip_serializing_if = "Option::is_none")]
262    pub cwd: Option<String>,
263    #[serde(rename = "exitCode", skip_serializing_if = "Option::is_none")]
264    pub exit_code: Option<Value>,
265    #[serde(skip_serializing_if = "Option::is_none")]
266    pub signal: Option<Value>,
267    #[serde(rename = "durationMs", skip_serializing_if = "Option::is_none")]
268    pub duration_ms: Option<f64>,
269}
270
271#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
272pub struct ConversationRange {
273    #[serde(rename = "firstEntryId")]
274    pub first_entry_id: String,
275    #[serde(rename = "lastEntryId")]
276    pub last_entry_id: String,
277}
278
279#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
280pub struct TraceEvent {
281    pub seq: u64,
282    pub at: String,
283    pub scope: String,
284    #[serde(rename = "type")]
285    pub event_type: String,
286    #[serde(rename = "runId")]
287    pub run_id: String,
288    #[serde(rename = "nodeId", skip_serializing_if = "Option::is_none")]
289    pub node_id: Option<String>,
290    #[serde(rename = "attemptId", skip_serializing_if = "Option::is_none")]
291    pub attempt_id: Option<String>,
292    pub payload: Value,
293}
294
295#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
296pub struct SessionBinding {
297    pub schema: String,
298    #[serde(rename = "runId")]
299    pub run_id: String,
300    #[serde(rename = "piSessionId")]
301    pub pi_session_id: String,
302    #[serde(rename = "piSessionFile", skip_serializing_if = "Option::is_none")]
303    pub pi_session_file: Option<String>,
304    pub cwd: String,
305    #[serde(rename = "boundAt")]
306    pub bound_at: String,
307}
308
309#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
310pub struct SessionEntryRecord {
311    pub seq: u64,
312    pub at: String,
313    pub entry: Value,
314}
315
316#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
317pub struct SessionEventRecord {
318    pub seq: u64,
319    pub at: String,
320    #[serde(rename = "nodeId")]
321    pub node_id: String,
322    #[serde(rename = "attemptId")]
323    pub attempt_id: String,
324    #[serde(rename = "turnId", skip_serializing_if = "Option::is_none")]
325    pub turn_id: Option<String>,
326    #[serde(rename = "messageId", skip_serializing_if = "Option::is_none")]
327    pub message_id: Option<String>,
328    #[serde(rename = "toolCallId", skip_serializing_if = "Option::is_none")]
329    pub tool_call_id: Option<String>,
330    #[serde(rename = "type")]
331    pub event_type: String,
332    pub payload: Value,
333}
334
335#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
336#[serde(rename_all = "snake_case")]
337pub enum SessionCaptureStatus {
338    Recording,
339    Complete,
340    Failed,
341}
342
343#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
344pub struct SessionCaptureFailure {
345    #[serde(rename = "failedAt")]
346    pub failed_at: String,
347    pub code: String,
348    pub message: String,
349}
350
351#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
352pub struct SessionCapture {
353    pub schema: String,
354    #[serde(rename = "eventSchema")]
355    pub event_schema: String,
356    pub status: SessionCaptureStatus,
357    #[serde(rename = "eventCount")]
358    pub event_count: u64,
359    #[serde(rename = "entryCount")]
360    pub entry_count: u64,
361    #[serde(rename = "lastEventSeq")]
362    pub last_event_seq: u64,
363    #[serde(skip_serializing_if = "Option::is_none")]
364    pub failure: Option<SessionCaptureFailure>,
365}
366
367// --- Definition snapshot ---
368
369#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
370pub struct DefinitionSnapshot {
371    pub schema: String,
372    pub name: String,
373    #[serde(rename = "startAt")]
374    pub start_at: String,
375    /// Insertion order matters (BFS fallback order, display order), so the
376    /// map preserves the document order of the JSON object.
377    pub nodes: serde_json::Map<String, Value>,
378    pub edges: Vec<EdgeDef>,
379}
380
381impl DefinitionSnapshot {
382    pub fn node_type(&self, node_id: &str) -> Option<&str> {
383        self.nodes
384            .get(node_id)?
385            .get("nodeType")
386            .and_then(Value::as_str)
387    }
388
389    pub fn node_action_execution(&self, node_id: &str) -> Option<&str> {
390        self.nodes
391            .get(node_id)?
392            .get("actionExecution")
393            .and_then(Value::as_str)
394    }
395
396    pub fn node_ids(&self) -> impl Iterator<Item = &str> {
397        self.nodes.keys().map(String::as_str)
398    }
399}
400
401/// A workflow edge: either a simple `from -> to` or a labelled switch.
402#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
403#[serde(untagged)]
404pub enum EdgeDef {
405    Simple { from: String, to: String },
406    Switch { from: String, switch: SwitchDef },
407}
408
409#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
410pub struct SwitchDef {
411    pub on: String,
412    /// Case order matters for edge expansion; preserve document order.
413    pub cases: serde_json::Map<String, Value>,
414}
415
416/// An artifact reference extracted from a `{"$artifact": …}` sentinel.
417#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
418pub struct ArtifactRef {
419    pub path: String,
420    #[serde(rename = "mediaType")]
421    pub media_type: String,
422    pub bytes: u64,
423    pub sha256: String,
424}
425
426/// Detect the `$artifact` sentinel: an object whose only key is `$artifact`.
427pub fn as_artifact_ref(value: &Value) -> Option<ArtifactRef> {
428    let object = value.as_object()?;
429    if object.len() != 1 {
430        return None;
431    }
432    serde_json::from_value(object.get("$artifact")?.clone()).ok()
433}
434
435/// Unwrap one level of `{"$escaped": …}` if present.
436pub fn as_escaped(value: &Value) -> Option<&Value> {
437    let object = value.as_object()?;
438    if object.len() != 1 {
439        return None;
440    }
441    object.get("$escaped")
442}
piw/state/types.rs

piw/state/
types.rs
