fix: follow lost worker leases and retried attempts instead of abandoning the run (customer issue 1)

kernel/DESIGN.md

@@ -386,7 +386,7 @@ Minimal verb set for gate 1:
 | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility |
 | `run.watch` | `{run_id}` → stream of `{event: "entry", data: Entry}` | every appended entry, pushed |
 | `worker.attach` | `{worker_id, step_types: ["llm","agent"], pins}` → `{}` | connection becomes a worker; agent workers **must** supply opaque initial workspace revisions/stream offsets (refused otherwise) and receive `step.dispatch` events with pins plus recovery context. A step whose declared surfaces no attached worker holds parks — it is not dispatched |
-| `step.heartbeat` | `{run_id, step_id, attempt, lease_id}` → `{lease_deadline_ms}` | renew the lease; the one lease primitive |
+| `step.heartbeat` | `{run_id, step_id, attempt, lease_id}` → `{lease_deadline_ms, lease_ttl_ms}` | renew the lease; the one lease primitive. `lease_ttl_ms` (also on `step.dispatch`) is the same deadline as a duration on the daemon's clock: workers time the lease on their own monotonic clock from it, never by comparing `lease_deadline_ms` with their wall clock |
 | `effect.record` | `{run_id, step_id, attempt, idempotency_key, surface_path, revision_before, revision_after}` → `{deduped}` | phase one: agent records a writeback before performing it; the journal boundary elects one provider-call winner. `deduped:false` means this attempt owes the call **and** the `effect.confirm` that closes it |
 | `effect.confirm` | `{run_id, step_id, attempt, idempotency_key, surface_path}` → `{confirmed}` | phase two: the elected attempt made the provider call. Until it lands the election is reclaimable by a later attempt, so a worker that dies mid-writeback loses nothing; only the election holder may confirm |
 | `step.complete` | `{run_id, step_id, attempt, idempotency_key, completionReason, output, usage, started_pins, end_pins, effects, trajectory_tail}` → `{}` | completes a dispatched step — **also the out-of-band path**: the lease holder reports its actual start/end pins and journaled effect refs; kernel verifies them, runs the gate, and decides the edge |

kernel/relayflowd/src/server.rs

@@ -381,12 +381,13 @@ fn handle_request(
             let lock = hub.run_lock(&params.run_id);
             let _guard = lock.lock().expect("run lock");
             ensure_mutable(&engine, &params.run_id)?;
+            let now = now_ms();
             let (deadline, run_deadline) = hub
                 .heartbeat(
                     connection_id,
                     &(params.run_id.clone(), params.step_id, params.attempt),
                     &params.lease_id,
-                    now_ms(),
+                    now,
                 )
                 .map_err(protocol_conflict)?;
             // Durable renewal: persist the extended deadline so the lease the
@@ -396,7 +397,12 @@ fn handle_request(
             engine
                 .renew_lease(&params.run_id, run_deadline)
                 .map_err(internal_error)?;
-            Ok(json!({"lease_deadline_ms": deadline}))
+            // `lease_ttl_ms`: the same deadline as a duration, so the worker can
+            // time it on its own clock (see the dispatch frame in assignments.rs).
+            Ok(json!({
+                "lease_deadline_ms": deadline,
+                "lease_ttl_ms": deadline.saturating_sub(now_ms()).max(0),
+            }))
         }
         "step.complete" => {
             let params: StepCompleteParams = decode_params(request.params)?;

kernel/relayflowd/src/server/session/assignments.rs

@@ -28,6 +28,13 @@ impl ProtocolHub {
             if assignment.connection_id != connection_id || assignment.lease_id != lease_id {
                 bail!("heartbeat does not match the active worker lease")
             }
+            // A late heartbeat cannot revive ownership after the daemon-side
+            // deadline. In particular, this closes the transport-delay gap
+            // between a dispatch TTL being written and the worker receiving
+            // it; workers heartbeat before beginning effectful execution.
+            if now_ms >= assignment.lease_deadline_ms {
+                bail!("attempt worker lease expired before heartbeat")
+            }
             assignment.lease_deadline_ms = now_ms.saturating_add(LEASE_RENEWAL_MS);
             assignment.lease_deadline_ms
         };
@@ -243,11 +250,20 @@ impl StepDispatcher for ProtocolHub {
             sessions.reservations.remove(&key);
             return Ok(DispatchOutcome::PinMismatch { detail });
         }
-        if let Err(error) = write_frame(
-            &worker.writer,
-            &json!({"event": "step.dispatch", "data": dispatch}),
-        )
-        .with_context(|| format!("dispatch step to worker {}", worker.worker_id))
+        // `lease_ttl_ms` restates the deadline as a duration on this daemon's
+        // clock. The worker times its lease against its OWN clock, so an
+        // absolute deadline alone is only as good as the two clocks' agreement
+        // -- a skew of one lease length refused every dispatch locally as
+        // already expired, or renewed after the daemon had swept the lease.
+        let mut frame = json!({"event": "step.dispatch", "data": dispatch});
+        frame["data"]["lease_ttl_ms"] = json!(
+            dispatch
+                .lease_deadline_ms
+                .saturating_sub(super::super::protocol::now_ms())
+                .max(0)
+        );
+        if let Err(error) = write_frame(&worker.writer, &frame)
+            .with_context(|| format!("dispatch step to worker {}", worker.worker_id))
         {
             sessions.reservations.remove(&key);
             return Err(error);

kernel/relayflowd/src/server/tests.rs

@@ -572,6 +572,88 @@ fn run_resume_refuses_a_valid_journal_that_belongs_to_another_run() {
     );
 }
 
+/// The dispatch restates its lease as a duration on the daemon's clock, so a
+/// worker can time the lease on its own clock instead of comparing the
+/// absolute deadline with a wall clock that may disagree (customer
+/// rw_3a0fcb71: a skewed worker refused or under-renewed every lease).
+#[test]
+fn dispatch_carries_the_lease_as_a_duration() {
+    let directory = tempdir().unwrap();
+    let data_dir = directory.path();
+    let hub = Arc::new(ProtocolHub::default());
+    let worker_peer = attach_llm_worker(data_dir, &hub, 1);
+    let mut worker_reader = BufReader::new(worker_peer);
+    start_llm_run(data_dir, &hub);
+    let dispatch = read_frame(&mut worker_reader);
+    let after = now_ms();
+    assert_eq!(dispatch["event"], "step.dispatch");
+    let ttl = dispatch["data"]["lease_ttl_ms"]
+        .as_i64()
+        .expect("lease_ttl_ms");
+    let deadline = dispatch["data"]["lease_deadline_ms"].as_i64().unwrap();
+    let observed_remaining = deadline.saturating_sub(after).max(0);
+    assert!(
+        ttl > 0 && ttl >= observed_remaining && ttl - observed_remaining <= 1_000,
+        "ttl {ttl} deadline {deadline} after {after} observed {observed_remaining}"
+    );
+}
+
+#[test]
+fn heartbeat_cannot_revive_an_expired_assignment() {
+    let directory = tempdir().unwrap();
+    let data_dir = directory.path();
+    let hub = Arc::new(ProtocolHub::default());
+    let worker_peer = attach_llm_worker(data_dir, &hub, 1);
+    let mut worker_reader = BufReader::new(worker_peer);
+    let run_id = start_llm_run(data_dir, &hub);
+    let dispatch = read_frame(&mut worker_reader);
+    let step_id = dispatch["data"]["step_id"].as_str().unwrap().to_owned();
+    let attempt = dispatch["data"]["attempt"].as_u64().unwrap() as u32;
+    let lease_id = dispatch["data"]["lease_id"].as_str().unwrap();
+    let deadline = dispatch["data"]["lease_deadline_ms"].as_i64().unwrap();
+
+    let error = hub
+        .heartbeat(1, &(run_id, step_id, attempt), lease_id, deadline)
+        .unwrap_err();
+
+    assert!(error.to_string().contains("lease expired"), "{error:#}");
+}
+
+#[test]
+fn heartbeat_carries_the_remaining_lease_as_a_duration() {
+    let directory = tempdir().unwrap();
+    let data_dir = directory.path();
+    let hub = Arc::new(ProtocolHub::default());
+    let worker_peer = attach_llm_worker(data_dir, &hub, 1);
+    let mut worker_reader = BufReader::new(worker_peer);
+    let run_id = start_llm_run(data_dir, &hub);
+    let dispatch = read_frame(&mut worker_reader);
+    let (writer, _peer) = shared_writer();
+    let response = request(
+        data_dir,
+        &hub,
+        1,
+        &writer,
+        &json!({"id":"heartbeat","verb":"step.heartbeat","params":{
+            "run_id":run_id,
+            "step_id":dispatch["data"]["step_id"],
+            "attempt":dispatch["data"]["attempt"],
+            "lease_id":dispatch["data"]["lease_id"],
+        }})
+        .to_string(),
+    );
+    let after = now_ms();
+    assert!(response.ok, "heartbeat failed: {:?}", response.error);
+    let result = response.result.unwrap();
+    let ttl = result["lease_ttl_ms"].as_i64().expect("lease_ttl_ms");
+    let deadline = result["lease_deadline_ms"].as_i64().unwrap();
+    let observed_remaining = deadline.saturating_sub(after).max(0);
+    assert!(
+        ttl > 0 && ttl >= observed_remaining && ttl - observed_remaining <= 1_000,
+        "ttl {ttl} deadline {deadline} after {after} observed {observed_remaining}"
+    );
+}
+
 /// Finding 3: a hung worker that stops heartbeating past its lease deadline —
 /// socket still open, so no disconnect fires — must not leave the run in
 /// waiting_worker forever. The reconciler journals a `lease_expired`

packages/sdk/src/authored-flow-error.ts

@@ -30,6 +30,12 @@ export type AuthoredFlowExecutionErrorCode =
   | 'operation_callback_failed'
   | 'step_failed'
   | 'lease_exceeded'
+  /**
+   * The CLI lost the authored root's worker lease and the kernel did not
+   * re-dispatch the root to it. The run is not failed: completed steps are
+   * journaled and `flows resume <rootRunId>` continues it.
+   */
+  | 'root_lease_lost'
   | 'unsupported_completion'
   | 'unsupported_gate'
   | 'gate_failed'

packages/sdk/src/authored-node-runner.ts

@@ -240,7 +240,8 @@ export function stepFailedFrame(value: unknown): StepFailedDetails | undefined {
   for (const [key, parsed] of [
     ['stepId', frameText(frame, 'stepId')], ['stepType', frameText(frame, 'stepType')],
     ['completionReason', frameText(frame, 'completionReason')], ['attempt', frameCount(frame, 'attempt')],
-    ['maxIterations', frameCount(frame, 'maxIterations')], ['exitCode', frameCount(frame, 'exitCode')],
+    ['maxIterations', frameCount(frame, 'maxIterations')],
+    ['unchargedAttempts', frameCount(frame, 'unchargedAttempts')], ['exitCode', frameCount(frame, 'exitCode')],
     ['stdoutTail', frameText(frame, 'stdoutTail')], ['stderrTail', frameText(frame, 'stderrTail')],
     ['detail', frameText(frame, 'detail')], ['transcriptPath', frameText(frame, 'transcriptPath')],
     ['attempts', attemptFrames(frame['attempts'])],