Made the prompt source optionally resilient to downstream errors

Author: travis-bauer

CHANGELOG.md

@@ -4,19 +4,25 @@
 ### Improvements
 - Changes several logging statements to make the consistent with the rest of the code
 - Refactored **lambda** to use AbstractFieldSegment for consistency
-- Removed fail_on_error parameter in **lambda** and **lambdaFilter**.  This is an API breaking change. 
+- Removed fail_on_error parameter in **lambda** and **lambdaFilter**.  This is an API breaking change.
   The segments will now always fail on an error and there is no option to silently fail.
 - Changed chatterlang_serve so that it raises and exception an exits if given a script that can't be
   compiled.  The previous behavior was that it would issue a log message and fall back to the default
   script.
-- Added unit test for parse_unknown_args and expanded it to support boolean flag parameters 
+- Added unit test for parse_unknown_args and expanded it to support boolean flag parameters
 - Marked UMAP as deprecated.  It will be removed in 1.0.  It requires additional dependencies that no
-  other core modules need.  
+  other core modules need.
 - Updated the ollama embedding and chat connector so that it uses the OLLAMA_SERVER_URL configuration
   variable as the host where ollama is installed.  So OLLAMA_SERVER_URL can be set in the TOML configuration
   file or TALKPIPE_OLLAMA_SERVER_URL can be set as an environment variable.
 - Implemented better compile errors for when sources or segments are not found.  It had been a key error.
   It will now be a compile error.
+- Added batch files for the tutorials so it is easier to run them on Windows
+- Enhanced **prompt** source to catch downstream pipeline errors and continue prompting instead of
+  crashing. When a user enters invalid input that causes an exception in downstream segments, the error
+  is displayed with a full traceback, and the prompt continues accepting input. This makes interactive
+  pipelines more robust and user-friendly. The error-resilient behavior is enabled by default but can
+  be disabled by setting `error_resilient=False`.
 
 ## 0.9.3 
 ### Improvements

src/talkpipe/pipe/io.py

@@ -5,15 +5,61 @@
 import os
 import pickle  # nosec B403 - Used only for write operations, not loading untrusted data
 import json
+import traceback
 from pprint import pformat
 from prompt_toolkit import PromptSession
 
 from talkpipe.chatterlang.registry import register_source, register_segment
 import talkpipe.chatterlang.registry as registry
-from talkpipe.pipe.core import AbstractSource, source, AbstractSegment, segment
+from talkpipe.pipe.core import AbstractSource, source, AbstractSegment, segment, Pipeline
 from talkpipe.util import data_manipulation
 
 
+class ErrorResilientPromptPipeline(Pipeline):
+    """A special pipeline that handles errors in interactive prompt workflows.
+
+    When a downstream segment raises an exception, this pipeline catches it,
+    displays the error, and continues processing the next item from the prompt.
+    """
+
+    def __init__(self, prompt_source, *operations):
+        super().__init__(prompt_source, *operations)
+        self.prompt_source = prompt_source
+
+    def transform(self, input_iter=None):
+        """Execute pipeline with error resilience for prompt-based workflows."""
+        # Get the prompt generator
+        prompt_iter = self.prompt_source()
+
+        # Build the downstream pipeline (everything after the prompt)
+        downstream_ops = [op for op in self.operations if op is not self.prompt_source]
+
+        # Process each prompt input with error handling
+        for user_input in prompt_iter:
+            try:
+                # Create a single-item iterator for this input
+                current_iter = iter([user_input])
+
+                # Pass through each downstream operation
+                for op in downstream_ops:
+                    current_iter = op(current_iter)
+
+                # Consume and yield results
+                for result in current_iter:
+                    yield result
+
+            except Exception as e:
+                # Catch and display errors, but continue prompting
+                print(f"Error: {e}")
+                traceback.print_exc()
+                # Don't yield anything for this failed input, just continue to next prompt
+
+    def __or__(self, other):
+        """Support chaining additional operations to the error-resilient pipeline."""
+        # Add the new operation to our operations list
+        return ErrorResilientPromptPipeline(self.prompt_source, *self.operations[1:], other)
+
+
 @registry.register_segment(name="print")
 class Print(AbstractSegment):
     """
@@ -73,22 +119,44 @@ def transform(self, input_iter: Annotated[Iterable[int], "Iterable input data"])
 @register_source('prompt')
 class Prompt(AbstractSource):
     """A source that generates input from a prompt.
-    
+
     This source will generate input from a prompt until the user enters an EOF.
     It is for creating interactive pipelines.  It uses prompt_toolkit under the
     hood to provide a nice prompt experience.
+
+    To enable error recovery (continue prompting after downstream errors), this source
+    needs to actively consume the downstream pipeline with error handling. This is done
+    by overriding the __or__ method to wrap the downstream in error handling.
     """
 
-    def __init__(self):
+    def __init__(self, error_resilient: Annotated[bool, "If True, catches downstream errors and continues prompting"] = True):
         super().__init__()
         self.session = PromptSession()
+        self.error_resilient = error_resilient
 
     def generate(self) -> Iterable[str]:
         while True:
             try:
-                yield self.session.prompt('> ')
+                user_input = self.session.prompt('> ')
+                yield user_input
             except EOFError:
                 break
+            except KeyboardInterrupt:
+                print("\nInterrupted. Press Ctrl+D to exit or continue entering input.")
+                continue
+
+    def __or__(self, other):
+        """Override to add error handling when chaining with other segments."""
+        if self.error_resilient:
+            # Register the downstream relationship
+            self.registerDownstream(other)
+            other.registerUpstream(self)
+
+            # Return an error-resilient pipeline instead of a regular one
+            return ErrorResilientPromptPipeline(self, other)
+        else:
+            # Use default behavior
+            return super().__or__(other)
 
 @register_source('echo')
 @source(delimiter=',')

tests/talkpipe/pipe/test_io.py

@@ -1,4 +1,5 @@
 import json, pickle, os
+from unittest.mock import Mock, patch
 from talkpipe.pipe import io
 from talkpipe.util import config
 from talkpipe.chatterlang import compiler
@@ -326,13 +327,91 @@ def test_writeString_compiler_integration(tmpdir):
 def test_writeString_single_item(tmpdir):
     """Test writeString with single item."""
     data = ["single_item"]
-    
+
     temp_file_path = tmpdir.join("test_single.txt")
     f = io.writeString(fname=str(temp_file_path))
-    
+
     result = list(f(data))
     assert result == data
-    
+
     with open(str(temp_file_path), 'r') as file:
         content = file.read()
     assert content == "single_item\n"
+
+def test_prompt_error_handling_continues_after_exception(capsys):
+    """Test that Prompt with error-resilient pipeline catches downstream exceptions and continues prompting."""
+    # Mock the PromptSession to return specific values
+    mock_session = Mock()
+    mock_session.prompt.side_effect = [
+        "1",      # Valid input
+        "abc",    # Invalid input that will cause error
+        "2",      # Valid input after error
+        EOFError()  # Simulate user exiting
+    ]
+
+    # Create a lambda segment that will fail on non-numeric input
+    from talkpipe.pipe.basic import EvalExpression
+
+    with patch('talkpipe.pipe.io.PromptSession', return_value=mock_session):
+        prompt = io.Prompt()
+        # Chain with a lambda that converts to int (will fail on "abc")
+        pipeline = prompt | EvalExpression(expression="int(item)")
+
+        # Convert to function and execute
+        func = pipeline.as_function()
+        results = func()
+
+        # Should get results for valid inputs only
+        assert 1 in results
+        assert 2 in results
+        # "abc" should have caused an error but not stopped the pipeline
+
+        # Verify error was printed
+        captured = capsys.readouterr()
+        assert "Error:" in captured.out or "invalid literal" in captured.out.lower()
+
+def test_prompt_keyboard_interrupt_continues(capsys):
+    """Test that Prompt handles KeyboardInterrupt and continues prompting."""
+    mock_session = Mock()
+    mock_session.prompt.side_effect = [
+        "first input",
+        KeyboardInterrupt(),
+        "second input",
+        EOFError()
+    ]
+
+    with patch('talkpipe.pipe.io.PromptSession', return_value=mock_session):
+        prompt = io.Prompt()
+        generator = prompt.generate()
+
+        # First input should work normally
+        assert next(generator) == "first input"
+
+        # After KeyboardInterrupt, generator should continue
+        assert next(generator) == "second input"
+
+        # Verify message was printed
+        captured = capsys.readouterr()
+        assert "Interrupted" in captured.out
+
+def test_prompt_eof_terminates():
+    """Test that Prompt terminates on EOFError."""
+    mock_session = Mock()
+    mock_session.prompt.side_effect = [
+        "first input",
+        EOFError()
+    ]
+
+    with patch('talkpipe.pipe.io.PromptSession', return_value=mock_session):
+        prompt = io.Prompt()
+        generator = prompt.generate()
+
+        # First input should work
+        assert next(generator) == "first input"
+
+        # EOFError should cause generator to stop
+        try:
+            next(generator)
+            assert False, "Generator should have stopped on EOFError"
+        except StopIteration:
+            pass  # Expected behavior