Record which liveness source owns a session, instead of inferring it from the ref

crates/voro-core/migrations/0019_session_liveness_source.sql

@@ -0,0 +1,14 @@
+-- Which liveness source is authoritative for a session (DESIGN.md §8, task
+-- #387), recorded at launch by the code that spawned the process instead of
+-- inferred by reconciliation from the presence of a session ref. 'pid' means
+-- the recorded pid is the work itself (a foreground child, or an agent with no
+-- 'sessions' verb); 'listing' means the launch handed the work to a supervisor,
+-- so only the agent's own listing can answer. Purely additive.
+--
+-- Sessions open across the upgrade default to 'listing', which is what every
+-- dispatch of an agent with a 'sessions' verb already was. A pre-migration
+-- interactive refine round is therefore left alone rather than pid-checked —
+-- unprobeable in the direction that never finalises a live session wrongly, and
+-- the operator's next transition closes it.
+ALTER TABLE sessions ADD COLUMN liveness_source TEXT NOT NULL DEFAULT 'listing'
+  CHECK (liveness_source IN ('pid','listing'));

crates/voro-core/src/agent.rs

@@ -18,6 +18,7 @@ use std::sync::LazyLock;
 use serde::Deserialize;
 
 use crate::error::{Error, Result};
+use crate::model::LivenessSource;
 use crate::scheduler::{AttentionCosts, DEFAULT_MAX_RUNNING};
 use crate::template::{render, shell_quote};
 
@@ -800,6 +801,24 @@ impl ResolvedAgent {
         render_launch(&self.dispatch, spec, model)
     }
 
+    /// Which liveness source a session launched through this agent's
+    /// `dispatch` template must be read by (DESIGN.md §8), recorded on the
+    /// session row at launch. An agent defining a `sessions` verb is one whose
+    /// launch may hand the work to a supervisor — `claude --bg` does — leaving
+    /// Voro holding a launcher pid that dies at birth, so its listing is the
+    /// only source that can answer. An agent without the verb has no listing to
+    /// consult, and its spawned pid is all there is.
+    ///
+    /// This is the *headless* launch's answer, which the interactive `plan`
+    /// verb does not share: that one is a foreground child Voro owns, so its
+    /// caller records [`LivenessSource::Pid`] itself.
+    pub fn dispatch_liveness_source(&self) -> LivenessSource {
+        match self.sessions {
+            Some(_) => LivenessSource::Listing,
+            None => LivenessSource::Pid,
+        }
+    }
+
     /// The plan template rendered the same way, when the agent defines the
     /// verb, with `{model}` resolved to `model_plan` falling back to `model`.
     /// Planning has no depth: it is interactive reasoning either way, so
@@ -2457,6 +2476,37 @@ mod tests {
         );
     }
 
+    /// What a headless launch records on its session row (DESIGN.md §8, task
+    /// #387): an agent with a `sessions` verb may hand the work to a supervisor,
+    /// so its listing is the authority; one without has only the pid Voro
+    /// spawned.
+    #[test]
+    fn a_sessions_verb_makes_a_launch_listing_authoritative() {
+        let text = r#"
+            [agents.supervised]
+            dispatch = "run --bg {prompt_file}"
+            sessions = "run sessions --json"
+
+            [agents.plain]
+            dispatch = "run {prompt_file}"
+        "#;
+        let config = AgentsConfig::parse(text, Path::new("/tmp/voro.toml")).unwrap();
+        assert_eq!(
+            config
+                .resolve(Some("supervised"))
+                .unwrap()
+                .dispatch_liveness_source(),
+            LivenessSource::Listing
+        );
+        assert_eq!(
+            config
+                .resolve(Some("plain"))
+                .unwrap()
+                .dispatch_liveness_source(),
+            LivenessSource::Pid
+        );
+    }
+
     #[test]
     fn default_agent_key_sets_the_default() {
         let text = r#"

crates/voro-core/src/lib.rs

@@ -28,8 +28,8 @@ pub use cap::{CAP_SIGNATURES, CapReading, read_cap, strip_ansi};
 pub use error::{Error, Result};
 pub use import::{GithubIssue, already_imported, issue_new_task, issue_task_body};
 pub use model::{
-    Dep, DepKind, DepRef, Doc, Event, NextAction, Priority, Project, RefineOutcome, Repo,
-    RunningRow, Session, SessionOutcome, Task, TaskState, location_is_url,
+    Dep, DepKind, DepRef, Doc, Event, LivenessSource, NextAction, Priority, Project, RefineOutcome,
+    Repo, RunningRow, Session, SessionOutcome, Task, TaskState, location_is_url,
 };
 pub use pr::{Mergeability, PrPlan, PrRef, format_review_feedback, parse_mergeable, plan_pr};
 pub use review::{

crates/voro-core/src/model.rs

@@ -199,6 +199,58 @@ impl ToSql for DepKind {
     }
 }
 
+/// Which source of liveness is authoritative for a session (DESIGN.md §8),
+/// recorded by the code that spawned the process because only it knows what it
+/// spawned. The two differ in one respect: whether the pid the session row
+/// holds is the work itself or a launcher that spawned it and exited.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum LivenessSource {
+    /// The recorded pid *is* the work — a foreground child Voro owns, or an
+    /// agent with no `sessions` verb, where the spawned pid is the only source
+    /// there is. `kill -0` answers.
+    Pid,
+    /// The work belongs to a supervisor the launch handed it to, so the
+    /// recorded pid dies at birth and only the agent's own `sessions` listing
+    /// can say whether the session is still working.
+    Listing,
+}
+
+impl LivenessSource {
+    pub const ALL: [LivenessSource; 2] = [LivenessSource::Pid, LivenessSource::Listing];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            LivenessSource::Pid => "pid",
+            LivenessSource::Listing => "listing",
+        }
+    }
+
+    pub fn parse(s: &str) -> Result<LivenessSource> {
+        Self::ALL
+            .into_iter()
+            .find(|source| source.as_str() == s)
+            .ok_or_else(|| Error::Invalid(format!("unknown liveness source '{s}'")))
+    }
+}
+
+impl fmt::Display for LivenessSource {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+impl FromSql for LivenessSource {
+    fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
+        LivenessSource::parse(value.as_str()?).map_err(|e| FromSqlError::Other(Box::new(e)))
+    }
+}
+
+impl ToSql for LivenessSource {
+    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
+        Ok(self.as_str().into())
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum SessionOutcome {
     Completed,
@@ -556,6 +608,9 @@ pub struct Session {
     /// substituted into the agent's attach/resume/continue verb templates.
     /// `None` when the agent has no capture story or capture failed.
     pub session_ref: Option<String>,
+    /// Which source reconciliation must read this session's liveness by,
+    /// recorded at launch by whichever code spawned the process (DESIGN.md §8).
+    pub liveness_source: LivenessSource,
     pub log_path: Option<String>,
     pub started_at: String,
     pub ended_at: Option<String>,

crates/voro-core/src/scheduler.rs

@@ -473,7 +473,7 @@ pub struct StateCounts {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::model::TaskState;
+    use crate::model::{LivenessSource, TaskState};
     use crate::store::NewTask;
 
     #[test]
@@ -661,7 +661,9 @@ mod tests {
     }
 
     fn to_stalled(s: &mut Store, id: i64) {
-        let (_, session) = s.record_dispatch(id, "claude", Some(1), None).unwrap();
+        let (_, session) = s
+            .record_dispatch(id, "claude", Some(1), LivenessSource::Pid, None)
+            .unwrap();
         s.reconcile_session(session.id, false, false).unwrap();
     }
 
@@ -1098,8 +1100,15 @@ mod tests {
         let mut s = setup();
         let p = add_project(&mut s, "p", 5);
         let refining = add_proposed(&mut s, p, "being rewritten", Priority::P0);
-        s.record_refine_launch(refining, "thin body", "claude", Some(1), None)
-            .unwrap();
+        s.record_refine_launch(
+            refining,
+            "thin body",
+            "claude",
+            Some(1),
+            LivenessSource::Pid,
+            None,
+        )
+        .unwrap();
         let ready = add_task(&mut s, p, "startable", Priority::P3);
 
         let candidates = s.candidates().unwrap();

crates/voro-core/src/seed.rs

@@ -9,7 +9,7 @@
 //! one by hand when a dev build needs to exercise dispatch.
 
 use crate::error::Result;
-use crate::model::{DepKind, Priority, SessionOutcome, TaskState};
+use crate::model::{DepKind, LivenessSource, Priority, SessionOutcome, TaskState};
 use crate::store::{NewTask, Store};
 use crate::transition::{Action, Triage};
 
@@ -70,7 +70,13 @@ pub fn seed(store: &mut Store) -> Result<SeedSummary> {
     // Fixture sessions record no pid: reconcile-on-read finalises a live
     // session whose process is gone, and an absent pid reads as liveness
     // unknown, which leaves the row standing.
-    store.create_session(running.id, "claude", None, Some("/tmp/voro-dev/import.log"))?;
+    store.create_session(
+        running.id,
+        "claude",
+        None,
+        LivenessSource::Listing,
+        Some("/tmp/voro-dev/import.log"),
+    )?;
     age(store, running.id, "-40 minutes")?;
     tasks += 1;
 
@@ -82,7 +88,8 @@ pub fn seed(store: &mut Store) -> Result<SeedSummary> {
         Priority::P2,
     )?;
     store.apply(asked.id, Action::Start)?;
-    let asked_session = store.create_session(asked.id, "claude", None, None)?;
+    let asked_session =
+        store.create_session(asked.id, "claude", None, LivenessSource::Listing, None)?;
     store.apply(
         asked.id,
         Action::Ask("Should the strip keep showing waiting rows once they have a PR?".into()),
@@ -100,8 +107,13 @@ pub fn seed(store: &mut Store) -> Result<SeedSummary> {
     )?;
     store.apply(review.id, Action::Start)?;
     store.set_branch(review.id, Some("split-review-keys"))?;
-    let review_session =
-        store.create_session(review.id, "claude", None, Some("/tmp/voro-dev/keys.log"))?;
+    let review_session = store.create_session(
+        review.id,
+        "claude",
+        None,
+        LivenessSource::Listing,
+        Some("/tmp/voro-dev/keys.log"),
+    )?;
     store.apply(
         review.id,
         Action::Complete(Some(
@@ -144,6 +156,7 @@ pub fn seed(store: &mut Store) -> Result<SeedSummary> {
         stalled.id,
         "claude",
         Some(0),
+        LivenessSource::Listing,
         Some("/tmp/voro-dev/backfill.log"),
     )?;
     store.reconcile_session(dead.id, false, false)?;
@@ -189,7 +202,7 @@ pub fn seed(store: &mut Store) -> Result<SeedSummary> {
         refining.id,
         Action::Refine("Narrow this to a decision with a threshold, not an open question.".into()),
     )?;
-    store.create_session(refining.id, "claude", None, None)?;
+    store.create_session(refining.id, "claude", None, LivenessSource::Listing, None)?;
     tasks += 1;
 
     let parked = ready_task(