feat(container): emit canonical argus-results.json + persist raw scanner output

Two related fixes for the container-scan flow, addressing user
reports that:

1. ``argus view`` doesn't display container vulnerabilities — the
   container-scan flow only wrote a domain-shaped
   ``container-scan.json`` (per-image counts, ``container_count``,
   etc.) which the viewers don't know how to render. The viewers
   consume the canonical ``argus-results.json`` shape produced by
   source scans.
2. The ``argus-results/`` dir doesn't preserve the raw per-scanner
   output files (``trivy-results.json``, ``grype-results.json``,
   ``syft-sbom.json``) — they live in a tempdir that gets wiped at
   the end of ``scan_image``. Users who want forensics, audit
   trails, or manual triage have nowhere to look.

Both rooted in the same architectural drift: the container flow
diverged from the source-scan output contract. This PR re-aligns it.

Canonical ScanSummary for container scans
- ``_cmd_container_scan`` now builds a canonical ``ScanSummary``
  alongside the existing ``ContainerScanSummary``: each container
  target maps to ``ScanResult(scanner=f"container/<name>",
  findings=combined, metadata={image_ref, build_success,
  scanner_errors, scan_error})``.
- The JSON reporter writes that to ``argus-results.json``
  unconditionally (matches the source-scan canonical-artifact
  contract from PR #111).
- The SARIF reporter now consumes the same canonical summary
  instead of building a one-off conversion locally.
- The domain-shaped ``container-scan.json`` (with ``container_count``,
  per-image stats) is preserved for backward compat with downstream
  tooling that consumes it; it just lives alongside the canonical
  artifact rather than instead of it.
- ``argus view`` opens container scan results without any new code
  on the viewer side — it just sees ``ScanResult`` rows named
  ``container/<image>`` and renders them like any other scanner.

Raw scanner output persistence
- ``scan_image`` gains a ``raw_output_dir: Path | None`` parameter.
  When set, copies ``trivy-results.json``, ``grype-results.json``,
  and ``syft-sbom.json`` into that directory before the tempdir is
  cleaned up. Best-effort — copy errors log a warning but don't
  fail the scan.
- ``ContainerEngine`` reads ``_raw_output_root`` from its config
  dict (the dispatcher sets this) and threads a per-target subdir
  to ``scan_image`` as ``<root>/<target.name>/``.
- ``_cmd_container_scan`` defaults to ON: raw outputs land at
  ``<output_dir>/raw/<image>/``. Opt out via
  ``--no-keep-raw`` flag or ``containers.keep_raw: false`` in
  argus.yml. CLI flag wins on conflict (explicit > implicit).
- 0-byte files are explicitly skipped (they're failure signals
  upstream; persisting them would make a known-bad output look
  authoritative on disk).

Documentation
- ``argus.example.yml`` documents ``containers.keep_raw: true`` in
  the commented schema block, alongside the existing ``images``,
  ``discover``, and ``scanners`` keys.

Tests (+5)
- ``TestScanImageRawOutputPersistence`` (4 cases): all artifacts
  copied when dir supplied, no copy when ``raw_output_dir=None``,
  partial coverage (only trivy ran) doesn't block the others, 0-byte
  files are explicitly skipped.
- ``TestContainerCanonicalScanSummary`` (1 case): each
  ContainerScanResult maps to a canonical ScanResult(scanner=
  "container/<name>") with combined findings; metadata lifts onto
  the ScanResult; round-trips through ``ScanSummary.to_dict()``
  unchanged so the viewer gets the same shape it expects.

Validation
- Full SDK suite: 1464 passed (+5 net), 8 skipped.

Author: eFAILution

argus.example.yml

@@ -101,6 +101,15 @@ reporting:
 #
 #   # Override which sub-scanners run; default is trivy + grype.
 #   scanners: [trivy, grype, syft]
+#
+#   # Persist raw per-scanner artifacts (trivy-results.json,
+#   # grype-results.json, syft-sbom.json) under
+#   # ``<output_dir>/raw/<image>/`` alongside the canonical
+#   # argus-results.json. Default is true so the artifacts are
+#   # available for forensics, audit, or manual triage. Set false
+#   # (or pass --no-keep-raw on the CLI) to skip — saves a few MB
+#   # per image when running in tight CI environments.
+#   keep_raw: true
 
 # Execution backend configuration
 execution:

argus/cli.py

@@ -616,6 +616,18 @@ def _build_scan_parser(subparsers: argparse._SubParsersAction) -> None:
         default=None,
         help="Sub-scanners for container scanning: trivy,grype,syft (default: trivy,grype)",
     )
+    container_group.add_argument(
+        "--no-keep-raw",
+        action="store_true",
+        dest="no_keep_raw",
+        help="Do not persist raw per-scanner output (trivy-results.json, "
+             "grype-results.json, syft-sbom.json) under "
+             "<output_dir>/raw/<image>/. By default raw artifacts are "
+             "kept alongside the canonical argus-results.json so users "
+             "can drill into individual scanner output for forensics or "
+             "manual triage. Set ``containers.keep_raw: false`` in argus.yml "
+             "for the same effect via config.",
+    )
 
     # ZAP DAST flags (used with: argus scan zap)
     dast_group = scan_parser.add_argument_group(
@@ -1733,6 +1745,18 @@ def _cmd_container_scan(
     output_dir = _make_run_dir(base_dir)
     formats = args.formats or ["terminal", "markdown"]
 
+    # Decide whether to persist raw per-scanner outputs alongside the
+    # canonical argus-results.json. Default is ON — the user just ran
+    # a scan and would expect those artifacts to be available for
+    # manual triage. Opt out via ``--no-keep-raw`` (CLI) or
+    # ``containers.keep_raw: false`` (argus.yml). CLI flag wins on
+    # conflict, matching the rest of the dispatcher's
+    # explicit-over-implicit posture.
+    keep_raw_config = config.get("keep_raw", True)
+    keep_raw = bool(keep_raw_config) and not getattr(args, "no_keep_raw", False)
+    if keep_raw:
+        config["_raw_output_root"] = str(Path(output_dir) / "raw")
+
     # Run
     try:
         engine = ContainerEngine(config)
@@ -1745,6 +1769,42 @@ def _cmd_container_scan(
         print(f"Error: container scan failed: {exc}", file=sys.stderr)
         return EXIT_ERROR
 
+    # Build a canonical ScanSummary view of the container results so
+    # the standard reporters (json → argus-results.json, sarif) and
+    # ``argus view`` can consume container scans the same way they
+    # consume source scans. Each container target becomes a
+    # ScanResult; the per-image domain metadata (image_ref, build
+    # status, scanner_errors) lifts onto ScanResult.metadata so the
+    # browser dashboard and exporters surface it.
+    from argus.core.models import ScanResult, ScanSummary
+    canonical_results = [
+        ScanResult(
+            scanner=f"container/{r.name}",
+            findings=list(r.combined_findings),
+            metadata={
+                "image_ref": r.image_ref,
+                "build_success": r.build_success,
+                **(
+                    {"scanner_errors": dict(r.scanner_errors)}
+                    if r.scanner_errors else {}
+                ),
+                **(
+                    {"scan_error": r.scan_error}
+                    if getattr(r, "scan_error", None) else {}
+                ),
+            },
+        )
+        for r in summary.results
+    ]
+    canonical_summary = ScanSummary(results=canonical_results)
+
+    # Always emit argus-results.json — same canonical-artifact
+    # contract the source-scan flow established. ``argus view`` and
+    # the audit manifest both consume this regardless of what the
+    # user listed in ``formats``.
+    from argus.reporters import get_reporter
+    get_reporter("json").report(canonical_summary, output_dir)
+
     # Reports
     for fmt in formats:
         if fmt == "markdown":
@@ -1755,16 +1815,15 @@ def _cmd_container_scan(
         elif fmt == "terminal":
             _print_container_terminal(summary)
         elif fmt == "json":
+            # Domain-shaped per-image summary (container_count etc.)
+            # lives at container-scan.json. The canonical
+            # argus-results.json was already written above; this
+            # is the supplementary domain artifact for tooling that
+            # wants per-image stats without parsing findings.
             _write_container_json(summary, output_dir)
         elif fmt == "sarif":
-            from argus.core.models import ScanResult, ScanSummary
-            from argus.reporters import get_reporter
-            results = [
-                ScanResult(scanner=f"container/{r.name}", findings=r.combined_findings)
-                for r in summary.results
-            ]
             sarif_reporter = get_reporter("sarif")
-            sarif_reporter.report(ScanSummary(results=results), output_dir)
+            sarif_reporter.report(canonical_summary, output_dir)
 
     # Exit code — scanner failures are always non-zero
     scan_failures = getattr(summary, "scan_failures", 0)

argus/container/engine.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+from pathlib import Path
 
 from .builder import build_image
 from .discovery import (
@@ -130,10 +131,20 @@ def _process_target(self, target: ContainerTarget) -> ContainerScanResult:
             self._built_images.append(target.image_ref)
 
         try:
+            # If the dispatcher set ``_raw_output_root`` in the
+            # config dict, persist this target's raw scanner outputs
+            # under ``<root>/<target.name>/``. Caller controls
+            # whether this is set (CLI flag + config opt-out); the
+            # engine just threads it through.
+            raw_root = self.config.get("_raw_output_root")
+            target_raw_dir = (
+                Path(raw_root) / target.name if raw_root else None
+            )
             return scan_image(
                 target,
                 scanners=self._scanners(),
                 sbom=self._sbom_enabled(),
+                raw_output_dir=target_raw_dir,
             )
         except OSError as exc:
             # Disk full, permission denied, etc.

argus/container/scanner.py

@@ -123,6 +123,7 @@ def scan_image(
     target: ContainerTarget,
     scanners: tuple[str, ...] = ("trivy", "grype"),
     sbom: bool = True,
+    raw_output_dir: Path | None = None,
 ) -> ContainerScanResult:
     """Scan a single container image with trivy and/or grype.
 
@@ -132,7 +133,19 @@ def scan_image(
 
     For locally-built images, scanners reference the local Docker daemon.
     Per-scanner errors are caught and recorded, not swallowed.
+
+    ``raw_output_dir``: when supplied, the raw scanner output files
+    (``trivy-results.json``, ``grype-results.json``, ``syft-sbom.json``)
+    are copied into this directory before the temp dir is cleaned up.
+    Lets users preserve full per-scanner artifacts for forensics,
+    audit, or manual triage workflows alongside the canonical
+    ``argus-results.json``. ``None`` (the default) means transient
+    output — historic behavior.
     """
+    import shutil as _shutil  # local import to avoid shadowing the
+                              # module-level ``shutil`` reference used
+                              # by ``shutil.which`` checks below.
+
     trivy_findings: list[Finding] = []
     grype_findings: list[Finding] = []
     scanner_errors: dict[str, str] = {}
@@ -164,6 +177,29 @@ def scan_image(
         if sbom and "syft" not in scanners:
             _run_syft(target.image_ref, tmp_path)
 
+        # Persist raw scanner artifacts (best-effort) before the
+        # tempdir is wiped. We copy whatever files exist; missing
+        # files (e.g. grype failed before writing) just don't get
+        # copied — the structured ``scanner_errors`` already records
+        # why. Errors during copy are non-fatal: the scan succeeded,
+        # the canonical JSON is still emitted upstream.
+        if raw_output_dir is not None:
+            try:
+                raw_output_dir.mkdir(parents=True, exist_ok=True)
+                for fname in (
+                    "trivy-results.json",
+                    "grype-results.json",
+                    "syft-sbom.json",
+                ):
+                    src = tmp_path / fname
+                    if src.exists() and src.stat().st_size > 0:
+                        _shutil.copy2(src, raw_output_dir / fname)
+            except OSError as exc:
+                logger.warning(
+                    "Failed to persist raw scanner outputs to %s: %s",
+                    raw_output_dir, exc,
+                )
+
     combined = deduplicate_findings(trivy_findings, grype_findings)
 
     return ContainerScanResult(

argus/tests/test_container_scanner_runners.py

@@ -333,6 +333,177 @@ def fake_run(cmd, **_kwargs):
 # ───────────────────────────────────────────────
 
 
+class TestScanImageRawOutputPersistence:
+    """``scan_image(raw_output_dir=...)`` copies raw scanner artifacts
+    into a caller-supplied directory so ``argus-results/<run>/raw/``
+    can preserve trivy/grype/syft per-scanner output for forensics
+    after the underlying tempdir is cleaned up."""
+
+    def _stub_runners(self, monkeypatch, write_files=("trivy", "grype")):
+        """Replace the live scanner runners with stubs that drop the
+        files we'd expect to see on a successful real run. Lets these
+        tests focus on the copy/persistence layer without touching
+        the actual binaries."""
+        from argus.container import scanner as scanner_mod
+
+        def fake_trivy(image_ref, tmp_path, local=False):
+            if "trivy" in write_files:
+                (tmp_path / "trivy-results.json").write_text('{"Results": []}')
+            return []
+
+        def fake_grype(image_ref, tmp_path, local=False):
+            if "grype" in write_files:
+                (tmp_path / "grype-results.json").write_text('{"matches": []}')
+            return []
+
+        def fake_syft(image_ref, tmp_path):
+            if "syft" in write_files:
+                (tmp_path / "syft-sbom.json").write_text('{"artifacts": []}')
+
+        monkeypatch.setattr(scanner_mod, "_run_trivy", fake_trivy)
+        monkeypatch.setattr(scanner_mod, "_run_grype", fake_grype)
+        monkeypatch.setattr(scanner_mod, "_run_syft", fake_syft)
+
+    def test_raw_outputs_copied_when_dir_supplied(self, tmp_path, monkeypatch):
+        from argus.container.scanner import scan_image
+        from argus.container.discovery import ContainerTarget
+
+        self._stub_runners(monkeypatch, write_files=("trivy", "grype", "syft"))
+
+        target = ContainerTarget(name="app", image_ref="myapp:dev")
+        raw_dir = tmp_path / "raw" / "app"
+
+        scan_image(target, sbom=True, raw_output_dir=raw_dir)
+
+        # All three artifacts persisted at the expected names.
+        assert (raw_dir / "trivy-results.json").exists()
+        assert (raw_dir / "grype-results.json").exists()
+        assert (raw_dir / "syft-sbom.json").exists()
+        # Contents survived intact.
+        assert "Results" in (raw_dir / "trivy-results.json").read_text()
+
+    def test_no_copy_when_raw_output_dir_is_none(self, tmp_path, monkeypatch):
+        # Default path — historic behavior — leaves no artifacts on
+        # disk after the tempdir cleanup.
+        from argus.container.scanner import scan_image
+        from argus.container.discovery import ContainerTarget
+
+        self._stub_runners(monkeypatch)
+
+        target = ContainerTarget(name="app", image_ref="myapp:dev")
+        scan_image(target, sbom=False, raw_output_dir=None)
+
+        # No `raw/` directory was created (the test's tmp_path is
+        # otherwise empty).
+        assert not (tmp_path / "raw").exists()
+
+    def test_partial_outputs_persisted_when_some_scanners_skipped(
+        self, tmp_path, monkeypatch,
+    ):
+        # Only trivy ran (grype skipped or failed); the raw dir
+        # contains just trivy's file. Missing files don't block the
+        # copy of the ones that exist.
+        from argus.container.scanner import scan_image
+        from argus.container.discovery import ContainerTarget
+
+        self._stub_runners(monkeypatch, write_files=("trivy",))
+
+        target = ContainerTarget(name="app", image_ref="myapp:dev")
+        raw_dir = tmp_path / "raw" / "app"
+
+        scan_image(
+            target, scanners=("trivy",), sbom=False,
+            raw_output_dir=raw_dir,
+        )
+
+        assert (raw_dir / "trivy-results.json").exists()
+        assert not (raw_dir / "grype-results.json").exists()
+        assert not (raw_dir / "syft-sbom.json").exists()
+
+    def test_zero_byte_files_are_not_persisted(self, tmp_path, monkeypatch):
+        # 0-byte files are an explicit failure signal upstream
+        # (``_validate_scanner_output`` rejects them). Don't copy
+        # them — the persistence layer should never make a 0-byte
+        # file look authoritative on disk.
+        from argus.container import scanner as scanner_mod
+        from argus.container.scanner import scan_image
+        from argus.container.discovery import ContainerTarget
+
+        def fake_trivy(image_ref, tmp_path, local=False):
+            (tmp_path / "trivy-results.json").touch()  # 0-byte
+            return []
+
+        monkeypatch.setattr(scanner_mod, "_run_trivy", fake_trivy)
+        monkeypatch.setattr(scanner_mod, "_run_grype", lambda *a, **kw: [])
+        monkeypatch.setattr(scanner_mod, "_run_syft", lambda *a, **kw: None)
+
+        raw_dir = tmp_path / "raw" / "app"
+        scan_image(
+            ContainerTarget(name="app", image_ref="myapp:dev"),
+            scanners=("trivy",), sbom=False, raw_output_dir=raw_dir,
+        )
+        # Either the dir doesn't exist (nothing copied) or it's empty.
+        if raw_dir.exists():
+            assert not list(raw_dir.iterdir())
+
+
+class TestContainerCanonicalScanSummary:
+    """The container scan flow now also emits the canonical
+    ScanSummary shape (the same one source scans use), so
+    ``argus view`` and the JSON reporter can render container
+    findings without a separate code path."""
+
+    def test_each_target_becomes_a_scanresult_with_combined_findings(
+        self, tmp_path, monkeypatch,
+    ):
+        # Exercises the cli.py snippet that maps ContainerScanResult
+        # → ScanResult(scanner=f"container/{name}", ...). Tests a
+        # representative subset of the conversion in isolation.
+        from argus.core.models import ScanResult, ScanSummary, Finding, Severity
+        from argus.container.scanner import (
+            ContainerScanResult, ContainerScanSummary,
+        )
+
+        f1 = Finding(id="CVE-2024-1", severity=Severity.HIGH, title="t1",
+                     cve="CVE-2024-1", scanner="trivy")
+        f2 = Finding(id="CVE-2024-2", severity=Severity.MEDIUM, title="t2",
+                     cve="CVE-2024-2", scanner="grype")
+
+        container_summary = ContainerScanSummary(
+            results=[
+                ContainerScanResult(
+                    name="webapp",
+                    image_ref="myorg/webapp:1.0",
+                    combined_findings=[f1, f2],
+                    scanner_errors={},
+                ),
+            ],
+        )
+
+        # Mirror cli.py's mapping logic.
+        canonical = ScanSummary(results=[
+            ScanResult(
+                scanner=f"container/{r.name}",
+                findings=list(r.combined_findings),
+                metadata={
+                    "image_ref": r.image_ref,
+                    "build_success": r.build_success,
+                },
+            )
+            for r in container_summary.results
+        ])
+
+        # The canonical summary round-trips through the same
+        # serialization the source-scan flow uses, so ``argus view``
+        # treats container findings identically.
+        as_dict = canonical.to_dict()
+        assert "results" in as_dict
+        assert as_dict["results"][0]["scanner"] == "container/webapp"
+        assert len(as_dict["results"][0]["findings"]) == 2
+        # Per-image metadata lifts onto the ScanResult.
+        assert as_dict["results"][0]["metadata"]["image_ref"] == "myorg/webapp:1.0"
+
+
 class TestOrchestratorRecordsScannerError:
     """Closing the loop: when ``_run_grype`` raises RuntimeError, the
     orchestrator must catch it and record under ``scanner_errors`` so