perspt_agent/runtime/
recorder.rs

1//! Durable event recording for the PSP-9 runtime.
2
3use super::*;
4
5/// Durable sink for all live PSP-9 events.
6pub struct Psp9Recorder {
7    pub(crate) session_id: String,
8    pub(crate) store: Arc<SessionStore>,
9    ledger: Mutex<Ledger>,
10    event_sender: Option<perspt_core::events::channel::EventSender>,
11}
12
13impl std::fmt::Debug for Psp9Recorder {
14    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15        f.debug_struct("Psp9Recorder")
16            .field("session_id", &self.session_id)
17            .field("head", &self.head())
18            .finish_non_exhaustive()
19    }
20}
21
22impl Psp9Recorder {
23    pub(crate) fn create(
24        session_id: &str,
25        task: &str,
26        working_dir: &Path,
27        database_path: Option<&Path>,
28        shared_store: Option<Arc<SessionStore>>,
29        event_sender: Option<perspt_core::events::channel::EventSender>,
30    ) -> Result<Self> {
31        // An embedder-supplied handle wins: sharing one connection with a
32        // dashboard avoids a second live handle on the same database file.
33        let store = match (shared_store, database_path) {
34            (Some(store), _) => store,
35            (None, Some(path)) => Arc::new(SessionStore::open(path).with_context(|| {
36                format!(
37                    "opening session database {}; if it is corrupted (for example a \
38                         poisoned WAL), run `perspt db repair {}`",
39                    path.display(),
40                    path.display()
41                )
42            })?),
43            (None, None) => Arc::new(SessionStore::new()?),
44        };
45        store.create_session(&SessionRecord {
46            session_id: session_id.into(),
47            task: task.into(),
48            working_dir: working_dir.display().to_string(),
49            merkle_root: None,
50            detected_toolchain: None,
51            status: "RUNNING_PSP9".into(),
52        })?;
53        store.initialize_authority_epoch(session_id, 0)?;
54        Ok(Self {
55            session_id: session_id.into(),
56            store,
57            ledger: Mutex::new(Ledger::new()),
58            event_sender,
59        })
60    }
61
62    /// Attach to an existing session for mid-loop resume: rebuild the
63    /// in-memory chain from the durable rows (verifying it in the process),
64    /// so subsequent appends extend the same ledger head.
65    pub(crate) fn attach(
66        session_id: &str,
67        database_path: Option<&Path>,
68        shared_store: Option<Arc<SessionStore>>,
69        event_sender: Option<perspt_core::events::channel::EventSender>,
70    ) -> Result<Self> {
71        let store = match (shared_store, database_path) {
72            (Some(store), _) => store,
73            (None, Some(path)) => Arc::new(SessionStore::open(path).with_context(|| {
74                format!(
75                    "opening session database {}; if it is corrupted (for example a \
76                         poisoned WAL), run `perspt db repair {}`",
77                    path.display(),
78                    path.display()
79                )
80            })?),
81            (None, None) => Arc::new(SessionStore::new()?),
82        };
83        let rows = store.get_psp9_events(session_id)?;
84        anyhow::ensure!(!rows.is_empty(), "session has no durable ledger to resume");
85        let mut ledger = Ledger::new();
86        for row in &rows {
87            let event: LedgerEvent = serde_json::from_str(&row.event_json)?;
88            let head = ledger
89                .append(event)
90                .map_err(|e| anyhow::anyhow!(e.to_string()))?;
91            anyhow::ensure!(
92                head == row.hash,
93                "ledger hash mismatch at sequence {}; refusing to resume",
94                row.sequence
95            );
96        }
97        store.update_session_status(session_id, "RUNNING_PSP9")?;
98        Ok(Self {
99            session_id: session_id.into(),
100            store,
101            ledger: Mutex::new(ledger),
102            event_sender,
103        })
104    }
105
106    pub fn record_custom(&self, kind: &str, payload: serde_json::Value) -> Result<()> {
107        self.append(LedgerEvent::Custom {
108            kind: kind.into(),
109            payload,
110        })
111    }
112
113    /// Write-ahead append: stage the record, persist it durably, and only
114    /// then extend the in-memory chain. Staging avoids the previous
115    /// clone-the-whole-ledger transaction, which made session recording
116    /// O(n²) in events.
117    pub(crate) fn append(&self, event: LedgerEvent) -> Result<()> {
118        let mut guard = self.ledger.lock().unwrap();
119        let record = guard
120            .stage(event)
121            .map_err(|e| anyhow::anyhow!(e.to_string()))?;
122        self.store.record_psp9_event(&Psp9LedgerRow {
123            session_id: self.session_id.clone(),
124            sequence: record.sequence as i64,
125            event_json: serde_json::to_string(&record.event)?,
126            prev_hash: record.prev_hash.clone(),
127            hash: record.hash.clone(),
128        })?;
129        guard
130            .commit_staged(record)
131            .map_err(|e| anyhow::anyhow!(e.to_string()))?;
132        Ok(())
133    }
134
135    pub fn head(&self) -> String {
136        self.ledger.lock().unwrap().head()
137    }
138
139    pub(crate) fn finish(&self, status: &str) -> Result<()> {
140        self.store.update_session_status(&self.session_id, status)
141    }
142
143    pub(crate) fn authority_epoch(&self) -> Result<u64> {
144        self.store.authority_epoch(&self.session_id)
145    }
146
147    pub(crate) fn record_grant_policy(
148        &self,
149        policy: &GrantPolicy,
150        signed: Option<&perspt_sdk::SignedGrantPolicy>,
151    ) -> Result<()> {
152        let durable = signed
153            .map(serde_json::to_string)
154            .transpose()?
155            .unwrap_or_else(|| serde_json::to_string(policy).expect("serializable grant policy"));
156        self.store
157            .record_grant_policy(&self.session_id, &policy.policy_id, &durable)?;
158        self.record_custom(
159            "grant_policy",
160            serde_json::json!({
161                "policy": policy,
162                "signed": signed.is_some(),
163            }),
164        )
165    }
166
167    pub(crate) fn record_external_intent(
168        &self,
169        key: &str,
170        intent: &serde_json::Value,
171    ) -> Result<()> {
172        let bytes = serde_json::to_vec(intent)?;
173        self.store.record_external_effect_intent(
174            &self.session_id,
175            key,
176            &perspt_sdk::ledger::content_hash(&bytes),
177            &String::from_utf8(bytes)?,
178        )?;
179        self.record_custom("external_effect_intent", intent.clone())
180    }
181
182    pub(crate) fn complete_external_effect(
183        &self,
184        key: &str,
185        result: &serde_json::Value,
186    ) -> Result<()> {
187        self.store.complete_external_effect(
188            &self.session_id,
189            key,
190            &serde_json::to_string(result)?,
191        )?;
192        self.record_custom("external_effect_completed", result.clone())
193    }
194}
195
196impl LoopRecorder for Psp9Recorder {
197    fn external_intent(&self, key: &str, intent: &serde_json::Value) -> Result<()> {
198        self.record_external_intent(key, intent)
199    }
200
201    fn external_result(&self, key: &str, result: &serde_json::Value) -> Result<()> {
202        self.complete_external_effect(key, result)
203    }
204
205    fn record(&self, event: &LoopEvent) -> Result<()> {
206        // PSP-10 Gate AD: every post-cutover runtime row is written once,
207        // wrapped in the versioned envelope.
208        let envelope = crate::toolloop::LoopEventEnvelopeV1::new(event.clone());
209        self.record_custom("tool_loop", serde_json::to_value(&envelope)?)?;
210        if let LoopEvent::ContextCheckpointCreated { checkpoint } = event {
211            self.store.record_psp9_checkpoint(
212                &self.session_id,
213                &checkpoint.covered_event_root,
214                &serde_json::to_string(checkpoint)?,
215            )?;
216        }
217        if let LoopEvent::DurableCandidateCheckpoint {
218            candidate_id,
219            state_root,
220            control,
221            conversation,
222            birth_deps,
223            canonical_scope,
224            files,
225        } = event
226        {
227            // The durable resume point: `perspt resume` rebuilds the
228            // accepted candidate from these rows.
229            self.store.record_psp9_checkpoint(
230                &self.session_id,
231                state_root,
232                &serde_json::to_string(&serde_json::json!({
233                    "kind": "candidate",
234                    "candidate_id": candidate_id,
235                    "state_root": state_root,
236                    "control": control,
237                    "conversation": conversation,
238                    "birth_deps": birth_deps,
239                    "canonical_scope": canonical_scope,
240                    "files": files,
241                }))?,
242            )?;
243        }
244        if let Some(sender) = &self.event_sender {
245            // Structured surfaces first: the TUI's energy panel consumes the
246            // typed event, not the narration line.
247            if let LoopEvent::CandidateMeasured {
248                node_id, energy, ..
249            } = event
250            {
251                let _ = sender.send(perspt_core::AgentEvent::EnergyUpdated {
252                    node_id: node_id.clone(),
253                    energy: *energy as f32,
254                });
255            }
256            let message = narration(event);
257            if let Some(message) = message {
258                let _ = sender.send(perspt_core::AgentEvent::Log(message));
259            }
260        }
261        Ok(())
262    }
263
264    fn record_artifact(&self, content: &[u8], media_type: &str) -> Result<String> {
265        let handle = perspt_sdk::ledger::content_hash(content);
266        self.store.put_psp9_artifact(&handle, content, media_type)?;
267        Ok(handle)
268    }
269
270    fn fetch_artifact(&self, handle: &str) -> Result<Option<Vec<u8>>> {
271        self.store.get_psp9_artifact(handle)
272    }
273}
274
275/// The one-line human narration for a loop event, if any.
276fn narration(event: &LoopEvent) -> Option<String> {
277    match event {
278        LoopEvent::CandidateMeasured {
279            node_id,
280            generation,
281            energy,
282            hard_pass,
283            residuals,
284            ..
285        } => Some(format!(
286            "Measured {node_id} generation {generation}: V={energy:.3}, \
287                 hard_pass={hard_pass}, residuals={}",
288            residuals.len()
289        )),
290        LoopEvent::GateDecisionRecorded {
291            node_id,
292            generation,
293            decision,
294            ..
295        } => Some(format!(
296            "Gate {node_id} generation {generation}: {decision:?}"
297        )),
298        LoopEvent::EffectDenied {
299            call_id, reason, ..
300        } => Some(format!("Effect {call_id} denied: {reason}")),
301        LoopEvent::EffectApplied {
302            call_id, mutated, ..
303        } => Some(format!(
304            "Effect {call_id} applied to candidate (mutated={mutated})"
305        )),
306        LoopEvent::RouteFailover {
307            from_model,
308            to_model,
309            cause,
310        } => Some(format!(
311            "Route failover {from_model} -> {to_model}: {cause}"
312        )),
313        LoopEvent::ContextCheckpointCreated { checkpoint } => Some(format!(
314            "Context checkpoint {} covers events {}..{}",
315            &checkpoint.covered_event_root[..12.min(checkpoint.covered_event_root.len())],
316            checkpoint.covered_from,
317            checkpoint.covered_to
318        )),
319        LoopEvent::RecoveryContained { reason, .. } => {
320            Some(format!("Recovery contained the node: {reason}"))
321        }
322        _ => None,
323    }
324}
325
326impl super::Psp9AgentRuntime {
327    pub(super) fn record_route_capabilities(&self, recorder: &Psp9Recorder) -> Result<()> {
328        for (position, model) in std::iter::once(&self.model)
329            .chain(self.fallback_models.iter())
330            .enumerate()
331        {
332            let capabilities = self.transport.capabilities(model);
333            let mut degradations = Vec::new();
334            if !capabilities.strict_schema {
335                degradations.push("strict_schema:local_validation");
336            }
337            if !capabilities.parallel_tool_calls {
338                degradations.push("parallel_tool_calls:sequential_execution");
339            }
340            if !capabilities.streaming_tool_calls {
341                degradations.push("streaming_tool_calls:turn_granular_progress");
342            }
343            if !capabilities.prompt_caching {
344                degradations.push("prompt_caching:cache_cold_accounting");
345            }
346            if !capabilities.structured_output {
347                degradations.push("structured_output:local_parse_and_validation");
348            }
349            recorder.record_custom(
350                "provider_capability_evidence",
351                serde_json::json!({
352                    "model": model,
353                    "route_position": position,
354                    "source": "declared",
355                    "capabilities": capabilities,
356                    "degradations": degradations,
357                }),
358            )?;
359        }
360        Ok(())
361    }
362}
363
364impl Psp9Recorder {
365    /// Ledger one compiled prompt program exactly (PSP-10 Gate Z): route,
366    /// dialect, ordered section provenance, budget-dropped sections, the
367    /// tool-surface hash, and the program digest.
368    pub(crate) fn record_prompt_program(
369        &self,
370        program: &perspt_sdk::prompt::CompiledPromptProgram,
371    ) -> Result<()> {
372        let sections: Vec<&perspt_sdk::prompt::SectionProvenance> = program
373            .messages
374            .iter()
375            .flat_map(|message| message.sections.iter())
376            .collect();
377        self.record_custom(
378            "prompt_program_compiled",
379            serde_json::json!({
380                "stage": program.stage.dir_name(),
381                "route": program.route,
382                "dialect": program.dialect,
383                "sections": sections,
384                "dropped_sections": program.dropped_sections,
385                "tool_spec_hash": program.tool_spec_hash,
386                "digest": program.program_digest,
387            }),
388        )
389    }
390
391    /// Ledger one model call's exact prompt binding: both program digests
392    /// and the invocation digest (Gate Z, one record per call).
393    pub(crate) fn record_prompt_invocation(
394        &self,
395        actor: &str,
396        turn: u32,
397        invocation: &perspt_sdk::prompt::CompiledPromptInvocation,
398    ) -> Result<()> {
399        self.record_custom(
400            "prompt_program_invoked",
401            serde_json::json!({
402                "actor": actor,
403                "turn": turn,
404                "platform_digest": invocation.platform.program_digest,
405                "domain_digest": invocation.domain.program_digest,
406                "invocation_digest": invocation.invocation_digest,
407                "tool_spec_hash": invocation.platform.tool_spec_hash,
408            }),
409        )
410    }
411}
perspt_agent/runtime/recorder.rs

perspt_agent/runtime/
recorder.rs
