Replace the shell's crashed flag with an attempt status enum

shell/src/main/scala/orca/shell/actions/SessionAction.scala

@@ -43,8 +43,8 @@ private[shell] object SessionAction:
     val name = SessionNaming.displayName(session)
     val stage = session.stage.fold("")(s => s", stage '$s'")
     val branch = selection.manifest.branch.fold("")(b => s", on branch '$b'")
-    val crashedSuffix = if selection.crashed then " (crashed)" else ""
-    s"resuming session '$name' [$harnessName]$stage$branch, in ${selection.manifest.workDir}$crashedSuffix"
+    val crashed = SessionNaming.crashedSuffix(selection.observedStatus)
+    s"resuming session '$name' [$harnessName]$stage$branch, in ${selection.manifest.workDir}$crashed"
 
   /** Parses the manifest's stored `workDir` and confirms it's still a directory
     * — a checkout deleted after its run finished otherwise crashes resume:

shell/src/main/scala/orca/shell/cli/Cli.scala

@@ -13,7 +13,7 @@ import org.jline.terminal.Terminal
 import orca.RawArgs
 import orca.shell.{ShellEnv, Tier, WorktreeScan}
 import orca.shell.run.LaunchResult
-import orca.shell.sessions.ManifestReader
+import orca.shell.sessions.ObservedStatus
 import orca.shell.ui.ShellUi
 import orca.subprocess.TtyProbe
 
@@ -309,7 +309,7 @@ private[cli] class CliCommands(using env: ShellEnv):
       list.value,
       json.value,
       isTty,
-      ManifestReader.pidAlive
+      ObservedStatus.processAlive
     )
 
   @main(doc =

shell/src/main/scala/orca/shell/cli/ContinueCli.scala

@@ -1,5 +1,6 @@
 package orca.shell.cli
 
+import orca.runner.manifest.AttemptManifest
 import orca.shell.ScanDirs
 import orca.shell.actions.SessionAction
 import orca.shell.sessions.{
@@ -17,10 +18,10 @@ import Cli.{actionFailure, complete, requireTty, usageFailure, withTerminal}
   */
 private[cli] object ContinueCli:
 
-  /** `continue`'s full behavior over explicit `dirs`/`tty`/`pidAlive` (test
+  /** `continue`'s full behavior over explicit `dirs`/`tty`/`processAlive` (test
     * seam) — tests seed each directory with `.orca/cache/attempts/` manifests,
-    * simulate a terminal or a pipe via `tty`, and fake pid liveness via
-    * `pidAlive`. The directories arrive resolved
+    * simulate a terminal or a pipe via `tty`, and fake process liveness via
+    * `processAlive`. The directories arrive resolved
     * ([[orca.shell.WorktreeScan.dirs]], at the real entry point), so nothing
     * here spawns git.
     */
@@ -30,10 +31,10 @@ private[cli] object ContinueCli:
       list: Boolean,
       json: Boolean,
       tty: Boolean,
-      pidAlive: Long => Boolean
+      processAlive: AttemptManifest => Boolean
   ): Int =
     val AttemptListing(attempts, warnings) =
-      ManifestReader.list(dirs.own, dirs.worktrees, pidAlive)
+      ManifestReader.list(dirs.own, dirs.worktrees, processAlive)
     warnings.foreach(Cli.diagnostic)
     val index = SessionIndex.of(attempts)
     if list then

shell/src/main/scala/orca/shell/cli/Tables.scala

@@ -12,7 +12,12 @@ import orca.StagePath
 import orca.runner.manifest.SessionKind
 import orca.shell.flows.DiscoveredFlow
 import orca.settings.AgentSpec
-import orca.shell.sessions.{ResumeCommand, SessionIndex, SessionNaming}
+import orca.shell.sessions.{
+  ObservedStatus,
+  ResumeCommand,
+  SessionIndex,
+  SessionNaming
+}
 
 /** The CLI's table/JSON rendering (ADR 0021 §10) — the row shapes `list` and
   * `continue --list` emit, their jsoniter codecs, and the shared space-padded
@@ -50,8 +55,13 @@ private[cli] object Tables:
       lastActiveAt: String,
       resumable: Boolean,
       reason: Option[String],
-      crashed: Boolean
+      /** The attempt's status, `Crashed` included. */
+      attemptStatus: ObservedStatus
   )
+  private given ConfiguredJsonValueCodec[ObservedStatus] =
+    ConfiguredJsonValueCodec.derived[ObservedStatus](using
+      CodecMakerConfig.withDiscriminatorFieldName(None)
+    )
   // `withTransientEmpty`/`withTransientNone` false: `--json` output is for
   // scripts, which should see an always-present `reason` key (null when
   // unset) and `shadows`/similar fields rather than a silently vanishing one.
@@ -76,7 +86,7 @@ private[cli] object Tables:
         lastActiveAt = session.lastActiveAt.toString,
         resumable = gate.isRight,
         reason = gate.left.toOption,
-        crashed = selection.crashed
+        attemptStatus = selection.observedStatus
       )
 
   private[cli] def printSessionListing(
@@ -94,7 +104,7 @@ private[cli] object Tables:
           if r.resumable then ""
           else s"  not resumable: ${r.reason.getOrElse("")}"
         val sessionName =
-          r.sessionName + (if r.crashed then " (crashed)" else "") +
+          r.sessionName + SessionNaming.crashedSuffix(r.attemptStatus) +
             tag(r.workDir, r.branch)
         (
           r.id,

shell/src/main/scala/orca/shell/menu/ShellMenu.scala

@@ -8,6 +8,7 @@ import orca.shell.run.FlowLauncher
 import orca.shell.sessions.{
   AttemptListing,
   ManifestReader,
+  ObservedStatus,
   SessionIndex,
   SessionPicker
 }
@@ -49,7 +50,7 @@ private[shell] object ShellMenu:
       ManifestReader.list(
         scanDirs.own,
         scanDirs.worktrees,
-        ManifestReader.pidAlive
+        ObservedStatus.processAlive
       )
     warnings.foreach(ShellOutput.info)
     val continueSessionCount =

shell/src/main/scala/orca/shell/sessions/ManifestReader.scala

@@ -1,20 +1,19 @@
 package orca.shell.sessions
 
 import orca.{AttemptId, OrcaDir}
-import orca.runner.manifest.{AttemptStatus, AttemptManifest}
+import orca.runner.manifest.AttemptManifest
 import orca.util.JsonFile
 
 import scala.util.control.NonFatal
 
-/** A manifest paired with its attempt's id (from the file name) and whether the
-  * attempt is now known to have crashed (status [[AttemptStatus.Running]] with
-  * a dead pid, ADR 0021 §8) — computed once here rather than re-derived by
-  * every caller.
+/** A manifest paired with its attempt's id (from the file name) and its
+  * [[ObservedStatus]] — read that rather than `manifest.status`, which cannot
+  * tell a crashed attempt from a running one.
   */
 private[shell] case class RecordedAttempt(
     id: AttemptId,
     manifest: AttemptManifest,
-    crashed: Boolean
+    observedStatus: ObservedStatus
 )
 
 /** What [[ManifestReader.list]] found: the continuable attempts, newest first,
@@ -46,33 +45,35 @@ private[shell] object ManifestReader:
     * parameters rather than one list, because that is the whole difference
     * between them.
     *
-    * A manifest with [[AttemptStatus.Running]] whose `pid` is no longer alive
-    * is a crashed attempt — its sessions are still offered, per ADR 0021 §8. An
-    * attempt that committed no session is left out: it has nothing to continue.
-    * Each directory's `.orca/cache/attempts/` is read passively
-    * ([[OrcaDir.attemptsPath]], not [[OrcaDir.ensureAttempts]]) — absent or
-    * empty contributes nothing and creates nothing on disk. A file that fails
-    * to parse as JSON, or doesn't match the `AttemptManifest` schema — which
-    * includes a timestamp that isn't an `Instant` — is skipped with a warning
-    * naming the file rather than aborting the whole listing.
+    * A crashed attempt ([[ObservedStatus.Crashed]], decided by `processAlive`;
+    * [[ObservedStatus.processAlive]] in production) still has its sessions
+    * offered, per ADR 0021 §8. An attempt that committed no session is left
+    * out: it has nothing to continue. Each directory's `.orca/cache/attempts/`
+    * is read passively ([[OrcaDir.attemptsPath]], not
+    * [[OrcaDir.ensureAttempts]]) — absent or empty contributes nothing and
+    * creates nothing on disk. A file that fails to parse as JSON, or doesn't
+    * match the `AttemptManifest` schema — which includes a timestamp that isn't
+    * an `Instant` — is skipped with a warning naming the file rather than
+    * aborting the whole listing.
     */
   def list(
       own: os.Path,
       otherWorktrees: List[os.Path],
-      pidAlive: Long => Boolean
+      processAlive: AttemptManifest => Boolean
   ): AttemptListing =
     val perDir =
-      readAttemptsDir(own, pidAlive) :: otherWorktrees.map(guarded(_, pidAlive))
+      readAttemptsDir(own, processAlive) ::
+        otherWorktrees.map(guarded(_, processAlive))
     AttemptListing(
       perDir.flatMap(_.attempts).sortBy(_.manifest.startedAt).reverse,
       perDir.flatMap(_.warnings)
     )
 
   private def guarded(
       workDir: os.Path,
-      pidAlive: Long => Boolean
+      processAlive: AttemptManifest => Boolean
   ): AttemptListing =
-    try readAttemptsDir(workDir, pidAlive)
+    try readAttemptsDir(workDir, processAlive)
     catch
       case NonFatal(e) =>
         AttemptListing(Nil, List(s"skipping $workDir: ${firstLine(e)}"))
@@ -88,7 +89,7 @@ private[shell] object ManifestReader:
     */
   private def readAttemptsDir(
       workDir: os.Path,
-      pidAlive: Long => Boolean
+      processAlive: AttemptManifest => Boolean
   ): AttemptListing =
     val dir = OrcaDir.attemptsPath(workDir)
     OrcaDir.assertNoOrcaSymlinks(workDir, dir)
@@ -98,20 +99,12 @@ private[shell] object ManifestReader:
         os.list(dir).filter(OrcaDir.isManifest).toList.map(readManifest)
       val attempts = results.collect:
         case Right((id, m)) if m.continuable =>
-          RecordedAttempt(id, m, crashed(m, pidAlive))
+          RecordedAttempt(id, m, ObservedStatus.of(m, processAlive))
       AttemptListing(
         attempts,
         results.collect { case Left(warning) => warning }
       )
 
-  private def crashed(
-      manifest: AttemptManifest,
-      pidAlive: Long => Boolean
-  ): Boolean =
-    manifest.status match
-      case AttemptStatus.Running => !pidAlive(manifest.pid)
-      case AttemptStatus.Succeeded | AttemptStatus.Failed => false
-
   /** A missing file is a warning too: the listing saw it and the read found
     * nothing, which for a directory the shell does not control is a race with
     * pruning or removal worth a line.
@@ -132,12 +125,3 @@ private[shell] object ManifestReader:
       case JsonFile.Read.Unreadable(reason) =>
         Left(s"skipping $file: $reason")
       case JsonFile.Read.Corrupt(reason) => Left(s"skipping $file: $reason")
-
-  /** The production value of [[list]]'s `pidAlive` parameter (ADR 0021 §8):
-    * `ProcessHandle.of` finds nothing for a pid that's been reaped — treated as
-    * not alive, same as a live handle reporting `isAlive == false`. Shared by
-    * the interactive menu and the CLI's `continue`, so both derive an attempt's
-    * crashed status the same way.
-    */
-  private[shell] def pidAlive(pid: Long): Boolean =
-    ProcessHandle.of(pid).map[Boolean](_.isAlive).orElse(false)

shell/src/main/scala/orca/shell/sessions/ObservedStatus.scala

@@ -0,0 +1,47 @@
+package orca.shell.sessions
+
+import orca.runner.manifest.{AttemptManifest, AttemptStatus}
+
+import java.time.Duration
+import scala.jdk.OptionConverters.*
+
+/** An attempt's [[AttemptStatus]] as the shell sees it now: a manifest still
+  * [[AttemptStatus.Running]] whose process is gone is `Crashed` (ADR 0021 §8).
+  */
+private[shell] enum ObservedStatus:
+  case Running, Succeeded, Failed, Crashed
+
+private[shell] object ObservedStatus:
+
+  /** `processAlive` answers whether the process that wrote `manifest` still
+    * runs — [[processAlive]] in production.
+    */
+  def of(
+      manifest: AttemptManifest,
+      processAlive: AttemptManifest => Boolean
+  ): ObservedStatus =
+    manifest.status match
+      case AttemptStatus.Running =>
+        if processAlive(manifest) then Running else Crashed
+      case AttemptStatus.Succeeded => Succeeded
+      case AttemptStatus.Failed    => Failed
+
+  /** Whether `manifest.pid` names a live process that started no later than
+    * `startedAt` (which the attempt takes inside that process) — a later start
+    * means the pid was reused. The slack absorbs wall-clock steps, which shift
+    * the start instants the OS reports; a crashed attempt's pid being reused
+    * within it is negligible. An unknown start instant counts as alive.
+    */
+  def processAlive(manifest: AttemptManifest): Boolean =
+    ProcessHandle
+      .of(manifest.pid)
+      .toScala
+      .filter(_.isAlive)
+      .exists: handle =>
+        handle
+          .info()
+          .startInstant()
+          .toScala
+          .forall(!_.isAfter(manifest.startedAt.plus(ProcessStartSlack)))
+
+  private val ProcessStartSlack = Duration.ofMinutes(1)

shell/src/main/scala/orca/shell/sessions/SessionIndex.scala

@@ -166,7 +166,7 @@ private[shell] object SessionIndex:
         SessionRef(attempt.id, i + 1),
         attempt.manifest,
         session,
-        attempt.crashed
+        attempt.observedStatus
       )
     val byLineage = all.groupBy(LineageKey.of)
     val lineages = byLineage.collect:

shell/src/main/scala/orca/shell/sessions/SessionNaming.scala

@@ -38,3 +38,10 @@ private[shell] object SessionNaming:
     */
   def displayName(session: ManifestSession): String =
     session.minted.fold(session.agent)(_.name)
+
+  /** ` (crashed)` for a crashed attempt's session; nothing otherwise. */
+  def crashedSuffix(status: ObservedStatus): String = status match
+    case ObservedStatus.Crashed => " (crashed)"
+    case ObservedStatus.Running | ObservedStatus.Succeeded |
+        ObservedStatus.Failed =>
+      ""

shell/src/main/scala/orca/shell/sessions/SessionPicker.scala

@@ -151,5 +151,5 @@ private[shell] object SessionPicker:
   private def rowTail(selection: SessionSelection, marker: String): String =
     val harness = AgentSpec.harnessNameFor(selection.session.backend)
     val branch = selection.manifest.branch.fold("")(b => s" on $b")
-    val crashed = if selection.crashed then " (crashed)" else ""
+    val crashed = SessionNaming.crashedSuffix(selection.observedStatus)
     s"[$harness]$marker$branch$crashed"

shell/src/main/scala/orca/shell/sessions/SessionSelection.scala

@@ -5,10 +5,9 @@ import orca.runner.manifest.{ManifestSession, AttemptManifest}
 /** One recorded session with the attempt manifest it came from — everything
   * [[orca.shell.actions.SessionAction.resume]] needs (the harness command comes
   * from the session; the working directory comes from the manifest, which may
-  * differ from the shell's own cwd). `crashed` carries the attempt's crashed
-  * status (status `"Running"` with a dead pid) through to display — resuming
-  * still offers a crashed attempt's sessions (ADR 0021 §8), but the notice
-  * should say so.
+  * differ from the shell's own cwd). `observedStatus` is the attempt's, carried
+  * through to display — resuming still offers a crashed attempt's sessions (ADR
+  * 0021 §8), but the notice should say so.
   *
   * Lives in `sessions` so [[SessionIndex]] can construct it without `sessions`
   * depending back on `actions`, which — with `actions/SessionAction` already
@@ -18,5 +17,5 @@ private[shell] case class SessionSelection(
     ref: SessionRef,
     manifest: AttemptManifest,
     session: ManifestSession,
-    crashed: Boolean
+    observedStatus: ObservedStatus
 )