sonny acp: a task opens as the lead thread's child in T3 Code

sonny/README.md

@@ -229,13 +229,18 @@ waits for you is asked in the thread (Allow or Deny); one the judge passes
 runs without asking. The turns the lead takes on its own, as when the lookout
 wakes it because a task finished, show in the thread between yours, as work
 the agent started (the protocol's version 2 draft, which T3 Code speaks).
-Each task the lead starts opens as a T3 Code thread of its own, in the task's
-worktree (in the project's folder when T3 Code does not take the worktree as
-one of the project's), with the task's brief as its first message; that
-thread is the task's, so it shows the task's work and steers it. Sonny opens
-it with the `t3_thread_launch` tool T3 Code gives each thread it runs, by the
-credential T3 Code scopes to that thread, which no program Sonny starts
-inherits. Sonny takes the person's words out of the instructions T3 Code
+Each task the lead starts opens as a T3 Code thread with the task's brief as
+its first message; that thread is the task's, so it shows the task's work and
+steers it. A task started while your prompt runs is the lead thread's child,
+nested under it, opened with T3 Code's `delegate_task` (T3 Code opens it in
+the lead's folder; the task still works in its own worktree), and your
+prompt's answer waits until it is open. T3 Code's message that a child ended
+starts no turn: the lead hears of its tasks from the lookout. A task started
+in a turn the lead took on its own, which T3 Code takes no child of, opens as
+a thread of its own with `t3_thread_launch`, in the task's worktree (in the
+project's folder when T3 Code does not take the worktree as one of the
+project's). Sonny calls these tools by the credential T3 Code scopes to the
+lead's thread, which no program Sonny starts inherits. Sonny takes the person's words out of the instructions T3 Code
 wraps a prompt in, which name tools Sonny's agents do not have, so T3 Code's
 Plan mode does not reach them; plan with the lead instead. What thread is
 which bot is kept in `~/.agent/sonny/t3`, so a thread T3 Code resumes is the

sonny/src/acp.rs

@@ -3,8 +3,8 @@
 //! thread. A thread in a folder is that folder's lead, as `sonny` there
 //! finds or makes it, set up the same way: its rules and skills, its tools,
 //! its approval mode, and the lookout. Each task the lead starts opens as a
-//! thread of its own, through T3 Code's tools, and that thread is the
-//! task's.
+//! thread through T3 Code's tools, the lead thread's child while the
+//! person's prompt runs, and that thread is the task's.
 //!
 //! It speaks the protocol's version 2 draft as T3 Code does: the turns a
 //! bot takes on its own, as when the lookout wakes a lead because a task
@@ -29,6 +29,9 @@ const VERSION: i64 = 2;
 /// The most of a call's output a thread is shown.
 const OUTPUT_BYTES: usize = 8 * 1024;
 
+/// The longest the person's prompt waits for its tasks' threads to open.
+const OPENING: std::time::Duration = std::time::Duration::from_secs(30);
+
 /// JSON-RPC's codes for what Sonny refuses.
 const INVALID_PARAMS: i64 = -32602;
 const NOT_FOUND: i64 = -32601;
@@ -406,6 +409,11 @@ struct State {
     /// The model the person picked, which the bot takes when it is next
     /// idle; a new lead starts on it.
     picked: Option<String>,
+    /// Tasks whose threads are opening, as children of the person's prompt
+    /// while it runs, and the answers of the prompts whose turns ended
+    /// meanwhile, held until they are open.
+    opening: usize,
+    held: Vec<(Value, Value)>,
 }
 
 struct Bound {
@@ -683,6 +691,9 @@ impl Session {
 
     async fn prompt(self: &Arc<Self>, agent: &Arc<Agent>, id: &Value, params: &Value) -> Reply {
         let (text, images) = words(&params["prompt"]);
+        if agent.t3.is_some() && images.is_empty() && t3::is_completion(&text) {
+            return Ok(json!({"stopReason": "end_turn"}));
+        }
         let bound = {
             let mut state = self.state.lock().await;
             state.cancelled = false;
@@ -1088,13 +1099,35 @@ impl Session {
                         .as_ref()
                         .map(|b| b.workspace.clone())
                         .unwrap_or_default();
-                    let (agent, session) = (agent.clone(), self.id.clone());
-                    tokio::spawn(async move {
-                        open_task(
-                            &agent, &session, &client, &project, &arguments, &workspace, &call,
-                        )
-                        .await
-                    });
+                    let this = (agent.sessions.lock().ok())
+                        .and_then(|sessions| sessions.get(&self.id).cloned());
+                    if let Some(session) = this {
+                        state.opening += 1;
+                        let session_id = session.id.clone();
+                        let agent = agent.clone();
+                        tokio::spawn(async move {
+                            let (opener, thread) = (agent.clone(), session.cwd.clone());
+                            let opening = tokio::spawn(async move {
+                                let at = Where {
+                                    project: &project,
+                                    workspace: &workspace,
+                                    thread: &thread,
+                                };
+                                let session = &session_id;
+                                open_task(&opener, session, &client, &at, &arguments, &call).await;
+                            });
+                            // The person's prompt waits for its tasks' threads,
+                            // not for ever: one still opening opens on its own.
+                            let _ = tokio::time::timeout(OPENING, opening).await;
+                            let mut state = session.state.lock().await;
+                            state.opening -= 1;
+                            if state.opening == 0 {
+                                for (prompt, result) in std::mem::take(&mut state.held) {
+                                    agent.out.answer(&prompt, result);
+                                }
+                            }
+                        });
+                    }
                 }
             }
             "approval_requested" | "approval_asked" => {
@@ -1144,7 +1177,13 @@ impl Session {
                         result["error"] = json!({"code": FAILED,
                             "message": data["error"].as_str().unwrap_or("the turn failed")});
                     }
-                    out.answer(&prompt, result);
+                    // A task T3 Code opens as a child of this prompt opens
+                    // while its run is on.
+                    if state.opening > 0 {
+                        state.held.push((prompt, result));
+                    } else {
+                        out.answer(&prompt, result);
+                    }
                 }
             }
             "deleted" => {
@@ -1430,18 +1469,28 @@ async fn setup_lead(
     Ok(())
 }
 
+/// Where a lead's thread is: its project, the lead's workspace, and the
+/// folder T3 Code opened the thread in.
+struct Where<'a> {
+    project: &'a str,
+    workspace: &'a str,
+    thread: &'a Path,
+}
+
 /// Open the task a lead's `start_task` call started as a thread of its
 /// own, with the task's brief as its first message; the thread is the
-/// task's when that message reaches it. The lead's call says where.
+/// task's when that message reaches it. While the person's prompt runs it
+/// is the lead thread's child, which T3 Code opens in the lead's folder;
+/// otherwise a thread of its own, where the lead's call says.
 async fn open_task(
     agent: &Agent,
     session: &str,
     client: &Client,
-    project: &str,
+    at: &Where<'_>,
     arguments: &Value,
-    lead_workspace: &str,
     call: &str,
 ) {
+    let (project, lead_workspace) = (at.project, at.workspace);
     let Some(t3) = &agent.t3 else { return };
     let Some(name) = arguments["name"].as_str() else {
         return;
@@ -1472,6 +1521,17 @@ async fn open_task(
             .ok_or("the task has no brief")?
             .to_owned();
         let workspace = record["workspace"].as_str().unwrap_or_default();
+        let message = t3::expect(&bot, &brief, at.thread)?;
+        let child = json!({"task": message, "title": name, "clientRequestId": bot});
+        // The person's prompt waits for it: not for ever. One that is late
+        // may open yet, so its task still waits for it and no other thread
+        // is opened for it.
+        let delegating = t3.call("delegate_task", child);
+        match tokio::time::timeout(OPENING, delegating).await {
+            Ok(Ok(delegated)) => return Ok(delegated),
+            Ok(Err(_)) => {}
+            Err(_) => return Ok(Value::Null),
+        }
         let message = t3::expect(&bot, &brief, Path::new(workspace))?;
         let mut launch = json!({"title": name, "message": message, "clientRequestId": bot});
         if workspace != lead_workspace {

sonny/src/t3.rs

@@ -1,7 +1,7 @@
 //! T3 Code, as `sonny acp` meets it: the tools T3 Code gives the agents it
 //! runs, over an MCP endpoint it names in their environment, with which a
-//! lead's thread opens a thread for each task the lead starts; the tasks
-//! waiting for their thread, and the bot each thread is.
+//! lead's thread opens a child thread for each task the lead starts; the
+//! tasks waiting for their thread, and the bot each thread is.
 use serde_json::{Value, json};
 use std::path::{Path, PathBuf};
 
@@ -184,6 +184,15 @@ pub fn is_note(block: &str) -> bool {
     block.starts_with("<runtime_info>")
 }
 
+/// The message T3 Code sends a thread when a task it delegated ends,
+/// which asks it to read the task's result with T3 Code's tools: a lead
+/// hears of its tasks from the lookout instead.
+pub fn is_completion(prompt: &str) -> bool {
+    let prompt = prompt.trim();
+    (prompt.starts_with("Delegated task ") && prompt.contains(" reached a terminal state."))
+        || (prompt.starts_with("Delegated tasks ") && prompt.contains(" reached terminal states."))
+}
+
 /// What the person wrote, out of a prompt T3 Code wrapped in its own
 /// instructions: those name tools the bot does not have.
 pub fn request(prompt: &str) -> &str {
@@ -330,6 +339,17 @@ mod tests {
         assert_eq!(request("Fix it."), "Fix it.");
     }
 
+    #[test]
+    fn a_delegated_tasks_end_is_t3_codes_message() {
+        assert!(is_completion(
+            "Delegated task node-1 reached a terminal state. Use task_status with taskId node-1 to read the result."
+        ));
+        assert!(is_completion(
+            "Delegated tasks node-1, node-2 reached terminal states. Use task_status with each taskId to read the results."
+        ));
+        assert!(!is_completion("Delegated task review: is it done?"));
+    }
+
     #[test]
     fn a_thread_claims_the_task_waiting_for_its_message_in_its_folder() {
         let home = std::env::temp_dir().join(format!("sonny-t3-{}", std::process::id()));

sonny/src/trigger.rs

@@ -2374,15 +2374,25 @@ pub fn restamp(places: &Places, identity: &str, lineage: &str, launchd: Loader)
             return;
         };
         // Read again under the lock: an `rm`, `add` or fire may have come first.
-        let path = places.plist(&trigger.name);
+        let path = places.job(&trigger.name);
         let Ok(text) = std::fs::read_to_string(&path) else {
             continue;
         };
-        let stamped = text.replacen(
-            &format!("<string>{}</string>", escape(identity)),
-            &format!("<string>{}</string>", escape(lineage)),
-            1,
-        );
+        let stamped = match is_plist(&text) {
+            true => text.replacen(
+                &format!("<string>{}</string>", escape(identity)),
+                &format!("<string>{}</string>", escape(lineage)),
+                1,
+            ),
+            false => {
+                let flag = quoted("--store-id", true);
+                text.replacen(
+                    &format!("{flag} {}", quoted(identity, true)),
+                    &format!("{flag} {}", quoted(lineage, true)),
+                    1,
+                )
+            }
+        };
         if stamped == text {
             continue;
         }
@@ -2398,14 +2408,11 @@ pub fn restamp(places: &Places, identity: &str, lineage: &str, launchd: Loader)
 
 /// Once Sonny knows its daemon: [`restamp`] the triggers made on its store.
 pub fn restamp_at_start(identity: Option<&str>, lineage: Option<&str>) {
-    if !cfg!(target_os = "macos") {
-        return;
-    }
     let (Ok(places), Some(identity), Some(lineage)) = (Places::home(), identity, lineage) else {
         return;
     };
     let (identity, lineage) = (identity.to_owned(), lineage.to_owned());
-    std::thread::spawn(move || restamp(&places, &identity, &lineage, &launchctl));
+    std::thread::spawn(move || restamp(&places, &identity, &lineage, &schedule));
 }
 
 /// What `trigger add` was asked.
@@ -4288,7 +4295,12 @@ mod tests {
 
     #[test]
     fn a_trigger_made_by_its_store_file_s_identity_names_its_lineage_instead() {
-        let w = World::new("restamp");
+        for w in [World::new("restamp"), World::systemd("restamp-unit")] {
+            restamps(w);
+        }
+    }
+
+    fn restamps(w: World) {
         let elsewhere = Trigger {
             name: "q.other".into(),
             store_id: "another-store".into(),