Keep the typed answer deadline when an answer wait starts after recovery's budget

CHANGELOG.md

@@ -14,4 +14,5 @@ First public release. Industrious is a Rust port of Calvino, a maintained fork o
 - File selection fails closed: `.gitignore` rules apply from the root of the repository holding each selected path, a `.gitignore` that cannot be read or parsed is an error, files read into an API prompt or a browser text bundle must be UTF-8 text rather than being garbled (browser uploads and ZIP bundles still send the original bytes), and credential and private-key files are refused by name and skipped in expansion.
 - File sections in rendered prompts and bundles keep every line that ended, dropping only the final line ending, so a file ending in a whitespace-only line (as a diff can) arrives whole.
 - A tab retirement that has reached its outcome always records it: a slow disk write at the end of the 5 s budget no longer reports a false cleanup failure and leaves the record undecided, with the tab already closed.
+- An answer wait reached just after the bounded original-target observation ended reports the answer deadline, as it does just before: a stall there no longer replaces that failure with an untyped error that withholds the Grok fallback.
 - Licensed under FSL-1.1-ALv2; each version also becomes available under Apache-2.0 two years after its release.

src/browser_runner.rs

@@ -4658,7 +4658,7 @@ async fn wait_original_answer(
     )?;
     let bounded = recovery.limits(limits);
     let initial = recovery
-        .within(wait_answer(
+        .answer_within(wait_answer(
             store,
             id,
             reservation,

src/browser_runner/read_recovery.rs

@@ -37,6 +37,26 @@ impl Budget {
             Instant::now() < deadline,
             "Original-target observation budget expired; recovery retained"
         );
+        Self::settle(deadline, future).await
+    }
+    /// An answer wait due to start after the episode ended has met its deadline, so it never
+    /// starts. The typed failure decides provider eligibility; `observe` revalidates authority.
+    pub(super) async fn answer_within<T>(
+        &self,
+        future: impl std::future::Future<Output = Result<T>>,
+    ) -> Result<T> {
+        let Some(deadline) = self.deadline else {
+            return future.await;
+        };
+        if Instant::now() >= deadline {
+            return Err(crate::browser_provider::Failure::AnswerDeadline.into());
+        }
+        Self::settle(deadline, future).await
+    }
+    async fn settle<T>(
+        deadline: Instant,
+        future: impl std::future::Future<Output = Result<T>>,
+    ) -> Result<T> {
         tokio::time::timeout_at(deadline, future).await.context(
             "Original-target observation did not settle before its deadline; recovery retained",
         )?
@@ -269,3 +289,45 @@ impl<'a> Context<'a> {
         self.require(cancel)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{cell::Cell, task::Poll};
+
+    fn budget(deadline: Instant) -> Budget {
+        Budget {
+            enabled: true,
+            deadline: Some(deadline),
+        }
+    }
+    #[tokio::test]
+    async fn an_answer_wait_after_the_budget_is_not_polled_and_reports_the_answer_deadline() {
+        let polls = Cell::new(0);
+        let wait = std::future::poll_fn(|_| {
+            polls.set(polls.get() + 1);
+            Poll::<Result<()>>::Pending
+        });
+        let error = budget(Instant::now())
+            .answer_within(wait)
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(
+                error.downcast_ref::<crate::browser_provider::Failure>(),
+                Some(crate::browser_provider::Failure::AnswerDeadline)
+            ),
+            "{error:#}"
+        );
+        assert_eq!(polls.get(), 0);
+    }
+    #[tokio::test]
+    async fn an_answer_wait_still_unsettled_at_the_deadline_stays_ineligible() {
+        let deadline = Instant::now() + Duration::from_millis(50);
+        let error = budget(deadline)
+            .answer_within(std::future::pending::<Result<()>>())
+            .await
+            .unwrap_err();
+        assert!(!eligible(&error), "{error:#}");
+    }
+}

src/browser_runner_tests.rs

@@ -37,6 +37,7 @@ enum Fault {
     RetryOffered,
     ProviderLateCommit,
     ProviderLateCommitPendingAnswer,
+    ProviderLateCommitStalledBeforeAnswer,
     ProviderLateCommitLostAck,
     ProviderLateCommitChangedOwner,
     ProviderLateAnswer,
@@ -558,6 +559,7 @@ impl Exchange {
                 self.fault,
                 Fault::ProviderLateCommit
                     | Fault::ProviderLateCommitPendingAnswer
+                    | Fault::ProviderLateCommitStalledBeforeAnswer
                     | Fault::ProviderLateCommitLostAck
                     | Fault::ProviderLateCommitChangedOwner
             ) && !self.provider_recovery_started()
@@ -678,6 +680,9 @@ impl Exchange {
             }
             if cfg["action"] == "acknowledge" {
                 self.commit = cfg["receipt"].clone();
+                if matches!(self.fault, Fault::ProviderLateCommitStalledBeforeAnswer) {
+                    hold_session_lock(&self.store, &self.id, RECOVERY_BUDGET);
+                }
             }
             return json!(true);
         }
@@ -1023,7 +1028,9 @@ impl Exchange {
             let recovering = self.provider_recovery_started();
             if matches!(
                 self.fault,
-                Fault::ProviderLateCommitPendingAnswer | Fault::ProviderLateCommitChangedOwner
+                Fault::ProviderLateCommitPendingAnswer
+                    | Fault::ProviderLateCommitStalledBeforeAnswer
+                    | Fault::ProviderLateCommitChangedOwner
             ) || matches!(self.fault, Fault::ProviderLateAnswerDeadline) && !recovering
             {
                 result["completionVisible"] = json!(false);
@@ -2068,13 +2075,29 @@ fn limits() -> RunnerLimits {
 const WAITED_OUT_ATTACHMENT_FLOOR: Duration = Duration::from_secs(5);
 /// For closes the fixture refuses, which production watches for its whole budget.
 const REFUSED_CLOSE_WAIT: Duration = Duration::from_secs(1);
+/// Keeps the session's writes waiting for `hold` from another thread, as a contended store would.
+fn hold_session_lock(store: &SessionStore, id: &str, hold: Duration) {
+    let lock = DirectoryLock::acquire(
+        &store
+            .home()
+            .join("sessions-write-locks")
+            .join(format!("{id}.lock")),
+        Duration::from_secs(1),
+    )
+    .unwrap();
+    std::thread::spawn(move || {
+        std::thread::sleep(hold);
+        drop(lock);
+    });
+}
+const RECOVERY_BUDGET: Duration = Duration::from_secs(3);
 /// Limits for cases whose commit or answer deadline expires before a bounded provider read.
 /// That read's whole budget is min(commit, answer), so the fixture's 500 ms leaves too little
 /// for its identity checks and receipts once a parallel suite stalls native publication.
 fn recovery_limits() -> RunnerLimits {
     RunnerLimits {
-        commit: Duration::from_secs(3),
-        answer: Duration::from_secs(3),
+        commit: RECOVERY_BUDGET,
+        answer: RECOVERY_BUDGET,
         ..limits()
     }
 }
@@ -3994,6 +4017,26 @@ async fn bounded_provider_read_does_not_restart_full_answer_timeout_after_commit
     stop(f).await;
 }
 #[tokio::test]
+async fn an_answer_wait_reached_after_the_recovery_budget_keeps_the_typed_answer_deadline() {
+    let f = fixture(Fault::ProviderLateCommitStalledBeforeAnswer, false).await;
+    let mut bounded = recovery_limits();
+    bounded.answer = Duration::from_secs(60);
+    // The session lock is held from the acknowledgement for the whole budget, so the answer wait
+    // starts after the episode that recovered the commitment has ended.
+    let error = run_within(&f, &bounded).await.unwrap_err();
+    assert!(
+        matches!(
+            error.downcast_ref::<crate::browser_provider::Failure>(),
+            Some(crate::browser_provider::Failure::AnswerDeadline)
+        ),
+        "{error:#}"
+    );
+    assert_eq!(f.sends.load(Ordering::SeqCst), 1);
+    assert_eq!(f.closes.load(Ordering::SeqCst), 0);
+    assert!(captured(&f.store, &f.id).is_err());
+    stop(f).await;
+}
+#[tokio::test]
 async fn bounded_provider_read_caps_recovered_ack_and_revalidates_used_episode() {
     for fault in [
         Fault::ProviderLateCommitLostAck,