Retire the pipeline integration testing subsystem (#435)

docs/adr/0001-product-contract.md

@@ -139,13 +139,16 @@ collection is prohibited.
 
 ## Golden pipelines
 
-Three executable acceptance specifications, all hermetic:
+Four executable acceptance specifications. The first two are hermetic; the last
+two are `live` and skip without a credential:
 
 1. **`basic`** — deterministic local tools, sequential steps, template
    interpolation, typed outputs.
-2. **`control-flow`** — conditional branching and parallel fan-out with
-   dependency ordering, plus a deliberately failing step to verify failure
-   propagation and exit codes.
+2. **`control-flow`** — parallel fan-out and a dependent fan-in join, with
+   template interpolation across step results, plus a deliberately failing step
+   to verify failure propagation and exit codes. Conditional branching and
+   loops are *not* covered by this fixture and are not yet part of the
+   supported contract; see #333 (`on_false` / `on_success`) and #320.
 3. **`live-anthropic`** — the same shape as `basic` but with one real Anthropic
    call. Marked `live`, skipped unless `ANTHROPIC_API_KEY` is set.
 4. **`live-dartmouth`** — the same shape, against a free Dartmouth Chat model.

docs/adr/0002-retire-pipeline-integration-infrastructure.md

@@ -0,0 +1,113 @@
+# ADR 0002: Retire the pipeline integration testing subsystem
+
+**Status:** accepted
+**Date:** 2026-08-01
+**Closes:** #435
+**Relates to:** #430 (retire the frozen competing implementations), #429
+(consolidate the two parallel model-adapter layers), #354 (legacy suite),
+#241 / #104 (`run` and `validate` disagree)
+
+## Decision
+
+Delete, rather than repair:
+
+- `src/orchestrator/testing/pipeline_integration_infrastructure.py` (958 lines)
+- `src/orchestrator/testing/pipeline_integration_demo.py` (544 lines)
+- `tests/integration/test_pipeline_integration_infrastructure.py` (721 lines)
+
+Replace only the intent with six tests against the canonical compiler and
+orchestrator, in `tests/test_pipeline_model_contracts.py`.
+
+## Why not repair it
+
+The module could not construct its own central model. Its constructor passed
+four separate keyword arguments that no core dataclass defines —
+`ModelCapabilities(max_output_tokens=)`, `ModelRequirements(network_access=)`,
+`ModelMetrics(tokens_per_second=)`, `ModelCost(input_cost_per_token=)`.
+Correcting one revealed the next. It was written against an API that has never
+existed, and therefore had never run: **27 tests, permanently red**.
+
+Renaming those arguments was explicitly rejected. It would have made the first
+layer pass while preserving a second testing architecture — its own model,
+provider, validator, scoring system and orchestration facade — inside the
+shipped package, with no user-facing consumer. That is the same
+parallel-implementation problem #430 exists to remove, and repairing it would
+have grown the surface that work has to retire.
+
+Supporting facts, all verified rather than assumed:
+
+- **No external consumers.** `rg` for the module names and for every symbol
+  they export (`PipelineTestModel`, `PipelineTestProvider`,
+  `PipelineIntegrationResult`, …) returns nothing outside the three files
+  themselves. `src/orchestrator/testing/__init__.py` does not import them.
+- **It duplicated working infrastructure.** `tests/test_infrastructure.py`
+  already provides a deterministic model and provider.
+- **It shipped pretend models.** It hardcoded fictional `openai/gpt-4` and
+  `anthropic/claude-*` entries into a package that end users install.
+- **It contradicted project policy.** It was built around `mock_responses`,
+  while CONTRIBUTING states "No mock objects. Ever."
+
+Git history preserves the code. An archive copy inside the tree would add
+noise without reducing risk.
+
+## What replaced it
+
+Six tests, hermetic, using the existing deterministic model:
+
+| Test | Pins |
+|-|-|
+| `test_a_valid_pipeline_compiles` | the canonical compiler accepts a well-formed pipeline |
+| `test_a_malformed_pipeline_is_refused` | validation failure raises, not compiles-to-broken |
+| `test_a_model_pipeline_executes` | a model step actually runs |
+| `test_structured_output_pipeline_returns_an_object` | structured output is a mapping |
+| `test_execution_failure_is_reported_not_swallowed` | a failing step reports failure |
+| `test_a_model_pipeline_compiles` | **xfail(strict)** — documents #241 |
+
+Test-only models and providers stay under `tests/`. They are not product code
+and must not be shipped again.
+
+## Defects this uncovered
+
+Deleting the parallel architecture forced the replacement tests through the
+canonical path, which immediately exposed three real bugs that the retired
+subsystem had been routing around:
+
+1. **`select_model()` results were looked up again.** Five call sites in
+   `hybrid_control_system.py` and `model_based_control_system.py` did
+   `model = get_model(await select_model(...))`. `select_model` already
+   returns a `Model`, so `get_model()`'s `":" in model_name` check raised
+   `TypeError: argument of type 'Model' is not a container`. Every
+   registry-selected `action: generate` step failed.
+2. **`create_test_orchestrator()` built the wrong registry.** Two classes
+   share the name `ModelRegistry`; only `models.registry` has
+   `register_provider`, and only `models.model_registry` has
+   `can_provide_models`, which `Orchestrator.__init__` calls. The helper used
+   the first, so it raised on every call and could not construct an
+   orchestrator at all. (The duplicate-registry split itself is #429.)
+3. **The deterministic model advertised the wrong task names.** It offered
+   `text-generation`, while the control system asks for `generate` and real
+   providers advertise `generate` — so it was ineligible for every generate
+   step and selection failed with `NoEligibleModelsError`.
+
+None of these were visible while a second testing stack stood in for the real
+one. That is the strongest argument for the deletion: the parallel
+architecture was not merely unused, it was hiding the state of the supported
+path.
+
+## Consequences
+
+- 2,223 lines removed; 6 tests added, in the **blocking** layer rather than a
+  permanently red one.
+- 27 tests were **retired, not fixed**. #354's backlog shrinks by that amount
+  for that reason, and must not be read as progress on the legacy suite.
+- `validate` still rejects model pipelines that `run` accepts. That divergence
+  is now pinned by a strict xfail instead of being absorbed by a bespoke
+  validator, and closing #241 will make that test flip to a visible failure,
+  prompting its removal.
+
+## If systematic example-pipeline validation is wanted again
+
+Build it around the canonical `orchestrator validate` path so it returns the
+same diagnostics as the CLI. It must not introduce another model, provider,
+validator, scoring system, or orchestration facade — that is what was just
+removed.

src/orchestrator/control_systems/hybrid_control_system.py

@@ -971,13 +971,13 @@ async def _handle_prompt_optimization_placeholder(self, task: Task, context: Dic
         goal = task.parameters.get("optimization_goal", "quality")
 
         # Use the model to enhance the prompt
-        model_name = await self.model_registry.select_model({"tasks": ["generate"]})
-        if not model_name:
+        # select_model returns a Model, so there is nothing further to look up.
+        model = await self.model_registry.select_model({"tasks": ["generate"]})
+        if not model:
             return {
                 "success": False,
                 "error": "No model available for prompt optimization"
             }
-        model = self.model_registry.get_model(model_name)
 
         optimization_prompt = f"""Optimize this prompt for {task_type}:
 Original prompt: "{prompt}"
@@ -1137,9 +1137,9 @@ async def _handle_task_delegation(self, task: Task, context: Dict[str, Any]) ->
                         if match:
                             # Use a model to resolve the AUTO tag
                             auto_prompt = match.group(1)
-                            model_name = await self.model_registry.select_model({"tasks": ["generate"]})
-                            if model_name:
-                                model = self.model_registry.get_model(model_name)
+                            # select_model returns a Model already.
+                            model = await self.model_registry.select_model({"tasks": ["generate"]})
+                            if model:
                                 response = await model.generate(
                                     f"{auto_prompt}\nBased on the task: '{params.get('task', '')}'\nRespond with only: simple, moderate, or complex"
                                 )
@@ -1208,8 +1208,8 @@ async def _handle_analyze_text(self, task: Task, context: Dict[str, Any]) -> Any
                     "tasks": ["analyze", "generate"],
                     "context_window": len(prompt.encode()) // 4  # Rough token estimate
                 }
-                model_name = await self.model_registry.select_model(requirements)
-                model = self.model_registry.get_model(model_name)
+                # select_model returns a Model already.
+                model = await self.model_registry.select_model(requirements)
             else:
                 # Get specific model
                 model = self.model_registry.get_model(model_spec)
@@ -1318,13 +1318,16 @@ async def _handle_generate_text(self, task: Task, context: Dict[str, Any]) -> An
             elif isinstance(model_spec, str) and model_spec.startswith("<AUTO"):
                 # AUTO tag - let model registry select
                 requirements = {"tasks": ["generate"], "context_window": len(prompt) // 4}
-                model_name = await self.model_registry.select_model(requirements)
-                model = self.model_registry.get_model(model_name)
+                # select_model already returns a Model, not a name. Feeding it
+                # back into get_model() raised
+                #   TypeError: argument of type 'Model' is not a container
+                # from get_model()'s `":" in model_name` check, so every
+                # registry-selected `action: generate` step failed.
+                model = await self.model_registry.select_model(requirements)
         else:
             # No model specified, select appropriate model
             requirements = {"tasks": ["generate"], "context_window": len(prompt) // 4}
-            model_name = await self.model_registry.select_model(requirements)
-            model = self.model_registry.get_model(model_name)
+            model = await self.model_registry.select_model(requirements)
 
         if not model:
             return {

src/orchestrator/control_systems/model_based_control_system.py

@@ -15,6 +15,14 @@
 
 logger = logging.getLogger(__name__)
 
+# Every other action in this project is spelled with underscores
+# (`generate_text`, `analyze_text`, `evaluate_condition`, `loop_complete`).
+# `generate-structured` was the lone hyphenated one, so `generate_structured`
+# fell through to the natural-language branch below, which turns an unknown
+# action into a *prompt*: the step returned a sentence instead of an object and
+# still reported success. Both spellings dispatch here.
+STRUCTURED_ACTIONS = frozenset({"generate-structured", "generate_structured"})
+
 
 class ModelBasedControlSystem(ControlSystem):
     """Control system that executes tasks using real AI models."""
@@ -39,6 +47,7 @@ def __init__(
                     "supported_actions": [
                         "generate",
                         "generate_text",
+                        "generate_structured",
                         "analyze",
                         "transform",
                         "execute",
@@ -178,9 +187,10 @@ async def _execute_task_impl(self, task: Task, context: Dict[str, Any]) -> Any:
         # If still no model, select one based on task requirements
         if not model:
             requirements = self._get_task_requirements(task)
-            model_name = await self.model_registry.select_model(requirements)
-            if model_name:
-                model = await self.model_registry.get_model(model_name)
+            # select_model returns a Model, not a name -- passing its result
+            # back into get_model() raised TypeError from get_model()'s
+            # `":" in model_name` check.
+            model = await self.model_registry.select_model(requirements)
 
         # Extract the actual action/prompt from the task
         if task.action in ["generate", "generate_text"] and task.parameters.get(
@@ -228,11 +238,11 @@ async def _execute_task_impl(self, task: Task, context: Dict[str, Any]) -> Any:
 
         try:
             # Check if this is a structured generation task
-            if task.action == "generate-structured":
+            if task.action in STRUCTURED_ACTIONS:
                 # Use structured generation
                 if not task.parameters or "schema" not in task.parameters:
                     raise ValueError(
-                        f"Task '{task.id}' with action 'generate-structured' requires a 'schema' parameter"
+                        f"Task '{task.id}' with action '{task.action}' requires a 'schema' parameter"
                     )
 
                 temperature = (
@@ -442,8 +452,8 @@ def _build_prompt(
 
     def _parse_result(self, result: Any, task: Task) -> Any:
         """Parse model result based on expected format."""
-        # For generate-structured actions, the result should already be a structured object
-        if task.action == "generate-structured":
+        # For structured actions, the result should already be a structured object
+        if task.action in STRUCTURED_ACTIONS:
             # If it's still a string, try to parse it as JSON
             if isinstance(result, str):
                 try: