piw/state/
types.rs

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

piw/state/
types.rs
