Simplify `BaseRunner.run` method

[MattBurn]

I am working through issue #224 and I think the current BaseRunner.run output API could be simplified.

My understanding of the current behavior is:

• capture=True uses subprocess.PIPE so stdout and stderr are returned on the CompletedProcess.
• silent=True redirects output to os.devnull, unless overridden.
• stdout / stderr can be set to Path objects to write the corresponding stream to disk.

This creates a few ambiguous combinations:

• capture=True, silent=True silently ignores silent.
• capture=False, stdout=path or stderr=path still redirects that stream to a file, despite capture being disabled.
• stdout / stderr are currently path-oriented, so users cannot pass an open file object, in-memory buffer, callback, logger, or multiple destinations.
• Internal code that needs to inspect stderr must either force capture=True or write stderr to a temporary file and read it back.

This is particularly awkward for OrcaMmRunner.run_orca_mm, which needs to inspect stderr to decide whether to raise OrcaMmException. At the moment, that requires coupling the internal error-checking behavior to the public output-routing behavior.

I think the issue is that capture, silent, stdout, and stderr are all competing controls for the same underlying concern: stream handling.

Proposal

Replace the current flag-based API with a stream-target API.

The runner would always capture stdout and stderr internally and return them on the result. The stdout and stderr arguments would only describe additional destinations for each stream.

For example:

StreamTarget = (
    None
    | Path
    | IO[str]
    | Callable[[str], None]
    | Sequence["StreamTarget"]
)

Proposed semantics:

• None: no external output destination
• Path: write the stream to a file
• IO[str]: write the stream to an open file-like object
• Callable[[str], None]: stream chunks to a callback
• Sequence[StreamTarget]: tee the stream to multiple destinations

The run method could then look something like:

def run(
    self,
    binary: OrcaBinary,
    args: Sequence[str] = (),
    *,
    stdout: StreamTarget = None,
    stderr: StreamTarget = None,
    stdin: str | None = None,
    cwd: Path | None = None,
    timeout: int | None = None,
) -> RunResult:
    ...

Where RunResult exposes the captured output:

@dataclass
class RunResult:
    binary: str
    args: Sequence[str]
    returncode: int
    stdout: str
    stderr: str

This removes the need for both capture and silent:

# Current silent behavior
runner.run(..., stdout=None, stderr=None)

# Write stderr to a file while still allowing internal inspection
result = runner.run(..., stderr=Path("orca_mm.err"))

# Print stdout live
result = runner.run(..., stdout=lambda chunk: print(chunk, end=""))

# Print stdout live and write it to disk
result = runner.run(
    ...,
    stdout=[
        lambda chunk: print(chunk, end=""),
        Path("orca.out"),
    ],
)

For OrcaMmRunner.run_orca_mm, this would simplify the implementation because stderr would always be available on the returned result:

result = self.run(
    OrcaBinary.ORCA_MM,
    args,
    stdout=stdout,
    stderr=stderr,
)

if raise_on_error and result.stderr:
    raise OrcaMmException(command, arguments, result.stderr)

This would decouple internal validation from user-facing output routing. Users can decide where output should go, while the runner can still inspect process output consistently.

One possible caveat is memory usage for very large outputs, since this proposal implies always retaining stdout and stderr in memory. If that becomes a concern, we could add a later escape hatch such as max_capture_bytes, but I think the simpler default API is preferable unless large-output cases become a real problem.

[timmyte]

@MattBurn
Thanks for this detailed issue and also pointing out any caveats directly.

I like the idea and do see the drawbacks of the current API.
Feel free to open another PR with your proposed changes.

I would have some additions to your proposal:

1. Instead of using None in StreamTarget, I would go with a boolean with behaves like capture.
   Where True is the default and means capturing the output and False redirect to subprocess.DEVNULL. This would offer one way to avoid a huge memory consumption.
   This also does not conflict with your proposal that the a runner can investigate the output independently, because for each subclass of Runner it can be decided what parts of the .run() API are exposed to higher layers.
   1.2) Regarding a max_capture_bytes argument: I would instead offer a custom buffer that acts like a sliding window buffer with a fixed but settable size -> SlideTextIO(size=500)
   But agreed, this is a nice-to-have-in-the-future thing.
2. I also like the idea of RunResult. However, I would keep it's design close to subprocess.CompletedProcess, so how about adding check_returncode(). Furthermore, instances of that class should not be mutable.

@dataclass(frozen=True)
class RunResult:
    binary: str | OrcaBinary
    args: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    
    def check_returncode(self) -> bool:
        return self.returncode == 0 

    def get_signal(self) -> int | None:
        """Check and return IPC signals."""
        if self.returncode < 0:
            return abs(self.returncode)
        else:
            return None

3. I think a callback function like Callable[[str], None] wouldn't work out-of-box, as subprocess's stdout and stderr expect an IO object. I guess, one would to have overload the .write() function of an IO object, which introduces a custom class. What do you think?
4. If StreamTarget can also a be Sequence of itself, then it should not contain None, as this would allow [Path, None] by the type definition but it is nonsensical.

[MattBurn]

Having a Boolean also leads to the problem with a sequence where we could have [True, False] as the stream target what if we instead use a Literal["capture"] and default to an empty sequence to denote the no capture case?

Good catch, the data class should definitely be immutable. I can look into the CompletedProcess API to make it as identical as possible.

The plan would be to take the stream targets and turn them all into implementations of TextIO so for each StreamTarget we can implement the conversion and implement something like a MultiTextIO object which dispatches over all of the targets.

[timmyte]

Having a Boolean also leads to the problem with a sequence where we could have [True, False] as the stream target what if we instead use a Literal["capture"] and default to an empty sequence to denote the no capture case?

Darn, you are right.
I'm not a fan of literals. I would prefer to rather slim down StreamTarget a little:

def run(stdout: bool | StreamTarget = True, ...)

# where 
type StreamTarget = (
    Path
    | IO[str]
    | Callable[[str], None]
    | Sequence["StreamTarget"]
)

To avoid my personally bias here, I also gonna ask my colleague:
@haneug
Any preference from your side?

Good catch, the data class should definitely be immutable. I can look into the CompletedProcess API to make it as identical as possible.

Thanks!!

The plan would be to take the stream targets and turn them all into implementations of TextIO so for each StreamTarget we can implement the conversion and implement something like a MultiTextIO object which dispatches over all of the targets.

Sounds like a plan. And with that the implementation of max_capture_bytes would also be easier.
Though, I still don't see the point for a callback function. IMO this would only be useful if we would work with threads or asynchronous calls (which is on the todo list: #73)

[MattBurn]

I took a stab at an initial implementation and took the same opinion as you by dropping the literal, I did still allow str which gets converted to a Path but that will be easy enough to remove if it does not fit well with the rest of the codebase.

I took the approach of solving the capture problem by allowing subprocess.PIPE to be passed in. I did want to only allow the subprocess.PIPE value (-1) to be passed in using the following:

CaptureType = Literal[-1]
if TYPE_CHECKING is True:
    assert subprocess.PIPE == -1
CAPTURE: Final[CaptureType] = subprocess.PIPE

And this passed my local ty and pyright type checks but not the mypy CI checks so I had to loosen CaptureType to be an int. This means that passing any int as a target will pass type checking but will still cause a ValueError at runtime.

max_capture_bytes should be easy to add to the TextStreamFanout I can add this to the current implementation or we can defer it to a later issue.

I did not know about #73. I have implemented an AsyncRunner and AsyncCalculator for an internal tool and implemented callbacks so that I could pipe stdout to various places such as a logger and provide heartbeats to a job orchestrator that the tool uses. Allowing callbacks gives you the ability to pass in something like stdout=logger.info. With the async implementation, a callback was trivial but you are right that the sync implementation required the use of threads.

In the past when running orca manually I have been able to determine that a calculation is not going to be successful and terminated the program early. If I were to automate this type of use case then I would need to be able to monitor the output and send a signal to the runner to terminate. Not sure if something like this fits with your plans for opi but using a callback that tails the output alongside a threading.Event would be a simple way to implement it.