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/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..3e2206370 --- /dev/null +++ b/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 => () 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..0257bb696 --- /dev/null +++ b/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)) 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..d3c8068d5 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,17 @@ 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]: + 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. 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))