cowork-bot: automated improvements (cowork/improve-sha-pin-checkout-20260810)

.github/workflows/cowork-auto-pr.yml

@@ -16,7 +16,7 @@ jobs:
       # without this step every run failed with "not a git repository" and no
       # PR was ever opened (fleet-wide defect: 11/11 seeded copies lacked it).
       - name: Check out the pushed branch
-        uses: actions/checkout@v4
+        uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
         with:
           ref: ${{ github.ref_name }}
           fetch-depth: 0

src/devforge/cli.py

@@ -172,16 +172,13 @@ def dispatch(ctx: typer.Context):
         # `--config file.yaml`) reach the underlying CLI instead of being
         # rejected by typer as "No such option".
         forwarded = list(ctx.args)
-        result = subprocess.run(
+        # Stream output in real time via Popen with inherited file descriptors.
+        # The previous subprocess.run(capture_output=True) buffered all output
+        # in memory, causing UX lag and potential OOM on large tool output.
+        proc = subprocess.Popen(
             [sys.executable, "-m", module_name] + forwarded,
-            capture_output=True,
-            text=True,
         )
-        if result.stdout:
-            sys.stdout.write(result.stdout)
-        if result.stderr:
-            sys.stderr.write(result.stderr)
-        sys.exit(result.returncode)
+        sys.exit(proc.wait())
 
     dispatch.__name__ = tool_name
     dispatch.__doc__ = f"Run `{pkg}` commands via the {tool_name} subcommand."

tests/test_ci_hygiene.py

@@ -0,0 +1,85 @@
+"""CI hygiene regression tests.
+
+Ensures workflow files follow security best practices:
+- All GitHub Actions are SHA-pinned (no mutable tags like @v4)
+- No silent-failure traps (|| true on validation steps)
+"""
+
+from __future__ import annotations
+
+import pytest
+import re
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+WORKFLOWS_DIR = REPO_ROOT / ".github" / "workflows"
+
+# Pattern: uses: OWNER/ACTION@REF
+# SHA-pinned refs are exactly 40 hex chars.
+# Mutable tags look like @v4, @v4.2.2, @main, @release/v1, etc.
+USES_PATTERN = re.compile(r"uses:\s*([^@\s]+)@(\S+)")
+SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$")
+
+# Local/composite actions (e.g. ./.github/actions/foo) don't need SHA pins.
+LOCAL_ACTION_PREFIX = "./"
+
+
+class TestWorkflowHygiene:
+    """Regression guards for CI workflow security and correctness."""
+
+    @pytest.fixture
+    def workflow_files(self) -> list[Path]:
+        files = list(WORKFLOWS_DIR.glob("*.yml")) + list(WORKFLOWS_DIR.glob("*.yaml"))
+        if not files:
+            pytest.skip("No workflow files found")
+        return files
+
+    def test_all_actions_sha_pinned(self, workflow_files: list[Path]) -> None:
+        """Every remote action reference must use a 40-char SHA, not a mutable tag.
+
+        Mutable tags like @v4 can be silently moved to point at different commits,
+        creating a supply-chain attack vector. SHA pins lock the exact commit.
+        """
+        violations: list[str] = []
+        for wf in workflow_files:
+            for lineno, line in enumerate(wf.read_text(encoding="utf-8").splitlines(), 1):
+                match = USES_PATTERN.search(line)
+                if not match:
+                    continue
+                action, ref = match.group(1), match.group(2)
+                # Strip inline comments (e.g. "# v4.2.2")
+                ref = ref.split("#")[0].strip()
+                if action.startswith(LOCAL_ACTION_PREFIX):
+                    continue
+                if not SHA_PATTERN.match(ref):
+                    violations.append(f"{wf.name}:{lineno} {action}@{ref}")
+
+        assert not violations, (
+            f"Found {len(violations)} mutable action reference(s). "
+            "Pin to a 40-char SHA instead:\n" + "\n".join(violations)
+        )
+
+    def test_no_silent_failure_on_validation_steps(self, workflow_files: list[Path]) -> None:
+        """Validation/lint/test steps must not suppress failures with '|| true'.
+
+        A step whose purpose is to fail the build on defects (linters, type
+        checkers, security scanners) must not hide failures. This catches the
+        'validation theater' trap where a real check is neutered.
+        """
+        validation_keywords = ("lint", "check", "test", "audit", "scan", "format", "typecheck")
+        violations: list[str] = []
+        for wf in workflow_files:
+            lines = wf.read_text(encoding="utf-8").splitlines()
+            for lineno, line in enumerate(lines, 1):
+                stripped = line.strip()
+                if "|| true" not in stripped:
+                    continue
+                # Check if this line or the step name above contains a validation keyword
+                context = " ".join(lines[max(0, lineno - 5) : lineno]).lower()
+                if any(kw in context for kw in validation_keywords):
+                    violations.append(f"{wf.name}:{lineno} {stripped[:80]}")
+
+        assert not violations, (
+            f"Found {len(violations)} validation step(s) with '|| true' suppression. "
+            "Remove the suppression so failures are visible:\n" + "\n".join(violations)
+        )

tests/test_cli.py

@@ -125,31 +125,35 @@ def test_dispatch_not_installed_shows_install_hint(self, _mock):
         assert 'pip install "git+https://github.com/Coding-Dev-Tools/devforge-cli.git[guard]"' in result.stdout
 
     @mock.patch("devforge.cli._is_tool_installed", return_value=True)
-    @mock.patch("devforge.cli.subprocess.run")
-    def test_dispatch_installed_tool_runs(self, mock_run, _mock_installed):
+    @mock.patch("devforge.cli.subprocess.Popen")
+    def test_dispatch_installed_tool_runs(self, mock_popen, _mock_installed):
         """When a tool is installed, dispatch calls the subprocess."""
-        mock_run.return_value = mock.MagicMock(returncode=0)
+        mock_proc = mock.MagicMock()
+        mock_proc.wait.return_value = 0
+        mock_popen.return_value = mock_proc
         with mock.patch("devforge.cli.sys.exit"):
             runner.invoke(app, ["guard"])
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
+        mock_popen.assert_called_once()
+        cmd = mock_popen.call_args[0][0]
         assert "api_contract_guardian" in cmd
 
     @mock.patch("devforge.cli._is_tool_installed", return_value=True)
-    @mock.patch("devforge.cli.subprocess.run")
-    def test_dispatch_forwards_tool_flags(self, mock_run, _mock_installed):
+    @mock.patch("devforge.cli.subprocess.Popen")
+    def test_dispatch_forwards_tool_flags(self, mock_popen, _mock_installed):
         """Tool flags (e.g. `--config file.yaml`) must reach the underlying CLI.
 
         Regression guard for the silent-failure trap where typer rejected any
         argument beginning with `-` as 'No such option' before the tool ran.
         With ignore_unknown_options/allow_extra_args, such flags are forwarded
         via ctx.args.
         """
-        mock_run.return_value = mock.MagicMock(returncode=0)
+        mock_proc = mock.MagicMock()
+        mock_proc.wait.return_value = 0
+        mock_popen.return_value = mock_proc
         with mock.patch("devforge.cli.sys.exit"):
             runner.invoke(app, ["guard", "--config", "x.yaml", "--verbose"])
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
+        mock_popen.assert_called_once()
+        cmd = mock_popen.call_args[0][0]
         # Underlying module is launched...
         assert "api_contract_guardian" in cmd
         # ...and the tool flags are forwarded, not swallowed by typer.

tests/test_dispatch_streaming.py

@@ -0,0 +1,97 @@
+"""Regression tests for subprocess output streaming in dispatch.
+
+The dispatch command must stream stdout/stderr in real time rather than
+buffering the entire output via capture_output=True. Long-running tools
+(deploydiff, schemaforge, configdrift on large datasets) can produce
+megabytes of output that should reach the user's terminal immediately.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from devforge.cli import app
+from typer.testing import CliRunner
+from unittest import mock
+
+runner = CliRunner()
+
+
+class TestDispatchStreaming:
+    """dispatch must NOT use subprocess.run with capture_output=True.
+
+    Real-time streaming requires subprocess.Popen (or subprocess.run with
+    stdout=None, stderr=None) so the child process inherits the parent's
+    file descriptors directly.
+    """
+
+    @mock.patch("devforge.cli._is_tool_installed", return_value=True)
+    def test_dispatch_does_not_buffer_output(self, _mock_installed):
+        """subprocess.run must NOT be called with capture_output=True.
+
+        capture_output=True buffers the entire child output in memory before
+        the parent can write anything. For tools that produce large or
+        long-running output, this is a UX regression: the user sees nothing
+        until the tool finishes, and memory usage grows unbounded.
+        """
+        with mock.patch("devforge.cli.subprocess.run") as mock_run:
+            mock_run.return_value = mock.MagicMock(returncode=0, stdout="", stderr="")
+            with mock.patch("devforge.cli.sys.exit"):
+                runner.invoke(app, ["guard", "--help"])
+
+        if mock_run.called:
+            # If subprocess.run is used, it must NOT capture output
+            call_kwargs = mock_run.call_args[1] if mock_run.call_args[1] else {}
+            assert call_kwargs.get("capture_output") is not True, (
+                "dispatch uses subprocess.run(capture_output=True) which buffers "
+                "all output. Use subprocess.Popen or stdout=None to stream."
+            )
+            assert call_kwargs.get("stdout") is not subprocess.PIPE, (
+                "dispatch uses stdout=PIPE which buffers output. Use stdout=None to inherit the parent's stdout."
+            )
+
+    @mock.patch("devforge.cli._is_tool_installed", return_value=True)
+    def test_dispatch_uses_popen_or_inherited_fds(self, _mock_installed):
+        """dispatch should use subprocess.Popen for real-time streaming,
+        or subprocess.run without capture (stdout=None, stderr=None).
+        """
+        with (
+            mock.patch("devforge.cli.subprocess.Popen") as mock_popen,
+            mock.patch("devforge.cli.subprocess.run") as mock_run,
+        ):
+            # Set up Popen mock to simulate a successful run
+            mock_proc = mock.MagicMock()
+            mock_proc.wait.return_value = 0
+            mock_popen.return_value = mock_proc
+
+            with mock.patch("devforge.cli.sys.exit"):
+                runner.invoke(app, ["guard"])
+
+            # Either Popen was used (preferred for streaming)
+            # or subprocess.run was used WITHOUT capture_output
+            if mock_popen.called:
+                # Good: Popen streams by default
+                assert True
+            elif mock_run.called:
+                kwargs = mock_run.call_args[1] if mock_run.call_args[1] else {}
+                assert kwargs.get("capture_output") is not True
+                assert kwargs.get("stdout") is not subprocess.PIPE
+            else:
+                raise AssertionError("Neither subprocess.Popen nor subprocess.run was called")
+
+    @mock.patch("devforge.cli._is_tool_installed", return_value=True)
+    def test_dispatch_exit_code_propagates(self, _mock_installed):
+        """The child process exit code must propagate to the parent."""
+        with mock.patch("devforge.cli.subprocess.Popen") as mock_popen:
+            mock_proc = mock.MagicMock()
+            mock_proc.wait.return_value = 42
+            mock_popen.return_value = mock_proc
+
+            with mock.patch("devforge.cli.sys.exit") as mock_exit:
+                runner.invoke(app, ["guard"])
+
+            if mock_popen.called:
+                # sys.exit(42) raises SystemExit; CliRunner catches it and
+                # may call sys.exit(0) afterward. Check that 42 was among
+                # the calls rather than asserting exactly one call.
+                exit_codes = [c.args[0] for c in mock_exit.call_args_list]
+                assert 42 in exit_codes, f"Expected exit code 42 in {exit_codes}"