runtime/turn/
speculation.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! Read-only tool calls started while the model is still generating,
7//! and the cache that holds what they returned until the answer settles.
8//!
9//! `crates/runtime/spec/Turn/Speculation.lean` is the authority on what may start
10//! early and in which order the results reach the Ledger: only the reads
11//! before the answer's first writing call, cached by position, recorded
12//! in emission order, and a failed answer's cache discarded whole.
13
14use kernel::{AxError, Increment, Model, ModelRequest, ModelReturn, ToolCall, ToolOutcome};
15
16use super::wave::{ConcurrentInvoke, lost_answer, reads_only};
17
18/// Who acts on a model's answer while it is still being generated. The
19/// one place that decides which door of the model port a turn takes.
20pub enum Generating<'a, 'sink> {
21    /// Nobody: the blocking door.
22    Unwatched,
23    /// A page reads the text as it arrives: the streaming door.
24    Watched(&'a mut (dyn FnMut(&Increment) + 'sink)),
25    /// The tool face also starts each read-only call the model hands
26    /// over before its first writing call: the speculating door.
27    Speculating {
28        deltas: Option<&'a mut (dyn FnMut(&Increment) + 'sink)>,
29        tools: &'a dyn ConcurrentInvoke,
30    },
31}
32
33/// What the reads started early returned, by their position in the
34/// answer. Each result is held beside the call that was started, so it
35/// answers only the settled call equal to it: the model port owes that
36/// equality, and holding the call keeps a port that broke it from
37/// answering one call with another's result.
38#[derive(Debug, Default)]
39pub(super) struct Speculated {
40    held: Vec<(ToolCall, Result<ToolOutcome, AxError>)>,
41}
42
43impl Speculated {
44    /// The cached answer for each of `calls`, in order: present where the
45    /// call at that position was started early and is the one that
46    /// settled there.
47    pub(super) fn answers_for(
48        self,
49        calls: &[&ToolCall],
50    ) -> Vec<Option<Result<ToolOutcome, AxError>>> {
51        let mut held = self.held.into_iter();
52        calls
53            .iter()
54            .map(|call| {
55                held.next()
56                    .and_then(|(started, answered)| (started == **call).then_some(answered))
57            })
58            .collect()
59    }
60}
61
62/// Whether the calls handed over so far are all reads.
63#[derive(Clone, Copy)]
64enum Prefix {
65    Reading,
66    Written,
67}
68
69/// Makes the model call through the speculating door, starting each
70/// read-only call handed over before the first writing call on a scoped
71/// thread of its own. Every thread is joined before this returns, so the
72/// turn waits for the longer of the reads and the generation rather than
73/// their sum. The scope is this module's exception to the one spawn point
74/// (ARCHITECTURE §10 rule 3), as `wave`'s is.
75///
76/// # Errors
77/// Whatever the model call fails with; the reads it started are joined
78/// and their results dropped with it.
79pub(super) fn call_ahead(
80    model: &mut dyn Model,
81    request: &ModelRequest,
82    onto: &mut dyn FnMut(&Increment),
83    tools: &dyn ConcurrentInvoke,
84) -> Result<(ModelReturn, Speculated), AxError> {
85    std::thread::scope(|scope| {
86        let mut started = Vec::new();
87        let mut prefix = Prefix::Reading;
88        let returned = model.call_speculating(request, onto, &mut |call: &ToolCall| {
89            // Resolved as the wave resolves it, so what starts early is
90            // the call the wave will admit and compare against.
91            let call = tools.resolve_call(call.clone());
92            if let Prefix::Reading = prefix
93                && reads_only(tools, &call)
94                && let Some(tool) = tools.ahead(&call)
95            {
96                let handed = call.clone();
97                started.push((call, scope.spawn(move || tool.invoke(&handed))));
98            } else {
99                prefix = Prefix::Written;
100            }
101        });
102        let held = started
103            .into_iter()
104            .map(|(call, worker)| (call, worker.join().unwrap_or_else(|_| Err(lost_answer()))))
105            .collect();
106        returned.map(|value| (value, Speculated { held }))
107    })
108}
runtime/turn/speculation.rs

runtime/turn/
speculation.rs
