diff --git a/CHANGELOG.md b/CHANGELOG.md index f7d9966a..9b5605eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,17 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/). ## [Unreleased] +### Added + +- Structured control execution records for enforcement observability ([#130](https://github.com/contextforge-org/cpex/issues/130)) + - `ControlExecutionStatus` enum and `ControlExecutionRecord` Pydantic model in `cpex.framework.models` + - `PluginResult.executions: list[ControlExecutionRecord]` — one record per plugin evaluated, always present + - All five execution phases instrumented: Sequential, Transform, Audit, Concurrent, Fire-and-forget + - Identity fields (`plugin_id`, `plugin_name`, `plugin_kind`, `mode`) sourced from trusted `PluginRef` config — plugins cannot forge these + - Monotonic per-plugin timing (`duration_ns`); fire-and-forget records use `duration_ns=0` at spawn time + - Security bounds: string fields capped at 256 bytes, config key lists capped at 64 entries, config values never stored + - `ControlExecutionRecord` and `ControlExecutionStatus` exported from `cpex.framework` + ## [0.1.1] - 2026-06-04 ### Added diff --git a/cpex/framework/__init__.py b/cpex/framework/__init__.py index aa3a8399..1b0d126a 100644 --- a/cpex/framework/__init__.py +++ b/cpex/framework/__init__.py @@ -68,6 +68,8 @@ from cpex.framework.loader.plugin import PluginLoader from cpex.framework.manager import PluginManager, TenantPluginManager from cpex.framework.models import ( + ControlExecutionRecord, + ControlExecutionStatus, GlobalContext, MCPClientConfig, MCPServerConfig, @@ -135,6 +137,8 @@ def get_plugin_manager( "AgentPreInvokePayload", "AgentPreInvokeResult", "ConfigLoader", + "ControlExecutionRecord", + "ControlExecutionStatus", "ExternalPluginServer", "get_attr", "get_hook_registry", @@ -181,12 +185,12 @@ def get_plugin_manager( "ResourcePostFetchResult", "ResourcePreFetchPayload", "ResourcePreFetchResult", + "TenantPluginManager", "ToolHookType", "ToolPostInvokePayload", "ToolPostInvokeResult", - "ToolPreInvokeResult", - "TenantPluginManager", "ToolPreInvokePayload", + "ToolPreInvokeResult", "TransportType", "UserContext", ] diff --git a/cpex/framework/manager.py b/cpex/framework/manager.py index d67b1a9c..8e2b740b 100644 --- a/cpex/framework/manager.py +++ b/cpex/framework/manager.py @@ -31,6 +31,7 @@ import asyncio import logging import threading +import time from dataclasses import dataclass from typing import Any, Literal, Optional, Union @@ -49,6 +50,8 @@ from cpex.framework.memory import _safe_deepcopy, copyonwrite, wrap_payload_for_isolation from cpex.framework.models import ( Config, + ControlExecutionRecord, + ControlExecutionStatus, GlobalContext, OnError, PluginContext, @@ -57,6 +60,9 @@ PluginMode, PluginPayload, PluginResult, + _collect_config_keys, + _truncate, + _truncate_opt, ) from cpex.framework.observability import ObservabilityProvider, current_trace_id from cpex.framework.registry import PluginInstanceRegistry @@ -127,6 +133,48 @@ class PayloadSizeError(ValueError): """Raised when a payload exceeds the maximum allowed size.""" +def _make_execution_record( + hook_ref: "HookRef", + hook_type: str, + status: "ControlExecutionStatus", + *, + effective_allow: bool, + duration_ns: int = 0, + requested_allow: Optional[bool] = None, + matched: Optional[bool] = None, + applied: bool = False, + payload_modified: bool = False, + extensions_modified: bool = False, + reason: Optional[str] = None, + error_code: Optional[str] = None, +) -> "ControlExecutionRecord": + """Build a ``ControlExecutionRecord`` from trusted ``PluginRef`` state. + + All identity fields are sourced from ``hook_ref.plugin_ref`` — never from + plugin-returned data. Centralising construction here prevents the field- + drift that caused the original ``matched=None`` inconsistency. + """ + trusted = hook_ref.plugin_ref.trusted_config + return ControlExecutionRecord( + plugin_id=hook_ref.plugin_ref.uuid, + plugin_name=hook_ref.plugin_ref.name, + plugin_kind=_truncate(trusted.kind), + hook_name=hook_type, + mode=trusted.mode, + status=status, + requested_allow=requested_allow, + effective_allow=effective_allow, + matched=matched, + applied=applied, + payload_modified=payload_modified, + extensions_modified=extensions_modified, + duration_ns=duration_ns, + reason=reason, + error_code=error_code, + config_keys=_collect_config_keys(trusted.config), + ) + + class PluginExecutor: """Executes a list of plugins with timeout protection and error handling. @@ -242,6 +290,7 @@ async def execute( current_extensions: Extensions | None = None decision_plugin_name: Optional[str] = None ctx = ExecutionContext() + executions: list[ControlExecutionRecord] = [] # Start hook-chain observability span trace_id = current_trace_id.get() @@ -289,11 +338,14 @@ async def execute( 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 @@ -316,6 +368,7 @@ async def execute( ctx=ctx, current_extensions=current_extensions, extensions=extensions, + executions=executions, ) current_payload = phase.payload decision_plugin_name = phase.decision_plugin @@ -340,6 +393,7 @@ async def execute( ctx=ctx, current_extensions=current_extensions, extensions=extensions, + executions=executions, ) # CONCURRENT: parallel execution with fail-fast on first blocking result @@ -366,11 +420,24 @@ async def execute( concurrent_tasks.append(asyncio.create_task(self._tagged(coro, idx))) for completed_coro in asyncio.as_completed(concurrent_tasks): - result, idx = await completed_coro + 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( + 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; " @@ -378,6 +445,19 @@ async def execute( 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 = ( @@ -407,6 +487,7 @@ async def execute( hook_type, decision_plugin_name, extensions=extensions, + executions=executions, ) self._end_hook_chain_span(ctx, status="ok") return halt @@ -419,6 +500,8 @@ async def execute( 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: @@ -435,6 +518,7 @@ async def execute( metadata=combined_metadata, background_tasks=bg_tasks, retry_delay_ms=ctx.max_retry_delay_ms, + executions=list(executions), ), res_local_contexts, ) @@ -543,6 +627,7 @@ async def _run_serial_phase( fire_and_forget_refs: Optional[list[HookRef]] = None, fire_and_forget_semaphore: Optional[asyncio.Semaphore] = None, extensions: Optional[Extensions] = None, + executions: Optional[list[ControlExecutionRecord]] = None, ) -> tuple[ Optional[tuple[PluginResult, PluginContextTable | None]], PhaseState, @@ -567,6 +652,7 @@ async def _run_serial_phase( ctx: Per-call execution context; counters and stop reason accumulate here. fire_and_forget_refs: Fire-and-forget refs to schedule on halt (only used when allow_blocking=True). fire_and_forget_semaphore: Semaphore for fire-and-forget tasks (only used when allow_blocking=True). + executions: Accumulator list for ControlExecutionRecords; appended in-place. Returns: A tuple of (halt_result, phase_state). halt_result is None if pipeline continues. @@ -576,22 +662,66 @@ async def _run_serial_phase( effective_payload = current_payload if current_payload is not None else payload plugin_input = self._isolate_payload(effective_payload, policy) - result = await self.execute_plugin( - hook_ref, - plugin_input, - local_context, - violations_as_exceptions, - global_context, - combined_metadata, - extensions=extensions, - ) + t_start = time.monotonic_ns() + try: + result = await self.execute_plugin( + hook_ref, + plugin_input, + local_context, + violations_as_exceptions, + global_context, + combined_metadata, + extensions=extensions, + ) + duration_ns = time.monotonic_ns() - t_start + 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 + 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", + )) + raise + except PluginTimeoutError as _te: + # on_error=IGNORE or DISABLE timeout — pipeline continues, record as TIMEOUT. + duration_ns = time.monotonic_ns() - t_start + exec_status = ControlExecutionStatus.TIMEOUT + exec_error_code = "plugin_timeout" + exec_reason = str(_te) + result = PluginResult(continue_processing=True) + except Exception as _exc: + # Unexpected exception that escaped execute_plugin entirely. + duration_ns = time.monotonic_ns() - t_start + exec_status = ControlExecutionStatus.ERROR + exec_error_code = "plugin_error" + exec_reason = str(_exc) + result = PluginResult(continue_processing=True) + ctx.hook_chain_executed += 1 # Propagate retry signal — take the largest delay requested by any plugin ctx.max_retry_delay_ms = max(ctx.max_retry_delay_ms, result.retry_delay_ms) + payload_modified = False + extensions_modified = False + if result.modified_payload is not None: if apply_modifications: + prev_payload = current_payload current_payload, decision_plugin_name = self._apply_payload_modification( hook_ref, result, @@ -602,6 +732,7 @@ async def _run_serial_phase( decision_plugin_name, apply_to=effective_payload, ) + payload_modified = current_payload is not prev_payload else: logger.debug( "%s plugin %s returned modified_payload on hook %s; discarding (%s is observe-only)", @@ -613,7 +744,50 @@ async def _run_serial_phase( # Accumulate modified_extensions (last writer wins) if result.modified_extensions is not None: + prev_ext = current_extensions current_extensions = result.modified_extensions + extensions_modified = current_extensions is not prev_ext + + # Derive matched / applied / effective_allow for the record + denied = not result.continue_processing + if exec_status == ControlExecutionStatus.COMPLETED: + 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 + ) + 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) + ) + 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 + 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, + )) if not result.continue_processing: violation_detail = f": [{result.violation.code}] {result.violation.reason}" if result.violation else "" @@ -641,6 +815,7 @@ async def _run_serial_phase( hook_type, decision_plugin_name, extensions=extensions, + executions=executions, ) return halt, state else: @@ -780,6 +955,7 @@ def _build_halt_result( hook_type: str, decision_plugin_name: Optional[str], extensions: Optional[Extensions] = None, + executions: Optional[list[ControlExecutionRecord]] = None, ) -> tuple[PluginResult, dict]: """Schedule fire-and-forget tasks and build a pipeline-halting result.""" bg_tasks = self._fire_and_forget_tasks( @@ -789,6 +965,8 @@ def _build_halt_result( 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 @@ -799,6 +977,7 @@ def _build_halt_result( violation=violation, metadata=combined_metadata, background_tasks=bg_tasks, + executions=list(executions) if executions is not None else [], ), res_local_contexts, ) @@ -810,10 +989,20 @@ async def _with_semaphore(semaphore: asyncio.Semaphore, coro: Any) -> Any: return await coro @staticmethod - async def _tagged(coro: Any, tag: Any) -> tuple[Any, Any]: - """Await *coro* and pair the result with *tag* for use with as_completed.""" - result = await coro - return result, tag + async def _tagged(coro: Any, tag: Any) -> tuple[Any, Any, Optional["PluginTimeoutError"]]: + """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. + """ + try: + result = await coro + return result, tag, None + except PluginTimeoutError as te: + return PluginResult(continue_processing=True), tag, te def _fire_and_forget_tasks( self, @@ -823,6 +1012,8 @@ def _fire_and_forget_tasks( res_local_contexts: dict, semaphore: Optional[asyncio.Semaphore], extensions: Optional[Extensions] = None, + executions: Optional[list[ControlExecutionRecord]] = None, + hook_type: str = "", ) -> list[asyncio.Task]: """Schedule all FIRE_AND_FORGET plugins as fire-and-forget background tasks. @@ -851,6 +1042,18 @@ def _fire_and_forget_tasks( ) local_context = PluginContext(global_context=tmp_gc) res_local_contexts[local_context_key] = local_context + + # FAF record appended at spawn time — outcome is unknowable when pipeline returns. + # 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, + )) + task = asyncio.create_task( self._run_fire_and_forget_task(ref, task_input, local_context, semaphore, extensions=extensions) ) @@ -1025,6 +1228,11 @@ async def execute_plugin( if on_error == OnError.DISABLE: async with self._runtime_disabled_lock: 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 except PluginViolationError: raise except PluginError as pe: diff --git a/cpex/framework/models.py b/cpex/framework/models.py index e85ffcf3..7ae29974 100644 --- a/cpex/framework/models.py +++ b/cpex/framework/models.py @@ -1656,6 +1656,166 @@ def create_instance_config( ) + +# --------------------------------------------------------------------------- +# Execution record types (issue #130) +# --------------------------------------------------------------------------- + +#: Maximum byte length for free-form string fields in a record. +_MAX_STRING_LEN: int = 256 + +#: Maximum number of config key names per record. +_MAX_CONFIG_KEYS: int = 64 + + +_ELLIPSIS = "…" +_ELLIPSIS_BYTES = len(_ELLIPSIS.encode("utf-8")) # 3 + + +def _truncate(s: str, max_len: int = _MAX_STRING_LEN) -> str: + """Truncate *s* so the UTF-8 encoding stays within *max_len* bytes. + + The result (including the trailing ellipsis when truncated) never exceeds + *max_len* bytes when encoded as UTF-8. + """ + encoded = s.encode("utf-8") + if len(encoded) <= max_len: + return s + # Reserve space for the ellipsis suffix + cap = max_len - _ELLIPSIS_BYTES + return encoded[:cap].decode("utf-8", errors="ignore") + _ELLIPSIS + + +def _truncate_opt(s: Optional[str]) -> Optional[str]: + """Truncate an optional string; returns None unchanged.""" + return _truncate(s) if s is not None else None + + +def _collect_config_keys(config: Optional[dict[str, Any]]) -> list[str]: + """Return key names from *config*, bounded to *_MAX_CONFIG_KEYS*. Values are never included.""" + if not isinstance(config, dict): + return [] + return [_truncate(k) for k in list(config.keys())[:_MAX_CONFIG_KEYS]] + + +class ControlExecutionStatus(StrEnum): + """Execution health of a single control invocation. + + Separate from the allow/deny policy decision — a control can complete + successfully and still deny the operation. + + Examples: + >>> ControlExecutionStatus.COMPLETED + + >>> ControlExecutionStatus.ERROR.value + 'error' + """ + + COMPLETED = "completed" + """Plugin ran to completion (may have allowed or denied). + Also used as a spawn-time placeholder for FIRE_AND_FORGET records whose + outcome is unknowable when the pipeline result is returned — identify FAF + records by ``mode == fire_and_forget``, not by this status.""" + ERROR = "error" + """Plugin returned an error (on_error=ignore/disable) or raised with on_error=fail.""" + TIMEOUT = "timeout" + """Plugin timed out with on_error=ignore or on_error=disable (pipeline continues). + on_error=fail timeouts are converted to PluginError and recorded as ERROR.""" + SKIPPED = "skipped" + """Reserved — not currently emitted. Disabled plugins are silently dropped by group_by_mode.""" + CANCELLED = "cancelled" + """Reserved — not currently emitted. Cancelled concurrent siblings produce no record.""" + DISABLED = "disabled" + """Reserved — not currently emitted. Runtime-disabled plugins are silently dropped.""" + + +class ControlExecutionRecord(BaseModel): + """Trusted execution record for one control/plugin evaluation. + + All identity, mode, status, duration, and effective decision fields are + populated by the executor from the trusted ``PluginRef`` configuration — + never from plugin-returned metadata. A plugin cannot forge these fields. + + Cardinality: free-form string fields (``reason``, ``error_code``) are + bounded to ``_MAX_STRING_LEN`` bytes. Config key lists are bounded to + ``_MAX_CONFIG_KEYS`` entries. Config *values* are never stored. + + Examples: + >>> rec = ControlExecutionRecord( + ... plugin_id="abc", + ... plugin_name="pii-guard", + ... plugin_kind="builtin", + ... hook_name="tool_pre_invoke", + ... mode=PluginMode.SEQUENTIAL, + ... status=ControlExecutionStatus.COMPLETED, + ... effective_allow=True, + ... ) + >>> rec.plugin_name + 'pii-guard' + >>> rec.effective_allow + True + """ + + plugin_id: str + """UUID (hex) assigned to this PluginRef at registration time. + Stable within one process only — do not use for cross-process joins.""" + + plugin_name: str + """Human-readable plugin name from the trusted config. + Stable across restarts — use as primary grouping key.""" + + plugin_kind: str + """Plugin kind from the trusted config (e.g. ``"builtin"``, ``"cpex.opa.OpaPlugin"``).""" + + hook_name: str + """Hook name this invocation was dispatched for.""" + + mode: PluginMode + """Execution mode from the trusted config.""" + + status: ControlExecutionStatus + """Execution health — separate from the allow/deny decision.""" + + requested_allow: Optional[bool] = None + """What the plugin result requested (``True`` = allow, ``False`` = deny). + ``None`` when no result was obtained (error / timeout / cancelled).""" + + effective_allow: bool = True + """Effective decision after execution-mode semantics and ``on_error`` policy. + This is the authoritative per-control allow/deny outcome.""" + + matched: Optional[bool] = None + """Whether the control condition matched, when determinable. + ``True`` — condition fired. + ``False`` — condition did not fire (clean allow, no mutation). + ``None`` — result not obtained (error/timeout only); do not treat as ``False``.""" + + applied: bool = False + """Whether this control changed the payload, extensions, or effective decision.""" + + payload_modified: bool = False + """Whether a payload modification was accepted by the framework.""" + + extensions_modified: bool = False + """Whether an extension modification was accepted by the framework.""" + + duration_ns: int = 0 + """Wall-clock execution duration in nanoseconds (monotonic). + ``0`` for fire-and-forget records (spawn time) and concurrent records.""" + + reason: Optional[str] = None + """Bounded reason string from a violation or error. + May contain user-provided text — do not log at high verbosity without review.""" + + error_code: Optional[str] = None + """Stable low-cardinality error/violation code. + Preferred over free-form text for dashboards and alerting.""" + + config_keys: list[str] = Field(default_factory=list) + """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. @@ -1822,6 +1982,12 @@ class PluginResult(BaseModel, Generic[T]): to await them and collect any errors. This field is excluded from model serialization. http_headers (Optional[dict[str, str]]): HTTP headers to include in successful responses. retry_delay_ms (int): Milliseconds the gateway should wait before retrying the tool call. + executions (list[ControlExecutionRecord]): Structured execution records, one per + *evaluated* control during this invocation. Disabled, skipped (condition unmet), + and cancelled concurrent siblings produce no record in 0.1.x (see + ``ControlExecutionStatus.SKIPPED / DISABLED / CANCELLED``). Populated by the + executor from trusted framework state — plugins cannot forge identity, duration, + or effective decision fields. Examples: >>> result = PluginResult() @@ -1859,6 +2025,9 @@ class PluginResult(BaseModel, Generic[T]): background_tasks: list[asyncio.Task] = Field(default_factory=list, exclude=True) http_headers: Optional[dict[str, str]] = None retry_delay_ms: int = 0 + executions: list[ControlExecutionRecord] = Field(default_factory=list) + """Structured execution records — one per plugin evaluated. + Always present; empty when no plugins ran for the given hook.""" async def wait_for_background_tasks(self) -> "list[PluginErrorModel]": """Await all FIRE_AND_FORGET background tasks and return any errors. diff --git a/tests/unit/cpex/fixtures/configs/execution_records_modes.yaml b/tests/unit/cpex/fixtures/configs/execution_records_modes.yaml new file mode 100644 index 00000000..0c5d5b1d --- /dev/null +++ b/tests/unit/cpex/fixtures/configs/execution_records_modes.yaml @@ -0,0 +1,116 @@ +plugins: + # sequential deny-list: tests concurrent allow/deny and transform via separate plugins + - name: "ConcurrentDenyPlugin" + kind: "plugins.deny_filter.DenyListPlugin" + description: "Deny-list in concurrent mode — blocks on deny-listed words." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "concurrent" + on_error: "fail" + priority: 100 + conditions: + - prompts: ["test_prompt"] + config: + words: + - innovative + - groundbreaking + - revolutionary + + - name: "ConcurrentAllowPlugin" + kind: "plugins.passthrough.PassThroughPlugin" + description: "Passthrough in concurrent mode — always allows." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "concurrent" + on_error: "fail" + priority: 200 + conditions: + - prompts: ["test_prompt_allow"] + + - name: "TransformPlugin" + kind: "plugins.search_replace.SearchReplacePlugin" + description: "Search-replace in transform mode — modifies payload." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "transform" + on_error: "fail" + priority: 300 + conditions: + - prompts: ["test_prompt_transform"] + config: + words: + - search: crap + replace: crud + + - name: "AuditPlugin" + kind: "plugins.passthrough.PassThroughPlugin" + description: "Passthrough in audit mode — observe only." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "audit" + on_error: "fail" + priority: 400 + conditions: + - prompts: ["test_prompt_audit"] + + - name: "FafPlugin" + kind: "plugins.passthrough.PassThroughPlugin" + description: "Passthrough in fire_and_forget mode." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "fire_and_forget" + on_error: "fail" + priority: 500 + conditions: + - prompts: ["test_prompt_faf"] + + - name: "TimeoutPlugin" + kind: "plugins.timeout_plugin.TimeoutPlugin" + description: "Always times out — tests on_error=ignore timeout recording." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "sequential" + on_error: "ignore" + priority: 600 + conditions: + - prompts: ["test_prompt_timeout"] + + - name: "ErrorPlugin" + kind: "plugins.error.ErrorPlugin" + description: "Error plugin — sequential, on_error=fail. Always raises." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "sequential" + on_error: "fail" + priority: 700 + conditions: + - prompts: ["test_prompt_error"] + + - name: "ConcurrentTimeoutPlugin" + kind: "plugins.timeout_plugin.TimeoutPlugin" + description: "Always times out — tests concurrent on_error=ignore timeout recording." + version: "0.1" + author: "ContextForge Team" + hooks: ["prompt_pre_fetch"] + mode: "concurrent" + on_error: "ignore" + priority: 800 + conditions: + - prompts: ["test_prompt_concurrent_timeout"] + +plugin_dirs: + - "tests/unit/cpex/fixtures" + +plugin_settings: + parallel_execution_within_band: true + plugin_timeout: 30 + fail_on_plugin_error: false + enable_plugin_api: true + plugin_health_check_interval: 60 diff --git a/tests/unit/cpex/fixtures/plugins/timeout_plugin.py b/tests/unit/cpex/fixtures/plugins/timeout_plugin.py new file mode 100644 index 00000000..86aee21a --- /dev/null +++ b/tests/unit/cpex/fixtures/plugins/timeout_plugin.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/unit/cpex/fixtures/plugins/timeout_plugin.py +Copyright 2025 +SPDX-License-Identifier: Apache-2.0 + +Timeout plugin — sleeps indefinitely to trigger asyncio.wait_for timeout. +Used to test on_error=ignore/disable timeout recording behaviour. +""" + +# Standard +import asyncio + +# First-Party +from cpex.framework import ( + Plugin, + PluginContext, + PromptPrehookPayload, + PromptPrehookResult, +) + + +class TimeoutPlugin(Plugin): + """A plugin that always times out (sleeps indefinitely).""" + + async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult: + """Sleep indefinitely — always triggers the executor timeout.""" + await asyncio.sleep(3600) + return PromptPrehookResult(continue_processing=True) # never reached diff --git a/tests/unit/cpex/framework/test_execution_records.py b/tests/unit/cpex/framework/test_execution_records.py new file mode 100644 index 00000000..5a475089 --- /dev/null +++ b/tests/unit/cpex/framework/test_execution_records.py @@ -0,0 +1,489 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/unit/cpex/framework/test_execution_records.py +Copyright 2025 +SPDX-License-Identifier: Apache-2.0 + +Unit tests for ControlExecutionRecord generation on PluginResult.executions (issue #130). + +These tests verify: +- executions is always present (empty when no plugins ran) +- sequential deny path produces a record with correct fields +- sequential allow path produces a record with correct fields (matched=False on clean allow) +- transform plugin produces a record with payload_modified=True +- audit plugin produces a record with applied=False regardless of result +- concurrent deny path produces a record with correct fields +- fire-and-forget plugins produce spawn-time records (mode=fire_and_forget) +- plugin identity fields come from trusted config, not plugin-returned metadata +- string truncation helpers work correctly +- on_error=ignore/disable timeout is recorded as TIMEOUT, not COMPLETED +""" + +# Standard +import asyncio + +# Third-Party +import pytest + +# First-Party +from cpex.framework import ( + ControlExecutionRecord, + ControlExecutionStatus, + GlobalContext, + PluginManager, + PluginMode, + PluginResult, + PromptHookType, + PromptPrehookPayload, +) +from cpex.framework.models import ( + _MAX_CONFIG_KEYS, + _MAX_STRING_LEN, + _collect_config_keys, + _truncate, + _truncate_opt, +) + + +# --------------------------------------------------------------------------- +# Helper / truncation unit tests +# --------------------------------------------------------------------------- + + +def test_truncate_short_string_unchanged(): + assert _truncate("hello") == "hello" + + +def test_truncate_long_string_is_bounded(): + long = "a" * (_MAX_STRING_LEN + 20) + result = _truncate(long) + # Result (including ellipsis) must not exceed MAX_STRING_LEN bytes + assert len(result.encode("utf-8")) <= _MAX_STRING_LEN + assert result.endswith("…") + + +def test_truncate_unicode_does_not_produce_invalid_utf8(): + # Each emoji is 4 bytes — truncation must not split a code point + s = "🎉" * 100 + result = _truncate(s) + # Must be valid string (no UnicodeDecodeError) + result.encode("utf-8") + assert len(result) > 0 + + +def test_truncate_opt_none_passthrough(): + assert _truncate_opt(None) is None + + +def test_truncate_opt_non_none(): + assert _truncate_opt("hi") == "hi" + + +def test_collect_config_keys_extracts_keys_only(): + config = {"policy_file": "apl/demo/hr.yaml", "timeout": 30} + keys = _collect_config_keys(config) + assert set(keys) == {"policy_file", "timeout"} + + +def test_collect_config_keys_bounded(): + config = {f"key_{i}": i for i in range(_MAX_CONFIG_KEYS + 10)} + keys = _collect_config_keys(config) + assert len(keys) == _MAX_CONFIG_KEYS + + +def test_collect_config_keys_non_dict_returns_empty(): + assert _collect_config_keys(None) == [] + assert _collect_config_keys("not a dict") == [] # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# ControlExecutionRecord model tests +# --------------------------------------------------------------------------- + + +def test_execution_record_default_values(): + rec = ControlExecutionRecord( + plugin_id="abc", + plugin_name="test-plugin", + plugin_kind="builtin", + hook_name="tool_pre_invoke", + mode=PluginMode.SEQUENTIAL, + status=ControlExecutionStatus.COMPLETED, + effective_allow=True, + ) + assert rec.requested_allow is None + assert rec.matched is None + assert rec.applied is False + assert rec.payload_modified is False + assert rec.extensions_modified is False + assert rec.duration_ns == 0 + assert rec.reason is None + assert rec.error_code is None + assert rec.config_keys == [] + + +def test_execution_record_serialises_to_dict(): + rec = ControlExecutionRecord( + plugin_id="abc", + plugin_name="pii-guard", + plugin_kind="builtin", + hook_name="tool_pre_invoke", + mode=PluginMode.SEQUENTIAL, + status=ControlExecutionStatus.COMPLETED, + effective_allow=False, + matched=True, + applied=True, + error_code="pii_access_denied", + reason="PII clearance required", + ) + d = rec.model_dump() + assert d["plugin_name"] == "pii-guard" + assert d["effective_allow"] is False + assert d["matched"] is True + assert d["error_code"] == "pii_access_denied" + + +# --------------------------------------------------------------------------- +# Integration: executions on PluginResult (via PluginManager) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_executions_empty_when_no_plugins_registered(): + """PluginResult.executions is always present and empty when no hooks are registered.""" + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_no_plugin.yaml") + await manager.initialize() + prompt = PromptPrehookPayload(prompt_id="p1", args={"user": "hello"}) + context = GlobalContext(request_id="req1") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + assert hasattr(result, "executions") + assert result.executions == [] + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_sequential_deny_record(): + """A sequential plugin that denies produces a record with expected fields.""" + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_single_filter_plugin.yaml") + await manager.initialize() + + # "innovative" is in the deny list + prompt = PromptPrehookPayload(prompt_id="test_prompt", args={"user": "innovative"}) + context = GlobalContext(request_id="req2", server_id="s1") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert not result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert isinstance(rec, ControlExecutionRecord) + assert rec.plugin_name == "DenyListPlugin" + assert rec.hook_name == PromptHookType.PROMPT_PRE_FETCH + assert rec.mode == PluginMode.SEQUENTIAL + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is False + assert rec.requested_allow is False + assert rec.matched is True + assert rec.applied is True + assert rec.payload_modified is False + assert rec.duration_ns > 0 # real execution — must have elapsed some time + assert rec.error_code is not None # deny path always has an error_code + # Identity fields come from trusted config, not from plugin + assert rec.plugin_id # non-empty UUID hex + assert rec.plugin_kind # non-empty kind string + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_sequential_allow_record(): + """A sequential plugin that allows produces a record with correct allow fields.""" + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_single_filter_plugin.yaml") + await manager.initialize() + + # "hello" is NOT in the deny list — should pass + prompt = PromptPrehookPayload(prompt_id="test_prompt", args={"user": "hello"}) + context = GlobalContext(request_id="req3", server_id="s1") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "DenyListPlugin" + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is True + assert rec.requested_allow is True + # DenyListPlugin unconditionally returns modified_payload=payload on allow (line 75 of + # deny_filter.py), so the executor detects a payload identity change → payload_modified=True + # → matched=True even on a clean allow. The meaningful invariant is no denial occurred. + assert rec.matched is not None, ( + f"clean allow must have a deterministic matched value, not None; got {rec.matched!r}" + ) + assert isinstance(rec.applied, bool) + assert rec.duration_ns > 0 + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_records_are_per_invocation_not_shared(): + """Two separate invoke_hook calls produce independent execution record lists.""" + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_single_filter_plugin.yaml") + await manager.initialize() + + prompt = PromptPrehookPayload(prompt_id="test_prompt", args={"user": "hello"}) + context = GlobalContext(request_id="req4", server_id="s1") + + result1, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + result2, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + # Each result has its own list — mutating one must not affect the other + assert result1.executions is not result2.executions + result1.executions.clear() + assert len(result2.executions) >= 1 + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_plugin_id_is_non_empty_hex(): + """plugin_id is a non-empty hex string (UUID without hyphens).""" + manager = PluginManager("./tests/unit/cpex/fixtures/configs/valid_single_filter_plugin.yaml") + await manager.initialize() + + prompt = PromptPrehookPayload(prompt_id="test_prompt", args={"user": "hello"}) + context = GlobalContext(request_id="req5", server_id="s1") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + rec = result.executions[0] + assert len(rec.plugin_id) == 32 # UUID hex = 32 chars + int(rec.plugin_id, 16) # must be valid hex + + await manager.shutdown() + PluginManager.reset() + + +_MODES_CONFIG = "./tests/unit/cpex/fixtures/configs/execution_records_modes.yaml" + + +@pytest.mark.asyncio +async def test_executions_fail_closed_on_error_fail(): + """Critical: a sequential plugin that raises with on_error=fail must propagate the error + (fail-closed), not swallow it and return continue_processing=True (fail-open). + The record must be written before the re-raise with status=ERROR and effective_allow=False.""" + from cpex.framework.errors import PluginError + + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt_error routes to ErrorPlugin (sequential, on_error=fail, always raises) + prompt = PromptPrehookPayload(prompt_id="test_prompt_error", args={"user": "hello"}) + context = GlobalContext(request_id="req-fail-closed") + + with pytest.raises(PluginError): + await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_concurrent_clean_allow_matched_false(): + """Concurrent clean-allow record must have matched=False (consistent with serial path).""" + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt_allow routes only to ConcurrentAllowPlugin (passthrough, concurrent) + prompt = PromptPrehookPayload(prompt_id="test_prompt_allow", args={"user": "hello"}) + context = GlobalContext(request_id="req-concurrent-allow") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + rec = result.executions[0] + assert rec.plugin_name == "ConcurrentAllowPlugin" + assert rec.mode == PluginMode.CONCURRENT + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is True + assert rec.matched is False, ( + f"concurrent clean allow: matched must be False, got {rec.matched!r}" + ) + assert rec.applied is False + assert rec.payload_modified is False + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_concurrent_deny_record(): + """A concurrent plugin that denies produces a record with correct fields.""" + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt routes only to ConcurrentDenyPlugin; "innovative" triggers denial + prompt = PromptPrehookPayload(prompt_id="test_prompt", args={"user": "innovative"}) + context = GlobalContext(request_id="req-concurrent-deny") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert not result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "ConcurrentDenyPlugin" + assert rec.mode == PluginMode.CONCURRENT + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is False + assert rec.requested_allow is False + assert rec.matched is True + assert rec.applied is True + assert rec.payload_modified is False # concurrent plugins cannot modify payload + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_transform_record(): + """A transform plugin produces a record: allow=True, payload_modified=True, applied=True.""" + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt_transform routes only to TransformPlugin; "crap" triggers substitution + prompt = PromptPrehookPayload(prompt_id="test_prompt_transform", args={"user": "crap input"}) + context = GlobalContext(request_id="req-transform") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "TransformPlugin" + assert rec.mode == PluginMode.TRANSFORM + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is True + assert rec.payload_modified is True + assert rec.applied is True # applied because payload was modified + assert rec.duration_ns > 0 + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_audit_record(): + """An audit plugin produces a record: allow=True, applied=False, payload_modified=False.""" + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt_audit routes only to AuditPlugin (passthrough, audit) + prompt = PromptPrehookPayload(prompt_id="test_prompt_audit", args={"user": "hello"}) + context = GlobalContext(request_id="req-audit") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "AuditPlugin" + assert rec.mode == PluginMode.AUDIT + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is True + assert rec.payload_modified is False # audit plugins cannot modify payloads + assert rec.applied is False # no denial, no mutation → not applied + assert rec.duration_ns > 0 + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_faf_record(): + """A fire-and-forget plugin produces a spawn-time record with mode=fire_and_forget.""" + manager = PluginManager(_MODES_CONFIG) + await manager.initialize() + + # test_prompt_faf routes only to FafPlugin (passthrough, fire_and_forget) + prompt = PromptPrehookPayload(prompt_id="test_prompt_faf", args={"user": "hello"}) + context = GlobalContext(request_id="req-faf") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "FafPlugin" + assert rec.mode == PluginMode.FIRE_AND_FORGET + # Spawn-time record: status is COMPLETED as an optimistic placeholder. + # Identify FAF records by mode, not by status. + assert rec.status == ControlExecutionStatus.COMPLETED + assert rec.effective_allow is True + assert rec.duration_ns == 0 # not yet executed at pipeline return time + assert rec.requested_allow is None # outcome unknowable at spawn + + # Wait for the background task to confirm it ran without error + errors = await result.wait_for_background_tasks() + assert errors == [] + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_timeout_ignore_recorded_as_timeout(): + """A plugin that times out with on_error=ignore must produce a TIMEOUT record, not COMPLETED.""" + # timeout=1 so the 3600s sleep in TimeoutPlugin fires within the test + manager = PluginManager(_MODES_CONFIG, timeout=1) + await manager.initialize() + + # test_prompt_timeout routes only to TimeoutPlugin (sequential, on_error=ignore) + prompt = PromptPrehookPayload(prompt_id="test_prompt_timeout", args={"user": "hello"}) + context = GlobalContext(request_id="req-timeout-ignore") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "TimeoutPlugin" + assert rec.status == ControlExecutionStatus.TIMEOUT, ( + f"expected TIMEOUT, got {rec.status!r} — timeout under on_error=ignore must not look like a clean allow" + ) + assert rec.effective_allow is True # pipeline continued + assert rec.error_code == "plugin_timeout" + + await manager.shutdown() + PluginManager.reset() + + +@pytest.mark.asyncio +async def test_executions_concurrent_timeout_ignore_does_not_escape(): + """A concurrent plugin that times out with on_error=ignore must NOT let PluginTimeoutError + escape invoke_hook: the pipeline continues and the record is TIMEOUT (mirrors serial phase).""" + # timeout=1 so the 3600s sleep in TimeoutPlugin fires within the test + manager = PluginManager(_MODES_CONFIG, timeout=1) + await manager.initialize() + + # test_prompt_concurrent_timeout routes only to ConcurrentTimeoutPlugin (concurrent, on_error=ignore) + prompt = PromptPrehookPayload(prompt_id="test_prompt_concurrent_timeout", args={"user": "hello"}) + context = GlobalContext(request_id="req-concurrent-timeout") + result, _ = await manager.invoke_hook(PromptHookType.PROMPT_PRE_FETCH, prompt, global_context=context) + + assert result.continue_processing # must not raise / must fail-open-continue for ignore + assert len(result.executions) >= 1 + + rec = result.executions[0] + assert rec.plugin_name == "ConcurrentTimeoutPlugin" + assert rec.mode == PluginMode.CONCURRENT + assert rec.status == ControlExecutionStatus.TIMEOUT, ( + f"expected TIMEOUT, got {rec.status!r} — concurrent ignore-timeout must not look like a clean allow" + ) + assert rec.effective_allow is True + assert rec.error_code == "plugin_timeout" + + await manager.shutdown() + PluginManager.reset()