langchain-hs-core 0.0.6.0 → 0.0.6.2

raw patch · 15 files changed

+2092/−243 lines, 15 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−0
• README.md +1/−1
• langchain-hs-core.cabal +5/−3
• src/Langchain/Agent/PlanAndExecute.hs +142/−36
• src/Langchain/Agent/ReAct.hs +651/−138
• src/Langchain/Eval.hs +308/−0
• src/Langchain/Prelude.hs +54/−1
• src/Langchain/Stream.hs +213/−9
• src/Langchain/Tool.hs +95/−0
• src/Langchain/Tool/Security.hs +45/−13
• src/Langchain/Tool/Shell.hs +227/−41
• test/Spec.hs +2/−0
• test/Test/Langchain/Eval.hs +109/−0
• test/Test/Langchain/Stream.hs +103/−1
• test/Test/Langchain/Tool.hs +114/−0

@@ -2,6 +2,29 @@  All notable changes to this package will be documented in this file. +## 0.0.6.2 - 2026-11-14++> **Note**: `0.0.6.1` was skipped globally after `langchain-hs-ollama-v0.0.6.1`.++- **Tool-Call & Agent Streaming**:+  - Extended `StreamEvent` in `Langchain.Stream` with `ToolStart`, `ToolChunk`, `ToolEnd`, and `ToolErrorEvent`, plus `liftChatStream` and `tapStreamMessage`.+  - Added cancellation-safe streaming combinators `liftChatStreamResource` and `liftChatStreamWithCancel` in `Langchain.Stream`.+  - Added `StreamingTool`, `createStreamingTool`, `fromTool`, `streamToolCall`, and `streamStreamingToolCall` to `Langchain.Tool`.+  - Added `streamingShellTool` (`Langchain.Tool.Shell`) and `withStreamingSecurity` (`Langchain.Tool.Security`).+  - Added `withStreamingTools`, `reactStepStream`, `streamReActAgent`, `streamReActAgentWithTrace`, and `runReActAgentStream` to `Langchain.Agent.ReAct`.+  - Added `streamPlanAndExecute` and `runPlanAndExecuteStream` to `Langchain.Agent.PlanAndExecute`.+  - Updated `Langchain.Prelude` exports for streaming tools and agents.+- **Multi-Agent Primitives & ReAct Loop Detection**:+  - Added `LoopDetectionStrategy` (`NoLoopDetection`, `WarnOnDuplicateAction`, `FailOnLoop`, `BreakOnLoop`) and iteration telemetry (`traceIterations`, `traceTerminatedBy`).+  - Added multi-agent handoff tools `asAgentTool` and `asStreamingAgentTool`.+- **Evaluation Harness (`Langchain.Eval`)**:+  - Added `EvalCase`, `EvalScore`, `Scorer`, and `EvalReport` with Aeson support.+  - Implemented `exactMatchScorer`, `containsScorer`, `toolCallExactScorer`, `llmJudgeScorer`, and `runEvalSuite`.+- **Security, Checkpointing, Memory & Parsers (consolidated post-`0.0.6.0` alignment)**:+  - `SecurityPolicy`, `FileSystem`, `Process`, and `Checkpoint` tool suites (`Langchain.Tool.*`).+  - `InVariableMemory`, `SessionMemory`, `EntityMemory`, `CombinedMemory`, `FileMemory`, and buffer reset helpers.+  - Updated `sqlite-simple` bounds to `>= 0.4.18 && < 0.5`.+ ## 0.0.6.0 - 2026-10-02  - Updated bounds for GHC 9.12 and modern Stackage LTS releases.

@@ -32,7 +32,7 @@  ```cabal build-depends:-    langchain-hs-core >= 0.0.6.0 && < 0.0.7.0+    langchain-hs-core >= 0.0.6.2 && < 0.0.7.0 ```  To connect concrete LLMs, combine with satellite provider packages:

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           langchain-hs-core-version:        0.0.6.0+version:        0.0.6.2 synopsis:       Core typeclasses and pure AST primitives for langchain-hs description:    Zero HTTP dependency core package containing ChatModel, RunnableTree, ContentBlock, Tool, and StreamEvent primitives. category:       Web, AI, Control@@ -91,6 +91,7 @@       Langchain.Agent.PlanAndExecute       Langchain.Guardrail       Langchain.Guardrail.Core+      Langchain.Eval       Langchain.Prelude   other-modules:       Paths_langchain_hs_core@@ -127,7 +128,7 @@     , sqlite-simple >=0.4.18 && <0.5     , stm ==2.5.*     , text >=1.2 && <3-    , time >=1.9 && <1.15+    , time >=1.9 && <1.16   default-language: Haskell2010  test-suite langchain-hs-core-test@@ -135,6 +136,7 @@   main-is: Spec.hs   other-modules:       Test.Langchain.Callback+      Test.Langchain.Eval       Test.Langchain.Model       Test.Langchain.Monad       Test.Langchain.Observability@@ -185,5 +187,5 @@     , tasty-hunit >=0.10     , tasty-quickcheck >=0.10     , text >=1.2 && <3-    , time >=1.9 && <1.15+    , time >=1.9 && <1.16   default-language: Haskell2010

@@ -27,22 +27,36 @@   ( PlanStep (..)   , Plan (..)   , StepExecutor (..)+  , StreamingStepExecutor (..)   , PlanAndExecuteAgent (..)   , newPlanAndExecuteAgent   , newPlanAndExecuteAgentWithTools   , runPlanAndExecute+  , streamPlanAndExecute+  , streamPlanAndExecuteWithResult+  , runPlanAndExecuteStream   ) where  import Control.Applicative ((<|>))+import Control.Monad (void) import Control.Monad.Except (MonadError, throwError) import Control.Monad.IO.Class (MonadIO)-import Data.Aeson (FromJSON (..), ToJSON, Value (..), withObject, (.!=), (.:), (.:?))+import Control.Monad.Trans (lift)+import Data.Aeson (FromJSON (..), ToJSON (toJSON), Value (..), withObject, (.!=), (.:), (.:?)) import Data.Aeson.Types (prependFailure, typeMismatch)+import Data.Conduit (ConduitT, fuseUpstream, runConduit, yield, (.|))+import qualified Data.Conduit.List as CL import Data.Text (Text) import qualified Data.Text as T import GHC.Generics (Generic) -import Langchain.Agent.ReAct (ReActAgent, defaultReActAgent, runReActAgent)+import Langchain.Agent.ReAct+  ( AgentTrace (traceFinalAnswer)+  , ReActAgent+  , defaultReActAgent+  , runReActAgent+  , streamReActAgentWithTrace+  ) import Langchain.Error (LangchainError, agentError) import Langchain.Model   ( ChatModel (..)@@ -50,6 +64,7 @@   , userMessage   ) import Langchain.OutputParser.Structured (StructuredOutput, TypeSchema, structuredInvoke)+import Langchain.Stream (EventStream, StreamEvent (..), liftChatStream, tapStreamMessage) import Langchain.Tool (Tool) import Langchain.Tool.Binding (ToolBinder (..)) @@ -129,6 +144,45 @@     msg <- invoke model [userMessage prompt] Nothing     pure $ extractMessageText msg +-- | Streaming abstraction for executing individual steps of a plan.+class StreamingStepExecutor e m where+  executeStepStream ::+    -- | Step executor instance (agent, model with tools, function, or model)+    e ->+    -- | Step instruction prompt to execute+    Text ->+    -- | Stream of execution events returning the textual step result+    ConduitT () StreamEvent m Text++instance+  {-# OVERLAPPING #-}+  (ToolBinder model n, MonadIO n, MonadError LangchainError n) =>+  StreamingStepExecutor (ReActAgent model n) n+  where+  executeStepStream agent prompt = do+    trace <- streamReActAgentWithTrace agent [userMessage prompt]+    pure $ extractMessageText (traceFinalAnswer trace)++instance+  {-# OVERLAPPING #-}+  (ToolBinder model n, MonadIO n, MonadError LangchainError n) =>+  StreamingStepExecutor (model, [Tool n]) n+  where+  executeStepStream (model, tools) =+    executeStepStream (defaultReActAgent model tools)++instance {-# OVERLAPPING #-} (m ~ n, Monad m) => StreamingStepExecutor (Text -> m Text) n where+  executeStepStream f prompt = lift (f prompt)++instance+  {-# OVERLAPPABLE #-}+  (ChatModel model, MonadIO m, MonadError LangchainError m) =>+  StreamingStepExecutor model m+  where+  executeStepStream model prompt = do+    msg <- liftChatStream (stream model [userMessage prompt] Nothing) .| tapStreamMessage+    pure $ extractMessageText msg+ -- | Plan-and-Execute agent container data PlanAndExecuteAgent planner executor = PlanAndExecuteAgent   { plannerModel :: planner@@ -163,6 +217,41 @@ newPlanAndExecuteAgentWithTools planner model tools =   PlanAndExecuteAgent planner (defaultReActAgent model tools) +buildPlanPrompt :: Maybe Text -> Text -> Text+buildPlanPrompt (Just p) userGoal = p <> "\nGoal: " <> userGoal+buildPlanPrompt Nothing userGoal =+  "You are an expert planner. For the following goal, generate a concise step-by-step execution plan.\n"+    <> "Output JSON format: {\"planSteps\": [{\"stepNumber\": 1, \"stepDescription\": \"...\"}]}\n"+    <> "Keep the plan focused and minimal (between 2 to 3 distinct, actionable steps).\n"+    <> "Goal: "+    <> userGoal++buildStepPrompt :: Text -> [(PlanStep, Text)] -> PlanStep -> Text+buildStepPrompt userGoal prevOutputs currStep =+  "User Goal: "+    <> userGoal+    <> ( if null prevOutputs+           then ""+           else+             "\n\nCompleted Steps So Far:\n"+               <> T.unlines+                 [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- prevOutputs]+       )+    <> "\n\nCurrent Task To Execute (Step "+    <> T.pack (show (stepNumber currStep))+    <> "): "+    <> stepDescription currStep+    <> "\nExecute this task using any appropriate tools available and provide the outcome:"++buildSynthesisPrompt :: Text -> [(PlanStep, Text)] -> Text+buildSynthesisPrompt userGoal stepOutputs =+  "User Goal: "+    <> userGoal+    <> "\n\nStep Execution History:\n"+    <> T.unlines+      [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- stepOutputs]+    <> "\n\nProvide the final synthesized answer satisfying the goal:"+ -- | Execute a goal using the Plan-and-Execute workflow with structured JSON planning runPlanAndExecute ::   (ChatModel planner, StepExecutor executor m, MonadIO m, MonadError LangchainError m) =>@@ -173,43 +262,60 @@   -- | Final synthesized answer solving the user goal   m Text runPlanAndExecute PlanAndExecuteAgent {..} userGoal = do-  let planPrompt = case planPromptTemplate of-        Just p -> p <> "\nGoal: " <> userGoal-        Nothing ->-          "You are an expert planner. For the following goal, generate a concise step-by-step execution plan.\n"-            <> "Output JSON format: {\"planSteps\": [{\"stepNumber\": 1, \"stepDescription\": \"...\"}]}\n"-            <> "Keep the plan focused and minimal (between 2 to 3 distinct, actionable steps).\n"-            <> "Goal: "-            <> userGoal-  plan <- structuredInvoke plannerModel [userMessage planPrompt]+  plan <- structuredInvoke plannerModel [userMessage (buildPlanPrompt planPromptTemplate userGoal)]   if null (planSteps plan)     then throwError $ agentError "Planner generated an empty plan" (Just "PlanAndExecuteAgent") Nothing     else executeSteps (planSteps plan) []   where-    executeSteps [] stepOutputs = do-      let synthesisPrompt =-            "User Goal: "-              <> userGoal-              <> "\n\nStep Execution History:\n"-              <> T.unlines-                [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- stepOutputs]-              <> "\n\nProvide the final synthesized answer satisfying the goal:"-      executeStep stepExecutor synthesisPrompt+    executeSteps [] stepOutputs =+      executeStep stepExecutor (buildSynthesisPrompt userGoal stepOutputs)     executeSteps (currStep : restSteps) prevOutputs = do-      let stepPrompt =-            "User Goal: "-              <> userGoal-              <> ( if null prevOutputs-                     then ""-                     else-                       "\n\nCompleted Steps So Far:\n"-                         <> T.unlines-                           [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- prevOutputs]-                 )-              <> "\n\nCurrent Task To Execute (Step "-              <> T.pack (show (stepNumber currStep))-              <> "): "-              <> stepDescription currStep-              <> "\nExecute this task using any appropriate tools available and provide the outcome:"-      stepOut <- executeStep stepExecutor stepPrompt+      stepOut <- executeStep stepExecutor (buildStepPrompt userGoal prevOutputs currStep)       executeSteps restSteps (prevOutputs ++ [(currStep, stepOut)])++{- | Execute a goal using the Plan-and-Execute workflow while streaming 'StreamEvent's+and returning the final synthesized answer.+-}+streamPlanAndExecuteWithResult ::+  (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) =>+  PlanAndExecuteAgent planner executor ->+  Text ->+  ConduitT () StreamEvent m Text+streamPlanAndExecuteWithResult PlanAndExecuteAgent {..} userGoal = do+  let pRunId = "plan-and-execute-run"+  yield $ ChainStart pRunId "PlanAndExecuteAgent" (toJSON userGoal)+  plan <-+    lift $ structuredInvoke plannerModel [userMessage (buildPlanPrompt planPromptTemplate userGoal)]+  if null (planSteps plan)+    then throwError $ agentError "Planner generated an empty plan" (Just "PlanAndExecuteAgent") Nothing+    else do+      finalOut <- executeSteps pRunId (planSteps plan) []+      yield $ ChainEnd pRunId "PlanAndExecuteAgent" (toJSON finalOut)+      pure finalOut+  where+    executeSteps _ [] stepOutputs =+      executeStepStream stepExecutor (buildSynthesisPrompt userGoal stepOutputs)+    executeSteps pRunId (currStep : restSteps) prevOutputs = do+      let sNodeId = "step-" <> T.pack (show (stepNumber currStep))+      yield $ NodeStart pRunId sNodeId (toJSON currStep)+      stepOut <- executeStepStream stepExecutor (buildStepPrompt userGoal prevOutputs currStep)+      yield $ NodeEnd pRunId sNodeId (toJSON stepOut)+      executeSteps pRunId restSteps (prevOutputs ++ [(currStep, stepOut)])++-- | Execute a goal using the Plan-and-Execute workflow as an 'EventStream'.+streamPlanAndExecute ::+  (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) =>+  PlanAndExecuteAgent planner executor ->+  Text ->+  EventStream m+streamPlanAndExecute agent userGoal = void (streamPlanAndExecuteWithResult agent userGoal)++-- | Execute a Plan-and-Execute agent in streaming mode with an event callback.+runPlanAndExecuteStream ::+  (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) =>+  PlanAndExecuteAgent planner executor ->+  Text ->+  (StreamEvent -> m ()) ->+  m Text+runPlanAndExecuteStream agent userGoal onEvent =+  runConduit (streamPlanAndExecuteWithResult agent userGoal `fuseUpstream` CL.mapM_ onEvent)

@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -49,6 +50,7 @@   , AgentTrace (..)   , ReActAgent (..)   , ToolErrorStrategy (..)+  , LoopDetectionStrategy (..)      -- * Construction   , defaultReActAgent@@ -62,26 +64,67 @@   , withCallbackManager   , withStopCondition   , withToolErrorStrategy+  , withStreamingTools+  , withLoopDetection+  , withMaxConsecutiveIdenticalToolCalls+  , withMaxDelegationDepth +    -- * Multi-Agent Handoff & Delegation Tools+  , asAgentTool+  , asStreamingAgentTool+     -- * Execution   , reactStep   , runReActAgent   , runReActAgentWithTrace++    -- * Streaming Execution+  , InterruptibleResult (..)+  , reactStepStream+  , streamReActAgent+  , streamReActAgentWithTrace+  , runReActAgentStream+  , runReActAgentStreamInterruptible   ) where -import Control.Monad (forM, when)+import Control.Monad (unless, when) import Control.Monad.Except (MonadError, catchError, throwError) import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans (lift)+import Data.Aeson (Value (..), object, toJSON, (.=))+import Data.Aeson.Types (parseEither, (.:?))+import Data.Conduit (ConduitT, await, fuseUpstream, mapOutput, runConduit, yield, (.|))+import qualified Data.Conduit.List as CL+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef) import Data.List (find) import Data.Text (Text) import qualified Data.Text as T import Data.Time.Clock (diffUTCTime, getCurrentTime)  import Langchain.Callback.Manager (CallbackEvent (..), CallbackManager, dispatchEvent)-import Langchain.Error (LangchainError, agentError, errorMessage)+import Langchain.Error (LangchainError, agentError, errorMessage, toolError) import Langchain.Memory.Core (BaseMemory (..), SomeMemory (..)) import Langchain.Model import qualified Langchain.Model.Types as M+import Langchain.Stream+  ( CancelToken+  , EventStream+  , StreamEvent+    ( ChainEnd+    , ChainStart+    , LLMChunk+    , LLMEnd+    , LLMStart+    , ToolChunk+    , ToolEnd+    , ToolErrorEvent+    , ToolStart+    )+  , isCancelled+  , liftChatStream+  , liftChatStreamWithCancel+  , tapStreamMessage+  ) import Langchain.Tool import Langchain.Tool.Binding (ToolBinder (..)) @@ -93,6 +136,16 @@     FailOnError   deriving (Eq, Show) +-- | Strategy for detecting runaway loops where the model repeatedly invokes identical tools with identical arguments.+data LoopDetectionStrategy+  = -- | Allow consecutive repeat tool calls without intervention (default for backwards compatibility).+    AllowRepeats+  | -- | After @n@ consecutive identical tool call signatures, skip tool execution and inject an observation warning to the LLM.+    WarnOnRepeat !Int+  | -- | After @n@ consecutive identical tool call signatures, abort execution immediately with an 'AgentError' and notify 'CallbackManager'.+    AbortOnRepeat !Int+  deriving (Eq, Show)+ -- | A single step in the ReAct reasoning chain. data AgentStep   = -- | The model decided to call one or more tools.@@ -107,6 +160,8 @@   -- ^ Each pair is (the step, observation messages produced)   , traceFinalAnswer :: Message   -- ^ The final answer message+  , traceIterations :: !Int+  -- ^ Total reasoning iterations executed   }   deriving (Eq, Show) @@ -135,6 +190,12 @@   -- ^ Optional early-stop predicate checked after each LLM response   , agentToolErrorStrategy :: ToolErrorStrategy   -- ^ How to handle tool execution errors (default: 'ContinueOnError')+  , agentStreamingTools :: [StreamingTool m]+  -- ^ Optional streaming tools capable of emitting incremental 'ToolChunk' events+  , agentLoopDetection :: !LoopDetectionStrategy+  -- ^ Strategy for runaway consecutive duplicate tool-call loop detection (default: 'AllowRepeats')+  , agentMaxDelegationDepth :: !Int+  -- ^ Maximum allowed delegation depth for nested sub-agents (default: 5)   }  {- | Construct a ReAct agent with sensible defaults.@@ -159,6 +220,9 @@     , agentCallbackManager = Nothing     , agentStopCondition = Nothing     , agentToolErrorStrategy = ContinueOnError+    , agentStreamingTools = []+    , agentLoopDetection = AllowRepeats+    , agentMaxDelegationDepth = 5     }  {- | Attach a 'BaseMemory' instance for persistent conversation history.@@ -263,6 +327,49 @@   ReActAgent model m withToolErrorStrategy strategy agent = agent {agentToolErrorStrategy = strategy} +-- | Attach 'StreamingTool's that emit incremental 'ToolChunk' events during streaming execution.+withStreamingTools ::+  [StreamingTool m] ->+  ReActAgent model m ->+  ReActAgent model m+withStreamingTools sTools agent = agent {agentStreamingTools = sTools}++{- | Set the runaway loop detection strategy (default: 'AllowRepeats').++* 'AllowRepeats': no intervention on repeated identical tool calls.+* 'WarnOnRepeat': skip execution and inject a warning observation once consecutive duplicates reach @n@.+* 'AbortOnRepeat': immediately terminate the reasoning loop and throw 'AgentError' once consecutive duplicates reach @n@.+-}+withLoopDetection ::+  LoopDetectionStrategy ->+  ReActAgent model m ->+  ReActAgent model m+withLoopDetection strat agent = agent {agentLoopDetection = strat}++-- | Set runaway loop detection to abort after @n@ consecutive duplicate tool calls.+withMaxConsecutiveIdenticalToolCalls ::+  Int ->+  ReActAgent model m ->+  ReActAgent model m+withMaxConsecutiveIdenticalToolCalls n = withLoopDetection (AbortOnRepeat n)++-- | Set the maximum allowed agent delegation depth (default: 5).+withMaxDelegationDepth ::+  Int ->+  ReActAgent model m ->+  ReActAgent model m+withMaxDelegationDepth d agent = agent {agentMaxDelegationDepth = d}++-- | Classify a model response 'Message' as 'AgentAction' or 'AgentFinish'.+classifyStepResponse :: Message -> AgentStep+classifyStepResponse responseMsg =+  case messageToolCalls responseMsg of+    Just tcs@(_ : _) -> AgentAction responseMsg tcs+    _ -> AgentFinish responseMsg++allAgentTools :: ReActAgent model m -> [Tool m]+allAgentTools ReActAgent {..} = agentTools ++ map streamingToolDef agentStreamingTools+ {- | Run a single step of ReAct reasoning.  Sends the current message history (with tools bound) to the LLM and@@ -285,51 +392,51 @@ reactStep model tools history userCfg = do   let toolCfg = bindToolsConfig @model tools userCfg   responseMsg <- invoke model history toolCfg-  case messageToolCalls responseMsg of-    Just tcs@(_ : _) -> pure $ AgentAction responseMsg tcs-    _ -> pure $ AgentFinish responseMsg--{- | Execute the full ReAct reasoning loop, returning only the final answer.--For the full reasoning trace, use 'runReActAgentWithTrace'.--When the agent has a 'BaseMemory' attached:+  pure $ classifyStepResponse responseMsg -1. Existing conversation history is loaded from memory.-2. The user's input messages are appended.-3. After the loop finishes, both the user input and final answer are saved to memory.--}-runReActAgent ::+reactStepStreamWith ::   forall model m.   (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>-  -- | Configured ReAct agent instance-  ReActAgent model m ->-  -- | Input user messages for this turn+  Maybe CancelToken ->+  model ->+  [Tool m] ->   [Message] ->-  -- | The final assistant answer message-  m Message-runReActAgent agent userMsgs = traceFinalAnswer <$> runReActAgentWithTrace agent userMsgs--{- | Execute the full ReAct reasoning loop, returning both the final answer-and the complete intermediate reasoning trace.+  Maybe (ModelConfig model) ->+  ConduitT () StreamEvent m AgentStep+reactStepStreamWith mbTok model tools history userCfg = do+  let toolCfg = bindToolsConfig @model tools userCfg+      liftFn = maybe liftChatStream liftChatStreamWithCancel mbTok+  responseMsg <- liftFn (stream model history toolCfg) .| tapStreamMessage+  pure $ classifyStepResponse responseMsg -Tool calls within a single step are executed in sequence within the @m@ monad,-and the observations are fed back to the LLM for the next reasoning step.+{- | Run a single step of ReAct reasoning using streaming LLM invocation. -The trace includes every 'AgentStep' and the observation messages produced-at each iteration, which is invaluable for debugging and observability.+Forwards all 'StreamEvent's from the underlying 'ChatModel' 'stream'+(including 'LLMStart', 'LLMChunk' with text and 'toolCallDelta', and 'LLMEnd')+downstream in real time, and returns the resulting 'AgentStep'. -}-runReActAgentWithTrace ::+reactStepStream ::   forall model m.   (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>-  -- | Configured ReAct agent instance+  -- | The chat model to stream+  model ->+  -- | Available tools to bind to the model+  [Tool m] ->+  -- | Current conversation message history+  [Message] ->+  -- | Optional provider-specific model configuration+  Maybe (ModelConfig model) ->+  -- | Stream of LLM events returning the step classification+  ConduitT () StreamEvent m AgentStep+reactStepStream = reactStepStreamWith Nothing++buildInitialHistory ::+  (MonadIO m, MonadError LangchainError m) =>   ReActAgent model m ->-  -- | Input user messages for this turn   [Message] ->-  -- | Full reasoning trace containing all intermediate steps and final answer-  m AgentTrace-runReActAgentWithTrace ReActAgent {..} userMsgs = do-  -- 1. Build initial history: system prompt + memory + user messages+  Text ->+  m [Message]+buildInitialHistory ReActAgent {..} userMsgs caller = do   memoryMsgs <- case agentMemory of     Nothing -> pure []     Just (SomeMemory mem) -> messages mem@@ -347,135 +454,541 @@       agentError         "Cannot invoke ReAct agent with empty message history (no user query, memory, or system prompt)"         (Just "ReActAgent")-        (Just "runReActAgentWithTrace")+        (Just caller) -  -- 2. Run the reasoning loop-  trace <- go initialHistory agentMaxIterations []+  pure initialHistory -  -- 3. Persist to memory if available+persistTurnMemory ::+  (MonadIO m, MonadError LangchainError m) =>+  ReActAgent model m ->+  [Message] ->+  Message ->+  m ()+persistTurnMemory ReActAgent {..} userMsgs finalMsg =   case agentMemory of     Nothing -> pure ()     Just (SomeMemory mem) -> do-      -- Save each user message       mapM_ (addMessage mem) userMsgs-      -- Save the final answer-      addMessage mem (traceFinalAnswer trace)+      addMessage mem finalMsg -  pure trace+toolObservationMessage :: Text -> Text -> Text -> Message+toolObservationMessage tName tcId txt =+  (textMessage M.Tool txt)+    { M.messageName = Just tName+    , M.messageToolId = Just tcId+    }++dispatchCb :: MonadIO m => Maybe CallbackManager -> CallbackEvent -> m ()+dispatchCb Nothing _ = pure ()+dispatchCb (Just mgr) ev = dispatchEvent mgr ev++reactRunId :: Text+reactRunId = "react-agent-run"++-- | Execute a ReAct loop step-by-step, parameterized over the LLM step and tool executors.+runLoopWith ::+  forall model m n.+  (MonadIO n, MonadError LangchainError n) =>+  Maybe CancelToken ->+  ReActAgent model m ->+  ([Message] -> n AgentStep) ->+  (ToolCall -> n Message) ->+  [Message] ->+  Int ->+  [(AgentStep, [Message])] ->+  n AgentTrace+runLoopWith mbTok agent@ReActAgent {..} runStep execTool = go Nothing 0   where-    go :: [Message] -> Int -> [(AgentStep, [Message])] -> m AgentTrace-    go history maxIter accSteps+    checkCancelled = maybe (pure False) isCancelled mbTok++    execToolsWithCancel [] acc = pure (reverse acc)+    execToolsWithCancel (tc : rest) acc = do+      cancelled <- checkCancelled+      if cancelled+        then pure (reverse acc)+        else do+          obs <- execTool tc+          execToolsWithCancel rest (obs : acc)++    toolSignature tcs = [(toolCallName tc, toolCallArguments tc) | tc <- tcs]++    go mbLastSig repeatCount history maxIter accSteps       | maxIter <= 0 = do           let maxErr =                 "ReAct Agent exceeded maximum iterations ("                   <> T.pack (show agentMaxIterations)                   <> ")"-          case agentCallbackManager of-            Nothing -> pure ()-            Just mgr -> do-              now <- liftIO getCurrentTime-              dispatchEvent mgr (OnError "ReActAgent" maxErr now)+          now <- liftIO getCurrentTime+          dispatchCb agentCallbackManager (OnError "ReActAgent" maxErr now)           throwError $             agentError               maxErr               (Just "ReActAgent")               (Just "runReActAgent")       | otherwise = do-          -- Dispatch OnLLMStart callback before calling model-          llmStartTime <- liftIO getCurrentTime-          case agentCallbackManager of-            Nothing -> pure ()-            Just mgr ->-              dispatchEvent mgr (OnLLMStart "ReActAgent" (map extractMessageText history) llmStartTime)+          cancelledBefore <- checkCancelled+          if cancelledBefore+            then pure $ AgentTrace accSteps (assistantMessage "") (length accSteps)+            else do+              llmStartTime <- liftIO getCurrentTime+              dispatchCb+                agentCallbackManager+                (OnLLMStart "ReActAgent" (map extractMessageText history) llmStartTime) -          -- Invoke the model with error interception for observability-          eStep <--            (Right <$> reactStep agentModel agentTools history agentModelConfig)-              `catchError` (pure . Left)-          llmEndTime <- liftIO getCurrentTime-          let llmLatency = round (realToFrac (diffUTCTime llmEndTime llmStartTime) * 1000000 :: Double) :: Int-          case eStep of-            Left err -> do-              case agentCallbackManager of-                Nothing -> pure ()-                Just mgr ->-                  dispatchEvent mgr (OnError "ReActAgent" (errorMessage err) llmEndTime)-              throwError err-            Right step -> case step of-              AgentFinish finalMsg -> do-                case agentCallbackManager of-                  Nothing -> pure ()-                  Just mgr ->-                    dispatchEvent mgr (OnLLMEnd "ReActAgent" (extractMessageText finalMsg) llmLatency llmEndTime)-                let allSteps = accSteps ++ [(step, [])]-                pure $ AgentTrace allSteps finalMsg-              AgentAction respMsg tcs -> do-                case agentCallbackManager of-                  Nothing -> pure ()-                  Just mgr ->-                    dispatchEvent mgr (OnLLMEnd "ReActAgent" (extractMessageText respMsg) llmLatency llmEndTime)+              eStep <- (Right <$> runStep history) `catchError` (pure . Left)+              llmEndTime <- liftIO getCurrentTime+              let llmLatency = round (realToFrac (diffUTCTime llmEndTime llmStartTime) * 1000000 :: Double) :: Int+              case eStep of+                Left err -> do+                  dispatchCb agentCallbackManager (OnError "ReActAgent" (errorMessage err) llmEndTime)+                  throwError err+                Right step -> case step of+                  AgentFinish finalMsg -> do+                    dispatchCb+                      agentCallbackManager+                      (OnLLMEnd "ReActAgent" (extractMessageText finalMsg) llmLatency llmEndTime)+                    let finalSteps = accSteps ++ [(step, [])]+                    pure $ AgentTrace finalSteps finalMsg (length finalSteps)+                  AgentAction respMsg tcs -> do+                    dispatchCb+                      agentCallbackManager+                      (OnLLMEnd "ReActAgent" (extractMessageText respMsg) llmLatency llmEndTime)+                    cancelledAfterStep <- checkCancelled+                    if cancelledAfterStep+                      then do+                        let finalSteps = accSteps ++ [(step, [])]+                        pure $ AgentTrace finalSteps respMsg (length finalSteps)+                      else case agentStopCondition of+                        Just cond+                          | cond respMsg -> do+                              let finalSteps = accSteps ++ [(step, [])]+                              pure $ AgentTrace finalSteps respMsg (length finalSteps)+                        _ -> do+                          let currentSig = toolSignature tcs+                              nextRepeatCount =+                                case mbLastSig of+                                  Just prevSig | prevSig == currentSig -> repeatCount + 1+                                  _ -> 1+                          case agentLoopDetection of+                            AbortOnRepeat maxR+                              | nextRepeatCount >= max 1 maxR -> do+                                  let abortErr =+                                        "Runaway loop detected: consecutive identical tool calls ("+                                          <> T.pack (show nextRepeatCount)+                                          <> " repetitions)"+                                  now <- liftIO getCurrentTime+                                  dispatchCb agentCallbackManager (OnError "ReActAgent" abortErr now)+                                  throwError $+                                    agentError+                                      abortErr+                                      (Just "ReActAgent")+                                      (Just "runLoopWith:loopDetection")+                            WarnOnRepeat maxR+                              | nextRepeatCount >= max 1 maxR -> do+                                  let warnText tc =+                                        "Loop warning: tool '"+                                          <> toolCallName tc+                                          <> "' called with identical arguments "+                                          <> T.pack (show nextRepeatCount)+                                          <> " times consecutively. Please change your approach, adjust arguments, or provide your final answer."+                                      warnMsgs =+                                        [ toolObservationMessage+                                            (toolCallName tc)+                                            (toolCallId tc)+                                            (warnText tc)+                                        | tc <- tcs+                                        ]+                                      newHistory = history ++ [respMsg] ++ warnMsgs+                                      newSteps = accSteps ++ [(step, warnMsgs)]+                                  go (Just currentSig) nextRepeatCount newHistory (maxIter - 1) newSteps+                            _ -> do+                              obsMsgs <- execToolsWithCancel tcs []+                              let newHistory = history ++ [respMsg] ++ obsMsgs+                                  newSteps = accSteps ++ [(step, obsMsgs)]+                              let _ = agent+                              go (Just currentSig) nextRepeatCount newHistory (maxIter - 1) newSteps -                -- Check custom stop condition-                case agentStopCondition of-                  Just cond | cond respMsg -> do-                    let allSteps = accSteps ++ [(step, [])]-                    pure $ AgentTrace allSteps respMsg-                  _ -> do-                    obsMsgs <- forM tcs executeTool-                    let newHistory = history ++ [respMsg] ++ obsMsgs-                        newSteps = accSteps ++ [(step, obsMsgs)]-                    go newHistory (maxIter - 1) newSteps+handleToolOutcome ::+  (MonadIO n, MonadError LangchainError n) =>+  ReActAgent model m ->+  ToolCall ->+  (StreamEvent -> n ()) ->+  Either LangchainError Text ->+  Int ->+  n Message+handleToolOutcome ReActAgent {..} tc emitStream eRes durationMicros = do+  let tName = toolCallName tc+  endTime <- liftIO getCurrentTime+  case eRes of+    Left err -> do+      let errTxt = "Error executing tool " <> tName <> ": " <> errorMessage err+      dispatchCb agentCallbackManager (OnError tName errTxt endTime)+      emitStream $ ToolErrorEvent reactRunId tName err+      when (agentToolErrorStrategy == FailOnError) $+        throwError $+          agentError errTxt (Just "ReActAgent") (Just $ "executeTool:" <> tName)+      pure $ toolObservationMessage tName (toolCallId tc) errTxt+    Right outTxt -> do+      dispatchCb agentCallbackManager (OnToolEnd tName outTxt durationMicros endTime)+      emitStream $ ToolEnd reactRunId tName (toJSON outTxt)+      pure $ toolObservationMessage tName (toolCallId tc) outTxt -    executeTool :: ToolCall -> m Message-    executeTool tc = do-      let tName = toolCallName tc-      case agentCallbackManager of-        Nothing -> pure ()-        Just mgr -> do-          now <- liftIO getCurrentTime-          dispatchEvent mgr (OnToolStart tName (toolCallArguments tc) now)+executeToolSync ::+  forall model m.+  (MonadIO m, MonadError LangchainError m) =>+  ReActAgent model m ->+  ToolCall ->+  m Message+executeToolSync agent@ReActAgent {..} tc = do+  let tName = toolCallName tc+      tools = allAgentTools agent+  now <- liftIO getCurrentTime+  dispatchCb agentCallbackManager (OnToolStart tName (toolCallArguments tc) now)+  case find (\t -> toolName t == tName) tools of+    Nothing -> do+      let notFoundMsg = "Error: Tool not found: " <> tName+          err = agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName)+      handleToolOutcome agent tc (const (pure ())) (Left err) 0+    Just tool -> do+      startTime <- liftIO getCurrentTime+      eRes <- toolExecute tool (toolCallArguments tc)+      endTime <- liftIO getCurrentTime+      let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int+      handleToolOutcome agent tc (const (pure ())) eRes durationMicros +executeToolStream ::+  forall model m.+  (MonadIO m, MonadError LangchainError m) =>+  ReActAgent model m ->+  ToolCall ->+  ConduitT () StreamEvent m Message+executeToolStream agent@ReActAgent {..} tc = do+  let tName = toolCallName tc+  now <- liftIO getCurrentTime+  dispatchCb agentCallbackManager (OnToolStart tName (toolCallArguments tc) now)+  yield $ ToolStart reactRunId tName (toolCallArguments tc)+  case find (\st -> toolName (streamingToolDef st) == tName) agentStreamingTools of+    Just sTool -> do+      startTime <- liftIO getCurrentTime+      eRes <- mapOutput (ToolChunk reactRunId tName) (streamingToolExecute sTool (toolCallArguments tc))+      endTime <- liftIO getCurrentTime+      let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int+      handleToolOutcome agent tc yield eRes durationMicros+    Nothing ->       case find (\t -> toolName t == tName) agentTools of         Nothing -> do           let notFoundMsg = "Error: Tool not found: " <> tName-          case agentCallbackManager of-            Nothing -> pure ()-            Just mgr -> do-              now <- liftIO getCurrentTime-              dispatchEvent mgr (OnError tName notFoundMsg now)-          when (agentToolErrorStrategy == FailOnError) $-            throwError $-              agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName)-          pure $-            (textMessage M.Tool notFoundMsg)-              { M.messageName = Just tName-              , M.messageToolId = Just (toolCallId tc)-              }+              err = agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName)+          handleToolOutcome agent tc yield (Left err) 0         Just tool -> do           startTime <- liftIO getCurrentTime-          eRes <- toolExecute tool (toolCallArguments tc)+          eRes <- lift $ toolExecute tool (toolCallArguments tc)           endTime <- liftIO getCurrentTime           let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int-          case eRes of-            Left err -> do-              let errTxt = "Error executing tool " <> tName <> ": " <> errorMessage err-              case agentCallbackManager of-                Nothing -> pure ()-                Just mgr -> dispatchEvent mgr (OnError tName errTxt endTime)-              when (agentToolErrorStrategy == FailOnError) $-                throwError $-                  agentError errTxt (Just "ReActAgent") (Just $ "executeTool:" <> tName)-              pure $-                (textMessage M.Tool errTxt)-                  { M.messageName = Just tName-                  , M.messageToolId = Just (toolCallId tc)-                  }-            Right outTxt -> do-              case agentCallbackManager of-                Nothing -> pure ()-                Just mgr -> dispatchEvent mgr (OnToolEnd tName outTxt durationMicros endTime)-              pure $-                (textMessage M.Tool outTxt)-                  { M.messageName = Just tName-                  , M.messageToolId = Just (toolCallId tc)-                  }+          handleToolOutcome agent tc yield eRes durationMicros++{- | Execute the full ReAct reasoning loop, returning only the final answer.++For the full reasoning trace, use 'runReActAgentWithTrace'.+For streaming execution, use 'streamReActAgent' or 'runReActAgentStream'.++When the agent has a 'BaseMemory' attached:++1. Existing conversation history is loaded from memory.+2. The user's input messages are appended.+3. After the loop finishes, both the user input and final answer are saved to memory.+-}+runReActAgent ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Configured ReAct agent instance+  ReActAgent model m ->+  -- | Input user messages for this turn+  [Message] ->+  -- | The final assistant answer message+  m Message+runReActAgent agent userMsgs = traceFinalAnswer <$> runReActAgentWithTrace agent userMsgs++{- | Execute the full ReAct reasoning loop, returning both the final answer+and the complete intermediate reasoning trace.++Tool calls within a single step are executed in sequence within the @m@ monad,+and the observations are fed back to the LLM for the next reasoning step.++The trace includes every 'AgentStep' and the observation messages produced+at each iteration, which is invaluable for debugging and observability.+-}+runReActAgentWithTrace ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Configured ReAct agent instance+  ReActAgent model m ->+  -- | Input user messages for this turn+  [Message] ->+  -- | Full reasoning trace containing all intermediate steps and final answer+  m AgentTrace+runReActAgentWithTrace agent@ReActAgent {..} userMsgs = do+  initialHistory <- buildInitialHistory agent userMsgs "runReActAgentWithTrace"+  let tools = allAgentTools agent+  trace <-+    runLoopWith+      Nothing+      agent+      (\hist -> reactStep agentModel tools hist agentModelConfig)+      (executeToolSync agent)+      initialHistory+      agentMaxIterations+      []+  persistTurnMemory agent userMsgs (traceFinalAnswer trace)+  pure trace++streamReActAgentTraceWith ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  Maybe CancelToken ->+  ReActAgent model m ->+  [Message] ->+  ConduitT () StreamEvent m AgentTrace+streamReActAgentTraceWith mbTok agent@ReActAgent {..} userMsgs = do+  initialHistory <- lift $ buildInitialHistory agent userMsgs "streamReActAgentWithTrace"+  let tools = allAgentTools agent+  yield $ ChainStart reactRunId "ReActAgent" (toJSON initialHistory)+  trace <-+    runLoopWith+      mbTok+      agent+      (\hist -> reactStepStreamWith mbTok agentModel tools hist agentModelConfig)+      (executeToolStream agent)+      initialHistory+      agentMaxIterations+      []+  cancelled <- maybe (pure False) isCancelled mbTok+  unless cancelled $ do+    lift $ persistTurnMemory agent userMsgs (traceFinalAnswer trace)+    yield $ ChainEnd reactRunId "ReActAgent" (toJSON (traceFinalAnswer trace))+  pure trace++{- | Execute the full ReAct reasoning loop as a Conduit stream of 'StreamEvent's,+returning the complete 'AgentTrace'.++Emits 'ChainStart', followed by 'LLMStart' / 'LLMChunk' / 'LLMEnd' for each+reasoning turn and 'ToolStart' / 'ToolChunk' / 'ToolEnd' / 'ToolErrorEvent' for+each tool invocation, concluding with 'ChainEnd'.+-}+streamReActAgentWithTrace ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Configured ReAct agent instance+  ReActAgent model m ->+  -- | Input user messages for this turn+  [Message] ->+  -- | Event stream returning the full reasoning trace+  ConduitT () StreamEvent m AgentTrace+streamReActAgentWithTrace = streamReActAgentTraceWith Nothing++{- | Execute the full ReAct reasoning loop as an 'EventStream'.++Suitable for composition with 'collectEvents', 'runStreamWith', or any Conduit sink.+-}+streamReActAgent ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Configured ReAct agent instance+  ReActAgent model m ->+  -- | Input user messages for this turn+  [Message] ->+  -- | Stream of agent lifecycle, LLM, and tool events+  EventStream m+streamReActAgent agent userMsgs = do+  _ <- streamReActAgentWithTrace agent userMsgs+  pure ()++-- | Outcome of an interruptible streaming ReAct turn.+data InterruptibleResult+  = -- | Turn completed normally with the final assistant message.+    Completed !Message+  | -- | Turn was cancelled via 'CancelToken', returning any partial assistant text emitted.+    Cancelled !Message+  deriving (Eq, Show)++{- | Execute an interruptible streaming ReAct turn using a cooperative 'CancelToken'.++Safe for @LangchainT () IO@ callers without @ResourceT@. When cancelled mid-turn,+upstream streaming stops promptly, conversation memory is left unmodified, and+'Cancelled' returns the partial assistant message streamed so far.+-}+runReActAgentStreamInterruptible ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  CancelToken ->+  ReActAgent model m ->+  [Message] ->+  (StreamEvent -> m ()) ->+  m InterruptibleResult+runReActAgentStreamInterruptible tok agent userMsgs onEvent = do+  cancelledStart <- isCancelled tok+  if cancelledStart+    then pure $ Cancelled (assistantMessage "")+    else do+      chunksRef <- liftIO $ newIORef ([] :: [Text])+      let trackAndForward ev = do+            liftIO $ case ev of+              LLMStart {} -> writeIORef chunksRef []+              LLMChunk _ txt _ | not (T.null txt) -> modifyIORef' chunksRef (txt :)+              LLMEnd _ msg _ -> writeIORef chunksRef [extractMessageText msg]+              _ -> pure ()+            onEvent ev+      trace <-+        runConduit+          ( streamReActAgentTraceWith (Just tok) agent userMsgs+              `fuseUpstream` CL.mapM_ trackAndForward+          )+      cancelledEnd <- isCancelled tok+      if cancelledEnd+        then do+          partialTxt <- liftIO (T.concat . reverse <$> readIORef chunksRef)+          pure $ Cancelled (assistantMessage partialTxt)+        else pure $ Completed (traceFinalAnswer trace)++{- | Execute the ReAct agent in streaming mode, invoking a callback for every+'StreamEvent' as it arrives and returning the final assistant 'Message'.+-}+runReActAgentStream ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Configured ReAct agent instance+  ReActAgent model m ->+  -- | Input user messages for this turn+  [Message] ->+  -- | Callback invoked on each 'StreamEvent' in real time+  (StreamEvent -> m ()) ->+  -- | The final assistant answer message+  m Message+runReActAgentStream agent userMsgs onEvent =+  traceFinalAnswer+    <$> runConduit (streamReActAgentWithTrace agent userMsgs `fuseUpstream` CL.mapM_ onEvent)++-- | JSON Schema describing the input expected by agent handoff tools.+handoffSchema :: Value+handoffSchema =+  object+    [ "type" .= ("object" :: Text)+    , "properties"+        .= object+          [ "task"+              .= object+                [ "type" .= ("string" :: Text)+                , "description" .= ("The task or query for the specialist agent" :: Text)+                ]+          ]+    , "required" .= (["task"] :: [Text])+    ]++-- | Extract the task query string from tool arguments.+extractAgentTask :: Text -> Value -> Either LangchainError Text+extractAgentTask toolNm = \case+  String s | not (T.null (T.strip s)) -> Right s+  Object o -> case parseEither (.:? "task") o of+    Right (Just task) | not (T.null (T.strip task)) -> Right task+    _ -> case parseEither (.:? "query") o of+      Right (Just q) | not (T.null (T.strip q)) -> Right q+      _ -> case parseEither (.:? "prompt") o of+        Right (Just p) | not (T.null (T.strip p)) -> Right p+        _ ->+          Left $+            toolError+              ("Missing or invalid 'task' parameter for agent tool: " <> toolNm)+              (Just toolNm)+              Nothing+  _ ->+    Left $+      toolError+        ("Invalid arguments object for agent tool: " <> toolNm)+        (Just toolNm)+        Nothing++{- | Wrap a 'ReActAgent' as a callable 'Tool' that delegates tasks to the sub-agent.++Accepts @{ "task": "<text>" }@ (or a raw JSON string) and returns the sub-agent's+final answer text. Bounded by 'agentMaxDelegationDepth' to prevent unbounded recursion.+-}+asAgentTool ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Tool name (e.g., "researcher")+  Text ->+  -- | Tool description explaining the specialist agent's role+  Text ->+  -- | Specialist sub-agent to delegate tasks to+  ReActAgent model m ->+  -- | Wrapped tool ready to be attached to a supervisor agent+  Tool m+asAgentTool name desc subAgent =+  createTool name desc handoffSchema $ \args -> do+    case extractAgentTask name args of+      Left err -> pure (Left err)+      Right task+        | agentMaxDelegationDepth subAgent <= 0 ->+            pure $+              Left $+                agentError+                  ("Maximum agent delegation depth exceeded for sub-agent '" <> name <> "'")+                  (Just "ReActAgent")+                  (Just "asAgentTool")+        | otherwise -> do+            let delegatedAgent =+                  subAgent {agentMaxDelegationDepth = agentMaxDelegationDepth subAgent - 1}+            eRes <- (Right <$> runReActAgent delegatedAgent [userMessage task]) `catchError` (pure . Left)+            pure $ fmap extractMessageText eRes++{- | Wrap a 'ReActAgent' as a 'StreamingTool' forwarding intermediate progress chunks to the caller.++Emits incremental sub-agent text and tool chunks downstream wrapped as 'ToolChunk' events.+-}+asStreamingAgentTool ::+  forall model m.+  (ToolBinder model m, MonadIO m, MonadError LangchainError m) =>+  -- | Tool name (e.g., "researcher")+  Text ->+  -- | Tool description explaining the specialist agent's role+  Text ->+  -- | Specialist sub-agent to delegate tasks to+  ReActAgent model m ->+  -- | Wrapped streaming tool emitting incremental sub-agent chunks+  StreamingTool m+asStreamingAgentTool name desc subAgent =+  createStreamingTool name desc handoffSchema $ \args -> do+    case extractAgentTask name args of+      Left err -> pure (Left err)+      Right task+        | agentMaxDelegationDepth subAgent <= 0 ->+            pure $+              Left $+                agentError+                  ("Maximum agent delegation depth exceeded for sub-agent '" <> name <> "'")+                  (Just "ReActAgent")+                  (Just "asStreamingAgentTool")+        | otherwise -> do+            let delegatedAgent =+                  subAgent {agentMaxDelegationDepth = agentMaxDelegationDepth subAgent - 1}+            ansRef <- liftIO $ newIORef ""+            let captureSub = do+                  trace <- streamReActAgentWithTrace delegatedAgent [userMessage task]+                  liftIO $ writeIORef ansRef (extractMessageText (traceFinalAnswer trace))+                forwardEvents = do+                  mEv <- await+                  case mEv of+                    Nothing -> pure ()+                    Just ev -> do+                      case ev of+                        LLMChunk _ txt Nothing | not (T.null txt) -> yield txt+                        ToolChunk _ _ chunk | not (T.null chunk) -> yield chunk+                        _ -> pure ()+                      forwardEvents+            catchError+              ( do+                  captureSub .| forwardEvents+                  ans <- liftIO $ readIORef ansRef+                  pure $ Right ans+              )+              (pure . Left)

@@ -0,0 +1,308 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- |+Module      : Langchain.Eval+Description : Lightweight evaluation harness for chains, agents, and models+Copyright   : (c) 2025-2026 Tushar Adhatrao+License     : MIT+Maintainer  : Tushar Adhatrao <tusharadhatrao@gmail.com>+Stability   : experimental++Quantitative evaluation and regression testing framework for LLM applications.+Provides typed datasets ('EvalCase'), deterministic scorers (exact match, substring,+tool call trace accuracy), LLM-as-a-judge scoring, and JSON evaluation report generation.+-}+module Langchain.Eval+  ( -- * Core Types+    EvalCase (..)+  , EvalScore (..)+  , Scorer (..)+  , createScorer+  , CaseResult (..)+  , CaseScoreResult (..)+  , EvalReport (..)++    -- * Deterministic Scorers+  , exactMatchScorer+  , containsScorer+  , toolCallExactScorer++    -- * Model-based Scorers+  , llmJudgeScorer++    -- * Evaluation Runner+  , runEvalSuite+  ) where++import Control.Monad.Except (MonadError, catchError)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Aeson+  ( FromJSON (..)+  , ToJSON (..)+  , Value (..)+  , decode+  , withObject+  , (.:)+  , (.:?)+  )+import Data.Aeson.Types (parseEither)+import qualified Data.ByteString.Lazy as LBS+import Data.Map.Strict (Map)+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Data.Time.Clock (diffUTCTime, getCurrentTime)+import GHC.Generics (Generic)++import Langchain.Agent.ReAct (AgentStep (..), AgentTrace (..))+import Langchain.Error (LangchainError, errorMessage)+import Langchain.Model (ChatModel (..))+import Langchain.Model.Types (extractMessageText, toolCallName, userMessage)++-- | A single evaluation test case with input, optional reference target, and metadata.+data EvalCase i o = EvalCase+  { evalInput :: !i+  -- ^ Input supplied to the subject under test+  , evalExpected :: !(Maybe o)+  -- ^ Optional ground-truth or expected reference output+  , evalMetadata :: !(Maybe (Map Text Text))+  -- ^ Optional contextual tags or metadata+  }+  deriving (Show, Eq, Generic, ToJSON, FromJSON)++-- | Assessment produced by a 'Scorer' on an individual test case.+data EvalScore = EvalScore+  { scoreValue :: !Double+  -- ^ Normalized score between 0.0 and 1.0+  , scorePassed :: !Bool+  -- ^ Whether the case satisfied the threshold or assertion+  , scoreReason :: !(Maybe Text)+  -- ^ Diagnostic or qualitative reasoning for the score+  }+  deriving (Show, Eq, Generic, ToJSON, FromJSON)++-- | Named evaluation metric evaluating actual output against a test case.+data Scorer i o m = Scorer+  { scorerName :: !Text+  -- ^ Unique identifier for this evaluation metric+  , scoreCase :: EvalCase i o -> o -> m EvalScore+  -- ^ Evaluation function producing 'EvalScore'+  }++-- | Construct a custom 'Scorer'.+createScorer :: Text -> (EvalCase i o -> o -> m EvalScore) -> Scorer i o m+createScorer = Scorer++-- | Score result produced by a specific named scorer.+data CaseScoreResult = CaseScoreResult+  { caseScoreName :: !Text+  , caseScore :: !EvalScore+  }+  deriving (Show, Eq, Generic, ToJSON, FromJSON)++-- | Outcome of running all scorers on a single test case.+data CaseResult = CaseResult+  { caseInput :: !Value+  , caseExpected :: !(Maybe Value)+  , caseActual :: !(Either Text Value)+  , caseScores :: ![CaseScoreResult]+  , caseLatencyMicros :: !Int+  , casePassed :: !Bool+  }+  deriving (Show, Eq, Generic, ToJSON, FromJSON)++-- | Aggregated summary report of an evaluation suite run.+data EvalReport = EvalReport+  { reportTotalCases :: !Int+  -- ^ Total number of evaluated cases+  , reportPassedCases :: !Int+  -- ^ Number of cases that passed all scorers+  , reportPassRate :: !Double+  -- ^ Ratio of passed cases (0.0 to 1.0)+  , reportMeanScore :: !Double+  -- ^ Average score value across all evaluations (0.0 to 1.0)+  , reportMeanLatencyMicros :: !Int+  -- ^ Average execution latency per case in microseconds+  , reportCases :: ![CaseResult]+  -- ^ Detailed per-case scoring breakdown+  }+  deriving (Show, Eq, Generic, ToJSON, FromJSON)++-- | Deterministic scorer verifying exact equality between actual and expected output.+exactMatchScorer :: (Eq o, Show o, Monad m) => Scorer i o m+exactMatchScorer = createScorer "exact_match" $ \c actual ->+  case evalExpected c of+    Nothing -> pure $ EvalScore 1.0 True (Just "No expected output specified")+    Just expected ->+      if actual == expected+        then pure $ EvalScore 1.0 True (Just "Exact match")+        else+          pure $+            EvalScore+              0.0+              False+              (Just $ "Expected: " <> T.pack (show expected) <> ", Actual: " <> T.pack (show actual))++-- | Deterministic scorer checking if expected text is contained within actual text.+containsScorer :: Monad m => Scorer Text Text m+containsScorer = createScorer "contains" $ \c actual ->+  case evalExpected c of+    Nothing -> pure $ EvalScore 1.0 True (Just "No expected substring specified")+    Just expectedSub ->+      if expectedSub `T.isInfixOf` actual+        then pure $ EvalScore 1.0 True (Just "Substring found")+        else+          pure $+            EvalScore+              0.0+              False+              (Just $ "Substring '" <> expectedSub <> "' not found in actual output")++-- | Deterministic scorer verifying that an 'AgentTrace' invoked an exact sequence of tools.+toolCallExactScorer :: Monad m => [Text] -> Scorer i AgentTrace m+toolCallExactScorer expectedTools = createScorer "tool_call_exact" $ \_ trace -> do+  let actualTools =+        [ toolCallName tc+        | (AgentAction _ tcs, _) <- traceSteps trace+        , tc <- tcs+        ]+  if actualTools == expectedTools+    then pure $ EvalScore 1.0 True (Just "Tool call sequence matched exactly")+    else+      pure $+        EvalScore+          0.0+          False+          ( Just $+              "Expected tool sequence: "+                <> T.pack (show expectedTools)+                <> ", but got: "+                <> T.pack (show actualTools)+          )++-- | LLM-as-a-judge scorer evaluating output against a custom qualitative rubric.+llmJudgeScorer ::+  forall model m.+  (ChatModel model, MonadIO m, MonadError LangchainError m) =>+  model ->+  Text ->+  Scorer Text Text m+llmJudgeScorer model rubric = createScorer "llm_judge" $ \c actual -> do+  let inputTxt = evalInput c+      expectedTxt = fromMaybe "" (evalExpected c)+      prompt =+        "You are an expert AI evaluator. Assess the following actual output against the rubric.\n\n"+          <> "Evaluation Rubric: "+          <> rubric+          <> "\n\n"+          <> "Test Input: "+          <> inputTxt+          <> "\n"+          <> "Reference Target: "+          <> expectedTxt+          <> "\n"+          <> "Actual Output: "+          <> actual+          <> "\n\n"+          <> "Reply with a JSON object: {\"passed\": true/false, \"score\": 0.0-1.0, \"reason\": \"<explanation>\"}"+  eResp <- (Right <$> invoke model [userMessage prompt] Nothing) `catchError` (pure . Left)+  case eResp of+    Left err -> pure $ EvalScore 0.0 False (Just $ "LLM judge invocation failed: " <> errorMessage err)+    Right resp -> do+      let outTxt = extractMessageText resp+      case parseJudgeJson outTxt of+        Just (passed, sc, rsn) -> pure $ EvalScore sc passed (Just rsn)+        Nothing ->+          let passed = not ("fail" `T.isInfixOf` T.toLower outTxt)+              sc = if passed then 1.0 else 0.0+           in pure $ EvalScore sc passed (Just outTxt)++parseJudgeJson :: Text -> Maybe (Bool, Double, Text)+parseJudgeJson txt = do+  val <- decode (LBS.fromStrict (TE.encodeUtf8 txt))+  case parseEither parser val of+    Right res -> Just res+    Left _ -> Nothing+  where+    parser = withObject "JudgeResponse" $ \o -> do+      passed <- o .: "passed"+      sc <-+        o .:? "score" >>= \case+          Just s -> pure s+          Nothing -> pure (if passed then 1.0 else 0.0)+      rsn <-+        o .:? "reason" >>= \case+          Just r -> pure r+          Nothing -> pure ""+      pure (passed, sc, rsn)++-- | Execute an evaluation suite over a set of test cases and scorers.+runEvalSuite ::+  forall i o m.+  (MonadIO m, ToJSON i, ToJSON o) =>+  (i -> m (Either LangchainError o)) ->+  [Scorer i o m] ->+  [EvalCase i o] ->+  m EvalReport+runEvalSuite runner scorers cases = do+  caseResults <- mapM evaluateCase cases+  let total = length caseResults+      passedCount = length (filter casePassed caseResults)+      passRate = if total == 0 then 1.0 else fromIntegral passedCount / fromIntegral total+      allScores = [scoreValue (caseScore sr) | cr <- caseResults, sr <- caseScores cr]+      meanScore =+        if null allScores+          then if total == 0 then 1.0 else 0.0+          else sum allScores / fromIntegral (length allScores)+      allLatencies = map caseLatencyMicros caseResults+      meanLatency =+        if null allLatencies+          then 0+          else sum allLatencies `div` total+  pure+    EvalReport+      { reportTotalCases = total+      , reportPassedCases = passedCount+      , reportPassRate = passRate+      , reportMeanScore = meanScore+      , reportMeanLatencyMicros = meanLatency+      , reportCases = caseResults+      }+  where+    evaluateCase c = do+      startTime <- liftIO getCurrentTime+      eActual <- runner (evalInput c)+      endTime <- liftIO getCurrentTime+      let latencyMicros =+            round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int+      case eActual of+        Left err -> do+          let failedScore = EvalScore 0.0 False (Just $ errorMessage err)+              scores = [CaseScoreResult (scorerName s) failedScore | s <- scorers]+          pure+            CaseResult+              { caseInput = toJSON (evalInput c)+              , caseExpected = fmap toJSON (evalExpected c)+              , caseActual = Left (errorMessage err)+              , caseScores = scores+              , caseLatencyMicros = latencyMicros+              , casePassed = False+              }+        Right actual -> do+          scores <- mapM (\s -> CaseScoreResult (scorerName s) <$> scoreCase s c actual) scorers+          let caseOk = all (scorePassed . caseScore) scores+          pure+            CaseResult+              { caseInput = toJSON (evalInput c)+              , caseExpected = fmap toJSON (evalExpected c)+              , caseActual = Right (toJSON actual)+              , caseScores = scores+              , caseLatencyMicros = latencyMicros+              , casePassed = caseOk+              }

@@ -56,6 +56,7 @@     , LLMChunk     , LLMEnd     , ToolStart+    , ToolChunk     , ToolEnd     , ToolErrorEvent     , ChainStart@@ -65,6 +66,18 @@     )   , TokenUsage (..)   , EventStream+  , StreamM+  , ChatStream+  , CancelToken+  , newCancelToken+  , triggerCancel+  , isCancelled+  , liftChatStream+  , liftChatStreamResource+  , liftChatStreamWithCancel+  , tapStreamMessage+  , consumeStreamWith+  , runStreamWith   , collectEvents   , printEvents @@ -100,6 +113,11 @@   , Tool (..)   , createTool   , toolToValue+  , StreamingTool (..)+  , createStreamingTool+  , fromTool+  , streamToolCall+  , streamStreamingToolCall   , ToolBinder (..)   , DeriveToolSchema (..)   , deriveToolParametersSchema@@ -108,10 +126,14 @@   , PlanStep (..)   , Plan (..)   , StepExecutor (..)+  , StreamingStepExecutor (..)   , PlanAndExecuteAgent (..)   , newPlanAndExecuteAgent   , newPlanAndExecuteAgentWithTools   , runPlanAndExecute+  , streamPlanAndExecute+  , streamPlanAndExecuteWithResult+  , runPlanAndExecuteStream      -- * Guardrails & Safety   , GuardrailResult (..)@@ -293,6 +315,7 @@   , AgentStep (..)   , AgentTrace (..)   , ToolErrorStrategy (..)+  , LoopDetectionStrategy (..)   , defaultReActAgent   , withMemory   , withoutMemory@@ -302,9 +325,21 @@   , withCallbackManager   , withStopCondition   , withToolErrorStrategy+  , withStreamingTools+  , withLoopDetection+  , withMaxConsecutiveIdenticalToolCalls+  , withMaxDelegationDepth+  , asAgentTool+  , asStreamingAgentTool   , reactStep   , runReActAgent   , runReActAgentWithTrace+  , reactStepStream+  , streamReActAgent+  , streamReActAgentWithTrace+  , runReActAgentStream+  , InterruptibleResult (..)+  , runReActAgentStreamInterruptible      -- * Tool Suite & Security   , ToolTier (..)@@ -313,6 +348,7 @@   , readOnlySecurityPolicy   , workspaceSandboxPolicy   , withSecurity+  , withStreamingSecurity   , newFileSystemSuite   , findFilesTool   , grepSearchTool@@ -325,6 +361,8 @@   , commandStatusTool   , commandKillTool   , shellTool+  , streamingShellTool+  , streamingShellToolWithCancel      -- * Hybrid Retrieval & BM25   , BM25Index (..)@@ -339,6 +377,20 @@   , searchHybrid   , searchHybridWithScores   , reciprocalRankFusion++    -- * Evaluation & Regression Testing+  , EvalCase (..)+  , EvalScore (..)+  , Scorer (..)+  , createScorer+  , CaseResult (..)+  , CaseScoreResult (..)+  , EvalReport (..)+  , exactMatchScorer+  , containsScorer+  , toolCallExactScorer+  , llmJudgeScorer+  , runEvalSuite   ) where  import Langchain.Agent.PlanAndExecute@@ -351,6 +403,7 @@ import Langchain.DocumentLoader.FileLoader import Langchain.Embeddings.Core import Langchain.Error+import Langchain.Eval import Langchain.Guardrail.Core import Langchain.Memory import Langchain.Model@@ -377,6 +430,6 @@ import Langchain.Tool.GenericSchema import Langchain.Tool.Process import Langchain.Tool.Security-import Langchain.Tool.Shell (shellTool)+import Langchain.Tool.Shell (shellTool, streamingShellTool, streamingShellToolWithCancel) import Langchain.VectorStore.Core import Langchain.VectorStore.InMemory

@@ -21,32 +21,60 @@   , StreamCallback   , StreamSource   , callbackSource+  , liftChatStream+  , liftChatStreamResource+  , CancelToken+  , newCancelToken+  , triggerCancel+  , isCancelled+  , waitCancel+  , liftChatStreamWithCancel+  , tapStreamMessage+  , consumeStreamWith+  , runStreamWith   , collectEvents   , printEvents   ) where -import Control.Concurrent.Async (async, cancel)+import Control.Concurrent.Async (Async, AsyncCancelled (..), async, cancel, wait) import Control.Concurrent.STM-  ( atomically+  ( STM+  , TBQueue+  , TMVar+  , atomically   , newEmptyTMVarIO   , newTBQueueIO   , orElse   , putTMVar   , readTBQueue   , readTMVar+  , tryPutTMVar+  , tryReadTMVar   , writeTBQueue   )-import Control.Exception (finally)-import Control.Monad.Except (ExceptT, runExceptT)-import Control.Monad.IO.Class (liftIO)-import Control.Monad.Trans.Resource (ResourceT, runResourceT)+import Control.Exception+  ( Exception (displayException)+  , SomeException+  , catch+  , finally+  , fromException+  , onException+  , throwIO+  )+import Control.Monad (void)+import Control.Monad.Except (ExceptT, MonadError (throwError), runExceptT)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Resource (MonadResource, ResourceT, runResourceT) import Data.Aeson (FromJSON, ToJSON, Value)-import Data.Conduit (ConduitT, bracketP, runConduit, yield, (.|))+import Data.Conduit (ConduitT, Void, await, bracketP, fuseUpstream, runConduit, yield, (.|)) import qualified Data.Conduit.List as CL+import Data.IORef (newIORef, readIORef, writeIORef)+import Data.Maybe (isJust) import Data.Text (Text)+import qualified Data.Text as T import GHC.Generics (Generic)-import Langchain.Error (LangchainError)-import Langchain.Model.Types (Message, ToolCall)+import Langchain.Error (LangchainError, llmError)+import Langchain.Model.Types (Message (..), ToolCall, assistantMessage)  -- | Token usage accounting for LLM execution. data TokenUsage = TokenUsage@@ -84,6 +112,12 @@       , toolName :: Text       , toolInput :: Value       }+  | -- | Tool incremental streaming output chunk+    ToolChunk+      { runId :: Text+      , toolName :: Text+      , toolChunkText :: Text+      }   | -- | Tool execution completion     ToolEnd       { runId :: Text@@ -159,6 +193,176 @@             Nothing -> pure ()      third (_, _, worker) = worker++-- | Cooperative single-shot cancellation token backed by an STM 'TMVar'.+newtype CancelToken = CancelToken (TMVar ())+  deriving (Eq)++-- | Allocate a new untriggered 'CancelToken'.+newCancelToken :: MonadIO m => m CancelToken+newCancelToken = CancelToken <$> liftIO newEmptyTMVarIO++-- | Signal cancellation on a 'CancelToken' (idempotent).+triggerCancel :: MonadIO m => CancelToken -> m ()+triggerCancel (CancelToken var) = liftIO . atomically . void $ tryPutTMVar var ()++-- | Check whether a 'CancelToken' has been triggered.+isCancelled :: MonadIO m => CancelToken -> m Bool+isCancelled (CancelToken var) = liftIO . atomically $ isJust <$> tryReadTMVar var++waitCancel :: CancelToken -> STM ()+waitCancel (CancelToken var) = readTMVar var++startChatWorker ::+  ChatStream ->+  IO (TBQueue StreamEvent, TMVar (Maybe LangchainError), Async ())+startChatWorker chatStream = do+  q <- newTBQueueIO 64+  done <- newEmptyTMVarIO+  errRef <- newIORef Nothing+  w <-+    async $+      ( do+          eRes <-+            runResourceT $+              runExceptT $+                runConduit $+                  chatStream .| CL.mapM_ (liftIO . atomically . writeTBQueue q)+          case eRes of+            Right () -> pure ()+            Left err -> writeIORef errRef (Just err)+      )+        `catch` ( \ex -> case fromException ex of+                    Just AsyncCancelled -> throwIO (ex :: SomeException)+                    Nothing ->+                      writeIORef+                        errRef+                        (Just $ llmError (T.pack (displayException ex)) (Just "Stream") Nothing)+                )+        `finally` (readIORef errRef >>= atomically . void . tryPutTMVar done)+  pure (q, done, w)++consumeChatWorker ::+  (MonadIO m, MonadError LangchainError m) =>+  Maybe CancelToken ->+  (TBQueue StreamEvent, TMVar (Maybe LangchainError), Async ()) ->+  EventStream m+consumeChatWorker mbTok (queue, doneVar, worker) = loop+  where+    loop = do+      next <-+        liftIO $+          atomically+            ( case mbTok of+                Just tok ->+                  (Left Nothing <$ waitCancel tok)+                    `orElse` (Right <$> readTBQueue queue)+                    `orElse` (Left . Just <$> readTMVar doneVar)+                Nothing ->+                  (Right <$> readTBQueue queue)+                    `orElse` (Left . Just <$> readTMVar doneVar)+            )+            `onException` cancel worker+      case next of+        Right ev -> do+          yield ev+          loop+        Left Nothing ->+          liftIO $ cancel worker+        Left (Just Nothing) ->+          liftIO $ wait worker+        Left (Just (Just err)) -> do+          liftIO $ cancel worker+          throwError err++{- | Lift a resource-safe 'ChatStream' into any 'MonadIO' and 'MonadError LangchainError'+'EventStream'. Events are forwarded incrementally through a bounded STM queue,+and cancelling or throwing an exception cancels the upstream stream worker promptly.+-}+liftChatStream ::+  (MonadIO m, MonadError LangchainError m) =>+  ChatStream ->+  EventStream m+liftChatStream chatStream = do+  st <- liftIO $ startChatWorker chatStream+  consumeChatWorker Nothing st++{- | Lift a resource-safe 'ChatStream' into any 'MonadResource' and+'MonadError LangchainError' 'EventStream'. Uses 'bracketP' so stopping+consumption early (e.g. @src .| CL.take 1@) cancels the upstream producer+immediately.+-}+liftChatStreamResource ::+  (MonadResource m, MonadError LangchainError m) =>+  ChatStream ->+  EventStream m+liftChatStreamResource chatStream =+  bracketP (startChatWorker chatStream) (\(_, _, w) -> cancel w) (consumeChatWorker Nothing)++{- | Lift a resource-safe 'ChatStream' with cooperative cancellation via+'CancelToken'. When cancelled, the upstream stream worker is terminated promptly+and the Conduit exits cleanly so downstream accumulators preserve partial text.+-}+liftChatStreamWithCancel ::+  (MonadIO m, MonadError LangchainError m) =>+  CancelToken ->+  ChatStream ->+  EventStream m+liftChatStreamWithCancel tok chatStream = do+  st <- liftIO $ startChatWorker chatStream+  consumeChatWorker (Just tok) st++{- | Pass every 'StreamEvent' downstream unchanged while accumulating the final 'Message'+from the last 'LLMEnd' event (or synthesizing one from accumulated 'LLMChunk's if+no 'LLMEnd' event was emitted).+-}+tapStreamMessage :: Monad m => ConduitT StreamEvent StreamEvent m Message+tapStreamMessage = loop [] [] Nothing+  where+    loop accText accTools mbFinal = do+      mEv <- await+      case mEv of+        Nothing ->+          pure $ case mbFinal of+            Just msg ->+              case (messageToolCalls msg, reverse accTools) of+                (Nothing, ts@(_ : _)) -> msg {messageToolCalls = Just ts}+                _ -> msg+            Nothing ->+              let fullText = T.concat (reverse accText)+                  ts = reverse accTools+               in (assistantMessage fullText)+                    { messageToolCalls = if null ts then Nothing else Just ts+                    }+        Just ev -> do+          yield ev+          case ev of+            LLMStart {} -> loop [] [] Nothing+            LLMChunk {..} ->+              let nextText = if T.null chunkText then accText else chunkText : accText+                  nextTools = maybe accTools (: accTools) toolCallDelta+               in loop nextText nextTools mbFinal+            LLMEnd {..} ->+              loop accText accTools (Just finalMessage)+            _ -> loop accText accTools mbFinal++{- | Consume a 'StreamEvent' pipeline, invoking a handler for every event, and+return the final 'Message' from the last 'LLMEnd' event (or synthesize one from+accumulated 'LLMChunk's if no 'LLMEnd' event was emitted).+-}+consumeStreamWith ::+  Monad m =>+  (StreamEvent -> m ()) ->+  ConduitT StreamEvent Void m Message+consumeStreamWith onEvent = tapStreamMessage `fuseUpstream` CL.mapM_ onEvent++-- | Run an 'EventStream' with an event handler and return the resulting 'Message'.+runStreamWith ::+  Monad m =>+  (StreamEvent -> m ()) ->+  EventStream m ->+  m Message+runStreamWith onEvent src = runConduit (src .| consumeStreamWith onEvent)  -- | Collect all emitted events from a stream into a list. collectEvents :: Monad m => EventStream m -> m [StreamEvent]

@@ -14,12 +14,20 @@   ( Tool (..)   , createTool   , toolToValue+  , StreamingTool (..)+  , createStreamingTool+  , fromTool+  , streamToolCall+  , streamStreamingToolCall   ) where +import Control.Monad.Trans (lift) import Data.Aeson+import Data.Conduit (ConduitT, await, fuseUpstream, mapOutput, runConduit, yield) import Data.Text (Text)  import Langchain.Error (LangchainError)+import Langchain.Stream (StreamEvent (ToolChunk, ToolEnd, ToolErrorEvent, ToolStart))  -- | Effect-polymorphic Tool abstraction data Tool m = Tool@@ -63,3 +71,90 @@           , "parameters" .= toolSchema           ]     ]++-- | Effect-polymorphic streaming Tool abstraction supporting incremental 'Text' output chunks.+data StreamingTool m = StreamingTool+  { streamingToolDef :: Tool m+  -- ^ Underlying tool specification and synchronous fallback executor+  , streamingToolExecute :: Value -> ConduitT () Text m (Either LangchainError Text)+  -- ^ Monadic streaming execution yielding output chunks and returning final text output+  }++instance Show (StreamingTool m) where+  show st = "StreamingTool { toolName = " ++ show (toolName (streamingToolDef st)) ++ " }"++-- | Construct a 'StreamingTool' from a name, description, schema, and streaming executor.+createStreamingTool ::+  Monad m =>+  Text ->+  Text ->+  Value ->+  (Value -> ConduitT () Text m (Either LangchainError Text)) ->+  StreamingTool m+createStreamingTool name desc schema streamExec =+  StreamingTool+    { streamingToolDef =+        createTool name desc schema $ \args ->+          runConduit (streamExec args `fuseUpstream` drainSink)+    , streamingToolExecute = streamExec+    }+  where+    drainSink = do+      mItem <- await+      case mItem of+        Nothing -> pure ()+        Just _ -> drainSink++-- | Lift a standard 'Tool' into a 'StreamingTool' that emits its full output as a single chunk.+fromTool :: Monad m => Tool m -> StreamingTool m+fromTool t =+  StreamingTool+    { streamingToolDef = t+    , streamingToolExecute = \args -> do+        res <- lift $ toolExecute t args+        case res of+          Right out -> yield out >> pure (Right out)+          Left err -> pure (Left err)+    }++{- | Execute a 'Tool' within a 'StreamEvent' pipeline, emitting 'ToolStart'+followed by 'ToolEnd' or 'ToolErrorEvent'.+-}+streamToolCall ::+  Monad m =>+  Text ->+  Tool m ->+  Value ->+  ConduitT () StreamEvent m (Either LangchainError Text)+streamToolCall rId tool args = do+  let tName = toolName tool+  yield $ ToolStart rId tName args+  eRes <- lift $ toolExecute tool args+  case eRes of+    Left err -> do+      yield $ ToolErrorEvent rId tName err+      pure (Left err)+    Right out -> do+      yield $ ToolEnd rId tName (toJSON out)+      pure (Right out)++{- | Execute a 'StreamingTool' within a 'StreamEvent' pipeline, emitting+'ToolStart', 'ToolChunk' for each streamed text chunk, and 'ToolEnd' or 'ToolErrorEvent'.+-}+streamStreamingToolCall ::+  Monad m =>+  Text ->+  StreamingTool m ->+  Value ->+  ConduitT () StreamEvent m (Either LangchainError Text)+streamStreamingToolCall rId sTool args = do+  let tName = toolName (streamingToolDef sTool)+  yield $ ToolStart rId tName args+  eRes <- mapOutput (ToolChunk rId tName) (streamingToolExecute sTool args)+  case eRes of+    Left err -> do+      yield $ ToolErrorEvent rId tName err+      pure (Left err)+    Right out -> do+      yield $ ToolEnd rId tName (toJSON out)+      pure (Right out)

@@ -38,6 +38,7 @@      -- * Tool Security Combinator   , withSecurity+  , withStreamingSecurity   ) where  import Control.Monad.IO.Class (MonadIO, liftIO)@@ -51,8 +52,9 @@ import System.Directory (canonicalizePath, doesDirectoryExist, doesFileExist) import System.FilePath (isAbsolute, normalise, pathSeparator, takeDirectory, takeFileName, (</>)) +import Control.Monad.Trans (lift) import Langchain.Error (LangchainError, toolError)-import Langchain.Tool (Tool (..))+import Langchain.Tool (StreamingTool (..), Tool (..))  -- | Tier levels representing the risk and side-effects of tool execution. data ToolTier@@ -278,6 +280,26 @@                   Nothing                   Nothing +checkPolicyApproval ::+  MonadIO m =>+  ToolTier ->+  SecurityPolicy ->+  Text ->+  Text ->+  Value ->+  m (Either LangchainError ())+checkPolicyApproval tier policy tName tDesc args =+  case checkTierPermission policy tier of+    Left err -> pure $ Left err+    Right () -> do+      mApproval <- case policyApprovalHook policy of+        Nothing -> pure Approved+        Just hook -> liftIO $ hook (ApprovalRequest tName tier args tDesc)+      case mApproval of+        Denied reason ->+          pure (Left (toolError ("Execution rejected by approval policy: " <> reason) (Just tName) Nothing))+        Approved -> pure $ Right ()+ -- | Wrap a 'Tool m' with permission tier enforcement, sandboxing, and optional HITL approval. withSecurity ::   MonadIO m =>@@ -288,17 +310,27 @@ withSecurity tier policy tool@Tool {..} =   tool     { toolExecute = \args -> do-        case checkTierPermission policy tier of+        eOk <- checkPolicyApproval tier policy toolName toolDescription args+        case eOk of           Left err -> pure $ Left err-          Right () -> do-            -- Check HITL approval if hook is defined-            mApproval <- case policyApprovalHook policy of-              Nothing -> pure Approved-              Just hook -> liftIO $ hook (ApprovalRequest toolName tier args toolDescription)-            case mApproval of-              Denied reason ->-                pure $-                  Left $-                    toolError ("Execution rejected by approval policy: " <> reason) (Just toolName) Nothing-              Approved -> toolExecute args+          Right () -> toolExecute args     }++-- | Wrap a 'StreamingTool m' with permission tier enforcement, sandboxing, and optional HITL approval.+withStreamingSecurity ::+  MonadIO m =>+  ToolTier ->+  SecurityPolicy ->+  StreamingTool m ->+  StreamingTool m+withStreamingSecurity tier policy st@StreamingTool {..} =+  let securedDef = withSecurity tier policy streamingToolDef+   in st+        { streamingToolDef = securedDef+        , streamingToolExecute = \args -> do+            eOk <-+              lift $ checkPolicyApproval tier policy (toolName securedDef) (toolDescription securedDef) args+            case eOk of+              Left err -> pure $ Left err+              Right () -> streamingToolExecute args+        }

@@ -1,5 +1,6 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeApplications #-}  {- | Module      : Langchain.Tool.Shell@@ -15,27 +16,96 @@ module Langchain.Tool.Shell   ( shellTool   , secureShellTool+  , streamingShellTool+  , streamingShellToolWithCancel   ) where -import Control.Exception (SomeException, try)+import Control.Concurrent.Async (AsyncCancelled (..), async, cancel, concurrently, wait)+import Control.Concurrent.STM+  ( TBQueue+  , atomically+  , newEmptyTMVarIO+  , newTBQueueIO+  , orElse+  , readTBQueue+  , readTMVar+  , tryPutTMVar+  , writeTBQueue+  )+import Control.Exception (SomeException, catch, finally, fromException, onException, throwIO, try)+import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value (..), object, (.=)) import Data.Aeson.Types (parseEither, (.:?))+import Data.Conduit (ConduitT, yield)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef) import Data.Text (Text) import qualified Data.Text as T+import qualified Data.Text.IO as TIO import System.Exit (ExitCode (..))-import System.Process (readProcessWithExitCode)+import System.IO (BufferMode (LineBuffering), Handle, hClose, hIsEOF, hSetBuffering)+import System.Process+  ( CreateProcess (..)+  , StdStream (..)+  , createProcess+  , interruptProcessGroupOf+  , proc+  , readProcessWithExitCode+  , terminateProcess+  , waitForProcess+  ) -import Langchain.Error (toolError)-import Langchain.Tool (Tool (..), createTool)+import Langchain.Error (LangchainError, toolError)+import Langchain.Stream (CancelToken, isCancelled, waitCancel)+import Langchain.Tool (StreamingTool, Tool (..), createStreamingTool, createTool) import Langchain.Tool.Security   ( SecurityPolicy   , ToolTier (..)   , defaultSecurityPolicy   , validateCommandSafety   , withSecurity+  , withStreamingSecurity   ) +shellToolDescription :: Text+shellToolDescription =+  "Execute a shell command line (e.g. bash/sh) and return stdout and stderr output."++shellToolSchema :: Value+shellToolSchema =+  object+    [ "type" .= ("object" :: Text)+    , "properties"+        .= object+          [ "command"+              .= object+                [ "type" .= ("string" :: Text)+                , "description" .= ("The shell command line to execute" :: Text)+                ]+          ]+    , "required" .= (["command"] :: [Text])+    ]++extractShellCommand :: Value -> Either LangchainError Text+extractShellCommand = \case+  Object o -> case parseEither (.:? "command") o of+    Right (Just cmd) -> Right cmd+    _ -> Left $ toolError "Missing 'command' parameter" (Just "shell_command") Nothing+  _ -> Left $ toolError "Invalid arguments object" (Just "shell_command") Nothing++formatShellResult :: ExitCode -> Text -> Text -> Text+formatShellResult ExitSuccess stdoutTxt stderrTxt =+  let out = T.strip stdoutTxt+      err = T.strip stderrTxt+   in if T.null out+        then if T.null err then "Command completed with no output." else err+        else out+formatShellResult (ExitFailure code) stdoutTxt stderrTxt =+  let combined = T.strip (stdoutTxt <> "\n" <> stderrTxt)+   in "Command exited with code "+        <> T.pack (show code)+        <> (if T.null combined then "" else ": " <> combined)+ -- | Tool that executes a shell command with default security policy. shellTool :: MonadIO m => Tool m shellTool = secureShellTool defaultSecurityPolicy@@ -46,42 +116,158 @@   withSecurity TierExecution policy $     createTool       "shell_command"-      "Execute a shell command line (e.g. bash/sh) and return stdout and stderr output."-      ( object-          [ "type" .= ("object" :: Text)-          , "properties"-              .= object-                [ "command"-                    .= object-                      [ "type" .= ("string" :: Text)-                      , "description" .= ("The shell command line to execute" :: Text)-                      ]-                ]-          , "required" .= (["command"] :: [Text])-          ]+      shellToolDescription+      shellToolSchema+      ( \args -> case extractShellCommand args of+          Left err -> pure $ Left err+          Right cmd -> case validateCommandSafety policy cmd of+            Left err -> pure $ Left err+            Right () -> do+              eRes <- liftIO $ try (readProcessWithExitCode "sh" ["-c", T.unpack cmd] "")+              case eRes of+                Left err ->+                  pure $ Left $ toolError (T.pack $ show (err :: SomeException)) (Just "shell_command") Nothing+                Right (code, stdoutStr, stderrStr) ->+                  pure $ Right $ formatShellResult code (T.pack stdoutStr) (T.pack stderrStr)       )-      ( \case-          Object o -> case parseEither (.:? "command") o of-            Right (Just cmd) -> case validateCommandSafety policy cmd of-              Left err -> pure $ Left err-              Right () -> do-                eRes <- liftIO $ try (readProcessWithExitCode "sh" ["-c", T.unpack cmd] "")-                case eRes of-                  Left err ->-                    pure $ Left $ toolError (T.pack $ show (err :: SomeException)) (Just "shell_command") Nothing-                  Right (ExitSuccess, stdoutStr, stderrStr) ->-                    let out = T.strip (T.pack stdoutStr)-                        err = T.strip (T.pack stderrStr)-                     in if T.null out-                          then if T.null err then pure $ Right "Command completed with no output." else pure $ Right err-                          else pure $ Right out-                  Right (ExitFailure code, stdoutStr, stderrStr) ->-                    let combined = T.strip (T.pack (stdoutStr <> "\n" <> stderrStr))-                     in pure $-                          Right $-                            "Command exited with code "-                              <> T.pack (show code)-                              <> (if T.null combined then "" else ": " <> combined)-            _ -> pure $ Left $ toolError "Missing 'command' parameter" (Just "shell_command") Nothing-          _ -> pure $ Left $ toolError "Invalid arguments object" (Just "shell_command") Nothing++-- | Streaming shell tool that yields stdout and stderr lines incrementally while enforcing 'SecurityPolicy'.+streamingShellTool :: MonadIO m => SecurityPolicy -> StreamingTool m+streamingShellTool = makeStreamingShellTool Nothing++{- | Streaming shell tool that monitors a 'CancelToken' and terminates the+spawned shell child process immediately on cancellation.+-}+streamingShellToolWithCancel :: MonadIO m => CancelToken -> SecurityPolicy -> StreamingTool m+streamingShellToolWithCancel tok = makeStreamingShellTool (Just tok)++makeStreamingShellTool :: MonadIO m => Maybe CancelToken -> SecurityPolicy -> StreamingTool m+makeStreamingShellTool mbTok policy =+  withStreamingSecurity TierExecution policy $+    createStreamingTool+      "shell_command"+      shellToolDescription+      shellToolSchema+      ( \args -> case extractShellCommand args of+          Left err -> pure $ Left err+          Right cmd -> case validateCommandSafety policy cmd of+            Left err -> pure $ Left err+            Right () -> streamShellCommand mbTok cmd       )++streamShellCommand ::+  MonadIO m =>+  Maybe CancelToken ->+  Text ->+  ConduitT () Text m (Either LangchainError Text)+streamShellCommand mbTok cmd = do+  cancelledNow <- maybe (pure False) isCancelled mbTok+  if cancelledNow+    then pure $ Left $ toolError "Shell command cancelled" (Just "shell_command") Nothing+    else startShellProcess mbTok cmd++startShellProcess ::+  MonadIO m =>+  Maybe CancelToken ->+  Text ->+  ConduitT () Text m (Either LangchainError Text)+startShellProcess mbTok cmd = do+  eProc <-+    liftIO $+      try $+        createProcess+          (proc "sh" ["-c", T.unpack cmd])+            { std_in = NoStream+            , std_out = CreatePipe+            , std_err = CreatePipe+            , create_group = True+            }+  case eProc of+    Left ex ->+      pure $ Left $ toolError (T.pack $ show (ex :: SomeException)) (Just "shell_command") Nothing+    Right (_, Just hOut, Just hErr, ph) -> do+      (queue, doneVar, worker) <- liftIO $ do+        hSetBuffering hOut LineBuffering+        hSetBuffering hErr LineBuffering+        q <- newTBQueueIO 128+        done <- newEmptyTMVarIO+        outRef <- newIORef ([] :: [Text])+        errRef <- newIORef ([] :: [Text])+        resRef <- newIORef (Right "" :: Either LangchainError Text)+        w <-+          async $+            ( do+                _ <- concurrently (pumpHandle hOut q outRef) (pumpHandle hErr q errRef)+                code <- waitForProcess ph+                outChunks <- T.concat . reverse <$> readIORef outRef+                errChunks <- T.concat . reverse <$> readIORef errRef+                writeIORef resRef (Right $ formatShellResult code outChunks errChunks)+            )+              `catch` ( \ex -> case fromException ex of+                          Just AsyncCancelled -> do+                            void (try @SomeException (interruptProcessGroupOf ph))+                            void (try @SomeException (terminateProcess ph))+                            void (try @SomeException (waitForProcess ph))+                            throwIO (ex :: SomeException)+                          Nothing ->+                            writeIORef+                              resRef+                              (Left $ toolError (T.pack $ show (ex :: SomeException)) (Just "shell_command") Nothing)+                      )+              `finally` ( do+                            void (try @SomeException (hClose hOut))+                            void (try @SomeException (hClose hErr))+                            readIORef resRef >>= atomically . void . tryPutTMVar done+                        )+        pure (q, done, w)+      drainLoop queue doneVar worker+    Right (_, mbOut, mbErr, ph) -> do+      liftIO $ do+        mapM_ (try @SomeException . hClose) mbOut+        mapM_ (try @SomeException . hClose) mbErr+        void (try @SomeException (interruptProcessGroupOf ph))+        void (try @SomeException (terminateProcess ph))+      pure $+        Left $+          toolError "Failed to open stdout/stderr pipes for shell_command" (Just "shell_command") Nothing+  where+    pumpHandle :: Handle -> TBQueue Text -> IORef [Text] -> IO ()+    pumpHandle h q ref = loop+      where+        loop = do+          eof <- hIsEOF h+          if eof+            then pure ()+            else do+              line <- TIO.hGetLine h+              let chunk = line <> "\n"+              modifyIORef' ref (chunk :)+              atomically $ writeTBQueue q chunk+              loop++    drainLoop queue doneVar worker = loop+      where+        loop = do+          next <-+            liftIO $+              atomically+                ( case mbTok of+                    Just tok ->+                      (Left Nothing <$ waitCancel tok)+                        `orElse` (Right <$> readTBQueue queue)+                        `orElse` (Left . Just <$> readTMVar doneVar)+                    Nothing ->+                      (Right <$> readTBQueue queue)+                        `orElse` (Left . Just <$> readTMVar doneVar)+                )+                `onException` cancel worker+          case next of+            Right chunk -> do+              yield chunk+              loop+            Left Nothing -> do+              liftIO $ cancel worker+              pure $ Left $ toolError "Shell command cancelled" (Just "shell_command") Nothing+            Left (Just res) -> do+              void $ liftIO $ wait worker+              pure res

@@ -3,6 +3,7 @@ import Test.Tasty  import qualified Test.Langchain.Callback as CallbackTest+import qualified Test.Langchain.Eval as EvalTest import qualified Test.Langchain.Model as ModelTest import qualified Test.Langchain.Monad as MonadTest import qualified Test.Langchain.Observability as ObservabilityTest@@ -30,4 +31,5 @@     , CallbackTest.tests     , RetrieverTest.tests     , ResilienceTest.tests+    , EvalTest.tests     ]

@@ -0,0 +1,109 @@+{-# LANGUAGE OverloadedStrings #-}++module Test.Langchain.Eval (tests) where++import Control.Monad.Except (runExceptT)+import qualified Data.Aeson as Aeson+import Data.Text (Text)+import Test.Tasty+import Test.Tasty.HUnit++import Langchain.Agent.ReAct (AgentStep (..), AgentTrace (..))+import Langchain.Error (LangchainError)+import Langchain.Eval+import Langchain.Model.Types (Role (..), ToolCall (..), assistantMessage, textMessage)+import Test.Langchain.TestModel (TestChatModel (..))++tests :: TestTree+tests =+  testGroup+    "Langchain.Eval"+    [ testCase "runEvalSuite computes pass rate, mean score, and latency across mixed cases" $ do+        let cases :: [EvalCase Text Text]+            cases =+              [ EvalCase "query-1" (Just "answer-1") Nothing+              , EvalCase "query-2" (Just "answer-2") Nothing+              ]+            mockRunner :: Text -> IO (Either LangchainError Text)+            mockRunner "query-1" = pure $ Right "answer-1"+            mockRunner "query-2" = pure $ Right "wrong-answer"+            mockRunner _ = pure $ Right ""+        report <- runEvalSuite mockRunner [exactMatchScorer] cases+        reportTotalCases report @?= 2+        reportPassedCases report @?= 1+        reportPassRate report @?= 0.5+        reportMeanScore report @?= 0.5+        assertBool "Non-negative latency" (reportMeanLatencyMicros report >= 0)+        length (reportCases report) @?= 2+    , testCase "containsScorer verifies substring presence" $ do+        let cPass :: EvalCase Text Text+            cPass = EvalCase "in" (Just "needle") Nothing+            cFail :: EvalCase Text Text+            cFail = EvalCase "in" (Just "missing") Nothing+        scPass <- scoreCase containsScorer cPass ("haystack needle haystack" :: Text)+        scorePassed scPass @?= True+        scoreValue scPass @?= 1.0+        scFail <- scoreCase containsScorer cFail ("haystack other haystack" :: Text)+        scorePassed scFail @?= False+        scoreValue scFail @?= 0.0+    , testCase "toolCallExactScorer verifies sequence of invoked tool names in AgentTrace" $ do+        let tc1 = ToolCall "c1" "function" "search" (Aeson.object [])+            tc2 = ToolCall "c2" "function" "calculator" (Aeson.object [])+            action1 = AgentAction (assistantMessage "") [tc1]+            action2 = AgentAction (assistantMessage "") [tc2]+            finish = AgentFinish (assistantMessage "Done")+            obs = textMessage Tool "obs"+            trace =+              AgentTrace+                { traceSteps = [(action1, [obs]), (action2, [obs]), (finish, [])]+                , traceFinalAnswer = assistantMessage "Done"+                , traceIterations = 3+                }+            matchingScorer = toolCallExactScorer ["search", "calculator"]+            mismatchScorer = toolCallExactScorer ["calculator", "search"]+            evalCase = EvalCase ("test" :: Text) Nothing Nothing+        resMatch <- scoreCase matchingScorer evalCase trace+        scorePassed resMatch @?= True+        scoreValue resMatch @?= 1.0+        resMismatch <- scoreCase mismatchScorer evalCase trace+        scorePassed resMismatch @?= False+        scoreValue resMismatch @?= 0.0+    , testCase "llmJudgeScorer parses JSON judge evaluation and score" $ do+        let judgeModel =+              TestChatModel+                "{\"passed\": true, \"score\": 0.95, \"reason\": \"Accurate and concise.\"}"+                "judge-mock"+            scorer = llmJudgeScorer judgeModel "Evaluate answer accuracy"+            evalCase :: EvalCase Text Text+            evalCase = EvalCase "What is 2+2?" (Just "4") Nothing+        eScore <- runExceptT $ scoreCase scorer evalCase "4"+        case eScore of+          Left err -> assertFailure ("Judge scorer failed: " ++ show err)+          Right score -> do+            scorePassed score @?= True+            scoreValue score @?= 0.95+            scoreReason score @?= Just "Accurate and concise."+    , testCase "EvalReport round-trips cleanly via Aeson encode and decode" $ do+        let score = EvalScore 1.0 True (Just "Matched perfectly")+            caseRes =+              CaseResult+                { caseInput = Aeson.String "input-1"+                , caseExpected = Just (Aeson.String "expected-1")+                , caseActual = Right (Aeson.String "actual-1")+                , caseScores = [CaseScoreResult "exact_match" score]+                , caseLatencyMicros = 1200+                , casePassed = True+                }+            report =+              EvalReport+                { reportTotalCases = 1+                , reportPassedCases = 1+                , reportPassRate = 1.0+                , reportMeanScore = 1.0+                , reportMeanLatencyMicros = 1200+                , reportCases = [caseRes]+                }+            encoded = Aeson.encode report+            decoded = Aeson.decode encoded+        decoded @?= Just report+    ]

@@ -1,3 +1,4 @@+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}  module Test.Langchain.Stream (tests) where@@ -5,10 +6,21 @@ import Test.Tasty import Test.Tasty.HUnit -import Control.Monad.Except (runExceptT)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, tryTakeMVar)+import Control.Exception (finally)+import Control.Monad.Except (runExceptT, throwError)+import Control.Monad.IO.Class (liftIO) import Control.Monad.Trans.Resource (runResourceT)+import Data.Aeson (object, (.=))+import Data.Conduit (runConduit, yield, (.|))+import qualified Data.Conduit.List as CL+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.Time.Clock (diffUTCTime, getCurrentTime) +import Langchain.Error (llmError) import Langchain.Model+import Langchain.Monad (runLangchainT) import Langchain.Stream import Test.Langchain.TestModel (TestChatModel (..)) @@ -30,4 +42,94 @@                 chunkText c @?= "Streamed content"                 extractMessageText (finalMessage e) @?= "Streamed content"               _ -> assertFailure $ "Unexpected event sequence: " ++ show events+    , testCase "liftChatStream forwards events into LangchainT () IO" $ do+        let model = TestChatModel "Bridged stream" "mock-gpt"+            input = [userMessage "Bridge test"]+        res <- runLangchainT () $ collectEvents $ liftChatStream (stream model input Nothing)+        case res of+          Left err -> assertFailure $ "Unexpected error: " ++ show err+          Right events -> do+            length events @?= 3+            case reverse events of+              LLMEnd _ msg _ : _ -> extractMessageText msg @?= "Bridged stream"+              _ -> assertFailure $ "Expected LLMEnd, got: " ++ show events+    , testCase "liftChatStream propagates LangchainError mid-stream" $ do+        let errStream :: ChatStream+            errStream = do+              yield $ LLMStart "r1" "mock" [userMessage "Hi"]+              yield $ LLMChunk "r1" "partial" Nothing+              throwError $ llmError "mid-stream boom" Nothing Nothing+        res <- runExceptT $ collectEvents $ liftChatStream errStream+        case res of+          Left _ -> pure ()+          Right events -> assertFailure $ "Expected failure, got: " ++ show events+    , testCase "runStreamWith invokes callback and resets across LLMStart turns" $ do+        let tc = ToolCall "call_1" "function" "calc" (object ["x" .= (1 :: Int)])+            multiTurnStream :: EventStream IO+            multiTurnStream = do+              yield $ LLMStart "r1" "mock" []+              yield $ LLMChunk "r1" "Calling tool..." Nothing+              yield $ LLMChunk "r1" "" (Just tc)+              yield $ LLMEnd "r1" ((assistantMessage "Calling tool...") {messageToolCalls = Just [tc]}) Nothing+              yield $ ToolStart "r1" "calc" (object ["x" .= (1 :: Int)])+              yield $ ToolEnd "r1" "calc" (object ["res" .= (2 :: Int)])+              yield $ LLMStart "r2" "mock" []+              yield $ LLMChunk "r2" "Final " Nothing+              yield $ LLMChunk "r2" "answer" Nothing+        seenRef <- newIORef ([] :: [StreamEvent])+        finalMsg <- runStreamWith (\ev -> liftIO $ modifyIORef' seenRef (ev :)) multiTurnStream+        seen <- readIORef seenRef+        length seen @?= 9+        extractMessageText finalMsg @?= "Final answer"+        messageToolCalls finalMsg @?= Nothing+    , testCase "liftChatStreamResource cancels upstream within 200ms on take 1" $ do+        cleanedUp <- newEmptyMVar+        let slowUpstream :: ChatStream+            slowUpstream = do+              yield $ LLMStart "slow" "mock" []+              liftIO $+                threadDelay 5000000+                  `finally` putMVar cleanedUp ()+              yield $ LLMChunk "slow" "late" Nothing+        t0 <- getCurrentTime+        res <-+          runResourceT $+            runExceptT $+              runConduit $+                liftChatStreamResource slowUpstream .| CL.take 1+        t1 <- getCurrentTime+        case res of+          Left err -> assertFailure $ "Unexpected error: " ++ show err+          Right evs -> length evs @?= 1+        threadDelay 50000+        mbDone <- tryTakeMVar cleanedUp+        mbDone @?= Just ()+        let elapsed = realToFrac (diffUTCTime t1 t0) :: Double+        assertBool ("Expected cancellation < 0.2s, got: " ++ show elapsed) (elapsed < 0.2)+    , testCase "liftChatStreamWithCancel stops on CancelToken and keeps partial text" $ do+        tok <- newCancelToken+        cleanedUp <- newEmptyMVar+        gate <- newEmptyMVar :: IO (MVar ())+        let partialUpstream :: ChatStream+            partialUpstream = do+              yield $ LLMStart "c1" "mock" []+              yield $ LLMChunk "c1" "partial-" Nothing+              liftIO $+                takeMVar gate+                  `finally` putMVar cleanedUp ()+              yield $ LLMChunk "c1" "never" Nothing+        res <-+          runLangchainT () $+            runStreamWith+              ( \case+                  LLMChunk {} -> triggerCancel tok+                  _ -> pure ()+              )+              (liftChatStreamWithCancel tok partialUpstream)+        case res of+          Left err -> assertFailure $ "Unexpected error: " ++ show err+          Right msg -> extractMessageText msg @?= "partial-"+        threadDelay 50000+        mbDone <- tryTakeMVar cleanedUp+        mbDone @?= Just ()     ]

@@ -3,11 +3,25 @@  module Test.Langchain.Tool (tests) where +import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, cancel, wait) import Data.Aeson import Data.Aeson.Types (parseEither)+import Data.Conduit (fuseUpstream, runConduit, yield)+import qualified Data.Conduit.List as CL+import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Text (Text)+import qualified Data.Text as T+import Data.Time.Clock (diffUTCTime, getCurrentTime) import Langchain.Error+import Langchain.Stream+  ( StreamEvent (ToolChunk, ToolEnd, ToolErrorEvent, ToolStart)+  , newCancelToken+  , triggerCancel+  ) import Langchain.Tool+import Langchain.Tool.Security (readOnlySecurityPolicy, unrestrictedSecurityPolicy)+import Langchain.Tool.Shell (streamingShellTool, streamingShellToolWithCancel) import Test.Tasty import Test.Tasty.HUnit @@ -36,4 +50,104 @@         let t = createTool "calculator" "Performs math" (object []) calcExec         res <- toolExecute t (object [])         res @?= Right "42"+    , testCase "streamToolCall emits ToolStart and ToolEnd on success" $ do+        let t = createTool "calculator" "Performs math" (object []) calcExec+        eventsRef <- newIORef ([] :: [StreamEvent])+        res <-+          runConduit $+            streamToolCall "run-1" t (object [])+              `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev]))+        res @?= Right "42"+        events <- readIORef eventsRef+        events+          @?= [ ToolStart "run-1" "calculator" (object [])+              , ToolEnd "run-1" "calculator" (toJSON ("42" :: Text))+              ]+    , testCase "streamToolCall emits ToolStart and ToolErrorEvent on failure" $ do+        let t = createTool "calculator" "Performs math" (object []) calcExec+        eventsRef <- newIORef ([] :: [StreamEvent])+        res <-+          runConduit $+            streamToolCall "run-2" t (Number 1)+              `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev]))+        case res of+          Left _ -> pure ()+          Right out -> assertFailure $ "Expected error, got: " ++ show out+        events <- readIORef eventsRef+        case events of+          [ToolStart "run-2" "calculator" (Number 1), ToolErrorEvent "run-2" "calculator" _] -> pure ()+          _ -> assertFailure $ "Unexpected events: " ++ show events+    , testCase "streamStreamingToolCall emits incremental ToolChunk events" $ do+        let sTool =+              createStreamingTool "streamer" "Streams chunks" (object []) $ \_ -> do+                yield "chunk-1 "+                yield "chunk-2"+                pure (Right "chunk-1 chunk-2")+        eventsRef <- newIORef ([] :: [StreamEvent])+        res <-+          runConduit $+            streamStreamingToolCall "run-3" sTool (object [])+              `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev]))+        res @?= Right "chunk-1 chunk-2"+        events <- readIORef eventsRef+        events+          @?= [ ToolStart "run-3" "streamer" (object [])+              , ToolChunk "run-3" "streamer" "chunk-1 "+              , ToolChunk "run-3" "streamer" "chunk-2"+              , ToolEnd "run-3" "streamer" (toJSON ("chunk-1 chunk-2" :: Text))+              ]+    , testCase "streamingShellTool streams stdout lines and enforces SecurityPolicy" $ do+        let sShell = streamingShellTool unrestrictedSecurityPolicy+        eventsRef <- newIORef ([] :: [StreamEvent])+        res <-+          runConduit $+            streamStreamingToolCall+              "run-sh"+              sShell+              (object ["command" .= ("printf 'line1\\nline2\\n'" :: Text)])+              `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev]))+        res @?= Right "line1\nline2"+        events <- readIORef eventsRef+        let chunks = [txt | ToolChunk _ _ txt <- events]+        chunks @?= ["line1\n", "line2\n"]++        let roShell = streamingShellTool readOnlySecurityPolicy+        roRes <- toolExecute (streamingToolDef roShell) (object ["command" .= ("echo hi" :: Text)])+        case roRes of+          Left err -> assertBool "Mentions permission" ("permission" `T.isInfixOf` T.toLower (errorMessage err))+          Right ok -> assertFailure $ "Expected security denial, got: " ++ show ok+    , testCase "cancelling streamingShellTool via Async.cancel terminates sleep child promptly" $ do+        let sShell = streamingShellTool unrestrictedSecurityPolicy+        t0 <- getCurrentTime+        a <-+          async $+            runConduit $+              streamStreamingToolCall+                "run-cancel-async"+                sShell+                (object ["command" .= ("sleep 10" :: Text)])+                `fuseUpstream` CL.sinkNull+        threadDelay 100000+        cancel a+        t1 <- getCurrentTime+        let elapsed = realToFrac (diffUTCTime t1 t0) :: Double+        assertBool ("Expected prompt Async cancellation (< 2s), got: " ++ show elapsed) (elapsed < 2.0)+    , testCase "cancelling streamingShellToolWithCancel via CancelToken terminates promptly" $ do+        tok <- newCancelToken+        let sShell = streamingShellToolWithCancel tok unrestrictedSecurityPolicy+        t0 <- getCurrentTime+        a <-+          async $+            runConduit $+              streamStreamingToolCall+                "run-cancel-tok"+                sShell+                (object ["command" .= ("sleep 10" :: Text)])+                `fuseUpstream` CL.sinkNull+        threadDelay 100000+        triggerCancel tok+        _ <- wait a+        t1 <- getCurrentTime+        let elapsed = realToFrac (diffUTCTime t1 t0) :: Double+        assertBool ("Expected prompt CancelToken termination (< 2s), got: " ++ show elapsed) (elapsed < 2.0)     ]