use crate::protocol::{PageKind, PatchOp};
use crate::state::types::{
DefinitionSnapshot, Manifest, ManifestPaths, RunState, SessionBinding, SessionCapture,
SessionEntryRecord, SessionEventRecord, TraceEvent, DEFINITION_SNAPSHOT_SCHEMA,
RUN_STATE_SCHEMA,
};
use anyhow::{bail, Context, Result};
use chrono::{TimeZone, Utc};
use rusqlite::{Connection, OpenFlags, OptionalExtension};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::path::Path;
pub const VIEWER_PAGE_SIZE: u64 = 256;
const APPLICATION_ID: i64 = 0x5049_5746;
const USER_VERSION: i64 = 1;
const SCHEMA_NAME: &str = "pi-workflows-state";
const APP_VERSION: &str = "0.13.3";
pub const SCHEMA_DIGEST: [u8; 32] = [
0xb5, 0xbc, 0xda, 0x10, 0x34, 0xe2, 0xee, 0x30, 0x13, 0xd8, 0xee, 0x2c, 0xc6, 0x5d, 0xcf, 0xe8,
0xa2, 0xad, 0x6e, 0x6a, 0xd8, 0xab, 0x1e, 0xeb, 0xab, 0x7f, 0x89, 0x0d, 0xe6, 0x00, 0x31, 0xed,
];
const RESET_INSTRUCTION: &str = "Pi Workflows durable state is incompatible. Move or remove the old workflow state, then create a new state.sqlite database.";
type LoadedSession = (
Option<SessionBinding>,
Vec<SessionEntryRecord>,
Vec<SessionEventRecord>,
Option<SessionCapture>,
);
#[derive(Clone, Copy, Debug, Default)]
pub struct ProjectionCursors {
pub step: Option<u64>,
pub trace: Option<u64>,
pub trace_step: Option<u64>,
pub session_entry: Option<u64>,
pub session_event: Option<u64>,
pub settings: Option<u64>,
pub follow_ups: Option<u64>,
pub updates: Option<u64>,
}
type LoadedSessionWindow = (
Option<SessionBinding>,
Vec<SessionEntryRecord>,
u64,
u64,
Vec<SessionEventRecord>,
u64,
u64,
Option<SessionCapture>,
Option<Value>,
);
#[derive(Debug, Clone)]
pub struct LoadedRun {
pub manifest: Manifest,
pub state: RunState,
pub graph_steps: Vec<crate::state::types::StepRecord>,
pub taken_transitions: Vec<String>,
pub graph_cursor: u64,
pub step_start: u64,
pub step_total: u64,
pub snapshot: Option<DefinitionSnapshot>,
pub trace: Vec<TraceEvent>,
pub trace_start: u64,
pub trace_total: u64,
pub session_binding: Option<SessionBinding>,
pub session_entries: Vec<SessionEntryRecord>,
pub session_entry_start: u64,
pub session_entry_total: u64,
pub session_events: Vec<SessionEventRecord>,
pub session_event_start: u64,
pub session_event_total: u64,
pub session_capture: Option<SessionCapture>,
pub session_replay_checkpoint: Option<Value>,
pub settings_scopes: Vec<Value>,
pub settings_start: u64,
pub settings_total: u64,
pub follow_up_queue: Option<Value>,
pub follow_up_start: u64,
pub follow_up_total: u64,
pub update_start: u64,
pub update_total: u64,
pub possibly_interrupted: bool,
pub presentation_revision: u64,
}
#[derive(Debug, Clone)]
pub struct RunIndexRow {
pub manifest: Manifest,
pub live: bool,
pub possibly_interrupted: bool,
pub presentation_revision: u64,
pub retained_from_revision: u64,
pub lease_owner_id: Option<String>,
pub lease_expires_at: Option<i64>,
}
#[derive(Debug, Clone)]
pub struct ViewerTargetDelta {
pub target_type: String,
pub target_key: String,
pub patch: Vec<PatchOp>,
}
#[derive(Debug, Clone)]
pub struct ViewerRevisionDelta {
pub revision: u64,
pub targets: Vec<ViewerTargetDelta>,
}
#[derive(Debug, Clone)]
pub struct ProjectionPage {
pub start: u64,
pub total: u64,
pub items: Vec<Value>,
pub graph_cursor: Option<u64>,
pub graph_steps: Option<Vec<crate::state::types::StepRecord>>,
pub taken_transitions: Option<Vec<String>>,
pub replay_checkpoint: Option<Value>,
}
pub enum ViewerDeltaRead {
Deltas {
current_revision: u64,
deltas: Vec<ViewerRevisionDelta>,
},
SnapshotRequired {
current_revision: u64,
retained_from_revision: u64,
},
}
pub struct ProjectionReader {
connection: Connection,
}
impl ProjectionReader {
pub fn open(database_path: &Path) -> Result<Self> {
Ok(Self {
connection: open(database_path)?,
})
}
pub fn data_version(&self) -> Result<u64> {
Ok(self
.connection
.pragma_query_value(None, "data_version", |row| row.get(0))?)
}
pub fn list_run_index(&self) -> Result<Vec<RunIndexRow>> {
list_run_index(&self.connection)
}
pub fn read_window(&self, run_id: &str, cursors: ProjectionCursors) -> Result<LoadedRun> {
let transaction = self.connection.unchecked_transaction()?;
let loaded = read_run_from_connection(&transaction, run_id, Some(cursors))?;
transaction.commit()?;
Ok(loaded)
}
pub fn read_page(
&self,
run_id: &str,
kind: PageKind,
cursor: u64,
) -> Result<(u64, ProjectionPage)> {
let transaction = self.connection.unchecked_transaction()?;
let revision = transaction.query_row(
"SELECT presentation_revision FROM viewer_runs WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let page = match kind {
PageKind::Steps => read_step_page(&transaction, run_id, Some(cursor))?,
PageKind::Trace | PageKind::TraceAtStep => {
let trace_cursor = if kind == PageKind::TraceAtStep {
trace_cursor_for_step(&transaction, run_id, cursor)?
} else {
Some(cursor)
};
let (items, start, total) = read_trace_window(&transaction, run_id, trace_cursor)?;
ProjectionPage {
start,
total,
items: items
.into_iter()
.map(serde_json::to_value)
.collect::<Result<Vec<_>, _>>()?,
graph_cursor: None,
graph_steps: None,
taken_transitions: None,
replay_checkpoint: None,
}
}
PageKind::SessionEntries => {
read_session_entry_page(&transaction, run_id, Some(cursor))?
}
PageKind::SessionEvents => read_session_event_page(&transaction, run_id, Some(cursor))?,
PageKind::Settings => read_settings_page(&transaction, run_id, Some(cursor))?,
PageKind::FollowUps => read_follow_up_page(&transaction, run_id, Some(cursor))?,
PageKind::Updates => read_update_page(&transaction, run_id, Some(cursor))?,
};
transaction.commit()?;
Ok((revision, page))
}
pub fn read_deltas(&self, run_id: &str, after_revision: u64) -> Result<ViewerDeltaRead> {
let (current_revision, retained_from_revision): (u64, u64) = self.connection.query_row(
"SELECT presentation_revision, retained_from_revision
FROM viewer_runs WHERE run_id = ?1",
[run_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
if after_revision == 0
|| after_revision > current_revision
|| after_revision < retained_from_revision.saturating_sub(1)
{
return Ok(ViewerDeltaRead::SnapshotRequired {
current_revision,
retained_from_revision,
});
}
let mut statement = self.connection.prepare(
"SELECT presentation_revision, target_type, target_key, patch_hash
FROM viewer_deltas
WHERE run_id = ?1 AND presentation_revision > ?2
ORDER BY presentation_revision, delta_index",
)?;
let rows = statement.query_map(rusqlite::params![run_id, after_revision], |row| {
Ok((
row.get::<_, u64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Vec<u8>>(3)?,
))
})?;
let mut grouped: BTreeMap<u64, Vec<ViewerTargetDelta>> = BTreeMap::new();
for row in rows {
let (revision, target_type, target_key, patch_hash) = row?;
let patch = serde_json::from_value(read_json_blob(&self.connection, &patch_hash)?)?;
grouped
.entry(revision)
.or_default()
.push(ViewerTargetDelta {
target_type,
target_key,
patch,
});
}
let contiguous = grouped
.keys()
.copied()
.eq((after_revision + 1)..=current_revision);
if !contiguous {
return Ok(ViewerDeltaRead::SnapshotRequired {
current_revision,
retained_from_revision,
});
}
Ok(ViewerDeltaRead::Deltas {
current_revision,
deltas: grouped
.into_iter()
.map(|(revision, targets)| ViewerRevisionDelta { revision, targets })
.collect(),
})
}
}
pub fn read_run(database_path: &Path, run_id: &str) -> Result<LoadedRun> {
let connection = open(database_path)?;
read_run_from_connection(&connection, run_id, None)
}
fn read_run_from_connection(
connection: &Connection,
run_id: &str,
cursors: Option<ProjectionCursors>,
) -> Result<LoadedRun> {
let row = connection
.query_row(
"SELECT d.definition_hash, l.owner_id, l.expires_at
FROM runs r
JOIN workflow_definitions d ON d.definition_digest = r.definition_digest
JOIN leases l ON l.resource_id = r.resource_id
WHERE r.run_id = ?1",
[run_id],
|row| {
Ok((
row.get::<_, Vec<u8>>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, Option<i64>>(2)?,
))
},
)
.optional()?;
let Some((definition_hash, owner_id, lease_expires_at)) = row else {
bail!("workflow run not found: {run_id}");
};
let definition_value = read_json_blob(connection, &definition_hash)?;
let snapshot: DefinitionSnapshot = serde_json::from_value(definition_value.clone())?;
if snapshot.schema != DEFINITION_SNAPSHOT_SCHEMA {
bail!(
"unsupported workflow definition schema: {}",
snapshot.schema
);
}
let step_total: u64 = connection.query_row(
"SELECT count(*) FROM run_steps WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let step_start = cursors.map_or(0, |cursors| page_start(step_total, cursors.step));
let mut update_page = read_update_page(
connection,
run_id,
cursors.and_then(|cursors| cursors.updates),
)?;
if cursors.is_none() && update_page.total > update_page.items.len() as u64 {
update_page.items = read_updates_range(connection, run_id, 0, -1)?;
update_page.start = 0;
}
let update_start = update_page.start;
let update_total = update_page.total;
let state = read_state(
connection,
run_id,
&definition_value,
cursors.map(|_| step_start),
update_page.items,
)?;
let graph_cursor = cursors.map_or_else(
|| step_total.saturating_sub(1),
|cursors| {
cursors
.step
.unwrap_or_else(|| step_total.saturating_sub(1))
.min(step_total.saturating_sub(1))
},
);
let (graph_steps, taken_transitions) = match cursors {
Some(_) => (
read_graph_steps(connection, run_id, graph_cursor)?,
read_taken_transitions(connection, run_id, graph_cursor)?,
),
None => (
state.steps.clone(),
state
.steps
.windows(2)
.map(|pair| format!("{}->{}", pair[0].node_id, pair[1].node_id))
.collect::<std::collections::BTreeSet<_>>()
.into_iter()
.collect(),
),
};
let (trace, trace_start, trace_total) = match cursors {
Some(cursors) => {
let trace_cursor = match cursors.trace_step {
Some(step) => trace_cursor_for_step(connection, run_id, step)?,
None => cursors.trace,
};
read_trace_window(connection, run_id, trace_cursor)?
}
None => {
let trace = read_trace(connection, run_id)?;
let total = trace.len() as u64;
(trace, 0, total)
}
};
let (
session_binding,
session_entries,
session_entry_start,
session_entry_total,
session_events,
session_event_start,
session_event_total,
session_capture,
session_replay_checkpoint,
) = match cursors {
Some(cursors) => read_session_window(
connection,
run_id,
cursors.session_entry,
cursors.session_event,
)?,
None => {
let (binding, entries, events, capture) = read_session(connection, run_id)?;
let entry_total = entries.len() as u64;
let event_total = events.len() as u64;
(
binding,
entries,
0,
entry_total,
events,
0,
event_total,
capture,
None,
)
}
};
let mut settings_page = read_settings_page(
connection,
run_id,
cursors.and_then(|cursors| cursors.settings),
)?;
if cursors.is_none() && settings_page.total > settings_page.items.len() as u64 {
settings_page.items = read_settings_range(connection, run_id, 0, -1)?;
settings_page.start = 0;
}
let mut follow_up_page = read_follow_up_page(
connection,
run_id,
cursors.and_then(|cursors| cursors.follow_ups),
)?;
if cursors.is_none() && follow_up_page.total > follow_up_page.items.len() as u64 {
follow_up_page.items = read_follow_up_range(connection, run_id, 0, -1)?;
follow_up_page.start = 0;
}
let follow_up_queue = read_follow_up_state(connection, run_id)?.map(|presentation_state| {
json!({
"presentationState": presentation_state,
"items": follow_up_page.items,
})
});
let presentation_revision = connection.query_row(
"SELECT presentation_revision FROM viewer_runs WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let manifest = manifest_from_state(&state);
let possibly_interrupted = state.status.label() == "running"
&& (owner_id.is_none()
|| lease_expires_at
.is_none_or(|expires_at| expires_at <= Utc::now().timestamp_millis()));
Ok(LoadedRun {
manifest,
state,
graph_steps,
taken_transitions,
graph_cursor,
step_start,
step_total,
snapshot: Some(snapshot),
trace,
trace_start,
trace_total,
session_binding,
session_entries,
session_entry_start,
session_entry_total,
session_events,
session_event_start,
session_event_total,
session_capture,
session_replay_checkpoint,
settings_scopes: settings_page.items,
settings_start: settings_page.start,
settings_total: settings_page.total,
follow_up_queue,
follow_up_start: follow_up_page.start,
follow_up_total: follow_up_page.total,
update_start,
update_total,
possibly_interrupted,
presentation_revision,
})
}
pub fn list_runs(database_path: &Path) -> Vec<(String, Manifest)> {
let Ok(connection) = open(database_path) else {
return Vec::new();
};
list_run_index(&connection)
.unwrap_or_default()
.into_iter()
.map(|row| (row.manifest.run_id.clone(), row.manifest))
.collect()
}
fn list_run_index(connection: &Connection) -> Result<Vec<RunIndexRow>> {
let now = Utc::now().timestamp_millis();
let mut statement = connection.prepare(
"SELECT r.run_id, d.workflow_name, r.title, r.status,
r.created_at, r.finished_at,
v.presentation_revision, v.retained_from_revision,
l.owner_id, l.expires_at,
s.source_type, s.source_ref, s.source_revision
FROM runs r
JOIN workflow_definitions d ON d.definition_digest = r.definition_digest
JOIN viewer_runs v ON v.run_id = r.run_id
JOIN leases l ON l.resource_id = r.resource_id
LEFT JOIN run_sources s ON s.run_id = r.run_id AND s.mount_path = ''
ORDER BY r.created_at DESC, r.run_id DESC",
)?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, String>(3)?,
row.get::<_, i64>(4)?,
row.get::<_, Option<i64>>(5)?,
row.get::<_, u64>(6)?,
row.get::<_, u64>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<i64>>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<String>>(12)?,
))
})?;
let mut index = Vec::new();
for row in rows {
let (
run_id,
workflow_name,
run_title,
status_value,
started_at,
finished_at,
presentation_revision,
retained_from_revision,
owner_id,
expires_at,
source_type,
source_ref,
source_revision,
) = row?;
let status = parse_run_status(&status_value)?;
let workflow_source = match (source_type.as_deref(), source_ref, source_revision) {
(Some("builtin"), Some(id), Some(revision)) => {
Some(crate::state::types::WorkflowSource::Builtin { id, revision })
}
(Some("file"), Some(path), Some(hash)) => {
Some(crate::state::types::WorkflowSource::File { path, hash })
}
(None, None, None) => None,
_ => bail!("workflow run source is incomplete: {run_id}"),
};
index.push(RunIndexRow {
manifest: Manifest {
schema: "pi-workflows.sqlite-view.v1".to_string(),
run_id,
workflow_name,
run_title,
workflow_source,
started_at: timestamp(started_at),
finished_at: finished_at.map(timestamp),
status,
trace_schema: "pi-workflows.event.v1".to_string(),
paths: ManifestPaths {
workflow: String::new(),
state: String::new(),
trace: String::new(),
session: None,
artifacts: None,
},
},
live: !status.is_terminal(),
possibly_interrupted: status.label() == "running"
&& (owner_id.is_none() || expires_at.is_none_or(|value| value <= now)),
presentation_revision,
retained_from_revision,
lease_owner_id: owner_id,
lease_expires_at: expires_at,
});
}
Ok(index)
}
fn parse_run_status(value: &str) -> Result<crate::state::types::RunStatus> {
use crate::state::types::RunStatus;
match value {
"queued" => Ok(RunStatus::Queued),
"running" => Ok(RunStatus::Running),
"waiting" => Ok(RunStatus::Waiting),
"completed" => Ok(RunStatus::Completed),
"failed" => Ok(RunStatus::Failed),
"timed_out" => Ok(RunStatus::TimedOut),
"cancelled" => Ok(RunStatus::Cancelled),
_ => bail!("workflow run status is invalid: {value}"),
}
}
pub fn with_artifact_placeholders(value: &Value) -> Value {
value.clone()
}
pub fn resolve_artifacts(value: &Value, _database_path: &Path, _max_bytes: u64) -> Value {
value.clone()
}
pub fn validate_database(database_path: &Path) -> Result<()> {
open(database_path).map(drop)
}
fn open(path: &Path) -> Result<Connection> {
let connection = Connection::open_with_flags(
path,
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
)
.with_context(|| format!("could not open {}", path.display()))?;
connection.pragma_update(None, "query_only", true)?;
connection.pragma_update(None, "foreign_keys", true)?;
let application_id: i64 =
connection.pragma_query_value(None, "application_id", |row| row.get(0))?;
let user_version: i64 =
connection.pragma_query_value(None, "user_version", |row| row.get(0))?;
if application_id != APPLICATION_ID || user_version != USER_VERSION {
bail!(RESET_INSTRUCTION);
}
let schema = connection
.query_row(
"SELECT schema_name, schema_version, schema_digest, app_version
FROM schema_meta WHERE id = 1",
[],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, Vec<u8>>(2)?,
row.get::<_, String>(3)?,
))
},
)
.optional()
.unwrap_or(None);
if !matches!(
schema,
Some((name, version, digest, app_version))
if name == SCHEMA_NAME
&& version == USER_VERSION
&& digest.as_slice() == SCHEMA_DIGEST
&& app_version == APP_VERSION
) {
bail!(RESET_INSTRUCTION);
}
Ok(connection)
}
fn read_json_blob(connection: &Connection, hash: &[u8]) -> Result<Value> {
let content: Vec<u8> = connection.query_row(
"SELECT content FROM blobs WHERE blob_hash = ?1 AND media_type = 'application/json'",
[hash],
|row| row.get(0),
)?;
Ok(serde_json::from_slice(&content)?)
}
fn read_text_blob(connection: &Connection, hash: &[u8]) -> Result<String> {
let content: Vec<u8> = connection.query_row(
"SELECT content FROM blobs WHERE blob_hash = ?1 AND media_type = 'text/plain'",
[hash],
|row| row.get(0),
)?;
Ok(String::from_utf8(content)?)
}
fn read_settings_range(
connection: &Connection,
run_id: &str,
start: u64,
limit: i64,
) -> Result<Vec<Value>> {
let mut statement = connection.prepare(
"SELECT s.scope_id, s.mount_path, s.invocation, s.current_hash, r.revision
FROM workflow_settings s
JOIN resources r ON r.resource_id = s.resource_id
WHERE s.active_run_id = ?1
ORDER BY s.mount_path, s.invocation LIMIT ?2 OFFSET ?3",
)?;
let rows = statement.query_map(rusqlite::params![run_id, limit, start], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, u64>(2)?,
row.get::<_, Vec<u8>>(3)?,
row.get::<_, u64>(4)?,
))
})?;
let mut items = Vec::new();
for row in rows {
let (scope_id, mount_path, invocation, settings_hash, change_number) = row?;
items.push(json!({
"scopeId": scope_id,
"mountPath": mount_path,
"invocation": invocation,
"changeNumber": change_number,
"settingsHash": encode_hex(&settings_hash),
}));
}
Ok(items)
}
fn read_settings_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM workflow_settings WHERE active_run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
Ok(projection_page(
start,
total,
read_settings_range(connection, run_id, start, VIEWER_PAGE_SIZE as i64)?,
))
}
fn read_follow_up_range(
connection: &Connection,
run_id: &str,
start: u64,
limit: i64,
) -> Result<Vec<Value>> {
let mut statement = connection.prepare(
"SELECT follow_up_id, order_number, status, source_type, session_entry_id
FROM workflow_follow_ups
WHERE run_id = ?1 ORDER BY order_number LIMIT ?2 OFFSET ?3",
)?;
let rows = statement.query_map(rusqlite::params![run_id, limit, start], |row| {
Ok(json!({
"followUpId": row.get::<_, String>(0)?,
"order": row.get::<_, u64>(1)?,
"state": row.get::<_, String>(2)?,
"source": row.get::<_, String>(3)?,
"sessionEntryId": row.get::<_, Option<String>>(4)?,
}))
})?;
Ok(rows.collect::<Result<Vec<_>, _>>()?)
}
fn read_follow_up_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM workflow_follow_ups WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
Ok(projection_page(
start,
total,
read_follow_up_range(connection, run_id, start, VIEWER_PAGE_SIZE as i64)?,
))
}
fn read_follow_up_state(connection: &Connection, run_id: &str) -> Result<Option<String>> {
Ok(connection
.query_row(
"SELECT presentation_state FROM workflow_follow_up_queues WHERE run_id = ?1",
[run_id],
|row| row.get::<_, String>(0),
)
.optional()?)
}
fn read_state(
connection: &Connection,
run_id: &str,
definition: &Value,
step_start: Option<u64>,
updates: Vec<Value>,
) -> Result<RunState> {
let row = connection.query_row(
"SELECT r.resource_id, d.workflow_name, r.parent_run_id, r.title, r.status,
r.paused, r.status_detail, r.input_hash, r.final_output_hash, r.error_hash,
r.definition_digest, r.created_at, r.updated_at, r.finished_at,
resources.revision
FROM runs r
JOIN workflow_definitions d ON d.definition_digest = r.definition_digest
JOIN resources ON resources.resource_id = r.resource_id
WHERE r.run_id = ?1",
[run_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, i64>(5)?,
row.get::<_, Option<String>>(6)?,
row.get::<_, Vec<u8>>(7)?,
row.get::<_, Option<Vec<u8>>>(8)?,
row.get::<_, Option<Vec<u8>>>(9)?,
row.get::<_, Vec<u8>>(10)?,
row.get::<_, i64>(11)?,
row.get::<_, i64>(12)?,
row.get::<_, Option<i64>>(13)?,
row.get::<_, u64>(14)?,
))
},
)?;
let (
_resource_id,
workflow_name,
parent_run_id,
title,
status,
paused,
status_detail,
input_hash,
final_output_hash,
error_hash,
definition_digest,
created_at,
updated_at,
finished_at,
revision,
) = row;
let steps = read_steps(connection, run_id, step_start)?;
let mut outputs = serde_json::Map::new();
let mut results = serde_json::Map::new();
for step in &steps {
let outcome = step
.get("outcome")
.and_then(Value::as_str)
.unwrap_or("failed");
let node_id = step
.get("nodeId")
.and_then(Value::as_str)
.unwrap_or_default();
let started = step
.get("startedAt")
.and_then(Value::as_str)
.unwrap_or_default();
let finished = step
.get("finishedAt")
.and_then(Value::as_str)
.unwrap_or_default();
let duration = chrono::DateTime::parse_from_rfc3339(finished)
.ok()
.zip(chrono::DateTime::parse_from_rfc3339(started).ok())
.map_or(0, |(end, start)| (end - start).num_milliseconds());
let mut result = json!({
"attemptId": step.get("attemptId").cloned().unwrap_or(Value::Null),
"nodeId": node_id,
"nodeType": step.get("nodeType").cloned().unwrap_or(Value::Null),
"outcome": outcome,
"startedAt": started,
"finishedAt": finished,
"durationMs": duration,
});
if outcome == "ok" {
let output = step.get("output").cloned().unwrap_or(Value::Null);
outputs.insert(node_id.to_string(), output.clone());
result["output"] = output;
} else {
outputs.remove(node_id);
}
if let Some(error) = step.get("error") {
result["error"] = error.clone();
}
results.insert(node_id.to_string(), result.clone());
if let Some(mount_path) = exit_mount_path(definition, node_id) {
if outcome == "ok" {
let output = step.get("output").cloned().unwrap_or(Value::Null);
outputs.insert(mount_path.clone(), output.clone());
result["nodeId"] = json!(mount_path);
result["output"] = output;
results.insert(mount_path, result);
}
}
}
let mut state = json!({
"schema": RUN_STATE_SCHEMA,
"traceSeq": revision,
"runId": run_id,
"workflowName": workflow_name,
"startedAt": timestamp(created_at),
"updatedAt": timestamp(updated_at),
"status": status,
"input": read_json_blob(connection, &input_hash)?,
"outputs": outputs,
"results": results,
"steps": steps,
});
if let Some(value) = parent_run_id {
state["parentRunId"] = json!(value);
}
if let Some(value) = title {
state["runTitle"] = json!(value);
}
if let Some(value) = status_detail {
state["statusDetail"] = json!(value);
}
if paused != 0 {
state["paused"] = json!(true);
}
if let Some(value) = finished_at {
state["finishedAt"] = json!(timestamp(value));
}
if let Some(hash) = final_output_hash {
state["finalOutput"] = read_json_blob(connection, &hash)?;
}
if let Some(hash) = error_hash {
state["error"] = json!(read_text_blob(connection, &hash)?);
}
let carried: u64 = connection.query_row(
"SELECT count(*) FROM run_steps s
JOIN node_attempts a ON a.attempt_id = s.attempt_id
WHERE s.run_id = ?1 AND a.run_id <> s.run_id",
[run_id],
|row| row.get(0),
)?;
if carried != 0 {
state["carriedStepCount"] = json!(carried);
}
if status == "running" {
if let Some((attempt_id, node_id, started_at, scope_id, change_number, settings_hash)) =
connection
.query_row(
"SELECT attempt_id, node_id, started_at,
settings_scope_id, settings_change_number, settings_hash
FROM node_attempts
WHERE run_id = ?1 AND status IN ('pending', 'running', 'waiting', 'interrupted')",
[run_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<i64>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<u64>>(4)?,
row.get::<_, Option<Vec<u8>>>(5)?,
))
},
)
.optional()?
{
state["currentAttemptId"] = json!(attempt_id);
state["currentNode"] = json!(node_id);
if let Some(value) = started_at {
state["currentNodeStartedAt"] = json!(timestamp(value));
}
match (scope_id, change_number, settings_hash) {
(Some(scope_id), Some(change_number), Some(settings_hash)) => {
state["currentSettingsScopeId"] = json!(scope_id);
state["currentSettingsChangeNumber"] = json!(change_number);
state["currentSettingsHash"] = json!(encode_hex(&settings_hash));
}
(None, None, None) => {}
_ => bail!("active workflow settings binding is incomplete"),
}
}
}
if status == "waiting" {
if let Some(node_id) = state["steps"]
.as_array()
.and_then(|values| values.last())
.and_then(|step| step.get("nodeId"))
{
state["waitingOn"] = node_id.clone();
}
}
let (root_source, mounted_sources) = read_sources(connection, run_id, definition)?;
if let Some(source) = root_source {
state["workflowSource"] = source;
}
if !mounted_sources.is_empty() {
state["workflowSources"] = json!(mounted_sources);
}
let has_composed_mounts = definition
.pointer("/composition/mounts")
.and_then(Value::as_array)
.is_some_and(|mounts| !mounts.is_empty());
if !state["workflowSources"].is_null() || has_composed_mounts {
state["definitionDigest"] = json!(format!("sha256:{}", encode_hex(&definition_digest)));
}
if !updates.is_empty() {
state["updates"] = json!(updates);
}
if let Some(receipt) = read_human_decision_receipt(connection, run_id)? {
state["humanDecision"] = receipt;
}
Ok(serde_json::from_value(state)?)
}
fn read_graph_steps(
connection: &Connection,
run_id: &str,
cutoff: u64,
) -> Result<Vec<crate::state::types::StepRecord>> {
let mut statement = connection.prepare(
"WITH ranked AS (
SELECT s.step_index, a.attempt_id, a.node_id, a.node_type, a.status,
a.settings_scope_id, a.settings_change_number, a.settings_hash,
a.started_at, a.finished_at,
row_number() OVER (
PARTITION BY a.node_id ORDER BY s.step_index DESC
) AS position
FROM run_steps s
JOIN node_attempts a ON a.attempt_id = s.attempt_id
WHERE s.run_id = ?1 AND s.step_index <= ?2
)
SELECT attempt_id, node_id, node_type, status,
settings_scope_id, settings_change_number, settings_hash,
started_at, finished_at
FROM ranked WHERE position = 1 ORDER BY step_index",
)?;
let rows = statement.query_map(rusqlite::params![run_id, cutoff], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<u64>>(5)?,
row.get::<_, Option<Vec<u8>>>(6)?,
row.get::<_, i64>(7)?,
row.get::<_, i64>(8)?,
))
})?;
let mut steps = Vec::new();
for row in rows {
let (
attempt_id,
node_id,
node_type,
status,
settings_scope_id,
settings_change_number,
settings_hash,
started_at,
finished_at,
) = row?;
let mut value = json!({
"attemptId": attempt_id,
"nodeId": node_id,
"nodeType": node_type,
"outcome": outcome_for_status(&status)?,
"startedAt": timestamp(started_at),
"finishedAt": timestamp(finished_at),
"prompt": null,
"output": null,
});
if let Some(scope_id) = settings_scope_id {
value["settingsScopeId"] = json!(scope_id);
}
if let Some(change_number) = settings_change_number {
value["settingsChangeNumber"] = json!(change_number);
}
if let Some(hash) = settings_hash {
value["settingsHash"] = json!(encode_hex(&hash));
}
steps.push(serde_json::from_value(value)?);
}
Ok(steps)
}
fn read_taken_transitions(
connection: &Connection,
run_id: &str,
cutoff: u64,
) -> Result<Vec<String>> {
let mut statement = connection.prepare(
"WITH ordered AS (
SELECT s.step_index, a.node_id,
lag(a.node_id) OVER (ORDER BY s.step_index) AS previous_node
FROM run_steps s
JOIN node_attempts a ON a.attempt_id = s.attempt_id
WHERE s.run_id = ?1 AND s.step_index <= ?2
)
SELECT DISTINCT previous_node, node_id
FROM ordered WHERE previous_node IS NOT NULL
ORDER BY previous_node, node_id",
)?;
let rows = statement.query_map(rusqlite::params![run_id, cutoff], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?;
rows.map(|row| row.map(|(from, to)| format!("{from}->{to}")))
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
fn read_steps(connection: &Connection, run_id: &str, start: Option<u64>) -> Result<Vec<Value>> {
let mut statement = connection.prepare(
"SELECT a.attempt_id, a.node_id, a.node_type, a.status,
a.prompt_hash, a.output_hash, s.output_override_hash, a.receipt_hash, a.error_hash,
prompt_entry.entry_hash, response_entry.entry_hash,
first_link.entry_id, last_link.entry_id,
a.settings_scope_id, a.settings_change_number, a.settings_hash,
a.started_at, a.finished_at
FROM run_steps s JOIN node_attempts a ON a.attempt_id = s.attempt_id
LEFT JOIN attempt_entries prompt_link
ON prompt_link.attempt_id = a.attempt_id AND prompt_link.role = 'prompt'
LEFT JOIN session_entries prompt_entry
ON prompt_entry.segment_id = prompt_link.segment_id AND prompt_entry.entry_id = prompt_link.entry_id
LEFT JOIN attempt_entries response_link
ON response_link.attempt_id = a.attempt_id AND response_link.role = 'response'
LEFT JOIN session_entries response_entry
ON response_entry.segment_id = response_link.segment_id AND response_entry.entry_id = response_link.entry_id
LEFT JOIN attempt_entries first_link
ON first_link.attempt_id = a.attempt_id AND first_link.role = 'first'
LEFT JOIN attempt_entries last_link
ON last_link.attempt_id = a.attempt_id AND last_link.role = 'last'
WHERE s.run_id = ?1 AND (?2 IS NULL OR s.step_index >= ?2)
ORDER BY s.step_index LIMIT ?3",
)?;
let limit = start.map_or(-1_i64, |_| VIEWER_PAGE_SIZE as i64);
let rows = statement.query_map(rusqlite::params![run_id, start, limit], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<Vec<u8>>>(4)?,
row.get::<_, Option<Vec<u8>>>(5)?,
row.get::<_, Option<Vec<u8>>>(6)?,
row.get::<_, Option<Vec<u8>>>(7)?,
row.get::<_, Option<Vec<u8>>>(8)?,
row.get::<_, Option<Vec<u8>>>(9)?,
row.get::<_, Option<Vec<u8>>>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<u64>>(14)?,
row.get::<_, Option<Vec<u8>>>(15)?,
row.get::<_, i64>(16)?,
row.get::<_, i64>(17)?,
))
})?;
let mut steps = Vec::new();
for row in rows {
let (
attempt_id,
node_id,
node_type,
status,
stored_prompt_hash,
output_hash,
override_hash,
receipt_hash,
error_hash,
prompt_hash,
response_hash,
first_entry_id,
last_entry_id,
settings_scope_id,
settings_change_number,
settings_hash,
started_at,
finished_at,
) = row?;
let receipt = receipt_hash
.as_deref()
.map(|hash| read_json_blob(connection, hash))
.transpose()?
.unwrap_or_else(|| json!({}));
let prompt = if let Some(hash) = prompt_hash.as_deref() {
let entry = read_json_blob(connection, hash)?;
prompt_from_entry(&entry).map_or(Value::Null, Value::String)
} else if let Some(hash) = stored_prompt_hash.as_deref() {
Value::String(read_text_blob(connection, hash)?)
} else {
Value::Null
};
let output = if let Some(hash) = override_hash.as_deref().or(output_hash.as_deref()) {
read_json_blob(connection, hash)?
} else if let Some(hash) = response_hash.as_deref() {
assistant_output_from_entry(&read_json_blob(connection, hash)?)?
} else {
Value::Null
};
let mut step = json!({
"attemptId": attempt_id,
"nodeId": node_id,
"nodeType": node_type,
"outcome": outcome_for_status(&status)?,
"startedAt": timestamp(started_at),
"finishedAt": timestamp(finished_at),
"prompt": prompt,
"output": output,
});
if let Some(hash) = error_hash {
step["error"] = json!(read_text_blob(connection, &hash)?);
}
if let Some(value) = receipt.get("action") {
step["action"] = value.clone();
}
if let Some(value) = receipt.get("assistantMessage") {
step["assistantMessage"] = value.clone();
}
if let (Some(first), Some(last)) = (first_entry_id, last_entry_id) {
step["conversation"] = json!({ "firstEntryId": first, "lastEntryId": last });
}
match (settings_scope_id, settings_change_number, settings_hash) {
(Some(scope_id), Some(change_number), Some(settings_hash)) => {
step["settingsScopeId"] = json!(scope_id);
step["settingsChangeNumber"] = json!(change_number);
step["settingsHash"] = json!(encode_hex(&settings_hash));
}
(None, None, None) => {}
_ => bail!("saved workflow settings binding is incomplete"),
}
steps.push(step);
}
Ok(steps)
}
fn read_sources(
connection: &Connection,
run_id: &str,
definition: &Value,
) -> Result<(Option<Value>, Vec<Value>)> {
let mut statement = connection.prepare(
"SELECT mount_path, source_type, source_ref, source_revision
FROM run_sources WHERE run_id = ?1 ORDER BY mount_path",
)?;
let rows = statement
.query_map([run_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
})?
.collect::<Result<Vec<_>, _>>()?;
let mut root = None;
let mut mounted = Vec::new();
for (mount_path, source_type, source_ref, source_revision) in rows {
let source = if source_type == "builtin" {
json!({ "kind": "builtin", "id": source_ref, "revision": source_revision })
} else {
json!({ "kind": "file", "path": source_ref, "hash": source_revision })
};
if mount_path.is_empty() {
if source["kind"] != "file"
|| !source["path"]
.as_str()
.is_some_and(|value| value.starts_with("inline:"))
{
root = Some(source);
}
continue;
}
let workflow_name = definition["composition"]["mounts"]
.as_array()
.and_then(|mounts| {
mounts.iter().find(|mount| {
mount["mountPath"].as_array().is_some_and(|parts| {
parts
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>()
.join("/")
== mount_path
})
})
})
.and_then(|mount| mount["workflowName"].as_str())
.unwrap_or(&mount_path);
mounted.push(json!({
"mountPath": mount_path.split('/').collect::<Vec<_>>(),
"workflowName": workflow_name,
"source": source,
}));
}
Ok((root, mounted))
}
fn read_updates_range(
connection: &Connection,
run_id: &str,
start: u64,
limit: i64,
) -> Result<Vec<Value>> {
let mut statement = connection.prepare(
"WITH latest AS (
SELECT u.update_id, u.run_revision, a.node_id, u.attempt_id,
u.update_type, u.update_key, u.data_hash, u.recorded_at,
row_number() OVER (
PARTITION BY u.update_type, u.update_key
ORDER BY u.run_revision DESC
) AS position
FROM workflow_updates u
JOIN node_attempts a ON a.attempt_id = u.attempt_id
WHERE a.run_id = ?1
)
SELECT update_id, run_revision, node_id, attempt_id,
update_type, update_key, data_hash, recorded_at
FROM latest WHERE position = 1
ORDER BY run_revision LIMIT ?2 OFFSET ?3",
)?;
let rows = statement.query_map(rusqlite::params![run_id, limit, start], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, u64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, Vec<u8>>(6)?,
row.get::<_, i64>(7)?,
))
})?;
let mut values = Vec::new();
for row in rows {
let (update_id, seq, node_id, attempt_id, kind, key, hash, at) = row?;
values.push(json!({
"updateId": update_id, "seq": seq, "at": timestamp(at), "runId": run_id,
"nodeId": node_id, "attemptId": attempt_id, "type": kind, "key": key,
"data": read_json_blob(connection, &hash)?,
}));
}
values.sort_by_key(|value| value["seq"].as_u64().unwrap_or_default());
Ok(values)
}
fn read_update_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM (
SELECT 1 FROM workflow_updates u
JOIN node_attempts a ON a.attempt_id = u.attempt_id
WHERE a.run_id = ?1 GROUP BY u.update_type, u.update_key
)",
[run_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
Ok(projection_page(
start,
total,
read_updates_range(connection, run_id, start, VIEWER_PAGE_SIZE as i64)?,
))
}
fn read_human_decision_receipt(connection: &Connection, run_id: &str) -> Result<Option<Value>> {
let row = connection
.query_row(
"SELECT d.request_hash, r.response_hash FROM continuations c
JOIN human_decisions d ON d.decision_id = c.decision_id
JOIN human_decision_resolutions r ON r.decision_id = c.decision_id
WHERE c.continuation_run_id = ?1 AND r.outcome = 'accepted'",
[run_id],
|row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, Vec<u8>>(1)?)),
)
.optional()?;
let Some((request_hash, decision_hash)) = row else {
return Ok(None);
};
let request = read_json_blob(connection, &request_hash)?;
let decision = read_json_blob(connection, &decision_hash)?;
Ok(Some(json!({
"schema": "pi-workflows.human-decision-receipt.v1",
"decisionId": request["decisionId"], "requestDigest": request["requestDigest"],
"nodeId": request["nodeId"], "response": decision["response"],
"provenance": decision["provenance"], "acceptedAt": decision["acceptedAt"],
"answerDigest": decision["answerDigest"], "subjectDigest": decision["subjectDigest"],
"presentationDigest": decision["presentationDigest"], "revision": decision["revision"],
})))
}
fn prompt_from_entry(entry: &Value) -> Option<String> {
if let Some(text) = entry.get("content").and_then(Value::as_str) {
return Some(text.to_string());
}
entry.get("content")?.as_array().map(|parts| {
parts
.iter()
.filter_map(|part| part.get("text").and_then(Value::as_str))
.collect::<Vec<_>>()
.join("\n")
})
}
fn assistant_output_from_entry(entry: &Value) -> Result<Value> {
let Some(parts) = entry["message"]["content"].as_array() else {
bail!("assistant response entry is invalid");
};
let text = parts
.iter()
.filter(|part| part["type"] == "text")
.filter_map(|part| part["text"].as_str())
.collect::<Vec<_>>()
.join("\n");
if text.trim().is_empty() {
bail!("assistant response entry has no visible text");
}
Ok(Value::String(text))
}
fn outcome_for_status(status: &str) -> Result<&'static str> {
match status {
"completed" => Ok("ok"),
"failed" => Ok("failed"),
"timed_out" => Ok("timed_out"),
"cancelled" => Ok("cancelled"),
_ => bail!("workflow step has nonterminal status: {status}"),
}
}
fn exit_mount_path(definition: &Value, node_id: &str) -> Option<String> {
let node = definition.get("nodes")?.get(node_id)?;
if node.get("includeTransition")?.as_str()? != "exit" {
return None;
}
Some(
node.get("mountPath")?
.as_array()?
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>()
.join("/"),
)
}
fn encode_hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push(HEX[(byte >> 4) as usize] as char);
output.push(HEX[(byte & 0x0f) as usize] as char);
}
output
}
fn read_trace(connection: &Connection, run_id: &str) -> Result<Vec<TraceEvent>> {
let resource_id: String = connection.query_row(
"SELECT resource_id FROM runs WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let mut statement = connection.prepare(
"SELECT resource_revision, event_type, payload_hash, recorded_at
FROM events WHERE resource_id = ?1 ORDER BY resource_revision",
)?;
let rows = statement.query_map([resource_id], |row| {
Ok((
row.get::<_, u64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<Vec<u8>>>(2)?,
row.get::<_, i64>(3)?,
))
})?;
let mut events = Vec::new();
for row in rows {
let (seq, event_type, payload_hash, recorded_at) = row?;
let envelope = match payload_hash {
Some(hash) => read_json_blob(connection, &hash)?,
None => json!({}),
};
let payload = envelope
.get("payload")
.cloned()
.unwrap_or_else(|| json!({}));
let mut event = json!({
"seq": seq,
"at": timestamp(recorded_at),
"runId": run_id,
"scope": envelope.get("scope").and_then(Value::as_str).unwrap_or("run"),
"type": event_type,
"payload": payload,
});
if let Some(node_id) = envelope.get("nodeId") {
event["nodeId"] = node_id.clone();
}
if let Some(attempt_id) = envelope.get("attemptId") {
event["attemptId"] = attempt_id.clone();
}
events.push(serde_json::from_value(event)?);
}
Ok(events)
}
fn read_trace_window(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<(Vec<TraceEvent>, u64, u64)> {
let resource_id: String = connection.query_row(
"SELECT resource_id FROM runs WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let total: u64 = connection.query_row(
"SELECT count(*) FROM events WHERE resource_id = ?1",
[&resource_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
let mut statement = connection.prepare(
"SELECT resource_revision, event_type, payload_hash, recorded_at
FROM events
WHERE resource_id = ?1 AND resource_revision > ?2
ORDER BY resource_revision LIMIT ?3",
)?;
let rows = statement.query_map(
rusqlite::params![resource_id, start, VIEWER_PAGE_SIZE],
|row| {
Ok((
row.get::<_, u64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<Vec<u8>>>(2)?,
row.get::<_, i64>(3)?,
))
},
)?;
let mut events = Vec::new();
for row in rows {
let (seq, event_type, payload_hash, recorded_at) = row?;
let envelope = match payload_hash {
Some(hash) => read_json_blob(connection, &hash)?,
None => json!({}),
};
let mut event = json!({
"seq": seq,
"at": timestamp(recorded_at),
"runId": run_id,
"scope": envelope.get("scope").and_then(Value::as_str).unwrap_or("run"),
"type": event_type,
"payload": envelope.get("payload").cloned().unwrap_or_else(|| json!({})),
});
if let Some(node_id) = envelope.get("nodeId") {
event["nodeId"] = node_id.clone();
}
if let Some(attempt_id) = envelope.get("attemptId") {
event["attemptId"] = attempt_id.clone();
}
events.push(serde_json::from_value(event)?);
}
Ok((events, start, total))
}
fn trace_cursor_for_step(
connection: &Connection,
run_id: &str,
step_index: u64,
) -> Result<Option<u64>> {
let timestamp = connection
.query_row(
"SELECT COALESCE(a.finished_at, a.started_at)
FROM run_steps s
JOIN node_attempts a ON a.attempt_id = s.attempt_id
WHERE s.run_id = ?1 AND s.step_index = ?2",
rusqlite::params![run_id, step_index],
|row| row.get::<_, Option<i64>>(0),
)
.optional()?
.flatten();
let Some(timestamp) = timestamp else {
return Ok(None);
};
let count: u64 = connection.query_row(
"SELECT count(*)
FROM events e JOIN runs r ON r.resource_id = e.resource_id
WHERE r.run_id = ?1 AND e.recorded_at <= ?2",
rusqlite::params![run_id, timestamp],
|row| row.get(0),
)?;
Ok(count.checked_sub(1))
}
fn read_session(connection: &Connection, run_id: &str) -> Result<LoadedSession> {
let mut segments_statement = connection.prepare(
"SELECT segment_id, binding_hash, status, entry_count, event_count,
failure_hash
FROM session_segments
WHERE run_id = ?1
ORDER BY created_at, segment_id",
)?;
let segment_rows = segments_statement
.query_map([run_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<Vec<u8>>>(1)?,
row.get::<_, String>(2)?,
row.get::<_, u64>(3)?,
row.get::<_, u64>(4)?,
row.get::<_, Option<Vec<u8>>>(5)?,
))
})?
.collect::<Result<Vec<_>, _>>()?;
if segment_rows.is_empty() {
return Ok((None, Vec::new(), Vec::new(), None));
}
let mut binding = None;
let mut entries = Vec::new();
let mut events = Vec::new();
let mut status = "complete".to_string();
let mut failure = None;
for (segment_id, binding_hash, segment_status, _entry_count, _event_count, failure_hash) in
segment_rows
{
if let Some(hash) = binding_hash {
binding = Some(serde_json::from_value(read_json_blob(connection, &hash)?)?);
}
if segment_status == "failed" {
status = "failed".to_string();
if failure.is_none() {
if let Some(hash) = failure_hash {
failure = Some(read_json_blob(connection, &hash)?);
}
}
} else if segment_status == "recording" && status != "failed" {
status = "recording".to_string();
}
let mut entries_statement = connection.prepare(
"SELECT entry_hash, recorded_at
FROM session_entries WHERE segment_id = ?1 ORDER BY entry_seq",
)?;
let segment_entries = entries_statement
.query_map([&segment_id], |row| {
Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?))
})?
.collect::<Result<Vec<_>, _>>()?;
for (hash, at) in segment_entries {
entries.push(serde_json::from_value(json!({
"seq": entries.len() + 1,
"at": timestamp(at),
"entry": read_json_blob(connection, &hash)?,
}))?);
}
let mut events_statement = connection.prepare(
"SELECT event_type, node_id, attempt_id, turn_id,
message_id, tool_call_id, payload_hash, recorded_at
FROM session_events WHERE segment_id = ?1 ORDER BY event_seq",
)?;
let segment_events = events_statement
.query_map([&segment_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, Vec<u8>>(6)?,
row.get::<_, i64>(7)?,
))
})?
.collect::<Result<Vec<_>, _>>()?;
for (event_type, node_id, attempt_id, turn_id, message_id, tool_call_id, hash, at) in
segment_events
{
let mut value = json!({
"seq": events.len() + 1,
"at": timestamp(at),
"nodeId": node_id,
"attemptId": attempt_id,
"type": event_type,
"payload": read_json_blob(connection, &hash)?,
});
if let Some(turn_id) = turn_id {
value["turnId"] = json!(turn_id);
}
if let Some(message_id) = message_id {
value["messageId"] = json!(message_id);
}
if let Some(tool_call_id) = tool_call_id {
value["toolCallId"] = json!(tool_call_id);
}
events.push(serde_json::from_value(value)?);
}
}
let mut capture = json!({
"schema": "pi-workflows.session-capture.v1",
"eventSchema": "pi-workflows.session-event.v1",
"status": status,
"eventCount": events.len(),
"entryCount": entries.len(),
"lastEventSeq": events.len(),
});
if let Some(failure) = failure {
capture["failure"] = failure;
}
Ok((
binding,
entries,
events,
Some(serde_json::from_value(capture)?),
))
}
fn read_step_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM run_steps WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
let graph_cursor = cursor
.unwrap_or_else(|| total.saturating_sub(1))
.min(total.saturating_sub(1));
Ok(ProjectionPage {
start,
total,
items: read_steps(connection, run_id, Some(start))?,
graph_cursor: Some(graph_cursor),
graph_steps: Some(read_graph_steps(connection, run_id, graph_cursor)?),
taken_transitions: Some(read_taken_transitions(connection, run_id, graph_cursor)?),
replay_checkpoint: None,
})
}
fn read_session_entry_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM session_entries WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let start = page_start(total, cursor);
let mut statement = connection.prepare(
"SELECT entry_hash, recorded_at
FROM session_entries
WHERE run_id = ?1 AND run_seq > ?2
ORDER BY run_seq LIMIT ?3",
)?;
let rows = statement.query_map(rusqlite::params![run_id, start, VIEWER_PAGE_SIZE], |row| {
Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?))
})?;
let mut items = Vec::new();
for (index, row) in rows.enumerate() {
let (hash, at) = row?;
items.push(json!({
"seq": start + index as u64 + 1,
"at": timestamp(at),
"entry": read_json_blob(connection, &hash)?,
}));
}
Ok(ProjectionPage {
start,
total,
items,
graph_cursor: None,
graph_steps: None,
taken_transitions: None,
replay_checkpoint: None,
})
}
fn read_session_event_page(
connection: &Connection,
run_id: &str,
cursor: Option<u64>,
) -> Result<ProjectionPage> {
let total: u64 = connection.query_row(
"SELECT count(*) FROM session_events WHERE run_id = ?1",
[run_id],
|row| row.get(0),
)?;
let start = session_page_start(total, cursor);
let items = read_session_events_range(connection, run_id, start, VIEWER_PAGE_SIZE as i64)?
.into_iter()
.map(serde_json::to_value)
.collect::<Result<Vec<_>, _>>()?;
Ok(ProjectionPage {
start,
total,
items,
graph_cursor: None,
graph_steps: None,
taken_transitions: None,
replay_checkpoint: session_replay_checkpoint(connection, run_id, start)?,
})
}
fn read_session_events_range(
connection: &Connection,
run_id: &str,
start: u64,
limit: i64,
) -> Result<Vec<SessionEventRecord>> {
let mut statement = connection.prepare(
"SELECT e.event_type, e.node_id, e.attempt_id, e.turn_id,
e.message_id, e.tool_call_id, e.payload_hash, e.recorded_at,
s.step_index
FROM session_events e
LEFT JOIN run_steps s ON s.run_id = e.run_id AND s.attempt_id = e.attempt_id
WHERE e.run_id = ?1 AND e.run_seq > ?2
ORDER BY e.run_seq LIMIT ?3",
)?;
let rows = statement.query_map(rusqlite::params![run_id, start, limit], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, Vec<u8>>(6)?,
row.get::<_, i64>(7)?,
row.get::<_, Option<u64>>(8)?,
))
})?;
let mut events = Vec::new();
for (index, row) in rows.enumerate() {
let (
event_type,
node_id,
attempt_id,
turn_id,
message_id,
tool_call_id,
hash,
at,
step_index,
) = row?;
let mut value = json!({
"seq": start + index as u64 + 1,
"at": timestamp(at),
"nodeId": node_id,
"attemptId": attempt_id,
"type": event_type,
"payload": read_json_blob(connection, &hash)?,
});
if let Some(step_index) = step_index {
value["stepIndex"] = json!(step_index);
}
if let Some(turn_id) = turn_id {
value["turnId"] = json!(turn_id);
}
if let Some(message_id) = message_id {
value["messageId"] = json!(message_id);
}
if let Some(tool_call_id) = tool_call_id {
value["toolCallId"] = json!(tool_call_id);
}
events.push(serde_json::from_value(value)?);
}
Ok(events)
}
fn session_replay_checkpoint(
connection: &Connection,
run_id: &str,
start: u64,
) -> Result<Option<Value>> {
if start == 0 {
return Ok(None);
}
let hash = connection
.query_row(
"SELECT state_hash FROM viewer_session_checkpoints
WHERE run_id = ?1 AND event_seq = ?2",
rusqlite::params![run_id, start],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()?;
let Some(hash) = hash else {
bail!("session replay checkpoint is missing for run {run_id} at {start}");
};
Ok(Some(read_json_blob(connection, &hash)?))
}
fn read_session_window(
connection: &Connection,
run_id: &str,
entry_cursor: Option<u64>,
event_cursor: Option<u64>,
) -> Result<LoadedSessionWindow> {
let mut segments_statement = connection.prepare(
"SELECT binding_hash, status, entry_count, event_count, failure_hash
FROM session_segments
WHERE run_id = ?1
ORDER BY created_at, segment_id",
)?;
let segment_rows = segments_statement
.query_map([run_id], |row| {
Ok((
row.get::<_, Option<Vec<u8>>>(0)?,
row.get::<_, String>(1)?,
row.get::<_, u64>(2)?,
row.get::<_, u64>(3)?,
row.get::<_, Option<Vec<u8>>>(4)?,
))
})?
.collect::<Result<Vec<_>, _>>()?;
if segment_rows.is_empty() {
return Ok((None, Vec::new(), 0, 0, Vec::new(), 0, 0, None, None));
}
let mut binding = None;
let mut status = "complete".to_string();
let mut failure = None;
let mut entry_total = 0u64;
let mut event_total = 0u64;
for (binding_hash, segment_status, entry_count, event_count, failure_hash) in &segment_rows {
entry_total += entry_count;
event_total += event_count;
if let Some(hash) = binding_hash {
binding = Some(serde_json::from_value(read_json_blob(connection, hash)?)?);
}
if segment_status == "failed" {
status = "failed".to_string();
if failure.is_none() {
if let Some(hash) = failure_hash {
failure = Some(read_json_blob(connection, hash)?);
}
}
} else if segment_status == "recording" && status != "failed" {
status = "recording".to_string();
}
}
let entry_start = page_start(entry_total, entry_cursor);
let mut entry_statement = connection.prepare(
"SELECT entry_hash, recorded_at
FROM session_entries
WHERE run_id = ?1 AND run_seq > ?2
ORDER BY run_seq
LIMIT ?3",
)?;
let entry_rows = entry_statement.query_map(
rusqlite::params![run_id, entry_start, VIEWER_PAGE_SIZE],
|row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?)),
)?;
let mut entries = Vec::new();
for (index, row) in entry_rows.enumerate() {
let (hash, at) = row?;
entries.push(serde_json::from_value(json!({
"seq": entry_start + index as u64 + 1,
"at": timestamp(at),
"entry": read_json_blob(connection, &hash)?,
}))?);
}
let event_start = session_page_start(event_total, event_cursor);
let mut event_statement = connection.prepare(
"SELECT e.event_type, e.node_id, e.attempt_id, e.turn_id,
e.message_id, e.tool_call_id, e.payload_hash, e.recorded_at,
s.step_index
FROM session_events e
LEFT JOIN run_steps s ON s.run_id = e.run_id AND s.attempt_id = e.attempt_id
WHERE e.run_id = ?1 AND e.run_seq > ?2
ORDER BY e.run_seq
LIMIT ?3",
)?;
let event_rows = event_statement.query_map(
rusqlite::params![run_id, event_start, VIEWER_PAGE_SIZE],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, Vec<u8>>(6)?,
row.get::<_, i64>(7)?,
row.get::<_, Option<u64>>(8)?,
))
},
)?;
let mut events = Vec::new();
for (index, row) in event_rows.enumerate() {
let (
event_type,
node_id,
attempt_id,
turn_id,
message_id,
tool_call_id,
hash,
at,
step_index,
) = row?;
let mut value = json!({
"seq": event_start + index as u64 + 1,
"at": timestamp(at),
"nodeId": node_id,
"attemptId": attempt_id,
"type": event_type,
"payload": read_json_blob(connection, &hash)?,
});
if let Some(step_index) = step_index {
value["stepIndex"] = json!(step_index);
}
if let Some(turn_id) = turn_id {
value["turnId"] = json!(turn_id);
}
if let Some(message_id) = message_id {
value["messageId"] = json!(message_id);
}
if let Some(tool_call_id) = tool_call_id {
value["toolCallId"] = json!(tool_call_id);
}
events.push(serde_json::from_value(value)?);
}
let mut capture = json!({
"schema": "pi-workflows.session-capture.v1",
"eventSchema": "pi-workflows.session-event.v1",
"status": status,
"eventCount": event_total,
"entryCount": entry_total,
"lastEventSeq": event_total,
});
if let Some(failure) = failure {
capture["failure"] = failure;
}
Ok((
binding,
entries,
entry_start,
entry_total,
events,
event_start,
event_total,
Some(serde_json::from_value(capture)?),
session_replay_checkpoint(connection, run_id, event_start)?,
))
}
fn projection_page(start: u64, total: u64, items: Vec<Value>) -> ProjectionPage {
ProjectionPage {
start,
total,
items,
graph_cursor: None,
graph_steps: None,
taken_transitions: None,
replay_checkpoint: None,
}
}
fn session_page_start(total: u64, cursor: Option<u64>) -> u64 {
if total <= VIEWER_PAGE_SIZE {
return 0;
}
let selected = cursor
.unwrap_or(total.saturating_sub(1))
.min(total.saturating_sub(1));
(selected / VIEWER_PAGE_SIZE) * VIEWER_PAGE_SIZE
}
fn page_start(total: u64, cursor: Option<u64>) -> u64 {
if total <= VIEWER_PAGE_SIZE {
return 0;
}
let center = cursor
.unwrap_or(total.saturating_sub(1))
.min(total.saturating_sub(1));
center
.saturating_sub(VIEWER_PAGE_SIZE / 2)
.min(total - VIEWER_PAGE_SIZE)
}
fn manifest_from_state(state: &RunState) -> Manifest {
Manifest {
schema: "pi-workflows.sqlite-view.v1".to_string(),
run_id: state.run_id.clone(),
workflow_name: state.workflow_name.clone(),
run_title: state.run_title.clone(),
workflow_source: state.workflow_source.clone(),
started_at: state.started_at.clone(),
finished_at: state.finished_at.clone(),
status: state.status,
trace_schema: "pi-workflows.event.v1".to_string(),
paths: ManifestPaths {
workflow: String::new(),
state: String::new(),
trace: String::new(),
session: None,
artifacts: None,
},
}
}
fn timestamp(milliseconds: i64) -> String {
Utc.timestamp_millis_opt(milliseconds)
.single()
.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
.unwrap_or_else(|| "1970-01-01T00:00:00.000Z".to_string())
}