feat: split plugin execution mode from error handling; add observe mode

[araujof]

Summary

This PR refactors PluginMode to cleanly separate two concerns that were previously conflated in the now-removed ENFORCE_IGNORE_ERROR mode:

• PluginMode — defines the plugin's scheduling authority: how and when it runs relative to the pipeline
• OnError — defines error handling behaviour independently, per plugin

The refactor also clarifies scheduling semantics by replacing the ambiguous ENFORCE name with explicit mode names:

• ENFORCE is renamed to SEQUENTIAL (parallel, fail-fast — the scheduling behaviour it always had)
• CONCURRENT mode is added (parallel, fail-fast — the scheduling behaviour it always had)
• PERMISSIVEis renamed to AUDIT (sequential, chained, cannot halt)
• FIRE_AND_FORGET mode is added (fire-and-forget background)

The PluginSettings Pydantic model is also removed from models.py; its live runtime fields were migrated to settings.py alongside the existing env-var–backed settings.

Closes #2

Changes

New OnError enum (models.py)

Replaces the blunt enforce_ignore_error escape hatch with a first-class, per-plugin error policy:

Value	Behaviour
fail	Pipeline halts; error propagates (default)
ignore	Error logged; pipeline continues
disable	Error logged; plugin auto-disabled for the lifetime of the executor; pipeline continues

# Example: a plugin that degrades gracefully on failure
- name: "AuditPlugin"
  mode: "fire_and_forget"
  on_error: "disable"

Updated PluginMode enum (models.py)

PluginMode now uses StrEnum and contains five values. Execution order: SEQUENTIAL → AUDIT → CONCURRENT → FIRE_AND_FORGET.

Mode	Order	Concurrency	Payload	Can Halt	Global State Merged
sequential	Priority	Sequential	Chained (receives prev output)	Yes	Yes
permissive	Priority	Sequential	Chained (receives prev output)	No (violations logged)	No
concurrent	Priority band	Parallel	Snapshot (modifications merged)	Yes (fail-fast)	Yes
fire_and_forget	Background	Bounded by pool	Isolated snapshot	Never	No
disabled	Skipped	—	—	—	—

ENFORCE_IGNORE_ERROR is removed and ENFORCE is replaced by SEQUENTIAL for backwards-compat YAML configs (see Migration Guide). Existing YAML configs using mode: enforce or mode: enforce_ignore_error are automatically migrated via a model_validator — no YAML changes required for existing deployments.

The default mode is SEQUENTIAL.

New PluginConfig.on_error field (models.py)

on_error: OnError = OnError.FAIL

A model_validator (_migrate_legacy_modes) converts legacy configs at parse time:

mode: enforce_ignore_error  →  mode: sequential, on_error: ignore
mode: enforce               →  mode: sequential

Removed PluginSettings model; fields migrated to settings.py

The PluginSettings Pydantic model (previously embedded in Config and populated from YAML) is removed. Its unused fields (parallel_execution_within_band, plugin_timeout, enable_plugin_api, plugin_health_check_interval, include_user_info) are dropped. The two fields still in active use are migrated to the env-var–backed settings.py:

Field	Type	Default	Env var
fail_on_plugin_error	bool	False	PLUGINS_FAIL_ON_PLUGIN_ERROR
execution_pool	int | None	None (unlimited)	PLUGINS_EXECUTION_POOL

Both fields are added to PluginsSettings (full settings) and PluginsStartupSettings (lightweight early-read class used at executor init), with corresponding lazy properties on LazySettingsWrapper. The executor reads them via settings.execution_pool and settings.fail_on_plugin_error.

Config.plugin_settings is removed and Config gains model_config = ConfigDict(extra="ignore") so that YAML files that still carry a plugin_settings: section parse without error.

execution_pool — FIRE_AND_FORGET and CONCURRENT concurrency caps

FIRE_AND_FORGET- and CONCURRENT-mode tasks are dispatched concurrently. The execution_pool setting bounds concurrent tasks through two independent asyncio.Semaphore objects (one per mode):

# Cap tasks to 5 workers per FIRE_AND_FORGET/CONCURRENT mode (max of 10 concurrent workers combined)
PLUGINS_EXECUTION_POOL=5

Refactored PluginExecutor (manager.py)

The previously monolithic execution loop is replaced with a four-phase scheduling model, executed in strict order:

1. SEQUENTIAL phase — plugins run one-at-a-time in priority order. Each plugin receives current_payload (the output of the previous plugin). Any plugin returning continue_processing=False halts the pipeline immediately; FIRE_AND_FORGET tasks are fired before returning.

2. AUDIT phase — plugins run sequentially, each receiving the output of the previous plugin (chained transformation semantics). A continue_processing=False result is swallowed — the violation is logged but the pipeline continues.

3. CONCURRENT phase — all CONCURRENT plugins receive a snapshot of current_payload at phase start and execute in parallel via asyncio.as_completed, bounded by Semaphore(execution_pool). Any plugin returning continue_processing=False cancels remaining tasks, fires FIRE_AND_FORGET tasks, and returns early.

4. FIRE_AND_FORGET phase — each plugin gets an isolated payload snapshot and is dispatched as a fire-and-forget asyncio.Task, bounded by Semaphore(execution_pool). The pipeline does not await these tasks. Errors are swallowed (with logging); on_error: disable adds the plugin to a _runtime_disabled set. FIRE_AND_FORGET always fires last — using the final payload state after all foreground phases complete.

New and updated private methods:

• _group_by_mode() — now returns a 4-tuple (sequential_refs, permissive_refs, concurrent_refs, fire_and_forget_refs), buckets and priority-sorts refs, respecting DISABLED and _runtime_disabled
• _fire_fire_and_forget_tasks() — extracted helper that schedules FIRE_AND_FORGET plugins; called from both early-exit paths (SEQUENTIAL/CONCURRENT halt) and the normal completion path, ensuring FIRE_AND_FORGET always fires exactly once
• _run_fire_and_forget_task() — wraps individual background execution with error handling and optional auto-disable
• _apply_payload_modification() — extracted helper for payload-merge policy logic

Error handling in execute_plugin() now dispatches on hook_ref.plugin_ref.on_error rather than mode. The halt check in execute_plugin() now covers both CONCURRENT and SEQUENTIAL modes (both can halt; AUDIT cannot).

Global state isolation

Only CONCURRENT and SEQUENTIAL plugins merge their local global_context mutations back to the shared context. AUDIT and FIRE_AND_FORGET plugins operate on isolated copy-on-write snapshots. CONCURRENT and SEQUENTIAL plugins are authoritative checks whose side-effects are intentional; AUDIT plugins are transformers that chain payload changes only.

Exports (__init__.py)

OnError is exported from cpex.framework alongside the existing PluginMode.

Breaking Changes

• PluginMode.ENFORCE_IGNORE_ERROR is removed from the enum. Python code that references it directly will raise AttributeError. YAML configs are migrated automatically.
• PluginMode.ENFORCE is removed. Python code that references PluginMode.ENFORCE must be updated to PluginMode.CONCURRENT (parallel, fail-fast) or PluginMode.SEQUENTIAL (sequential, can halt) depending on intent.
• PluginMode.PERMISSIVE is renamed PluginMode.AUDIT. Python code that references PluginMode.PERMISSIVE must be updated to PluginMode.AUDIT.
• Execution order has changed: SEQUENTIAL → AUDIT → CONCURRENT → FIRE_AND_FORGET.
• PluginSettings is removed from models.py. Code that imports or instantiates it must be updated.
• Config.plugin_settings is removed. Any code that reads fields from it (e.g. config.plugin_settings.plugin_timeout) must be updated to read from settings.* instead.
• CONCURRENT plugins execute in parallel and receive payload snapshots; they cannot chain payload modifications. Plugins that relied on sequential ordering or chained transforms within the old ENFORCE band must be changed to mode: sequential or mode: permissive.

Migration Guide

Plugin mode — Python code

# Before (ENFORCE → parallel, fail-fast)
plugin_config.mode = PluginMode.ENFORCE

# After — use CONCURRENT for parallel/fail-fast (same scheduling as old ENFORCE)
plugin_config.mode = PluginMode.CONCURRENT

# Or SEQUENTIAL if you need sequential, chained, can-halt semantics
plugin_config.mode = PluginMode.SEQUENTIAL

# Before
plugin_config.mode = PluginMode.ENFORCE_IGNORE_ERROR

# After
plugin_config.mode = PluginMode.SEQUENTIAL
plugin_config.on_error = OnError.IGNORE

# Before
plugin_config.mode = PluginMode.PERMISSIVE

# After
plugin_config.mode = PluginMode.AUDIT

YAML plugin config

# Before (parallel, fail-fast)
mode: enforce

# After — explicit canonical form
mode: sequential

# Legacy value is also accepted — auto-migrated to SEQUENTIAL at parse time
# (use `mode: concurrent` if you want parallel semantics)
mode: enforce

# Before
mode: enforce_ignore_error

# After (explicit)
mode: sequential
on_error: ignore

# Legacy value is also accepted — auto-migrated at parse time
mode: enforce_ignore_error

# Before 
mode: permissive

# After — explicit canonical form
mode: audit

# Legacy value is also accepted — auto-migrated to AUDIT at parse time
mode: permissive

Runtime settings (previously in YAML plugin_settings:)

# Before: plugins/config.yaml
# plugin_settings:
#   fail_on_plugin_error: true
#   execution_pool: 10

# After: environment variable or .env file
PLUGINS_FAIL_ON_PLUGIN_ERROR=true
PLUGINS_EXECUTION_POOL=10

[terylt]

LGTM