pointlock_store/projection/
inbox.rs

1//! `HumanInboxEntry` — the unified inbox projection (spine §10.1,
2//! 08 §2.6): every `humanRequested` without a paired FINAL
3//! `humanResponded`, across all runs, human steps and supervision gates
4//! in the SAME box (R13 — one arbitration, one channel, one inbox). A
5//! supervision `suspend` answer is non-final and keeps its request
6//! pending (spine §6.9).
7//!
8//! v0.1 is notify-side only: entries render, responses go through
9//! `pointlock-human-cli` (06 §4.2 — the `webUi` collect channel is a
10//! reserved v0.2 surface).
11
12use pointlock_ir::{JsonSchemaDocument, RunLogPayload, RunPath, render_run_path};
13use schemars::JsonSchema;
14use serde::{Deserialize, Serialize};
15use serde_json::Value;
16
17use super::ProjectionVersion;
18use crate::error::StoreError;
19use crate::store::Store;
20
21/// One pending human request (step or supervision — `purpose` is the
22/// discriminator, spine A.4).
23#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
24#[serde(rename_all = "camelCase", deny_unknown_fields)]
25pub struct HumanInboxEntry {
26    /// Protocol version (spine §10.3).
27    pub projection_version: ProjectionVersion,
28    /// The run awaiting the response.
29    pub run_id: String,
30    /// The run's flow.
31    pub flow_id: String,
32    /// The pairing id a response must carry.
33    pub request_id: String,
34    /// `step` vs `supervision` (R13).
35    pub purpose: String,
36    /// Interaction mode (`purpose = step` only).
37    #[serde(skip_serializing_if = "Option::is_none")]
38    pub mode: Option<String>,
39    /// The prompt.
40    pub prompt: String,
41    /// Materialized exhibits (values render directly; evidence refs go
42    /// through the gallery route — 08 §2.6).
43    pub presents: Value,
44    /// Confirm labels, when the mode declares them.
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub decisions: Option<Vec<String>>,
47    /// The provideInput contract, when the mode declares one.
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub output_schema: Option<JsonSchemaDocument>,
50    /// Absolute response deadline (ms); absent for supervision gates —
51    /// they never time out (spine §6.9). After the deadline the runner
52    /// settles the request to `unknown` (fixed vocabulary).
53    #[serde(skip_serializing_if = "Option::is_none")]
54    pub deadline_at_ms: Option<u64>,
55    /// When the request was recorded.
56    pub requested_at_ms: u64,
57    /// Ledger seq of the request.
58    pub requested_seq: u64,
59    /// Canonical string of the awaiting/gated step's path.
60    pub run_path: String,
61    /// The structured path (frames are the authority — spine §9).
62    pub run_path_frames: RunPath,
63}
64
65/// Scans one run's ledger for pending entries, in request order.
66pub fn run_inbox(store: &Store, run_id: &str) -> Result<Vec<HumanInboxEntry>, StoreError> {
67    let meta = store.run_meta(run_id)?;
68    let events = store.events(run_id)?;
69    let mut pending: Vec<HumanInboxEntry> = Vec::new();
70    for event in &events {
71        match &event.payload {
72            RunLogPayload::HumanRequested {
73                request_id,
74                purpose,
75                mode,
76                prompt,
77                presents,
78                decisions,
79                output_schema,
80                deadline_at_ms,
81            } => pending.push(HumanInboxEntry {
82                projection_version: ProjectionVersion,
83                run_id: run_id.to_owned(),
84                flow_id: meta.flow_id.to_string(),
85                request_id: request_id.clone(),
86                purpose: wire(purpose),
87                mode: mode.as_ref().map(wire),
88                prompt: prompt.clone(),
89                presents: presents.clone(),
90                decisions: decisions.clone(),
91                output_schema: output_schema.clone(),
92                deadline_at_ms: *deadline_at_ms,
93                requested_at_ms: event.at_ms,
94                requested_seq: event.seq,
95                run_path: render_run_path(&event.run_path),
96                run_path_frames: event.run_path.clone(),
97            }),
98            RunLogPayload::HumanResponded {
99                request_id,
100                purpose,
101                response,
102                ..
103            } => {
104                let non_final = *purpose == pointlock_ir::HumanPurpose::Supervision
105                    && response.get("decision").and_then(Value::as_str) == Some("suspend");
106                if !non_final {
107                    pending.retain(|entry| entry.request_id != *request_id);
108                }
109            }
110            // A terminal exit of the awaiting step settles its request
111            // WITHOUT a response — the lazy timeout settlement (verdict
112            // unknown) and the aborted disposition both take this path
113            // (06 §5.3; mirrors the checkpoint fold's rule).
114            RunLogPayload::StepExited { .. } => {
115                pending.retain(|entry| entry.run_path_frames != event.run_path);
116            }
117            _ => {}
118        }
119    }
120    Ok(pending)
121}
122
123/// The cross-run inbox (08 §2.6): pending entries of every run, ordered
124/// by run creation then request seq.
125pub fn human_inbox(store: &Store) -> Result<Vec<HumanInboxEntry>, StoreError> {
126    let mut entries = Vec::new();
127    for run in store.list_runs()? {
128        entries.extend(run_inbox(store, &run.run_id)?);
129    }
130    Ok(entries)
131}
132
133/// Serializes a unit-enum value to its wire literal.
134fn wire<T: Serialize>(value: &T) -> String {
135    serde_json::to_value(value)
136        .ok()
137        .and_then(|v| v.as_str().map(str::to_owned))
138        .unwrap_or_default()
139}
pointlock_store/projection/inbox.rs

pointlock_store/projection/
inbox.rs
