core: Extract duplicated subprocess spawn/wait boilerplate into a shared helper

[rafael-graunke]

Fixes #3194

Summary

Adds run_subprocess() to commonwealth.utils.general and replaces the
asyncio.create_subprocess_exec + communicate() boilerplate that was
copy-pasted across the codebase.

The issue mentions AbstractRouter.start and helper/main.py:ping as the
places to share this with. In practice the duplication is bigger than that:
9 call sites had the identical spawn/wait/decode pattern:

• commonwealth/utils/DHCPDiscovery.py
• services/helper/main.py (ping)
• services/disk_usage/main.py (x2: du scan, disktest speed test)
• services/recorder_extractor/main.py (x5: mcap doctor, mcap recover,
  gst-discoverer-1.0, gst-play-1.0, mcap-foxglove-video-extract)

AbstractRouter.start was deliberately not touched: it spawns a long-lived
router process, stores the handle on self._subprocess for exit() and
start_house_keepers() to use afterward, and only calls communicate() on
the failure path. That's a "supervise a daemon" lifecycle, not "run to
completion and collect output", so forcing it through the new helper would
mean either killing the daemon immediately or leaking the Process object
back out, which defeats the point of the abstraction.

Bug fix included

While extracting recorder_extractor/main.py's build_thumbnail_bytes, found
that the success path referenced an undefined stdout variable
(if play_proc.returncode != 0 or not stdout:, only stdout_bytes was ever
assigned). On success (returncode == 0) this raised NameError, since or
only short-circuits past it on the failure path. Fixed as part of the
extraction (now checks stdout_bytes), since the refactor touches that exact
line anyway.

Testing

• Added unit tests for run_subprocess (returncode, stdout/stderr split, and
  the merge_stderr bytes-vs-None case), ran locally against this repo's
  asyncio_mode=STRICT pytest config, all pass.
• No existing test coverage for disk_usage, recorder_extractor,
  helper/main.py, or DHCPDiscovery.py. Verified behavior preservation by
  inspection: each diff only replaces the spawn+communicate block, every
  downstream .decode(...) call and parsing/log line is outside the diff,
  unchanged. No automated test harness exists for these 4 services to verify
  at runtime.

Summary by Sourcery

Introduce a shared async subprocess helper and adopt it across services that previously duplicated spawn/wait logic.

New Features:

• Add run_subprocess utility to commonwealth.utils.general for running async commands with optional stderr merging.

Bug Fixes:

• Fix recorder_extractor thumbnail generation by checking the correct stdout variable when validating gst-play-1.0 output.

Enhancements:

• Refactor recorder_extractor, disk_usage, DHCPDiscovery, and helper ping to use the shared run_subprocess helper instead of inline asyncio subprocess boilerplate.

Tests:

• Add unit tests for run_subprocess covering stdout/stderr capture, non-zero return codes, and merged stderr behavior.

[cursor]

Bugbot is not enabled for this team, so this pull request was not reviewed.

Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs.

[sourcery-ai]

Hey - I've found 1 issue, and left some high level feedback:

• In run_subprocess, the return type is annotated as Optional[int] for the return code, but the implementation always returns an int; either adjust the annotation to int or make the None case explicit if you intend to represent a failure-to-start scenario.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `run_subprocess`, the return type is annotated as `Optional[int]` for the return code, but the implementation always returns an `int`; either adjust the annotation to `int` or make the `None` case explicit if you intend to represent a failure-to-start scenario.

## Individual Comments

### Comment 1
<location path="core/libs/commonwealth/src/commonwealth/utils/general.py" line_range="225-234" />
<code_context>
     return _file_is_open_logic_lsof(result.returncode, result.stdout.strip(), result.stderr.strip())


+async def run_subprocess(command: List[str], merge_stderr: bool = False) -> Tuple[Optional[int], bytes, bytes]:
+    """Run a command asynchronously and wait for it to finish.
+
+    Args:
+        command: Command and arguments to execute.
+        merge_stderr: If True, redirect stderr into stdout (mirrors `subprocess.STDOUT`).
+            The returned stderr is empty bytes in this case.
+
+    Returns:
+        Tuple of (returncode, stdout, stderr) as raw bytes. Callers that expect text output
+        are responsible for decoding it themselves.
+    """
+    process = await asyncio.create_subprocess_exec(
+        *command,
+        stdout=asyncio.subprocess.PIPE,
</code_context>
<issue_to_address>
**suggestion:** Return code type should be non-optional to better reflect actual behavior and simplify callers.

Since the function always returns `process.returncode`, `None` should not occur in normal operation. Typing this as `Tuple[int, bytes, bytes]` instead of `Tuple[Optional[int], bytes, bytes]` makes the API more accurate and avoids burdening callers with an impossible `None` case. If you need to represent failures before the process is spawned, consider propagating the exception or modeling that failure mode explicitly rather than using `None` for the return code.

Suggested implementation:

```python
async def run_subprocess(command: List[str], merge_stderr: bool = False) -> Tuple[int, bytes, bytes]:

```

If `Optional[int>` is not used elsewhere in this module, you can remove it from the `from typing import ...` import list to keep the imports clean. No other behavioral changes are required since `process.returncode` is already always an `int` after `communicate()` completes.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion: Return code type should be non-optional to better reflect actual behavior and simplify callers.

Since the function always returns process.returncode, None should not occur in normal operation. Typing this as Tuple[int, bytes, bytes] instead of Tuple[Optional[int], bytes, bytes] makes the API more accurate and avoids burdening callers with an impossible None case. If you need to represent failures before the process is spawned, consider propagating the exception or modeling that failure mode explicitly rather than using None for the return code.

Suggested implementation:

async def run_subprocess(command: List[str], merge_stderr: bool = False) -> Tuple[int, bytes, bytes]:

If Optional[int> is not used elsewhere in this module, you can remove it from the from typing import ... import list to keep the imports clean. No other behavioral changes are required since process.returncode is already always an int after communicate() completes.

[rafael-graunke]

Coverage gate failing (71.91%, need 75%), not what it looks like

Not that the refactored code is undertested. It's a side effect of how coverage.py scopes files.

commonwealth/utils/general.py had zero test coverage before this PR, but it also wasn't in the coverage denominator. Nothing in the primary suite imported it (its only consumers, disk_usage/recorder_extractor, have no tests; helper is secondary-suite scope). Coverage.py only reports files something actually imports during the run.

This PR's test_general.py (added to test run_subprocess) is the first thing to import commonwealth.utils.general in the primary suite. That flips ~195 previously-invisible statements (plus ~68 more in the transitively
imported commands.py) from "absent" to "mostly uncovered," dragging the total down. The code I actually wrote is fully tested; the regression is pre-existing untested code becoming visible for the first time.

Simplest fix: move run_subprocess out of general.py into its own module (e.g. commonwealth/utils/subprocess_runner.py). Then test_general.py goes away, the 9 call sites import from the new module instead, and nothing left imports general.py/commands.py in the primary suite, so both go back to
invisible, same as master. No coverage-gate discussion needed.

Alternative, if you'd rather have real coverage on that file regardless: I wrote a full test suite for the rest of both files locally (55 new tests, 94%/96% file coverage). Happy to bundle it in or split it into a follow-up PR.

Let me know which you'd prefer.