Don't report a failed subprocess start on os-lib's threads

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

@@ -4,7 +4,7 @@ import org.jline.terminal.Terminal
 import orca.shell.run.ChildTerminal
 import orca.shell.sessions.{ResumeCommand, SessionNaming, SessionSelection}
 import orca.shell.ui.ShellOutput
-import orca.subprocess.QuietProc
+import orca.subprocess.{QuietProc, Spawn}
 import orca.backend.gemini.GeminiSessionList
 import orca.backend.pi.PiSessionStore
 
@@ -126,15 +126,7 @@ private[shell] object SessionAction:
           case Right(argv) =>
             try
               val exitCode = ChildTerminal.withChild(terminal):
-                os.proc(argv)
-                  .call(
-                    cwd = workDir,
-                    stdin = os.Inherit,
-                    stdout = os.Inherit,
-                    stderr = os.Inherit,
-                    check = false
-                  )
-                  .exitCode
+                Spawn.inheritingStdio(argv, workDir)
               ShellOutput.info(s"session ended (exit code $exitCode)")
               Right(exitCode)
             catch case NonFatal(e) => Left(s"resume failed — ${e.getMessage}")

shell/src/main/scala/orca/shell/create/FlowAuthoring.scala

@@ -3,6 +3,7 @@ package orca.shell.create
 import orca.agents.BackendTag
 import orca.settings.{AgentSpec, SettingsFile, SettingsScope}
 import orca.shell.{OrcaBuild, ShellEnv, Tier}
+import orca.subprocess.Spawn
 import orca.util.PromptResource
 
 import scala.util.control.NonFatal
@@ -263,16 +264,16 @@ private[shell] object FlowAuthoring:
     * a zero exit; `None` on a timeout (the process tree is killed — os-lib's
     * `destroy` recurses into `children()`, so the agent CLI's own subprocesses
     * go with it — rather than left running), a non-zero exit, or any failure to
-    * even start (a missing binary throws from `os.proc`'s spawn).
-    * [[suggestFilename]]'s production runner.
+    * even start (a missing binary throws from `Spawn`). [[suggestFilename]]'s
+    * production runner.
     */
   private def runSlugProc(
       argv: Seq[String],
       timeoutMillis: Long
   ): Option[String] =
     try
       val proc =
-        os.proc(argv).spawn(stdin = os.Pipe, stdout = os.Pipe, stderr = os.Pipe)
+        Spawn(argv, os.pwd, Map.empty, os.Pipe, os.Pipe, os.Pipe)
       try
         if proc.waitFor(timeoutMillis) && proc.exitCode() == 0 then
           Some(proc.stdout.text())

shell/src/main/scala/orca/shell/flows/FlowEditor.scala

@@ -1,6 +1,7 @@
 package orca.shell.flows
 
 import orca.shell.{ShellEnv, Tier}
+import orca.subprocess.Spawn
 
 /** Opens a flow in the user's editor (ADR 0021 §6). */
 private[shell] object FlowEditor:
@@ -25,14 +26,7 @@ private[shell] object FlowEditor:
     * `ChildTerminal.withChild`. Returns the exit code.
     */
   def edit(editor: String, path: os.Path): Int =
-    os.proc(editArgv(editor, path))
-      .call(
-        stdin = os.Inherit,
-        stdout = os.Inherit,
-        stderr = os.Inherit,
-        check = false
-      )
-      .exitCode
+    Spawn.inheritingStdio(editArgv(editor, path), os.pwd)
 
   /** Copies a built-in flow into `tier` (Project or Global) so it can be edited
     * without touching the immutable built-in cache — the copy thereafter

shell/src/main/scala/orca/shell/run/FlowLauncher.scala

@@ -5,7 +5,7 @@ import orca.{FlowSourceProperty, OrcaArgs}
 import orca.progress.FlowSource
 import orca.shell.{OrcaBuild, ShellEnv}
 import orca.shell.ui.{ShellOutput, ShellUi, UiOutcome}
-import orca.subprocess.QuietProc
+import orca.subprocess.{QuietProc, Spawn}
 
 /** Outcome of [[FlowLauncher.runAnnounced]]. */
 private[shell] enum LaunchResult:
@@ -175,17 +175,6 @@ private[shell] object FlowLauncher:
     os.makeDir.all(dir)
     dir
 
-  private def spawnInherited(argv: Seq[String], workDir: os.Path): Int =
-    os.proc(argv)
-      .call(
-        cwd = workDir,
-        stdin = os.Inherit,
-        stdout = os.Inherit,
-        stderr = os.Inherit,
-        check = false
-      )
-      .exitCode
-
   /** A snapshot that isn't in the local Ivy repository fails the forced compile
     * the same way an incompatible flow does, so its question names both.
     */
@@ -250,7 +239,7 @@ private[shell] object FlowLauncher:
             runForced(onIncompatible, flow, args, workDir)
           case PinPolicy.Honor =>
             toLaunchResult(
-              spawnInherited(
+              Spawn.inheritingStdio(
                 argv(flow, None, args, resolveWorkspaceDir()),
                 workDir
               )
@@ -275,7 +264,7 @@ private[shell] object FlowLauncher:
   )(using ShellEnv): LaunchResult =
     val build = OrcaBuild.current
     val workspaceDir = resolveWorkspaceDir()
-    val forcedExit = spawnInherited(
+    val forcedExit = Spawn.inheritingStdio(
       argv(flow, Some(build), args, workspaceDir),
       workDir
     )
@@ -297,7 +286,7 @@ private[shell] object FlowLauncher:
                   flow.fileName
                 )(
                   toLaunchResult(
-                    spawnInherited(
+                    Spawn.inheritingStdio(
                       argv(flow, None, args, workspaceDir),
                       workDir
                     )

tools/src/main/scala/orca/subprocess/OsProcCliRunner.scala

@@ -42,15 +42,14 @@ object OsProcCliRunner extends CliRunner:
     // only with `propagateEnv = false`), so this adds one variable and changes
     // nothing else the CLI sees.
     val cookie = EnvCookie.mint()
-    val sub = os
-      .proc(args)
-      .spawn(
-        cwd = cwd,
-        env = env + (EnvCookie.VarName -> cookie.value),
-        stdin = os.Pipe,
-        stdout = os.Pipe,
-        stderr = os.Pipe
-      )
+    val sub = Spawn(
+      args,
+      cwd = cwd,
+      env = env + (EnvCookie.VarName -> cookie.value),
+      stdin = os.Pipe,
+      stdout = os.Pipe,
+      stderr = os.Pipe
+    )
     new OsPipedSubProcess(sub, cookie)
 
 /** Takes the cookie unwrapped: a process orca actually spawned always carries

tools/src/main/scala/orca/subprocess/QuietProc.scala

@@ -1,5 +1,10 @@
 package orca.subprocess
 
+import ox.discard
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import scala.jdk.CollectionConverters.given
+
 /** Subprocess invocation that **guarantees captured stderr**.
   *
   * os-lib defaults `os.proc(...).call(...)`'s `stderr` to `os.Inherit`, which
@@ -33,15 +38,18 @@ private[orca] object QuietProc:
       stdin: os.ProcessInput = os.Pipe
   ): os.CommandResult =
     log.debug("exec: {}", args.mkString(" "))
-    os.proc(args)
-      .call(
-        cwd = cwd,
-        env = env,
-        stdin = stdin,
-        stdout = os.Pipe,
-        stderr = os.Pipe,
-        check = false
-      )
+    val chunks = ConcurrentLinkedQueue[Either[geny.Bytes, geny.Bytes]]()
+    val exitCode = runToExit(
+      args,
+      cwd,
+      env,
+      stdin,
+      stdout =
+        os.ProcessOutput((buf, n) => chunks.add(Left(bytes(buf, n))).discard),
+      stderr =
+        os.ProcessOutput((buf, n) => chunks.add(Right(bytes(buf, n))).discard)
+    )
+    os.CommandResult(args, exitCode, chunks.asScala.toIndexedSeq)
 
   /** Run `args` to completion, keeping at most `maxBytes` of each of stdout and
     * stderr: the rest is drained and dropped, so the exit code is the one
@@ -62,23 +70,38 @@ private[orca] object QuietProc:
     log.debug("exec: {}", args.mkString(" "))
     val outKept = new java.io.ByteArrayOutputStream()
     val errKept = new java.io.ByteArrayOutputStream()
-    val result = os
-      .proc(args)
-      .call(
-        cwd = cwd,
-        env = env,
-        stdin = os.Pipe,
-        stdout = os.ProcessOutput(keepFirst(maxBytes, outKept)),
-        stderr = os.ProcessOutput(keepFirst(maxBytes, errKept)),
-        check = false
-      )
+    val exitCode = runToExit(
+      args,
+      cwd,
+      env,
+      stdin = os.Pipe,
+      stdout = os.ProcessOutput(keepFirst(maxBytes, outKept)),
+      stderr = os.ProcessOutput(keepFirst(maxBytes, errKept))
+    )
     CappedResult(
-      exitCode = result.exitCode,
+      exitCode = exitCode,
       out = utf8(withoutSentinel(outKept.toByteArray, maxBytes)),
       truncated = outKept.size() > maxBytes,
       err = utf8(withoutSentinel(errKept.toByteArray, maxBytes))
     )
 
+  /** Joins the output pump threads too, so their writes are visible on return.
+    */
+  private def runToExit(
+      args: Seq[String],
+      cwd: os.Path,
+      env: Map[String, String],
+      stdin: os.ProcessInput,
+      stdout: os.ProcessOutput,
+      stderr: os.ProcessOutput
+  ): Int =
+    val sub = Spawn(args, cwd, env, stdin, stdout, stderr)
+    sub.join().discard
+    sub.exitCode()
+
+  private def bytes(buf: Array[Byte], n: Int): geny.Bytes =
+    geny.Bytes(java.util.Arrays.copyOf(buf, n))
+
   /** What a sink retained, less the sentinel byte. Reads the answer off the
     * sink rather than re-cutting at `maxBytes`, so a sink that stopped bounding
     * anything returns too much rather than the right prefix by luck — which is
@@ -91,9 +114,8 @@ private[orca] object QuietProc:
     * sentinel byte — that extra byte is how "the child wrote more than the cap"
     * is told from "it wrote exactly the cap", and it never reaches the result.
     *
-    * os-lib writes this from a pump thread of its own and joins that thread
-    * before `call` returns; the join is what publishes `into`'s contents to the
-    * caller.
+    * os-lib writes this from a pump thread of its own, which `runToExit` joins;
+    * the join is what publishes `into`'s contents to the caller.
     */
   private[subprocess] def keepFirst(
       maxBytes: Int,

tools/src/main/scala/orca/subprocess/Spawn.scala

@@ -0,0 +1,68 @@
+package orca.subprocess
+
+import ox.discard
+
+import java.util.concurrent.CompletableFuture
+
+/** `os.proc(args).spawn(...)`, with the process destroyed on JVM exit like
+  * os-lib's `destroyOnExit`, but a failed start (e.g. a program not on PATH)
+  * only throws to the caller.
+  *
+  * os-lib's `destroyOnExit` registers its shutdown hook and starts a monitor
+  * thread before starting the process. When the start fails, the monitor and,
+  * at JVM exit, the hook each retry the start and report the failure as an
+  * uncaught exception. Here the hook is registered only once the process runs.
+  */
+private[orca] object Spawn:
+
+  def apply(
+      args: Seq[String],
+      cwd: os.Path,
+      env: Map[String, String],
+      stdin: os.ProcessInput,
+      stdout: os.ProcessOutput,
+      stderr: os.ProcessOutput
+  ): os.SubProcess =
+    val sub = os
+      .proc(args)
+      .spawn(
+        cwd = cwd,
+        env = env,
+        stdin = stdin,
+        stdout = stdout,
+        stderr = stderr,
+        destroyOnExit = false
+      )
+    destroyOnJvmExit(sub).discard
+    sub
+
+  /** Runs `args` to completion on the parent's stdin/stdout/stderr, returning
+    * the exit code.
+    */
+  def inheritingStdio(args: Seq[String], cwd: os.Path): Int =
+    val sub = apply(args, cwd, Map.empty, os.Inherit, os.Inherit, os.Inherit)
+    sub.join().discard
+    sub.exitCode()
+
+  /** `removed` completes once the process has exited and `hook` is
+    * unregistered.
+    */
+  private[subprocess] final case class ExitHook(
+      hook: Thread,
+      removed: CompletableFuture[Void]
+  )
+
+  private[subprocess] def destroyOnJvmExit(sub: os.SubProcess): ExitHook =
+    val hook = Thread(() => sub.destroy())
+    try Runtime.getRuntime.addShutdownHook(hook)
+    catch
+      // The JVM is already shutting down. os-lib refuses the spawn then too.
+      case e: IllegalStateException =>
+        sub.wrapped.destroyForcibly().discard
+        throw e
+    ExitHook(hook, sub.wrapped.onExit().thenRun(() => removeHook(hook)))
+
+  private def removeHook(hook: Thread): Unit =
+    // Throws once the JVM is shutting down, when the hook runs anyway.
+    try Runtime.getRuntime.removeShutdownHook(hook).discard
+    catch case _: IllegalStateException => ()

tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala

@@ -0,0 +1,37 @@
+package orca.subprocess
+
+import ox.discard
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import scala.jdk.CollectionConverters.given
+
+/** Names of os-lib's subprocess shutdown-hook threads that died of an uncaught
+  * exception while `body` ran. Swaps the JVM-wide default handler for the
+  * duration.
+  */
+object OsLibThreadFailures:
+
+  private val ThreadPrefix = "subprocess-shutdown-hook"
+
+  def during(body: => Unit): List[String] =
+    val failed = ConcurrentLinkedQueue[String]()
+    val previous = Thread.getDefaultUncaughtExceptionHandler
+    Thread.setDefaultUncaughtExceptionHandler: (thread, error) =>
+      if thread.getName.startsWith(ThreadPrefix) then
+        failed.add(thread.getName).discard
+      else if previous != null then previous.uncaughtException(thread, error)
+      // What the JVM prints with no default handler. Not the thread group's
+      // `uncaughtException`: the root group calls back into this handler.
+      else
+        System.err.print(s"Exception in thread \"${thread.getName}\" ")
+        error.printStackTrace(System.err)
+    try
+      body
+      // Bounded: other suites' live processes have monitor threads too.
+      liveOsLibThreads().foreach(_.join(2000))
+    finally Thread.setDefaultUncaughtExceptionHandler(previous)
+    failed.asScala.toList
+
+  private def liveOsLibThreads(): List[Thread] =
+    Thread.getAllStackTraces.keySet.asScala.toList
+      .filter(_.getName.startsWith(ThreadPrefix))

tools/src/test/scala/orca/subprocess/OsProcCliRunnerTest.scala

@@ -174,3 +174,14 @@ class OsProcCliRunnerTest extends munit.FunSuite:
     )
     try assertEquals(proc.stdoutLines.toList, List("late"))
     finally proc.destroyForciblyTree()
+
+  test("spawnPiped of a missing program fails only in the caller"):
+    val failedThreads = OsLibThreadFailures.during:
+      intercept[java.io.IOException]:
+        OsProcCliRunner.spawnPiped(
+          Seq("orca-test-no-such-program"),
+          env = Map.empty,
+          cwd = os.pwd
+        )
+      .discard
+    assertEquals(failedThreads, Nil)