fix: violation record on exception

.github/workflows/ci.yml

@@ -6,10 +6,11 @@ name: CI
 
 on:
   push:
-    branches: ["main"]
+    # 0.1.x is the long-lived maintenance base branch for the 0.1 series
+    branches: ["main", "dev", "0.1.x"]
   pull_request:
     types: [opened, synchronize, ready_for_review]
-    branches: ["main"]
+    branches: ["main", "dev", "0.1.x"]
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
@@ -27,6 +28,7 @@ jobs:
     name: Tests
     uses: ./.github/workflows/tests.yaml
 
-  docs:
-    name: Docs Build
-    uses: ./.github/workflows/docs-build.yaml
+  # No docs job on 0.1.x: docs are built and published from main only.
+  # Do not re-add a `uses: ./.github/workflows/docs-build.yaml` job here —
+  # workflow_call bypasses that workflow's own main-only branch filter, so it
+  # would run on this branch regardless.

.github/workflows/lint.yml

@@ -9,21 +9,11 @@
 
 name: Lint & Static Analysis
 
+# Reusable workflow only — invoked by ci.yml, which owns the branch filters.
+# Do not add push/pull_request triggers here: ci.yml already calls this on those
+# events, so its own triggers would run every job a second time.
 on:
   workflow_call:
-  push:
-    branches: ["main"]
-    paths:
-      - "cpex/**"
-      - "pyproject.toml"
-      - ".github/workflows/lint.yml"
-  pull_request:
-    types: [opened, synchronize, ready_for_review]
-    branches: ["main"]
-    paths:
-      - "cpex/**"
-      - "pyproject.toml"
-      - ".github/workflows/lint.yml"
 
 permissions:
   contents: read

.github/workflows/tests.yaml

@@ -4,23 +4,11 @@
 
 name: Tests
 
+# Reusable workflow only — invoked by ci.yml, which owns the branch filters.
+# Do not add push/pull_request triggers here: ci.yml already calls this on those
+# events, so its own triggers would run every job a second time.
 on:
   workflow_call:
-  push:
-    branches: ["main"]
-    paths:
-      - "cpex/**"
-      - "tests/**"
-      - "pyproject.toml"
-      - ".github/workflows/tests.yaml"
-  pull_request:
-    types: [opened, synchronize, ready_for_review]
-    branches: ["main"]
-    paths:
-      - "cpex/**"
-      - "tests/**"
-      - "pyproject.toml"
-      - ".github/workflows/tests.yaml"
   workflow_dispatch:
 
 permissions:

cpex/framework/errors.py

@@ -10,7 +10,7 @@
 """
 
 # First-Party
-from cpex.framework.models import PluginErrorModel, PluginViolation
+from cpex.framework.models import ControlExecutionRecord, PluginErrorModel, PluginViolation
 
 
 class PluginViolationError(Exception):
@@ -19,6 +19,12 @@ class PluginViolationError(Exception):
     Attributes:
         violation (PluginViolation): the plugin violation.
         message (str): the plugin violation reason.
+        executions (list[ControlExecutionRecord] | None): execution records collected up to and
+            including the denying plugin.  ``None`` when the exception is raised outside of a
+            hook chain (e.g. a direct ``execute_plugin`` call, or in unit tests).  Populated by
+            the executor before the exception propagates to callers (fix for issue #147).
+            Note: fire-and-forget plugins do not run on this path, so their records are absent
+            here — unlike ``PluginResult.executions`` on a non-exception halt.
     """
 
     def __init__(self, message: str, violation: PluginViolation | None = None):
@@ -38,6 +44,7 @@ def __init__(self, message: str, violation: PluginViolation | None = None):
         """
         self.message = message
         self.violation = violation
+        self.executions: list[ControlExecutionRecord] | None = None
         super().__init__(self.message)
 
 

cpex/framework/manager.py

@@ -420,7 +420,7 @@ async def execute(
                 concurrent_tasks.append(asyncio.create_task(self._tagged(coro, idx)))
 
             for completed_coro in asyncio.as_completed(concurrent_tasks):
-                result, idx, timeout_err = await completed_coro
+                result, idx, timeout_err, violation_err = await completed_coro
                 ref, _, _ = concurrent_ctx_list[idx]
                 ctx.hook_chain_executed += 1
                 # Propagate retry signal from concurrent plugins
@@ -429,15 +429,42 @@ async def execute(
                     # on_error=ignore/disable timeout: pipeline continues, record TIMEOUT
                     # (mirrors the serial-phase handler; on_error=fail raises PluginError,
                     # which is not caught here and propagates fail-closed).
-                    executions.append(_make_execution_record(
-                        ref,
-                        hook_type,
-                        ControlExecutionStatus.TIMEOUT,
-                        effective_allow=True,
-                        reason=_truncate_opt(str(timeout_err)),
-                        error_code="plugin_timeout",
-                    ))
+                    executions.append(
+                        _make_execution_record(
+                            ref,
+                            hook_type,
+                            ControlExecutionStatus.TIMEOUT,
+                            effective_allow=True,
+                            reason=_truncate_opt(str(timeout_err)),
+                            error_code="plugin_timeout",
+                        )
+                    )
                     continue
+                if violation_err is not None:
+                    # violations_as_exceptions=True concurrent denial: write the record,
+                    # cancel sibling tasks, attach the accumulated records to the exception,
+                    # then re-raise (fix for #147). FAF plugins are not scheduled on this path,
+                    # so their records are absent — unlike the non-exception halt result.
+                    _pve = violation_err
+                    executions.append(
+                        _make_execution_record(
+                            ref,
+                            hook_type,
+                            ControlExecutionStatus.COMPLETED,
+                            effective_allow=False,
+                            requested_allow=False,
+                            matched=True,
+                            applied=True,
+                            reason=_truncate_opt(_pve.violation.reason if _pve.violation else str(_pve)),
+                            error_code=_truncate(_pve.violation.code) if _pve.violation else "plugin_violation",
+                        )
+                    )
+                    for task in concurrent_tasks:
+                        if not task.done():
+                            task.cancel()
+                    await asyncio.gather(*concurrent_tasks, return_exceptions=True)
+                    _pve.executions = list(executions)
+                    raise _pve
                 if result.modified_payload is not None:
                     logger.debug(
                         "CONCURRENT plugin %s returned modified_payload on hook %s; "
@@ -447,17 +474,19 @@ async def execute(
                     )
                 # Build the concurrent execution record (duration=0: no per-branch timing)
                 _concurrent_denied = not result.continue_processing
-                executions.append(_make_execution_record(
-                    ref,
-                    hook_type,
-                    ControlExecutionStatus.COMPLETED,
-                    effective_allow=not _concurrent_denied,
-                    requested_allow=result.continue_processing,
-                    matched=True if _concurrent_denied else False,
-                    applied=_concurrent_denied,
-                    reason=_truncate_opt(result.violation.reason if result.violation else None),
-                    error_code=_truncate(result.violation.code) if result.violation else None,
-                ))
+                executions.append(
+                    _make_execution_record(
+                        ref,
+                        hook_type,
+                        ControlExecutionStatus.COMPLETED,
+                        effective_allow=not _concurrent_denied,
+                        requested_allow=result.continue_processing,
+                        matched=True if _concurrent_denied else False,
+                        applied=_concurrent_denied,
+                        reason=_truncate_opt(result.violation.reason if result.violation else None),
+                        error_code=_truncate(result.violation.code) if result.violation else None,
+                    )
+                )
                 if not result.continue_processing:
                     pending = sum(1 for t in concurrent_tasks if not t.done())
                     violation_detail = (
@@ -677,24 +706,46 @@ async def _run_serial_phase(
                 exec_status = ControlExecutionStatus.COMPLETED
                 exec_error_code: Optional[str] = None
                 exec_reason: Optional[str] = None
-            except PluginViolationError:
-                # violations_as_exceptions=True — propagate immediately, no record needed
+            except PluginViolationError as _pve:
+                # violations_as_exceptions=True — append the denial record and attach the
+                # accumulated records to the exception before propagating, so callers can
+                # identify the denying plugin (fix for #147). FAF plugins are not scheduled
+                # on this path, so their records are absent — unlike the non-exception halt.
+                duration_ns = time.monotonic_ns() - t_start
+                if executions is not None:
+                    executions.append(
+                        _make_execution_record(
+                            hook_ref,
+                            hook_type,
+                            ControlExecutionStatus.COMPLETED,
+                            effective_allow=False,
+                            duration_ns=duration_ns,
+                            requested_allow=False,
+                            matched=True,
+                            applied=True,
+                            reason=_truncate_opt(_pve.violation.reason if _pve.violation else str(_pve)),
+                            error_code=_truncate(_pve.violation.code) if _pve.violation else "plugin_violation",
+                        )
+                    )
+                    _pve.executions = list(executions)
                 raise
             except PluginError as _pe:
                 # execute_plugin re-raises PluginError when on_error=FAIL — must not swallow.
                 # Record the error then re-raise to preserve fail-closed behaviour.
                 duration_ns = time.monotonic_ns() - t_start
                 if executions is not None:
-                    executions.append(_make_execution_record(
-                        hook_ref,
-                        hook_type,
-                        ControlExecutionStatus.ERROR,
-                        effective_allow=False,
-                        duration_ns=duration_ns,
-                        applied=True,
-                        reason=_truncate_opt(str(_pe)),
-                        error_code="plugin_error",
-                    ))
+                    executions.append(
+                        _make_execution_record(
+                            hook_ref,
+                            hook_type,
+                            ControlExecutionStatus.ERROR,
+                            effective_allow=False,
+                            duration_ns=duration_ns,
+                            applied=True,
+                            reason=_truncate_opt(str(_pe)),
+                            error_code="plugin_error",
+                        )
+                    )
                 raise
             except PluginTimeoutError as _te:
                 # on_error=IGNORE or DISABLE timeout — pipeline continues, record as TIMEOUT.
@@ -754,40 +805,40 @@ async def _run_serial_phase(
                 requested_allow: Optional[bool] = result.continue_processing
                 # matched: True if denied, or if payload/extensions were changed
                 # False if clean allow with no mutation, None if error/timeout
-                matched: Optional[bool] = True if denied else (
-                    True if (payload_modified or extensions_modified) else False
+                matched: Optional[bool] = (
+                    True if denied else (True if (payload_modified or extensions_modified) else False)
                 )
                 effective_allow = not denied
                 rec_applied = denied or payload_modified or extensions_modified
-                rec_error_code = (
-                    _truncate(result.violation.code) if (denied and result.violation) else exec_error_code
-                )
-                rec_reason = (
-                    _truncate_opt(result.violation.reason if (denied and result.violation) else exec_reason)
-                )
+                rec_error_code = _truncate(result.violation.code) if (denied and result.violation) else exec_error_code
+                rec_reason = _truncate_opt(result.violation.reason if (denied and result.violation) else exec_reason)
             else:
                 requested_allow = None
                 matched = None
-                effective_allow = True   # error/timeout in non-blocking phase → pipeline continues
-                rec_applied = exec_status in (ControlExecutionStatus.ERROR, ControlExecutionStatus.TIMEOUT) and allow_blocking
+                effective_allow = True  # error/timeout in non-blocking phase → pipeline continues
+                rec_applied = (
+                    exec_status in (ControlExecutionStatus.ERROR, ControlExecutionStatus.TIMEOUT) and allow_blocking
+                )
                 rec_error_code = exec_error_code
                 rec_reason = _truncate_opt(exec_reason)
 
             if executions is not None:
-                executions.append(_make_execution_record(
-                    hook_ref,
-                    hook_type,
-                    exec_status,
-                    effective_allow=effective_allow,
-                    duration_ns=duration_ns,
-                    requested_allow=requested_allow,
-                    matched=matched,
-                    applied=rec_applied,
-                    payload_modified=payload_modified,
-                    extensions_modified=extensions_modified,
-                    reason=rec_reason,
-                    error_code=rec_error_code,
-                ))
+                executions.append(
+                    _make_execution_record(
+                        hook_ref,
+                        hook_type,
+                        exec_status,
+                        effective_allow=effective_allow,
+                        duration_ns=duration_ns,
+                        requested_allow=requested_allow,
+                        matched=matched,
+                        applied=rec_applied,
+                        payload_modified=payload_modified,
+                        extensions_modified=extensions_modified,
+                        reason=rec_reason,
+                        error_code=rec_error_code,
+                    )
+                )
 
             if not result.continue_processing:
                 violation_detail = f": [{result.violation.code}] {result.violation.reason}" if result.violation else ""
@@ -989,20 +1040,29 @@ async def _with_semaphore(semaphore: asyncio.Semaphore, coro: Any) -> Any:
             return await coro
 
     @staticmethod
-    async def _tagged(coro: Any, tag: Any) -> tuple[Any, Any, Optional["PluginTimeoutError"]]:
+    async def _tagged(
+        coro: Any, tag: Any
+    ) -> tuple[Any, Any, Optional["PluginTimeoutError"], Optional["PluginViolationError"]]:
         """Await *coro* and pair the result with *tag* for use with as_completed.
 
         Catches PluginTimeoutError (raised by execute_plugin for on_error=ignore/disable
         timeouts) so it never escapes the concurrent as_completed loop unpaired with its
-        tag. The loop records a TIMEOUT execution record and continues. PluginError
-        (on_error=fail) is intentionally not caught here — it must propagate to preserve
-        fail-closed behaviour.
+        tag. The loop records a TIMEOUT execution record and continues.
+
+        Catches PluginViolationError (raised by execute_plugin for concurrent-mode denials
+        when violations_as_exceptions=True) so it never escapes the loop unpaired with its
+        tag. The loop writes the denial record, cancels sibling tasks, and re-raises.
+
+        PluginError (on_error=fail) is intentionally not caught here — it must propagate
+        to preserve fail-closed behaviour.
         """
         try:
             result = await coro
-            return result, tag, None
+            return result, tag, None, None
         except PluginTimeoutError as te:
-            return PluginResult(continue_processing=True), tag, te
+            return PluginResult(continue_processing=True), tag, te, None
+        except PluginViolationError as pve:
+            return PluginResult(continue_processing=True), tag, None, pve
 
     def _fire_and_forget_tasks(
         self,
@@ -1047,12 +1107,14 @@ def _fire_and_forget_tasks(
             # status=COMPLETED is an optimistic placeholder; duration_ns=0 (not yet run).
             # Identify FAF records by mode == "fire_and_forget", not by status.
             if executions is not None:
-                executions.append(_make_execution_record(
-                    ref,
-                    hook_type,
-                    ControlExecutionStatus.COMPLETED,
-                    effective_allow=True,
-                ))
+                executions.append(
+                    _make_execution_record(
+                        ref,
+                        hook_type,
+                        ControlExecutionStatus.COMPLETED,
+                        effective_allow=True,
+                    )
+                )
 
             task = asyncio.create_task(
                 self._run_fire_and_forget_task(ref, task_input, local_context, semaphore, extensions=extensions)
@@ -1230,9 +1292,7 @@ async def execute_plugin(
                     self._runtime_disabled.add(hook_ref.plugin_ref.name)
             # on_error=IGNORE or DISABLE: pipeline continues, but raise PluginTimeoutError so
             # _run_serial_phase can record status=TIMEOUT rather than COMPLETED.
-            raise PluginTimeoutError(
-                f"Plugin {hook_ref.plugin_ref.name} exceeded {self.timeout}s timeout"
-            ) from exc
+            raise PluginTimeoutError(f"Plugin {hook_ref.plugin_ref.name} exceeded {self.timeout}s timeout") from exc
         except PluginViolationError:
             raise
         except PluginError as pe: