Close idle tasks and let clients start fresh after a task closes

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.

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
 

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.

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<String> = 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<ResumeRejection, JournalError> {
+        let hash = capability_hash(capability);
+        let connection = self.connection.lock();
+        let closed: Option<i64> = 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();