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

piw/state/
types.rs
