diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +## 0.3.2.0 — 2026-11-03 + +### Added + +- Capture valid atomic structured streams through the existing response codec. +- Replay structured recordings through either Complete or Stream, including + changed local decision calibration, without provider dispatch or fresh evidence. + +### Changed + +- Require `shikumi ^>=0.5.1.0` for pure checked response/event helpers. +- Keep text streams usage-only; missing, malformed and text-only stream recordings + fail closed with ReplayDivergence. Serialized responses still omit call evidence. + ## 0.3.1.0 — 2026-11-02 ### Changed diff --git a/shikumi-trace.cabal b/shikumi-trace.cabal --- a/shikumi-trace.cabal +++ b/shikumi-trace.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: shikumi-trace -version: 0.3.1.0 +version: 0.3.2.0 synopsis: Hierarchical tracing, observability, and deterministic replay for shikumi (EP-7) @@ -67,7 +67,7 @@ generic-lens >=2.2 && <2.4, lens ^>=5.3, scientific >=0.3 && <0.4, - shikumi ^>=0.5.0.0, + shikumi ^>=0.5.1.0, shikumi-cache ^>=0.2.1.0, text ^>=2.1, time >=1.12 && <1.17, @@ -83,7 +83,7 @@ build-depends: base, - shikumi-trace ^>=0.3.1.0, + shikumi-trace ^>=0.3.2.0, test-suite shikumi-trace-test import: common-options @@ -109,10 +109,10 @@ effectful, generic-lens, lens ^>=5.3, - shikumi ^>=0.5.0.0, + shikumi ^>=0.5.1.0, shikumi-cache ^>=0.2.1.0, shikumi-testing, - shikumi-trace ^>=0.3.1.0, + shikumi-trace ^>=0.3.2.0, tasty, tasty-hunit, tasty-quickcheck, diff --git a/src/Shikumi/Trace.hs b/src/Shikumi/Trace.hs --- a/src/Shikumi/Trace.hs +++ b/src/Shikumi/Trace.hs @@ -92,6 +92,7 @@ import Shikumi.Effect.Time (Time, getCurrentTime) import Shikumi.LLM (LLM (..), complete, stream) import Shikumi.LLM.Observation (BillingSummary, UsageRecord (..), observedModelOf) +import Shikumi.Stream (atomicEventsFromResponse, responseFromEvents) import Shikumi.Trace.Node (NodePath (..)) import Shikumi.Trace.ResponseJSON () import Text.Read (readMaybe) @@ -356,8 +357,8 @@ -- | Capture every LM call as a leaf 'LlmCallSpan' under the active span. Interpose -- on EP-1's @LLM@ effect: open a span, delegate to the underlying handler, then --- fill the span's attributes from the returned 'Response' (and the request). Streaming terminals contribute usage and observed identity; only successful --- blocking responses produce replay records. +-- fill the span's attributes from the returned 'Response' (and the request). Streaming terminals contribute usage and observed identity; successful blocking responses and atomic structured +-- streams produce replay records. Text streams retain usage-only capture. -- -- This capture uses 'runTrace'\'s sequential span stack. Compose it with -- 'Shikumi.Program.runProgram'; concurrent program execution is intentionally @@ -370,20 +371,26 @@ pure resp Stream m c o -> withSpan LlmCallSpan (llmLabel m) $ do evs <- stream m c o - case [t | EventDone t <- evs] of - t : _ -> case t ^. #message of - AssistantMessage p -> annotateSpan $ \a -> - a - & #model ?~ (m ^. #modelId) - & #provider ?~ (m ^. #provider) - & #inputTokens ?~ (p ^. #usage . #inputTokens) - & #outputTokens ?~ (p ^. #usage . #outputTokens) - & #costUsd ?~ realToFrac (p ^. #usage . #cost . #usd :: Rational) - & #billingQuality ?~ UsageRecord (p ^. #usage) - & #observedModel .~ observedModelOf (t ^. #evidence) - _ -> pure () - [] -> pure () + case responseFromEvents m evs of + Right resp | Just _ <- atomicEventsFromResponse resp -> annotateSpan (const (llmAttrs m c o resp)) + _ -> usageOnly m evs pure evs + +usageOnly :: (Trace :> es) => Model -> [AssistantMessageEvent] -> Eff es () +usageOnly m evs = + case [t | EventDone t <- evs] of + t : _ -> case t ^. #message of + AssistantMessage p -> annotateSpan $ \a -> + a + & #model ?~ (m ^. #modelId) + & #provider ?~ (m ^. #provider) + & #inputTokens ?~ (p ^. #usage . #inputTokens) + & #outputTokens ?~ (p ^. #usage . #outputTokens) + & #costUsd ?~ realToFrac (p ^. #usage . #cost . #usd :: Rational) + & #billingQuality ?~ UsageRecord (p ^. #usage) + & #observedModel .~ observedModelOf (t ^. #evidence) + _ -> pure () + [] -> pure () -- | The label for an LM-call span: @provider/model-id@. llmLabel :: Model -> Text diff --git a/src/Shikumi/Trace/Replay.hs b/src/Shikumi/Trace/Replay.hs --- a/src/Shikumi/Trace/Replay.hs +++ b/src/Shikumi/Trace/Replay.hs @@ -37,6 +37,7 @@ import GHC.Generics (Generic) import Shikumi.Cache.Key (CacheKey (..), cacheKey) import Shikumi.LLM (LLM (..)) +import Shikumi.Stream (atomicEventsFromResponse) import Shikumi.Trace.ResponseJSON () -- | Raised when a replayed program issues an LM call that the recorded trace @@ -54,7 +55,8 @@ -- | Interpret the @LLM@ effect by lookup in a replay index instead of calling a -- provider. A hit returns the recorded response; a miss raises t'ReplayDivergence'. --- Streaming completions are not replayable and raise a divergence naming the key. +-- Structured recordings can also supply atomic streams; text-only recordings +-- raise a divergence on Stream. Both operations share the same request key. -- -- No @IOE@ is required: the lookup and decode are pure, and t'ReplayDivergence' is -- raised with @effectful@'s pure-in-@Eff@ 'throwIO'. The contract the plan sketched @@ -70,9 +72,11 @@ Error e -> throwIO $ (divergence key m c) & #promptSummary .~ ("recorded response failed to decode: " <> T.pack e) - Stream m c o -> - throwIO $ - (divergence (cacheKey m c o) m c) & #promptSummary .~ "replay does not support streaming completions" + Stream m c o -> do + let key = cacheKey m c o + case Map.lookup key idx of + Just v | Success r <- fromJSON v, Just events <- atomicEventsFromResponse r -> pure events + _ -> throwIO $ (divergence key m c) & #promptSummary .~ "missing or invalid structured stream recording" -- | Build a t'ReplayDivergence' for a request. divergence :: CacheKey -> Model -> Context -> ReplayDivergence diff --git a/test/DecisionReplaySpec.hs b/test/DecisionReplaySpec.hs --- a/test/DecisionReplaySpec.hs +++ b/test/DecisionReplaySpec.hs @@ -1,44 +1,71 @@ module DecisionReplaySpec (tests) where +import Baikai qualified as B +import Control.Lens ((^.)) +import Data.Aeson (eitherDecode, encode) +import Data.Generics.Labels () import Data.Map.Strict qualified as Map import Data.Vector qualified as V import Effectful (runEff) +import Effectful.Dispatch.Dynamic (interpret) import Effectful.Error.Static (runErrorNoCallStack) +import Effectful.Exception (try) import Effectful.Prim (runPrim) import Shikumi.Decision import Shikumi.Decision.Setting import Shikumi.Effect.Time (runTime) import Shikumi.Error (ShikumiError) +import Shikumi.LLM (LLM (..)) +import Shikumi.LLM qualified import Shikumi.Program import Shikumi.Routing import Shikumi.Schema (field) +import Shikumi.Stream import Shikumi.Testing.DecisionFixtures -import Shikumi.Testing.StubLLM (runStubLLM) +import Shikumi.Testing.Response (mkTextResponse) import Shikumi.Testing.SystemOne import Shikumi.Trace (runTrace, tracedLLM) -import Shikumi.Trace.Replay (runLLMReplay) +import Shikumi.Trace.Replay (ReplayDivergence, runLLMReplay) import Shikumi.Trace.Store (replayIndex) import Test.Tasty import Test.Tasty.HUnit tests :: TestTree tests = - testGroup - "Decision replay" - [ testCase "new settings derive a different answer from a recorded response" $ do + testGroup "Decision replay" $ + [ testCase ("capture " ++ show capturedStream ++ ", replay " ++ show replayStream) $ do let program = Predict assessmentSig emptyParams input = Ticket (field "x") - responder = systemOneResponder (const [("urgent", FieldNoul (NoulEvidence 0.6)), ("severity", FieldScore (ScoreEvidence (V.fromList [0.1, 0.3, 0.6]) Nothing)), ("category", FieldChoice (ChoiceEvidence (Map.fromList [("billing", 0.2), ("technical", 0.8)]) Nothing))]) - recorded <- runEff . runPrim . runTime . runErrorNoCallStack @ShikumiError . runRouting (systemOneModel "jev-latest") . runTrace . runStubLLM responder . tracedLLM . routeLLM $ runProgram program input + response = systemOneResponse [("urgent", FieldNoul (NoulEvidence 0.6)), ("severity", FieldScore (ScoreEvidence (V.fromList [0.1, 0.3, 0.6]) (Just 0.55))), ("category", FieldChoice (ChoiceEvidence (Map.fromList [("billing", 0.2), ("technical", 0.8)]) (Just 0.65)))] + execute streaming p = if streaming then streamProgram p input (const (pure ())) else runProgram p input + handler = interpret $ \_ -> \case + Complete {} -> pure response + Stream {} -> maybe (error "fixture") pure (atomicEventsFromResponse response) + recorded <- runEff . runPrim . runTime . runErrorNoCallStack @ShikumiError . runRouting (systemOneModel "jev-latest") . runTrace . handler . tracedLLM . routeLLM $ execute capturedStream program case recorded of Left err -> assertFailure (show err) - Right (before, tree) -> case replayIndex tree of - Left err -> assertFailure (show err) - Right index -> do - replayed <- - runEff . runErrorNoCallStack @ShikumiError . runRouting (systemOneModel "jev-latest") . runLLMReplay index . routeLLM $ - runProgram (mapParams (setDecisionSetting "severity" (ScoreSetting [0.5, 1.6])) program) input - case replayed of - Left err -> assertFailure (show err) - Right after -> assertBool "cuts affect replayed level" (after /= before) + Right (before, tree) -> do + -- Serialization is part of acceptance, including distributions, + -- confidence and method. The response codec deliberately drops evidence. + eitherDecode (encode response) @?= Right response + restored <- either assertFailure pure (eitherDecode (encode tree)) + case replayIndex restored of + Left err -> assertFailure (show err) + Right index -> do + replayed <- runEff . runErrorNoCallStack @ShikumiError . runRouting (systemOneModel "jev-latest") . runLLMReplay index . routeLLM $ execute replayStream program + replayed @?= Right before + calibrated <- + runEff . runErrorNoCallStack @ShikumiError . runRouting (systemOneModel "jev-latest") . runLLMReplay index . routeLLM $ + execute replayStream (mapParams (setDecisionSetting "severity" (ScoreSetting [0.5, 1.6])) program) + case calibrated of + Left err -> assertFailure (show err) + Right after -> assertBool "cuts rederive level" (after /= before) + | capturedStream <- [False, True], + replayStream <- [False, True] ] + ++ [ testCase "text-only recording cannot become a structured stream" $ do + let text = mkTextResponse "hello" + atomicEventsFromResponse text @?= Nothing + result <- runEff . try @ReplayDivergence . runLLMReplay Map.empty $ B.emptyResponse <$ Shikumi.LLM.stream B.emptyModel B.emptyContext B.emptyOptions + assertBool "divergence" (case result of Left _ -> True; _ -> False) + ]