feat(validation): add validation pipeline and results model

ISSUES CLOSED: #175

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

@@ -1,3 +1,154 @@
+# 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 |
+
+---
+
 # Validation Pipeline — Apply Gating
 
 ## Overview

features/steps/skill_registry_steps.py

@@ -6,7 +6,8 @@ integration layer.
 
 from __future__ import annotations
 
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from behave import given, then, when
 from behave.runner import Context

features/steps/tool_registry_service_fallback_coverage_steps.py

@@ -0,0 +1,429 @@
+"""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,221 @@
+"""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
+
+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

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

@@ -38,6 +38,12 @@ from cleveragents.application.services.validation_apply import (
     ValidationAttachment,
     ValidationRunner,
 )
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    ValidationResult,
+    ValidationSummary,
+)
 
 __all__ = [
     "ApplyValidationGate",
@@ -58,5 +64,9 @@ __all__ = [
     "SkillRegistryService",
     "ToolRegistryService",
     "ValidationAttachment",
+    "ValidationCommand",
+    "ValidationPipeline",
+    "ValidationResult",
     "ValidationRunner",
+    "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

@@ -277,3 +277,14 @@ SubgraphCycleError  # noqa: B018, F821
 MissingNodeError  # noqa: B018, F821
 InvalidEntryExitError  # noqa: B018, F821
 ActorResolver  # 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