diff --git a/apps/photocraft/src/control_server.rs b/apps/photocraft/src/control_server.rs index 8656cafd00..33923d9fac 100644 --- a/apps/photocraft/src/control_server.rs +++ b/apps/photocraft/src/control_server.rs @@ -5,7 +5,7 @@ use std::io::{BufReader, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::Arc; use std::sync::mpsc::{Receiver, Sender, channel}; -use std::time::Duration; +use std::time::{Duration, Instant}; use photocraft_automation::budgets::write_reply; use photocraft_automation::security::{ @@ -83,12 +83,14 @@ fn serve(stream: TcpStream, token: &str, tx: Sender, ctx: egui:: let id = msg.get("id").cloned().unwrap_or(Value::Null); let method = msg.get("method").and_then(Value::as_str).unwrap_or("").to_string(); let params = msg.get("params").cloned().unwrap_or(json!({})); - let (req, rrx) = ControlRequest::new(method, params); + let deadline = Instant::now() + Duration::from_secs(60); + let (mut req, rrx) = ControlRequest::new(method, params); + req.deadline = Some(deadline); if tx.send(req).is_err() { break; } ctx.request_repaint(); - let mut r = rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"})); + let mut r = rrx.recv_timeout(deadline.saturating_duration_since(Instant::now())).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"})); if let Some(o) = r.as_object_mut() { o.insert("id".into(), id); } @@ -124,6 +126,7 @@ mod tests { let handler = std::thread::spawn(move || { let first = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(first.method, "test.large"); + assert!(first.deadline.is_some_and(|deadline| deadline > std::time::Instant::now()), "native requests carry their reply deadline"); first.reply.send(json!({"ok": true, "result": "x".repeat(MAX_RESPONSE_BYTES)})).unwrap(); let second = rx.recv_timeout(Duration::from_secs(5)).unwrap(); assert_eq!(second.method, "test.small"); diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index 800f05a112..667a1e3d33 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -45,12 +45,14 @@ pub struct ControlRequest { pub method: String, pub params: Value, pub reply: Sender, + /// Requests still queued at this instant are rejected; already dispatched work continues. + pub deadline: Option, } impl ControlRequest { pub fn new(method: impl Into, params: Value) -> (Self, std::sync::mpsc::Receiver) { let (tx, rx) = std::sync::mpsc::channel(); - (Self { method: method.into(), params, reply: tx }, rx) + (Self { method: method.into(), params, reply: tx, deadline: None }, rx) } } @@ -816,6 +818,45 @@ mod tests { } } + #[test] + fn expired_queued_edit_does_not_run_and_a_live_retry_runs_once() { + use std::time::{Duration, Instant}; + let mut app = PhotocraftApp::new(photocraft_engine::Session::new(), crate::Services::default()); + app.run("file.new", json!({"width": 8, "height": 8})).unwrap(); + let before = app.session.active().unwrap(); + let layers = before.doc.layers.len(); + let steps = before.history.past_len(); + let (tx, rx) = std::sync::mpsc::channel(); + app = app.with_control(rx); + let (mut expired, expired_reply) = ControlRequest::new("engine.execute", json!({"command": "layer.new.layer"})); + expired.deadline = Some(Instant::now() - Duration::from_secs(1)); + tx.send(expired).unwrap(); + let (mut retry, retry_reply) = ControlRequest::new("engine.execute", json!({"command": "layer.new.layer"})); + retry.deadline = Some(Instant::now() + Duration::from_secs(60)); + tx.send(retry).unwrap(); + app.drain_control(&egui::Context::default()); + assert_eq!(expired_reply.try_recv().unwrap(), json!({"ok": false, "error": "timeout"})); + assert_eq!(retry_reply.try_recv().unwrap()["ok"], true); + let after = app.session.active().unwrap(); + assert_eq!(after.doc.layers.len(), layers + 1, "only the live retry adds a layer"); + assert_eq!(after.history.past_len(), steps + 1, "the expired edit adds no undo step"); + } + + #[test] + fn deadline_free_requests_run_even_after_the_receiver_is_dropped() { + let mut app = PhotocraftApp::new(photocraft_engine::Session::new(), crate::Services::default()); + app.run("file.new", json!({"width": 8, "height": 8})).unwrap(); + let layers = app.session.active().unwrap().doc.layers.len(); + let (tx, rx) = std::sync::mpsc::channel(); + app = app.with_control(rx); + let (req, reply) = ControlRequest::new("engine.execute", json!({"command": "layer.new.layer"})); + assert!(req.deadline.is_none()); + drop(reply); + tx.send(req).unwrap(); + app.drain_control(&egui::Context::default()); + assert_eq!(app.session.active().unwrap().doc.layers.len(), layers + 1); + } + #[test] fn right_pointer_opens_agent_visible_selection_menu() { let mut app = PhotocraftApp::new(photocraft_engine::Session::new(), crate::Services::default()); diff --git a/crates/ui-egui/src/jobs_ui_tests.rs b/crates/ui-egui/src/jobs_ui_tests.rs index 946a538293..e02df99ceb 100644 --- a/crates/ui-egui/src/jobs_ui_tests.rs +++ b/crates/ui-egui/src/jobs_ui_tests.rs @@ -170,6 +170,31 @@ fn control_requests_wait_for_jobs_unless_asked_not_to() { assert_eq!(v["ok"], false); } +#[test] +fn accepted_control_job_applies_after_its_dispatch_deadline() { + let mut h = app_harness(); + let ctx = h.ctx.clone(); + let steps = h.state().session.active().unwrap().history.past_len(); + let deadline = Instant::now() + Duration::from_secs(1); + let (mut req, reply) = crate::ControlRequest::new("engine.execute", json!({"command": "filter.blur.gaussianBlur", "params": {"radius": 2}})); + req.deadline = Some(deadline); + let (tx, rx) = std::sync::mpsc::channel(); + h.state_mut().control_rx = Some(rx); + tx.send(req).unwrap(); + h.state_mut().drain_control(&ctx); + let job = h.state().jobs.waiters.first().expect("the accepted request waits for its job").0; + assert_eq!(h.state().session.active().unwrap().history.past_len(), steps, "the job has not been applied yet"); + assert!(reply.try_recv().is_err()); + // No frames poll the worker until the transport's deadline has elapsed. + std::thread::sleep(deadline.saturating_duration_since(Instant::now())); + assert!(Instant::now() >= deadline); + drop(reply); + let result = job_reply(&mut h, job); + assert_eq!(result["ok"], true, "{result}"); + assert!(result["result"]["filter"].is_object(), "{result}"); + assert_eq!(h.state().session.active().unwrap().history.past_len(), steps + 1); +} + /// The control reply for `job`, once the app's frame loop has applied it. fn job_reply(h: &mut Harness<'static, PhotocraftApp>, job: JobId) -> Value { let (tx, rx) = std::sync::mpsc::channel(); diff --git a/crates/ui-egui/src/lib.rs b/crates/ui-egui/src/lib.rs index 037858b448..369ab389f6 100644 --- a/crates/ui-egui/src/lib.rs +++ b/crates/ui-egui/src/lib.rs @@ -942,6 +942,10 @@ impl PhotocraftApp { let Some(rx) = self.control_rx.take() else { return }; while let Ok(req) = rx.try_recv() { let reply = req.reply.clone(); + if req.deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { + let _ = reply.send(serde_json::json!({"ok": false, "error": "timeout"})); + continue; + } match control::handle(self, ctx, &req) { control::Outcome::Done(v) => { let _ = reply.send(v); diff --git a/docs/control-protocol.md b/docs/control-protocol.md index cf0d302816..18ca342d34 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -28,6 +28,8 @@ Each reply is one JSON line with the same `id`: {"id": 2, "ok": false, "error": "unknown tool `foo`"} ``` +The desktop server waits up to 60 seconds for a reply. A request still queued at that deadline is rejected before dispatch; a timeout does not cancel work that has already started. + The transport is `apps/photocraft/src/control_server.rs`, and the handlers are in `crates/ui-egui/src/control.rs`. The MCP server (`photocraft-cli mcp --bridge 127.0.0.1:`, crate `photocraft-automation`) wraps this same protocol. See [MCP bridge](#mcp-bridge) below. ## Methods