Fix of Model HTTP retry/backoff + real background-job cancellation

docs/AGENT_RUNTIME_V1.md

@@ -87,6 +87,25 @@ 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 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.
+
 ## Safety
 
 Background jobs may call observe/read tools and terminal tools. State-changing

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

overlay/packages/apps/OpenPhoneAssistant/src/org/openphone/assistant/jobs/AgentJobStore.java

@@ -19,52 +19,61 @@ 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) {
         mPrefs = context.getApplicationContext().getSharedPreferences(PREFS,
                 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();
         if (cleanTitle.isEmpty()) {
             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<AgentJobRecord> due(long nowMillis, int limit) {
@@ -149,8 +158,47 @@ public synchronized boolean markRunning(long id, long nowMillis) {
         });
     }
 
-    public synchronized boolean markCompleted(long id, String result, long nowMillis) {
+    /**
+     * 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 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 isLiveRun(job, runToken);
+            }
+        }
+        return false;
+    }
+
+    /** 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 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");
             long interval = schedule == null ? 0L : schedule.optLong("interval_ms", 0L);
             job.put("status", interval > 0 ? "active" : "completed")
@@ -169,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)
@@ -182,8 +236,13 @@ 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 (!isLiveRun(job, runToken)) {
+                return false;
+            }
             job.put("status", "active")
                     .put("updated_at", nowMillis)
                     .put("last_run_at", nowMillis)
@@ -233,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 {
@@ -275,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() {