piw/state/

reader.rs

//! Read-only access to the canonical Pi Workflows SQLite database.

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::path::Path;

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] = [
    0x46, 0x6d, 0xdb, 0xae, 0x92, 0x1f, 0x26, 0x19, 0xff, 0x81, 0xdb, 0x56, 0x73, 0x95, 0x18, 0xe7,
    0xef, 0x68, 0xfa, 0x0a, 0x23, 0xbf, 0xf8, 0x0d, 0xf0, 0x29, 0x6b, 0x3a, 0xfd, 0x01, 0xc8, 0x56,
];
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(Debug, Clone)]
pub struct LoadedRun {
    pub manifest: Manifest,
    pub state: RunState,
    pub snapshot: Option<DefinitionSnapshot>,
    pub trace: Vec<TraceEvent>,
    pub session_binding: Option<SessionBinding>,
    pub session_entries: Vec<SessionEntryRecord>,
    pub session_events: Vec<SessionEventRecord>,
    pub session_capture: Option<SessionCapture>,
    pub settings_scopes: Vec<Value>,
    pub follow_up_queue: Option<Value>,
    pub possibly_interrupted: bool,
}

pub fn read_run(database_path: &Path, run_id: &str) -> Result<LoadedRun> {
    let connection = open(database_path)?;
    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 state = read_state(&connection, run_id, &definition_value)?;
    let trace = read_trace(&connection, run_id)?;
    let (session_binding, session_entries, session_events, session_capture) =
        read_session(&connection, run_id)?;
    let settings_scopes = read_settings(&connection, run_id)?;
    let follow_up_queue = read_follow_ups(&connection, run_id)?;
    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,
        snapshot: Some(snapshot),
        trace,
        session_binding,
        session_entries,
        session_events,
        session_capture,
        settings_scopes,
        follow_up_queue,
        possibly_interrupted,
    })
}

pub fn list_runs(database_path: &Path) -> Vec<(String, Manifest)> {
    let Ok(connection) = open(database_path) else {
        return Vec::new();
    };
    let Ok(mut statement) = connection.prepare("SELECT run_id FROM runs ORDER BY created_at DESC")
    else {
        return Vec::new();
    };
    let Ok(ids) = statement.query_map([], |row| row.get::<_, String>(0)) else {
        return Vec::new();
    };
    ids.filter_map(|id| {
        let id = id.ok()?;
        let run = read_run(database_path, &id).ok()?;
        Some((id, run.manifest))
    })
    .collect()
}

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(connection: &Connection, run_id: &str) -> 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",
    )?;
    let rows = statement.query_map([run_id], |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 scopes = Vec::new();
    for row in rows {
        let (scope_id, mount_path, invocation, settings_hash, change_number) = row?;
        scopes.push(json!({
            "scopeId": scope_id,
            "mountPath": mount_path,
            "invocation": invocation,
            "changeNumber": change_number,
            "settingsHash": encode_hex(&settings_hash),
        }));
    }
    Ok(scopes)
}

fn read_follow_ups(connection: &Connection, run_id: &str) -> Result<Option<Value>> {
    let queue = connection
        .query_row(
            "SELECT presentation_state FROM workflow_follow_up_queues WHERE run_id = ?1",
            [run_id],
            |row| row.get::<_, String>(0),
        )
        .optional()?;
    let Some(presentation_state) = queue else {
        return Ok(None);
    };
    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",
    )?;
    let rows = statement.query_map([run_id], |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)?,
        }))
    })?;
    let items = rows.collect::<Result<Vec<_>, _>>()?;
    Ok(Some(json!({
        "presentationState": presentation_state,
        "items": items,
    })))
}

fn read_state(connection: &Connection, run_id: &str, definition: &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)?;
    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)));
    }
    let updates = read_updates(connection, run_id)?;
    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_steps(connection: &Connection, run_id: &str) -> 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 ORDER BY s.step_index",
    )?;
    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)?,
            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(connection: &Connection, run_id: &str) -> Result<Vec<Value>> {
    let mut statement = connection.prepare(
        "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
         FROM workflow_updates u JOIN node_attempts a ON a.attempt_id = u.attempt_id
         WHERE a.run_id = ?1 ORDER BY u.run_revision",
    )?;
    let rows = statement.query_map([run_id], |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 current = std::collections::BTreeMap::new();
    for row in rows {
        let (update_id, seq, node_id, attempt_id, kind, key, hash, at) = row?;
        current.insert(
            (kind.clone(), key.clone()),
            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)?,
            }),
        );
    }
    let mut values = current.into_values().collect::<Vec<_>>();
    values.sort_by_key(|value| value["seq"].as_u64().unwrap_or_default());
    Ok(values)
}

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_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 binding.is_none() {
            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 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())
}