diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,32 @@ ## Unreleased +## 0.5.1.0 — 2026-11-03 + +### Added + +- Support typed decision outputs through `streamProgram` and `routeLLM`, with + shared request preparation, immutable calibration, and checked atomic terminals. +- Add `preparePredict`, `responseFromEvents`, and `atomicEventsFromResponse`. +- Add `Shikumi.LLM.Evidence`, effectful evidence-aware bare and resilient runners, + and `withEvidenceRuns` for per-operation run IDs. Acknowledgement occurs after + budget and billing settlement and before retry, failure or successful return. + +### Fixed + +- Increment Baikai attempt provenance across retries and link superseded records. +- Make typed evidence-observer failures terminal across both retry layers. +- Fail closed when required evidence cannot accompany a thrown transport error. + +### Limits + +- Decision streams deliver status only; TypeSafe answers atomically. The LLM + effect collects events before invoking program callbacks. +- Cancelled attempts have no receipt until upstream exposure is released at + `mori://shinzui/baikai/okf/improvement-requests/concepts/IR-13`. +- Full evidence can contain provider response text; AttemptReceipt deliberately + has no JSON instances. Existing LLMConfig and runner signatures are unchanged. + ## 0.5.0.0 — 2026-11-02 ### Added diff --git a/shikumi.cabal b/shikumi.cabal --- a/shikumi.cabal +++ b/shikumi.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: shikumi -version: 0.5.0.0 +version: 0.5.1.0 synopsis: Typed, structured, evaluable LM programs over baikai category: AI description: @@ -51,6 +51,7 @@ Shikumi.LLM.Budget Shikumi.LLM.Continuation Shikumi.LLM.Defaults + Shikumi.LLM.Evidence Shikumi.LLM.Observation Shikumi.Module Shikumi.Multimodal @@ -99,6 +100,7 @@ other-modules: AdapterSpec + AttemptEvidenceSpec CliSchemaSpec CombinatorSpec ConstraintSpec @@ -148,11 +150,12 @@ effectful, generic-lens, lens, - shikumi ^>=0.5.0.0, + shikumi ^>=0.5.1.0, stm, streamly-core, tasty, tasty-hunit, tasty-quickcheck, text, + time, vector, diff --git a/src/Shikumi/LLM.hs b/src/Shikumi/LLM.hs --- a/src/Shikumi/LLM.hs +++ b/src/Shikumi/LLM.hs @@ -22,11 +22,13 @@ LLM (..), complete, stream, + streamTerminalError, -- * Bare interpreters runLLM, runLLMWith, runLLMWithObserver, + runLLMWithEvidence, -- * Resilience RetryPolicy (..), @@ -36,6 +38,7 @@ LLMConfig (..), defaultLLMConfig, runLLMResilient, + runLLMResilientWithEvidence, -- * Re-exports of the baikai request/response vocabulary used at call sites Model, @@ -59,8 +62,10 @@ import Baikai.Effectful (Baikai, runBaikai, runBaikaiWith) import Baikai.Effectful qualified as BE import Baikai.Error (BaikaiError) +import Baikai.Evidence qualified as E import Baikai.Provider.Registry (ProviderRegistry) import Baikai.Usage qualified as U +import Control.Applicative ((<|>)) import Control.Concurrent.STM ( TVar, modifyTVar', @@ -69,13 +74,15 @@ retry, writeTVar, ) -import Control.Lens ((^.)) +import Control.Exception (throwIO) +import Control.Lens ((&), (.~), (^.)) import Data.Generics.Labels () +import Data.IORef (IORef, newIORef, readIORef, writeIORef) import Data.Maybe (fromMaybe, mapMaybe) import Data.Text qualified as T import Data.Time.Clock (getCurrentTime) import Data.Unique (hashUnique, newUnique) -import Effectful (Dispatch (Dynamic), DispatchOf, Eff, Effect, IOE, liftIO, (:>)) +import Effectful (Dispatch (Dynamic), DispatchOf, Eff, Effect, IOE, liftIO, raise, (:>)) import Effectful.Concurrent (Concurrent, threadDelay) import Effectful.Concurrent.STM (atomically) import Effectful.Dispatch.Dynamic (reinterpret_, send) @@ -84,6 +91,7 @@ import Shikumi.Error (ShikumiError (..), fromBaikaiError, isTransient) import Shikumi.LLM.Budget (Budget, admitCall, recordCost) import Shikumi.LLM.Continuation +import Shikumi.LLM.Evidence import Shikumi.LLM.Observation qualified as O -- | The provider-neutral LM effect. 'Complete' is a blocking completion; @@ -122,7 +130,7 @@ (IOE :> es, Error ShikumiError :> es) => Eff (LLM : es) a -> Eff es a -runLLM = reinterpret_ runBaikai (bareHandler O.noObservation) +runLLM = reinterpret_ runBaikai (bareHandler O.noObservation (const (pure ()))) -- | Bare interpreter over an explicit registry. runLLMWith :: @@ -134,8 +142,21 @@ -- | Observe bare transport calls. Callback exceptions propagate without retries. runLLMWithObserver :: (IOE :> es, Error ShikumiError :> es) => ProviderRegistry -> O.LLMObserver -> Eff (LLM : es) a -> Eff es a -runLLMWithObserver reg observer = reinterpret_ (runBaikaiWith reg) (bareHandler observer) +runLLMWithObserver reg observer = runLLMWithEvidence reg observer (const (pure ())) +-- | Bare interpreter with an effectful evidence acknowledgement boundary. +-- The observer does not enable evidence; request it in options or defaults. +-- Callbacks run outside the transport exception handler and cannot cause retries. +runLLMWithEvidence :: + (IOE :> es, Error ShikumiError :> es) => + ProviderRegistry -> O.LLMObserver -> EvidenceObserver es -> Eff (LLM : es) a -> Eff es a +runLLMWithEvidence reg observer observe = + reinterpret_ (runBaikaiWith reg) (bareHandler observer (raise . acknowledge observe)) + +acknowledge :: (IOE :> es, Error ShikumiError :> es) => EvidenceObserver es -> AttemptReceipt -> Eff es () +acknowledge observe receipt = + observe receipt `catchError` \_ e -> liftIO (throwIO (EvidenceObserverFailed e)) + -- | The bare handler, shared by both interpreters. It runs in the handler stack -- (@Baikai : es@), so it can call the @Baikai@ transport effect and throw -- through the @Error ShikumiError@ effect. The blocking path routes baikai's @@ -147,10 +168,10 @@ -- both operations identically. bareHandler :: (IOE :> es, Baikai :> es, Error ShikumiError :> es) => - O.LLMObserver -> LLM (Eff localEs) a -> Eff es a -bareHandler observer op = do + O.LLMObserver -> EvidenceObserver es -> LLM (Eff localEs) a -> Eff es a +bareHandler observer observe op = do cid <- liftIO freshCallId - transportAttempt observer Nothing cid 1 op + transportAttempt observer observe Nothing Nothing cid 1 op freshCallId :: IO T.Text freshCallId = (T.pack . ("call-" <>) . show . hashUnique) <$> newUnique @@ -159,16 +180,17 @@ -- become retryable provider errors, even if the callback throws BaikaiError. transportAttempt :: (IOE :> es, Baikai :> es, Error ShikumiError :> es) => - O.LLMObserver -> Maybe Budget -> T.Text -> Int -> LLM (Eff localEs) a -> Eff es a -transportAttempt observer mb cid ordinal op = case op of - Complete m c o -> do + O.LLMObserver -> EvidenceObserver es -> Maybe Budget -> Maybe (IORef (Maybe T.Text)) -> T.Text -> Int -> LLM (Eff localEs) a -> Eff es a +transportAttempt observer observe mb previous cid ordinal op = case op of + Complete m c original -> do + o <- liftIO (attemptOptions previous ordinal original) either throwError pure (validateRequestContinuation m c o) start <- liftIO getCurrentTime res <- try @BaikaiError (BE.complete m c (stripContinuationMetadata o)) case res of Left be -> do emit m O.CompletionCall start (Just (fromBaikaiError be)) Nothing Nothing - throwError (fromBaikaiError be) + failThrown o be Right resp -> do liftIO (chargeBudget mb resp) emit @@ -178,15 +200,17 @@ (fromBaikaiError <$> responseError resp) (availableUsage (resp ^. #message . #usage)) (O.observedModelOf (resp ^. #evidence)) + deliver o O.CompletionCall (resp ^. #evidence) raiseResponseError resp - Stream m c o -> do + Stream m c original -> do + o <- liftIO (attemptOptions previous ordinal original) either throwError pure (validateRequestContinuation m c o) start <- liftIO getCurrentTime res <- try @BaikaiError (BE.streamCollect m c (stripContinuationMetadata o)) case res of Left be -> do emit m O.StreamCall start (Just (fromBaikaiError be)) Nothing Nothing - throwError (fromBaikaiError be) + failThrown o be Right evs -> do liftIO (chargeBudgetFromEvents mb evs) let terminals = [tp | ev <- evs, tp <- case ev of EventDone t -> [t]; EventError t -> [t]; _ -> []] @@ -194,8 +218,19 @@ usage = terminal >>= \t -> case t ^. #message of AssistantMessage p -> availableUsage (p ^. #usage); _ -> Nothing err = case [streamTerminalError t | EventError t <- evs] of e : _ -> Just e; [] -> Nothing emit m O.StreamCall start err usage (terminal >>= O.observedModelOf . (^. #evidence)) + deliver o O.StreamCall (terminal >>= (^. #evidence)) raiseStreamError evs where + failThrown o be = case o ^. #evidence of + Just req + | E.EvidenceRequired _ <- req ^. #strictness -> + throwError (ValidationFailure ("required attempt evidence unavailable: " <> O.errorClass (fromBaikaiError be))) + _ -> throwError (fromBaikaiError be) + deliver o kind record = case (o ^. #evidence, record) of + (Just _, Just ev) -> do + liftIO (mapM_ (\ref -> writeIORef ref (Just (ev ^. #callId))) previous) + observe (AttemptReceipt cid ordinal kind ev) + _ -> pure () emit m kind start err usage observed = liftIO $ do end <- getCurrentTime observer @@ -280,10 +315,43 @@ LLMConfig -> Eff (LLM : es) a -> Eff es a -runLLMResilient cfg = reinterpret_ (runBaikaiWith (registry cfg)) $ \op -> do +runLLMResilient = runLLMResilientWithEvidence (const (pure ())) + +-- | Resilient calls with acknowledgement of each provider-built attempt record. +-- Budget charging precedes billing observation, which precedes evidence +-- acknowledgement. A failed acknowledgement never redispatches. Cancelled calls +-- have no receipt on the current Baikai transport path. +runLLMResilientWithEvidence :: + (IOE :> es, Concurrent :> es, Error ShikumiError :> es) => + EvidenceObserver es -> LLMConfig -> Eff (LLM : es) a -> Eff es a +runLLMResilientWithEvidence observe cfg = reinterpret_ (runBaikaiWith (registry cfg)) $ \op -> do cid <- liftIO freshCallId + previous <- liftIO $ case operationOptions op ^. #evidence of + Nothing -> pure Nothing + Just _ -> Just <$> newIORef Nothing withBudget (budget cfg) . withRateLimit (rateLimit cfg) $ - retrying (retryPolicy cfg) (\ordinal -> transportAttempt (fromMaybe O.noObservation (observer cfg)) (budget cfg) cid ordinal op) + retrying (retryPolicy cfg) (\ordinal -> transportAttempt (fromMaybe O.noObservation (observer cfg)) (raise . acknowledge observe) (budget cfg) previous cid ordinal op) + +operationOptions :: LLM m a -> Options +operationOptions (Complete _ _ o) = o +operationOptions (Stream _ _ o) = o + +-- The first request is unchanged. Later attempts retain the caller's base and +-- link to the last actual record, never to the process-local billing identity. +attemptOptions :: Maybe (IORef (Maybe T.Text)) -> Int -> Options -> IO Options +attemptOptions previous ordinal opts = case (previous, opts ^. #evidence) of + (Just ref, Just req) | ordinal > 1 -> do + latest <- readIORef ref + pure + ( opts + & #evidence + .~ Just + ( req + & #attempt .~ (req ^. #attempt + fromIntegral (ordinal - 1)) + & #supersedes .~ (latest <|> req ^. #supersedes) + ) + ) + _ -> pure opts -- | Optimistic pre-call budget gate (admission, not reservation). Refuses the -- call with 'BudgetExceeded' when the recorded running total has already reached diff --git a/src/Shikumi/LLM/Defaults.hs b/src/Shikumi/LLM/Defaults.hs --- a/src/Shikumi/LLM/Defaults.hs +++ b/src/Shikumi/LLM/Defaults.hs @@ -6,11 +6,12 @@ emptyRequestDefaults, applyRequestDefaults, withRequestDefaults, + withEvidenceRuns, ) where import Baikai (Options) -import Baikai.Evidence (EvidenceRequest) +import Baikai.Evidence (EvidenceRequest, EvidenceStrictness, evidenceRequest) import Baikai.Model (isJudgmentOnly) import Baikai.Speed (Speed) import Baikai.ThinkingLevel (ThinkingLevel) @@ -18,6 +19,7 @@ import Control.Lens ((&), (.~), (^.)) import Control.Monad (when) import Data.Generics.Labels () +import Data.Text (Text) import Effectful (Eff, (:>)) import Effectful.Dispatch.Dynamic (interpose) import Effectful.Error.Static (Error, throwError) @@ -67,3 +69,17 @@ Stream model ctx opts -> stream model ctx (applyRequestDefaults (forModel model) opts) ) action + +-- | Mint one evidence run ID per logical LLM operation, above transport retries. +-- Explicit per-call requests win. Unlike a fixed defaultEvidence, a multi-Predict +-- program gets distinct run IDs. Compose below routing and above cache/transport. +withEvidenceRuns :: (LLM :> es) => EvidenceStrictness -> Eff es Text -> Eff es a -> Eff es a +withEvidenceRuns strictness mint = interpose $ \_ -> \case + Complete m c o -> fill o >>= complete m c + Stream m c o -> fill o >>= stream m c + where + fill o = case o ^. #evidence of + Just _ -> pure o + Nothing -> do + rid <- mint + pure (o & #evidence .~ Just (evidenceRequest rid & #strictness .~ strictness)) diff --git a/src/Shikumi/LLM/Evidence.hs b/src/Shikumi/LLM/Evidence.hs new file mode 100644 --- /dev/null +++ b/src/Shikumi/LLM/Evidence.hs @@ -0,0 +1,48 @@ +-- | Acknowledged, provider-built evidence for actual transport attempts. +module Shikumi.LLM.Evidence + ( AttemptReceipt (..), + EvidenceObserver, + EvidenceObserverFailed (..), + ) +where + +import Baikai.Evidence (ModelCallEvidence) +import Control.Exception (Exception) +import Data.Text (Text) +import Effectful (Eff) +import GHC.Generics (Generic) +import Shikumi.Error (ShikumiError) +import Shikumi.LLM.Observation (CallKind) + +-- | The original provider record, delivered before retry, error propagation or +-- successful return. No record is fabricated for cancellation. Installing an +-- observer does not request evidence: use request options or scoped defaults. +-- Best-effort requests may produce no record; Baikai enforces required evidence. +-- +-- There are deliberately no JSON instances. Provider error messages inside +-- 'ModelCallEvidence' can include provider response text; applications must choose +-- what is appropriate to persist rather than serializing this envelope blindly. +data AttemptReceipt = AttemptReceipt + { -- | Process-local identity, shared with billing observations across retries. + -- Persist the enclosed evidence's callId as the durable attempt identity. + logicalCallId :: !Text, + -- | One-based attempt ordinal, shared with billing observations. + attemptOrdinal :: !Int, + callKind :: !CallKind, + -- | The exact record built by Baikai, including its own per-attempt callId. + evidence :: !ModelCallEvidence + } + deriving stock (Eq, Show, Generic) + +-- | Normal return acknowledges the record. Runs after budget charging and billing +-- observation. Must be safe for concurrent use: concurrent programs invoke it on +-- worker threads, and worker-local State changes do not reach the parent. +-- Install the observer's effects outside the LLM interpreter. +type EvidenceObserver es = AttemptReceipt -> Eff es () + +-- | Typed observer failures become exceptions so transport retries and Program +-- Retry nodes cannot turn a persistence failure into another provider invocation. +-- Other exceptions, including cancellation, propagate unchanged. +newtype EvidenceObserverFailed = EvidenceObserverFailed ShikumiError + deriving stock (Show) + deriving anyclass (Exception) diff --git a/src/Shikumi/Program.hs b/src/Shikumi/Program.hs --- a/src/Shikumi/Program.hs +++ b/src/Shikumi/Program.hs @@ -72,6 +72,7 @@ sampleTemps, withSampleTemp, effectiveSignature, + preparePredict, parseResponse, -- * Parameter interface (the optimizer/compiler contract) @@ -89,7 +90,7 @@ ) where -import Baikai (Model, emptyModel) +import Baikai (Context, Model, Options, emptyModel) import Data.Aeson (FromJSON (..), ToJSON (..), Value (..), object, withObject, (.!=), (.:), (.:?), (.=)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KM @@ -370,6 +371,17 @@ i -> Eff es o runPredict sig ps i = do + (sig', ctx, opts) <- preparePredict sig ps i + resp <- complete placeholderModel ctx opts + either throwError pure (parseResponseWithSettings (decisionSettings ps) sig' resp) + +-- | Validate local settings and prepare the same request for both execution modes. +-- Calibration values stay local; the returned options carry only rendering metadata. +preparePredict :: + forall i o es. + (FromModel i, FromModel o, ToSchema o, Validatable o, ToPrompt i, ToPrompt o, Error ShikumiError :> es) => + Signature i o -> Params -> i -> Eff es (Signature i o, Context, Options) +preparePredict sig ps i = do let fields = fst (judgmentsInSchema (deriveSchema @o)) either throwError pure $ mapM_ @@ -398,8 +410,7 @@ opts | null (fst (judgmentsInSchema schema)) = renderedOpts | otherwise = attachDecisionState (renderDecisionState sig' i) renderedOpts - resp <- complete placeholderModel ctx opts - either throwError pure (parseResponseWithSettings (decisionSettings ps) sig' resp) + pure (sig', ctx, opts) -- | Decode a response back into the typed output, tolerant of /either/ wire shape: -- a native model (whose @responseFormat@ the router enforced) replies with a JSON diff --git a/src/Shikumi/Routing.hs b/src/Shikumi/Routing.hs --- a/src/Shikumi/Routing.hs +++ b/src/Shikumi/Routing.hs @@ -111,7 +111,6 @@ complete m ctx' opts' Stream _placeholder ctx opts -> do m <- currentModel - if capabilityFor m == DecisionOnly || Map.member metaDecisionEvidenceKey (opts ^. #metadata) then throwError (ValidationFailure "decision outputs do not support streaming") else pure () (ctx', opts') <- either throwError pure (translateForWire m ctx opts) either throwError pure (validateRequestContinuation m ctx' opts') stream m ctx' opts' diff --git a/src/Shikumi/Stream.hs b/src/Shikumi/Stream.hs --- a/src/Shikumi/Stream.hs +++ b/src/Shikumi/Stream.hs @@ -36,6 +36,8 @@ -- * Entry points streamComplete, streamProgram, + responseFromEvents, + atomicEventsFromResponse, ) where @@ -44,38 +46,44 @@ AssistantMessageEvent (..), BlockEndPayload (..), Context, + DataEndPayload (..), DeltaPayload (..), + IndexPayload (..), Message (..), Model, Options, Response, + StartPayload (..), + StopReason (ErrorReason), TerminalPayload (..), + doneTerminal, emptyModel, emptyResponse, emptyTextContent, + responseError, ) -import Control.Lens ((&), (.~)) -import Control.Monad (when) +import Control.Lens ((&), (.~), (^.)) import Data.Generics.Labels () import Data.Maybe (fromMaybe, listToMaybe) import Data.Text (Text) +import Data.Time.Clock (diffUTCTime) import Data.Vector qualified as V import Effectful (Eff, (:>)) import Effectful.Error.Static (Error, throwError) import GHC.Generics (Generic) -import Shikumi.Adapter (Adapter (..), ToPrompt, adapterFor, attachNativeRender, attachSchema, evidenceFields, nativeRenderPieces) +import Shikumi.Adapter (ToPrompt, evidenceFields) import Shikumi.Error (ShikumiError (..)) -import Shikumi.LLM (LLM, stream) +import Shikumi.LLM (LLM, stream, streamTerminalError) import Shikumi.Program ( Params (..), Program (Compose, Embed, FMap, MajorityVote, Map, Parallel, Predict, PredictCaptured, Retry, RetryWhen, Validate), acceptOrReject, - effectiveSignature, - parseResponse, + parseResponseWithSettings, + preparePredict, retryWith, runProgram, ) -import Shikumi.Schema (FromModel, ToSchema, Validatable, deriveSchema) +import Shikumi.Schema (FromModel, ToSchema, Validatable) import Shikumi.Schema.Types qualified as ST import Shikumi.Signature (Signature, outputFields) @@ -130,7 +138,7 @@ -- | Drive one streaming LM call, folding baikai events into 'StreamEvent's -- delivered to the callback in order, and return the fully assembled 'Response' --- (reassembled from the terminal event) so the caller can 'parse' it into a typed +-- (reassembled from the terminal event) so the caller can decode it into a typed -- value exactly as the blocking path would. -- -- Each @TextDelta@ becomes a 'StreamFieldChunk' attributed to @fieldNm@; the final @@ -237,20 +245,12 @@ Embed _ -> blockingNode prog i cb _ -> blockingNode prog i cb --- | Stream a single 'Predict', mirroring 'Shikumi.Program.runPredict' exactly so --- the streamed and blocking paths share one wire + decode definition. It overlays --- t'Params' via the exported 'effectiveSignature', renders model-agnostically, stamps --- the derived schema and the native render alternative onto the metadata channel --- (so the router can attach a native @responseFormat@ and swap the native prompt for --- native-capable models), streams the first output field's chunks bracketed by --- @LmStart@/@LmEnd@, and decodes the reassembled response through the shared --- 'Shikumi.Program.parseResponse' (dual-format: JSON body → native parser keeping --- its located error, marker body → fallback parser). --- --- Chunk-attribution caveat (unchanged): field chunks are honest only for the --- marker/raw-text wire shape; a routed native (JSON) stream still delivers chunks of --- the raw JSON text, though the /final typed value/ now decodes correctly via --- 'parseResponse'. +-- | Prepare exactly the same request and apply the same local settings as the +-- blocking runner. Decision outputs validate the complete terminal before decode +-- and emit only LmStart and LmEnd, including mixed generative decision and text outputs. +-- TypeSafe delivers one atomic structured response. The LLM operation collects +-- events before invoking the program callback; it is not a live transport iterator. +-- Ordinary text outputs retain their field chunks and status timing. streamPredict :: forall i o es. (FromModel i, FromModel o, ToSchema o, Validatable o, ToPrompt i, ToPrompt o) => @@ -261,19 +261,78 @@ (StreamEvent -> Eff es ()) -> Eff es o streamPredict sig ps i cb = do - when (not (null (evidenceFields @o (decisionSettings ps)))) (throwError (ValidationFailure "decision outputs do not support streaming")) - sig' <- effectiveSignature sig ps - let adapter = adapterFor emptyModel - (ctx, opts0) = render adapter sig' i - (nativeSys, nativeDemos) = nativeRenderPieces @i @o sig' - opts = attachNativeRender nativeSys nativeDemos (attachSchema (deriveSchema @o) opts0) - fieldNm = case map ST.fieldName (outputFields sig') of + (sig', ctx, opts) <- preparePredict sig ps i + let fieldNm = case map ST.fieldName (outputFields sig') of (x : _) -> x [] -> "" + checked = not (null (evidenceFields @o (decisionSettings ps))) + decode = either throwError pure . parseResponseWithSettings (decisionSettings ps) sig' cb (StreamStatus (Status LmStart "LM call started")) - resp <- streamComplete fieldNm emptyModel ctx opts cb - cb (StreamStatus (Status LmEnd "LM call finished")) - either throwError pure (parseResponse sig' resp) + if checked + then do + evs <- stream emptyModel ctx opts + resp <- either throwError pure (responseFromEvents emptyModel evs) + result <- decode resp + cb (StreamStatus (Status LmEnd "LM call finished")) + pure result + else do + resp <- streamComplete fieldNm emptyModel ctx opts cb + cb (StreamStatus (Status LmEnd "LM call finished")) + decode resp + +-- | Validate terminal integrity and preserve the complete response envelope. +-- Answer validity is checked separately by the settings-aware program decoder. +responseFromEvents :: Model -> [AssistantMessageEvent] -> Either ShikumiError Response +responseFromEvents model evs = case reverse evs of + EventDone tp : rest | not (any terminal rest) -> case tp ^. #message of + AssistantMessage payload + | tp ^. #errorInfo /= Nothing || payload ^. #stopReason == ErrorReason -> Left (streamTerminalError tp) + | otherwise -> do + let expected = [(i, d) | (i, AssistantData d) <- zip [0 ..] (V.toList (payload ^. #content))] + actual = [(i, d) | DataEnd (DataEndPayload i d) <- evs] + starts = [t | EventStart (StartPayload (AssistantMessage p) _) <- evs, Just t <- [p ^. #timestamp]] + latency = case (starts, payload ^. #timestamp) of + (t : _, Just end) -> max 0 (floor (1000 * diffUTCTime end t)) + _ -> 0 + if actual /= expected + then malformed "structured blocks disagree with terminal" + else + Right + ( emptyResponse + & #model .~ model + & #api .~ (model ^. #api) + & #provider .~ (model ^. #provider) + & #message .~ payload + & #responseId .~ (tp ^. #responseId) + & #evidence .~ (tp ^. #evidence) + & #latencyMs .~ latency + ) + _ -> malformed "terminal is not an assistant message" + EventError tp : rest | not (any terminal rest) -> Left (streamTerminalError tp) + _ -> malformed "expected one final success terminal" + where + terminal EventDone {} = True + terminal EventError {} = True + terminal _ = False + malformed = Left . SchemaMismatch + +-- | Reconstruct an atomic structured stream, without IO or new provider evidence. +-- Text-only and failed recordings cannot supply structured stream replay. +atomicEventsFromResponse :: Response -> Maybe [AssistantMessageEvent] +atomicEventsFromResponse resp + | responseError resp /= Nothing || null structured = Nothing + | otherwise = + Just $ + [EventStart (StartPayload (AssistantMessage skeleton) (resp ^. #responseId))] + ++ concat [block i b | (i, b) <- zip [0 ..] (V.toList (payload ^. #content))] + ++ [EventDone (doneTerminal (resp ^. #evidence) (resp ^. #responseId) (payload ^. #stopReason) (AssistantMessage payload))] + where + payload = resp ^. #message + skeleton = payload & #content .~ V.empty + structured = [d | AssistantData d <- V.toList (payload ^. #content)] + block i (AssistantData d) = [DataStart (IndexPayload i), DataEnd (DataEndPayload i d)] + block i (AssistantText t) = [TextStart (IndexPayload i), TextEnd (BlockEndPayload i (t ^. #text))] + block _ _ = [] -- | Bracket an action with @NodeStart@/@NodeEnd@ status messages. bracketNode :: (StreamEvent -> Eff es ()) -> Eff es a -> Eff es a diff --git a/test/AttemptEvidenceSpec.hs b/test/AttemptEvidenceSpec.hs new file mode 100644 --- /dev/null +++ b/test/AttemptEvidenceSpec.hs @@ -0,0 +1,234 @@ +module AttemptEvidenceSpec (tests) where + +import Baikai qualified as B +import Baikai.Error qualified as BE +import Baikai.Evidence qualified as E +import Control.Concurrent.MVar +import Control.Exception (throwIO, try) +import Control.Lens ((&), (.~), (^.)) +import Control.Monad (void) +import Data.Generics.Labels () +import Data.IORef +import Data.Map.Strict qualified as Map +import Effectful (Eff, IOE, liftIO, runEff, (:>)) +import Effectful.Concurrent (Concurrent, runConcurrent) +import Effectful.Concurrent.Async (cancel, wait, waitCatch, withAsync) +import Effectful.Error.Static (Error, runErrorNoCallStack, throwError) +import Shikumi.Error +import Shikumi.LLM qualified as L +import Shikumi.LLM.Budget (newBudget) +import Shikumi.LLM.Defaults +import Shikumi.LLM.Evidence +import Shikumi.LLM.Observation qualified as O +import Shikumi.Program (Program (Compose, Embed, Parallel), retryWith, runProgram, runProgramConc) +import StubProvider +import System.Timeout (timeout) +import Test.Tasty +import Test.Tasty.HUnit + +transient :: BE.BaikaiError +transient = (BE.providerError "fixture failure") {BE.category = BE.TransientError} + +opts :: B.Options +opts = stubOptions & #evidence .~ Just (E.evidenceRequest "run" & #attempt .~ 4 & #supersedes .~ Just "before") + +invoke :: Bool -> B.Options -> Eff '[L.LLM, Error ShikumiError, Concurrent, IOE] () +invoke streaming o = if streaming then void (L.stream stubModel stubContext o) else void (L.complete stubModel stubContext o) + +config :: B.ProviderRegistry -> L.LLMConfig +config reg = (L.defaultLLMConfig reg) {L.retryPolicy = L.RetryPolicy 3 0 0} + +record :: (IOE :> es) => IORef [AttemptReceipt] -> EvidenceObserver es +record ref receipt = liftIO (atomicModifyIORef' ref (\xs -> (xs ++ [receipt], ()))) + +runCall :: EvidenceObserver '[Error ShikumiError, Concurrent, IOE] -> L.LLMConfig -> Eff '[L.LLM, Error ShikumiError, Concurrent, IOE] a -> IO (Either ShikumiError a) +runCall observe cfg action = runEff . runConcurrent . runErrorNoCallStack @ShikumiError . L.runLLMResilientWithEvidence observe cfg $ action + +bounded :: IO a -> IO a +bounded action = timeout 3000000 action >>= maybe (assertFailure "test timeout") pure + +tests :: TestTree +tests = + testGroup "AttemptEvidence" $ + [ testGroup + (if streaming then "Stream" else "Complete") + [ testCase "retry acknowledges exact records and preserves caller provenance" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + result <- runCall (record receipts) (config reg) (invoke streaming opts) + result @?= Right () + sent <- readIORef calls + rs <- readIORef receipts + map attemptOrdinal rs @?= [1, 2] + case rs of + [a, b] -> do + logicalCallId a @?= logicalCallId b + evidence a ^. #attempt @?= 4 + evidence b ^. #attempt @?= 5 + evidence b ^. #supersedes @?= Just (evidence a ^. #callId) + map ((^. #runId) . evidence) rs @?= ["run", "run"] + evidence a ^. #supersedes @?= Just "before" + (sent !! 1) ^. #evidence @?= Just (E.evidenceRequest "run" & #attempt .~ 5 & #supersedes .~ Just (evidence a ^. #callId)) + map callKind rs @?= replicate 2 (if streaming then O.StreamCall else O.CompletionCall) + _ -> assertFailure (show rs), + testCase "terminal failure is acknowledged before it is raised" $ do + calls <- newIORef [] + receipts <- newIORef [] + let err = BE.invalidRequest "terminal" + reg <- evidenceStubRegistry calls 10 err + result <- runCall (record receipts) (config reg) (invoke streaming opts) + result @?= Left (fromBaikaiError err) + readIORef receipts >>= \rs -> map ((^. #status) . evidence) rs @?= [E.CallFailed], + testCase "acknowledgement blocks return and the next dispatch" $ bounded $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + entered <- newEmptyMVar + release <- newEmptyMVar + result <- runEff + . runConcurrent + . runErrorNoCallStack @ShikumiError + . L.runLLMResilientWithEvidence (\r -> if attemptOrdinal r == 1 then liftIO (putMVar entered () >> takeMVar release) else pure ()) (config reg) + $ withAsync (invoke streaming opts) + $ \worker -> do + liftIO (takeMVar entered) + liftIO (readIORef calls >>= \xs -> length xs @?= 1) + liftIO (putMVar release ()) + wait worker + result @?= Right (), + testCase "successful acknowledgement blocks typed return" $ bounded $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + entered <- newEmptyMVar + release <- newEmptyMVar + returned <- newIORef False + result <- runEff + . runConcurrent + . runErrorNoCallStack @ShikumiError + . L.runLLMResilientWithEvidence (\_ -> liftIO (putMVar entered () >> takeMVar release)) (config reg) + $ withAsync (invoke streaming opts >> liftIO (writeIORef returned True)) + $ \worker -> do + liftIO (takeMVar entered >> readIORef returned >>= (@?= False)) + liftIO (putMVar release ()) + wait worker + result @?= Right (), + testCase "typed observer failure escapes both retry layers" $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + result <- + try @EvidenceObserverFailed $ + runCall + (\_ -> throwError (ProviderFailure "persist failed")) + (config reg) + (retryWith (\_ () -> invoke streaming opts) (const True) 3 (Embed (\() -> pure ())) ()) + assertBool "observer exception" (case result of Left _ -> True; _ -> False) + readIORef calls >>= \xs -> length xs @?= 1, + testCase "observer exception after failed attempt never retries" $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + result <- try @BE.BaikaiError $ runCall (\_ -> liftIO (throwIO transient)) (config reg) (invoke streaming opts) + result @?= Left transient + readIORef calls >>= \xs -> length xs @?= 1, + testCase "no evidence request means identical options and no receipts" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + runCall (record receipts) (config reg) (invoke streaming stubOptions) >>= (@?= Right ()) + readIORef calls >>= (@?= [stubOptions, stubOptions]) + readIORef receipts >>= (@?= []), + testCase "budget refusal creates neither dispatch nor receipt" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + budget <- newBudget (Just 0) + result <- runCall (record receipts) ((config reg) {L.budget = Just budget}) (invoke streaming opts) + assertBool "refused" (case result of Left BudgetExceeded {} -> True; _ -> False) + readIORef calls >>= (@?= []) + readIORef receipts >>= (@?= []), + testCase "bare runner acknowledges the same record" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + result <- runEff . runConcurrent . runErrorNoCallStack @ShikumiError . L.runLLMWithEvidence reg O.noObservation (record receipts) $ invoke streaming opts + result @?= Right () + readIORef receipts >>= \rs -> map attemptOrdinal rs @?= [1], + testCase "strict refusal delivers a real record with zero provider invocations" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + let strict = opts & #evidence .~ Just (E.evidenceRequest "strict" & #strictness .~ E.EvidenceRequired E.EvidenceModelObserved) + result <- runCall (record receipts) (config reg) (invoke streaming strict) + assertBool "refused" (case result of Left _ -> True; _ -> False) + readIORef calls >>= (@?= []) + readIORef receipts >>= \rs -> length rs @?= 1, + testCase "billing settles before evidence acknowledgement" $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + (billing, snapshot) <- O.newBillingCollector + result <- runCall (\_ -> liftIO (snapshot >>= \s -> O.completedAttempts s @?= 1)) ((config reg) {L.observer = Just billing}) (invoke streaming opts) + result @?= Right (), + testCase "cancelled rate-limit waiter never dispatches or emits a receipt" $ bounded $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + limiter <- L.newRateLimiter 0 + result <- runEff + . runConcurrent + . runErrorNoCallStack @ShikumiError + . L.runLLMResilientWithEvidence (record receipts) ((config reg) {L.rateLimit = Just limiter}) + $ withAsync (invoke streaming opts) + $ \worker -> cancel worker >> waitCatch worker + assertBool "cancelled" (case result of Right (Left _) -> True; _ -> False) + readIORef calls >>= (@?= []) + readIORef receipts >>= (@?= []), + testCase "concurrent programs keep retry provenance isolated" $ do + calls <- newIORef [] + receipts <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + let leaf = Embed (\() -> void (L.complete stubModel stubContext opts)) + result <- runCall (record receipts) (config reg) (runProgramConc (Parallel leaf leaf) ()) + result @?= Right ((), ()) + rs <- readIORef receipts + let groups = foldr (\r -> Map.insertWith (++) (logicalCallId r) [r]) Map.empty rs + Map.size groups @?= 2 + mapM_ + ( \group -> case group of + [a, b] -> evidence b ^. #supersedes @?= Just (evidence a ^. #callId) + [_] -> pure () + _ -> assertFailure (show group) + ) + (Map.elems groups), + testCase "Compose mints a distinct run for each operation" $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 0 transient + count <- newIORef (0 :: Int) + let mint = liftIO (atomicModifyIORef' count (\n -> (n + 1, if n == 0 then "first" else "second"))) + leaf = Embed (\() -> void (L.complete stubModel stubContext stubOptions)) + result <- + runCall (const (pure ())) (config reg) . withEvidenceRuns E.EvidenceBestEffort mint $ + runProgram (Compose leaf leaf) () + result @?= Right () + readIORef calls >>= \xs -> map (fmap (^. #runId) . (^. #evidence)) xs @?= [Just "first", Just "second"], + testCase "per-operation run minting survives retries and honors explicit requests" $ do + calls <- newIORef [] + reg <- evidenceStubRegistry calls 1 transient + minted <- newIORef (0 :: Int) + let mint = liftIO (atomicModifyIORef' minted (\n -> (n + 1, if n == 0 then "first" else "second"))) + result <- runCall (const (pure ())) (config reg) . withEvidenceRuns E.EvidenceBestEffort mint $ do + invoke streaming stubOptions + invoke streaming stubOptions + invoke streaming opts + result @?= Right () + readIORef calls >>= \xs -> map (fmap (^. #runId) . (^. #evidence)) xs @?= map Just ["first", "first", "second", "run"] + readIORef minted >>= (@?= 2) + ] + | streaming <- [False, True] + ] + ++ [ testCase "strict thrown transport failure fails closed without retry" $ do + calls <- newIORef 0 + reg <- failingStubRegistry calls 10 "ok" + let strict = opts & #evidence .~ Just (E.evidenceRequest "strict" & #strictness .~ E.EvidenceRequired E.EvidenceRequestedOnly) + result <- runCall (const (pure ())) (config reg) (invoke False strict) + assertBool "terminal validation failure" (case result of Left ValidationFailure {} -> True; _ -> False) + readIORef calls >>= (@?= 1) + ] diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -3,6 +3,7 @@ module Main (main) where import AdapterSpec qualified +import AttemptEvidenceSpec qualified import CliSchemaSpec qualified import CombinatorSpec qualified import ConstraintSpec qualified @@ -38,7 +39,8 @@ defaultMain $ testGroup "shikumi" - [ ResponsesSpec.tests, + [ AttemptEvidenceSpec.tests, + ResponsesSpec.tests, DecisionDeriveSpec.tests, RequestDefaultsSpec.tests, ContinuationSpec.tests, diff --git a/test/StreamSpec.hs b/test/StreamSpec.hs --- a/test/StreamSpec.hs +++ b/test/StreamSpec.hs @@ -12,6 +12,8 @@ ( AssistantContent (..), AssistantMessageEvent (..), BlockEndPayload (..), + DataContent (..), + DataEndPayload (..), DeltaPayload (..), IndexPayload (..), Message (..), @@ -27,6 +29,7 @@ emptyTextContent, ) import Control.Lens ((&), (.~), (^.)) +import Data.Aeson (Value (Null)) import Data.Generics.Labels () import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef) import Data.Text (Text) @@ -50,11 +53,13 @@ Status (..), StatusPhase (..), StreamEvent (..), + atomicEventsFromResponse, + responseFromEvents, streamComplete, streamProgram, ) import Test.Tasty (TestTree, testGroup) -import Test.Tasty.HUnit (testCase, (@?=)) +import Test.Tasty.HUnit (assertBool, testCase, (@?=)) -- A single-text-field signature, the honest field-chunk case. newtype Question = Question {question :: Text} @@ -139,7 +144,18 @@ tests = testGroup "StreamSpec" - [ testCase "M1: streamComplete folds deltas into field chunks and assembles the response" $ do + [ testCase "checked response rejects missing, duplicate, misplaced or inconsistent terminals" $ do + let d = DataContent Null Nothing Nothing Nothing + r = emptyResponse & #message . #content .~ V.singleton (AssistantData d) + terminal = EventDone (doneTerminal Nothing (Just "response-id") Stop (AssistantMessage (r ^. #message))) + block = DataEnd (DataEndPayload 0 d) + malformed = [[], [block], [block, terminal, terminal], [terminal, block], [terminal], [DataEnd (DataEndPayload 1 d), terminal]] + mapM_ (\events -> assertBool "terminal integrity" (case responseFromEvents emptyModel events of Left (SchemaMismatch _) -> True; _ -> False)) malformed + fmap (\r' -> r' ^. #responseId) (responseFromEvents emptyModel [block, terminal]) @?= Right (Just "response-id") + case atomicEventsFromResponse r of + Just events -> fmap (\r' -> r' ^. #message . #content) (responseFromEvents emptyModel events) @?= Right (r ^. #message . #content) + Nothing -> error "structured replay fixture", + testCase "M1: streamComplete folds deltas into field chunks and assembles the response" $ do ref <- newIORef [streamEventsFor ["Hel", "lo"] "Hello"] rec <- newIORef [] resp <- diff --git a/test/StubProvider.hs b/test/StubProvider.hs --- a/test/StubProvider.hs +++ b/test/StubProvider.hs @@ -22,10 +22,12 @@ concurrencyStubRegistry, classifiedStubRegistry, exceptionStubRegistry, + evidenceStubRegistry, ) where import Baikai hiding (flattenAssistantText) +import Baikai.Evidence qualified as E import Baikai.Prelude import Control.Concurrent (threadDelay) import Control.Concurrent.STM @@ -39,6 +41,7 @@ import Data.Generics.Labels () import Data.IORef (IORef, atomicModifyIORef') import Data.Text qualified as T +import Data.Time.Clock (getCurrentTime) import Data.Vector qualified as V import Streamly.Data.Stream qualified as Stream @@ -302,3 +305,35 @@ exceptionStubRegistry :: IO Response -> IO ProviderRegistry exceptionStubRegistry action = mkRegistry "" (\_ _ _ -> action) + +-- | One actual record per scripted attempt, built from the dispatched request. +-- Both operations share the same complete implementation and evidence provenance. +evidenceStubRegistry :: IORef [Options] -> Int -> BaikaiError -> IO ProviderRegistry +evidenceStubRegistry requests failures err = do + let completeFn m _ o = do + n <- atomicModifyIORef' requests (\xs -> (xs ++ [o], length xs)) + now <- getCurrentTime + cid <- E.newCallId + let failed = n < failures + receipt req = + E.baseEvidence + req + cid + (E.EndpointIdentity "stub" "custom" E.TransportHttpApi Nothing "fixture" Nothing) + (m ^. #modelId) + E.noThinkingRequested + now + now + (if failed then E.CallFailed else E.CallSucceeded) + "digest" + "configuration" + response = if failed then errorResponse m now 0 err else stubResponse "ok" + pure (response & #evidence .~ (receipt <$> o ^. #evidence)) + reg <- newProviderRegistry + registerApiProviderWith + reg + ( (apiProviderWith stubApi (liftCompleteToStream completeFn) completeFn) + { describeThinking = stubDescribeThinking + } + ) + pure reg