diff --git a/docs/guides/pipeline.md b/docs/guides/pipeline.md index e8e86107..523dca8c 100644 --- a/docs/guides/pipeline.md +++ b/docs/guides/pipeline.md @@ -241,7 +241,18 @@ engine = ExecutionEngine(max_workers=2, retry_on_failure=True) result = engine.execute_pipeline(pipeline) ``` -`set_parallelism(n)` tells the engine how many steps it may run simultaneously. The topological sort guarantees that only steps whose dependencies are all completed are eligible for concurrent execution — you cannot accidentally run a step before its inputs are ready. +`set_parallelism(n)` tells the engine how many steps it may run simultaneously; `n` must be a positive integer. The topological sort guarantees that only steps whose dependencies are all completed are eligible for concurrent execution — you cannot accidentally run a step before its inputs are ready. The effective concurrency is capped at `min(n, max_workers)`, so the engine's `max_workers` setting remains a hard resource ceiling. + +Concurrency is opt-in per step. A dependency layer only runs in parallel when every step in that layer is marked `parallel_safe`, the layer has more than one step, and the data flowing into the layer is a dict: + +```python +builder.add_step("ner", "ner_extract", parallel_safe=True, confidence_threshold=0.75) +builder.add_step("triplets", "triplet_extract", parallel_safe=True, include_temporal=True) +``` + +If any step in a layer is not marked `parallel_safe`, or if a step runs in delta mode, the entire layer falls back to sequential execution — parallelism never silently bypasses a step that was not declared safe. `parallel_safe` is a control field: like `dependencies`, it is consumed by the builder and never reaches your handler's config. + +Parallel-safe handlers must return a dict. Each step in a parallel layer receives an isolated deep copy of the layer's input, so steps cannot see each other's mutations. The per-step results are merged key by key in step declaration order: a key written by one step is added to the merged output, a key written by several steps with equal values is kept, and two steps writing different values for the same key fail the pipeline with a `ProcessingError` naming the conflicting key and both steps. Handlers that touch shared mutable resources — database connections, in-memory stores, global caches — should not be marked `parallel_safe`. ## Common Pitfalls diff --git a/semantica/pipeline/execution_engine.py b/semantica/pipeline/execution_engine.py index 63f382ea..e0067f99 100644 --- a/semantica/pipeline/execution_engine.py +++ b/semantica/pipeline/execution_engine.py @@ -32,8 +32,10 @@ Author: Semantica Contributors License: MIT """ +import copy import threading import time +from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import datetime from enum import Enum @@ -59,6 +61,14 @@ class PipelineStatus(Enum): STOPPED = "stopped" +class _ParallelResultContractError(ProcessingError): + """Raised when a parallel step violates the dict-result contract. + + Contract violations are deterministic: re-running the handler cannot + change its return type, so the retry loop must be skipped entirely. + """ + + @dataclass class ExecutionResult: """Pipeline execution result.""" @@ -242,11 +252,67 @@ class ExecutionEngine: return ExecutionResult(success=False, output=None, errors=[str(e)]) def _execute_steps(self, pipeline: Pipeline, data: Any, **options) -> Any: - """Execute pipeline steps.""" - # Sort steps by dependencies (topological sort) - sorted_steps = self._topological_sort(pipeline.steps) + """ + Execute pipeline steps. + + Steps are grouped into dependency layers. When a pipeline is + configured with parallelism > 1, a layer whose steps are all marked + ``parallel_safe`` and whose input is a dict is executed concurrently + (bounded by the effective parallelism); every other layer runs + sequentially, preserving the default serial behaviour. + """ + effective_parallelism = self._get_effective_parallelism(pipeline) + + if effective_parallelism <= 1: + return self._execute_steps_sequential(pipeline, data, **options) + + layers = self._group_steps_by_dependency_level(pipeline.steps) + + current_data = data + for layer in layers: + if self.pipeline_status.get(pipeline.name) == PipelineStatus.STOPPED: + break + + # Wait if paused + while self.pipeline_status.get(pipeline.name) == PipelineStatus.PAUSED: + time.sleep(0.1) + + if self._can_run_layer_in_parallel(layer, current_data): + merged = self._execute_parallel_group( + layer, + current_data, + effective_parallelism, + pipeline_name=pipeline.name, + **options, + ) + if merged is None: + # Input isolation failed before any handler started; + # run this layer sequentially instead. + current_data = self._execute_steps_sequential( + pipeline, current_data, steps=layer, **options + ) + else: + current_data = merged + else: + current_data = self._execute_steps_sequential( + pipeline, current_data, steps=layer, **options + ) + + return current_data + + def _execute_steps_sequential( + self, + pipeline: Pipeline, + data: Any, + steps: Optional[List[PipelineStep]] = None, + **options, + ) -> Any: + """Execute steps sequentially following dependency order.""" + if steps is None: + sorted_steps = self._topological_sort(pipeline.steps) + else: + sorted_steps = list(steps) - # Execute steps current_data = data total_steps = len(sorted_steps) @@ -258,76 +324,320 @@ class ExecutionEngine: while self.pipeline_status.get(pipeline.name) == PipelineStatus.PAUSED: time.sleep(0.1) - # Track step execution - step_tracking_id = self.progress_tracker.start_tracking( - module="pipeline", - submodule=step.step_type or step.name, - message=f"Step {step_idx + 1}/{total_steps}: {step.name}", + current_data = self._execute_step_with_retries( + step, + current_data, + step_label=f"Step {step_idx + 1}/{total_steps}: {step.name}", + pipeline_name=pipeline.name, + **options, ) - try: - # Execute step - step.status = StepStatus.RUNNING - step_result = self._execute_step(step, current_data, **options) - step.status = StepStatus.COMPLETED - step.result = step_result - current_data = step_result + return current_data + def _execute_step_with_retries( + self, + step: PipelineStep, + data: Any, + step_label: Optional[str] = None, + pipeline_name: Optional[str] = None, + require_dict_result: bool = False, + **options, + ) -> Any: + """ + Execute a single step with retry handling. + + Shared by the sequential and parallel execution paths so that retry + policies, step status tracking and progress reporting behave + identically. Returns the step result, or raises the final error + after retries are exhausted. + + When ``require_dict_result`` is set (parallel layers), a handler + returning a non-dict raises ProcessingError before the step is + marked completed, so status and progress reporting stay consistent. + Such contract violations are never retried: the handler's return + type cannot change between attempts. + """ + step_tracking_id = self.progress_tracker.start_tracking( + module="pipeline", + # Pipeline identity + step name keep tracking IDs unique so + # concurrent steps of the same step_type cannot overwrite each + # other's progress records. + submodule=( + f"{pipeline_name or 'pipeline'}:" + f"{step.step_type or 'step'}:{step.name}" + ), + message=step_label or f"Executing step: {step.name}", + ) + + try: + step.status = StepStatus.RUNNING + step_result = self._execute_step(step, data, **options) + if require_dict_result and not isinstance(step_result, dict): + raise _ParallelResultContractError( + f"Step '{step.name}' is marked parallel_safe and must " + f"return a dict so parallel results can be merged, got " + f"{type(step_result).__name__}" + ) + step.status = StepStatus.COMPLETED + step.result = step_result + + self.progress_tracker.stop_tracking( + step_tracking_id, + status="completed", + message=f"Completed step: {step.name}", + ) + return step_result + + except Exception as e: + step.status = StepStatus.FAILED + step.error = e + + # Contract violations are deterministic failures: re-running + # the handler cannot change its return type, so never consult + # the retry policy for them. + if isinstance(e, _ParallelResultContractError): + self.progress_tracker.stop_tracking( + step_tracking_id, status="failed", message=str(e) + ) + raise + + # Retry loop respecting max_retries from the policy + retry_policy = self.failure_handler.get_retry_policy(step.step_type) + max_retries = retry_policy.max_retries if retry_policy else 0 + retry_count = 0 + success = False + + while retry_count < max_retries: + recovery_result = self.failure_handler.handle_step_failure(step, e) + if not recovery_result.get("retry", False): + break + retry_delay = recovery_result.get("retry_delay", 0.0) + if retry_delay > 0: + time.sleep(retry_delay) + self.progress_tracker.update_tracking( + step_tracking_id, + status="running", + message=f"Retrying step: {step.name} (attempt {retry_count + 1})", + ) + step.status = StepStatus.RUNNING + try: + step_result = self._execute_step(step, data, **options) + if require_dict_result and not isinstance( + step_result, dict + ): + raise _ParallelResultContractError( + f"Step '{step.name}' is marked parallel_safe " + f"and must return a dict so parallel results " + f"can be merged, got " + f"{type(step_result).__name__}" + ) + step.status = StepStatus.COMPLETED + step.result = step_result + success = True + break + except Exception as retry_e: + step.status = StepStatus.FAILED + step.error = retry_e + e = retry_e + retry_count += 1 + + if success: self.progress_tracker.stop_tracking( step_tracking_id, status="completed", - message=f"Completed step: {step.name}", + message=f"Retry successful: {step.name}", ) + return step_result + else: + self.progress_tracker.stop_tracking( + step_tracking_id, status="failed", message=str(e) + ) + raise e - except Exception as e: - step.status = StepStatus.FAILED - step.error = e + def _get_effective_parallelism(self, pipeline: Pipeline) -> int: + """Return the parallelism actually used for this pipeline.""" + configured = pipeline.config.get("parallelism", 1) + if not isinstance(configured, int) or configured <= 0: + return 1 + return min(configured, self.parallelism_manager.max_workers) - # Retry loop respecting max_retries from the policy - retry_policy = self.failure_handler.get_retry_policy(step.step_type) - max_retries = retry_policy.max_retries if retry_policy else 0 - retry_count = 0 - success = False + def _group_steps_by_dependency_level( + self, steps: List[PipelineStep] + ) -> List[List[PipelineStep]]: + """ + Group steps into dependency layers, preserving declaration order. - while retry_count < max_retries: - recovery_result = self.failure_handler.handle_step_failure(step, e) - if not recovery_result.get("retry", False): - break - retry_delay = recovery_result.get("retry_delay", 0.0) - if retry_delay > 0: - time.sleep(retry_delay) - self.progress_tracker.update_tracking( - step_tracking_id, - status="running", - message=f"Retrying step: {step.name} (attempt {retry_count + 1})", + Circular or unknown dependencies raise ValidationError so the + parallel path fails deterministically, matching the validation + behaviour of the serial topological sort. + """ + step_map = {step.name: step for step in steps} + levels: Dict[str, int] = {} + visiting: set = set() + + for step in steps: + for dep in step.dependencies: + if dep not in step_map: + raise ValidationError( + f"Step '{step.name}' depends on unknown step '{dep}'" ) - step.status = StepStatus.RUNNING - try: - step_result = self._execute_step(step, current_data, **options) - step.status = StepStatus.COMPLETED - step.result = step_result - current_data = step_result - success = True - break - except Exception as retry_e: - step.status = StepStatus.FAILED - step.error = retry_e - e = retry_e - retry_count += 1 - if success: - self.progress_tracker.stop_tracking( - step_tracking_id, - status="completed", - message=f"Retry successful: {step.name}", - ) + def get_level(step_name: str) -> int: + if step_name in levels: + return levels[step_name] + if step_name in visiting: + raise ValidationError( + "Circular dependency detected in pipeline " + f"(cycle passes through step '{step_name}')" + ) + visiting.add(step_name) + step = step_map[step_name] + if not step.dependencies: + level = 0 + else: + level = max(get_level(dep) for dep in step.dependencies) + 1 + visiting.discard(step_name) + levels[step_name] = level + return level + + for step in steps: + get_level(step.name) + + grouped: Dict[int, List[PipelineStep]] = {} + for step in steps: + grouped.setdefault(levels[step.name], []).append(step) + return [grouped[level] for level in sorted(grouped)] + + def _can_run_layer_in_parallel(self, layer: List[PipelineStep], data: Any) -> bool: + """Check whether a dependency layer can safely run in parallel.""" + if len(layer) <= 1: + return False + if not isinstance(data, dict): + return False + for step in layer: + # Strict boolean check: truthy non-bool values (e.g. the + # string "false") must never opt a step into concurrency. + if getattr(step, "parallel_safe", False) is not True: + return False + if getattr(step, "delta_mode", False): + return False + return True + + def _execute_parallel_group( + self, + layer: List[PipelineStep], + data: Any, + effective_parallelism: int, + pipeline_name: Optional[str] = None, + **options, + ) -> Optional[Any]: + """ + Execute a dependency layer concurrently. + + Per-step inputs are deep-copied before any handler starts so that + parallel steps do not share mutable state. Returns the merged dict + result, or None when input isolation failed (before any handler + ran) and the layer should fall back to sequential execution. + """ + # Isolate per-step inputs before starting any handler + try: + step_inputs = {step.name: copy.deepcopy(data) for step in layer} + except Exception as e: + self.logger.warning( + f"Falling back to sequential execution: input for parallel " + f"layer could not be isolated ({e})" + ) + return None + + step_results: Dict[str, Any] = {} + failure: Optional[BaseException] = None + max_workers = min(effective_parallelism, len(layer)) + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = { + executor.submit( + self._execute_step_with_retries, + step, + step_inputs[step.name], + # Validate the dict-result contract inside the shared + # lifecycle path, before completion is reported. + pipeline_name=pipeline_name, + require_dict_result=True, + **options, + ): step + for step in layer + } + + for future in as_completed(futures): + step = futures[future] + try: + step_result = future.result() + except Exception as e: + if failure is None: + failure = e + # Cancel steps that have not started yet + for pending in futures: + pending.cancel() else: - self.progress_tracker.stop_tracking( - step_tracking_id, status="failed", message=str(e) + step_results[step.name] = step_result + + if failure is not None: + raise failure + + return self._merge_parallel_results(data, layer, step_results) + + def _merge_parallel_results( + self, + base: Dict[str, Any], + layer: List[PipelineStep], + step_results: Dict[str, Any], + ) -> Dict[str, Any]: + """ + Merge the results of a parallel layer into a single dict. + + Steps are processed in declaration order (never by completion + order). Keys whose values are unchanged from the shared base input + are skipped (handlers commonly return complete dicts such as + ``{**data, ...}``); only added or changed keys count as branch + writes. Keys written with equal values by multiple steps are + allowed; conflicting values for the same key raise a + ProcessingError naming the key and both steps. Ambiguous equality + comparisons count as changed, never as unchanged. + """ + merged = dict(base) + key_sources: Dict[str, str] = {} + + for step in layer: + step_result = step_results.get(step.name) + if step_result is None: + continue + for key, value in step_result.items(): + if key in base and self._values_equal(base[key], value): + # Echo of the shared input: not a branch write, so it + # cannot conflict with a sibling that changes the key. + continue + if key in key_sources and not self._values_equal( + merged.get(key), value + ): + raise ProcessingError( + f"Conflicting values for key '{key}' in parallel step " + f"results: step '{step.name}' produced {value!r}, but " + f"step '{key_sources[key]}' previously produced " + f"{merged.get(key)!r}" ) - raise e + key_sources[key] = step.name + merged[key] = value + + return merged + + @staticmethod + def _values_equal(left: Any, right: Any) -> bool: + """Safely compare two values; ambiguous comparisons count as conflicts.""" + try: + return bool(left == right) + except (TypeError, ValueError): + return False - return current_data def _execute_step(self, step: PipelineStep, data: Any, **options) -> Any: """ diff --git a/semantica/pipeline/pipeline_builder.py b/semantica/pipeline/pipeline_builder.py index b38dc23c..bb3ed84c 100644 --- a/semantica/pipeline/pipeline_builder.py +++ b/semantica/pipeline/pipeline_builder.py @@ -67,6 +67,7 @@ class PipelineStep: delta_mode: bool = False base_version_id: Optional[str] = None target_version_id: Optional[str] = None + parallel_safe: bool = False @dataclass @@ -133,6 +134,12 @@ class PipelineBuilder: target_version_id = config.pop("target_version_id", None) dependencies = config.pop("dependencies", []) handler = config.pop("handler", None) + parallel_safe = config.pop("parallel_safe", False) + if not isinstance(parallel_safe, bool): + raise ValidationError( + f"parallel_safe must be a boolean, got " + f"{type(parallel_safe).__name__} for step '{step_name}'" + ) if handler is None: handler = self.step_registry.get(step_type) @@ -145,6 +152,7 @@ class PipelineBuilder: delta_mode = delta_mode, base_version_id=base_version_id, target_version_id=target_version_id, + parallel_safe=parallel_safe, ) self.steps.append(step) @@ -186,6 +194,14 @@ class PipelineBuilder: Returns: Self for method chaining """ + if ( + isinstance(level, bool) + or not isinstance(level, int) + or level <= 0 + ): + raise ValidationError( + f"Parallelism level must be a positive integer, got {level!r}" + ) self.pipeline_config["parallelism"] = level return self @@ -289,6 +305,16 @@ class PipelineBuilder: step.target_version_id = step_config.get( "target_version_id", step.target_version_id ) + raw_parallel_safe = step_config.get( + "parallel_safe", step.parallel_safe + ) + if not isinstance(raw_parallel_safe, bool): + raise ValidationError( + "parallel_safe must be a boolean for step " + f"'{step_name}', got " + f"{type(raw_parallel_safe).__name__}" + ) + step.parallel_safe = raw_parallel_safe # Set parallelism if specified if "parallelism" in pipeline_config: @@ -344,6 +370,7 @@ class PipelineBuilder: "type": step.step_type, "config": step.config, "dependencies": step.dependencies, + "parallel_safe": step.parallel_safe, } for step in self.steps ], @@ -425,6 +452,7 @@ class PipelineSerializer: "delta_mode", "base_version_id", "target_version_id", + "parallel_safe", } pipeline_data = { "name": pipeline.name, @@ -441,6 +469,7 @@ class PipelineSerializer: "delta_mode": getattr(step, "delta_mode", False), "base_version_id": getattr(step, "base_version_id", None), "target_version_id": getattr(step, "target_version_id", None), + "parallel_safe": getattr(step, "parallel_safe", False), } for step in pipeline.steps ], @@ -488,6 +517,12 @@ class PipelineSerializer: sanitized_steps.append(sanitized_step) pipeline_data["steps"] = sanitized_steps + # Reapply pipeline-level config (e.g. parallelism) at the top level + # so build_pipeline picks it up + serialized_config = pipeline_data.pop("config", None) or {} + for key, value in serialized_config.items(): + pipeline_data.setdefault(key, value) + # Reconstruct pipeline builder = PipelineBuilder(**self.config) pipeline = builder.build_pipeline(pipeline_data, **options) diff --git a/tests/pipeline/test_pipeline_parallel.py b/tests/pipeline/test_pipeline_parallel.py new file mode 100644 index 00000000..a66ebee8 --- /dev/null +++ b/tests/pipeline/test_pipeline_parallel.py @@ -0,0 +1,785 @@ +"""Focused tests for pipeline parallel execution (issue #1223). + +Covers: + - ``PipelineBuilder.set_parallelism()`` validation + - dependency-layer parallel execution gated on ``parallel_safe`` + - input isolation via deep copies + - incremental dict merging of parallel outputs + - failure handling and cancellation semantics + - ``parallel_safe`` not leaking into handler kwargs + - ``parallel_safe`` surviving serialization round-trips +""" + +import threading +import time +import unittest +from unittest.mock import MagicMock, patch + +from semantica.pipeline.execution_engine import ExecutionEngine +from semantica.pipeline.failure_handler import RetryPolicy, RetryStrategy +from semantica.pipeline.pipeline_builder import ( + Pipeline, + PipelineBuilder, + PipelineSerializer, + PipelineStep, + StepStatus, +) +from semantica.utils.exceptions import ProcessingError, ValidationError + + +class ConcurrencyProbe: + """Thread-safe tracker of handler invocations and concurrency level.""" + + def __init__(self): + self._lock = threading.Lock() + self.active = 0 + self.max_active = 0 + self.calls = 0 + + def __enter__(self): + with self._lock: + self.active += 1 + self.calls += 1 + self.max_active = max(self.max_active, self.active) + return self + + def __exit__(self, *exc): + with self._lock: + self.active -= 1 + return False + + +def branch_handler(probe, key, value=True, delay=0.0): + """Build a handler that records concurrency and adds one output key.""" + + def handler(data, **kwargs): + with probe: + if delay: + time.sleep(delay) + return {**data, key: value} + + return handler + + +class Undeepcopyable: + """Object whose deepcopy always fails.""" + + def __deepcopy__(self, memo): + raise TypeError("cannot deepcopy this object") + + +class TestSetParallelismValidation(unittest.TestCase): + """set_parallelism() must only accept positive integers.""" + + def test_rejects_invalid_levels(self): + builder = PipelineBuilder() + for invalid in (0, -1, 1.5, True, False, "2", None): + with self.assertRaises(ValidationError): + builder.set_parallelism(invalid) + + def test_accepts_positive_integers(self): + builder = PipelineBuilder() + builder.set_parallelism(4) + self.assertEqual(builder.pipeline_config["parallelism"], 4) + + +class TestPipelineParallelExecution(unittest.TestCase): + + def setUp(self): + self.mock_tracker_patcher = patch( + "semantica.utils.progress_tracker.get_progress_tracker" + ) + self.mock_get_tracker = self.mock_tracker_patcher.start() + self.mock_get_tracker.return_value = MagicMock() + + def tearDown(self): + self.mock_tracker_patcher.stop() + + def _build(self, steps, parallelism=None, name="parallel_pipeline"): + """steps: list of (step_name, step_type, handler, parallel_safe).""" + builder = PipelineBuilder() + for step_name, step_type, handler, parallel_safe in steps: + builder.add_step( + step_name, step_type, handler=handler, parallel_safe=parallel_safe + ) + if parallelism is not None: + builder.set_parallelism(parallelism) + return builder.build(name) + + def test_unconfigured_pipeline_stays_serial(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), True), + ] + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + + def test_parallelism_one_stays_serial(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), True), + ], + parallelism=1, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + + def test_parallel_safe_steps_overlap_with_barrier(self): + barrier = threading.Barrier(2) + done = {"a": False, "b": False} + + def handler(key): + def inner(data, **kwargs): + barrier.wait(timeout=5) + done[key] = True + return {**data, key: True} + + return inner + + pipeline = self._build( + [ + ("a", "branch", handler("a"), True), + ("b", "branch", handler("b"), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success, msg=str(result.errors)) + self.assertTrue(done["a"]) + self.assertTrue(done["b"]) + + def test_active_workers_do_not_exceed_parallelism(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a", delay=0.15), True), + ("b", "branch", branch_handler(probe, "b", delay=0.15), True), + ("c", "branch", branch_handler(probe, "c", delay=0.15), True), + ("d", "branch", branch_handler(probe, "d", delay=0.15), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertLessEqual(probe.max_active, 2) + self.assertGreaterEqual(probe.max_active, 2) + + def test_unmarked_steps_stay_serial(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), False), + ("b", "branch", branch_handler(probe, "b"), False), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + + def test_layer_with_one_unsafe_step_is_serial(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), False), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + + def test_pure_dependency_chain_stays_serial(self): + probe = ConcurrencyProbe() + + def chain_handler(key): + def inner(data, **kwargs): + with probe: + return {**data, key: True} + + return inner + + builder = PipelineBuilder() + builder.add_step( + "a", "branch", handler=chain_handler("a"), parallel_safe=True + ) + builder.add_step( + "b", + "branch", + handler=chain_handler("b"), + parallel_safe=True, + dependencies=["a"], + ) + builder.add_step( + "c", + "branch", + handler=chain_handler("c"), + parallel_safe=True, + dependencies=["b"], + ) + builder.set_parallelism(4) + pipeline = builder.build("chain_pipeline") + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + self.assertEqual(result.output, {"text": "hi", "a": True, "b": True, "c": True}) + + def test_non_dict_input_falls_back_to_serial(self): + probe = ConcurrencyProbe() + + def passthrough(data, **kwargs): + with probe: + return data + + pipeline = self._build( + [ + ("a", "branch", passthrough, True), + ("b", "branch", passthrough, True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline( + pipeline, data="plain-text-input" + ) + + self.assertTrue(result.success) + self.assertEqual(probe.calls, 2) + self.assertEqual(probe.max_active, 1) + self.assertEqual(result.output, "plain-text-input") + + def test_deepcopy_failure_falls_back_to_serial(self): + probe = ConcurrencyProbe() + + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline( + pipeline, data={"text": "hi", "obj": Undeepcopyable()} + ) + + self.assertTrue(result.success, msg=str(result.errors)) + self.assertEqual(probe.calls, 2) + self.assertEqual(probe.max_active, 1) + self.assertIn("a", result.output) + self.assertIn("b", result.output) + + def test_parallel_results_merge_different_fields(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "entities", ["Alice"]), True), + ("b", "branch", branch_handler(probe, "triplets", [(1, 2)]), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline( + pipeline, data={"text": "Alice works at Acme"} + ) + + self.assertTrue(result.success) + self.assertEqual( + result.output, + { + "text": "Alice works at Acme", + "entities": ["Alice"], + "triplets": [(1, 2)], + }, + ) + + def test_parallel_results_same_key_same_value_allowed(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "shared", [1, 2]), True), + ("b", "branch", branch_handler(probe, "shared", [1, 2]), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success, msg=str(result.errors)) + self.assertEqual(result.output["shared"], [1, 2]) + + def test_parallel_results_same_key_different_values_raises(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "shared", 1), True), + ("b", "branch", branch_handler(probe, "shared", 2), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + error_text = result.errors[0] if result.errors else "" + self.assertIn("shared", error_text) + self.assertIn("a", error_text) + self.assertIn("b", error_text) + + def test_parallel_handler_non_dict_return_fails(self): + probe = ConcurrencyProbe() + + def non_dict_handler(data, **kwargs): + with probe: + return "not-a-dict" + + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", non_dict_handler, True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + error_text = result.errors[0] if result.errors else "" + self.assertIn("b", error_text) + self.assertIn("str", error_text) + + def test_non_dict_handler_executed_only_once(self): + probe = ConcurrencyProbe() + + def non_dict_handler(data, **kwargs): + with probe: + return "not-a-dict" + + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", non_dict_handler, True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + self.assertEqual(probe.calls, 2) # one invocation per handler, no re-runs + + def test_parallel_step_retry_succeeds(self): + engine = ExecutionEngine() + engine.failure_handler.set_retry_policy( + "flaky", + RetryPolicy( + max_retries=2, initial_delay=0.0, strategy=RetryStrategy.FIXED + ), + ) + + probe = ConcurrencyProbe() + attempts = {"flaky": 0} + + def flaky_handler(data, **kwargs): + with probe: + attempts["flaky"] += 1 + if attempts["flaky"] == 1: + raise RuntimeError("transient failure") + return {**data, "flaky": True} + + pipeline = self._build( + [ + ("a", "flaky", flaky_handler, True), + ("b", "branch", branch_handler(probe, "b"), True), + ], + parallelism=2, + ) + + result = engine.execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success, msg=str(result.errors)) + self.assertEqual(attempts["flaky"], 2) + + def test_failed_branch_skips_downstream_layer(self): + engine = ExecutionEngine() + engine.failure_handler.set_retry_policy( + "always_fail", RetryPolicy(max_retries=0) + ) + + probe = ConcurrencyProbe() + + def failing_handler(data, **kwargs): + with probe: + raise RuntimeError("permanent failure") + + downstream_calls = {"c": 0} + + def downstream_handler(data, **kwargs): + downstream_calls["c"] += 1 + return {**data, "c": True} + + builder = PipelineBuilder() + builder.add_step("a", "always_fail", handler=failing_handler, parallel_safe=True) + builder.add_step("b", "branch", handler=branch_handler(probe, "b"), parallel_safe=True) + builder.add_step( + "c", "branch", handler=downstream_handler, dependencies=["a", "b"] + ) + builder.set_parallelism(2) + pipeline = builder.build("failure_pipeline") + + result = engine.execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + self.assertEqual(downstream_calls["c"], 0) + + failed_step = next(s for s in pipeline.steps if s.name == "a") + self.assertEqual(failed_step.status, StepStatus.FAILED) + self.assertIsNotNone(failed_step.error) + + def test_parallel_safe_not_passed_to_handler_kwargs(self): + received_kwargs = {} + + def capturing_handler(data, **kwargs): + received_kwargs.update(kwargs) + return data + + builder = PipelineBuilder() + builder.add_step( + "a", "branch", handler=capturing_handler, parallel_safe=True, batch_size=2 + ) + pipeline = builder.build("kwargs_pipeline") + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertIn("batch_size", received_kwargs) + self.assertNotIn("parallel_safe", received_kwargs) + self.assertNotIn("parallel_safe", pipeline.steps[0].config) + + def test_unchanged_echoed_key_does_not_conflict(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + # "a" echoes the base value of "shared" (unchanged), + # "b" legitimately changes it: no false conflict. + ("a", "branch", branch_handler(probe, "shared", 1), True), + ("b", "branch", branch_handler(probe, "shared", 2), True), + ], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline( + pipeline, data={"text": "hi", "shared": 1} + ) + + self.assertTrue(result.success, msg=str(result.errors)) + self.assertEqual(result.output["shared"], 2) + + def test_ambiguous_equality_counts_as_changed(self): + sentinel = Uncomparable() + + def echo(data, **kwargs): + return {**data, "shared": data["shared"]} + + def change(data, **kwargs): + return {**data, "shared": "changed"} + + pipeline = self._build( + [("a", "branch", echo, True), ("b", "branch", change, True)], + parallelism=2, + ) + + result = ExecutionEngine().execute_pipeline( + pipeline, data={"text": "hi", "shared": sentinel} + ) + + # Ambiguous equality must not be treated as unchanged, so both + # branches count as writes and the merge reports the conflict. + self.assertFalse(result.success) + error_text = result.errors[0] if result.errors else "" + self.assertIn("shared", error_text) + + def test_non_boolean_parallel_safe_stays_serial(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), True), + ], + parallelism=2, + ) + # Simulate a truthy non-bool value reaching the engine (e.g. set + # directly on the step attribute): must not enable concurrency. + for step in pipeline.steps: + step.parallel_safe = "false" + + result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success) + self.assertEqual(probe.max_active, 1) + + def test_non_dict_result_reports_failure_not_completion(self): + probe = ConcurrencyProbe() + tracking_ids = {} + counter = {"n": 0} + + def fake_start(*args, **kwargs): + tid = f"tid_{counter['n']}" + counter["n"] += 1 + tracking_ids[kwargs.get("submodule")] = tid + return tid + + with patch( + "semantica.pipeline.execution_engine.get_progress_tracker" + ) as mock_get: + tracker = MagicMock() + tracker.start_tracking.side_effect = fake_start + mock_get.return_value = tracker + engine = ExecutionEngine() + + def non_dict_handler(data, **kwargs): + with probe: + return "not-a-dict" + + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", non_dict_handler, True), + ], + parallelism=2, + name="progress_pipeline", + ) + + result = engine.execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + failed_step = next(s for s in pipeline.steps if s.name == "b") + self.assertEqual(failed_step.status, StepStatus.FAILED) + self.assertIsInstance(failed_step.error, ProcessingError) + # Handler ran exactly once: never re-run serially. + self.assertEqual(probe.calls, 2) + + b_tid = tracking_ids.get("progress_pipeline:branch:b") + self.assertIsNotNone( + b_tid, f"expected tracking for step b, got {tracking_ids}" + ) + b_stops = [ + c + for c in tracker.stop_tracking.call_args_list + if (c.args[0] if c.args else c.kwargs.get("tracking_id")) == b_tid + ] + self.assertTrue(b_stops) + for call in b_stops: + status = call.kwargs.get("status") + self.assertEqual(status, "failed") + self.assertNotEqual(status, "completed") + + def test_same_type_steps_get_distinct_tracking_records(self): + probe = ConcurrencyProbe() + pipeline = self._build( + [ + ("a", "branch", branch_handler(probe, "a"), True), + ("b", "branch", branch_handler(probe, "b"), True), + ], + parallelism=2, + name="tracking_pipeline", + ) + + with patch( + "semantica.pipeline.execution_engine.get_progress_tracker" + ) as mock_get: + tracker = MagicMock() + mock_get.return_value = tracker + engine = ExecutionEngine() + + result = engine.execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertTrue(result.success, msg=str(result.errors)) + submodules = [ + call.kwargs.get("submodule") + for call in tracker.start_tracking.call_args_list + if call.kwargs.get("module") == "pipeline" + ] + # Concurrent steps of the same step_type get distinct submodules + # (and therefore distinct tracking IDs), including pipeline identity. + self.assertIn("tracking_pipeline:branch:a", submodules) + self.assertIn("tracking_pipeline:branch:b", submodules) + + +class TestParallelSafeSerialization(unittest.TestCase): + """parallel_safe must survive dict and JSON round-trips.""" + + def _build_pipeline(self): + builder = PipelineBuilder() + builder.add_step("extract", "source", parallel_safe=True, batch_size=10) + builder.add_step("index", "sink", dependencies=["extract"]) + builder.set_parallelism(3) + return builder.build("parallel-serialization") + + def test_serializer_roundtrip_preserves_parallel_safe(self): + pipeline = self._build_pipeline() + serializer = PipelineSerializer() + + serialized = serializer.serialize_pipeline(pipeline, format="dict") + self.assertTrue(serialized["steps"][0]["parallel_safe"]) + self.assertFalse(serialized["steps"][1]["parallel_safe"]) + + restored = serializer.deserialize_pipeline(serialized) + self.assertTrue(restored.steps[0].parallel_safe) + self.assertFalse(restored.steps[1].parallel_safe) + self.assertEqual(restored.config.get("parallelism"), 3) + + def test_serializer_json_roundtrip_preserves_parallel_safe(self): + pipeline = self._build_pipeline() + serializer = PipelineSerializer() + + serialized = serializer.serialize_pipeline(pipeline, format="json") + restored = serializer.deserialize_pipeline(serialized) + + self.assertTrue(restored.steps[0].parallel_safe) + self.assertFalse(restored.steps[1].parallel_safe) + + def test_builder_serialize_outputs_parallel_safe(self): + builder = PipelineBuilder() + builder.add_step("extract", "source", parallel_safe=True) + builder.add_step("index", "sink", dependencies=["extract"]) + builder.set_parallelism(2) + + serialized = builder.serialize(format="dict") + + self.assertTrue(serialized["steps"][0]["parallel_safe"]) + self.assertFalse(serialized["steps"][1]["parallel_safe"]) + self.assertEqual(serialized["config"]["parallelism"], 2) + + +class Uncomparable: + """Object whose equality comparison always raises TypeError.""" + + def __eq__(self, other): + raise TypeError("cannot compare") + + +class TestParallelDependencyValidation(unittest.TestCase): + """Cyclic/unknown dependencies must fail with ValidationError in the + parallel path, not RecursionError/KeyError (review finding #1).""" + + def _engine(self): + with patch( + "semantica.pipeline.execution_engine.get_progress_tracker" + ) as mock_get: + mock_get.return_value = MagicMock() + engine = ExecutionEngine() + return engine + + def _handler(self): + return lambda data, **kwargs: data + + def test_cyclic_dependencies_raise_validation_error(self): + # Construct the pipeline directly: the builder already rejects + # cycles at build time, so bypassing it exercises the engine-level + # grouping guard (which otherwise hits RecursionError). + steps = [ + PipelineStep( + name="a", + step_type="branch", + handler=self._handler(), + dependencies=["b"], + parallel_safe=True, + ), + PipelineStep( + name="b", + step_type="branch", + handler=self._handler(), + dependencies=["a"], + parallel_safe=True, + ), + ] + pipeline = Pipeline( + name="cyclic_pipeline", + steps=steps, + config={"parallelism": 4}, + ) + + result = self._engine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + error_text = result.errors[0] if result.errors else "" + self.assertIn("Circular dependency", error_text) + + def test_unknown_dependency_raises_validation_error(self): + # Construct the pipeline directly so the engine-level unknown + # dependency guard is exercised (instead of a builder-time + # KeyError from the grouping DFS). + steps = [ + PipelineStep( + name="a", + step_type="branch", + handler=self._handler(), + dependencies=["missing"], + parallel_safe=True, + ), + ] + pipeline = Pipeline( + name="unknown_dep_pipeline", + steps=steps, + config={"parallelism": 4}, + ) + + result = self._engine().execute_pipeline(pipeline, data={"text": "hi"}) + + self.assertFalse(result.success) + error_text = result.errors[0] if result.errors else "" + self.assertIn("unknown step 'missing'", error_text) + self.assertIn("'a'", error_text) + + +class TestParallelSafeMustBeBoolean(unittest.TestCase): + """parallel_safe must be an explicit boolean everywhere (review #4).""" + + def test_add_step_rejects_non_boolean_values(self): + builder = PipelineBuilder() + for invalid in ("false", "true", 1, 0, None, [True]): + with self.assertRaises(ValidationError): + builder.add_step( + "a", + "branch", + handler=lambda data, **kwargs: data, + parallel_safe=invalid, + ) + + def test_build_pipeline_rejects_non_boolean_values(self): + builder = PipelineBuilder() + config = { + "name": "invalid_parallel_safe", + "steps": [ + {"name": "a", "type": "branch", "parallel_safe": "false"} + ], + } + with self.assertRaises(ValidationError): + builder.build_pipeline(config) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file