perspt_agent/turn/
runner.rs

1//! The actor turn runner (PSP-10 system 27).
2
3use anyhow::Result;
4use perspt_sdk::prompt::{ContextBudget, TokenAccountantRef};
5use perspt_sdk::{
6    Conversation, FailureKind, ModelId, ModelTransport, ToolChoicePolicy, ToolSpec, TurnOutput,
7};
8
9use crate::toolloop::{LoopEvent, LoopRecorder};
10
11/// The stochastic actors one runner serves (system 27).
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum ActorKind {
14    Worker,
15    Explorer,
16    Architect,
17    Adjudicator,
18    Summarizer,
19    CapabilityProbe,
20}
21
22impl ActorKind {
23    pub fn tag(&self) -> &'static str {
24        match self {
25            ActorKind::Worker => "worker",
26            ActorKind::Explorer => "explorer",
27            ActorKind::Architect => "architect",
28            ActorKind::Adjudicator => "adjudicator",
29            ActorKind::Summarizer => "summarizer",
30            ActorKind::CapabilityProbe => "capability_probe",
31        }
32    }
33}
34
35/// The default per-turn wall-clock deadline when none is configured.
36pub const DEFAULT_TURN_DEADLINE_SECS: u64 = 120;
37
38/// The prompt route and dialect of one resolved model (PSP-10 system 24):
39/// route identity from the transport, the single-slot genai dialect fitted
40/// to the route's declared context window. Scripted transports get the
41/// same structural dialect, so tests exercise the production layout.
42pub fn route_dialect(
43    transport: &dyn ModelTransport,
44    model: &ModelId,
45) -> (
46    perspt_sdk::prompt::PromptRoute,
47    perspt_sdk::prompt::ModelDialect,
48) {
49    let route = transport.prompt_route(model);
50    let window = u64::from(transport.capabilities(model).max_context_tokens.max(1));
51    let dialect =
52        perspt_sdk::prompt::ModelDialect::genai_single_slot_v1().with_context_window(window);
53    (route, dialect)
54}
55
56/// One transport call under a hard wall-clock deadline — the one door every
57/// actor's model turn goes through. A finite turn count alone never bounds
58/// wall time; the deadline does. Exceeding it is an ordinary transport
59/// failure (it consumes sticky failover like a provider outage), and
60/// dropping the timed future cancels the underlying request.
61pub async fn chat_turn_with_deadline(
62    transport: &dyn ModelTransport,
63    model: &ModelId,
64    conversation: &Conversation,
65    tools: &[ToolSpec],
66    choice: ToolChoicePolicy,
67    deadline_secs: u64,
68) -> perspt_sdk::Result<TurnOutput> {
69    let deadline = std::time::Duration::from_secs(deadline_secs.max(1));
70    match tokio::time::timeout(
71        deadline,
72        transport.chat_turn(model, conversation, tools, choice),
73    )
74    .await
75    {
76        Ok(result) => result,
77        Err(_) => Err(perspt_sdk::SdkError::Domain(format!(
78            "transport deadline: no turn from {model} within {deadline_secs}s"
79        ))),
80    }
81}
82
83/// Classify a transport failure cause — the one definition the worker loop
84/// and the runner share, so no actor invents its own retry taxonomy.
85pub fn transport_failure_kind(cause: &str) -> FailureKind {
86    let lowered = cause.to_ascii_lowercase();
87    let rate_limited = ["rate limit", "429", "too many requests"]
88        .iter()
89        .any(|marker| lowered.contains(marker));
90    if rate_limited {
91        FailureKind::ProviderRateLimit
92    } else {
93        FailureKind::ProviderTransport
94    }
95}
96
97/// One actor's turn discipline: budget feasibility, sticky failover, and
98/// raw-before-parse observation recording.
99pub struct ActorTurnRunner<'a> {
100    pub transport: &'a dyn ModelTransport,
101    pub model: ModelId,
102    /// Sticky failover chain, consumed only on observed transport failure.
103    pub fallbacks: Vec<ModelId>,
104    pub recorder: Option<&'a dyn LoopRecorder>,
105    pub actor: ActorKind,
106    /// Turn ordinal within this actor's own exchange.
107    pub turn: u32,
108    /// Per-call wall-clock deadline in seconds ([`DEFAULT_TURN_DEADLINE_SECS`]
109    /// when unconfigured).
110    pub deadline_secs: u64,
111}
112
113impl ActorTurnRunner<'_> {
114    fn emit(&self, event: LoopEvent) -> Result<()> {
115        if let Some(recorder) = self.recorder {
116            recorder.record(&event)?;
117        }
118        Ok(())
119    }
120
121    /// Definition 6's refusal: the composed request must fit the route's
122    /// input allowance under the versioned accountant; an infeasible
123    /// request makes no model call and records `ContextInfeasible`.
124    fn ensure_feasible(&self, conversation: &Conversation, tools: &[ToolSpec]) -> Result<()> {
125        let capabilities = self.transport.capabilities(&self.model);
126        let accountant = TokenAccountantRef::approx_bytes_v1();
127        let tool_reserve: u64 = tools
128            .iter()
129            .map(|spec| {
130                accountant.count_text(&format!("{}{}{}", spec.name, spec.description, spec.schema))
131            })
132            .sum();
133        let budget = ContextBudget {
134            window_tokens: u64::from(capabilities.max_context_tokens.max(1)),
135            output_reserve: 1_024,
136            tool_reserve,
137            guard_reserve: 256,
138        };
139        let allowance = budget
140            .input_allowance()
141            .map_err(|e| anyhow::anyhow!("{e}"))?;
142        let required: u64 = conversation
143            .messages()
144            .iter()
145            .map(|message| accountant.count_message(&format!("{message:?}")))
146            .sum();
147        if required > allowance {
148            self.emit(LoopEvent::ContextInfeasible {
149                forest_id: String::new(),
150                branch_id: String::new(),
151                turn: self.turn,
152                required,
153                allowance,
154            })?;
155            anyhow::bail!(
156                "context budget infeasible for {}: {required} tokens over the \
157                 {allowance}-token input allowance; no model call was made",
158                self.actor.tag()
159            );
160        }
161        Ok(())
162    }
163
164    /// Run one turn: feasibility, transport with the shared sticky
165    /// failover, then the raw actor-tagged observation — recorded before
166    /// any caller inspects the output (system 27 step 4).
167    pub async fn run_turn(
168        &mut self,
169        conversation: &Conversation,
170        tools: &[ToolSpec],
171        choice: ToolChoicePolicy,
172    ) -> Result<TurnOutput> {
173        self.ensure_feasible(conversation, tools)?;
174        let output = loop {
175            match chat_turn_with_deadline(
176                self.transport,
177                &self.model,
178                conversation,
179                tools,
180                choice.clone(),
181                self.deadline_secs,
182            )
183            .await
184            {
185                Ok(output) => break output,
186                Err(error) => {
187                    let cause = error.to_string();
188                    let _failure = transport_failure_kind(&cause);
189                    let Some(next) = self.fallbacks.first().cloned() else {
190                        anyhow::bail!(
191                            "{} transport failed with no remaining fallback: {cause}",
192                            self.actor.tag()
193                        );
194                    };
195                    self.fallbacks.remove(0);
196                    self.emit(LoopEvent::RouteFailover {
197                        from_model: self.model.clone(),
198                        to_model: next.clone(),
199                        cause,
200                    })?;
201                    self.model = next;
202                }
203            }
204        };
205        self.emit(LoopEvent::TurnObserved {
206            turn: self.turn,
207            actor: self.actor.tag().to_string(),
208            output: output.clone(),
209        })?;
210        Ok(output)
211    }
212}
perspt_agent/turn/runner.rs

perspt_agent/turn/
runner.rs
