diff --git a/sonny/README.md b/sonny/README.md index c58e4bb83..b3804553f 100644 --- a/sonny/README.md +++ b/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 diff --git a/sonny/src/acp.rs b/sonny/src/acp.rs index f315a8597..493de34dc 100644 --- a/sonny/src/acp.rs +++ b/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, + /// 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, agent: &Arc, id: &Value, params: &Value) -> Reply { let (text, images) = words(¶ms["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 { diff --git a/sonny/src/t3.rs b/sonny/src/t3.rs index dfa7641c2..d9becaaa1 100644 --- a/sonny/src/t3.rs +++ b/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("") } +/// 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())); diff --git a/sonny/src/trigger.rs b/sonny/src/trigger.rs index 35ed6cacb..b6385e408 100644 --- a/sonny/src/trigger.rs +++ b/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!("{}", escape(identity)), - &format!("{}", escape(lineage)), - 1, - ); + let stamped = match is_plist(&text) { + true => text.replacen( + &format!("{}", escape(identity)), + &format!("{}", 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(), diff --git a/tests/test_acp.py b/tests/test_acp.py index ecd0d69ff..4d70c79ad 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -18,7 +18,9 @@ class T3Tools(http.server.BaseHTTPRequestHandler): - """T3 Code's MCP endpoint, as small as `t3_thread_launch` needs.""" + """T3 Code's MCP endpoint, as small as `delegate_task` and + `t3_thread_launch` need. `delegate_task` fails, as when the lead's + thread has no run on, while `server.delegating` is false.""" def log_message(self, *_): pass @@ -43,7 +45,12 @@ def do_POST(self): if 'id' not in message: return self.reply(202) self.server.calls.put(message['params']) + if message['params']['name'] == 'delegate_task' and not self.server.delegating: + refused = {'content': [{'type': 'text', 'text': 'parent_not_active'}], 'isError': True} + return self.reply(200, {'jsonrpc': '2.0', 'id': message['id'], 'result': refused}) result = {'threadId': 'thread-2', 'link': 't3://thread/thread-2', 'status': 'started'} + if message['params']['name'] == 'delegate_task': + result = {'taskId': 'node-1', 'childThreadId': 'thread-2', 'status': 'running'} self.reply(200, {'jsonrpc': '2.0', 'id': message['id'], 'result': {'content': [{'type': 'text', 'text': json.dumps(result)}], 'structuredContent': result}}) @@ -114,7 +121,7 @@ def setUp(self): self.daemon = SocketClient(self.binary, self.store, self.url, 'echo') self.addCleanup(self.daemon.close) self.t3 = http.server.ThreadingHTTPServer(('127.0.0.1', 0), T3Tools) - self.t3.calls, self.t3.headers = queue.Queue(), [] + self.t3.calls, self.t3.headers, self.t3.delegating = queue.Queue(), [], True threading.Thread(target=self.t3.serve_forever, daemon=True).start() self.addCleanup(self.t3.server_close) self.addCleanup(self.t3.shutdown) @@ -172,6 +179,8 @@ def test_a_thread_is_its_folders_lead_and_a_task_it_starts_opens_as_a_thread_of_ thread.send({'id': asked['id'], 'result': {'outcome': {'outcome': 'selected', 'optionId': 'allow'}}}) answered, updates = thread.answer_of(prompt) self.assertEqual(answered['result'], {'stopReason': 'end_turn'}) + # The task was T3 Code's child of this prompt before it was answered. + delegated = self.t3.calls.get_nowait() kinds = [u['sessionUpdate'] for u in updates] self.assertEqual(kinds[-1], 'state_update') self.assertEqual(updates[-1]['state'], 'idle') @@ -188,23 +197,18 @@ def test_a_thread_is_its_folders_lead_and_a_task_it_starts_opens_as_a_thread_of_ prompts = self.daemon.request('prompts', bot='calc.lead', turn=turn['turn'])['result']['prompts'] self.assertEqual(prompts[0]['text'], 'script') - # The task opens as a thread of its own, in its worktree, with its - # brief as the first message, through this thread's own tools. - launch = self.t3.calls.get(timeout=30) - self.assertEqual(launch['name'], 't3_thread_launch') - worktree = self.path / '.agent/worktrees/calc.fix' + # The task opens as the lead thread's child, with its brief as the + # first message, through this thread's own tools. + self.assertEqual(delegated['name'], 'delegate_task') turns = self.daemon.request('turns', bot='calc.fix')['result']['turns'] brief = self.daemon.request('prompts', bot='calc.fix', turn=turns[0]['turn'])['result']['prompts'][0]['text'] - self.assertEqual(launch['arguments']['title'], 'fix') - self.assertEqual(launch['arguments']['message'], brief) - self.assertEqual(launch['arguments']['workspaceStrategy'], - {'type': 'existing_worktree', 'worktreePath': str(worktree)}) + self.assertEqual(delegated['arguments'], {'task': brief, 'title': 'fix', 'clientRequestId': 'calc.fix'}) self.assertTrue(all(h.get('authorization') == 'Bearer thread-1' or h.get('Authorization') == 'Bearer thread-1' for h in self.t3.headers)) - # That thread is the task's, not a lead made for its folder: it - # answers with the task's work. - child, child_session = self.open(worktree) + # That thread, which T3 Code opens in the lead's folder, is the + # task's, not the folder's lead: it answers with the task's work. + child, child_session = self.open(repo) answered, updates = child.answer_of(child.ask('session/prompt', { 'sessionId': child_session, 'prompt': self.wrapped(brief)})) self.assertEqual(answered['result']['stopReason'], 'end_turn') @@ -213,6 +217,12 @@ def test_a_thread_is_its_folders_lead_and_a_task_it_starts_opens_as_a_thread_of_ self.assertEqual(kept.read_text(), 'calc.fix') self.assertEqual([b['name'] for b in self.daemon.request('bots')['result']['bots']], ['calc.fix', 'calc.lead']) + # T3 Code tells the lead's thread its child ended; the lead hears of + # its task from the lookout, so that message starts no turn. + answered, _ = thread.answer_of(thread.ask('session/prompt', {'sessionId': session, 'prompt': self.wrapped( + 'Delegated task node-1 reached a terminal state. Use task_status with taskId node-1 to read the result.')})) + self.assertEqual(answered['result'], {'stopReason': 'end_turn'}) + self.assertEqual(len(self.daemon.request('turns', bot='calc.lead')['result']['turns']), 1) def test_a_prompt_that_steers_is_answered_when_the_turn_it_joined_ends(self): repo = self.repo() @@ -220,6 +230,7 @@ def test_a_prompt_that_steers_is_answered_when_the_turn_it_joined_ends(self): instructions='Lead.', declare=[declared('start_task', inherit=False)], approve=['start_task'], approver='manual') thread, session = self.open(repo) + self.t3.delegating = False self.model.call_script = [('start_task', {'name': 'fix', 'brief': 'Fix calc.py.'})] first = thread.ask('session/prompt', {'sessionId': session, 'prompt': self.wrapped('script')}) [asked] = [m for m in thread.until(lambda m: m.get('method') == 'session/request_permission') @@ -232,10 +243,17 @@ def test_a_prompt_that_steers_is_answered_when_the_turn_it_joined_ends(self): thread.send({'id': asked['id'], 'result': {'outcome': {'outcome': 'selected', 'optionId': 'allow'}}}) answered, updates = thread.answer_of(second) self.assertEqual(answered['result'], {'stopReason': 'end_turn'}) - self.assertEqual(updates[-1]['state'], 'idle') + # Idle before the answer, which waited for the task's thread. + self.assertIn('idle', [u.get('state') for u in updates if u['sessionUpdate'] == 'state_update']) turns = self.daemon.request('turns', bot='calc.lead')['result']['turns'] self.assertEqual(sorted(t['status'] for t in turns), ['completed', 'steered']) - self.assertEqual(self.t3.calls.get(timeout=30)['name'], 't3_thread_launch') + # T3 Code would not take the task as a child: it opens as a thread of + # its own, in its worktree. + self.assertEqual(self.t3.calls.get(timeout=30)['name'], 'delegate_task') + launch = self.t3.calls.get(timeout=30) + self.assertEqual(launch['name'], 't3_thread_launch') + self.assertEqual(launch['arguments']['workspaceStrategy'], + {'type': 'existing_worktree', 'worktreePath': str(self.path / '.agent/worktrees/calc.fix')}) def test_a_turn_the_lead_takes_on_its_own_shows_in_the_thread(self): repo = self.repo()