perspt_agent/runtime/
mod.rs

1//! Authoritative PSP-9 agent runtime.
2
3use std::path::{Path, PathBuf};
4use std::str::FromStr;
5use std::sync::{Arc, Mutex};
6
7use anyhow::{Context, Result};
8use perspt_coding::{CodingDomain, OperationalSafetyBarrier};
9use perspt_sdk::capability::CommandPattern;
10use perspt_sdk::capability::PathPattern;
11use perspt_sdk::{
12    ActorId, AgentDomainPackage, ApprovalPolicy, CandidateTransition, Capability, CapabilityRole,
13    ContractEvaluator, ContractWitness, Conversation, EffectKind, Footprint, GrantPolicy, Ledger,
14    LedgerEvent, ModelId, ModelTransport, NodeClass, NodeTerminalOutcome, Resource, RiskBudget,
15    Scheduler, StaticCatalog, ToolCatalog, ToolChoicePolicy, TurnOutput, WorkGraphRevision,
16    WorkNode, WorkNodeState,
17};
18use perspt_store::{Psp9LedgerRow, Psp9VerdictRow, SessionRecord, SessionStore};
19
20use crate::candidate::CandidateWorkspace;
21use crate::toolloop::{EffectExecutor, LoopBudgets, LoopEvent, LoopRecorder};
22use crate::transport::GenAiTransport;
23
24mod adjudicate;
25mod attempt;
26mod dispatch;
27mod explore;
28mod external;
29mod integrate;
30mod node;
31mod plan;
32mod prompt_bind;
33mod recorder;
34mod resume;
35mod search;
36mod settings;
37
38use external::{external_runtime_from, registry_with_external};
39use node::*;
40pub use recorder::Psp9Recorder;
41pub use settings::{Psp9ModelRoutes, Psp9RunConfig, Psp9RunSummary};
42
43/// The production runtime for one coding task.
44pub struct Psp9AgentRuntime {
45    working_dir: PathBuf,
46    transport: Arc<dyn ModelTransport>,
47    model: ModelId,
48    fallback_models: Vec<ModelId>,
49    explorer_model: Option<ModelId>,
50    adjudicator_model: Option<ModelId>,
51    /// Higher-capability route for the recovery ladder's level-3 handoff.
52    handoff_model: Option<ModelId>,
53    config: Psp9RunConfig,
54    event_sender: Option<perspt_core::events::channel::EventSender>,
55    action_receiver: tokio::sync::Mutex<Option<perspt_core::events::channel::ActionReceiver>>,
56    database_path: Option<PathBuf>,
57    shared_store: Option<Arc<SessionStore>>,
58    /// The open execution plane shared by every candidate this runtime
59    /// creates. The composition root may extend it with registered families.
60    tool_handlers: Arc<crate::tools::handlers::CandidateHandlerRegistry>,
61    /// Catalog entries for registered tool families, appended to the
62    /// domain's entries at node assembly.
63    extra_tool_entries: Vec<perspt_sdk::ToolEntry>,
64    /// The selected domain package. Defaults to coding; the composition
65    /// root selects from a `DomainRegistry` (explicit `--domain` or
66    /// detection).
67    domain: Arc<dyn AgentDomainPackage>,
68    /// Optional MCP edge adapter (system 13). Servers come from
69    /// `[[external_tools]]`; nothing changes when none are configured.
70    external: Option<Arc<tokio::sync::Mutex<crate::external_tools::ExternalToolRuntime>>>,
71    /// Admitted external entries, discovered once per session.
72    external_entries: Mutex<Option<Vec<perspt_sdk::ToolEntry>>>,
73    /// Bounded-search settings from `[exploration]` (PSP-10 system 20).
74    search: search::SearchSettings,
75    /// Prior search state folded from the ledger at resume. The folded
76    /// no-goods seed every forest in the session; the interrupted
77    /// forest's consumption is claimed once, by its exact owner.
78    search_seed: Mutex<Option<search::SearchSeed>>,
79    /// Experimental platform-section overrides from validated
80    /// `[prompts].bundles`, loaded by the composition root only under
81    /// `--allow-experimental-prompts` (PSP-10 system 25, Gate AE).
82    prompt_overrides: perspt_core::prompts::SectionOverrides,
83}
84
85impl std::fmt::Debug for Psp9AgentRuntime {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("Psp9AgentRuntime")
88            .field("working_dir", &self.working_dir)
89            .field("model", &self.model)
90            .field("config", &self.config)
91            .finish_non_exhaustive()
92    }
93}
94
95impl Psp9AgentRuntime {
96    pub fn from_config(
97        working_dir: PathBuf,
98        config: &perspt_core::Config,
99        routes: Psp9ModelRoutes,
100        run_config: Psp9RunConfig,
101    ) -> Result<Self> {
102        let portfolio = Arc::new(perspt_core::ModelPortfolio::from_config(config)?);
103        let route = routes
104            .primary
105            .as_deref()
106            .or(routes.actuator.as_deref())
107            .or_else(|| config.models.as_ref().and_then(|m| m.actuator.as_deref()))
108            .or(config.actuator_model.as_deref())
109            .or(config.model.as_deref())
110            .unwrap_or_else(|| perspt_core::ModelTier::default_model_name());
111        let model = qualify_model(route, config, &portfolio)?;
112        // Resolve eagerly so a typo fails before a session is created.
113        portfolio.resolve(&model.provider)?;
114        let transport = Arc::new(GenAiTransport::new(portfolio));
115        let primary_capabilities = transport.capabilities(&model);
116        anyhow::ensure!(
117            primary_capabilities.tool_calling,
118            "route {model} does not declare native tool calling; PSP-9 \
119             tool-loop mode never emulates tool calls in text"
120        );
121        let (explorer_model, adjudicator_model, handoff_model) =
122            resolve_role_routes(&routes, config, &transport)?;
123        let fallback_models = resolve_fallbacks(&routes.fallbacks, &model, config, &transport)?;
124        let sampling = Arc::new(crate::external_tools::ModelTransportSamplingProvider::new(
125            transport.clone(),
126            model.clone(),
127        ));
128        let external = external_runtime_from(config, Some(sampling))?;
129        let handlers = registry_with_external(&external);
130        let run_config = settings::apply_config_overrides(run_config, config);
131        Ok(Self {
132            working_dir,
133            transport,
134            model,
135            fallback_models,
136            explorer_model,
137            adjudicator_model,
138            handoff_model,
139            config: run_config,
140            event_sender: None,
141            action_receiver: tokio::sync::Mutex::new(None),
142            database_path: None,
143            shared_store: None,
144            tool_handlers: Arc::new(handlers),
145            extra_tool_entries: Vec::new(),
146            domain: Arc::new(CodingDomain::new()),
147            external,
148            external_entries: Mutex::new(None),
149            search: search::SearchSettings::from_config(config.exploration.as_ref()),
150            search_seed: Mutex::new(None),
151            prompt_overrides: Default::default(),
152        })
153    }
154
155    /// Assemble the same runtime with an injected provider-neutral transport.
156    /// Used by conformance tests and embedders that supply their own model plane.
157    pub fn with_transport(
158        working_dir: PathBuf,
159        transport: Arc<dyn ModelTransport>,
160        model: ModelId,
161        run_config: Psp9RunConfig,
162    ) -> Self {
163        Self {
164            working_dir,
165            transport,
166            model,
167            fallback_models: Vec::new(),
168            explorer_model: None,
169            adjudicator_model: None,
170            handoff_model: None,
171            config: run_config,
172            event_sender: None,
173            action_receiver: tokio::sync::Mutex::new(None),
174            database_path: None,
175            shared_store: None,
176            tool_handlers: Arc::new(
177                crate::tools::handlers::CandidateHandlerRegistry::with_builtins(),
178            ),
179            extra_tool_entries: Vec::new(),
180            domain: Arc::new(CodingDomain::new()),
181            external: None,
182            external_entries: Mutex::new(None),
183            search: search::SearchSettings::default(),
184            search_seed: Mutex::new(None),
185            prompt_overrides: Default::default(),
186        }
187    }
188
189    /// Extend or replace the execution plane with a registry assembled at
190    /// the composition root (registered tool families beyond the builtins).
191    /// A configured MCP dispatcher is preserved as the fallback.
192    pub fn with_tool_handlers(
193        mut self,
194        mut handlers: crate::tools::handlers::CandidateHandlerRegistry,
195    ) -> Self {
196        if let (Some(external), false) = (&self.external, handlers.has_fallback()) {
197            handlers.set_fallback(Arc::new(
198                crate::tools::handlers::external::ExternalDispatcher::new(external.clone()),
199            ));
200        }
201        self.tool_handlers = Arc::new(handlers);
202        self
203    }
204
205    /// Add a registered tool family's catalog entries. Pair with
206    /// [`Self::with_tool_handlers`]; the entries enter every node's
207    /// assembled catalog and therefore its derived grant.
208    pub fn with_tool_family(mut self, entries: Vec<perspt_sdk::ToolEntry>) -> Self {
209        self.extra_tool_entries.extend(entries);
210        self
211    }
212
213    /// Effects the user explicitly opted into beyond the default grant.
214    fn opted_in_effects(&self) -> Vec<EffectKind> {
215        if self.config.allow_dependency_mutation {
216            vec![EffectKind::MutateDependencies]
217        } else {
218            Vec::new()
219        }
220    }
221
222    /// Select the domain package (from a `DomainRegistry` at the
223    /// composition root). Defaults to the coding domain.
224    pub fn with_domain(mut self, domain: Arc<dyn AgentDomainPackage>) -> Self {
225        self.domain = domain;
226        self
227    }
228
229    pub fn with_database_path(mut self, path: PathBuf) -> Self {
230        self.database_path = Some(path);
231        self
232    }
233
234    /// Record into an existing store handle instead of opening a second live
235    /// handle on the same database file (used when an in-process dashboard
236    /// already holds one).
237    pub fn with_session_store(mut self, store: Arc<SessionStore>) -> Self {
238        self.database_path = Some(store.database_path().to_path_buf());
239        self.shared_store = Some(store);
240        self
241    }
242
243    /// Install experimental platform-section overrides (validated bundle
244    /// replacements). The composition root gates this behind
245    /// `--allow-experimental-prompts`; active overrides are ledgered at
246    /// session start.
247    pub fn with_prompt_overrides(
248        mut self,
249        overrides: perspt_core::prompts::SectionOverrides,
250    ) -> Self {
251        self.prompt_overrides = overrides;
252        self
253    }
254
255    /// Override the forest's Definition 5 limit vector (tests, embedders).
256    pub fn with_search_limits(mut self, limits: perspt_sdk::SearchLimits) -> Self {
257        self.search.limits = limits;
258        self
259    }
260
261    pub fn with_fallback_models(mut self, models: Vec<ModelId>) -> Self {
262        self.fallback_models = models;
263        self
264    }
265
266    pub fn with_explorer_model(mut self, model: ModelId) -> Self {
267        self.explorer_model = Some(model);
268        self
269    }
270
271    pub fn with_adjudicator_model(mut self, model: ModelId) -> Self {
272        self.adjudicator_model = Some(model);
273        self
274    }
275
276    pub fn connect_tui(
277        mut self,
278        sender: perspt_core::events::channel::EventSender,
279        receiver: perspt_core::events::channel::ActionReceiver,
280    ) -> Self {
281        self.event_sender = Some(sender);
282        self.action_receiver = tokio::sync::Mutex::new(Some(receiver));
283        self
284    }
285
286    /// Open the durable session, run read-only exploration, and dispatch the
287    /// initial work graph. Returns the recorder, the compiled goal, and the
288    /// running graph revision.
289    async fn open_session(
290        &mut self,
291        task: &str,
292        session_id: &str,
293        node_id: &str,
294    ) -> Result<(Psp9Recorder, String, Scheduler, WorkGraphRevision)> {
295        let recorder = Psp9Recorder::create(
296            session_id,
297            task,
298            &self.working_dir,
299            self.database_path.as_deref(),
300            self.shared_store.clone(),
301            self.event_sender.clone(),
302        )?;
303        recorder.record_custom(
304            "session_started",
305            serde_json::json!({
306                "task": task,
307                "model": self.model,
308                "fallback_models": self.fallback_models,
309                "mode": "tool_loop",
310                "test_policy": self.config.test_policy.as_str(),
311            }),
312        )?;
313        self.record_route_capabilities(&recorder)?;
314        self.record_prompt_overrides(&recorder)?;
315
316        let agent_goal = self.explore(&recorder, task).await?;
317
318        self.emit(perspt_core::AgentEvent::Log(format!(
319            "PSP-9 session {} using {}",
320            &session_id[..8],
321            self.model
322        )));
323        self.emit(perspt_core::AgentEvent::TaskStatusChanged {
324            node_id: node_id.to_string(),
325            status: perspt_core::NodeStatus::Coding,
326        });
327
328        let graph = self.plan_initial_graph(&recorder, node_id, task).await?;
329        recorder.record_custom("graph_revision", serde_json::to_value(&graph)?)?;
330        if self.config.max_parallel_nodes > 1 {
331            // The dispatcher owns scheduling; this scheduler is unused.
332            return Ok((recorder, agent_goal, Scheduler::new(1), graph));
333        }
334        let mut scheduler = Scheduler::new(1);
335        let ready = scheduler.ready_nodes(&graph, node_footprint);
336        let selected = ready
337            .first()
338            .context("initial work graph has no schedulable node")?;
339        scheduler.start(selected, node_footprint(selected));
340        recorder.record_custom(
341            "scheduler_dispatch",
342            serde_json::json!({
343                "node_id": selected.node_id,
344                "generation": selected.generation,
345                "parallel_slot": 0,
346            }),
347        )?;
348        let running_graph = execution_revision(&graph, node_id, WorkNodeState::Running)?;
349        recorder.record_custom("graph_revision", serde_json::to_value(&running_graph)?)?;
350        Ok((recorder, agent_goal, scheduler, running_graph))
351    }
352
353    /// Gate AE: experimental section overrides are ledgered at session
354    /// start, never silent.
355    fn record_prompt_overrides(&self, recorder: &Psp9Recorder) -> Result<()> {
356        if self.prompt_overrides.is_empty() {
357            return Ok(());
358        }
359        recorder.record_custom(
360            "prompt_overrides_active",
361            serde_json::json!({
362                "experimental": true,
363                "sections": self
364                    .prompt_overrides
365                    .provenance()
366                    .iter()
367                    .map(|(id, hash)| serde_json::json!({"id": id, "content_hash": hash}))
368                    .collect::<Vec<_>>(),
369            }),
370        )
371    }
372
373    /// Mint (and for persistent grants, sign and verify) the session grant.
374    fn mint_grant(
375        &self,
376        recorder: &Psp9Recorder,
377        revision_id: &str,
378        catalog: &dyn ToolCatalog,
379    ) -> Result<perspt_sdk::GrantPolicy> {
380        let grant_policy = session_grant_policy(
381            &self.working_dir,
382            revision_id,
383            self.config.persistent_grants,
384            catalog,
385            &self.opted_in_effects(),
386        )?;
387        let signed_grant = if self.config.persistent_grants {
388            let key = crate::grant::GrantSigningKey::resolve()?;
389            let signed = perspt_sdk::SignedGrantPolicy::sign(grant_policy.clone(), &key.bytes)?;
390            signed.verify_against(&key.public_key())?;
391            recorder.record_custom(
392                "grant_signing_key_resolved",
393                serde_json::json!({"source": format!("{:?}", key.source)}),
394            )?;
395            Some(signed)
396        } else {
397            None
398        };
399        recorder.record_grant_policy(&grant_policy, signed_grant.as_ref())?;
400        Ok(grant_policy)
401    }
402
403    fn open_candidate(
404        &self,
405        node_id: &str,
406        generation: u32,
407        revision_id: &str,
408    ) -> Result<CandidateWorkspace> {
409        let exclusions = self.candidate_exclusions();
410        let mut candidate = CandidateWorkspace::create_filtered(
411            &self.working_dir,
412            node_id,
413            generation,
414            revision_id,
415            self.config.allow_unisolated_verifiers,
416            &exclusions,
417        )?;
418        candidate.set_tool_handlers(self.tool_handlers.clone());
419        candidate.set_verifier_timeouts(self.config.verifier_timeouts);
420        Ok(candidate)
421    }
422
423    /// The live ledger is runtime state, not a coding artifact. Keeping it out
424    /// of candidates also avoids copying an exclusively locked DuckDB file on
425    /// Windows when `--db-path` names a file inside the workspace.
426    fn candidate_exclusions(&self) -> Vec<PathBuf> {
427        let Some(configured) = &self.database_path else {
428            return Vec::new();
429        };
430        let absolute = if configured.is_absolute() {
431            configured.clone()
432        } else {
433            std::env::current_dir()
434                .unwrap_or_else(|_| self.working_dir.clone())
435                .join(configured)
436        };
437        let target = absolute.canonicalize().unwrap_or(absolute);
438        let Some((parent, name)) = target.parent().zip(target.file_name()) else {
439            return vec![target];
440        };
441        let prefix = format!("{}.", name.to_string_lossy());
442        let mut excluded = vec![target.clone()];
443        if let Ok(entries) = std::fs::read_dir(parent) {
444            excluded.extend(
445                entries
446                    .filter_map(|entry| entry.ok())
447                    .map(|entry| entry.path())
448                    .filter(|path| {
449                        path.file_name()
450                            .is_some_and(|file| file.to_string_lossy().starts_with(&prefix))
451                    }),
452            );
453        }
454        excluded
455    }
456
457    /// Recovery ladder (Theorem 6): the loop already consumed its retry and
458    /// fallback levels; the runtime holds the higher rungs. Level 2 refines
459    /// the work graph with the attempt's evidence and re-runs at a new
460    /// generation; level 3 hands the node to the configured higher-capability
461    /// route; level 4 revokes the session's authority epoch after
462    /// restore-best containment.
463    async fn recovery_ladder(
464        &self,
465        session: LadderSession<'_>,
466        initial_budget: u32,
467        mut graph: WorkGraphRevision,
468        mut goal: String,
469        mut attempt: NodeAttempt,
470    ) -> Result<(NodeAttempt, WorkGraphRevision, u32)> {
471        let LadderSession {
472            recorder,
473            session_id,
474            node_id,
475            scheduler,
476        } = session;
477        // The ladder continues the node's recorded generation (a node
478        // dispatched at generation >= 1 after a graph resume must refine to
479        // generation + 1, or `WorkGraphRevision::revise` rejects it stale).
480        let mut generation = graph.node(node_id).map_or(0, |node| node.generation);
481        let mut model = self.model.clone();
482        let mut remaining_budget = initial_budget.saturating_sub(spent_of(&attempt));
483        for level in [
484            perspt_sdk::CascadeLevel::Refine,
485            perspt_sdk::CascadeLevel::Escalate,
486        ] {
487            if !matches!(
488                attempt.outcome.outcome,
489                NodeTerminalOutcome::Escalated { .. }
490            ) {
491                break;
492            }
493            match level {
494                perspt_sdk::CascadeLevel::Refine => {
495                    let refined = refine_rung(
496                        recorder, scheduler, &graph, node_id, generation, &goal, &attempt,
497                    )?;
498                    let Some((revised, refined_goal)) = refined else {
499                        continue;
500                    };
501                    graph = revised;
502                    generation += 1;
503                    goal = refined_goal;
504                }
505                perspt_sdk::CascadeLevel::Escalate => {
506                    let Some(handoff) = self.escalation_handoff(recorder, &model)? else {
507                        continue;
508                    };
509                    model = handoff;
510                }
511                _ => unreachable!("ladder iterates refine and escalate only"),
512            }
513            let next = self
514                .rung_forest(
515                    recorder,
516                    session_id,
517                    &goal,
518                    node_id,
519                    generation,
520                    &model,
521                    &graph,
522                    &attempt,
523                    remaining_budget,
524                )
525                .await?;
526            let spent = spent_of(&next);
527            attempt = next;
528            remaining_budget = remaining_budget.saturating_sub(spent);
529        }
530        contain_if_escalated(recorder, session_id, &attempt)?;
531        Ok((attempt, graph, generation))
532    }
533
534    /// One rung's re-attempt: restore-best, then open a bounded search
535    /// forest (PSP-10 system 20) — isolated branches against the same
536    /// accepted root, one candidate committed through the ordinary gate
537    /// by the deterministic rule.
538    #[allow(clippy::too_many_arguments)]
539    async fn rung_forest(
540        &self,
541        recorder: &Psp9Recorder,
542        session_id: &str,
543        goal: &str,
544        node_id: &str,
545        generation: u32,
546        model: &ModelId,
547        graph: &WorkGraphRevision,
548        attempt: &NodeAttempt,
549        remaining_budget: u32,
550    ) -> Result<NodeAttempt> {
551        let seed = seed_from_attempt(recorder, node_id, attempt).await?;
552        let baseline_energy = attempt.outcome.trajectory.best_accepted_energy;
553        let rho_gate = attempt.assembly.energy.rho_gate;
554        self.run_search_forest(
555            recorder,
556            session_id,
557            goal,
558            node_id,
559            generation,
560            model,
561            graph,
562            seed.as_ref(),
563            remaining_budget,
564            baseline_energy,
565            rho_gate,
566        )
567        .await
568    }
569
570    /// Multi-node session: dispatch the planned graph, then close the
571    /// session with the aggregate outcome.
572    async fn run_graph_session(
573        &self,
574        recorder: &Psp9Recorder,
575        session_id: &str,
576        graph: WorkGraphRevision,
577    ) -> Result<Psp9RunSummary> {
578        let dispatched = match self.run_dispatched(recorder, session_id, graph).await {
579            Ok(dispatched) => dispatched,
580            Err(error) => {
581                recorder.finish("FAILED_PSP9").ok();
582                return Err(error);
583            }
584        };
585        recorder.finish(dispatched.status)?;
586        self.emit(perspt_core::AgentEvent::Complete {
587            success: matches!(dispatched.outcome, NodeTerminalOutcome::HardPass),
588            message: format!("PSP-9 outcome: {:?}", dispatched.outcome),
589        });
590        Ok(Psp9RunSummary {
591            session_id: session_id.to_string(),
592            node_id: "graph".into(),
593            outcome: dispatched.outcome,
594            turns_used: dispatched.turns_used,
595            ledger_head: recorder.head(),
596            promoted_paths: dispatched.promoted_paths,
597        })
598    }
599
600    /// Assemble the conclusion context for a finished attempt and decide the
601    /// node's terminal fate.
602    async fn conclude_attempt(
603        &self,
604        recorder: &Psp9Recorder,
605        attempt: &NodeAttempt,
606        node_id: &str,
607        task: &str,
608    ) -> Result<(NodeTerminalOutcome, &'static str, Vec<String>)> {
609        self.conclude_run(ConcludeContext {
610            recorder,
611            candidate: &attempt.candidate,
612            node_id,
613            task,
614            grant_policy: &attempt.assembly.grant_policy,
615            capability: &attempt.assembly.capability,
616            contract: &attempt.assembly.contract,
617            barrier: &attempt.assembly.barrier,
618            kernel_state: &attempt.kernel_state,
619            loop_outcome: &attempt.outcome.outcome,
620            calibration: &attempt.assembly.calibration,
621        })
622        .await
623    }
624
625    /// Adjudicate and approve one attempt without promoting it — the
626    /// staging path's validation half (PSP-10 system 22).
627    async fn validate_and_approve_attempt(
628        &self,
629        recorder: &Psp9Recorder,
630        attempt: &NodeAttempt,
631        node_id: &str,
632        task: &str,
633    ) -> Result<bool> {
634        let ctx = ConcludeContext {
635            recorder,
636            candidate: &attempt.candidate,
637            node_id,
638            task,
639            grant_policy: &attempt.assembly.grant_policy,
640            capability: &attempt.assembly.capability,
641            contract: &attempt.assembly.contract,
642            barrier: &attempt.assembly.barrier,
643            kernel_state: &attempt.kernel_state,
644            loop_outcome: &attempt.outcome.outcome,
645            calibration: &attempt.assembly.calibration,
646        };
647        self.validate_and_approve(&ctx, true).await
648    }
649
650    /// Level-3 handoff: returns the higher-capability route when one is
651    /// configured and distinct from the current model.
652    fn escalation_handoff(
653        &self,
654        recorder: &Psp9Recorder,
655        current: &perspt_sdk::ModelId,
656    ) -> Result<Option<perspt_sdk::ModelId>> {
657        let Some(handoff) = self.handoff_model.clone() else {
658            return Ok(None);
659        };
660        if handoff == *current {
661            return Ok(None);
662        }
663        recorder.record_custom(
664            "recovery_handoff",
665            serde_json::json!({
666                "level": "escalate",
667                "from_model": current,
668                "to_model": handoff,
669            }),
670        )?;
671        Ok(Some(handoff))
672    }
673
674    pub async fn run(mut self, task: String) -> Result<Psp9RunSummary> {
675        let session_id = uuid::Uuid::new_v4().to_string();
676        let node_id = "implement-1".to_string();
677        let (recorder, agent_goal, mut scheduler, running_graph) =
678            self.open_session(&task, &session_id, &node_id).await?;
679
680        if self.config.max_parallel_nodes > 1 {
681            return self
682                .run_graph_session(&recorder, &session_id, running_graph)
683                .await;
684        }
685
686        let first = match self
687            .attempt_node(
688                &recorder,
689                &session_id,
690                &agent_goal,
691                &node_id,
692                0,
693                &self.model.clone(),
694                &running_graph,
695                None,
696                self.config.rejection_budget,
697            )
698            .await
699        {
700            Ok(attempt) => attempt,
701            Err(error) => return Err(self.fail_session(&recorder, error)),
702        };
703        let (attempt, graph, final_generation) = match self
704            .recovery_ladder(
705                LadderSession {
706                    recorder: &recorder,
707                    session_id: &session_id,
708                    node_id: &node_id,
709                    scheduler: &mut scheduler,
710                },
711                self.config.rejection_budget,
712                running_graph,
713                agent_goal,
714                first,
715            )
716            .await
717        {
718            Ok(result) => result,
719            Err(error) => return Err(self.fail_session(&recorder, error)),
720        };
721
722        let verdict = match self
723            .conclude_attempt(&recorder, &attempt, &node_id, &task)
724            .await
725        {
726            Ok(verdict) => verdict,
727            Err(error) => return Err(self.fail_session(&recorder, error)),
728        };
729        let (final_outcome, status, promoted_paths) = verdict;
730        scheduler.finish(&node_id, final_generation);
731        if let Err(error) = self.finish_node(&recorder, &graph, &node_id, &final_outcome, status) {
732            return Err(self.fail_session(&recorder, error));
733        }
734
735        Ok(Psp9RunSummary {
736            session_id,
737            node_id,
738            outcome: final_outcome,
739            turns_used: attempt.outcome.turns_used,
740            ledger_head: recorder.head(),
741            promoted_paths,
742        })
743    }
744
745    fn emit(&self, event: perspt_core::AgentEvent) {
746        if let Some(sender) = &self.event_sender {
747            let _ = sender.send(event);
748        }
749    }
750
751    /// Record a loop failure durably and close the session as failed.
752    fn fail_session(&self, recorder: &Psp9Recorder, error: anyhow::Error) -> anyhow::Error {
753        let recording = recorder
754            .record_custom(
755                "node_terminal",
756                serde_json::json!({"class": "failed", "error": error.to_string()}),
757            )
758            .and_then(|()| recorder.finish("FAILED_PSP9"));
759        if let Err(recording_error) = recording {
760            log::warn!("failed to record session failure: {recording_error}");
761        }
762        self.emit(perspt_core::AgentEvent::Error(error.to_string()));
763        error
764    }
765
766    fn loop_budgets(&self, domain_rho_gate: f64, shared_recovery_budget: u32) -> LoopBudgets {
767        LoopBudgets {
768            max_turns: self.config.max_turns,
769            max_calls_per_turn: self.config.max_calls_per_turn,
770            rejection_budget: shared_recovery_budget,
771            rho_gate: self.config.rho_gate.max(domain_rho_gate),
772            declared_energy_floor: None,
773            context_soft_limit_chars: 240_000,
774            recovery_budget: shared_recovery_budget,
775            turn_deadline_secs: self.config.turn_deadline_secs,
776            resident: crate::toolloop::ResidentReserves {
777                paging_enabled: self.config.resident.paging_enabled
778                    && !self.config.ablate_context_paging,
779                ..self.config.resident
780            },
781        }
782    }
783
784    /// Record the terminal graph revision and outcome, close the session.
785    fn finish_node(
786        &self,
787        recorder: &Psp9Recorder,
788        running_graph: &WorkGraphRevision,
789        node_id: &str,
790        final_outcome: &NodeTerminalOutcome,
791        status: &str,
792    ) -> Result<()> {
793        let terminal_state = if matches!(final_outcome, NodeTerminalOutcome::HardPass) {
794            WorkNodeState::Stable
795        } else {
796            WorkNodeState::Stopped {
797                certificate_id: match final_outcome {
798                    NodeTerminalOutcome::Escalated { certificate_id } => certificate_id.clone(),
799                    _ => uuid::Uuid::new_v4().to_string(),
800                },
801            }
802        };
803        let terminal_graph = execution_revision(running_graph, node_id, terminal_state)?;
804        recorder.record_custom("graph_revision", serde_json::to_value(&terminal_graph)?)?;
805        recorder.record_custom(
806            "node_terminal",
807            serde_json::json!({"node_id": node_id, "outcome": final_outcome}),
808        )?;
809        recorder.finish(status)?;
810        self.emit(perspt_core::AgentEvent::Complete {
811            success: matches!(final_outcome, NodeTerminalOutcome::HardPass),
812            message: format!("PSP-9 outcome: {final_outcome:?}"),
813        });
814        Ok(())
815    }
816
817    /// Build the node's governed surfaces: catalog, grant, capability,
818    /// contract, barrier, cadence, and energy model.
819    #[allow(clippy::too_many_arguments)]
820    fn assemble_node(
821        &self,
822        recorder: &Psp9Recorder,
823        session_id: &str,
824        running_graph: &WorkGraphRevision,
825        node_id: &str,
826        generation: u32,
827        external_entries: &[perspt_sdk::ToolEntry],
828    ) -> Result<NodeAssembly> {
829        let domain = self.domain.clone();
830        let scope = perspt_sdk::DomainScope {
831            label: node_id.to_string(),
832            paths: Vec::new(),
833        };
834        let catalog = self.assemble_catalog(node_id, external_entries)?;
835        let grant_policy = self.mint_grant(recorder, &running_graph.revision_id, &catalog)?;
836        // Every live capability is the intersection of the worker template
837        // with the grant ceilings — the ceilings are enforced, not decorative.
838        let mut capability = grant_policy
839            .mint(worker_capability(
840                session_id,
841                &running_graph.revision_id,
842                grant_policy.authority_epoch,
843                generation,
844                &catalog,
845                &self.opted_in_effects(),
846            ))
847            .map_err(|e| anyhow::anyhow!("grant intersection: {e}"))?;
848        // PSP-10 system 22: a node's declared output_targets become its
849        // enforced write ceiling — mutating proposals outside the declared
850        // footprint are denied while reads stay workspace-wide. Undeclared
851        // targets keep unrestricted writes (the opaque scheduler footprint
852        // still serializes such nodes).
853        if let Some(node) = running_graph.node(node_id) {
854            capability.write_scope = write_scope_from_targets(&node.output_targets);
855        }
856        let contract = CodingContract {
857            graph_revision: running_graph.revision_id.clone(),
858            node_id: node_id.to_string(),
859            generation,
860            policy: perspt_policy::engine::PolicyEngine::new()?,
861        };
862        let barrier = OperationalSafetyBarrier::default();
863        let cadence = domain.verifier_suite(&scope).cadence;
864        let energy = domain.energy_model(&scope);
865        let calibration =
866            self.record_calibration_readiness(recorder, &catalog, &capability, &cadence)?;
867        Ok(NodeAssembly {
868            catalog,
869            grant_policy,
870            capability,
871            contract,
872            barrier,
873            cadence,
874            energy,
875            calibration,
876        })
877    }
878
879    /// Decide the node's terminal fate after the loop: adjudication and
880    /// human approval first, then the kernel-certified promotion. Anything
881    /// short of a fully certified promotion escalates.
882    async fn conclude_run(
883        &self,
884        ctx: ConcludeContext<'_>,
885    ) -> Result<(NodeTerminalOutcome, &'static str, Vec<String>)> {
886        let hard_pass = matches!(ctx.loop_outcome, NodeTerminalOutcome::HardPass);
887        let promotion_approved = self.validate_and_approve(&ctx, hard_pass).await?;
888        let mut final_outcome = if hard_pass && !promotion_approved {
889            NodeTerminalOutcome::Escalated {
890                certificate_id: uuid::Uuid::new_v4().to_string(),
891            }
892        } else {
893            ctx.loop_outcome.clone()
894        };
895        let mut promoted_paths = Vec::new();
896        let status = if matches!(final_outcome, NodeTerminalOutcome::HardPass) {
897            let certified = self
898                .promote_certified(
899                    ctx.recorder,
900                    ctx.candidate,
901                    ctx.node_id,
902                    ctx.grant_policy,
903                    ctx.capability,
904                    ctx.contract,
905                    ctx.barrier,
906                    ctx.kernel_state,
907                    ctx.calibration,
908                )
909                .await?;
910            match certified {
911                Some(paths) => {
912                    promoted_paths = paths;
913                    self.emit(perspt_core::AgentEvent::NodeCompleted {
914                        node_id: ctx.node_id.to_string(),
915                        goal: ctx.task.to_string(),
916                    });
917                    "COMPLETED_PSP9"
918                }
919                None => {
920                    final_outcome = NodeTerminalOutcome::Escalated {
921                        certificate_id: uuid::Uuid::new_v4().to_string(),
922                    };
923                    self.emit(perspt_core::AgentEvent::TaskStatusChanged {
924                        node_id: ctx.node_id.to_string(),
925                        status: perspt_core::NodeStatus::Escalated,
926                    });
927                    "ESCALATED_PSP9"
928                }
929            }
930        } else {
931            self.emit(perspt_core::AgentEvent::TaskStatusChanged {
932                node_id: ctx.node_id.to_string(),
933                status: perspt_core::NodeStatus::Escalated,
934            });
935            "ESCALATED_PSP9"
936        };
937        Ok((final_outcome, status, promoted_paths))
938    }
939
940    /// Record the deterministic-suite verdict (system 8's second live
941    /// validator, joined to the adjudicator's row by candidate id and
942    /// stratum), then run adjudication and approval for a hard pass.
943    async fn validate_and_approve(
944        &self,
945        ctx: &ConcludeContext<'_>,
946        hard_pass: bool,
947    ) -> Result<bool> {
948        let candidate_id = ctx.candidate.checkpoint(&[]).await?.witness.state_root;
949        record_deterministic_verdict(
950            ctx.recorder,
951            &candidate_id,
952            &ctx.calibration.stratum,
953            hard_pass,
954        )?;
955        if !hard_pass {
956            return Ok(false);
957        }
958        let adjudicated = self
959            .adjudicate_candidate(
960                ctx.recorder,
961                ctx.candidate,
962                ctx.task,
963                &ctx.calibration.stratum,
964            )
965            .await?;
966        Ok(adjudicated && self.conformal_or_human_approval(ctx).await?)
967    }
968
969    /// Autonomous commitment (PSP-9 Gate Q): with an *activated* calibration
970    /// epoch for this exact stratum — reached only at the finite sample floor
971    /// through delayed audit labels — a hard-pass candidate above the
972    /// conformal threshold commits without a human prompt, and the certified
973    /// accept is ledgered. Any other state (shadow, insufficient samples,
974    /// stale) backs off to the configured approval policy.
975    async fn conformal_or_human_approval(&self, ctx: &ConcludeContext<'_>) -> Result<bool> {
976        // Score definition v1: hard pass ⇒ V = 0 ⇒ score 1/(1+V) = 1.0.
977        const HARD_PASS_SCORE: f64 = 1.0;
978        let certified = self.config.approval_policy == ApprovalPolicy::Ask
979            && ctx.calibration.state == "active"
980            && ctx
981                .calibration
982                .threshold
983                .is_some_and(|theta| HARD_PASS_SCORE > theta);
984        if certified {
985            ctx.recorder.record_custom(
986                "conformal_certified_accept",
987                serde_json::json!({
988                    "epoch_id": ctx.calibration.epoch_id,
989                    "threshold": ctx.calibration.threshold,
990                    "score": HARD_PASS_SCORE,
991                    "node_id": ctx.node_id,
992                }),
993            )?;
994            return Ok(true);
995        }
996        self.approve_promotion(
997            ctx.recorder,
998            ctx.node_id,
999            ctx.candidate.touched_paths(),
1000            ctx.candidate.realized_diff().ok(),
1001        )
1002        .await
1003    }
1004
1005    /// Certify and perform the run's one durable effect. Promotion passes
1006    /// through the same five-clause kernel as every candidate mutation: a
1007    /// `WriteArtifact` proposal over the realized mutated paths, evaluated
1008    /// with the contract and barrier on the realized witness. A kernel denial
1009    /// escalates instead of promoting (`None`).
1010    #[allow(clippy::too_many_arguments)]
1011    async fn promote_certified(
1012        &self,
1013        recorder: &Psp9Recorder,
1014        candidate: &CandidateWorkspace,
1015        node_id: &str,
1016        grant_policy: &perspt_sdk::GrantPolicy,
1017        capability: &Capability,
1018        contract: &CodingContract,
1019        barrier: &OperationalSafetyBarrier,
1020        kernel_state: &perspt_sdk::KernelState,
1021        calibration: &CalibrationBinding,
1022    ) -> Result<Option<Vec<String>>> {
1023        anyhow::ensure!(
1024            recorder.authority_epoch()? == grant_policy.authority_epoch,
1025            "authority epoch changed before promotion"
1026        );
1027        recorder.record_custom(
1028            "authority_epoch_rechecked",
1029            serde_json::json!({
1030                "node_id": node_id,
1031                "generation": capability.node_generation,
1032                "epoch": grant_policy.authority_epoch,
1033            }),
1034        )?;
1035        let touched = candidate.touched_paths();
1036        let realized = candidate.checkpoint(&[]).await?.witness;
1037        let certified = certify_promotion(
1038            recorder,
1039            node_id,
1040            &touched,
1041            &realized,
1042            capability,
1043            contract,
1044            barrier,
1045            kernel_state,
1046        )?;
1047        if !certified {
1048            return Ok(None);
1049        }
1050
1051        let promoted_paths = commit_promotion(
1052            recorder,
1053            candidate,
1054            node_id,
1055            grant_policy.authority_epoch,
1056            &realized.state_root,
1057            &touched,
1058            &self.working_dir,
1059        )?;
1060        record_promotion_sample(recorder, calibration, &realized.state_root)?;
1061        Ok(Some(promoted_paths))
1062    }
1063    /// The exact fingerprinted stratum this run calibrates under: model
1064    /// route, verifier suite, catalog, policy, and score definition.
1065    fn calibration_stratum(
1066        &self,
1067        catalog: &StaticCatalog,
1068        capability: &Capability,
1069        cadence: &perspt_sdk::VerificationCadence,
1070    ) -> Result<perspt_sdk::CalibrationStratum> {
1071        let verifier_suite_fingerprint =
1072            perspt_sdk::content_hash(&serde_json::to_vec(&serde_json::json!({
1073                "cadence": cadence,
1074                "plugins": perspt_core::PluginRegistry::new()
1075                    .detect_all(&self.working_dir)
1076                    .into_iter()
1077                    .map(|plugin| plugin.name().to_string())
1078                    .collect::<Vec<_>>(),
1079            }))?);
1080        let tool_catalog_fingerprint = perspt_sdk::content_hash(&serde_json::to_vec(
1081            &catalog.specs_for(std::slice::from_ref(capability), true),
1082        )?);
1083        Ok(perspt_sdk::CalibrationStratum {
1084            domain_package: "perspt-coding".into(),
1085            domain_version: env!("CARGO_PKG_VERSION").into(),
1086            effect_kind: "candidate_promotion".into(),
1087            risk_class: "workspace_mutation".into(),
1088            model_route: self.model.to_string(),
1089            verifier_suite_fingerprint,
1090            tool_catalog_fingerprint,
1091            policy_version: "coding-contract-v1".into(),
1092            score_definition: "hard-gate-plus-quadratic-energy-v1".into(),
1093        })
1094    }
1095
1096    fn record_calibration_readiness(
1097        &self,
1098        recorder: &Psp9Recorder,
1099        catalog: &StaticCatalog,
1100        capability: &Capability,
1101        cadence: &perspt_sdk::VerificationCadence,
1102    ) -> Result<CalibrationBinding> {
1103        const TARGET_RHO: f64 = 0.05;
1104        let stratum = self.calibration_stratum(catalog, capability, cadence)?;
1105        let serialized = serde_json::to_string(&stratum)?;
1106        let epoch = match recorder.store.latest_psp9_calibration_epoch(&serialized)? {
1107            Some(epoch) => epoch,
1108            None => {
1109                let epoch = perspt_store::Psp9CalibrationEpochRow {
1110                    epoch_id: uuid::Uuid::new_v4().to_string(),
1111                    stratum: serialized,
1112                    target_rho: TARGET_RHO,
1113                    threshold: None,
1114                    state: "insufficient_samples".into(),
1115                    sample_count: 0,
1116                };
1117                recorder.store.record_psp9_calibration_epoch(&epoch)?;
1118                epoch
1119            }
1120        };
1121        let need = perspt_sdk::sample_floor(epoch.target_rho)
1122            .map_err(|error| anyhow::anyhow!(error.to_string()))?;
1123        let certified = epoch.state == "active" && epoch.threshold.is_some();
1124        recorder.record_custom(
1125            "calibration_readiness",
1126            serde_json::json!({
1127                "epoch_id": epoch.epoch_id,
1128                "stratum": stratum,
1129                "state": epoch.state,
1130                "sample_count": epoch.sample_count,
1131                "sample_floor": need,
1132                "target_rho": epoch.target_rho,
1133                "threshold": epoch.threshold,
1134                "certified_for_promotion": certified,
1135                "reason": if certified {
1136                    "an activated epoch at the finite sample floor backs the \
1137                     conformal accepted-unsafe bound for this stratum"
1138                } else {
1139                    "coding promotion currently relies on deterministic contract, \
1140                     barrier, and verifier evidence; no probabilistic claim is used"
1141                },
1142            }),
1143        )?;
1144        Ok(CalibrationBinding {
1145            epoch_id: epoch.epoch_id,
1146            stratum: epoch.stratum,
1147            state: epoch.state,
1148            threshold: epoch.threshold,
1149        })
1150    }
1151
1152    async fn approve_promotion(
1153        &self,
1154        recorder: &Psp9Recorder,
1155        node_id: &str,
1156        paths: Vec<String>,
1157        diff: Option<String>,
1158    ) -> Result<bool> {
1159        match self.config.approval_policy {
1160            ApprovalPolicy::Auto | ApprovalPolicy::SessionApproved => return Ok(true),
1161            ApprovalPolicy::Deny => return Ok(false),
1162            ApprovalPolicy::Ask => {}
1163        }
1164        let request_id = uuid::Uuid::new_v4().to_string();
1165        recorder.record_custom(
1166            "approval_requested",
1167            serde_json::json!({"request_id": request_id, "node_id": node_id, "paths": paths}),
1168        )?;
1169        self.emit(perspt_core::AgentEvent::ApprovalRequest {
1170            request_id: request_id.clone(),
1171            node_id: node_id.into(),
1172            action_type: perspt_core::ActionType::BundleWrite {
1173                node_id: node_id.into(),
1174                files: paths,
1175            },
1176            description: "Promote the verified PSP-9 candidate to the workspace".into(),
1177            diff,
1178        });
1179        let mut receiver_slot = self.action_receiver.lock().await;
1180        let Some(receiver) = receiver_slot.as_mut() else {
1181            recorder.record_custom(
1182                "approval_unavailable",
1183                serde_json::json!({"request_id": request_id}),
1184            )?;
1185            return Ok(false);
1186        };
1187        while let Some(action) = receiver.recv().await {
1188            let approved = match action {
1189                perspt_core::AgentAction::Approve { request_id: id }
1190                | perspt_core::AgentAction::ApproveWithEdit { request_id: id, .. }
1191                    if id == request_id =>
1192                {
1193                    Some(true)
1194                }
1195                perspt_core::AgentAction::Reject { request_id: id, .. } if id == request_id => {
1196                    Some(false)
1197                }
1198                perspt_core::AgentAction::Abort => Some(false),
1199                _ => None,
1200            };
1201            if let Some(approved) = approved {
1202                recorder.record_custom(
1203                    "approval_resolved",
1204                    serde_json::json!({"request_id": request_id, "approved": approved}),
1205                )?;
1206                return Ok(approved);
1207            }
1208        }
1209        Ok(false)
1210    }
1211}
perspt_agent/runtime/mod.rs

perspt_agent/runtime/
mod.rs
