diff --git a/AGENTS.md b/AGENTS.md index c47749c..7a81f1b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -487,6 +487,16 @@ written, and the fold uses that same predicate to decide a reply is settled. `generate` now fails the turn for both -- a reply that stops early is not a reply, and the transcript keeps whatever arrived before it. +- **A cancel flag is only as prompt as the next place somebody looks.** A + llama turn waits on two things that look nowhere at all: a permission + question, and a tool call `llama-server` is running -- a shell command there + runs to its own timeout, up to a minute. Setting `cancel` left the turn + exactly where it was until that came back, so Pause did nothing on screen + for as long as the command took. `Shared::abandon_turn` sets the flag *and* + releases both waits, the tool call by answering its channel with + `tools::UNFINISHED` -- the call is left running over there, because nothing + in that protocol takes one back, and its answer is dropped. + - **A path is stored as it was typed, and `~` is expanded where it is used.** `~/repos/x` and `/home/someone/repos/x` are a path and a snapshot of where it pointed, and the snapshot is what breaks when an account is renamed or the diff --git a/server/src/session/llama/mod.rs b/server/src/session/llama/mod.rs index edf0e35..cf03069 100644 --- a/server/src/session/llama/mod.rs +++ b/server/src/session/llama/mod.rs @@ -384,8 +384,10 @@ struct Shared { /// `~`. `None` leaves the directory to `llama-server`, which is the honest /// answer rather than a guess at one. cwd: Option, - /// Set by [`Driver::interrupt`]; the streaming loop and the tool loop both - /// check it, leaving what was produced in the transcript. + /// Set by [`Shared::abandon_turn`]; the streaming loop and the tool loop + /// both check it, leaving what was produced in the transcript. A flag is + /// only as prompt as the next place somebody looks, which is why the two + /// waits that look nowhere are released by hand there as well. cancel: AtomicBool, /// Whether a turn is running, and what is waiting behind it. /// @@ -413,6 +415,9 @@ struct Shared { allowed: Mutex>, /// Questions a turn is blocked on, by question id. asked: Mutex>>>, + /// Tool calls a turn is blocked on, by the call's id, so that an + /// interrupt can let go of one rather than sit out whatever it is doing. + running_tools: Mutex>>, /// Which of the machine's tools this session offers its model. Live like /// the sampling settings and for the same reason -- it is applied to the /// next request rather than to anything that was started. @@ -516,6 +521,7 @@ impl LlamaDriver { ), allowed: Mutex::new(allowances(transcript)), asked: Mutex::new(HashMap::new()), + running_tools: Mutex::new(HashMap::new()), tools_wanted: Mutex::new(Chosen::from(meta.params.get(TOOLS).map(String::as_str))), watching: AtomicBool::new(false), thinking: Mutex::new(chosen_thinking(&meta.params)), @@ -826,6 +832,39 @@ impl Shared { } } + /// Lets go of every tool call this session is waiting on, telling the + /// model what a call that never came back was. + /// + /// The call itself is not stopped: `llama-server` does not take one back + /// once it has started it, so the thread that asked is left to collect an + /// answer nobody reads. What ends here is the *waiting*, which is what a + /// turn is actually made of. + fn abandon_tools(&self) { + for (_, finished) in std::mem::take( + &mut *self + .running_tools + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ) { + let _ = finished.send(tools::UNFINISHED.to_string()); + } + } + + /// Ends the running turn: the flag, and everything blocked on something + /// that will not read it. + /// + /// Both of those waits are on another thread with no deadline a reader + /// would sit through -- a permission nobody is going to answer, and a tool + /// call running a shell command, which is up to a minute of `llama-server` + /// not being asked anything. Setting the flag alone left the turn exactly + /// where it was until that came back, which on screen was a Pause button + /// that did nothing at all. + fn abandon_turn(&self) { + self.cancel.store(true, Ordering::SeqCst); + self.abandon_questions(); + self.abandon_tools(); + } + fn emit(&self, event: Event) { let _ = self.sink.send(event); } @@ -1127,7 +1166,7 @@ impl LlamaDriver { fn converse( shared: &Arc, serves: &Serves, - tools: &Tools, + tools: &Arc, mut messages: Vec, ) -> Result<()> { for _ in 0..MAX_STEPS { @@ -1219,7 +1258,7 @@ fn take_steers(shared: &Arc, messages: &mut Vec) { /// refused or interrupted. A call with no result is a conversation a chat /// template cannot render -- see [`tools::UNFINISHED`] -- so there is no path /// out of here that leaves one. -fn run_call(shared: &Arc, tools: &Tools, call: &Call) -> String { +fn run_call(shared: &Arc, tools: &Arc, call: &Call) -> String { let arguments = call.arguments(); shared.emit(Event::ToolStart { id: call.id.clone(), @@ -1238,19 +1277,7 @@ fn run_call(shared: &Arc, tools: &Tools, call: &Call) -> String { } else if !permitted(shared, call) { tools::REFUSED.to_string() } else { - match tools.execute( - &call.name, - &arguments, - shared.cwd.as_deref(), - &shared.tools_wanted.lock().unwrap(), - ) { - Ok(output) => output, - // Reaching the tool failed, which is this server's problem and - // not the model's work going wrong -- but the model is still what - // has to carry on, so it is told in the result rather than only - // in the log. - Err(err) => format!("This tool could not be run: {err:#}"), - } + run_tool(shared, tools, call, arguments) }; shared.emit(Event::ToolEnd { id: call.id.clone(), @@ -1259,6 +1286,58 @@ fn run_call(shared: &Arc, tools: &Tools, call: &Call) -> String { output } +/// Runs one tool on a thread of its own, and waits for it in a way an +/// interrupt can reach. +/// +/// The wait is a channel with two writers -- the call's own thread and +/// [`Shared::abandon_tools`] -- and whichever speaks first is what the model +/// is told. A call is the one part of a turn this server can neither hurry +/// nor take back: `llama-server` runs it to its own timeout, up to a minute +/// for a shell command, and nothing in the protocol says "stop that one". So +/// an interrupt leaves it running over there and drops its answer, which is +/// what makes Pause end the turn when it is pressed rather than when the +/// command it happened to be running is over. +fn run_tool(shared: &Arc, tools: &Arc, call: &Call, arguments: Value) -> String { + let (finished, waiting) = std::sync::mpsc::channel(); + shared + .running_tools + .lock() + .unwrap() + .insert(call.id.clone(), finished.clone()); + // Checked after registering, not before: the flag is set ahead of the + // drain, so a call that registered after that drain would be waiting on a + // wake-up already given out. Either the insert beat the drain and this + // will be released, or it did not and the flag is already true. + if shared.cancel.load(Ordering::SeqCst) { + shared.running_tools.lock().unwrap().remove(&call.id); + return tools::UNFINISHED.to_string(); + } + { + let tools = Arc::clone(tools); + let name = call.name.clone(); + let cwd = shared.cwd.clone(); + // Read here rather than in the thread, which is where it was read + // before: the filter is live, and this is the call it applies to. + let chosen = shared.tools_wanted.lock().unwrap().clone(); + std::thread::spawn(move || { + let output = match tools.execute(&name, &arguments, cwd.as_deref(), &chosen) { + Ok(output) => output, + // Reaching the tool failed, which is this server's problem and + // not the model's work going wrong -- but the model is still + // what has to carry on, so it is told in the result rather + // than only in the log. + Err(err) => format!("This tool could not be run: {err:#}"), + }; + let _ = finished.send(output); + }); + } + let output = waiting + .recv() + .unwrap_or_else(|_| tools::UNFINISHED.to_string()); + shared.running_tools.lock().unwrap().remove(&call.id); + output +} + /// Whether this call may go ahead, asking whoever is reading if it has to. /// /// Blocks the turn while the question is out, which is what the question is @@ -1408,8 +1487,7 @@ impl Driver for LlamaDriver { } fn interrupt(&self) { - self.shared.cancel.store(true, Ordering::SeqCst); - self.shared.abandon_questions(); + self.shared.abandon_turn(); } // Nothing to forward: this process has no notion of what the conversation @@ -1527,8 +1605,7 @@ impl Driver for LlamaDriver { /// Stops generating and leaves the machine's server alone. fn detach(&self) { - self.shared.cancel.store(true, Ordering::SeqCst); - self.shared.abandon_questions(); + self.shared.abandon_turn(); } /// Ends this session's claim on the model and nothing else. @@ -1539,8 +1616,7 @@ impl Driver for LlamaDriver { /// than a rule to remember. A model is taken out of memory from the /// machine's provider settings, where what it costs everybody is visible. fn stop(&self) { - self.shared.cancel.store(true, Ordering::SeqCst); - self.shared.abandon_questions(); + self.shared.abandon_turn(); process::clear(&self.shared.session_dir); } } diff --git a/server/src/session/llama/tools.rs b/server/src/session/llama/tools.rs index 8b811e9..1ee3a92 100644 --- a/server/src/session/llama/tools.rs +++ b/server/src/session/llama/tools.rs @@ -235,6 +235,7 @@ impl Tools { /// people actually write: everything, nothing, or these. "Nothing" is the one /// that has to be sayable at all -- an empty list would be indistinguishable /// from the setting being unset, which is what `all` means. +#[derive(Clone)] pub enum Chosen { All, None,