feat(validation): add validation pipeline and results model

benchmarks/validation_pipeline_bench.py

@@ -0,0 +1,188 @@
+"""ASV benchmarks for the validation pipeline.
+
+Measures the performance of:
+- Pipeline creation and command sorting
+- Pipeline execution with passing validations
+- Summary building and aggregation
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    ValidationResult,
+    ValidationSummary,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Trivial executor that always passes."""
+    return {"passed": True, "message": f"{validation_name} ok"}
+
+
+def _build_commands(count: int) -> list[ValidationCommand]:
+    """Build a list of N validation commands across multiple resources."""
+    commands: list[ValidationCommand] = []
+    for i in range(count):
+        resource_name = f"resource-{i % 5}"
+        mode = ValidationMode.REQUIRED if i % 3 else ValidationMode.INFORMATIONAL
+        commands.append(
+            ValidationCommand(
+                validation_name=f"check-{i:04d}",
+                resource_id=f"R{i:05d}",
+                resource_name=resource_name,
+                mode=mode,
+                arguments={"index": i},
+                timeout_seconds=30.0,
+            )
+        )
+    return commands
+
+
+# ---------------------------------------------------------------------------
+# Benchmark suites
+# ---------------------------------------------------------------------------
+
+
+class ValidationPipelineCreationSuite:
+    """Benchmarks for pipeline instantiation and command sorting."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.commands_10 = _build_commands(10)
+        self.commands_100 = _build_commands(100)
+
+    def time_create_pipeline_10_commands(self) -> None:
+        """Benchmark creating a pipeline with 10 commands."""
+        ValidationPipeline(
+            commands=self.commands_10,
+            executor=_mock_executor,
+            max_workers=4,
+        )
+
+    def time_create_pipeline_100_commands(self) -> None:
+        """Benchmark creating a pipeline with 100 commands."""
+        ValidationPipeline(
+            commands=self.commands_100,
+            executor=_mock_executor,
+            max_workers=4,
+        )
+
+    def time_sort_100_commands(self) -> None:
+        """Benchmark sorting 100 commands."""
+        ValidationPipeline._sort_commands(self.commands_100)
+
+
+class ValidationPipelineExecutionSuite:
+    """Benchmarks for pipeline execution with passing validations."""
+
+    timeout = 120
+
+    def setup(self) -> None:
+        self.commands_10 = _build_commands(10)
+        self.commands_50 = _build_commands(50)
+
+    def time_run_10_validations(self) -> None:
+        """Benchmark running 10 validations."""
+        pipeline = ValidationPipeline(
+            commands=self.commands_10,
+            executor=_mock_executor,
+            max_workers=4,
+        )
+        pipeline.run()
+
+    def time_run_50_validations_sequential(self) -> None:
+        """Benchmark running 50 validations with max_workers=1."""
+        pipeline = ValidationPipeline(
+            commands=self.commands_50,
+            executor=_mock_executor,
+            max_workers=1,
+        )
+        pipeline.run()
+
+    def time_run_10_validations_for_plan(self) -> None:
+        """Benchmark run_for_plan with 10 validations."""
+        pipeline = ValidationPipeline(
+            commands=self.commands_10,
+            executor=_mock_executor,
+            max_workers=4,
+        )
+        metadata: dict[str, Any] = {}
+        pipeline.run_for_plan(plan_metadata=metadata)
+
+
+class ValidationPipelineSummarySuite:
+    """Benchmarks for summary construction and grouping."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.results: list[ValidationResult] = []
+        for i in range(50):
+            self.results.append(
+                ValidationResult(
+                    validation_name=f"check-{i:04d}",
+                    resource_id=f"R{i:05d}",
+                    resource_name=f"resource-{i % 5}",
+                    mode=(
+                        ValidationMode.REQUIRED
+                        if i % 3
+                        else ValidationMode.INFORMATIONAL
+                    ),
+                    passed=i % 2 == 0,
+                    message=f"Result for check-{i:04d}",
+                    data=None,
+                    duration_ms=float(i),
+                    error=None,
+                    timed_out=False,
+                )
+            )
+
+    def time_group_by_resource_50_results(self) -> None:
+        """Benchmark grouping 50 results by resource."""
+        ValidationPipeline.group_by_resource(self.results)
+
+    def time_build_summary_model(self) -> None:
+        """Benchmark building a ValidationSummary from 50 results."""
+        req_p = sum(
+            1 for r in self.results if r.mode == ValidationMode.REQUIRED and r.passed
+        )
+        req_f = sum(
+            1
+            for r in self.results
+            if r.mode == ValidationMode.REQUIRED and not r.passed
+        )
+        info_p = sum(
+            1
+            for r in self.results
+            if r.mode == ValidationMode.INFORMATIONAL and r.passed
+        )
+        info_f = sum(
+            1
+            for r in self.results
+            if r.mode == ValidationMode.INFORMATIONAL and not r.passed
+        )
+        ValidationSummary(
+            total=len(self.results),
+            required_passed=req_p,
+            required_failed=req_f,
+            informational_passed=info_p,
+            informational_failed=info_f,
+            results=self.results,
+        )

docs/reference/validation_pipeline.md

@@ -0,0 +1,148 @@
+# Validation Pipeline
+
+The **Validation Pipeline** orchestrates deterministic execution of validations
+against plan resources, enforcing required vs informational semantics and
+producing aggregated summaries for the plan lifecycle.
+
+## Overview
+
+When a plan reaches the end of its Execute phase, the validation pipeline
+runs all attached validations.  Each validation is modelled as a
+`ValidationCommand` that specifies the target resource, execution mode,
+arguments, and timeout.  Results are collected into `ValidationResult`
+objects and aggregated into a `ValidationSummary`.
+
+```
+ValidationCommand(s) ──► ValidationPipeline.run() ──► ValidationSummary
+```
+
+## Ordering Rules
+
+Validations are sorted deterministically before execution:
+
+1. **Resource name** (alphabetical)
+2. **Mode** (`informational` before `required` by enum value)
+3. **Validation name** (alphabetical)
+
+This ensures repeatable execution order regardless of the order in which
+validations are registered or attached.
+
+## Required vs Informational Behaviour
+
+| Mode            | On Pass | On Fail                                      |
+|-----------------|---------|----------------------------------------------|
+| `required`      | OK      | Blocks: `all_required_passed` becomes `False` |
+| `informational` | OK      | Logged but does **not** block                 |
+
+- A `ValidationSummary` with `all_required_passed == False` signals that
+  the plan should not proceed to the Apply phase.
+- Informational failures are logged at INFO level for review but do not
+  affect the `all_required_passed` property.
+
+## Timeout Handling
+
+Each `ValidationCommand` carries a `timeout_seconds` field (default 30 s).
+Timeouts are enforced using daemon threads with `thread.join(timeout=N)`.
+If a validation exceeds its timeout:
+
+- The daemon thread is abandoned (not blocked on).
+- The result is marked `passed=False` and `timed_out=True`.
+- The error field contains `TimeoutError: exceeded <N>s`.
+- The duration reflects the wall-clock time up to the timeout.
+
+## Concurrency and Capture
+
+The pipeline uses a `concurrent.futures.ThreadPoolExecutor` with a
+configurable `max_workers` (default 4).  Validations are submitted
+concurrently but results are re-sorted to maintain deterministic ordering.
+
+Before execution, `sys.stdout` and `sys.stderr` are replaced with
+thread-local stream wrappers (`_ThreadLocalStream`).  Each worker thread
+captures its own output independently, avoiding cross-thread pollution.
+The original streams are restored in a `finally` block after all
+validations complete.
+
+Setting `max_workers=1` forces purely sequential execution, which is
+useful for debugging or when validations have interdependencies.
+
+## Output Normalisation
+
+Validation executors may return arbitrary output.  The pipeline normalises
+all results to a consistent schema:
+
+| Executor Returns       | Normalised To                               |
+|------------------------|---------------------------------------------|
+| `{"passed": True, ...}`| Used directly                               |
+| Non-dict value         | `passed=False`, message describes the error |
+| Missing `passed` key   | Defaults to `False`                         |
+| Missing `message` key  | Defaults to `"No message provided"`         |
+| Non-dict `data` value  | Wrapped in `{"raw": <value>}`               |
+
+When an executor raises an exception, the result is recorded as
+`passed=False` with a message describing the exception type and details
+(e.g., `Validation raised RuntimeError: ...`).  The exception
+information is preserved in the `error` field.
+
+Captured stdout/stderr during validation execution is stored in the
+result's `data` dict under `_captured_stdout` and `_captured_stderr` keys.
+
+## Read-Only Resource Guard
+
+If a validation targets a resource marked as read-only (via the
+`read_only_resources` parameter), the pipeline **skips** the validation
+and returns a passing result with a descriptive message.  A warning is
+logged when this occurs.
+
+## Integration with Plan Lifecycle
+
+The `run_for_plan()` method extends `run()` by persisting the validation
+summary into the plan's metadata dict:
+
+```python
+pipeline = ValidationPipeline(commands, executor)
+summary = pipeline.run_for_plan(plan_metadata=plan.metadata)
+# plan.metadata["validation_summary"] now contains the full summary
+```
+
+This allows downstream phases (Apply, CLI display) to inspect validation
+results without re-running the pipeline.
+
+## Models Reference
+
+### ValidationCommand
+
+| Field              | Type             | Default | Description                |
+|--------------------|------------------|---------|----------------------------|
+| `validation_name`  | `str`            | —       | Validation identifier      |
+| `resource_id`      | `str`            | —       | Target resource ULID       |
+| `resource_name`    | `str`            | —       | Human-readable name        |
+| `mode`             | `ValidationMode` | —       | `required`/`informational` |
+| `arguments`        | `dict`           | `{}`    | Executor arguments         |
+| `timeout_seconds`  | `float`          | `30.0`  | Timeout in seconds         |
+
+### ValidationResult
+
+| Field              | Type             | Default | Description                |
+|--------------------|------------------|---------|----------------------------|
+| `validation_name`  | `str`            | —       | Validation identifier      |
+| `resource_id`      | `str`            | —       | Resource ULID              |
+| `resource_name`    | `str`            | —       | Human-readable name        |
+| `mode`             | `ValidationMode` | —       | `required`/`informational` |
+| `passed`           | `bool`           | —       | Pass/fail status           |
+| `message`          | `str`            | —       | Result description         |
+| `data`             | `dict \| None`   | `None`  | Structured data            |
+| `duration_ms`      | `float`          | —       | Execution time (ms)        |
+| `error`            | `str \| None`    | `None`  | Error message              |
+| `timed_out`        | `bool`           | `False` | Timeout flag               |
+
+### ValidationSummary
+
+| Field                   | Type                    | Description              |
+|-------------------------|-------------------------|--------------------------|
+| `total`                 | `int`                   | Total validations        |
+| `required_passed`       | `int`                   | Required passes          |
+| `required_failed`       | `int`                   | Required failures        |
+| `informational_passed`  | `int`                   | Informational passes     |
+| `informational_failed`  | `int`                   | Informational failures   |
+| `results`               | `list[ValidationResult]`| All results              |
+| `all_required_passed`   | `bool` (property)       | True if no req. failures |

features/steps/skill_registry_steps.py

@@ -179,7 +179,7 @@ def step_register_skills_table(context: Context) -> None:
             description=row["description"],
         )
         svc.add_skill(skill)
-    _commit_pending(context)
+        _commit_pending(context)
 
 
 @given('a valid skill named "{name}" with overrides for "{tool_name}"')

features/steps/tool_registry_service_fallback_coverage_steps.py

@@ -0,0 +1,430 @@
+"""Step definitions for tool_registry_service_fallback_coverage.feature.
+
+Exercises the fallback method-resolution branches in ToolRegistryService:
+- register_tool: falls back from create -> add -> create (AttributeError)
+- update_tool: tool_config parameter paths (dict, Tool model)
+- remove_tool: falls back from delete -> remove -> delete (AttributeError)
+- list_tools: delegates to list_all
+- attach_validation: invalid mode raises ValidationError
+
+All repository dependencies are lightweight mocks — no database needed.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.tool_registry_service import (
+    ToolRegistryService,
+)
+from cleveragents.core.exceptions import ValidationError
+from cleveragents.domain.models.core.tool import Tool, ToolSource
+
+
+# ---------------------------------------------------------------------------
+# Mock repos with specific method presence
+# ---------------------------------------------------------------------------
+
+
+class _AddOnlyToolRepo:
+    """A repo that has add() but no create() method."""
+
+    def __init__(self) -> None:
+        self._added: Any = None
+
+    # Deliberately no create() method
+
+    def add(self, tool: Any) -> Any:
+        self._added = tool
+        return tool
+
+    def get_by_name(self, name: str) -> Any:
+        return None
+
+    def list_all(self, **kwargs: Any) -> list[Any]:
+        return []
+
+    def update(self, tool: Any) -> Any:
+        return tool
+
+    def delete(self, name: str) -> bool:
+        return True
+
+
+class _NoCreateNoAddToolRepo:
+    """A repo that has neither create() nor add() as callable methods.
+
+    Has create as a non-callable (string) to trigger the final fallback.
+    """
+
+    def __init__(self) -> None:
+        # create is set to a non-callable to force the fallback path
+        pass
+
+    def get_by_name(self, name: str) -> Any:
+        return None
+
+    def list_all(self, **kwargs: Any) -> list[Any]:
+        return []
+
+    def update(self, tool: Any) -> Any:
+        return tool
+
+
+class _RemoveOnlyToolRepo:
+    """A repo that has remove() but no delete() method."""
+
+    def __init__(self) -> None:
+        self._removed: str | None = None
+
+    # Deliberately no delete() method
+
+    def remove(self, name: str) -> bool:
+        self._removed = name
+        return True
+
+    def get_by_name(self, name: str) -> Any:
+        return None
+
+    def create(self, tool: Any) -> Any:
+        return tool
+
+    def list_all(self, **kwargs: Any) -> list[Any]:
+        return []
+
+    def update(self, tool: Any) -> Any:
+        return tool
+
+
+class _NoDeleteNoRemoveToolRepo:
+    """A repo that has neither delete() nor remove()."""
+
+    def __init__(self) -> None:
+        pass
+
+    def get_by_name(self, name: str) -> Any:
+        return None
+
+    def create(self, tool: Any) -> Any:
+        return tool
+
+    def list_all(self, **kwargs: Any) -> list[Any]:
+        return []
+
+    def update(self, tool: Any) -> Any:
+        return tool
+
+
+class _StandardMockToolRepo:
+    """Standard mock with all methods present."""
+
+    def __init__(self) -> None:
+        self._last_updated: Any = None
+        self._list_all_return: list[Any] = [
+            {"name": "local/tool-1"},
+            {"name": "local/tool-2"},
+        ]
+        self._get_by_name_return: Any = {"name": "local/some-check"}
+
+    def create(self, tool: Any) -> Any:
+        return tool
+
+    def get_by_name(self, name: str) -> Any:
+        return self._get_by_name_return
+
+    def list_all(
+        self,
+        namespace: str | None = None,
+        tool_type: str | None = None,
+        source: str | None = None,
+    ) -> list[Any]:
+        return self._list_all_return
+
+    def update(self, tool: Any) -> Any:
+        self._last_updated = tool
+        return tool
+
+    def delete(self, name: str) -> bool:
+        return True
+
+
+class _MinimalAttachmentRepo:
+    """Attachment repo that satisfies the service interface."""
+
+    def attach(self, **kwargs: Any) -> dict[str, str]:
+        return {"attachment_id": "fallback-att-001"}
+
+    def detach(self, attachment_id: str) -> bool:
+        return True
+
+    def list_for_resource(self, **kwargs: Any) -> list[Any]:
+        return []
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a tool registry service with a repo that only has add method")
+def step_given_add_only_repo(context: Context) -> None:
+    context.fb_tool_repo = _AddOnlyToolRepo()
+    context.fb_attachment_repo = _MinimalAttachmentRepo()
+    context.fb_service = ToolRegistryService(
+        tool_repo=context.fb_tool_repo,
+        attachment_repo=context.fb_attachment_repo,
+    )
+    context.fb_error: Exception | None = None
+    context.fb_result: Any = None
+
+
+@given("a tool registry service with a repo that has no create or add methods")
+def step_given_no_create_no_add(context: Context) -> None:
+    context.fb_tool_repo = _NoCreateNoAddToolRepo()
+    context.fb_attachment_repo = _MinimalAttachmentRepo()
+    context.fb_service = ToolRegistryService(
+        tool_repo=context.fb_tool_repo,
+        attachment_repo=context.fb_attachment_repo,
+    )
+    context.fb_error = None
+    context.fb_result = None
+
+
+@given("a tool registry service with a standard mock repo")
+def step_given_standard_repo(context: Context) -> None:
+    context.fb_tool_repo = _StandardMockToolRepo()
+    context.fb_attachment_repo = _MinimalAttachmentRepo()
+    context.fb_service = ToolRegistryService(
+        tool_repo=context.fb_tool_repo,
+        attachment_repo=context.fb_attachment_repo,
+    )
+    context.fb_error = None
+    context.fb_result = None
+
+
+@given("a tool registry service with a repo that only has remove method")
+def step_given_remove_only_repo(context: Context) -> None:
+    context.fb_tool_repo = _RemoveOnlyToolRepo()
+    context.fb_attachment_repo = _MinimalAttachmentRepo()
+    context.fb_service = ToolRegistryService(
+        tool_repo=context.fb_tool_repo,
+        attachment_repo=context.fb_attachment_repo,
+    )
+    context.fb_error = None
+    context.fb_result = None
+
+
+@given("a tool registry service with a repo that has no delete or remove methods")
+def step_given_no_delete_no_remove(context: Context) -> None:
+    context.fb_tool_repo = _NoDeleteNoRemoveToolRepo()
+    context.fb_attachment_repo = _MinimalAttachmentRepo()
+    context.fb_service = ToolRegistryService(
+        tool_repo=context.fb_tool_repo,
+        attachment_repo=context.fb_attachment_repo,
+    )
+    context.fb_error = None
+    context.fb_result = None
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I register a tool via the fallback service")
+def step_when_register_fallback(context: Context) -> None:
+    tool = {"name": "local/fallback-tool", "source": "builtin"}
+    try:
+        context.fb_result = context.fb_service.register_tool(tool)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when("I register a tool via the fallback service and expect error")
+def step_when_register_fallback_error(context: Context) -> None:
+    tool = {"name": "local/fallback-tool", "source": "builtin"}
+    try:
+        context.fb_result = context.fb_service.register_tool(tool)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I update tool "{name}" with dict config via the fallback service')
+def step_when_update_dict_config(context: Context, name: str) -> None:
+    try:
+        context.fb_result = context.fb_service.update_tool(
+            name,
+            tool_config={"description": "Updated description"},
+        )
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I update tool "{name}" with Tool model config via the fallback service')
+def step_when_update_tool_model(context: Context, name: str) -> None:
+    try:
+        tool_model = Tool(
+            name="local/my-tool",
+            description="A tool model",
+            source=ToolSource.BUILTIN,
+            tool_type="tool",
+            timeout=300,
+        )
+        context.fb_result = context.fb_service.update_tool(name, tool_config=tool_model)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when("I update with dict tool and dict config via the fallback service")
+def step_when_update_dict_tool_dict_config(context: Context) -> None:
+    try:
+        tool_dict = {"name": "local/dict-tool", "source": "builtin"}
+        context.fb_result = context.fb_service.update_tool(
+            tool_dict,
+            tool_config={"description": "Updated"},
+        )
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I remove tool "{name}" via the fallback service')
+def step_when_remove_fallback(context: Context, name: str) -> None:
+    try:
+        context.fb_result = context.fb_service.remove_tool(name)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I remove tool "{name}" via the fallback service and expect error')
+def step_when_remove_fallback_error(context: Context, name: str) -> None:
+    try:
+        context.fb_result = context.fb_service.remove_tool(name)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I list tools with namespace "{ns}" and type "{tt}" via the fallback service')
+def step_when_list_tools(context: Context, ns: str, tt: str) -> None:
+    try:
+        context.fb_result = context.fb_service.list_tools(namespace=ns, tool_type=tt)
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+@when('I attach validation with invalid mode "{mode}" via the fallback service')
+def step_when_attach_invalid_mode(context: Context, mode: str) -> None:
+    try:
+        context.fb_result = context.fb_service.attach_validation(
+            validation_name="local/check",
+            resource_id="res-001",
+            mode=mode,
+        )
+        context.fb_error = None
+    except Exception as exc:
+        context.fb_error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the tool should be registered via the add method")
+def step_then_registered_via_add(context: Context) -> None:
+    assert context.fb_result is not None, "Expected a result from register_tool"
+    assert hasattr(context.fb_tool_repo, "_added"), "Expected _added attribute on repo"
+    assert context.fb_tool_repo._added is not None, (
+        "Expected tool to be passed to add()"
+    )
+
+
+@then("no error should be raised by the fallback service")
+def step_then_no_fallback_error(context: Context) -> None:
+    assert context.fb_error is None, (
+        f"Expected no error, got {type(context.fb_error).__name__}: {context.fb_error}"
+    )
+
+
+@then("a fallback AttributeError should be raised")
+def step_then_attribute_error(context: Context) -> None:
+    assert context.fb_error is not None, "Expected an error but no error was raised"
+    assert isinstance(context.fb_error, AttributeError), (
+        f"Expected AttributeError, got {type(context.fb_error).__name__}: "
+        f"{context.fb_error}"
+    )
+
+
+@then('the updated tool should have name "{name}"')
+def step_then_updated_has_name(context: Context, name: str) -> None:
+    assert context.fb_result is not None, "Expected a result from update_tool"
+    if isinstance(context.fb_result, dict):
+        assert context.fb_result.get("name") == name, (
+            f"Expected name '{name}', got {context.fb_result.get('name')}"
+        )
+    else:
+        assert getattr(context.fb_result, "name", None) == name
+
+
+@then("the update should use the Tool model directly")
+def step_then_update_uses_model(context: Context) -> None:
+    assert context.fb_result is not None, "Expected a result"
+    assert isinstance(context.fb_result, Tool), (
+        f"Expected Tool instance, got {type(context.fb_result).__name__}"
+    )
+
+
+@then("the updated tool should be the original dict")
+def step_then_updated_is_dict(context: Context) -> None:
+    assert context.fb_result is not None, "Expected a result"
+    assert isinstance(context.fb_result, dict), (
+        f"Expected dict, got {type(context.fb_result).__name__}"
+    )
+
+
+@then("the tool should be removed via the remove method")
+def step_then_removed_via_remove(context: Context) -> None:
+    assert context.fb_result is True, (
+        f"Expected True from remove_tool, got {context.fb_result}"
+    )
+    assert hasattr(context.fb_tool_repo, "_removed"), (
+        "Expected _removed attribute on repo"
+    )
+    assert context.fb_tool_repo._removed is not None, (
+        "Expected tool name to be passed to remove()"
+    )
+
+
+@then("the list result should be returned from list_all")
+def step_then_list_result(context: Context) -> None:
+    assert context.fb_result is not None, "Expected a list result"
+    assert isinstance(context.fb_result, list), (
+        f"Expected list, got {type(context.fb_result).__name__}"
+    )
+    assert len(context.fb_result) == 2, (
+        f"Expected 2 items, got {len(context.fb_result)}"
+    )
+
+
+@then('a fallback ValidationError should be raised with message containing "{text}"')
+def step_then_validation_error(context: Context, text: str) -> None:
+    assert context.fb_error is not None, (
+        "Expected ValidationError but no error was raised"
+    )
+    assert isinstance(context.fb_error, ValidationError), (
+        f"Expected ValidationError, got {type(context.fb_error).__name__}: "
+        f"{context.fb_error}"
+    )
+    assert text in str(context.fb_error), (
+        f"Expected '{text}' in error message, got '{context.fb_error}'"
+    )

features/steps/validation_pipeline_steps.py

@@ -0,0 +1,545 @@
+"""Step definitions for validation pipeline scenarios.
+
+All step names are prefixed with 'validation pipeline' context to avoid
+AmbiguousStep collisions with existing step definitions.
+Uses the ``re`` step matcher for patterns that have optional trailing parts.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from typing import Any
+
+from behave import given, then, use_step_matcher, when
+
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    ValidationResult,
+    ValidationSummary,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+# ---------------------------------------------------------------------------
+# Mock executor
+# ---------------------------------------------------------------------------
+
+
+class MockValidationExecutor:
+    """Configurable mock executor for testing the validation pipeline."""
+
+    def __init__(self) -> None:
+        self._results: dict[str, dict[str, Any]] = {}
+        self._timeout_names: set[str] = set()
+        self._exception_names: dict[str, Exception] = {}
+        self._non_dict_names: set[str] = set()
+        self._missing_passed_names: set[str] = set()
+        self._default_passed: bool = False
+        self._stdout_names: set[str] = set()
+        self._stderr_names: set[str] = set()
+
+    def set_passed(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} passed"}
+
+    def set_passed_with_data(self, name: str) -> None:
+        self._results[name] = {
+            "passed": True,
+            "message": f"{name} passed with data",
+            "data": {"detail": "some-value", "count": 42},
+        }
+
+    def set_failed(self, name: str, message: str) -> None:
+        self._results[name] = {"passed": False, "message": message}
+
+    def set_timeout(self, name: str) -> None:
+        self._timeout_names.add(name)
+
+    def set_exception(self, name: str, exc: Exception) -> None:
+        self._exception_names[name] = exc
+
+    def set_non_dict(self, name: str) -> None:
+        self._non_dict_names.add(name)
+
+    def set_missing_passed(self, name: str) -> None:
+        self._missing_passed_names.add(name)
+
+    def set_default_passed(self) -> None:
+        self._default_passed = True
+
+    def set_non_bool_passed(self, name: str) -> None:
+        self._results[name] = {"passed": "yes", "message": f"{name} passed"}
+
+    def set_non_string_message(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": 12345}
+
+    def set_non_dict_data(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok", "data": [1, 2]}
+
+    def set_prints_stdout(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok"}
+        self._stdout_names.add(name)
+
+    def set_prints_stderr(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok"}
+        self._stderr_names.add(name)
+
+    def __call__(
+        self, validation_name: str, arguments: dict[str, Any]
+    ) -> dict[str, Any]:
+        if validation_name in self._timeout_names:
+            # Sleep longer than the test timeout (0.2 s) but not excessively
+            time.sleep(1)
+            return {"passed": True, "message": "should not reach here"}
+
+        if validation_name in self._exception_names:
+            raise self._exception_names[validation_name]
+
+        if validation_name in self._non_dict_names:
+            return "not-a-dict"  # type: ignore[return-value]
+
+        if validation_name in self._missing_passed_names:
+            return {"message": "no passed key here"}
+
+        if validation_name in self._stdout_names:
+            print(f"stdout from {validation_name}")
+
+        if validation_name in self._stderr_names:
+            print(f"stderr from {validation_name}", file=sys.stderr)
+
+        if validation_name in self._results:
+            return self._results[validation_name]
+
+        if self._default_passed:
+            return {"passed": True, "message": f"{validation_name} passed (default)"}
+
+        return {"passed": False, "message": f"{validation_name}: no result configured"}
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_RESOURCE_ID_COUNTER: int = 0
+
+
+def _next_resource_id() -> str:
+    global _RESOURCE_ID_COUNTER
+    _RESOURCE_ID_COUNTER += 1
+    return f"R{_RESOURCE_ID_COUNTER:05d}"
+
+
+# ---------------------------------------------------------------------------
+# Given steps — use "re" matcher for the required/informational patterns
+# ---------------------------------------------------------------------------
+
+use_step_matcher("re")
+
+
+@given(r"a validation pipeline test environment")
+def step_given_pipeline_env(context: Any) -> None:
+    context.vp_commands: list[ValidationCommand] = []
+    context.vp_executor = MockValidationExecutor()
+    context.vp_max_workers: int = 4
+    context.vp_read_only_resources: set[str] = set()
+    context.vp_summary: ValidationSummary | None = None
+    context.vp_plan_metadata: dict[str, Any] | None = None
+    context.vp_resource_id_map: dict[str, str] = {}
+    global _RESOURCE_ID_COUNTER
+    _RESOURCE_ID_COUNTER = 0
+
+
+@given(r"an empty list of validation commands")
+def step_given_empty_commands(context: Any) -> None:
+    context.vp_commands = []
+
+
+@given(
+    r'a required vp command "(?P<name>[^"]+)" on resource "(?P<resource>[^"]+)"'
+    r" with timeout (?P<timeout>[\d.]+)"
+)
+def step_given_required_command_timeout(
+    context: Any, name: str, resource: str, timeout: str
+) -> None:
+    rid = context.vp_resource_id_map.get(resource, _next_resource_id())
+    context.vp_resource_id_map[resource] = rid
+    context.vp_commands.append(
+        ValidationCommand(
+            validation_name=name,
+            resource_id=rid,
+            resource_name=resource,
+            mode=ValidationMode.REQUIRED,
+            arguments={},
+            timeout_seconds=float(timeout),
+        )
+    )
+
+
+@given(
+    r'a required vp command "(?P<name>[^"]+)" on resource "(?P<resource>[^"]+)"'
+    r' having resource_id "(?P<rid>[^"]+)"'
+)
+def step_given_required_command_with_rid(
+    context: Any, name: str, resource: str, rid: str
+) -> None:
+    context.vp_resource_id_map[resource] = rid
+    context.vp_commands.append(
+        ValidationCommand(
+            validation_name=name,
+            resource_id=rid,
+            resource_name=resource,
+            mode=ValidationMode.REQUIRED,
+            arguments={},
+        )
+    )
+
+
+@given(r'a required vp command "(?P<name>[^"]+)" on resource "(?P<resource>[^"]+)"')
+def step_given_required_command(context: Any, name: str, resource: str) -> None:
+    rid = context.vp_resource_id_map.get(resource, _next_resource_id())
+    context.vp_resource_id_map[resource] = rid
+    context.vp_commands.append(
+        ValidationCommand(
+            validation_name=name,
+            resource_id=rid,
+            resource_name=resource,
+            mode=ValidationMode.REQUIRED,
+            arguments={},
+        )
+    )
+
+
+@given(
+    r'an informational vp command "(?P<name>[^"]+)"'
+    r' on resource "(?P<resource>[^"]+)"'
+)
+def step_given_informational_command(context: Any, name: str, resource: str) -> None:
+    rid = context.vp_resource_id_map.get(resource, _next_resource_id())
+    context.vp_resource_id_map[resource] = rid
+    context.vp_commands.append(
+        ValidationCommand(
+            validation_name=name,
+            resource_id=rid,
+            resource_name=resource,
+            mode=ValidationMode.INFORMATIONAL,
+            arguments={},
+        )
+    )
+
+
+@given(r'the vp executor returns passed for "(?P<name>[^"]+)"')
+def step_given_executor_passed(context: Any, name: str) -> None:
+    context.vp_executor.set_passed(name)
+
+
+@given(
+    r'the vp executor returns failed for "(?P<name>[^"]+)"'
+    r' with message "(?P<message>[^"]+)"'
+)
+def step_given_executor_failed(context: Any, name: str, message: str) -> None:
+    context.vp_executor.set_failed(name, message)
+
+
+@given(r"the vp executor returns passed for all validations")
+def step_given_executor_all_passed(context: Any) -> None:
+    context.vp_executor.set_default_passed()
+
+
+@given(r'the vp executor will timeout for "(?P<name>[^"]+)"')
+def step_given_executor_timeout(context: Any, name: str) -> None:
+    context.vp_executor.set_timeout(name)
+
+
+@given(r'the vp executor returns non-dict output for "(?P<name>[^"]+)"')
+def step_given_executor_non_dict(context: Any, name: str) -> None:
+    context.vp_executor.set_non_dict(name)
+
+
+@given(r'the vp executor returns dict without passed key for "(?P<name>[^"]+)"')
+def step_given_executor_missing_passed(context: Any, name: str) -> None:
+    context.vp_executor.set_missing_passed(name)
+
+
+@given(r'the vp executor raises an exception for "(?P<name>[^"]+)"')
+def step_given_executor_exception(context: Any, name: str) -> None:
+    context.vp_executor.set_exception(name, RuntimeError("Test exception"))
+
+
+@given(r'the vp executor returns passed with data for "(?P<name>[^"]+)"')
+def step_given_executor_passed_with_data(context: Any, name: str) -> None:
+    context.vp_executor.set_passed_with_data(name)
+
+
+@given(r'the vp resource "(?P<rid>[^"]+)" is marked as read-only')
+def step_given_read_only_resource(context: Any, rid: str) -> None:
+    context.vp_read_only_resources.add(rid)
+
+
+@given(r"a validation summary with (?P<count>\d+) required failures")
+def step_given_summary_with_failures(context: Any, count: str) -> None:
+    cnt = int(count)
+    context.vp_summary = ValidationSummary(
+        total=cnt + 1,
+        required_passed=1,
+        required_failed=cnt,
+        informational_passed=0,
+        informational_failed=0,
+        results=[],
+    )
+
+
+@given(r"the pipeline max_workers is set to (?P<workers>\d+)")
+def step_given_max_workers(context: Any, workers: str) -> None:
+    context.vp_max_workers = int(workers)
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when(r"the validation pipeline runs for plan with metadata")
+def step_when_pipeline_runs_for_plan(context: Any) -> None:
+    context.vp_plan_metadata = {}
+    pipeline = ValidationPipeline(
+        commands=context.vp_commands,
+        executor=context.vp_executor,
+        max_workers=context.vp_max_workers,
+        read_only_resources=context.vp_read_only_resources,
+    )
+    context.vp_summary = pipeline.run_for_plan(plan_metadata=context.vp_plan_metadata)
+
+
+@when(r"the validation pipeline runs")
+def step_when_pipeline_runs(context: Any) -> None:
+    pipeline = ValidationPipeline(
+        commands=context.vp_commands,
+        executor=context.vp_executor,
+        max_workers=context.vp_max_workers,
+        read_only_resources=context.vp_read_only_resources,
+    )
+    context.vp_summary = pipeline.run()
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then(r"the validation summary total is (?P<count>\d+)")
+def step_then_summary_total(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None, "No summary available"
+    assert context.vp_summary.total == cnt, (
+        f"Expected total={cnt}, got {context.vp_summary.total}"
+    )
+
+
+@then(r"the validation summary required_passed is (?P<count>\d+)")
+def step_then_summary_required_passed(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    assert context.vp_summary.required_passed == cnt, (
+        f"Expected required_passed={cnt}, got {context.vp_summary.required_passed}"
+    )
+
+
+@then(r"the validation summary required_failed is (?P<count>\d+)")
+def step_then_summary_required_failed(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    assert context.vp_summary.required_failed == cnt, (
+        f"Expected required_failed={cnt}, got {context.vp_summary.required_failed}"
+    )
+
+
+@then(r"the validation summary informational_passed is (?P<count>\d+)")
+def step_then_summary_informational_passed(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    assert context.vp_summary.informational_passed == cnt, (
+        f"Expected informational_passed={cnt}, "
+        f"got {context.vp_summary.informational_passed}"
+    )
+
+
+@then(r"the validation summary informational_failed is (?P<count>\d+)")
+def step_then_summary_informational_failed(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    assert context.vp_summary.informational_failed == cnt, (
+        f"Expected informational_failed={cnt}, "
+        f"got {context.vp_summary.informational_failed}"
+    )
+
+
+@then(r"all required validations passed")
+def step_then_all_required_passed(context: Any) -> None:
+    assert context.vp_summary is not None
+    assert context.vp_summary.all_required_passed, (
+        f"Expected all_required_passed=True, got False "
+        f"(required_failed={context.vp_summary.required_failed})"
+    )
+
+
+@then(r"not all required validations passed")
+def step_then_not_all_required_passed(context: Any) -> None:
+    assert context.vp_summary is not None
+    assert not context.vp_summary.all_required_passed, (
+        "Expected all_required_passed=False, got True"
+    )
+
+
+@then(r"the validation results are sorted by resource then mode then name")
+def step_then_results_sorted(context: Any) -> None:
+    assert context.vp_summary is not None
+    results = context.vp_summary.results
+    keys = [(r.resource_name, r.mode.value, r.validation_name) for r in results]
+    assert keys == sorted(keys), f"Results not sorted: {keys}"
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has timed_out true')
+def step_then_result_timed_out(context: Any, name: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert result.timed_out, f"Expected timed_out=True for '{name}'"
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has passed false')
+def step_then_result_passed_false(context: Any, name: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert not result.passed, f"Expected passed=False for '{name}'"
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has passed true')
+def step_then_result_passed_true(context: Any, name: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert result.passed, f"Expected passed=True for '{name}'"
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has error containing "(?P<text>[^"]+)"')
+def step_then_result_error_contains(context: Any, name: str, text: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert result.error is not None, f"Expected error for '{name}', got None"
+    assert text in result.error, (
+        f"Expected error containing '{text}', got: {result.error}"
+    )
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has message containing "(?P<text>[^"]+)"')
+def step_then_result_message_contains(context: Any, name: str, text: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert text in result.message, (
+        f"Expected message containing '{text}', got: {result.message}"
+    )
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has data key "(?P<key>[^"]+)"')
+def step_then_result_has_data_key(context: Any, name: str, key: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert result.data is not None, f"Expected data for '{name}', got None"
+    assert key in result.data, (
+        f"Expected data key '{key}' in {list(result.data.keys())}"
+    )
+
+
+@then(r"the results grouped by resource have (?P<count>\d+) groups")
+def step_then_groups_count(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    groups = ValidationPipeline.group_by_resource(context.vp_summary.results)
+    assert len(groups) == cnt, f"Expected {cnt} groups, got {len(groups)}"
+
+
+@then(r'the vp resource group "(?P<resource>[^"]+)" has (?P<count>\d+) result')
+def step_then_group_result_count(context: Any, resource: str, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_summary is not None
+    groups = ValidationPipeline.group_by_resource(context.vp_summary.results)
+    assert resource in groups, f"Resource '{resource}' not in groups"
+    actual = len(groups[resource])
+    assert actual == cnt, f"Expected {cnt} results for '{resource}', got {actual}"
+
+
+@then(r"the summary all_required_passed property is true")
+def step_then_summary_property_true(context: Any) -> None:
+    assert context.vp_summary is not None
+    assert context.vp_summary.all_required_passed
+
+
+@then(r"the summary all_required_passed property is false")
+def step_then_summary_property_false(context: Any) -> None:
+    assert context.vp_summary is not None
+    assert not context.vp_summary.all_required_passed
+
+
+@then(r"the plan metadata contains validation_summary")
+def step_then_plan_metadata_has_summary(context: Any) -> None:
+    assert context.vp_plan_metadata is not None
+    assert "validation_summary" in context.vp_plan_metadata
+
+
+@then(r"the plan metadata validation_summary total is (?P<count>\d+)")
+def step_then_plan_metadata_total(context: Any, count: str) -> None:
+    cnt = int(count)
+    assert context.vp_plan_metadata is not None
+    vs = context.vp_plan_metadata["validation_summary"]
+    assert vs["total"] == cnt, f"Expected total={cnt}, got {vs['total']}"
+
+
+@given(r'the vp executor returns non-bool passed for "(?P<name>[^"]+)"')
+def step_given_executor_non_bool_passed(context: Any, name: str) -> None:
+    context.vp_executor.set_non_bool_passed(name)
+
+
+@given(r'the vp executor returns non-string message for "(?P<name>[^"]+)"')
+def step_given_executor_non_string_message(context: Any, name: str) -> None:
+    context.vp_executor.set_non_string_message(name)
+
+
+@given(r'the vp executor returns non-dict data for "(?P<name>[^"]+)"')
+def step_given_executor_non_dict_data(context: Any, name: str) -> None:
+    context.vp_executor.set_non_dict_data(name)
+
+
+@given(r'the vp executor prints to stdout for "(?P<name>[^"]+)"')
+def step_given_executor_prints_stdout(context: Any, name: str) -> None:
+    context.vp_executor.set_prints_stdout(name)
+
+
+@given(r'the vp executor prints to stderr for "(?P<name>[^"]+)"')
+def step_given_executor_prints_stderr(context: Any, name: str) -> None:
+    context.vp_executor.set_prints_stderr(name)
+
+
+@then(r'the vp result for "(?P<name>[^"]+)" has positive duration_ms')
+def step_then_result_positive_duration(context: Any, name: str) -> None:
+    assert context.vp_summary is not None
+    result = _find_result(context.vp_summary, name)
+    assert result.duration_ms > 0, (
+        f"Expected duration_ms > 0 for '{name}', got {result.duration_ms}"
+    )
+
+
+# Reset to default parse matcher
+use_step_matcher("parse")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _find_result(summary: ValidationSummary, name: str) -> ValidationResult:
+    """Find a validation result by name."""
+    for r in summary.results:
+        if r.validation_name == name:
+            return r
+    available = [r.validation_name for r in summary.results]
+    msg = f"No result for '{name}', available: {available}"
+    raise AssertionError(msg)

features/steps/validation_pipeline_stream_coverage_steps.py

@@ -0,0 +1,223 @@
+"""Step definitions for validation_pipeline_stream_coverage.feature.
+
+Exercises the _ThreadLocalStream helper methods that lack coverage:
+- encoding property (line 58)
+- writable() method (line 61)
+- readable() method (line 64)
+- flush() with active buffer (line 75)
+- isatty() method (line 79)
+
+All tests use a simple mock original stream — no real I/O needed.
+"""
+
+from __future__ import annotations
+
+import io
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.validation_pipeline import (
+    _ThreadLocalStream,
+)
+
+
+# ---------------------------------------------------------------------------
+# Mock original stream
+# ---------------------------------------------------------------------------
+
+
+class _MockOriginalStream:
+    """Configurable mock that mimics a real stdout/stderr."""
+
+    def __init__(self) -> None:
+        self.encoding: str = "utf-8"
+        self._has_encoding: bool = True
+        self._isatty_return: bool = False
+        self._written: list[str] = []
+        self._flushed: bool = False
+
+    def write(self, s: str) -> int:
+        self._written.append(s)
+        return len(s)
+
+    def flush(self) -> None:
+        self._flushed = True
+
+    def isatty(self) -> bool:
+        return self._isatty_return
+
+
+class _NoEncodingStream:
+    """Mock stream without an encoding attribute."""
+
+    def write(self, s: str) -> int:
+        return len(s)
+
+    def flush(self) -> None:
+        pass
+
+    def isatty(self) -> bool:
+        return False
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a thread-local stream wrapping a mock original stream")
+def step_given_tl_stream(context: Context) -> None:
+    context.tls_original = _MockOriginalStream()
+    context.tls_stream = _ThreadLocalStream(context.tls_original)
+    context.tls_result: Any = None
+    context.tls_captured: str = ""
+    context.tls_error: Exception | None = None
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given('the original stream has encoding "{enc}"')
+def step_given_encoding(context: Context, enc: str) -> None:
+    context.tls_original.encoding = enc
+
+
+@given("the original stream has no encoding attribute")
+def step_given_no_encoding(context: Context) -> None:
+    context.tls_stream = _ThreadLocalStream(_NoEncodingStream())
+
+
+@given("the original stream isatty returns True")
+def step_given_isatty_true(context: Context) -> None:
+    context.tls_original._isatty_return = True
+
+
+@given("the original stream isatty returns False")
+def step_given_isatty_false(context: Context) -> None:
+    context.tls_original._isatty_return = False
+
+
+@given("capture is started on the thread-local stream")
+def step_given_capture_started(context: Context) -> None:
+    context.tls_stream.start_capture()
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I read the thread-local stream encoding")
+def step_when_read_encoding(context: Context) -> None:
+    context.tls_result = context.tls_stream.encoding
+
+
+@when("I check if the thread-local stream is writable")
+def step_when_check_writable(context: Context) -> None:
+    context.tls_result = context.tls_stream.writable()
+
+
+@when("I check if the thread-local stream is readable")
+def step_when_check_readable(context: Context) -> None:
+    context.tls_result = context.tls_stream.readable()
+
+
+@when("I flush the thread-local stream")
+def step_when_flush(context: Context) -> None:
+    try:
+        context.tls_stream.flush()
+        context.tls_error = None
+    except Exception as exc:
+        context.tls_error = exc
+
+
+@when("I check if the thread-local stream isatty")
+def step_when_check_isatty(context: Context) -> None:
+    context.tls_result = context.tls_stream.isatty()
+
+
+@when('I write "{text}" to the thread-local stream')
+def step_when_write(context: Context, text: str) -> None:
+    context.tls_stream.write(text)
+
+
+@when("capture is stopped on the thread-local stream")
+def step_when_stop_capture(context: Context) -> None:
+    context.tls_captured = context.tls_stream.stop_capture()
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then('the encoding should be "{enc}"')
+def step_then_encoding(context: Context, enc: str) -> None:
+    assert context.tls_result == enc, (
+        f"Expected encoding '{enc}', got '{context.tls_result}'"
+    )
+
+
+@then("writable should return True")
+def step_then_writable_true(context: Context) -> None:
+    assert context.tls_result is True, (
+        f"Expected writable()=True, got {context.tls_result}"
+    )
+
+
+@then("readable should return False")
+def step_then_readable_false(context: Context) -> None:
+    assert context.tls_result is False, (
+        f"Expected readable()=False, got {context.tls_result}"
+    )
+
+
+@then("the flush should succeed without error")
+def step_then_flush_ok(context: Context) -> None:
+    assert context.tls_error is None, f"Expected no error, got {context.tls_error}"
+
+
+@then("the original stream flush should be called")
+def step_then_original_flushed(context: Context) -> None:
+    assert context.tls_original._flushed, (
+        "Expected original stream flush() to be called"
+    )
+
+
+@then("isatty should return True")
+def step_then_isatty_true(context: Context) -> None:
+    assert context.tls_result is True, (
+        f"Expected isatty()=True, got {context.tls_result}"
+    )
+
+
+@then("isatty should return False")
+def step_then_isatty_false(context: Context) -> None:
+    assert context.tls_result is False, (
+        f"Expected isatty()=False, got {context.tls_result}"
+    )
+
+
+@then('the original stream should have received "{text}"')
+def step_then_original_received(context: Context, text: str) -> None:
+    assert text in context.tls_original._written, (
+        f"Expected '{text}' in original writes, got {context.tls_original._written}"
+    )
+
+
+@then('the captured text should be "{text}"')
+def step_then_captured_text(context: Context, text: str) -> None:
+    assert context.tls_captured == text, (
+        f"Expected captured text '{text}', got '{context.tls_captured}'"
+    )
+
+
+@then("the captured text should be empty")
+def step_then_captured_text_empty(context: Context) -> None:
+    assert context.tls_captured == "", (
+        f"Expected empty captured text, got '{context.tls_captured}'"
+    )

features/tool_registry_service_fallback_coverage.feature

@@ -0,0 +1,66 @@
+@unit @coverage @tool_registry
+Feature: Tool Registry Service - fallback method paths
+  As a developer maintaining the tool registry service
+  I want to exercise the fallback method resolution branches
+  So that all delegation paths in register_tool, update_tool, and remove_tool are covered
+
+  # --- register_tool fallback: no create, uses add -------------------------
+
+  Scenario: register_tool falls back to add when create does not exist
+    Given a tool registry service with a repo that only has add method
+    When I register a tool via the fallback service
+    Then the tool should be registered via the add method
+    And no error should be raised by the fallback service
+
+  Scenario: register_tool falls back to create attribute when neither create nor add are callable
+    Given a tool registry service with a repo that has no create or add methods
+    When I register a tool via the fallback service and expect error
+    Then a fallback AttributeError should be raised
+
+  # --- update_tool with tool_config parameter (dict) ----------------------
+
+  Scenario: update_tool with string tool name and dict config merges them
+    Given a tool registry service with a standard mock repo
+    When I update tool "local/my-tool" with dict config via the fallback service
+    Then the updated tool should have name "local/my-tool"
+    And no error should be raised by the fallback service
+
+  Scenario: update_tool with string tool name and Tool model config uses model
+    Given a tool registry service with a standard mock repo
+    When I update tool "local/my-tool" with Tool model config via the fallback service
+    Then the update should use the Tool model directly
+    And no error should be raised by the fallback service
+
+  Scenario: update_tool with dict tool and dict config uses the dict tool
+    Given a tool registry service with a standard mock repo
+    When I update with dict tool and dict config via the fallback service
+    Then the updated tool should be the original dict
+    And no error should be raised by the fallback service
+
+  # --- remove_tool fallback: no delete, uses remove -----------------------
+
+  Scenario: remove_tool falls back to remove when delete does not exist
+    Given a tool registry service with a repo that only has remove method
+    When I remove tool "local/gone" via the fallback service
+    Then the tool should be removed via the remove method
+    And no error should be raised by the fallback service
+
+  Scenario: remove_tool falls back to delete attribute when neither delete nor remove are callable
+    Given a tool registry service with a repo that has no delete or remove methods
+    When I remove tool "local/gone" via the fallback service and expect error
+    Then a fallback AttributeError should be raised
+
+  # --- list_tools delegates to list_all -----------------------------------
+
+  Scenario: list_tools delegates to repo list_all with filters
+    Given a tool registry service with a standard mock repo
+    When I list tools with namespace "local" and type "validation" via the fallback service
+    Then the list result should be returned from list_all
+    And no error should be raised by the fallback service
+
+  # --- attach_validation with invalid mode --------------------------------
+
+  Scenario: attach_validation raises ValidationError for invalid mode
+    Given a tool registry service with a standard mock repo
+    When I attach validation with invalid mode "bogus" via the fallback service
+    Then a fallback ValidationError should be raised with message containing "Invalid mode"

features/validation_pipeline.feature

@@ -0,0 +1,235 @@
+Feature: Validation pipeline
+  As a plan executor
+  I want a validation pipeline that runs validations deterministically
+  So that required failures block and informational failures are reported
+
+  Background:
+    Given a validation pipeline test environment
+
+  # ── Empty pipeline ─────────────────────────────────────────────
+
+  Scenario: Empty validation list returns passing summary
+    Given an empty list of validation commands
+    When the validation pipeline runs
+    Then the validation summary total is 0
+    And all required validations passed
+
+  # ── Single required validation ─────────────────────────────────
+
+  Scenario: Single required validation passes
+    Given a required vp command "check-syntax" on resource "repo-main"
+    And the vp executor returns passed for "check-syntax"
+    When the validation pipeline runs
+    Then the validation summary total is 1
+    And the validation summary required_passed is 1
+    And all required validations passed
+
+  Scenario: Single required validation fails and blocks
+    Given a required vp command "check-syntax" on resource "repo-main"
+    And the vp executor returns failed for "check-syntax" with message "Syntax error on line 42"
+    When the validation pipeline runs
+    Then the validation summary total is 1
+    And the validation summary required_failed is 1
+    And not all required validations passed
+
+  # ── Single informational validation ────────────────────────────
+
+  Scenario: Single informational validation passes
+    Given an informational vp command "lint-check" on resource "repo-main"
+    And the vp executor returns passed for "lint-check"
+    When the validation pipeline runs
+    Then the validation summary total is 1
+    And the validation summary informational_passed is 1
+    And all required validations passed
+
+  Scenario: Single informational validation fails but does not block
+    Given an informational vp command "lint-check" on resource "repo-main"
+    And the vp executor returns failed for "lint-check" with message "Linting warnings found"
+    When the validation pipeline runs
+    Then the validation summary total is 1
+    And the validation summary informational_failed is 1
+    And all required validations passed
+
+  # ── Multiple validations sorted by resource/mode/name ──────────
+
+  Scenario: Multiple validations are sorted by resource name then mode then name
+    Given a required vp command "z-check" on resource "beta-repo"
+    And a required vp command "a-check" on resource "alpha-repo"
+    And an informational vp command "b-lint" on resource "alpha-repo"
+    And the vp executor returns passed for all validations
+    When the validation pipeline runs
+    Then the validation results are sorted by resource then mode then name
+    And the validation summary total is 3
+
+  # ── Timeout handling ───────────────────────────────────────────
+
+  Scenario: Validation that exceeds timeout is marked as timed out
+    Given a required vp command "slow-check" on resource "repo-main" with timeout 0.2
+    And the vp executor will timeout for "slow-check"
+    When the validation pipeline runs
+    Then the vp result for "slow-check" has timed_out true
+    And the vp result for "slow-check" has passed false
+    And the vp result for "slow-check" has error containing "TimeoutError"
+
+  # ── Error normalisation ────────────────────────────────────────
+
+  Scenario: Executor returning non-dict output is normalised to failure
+    Given a required vp command "broken-check" on resource "repo-main"
+    And the vp executor returns non-dict output for "broken-check"
+    When the validation pipeline runs
+    Then the vp result for "broken-check" has passed false
+    And the vp result for "broken-check" has message containing "non-dict"
+
+  Scenario: Executor missing passed key defaults to false
+    Given a required vp command "partial-check" on resource "repo-main"
+    And the vp executor returns dict without passed key for "partial-check"
+    When the validation pipeline runs
+    Then the vp result for "partial-check" has passed false
+
+  Scenario: Executor raising exception is captured as error
+    Given a required vp command "error-check" on resource "repo-main"
+    And the vp executor raises an exception for "error-check"
+    When the validation pipeline runs
+    Then the vp result for "error-check" has passed false
+    And the vp result for "error-check" has error containing "RuntimeError"
+
+  # ── Required + informational mixed results ─────────────────────
+
+  Scenario: Mixed required and informational with all required passing
+    Given a required vp command "req-check" on resource "repo-main"
+    And an informational vp command "info-lint" on resource "repo-main"
+    And the vp executor returns passed for "req-check"
+    And the vp executor returns failed for "info-lint" with message "Info warning"
+    When the validation pipeline runs
+    Then the validation summary required_passed is 1
+    And the validation summary informational_failed is 1
+    And all required validations passed
+
+  Scenario: Mixed validations where required fails blocks
+    Given a required vp command "req-check" on resource "repo-main"
+    And an informational vp command "info-lint" on resource "repo-main"
+    And the vp executor returns failed for "req-check" with message "Required failed"
+    And the vp executor returns passed for "info-lint"
+    When the validation pipeline runs
+    Then the validation summary required_failed is 1
+    And the validation summary informational_passed is 1
+    And not all required validations passed
+
+  # ── Summary counts verification ────────────────────────────────
+
+  Scenario: Summary counts are accurate for multiple validations
+    Given a required vp command "req-a" on resource "repo-1"
+    And a required vp command "req-b" on resource "repo-1"
+    And an informational vp command "info-a" on resource "repo-1"
+    And an informational vp command "info-b" on resource "repo-1"
+    And the vp executor returns passed for "req-a"
+    And the vp executor returns failed for "req-b" with message "Failed"
+    And the vp executor returns passed for "info-a"
+    And the vp executor returns failed for "info-b" with message "Info fail"
+    When the validation pipeline runs
+    Then the validation summary total is 4
+    And the validation summary required_passed is 1
+    And the validation summary required_failed is 1
+    And the validation summary informational_passed is 1
+    And the validation summary informational_failed is 1
+
+  # ── Grouping by resource ───────────────────────────────────────
+
+  Scenario: Validation results can be grouped by resource
+    Given a required vp command "check-a" on resource "repo-alpha"
+    And a required vp command "check-b" on resource "repo-beta"
+    And the vp executor returns passed for all validations
+    When the validation pipeline runs
+    Then the results grouped by resource have 2 groups
+    And the vp resource group "repo-alpha" has 1 result
+    And the vp resource group "repo-beta" has 1 result
+
+  # ── Read-only resource guard ───────────────────────────────────
+
+  Scenario: Skip validation on read-only resource
+    Given a required vp command "write-check" on resource "readonly-repo" having resource_id "RO001"
+    And the vp resource "RO001" is marked as read-only
+    And the vp executor returns passed for all validations
+    When the validation pipeline runs
+    Then the vp result for "write-check" has passed true
+    And the vp result for "write-check" has message containing "read-only"
+
+  # ── ValidationSummary all_required_passed property ─────────────
+
+  Scenario: ValidationSummary all_required_passed is true when no required failures
+    Given a validation summary with 0 required failures
+    Then the summary all_required_passed property is true
+
+  Scenario: ValidationSummary all_required_passed is false when required failures exist
+    Given a validation summary with 2 required failures
+    Then the summary all_required_passed property is false
+
+  # ── Pipeline with max_workers=1 ────────────────────────────────
+
+  Scenario: Pipeline with max_workers=1 executes sequentially
+    Given a required vp command "seq-a" on resource "repo-1"
+    And a required vp command "seq-b" on resource "repo-1"
+    And the vp executor returns passed for all validations
+    And the pipeline max_workers is set to 1
+    When the validation pipeline runs
+    Then the validation summary total is 2
+    And all required validations passed
+
+  # ── Validation with data payload ───────────────────────────────
+
+  Scenario: Validation result includes data payload
+    Given a required vp command "data-check" on resource "repo-main"
+    And the vp executor returns passed with data for "data-check"
+    When the validation pipeline runs
+    Then the vp result for "data-check" has data key "detail"
+
+  # ── run_for_plan integration ───────────────────────────────────
+
+  Scenario: run_for_plan stores summary in plan metadata
+    Given a required vp command "plan-check" on resource "repo-main"
+    And the vp executor returns passed for "plan-check"
+    When the validation pipeline runs for plan with metadata
+    Then the plan metadata contains validation_summary
+    And the plan metadata validation_summary total is 1
+
+  # ── Normalisation edge cases ───────────────────────────────────
+
+  Scenario: Executor returns non-bool passed value is normalised to false
+    Given a required vp command "nonbool-check" on resource "repo-main"
+    And the vp executor returns non-bool passed for "nonbool-check"
+    When the validation pipeline runs
+    Then the vp result for "nonbool-check" has passed false
+
+  Scenario: Executor returns non-string message is coerced to string
+    Given a required vp command "nonstr-msg" on resource "repo-main"
+    And the vp executor returns non-string message for "nonstr-msg"
+    When the validation pipeline runs
+    Then the vp result for "nonstr-msg" has passed true
+
+  Scenario: Executor returns non-dict data is wrapped in raw key
+    Given a required vp command "nondict-data" on resource "repo-main"
+    And the vp executor returns non-dict data for "nondict-data"
+    When the validation pipeline runs
+    Then the vp result for "nondict-data" has data key "raw"
+
+  # ── Stdout/stderr capture ──────────────────────────────────────
+
+  Scenario: Validation that prints to stdout has output captured
+    Given a required vp command "stdout-check" on resource "repo-main"
+    And the vp executor prints to stdout for "stdout-check"
+    When the validation pipeline runs
+    Then the vp result for "stdout-check" has data key "_captured_stdout"
+
+  Scenario: Validation that prints to stderr has output captured
+    Given a required vp command "stderr-check" on resource "repo-main"
+    And the vp executor prints to stderr for "stderr-check"
+    When the validation pipeline runs
+    Then the vp result for "stderr-check" has data key "_captured_stderr"
+
+  # ── Duration tracking ────────────────────────────────────────────
+
+  Scenario: Validation result has positive duration
+    Given a required vp command "dur-check" on resource "repo-main"
+    And the vp executor returns passed for "dur-check"
+    When the validation pipeline runs
+    Then the vp result for "dur-check" has positive duration_ms

features/validation_pipeline_stream_coverage.feature

@@ -0,0 +1,76 @@
+@unit @coverage @validation_pipeline
+Feature: Validation pipeline _ThreadLocalStream coverage
+  As a developer maintaining the validation pipeline
+  I want to exercise the _ThreadLocalStream helper methods
+  So that all stream-interface methods have coverage
+
+  Background:
+    Given a thread-local stream wrapping a mock original stream
+
+  # --- encoding property ---------------------------------------------------
+
+  Scenario: encoding property returns the original stream encoding
+    Given the original stream has encoding "utf-8"
+    When I read the thread-local stream encoding
+    Then the encoding should be "utf-8"
+
+  Scenario: encoding property falls back to utf-8 when original has no encoding
+    Given the original stream has no encoding attribute
+    When I read the thread-local stream encoding
+    Then the encoding should be "utf-8"
+
+  # --- writable method -----------------------------------------------------
+
+  Scenario: writable returns True
+    When I check if the thread-local stream is writable
+    Then writable should return True
+
+  # --- readable method -----------------------------------------------------
+
+  Scenario: readable returns False
+    When I check if the thread-local stream is readable
+    Then readable should return False
+
+  # --- flush method with active buffer -------------------------------------
+
+  Scenario: flush flushes the capture buffer when capturing is active
+    Given capture is started on the thread-local stream
+    When I write "hello" to the thread-local stream
+    And I flush the thread-local stream
+    Then the flush should succeed without error
+
+  # --- flush method without active buffer ----------------------------------
+
+  Scenario: flush flushes the original stream when no capture is active
+    When I flush the thread-local stream
+    Then the original stream flush should be called
+
+  # --- isatty method -------------------------------------------------------
+
+  Scenario: isatty delegates to the original stream
+    Given the original stream isatty returns True
+    When I check if the thread-local stream isatty
+    Then isatty should return True
+
+  Scenario: isatty returns False when original returns False
+    Given the original stream isatty returns False
+    When I check if the thread-local stream isatty
+    Then isatty should return False
+
+  # --- write with and without capture ------------------------------------
+
+  Scenario: write without capture goes to original stream
+    When I write "direct output" to the thread-local stream
+    Then the original stream should have received "direct output"
+
+  Scenario: write with capture goes to capture buffer
+    Given capture is started on the thread-local stream
+    When I write "captured output" to the thread-local stream
+    And capture is stopped on the thread-local stream
+    Then the captured text should be "captured output"
+
+  # --- stop_capture when no capture was started ----------------------------
+
+  Scenario: stop_capture returns empty string when no capture was started
+    When capture is stopped on the thread-local stream
+    Then the captured text should be empty

implementation_plan.md

@@ -3899,32 +3899,32 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [ ] Forgejo PR [Luis]: Open PR from `feature/m3-diff-review` to `master` with a suitable and thorough description
 
 **Parallel Group C6: Validation Pipeline [Luis + Jeff]** (depends on C5 and validation attachment config)
-- [ ] **COMMIT (Owner: Luis | Group: C6.pipeline | Branch: feature/m3-validation-pipeline | Planned: Day 21 | Expected: Day 22) - Commit message: "feat(validation): add validation pipeline and results model"**
-  - [ ] Git [Luis]: `git checkout master`
-  - [ ] Git [Luis]: `git pull origin master`
-  - [ ] Git [Luis]: `git checkout -b feature/m3-validation-pipeline`
-  - [ ] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Luis]: Implement ValidationCommand, ValidationResult, and ValidationPipeline using Validation attachments from Tool Registry.
-  - [ ] Code [Luis]: Run validations at end of Execute phase only; do not re-run during Apply per spec.
-  - [ ] Code [Luis]: Enforce required vs informational validation modes and fix-then-revalidate loop hooks.
-  - [ ] Code [Luis]: Persist validation summary into Plan metadata for later review.
-  - [ ] Code [Luis]: Sort validations by (resource, mode, validation name) for deterministic execution order.
-  - [ ] Code [Luis]: Add per-validation timeout and capture stdout/stderr in ValidationResult for troubleshooting.
-  - [ ] Code [Luis]: Group validations by resource and cap parallel validation concurrency per resource type.
-  - [ ] Code [Luis]: Normalize validation output to `passed/message/data` schema even for tool errors.
-  - [ ] Code [Luis]: Add aggregation helper to compute counts (required_passed, required_failed, info_failed) for CLI display.
-  - [ ] Code [Luis]: Add guard to skip validations on read-only resources when validation requires write access (should never happen).
-  - [ ] Docs [Luis]: Add `docs/reference/validation_pipeline.md` with ordering, timeouts, and failure handling.
-  - [ ] Docs [Luis]: Document concurrency limits and output normalization rules.
-  - [ ] Tests (Behave) [Luis]: Add `features/validation_pipeline.feature` for pass/fail paths and required/informational modes.
-  - [ ] Tests (Robot) [Luis]: Add `robot/validation_pipeline.robot` for pipeline smoke tests.
-  - [ ] Tests (ASV) [Luis]: Add `benchmarks/validation_pipeline_bench.py` for pipeline runtime.
-  - [ ] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
-  - [ ] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
-  - [ ] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
-  - [ ] Git [Luis]: `git commit -m "feat(validation): add validation pipeline and results model"`
-  - [ ] Git [Luis]: `git push -u origin feature/m3-validation-pipeline`
-  - [ ] Forgejo PR [Luis]: Open PR from `feature/m3-validation-pipeline` to `master` with a suitable and thorough description
+- [X] **COMMIT (Owner: Luis | Group: C6.pipeline | Branch: feature/m3-validation-pipeline | Planned: Day 21 | Expected: Day 22) - Commit message: "feat(validation): add validation pipeline and results model"** Done: Day 22, February 21, 2026
+  - [X] Git [Luis]: `git checkout master`
+  - [X] Git [Luis]: `git pull origin master`
+  - [X] Git [Luis]: `git checkout -b feature/m3-validation-pipeline`
+  - [X] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
+  - [X] Code [Luis]: Implement ValidationCommand, ValidationResult, and ValidationPipeline using Validation attachments from Tool Registry.
+  - [X] Code [Luis]: Run validations at end of Execute phase only; do not re-run during Apply per spec.
+  - [X] Code [Luis]: Enforce required vs informational validation modes and fix-then-revalidate loop hooks.
+  - [X] Code [Luis]: Persist validation summary into Plan metadata for later review.
+  - [X] Code [Luis]: Sort validations by (resource, mode, validation name) for deterministic execution order.
+  - [X] Code [Luis]: Add per-validation timeout and capture stdout/stderr in ValidationResult for troubleshooting.
+  - [X] Code [Luis]: Group validations by resource and cap parallel validation concurrency per resource type.
+  - [X] Code [Luis]: Normalize validation output to `passed/message/data` schema even for tool errors.
+  - [X] Code [Luis]: Add aggregation helper to compute counts (required_passed, required_failed, info_failed) for CLI display.
+  - [X] Code [Luis]: Add guard to skip validations on read-only resources when validation requires write access (should never happen).
+  - [X] Docs [Luis]: Add `docs/reference/validation_pipeline.md` with ordering, timeouts, and failure handling.
+  - [X] Docs [Luis]: Document concurrency limits and output normalization rules.
+  - [X] Tests (Behave) [Luis]: Add `features/validation_pipeline.feature` for pass/fail paths and required/informational modes.
+  - [X] Tests (Robot) [Luis]: Add `robot/validation_pipeline.robot` for pipeline smoke tests.
+  - [X] Tests (ASV) [Luis]: Add `benchmarks/validation_pipeline_bench.py` for pipeline runtime.
+  - [X] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
+  - [X] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
+  - [X] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
+  - [X] Git [Luis]: `git commit -m "feat(validation): add validation pipeline and results model"`
+  - [X] Git [Luis]: `git push -u origin feature/m3-validation-pipeline`
+  - [X] Forgejo PR [Luis]: Open PR from `feature/m3-validation-pipeline` to `master` with a suitable and thorough description
 - [ ] **COMMIT (Owner: Jeff | Group: C6.wraps | Branch: feature/m3-validation-wraps | Planned: Day 21 | Expected: Day 22) - Commit message: "feat(validation): support wrapped tools and transforms"**
   - [ ] Git [Jeff]: `git checkout master`
   - [ ] Git [Jeff]: `git pull origin master`

robot/helper_validation_pipeline.py

@@ -0,0 +1,196 @@
+"""Helper script for Robot Framework validation pipeline tests.
+
+Self-contained Python helper that exercises the ValidationPipeline
+and its models, printing sentinel strings on success and exiting
+with code 1 on failure.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+# Ensure src is importable when run from workspace root
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    ValidationResult,
+    ValidationSummary,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+# ---------------------------------------------------------------------------
+# Mock executor
+# ---------------------------------------------------------------------------
+
+
+def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Simple mock executor that always passes."""
+    return {"passed": True, "message": f"{validation_name} passed"}
+
+
+def _failing_executor(
+    validation_name: str, arguments: dict[str, Any]
+) -> dict[str, Any]:
+    """Mock executor that always fails."""
+    return {"passed": False, "message": f"{validation_name} failed"}
+
+
+# ---------------------------------------------------------------------------
+# Sub-commands
+# ---------------------------------------------------------------------------
+
+
+def test_models() -> None:
+    """Verify ValidationCommand, ValidationResult, ValidationSummary models."""
+    cmd = ValidationCommand(
+        validation_name="check-syntax",
+        resource_id="R00001",
+        resource_name="repo-main",
+        mode=ValidationMode.REQUIRED,
+        arguments={"key": "value"},
+        timeout_seconds=15.0,
+    )
+    assert cmd.validation_name == "check-syntax"
+    assert cmd.mode == ValidationMode.REQUIRED
+    assert cmd.timeout_seconds == 15.0
+
+    result = ValidationResult(
+        validation_name="check-syntax",
+        resource_id="R00001",
+        resource_name="repo-main",
+        mode=ValidationMode.REQUIRED,
+        passed=True,
+        message="All good",
+        data={"detail": "ok"},
+        duration_ms=42.5,
+        error=None,
+        timed_out=False,
+    )
+    assert result.passed is True
+    assert result.duration_ms == 42.5
+
+    summary = ValidationSummary(
+        total=2,
+        required_passed=1,
+        required_failed=1,
+        informational_passed=0,
+        informational_failed=0,
+        results=[],
+    )
+    assert summary.all_required_passed is False
+    print("models-ok")
+
+
+def test_empty_pipeline() -> None:
+    """Verify empty pipeline returns passing summary."""
+    pipeline = ValidationPipeline(
+        commands=[],
+        executor=_mock_executor,
+    )
+    summary = pipeline.run()
+    assert summary.total == 0
+    assert summary.all_required_passed is True
+    print("empty-pipeline-ok")
+
+
+def test_passing_pipeline() -> None:
+    """Verify pipeline with passing validations."""
+    commands = [
+        ValidationCommand(
+            validation_name="check-a",
+            resource_id="R1",
+            resource_name="repo",
+            mode=ValidationMode.REQUIRED,
+        ),
+        ValidationCommand(
+            validation_name="check-b",
+            resource_id="R1",
+            resource_name="repo",
+            mode=ValidationMode.INFORMATIONAL,
+        ),
+    ]
+    pipeline = ValidationPipeline(
+        commands=commands,
+        executor=_mock_executor,
+        max_workers=1,
+    )
+    summary = pipeline.run()
+    assert summary.total == 2
+    assert summary.required_passed == 1
+    assert summary.informational_passed == 1
+    assert summary.all_required_passed is True
+    print("passing-pipeline-ok")
+
+
+def test_failing_required() -> None:
+    """Verify failing required validation blocks."""
+    commands = [
+        ValidationCommand(
+            validation_name="check-fail",
+            resource_id="R1",
+            resource_name="repo",
+            mode=ValidationMode.REQUIRED,
+        ),
+    ]
+    pipeline = ValidationPipeline(
+        commands=commands,
+        executor=_failing_executor,
+        max_workers=1,
+    )
+    summary = pipeline.run()
+    assert summary.total == 1
+    assert summary.required_failed == 1
+    assert summary.all_required_passed is False
+    print("failing-required-ok")
+
+
+def test_run_for_plan() -> None:
+    """Verify run_for_plan stores summary in metadata."""
+    commands = [
+        ValidationCommand(
+            validation_name="plan-check",
+            resource_id="R1",
+            resource_name="repo",
+            mode=ValidationMode.REQUIRED,
+        ),
+    ]
+    pipeline = ValidationPipeline(
+        commands=commands,
+        executor=_mock_executor,
+        max_workers=1,
+    )
+    metadata: dict[str, Any] = {}
+    summary = pipeline.run_for_plan(plan_metadata=metadata)
+    assert summary.total == 1
+    assert "validation_summary" in metadata
+    assert metadata["validation_summary"]["total"] == 1
+    assert metadata["validation_summary"]["all_required_passed"] is True
+    print("run-for-plan-ok")
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "models"
+    commands_map = {
+        "models": test_models,
+        "empty_pipeline": test_empty_pipeline,
+        "passing_pipeline": test_passing_pipeline,
+        "failing_required": test_failing_required,
+        "run_for_plan": test_run_for_plan,
+    }
+    handler = commands_map.get(cmd)
+    if handler is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        sys.exit(1)
+    try:
+        handler()
+    except Exception as exc:
+        print(f"FAIL: {exc}", file=sys.stderr)
+        sys.exit(1)

robot/validation_pipeline.robot

@@ -0,0 +1,47 @@
+*** Settings ***
+Documentation    Smoke tests for the validation pipeline service
+Library          Process
+Library          OperatingSystem
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_validation_pipeline.py
+
+*** Test Cases ***
+Validation Pipeline Models Are Valid
+    [Documentation]    Verify ValidationCommand, ValidationResult, ValidationSummary Pydantic models
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    models    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    models-ok
+
+Validation Pipeline Empty Pipeline Returns Passing Summary
+    [Documentation]    Empty command list produces a zero-total passing summary
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    empty_pipeline    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    empty-pipeline-ok
+
+Validation Pipeline Passing Validations
+    [Documentation]    Pipeline with passing required and informational validations
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    passing_pipeline    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    passing-pipeline-ok
+
+Validation Pipeline Failing Required Blocks
+    [Documentation]    A failing required validation sets all_required_passed to False
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    failing_required    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    failing-required-ok
+
+Validation Pipeline Run For Plan Integration
+    [Documentation]    run_for_plan stores validation_summary in plan metadata dict
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    run_for_plan    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    run-for-plan-ok
+
+Validation Pipeline Unknown Command Returns Error
+    [Documentation]    Running with unknown command exits with code 1
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    nonexistent_cmd    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    1
+    Should Contain    ${result.stderr}    Unknown command

src/cleveragents/application/services/__init__.py

@@ -12,9 +12,19 @@ from cleveragents.application.services.skill_registry_service import (
 from cleveragents.application.services.tool_registry_service import (
     ToolRegistryService,
 )
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    ValidationResult,
+    ValidationSummary,
+)
 
 __all__ = [
     "PersistentSessionService",
     "SkillRegistryService",
     "ToolRegistryService",
+    "ValidationCommand",
+    "ValidationPipeline",
+    "ValidationResult",
+    "ValidationSummary",
 ]

src/cleveragents/application/services/validation_pipeline.py

@@ -0,0 +1,569 @@
+"""Validation pipeline service for CleverAgents.
+
+Orchestrates execution of validation commands against resources,
+enforces required vs informational semantics, and produces an
+aggregated validation summary.
+
+The pipeline:
+- Sorts validations deterministically by (resource_name, mode, validation_name)
+- Groups validations by resource for organised reporting
+- Executes each validation with a configurable timeout
+- Captures stdout/stderr for troubleshooting
+- Normalises output to a consistent passed/message/data schema
+- Builds a ValidationSummary with aggregated counts
+- Integrates with the plan lifecycle for metadata persistence
+
+Based on ``docs/specification.md`` and ADR-013 (Validation Abstraction).
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import sys
+import threading
+import time
+from collections import defaultdict
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Protocol
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cleveragents.domain.models.core.tool import ValidationMode
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Thread-safe stdout/stderr capture
+# ---------------------------------------------------------------------------
+
+
+class _ThreadLocalStream(io.TextIOBase):
+    """Stream wrapper providing thread-local output capture.
+
+    Non-capturing threads write through to the original stream.
+    Threads that call ``start_capture()`` collect writes in a
+    per-thread ``StringIO`` buffer until ``stop_capture()`` returns
+    the captured text.
+    """
+
+    def __init__(self, original: Any) -> None:
+        self._original = original
+        self._local = threading.local()
+
+    @property
+    def encoding(self) -> str:  # type: ignore[override]
+        return getattr(self._original, "encoding", "utf-8")
+
+    def writable(self) -> bool:
+        return True
+
+    def readable(self) -> bool:
+        return False
+
+    def write(self, s: str) -> int:
+        buf: io.StringIO | None = getattr(self._local, "buffer", None)
+        if buf is not None:
+            return buf.write(s)
+        return self._original.write(s)
+
+    def flush(self) -> None:
+        buf: io.StringIO | None = getattr(self._local, "buffer", None)
+        if buf is not None:
+            buf.flush()
+        self._original.flush()
+
+    def isatty(self) -> bool:
+        return self._original.isatty()
+
+    def start_capture(self) -> None:
+        """Begin capturing writes for the current thread."""
+        self._local.buffer = io.StringIO()
+
+    def stop_capture(self) -> str:
+        """Stop capturing and return captured text for the current thread."""
+        buf: io.StringIO | None = getattr(self._local, "buffer", None)
+        self._local.buffer = None
+        return buf.getvalue() if buf else ""
+
+
+# ---------------------------------------------------------------------------
+# Type alias for validation executor callables
+# ---------------------------------------------------------------------------
+
+
+class ValidationExecutor(Protocol):
+    """Protocol for a callable that executes a single validation."""
+
+    def __call__(
+        self, validation_name: str, arguments: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Execute a validation and return result dict.
+
+        Must return a dict with at least:
+        - ``passed`` (bool): whether the validation passed
+        - ``message`` (str): human-readable result description
+
+        May optionally include:
+        - ``data`` (dict): additional structured data
+        """
+        ...  # pragma: no cover
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
+
+class ValidationCommand(BaseModel):
+    """Command describing a single validation to execute.
+
+    Attributes:
+        validation_name: Name of the validation to run.
+        resource_id: ULID of the resource being validated.
+        resource_name: Human-readable resource name.
+        mode: Whether failure blocks (required) or is informational.
+        arguments: Arguments to pass to the validation executor.
+        timeout_seconds: Maximum execution time in seconds.
+    """
+
+    validation_name: str = Field(
+        ..., min_length=1, description="Name of the validation"
+    )
+    resource_id: str = Field(..., min_length=1, description="Resource ULID")
+    resource_name: str = Field(
+        ..., min_length=1, description="Human-readable resource name"
+    )
+    mode: ValidationMode = Field(..., description="Required or informational")
+    arguments: dict[str, Any] = Field(
+        default_factory=dict, description="Arguments for the validation"
+    )
+    timeout_seconds: float = Field(
+        default=30.0, ge=0.1, description="Timeout in seconds"
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+        use_enum_values=False,
+    )
+
+
+class ValidationResult(BaseModel):
+    """Result of executing a single validation.
+
+    Attributes:
+        validation_name: Name of the executed validation.
+        resource_id: ULID of the validated resource.
+        resource_name: Human-readable resource name.
+        mode: Whether the validation was required or informational.
+        passed: Whether the validation passed.
+        message: Human-readable result description.
+        data: Optional structured result data.
+        duration_ms: Execution duration in milliseconds.
+        error: Error message if the validation raised an exception.
+        timed_out: Whether the validation exceeded its timeout.
+    """
+
+    validation_name: str = Field(..., description="Name of the validation")
+    resource_id: str = Field(..., description="Resource ULID")
+    resource_name: str = Field(..., description="Human-readable resource name")
+    mode: ValidationMode = Field(..., description="Required or informational")
+    passed: bool = Field(..., description="Whether the validation passed")
+    message: str = Field(..., description="Human-readable result description")
+    data: dict[str, Any] | None = Field(
+        default=None, description="Optional structured data"
+    )
+    duration_ms: float = Field(..., ge=0.0, description="Duration in milliseconds")
+    error: str | None = Field(
+        default=None, description="Error message if validation raised"
+    )
+    timed_out: bool = Field(
+        default=False, description="Whether the validation timed out"
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+        use_enum_values=False,
+    )
+
+
+class ValidationSummary(BaseModel):
+    """Aggregated summary of all validation results.
+
+    Attributes:
+        total: Total number of validations executed.
+        required_passed: Count of required validations that passed.
+        required_failed: Count of required validations that failed.
+        informational_passed: Count of informational validations that passed.
+        informational_failed: Count of informational validations that failed.
+        results: Individual validation results.
+    """
+
+    total: int = Field(..., ge=0, description="Total validations executed")
+    required_passed: int = Field(..., ge=0, description="Required validations passed")
+    required_failed: int = Field(..., ge=0, description="Required validations failed")
+    informational_passed: int = Field(
+        ..., ge=0, description="Informational validations passed"
+    )
+    informational_failed: int = Field(
+        ..., ge=0, description="Informational validations failed"
+    )
+    results: list[ValidationResult] = Field(
+        default_factory=list, description="Individual results"
+    )
+
+    @property
+    def all_required_passed(self) -> bool:
+        """Return True if all required validations passed."""
+        return self.required_failed == 0
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+        use_enum_values=False,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pipeline
+# ---------------------------------------------------------------------------
+
+
+def _normalise_executor_output(raw: Any) -> dict[str, Any]:
+    """Normalise raw executor output to passed/message/data schema.
+
+    Handles various error conditions:
+    - Non-dict output is coerced to a failed result
+    - Missing ``passed`` defaults to False
+    - Missing ``message`` gets a default string
+    """
+    if not isinstance(raw, dict):
+        return {
+            "passed": False,
+            "message": f"Validation returned non-dict output: {raw!r}",
+            "data": None,
+        }
+    passed = raw.get("passed", False)
+    if not isinstance(passed, bool):
+        passed = False
+    message = raw.get("message", "No message provided")
+    if not isinstance(message, str):
+        message = str(message)
+    data = raw.get("data")
+    if data is not None and not isinstance(data, dict):
+        data = {"raw": data}
+    return {"passed": passed, "message": message, "data": data}
+
+
+class ValidationPipeline:
+    """Orchestrates validation execution with deterministic ordering.
+
+    The pipeline sorts validations by ``(resource_name, mode, validation_name)``
+    for deterministic execution, groups them by resource for reporting, and
+    enforces required vs informational semantics.
+
+    Args:
+        commands: List of validation commands to execute.
+        executor: Callable that runs a single validation.
+        max_workers: Maximum concurrent validation threads.
+        read_only_resources: Set of resource IDs that are read-only.
+    """
+
+    def __init__(
+        self,
+        commands: list[ValidationCommand],
+        executor: Callable[[str, dict[str, Any]], dict[str, Any]],
+        max_workers: int = 4,
+        read_only_resources: set[str] | None = None,
+    ) -> None:
+        self._commands = self._sort_commands(commands)
+        self._executor = executor
+        self._max_workers = max_workers
+        self._read_only_resources: set[str] = read_only_resources or set()
+
+    @staticmethod
+    def _sort_commands(
+        commands: list[ValidationCommand],
+    ) -> list[ValidationCommand]:
+        """Sort commands by (resource_name, mode, validation_name)."""
+        return sorted(
+            commands,
+            key=lambda c: (c.resource_name, c.mode.value, c.validation_name),
+        )
+
+    @staticmethod
+    def group_by_resource(
+        results: list[ValidationResult],
+    ) -> dict[str, list[ValidationResult]]:
+        """Group validation results by resource_name."""
+        groups: dict[str, list[ValidationResult]] = defaultdict(list)
+        for result in results:
+            groups[result.resource_name].append(result)
+        return dict(groups)
+
+    def _execute_single(self, cmd: ValidationCommand) -> ValidationResult:
+        """Execute a single validation command with timeout and output capture.
+
+        Uses a daemon thread for timeout enforcement so that timed-out
+        tasks are abandoned immediately without blocking the caller.
+        Stdout/stderr capture is thread-safe via ``_ThreadLocalStream``.
+        """
+        # Check read-only resource guard
+        if cmd.resource_id in self._read_only_resources:
+            logger.warning(
+                "Skipping validation '%s' on read-only resource '%s' (%s)",
+                cmd.validation_name,
+                cmd.resource_name,
+                cmd.resource_id,
+            )
+            return ValidationResult(
+                validation_name=cmd.validation_name,
+                resource_id=cmd.resource_id,
+                resource_name=cmd.resource_name,
+                mode=cmd.mode,
+                passed=True,
+                message=(f"Skipped: resource '{cmd.resource_name}' is read-only"),
+                data=None,
+                duration_ms=0.0,
+                error=None,
+                timed_out=False,
+            )
+
+        start = time.monotonic()
+
+        # Cross-thread communication holders
+        result_holder: list[Any] = []
+        error_holder: list[Exception] = []
+        stdout_holder: list[str] = [""]
+        stderr_holder: list[str] = [""]
+
+        def _run_in_thread() -> None:
+            # Activate per-thread capture if stream wrappers are installed
+            tl_out = sys.stdout
+            tl_err = sys.stderr
+            capture_out = isinstance(tl_out, _ThreadLocalStream)
+            capture_err = isinstance(tl_err, _ThreadLocalStream)
+            if capture_out:
+                tl_out.start_capture()
+            if capture_err:
+                tl_err.start_capture()
+            try:
+                raw = self._executor(cmd.validation_name, cmd.arguments)
+                result_holder.append(raw)
+            except Exception as exc:
+                error_holder.append(exc)
+            finally:
+                if capture_out:
+                    stdout_holder[0] = tl_out.stop_capture()
+                if capture_err:
+                    stderr_holder[0] = tl_err.stop_capture()
+
+        thread = threading.Thread(target=_run_in_thread, daemon=True)
+        thread.start()
+        thread.join(timeout=cmd.timeout_seconds)
+        elapsed_ms = (time.monotonic() - start) * 1000
+
+        if thread.is_alive():
+            # Timeout: daemon thread is abandoned (cleaned up at exit)
+            return ValidationResult(
+                validation_name=cmd.validation_name,
+                resource_id=cmd.resource_id,
+                resource_name=cmd.resource_name,
+                mode=cmd.mode,
+                passed=False,
+                message=(f"Validation timed out after {cmd.timeout_seconds}s"),
+                data=None,
+                duration_ms=elapsed_ms,
+                error=f"TimeoutError: exceeded {cmd.timeout_seconds}s",
+                timed_out=True,
+            )
+
+        if error_holder:
+            exc = error_holder[0]
+            return ValidationResult(
+                validation_name=cmd.validation_name,
+                resource_id=cmd.resource_id,
+                resource_name=cmd.resource_name,
+                mode=cmd.mode,
+                passed=False,
+                message=f"Validation raised {type(exc).__name__}: {exc}",
+                data=None,
+                duration_ms=elapsed_ms,
+                error=f"{type(exc).__name__}: {exc}",
+                timed_out=False,
+            )
+
+        raw_result = result_holder[0] if result_holder else None
+        normalised = _normalise_executor_output(raw_result)
+
+        # Merge captured output into data
+        stdout_text = stdout_holder[0]
+        stderr_text = stderr_holder[0]
+        result_data = normalised["data"]
+        if stdout_text or stderr_text:
+            if result_data is None:
+                result_data = {}
+            if stdout_text:
+                result_data["_captured_stdout"] = stdout_text
+            if stderr_text:
+                result_data["_captured_stderr"] = stderr_text
+
+        return ValidationResult(
+            validation_name=cmd.validation_name,
+            resource_id=cmd.resource_id,
+            resource_name=cmd.resource_name,
+            mode=cmd.mode,
+            passed=normalised["passed"],
+            message=normalised["message"],
+            data=result_data if result_data else None,
+            duration_ms=elapsed_ms,
+            error=None,
+            timed_out=False,
+        )
+
+    def _build_summary(self, results: list[ValidationResult]) -> ValidationSummary:
+        """Build an aggregated summary from individual results."""
+        required_passed = 0
+        required_failed = 0
+        informational_passed = 0
+        informational_failed = 0
+
+        for r in results:
+            if r.mode == ValidationMode.REQUIRED:
+                if r.passed:
+                    required_passed += 1
+                else:
+                    required_failed += 1
+            else:
+                if r.passed:
+                    informational_passed += 1
+                else:
+                    informational_failed += 1
+
+        return ValidationSummary(
+            total=len(results),
+            required_passed=required_passed,
+            required_failed=required_failed,
+            informational_passed=informational_passed,
+            informational_failed=informational_failed,
+            results=results,
+        )
+
+    def run(self) -> ValidationSummary:
+        """Execute all validations and return a summary.
+
+        Validations are executed concurrently up to ``max_workers``.
+        Required failures cause ``all_required_passed`` to be False.
+        Informational failures are logged but do not block.
+
+        Returns:
+            ValidationSummary with aggregated counts and results.
+        """
+        if not self._commands:
+            return ValidationSummary(
+                total=0,
+                required_passed=0,
+                required_failed=0,
+                informational_passed=0,
+                informational_failed=0,
+                results=[],
+            )
+
+        results: list[ValidationResult] = []
+
+        # Install thread-local stream wrappers so _execute_single can
+        # capture per-thread stdout/stderr without global races.
+        orig_stdout = sys.stdout
+        orig_stderr = sys.stderr
+        sys.stdout = _ThreadLocalStream(orig_stdout)
+        sys.stderr = _ThreadLocalStream(orig_stderr)
+        try:
+            with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
+                futures_map = {
+                    pool.submit(self._execute_single, cmd): cmd
+                    for cmd in self._commands
+                }
+                for future in futures_map:
+                    result = future.result()
+                    results.append(result)
+        finally:
+            sys.stdout = orig_stdout
+            sys.stderr = orig_stderr
+
+        # Re-sort results to match command ordering (deterministic)
+        results.sort(
+            key=lambda r: (
+                r.resource_name,
+                r.mode.value,
+                r.validation_name,
+            )
+        )
+
+        # Log informational failures
+        for r in results:
+            if r.mode == ValidationMode.INFORMATIONAL and not r.passed:
+                logger.info(
+                    "Informational validation '%s' on resource '%s' failed: %s",
+                    r.validation_name,
+                    r.resource_name,
+                    r.message,
+                )
+
+        summary = self._build_summary(results)
+
+        if not summary.all_required_passed:
+            logger.warning(
+                "Validation pipeline: %d required validation(s) failed",
+                summary.required_failed,
+            )
+
+        return summary
+
+    def run_for_plan(
+        self,
+        plan_metadata: dict[str, Any] | None = None,
+    ) -> ValidationSummary:
+        """Execute validations and persist summary to plan metadata.
+
+        This integrates the validation pipeline with the plan lifecycle
+        by storing the validation summary in the plan's metadata dict.
+
+        Args:
+            plan_metadata: Mutable dict for plan metadata. If provided,
+                ``validation_summary`` key will be updated with summary data.
+
+        Returns:
+            ValidationSummary with aggregated counts and results.
+        """
+        summary = self.run()
+
+        if plan_metadata is not None:
+            plan_metadata["validation_summary"] = {
+                "total": summary.total,
+                "required_passed": summary.required_passed,
+                "required_failed": summary.required_failed,
+                "informational_passed": summary.informational_passed,
+                "informational_failed": summary.informational_failed,
+                "all_required_passed": summary.all_required_passed,
+                "results": [
+                    {
+                        "validation_name": r.validation_name,
+                        "resource_id": r.resource_id,
+                        "resource_name": r.resource_name,
+                        "mode": r.mode.value,
+                        "passed": r.passed,
+                        "message": r.message,
+                        "data": r.data,
+                        "duration_ms": r.duration_ms,
+                        "error": r.error,
+                        "timed_out": r.timed_out,
+                    }
+                    for r in summary.results
+                ],
+            }
+
+        return summary

vulture_whitelist.py

@@ -234,3 +234,14 @@ normalize_tool_schema_for_provider  # noqa: B018, F821
 route_batch  # noqa: B018, F821
 route_streaming  # noqa: B018, F821
 export_schemas  # noqa: B018, F821
+
+# Validation pipeline — public API for validation orchestration (C6.pipeline)
+ValidationCommand  # noqa: B018, F821
+ValidationResult  # noqa: B018, F821
+ValidationSummary  # noqa: B018, F821
+ValidationPipeline  # noqa: B018, F821
+ValidationExecutor  # noqa: B018, F821
+_normalise_executor_output  # noqa: B018, F821
+group_by_resource  # noqa: B018, F821
+run_for_plan  # noqa: B018, F821
+all_required_passed  # noqa: B018, F821