feat(scanner): emit Actions poll analyzer findings without double-count

.github/workflows/actions-poll-analyzer-coverage.yml

@@ -59,6 +59,7 @@ jobs:
             --source=appguardrail_core.actions_poll_analyzer \
             -m pytest -q \
             tests/test_actions_poll_structural_analyzer.py \
+            tests/test_actions_poll_analyzer_emit.py \
             tests/test_github_actions_poll_bounds.py \
             tests/test_github_actions_poll_bound_control_flow_regression.py \
             tests/test_github_actions_poll_bound_aliases.py \
@@ -91,5 +92,6 @@ jobs:
           python -m compileall -q
           appguardrail_core/actions_poll_analyzer.py
           tests/test_actions_poll_structural_analyzer.py
+          tests/test_actions_poll_analyzer_emit.py
           tests/test_github_actions_poll_bounds.py
           tests/test_github_actions_poll_bound_control_flow_regression.py

CHANGELOG.d/1087-actions-poll-analyzer-emit.md

@@ -0,0 +1,3 @@
+# Security
+
+- Emit structural GitHub Actions poll-bound findings through production `_scan_file` using the existing `github-actions-transport-only-poll-bound` and `github-actions-transport-failure-budget-poll-bound` identities. Regex and analyzer hits for the same rule on the same file merge to one finding; helper loops, reversed comparisons, unreachable exits, and split `while`/`do` polls that regex adjacency misses can now be reported. Quoted or comment text stays negative. See issue #1087.

appguardrail_core/actions_poll_analyzer.py

@@ -140,6 +140,34 @@ def poll_bound_rule_ids(assessments: tuple[PollLoopAssessment, ...]) -> tuple[st
     return tuple(rule_ids)
 
 
+def additional_poll_bound_rule_ids(
+    workflow_text: str,
+    existing_rule_ids: tuple[str, ...] = (),
+) -> tuple[str, ...]:
+    """Return poll-bound rule IDs not already reported for the same file.
+
+    Merge is keyed by rule identity. Regex hits for an existing detector ID
+    are preserved; the classifier contributes an ID only when that identity
+    is absent so a file cannot double-count the packaged #1088 corpus.
+
+    Args:
+        workflow_text: Complete workflow document text.
+        existing_rule_ids: Detector identities already emitted for this file.
+
+    Returns:
+        Existing packaged rule IDs in first-seen analyzer order.
+    """
+    already = set(existing_rule_ids)
+    extra: list[str] = []
+    seen: set[str] = set()
+    for rule_id in poll_bound_rule_ids(classify_poll_loops(workflow_text)):
+        if rule_id in already or rule_id in seen:
+            continue
+        seen.add(rule_id)
+        extra.append(rule_id)
+    return tuple(extra)
+
+
 def _is_positive_timeout(value: str) -> bool:
     """Return whether a timeout-minutes value is a static positive bound."""
     stripped = value.split("#", 1)[0].strip()
@@ -231,6 +259,7 @@ def _classify_shell(job_name: str, shell: str, owning_timeout: bool) -> tuple[Po
         converges, exit_reachable = _total_bound_flags(assignments, frames, top_commands)
         loop_local = converges and exit_reachable
         historical = _uses_historical_names(assignments, commands)
+        terminates = _success_path_terminates(top_commands)
         assessments.append(
             PollLoopAssessment(
                 job_name=job_name,
@@ -239,7 +268,12 @@ def _classify_shell(job_name: str, shell: str, owning_timeout: bool) -> tuple[Po
                 owning_job_timeout=owning_timeout,
                 comparison_converges=converges,
                 exit_reachable=exit_reachable,
-                is_transport_only_unbounded=transport and not loop_local and not owning_timeout,
+                is_transport_only_unbounded=(
+                    transport
+                    and not loop_local
+                    and not owning_timeout
+                    and not terminates
+                ),
                 historical_transport_names=historical,
             )
         )
@@ -469,6 +503,20 @@ def _walk_if_frames(commands: tuple[str, ...]) -> tuple[tuple[_IfFrame, ...], tu
     return tuple(completed), tuple(top)
 
 
+def _success_path_terminates(top_commands: tuple[str, ...]) -> bool:
+    """Return whether a reachable top-level transfer ends the poll.
+
+    An unconditional ``break``, ``exit``, or ``return`` removes the back edge.
+    A top-level ``continue`` skips the rest of the body and repeats the loop.
+    """
+    for command in top_commands:
+        if _TRANSFER.fullmatch(command):
+            return not command.startswith("continue")
+        if _FAIL_EXIT.fullmatch(command):
+            return True
+    return False
+
+
 def _has_transport_failure_budget(frames: tuple[_IfFrame, ...], top_commands: tuple[str, ...]) -> bool:
     """Return whether a negated ``gh api`` branch increments a retry counter."""
     del top_commands

scanner/cli/appguardrail.py

@@ -88,6 +88,7 @@
     render_report,
     supported_report_types,
 )
+from appguardrail_core.actions_poll_analyzer import additional_poll_bound_rule_ids
 from appguardrail_core.rules import build_rule_metadata
 from appguardrail_core.scan_paths import ScanPathContext, build_scan_path_context
 
@@ -2916,6 +2917,53 @@ def _run_codegraph_index(scan_path: Path):
     return _run_codegraph_command([codegraph, "status"], workdir, "status")
 
 
+_POLL_WORKFLOW_INCLUDE = (
+    ".github/workflows/*.yml",
+    ".github/workflows/*.yaml",
+)
+
+
+def _poll_analyzer_location(content: str) -> tuple[int, str]:
+    """Return a display line and snippet for an analyzer-only poll finding."""
+    for index, raw_line in enumerate(content.splitlines(), start=1):
+        stripped = raw_line.strip()
+        if stripped.startswith("while"):
+            return index, stripped[:120]
+    return 1, ""
+
+
+def _append_actions_poll_analyzer_findings(
+    content: str, findings: list, rel_path_str: str, build_finding
+) -> None:
+    """Append structural poll-bound findings not already present for this file.
+
+    Merge is keyed by ``(rule_id, file)``. Regex hits keep their packaged
+    identity; the classifier adds an ID only when that identity is absent.
+    """
+    extra_ids = additional_poll_bound_rule_ids(
+        content, tuple(item["rule_id"] for item in findings)
+    )
+    if not extra_ids:
+        return
+    templates = {
+        rule["id"]: rule for rule in SCAN_RULES if rule["id"] in extra_ids
+    }
+    line_num, snippet = _poll_analyzer_location(content)
+    for rule_id in extra_ids:
+        rule = templates[rule_id]
+        findings.append(
+            build_finding(
+                "appguardrail-rule",
+                rule_id,
+                rule["severity"],
+                rule["message"],
+                rel_path_str,
+                line_num,
+                snippet,
+            )
+        )
+
+
 def _scan_file(
     file_path: Path,
     base_path: Path,
@@ -2927,6 +2975,10 @@ def _scan_file(
     Direct callers may omit ``path_context`` and retain the historical safe
     fallback. Batch callers should build one context and reuse it for every
     file so root classification and normalized prefix construction happen once.
+
+    GitHub Actions workflow paths also run the structural poll-loop classifier.
+    Analyzer identities merge with regex hits by ``(rule_id, file)`` so the
+    packaged transport-only family is emitted once.
     """
     findings = []
     context = path_context or build_scan_path_context(base_path)
@@ -3027,6 +3079,21 @@ def _scan_file(
                             snippet,
                         )
                     )
+            if ext in {".yml", ".yaml"}:
+                if rel_path_for_filters is None:
+                    rel_path_for_filters = _display_path(
+                        context.relative_candidate(file_path)
+                    )
+                if _path_allowed_by_rule(
+                    rel_path_for_filters, _POLL_WORKFLOW_INCLUDE, ()
+                ):
+                    if rel_path_str is None:
+                        rel_path_str = _sanitize_terminal_output(
+                            _display_path(context.relative_candidate(file_path))
+                        )
+                    _append_actions_poll_analyzer_findings(
+                        content, findings, rel_path_str, build_finding
+                    )
     except (OSError, PermissionError):
         pass