From 7d5161e489b2ee91699e4aed64349d7c7fc1ecdd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 04:58:44 +0000 Subject: [PATCH 1/2] Add model HTTP retry/backoff and real background-job cancellation OpenAiResponsesAgentAdapter now routes every Responses call through one HTTP path that retries 429/5xx with bounded exponential backoff plus jitter, honors Retry-After (seconds or HTTP-date), and polls for cancellation while backing off. runTask also catches RuntimeException so a misbehaving upstream surfaces as a failed task result instead of an uncaught exception. OpenPhoneAgentJobScheduler wires ToolExecutor.isCancelled() to the job store: a run counts as cancelled once its stored status leaves "running" (background_job_stop or stuck-run repair), and a watchdog thread hard-cancels the adapter so a stopped job stops even while blocked on a stuck upstream read. Job runner threads catch RuntimeException and record a failed run with backoff instead of crashing the assistant process. AgentJobStore gains statusOf() and live-run guards on markCompleted and markFailed so a finishing or failing run cannot resurrect a job the user already stopped; notifications follow the recorded outcome. Validation: scripts/check.sh passes (including check-assistant-java against android-36); a local JVM harness drove the real adapter against a stub HTTP server covering 429-then-success, Retry-After preference, bounded attempts on persistent 500, no retry on 400, and cancel during backoff. No device eval was run (no device available in this environment). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ASVb6mqLiGd7naEnLaoKnX --- docs/AGENT_RUNTIME_V1.md | 17 ++ docs/releases/CHANGELOG.md | 11 + .../assistant/jobs/AgentJobStore.java | 27 +++ .../jobs/OpenPhoneAgentJobScheduler.java | 160 +++++++++++--- .../model/OpenAiResponsesAgentAdapter.java | 196 +++++++++++------- 5 files changed, 307 insertions(+), 104 deletions(-) diff --git a/docs/AGENT_RUNTIME_V1.md b/docs/AGENT_RUNTIME_V1.md index 65f5725..08be573 100644 --- a/docs/AGENT_RUNTIME_V1.md +++ b/docs/AGENT_RUNTIME_V1.md @@ -87,6 +87,23 @@ The assistant schedules a single alarm for the next active job. On wake it: Jobs also wake during boot, locked boot, package replacement, and assistant service startup. +## Cancellation and Upstream Failures + +- `background_job_stop` marks the job `stopped`. A running job polls its + stored status between agent steps and while waiting out retry backoff; a + watchdog additionally aborts the in-flight model call, so a stopped job + halts even when it is blocked on a stuck upstream read. +- `stopped` is terminal for that run: completion and failure results from a + cancelled run are not recorded back onto the job, so a stop cannot be + overwritten or the job resurrected by a run that finishes late. +- Model HTTP calls retry 429 and 5xx responses with bounded exponential + backoff plus jitter, honoring `Retry-After` when the server provides it. + Persistent upstream failure still fails the run, which then follows the + job store's normal failure backoff. +- An unchecked exception inside the background runner (for example a broker + outage surfacing as a runtime error) marks the job failed with backoff + instead of killing the assistant process. + ## Safety Background jobs may call observe/read tools and terminal tools. State-changing diff --git a/docs/releases/CHANGELOG.md b/docs/releases/CHANGELOG.md index 5c54364..803b679 100644 --- a/docs/releases/CHANGELOG.md +++ b/docs/releases/CHANGELOG.md @@ -8,6 +8,17 @@ flow, and device support. ## [Unreleased] +### Changed + +- Cloud model calls now retry 429/5xx responses with bounded exponential + backoff plus jitter and honor `Retry-After`. +- `background_job_stop` now cancels a running job: the runner polls the stored + job status, aborts the in-flight model call, and a stopped run can no longer + resurrect the job by recording completion or failure. +- Background job runner threads survive unchecked exceptions (for example a + broker outage) by recording a failed run with backoff instead of crashing + the assistant process. + ## [0.0.3] - 2026-07-03 ### Added diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java index 9c9aaff..f8b551b 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java @@ -149,8 +149,30 @@ public synchronized boolean markRunning(long id, long nowMillis) { }); } + /** + * Current stored status for a job, or "" when the job no longer exists. + * Running jobs poll this so a user stop ({@link #stop}) or a stuck-run + * repair actually cancels the in-flight run instead of being ignored. + */ + public synchronized String statusOf(long id) { + JSONArray jobs = readJobs(); + for (int i = 0; i < jobs.length(); i++) { + JSONObject job = jobs.optJSONObject(i); + if (job != null && job.optLong("id", -1L) == id) { + return job.optString("status", ""); + } + } + return ""; + } + public synchronized boolean markCompleted(long id, String result, long nowMillis) { return updateJob(id, job -> { + // Only a run the store still considers live may record completion. + // Without this guard a finishing run would resurrect a job the + // user stopped (or that stuck-run repair already rescheduled). + if (!"running".equals(job.optString("status", ""))) { + return false; + } JSONObject schedule = job.optJSONObject("schedule_json"); long interval = schedule == null ? 0L : schedule.optLong("interval_ms", 0L); job.put("status", interval > 0 ? "active" : "completed") @@ -184,6 +206,11 @@ public synchronized boolean markDispatched(long id, String result, long nowMilli public synchronized boolean markFailed(long id, String reason, long nextRunAtMillis, int failureCount, long failureAlertAtMillis, long nowMillis) { return updateJob(id, job -> { + // Same live-run guard as markCompleted: a failure from a run the + // user already stopped must not reactivate the job for retry. + if (!"running".equals(job.optString("status", ""))) { + return false; + } job.put("status", "active") .put("updated_at", nowMillis) .put("last_run_at", nowMillis) diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java index e5a1162..e298e25 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java @@ -32,6 +32,7 @@ public final class OpenPhoneAgentJobScheduler { private static final long MIN_DELAY_MILLIS = 15_000L; private static final long STUCK_TIMEOUT_MILLIS = 10L * 60L * 1000L; private static final int MAX_DUE_PER_CHECK = 3; + private static final long CANCEL_POLL_MILLIS = 2_000L; private OpenPhoneAgentJobScheduler() {} @@ -74,7 +75,21 @@ private static void fireJob(Context context, AgentJobStore store, new Thread(new Runnable() { @Override public void run() { - runJob(context, job); + try { + runJob(context, job); + } catch (RuntimeException e) { + // An uncaught exception here (broker outage, store + // corruption, notification failure) would kill the + // whole assistant process, not just this job. Record + // the failure with backoff and keep the runtime alive. + Log.e(TAG, "Agent job crashed: " + job.id, e); + try { + failJob(context, new AgentJobStore(context), job, + "job_crash:" + e.getClass().getSimpleName()); + } catch (RuntimeException inner) { + Log.e(TAG, "Agent job crash handling failed: " + job.id, inner); + } + } } }, "OpenPhoneAgentJob-" + job.id).start(); } @@ -121,30 +136,41 @@ private static void runJob(Context context, AgentJobRecord job) { FrameworkToolExecutor toolExecutor = new FrameworkToolExecutor(context, agentManager); OpenAiResponsesAgentAdapter adapter = new OpenAiResponsesAgentAdapter(endpointConfig); final String activeTaskId = taskId; - String result = adapter.runTask(activeTaskId, job.prompt, - new ModelAdapter.ToolExecutor() { - @Override - public String callTool(String toolName, String argumentsJson) { - if (isStateChangingBackgroundTool(toolName)) { - return "{\"status\":\"background.confirmation_required\"," - + "\"reason\":\"state_changing_tool_blocked\"," - + "\"tool\":\"" + jsonEscape(toolName) + "\"}"; - } - try { - return toolExecutor.execute(activeTaskId, toolName, - new JSONObject(argumentsJson == null ? "{}" : argumentsJson)); - } catch (JSONException e) { - return "{\"status\":\"error\",\"reason\":\"bad_tool_json\"}"; + final JobCancellationSignal cancellation = new JobCancellationSignal(store, job.id); + Thread cancelWatchdog = startCancelWatchdog(adapter, cancellation, job.id); + String result; + try { + result = adapter.runTask(activeTaskId, job.prompt, + new ModelAdapter.ToolExecutor() { + @Override + public String callTool(String toolName, String argumentsJson) { + if (isStateChangingBackgroundTool(toolName)) { + return "{\"status\":\"background.confirmation_required\"," + + "\"reason\":\"state_changing_tool_blocked\"," + + "\"tool\":\"" + jsonEscape(toolName) + "\"}"; + } + try { + return toolExecutor.execute(activeTaskId, toolName, + new JSONObject(argumentsJson == null ? "{}" : argumentsJson)); + } catch (JSONException e) { + return "{\"status\":\"error\",\"reason\":\"bad_tool_json\"}"; + } } - } - @Override - public boolean isCancelled() { - return false; - } - }); - store.markCompleted(job.id, result, System.currentTimeMillis()); - if (shouldNotify(job)) { + @Override + public boolean isCancelled() { + return cancellation.isCancelled(); + } + }); + } finally { + cancellation.finish(); + cancelWatchdog.interrupt(); + } + // markCompleted refuses to record when the job is no longer + // "running" (user stop or stuck-run repair), so a cancelled run + // neither resurrects the job nor notifies as if it finished. + boolean recorded = store.markCompleted(job.id, result, System.currentTimeMillis()); + if (recorded && shouldNotify(job)) { OpenPhoneNotificationController.showAgentJobFinished(context, job, result); } } catch (RuntimeException e) { @@ -160,6 +186,86 @@ public boolean isCancelled() { } } + /** + * Aborts the in-flight model call once the job leaves the "running" + * status. The adapter checks {@code isCancelled()} between steps and + * while backing off between HTTP retries, but only a hard + * {@code adapter.cancel()} can break a thread blocked on a stuck + * upstream read, so a dedicated watcher does that. + */ + private static Thread startCancelWatchdog(OpenAiResponsesAgentAdapter adapter, + JobCancellationSignal cancellation, long jobId) { + Thread watchdog = new Thread(new Runnable() { + @Override + public void run() { + while (!cancellation.isFinished()) { + if (cancellation.isCancelled()) { + adapter.cancel(); + return; + } + try { + Thread.sleep(CANCEL_POLL_MILLIS); + } catch (InterruptedException e) { + return; + } + } + } + }, "OpenPhoneAgentJobCancel-" + jobId); + watchdog.setDaemon(true); + watchdog.start(); + return watchdog; + } + + /** + * Live cancellation signal for one job run, backed by the job store: + * the run counts as cancelled as soon as the stored status leaves + * "running" (user called background_job_stop, or stuck-run repair + * reclaimed the job). Store reads are throttled so per-step + * isCancelled() checks stay cheap, and store failures are treated as + * "not cancelled" rather than crashing the job thread. + */ + static final class JobCancellationSignal { + private final AgentJobStore mStore; + private final long mJobId; + private volatile boolean mCancelled; + private volatile boolean mFinished; + private long mLastPolledAtMillis; + + JobCancellationSignal(AgentJobStore store, long jobId) { + mStore = store; + mJobId = jobId; + } + + boolean isCancelled() { + if (mCancelled) { + return true; + } + long now = System.currentTimeMillis(); + synchronized (this) { + if (now - mLastPolledAtMillis < CANCEL_POLL_MILLIS) { + return false; + } + mLastPolledAtMillis = now; + } + try { + if (!"running".equals(mStore.statusOf(mJobId))) { + mCancelled = true; + } + } catch (RuntimeException e) { + Log.w(TAG, "Cancellation poll failed for job " + mJobId, e); + } + return mCancelled; + } + + void finish() { + mFinished = true; + } + + boolean isFinished() { + return mFinished; + } + } + private static boolean isStateChangingBackgroundTool(String toolName) { if (ToolCatalog.get().isTerminalTool(toolName)) { return false; @@ -210,9 +316,11 @@ private static void failJob(Context context, AgentJobStore store, int failures = job.failureCount + 1; long nextRunAt = now + AgentJobStore.backoffMillis(failures); long failureAlertAt = failures >= 3 ? now : job.failureAlertAtMillis; - store.markFailed(job.id, reason, nextRunAt, failures, failureAlertAt, now); - Log.w(TAG, "Agent job failed: " + job.id + " " + reason); - if (failures >= 3 && shouldNotify(job)) { + boolean recorded = store.markFailed(job.id, reason, nextRunAt, failures, + failureAlertAt, now); + Log.w(TAG, "Agent job failed: " + job.id + " " + reason + + (recorded ? "" : " (not recorded; job no longer running)")); + if (recorded && failures >= 3 && shouldNotify(job)) { OpenPhoneNotificationController.showAgentJobFailed(context, job, reason); } scheduleNext(context, store); diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java index 3df88ac..492f31a 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java @@ -9,6 +9,7 @@ import java.net.URL; import java.nio.charset.StandardCharsets; import java.util.Locale; +import java.util.concurrent.ThreadLocalRandom; import org.json.JSONArray; import org.json.JSONException; @@ -55,11 +56,21 @@ private static String resolvedModel() { private static final int MAX_CONSECUTIVE_TOOL_ERRORS = 2; private static final int MAX_CONSECUTIVE_NO_PROGRESS_ACTIONS = 2; private static final long MAX_DURATION_MS = 300000; + // Transient upstream failures (429/5xx) retry with exponential backoff + // plus jitter, honoring Retry-After when the server sends one. The + // attempt count is deliberately small so an interactive turn is not + // stalled for minutes behind a dead upstream; background jobs get their + // durability from the job store's own failure backoff on top of this. + private static final int MAX_HTTP_ATTEMPTS = 3; + private static final long INITIAL_BACKOFF_MILLIS = 1_000L; + private static final long MAX_BACKOFF_MILLIS = 30_000L; + private static final long BACKOFF_POLL_MILLIS = 250L; private final ModelEndpointConfig mEndpointConfig; private final boolean mFullYolo; private volatile boolean mCancelled; private volatile HttpURLConnection mActiveConnection; + private volatile ToolExecutor mActiveExecutor; public OpenAiResponsesAgentAdapter(String apiKey) { this(ModelEndpointConfig.directOpenAi(apiKey), false); @@ -239,6 +250,10 @@ public String runTask(String taskId, String userGoal, ToolExecutor executor) { } JSONArray steps = new JSONArray(); + // Publish the executor so the HTTP retry loop can observe external + // cancellation (e.g. a background job stopped by the user) while it + // is backing off between attempts. + mActiveExecutor = executor; try { long startedAtMillis = System.currentTimeMillis(); int consecutiveToolErrors = 0; @@ -407,6 +422,18 @@ public String runTask(String taskId, String userGoal, ToolExecutor executor) { return cancelledResult(userGoal, steps); } return error("network_error", e.getMessage(), steps); + } catch (RuntimeException e) { + // A misbehaving upstream (broker returning garbage, HTTP stack + // throwing unchecked) must surface as a failed task result so + // the calling agent thread survives and can apply its own + // retry/backoff instead of dying with an uncaught exception. + if (mCancelled || executor.isCancelled()) { + return cancelledResult(userGoal, steps); + } + return error("adapter_error", + e.getClass().getSimpleName() + ": " + e.getMessage(), steps); + } finally { + mActiveExecutor = null; } } @@ -430,44 +457,7 @@ private JSONObject callChatResponsesApi(String userMessage) throws IOException, .put("openphone_chat", "true") .put("mode", "conversation")) .put("max_output_tokens", 350); - - HttpURLConnection connection = (HttpURLConnection) new URL( - mEndpointConfig.responsesUrl()).openConnection(); - mActiveConnection = connection; - try { - connection.setConnectTimeout(15000); - connection.setReadTimeout(45000); - connection.setRequestMethod("POST"); - connection.setDoOutput(true); - connection.setRequestProperty("Authorization", "Bearer " - + mEndpointConfig.bearerToken()); - connection.setRequestProperty("Content-Type", "application/json"); - connection.setRequestProperty("Accept", "application/json"); - if (mEndpointConfig.isBrokerMode()) { - connection.setRequestProperty("X-OpenPhone-Model-Provider", "openai_responses"); - connection.setRequestProperty("X-OpenPhone-Request-Shape", "responses_proxy"); - } - - byte[] bodyBytes = body.toString().getBytes(StandardCharsets.UTF_8); - connection.setFixedLengthStreamingMode(bodyBytes.length); - try (OutputStream outputStream = connection.getOutputStream()) { - outputStream.write(bodyBytes); - } - - int statusCode = connection.getResponseCode(); - String responseBody = readAll(statusCode >= 200 && statusCode < 300 - ? connection.getInputStream() : connection.getErrorStream()); - if (statusCode < 200 || statusCode >= 300) { - throw new IOException("OpenAI HTTP " + statusCode + ": " - + summarizeError(responseBody)); - } - return new JSONObject(responseBody); - } finally { - if (mActiveConnection == connection) { - mActiveConnection = null; - } - connection.disconnect(); - } + return postResponses(body); } private JSONObject callOrchestratorResponsesApi(String userMessage, boolean hasActiveTask, @@ -727,6 +717,25 @@ private JSONObject callScreenAnswerResponsesApi(String userMessage, JSONObject s } private JSONObject postResponses(JSONObject body) throws IOException, JSONException { + TransientHttpException transientFailure = null; + for (int attempt = 1; attempt <= MAX_HTTP_ATTEMPTS; attempt++) { + if (attempt > 1 && !backoffBeforeRetry(attempt - 1, + transientFailure.retryAfterMillis)) { + break; + } + try { + return postResponsesOnce(body); + } catch (TransientHttpException e) { + transientFailure = e; + } + } + if (transientFailure == null) { + throw new IOException("request_cancelled"); + } + throw transientFailure; + } + + private JSONObject postResponsesOnce(JSONObject body) throws IOException, JSONException { HttpURLConnection connection = (HttpURLConnection) new URL( mEndpointConfig.responsesUrl()).openConnection(); mActiveConnection = connection; @@ -754,8 +763,12 @@ private JSONObject postResponses(JSONObject body) throws IOException, JSONExcept String responseBody = readAll(statusCode >= 200 && statusCode < 300 ? connection.getInputStream() : connection.getErrorStream()); if (statusCode < 200 || statusCode >= 300) { - throw new IOException("OpenAI HTTP " + statusCode + ": " - + summarizeError(responseBody)); + String message = "OpenAI HTTP " + statusCode + ": " + + summarizeError(responseBody); + if (isTransientStatus(statusCode)) { + throw new TransientHttpException(message, retryAfterMillis(connection)); + } + throw new IOException(message); } return new JSONObject(responseBody); } finally { @@ -766,6 +779,70 @@ private JSONObject postResponses(JSONObject body) throws IOException, JSONExcept } } + private static boolean isTransientStatus(int statusCode) { + return statusCode == 429 || (statusCode >= 500 && statusCode < 600); + } + + /** Parses Retry-After as delta seconds or an HTTP date; 0 when absent or invalid. */ + private static long retryAfterMillis(HttpURLConnection connection) { + String header = connection.getHeaderField("Retry-After"); + if (header == null || header.trim().isEmpty()) { + return 0L; + } + try { + return Math.max(0L, Long.parseLong(header.trim()) * 1000L); + } catch (NumberFormatException ignored) { + } + long retryAtMillis = connection.getHeaderFieldDate("Retry-After", 0L); + return retryAtMillis <= 0L ? 0L + : Math.max(0L, retryAtMillis - System.currentTimeMillis()); + } + + /** + * Waits before retry number {@code retryIndex} (1-based), preferring the + * server-provided Retry-After delay over exponential backoff with jitter. + * The wait polls for cancellation so a stopped task or a disconnected + * caller does not sit out the full backoff; returns false when the wait + * was cut short by cancellation. + */ + private boolean backoffBeforeRetry(int retryIndex, long retryAfterMillis) { + long backoff = INITIAL_BACKOFF_MILLIS << Math.min(retryIndex - 1, 5); + long delay = retryAfterMillis > 0 ? retryAfterMillis + : backoff / 2 + ThreadLocalRandom.current().nextLong(backoff / 2 + 1); + delay = Math.min(delay, MAX_BACKOFF_MILLIS); + long deadline = System.currentTimeMillis() + delay; + while (System.currentTimeMillis() < deadline) { + if (isRequestCancelled()) { + return false; + } + try { + Thread.sleep(Math.min(BACKOFF_POLL_MILLIS, + Math.max(1L, deadline - System.currentTimeMillis()))); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + return !isRequestCancelled(); + } + + private boolean isRequestCancelled() { + if (mCancelled) { + return true; + } + ToolExecutor executor = mActiveExecutor; + return executor != null && executor.isCancelled(); + } + + private static final class TransientHttpException extends IOException { + final long retryAfterMillis; + + TransientHttpException(String message, long retryAfterMillis) { + super(message); + this.retryAfterMillis = retryAfterMillis; + } + } + private String captureScreenWithRetry(ToolExecutor executor) { String request = "{\"include_screenshot\":true,\"include_activity\":true," + "\"include_ui_tree\":true," @@ -900,44 +977,7 @@ private JSONObject callResponsesApi(String userGoal, JSONArray steps, JSONObject .put("content", content))) .put("metadata", requestMetadata(screenJson)) .put("max_output_tokens", 500); - - HttpURLConnection connection = (HttpURLConnection) new URL( - mEndpointConfig.responsesUrl()).openConnection(); - mActiveConnection = connection; - try { - connection.setConnectTimeout(15000); - connection.setReadTimeout(45000); - connection.setRequestMethod("POST"); - connection.setDoOutput(true); - connection.setRequestProperty("Authorization", "Bearer " - + mEndpointConfig.bearerToken()); - connection.setRequestProperty("Content-Type", "application/json"); - connection.setRequestProperty("Accept", "application/json"); - if (mEndpointConfig.isBrokerMode()) { - connection.setRequestProperty("X-OpenPhone-Model-Provider", "openai_responses"); - connection.setRequestProperty("X-OpenPhone-Request-Shape", "responses_proxy"); - } - - byte[] bodyBytes = body.toString().getBytes(StandardCharsets.UTF_8); - connection.setFixedLengthStreamingMode(bodyBytes.length); - try (OutputStream outputStream = connection.getOutputStream()) { - outputStream.write(bodyBytes); - } - - int statusCode = connection.getResponseCode(); - String responseBody = readAll(statusCode >= 200 && statusCode < 300 - ? connection.getInputStream() : connection.getErrorStream()); - if (statusCode < 200 || statusCode >= 300) { - throw new IOException("OpenAI HTTP " + statusCode + ": " - + summarizeError(responseBody)); - } - return new JSONObject(responseBody); - } finally { - if (mActiveConnection == connection) { - mActiveConnection = null; - } - connection.disconnect(); - } + return postResponses(body); } private static JSONObject parseDecision(String outputText) throws JSONException { From 8c42a7b556279600c18cf3ab8be140893a00815b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 05:41:16 +0000 Subject: [PATCH 2/2] Harden job cancellation against races found in adversarial review Serialize all AgentJobStore mutations on a process-wide lock: every caller constructs its own store over the same SharedPreferences file, so the previous instance-level synchronized let a concurrent background_job_stop be erased by a job thread committing a stale jobs array (read-modify-write race). markCompleted/markFailed/markDispatched now also require a run-ownership token (the markRunning timestamp) so a zombie run reclaimed by stuck-run repair cannot record over its successor, and a stop that lands during runtime dispatch wins. The cancel watchdog now keeps re-cancelling every poll until the run finishes instead of exiting after the first cancel(), and the adapter checks cancellation before opening a request and again after publishing the connection, closing the window where a cancel saw no connection to disconnect and the following stuck read had nothing to break it. Background runs that end in an infrastructure error (network_error from exhausted retries, adapter_error from an unchecked crash) are now recorded as failures with the job store's backoff and repeated-failure alerting instead of counting as completed runs. A Retry-After longer than the in-call retry budget fails the call immediately rather than retrying inside the server's declared closed window. Backoff deadlines and cancellation-poll throttling use the monotonic clock, and scheduleNext keeps the alarm armed while any job is marked running so a run stranded by process death is repaired without waiting for a boot. Validation: scripts/check.sh passes (javac gate against android-36); the local JVM harness now also covers fail-fast on Retry-After beyond the retry budget, alongside 429-then-success, Retry-After preference, bounded attempts on persistent 500, no retry on 400, and cancel during backoff. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ASVb6mqLiGd7naEnLaoKnX --- docs/AGENT_RUNTIME_V1.md | 8 +- .../assistant/jobs/AgentJobStore.java | 153 +++++++++++------- .../jobs/OpenPhoneAgentJobScheduler.java | 106 +++++++----- .../model/OpenAiResponsesAgentAdapter.java | 30 +++- 4 files changed, 194 insertions(+), 103 deletions(-) diff --git a/docs/AGENT_RUNTIME_V1.md b/docs/AGENT_RUNTIME_V1.md index 08be573..2680493 100644 --- a/docs/AGENT_RUNTIME_V1.md +++ b/docs/AGENT_RUNTIME_V1.md @@ -97,9 +97,11 @@ service startup. cancelled run are not recorded back onto the job, so a stop cannot be overwritten or the job resurrected by a run that finishes late. - Model HTTP calls retry 429 and 5xx responses with bounded exponential - backoff plus jitter, honoring `Retry-After` when the server provides it. - Persistent upstream failure still fails the run, which then follows the - job store's normal failure backoff. + backoff plus jitter, honoring `Retry-After` when it fits the in-call retry + budget; a longer `Retry-After` fails the call immediately. A run that ends + in an infrastructure error (exhausted retries, adapter crash) is recorded + as a failure and follows the job store's normal failure backoff and + repeated-failure alerting instead of counting as a completion. - An unchecked exception inside the background runner (for example a broker outage surfacing as a runtime error) marks the job failed with backoff instead of killing the assistant process. diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java index f8b551b..1803a36 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java @@ -19,6 +19,13 @@ public final class AgentJobStore { private static final String KEY_NEXT_ID = "next_id"; private static final int MAX_JOBS = 200; + // Every caller constructs its own AgentJobStore over the same + // SharedPreferences file, so instance-level synchronized cannot + // serialize the read-parse-modify-commit cycles against each other. + // All mutations take this process-wide lock so a concurrent stop() + // cannot be erased by a job thread committing a stale jobs array. + private static final Object WRITE_LOCK = new Object(); + private final SharedPreferences mPrefs; public AgentJobStore(Context context) { @@ -26,7 +33,7 @@ public AgentJobStore(Context context) { Context.MODE_PRIVATE); } - public synchronized long createJob(String type, String title, String prompt, + public long createJob(String type, String title, String prompt, String payloadJson, String scheduleJson, String sessionTarget, String deliveryJson, long nextRunAtMillis) { String cleanTitle = safe(title).trim(); @@ -34,37 +41,39 @@ public synchronized long createJob(String type, String title, String prompt, return -1L; } long now = System.currentTimeMillis(); - long id = Math.max(1L, mPrefs.getLong(KEY_NEXT_ID, 1L)); - JSONArray jobs = readJobs(); - JSONObject job = new JSONObject(); - try { - job.put("id", id) - .put("type", normalizeType(type)) - .put("title", cleanTitle) - .put("prompt", safe(prompt)) - .put("payload_json", objectOrEmpty(payloadJson)) - .put("schedule_json", objectOrEmpty(scheduleJson)) - .put("session_target", safe(sessionTarget).isEmpty() - ? "main" : safe(sessionTarget)) - .put("delivery_json", objectOrEmpty(defaultDelivery(deliveryJson))) - .put("status", "active") - .put("created_at", now) - .put("updated_at", now) - .put("next_run_at", Math.max(0L, nextRunAtMillis)) - .put("running_at", 0L) - .put("last_run_at", 0L) - .put("last_result", "") - .put("failure_count", 0) - .put("failure_alert_at", 0L); - } catch (JSONException e) { - return -1L; - } - jobs.put(job); - trimOldTerminalJobs(jobs); - if (!writeJobs(jobs, id + 1L)) { - return -1L; + synchronized (WRITE_LOCK) { + long id = Math.max(1L, mPrefs.getLong(KEY_NEXT_ID, 1L)); + JSONArray jobs = readJobs(); + JSONObject job = new JSONObject(); + try { + job.put("id", id) + .put("type", normalizeType(type)) + .put("title", cleanTitle) + .put("prompt", safe(prompt)) + .put("payload_json", objectOrEmpty(payloadJson)) + .put("schedule_json", objectOrEmpty(scheduleJson)) + .put("session_target", safe(sessionTarget).isEmpty() + ? "main" : safe(sessionTarget)) + .put("delivery_json", objectOrEmpty(defaultDelivery(deliveryJson))) + .put("status", "active") + .put("created_at", now) + .put("updated_at", now) + .put("next_run_at", Math.max(0L, nextRunAtMillis)) + .put("running_at", 0L) + .put("last_run_at", 0L) + .put("last_result", "") + .put("failure_count", 0) + .put("failure_alert_at", 0L); + } catch (JSONException e) { + return -1L; + } + jobs.put(job); + trimOldTerminalJobs(jobs); + if (!writeJobs(jobs, id + 1L)) { + return -1L; + } + return id; } - return id; } public synchronized List due(long nowMillis, int limit) { @@ -150,27 +159,44 @@ public synchronized boolean markRunning(long id, long nowMillis) { } /** - * Current stored status for a job, or "" when the job no longer exists. - * Running jobs poll this so a user stop ({@link #stop}) or a stuck-run - * repair actually cancels the in-flight run instead of being ignored. + * True while the given run of a job is still live: the job exists, is + * "running", and running_at still carries this run's token (the + * markRunning timestamp). Running jobs poll this so a user stop + * ({@link #stop}) or a stuck-run repair actually cancels the in-flight + * run, and a zombie run cannot mistake its successor's execution for + * its own. */ - public synchronized String statusOf(long id) { + public synchronized boolean isRunLive(long id, long runToken) { JSONArray jobs = readJobs(); for (int i = 0; i < jobs.length(); i++) { JSONObject job = jobs.optJSONObject(i); if (job != null && job.optLong("id", -1L) == id) { - return job.optString("status", ""); + return isLiveRun(job, runToken); } } - return ""; + return false; } - public synchronized boolean markCompleted(long id, String result, long nowMillis) { + /** True when any job is currently marked running. */ + public synchronized boolean hasRunningJobs() { + JSONArray jobs = readJobs(); + for (int i = 0; i < jobs.length(); i++) { + JSONObject job = jobs.optJSONObject(i); + if (job != null && "running".equals(job.optString("status", ""))) { + return true; + } + } + return false; + } + + public synchronized boolean markCompleted(long id, String result, long nowMillis, + long runToken) { return updateJob(id, job -> { - // Only a run the store still considers live may record completion. - // Without this guard a finishing run would resurrect a job the - // user stopped (or that stuck-run repair already rescheduled). - if (!"running".equals(job.optString("status", ""))) { + // Only the run that markRunning stamped may record completion. + // Anything else means the user stopped the job or stuck-run + // repair reclaimed it (possibly for a successor run), and + // recording would resurrect or clobber that state. + if (!isLiveRun(job, runToken)) { return false; } JSONObject schedule = job.optJSONObject("schedule_json"); @@ -191,8 +217,14 @@ public synchronized boolean markCompleted(long id, String result, long nowMillis }); } - public synchronized boolean markDispatched(long id, String result, long nowMillis) { + public synchronized boolean markDispatched(long id, String result, long nowMillis, + long runToken) { return updateJob(id, job -> { + // Same live-run guard as markCompleted: a stop that lands while + // the job is being handed to an external runtime must win. + if (!isLiveRun(job, runToken)) { + return false; + } job.put("status", "dispatched") .put("updated_at", nowMillis) .put("last_run_at", nowMillis) @@ -204,11 +236,11 @@ public synchronized boolean markDispatched(long id, String result, long nowMilli } public synchronized boolean markFailed(long id, String reason, long nextRunAtMillis, - int failureCount, long failureAlertAtMillis, long nowMillis) { + int failureCount, long failureAlertAtMillis, long nowMillis, long runToken) { return updateJob(id, job -> { // Same live-run guard as markCompleted: a failure from a run the // user already stopped must not reactivate the job for retry. - if (!"running".equals(job.optString("status", ""))) { + if (!isLiveRun(job, runToken)) { return false; } job.put("status", "active") @@ -260,6 +292,11 @@ public static long backoffMillis(int failureCount) { return (1L << (bounded - 1)) * 60L * 1000L; } + private static boolean isLiveRun(JSONObject job, long runToken) { + return "running".equals(job.optString("status", "")) + && job.optLong("running_at", 0L) == runToken; + } + static JSONObject toJson(AgentJobRecord job) { JSONObject out = new JSONObject(); try { @@ -302,20 +339,22 @@ private boolean updateJob(long id, JobUpdater updater) { } private boolean updateAllJobs(JobUpdater updater) { - JSONArray jobs = readJobs(); - boolean changed = false; - for (int i = 0; i < jobs.length(); i++) { - JSONObject job = jobs.optJSONObject(i); - if (job == null) { - continue; - } - try { - changed |= updater.update(job); - } catch (JSONException e) { - Log.w(TAG, "job update failed", e); + synchronized (WRITE_LOCK) { + JSONArray jobs = readJobs(); + boolean changed = false; + for (int i = 0; i < jobs.length(); i++) { + JSONObject job = jobs.optJSONObject(i); + if (job == null) { + continue; + } + try { + changed |= updater.update(job); + } catch (JSONException e) { + Log.w(TAG, "job update failed", e); + } } + return !changed || writeJobs(jobs, mPrefs.getLong(KEY_NEXT_ID, 1L)); } - return !changed || writeJobs(jobs, mPrefs.getLong(KEY_NEXT_ID, 1L)); } private JSONArray readJobs() { diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java index e298e25..2873cda 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/OpenPhoneAgentJobScheduler.java @@ -72,11 +72,14 @@ private static void fireJob(Context context, AgentJobStore store, if (!store.markRunning(job.id, now)) { return; } + // markRunning stamped running_at with this timestamp; it doubles as + // the run-ownership token so only this run can record its outcome. + final long runToken = now; new Thread(new Runnable() { @Override public void run() { try { - runJob(context, job); + runJob(context, job, runToken); } catch (RuntimeException e) { // An uncaught exception here (broker outage, store // corruption, notification failure) would kill the @@ -85,7 +88,7 @@ public void run() { Log.e(TAG, "Agent job crashed: " + job.id, e); try { failJob(context, new AgentJobStore(context), job, - "job_crash:" + e.getClass().getSimpleName()); + "job_crash:" + e.getClass().getSimpleName(), runToken); } catch (RuntimeException inner) { Log.e(TAG, "Agent job crash handling failed: " + job.id, inner); } @@ -94,16 +97,16 @@ public void run() { }, "OpenPhoneAgentJob-" + job.id).start(); } - private static void runJob(Context context, AgentJobRecord job) { + private static void runJob(Context context, AgentJobRecord job, long runToken) { AgentJobStore store = new AgentJobStore(context); long now = System.currentTimeMillis(); if ("heartbeat".equals(job.type)) { - store.markCompleted(job.id, "heartbeat", now); + store.markCompleted(job.id, "heartbeat", now, runToken); scheduleNext(context, store); return; } if (!"agent_turn".equals(job.type)) { - store.markCompleted(job.id, "system_event_recorded", now); + store.markCompleted(job.id, "system_event_recorded", now, runToken); scheduleNext(context, store); return; } @@ -111,18 +114,19 @@ private static void runJob(Context context, AgentJobRecord job) { String backgroundRuntime = AssistantBrainConfig.routeBackgroundRuntime( context, runtimeConfig); if (!AssistantBrainConfig.BUILTIN.equals(backgroundRuntime)) { - sendBackgroundJobToRuntime(context, store, job, backgroundRuntime, runtimeConfig); + sendBackgroundJobToRuntime(context, store, job, backgroundRuntime, runtimeConfig, + runToken); scheduleNext(context, store); return; } OpenPhoneAgentManager agentManager = context.getSystemService(OpenPhoneAgentManager.class); if (agentManager == null) { - failJob(context, store, job, "framework_unavailable"); + failJob(context, store, job, "framework_unavailable", runToken); return; } ModelEndpointConfig endpointConfig = backgroundEndpointConfig(context); if (!endpointConfig.isConfigured()) { - failJob(context, store, job, "model_unconfigured"); + failJob(context, store, job, "model_unconfigured", runToken); return; } String taskId = null; @@ -130,13 +134,14 @@ private static void runJob(Context context, AgentJobRecord job) { String response = agentManager.startTask(taskRequestJson(job)); taskId = parseString(response, "task_id"); if (taskId == null || taskId.isEmpty()) { - failJob(context, store, job, "task_start_failed"); + failJob(context, store, job, "task_start_failed", runToken); return; } FrameworkToolExecutor toolExecutor = new FrameworkToolExecutor(context, agentManager); OpenAiResponsesAgentAdapter adapter = new OpenAiResponsesAgentAdapter(endpointConfig); final String activeTaskId = taskId; - final JobCancellationSignal cancellation = new JobCancellationSignal(store, job.id); + final JobCancellationSignal cancellation = + new JobCancellationSignal(store, job.id, runToken); Thread cancelWatchdog = startCancelWatchdog(adapter, cancellation, job.id); String result; try { @@ -166,15 +171,24 @@ public boolean isCancelled() { cancellation.finish(); cancelWatchdog.interrupt(); } - // markCompleted refuses to record when the job is no longer - // "running" (user stop or stuck-run repair), so a cancelled run + // Infrastructure failures (exhausted HTTP retries, unchecked + // adapter errors) follow the job store's failure backoff and its + // repeated-failure alert instead of recording as completed runs. + String resultStatus = parseString(result, "status"); + if ("network_error".equals(resultStatus) || "adapter_error".equals(resultStatus)) { + failJob(context, store, job, "job_error:" + resultStatus, runToken); + return; + } + // markCompleted refuses to record when this run is no longer the + // live one (user stop or stuck-run repair), so a cancelled run // neither resurrects the job nor notifies as if it finished. - boolean recorded = store.markCompleted(job.id, result, System.currentTimeMillis()); + boolean recorded = store.markCompleted(job.id, result, + System.currentTimeMillis(), runToken); if (recorded && shouldNotify(job)) { OpenPhoneNotificationController.showAgentJobFinished(context, job, result); } } catch (RuntimeException e) { - failJob(context, store, job, "job_error:" + e.getClass().getSimpleName()); + failJob(context, store, job, "job_error:" + e.getClass().getSimpleName(), runToken); } finally { if (agentManager != null && taskId != null && !taskId.isEmpty()) { try { @@ -187,11 +201,13 @@ public boolean isCancelled() { } /** - * Aborts the in-flight model call once the job leaves the "running" - * status. The adapter checks {@code isCancelled()} between steps and - * while backing off between HTTP retries, but only a hard - * {@code adapter.cancel()} can break a thread blocked on a stuck - * upstream read, so a dedicated watcher does that. + * Aborts the in-flight model call once the run stops being live. The + * adapter checks {@code isCancelled()} between steps and while backing + * off between HTTP retries, but only a hard {@code adapter.cancel()} + * can break a thread blocked on a stuck upstream read, so a dedicated + * watcher does that. It keeps re-cancelling every poll until the run + * finishes: a request opened just after one cancel() (which then saw no + * connection to disconnect) must also be torn down. */ private static Thread startCancelWatchdog(OpenAiResponsesAgentAdapter adapter, JobCancellationSignal cancellation, long jobId) { @@ -201,7 +217,6 @@ public void run() { while (!cancellation.isFinished()) { if (cancellation.isCancelled()) { adapter.cancel(); - return; } try { Thread.sleep(CANCEL_POLL_MILLIS); @@ -218,37 +233,41 @@ public void run() { /** * Live cancellation signal for one job run, backed by the job store: - * the run counts as cancelled as soon as the stored status leaves - * "running" (user called background_job_stop, or stuck-run repair - * reclaimed the job). Store reads are throttled so per-step - * isCancelled() checks stay cheap, and store failures are treated as - * "not cancelled" rather than crashing the job thread. + * the run counts as cancelled as soon as it stops being the live run + * (user called background_job_stop, or stuck-run repair reclaimed the + * job — possibly for a successor run). Store reads are throttled on a + * monotonic clock so per-step isCancelled() checks stay cheap, and + * store failures are treated as "not cancelled" rather than crashing + * the job thread. */ static final class JobCancellationSignal { private final AgentJobStore mStore; private final long mJobId; + private final long mRunToken; private volatile boolean mCancelled; private volatile boolean mFinished; - private long mLastPolledAtMillis; + private long mLastPolledAtNanos; - JobCancellationSignal(AgentJobStore store, long jobId) { + JobCancellationSignal(AgentJobStore store, long jobId, long runToken) { mStore = store; mJobId = jobId; + mRunToken = runToken; + mLastPolledAtNanos = System.nanoTime() - CANCEL_POLL_MILLIS * 1_000_000L; } boolean isCancelled() { if (mCancelled) { return true; } - long now = System.currentTimeMillis(); + long now = System.nanoTime(); synchronized (this) { - if (now - mLastPolledAtMillis < CANCEL_POLL_MILLIS) { + if (now - mLastPolledAtNanos < CANCEL_POLL_MILLIS * 1_000_000L) { return false; } - mLastPolledAtMillis = now; + mLastPolledAtNanos = now; } try { - if (!"running".equals(mStore.statusOf(mJobId))) { + if (!mStore.isRunLive(mJobId, mRunToken)) { mCancelled = true; } } catch (RuntimeException e) { @@ -274,11 +293,11 @@ private static boolean isStateChangingBackgroundTool(String toolName) { } private static void sendBackgroundJobToRuntime(Context context, AgentJobStore store, - AgentJobRecord job, String runtime, RuntimeConfig config) { + AgentJobRecord job, String runtime, RuntimeConfig config, long runToken) { String cleanRuntime = runtime == null ? "" : runtime.trim().toLowerCase(java.util.Locale.US); if (cleanRuntime.isEmpty() || !config.configured(cleanRuntime)) { failJob(context, store, job, - "runtime_background_dispatch_unavailable:" + cleanRuntime); + "runtime_background_dispatch_unavailable:" + cleanRuntime, runToken); return; } try { @@ -296,9 +315,10 @@ private static void sendBackgroundJobToRuntime(Context context, AgentJobStore st intent.putExtra(OpenPhoneAssistantService.EXTRA_RUNTIME_ATTENTION_INCLUDE_SCREEN, true); context.startService(intent); - store.markDispatched(job.id, "runtime_attention.sent:" + cleanRuntime, - System.currentTimeMillis()); - if (shouldNotify(job)) { + boolean recorded = store.markDispatched(job.id, + "runtime_attention.sent:" + cleanRuntime, System.currentTimeMillis(), + runToken); + if (recorded && shouldNotify(job)) { OpenPhoneNotificationController.showAgentJobFinished(context, job, AssistantBrainConfig.label(cleanRuntime) + " accepted this background job."); @@ -306,18 +326,18 @@ private static void sendBackgroundJobToRuntime(Context context, AgentJobStore st } catch (RuntimeException e) { failJob(context, store, job, "runtime_background_dispatch_failed:" + cleanRuntime + ":" - + e.getClass().getSimpleName()); + + e.getClass().getSimpleName(), runToken); } } private static void failJob(Context context, AgentJobStore store, - AgentJobRecord job, String reason) { + AgentJobRecord job, String reason, long runToken) { long now = System.currentTimeMillis(); int failures = job.failureCount + 1; long nextRunAt = now + AgentJobStore.backoffMillis(failures); long failureAlertAt = failures >= 3 ? now : job.failureAlertAtMillis; boolean recorded = store.markFailed(job.id, reason, nextRunAt, failures, - failureAlertAt, now); + failureAlertAt, now, runToken); Log.w(TAG, "Agent job failed: " + job.id + " " + reason + (recorded ? "" : " (not recorded; job no longer running)")); if (recorded && failures >= 3 && shouldNotify(job)) { @@ -334,6 +354,14 @@ private static void scheduleNext(Context context, AgentJobStore store) { PendingIntent pending = checkPendingIntent(context); long now = System.currentTimeMillis(); long nextDueAt = store.nextRunAt(now); + if (store.hasRunningJobs()) { + // Keep the alarm armed while anything is marked running so + // repairStuck can reclaim a run whose process died even when no + // other job is due; otherwise such a job is stranded until the + // next boot or job creation. + long repairAt = now + STUCK_TIMEOUT_MILLIS; + nextDueAt = nextDueAt <= 0 ? repairAt : Math.min(nextDueAt, repairAt); + } if (nextDueAt <= 0) { alarms.cancel(pending); return; diff --git a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java index 492f31a..6e99454 100644 --- a/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java +++ b/overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/model/OpenAiResponsesAgentAdapter.java @@ -726,6 +726,13 @@ private JSONObject postResponses(JSONObject body) throws IOException, JSONExcept try { return postResponsesOnce(body); } catch (TransientHttpException e) { + if (e.retryAfterMillis > MAX_BACKOFF_MILLIS) { + // The server closed the window for longer than this call + // is willing to wait; retrying sooner would land inside + // the declared closed window, so fail fast and let the + // caller's own backoff (e.g. the job store) reschedule. + throw e; + } transientFailure = e; } } @@ -736,10 +743,20 @@ private JSONObject postResponses(JSONObject body) throws IOException, JSONExcept } private JSONObject postResponsesOnce(JSONObject body) throws IOException, JSONException { + if (isRequestCancelled()) { + throw new IOException("request_cancelled"); + } HttpURLConnection connection = (HttpURLConnection) new URL( mEndpointConfig.responsesUrl()).openConnection(); mActiveConnection = connection; try { + // Re-check after publishing the connection: a cancel() that + // fired in between saw no connection to disconnect, so without + // this the request would proceed with nothing able to abort it + // until the watchdog's next cancel poll. + if (isRequestCancelled()) { + throw new IOException("request_cancelled"); + } connection.setConnectTimeout(15000); connection.setReadTimeout(45000); connection.setRequestMethod("POST"); @@ -810,14 +827,19 @@ private boolean backoffBeforeRetry(int retryIndex, long retryAfterMillis) { long delay = retryAfterMillis > 0 ? retryAfterMillis : backoff / 2 + ThreadLocalRandom.current().nextLong(backoff / 2 + 1); delay = Math.min(delay, MAX_BACKOFF_MILLIS); - long deadline = System.currentTimeMillis() + delay; - while (System.currentTimeMillis() < deadline) { + // Monotonic deadline: a wall-clock adjustment mid-backoff must not + // stretch or skip the wait. + long deadlineNanos = System.nanoTime() + delay * 1_000_000L; + while (true) { + long remainingMillis = (deadlineNanos - System.nanoTime()) / 1_000_000L; + if (remainingMillis <= 0) { + break; + } if (isRequestCancelled()) { return false; } try { - Thread.sleep(Math.min(BACKOFF_POLL_MILLIS, - Math.max(1L, deadline - System.currentTimeMillis()))); + Thread.sleep(Math.min(BACKOFF_POLL_MILLIS, Math.max(1L, remainingMillis))); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false;