From fc51c0dc9d0318639588d88fa60c125af690cde1 Mon Sep 17 00:00:00 2001 From: prakhar-singh1928 Date: Fri, 31 Jul 2026 13:35:43 +0100 Subject: [PATCH 1/6] fix(#147): append ControlExecutionRecord before raising PluginViolationError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When violations_as_exceptions=True and a plugin denies an invocation, the except PluginViolationError handler in _run_serial_phase previously re-raised immediately without appending a ControlExecutionRecord for the denying plugin to the executions accumulator. The same gap existed for CONCURRENT-mode plugins: _tagged only caught PluginTimeoutError, so PluginViolationError escaped the as_completed loop before the record-append and before sibling task cancellation. This meant PluginResult.executions only contained records for plugins that ran *before* the denial, making it impossible for consumers to identify which specific control blocked the invocation from telemetry alone. Changes: - cpex/framework/manager.py (_run_serial_phase): Append the denying plugin's ControlExecutionRecord (status=COMPLETED, effective_allow=False, matched=True, applied=True, error_code/reason from violation) before re-raising PluginViolationError. Covers SEQUENTIAL, TRANSFORM, AUDIT. - cpex/framework/manager.py (_tagged): Catch PluginViolationError and return it as a sentinel alongside the tag (mirrors PluginTimeoutError), so it never escapes the concurrent as_completed loop unpaired. - cpex/framework/manager.py (concurrent as_completed loop): Handle the new violation_err sentinel — write the denial record, cancel all sibling tasks, and re-raise. Fixes the pre-existing sibling-task leak on this path. - cpex/framework/manager.py (execute): Outer try/except PluginViolationError attaches list(executions) to pve.executions before propagating. FAF plugins are not scheduled on this raise path (noted in comment). - cpex/framework/errors.py (PluginViolationError): Add executions attribute (list[ControlExecutionRecord] | None, default None). Simplified import — ControlExecutionRecord added to existing runtime import line, dropping the unnecessary __future__/TYPE_CHECKING guard. - tests/unit/cpex/framework/test_execution_records.py: Four new tests: - test_executions_on_violation_exception_single_plugin - test_executions_on_violation_exception_two_plugin_chain - test_executions_on_violation_exception_concurrent_plugin (new — covers the concurrent-mode gap) - test_executions_on_violation_exception_list_is_independent_copy (new — asserts pve.executions is a distinct list from the internal accumulator) Fixes: #147 --- cpex/framework/errors.py | 7 +- cpex/framework/manager.py | 463 ++++++++++-------- .../cpex/framework/test_execution_records.py | 329 ++++++++++++- 3 files changed, 593 insertions(+), 206 deletions(-) diff --git a/cpex/framework/errors.py b/cpex/framework/errors.py index d1e41874..c719537f 100644 --- a/cpex/framework/errors.py +++ b/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,10 @@ 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 an + ``invoke_hook`` call (e.g. in unit tests). Populated by ``PluginExecutor.execute`` + before the exception propagates to callers (fix for issue #147). """ def __init__(self, message: str, violation: PluginViolation | None = None): @@ -38,6 +42,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) diff --git a/cpex/framework/manager.py b/cpex/framework/manager.py index 8e2b740b..1f4bb317 100644 --- a/cpex/framework/manager.py +++ b/cpex/framework/manager.py @@ -317,211 +317,242 @@ async def execute( fire_and_forget_semaphore = asyncio.Semaphore(pool) if pool else None concurrent_semaphore = asyncio.Semaphore(pool) if pool else None - # SEQUENTIAL: sequential, chained execution — can halt pipeline - halt_result, phase = await self._run_serial_phase( - hook_refs=sequential_refs, - mode_label="SEQUENTIAL", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=True, - allow_blocking=True, - ctx=ctx, - current_extensions=current_extensions, - fire_and_forget_refs=fire_and_forget_refs, - fire_and_forget_semaphore=fire_and_forget_semaphore, - extensions=extensions, - executions=executions, - ) - current_payload = phase.payload - decision_plugin_name = phase.decision_plugin - current_extensions = phase.extensions - if halt_result is not None: - # Attach records collected so far (including FAF scheduled on halt) - halt_result[0].executions = list(executions) - self._end_hook_chain_span(ctx, status="ok") - return halt_result - - # TRANSFORM: serial, chained execution — can modify payloads but cannot halt pipeline - _, phase = await self._run_serial_phase( - hook_refs=transform_refs, - mode_label="TRANSFORM", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=True, - allow_blocking=False, - ctx=ctx, - current_extensions=current_extensions, - extensions=extensions, - executions=executions, - ) - current_payload = phase.payload - decision_plugin_name = phase.decision_plugin - current_extensions = phase.extensions - - # AUDIT: serial execution — observe-only (no modifications, no blocking) - _, phase = await self._run_serial_phase( - hook_refs=audit_refs, - mode_label="AUDIT", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=False, - allow_blocking=False, - ctx=ctx, - current_extensions=current_extensions, - extensions=extensions, - executions=executions, - ) + try: + # SEQUENTIAL: sequential, chained execution — can halt pipeline + halt_result, phase = await self._run_serial_phase( + hook_refs=sequential_refs, + mode_label="SEQUENTIAL", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=True, + allow_blocking=True, + ctx=ctx, + current_extensions=current_extensions, + fire_and_forget_refs=fire_and_forget_refs, + fire_and_forget_semaphore=fire_and_forget_semaphore, + extensions=extensions, + executions=executions, + ) + current_payload = phase.payload + decision_plugin_name = phase.decision_plugin + current_extensions = phase.extensions + if halt_result is not None: + # Attach records collected so far (including FAF scheduled on halt) + halt_result[0].executions = list(executions) + self._end_hook_chain_span(ctx, status="ok") + return halt_result + + # TRANSFORM: serial, chained execution — can modify payloads but cannot halt pipeline + _, phase = await self._run_serial_phase( + hook_refs=transform_refs, + mode_label="TRANSFORM", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=True, + allow_blocking=False, + ctx=ctx, + current_extensions=current_extensions, + extensions=extensions, + executions=executions, + ) + current_payload = phase.payload + decision_plugin_name = phase.decision_plugin + current_extensions = phase.extensions + + # AUDIT: serial execution — observe-only (no modifications, no blocking) + _, phase = await self._run_serial_phase( + hook_refs=audit_refs, + mode_label="AUDIT", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=False, + allow_blocking=False, + ctx=ctx, + current_extensions=current_extensions, + extensions=extensions, + executions=executions, + ) - # CONCURRENT: parallel execution with fail-fast on first blocking result - if concurrent_refs: - concurrent_ctx_list: list[tuple[HookRef, PluginContext, PluginPayload]] = [] - concurrent_tasks: list[asyncio.Task] = [] - effective_payload = current_payload if current_payload is not None else payload - for ref in concurrent_refs: - plugin_input = self._isolate_payload(effective_payload, policy) - local_context = self._prepare_plugin_context(ref, global_context, local_contexts, res_local_contexts) - idx = len(concurrent_ctx_list) - concurrent_ctx_list.append((ref, local_context, effective_payload)) - coro = self.execute_plugin( - ref, - plugin_input, - local_context, - violations_as_exceptions, - global_context, - combined_metadata, - extensions=extensions, - ) - if concurrent_semaphore: - coro = self._with_semaphore(concurrent_semaphore, coro) - 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 - ref, _, _ = concurrent_ctx_list[idx] - ctx.hook_chain_executed += 1 - # Propagate retry signal from concurrent plugins - ctx.max_retry_delay_ms = max(ctx.max_retry_delay_ms, result.retry_delay_ms) - if timeout_err is not None: - # 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( + # CONCURRENT: parallel execution with fail-fast on first blocking result + if concurrent_refs: + concurrent_ctx_list: list[tuple[HookRef, PluginContext, PluginPayload]] = [] + concurrent_tasks: list[asyncio.Task] = [] + effective_payload = current_payload if current_payload is not None else payload + for ref in concurrent_refs: + plugin_input = self._isolate_payload(effective_payload, policy) + local_context = self._prepare_plugin_context(ref, global_context, local_contexts, res_local_contexts) + idx = len(concurrent_ctx_list) + concurrent_ctx_list.append((ref, local_context, effective_payload)) + coro = self.execute_plugin( ref, - hook_type, - ControlExecutionStatus.TIMEOUT, - effective_allow=True, - reason=_truncate_opt(str(timeout_err)), - error_code="plugin_timeout", - )) - continue - if result.modified_payload is not None: - logger.debug( - "CONCURRENT plugin %s returned modified_payload on hook %s; " - "discarding (concurrent plugins cannot modify payloads)", - ref.plugin_ref.name, - hook_type, - ) - # 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, - )) - if not result.continue_processing: - pending = sum(1 for t in concurrent_tasks if not t.done()) - violation_detail = ( - f": [{result.violation.code}] {result.violation.reason}" if result.violation else "" - ) - logger.warning( - "Pipeline halted by CONCURRENT plugin %s on hook %s%s; cancelling %d pending task(s)", - ref.plugin_ref.name, - hook_type, - violation_detail, - pending, - ) - for task in concurrent_tasks: - if not task.done(): - task.cancel() - await asyncio.gather(*concurrent_tasks, return_exceptions=True) - ctx.hook_chain_stopped_by = ref.plugin_ref.name - halt = self._build_halt_result( - current_payload, - result.violation, - combined_metadata, - fire_and_forget_refs, - payload, + plugin_input, + local_context, + violations_as_exceptions, global_context, - res_local_contexts, - fire_and_forget_semaphore, - hook_type, - decision_plugin_name, + combined_metadata, extensions=extensions, - executions=executions, ) - self._end_hook_chain_span(ctx, status="ok") - return halt + if concurrent_semaphore: + coro = self._with_semaphore(concurrent_semaphore, coro) + concurrent_tasks.append(asyncio.create_task(self._tagged(coro, idx))) + + for completed_coro in asyncio.as_completed(concurrent_tasks): + 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 + ctx.max_retry_delay_ms = max(ctx.max_retry_delay_ms, result.retry_delay_ms) + if timeout_err is not None: + # 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", + )) + continue + if violation_err is not None: + # violations_as_exceptions=True concurrent denial: write the record, + # cancel sibling tasks, then re-raise so the outer except attaches + # the full executions list to the exception (fix for #147). + _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) + raise _pve + if result.modified_payload is not None: + logger.debug( + "CONCURRENT plugin %s returned modified_payload on hook %s; " + "discarding (concurrent plugins cannot modify payloads)", + ref.plugin_ref.name, + hook_type, + ) + # 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, + )) + if not result.continue_processing: + pending = sum(1 for t in concurrent_tasks if not t.done()) + violation_detail = ( + f": [{result.violation.code}] {result.violation.reason}" if result.violation else "" + ) + logger.warning( + "Pipeline halted by CONCURRENT plugin %s on hook %s%s; cancelling %d pending task(s)", + ref.plugin_ref.name, + hook_type, + violation_detail, + pending, + ) + for task in concurrent_tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*concurrent_tasks, return_exceptions=True) + ctx.hook_chain_stopped_by = ref.plugin_ref.name + halt = self._build_halt_result( + current_payload, + result.violation, + combined_metadata, + fire_and_forget_refs, + payload, + global_context, + res_local_contexts, + fire_and_forget_semaphore, + hook_type, + decision_plugin_name, + extensions=extensions, + executions=executions, + ) + self._end_hook_chain_span(ctx, status="ok") + return halt + + # FIRE_AND_FORGET: fire-and-forget background tasks (fires last with final payload snapshot) + bg_tasks = self._fire_and_forget_tasks( + fire_and_forget_refs, + payload, + global_context, + res_local_contexts, + fire_and_forget_semaphore, + extensions=extensions, + executions=executions, + hook_type=hook_type, + ) - # FIRE_AND_FORGET: fire-and-forget background tasks (fires last with final payload snapshot) - bg_tasks = self._fire_and_forget_tasks( - fire_and_forget_refs, - payload, - global_context, - res_local_contexts, - fire_and_forget_semaphore, - extensions=extensions, - executions=executions, - hook_type=hook_type, - ) + if hook_type == HTTP_AUTH_CHECK_PERMISSION_HOOK and decision_plugin_name: + combined_metadata[DECISION_PLUGIN_METADATA_KEY] = decision_plugin_name - if hook_type == HTTP_AUTH_CHECK_PERMISSION_HOOK and decision_plugin_name: - combined_metadata[DECISION_PLUGIN_METADATA_KEY] = decision_plugin_name + self._end_hook_chain_span(ctx, status="ok") - self._end_hook_chain_span(ctx, status="ok") + return ( + PluginResult( + continue_processing=True, + modified_payload=current_payload, + modified_extensions=current_extensions, + violation=None, + metadata=combined_metadata, + background_tasks=bg_tasks, + retry_delay_ms=ctx.max_retry_delay_ms, + executions=list(executions), + ), + res_local_contexts, + ) - return ( - PluginResult( - continue_processing=True, - modified_payload=current_payload, - modified_extensions=current_extensions, - violation=None, - metadata=combined_metadata, - background_tasks=bg_tasks, - retry_delay_ms=ctx.max_retry_delay_ms, - executions=list(executions), - ), - res_local_contexts, - ) + except PluginViolationError as _pve: + # Attach the execution records accumulated so far (including the record for the + # denying plugin written in _run_serial_phase or the concurrent loop) to the + # exception before propagating. FAF plugins are not scheduled on this path. + # Callers can inspect pve.executions to identify which plugin denied the + # invocation — fix for #147. + _pve.executions = list(executions) + raise def _group_by_mode( self, @@ -677,8 +708,23 @@ 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 before propagating + # so that PluginResult.executions always contains an entry for the denying plugin. + 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", + )) raise except PluginError as _pe: # execute_plugin re-raises PluginError when on_error=FAIL — must not swallow. @@ -989,20 +1035,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, diff --git a/tests/unit/cpex/framework/test_execution_records.py b/tests/unit/cpex/framework/test_execution_records.py index 5a475089..186bc0ff 100644 --- a/tests/unit/cpex/framework/test_execution_records.py +++ b/tests/unit/cpex/framework/test_execution_records.py @@ -3,7 +3,7 @@ Copyright 2025 SPDX-License-Identifier: Apache-2.0 -Unit tests for ControlExecutionRecord generation on PluginResult.executions (issue #130). +Unit tests for ControlExecutionRecord generation on PluginResult.executions (issues #130, #147). These tests verify: - executions is always present (empty when no plugins ran) @@ -487,3 +487,330 @@ async def test_executions_concurrent_timeout_ignore_does_not_escape(): await manager.shutdown() PluginManager.reset() + + +# --------------------------------------------------------------------------- +# Issue #130 fix: violation record appended before exception is raised +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_executions_on_violation_exception_single_plugin(): + """When violations_as_exceptions=True and a single plugin denies the request, + the raised PluginViolationError must carry an ``executions`` list with one record + for the denying plugin (fix for cpex#130).""" + from unittest.mock import patch + + from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework.base import HookRef + from cpex.framework.errors import PluginViolationError + from cpex.framework.registry import PluginRef + + class DenyPlugin(Plugin): + async def prompt_pre_fetch(self, payload, context): + return PluginResult( + continue_processing=False, + violation=PluginViolation( + reason="Deny reason", + description="Deny description", + code="DENY_CODE", + ), + ) + + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_no_plugin.yaml") + await manager.initialize() + + config = PluginConfig( + name="DenyPlugin", + description="Always denies", + author="Test", + version="1.0", + tags=["test"], + kind="DenyPlugin", + mode=PluginMode.SEQUENTIAL, + hooks=["prompt_pre_fetch"], + config={}, + ) + plugin = DenyPlugin(config) + + with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get: + hook_ref = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin)) + mock_get.return_value = [hook_ref] + + prompt = PromptPrehookPayload(prompt_id="test", args={"user": "hello"}) + context = GlobalContext(request_id="req-vae-single") + + with pytest.raises(PluginViolationError) as pve: + await manager.invoke_hook( + PromptHookType.PROMPT_PRE_FETCH, + prompt, + global_context=context, + violations_as_exceptions=True, + ) + + # The exception must carry the violation details + assert pve.value.violation is not None + assert pve.value.violation.code == "DENY_CODE" + + # executions is attached to the exception (fix for cpex#130) + assert pve.value.executions is not None, ( + "PluginViolationError.executions must not be None when raised from invoke_hook" + ) + assert len(pve.value.executions) == 1 + rec = pve.value.executions[0] + assert rec.plugin_name == "DenyPlugin" + assert rec.effective_allow is False + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.requested_allow is False + assert rec.matched is True + assert rec.applied is True + assert rec.error_code == "DENY_CODE" + assert rec.duration_ns > 0 + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_on_violation_exception_two_plugin_chain(): + """When violations_as_exceptions=True and the second plugin in a two-plugin chain denies, + the raised PluginViolationError must carry an ``executions`` list with records for BOTH + plugins — the first plugin (allow) and the second plugin (deny) (fix for cpex#130).""" + from unittest.mock import patch + + from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework.base import HookRef + from cpex.framework.errors import PluginViolationError + from cpex.framework.registry import PluginRef + + class AllowPlugin(Plugin): + async def prompt_pre_fetch(self, payload, context): + return PluginResult(continue_processing=True) + + class DenyPlugin(Plugin): + async def prompt_pre_fetch(self, payload, context): + return PluginResult( + continue_processing=False, + violation=PluginViolation( + reason="Second plugin deny", + description="Blocked by second plugin", + code="SECOND_DENY", + ), + ) + + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_no_plugin.yaml") + await manager.initialize() + + allow_config = PluginConfig( + name="AllowPlugin", + description="Always allows", + author="Test", + version="1.0", + tags=["test"], + kind="AllowPlugin", + mode=PluginMode.SEQUENTIAL, + hooks=["prompt_pre_fetch"], + config={}, + ) + deny_config = PluginConfig( + name="DenyPlugin", + description="Always denies", + author="Test", + version="1.0", + tags=["test"], + kind="DenyPlugin", + mode=PluginMode.SEQUENTIAL, + hooks=["prompt_pre_fetch"], + config={}, + ) + allow_plugin = AllowPlugin(allow_config) + deny_plugin = DenyPlugin(deny_config) + + with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get: + ref_allow = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(allow_plugin)) + ref_deny = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(deny_plugin)) + mock_get.return_value = [ref_allow, ref_deny] + + prompt = PromptPrehookPayload(prompt_id="test", args={"user": "hello"}) + context = GlobalContext(request_id="req-vae-two-plugin") + + with pytest.raises(PluginViolationError) as pve: + await manager.invoke_hook( + PromptHookType.PROMPT_PRE_FETCH, + prompt, + global_context=context, + violations_as_exceptions=True, + ) + + assert pve.value.violation is not None + assert pve.value.violation.code == "SECOND_DENY" + + # Both plugins must have records on the exception (fix for cpex#130) + assert pve.value.executions is not None, ( + "PluginViolationError.executions must not be None when raised from invoke_hook" + ) + assert len(pve.value.executions) == 2, ( + f"expected 2 execution records (allow + deny), got {len(pve.value.executions)}" + ) + + allow_rec = pve.value.executions[0] + assert allow_rec.plugin_name == "AllowPlugin" + assert allow_rec.effective_allow is True + + deny_rec = pve.value.executions[1] + assert deny_rec.plugin_name == "DenyPlugin" + assert deny_rec.effective_allow is False + assert deny_rec.status == ControlExecutionStatus.COMPLETED + assert deny_rec.error_code == "SECOND_DENY" + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_on_violation_exception_concurrent_plugin(): + """When violations_as_exceptions=True and a CONCURRENT plugin denies, the raised + PluginViolationError must carry a record for the denying plugin (fix for #147). + This is the concurrent-mode analogue of test_executions_on_violation_exception_single_plugin.""" + from unittest.mock import patch + + from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework.base import HookRef + from cpex.framework.errors import PluginViolationError + from cpex.framework.registry import PluginRef + + class ConcurrentDenyPlugin(Plugin): + async def prompt_pre_fetch(self, payload, context): + return PluginResult( + continue_processing=False, + violation=PluginViolation( + reason="Concurrent deny reason", + description="Concurrent deny description", + code="CONC_DENY", + ), + ) + + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_no_plugin.yaml") + await manager.initialize() + + config = PluginConfig( + name="ConcurrentDenyPlugin", + description="Always denies in concurrent mode", + author="Test", + version="1.0", + tags=["test"], + kind="ConcurrentDenyPlugin", + mode=PluginMode.CONCURRENT, + hooks=["prompt_pre_fetch"], + config={}, + ) + plugin = ConcurrentDenyPlugin(config) + + with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get: + hook_ref = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin)) + mock_get.return_value = [hook_ref] + + prompt = PromptPrehookPayload(prompt_id="test", args={"user": "hello"}) + context = GlobalContext(request_id="req-vae-concurrent") + + with pytest.raises(PluginViolationError) as pve: + await manager.invoke_hook( + PromptHookType.PROMPT_PRE_FETCH, + prompt, + global_context=context, + violations_as_exceptions=True, + ) + + assert pve.value.violation is not None + assert pve.value.violation.code == "CONC_DENY" + + # Concurrent denial must also produce a record on the exception (fix for #147) + assert pve.value.executions is not None, ( + "PluginViolationError.executions must not be None for concurrent-mode denial" + ) + assert len(pve.value.executions) == 1 + rec = pve.value.executions[0] + assert rec.plugin_name == "ConcurrentDenyPlugin" + assert rec.mode == PluginMode.CONCURRENT + assert rec.effective_allow is False + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.requested_allow is False + assert rec.matched is True + assert rec.applied is True + assert rec.error_code == "CONC_DENY" + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_on_violation_exception_list_is_independent_copy(): + """pve.executions must be a distinct list from the manager's internal accumulator — + mutating it after the catch must not affect any other state.""" + from unittest.mock import patch + + from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework.base import HookRef + from cpex.framework.errors import PluginViolationError + from cpex.framework.registry import PluginRef + + class DenyPlugin(Plugin): + async def prompt_pre_fetch(self, payload, context): + return PluginResult( + continue_processing=False, + violation=PluginViolation(reason="r", description="d", code="COPY_CHECK"), + ) + + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_no_plugin.yaml") + await manager.initialize() + + config = PluginConfig( + name="DenyPlugin", + description="Always denies", + author="Test", + version="1.0", + tags=["test"], + kind="DenyPlugin", + mode=PluginMode.SEQUENTIAL, + hooks=["prompt_pre_fetch"], + config={}, + ) + plugin = DenyPlugin(config) + + with patch.object(manager._registry, "get_hook_refs_for_hook") as mock_get: + hook_ref = HookRef(PromptHookType.PROMPT_PRE_FETCH, PluginRef(plugin)) + mock_get.return_value = [hook_ref] + + prompt = PromptPrehookPayload(prompt_id="test", args={"user": "hello"}) + context = GlobalContext(request_id="req-vae-copy") + + with pytest.raises(PluginViolationError) as pve: + await manager.invoke_hook( + PromptHookType.PROMPT_PRE_FETCH, + prompt, + global_context=context, + violations_as_exceptions=True, + ) + + executions = pve.value.executions + assert executions is not None + original_len = len(executions) + + # Mutate the captured list — a second call must produce a fresh list unaffected + executions.clear() + assert len(executions) == 0 + + with pytest.raises(PluginViolationError) as pve2: + await manager.invoke_hook( + PromptHookType.PROMPT_PRE_FETCH, + prompt, + global_context=context, + violations_as_exceptions=True, + ) + + # The second exception's list must be independent + assert pve2.value.executions is not executions + assert len(pve2.value.executions) == original_len + + await manager.shutdown() + PluginManager.reset() From 1289f2c66b1326735e6fa92ad88c142c90b2d141 Mon Sep 17 00:00:00 2001 From: Frederico Araujo Date: Tue, 4 Aug 2026 14:58:21 -0400 Subject: [PATCH 2/6] fix(#147): correct issue refs, drop oversized try clause Attach executions at the two PluginViolationError raise sites instead of wrapping the whole pipeline in a try, which tripped ruff PLW0717. Retarget stale #130 references to #147 and note that FAF records are absent on the exception path. Signed-off-by: Frederico Araujo --- cpex/framework/errors.py | 8 +- cpex/framework/manager.py | 449 +++++++++--------- .../cpex/framework/test_execution_records.py | 18 +- 3 files changed, 236 insertions(+), 239 deletions(-) diff --git a/cpex/framework/errors.py b/cpex/framework/errors.py index c719537f..c29d2bba 100644 --- a/cpex/framework/errors.py +++ b/cpex/framework/errors.py @@ -20,9 +20,11 @@ class PluginViolationError(Exception): 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 an - ``invoke_hook`` call (e.g. in unit tests). Populated by ``PluginExecutor.execute`` - before the exception propagates to callers (fix for issue #147). + 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): diff --git a/cpex/framework/manager.py b/cpex/framework/manager.py index 1f4bb317..3e708d54 100644 --- a/cpex/framework/manager.py +++ b/cpex/framework/manager.py @@ -317,242 +317,234 @@ async def execute( fire_and_forget_semaphore = asyncio.Semaphore(pool) if pool else None concurrent_semaphore = asyncio.Semaphore(pool) if pool else None - try: - # SEQUENTIAL: sequential, chained execution — can halt pipeline - halt_result, phase = await self._run_serial_phase( - hook_refs=sequential_refs, - mode_label="SEQUENTIAL", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=True, - allow_blocking=True, - ctx=ctx, - current_extensions=current_extensions, - fire_and_forget_refs=fire_and_forget_refs, - fire_and_forget_semaphore=fire_and_forget_semaphore, - extensions=extensions, - executions=executions, - ) - current_payload = phase.payload - decision_plugin_name = phase.decision_plugin - current_extensions = phase.extensions - if halt_result is not None: - # Attach records collected so far (including FAF scheduled on halt) - halt_result[0].executions = list(executions) - self._end_hook_chain_span(ctx, status="ok") - return halt_result - - # TRANSFORM: serial, chained execution — can modify payloads but cannot halt pipeline - _, phase = await self._run_serial_phase( - hook_refs=transform_refs, - mode_label="TRANSFORM", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=True, - allow_blocking=False, - ctx=ctx, - current_extensions=current_extensions, - extensions=extensions, - executions=executions, - ) - current_payload = phase.payload - decision_plugin_name = phase.decision_plugin - current_extensions = phase.extensions - - # AUDIT: serial execution — observe-only (no modifications, no blocking) - _, phase = await self._run_serial_phase( - hook_refs=audit_refs, - mode_label="AUDIT", - payload=payload, - policy=policy, - hook_type=hook_type, - global_context=global_context, - local_contexts=local_contexts, - res_local_contexts=res_local_contexts, - violations_as_exceptions=violations_as_exceptions, - combined_metadata=combined_metadata, - current_payload=current_payload, - decision_plugin_name=decision_plugin_name, - apply_modifications=False, - allow_blocking=False, - ctx=ctx, - current_extensions=current_extensions, - extensions=extensions, - executions=executions, - ) + # SEQUENTIAL: sequential, chained execution — can halt pipeline + halt_result, phase = await self._run_serial_phase( + hook_refs=sequential_refs, + mode_label="SEQUENTIAL", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=True, + allow_blocking=True, + ctx=ctx, + current_extensions=current_extensions, + fire_and_forget_refs=fire_and_forget_refs, + fire_and_forget_semaphore=fire_and_forget_semaphore, + extensions=extensions, + executions=executions, + ) + current_payload = phase.payload + decision_plugin_name = phase.decision_plugin + current_extensions = phase.extensions + if halt_result is not None: + # Attach records collected so far (including FAF scheduled on halt) + halt_result[0].executions = list(executions) + self._end_hook_chain_span(ctx, status="ok") + return halt_result + + # TRANSFORM: serial, chained execution — can modify payloads but cannot halt pipeline + _, phase = await self._run_serial_phase( + hook_refs=transform_refs, + mode_label="TRANSFORM", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=True, + allow_blocking=False, + ctx=ctx, + current_extensions=current_extensions, + extensions=extensions, + executions=executions, + ) + current_payload = phase.payload + decision_plugin_name = phase.decision_plugin + current_extensions = phase.extensions + + # AUDIT: serial execution — observe-only (no modifications, no blocking) + _, phase = await self._run_serial_phase( + hook_refs=audit_refs, + mode_label="AUDIT", + payload=payload, + policy=policy, + hook_type=hook_type, + global_context=global_context, + local_contexts=local_contexts, + res_local_contexts=res_local_contexts, + violations_as_exceptions=violations_as_exceptions, + combined_metadata=combined_metadata, + current_payload=current_payload, + decision_plugin_name=decision_plugin_name, + apply_modifications=False, + allow_blocking=False, + ctx=ctx, + current_extensions=current_extensions, + extensions=extensions, + executions=executions, + ) - # CONCURRENT: parallel execution with fail-fast on first blocking result - if concurrent_refs: - concurrent_ctx_list: list[tuple[HookRef, PluginContext, PluginPayload]] = [] - concurrent_tasks: list[asyncio.Task] = [] - effective_payload = current_payload if current_payload is not None else payload - for ref in concurrent_refs: - plugin_input = self._isolate_payload(effective_payload, policy) - local_context = self._prepare_plugin_context(ref, global_context, local_contexts, res_local_contexts) - idx = len(concurrent_ctx_list) - concurrent_ctx_list.append((ref, local_context, effective_payload)) - coro = self.execute_plugin( + # CONCURRENT: parallel execution with fail-fast on first blocking result + if concurrent_refs: + concurrent_ctx_list: list[tuple[HookRef, PluginContext, PluginPayload]] = [] + concurrent_tasks: list[asyncio.Task] = [] + effective_payload = current_payload if current_payload is not None else payload + for ref in concurrent_refs: + plugin_input = self._isolate_payload(effective_payload, policy) + local_context = self._prepare_plugin_context(ref, global_context, local_contexts, res_local_contexts) + idx = len(concurrent_ctx_list) + concurrent_ctx_list.append((ref, local_context, effective_payload)) + coro = self.execute_plugin( + ref, + plugin_input, + local_context, + violations_as_exceptions, + global_context, + combined_metadata, + extensions=extensions, + ) + if concurrent_semaphore: + coro = self._with_semaphore(concurrent_semaphore, coro) + concurrent_tasks.append(asyncio.create_task(self._tagged(coro, idx))) + + for completed_coro in asyncio.as_completed(concurrent_tasks): + 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 + ctx.max_retry_delay_ms = max(ctx.max_retry_delay_ms, result.retry_delay_ms) + if timeout_err is not None: + # 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, - plugin_input, - local_context, - violations_as_exceptions, - global_context, - combined_metadata, - extensions=extensions, - ) - if concurrent_semaphore: - coro = self._with_semaphore(concurrent_semaphore, coro) - concurrent_tasks.append(asyncio.create_task(self._tagged(coro, idx))) - - for completed_coro in asyncio.as_completed(concurrent_tasks): - 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 - ctx.max_retry_delay_ms = max(ctx.max_retry_delay_ms, result.retry_delay_ms) - if timeout_err is not None: - # 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", - )) - continue - if violation_err is not None: - # violations_as_exceptions=True concurrent denial: write the record, - # cancel sibling tasks, then re-raise so the outer except attaches - # the full executions list to the exception (fix for #147). - _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) - raise _pve - if result.modified_payload is not None: - logger.debug( - "CONCURRENT plugin %s returned modified_payload on hook %s; " - "discarding (concurrent plugins cannot modify payloads)", - ref.plugin_ref.name, - hook_type, - ) - # Build the concurrent execution record (duration=0: no per-branch timing) - _concurrent_denied = not result.continue_processing + 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=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, + 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", )) - if not result.continue_processing: - pending = sum(1 for t in concurrent_tasks if not t.done()) - violation_detail = ( - f": [{result.violation.code}] {result.violation.reason}" if result.violation else "" - ) - logger.warning( - "Pipeline halted by CONCURRENT plugin %s on hook %s%s; cancelling %d pending task(s)", - ref.plugin_ref.name, - hook_type, - violation_detail, - pending, - ) - for task in concurrent_tasks: - if not task.done(): - task.cancel() - await asyncio.gather(*concurrent_tasks, return_exceptions=True) - ctx.hook_chain_stopped_by = ref.plugin_ref.name - halt = self._build_halt_result( - current_payload, - result.violation, - combined_metadata, - fire_and_forget_refs, - payload, - global_context, - res_local_contexts, - fire_and_forget_semaphore, - hook_type, - decision_plugin_name, - extensions=extensions, - executions=executions, - ) - self._end_hook_chain_span(ctx, status="ok") - return halt - - # FIRE_AND_FORGET: fire-and-forget background tasks (fires last with final payload snapshot) - bg_tasks = self._fire_and_forget_tasks( - fire_and_forget_refs, - payload, - global_context, - res_local_contexts, - fire_and_forget_semaphore, - extensions=extensions, - executions=executions, - hook_type=hook_type, - ) + 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; " + "discarding (concurrent plugins cannot modify payloads)", + ref.plugin_ref.name, + hook_type, + ) + # 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, + )) + if not result.continue_processing: + pending = sum(1 for t in concurrent_tasks if not t.done()) + violation_detail = ( + f": [{result.violation.code}] {result.violation.reason}" if result.violation else "" + ) + logger.warning( + "Pipeline halted by CONCURRENT plugin %s on hook %s%s; cancelling %d pending task(s)", + ref.plugin_ref.name, + hook_type, + violation_detail, + pending, + ) + for task in concurrent_tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*concurrent_tasks, return_exceptions=True) + ctx.hook_chain_stopped_by = ref.plugin_ref.name + halt = self._build_halt_result( + current_payload, + result.violation, + combined_metadata, + fire_and_forget_refs, + payload, + global_context, + res_local_contexts, + fire_and_forget_semaphore, + hook_type, + decision_plugin_name, + extensions=extensions, + executions=executions, + ) + self._end_hook_chain_span(ctx, status="ok") + return halt - if hook_type == HTTP_AUTH_CHECK_PERMISSION_HOOK and decision_plugin_name: - combined_metadata[DECISION_PLUGIN_METADATA_KEY] = decision_plugin_name + # FIRE_AND_FORGET: fire-and-forget background tasks (fires last with final payload snapshot) + bg_tasks = self._fire_and_forget_tasks( + fire_and_forget_refs, + payload, + global_context, + res_local_contexts, + fire_and_forget_semaphore, + extensions=extensions, + executions=executions, + hook_type=hook_type, + ) - self._end_hook_chain_span(ctx, status="ok") + if hook_type == HTTP_AUTH_CHECK_PERMISSION_HOOK and decision_plugin_name: + combined_metadata[DECISION_PLUGIN_METADATA_KEY] = decision_plugin_name - return ( - PluginResult( - continue_processing=True, - modified_payload=current_payload, - modified_extensions=current_extensions, - violation=None, - metadata=combined_metadata, - background_tasks=bg_tasks, - retry_delay_ms=ctx.max_retry_delay_ms, - executions=list(executions), - ), - res_local_contexts, - ) + self._end_hook_chain_span(ctx, status="ok") - except PluginViolationError as _pve: - # Attach the execution records accumulated so far (including the record for the - # denying plugin written in _run_serial_phase or the concurrent loop) to the - # exception before propagating. FAF plugins are not scheduled on this path. - # Callers can inspect pve.executions to identify which plugin denied the - # invocation — fix for #147. - _pve.executions = list(executions) - raise + return ( + PluginResult( + continue_processing=True, + modified_payload=current_payload, + modified_extensions=current_extensions, + violation=None, + metadata=combined_metadata, + background_tasks=bg_tasks, + retry_delay_ms=ctx.max_retry_delay_ms, + executions=list(executions), + ), + res_local_contexts, + ) def _group_by_mode( self, @@ -709,8 +701,10 @@ async def _run_serial_phase( exec_error_code: Optional[str] = None exec_reason: Optional[str] = None except PluginViolationError as _pve: - # violations_as_exceptions=True — append the denial record before propagating - # so that PluginResult.executions always contains an entry for the denying plugin. + # 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( @@ -725,6 +719,7 @@ async def _run_serial_phase( 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. diff --git a/tests/unit/cpex/framework/test_execution_records.py b/tests/unit/cpex/framework/test_execution_records.py index 186bc0ff..35347f4c 100644 --- a/tests/unit/cpex/framework/test_execution_records.py +++ b/tests/unit/cpex/framework/test_execution_records.py @@ -490,7 +490,7 @@ async def test_executions_concurrent_timeout_ignore_does_not_escape(): # --------------------------------------------------------------------------- -# Issue #130 fix: violation record appended before exception is raised +# Issue #147 fix: violation record appended before exception is raised # --------------------------------------------------------------------------- @@ -498,10 +498,10 @@ async def test_executions_concurrent_timeout_ignore_does_not_escape(): async def test_executions_on_violation_exception_single_plugin(): """When violations_as_exceptions=True and a single plugin denies the request, the raised PluginViolationError must carry an ``executions`` list with one record - for the denying plugin (fix for cpex#130).""" + for the denying plugin (fix for #147).""" from unittest.mock import patch - from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework import Plugin, PluginConfig, PluginViolation from cpex.framework.base import HookRef from cpex.framework.errors import PluginViolationError from cpex.framework.registry import PluginRef @@ -552,7 +552,7 @@ async def prompt_pre_fetch(self, payload, context): assert pve.value.violation is not None assert pve.value.violation.code == "DENY_CODE" - # executions is attached to the exception (fix for cpex#130) + # executions is attached to the exception (fix for #147) assert pve.value.executions is not None, ( "PluginViolationError.executions must not be None when raised from invoke_hook" ) @@ -575,10 +575,10 @@ async def prompt_pre_fetch(self, payload, context): async def test_executions_on_violation_exception_two_plugin_chain(): """When violations_as_exceptions=True and the second plugin in a two-plugin chain denies, the raised PluginViolationError must carry an ``executions`` list with records for BOTH - plugins — the first plugin (allow) and the second plugin (deny) (fix for cpex#130).""" + plugins — the first plugin (allow) and the second plugin (deny) (fix for #147).""" from unittest.mock import patch - from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework import Plugin, PluginConfig, PluginViolation from cpex.framework.base import HookRef from cpex.framework.errors import PluginViolationError from cpex.framework.registry import PluginRef @@ -645,7 +645,7 @@ async def prompt_pre_fetch(self, payload, context): assert pve.value.violation is not None assert pve.value.violation.code == "SECOND_DENY" - # Both plugins must have records on the exception (fix for cpex#130) + # Both plugins must have records on the exception (fix for #147) assert pve.value.executions is not None, ( "PluginViolationError.executions must not be None when raised from invoke_hook" ) @@ -674,7 +674,7 @@ async def test_executions_on_violation_exception_concurrent_plugin(): This is the concurrent-mode analogue of test_executions_on_violation_exception_single_plugin.""" from unittest.mock import patch - from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework import Plugin, PluginConfig, PluginViolation from cpex.framework.base import HookRef from cpex.framework.errors import PluginViolationError from cpex.framework.registry import PluginRef @@ -749,7 +749,7 @@ async def test_executions_on_violation_exception_list_is_independent_copy(): mutating it after the catch must not affect any other state.""" from unittest.mock import patch - from cpex.framework import Plugin, PluginConfig, PluginMode, PluginResult, PluginViolation + from cpex.framework import Plugin, PluginConfig, PluginViolation from cpex.framework.base import HookRef from cpex.framework.errors import PluginViolationError from cpex.framework.registry import PluginRef From d0e838e3d810e865c7cc8ea738815a4bfa7ee791 Mon Sep 17 00:00:00 2001 From: Frederico Araujo Date: Tue, 4 Aug 2026 14:58:54 -0400 Subject: [PATCH 3/6] ci: run lint and tests on dev and 0.1.x branches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PRs targeting 0.1.x triggered no checks — every workflow filtered on main only. Add dev and 0.1.x, the long-lived maintenance base branch for the 0.1 series. Drop the docs job from the CI pipeline on this branch: docs are built and published from main only. It has to go from ci.yml rather than docs-build.yaml, because workflow_call bypasses the called workflow's own branch filter. Signed-off-by: Frederico Araujo --- .github/workflows/ci.yml | 12 +++++++----- .github/workflows/lint.yml | 5 +++-- .github/workflows/tests.yaml | 5 +++-- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec838293..1d461294 100644 --- a/.github/workflows/ci.yml +++ b/.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. diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index f4f84c88..94740092 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -12,14 +12,15 @@ name: Lint & Static Analysis on: workflow_call: push: - branches: ["main"] + # 0.1.x is the long-lived maintenance base branch for the 0.1 series + branches: ["main", "dev", "0.1.x"] paths: - "cpex/**" - "pyproject.toml" - ".github/workflows/lint.yml" pull_request: types: [opened, synchronize, ready_for_review] - branches: ["main"] + branches: ["main", "dev", "0.1.x"] paths: - "cpex/**" - "pyproject.toml" diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 96ffa61a..206cd811 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -7,7 +7,8 @@ name: Tests on: workflow_call: push: - branches: ["main"] + # 0.1.x is the long-lived maintenance base branch for the 0.1 series + branches: ["main", "dev", "0.1.x"] paths: - "cpex/**" - "tests/**" @@ -15,7 +16,7 @@ on: - ".github/workflows/tests.yaml" pull_request: types: [opened, synchronize, ready_for_review] - branches: ["main"] + branches: ["main", "dev", "0.1.x"] paths: - "cpex/**" - "tests/**" From 26c34af5921324f7a7a692047f73bffbce5ef7ac Mon Sep 17 00:00:00 2001 From: Frederico Araujo Date: Tue, 4 Aug 2026 15:42:50 -0400 Subject: [PATCH 4/6] ci: run each job once instead of twice ci.yml calls lint.yml and tests.yaml via workflow_call, but both also had their own push/pull_request triggers, so every job ran twice per PR. Make them reusable-only; ci.yml owns the branch filters. Signed-off-by: Frederico Araujo --- .github/workflows/lint.yml | 17 +++-------------- .github/workflows/tests.yaml | 19 +++---------------- 2 files changed, 6 insertions(+), 30 deletions(-) diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 94740092..b4aa936a 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -9,22 +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: - # 0.1.x is the long-lived maintenance base branch for the 0.1 series - branches: ["main", "dev", "0.1.x"] - paths: - - "cpex/**" - - "pyproject.toml" - - ".github/workflows/lint.yml" - pull_request: - types: [opened, synchronize, ready_for_review] - branches: ["main", "dev", "0.1.x"] - paths: - - "cpex/**" - - "pyproject.toml" - - ".github/workflows/lint.yml" permissions: contents: read diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 206cd811..079b1689 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -4,24 +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: - # 0.1.x is the long-lived maintenance base branch for the 0.1 series - branches: ["main", "dev", "0.1.x"] - paths: - - "cpex/**" - - "tests/**" - - "pyproject.toml" - - ".github/workflows/tests.yaml" - pull_request: - types: [opened, synchronize, ready_for_review] - branches: ["main", "dev", "0.1.x"] - paths: - - "cpex/**" - - "tests/**" - - "pyproject.toml" - - ".github/workflows/tests.yaml" workflow_dispatch: permissions: From 3830c451825f45253686b199f52a56c9e9a6d117 Mon Sep 17 00:00:00 2001 From: Frederico Araujo Date: Tue, 4 Aug 2026 15:43:09 -0400 Subject: [PATCH 5/6] fix: cap mcp below 2.0 mcp 2.0.0 renamed McpError -> MCPError, breaking the import in cpex/framework/external/mcp/client.py and erroring out test collection. Verified McpError is present in 1.26.0 through 1.29.0 and absent in 2.0.0, so the cap is <2 rather than a tighter pin. Signed-off-by: Frederico Araujo --- pyproject.toml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index a7f91422..83bc579a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,10 @@ dependencies = [ "httpx>=0.28.1", "httpx[http2]>=0.28.1", "jinja2>=3.1.6", - "mcp>=1.26.0", + # Capped below 2.0: mcp 2.0 renamed McpError -> MCPError, which breaks + # cpex/framework/external/mcp/client.py. The 2.0 migration is tracked + # separately; the 0.1.x line stays on mcp 1.x. + "mcp>=1.26.0,<2", "orjson>=3.11.7", "prometheus-fastapi-instrumentator>=7.1.0", "prometheus_client>=0.24.1", From 8191f1f80acec35691c0b5187d31933a09652106 Mon Sep 17 00:00:00 2001 From: Frederico Araujo Date: Tue, 4 Aug 2026 15:43:09 -0400 Subject: [PATCH 6/6] style: apply ruff format to manager and models MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clears the two E303 errors that failed the ruff job and the outstanding ruff format diff. Formatting only — ASTs are unchanged. Signed-off-by: Frederico Araujo --- cpex/framework/manager.py | 180 ++++++++++++++++++++------------------ cpex/framework/models.py | 2 - 2 files changed, 95 insertions(+), 87 deletions(-) diff --git a/cpex/framework/manager.py b/cpex/framework/manager.py index 3e708d54..e5b7cb92 100644 --- a/cpex/framework/manager.py +++ b/cpex/framework/manager.py @@ -429,14 +429,16 @@ 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, @@ -444,17 +446,19 @@ async def execute( # 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", - )) + 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() @@ -470,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 = ( @@ -707,18 +713,20 @@ async def _run_serial_phase( # 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", - )) + 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: @@ -726,16 +734,18 @@ async def _run_serial_phase( # 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. @@ -795,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 "" @@ -1097,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) @@ -1280,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: diff --git a/cpex/framework/models.py b/cpex/framework/models.py index 7ae29974..46c0bace 100644 --- a/cpex/framework/models.py +++ b/cpex/framework/models.py @@ -1656,7 +1656,6 @@ def create_instance_config( ) - # --------------------------------------------------------------------------- # Execution record types (issue #130) # --------------------------------------------------------------------------- @@ -1815,7 +1814,6 @@ class ControlExecutionRecord(BaseModel): """Config key *names* from the plugin's trusted config. Values are never included.""" - class PluginErrorModel(BaseModel): """A plugin error, used to denote exceptions/errors inside external plugins.