From 62a24ce53cb68394cfd34b86a0c0a960570b6256 Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:42:03 -0700 Subject: [PATCH 1/4] Close idle disconnected tasks after 24 hours A task stayed active until an agent finished or closed it, so every session that disconnected without finishing left an open task in state.sqlite3 indefinitely. The journal now closes a task once it has had no connection, pending resume, owned tab, or staged Commit for 24 hours, at startup and before each mutation. Closed tasks keep their existing 24-hour deletion. Resuming a closed task already starts a new task, so adapters need no change. --- CHANGELOG.md | 1 + docs/multi-agent.md | 2 +- docs/rust-host.md | 2 + host-rs/crates/agenttab-host/src/journal.rs | 70 ++++++++++++++++++++- 4 files changed, 73 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 134ad55..5ecb0d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # 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. A later resume of such a task starts a new task, as resuming any closed task already did. Tasks that still own tabs are never closed this way. - 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..e8cd504 100644 --- a/docs/multi-agent.md +++ b/docs/multi-agent.md @@ -10,7 +10,7 @@ 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. diff --git a/docs/rust-host.md b/docs/rust-host.md index f25f23b..26ac0ac 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 and before each mutation, and deletes closed tasks 24 hours after they close. Resuming a closed task's capability starts a new task. + 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..fe7240b 100644 --- a/host-rs/crates/agenttab-host/src/journal.rs +++ b/host-rs/crates/agenttab-host/src/journal.rs @@ -17,6 +17,10 @@ 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 a missing or closed +/// capability already yields a fresh one. +const IDLE_TASK_RETENTION_MS: i64 = 24 * 60 * 60 * 1000; const MAX_STAGE_LIFETIME_MS: i64 = 5 * 60 * 1000; #[derive(Debug, Error)] @@ -1266,6 +1270,20 @@ 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], + )?; connection.execute( "DELETE FROM tasks WHERE state = 'closed' AND closed_at_ms IS NOT NULL AND closed_at_ms < ?1", @@ -1475,7 +1493,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 +1505,56 @@ 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()); + 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 completed_mutation_is_replayed_and_input_conflicts_fail() { let temp = tempfile::tempdir().unwrap(); From 4d01c14f85d0b1f7ccd66bfba445c3bb2d9a78b4 Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:53:43 -0700 Subject: [PATCH 2/4] Let clients start fresh when the host closed their task Clients refused to connect when a stored resume capability no longer resumed, so once the host closed an idle task, every adapter holding its capability failed until the state file was deleted by hand. The connection acknowledgement now carries resume_rejection: task_closed when the capability belongs to a finished or closed task, unknown_capability otherwise. The host keeps closed tasks' capability hashes for a year after deleting their rows so the reason stays known. On task_closed the TypeScript and Python SDKs clear the stored capability and connect as a new task; any other rejection still refuses and keeps the capability. A task still connected when the host stopped now counts its idle time from the restart instead of its last attach, so a restart cannot close it immediately. Also adds the missing clear() to the Python capability store, which finish_task already called. --- CHANGELOG.md | 4 +- docs/multi-agent.md | 4 +- docs/rust-host.md | 2 +- host-rs/crates/agenttab-host/src/journal.rs | 156 ++++++++++++++- host-rs/crates/agenttab-host/src/task.rs | 41 +++- host-rs/crates/agenttab-protocol/src/lib.rs | 15 ++ packages/sdk-python/agenttab/client.py | 40 +++- packages/sdk-python/tests/test_client.py | 188 ++++++++++++++++++ packages/sdk-typescript/src/index.ts | 13 +- packages/sdk-typescript/test/client.test.ts | 154 ++++++++++++++ schemas/rpc/v1/connection.schema.json | 10 + tests/architecture/verify_protocol_schemas.py | 24 ++- 12 files changed, 622 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ecb0d1..dab5b77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +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. A later resume of such a task starts a new task, as resuming any closed task already did. Tasks that still own tabs are never closed this way. +- 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 e8cd504..7fd95e9 100644 --- a/docs/multi-agent.md +++ b/docs/multi-agent.md @@ -12,7 +12,9 @@ A fresh adapter connection therefore gets a distinct task. Adapters that need to 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 26ac0ac..7a6a0fc 100644 --- a/docs/rust-host.md +++ b/docs/rust-host.md @@ -43,7 +43,7 @@ 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 and before each mutation, and deletes closed tasks 24 hours after they close. Resuming a closed task's capability starts a new task. +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 and before each mutation; 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. diff --git a/host-rs/crates/agenttab-host/src/journal.rs b/host-rs/crates/agenttab-host/src/journal.rs index fe7240b..99524e5 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; @@ -18,9 +20,12 @@ 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 a missing or closed -/// capability already yields a fresh one. +/// 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)] @@ -241,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, @@ -285,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; @@ -365,6 +383,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, @@ -1284,10 +1324,27 @@ fn cleanup_expired_tasks(connection: &Connection, now: i64) -> Result<(), Journa )", 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(()) } @@ -1539,6 +1596,10 @@ mod tests { .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"), @@ -1555,6 +1616,85 @@ mod tests { } } + #[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..edbe5d2 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, @@ -807,11 +822,24 @@ def connect( stream.close() if capability_store is not None and capability != active_capability: continue - source = "stored" if capability_store is not None else "supplied" - raise RuntimeError( - f"AgentTab rejected the {source} resume capability; " - "start without it only when creating a new task is intended" + if ( + capability_store is None + or connection.get("resume_rejection") != "task_closed" + ): + source = "stored" if capability_store is not None else "supplied" + raise RuntimeError( + f"AgentTab rejected the {source} resume capability; " + "start without it only when creating a new task is intended" + ) + capability_store.clear() + negotiated = cls._negotiate_connection( + address, + conversation_id=conversation_id, + resume_capability=None, + connect_timeout=connect_timeout, + request_timeout=request_timeout, ) + break negotiated = (stream, connection) attempted_capability = capability break @@ -919,6 +947,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..22b815a 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,191 @@ 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_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..be2d84f 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"; @@ -1019,9 +1020,14 @@ export class AgentTabClient { if (capability && !attempt.connected.resumed) { attempt.socket.destroy(); 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", - ); + if (attempt.connected.resume_rejection !== "task_closed") { + throw new Error( + "AgentTab rejected the stored resume capability; remove that client state only when starting a new task is intended", + ); + } + await store?.clear(); + negotiated = await negotiateConnection(endpoint, timeoutMs, options.conversationId, undefined); + break; } attemptedCapability = capability; negotiated = attempt; @@ -1325,6 +1331,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..f690944 100644 --- a/packages/sdk-typescript/test/client.test.ts +++ b/packages/sdk-typescript/test/client.test.ts @@ -474,6 +474,160 @@ 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(); +}); + +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: From 94053cc30b6b062dd904b03a4e06fe315def022a Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:56:17 -0700 Subject: [PATCH 3/4] Start fresh when a pending capability's task closed A client that crashed between the host confirming a rotated capability and activating it locally keeps the old capability as active and the confirmed one as pending. Once that task closed, the pending candidate reported task_closed but was skipped, and the rotated-away active capability was unknown, so connect refused until the state was deleted by hand. Both candidates belong to the same task, so task_closed on either one now clears the store and connects as a new task. --- packages/sdk-python/agenttab/client.py | 34 ++++++------- packages/sdk-python/tests/test_client.py | 53 +++++++++++++++++++++ packages/sdk-typescript/src/index.ts | 18 +++---- packages/sdk-typescript/test/client.test.ts | 44 +++++++++++++++++ 4 files changed, 125 insertions(+), 24 deletions(-) diff --git a/packages/sdk-python/agenttab/client.py b/packages/sdk-python/agenttab/client.py index edbe5d2..4b2a987 100644 --- a/packages/sdk-python/agenttab/client.py +++ b/packages/sdk-python/agenttab/client.py @@ -820,26 +820,28 @@ def connect( ) if capability is not None and connection.get("resumed") is not True: stream.close() - if capability_store is not None and capability != active_capability: - continue + # 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 None - or connection.get("resume_rejection") != "task_closed" + capability_store is not None + and connection.get("resume_rejection") == "task_closed" ): - source = "stored" if capability_store is not None else "supplied" - raise RuntimeError( - f"AgentTab rejected the {source} resume capability; " - "start without it only when creating a new task is intended" + capability_store.clear() + negotiated = cls._negotiate_connection( + address, + conversation_id=conversation_id, + resume_capability=None, + connect_timeout=connect_timeout, + request_timeout=request_timeout, ) - 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" + raise RuntimeError( + f"AgentTab rejected the {source} resume capability; " + "start without it only when creating a new task is intended" ) - break negotiated = (stream, connection) attempted_capability = capability break diff --git a/packages/sdk-python/tests/test_client.py b/packages/sdk-python/tests/test_client.py index 22b815a..05ab11a 100644 --- a/packages/sdk-python/tests/test_client.py +++ b/packages/sdk-python/tests/test_client.py @@ -1045,6 +1045,59 @@ def serve() -> None: 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: diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index be2d84f..76829b2 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -1019,15 +1019,17 @@ export class AgentTabClient { const attempt = await negotiateConnection(endpoint, timeoutMs, options.conversationId, capability); if (capability && !attempt.connected.resumed) { attempt.socket.destroy(); - if (capability !== activeCapability) continue; - if (attempt.connected.resume_rejection !== "task_closed") { - throw new Error( - "AgentTab rejected the stored resume capability; remove that client state only when starting a new task is intended", - ); + // 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; } - 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", + ); } attemptedCapability = capability; negotiated = attempt; diff --git a/packages/sdk-typescript/test/client.test.ts b/packages/sdk-typescript/test/client.test.ts index f690944..8e6869c 100644 --- a/packages/sdk-typescript/test/client.test.ts +++ b/packages/sdk-typescript/test/client.test.ts @@ -554,6 +554,50 @@ test("starts a fresh task when the host reports the stored capability's task clo 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", {}], From 4dde0855a52f84bd191ecc06a60c34fc157a8f00 Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:57:01 -0700 Subject: [PATCH 4/4] Expire idle tasks before accepting a resume Cleanup ran only at startup and before mutations, so on a host with no recent mutations a task idle past 24 hours could still resume, contrary to the documented contract. --- docs/rust-host.md | 2 +- host-rs/crates/agenttab-host/src/journal.rs | 27 +++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/docs/rust-host.md b/docs/rust-host.md index 7a6a0fc..c8c90da 100644 --- a/docs/rust-host.md +++ b/docs/rust-host.md @@ -43,7 +43,7 @@ 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 and before each mutation; 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. +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. diff --git a/host-rs/crates/agenttab-host/src/journal.rs b/host-rs/crates/agenttab-host/src/journal.rs index 99524e5..29a61f8 100644 --- a/host-rs/crates/agenttab-host/src/journal.rs +++ b/host-rs/crates/agenttab-host/src/journal.rs @@ -348,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( @@ -1519,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();