fix(runtime): settle a parked run on its leader's disconnect, never on silence alone

crates/protocol/src/event.rs

@@ -346,6 +346,12 @@ pub enum AnsweredBy {
 pub enum TimeoutKind {
     Inactivity,
     Total,
+    /// The run's owning leader has held no live connection for the full
+    /// disconnect grace window. Distinct from `inactivity`/`total`,
+    /// which describe the worker's own liveness -- this describes the
+    /// leader's, and is the fact a grace-window teardown journals before
+    /// the run is settled.
+    LeaderGone,
 }
 
 // Mirrored 1:1 from the TUI adapter's own internal `GateKind`

crates/runtime/src/adapter/activity.rs

@@ -130,6 +130,22 @@ impl ActivityClock {
             match kind {
                 TimeoutKind::Inactivity => entry.inactivity_journaled = true,
                 TimeoutKind::Total => entry.total_journaled = true,
+                // This clock tracks worker/PTY liveness only -- a
+                // leader's connection state is `ipc::leader_registry`'s
+                // concern, tracked separately and never through here.
+                // Should be unreachable on that invariant (the sweep's
+                // own `since_ms` computation already skips this kind
+                // before ever calling here), but this is a `pub fn` a
+                // future caller could reach directly with a bad `kind` --
+                // logging and doing nothing costs a call that flags
+                // nothing, never a panic in a long-lived background task.
+                TimeoutKind::LeaderGone => {
+                    tracing::error!(
+                        run_id = %run_id,
+                        "mark_journaled called with LeaderGone -- ActivityClock has no \
+                         leader-connection state to flag; ignoring"
+                    );
+                }
             }
         }
     }

crates/runtime/src/domain/repository.rs

@@ -1836,46 +1836,11 @@ impl<'c> DomainRepository<'c> {
         )?))
     }
 
-    /// ADR-0027 wave 3's inactivity backstop: settles a run whose vendor
-    /// finished a turn that the leader then never closed.
-    ///
-    /// Settles to `lost`, never `succeeded`. The runtime observed only
-    /// silence, and ADR-0015's precedent is to name that uncertainty rather
-    /// than guess an outcome -- a success reached by nobody looking would be
-    /// a fabricated fact. The run's journaled result text stays readable
-    /// either way, since `run/result` reads any terminal state.
-    ///
-    /// Returns `None` unless the run is parked in `waitingUser` **and**
-    /// flagged `turnSettled`. That pairing is the whole safety property: a
-    /// `waitingUser` that means "the worker asked a question" or "an
-    /// approval is pending" must never be settled by a timeout, and neither
-    /// must a `waitingPeer` or a `paused` run. The read and the transition
-    /// share one transaction so a leader finishing the run concurrently
-    /// cannot be raced.
-    pub fn settle_abandoned_turn(
-        &mut self,
-        run_id: RunId,
-    ) -> Result<Option<Committed>, DomainError> {
-        let Some((state, turn_settled)): Option<(String, i64)> = self
-            .conn
-            .query_row(
-                "SELECT state, flags_turn_settled FROM runs WHERE run_id = ?1",
-                [run_id.to_string()],
-                |row| Ok((row.get(0)?, row.get(1)?)),
-            )
-            .optional()?
-        else {
-            return Ok(None);
-        };
-        if state != "waitingUser" || turn_settled == 0 {
-            return Ok(None);
-        }
-        let to = RunState::try_from("lost").map_err(|_| DomainError::NotFound {
-            kind: "run-state",
-            id: "lost".to_string(),
-        })?;
-        self.transition_run(run_id, &to, None).map(Some)
-    }
+    // `settle_abandoned_turn` (ADR-0027 wave 3's inactivity backstop,
+    // which settled a leader-abandoned turn to `lost` after inactivity
+    // alone) is removed: the maintainer ruled that silence is never
+    // evidence a leader gave up, only its connection actually being gone
+    // is. See ADR-0036 and `crate::ipc::leader_registry`.
 
     /// Updates a message's delivery state. Emits the matching `Message*`
     /// event.
@@ -3706,86 +3671,6 @@ mod tests {
         assert_eq!(turns_used, 0, "answers must not consume turns");
     }
 
-    /// The backstop settles a turn nobody closed -- to `lost`, because
-    /// silence is not evidence of success.
-    #[test]
-    fn the_backstop_settles_an_abandoned_settled_turn_as_lost() {
-        let mut conn = open_test_db();
-        let project_id = ProjectId::new();
-        let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10);
-        let mut repo = DomainRepository::new(&mut conn, project_id);
-        for state in ["starting", "working", "waitingUser"] {
-            let target = RunState::try_from(state).unwrap();
-            repo.transition_run(run_id, &target, None)
-                .expect("drive to waitingUser");
-        }
-        repo.set_run_flag(run_id, RunFlag::TurnSettled, true)
-            .expect("mark the wait as a finished turn");
-
-        let committed = repo
-            .settle_abandoned_turn(run_id)
-            .expect("settle")
-            .expect("an abandoned settled turn must be settled");
-        match committed.envelope.event {
-            RuntimeEvent::RunEvent { ref state, .. } => assert_eq!(state.as_str(), "lost"),
-            ref other => panic!("expected a RunEvent, got {other:?}"),
-        }
-        // This layer only settles the state; clearing the marker is the
-        // sweep's job, so it is still set here.
-        assert!(repo.read_run_flags(run_id).unwrap().turn_settled);
-    }
-
-    /// The safety property: a wait that is not a finished turn -- a
-    /// question, an approval -- must survive the backstop untouched.
-    #[test]
-    fn the_backstop_never_settles_a_wait_that_is_not_a_finished_turn() {
-        let mut conn = open_test_db();
-        let project_id = ProjectId::new();
-        let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10);
-        let mut repo = DomainRepository::new(&mut conn, project_id);
-        for state in ["starting", "working", "waitingUser"] {
-            let target = RunState::try_from(state).unwrap();
-            repo.transition_run(run_id, &target, None)
-                .expect("drive to waitingUser");
-        }
-        // No turnSettled flag: this run is waiting on a human, not holding
-        // a finished answer.
-
-        assert!(
-            repo.settle_abandoned_turn(run_id)
-                .expect("settle")
-                .is_none(),
-            "a question wait must never be settled by a timeout"
-        );
-        assert!(
-            !repo.read_run_flags(run_id).unwrap().turn_settled,
-            "sanity: the flag really is unset"
-        );
-    }
-
-    /// A run still working is not abandoned, however quiet it has been.
-    #[test]
-    fn the_backstop_never_settles_a_working_run() {
-        let mut conn = open_test_db();
-        let project_id = ProjectId::new();
-        let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10);
-        let mut repo = DomainRepository::new(&mut conn, project_id);
-        for state in ["starting", "working"] {
-            let target = RunState::try_from(state).unwrap();
-            repo.transition_run(run_id, &target, None)
-                .expect("drive to working");
-        }
-        repo.set_run_flag(run_id, RunFlag::TurnSettled, true)
-            .expect("set the flag even though the state does not match");
-
-        assert!(
-            repo.settle_abandoned_turn(run_id)
-                .expect("settle")
-                .is_none(),
-            "the state and the flag must BOTH hold; the flag alone is not enough"
-        );
-    }
-
     #[test]
     fn timeout_facts_skip_terminal_runs_but_journal_for_live_ones() {
         let mut conn = open_test_db();

crates/runtime/src/ipc/connection.rs

@@ -88,6 +88,43 @@ pub(crate) async fn handle(stream: UnixStream, ctx: ConnContext, shared: Arc<Sha
         }
     };
 
+    // Registers this connection under its negotiated instance id for the
+    // disconnect-grace-window mechanism (`leader_registry`'s own module
+    // doc). Held for the rest of this function's scope, so it deregisters
+    // however the dispatch loop below ends -- normal close, an error
+    // break, or a panic unwind -- never only on the path that happened to
+    // exist when this was written. See `LeaderConnectionGuard`'s own doc
+    // comment for why that has to be structural rather than a manually
+    // paired decrement at each exit.
+    //
+    // `on_gone` fires when this is the LAST live connection for this
+    // instance id to disconnect -- never on an ordinary reconnect leaving
+    // others live. It arms the grace-window timer rather than settling
+    // immediately: the timer re-checks `gone_for_at_least` itself when it
+    // wakes, which is what actually distinguishes "gone for the whole
+    // window" from "gone right now" (see `LeaderRegistry::gone_for_at_least`'s
+    // own doc comment).
+    let on_gone_shared = Arc::clone(&shared);
+    let _leader_connection =
+        shared
+            .leader_registry
+            .register(principal.instance_id.clone(), move |instance_id| {
+                let shared = Arc::clone(&on_gone_shared);
+                let grace = shared.config.leader_disconnect_grace;
+                tokio::spawn(async move {
+                    tokio::time::sleep(grace).await;
+                    if shared
+                        .leader_registry
+                        .gone_for_at_least(&instance_id, grace)
+                    {
+                        shared
+                            .orchestration
+                            .settle_leader_gone(&instance_id, grace, &shared.leader_registry)
+                            .await;
+                    }
+                });
+            });
+
     // ---- switch both directions to the negotiated frame size. ----
     let _ = writer_tx.send(WriterMsg::SetMax(negotiated_frame)).await;
     let parts = framed.into_parts();