[codex] Surface policy parse failures in scenario metadata

iamscope/models.py

@@ -726,6 +726,12 @@ class ScenarioMetadata:
     # because `canonical_hash` excludes metadata — the non-deterministic
     # failure content will never perturb the hash.
     collection_failures: list[dict[str, Any]] = field(default_factory=list)
+    # BUG-024 fix: structured record of IAM policy parse failures
+    # collected during per-account IAM collection. Empty/absent in the
+    # happy path; any non-empty value means IAM policies were skipped
+    # before graph construction and downstream findings may be missing.
+    # Each entry is produced by `PolicyParseFailure.to_dict()`.
+    policy_parse_failures: list[dict[str, Any]] = field(default_factory=list)
     hash_scope: str = "canonical_hash excludes metadata block"
 
     def to_dict(self) -> dict[str, Any]:
@@ -744,4 +750,6 @@ def to_dict(self) -> dict[str, Any]:
             "noise_filter": self.noise_filter,
             "org_id": self.org_id,
         }
+        if self.policy_parse_failures:
+            d["policy_parse_failures"] = self.policy_parse_failures
         return _sorted_dict(d)

iamscope/pipeline.py

@@ -49,6 +49,7 @@
     TrustParseResult,
 )
 from iamscope.output.scenario_json import emit_binding_metadata, emit_scenario
+from iamscope.parser.parse_failures import PolicyParseFailure
 from iamscope.parser.resource_policy import parse_resource_policy_documents
 from iamscope.resolver.cross_account import build_trust_edges, resolve_synthetic_nodes
 from iamscope.resolver.identity_deny_binder import (
@@ -154,6 +155,12 @@ class PipelineResult:
     # read scenario.json still see the signal.
     collection_failures: list[CollectionFailure] = field(default_factory=list)
 
+    # BUG-024 fix: structured record of per-account IAM policy parse
+    # failures aggregated from AccountData.policy_parse_failures. Empty
+    # in the happy path; any non-empty value means one or more IAM
+    # policies could not be parsed before graph construction.
+    policy_parse_failures: list[PolicyParseFailure] = field(default_factory=list)
+
 
 def run_pipeline(
     session: boto3.Session,
@@ -353,6 +360,8 @@ def run_pipeline(
         result.accounts_collected = len(all_account_data)
         result.accounts_skipped = accounts_skipped
 
+    result.policy_parse_failures = _aggregate_policy_parse_failures(result.account_data)
+
     # --- Phase 3: Resolution ---
     logger.info("=== Phase 3: Resolution Pipeline ===")
     nodes, edges, constraints, edge_constraints, budget_hit = _run_resolution(
@@ -447,6 +456,15 @@ def run_pipeline(
             "the full list.",
             len(result.collection_failures),
         )
+    if result.policy_parse_failures:
+        logger.warning(
+            "IAM policy parsing was partial: %d policy document(s) failed "
+            "to parse during account collection. Fact graph may be "
+            "incomplete and downstream findings may be missing. "
+            "See PipelineResult.policy_parse_failures or the scenario.json "
+            "metadata.policy_parse_failures field for the full list.",
+            len(result.policy_parse_failures),
+        )
     metadata = ScenarioMetadata(
         collector="iamscope",
         collector_version="0.2.0",
@@ -468,6 +486,7 @@ def run_pipeline(
             "total_edge_constraints": len(edge_constraints),
         },
         collection_failures=[f.to_dict() for f in result.collection_failures],
+        policy_parse_failures=[f.to_dict() for f in result.policy_parse_failures],
     )
 
     scenario_bytes, canonical_hash = emit_scenario(
@@ -508,6 +527,28 @@ def run_pipeline(
     return result
 
 
+def _policy_parse_failure_sort_key(failure: PolicyParseFailure) -> tuple[str, str, str, str, str, str, str, str]:
+    """Stable ordering for policy parse failures across accounts and AWS pages."""
+    return (
+        failure.parser,
+        failure.source_arn,
+        failure.policy_source,
+        failure.policy_name,
+        failure.policy_arn,
+        failure.failure_kind,
+        failure.error_class,
+        failure.error_message,
+    )
+
+
+def _aggregate_policy_parse_failures(account_data: list[AccountData]) -> list[PolicyParseFailure]:
+    """Aggregate per-account IAM policy parse failures deterministically."""
+    failures: list[PolicyParseFailure] = []
+    for acct in sorted(account_data, key=lambda account: account.account_id):
+        failures.extend(sorted(acct.policy_parse_failures, key=_policy_parse_failure_sort_key))
+    return failures
+
+
 def _resolve_target_accounts(
     org_data: OrgData,
     config: PipelineConfig,

tests/test_pipeline.py

@@ -27,10 +27,19 @@
 
 from iamscope.auth.session import get_session
 from iamscope.collector.account import AccountData
+from iamscope.collector.failures import CollectionFailure
 from iamscope.constants import CONSTRAINT_TYPE_STALE_PRINCIPAL_DRIFT, NODE_TYPE_IAM_ROLE, PROVIDER_AWS
-from iamscope.models import AccountInfo, Node, OrgData
+from iamscope.models import AccountInfo, Node, OrgData, ScenarioMetadata
+from iamscope.output.scenario_json import emit_scenario
+from iamscope.parser.parse_failures import PolicyParseFailure
 from iamscope.parser.trust_policy import parse_trust_policy
-from iamscope.pipeline import PipelineConfig, _run_resolution, run_pipeline
+from iamscope.pipeline import (
+    PipelineConfig,
+    PipelineResult,
+    _aggregate_policy_parse_failures,
+    _run_resolution,
+    run_pipeline,
+)
 
 
 @pytest.fixture
@@ -317,6 +326,129 @@ def test_binding_metadata_sidecar(self, full_org_setup) -> None:
         assert isinstance(sidecar, list)
 
 
+class TestPolicyParseFailureMetadata:
+    """No-AWS tests for IAM policy parse failure aggregation and metadata emission."""
+
+    def _failure(
+        self,
+        *,
+        source_arn: str,
+        parser: str = "permission_policy",
+        policy_source: str = "inline",
+        policy_name: str = "BrokenInline",
+        policy_arn: str = "",
+        failure_kind: str = "json_decode_error",
+        error_class: str = "JSONDecodeError",
+        error_message: str = "Expecting property name enclosed in double quotes",
+    ) -> PolicyParseFailure:
+        return PolicyParseFailure(
+            parser=parser,
+            source_arn=source_arn,
+            policy_source=policy_source,
+            policy_name=policy_name,
+            policy_arn=policy_arn,
+            failure_kind=failure_kind,
+            error_class=error_class,
+            error_message=error_message,
+        )
+
+    def test_pipeline_result_has_policy_parse_failures_field(self) -> None:
+        result = PipelineResult()
+        assert result.policy_parse_failures == []
+
+    def test_aggregates_policy_parse_failures_deterministically(self) -> None:
+        account_a = "111111\u003111111"
+        account_b = "222222\u003222222"
+        role_failure = self._failure(
+            parser="trust_policy",
+            source_arn=f"arn:aws:iam::{account_a}:role/BrokenTrust",
+            policy_source="trust",
+            policy_name="AssumeRolePolicyDocument",
+            failure_kind="not_a_dict",
+            error_class="",
+            error_message="",
+        )
+        user_failure = self._failure(source_arn=f"arn:aws:iam::{account_a}:user/BrokenUser")
+        group_failure = self._failure(
+            source_arn=f"arn:aws:iam::{account_b}:group/BrokenGroup",
+            policy_source="group_inline",
+            policy_name="BrokenGroupInline",
+        )
+        acct_b = AccountData(account_id=account_b, policy_parse_failures=[group_failure])
+        acct_a = AccountData(account_id=account_a, policy_parse_failures=[user_failure, role_failure])
+
+        failures = _aggregate_policy_parse_failures([acct_b, acct_a])
+
+        assert failures == [user_failure, role_failure, group_failure]
+
+    def test_empty_policy_parse_failures_are_stable(self) -> None:
+        acct = AccountData(account_id="111111\u003111111")
+
+        failures = _aggregate_policy_parse_failures([acct])
+
+        assert failures == []
+        assert "policy_parse_failures" not in ScenarioMetadata().to_dict()
+
+    def test_scenario_metadata_includes_policy_parse_failures(self) -> None:
+        account_id = "111111\u003111111"
+        failure = self._failure(
+            source_arn=f"arn:aws:iam::{account_id}:role/BrokenRole",
+            policy_arn=f"arn:aws:iam::{account_id}:policy/BrokenPolicy",
+        )
+        collection_failure = CollectionFailure(
+            collector="lambda",
+            account_id=account_id,
+            region="us-east-1",
+            error_class="AccessDeniedException",
+            error_message="synthetic collector failure",
+        )
+        metadata = ScenarioMetadata(
+            collection_failures=[collection_failure.to_dict()],
+            policy_parse_failures=[failure.to_dict()],
+        )
+
+        scenario_bytes, _hash = emit_scenario(
+            nodes=[],
+            edges=[],
+            constraints=[],
+            edge_constraints=[],
+            metadata=metadata,
+        )
+        scenario = json.loads(scenario_bytes)
+
+        assert scenario["metadata"]["collection_failures"] == [collection_failure.to_dict()]
+        assert scenario["metadata"]["policy_parse_failures"] == [failure.to_dict()]
+        assert scenario["metadata"]["collection_failures"] != scenario["metadata"]["policy_parse_failures"]
+
+    def test_policy_parse_failure_serialization_keeps_useful_fields(self) -> None:
+        account_id = "111111\u003111111"
+        failure = self._failure(
+            source_arn=f"arn:aws:iam::{account_id}:role/BrokenRole",
+            policy_arn=f"arn:aws:iam::{account_id}:policy/BrokenPolicy",
+        )
+
+        record = failure.to_dict()
+
+        assert set(record) == {
+            "error_class",
+            "error_message",
+            "failure_kind",
+            "parser",
+            "policy_arn",
+            "policy_name",
+            "policy_source",
+            "source_arn",
+        }
+        assert record["parser"] == "permission_policy"
+        assert record["source_arn"] == f"arn:aws:iam::{account_id}:role/BrokenRole"
+        assert record["policy_source"] == "inline"
+        assert record["policy_name"] == "BrokenInline"
+        assert record["policy_arn"] == f"arn:aws:iam::{account_id}:policy/BrokenPolicy"
+        assert record["failure_kind"] == "json_decode_error"
+        assert record["error_class"] == "JSONDecodeError"
+        assert record["error_message"]
+
+
 # ---------------------------------------------------------------------------
 # Standalone (single-account) mode tests
 # ---------------------------------------------------------------------------