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 = "startedAt")]
135    pub started_at: String,
136    #[serde(rename = "finishedAt", skip_serializing_if = "Option::is_none")]
137    pub finished_at: Option<String>,
138    #[serde(rename = "updatedAt")]
139    pub updated_at: String,
140    pub status: RunStatus,
141    pub input: Value,
142    pub outputs: BTreeMap<String, Value>,
143    pub results: BTreeMap<String, NodeResult>,
144    pub steps: Vec<StepRecord>,
145    #[serde(rename = "currentNode", skip_serializing_if = "Option::is_none")]
146    pub current_node: Option<String>,
147    #[serde(rename = "currentAttemptId", skip_serializing_if = "Option::is_none")]
148    pub current_attempt_id: Option<String>,
149    #[serde(
150        rename = "currentNodeStartedAt",
151        skip_serializing_if = "Option::is_none"
152    )]
153    pub current_node_started_at: Option<String>,
154    #[serde(rename = "statusDetail", skip_serializing_if = "Option::is_none")]
155    pub status_detail: Option<String>,
156    #[serde(rename = "humanDecision", skip_serializing_if = "Option::is_none")]
157    pub human_decision: Option<Value>,
158    #[serde(skip_serializing_if = "Option::is_none")]
159    pub paused: Option<bool>,
160    #[serde(rename = "waitingOn", skip_serializing_if = "Option::is_none")]
161    pub waiting_on: Option<String>,
162    #[serde(rename = "finalOutput", skip_serializing_if = "Option::is_none")]
163    pub final_output: Option<Value>,
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub error: Option<String>,
166}
167
168#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
169pub struct NodeResult {
170    #[serde(rename = "attemptId")]
171    pub attempt_id: String,
172    #[serde(rename = "nodeId")]
173    pub node_id: String,
174    #[serde(rename = "nodeType")]
175    pub node_type: String,
176    pub outcome: NodeOutcome,
177    #[serde(rename = "startedAt")]
178    pub started_at: String,
179    #[serde(rename = "finishedAt")]
180    pub finished_at: String,
181    #[serde(rename = "durationMs")]
182    pub duration_ms: f64,
183    #[serde(skip_serializing_if = "Option::is_none")]
184    pub output: Option<Value>,
185    #[serde(skip_serializing_if = "Option::is_none")]
186    pub error: Option<String>,
187}
188
189#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
190pub struct StepRecord {
191    #[serde(rename = "attemptId")]
192    pub attempt_id: String,
193    #[serde(rename = "nodeId")]
194    pub node_id: String,
195    #[serde(rename = "nodeType")]
196    pub node_type: String,
197    pub outcome: NodeOutcome,
198    #[serde(rename = "startedAt")]
199    pub started_at: String,
200    #[serde(rename = "finishedAt")]
201    pub finished_at: String,
202    /// Full prompt for agent steps (`null` otherwise); may be an
203    /// externalized `$artifact` object in persisted form.
204    pub prompt: Value,
205    pub output: Value,
206    #[serde(skip_serializing_if = "Option::is_none")]
207    pub error: Option<String>,
208    #[serde(skip_serializing_if = "Option::is_none")]
209    pub action: Option<ActionReceipt>,
210    #[serde(skip_serializing_if = "Option::is_none")]
211    pub conversation: Option<ConversationRange>,
212}
213
214#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
215pub struct ActionReceipt {
216    #[serde(rename = "actionType")]
217    pub action_type: String,
218    #[serde(skip_serializing_if = "Option::is_none")]
219    pub command: Option<String>,
220    #[serde(skip_serializing_if = "Option::is_none")]
221    pub args: Option<Vec<String>>,
222    #[serde(skip_serializing_if = "Option::is_none")]
223    pub cwd: Option<String>,
224    #[serde(rename = "exitCode", skip_serializing_if = "Option::is_none")]
225    pub exit_code: Option<Value>,
226    #[serde(skip_serializing_if = "Option::is_none")]
227    pub signal: Option<Value>,
228    #[serde(rename = "durationMs", skip_serializing_if = "Option::is_none")]
229    pub duration_ms: Option<f64>,
230}
231
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct ConversationRange {
234    #[serde(rename = "firstEntryId")]
235    pub first_entry_id: String,
236    #[serde(rename = "lastEntryId")]
237    pub last_entry_id: String,
238}
239
240#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
241pub struct TraceEvent {
242    pub seq: u64,
243    pub at: String,
244    pub scope: String,
245    #[serde(rename = "type")]
246    pub event_type: String,
247    #[serde(rename = "runId")]
248    pub run_id: String,
249    #[serde(rename = "nodeId", skip_serializing_if = "Option::is_none")]
250    pub node_id: Option<String>,
251    #[serde(rename = "attemptId", skip_serializing_if = "Option::is_none")]
252    pub attempt_id: Option<String>,
253    pub payload: Value,
254}
255
256#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
257pub struct SessionBinding {
258    pub schema: String,
259    #[serde(rename = "runId")]
260    pub run_id: String,
261    #[serde(rename = "piSessionId")]
262    pub pi_session_id: String,
263    #[serde(rename = "piSessionFile", skip_serializing_if = "Option::is_none")]
264    pub pi_session_file: Option<String>,
265    pub cwd: String,
266    #[serde(rename = "boundAt")]
267    pub bound_at: String,
268}
269
270#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
271pub struct SessionEntryRecord {
272    pub seq: u64,
273    pub at: String,
274    pub entry: Value,
275}
276
277#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
278pub struct SessionEventRecord {
279    pub seq: u64,
280    pub at: String,
281    #[serde(rename = "nodeId")]
282    pub node_id: String,
283    #[serde(rename = "attemptId")]
284    pub attempt_id: String,
285    #[serde(rename = "turnId", skip_serializing_if = "Option::is_none")]
286    pub turn_id: Option<String>,
287    #[serde(rename = "messageId", skip_serializing_if = "Option::is_none")]
288    pub message_id: Option<String>,
289    #[serde(rename = "toolCallId", skip_serializing_if = "Option::is_none")]
290    pub tool_call_id: Option<String>,
291    #[serde(rename = "type")]
292    pub event_type: String,
293    pub payload: Value,
294}
295
296#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
297#[serde(rename_all = "snake_case")]
298pub enum SessionCaptureStatus {
299    Recording,
300    Complete,
301    Failed,
302}
303
304#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
305pub struct SessionCaptureFailure {
306    #[serde(rename = "failedAt")]
307    pub failed_at: String,
308    pub code: String,
309    pub message: String,
310}
311
312#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
313pub struct SessionCapture {
314    pub schema: String,
315    #[serde(rename = "eventSchema")]
316    pub event_schema: String,
317    pub status: SessionCaptureStatus,
318    #[serde(rename = "eventCount")]
319    pub event_count: u64,
320    #[serde(rename = "entryCount")]
321    pub entry_count: u64,
322    #[serde(rename = "lastEventSeq")]
323    pub last_event_seq: u64,
324    #[serde(skip_serializing_if = "Option::is_none")]
325    pub failure: Option<SessionCaptureFailure>,
326}
327
328// --- Definition snapshot ---
329
330#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
331pub struct DefinitionSnapshot {
332    pub schema: String,
333    pub name: String,
334    #[serde(rename = "startAt")]
335    pub start_at: String,
336    /// Insertion order matters (BFS fallback order, display order), so the
337    /// map preserves the document order of the JSON object.
338    pub nodes: serde_json::Map<String, Value>,
339    pub edges: Vec<EdgeDef>,
340}
341
342impl DefinitionSnapshot {
343    pub fn node_type(&self, node_id: &str) -> Option<&str> {
344        self.nodes
345            .get(node_id)?
346            .get("nodeType")
347            .and_then(Value::as_str)
348    }
349
350    pub fn node_action_execution(&self, node_id: &str) -> Option<&str> {
351        self.nodes
352            .get(node_id)?
353            .get("actionExecution")
354            .and_then(Value::as_str)
355    }
356
357    pub fn node_ids(&self) -> impl Iterator<Item = &str> {
358        self.nodes.keys().map(String::as_str)
359    }
360}
361
362/// A workflow edge: either a simple `from -> to` or a labelled switch.
363#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
364#[serde(untagged)]
365pub enum EdgeDef {
366    Simple { from: String, to: String },
367    Switch { from: String, switch: SwitchDef },
368}
369
370#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
371pub struct SwitchDef {
372    pub on: String,
373    /// Case order matters for edge expansion; preserve document order.
374    pub cases: serde_json::Map<String, Value>,
375}
376
377/// An artifact reference extracted from a `{"$artifact": …}` sentinel.
378#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
379pub struct ArtifactRef {
380    pub path: String,
381    #[serde(rename = "mediaType")]
382    pub media_type: String,
383    pub bytes: u64,
384    pub sha256: String,
385}
386
387/// Detect the `$artifact` sentinel: an object whose only key is `$artifact`.
388pub fn as_artifact_ref(value: &Value) -> Option<ArtifactRef> {
389    let object = value.as_object()?;
390    if object.len() != 1 {
391        return None;
392    }
393    serde_json::from_value(object.get("$artifact")?.clone()).ok()
394}
395
396/// Unwrap one level of `{"$escaped": …}` if present.
397pub fn as_escaped(value: &Value) -> Option<&Value> {
398    let object = value.as_object()?;
399    if object.len() != 1 {
400        return None;
401    }
402    object.get("$escaped")
403}
piw/state/types.rs

piw/state/
types.rs
