diff --git a/CHANGELOG.md b/CHANGELOG.md index 134ad55..dab5b77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,9 @@ # Changelog ## Unreleased +- The host now closes a task after 24 hours with no connection, owned tab, or staged Commit. Previously a task stayed open until an agent finished or closed it explicitly, so every session that disconnected without finishing left an open task in `state.sqlite3` indefinitely. Tasks that own tabs are never closed this way, and a task that was connected when the host stopped counts its idle time from the restart. +- A connection acknowledgement that does not resume a supplied capability now carries `resume_rejection`: `task_closed` when the task was finished or closed, or `unknown_capability` otherwise. The host keeps closed tasks' capability hashes for a year to tell the two apart. On `task_closed`, the TypeScript and Python SDKs, and through them the MCP, OMP, Pi, and GPT-Control adapters, clear the stored capability and connect as a new task instead of failing with "rejected the stored resume capability" until the state was deleted by hand. Any other rejection still refuses to connect and keeps the stored capability. +- Added `clear()` to the Python SDK's resume capability store. `finish_task` already called it, so finishing a task with the built-in file store raised `AttributeError`. - Fixed a task tab with a "Leave site?" prompt stalling the extension. Chrome's `tabs.remove` waits until the tab is gone, and a page's beforeunload prompt, which a background tab shows only once activated, can hold the tab open indefinitely. `browser_finish` and task close waited on that removal inside the ownership queue, so every later ownership change, `browser_open`, `browser_tabs`, tab event, and native reconnect stalled behind it until the extension was reloaded. Task cleanup still removes ownership before closing tabs, now waits at most two seconds for Chrome to close them, and leaves a tab still waiting on its prompt out of both `closed_tab_ids` and `retained_tab_ids`; the tab closes if the user confirms leaving. A `browser_act` `close` waits the same two seconds and then returns `close_pending: true` instead of holding its request open. - No single extension operation can stall the operations queued behind it any more. An ownership change or attention-notice change still running after 10 seconds, or a background tab or alarm event after 30 seconds, keeps running and is logged, but the next operation starts. Ten seconds covers the two-second tab removal wait with room for normal Chrome calls and stays well under the host's 30-second command deadline, so a command queued behind a stuck operation can still answer. - `browser_finish` and task close now report an extension that does not answer in time as `extension_timeout`, telling the agent that the task is still open and to reload the extension if it stays unresponsive, then finish or close again. They previously reported `state_store_error` and suggested repairing `~/.agenttab` or freeing disk space. When the extension refuses cleanup, such as for a `keep_tab_ids` entry the task does not own, the error now carries the extension's own code and message instead of `native_protocol_error`. Disconnects and protocol or transport failures during cleanup keep their own error codes too. diff --git a/docs/multi-agent.md b/docs/multi-agent.md index 05bb561..7fd95e9 100644 --- a/docs/multi-agent.md +++ b/docs/multi-agent.md @@ -10,9 +10,11 @@ A fresh adapter connection therefore gets a distinct task. Adapters that need to ## Resume capabilities -A resume capability is a 256-bit random bearer secret. The host persists only its hash. A client must durably persist and confirm an initial capability before the host accepts another RPC. On successful resume the host returns a replacement capability under the same persist-then-confirm barrier; if delivery or confirmation fails, a durably retained capability remains recoverable. Once a replacement is confirmed, the old capability no longer resumes the task. +A resume capability is a 256-bit random bearer secret. The host persists only its hash. A client must durably persist and confirm an initial capability before the host accepts another RPC. On successful resume the host returns a replacement capability under the same persist-then-confirm barrier; if delivery or confirmation fails, a durably retained capability remains recoverable. Once a replacement is confirmed, the old capability no longer resumes the task. A capability also stops resuming its task after the task is finished or closed, or after 24 hours with no connection, owned tab, or staged Commit. -Store the capability only in adapter-owned private state. Never put it in prompts, page content, logs, task titles, or shared configuration. A missing or invalid capability creates a new task rather than recovering another agent's task. +When a supplied capability does not resume, the host answers `resumed: false` with a `resume_rejection` reason and closes the connection. `task_closed` means the task was finished or closed, including for inactivity; the host remembers this for a year after the close. `unknown_capability` covers anything else, such as a superseded capability or one from a lost journal. The TypeScript and Python SDKs clear their stored capability and reconnect as a new task only for `task_closed`. For any other reason they refuse to connect and keep the stored capability, so a task is never abandoned silently. + +Store the capability only in adapter-owned private state. Never put it in prompts, page content, logs, task titles, or shared configuration. A connection without a capability gets a new task rather than recovering another agent's task. ## Task-owned tabs and visible groups diff --git a/docs/rust-host.md b/docs/rust-host.md index f25f23b..c8c90da 100644 --- a/docs/rust-host.md +++ b/docs/rust-host.md @@ -43,6 +43,8 @@ By default, Unix state lives under `$HOME/.agenttab`; Windows uses `%LOCALAPPDAT `state.sqlite3` uses WAL, full synchronous writes, foreign keys, and a busy timeout. It stores only hashes of resume capabilities and staged tokens. It also stores task ownership, monotonic page-revision floors, native-event receipts, staged Commit bindings, and idempotency entries. +A task stays resumable after its last connection detaches. The host closes a task once it has had no connection, owned tab, or staged Commit for 24 hours, checking at startup, before each mutation, and before each resume; a task still connected when the host stopped counts its idle time from the restart. Closed task rows are deleted 24 hours after they close, but their capability hashes are kept for a year so that resuming one is rejected with `resume_rejection: "task_closed"` rather than `unknown_capability`. See [multi-agent.md](multi-agent.md#resume-capabilities) for how clients react. + Mutation idempotency is keyed by task and UUIDv7 key with a canonical method/parameter hash. The host records `started` before native dispatch and a terminal response after completion. A matching completed record replays the cached response. A durable started record after a crash returns `unknown` and is never re-executed. Terminal records are retained for seven days, with at most 10,000 records per task. `audit.jsonl` is a separate local audit record. It contains structural summaries and digests rather than raw request and result values. See [security.md](security.md) for the precise privacy implications. diff --git a/host-rs/crates/agenttab-host/src/journal.rs b/host-rs/crates/agenttab-host/src/journal.rs index 1ba0f35..29a61f8 100644 --- a/host-rs/crates/agenttab-host/src/journal.rs +++ b/host-rs/crates/agenttab-host/src/journal.rs @@ -1,4 +1,6 @@ -use agenttab_protocol::{NativeStagedCommit, NativeTab, Outcome, RpcError, RpcMethod, RpcResponse}; +use agenttab_protocol::{ + NativeStagedCommit, NativeTab, Outcome, ResumeRejection, RpcError, RpcMethod, RpcResponse, +}; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use base64::Engine; use parking_lot::Mutex; @@ -17,6 +19,13 @@ const IDEMPOTENCY_RETENTION_MS: i64 = 7 * 24 * 60 * 60 * 1000; const IDEMPOTENCY_FUTURE_SKEW_MS: i64 = 5 * 60 * 1000; const MAX_RECORDS_PER_TASK: i64 = 10_000; const CLOSED_TASK_RETENTION_MS: i64 = 24 * 60 * 60 * 1000; +/// A task with no connection, tabs, or staged Commit that has been untouched this long +/// is closed: resuming it could only recover an empty task, and its capability then +/// reports `task_closed` so the client starts a new task. +const IDLE_TASK_RETENTION_MS: i64 = 24 * 60 * 60 * 1000; +/// How long a deleted closed task's capability hashes still report `task_closed` on resume, +/// letting a client that returns late start a new task instead of failing. +const CLOSED_CAPABILITY_RETENTION_MS: i64 = 365 * 24 * 60 * 60 * 1000; const MAX_STAGE_LIFETIME_MS: i64 = 5 * 60 * 1000; #[derive(Debug, Error)] @@ -237,7 +246,11 @@ impl Journal { FOREIGN KEY (task_id) REFERENCES tasks(task_id) ON DELETE CASCADE ); CREATE UNIQUE INDEX IF NOT EXISTS idx_staged_commits_native_token - ON staged_commits(native_token);", + ON staged_commits(native_token); + CREATE TABLE IF NOT EXISTS closed_task_capabilities ( + capability_hash BLOB PRIMARY KEY, + closed_at_ms INTEGER NOT NULL + );", )?; ensure_column( &connection, @@ -281,11 +294,20 @@ impl Journal { ON staged_commits(review_handle_hash)", [], )?; + let now = now_ms(); + // Connections held when the host stopped never detached; count their idle time + // from this restart, the latest moment they are known to have been connected. connection.execute( - "UPDATE tasks SET active_connections = 0, pending_resume_hash = NULL", - [], + "UPDATE tasks + SET updated_at_ms = CASE + WHEN active_connections > 0 OR pending_resume_hash IS NOT NULL THEN ?1 + ELSE updated_at_ms + END, + active_connections = 0, + pending_resume_hash = NULL", + params![now], )?; - cleanup_expired_tasks(&connection, now_ms())?; + cleanup_expired_tasks(&connection, now)?; #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; @@ -326,6 +348,8 @@ impl Journal { let next_hash = capability_hash(&next_capability); let now = now_ms(); let mut connection = self.connection.lock(); + // Expire first so an idle task cannot resume merely because no mutation ran lately. + cleanup_expired_tasks(&connection, now)?; let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let task_id: Option = transaction .query_row( @@ -361,6 +385,28 @@ impl Journal { })) } + /// Classifies a capability that `resume_task` did not resume. + pub fn resume_rejection(&self, capability: &str) -> Result { + let hash = capability_hash(capability); + let connection = self.connection.lock(); + let closed: Option = connection + .query_row( + "SELECT 1 FROM tasks + WHERE state = 'closed' AND (resume_hash = ?1 OR pending_resume_hash = ?1) + UNION ALL + SELECT 1 FROM closed_task_capabilities WHERE capability_hash = ?1 + LIMIT 1", + params![hash.as_slice()], + |row| row.get(0), + ) + .optional()?; + Ok(if closed.is_some() { + ResumeRejection::TaskClosed + } else { + ResumeRejection::UnknownCapability + }) + } + pub fn acknowledge_resume_capability( &self, task_id: Uuid, @@ -1266,10 +1312,41 @@ fn ensure_column( } fn cleanup_expired_tasks(connection: &Connection, now: i64) -> Result<(), JournalError> { + connection.execute( + "UPDATE tasks + SET state = 'closed', updated_at_ms = ?1, closed_at_ms = ?1 + WHERE state = 'active' + AND active_connections = 0 + AND pending_resume_hash IS NULL + AND tab_count = 0 + AND updated_at_ms < ?2 + AND NOT EXISTS (SELECT 1 FROM task_tabs WHERE task_tabs.task_id = tasks.task_id) + AND NOT EXISTS ( + SELECT 1 FROM staged_commits WHERE staged_commits.task_id = tasks.task_id + )", + params![now, now - IDLE_TASK_RETENTION_MS], + )?; + let expired_before = now - CLOSED_TASK_RETENTION_MS; + // Insert the tombstones before deleting the rows so a crash between the two statements + // never loses the closed status of a capability. + connection.execute( + "INSERT OR IGNORE INTO closed_task_capabilities(capability_hash, closed_at_ms) + SELECT resume_hash, closed_at_ms FROM tasks + WHERE state = 'closed' AND closed_at_ms IS NOT NULL AND closed_at_ms < ?1 + UNION ALL + SELECT pending_resume_hash, closed_at_ms FROM tasks + WHERE state = 'closed' AND closed_at_ms IS NOT NULL AND closed_at_ms < ?1 + AND pending_resume_hash IS NOT NULL", + params![expired_before], + )?; connection.execute( "DELETE FROM tasks WHERE state = 'closed' AND closed_at_ms IS NOT NULL AND closed_at_ms < ?1", - params![now - CLOSED_TASK_RETENTION_MS], + params![expired_before], + )?; + connection.execute( + "DELETE FROM closed_task_capabilities WHERE closed_at_ms < ?1", + params![now - CLOSED_CAPABILITY_RETENTION_MS], )?; Ok(()) } @@ -1444,6 +1521,31 @@ mod tests { .is_none()); } + #[test] + fn idle_task_expires_on_resume_without_an_intervening_mutation() { + let temp = tempfile::tempdir().unwrap(); + let journal = open_journal(&temp); + let task = journal.create_task(None).unwrap(); + journal.detach_connection(task.task_id).unwrap(); + journal + .connection + .lock() + .execute( + "UPDATE tasks SET updated_at_ms = ?1", + params![now_ms() - IDLE_TASK_RETENTION_MS - 1], + ) + .unwrap(); + + assert!(journal + .resume_task(&task.resume_capability) + .unwrap() + .is_none()); + assert_eq!( + journal.resume_rejection(&task.resume_capability).unwrap(), + ResumeRejection::TaskClosed + ); + } + #[test] fn host_restart_discards_an_unconfirmed_candidate_and_preserves_the_old_capability() { let temp = tempfile::tempdir().unwrap(); @@ -1475,7 +1577,7 @@ mod tests { } #[test] - fn disconnected_task_remains_resumable_until_explicit_close() { + fn disconnected_task_remains_resumable_within_idle_retention() { let temp = tempfile::tempdir().unwrap(); let journal = open_journal(&temp); let task = journal.create_task(None).unwrap(); @@ -1487,6 +1589,139 @@ mod tests { assert_eq!(resumed.task_id, task.task_id); } + #[test] + fn idle_disconnected_task_without_tabs_closes_after_retention() { + let temp = tempfile::tempdir().unwrap(); + let journal = open_journal(&temp); + let idle = journal.create_task(None).unwrap(); + journal.detach_connection(idle.task_id).unwrap(); + let recent = journal.create_task(None).unwrap(); + journal.detach_connection(recent.task_id).unwrap(); + let connected = journal.create_task(None).unwrap(); + let with_tab = journal.create_task(None).unwrap(); + journal + .reconcile_inventory(&[owned_tab(with_tab.task_id, 1)]) + .unwrap(); + journal.detach_connection(with_tab.task_id).unwrap(); + let stale = now_ms() - IDLE_TASK_RETENTION_MS - 1; + for task_id in [idle.task_id, connected.task_id, with_tab.task_id] { + journal + .connection + .lock() + .execute( + "UPDATE tasks SET updated_at_ms = ?1 WHERE task_id = ?2", + params![stale, task_id.to_string()], + ) + .unwrap(); + } + + journal + .begin_mutation(recent.task_id, Uuid::now_v7(), RpcMethod::BrowserAct, "a") + .unwrap(); + + assert!(journal + .resume_task(&idle.resume_capability) + .unwrap() + .is_none()); + assert_eq!( + journal.resume_rejection(&idle.resume_capability).unwrap(), + ResumeRejection::TaskClosed + ); + for (task, label) in [ + (&recent, "recently detached"), + (&connected, "connected"), + (&with_tab, "tab-owning"), + ] { + assert_eq!( + journal + .resume_task(&task.resume_capability) + .unwrap() + .unwrap_or_else(|| panic!("{label} task stays resumable")) + .task_id, + task.task_id + ); + } + } + + #[test] + fn host_restart_restarts_idle_clock_for_tasks_connected_at_shutdown() { + let temp = tempfile::tempdir().unwrap(); + let state_path = temp.path().join("state.sqlite3"); + let stale = now_ms() - IDLE_TASK_RETENTION_MS - 1; + let (connected, detached) = { + let journal = Journal::open(&state_path).unwrap(); + let connected = journal.create_task(None).unwrap(); + let detached = journal.create_task(None).unwrap(); + journal.detach_connection(detached.task_id).unwrap(); + journal + .connection + .lock() + .execute("UPDATE tasks SET updated_at_ms = ?1", params![stale]) + .unwrap(); + (connected, detached) + }; + + let restarted = Journal::open(&state_path).unwrap(); + assert!(restarted + .resume_task(&detached.resume_capability) + .unwrap() + .is_none()); + assert_eq!( + restarted + .resume_task(&connected.resume_capability) + .unwrap() + .expect("a task connected at shutdown stays resumable") + .task_id, + connected.task_id + ); + } + + #[test] + fn deleted_closed_task_capability_reports_closed_until_tombstone_expires() { + let temp = tempfile::tempdir().unwrap(); + let state_path = temp.path().join("state.sqlite3"); + let (recent, ancient) = { + let journal = Journal::open(&state_path).unwrap(); + let recent = journal.create_task(None).unwrap(); + let ancient = journal.create_task(None).unwrap(); + for (task, closed_at) in [ + (&recent, now_ms() - CLOSED_TASK_RETENTION_MS - 1), + (&ancient, now_ms() - CLOSED_CAPABILITY_RETENTION_MS - 1), + ] { + journal.close_task(task.task_id).unwrap(); + journal + .connection + .lock() + .execute( + "UPDATE tasks SET closed_at_ms = ?1 WHERE task_id = ?2", + params![closed_at, task.task_id.to_string()], + ) + .unwrap(); + } + (recent, ancient) + }; + + let restarted = Journal::open(&state_path).unwrap(); + let remaining: i64 = restarted + .connection + .lock() + .query_row("SELECT COUNT(*) FROM tasks", [], |row| row.get(0)) + .unwrap(); + assert_eq!(remaining, 0, "closed task rows are deleted after retention"); + assert_eq!( + restarted + .resume_rejection(&recent.resume_capability) + .unwrap(), + ResumeRejection::TaskClosed + ); + assert_eq!( + restarted + .resume_rejection(&ancient.resume_capability) + .unwrap(), + ResumeRejection::UnknownCapability + ); + } + #[test] fn completed_mutation_is_replayed_and_input_conflicts_fail() { let temp = tempfile::tempdir().unwrap(); diff --git a/host-rs/crates/agenttab-host/src/task.rs b/host-rs/crates/agenttab-host/src/task.rs index 2b4ff2a..d03760a 100644 --- a/host-rs/crates/agenttab-host/src/task.rs +++ b/host-rs/crates/agenttab-host/src/task.rs @@ -32,9 +32,12 @@ impl ConnectionContext { journal: &Arc, runtime_state: RuntimeState, ) -> Result<(Arc, ConnectionAck), JournalError> { - let resumed_lease = match init.resume_capability.as_deref() { - Some(capability) => journal.resume_task(capability)?, - None => None, + let (resumed_lease, resume_rejection) = match init.resume_capability.as_deref() { + Some(capability) => match journal.resume_task(capability)? { + Some(lease) => (Some(lease), None), + None => (None, Some(journal.resume_rejection(capability)?)), + }, + None => (None, None), }; let resumed = resumed_lease.is_some(); let ack = ConnectionAck { @@ -47,6 +50,7 @@ impl ConnectionContext { resume_capability: resumed_lease .as_ref() .map(|lease| lease.resume_capability.clone()), + resume_rejection, state: runtime_state, }; let context = Arc::new(Self { @@ -175,7 +179,9 @@ impl ConnectionContext { #[cfg(test)] mod tests { use super::*; - use agenttab_protocol::{ConnectKind, ConnectionInit, ResumeCapabilityConfirmKind}; + use agenttab_protocol::{ + ConnectKind, ConnectionInit, ResumeCapabilityConfirmKind, ResumeRejection, + }; fn init(capability: Option) -> ConnectionInit { ConnectionInit { @@ -304,4 +310,31 @@ mod tests { .unwrap() .is_some()); } + + #[test] + fn rejected_resume_reports_whether_the_task_was_closed() { + let temp = tempfile::tempdir().unwrap(); + let journal = Arc::new(Journal::open(&temp.path().join("state.sqlite3")).unwrap()); + let closed = journal.create_task(None).unwrap(); + journal.close_task(closed.task_id).unwrap(); + let live = journal.create_task(None).unwrap(); + let negotiate = |capability: Option| { + ConnectionContext::negotiate(init(capability), &journal, RuntimeState::Ready) + .unwrap() + .1 + }; + + let ack = negotiate(Some(closed.resume_capability)); + assert!(!ack.resumed); + assert_eq!(ack.resume_rejection, Some(ResumeRejection::TaskClosed)); + let ack = negotiate(Some("x".repeat(43))); + assert_eq!( + ack.resume_rejection, + Some(ResumeRejection::UnknownCapability) + ); + assert_eq!(negotiate(None).resume_rejection, None); + let ack = negotiate(Some(live.resume_capability)); + assert!(ack.resumed); + assert_eq!(ack.resume_rejection, None); + } } diff --git a/host-rs/crates/agenttab-protocol/src/lib.rs b/host-rs/crates/agenttab-protocol/src/lib.rs index 862001a..91d752c 100644 --- a/host-rs/crates/agenttab-protocol/src/lib.rs +++ b/host-rs/crates/agenttab-protocol/src/lib.rs @@ -1988,6 +1988,10 @@ pub struct ConnectionAck { pub task_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub resume_capability: Option, + /// Why a supplied resume capability did not resume; absent when none was supplied + /// or the task resumed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resume_rejection: Option, pub state: RuntimeState, } @@ -1997,6 +2001,17 @@ impl ConnectionAck { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ResumeRejection { + /// The capability's task was finished, closed, or closed for inactivity; starting a new + /// task loses nothing. + TaskClosed, + /// The capability is not known to this host, such as a superseded capability or one + /// from a lost journal. + UnknownCapability, +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ConnectedKind { diff --git a/packages/sdk-python/agenttab/client.py b/packages/sdk-python/agenttab/client.py index 9678091..4b2a987 100644 --- a/packages/sdk-python/agenttab/client.py +++ b/packages/sdk-python/agenttab/client.py @@ -561,6 +561,9 @@ def prepare_replacement(self, current_capability: str, replacement_capability: s def activate_replacement(self, replacement_capability: str) -> None: ... + def clear(self) -> None: + ... + def _validate_resume_capability(capability: str) -> None: if not 32 <= len(capability) <= 64: @@ -724,6 +727,18 @@ def activate_replacement(self, replacement_capability: str) -> None: {"schemaVersion": 1, "resumeCapability": replacement_capability} ) + def clear(self) -> None: + directory = self.path.parent + if not _prepare_capability_directory(directory, create=False): + return + self.path.unlink(missing_ok=True) + if os.name != "nt": + descriptor = os.open(directory, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + def create_resume_capability_store( namespace: str, @@ -805,6 +820,21 @@ def connect( ) if capability is not None and connection.get("resumed") is not True: stream.close() + # A pending candidate is a rotation of the active capability's task, so + # either one reporting the task closed means nothing remains to resume. + if ( + capability_store is not None + and connection.get("resume_rejection") == "task_closed" + ): + capability_store.clear() + negotiated = cls._negotiate_connection( + address, + conversation_id=conversation_id, + resume_capability=None, + connect_timeout=connect_timeout, + request_timeout=request_timeout, + ) + break if capability_store is not None and capability != active_capability: continue source = "stored" if capability_store is not None else "supplied" @@ -919,6 +949,10 @@ def _negotiate_connection( or connection.get("kind") != "connected" ): raise RuntimeError("AgentTab returned an invalid connection acknowledgement") + if "resume_rejection" in connection and not isinstance( + connection["resume_rejection"], str + ): + raise RuntimeError("AgentTab returned an invalid connection acknowledgement") if connection.get("resumed") and ( not isinstance(connection.get("connection_id"), str) or not isinstance(connection.get("task_id"), str) diff --git a/packages/sdk-python/tests/test_client.py b/packages/sdk-python/tests/test_client.py index 4d5133e..05ab11a 100644 --- a/packages/sdk-python/tests/test_client.py +++ b/packages/sdk-python/tests/test_client.py @@ -762,6 +762,9 @@ def prepare_replacement(self, _current: str, _replacement: str) -> None: def activate_replacement(self, _replacement: str) -> None: return None + def clear(self) -> None: + self.capability = None + with tempfile.TemporaryDirectory() as root: endpoint = str(Path(root) / "agenttab.sock") server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) @@ -947,6 +950,244 @@ def serve() -> None: worker.join(timeout=2) server.close() + def test_closed_task_capability_is_cleared_and_connect_starts_a_new_task( + self, + ) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + store = create_resume_capability_store( + "python-sdk", + scope="closed-task", + state_dir=root, + ) + active = "a" * 32 + issued = "c" * 32 + store.save(active) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(2) + attempted: list[object] = [] + stored_at_renegotiation: list[str | None] = [] + + def serve() -> None: + rejected, _ = server.accept() + try: + hello = read_frame(rejected) + attempted.append(hello.get("resume_capability")) + rejected.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da41", + "resumed": False, + "resume_rejection": "task_closed", + "state": "ready", + }, 1024 * 1024)) + finally: + rejected.close() + + fresh, _ = server.accept() + try: + hello = read_frame(fresh) + attempted.append(hello.get("resume_capability")) + stored_at_renegotiation.append(store.load()) + fresh.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da42", + "resumed": False, + "state": "ready", + }, 1024 * 1024)) + request = read_frame(fresh) + fresh.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "request_id": request["request_id"], + "ok": True, + "outcome": "completed", + "result": {"created": True}, + "task": { + "task_id": "018f22b2-4126-7c1a-8c31-3f45a783da44", + "resume_capability": issued, + }, + }, 1024 * 1024)) + confirmation = read_frame(fresh) + self.assertEqual(confirmation["kind"], "resume_confirm") + self.assertEqual(confirmation["resume_capability"], issued) + fresh.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "resume_confirmed", + "connection_id": confirmation["connection_id"], + }, 1024 * 1024)) + finally: + fresh.close() + + worker = threading.Thread(target=serve) + worker.start() + client = AgentTabClient.connect(endpoint=endpoint, capability_store=store) + self.assertFalse(client.connection["resumed"]) + self.assertIsNone(client.resume_capability) + self.assertIsNone(client.pending_resume_capability) + self.assertEqual(attempted, [active, None]) + self.assertEqual(stored_at_renegotiation, [None]) + self.assertFalse(store.path.exists()) + + self.assertEqual( + client.call("browser_open", {"mode": "create"}), + {"created": True}, + ) + self.assertEqual(client.resume_capability, issued) + self.assertEqual(store.load(), issued) + self.assertIsNone(store.load_pending()) + client.close() + worker.join(timeout=2) + server.close() + + def test_pending_capability_task_closed_starts_a_new_task(self) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + store = create_resume_capability_store( + "python-sdk", + scope="closed-pending", + state_dir=root, + ) + rotated_away = "a" * 32 + confirmed_pending = "c" * 32 + store.save(rotated_away) + store.prepare_replacement(rotated_away, confirmed_pending) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(2) + attempted: list[object] = [] + ack = { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da51", + "resumed": False, + "state": "ready", + } + accepted: list[socket.socket] = [] + + def serve() -> None: + rejected, _ = server.accept() + try: + attempted.append(read_frame(rejected).get("resume_capability")) + rejected.sendall( + encode_frame(dict(ack, resume_rejection="task_closed"), 1024 * 1024) + ) + finally: + rejected.close() + fresh, _ = server.accept() + accepted.append(fresh) + attempted.append(read_frame(fresh).get("resume_capability")) + fresh.sendall(encode_frame(ack, 1024 * 1024)) + + worker = threading.Thread(target=serve) + worker.start() + client = AgentTabClient.connect(endpoint=endpoint, capability_store=store) + worker.join(timeout=2) + self.assertFalse(client.connection["resumed"]) + self.assertEqual(attempted, [confirmed_pending, None]) + self.assertIsNone(store.load()) + self.assertIsNone(store.load_pending()) + client.close() + for connection in accepted: + connection.close() + server.close() + + def test_stored_capability_rejection_other_than_task_closed_keeps_the_store( + self, + ) -> None: + for rejection in ("unknown_capability", None, "future_reason"): + with self.subTest(resume_rejection=rejection), tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + store = create_resume_capability_store( + "python-sdk", + scope="rejected-capability", + state_dir=root, + ) + active = "a" * 32 + store.save(active) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(1) + acknowledgement: dict[str, object] = { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da42", + "resumed": False, + "state": "ready", + } + if rejection is not None: + acknowledgement["resume_rejection"] = rejection + + def serve() -> None: + connection, _ = server.accept() + try: + hello = read_frame(connection) + self.assertEqual(hello["resume_capability"], active) + connection.sendall(encode_frame(acknowledgement, 1024 * 1024)) + finally: + connection.close() + + worker = threading.Thread(target=serve) + worker.start() + with self.assertRaisesRegex( + RuntimeError, + "rejected the stored resume capability", + ): + AgentTabClient.connect( + endpoint=endpoint, + capability_store=store, + connect_timeout=1, + ) + worker.join(timeout=2) + server.close() + self.assertEqual(store.load(), active) + self.assertIsNone(store.load_pending()) + + def test_non_string_resume_rejection_invalidates_the_acknowledgement(self) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + store = create_resume_capability_store( + "python-sdk", + scope="invalid-rejection", + state_dir=root, + ) + active = "a" * 32 + store.save(active) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(1) + + def serve() -> None: + connection, _ = server.accept() + try: + read_frame(connection) + connection.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da42", + "resumed": False, + "resume_rejection": 1, + "state": "ready", + }, 1024 * 1024)) + finally: + connection.close() + + worker = threading.Thread(target=serve) + worker.start() + with self.assertRaisesRegex(RuntimeError, "invalid connection acknowledgement"): + AgentTabClient.connect(endpoint=endpoint, capability_store=store) + worker.join(timeout=2) + server.close() + self.assertEqual(store.load(), active) + def test_explicit_memory_only_resume_is_rejected_when_a_store_is_configured( self, diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index fd3f79d..76829b2 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -395,6 +395,7 @@ export interface ConnectionAck { task_id?: string; resume_capability?: string; state?: "starting" | "reconciling" | "ready" | "paused" | "terminal"; + resume_rejection?: "task_closed" | "unknown_capability"; } export type Outcome = "completed" | "not_started" | "unknown" | "needs_user" | "commit_required"; @@ -1018,6 +1019,13 @@ export class AgentTabClient { const attempt = await negotiateConnection(endpoint, timeoutMs, options.conversationId, capability); if (capability && !attempt.connected.resumed) { attempt.socket.destroy(); + // A pending candidate is a rotation of the active capability's task, so either one + // reporting the task closed means nothing remains to resume. + if (attempt.connected.resume_rejection === "task_closed") { + await store?.clear(); + negotiated = await negotiateConnection(endpoint, timeoutMs, options.conversationId, undefined); + break; + } if (capability !== activeCapability) continue; throw new Error( "AgentTab rejected the stored resume capability; remove that client state only when starting a new task is intended", @@ -1325,6 +1333,7 @@ function isConnectionAck(value: unknown): value is ConnectionAck { value.kind === "connected" && typeof value.connection_id === "string" && typeof value.resumed === "boolean" && + (value.resume_rejection === undefined || typeof value.resume_rejection === "string") && (!value.resumed || (typeof value.task_id === "string" && typeof value.resume_capability === "string")) ); diff --git a/packages/sdk-typescript/test/client.test.ts b/packages/sdk-typescript/test/client.test.ts index ac6ca23..8e6869c 100644 --- a/packages/sdk-typescript/test/client.test.ts +++ b/packages/sdk-typescript/test/client.test.ts @@ -474,6 +474,204 @@ test("does not create a new task when a stored resume capability is rejected", a expect(connections).toBe(1); }); +test("starts a fresh task when the host reports the stored capability's task closed", async () => { + const stateDir = mkdtempSync(join(tmpdir(), "agenttab-sdk-state-")); + roots.push(stateDir); + const store = createResumeCapabilityStore("mcp", { scope: "closed-task", stateDir }); + const closedCapability = "a".repeat(32); + const freshCapability = "b".repeat(32); + await store.save(closedCapability); + const connects: Array> = []; + let clearedBeforeFreshConnect = false; + let confirmed = false; + + const endpoint = await listen((socket) => { + socket.on("error", () => {}); + const decoder = new FrameDecoder(1024 * 1024); + socket.on("data", async (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind === "connect") { + connects.push(value); + if (value.resume_capability !== undefined) { + socket.end(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da50", + resumed: false, + state: "ready", + resume_rejection: "task_closed", + }, 1024 * 1024)); + continue; + } + clearedBeforeFreshConnect = await store.load() === undefined; + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da51", + resumed: false, + state: "ready", + }, 1024 * 1024)); + } else if (value.kind === "resume_confirm") { + confirmed = value.resume_capability === freshCapability; + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "resume_confirmed", + connection_id: value.connection_id, + }, 1024 * 1024)); + } else { + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + request_id: value.request_id, + ok: true, + outcome: "completed", + result: { task: "created" }, + task: { + task_id: "018f22b2-4126-7c1a-8c31-3f45a783da52", + resume_capability: freshCapability, + }, + }, 1024 * 1024)); + } + } + }); + }); + + const client = await AgentTabClient.connect({ endpoint, capabilityStore: store }); + expect(client.connection.resumed).toBe(false); + expect(client.connection.connection_id).toBe("018f22b2-4126-7c1a-8c31-3f45a783da51"); + expect(connects).toHaveLength(2); + expect(connects[0].resume_capability).toBe(closedCapability); + expect("resume_capability" in connects[1]).toBe(false); + expect(clearedBeforeFreshConnect).toBe(true); + + await expect(client.call("browser_open", { mode: "create" })).resolves.toEqual({ task: "created" }); + expect(confirmed).toBe(true); + expect(client.resumeCapability).toBe(freshCapability); + expect(await store.load()).toBe(freshCapability); + client.close(); +}); + +test("starts a fresh task when the pending capability's task closed after an interrupted rotation", async () => { + const stateDir = mkdtempSync(join(tmpdir(), "agenttab-sdk-state-")); + roots.push(stateDir); + const store = createResumeCapabilityStore("mcp", { scope: "closed-pending", stateDir }); + const rotatedAway = "a".repeat(32); + const confirmedPending = "c".repeat(32); + await store.save(rotatedAway); + await store.prepareReplacement(rotatedAway, confirmedPending); + const connects: Array> = []; + + const endpoint = await listen((socket) => { + socket.on("error", () => {}); + const decoder = new FrameDecoder(1024 * 1024); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind !== "connect") continue; + connects.push(value); + const ack = { + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: `018f22b2-4126-7c1a-8c31-3f45a783da6${connects.length}`, + resumed: false, + state: "ready", + }; + if (value.resume_capability === confirmedPending) { + socket.end(encodeFrame({ ...ack, resume_rejection: "task_closed" }, 1024 * 1024)); + } else if (value.resume_capability !== undefined) { + socket.end(encodeFrame({ ...ack, resume_rejection: "unknown_capability" }, 1024 * 1024)); + } else { + socket.write(encodeFrame(ack, 1024 * 1024)); + } + } + }); + }); + + const client = await AgentTabClient.connect({ endpoint, capabilityStore: store }); + expect(client.connection.resumed).toBe(false); + expect(connects.map((connect) => connect.resume_capability)).toEqual([confirmedPending, undefined]); + expect(await store.load()).toBeUndefined(); + expect(await store.loadPending()).toBeUndefined(); + client.close(); +}); + +for (const [label, rejection] of [ + ["unknown_capability", { resume_rejection: "unknown_capability" }], + ["an absent rejection from an older host", {}], + ["an unrecognized rejection value", { resume_rejection: "future_reason" }], +] as const) { + test(`refuses a fresh task and keeps the stored capability for ${label}`, async () => { + const stateDir = mkdtempSync(join(tmpdir(), "agenttab-sdk-state-")); + roots.push(stateDir); + const store = createResumeCapabilityStore("mcp", { scope: "rejected-capability", stateDir }); + const capability = "a".repeat(32); + await store.save(capability); + let connections = 0; + const endpoint = await listen((socket) => { + connections += 1; + socket.on("error", () => {}); + const decoder = new FrameDecoder(1024 * 1024); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind !== "connect") continue; + socket.end(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da53", + resumed: false, + state: "ready", + ...rejection, + }, 1024 * 1024)); + } + }); + }); + + await expect(AgentTabClient.connect({ endpoint, capabilityStore: store })).rejects.toThrow( + "AgentTab rejected the stored resume capability; remove that client state only when starting a new task is intended", + ); + expect(connections).toBe(1); + expect(await store.load()).toBe(capability); + }); +} + +test("treats a non-string resume_rejection as an invalid connection acknowledgement", async () => { + const stateDir = mkdtempSync(join(tmpdir(), "agenttab-sdk-state-")); + roots.push(stateDir); + const store = createResumeCapabilityStore("mcp", { scope: "invalid-rejection", stateDir }); + const capability = "a".repeat(32); + await store.save(capability); + let connections = 0; + const endpoint = await listen((socket) => { + connections += 1; + socket.on("error", () => {}); + const decoder = new FrameDecoder(1024 * 1024); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind !== "connect") continue; + socket.end(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da54", + resumed: false, + state: "ready", + resume_rejection: 1, + }, 1024 * 1024)); + } + }); + }); + + await expect(AgentTabClient.connect({ endpoint, capabilityStore: store })).rejects.toThrow( + "AgentTab sent a response before connection negotiation completed", + ); + expect(connections).toBe(1); + expect(await store.load()).toBe(capability); +}); + test("retries the active capability after a pending replacement is rejected", async () => { const activeCapability = "a".repeat(32); const pendingCapability = "b".repeat(32); diff --git a/schemas/rpc/v1/connection.schema.json b/schemas/rpc/v1/connection.schema.json index 553ac5d..a64178d 100644 --- a/schemas/rpc/v1/connection.schema.json +++ b/schemas/rpc/v1/connection.schema.json @@ -26,6 +26,7 @@ "resumed": { "type": "boolean" }, "task_id": { "type": "string", "format": "uuid" }, "resume_capability": { "type": "string", "minLength": 32, "maxLength": 64 }, + "resume_rejection": { "enum": ["task_closed", "unknown_capability"] }, "state": { "enum": ["starting", "reconciling", "ready", "paused", "terminal"] } }, "allOf": [ @@ -37,6 +38,15 @@ "then": { "required": ["task_id", "resume_capability"] } + }, + { + "if": { + "properties": { "resumed": { "const": true } }, + "required": ["resumed"] + }, + "then": { + "not": { "required": ["resume_rejection"] } + } } ], "additionalProperties": false diff --git a/tests/architecture/verify_protocol_schemas.py b/tests/architecture/verify_protocol_schemas.py index 9e0886b..6bb5809 100755 --- a/tests/architecture/verify_protocol_schemas.py +++ b/tests/architecture/verify_protocol_schemas.py @@ -461,14 +461,24 @@ def verify_core_messages(schemas: dict[Path, dict], registry: Registry) -> int: resumed_without_binding, "resumed connection task binding", ) - connection_validator.validate( - dict( - resumed_without_binding, - task_id="018f47a0-7b10-7abc-8def-0123456789ac", - resume_capability="r" * 32, - ) + resumed_connection = dict( + resumed_without_binding, + task_id="018f47a0-7b10-7abc-8def-0123456789ac", + resume_capability="r" * 32, + ) + connection_validator.validate(resumed_connection) + connection_validator.validate(dict(connected, resume_rejection="task_closed")) + expect_invalid( + connection_validator, + dict(connected, resume_rejection="expired"), + "unknown resume rejection", + ) + expect_invalid( + connection_validator, + dict(resumed_connection, resume_rejection="task_closed"), + "resume rejection on a resumed connection", ) - return len(requests) + 10 + return len(requests) + 13 def verify_native_messages(schemas: dict[Path, dict], registry: Registry) -> int: