From 1642561b9106fc8b29143a1eae7ed085f8c9538e Mon Sep 17 00:00:00 2001 From: Adam Warski Date: Thu, 22 Oct 2026 14:36:51 +0000 Subject: [PATCH 1/2] Don't report a failed subprocess start on os-lib's threads os-lib registers its destroy-on-exit shutdown hook, and starts a thread that removes it, before starting the process. When the start fails (for example the agent CLI is not on PATH), both retry the start and print the failure as an uncaught exception. Spawn with destroyOnExit = false and register the hook only after the process has started. --- .../orca/subprocess/OsProcCliRunner.scala | 17 +++-- .../scala/orca/subprocess/QuietProc.scala | 68 ++++++++++++------- .../main/scala/orca/subprocess/Spawn.scala | 51 ++++++++++++++ .../orca/subprocess/OsLibThreadFailures.scala | 32 +++++++++ .../orca/subprocess/OsProcCliRunnerTest.scala | 11 +++ .../scala/orca/subprocess/QuietProcTest.scala | 9 +++ 6 files changed, 156 insertions(+), 32 deletions(-) create mode 100644 tools/src/main/scala/orca/subprocess/Spawn.scala create mode 100644 tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala diff --git a/tools/src/main/scala/orca/subprocess/OsProcCliRunner.scala b/tools/src/main/scala/orca/subprocess/OsProcCliRunner.scala index 6e22ddd1a..f6471b05b 100644 --- a/tools/src/main/scala/orca/subprocess/OsProcCliRunner.scala +++ b/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 diff --git a/tools/src/main/scala/orca/subprocess/QuietProc.scala b/tools/src/main/scala/orca/subprocess/QuietProc.scala index 27282d422..031c8421b 100644 --- a/tools/src/main/scala/orca/subprocess/QuietProc.scala +++ b/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, diff --git a/tools/src/main/scala/orca/subprocess/Spawn.scala b/tools/src/main/scala/orca/subprocess/Spawn.scala new file mode 100644 index 000000000..72bcbce87 --- /dev/null +++ b/tools/src/main/scala/orca/subprocess/Spawn.scala @@ -0,0 +1,51 @@ +package orca.subprocess + +import ox.discard + +/** `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[subprocess] 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) + sub + + private def destroyOnJvmExit(sub: os.SubProcess): Unit = + val hook = Thread: () => + sub.destroy(sub.shutdownGracePeriod, async = false, recursive = true) + 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 + sub.wrapped.onExit().thenRun(() => removeHook(hook)).discard + + 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 => () diff --git a/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala b/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala new file mode 100644 index 000000000..cc90afdc2 --- /dev/null +++ b/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala @@ -0,0 +1,32 @@ +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) + 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)) diff --git a/tools/src/test/scala/orca/subprocess/OsProcCliRunnerTest.scala b/tools/src/test/scala/orca/subprocess/OsProcCliRunnerTest.scala index 5901e03f3..93cabfdea 100644 --- a/tools/src/test/scala/orca/subprocess/OsProcCliRunnerTest.scala +++ b/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) diff --git a/tools/src/test/scala/orca/subprocess/QuietProcTest.scala b/tools/src/test/scala/orca/subprocess/QuietProcTest.scala index caf525763..c2e9466d5 100644 --- a/tools/src/test/scala/orca/subprocess/QuietProcTest.scala +++ b/tools/src/test/scala/orca/subprocess/QuietProcTest.scala @@ -1,5 +1,7 @@ package orca.subprocess +import ox.discard + class QuietProcTest extends munit.FunSuite: /** A subprocess writing to stderr must have its output captured into @@ -24,6 +26,13 @@ class QuietProcTest extends munit.FunSuite: val result = QuietProc.call(Seq("bash", "-c", "exit 7"), cwd = os.pwd) assertEquals(result.exitCode, 7) + test("call of a missing program fails only in the caller"): + val failedThreads = OsLibThreadFailures.during: + intercept[java.io.IOException]: + QuietProc.call(Seq("orca-test-no-such-program"), cwd = os.pwd) + .discard + assertEquals(failedThreads, Nil) + /** The heap bound itself, which no result can show: every field of * `CappedResult` reads the same whether the sink stopped at the cap or * retained the lot. From 91984a0beacec3afe55b15c05be8360c32b68ec8 Mon Sep 17 00:00:00 2001 From: Adam Warski Date: Thu, 22 Oct 2026 14:45:37 +0000 Subject: [PATCH 2/2] Address review findings --- .../orca/shell/actions/SessionAction.scala | 12 ++------ .../orca/shell/create/FlowAuthoring.scala | 7 +++-- .../scala/orca/shell/flows/FlowEditor.scala | 10 ++----- .../scala/orca/shell/run/FlowLauncher.scala | 19 +++--------- .../main/scala/orca/subprocess/Spawn.scala | 29 +++++++++++++++---- .../orca/subprocess/OsLibThreadFailures.scala | 5 ++++ .../scala/orca/subprocess/QuietProcTest.scala | 4 +++ .../scala/orca/subprocess/SpawnTest.scala | 9 ++++++ 8 files changed, 53 insertions(+), 42 deletions(-) create mode 100644 tools/src/test/scala/orca/subprocess/SpawnTest.scala diff --git a/shell/src/main/scala/orca/shell/actions/SessionAction.scala b/shell/src/main/scala/orca/shell/actions/SessionAction.scala index dc678144c..97668623a 100644 --- a/shell/src/main/scala/orca/shell/actions/SessionAction.scala +++ b/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}") diff --git a/shell/src/main/scala/orca/shell/create/FlowAuthoring.scala b/shell/src/main/scala/orca/shell/create/FlowAuthoring.scala index ccc9b2f5c..a78a2d0f7 100644 --- a/shell/src/main/scala/orca/shell/create/FlowAuthoring.scala +++ b/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,8 +264,8 @@ 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], @@ -272,7 +273,7 @@ private[shell] object FlowAuthoring: ): 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()) diff --git a/shell/src/main/scala/orca/shell/flows/FlowEditor.scala b/shell/src/main/scala/orca/shell/flows/FlowEditor.scala index e5bb5cfa6..9c8b41839 100644 --- a/shell/src/main/scala/orca/shell/flows/FlowEditor.scala +++ b/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 diff --git a/shell/src/main/scala/orca/shell/run/FlowLauncher.scala b/shell/src/main/scala/orca/shell/run/FlowLauncher.scala index bc05394f8..ab9ed2310 100644 --- a/shell/src/main/scala/orca/shell/run/FlowLauncher.scala +++ b/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 ) diff --git a/tools/src/main/scala/orca/subprocess/Spawn.scala b/tools/src/main/scala/orca/subprocess/Spawn.scala index 72bcbce87..3e2206370 100644 --- a/tools/src/main/scala/orca/subprocess/Spawn.scala +++ b/tools/src/main/scala/orca/subprocess/Spawn.scala @@ -2,6 +2,8 @@ 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. @@ -11,7 +13,7 @@ import ox.discard * 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[subprocess] object Spawn: +private[orca] object Spawn: def apply( args: Seq[String], @@ -31,19 +33,34 @@ private[subprocess] object Spawn: stderr = stderr, destroyOnExit = false ) - destroyOnJvmExit(sub) + destroyOnJvmExit(sub).discard sub - private def destroyOnJvmExit(sub: os.SubProcess): Unit = - val hook = Thread: () => - sub.destroy(sub.shutdownGracePeriod, async = false, recursive = true) + /** 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 - sub.wrapped.onExit().thenRun(() => removeHook(hook)).discard + 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. diff --git a/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala b/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala index cc90afdc2..0257bb696 100644 --- a/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala +++ b/tools/src/test/scala/orca/subprocess/OsLibThreadFailures.scala @@ -20,6 +20,11 @@ object OsLibThreadFailures: 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. diff --git a/tools/src/test/scala/orca/subprocess/QuietProcTest.scala b/tools/src/test/scala/orca/subprocess/QuietProcTest.scala index c2e9466d5..d3c8068d5 100644 --- a/tools/src/test/scala/orca/subprocess/QuietProcTest.scala +++ b/tools/src/test/scala/orca/subprocess/QuietProcTest.scala @@ -26,6 +26,10 @@ class QuietProcTest extends munit.FunSuite: val result = QuietProc.call(Seq("bash", "-c", "exit 7"), cwd = os.pwd) assertEquals(result.exitCode, 7) + test("call feeds stdin to the child"): + val result = QuietProc.call(Seq("cat"), cwd = os.temp.dir(), stdin = "hi") + assertEquals(result.out.text(), "hi") + test("call of a missing program fails only in the caller"): val failedThreads = OsLibThreadFailures.during: intercept[java.io.IOException]: diff --git a/tools/src/test/scala/orca/subprocess/SpawnTest.scala b/tools/src/test/scala/orca/subprocess/SpawnTest.scala new file mode 100644 index 000000000..e2bc1a503 --- /dev/null +++ b/tools/src/test/scala/orca/subprocess/SpawnTest.scala @@ -0,0 +1,9 @@ +package orca.subprocess + +class SpawnTest extends munit.FunSuite: + + test("the shutdown hook is removed once the process exits"): + val sub = os.proc("true").spawn(destroyOnExit = false) + val exitHook = Spawn.destroyOnJvmExit(sub) + exitHook.removed.get() + assert(!Runtime.getRuntime.removeShutdownHook(exitHook.hook))