diff --git a/dataframe-csv.cabal b/dataframe-csv.cabal
--- a/dataframe-csv.cabal
+++ b/dataframe-csv.cabal
@@ -1,12 +1,11 @@
 cabal-version:      2.4
 name:               dataframe-csv
-version:            1.0.1.1
-
+version:            1.0.2.0
 synopsis:           CSV reader and writer for the dataframe ecosystem.
 description:
-    @DataFrame.IO.CSV@ — streaming CSV read/write built on top of
-    @cassava@. Plus @DataFrame.IO.Internal.MutableColumn@, the
-    mutable-column ingest helpers shared by the lazy CSV reader.
+    @DataFrame.IO.CSV@ — strict single-pass CSV read/write (pure
+    Haskell RFC 4180 scanner). Plus @DataFrame.IO.Internal.MutableColumn@,
+    the mutable-column ingest helpers shared by the lazy CSV reader.
 
 bug-reports:        https://github.com/mchav/dataframe/issues
 license:            MIT
@@ -29,15 +28,20 @@
     import:             warnings
     exposed-modules:
                         DataFrame.IO.CSV
+                        DataFrame.IO.CSV.Internal.Infer
+                        DataFrame.IO.CSV.Internal.Options
+                        DataFrame.IO.CSV.Internal.Read
+                        DataFrame.IO.CSV.Internal.Scanner
+                        DataFrame.IO.CSV.Internal.Sink
                         DataFrame.IO.Internal.MutableColumn
     build-depends:      base >= 4 && < 5,
                         bytestring >= 0.11 && < 0.13,
-                        cassava >= 0.1 && < 1,
                         containers >= 0.6.7 && < 0.9,
-                        dataframe-core ^>= 1.0,
-                        dataframe-operations ^>= 1.1,
-                        dataframe-parsing ^>= 1.0,
+                        dataframe-core ^>= 1.1,
+                        dataframe-operations ^>= 1.1.1,
+                        dataframe-parsing ^>= 1.0.2,
                         text >= 2.0 && < 3,
+                        time >= 1.12 && < 2,
                         vector ^>= 0.13
     hs-source-dirs:     src
     default-language:   Haskell2010
diff --git a/src/DataFrame/IO/CSV.hs b/src/DataFrame/IO/CSV.hs
--- a/src/DataFrame/IO/CSV.hs
+++ b/src/DataFrame/IO/CSV.hs
@@ -1,287 +1,49 @@
-{-# LANGUAGE BangPatterns #-}
-{-# LANGUAGE ExplicitNamespaces #-}
-{-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TypeApplications #-}
 
-module DataFrame.IO.CSV where
+{- | CSV reading and writing for dataframes. The reader is a strict,
+single-pass RFC 4180 scanner (cassava-compatible semantics, pinned by
+the golden suite in @tests/IO/CsvGolden.hs@) that parses fields straight
+into typed column builders. Ragged rows are padded with nulls (short
+rows) and extra fields dropped.
+-}
+module DataFrame.IO.CSV (
+    -- * Reading
+    readCsv,
+    readTsv,
+    readCsvWithOpts,
+    readSeparated,
+    readCsvWithSchema,
+    CsvReader,
+    decodeSeparated,
+    fromCsv,
+    fromCsvBytes,
 
+    -- * Options
+    module DataFrame.IO.CSV.Internal.Options,
+
+    -- * Writing
+    writeCsv,
+    writeTsv,
+    writeSeparated,
+
+    -- * Helpers
+    stripQuotes,
+) where
+
 import qualified Data.ByteString as BS
-import qualified Data.ByteString.Char8 as C
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.Map.Strict as M
-import qualified Data.Proxy as P
 import qualified Data.Text as T
 import qualified Data.Text.Encoding as TE
 import qualified Data.Text.IO as TIO
-import qualified Data.Vector as V
-import qualified Data.Vector.Mutable as VM
-import qualified Data.Vector.Unboxed as VU
-import qualified Data.Vector.Unboxed.Mutable as VUM
 
-import Data.Csv.Streaming (Records (..))
-import qualified Data.Csv.Streaming as CsvStream
-
 import Control.Exception (SomeException, catch)
-import Control.Monad
-import Control.Monad.ST (runST)
-import Data.Char
-import qualified Data.Csv as Csv
-import Data.Either
-import Data.Functor
-import Data.IORef
-import Data.Maybe
-import Data.Type.Equality (TestEquality (testEquality))
-import Data.Word (Word8)
-import DataFrame.Internal.Column
-import DataFrame.Internal.DataFrame (
-    DataFrame (..),
-    forceDataFrame,
-    toSeparated,
- )
-import DataFrame.Internal.Parsing
-import DataFrame.Internal.Schema
-import DataFrame.Operations.Typing
-import System.IO
-import Type.Reflection
-import Prelude hiding (concat, takeWhile)
-
-chunkSize :: Int
-chunkSize = 16_384
-
-data PagedVector a = PagedVector
-    { pvChunks :: !(IORef [V.Vector a])
-    -- ^ Finished chunks (reverse order)
-    , pvActive :: !(IORef (VM.IOVector a))
-    -- ^ Current mutable chunk
-    , pvCount :: !(IORef Int)
-    -- ^ Items written in current chunk
-    }
-
-data PagedUnboxedVector a = PagedUnboxedVector
-    { puvChunks :: !(IORef [VU.Vector a])
-    , puvActive :: !(IORef (VUM.IOVector a))
-    , puvCount :: !(IORef Int)
-    }
-
-data BuilderColumn
-    = BuilderInt !(PagedUnboxedVector Int) !(PagedUnboxedVector Word8)
-    | BuilderDouble !(PagedUnboxedVector Double) !(PagedUnboxedVector Word8)
-    | BuilderText !(PagedVector T.Text) !(PagedUnboxedVector Word8)
-    | BuilderBS !(PagedVector BS.ByteString) !(PagedUnboxedVector Word8)
-
-newPagedVector :: IO (PagedVector a)
-newPagedVector = do
-    active <- VM.unsafeNew chunkSize
-    PagedVector <$> newIORef [] <*> newIORef active <*> newIORef 0
-
-newPagedUnboxedVector :: (VUM.Unbox a) => IO (PagedUnboxedVector a)
-newPagedUnboxedVector = do
-    active <- VUM.unsafeNew chunkSize
-    PagedUnboxedVector <$> newIORef [] <*> newIORef active <*> newIORef 0
-
-appendPagedVector :: PagedVector a -> a -> IO ()
-appendPagedVector (PagedVector chunksRef activeRef countRef) !val = do
-    count <- readIORef countRef
-    active <- readIORef activeRef
-
-    if count < chunkSize
-        then do
-            VM.unsafeWrite active count val
-            writeIORef countRef $! count + 1
-        else do
-            frozen <- V.unsafeFreeze active
-            modifyIORef' chunksRef (frozen :)
-
-            newActive <- VM.unsafeNew chunkSize
-            VM.unsafeWrite newActive 0 val
-
-            writeIORef activeRef newActive
-            writeIORef countRef 1
-{-# INLINE appendPagedVector #-}
-
-appendPagedUnboxedVector :: (VUM.Unbox a) => PagedUnboxedVector a -> a -> IO ()
-appendPagedUnboxedVector (PagedUnboxedVector chunksRef activeRef countRef) !val = do
-    count <- readIORef countRef
-    active <- readIORef activeRef
-
-    if count < chunkSize
-        then do
-            VUM.unsafeWrite active count val
-            writeIORef countRef $! count + 1
-        else do
-            frozen <- VU.unsafeFreeze active
-            modifyIORef' chunksRef (frozen :)
-
-            newActive <- VUM.unsafeNew chunkSize
-            VUM.unsafeWrite newActive 0 val
-
-            writeIORef activeRef newActive
-            writeIORef countRef 1
-{-# INLINE appendPagedUnboxedVector #-}
-
-freezePagedVector :: PagedVector a -> IO (V.Vector a)
-freezePagedVector (PagedVector chunksRef activeRef countRef) = do
-    count <- readIORef countRef
-    active <- readIORef activeRef
-    chunks <- readIORef chunksRef
-
-    writeIORef chunksRef [] -- release chunk references
-    let frozenChunks = reverse chunks
-        totalLen = count + sum (map V.length frozenChunks)
-
-    mv <- VM.unsafeNew totalLen
-
-    let copyChunk !offset chunk = do
-            V.copy (VM.slice offset (V.length chunk) mv) chunk
-            pure (offset + V.length chunk)
-
-    offset <- foldM copyChunk 0 frozenChunks
-    VM.copy (VM.slice offset count mv) (VM.slice 0 count active)
-
-    V.unsafeFreeze mv
-
-freezePagedUnboxedVector ::
-    (VUM.Unbox a) => PagedUnboxedVector a -> IO (VU.Vector a)
-freezePagedUnboxedVector (PagedUnboxedVector chunksRef activeRef countRef) = do
-    count <- readIORef countRef
-    active <- readIORef activeRef
-    chunks <- readIORef chunksRef
-
-    writeIORef chunksRef [] -- release chunk references
-    let frozenChunks = reverse chunks
-        totalLen = count + sum (map VU.length frozenChunks)
-
-    mv <- VUM.unsafeNew totalLen
-
-    let copyChunk !offset chunk = do
-            VU.copy (VUM.slice offset (VU.length chunk) mv) chunk
-            pure (offset + VU.length chunk)
-
-    offset <- foldM copyChunk 0 frozenChunks
-    VUM.copy (VUM.slice offset count mv) (VUM.slice 0 count active)
-
-    VU.unsafeFreeze mv
-
--- | STANDARD CONFIG TYPES
-data HeaderSpec = NoHeader | UseFirstRow | ProvideNames [T.Text]
-    deriving (Eq, Show)
-
-data TypeSpec
-    = InferFromSample Int
-    | SpecifyTypes [(T.Text, SchemaType)] TypeSpec
-    | NoInference
-
-{- | How the fast reader should treat a row whose field count does not
-match the header row.  Only consulted by @dataframe-fastcsv@; the pure
-Haskell reader has its own semantics.
--}
-data RaggedRowPolicy
-    = {- | Fill missing cells with nulls; silently drop extras.
-      Matches pandas / polars lenient defaults.
-      -}
-      PadWithNull
-    | {- | Fill missing cells with nulls; silently drop extras.  Alias
-      kept for ergonomic naming when the caller only cares about the
-      "don't raise, just forget the extras" half of 'PadWithNull'.
-      -}
-      Truncate
-    | {- | Raise a 'DataFrame.IO.CSV.Fast.CsvParseError' on any row whose
-      field count differs from the header.  Matches polars' strict
-      schema-bound mode.
-      -}
-      RaiseOnRagged
-    deriving (Eq, Show)
-
--- | How the fast reader should treat an unclosed quoted field at EOF.
-data UnclosedQuotePolicy
-    = -- | Raise a 'DataFrame.IO.CSV.Fast.CsvParseError'.  Default.
-      RaiseOnUnclosedQuote
-    | {- | Return whatever rows were parsed before the stray quote; the
-      remainder is silently dropped.
-      -}
-      BestEffort
-    deriving (Eq, Show)
-
--- | CSV read parameters.
-data ReadOptions = ReadOptions
-    { headerSpec :: HeaderSpec
-    -- ^ Where to get the headers from. (default: UseFirstRow)
-    , typeSpec :: TypeSpec
-    -- ^ Whether/how to infer types. (default: InferFromSample 100)
-    , safeRead :: SafeReadMode
-    {- ^ Default 'SafeReadMode' for columns without an entry in
-    'safeReadOverrides'. (default: 'NoSafeRead')
-    -}
-    , safeReadOverrides :: [(T.Text, SafeReadMode)]
-    -- ^ Per-column 'SafeReadMode' overrides; takes precedence over 'safeRead'.
-    , dateFormat :: String
-    {- ^ Format of date fields as recognized by the Data.Time.Format module.
-
-    __Examples:__
-
-    @
-    > parseTimeM True defaultTimeLocale "%Y/%-m/%-d" "2010/3/04" :: Maybe Day
-    Just 2010-03-04
-    > parseTimeM True defaultTimeLocale "%d/%-m/%-Y" "04/3/2010" :: Maybe Day
-    Just 2010-03-04
-    @
-    -}
-    , columnSeparator :: Char
-    -- ^ Character that separates column values.
-    , numColumns :: Maybe Int
-    -- ^ Number of columns to read.
-    , missingIndicators :: [T.Text]
-    -- ^ Values that should be read as `Nothing`.
-    , fastCsvOnRaggedRow :: RaggedRowPolicy
-    {- ^ @dataframe-fastcsv@: how to treat rows with a non-header field count.
-    (default: 'PadWithNull')
-    -}
-    , fastCsvOnUnclosedQuote :: UnclosedQuotePolicy
-    {- ^ @dataframe-fastcsv@: how to treat an unclosed quoted field at EOF.
-    (default: 'RaiseOnUnclosedQuote')
-    -}
-    , fastCsvTrimUnquoted :: Bool
-    {- ^ @dataframe-fastcsv@: if 'True', leading/trailing whitespace is
-    stripped from unquoted fields after decoding.  RFC 4180 preserves
-    this whitespace, and that is the default ('False').
-    -}
-    }
-
-shouldInferFromSample :: TypeSpec -> Bool
-shouldInferFromSample (InferFromSample _) = True
-shouldInferFromSample (SpecifyTypes _ fallback) = shouldInferFromSample fallback
-shouldInferFromSample _ = False
-
-schemaTypeMap :: TypeSpec -> M.Map T.Text SchemaType
-schemaTypeMap (SpecifyTypes xs _) = M.fromList xs
-schemaTypeMap _ = M.empty
-
-typeInferenceSampleSize :: TypeSpec -> Int
-typeInferenceSampleSize (InferFromSample n) = n
-typeInferenceSampleSize (SpecifyTypes _ fallback) = typeInferenceSampleSize fallback
-typeInferenceSampleSize _ = 0
-
-defaultReadOptions :: ReadOptions
-defaultReadOptions =
-    ReadOptions
-        { headerSpec = UseFirstRow
-        , typeSpec = InferFromSample 100
-        , safeRead = NoSafeRead
-        , safeReadOverrides = []
-        , dateFormat = "%Y-%m-%d"
-        , columnSeparator = ','
-        , numColumns = Nothing
-        , missingIndicators =
-            ["Nothing", "NULL", "", " ", "nan", "null", "N/A", "NaN", "NAN", "NA"]
-        , fastCsvOnRaggedRow = PadWithNull
-        , fastCsvOnUnclosedQuote = RaiseOnUnclosedQuote
-        , fastCsvTrimUnquoted = False
-        }
+import Data.Maybe (fromMaybe)
+import DataFrame.IO.CSV.Internal.Options
+import DataFrame.IO.CSV.Internal.Read (decodeCsvStrict)
+import DataFrame.Internal.DataFrame (DataFrame (..), toSeparated)
+import DataFrame.Internal.Schema (Schema, elements)
 
 {- | Read CSV file from path and load it into a dataframe.
 
@@ -296,10 +58,10 @@
 
 type CsvReader = Schema -> FilePath -> IO DataFrame
 
-{- | Schema-driven attoparsec CSV reader.  Coerces each column to the
-type declared in 'Schema'; columns absent from the schema fall back to
-the default inference path.  Defined in terms of 'readSeparated' with
-the 'TypeSpec' filled in.
+{- | Schema-driven CSV reader.  Coerces each column to the type declared
+in 'Schema'; columns absent from the schema fall back to the default
+inference path.  Defined in terms of 'readSeparated' with the 'TypeSpec'
+filled in.
 
 @
 import qualified DataFrame as D
@@ -347,419 +109,16 @@
 @
 -}
 readSeparated :: ReadOptions -> FilePath -> IO DataFrame
-readSeparated opts !path = do
-    let stripUtf8Bom bs = fromMaybe bs (BL.stripPrefix "\xEF\xBB\xBF" bs)
-    csvData <- stripUtf8Bom <$> BL.readFile path
-    fmap forceDataFrame (decodeSeparated opts csvData)
-
-decodeSeparated :: ReadOptions -> BL.ByteString -> IO DataFrame
-decodeSeparated !opts csvData = do
-    let sep = columnSeparator opts
-    let decodeOpts = Csv.defaultDecodeOptions{Csv.decDelimiter = fromIntegral (ord sep)}
-    let stream = CsvStream.decodeWith decodeOpts Csv.NoHeader csvData
-
-    let peekStream (Cons (Right row) rest) = return (row, rest)
-        peekStream (Cons (Left err) _) = error $ "Error parsing CSV header: " ++ err
-        peekStream (Nil Nothing _) = error "Empty CSV file"
-        peekStream (Nil (Just err) _) = error err
-
-    (firstRowRaw, dataStream) <- peekStream stream
-
-    let (columnNames, rowsToProcess) = case headerSpec opts of
-            NoHeader ->
-                ( map (T.pack . show) [0 .. V.length firstRowRaw - 1]
-                , Cons (Right firstRowRaw) dataStream
-                )
-            UseFirstRow ->
-                ( map (T.strip . TE.decodeUtf8Lenient . BL.toStrict) (V.toList firstRowRaw)
-                , dataStream
-                )
-            ProvideNames ns ->
-                ( ns ++ drop (length ns) (map (T.pack . show) [0 .. V.length firstRowRaw - 1])
-                , Cons (Right firstRowRaw) dataStream
-                )
-
-    (sampleRow, _) <- peekStream rowsToProcess
-    builderCols <- initializeColumns columnNames (V.toList sampleRow) opts
-    let !builderColsV = V.fromList builderCols
-    let colNamesV = V.fromList columnNames
-        resolveMode =
-            effectiveSafeRead
-                (safeRead opts)
-                (safeReadOverrides opts)
-        -- If ANY column is EitherRead we keep every raw cell (including
-        -- "N/A" etc.) verbatim; otherwise the missing-indicator list applies.
-        anyEither =
-            any (\n -> resolveMode n == EitherRead) columnNames
-        missing = if anyEither then [] else missingIndicators opts
-    processStream missing rowsToProcess builderColsV (numColumns opts)
-
-    frozenCols <-
-        V.zipWithM
-            (\name bc -> finalizeBuilderColumn (resolveMode name) opts bc)
-            colNamesV
-            builderColsV
-    let numRows = maybe 0 columnLength (frozenCols V.!? 0)
-
-    let df =
-            DataFrame
-                frozenCols
-                (M.fromList (zip columnNames [0 ..]))
-                (numRows, V.length frozenCols)
-                M.empty -- TODO give typed column references
-    pure $ parseWithTypes resolveMode (schemaTypeMap (typeSpec opts)) df
-
-initializeColumns ::
-    [T.Text] -> [BL.ByteString] -> ReadOptions -> IO [BuilderColumn]
-initializeColumns names _row opts = zipWithM initColumn names (map lookupType names)
-  where
-    typeMap = schemaTypeMap (typeSpec opts)
-    -- Return Nothing for columns that should be inferred from BS
-    shouldInfer = case typeSpec opts of
-        InferFromSample _ -> True
-        SpecifyTypes _ fallback -> shouldInferFromSample fallback
-        NoInference -> False
-    lookupType name = M.lookup name typeMap
-    resolveMode =
-        effectiveSafeRead (safeRead opts) (safeReadOverrides opts)
-    initColumn :: T.Text -> Maybe SchemaType -> IO BuilderColumn
-    initColumn name _ | resolveMode name == EitherRead = do
-        validityRef <- newPagedUnboxedVector
-        BuilderBS <$> newPagedVector <*> pure validityRef
-    initColumn _ Nothing | shouldInfer = do
-        validityRef <- newPagedUnboxedVector
-        BuilderBS <$> newPagedVector <*> pure validityRef
-    initColumn _ mtype = do
-        validityRef <- newPagedUnboxedVector
-        let t = fromMaybe (schemaType @T.Text) mtype
-        case t of
-            SType (_ :: P.Proxy a) -> case testEquality (typeRep @a) (typeRep @Int) of
-                Just Refl -> BuilderInt <$> newPagedUnboxedVector <*> pure validityRef
-                Nothing -> case testEquality (typeRep @a) (typeRep @Double) of
-                    Just Refl -> BuilderDouble <$> newPagedUnboxedVector <*> pure validityRef
-                    Nothing -> BuilderText <$> newPagedVector <*> pure validityRef
-
-processStream ::
-    [T.Text] ->
-    CsvStream.Records (V.Vector BL.ByteString) ->
-    V.Vector BuilderColumn ->
-    Maybe Int ->
-    IO ()
-processStream _ _ _ (Just 0) = return ()
-processStream missing (Cons (Right row) rest) cols n =
-    processRow missing row cols
-        >> processStream missing rest cols (fmap (flip (-) 1) n)
-processStream _missing (Cons (Left err) _) _ _ = error ("CSV Parse Error: " ++ err)
-processStream _missing (Nil _ _) _ _ = return ()
-
-processRow ::
-    [T.Text] -> V.Vector BL.ByteString -> V.Vector BuilderColumn -> IO ()
-processRow missing !vals !cols = V.zipWithM_ processValue vals cols
-  where
-    processValue !bs !col = do
-        let !bs' = BL.toStrict bs
-        case col of
-            BuilderInt gv valid -> case readByteStringInt bs' of
-                Just !i -> appendPagedUnboxedVector gv i >> appendPagedUnboxedVector valid 1
-                Nothing -> appendPagedUnboxedVector gv 0 >> appendPagedUnboxedVector valid 0
-            BuilderDouble gv valid -> case readByteStringDouble bs' of
-                Just !d -> appendPagedUnboxedVector gv d >> appendPagedUnboxedVector valid 1
-                Nothing -> appendPagedUnboxedVector gv 0.0 >> appendPagedUnboxedVector valid 0
-            BuilderText gv valid -> do
-                let !val = T.strip (TE.decodeUtf8Lenient bs')
-                appendPagedVector gv val
-                appendPagedUnboxedVector valid (if val `elem` missing then 0 else 1)
-            BuilderBS gv valid -> do
-                let !bs'' = C.strip bs'
-                appendPagedVector gv bs''
-                appendPagedUnboxedVector
-                    valid
-                    (if TE.decodeUtf8Lenient bs'' `elem` missing then 0 else 1)
-
-freezeBuilderColumn :: BuilderColumn -> IO Column
-freezeBuilderColumn (BuilderInt gv validRef) = do
-    vec <- freezePagedUnboxedVector gv
-    valid <- freezePagedUnboxedVector validRef
-    if VU.all (== 1) valid
-        then return $! UnboxedColumn Nothing vec
-        else constructOptional vec valid
-freezeBuilderColumn (BuilderDouble gv validRef) = do
-    vec <- freezePagedUnboxedVector gv
-    valid <- freezePagedUnboxedVector validRef
-    if VU.all (== 1) valid
-        then return $! UnboxedColumn Nothing vec
-        else constructOptional vec valid
-freezeBuilderColumn (BuilderText gv validRef) = do
-    vec <- freezePagedVector gv
-    valid <- freezePagedUnboxedVector validRef
-    if VU.all (== 1) valid
-        then return $! BoxedColumn Nothing vec
-        else constructOptionalBoxed vec valid
-freezeBuilderColumn (BuilderBS _ _) =
-    error
-        "freezeBuilderColumn: BuilderBS must be finalized via finalizeBuilderColumn"
-
-finalizeBuilderColumn ::
-    SafeReadMode -> ReadOptions -> BuilderColumn -> IO Column
-finalizeBuilderColumn mode opts bc = do
-    col <- case bc of
-        BuilderBS gv validRef -> do
-            vec <- freezePagedVector gv
-            valid <- freezePagedUnboxedVector validRef
-            return $! inferColumnFromBS mode opts vec valid
-        _ -> freezeBuilderColumn bc
-    return $! case mode of
-        NoSafeRead -> col
-        MaybeRead -> ensureOptional col
-        EitherRead -> col
-
-inferColumnFromBS ::
-    SafeReadMode ->
-    ReadOptions ->
-    V.Vector BS.ByteString ->
-    VU.Vector Word8 ->
-    Column
-inferColumnFromBS mode opts vec valid =
-    let sampleN = let n = typeInferenceSampleSize (typeSpec opts) in if n == 0 then 100 else n
-        dfmt = dateFormat opts
-        -- The sample IS still Maybe-wrapped; it's bounded at 100 rows
-        -- by default, so the allocation is ignorable.  The previous
-        -- full-column `asMaybeFull = V.generate ...` allocation is
-        -- gone — handlers walk (vec, valid) directly.
-        samples = V.generate (min sampleN (V.length vec)) $ \i ->
-            if valid VU.! i == 1 then Just (vec V.! i) else Nothing
-        assumption = makeParsingAssumptionBS dfmt samples
-     in case mode of
-            EitherRead -> handleBSEither dfmt assumption vec valid
-            _ -> case assumption of
-                IntAssumption -> handleBSInt dfmt vec valid
-                DoubleAssumption -> handleBSDouble vec valid
-                BoolAssumption -> handleBSBool vec valid
-                DateAssumption -> handleBSDate dfmt vec valid
-                TextAssumption -> handleBSText vec valid
-                NoAssumption -> handleBSNo dfmt vec valid
-
-{- | 'EitherRead' wrap for the ByteString inference path: produce an
-@Either Text a@ column. Raw input bytes are preserved verbatim; rows that were
-marked invalid by the builder (e.g. empty cells before EitherRead disabled
-missing-indicator detection) become @Left \"\"@.
--}
-handleBSEither ::
-    String ->
-    ParsingAssumption ->
-    V.Vector BS.ByteString ->
-    VU.Vector Word8 ->
-    Column
-handleBSEither dfmt assumption vec valid = case assumption of
-    BoolAssumption -> wrap readByteStringBool
-    IntAssumption -> wrap readByteStringInt
-    DoubleAssumption -> wrap readByteStringDouble
-    DateAssumption -> wrap (readByteStringDate dfmt)
-    -- Text / No assumption: column is Either Text Text; empty cells become
-    -- Left "" to keep the "Left means missing/failure" convention.
-    TextAssumption -> fromVector (V.imap textEither vec)
-    NoAssumption -> fromVector (V.imap textEither vec)
-  where
-    wrap ::
-        forall a. (Columnable a) => (BS.ByteString -> Maybe a) -> Column
-    wrap p = fromVector (V.imap (toEither p) vec)
-
-    toEither ::
-        forall a.
-        (BS.ByteString -> Maybe a) ->
-        Int ->
-        BS.ByteString ->
-        Either T.Text a
-    toEither p i bs
-        | valid VU.! i == 0 = Left (TE.decodeUtf8Lenient bs)
-        | otherwise = case p bs of
-            Just v -> Right v
-            Nothing -> Left (TE.decodeUtf8Lenient bs)
-
-    textEither :: Int -> BS.ByteString -> Either T.Text T.Text
-    textEither i bs =
-        let t = TE.decodeUtf8Lenient bs
-         in if valid VU.! i == 0 || T.null t then Left t else Right t
-
-makeParsingAssumptionBS ::
-    String -> V.Vector (Maybe BS.ByteString) -> ParsingAssumption
-makeParsingAssumptionBS dfmt asMaybe
-    | V.all (== Nothing) asMaybe = NoAssumption
-    | vecSameConstructor asMaybe asMaybeBool = BoolAssumption
-    | vecSameConstructor asMaybe asMaybeInt
-        && vecSameConstructor asMaybe asMaybeDouble =
-        IntAssumption
-    | vecSameConstructor asMaybe asMaybeDouble = DoubleAssumption
-    | vecSameConstructor asMaybe asMaybeDate = DateAssumption
-    | otherwise = TextAssumption
-  where
-    asMaybeBool = V.map (>>= readByteStringBool) asMaybe
-    asMaybeInt = V.map (>>= readByteStringInt) asMaybe
-    asMaybeDouble = V.map (>>= readByteStringDouble) asMaybe
-    asMaybeDate = V.map (>>= readByteStringDate dfmt) asMaybe
-
--- All @handleBS*@ helpers now take the raw @V.Vector BS.ByteString@
--- plus the builder's @VU.Vector Word8@ validity vector, fusing the
--- parse + validity check into a single pass via
--- 'parseUnboxedColumnWithValid'.  The previous @V.Vector (Maybe
--- BS.ByteString)@ intermediate — allocated upstream in
--- 'inferColumnFromBS' — is gone, along with the paired Int/Double
--- parses and the two 'V.zipWith' Bool vectors from
--- 'vecSameConstructor'.
-
-handleBSBool ::
-    V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSBool vec valid =
-    case parseUnboxedColumnWithValid False readByteStringBool vec valid of
-        Just (mbm, out) -> UnboxedColumn mbm out
-        Nothing -> handleBSText vec valid
-
-handleBSInt ::
-    String -> V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSInt _dfmt vec valid =
-    case parseUnboxedColumnWithValid 0 readByteStringInt vec valid of
-        Just (mbm, out) -> UnboxedColumn mbm out
-        Nothing -> case parseUnboxedColumnWithValid 0 readByteStringDouble vec valid of
-            Just (mbm, out) -> UnboxedColumn mbm out
-            Nothing -> handleBSText vec valid
-
-handleBSDouble ::
-    V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSDouble vec valid =
-    case parseUnboxedColumnWithValid 0 readByteStringDouble vec valid of
-        Just (mbm, out) -> UnboxedColumn mbm out
-        Nothing -> handleBSText vec valid
-
--- Dates are boxed ('Day' isn't 'VU.Unbox'), so fuse into a V.Vector
--- (Maybe Day) in one pass.  Bails on the first non-null cell that
--- fails to parse.
-handleBSDate ::
-    String -> V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSDate dfmt vec valid =
-    case parseBoxedMaybeBSColumn valid (readByteStringDate dfmt) vec of
-        Just (anyNull, out)
-            | anyNull -> fromVector out
-            | otherwise -> fromVector (V.mapMaybe id out)
-        Nothing -> handleBSText vec valid
-
--- Fused Text handler: decode each cell's UTF-8 bytes once, mark nulls
--- directly from the validity vector.  Replaces the two-pass
--- `V.map (fmap decodeUtf8Lenient) ... sequenceA ...` pattern.
-handleBSText ::
-    V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSText vec valid
-    | VU.any (== 0) valid =
-        fromVector
-            ( V.imap
-                ( \i bs ->
-                    if valid VU.! i == 0
-                        then Nothing
-                        else Just (TE.decodeUtf8Lenient bs)
-                )
-                vec
-            )
-    | otherwise = fromVector (V.map TE.decodeUtf8Lenient vec)
-
-handleBSNo ::
-    String -> V.Vector BS.ByteString -> VU.Vector Word8 -> Column
-handleBSNo dfmt vec valid
-    | VU.all (== 0) valid =
-        fromVector (V.map (const (Nothing :: Maybe T.Text)) vec)
-    | Just (mbm, out) <-
-        parseUnboxedColumnWithValid False readByteStringBool vec valid =
-        UnboxedColumn mbm out
-    | Just (mbm, out) <- parseUnboxedColumnWithValid 0 readByteStringInt vec valid =
-        UnboxedColumn mbm out
-    | Just (mbm, out) <- parseUnboxedColumnWithValid 0 readByteStringDouble vec valid =
-        UnboxedColumn mbm out
-    | otherwise = case parseBoxedMaybeBSColumn valid (readByteStringDate dfmt) vec of
-        Just (anyNull, out)
-            | anyNull -> fromVector out
-            | otherwise -> fromVector (V.mapMaybe id out)
-        Nothing -> handleBSText vec valid
-
--- Boxed counterpart to 'parseUnboxedColumnWithValid' for types that
--- aren't 'VU.Unbox' (e.g. 'Day').  Same one-pass + early-bail shape.
-parseBoxedMaybeBSColumn ::
-    VU.Vector Word8 ->
-    (BS.ByteString -> Maybe a) ->
-    V.Vector BS.ByteString ->
-    Maybe (Bool, V.Vector (Maybe a))
-parseBoxedMaybeBSColumn valid parser vec = runST $ do
-    let n = V.length vec
-    out <- VM.new n
-    let loop !i !anyNull
-            | i >= n = do
-                frozen <- V.unsafeFreeze out
-                return (Just (anyNull, frozen))
-            | VU.unsafeIndex valid i == 0 = do
-                VM.unsafeWrite out i Nothing
-                loop (i + 1) True
-            | otherwise = case parser (V.unsafeIndex vec i) of
-                Just v -> do
-                    VM.unsafeWrite out i (Just v)
-                    loop (i + 1) anyNull
-                Nothing -> return Nothing
-    loop 0 False
-
-{- | One-pass fused parse into a typed unboxed column.  Avoids the
-@V.Vector (Maybe a)@ intermediate that the "parse then 'sequenceA'"
-idiom requires, and avoids the upstream @V.Vector (Maybe src)@
-classification by reading nullability from a precomputed validity
-vector (produced by the CSV builder alongside the raw cells).
-
-Returns @Just (mbm, vec)@ only when every non-null cell parses.  The
-first unparseable cell short-circuits to @Nothing@ so the caller can
-fall back to the next assumption (Int → Double → Text).  @mbm@ is
-@Nothing@ when no nulls exist (the column is non-nullable) and
-@Just bm@ otherwise.
-
-Null slots are filled with @nullValue@; downstream consumers only see
-them through the bitmap, so the sentinel never escapes.
-
-Memory shape for a length-@n@ input:
+readSeparated opts path = do
+    let stripUtf8Bom b = fromMaybe b (BS.stripPrefix "\xEF\xBB\xBF" b)
+    csvData <- stripUtf8Bom <$> BS.readFile path
+    decodeCsvStrict opts csvData
 
-  * 1 × 'VUM.STVector' of @a@        (final data, @sizeOf a × n@ bytes)
-  * 1 × 'VUM.STVector' of 'Word8'    (per-element validity, @n@ bytes)
-  * 1 × 'Bitmap' (bit-packed, @⌈n\/8⌉@ bytes) — only when nulls exist.
+{- | Decode in-memory CSV bytes into a dataframe. The result is fully
+forced. (Note: unlike 'readSeparated', no UTF-8 BOM is stripped.)
 -}
-parseUnboxedColumnWithValid ::
-    forall src a.
-    (VU.Unbox a) =>
-    a ->
-    (src -> Maybe a) ->
-    V.Vector src ->
-    VU.Vector Word8 ->
-    Maybe (Maybe Bitmap, VU.Vector a)
-parseUnboxedColumnWithValid nullValue parser vec valid = runST $ do
-    let n = V.length vec
-    values <- VUM.unsafeNew n
-    vmask <- VUM.unsafeNew n
-    let go !i !anyNull
-            | i >= n = finalizeParseResult values vmask anyNull
-            | VU.unsafeIndex valid i == 0 = do
-                VUM.unsafeWrite vmask i 0
-                VUM.unsafeWrite values i nullValue
-                go (i + 1) True
-            | otherwise = case parser (V.unsafeIndex vec i) of
-                Just v -> do
-                    VUM.unsafeWrite vmask i 1
-                    VUM.unsafeWrite values i v
-                    go (i + 1) anyNull
-                Nothing -> return Nothing
-    go 0 False
-{-# INLINE parseUnboxedColumnWithValid #-}
-
-constructOptional ::
-    (VU.Unbox a, Columnable a) => VU.Vector a -> VU.Vector Word8 -> IO Column
-constructOptional vec valid = do
-    let bm = buildBitmapFromValid valid
-    pure $ UnboxedColumn (Just bm) vec
-
-constructOptionalBoxed :: V.Vector T.Text -> VU.Vector Word8 -> IO Column
-constructOptionalBoxed vec valid = do
-    let bm = buildBitmapFromValid valid
-    pure $ BoxedColumn (Just bm) vec
+decodeSeparated :: ReadOptions -> BL.ByteString -> IO DataFrame
+decodeSeparated opts csvData = decodeCsvStrict opts (BL.toStrict csvData)
 
 writeCsv :: FilePath -> DataFrame -> IO ()
 writeCsv = writeSeparated ','
diff --git a/src/DataFrame/IO/CSV/Internal/Infer.hs b/src/DataFrame/IO/CSV/Internal/Infer.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Internal/Infer.hs
@@ -0,0 +1,239 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+{- | Column inference over raw field bytes for the default CSV reader:
+classify a bounded sample with the shared lattice
+("DataFrame.Operations.Inference"), then build the column in one fused
+pass with in-place Int -> Double promotion (audit T2). Cells arrive as
+'Cells' — unboxed offsets into the input buffer plus a builder validity
+vector — so the passes parse slices without materializing cell values.
+-}
+module DataFrame.IO.CSV.Internal.Infer (
+    inferColumnFromBS,
+) where
+
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Unsafe as BSU
+import qualified Data.IntMap.Strict as IM
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import qualified Data.Vector as V
+import qualified Data.Vector.Mutable as VM
+import qualified Data.Vector.Unboxed as VU
+import qualified Data.Vector.Unboxed.Mutable as VUM
+
+import Control.Monad.ST (runST, stToIO)
+import Data.Time (Day)
+import Data.Word (Word8)
+import DataFrame.IO.CSV.Internal.Options
+import DataFrame.IO.CSV.Internal.Sink (Cells (..), cellAt, withCell)
+import DataFrame.Internal.Column
+import DataFrame.Internal.ColumnBuilder
+import DataFrame.Internal.Parsing
+import DataFrame.Internal.Parsing.Fast (
+    parseBoolFieldSlice,
+    parseDateFieldSlice,
+    parseDoubleFieldSlice,
+    parseIntFieldSlice,
+ )
+import DataFrame.Operations.Typing
+import Foreign.Ptr (Ptr, castPtr, plusPtr)
+
+inferColumnFromBS :: SafeReadMode -> ReadOptions -> Cells -> IO Column
+inferColumnFromBS mode opts cells =
+    let sampleN = let n = typeInferenceSampleSize (typeSpec opts) in if n == 0 then 100 else n
+        dfmt = dateFormat opts
+        -- The sample IS still Maybe-wrapped; it's bounded at 100 rows
+        -- by default, so the allocation is ignorable.
+        samples = V.generate (min sampleN (cLen cells)) $ \i ->
+            if validAt cells i then Just (cellAt cells i) else Nothing
+        assumption = makeParsingAssumptionBytes dfmt samples
+        orText = maybe (textColumn cells) pure
+     in case mode of
+            EitherRead -> pure (handleBSEither dfmt assumption cells)
+            _ -> case assumption of
+                IntAssumption -> orText (handleBSInt cells)
+                DoubleAssumption -> orText (handleBSDouble cells)
+                BoolAssumption -> orText (handleBSBool cells)
+                DateAssumption -> orText (handleBSDate dfmt cells)
+                TextAssumption -> textColumn cells
+                NoAssumption -> handleBSNo dfmt cells
+
+validAt :: Cells -> Int -> Bool
+validAt cells i = VU.unsafeIndex (cValid cells) i /= 0
+{-# INLINE validAt #-}
+
+{- | @%Y-%m-%d@ takes the WS-B slice fast path; custom formats keep the
+reference parser over a materialized cell.
+-}
+dateSliceParser :: String -> Cells -> Int -> Maybe Day
+dateSliceParser "%Y-%m-%d" cells i = withCell cells i parseDateFieldSlice
+dateSliceParser fmt cells i = readByteStringDate fmt (cellAt cells i)
+{-# INLINE dateSliceParser #-}
+
+{- | 'EitherRead' wrap for the ByteString inference path: produce an
+@Either Text a@ column. Raw input bytes are preserved verbatim; rows that were
+marked invalid by the builder (e.g. empty cells before EitherRead disabled
+missing-indicator detection) become @Left \"\"@.
+-}
+handleBSEither :: String -> ParsingAssumption -> Cells -> Column
+handleBSEither dfmt assumption cells = case assumption of
+    BoolAssumption -> wrap readByteStringBool
+    IntAssumption -> wrap readByteStringInt
+    DoubleAssumption -> wrap readByteStringDouble
+    DateAssumption -> wrap (readByteStringDate dfmt)
+    -- Text / No assumption: column is Either Text Text; empty cells become
+    -- Left "" to keep the "Left means missing/failure" convention.
+    TextAssumption -> fromVector (V.generate (cLen cells) textEither)
+    NoAssumption -> fromVector (V.generate (cLen cells) textEither)
+  where
+    wrap ::
+        forall a. (Columnable a) => (BS.ByteString -> Maybe a) -> Column
+    wrap p = fromVector (V.generate (cLen cells) (toEither p))
+
+    toEither ::
+        forall a. (BS.ByteString -> Maybe a) -> Int -> Either T.Text a
+    toEither p i
+        | not (validAt cells i) = Left (TE.decodeUtf8Lenient bs)
+        | otherwise = case p bs of
+            Just v -> Right v
+            Nothing -> Left (TE.decodeUtf8Lenient bs)
+      where
+        bs = cellAt cells i
+
+    textEither :: Int -> Either T.Text T.Text
+    textEither i =
+        let t = TE.decodeUtf8Lenient (cellAt cells i)
+         in if not (validAt cells i) || T.null t then Left t else Right t
+
+handleBSBool :: Cells -> Maybe Column
+handleBSBool cells =
+    uncurry UnboxedColumn
+        <$> parseUnboxedCells False (\i -> withCell cells i parseBoolFieldSlice) cells
+
+{- | Int columns: one fused pass with in-place Int -> Double promotion
+('promoteIntColumnIndexed', audit T2) instead of the full-column Double
+re-parse; both parsers are the WS-B slice parsers (bit-exact with the
+old @readByteString*@ pair).
+-}
+handleBSInt :: Cells -> Maybe Column
+handleBSInt cells =
+    promoteIntColumnIndexed
+        (cLen cells)
+        (not . validAt cells)
+        (\i -> withCell cells i parseIntFieldSlice)
+        (\i -> withCell cells i parseDoubleFieldSlice)
+
+handleBSDouble :: Cells -> Maybe Column
+handleBSDouble cells =
+    uncurry UnboxedColumn
+        <$> parseUnboxedCells 0 (\i -> withCell cells i parseDoubleFieldSlice) cells
+
+-- Dates are boxed ('Day' isn't 'VU.Unbox'), so fuse into a V.Vector
+-- (Maybe Day) in one pass.  Bails on the first non-null cell that
+-- fails to parse.
+handleBSDate :: String -> Cells -> Maybe Column
+handleBSDate dfmt cells =
+    case parseBoxedMaybeCells (dateSliceParser dfmt cells) cells of
+        Just (anyNull, out)
+            | anyNull -> Just (fromVector out)
+            | otherwise -> Just (fromVector (V.mapMaybe id out))
+        Nothing -> Nothing
+
+{- | Text columns build through the WS-C 'TextBuilder': one shared byte
+array + offsets, 'Data.Text.Text' values sliced off it at freeze (UTF-8
+validated once, lenient per-field decode only when invalid).
+-}
+textColumn :: Cells -> IO Column
+textColumn cells = do
+    let n = cLen cells
+    b <- stToIO (newTextBuilder n (n * 8))
+    BSU.unsafeUseAsCStringLen (cBS cells) $ \(p, _) -> do
+        let base = castPtr p :: Ptr Word8
+            go !i
+                | i >= n = pure ()
+                | otherwise = do
+                    let s = VU.unsafeIndex (cOffs cells) (2 * i)
+                        e = VU.unsafeIndex (cOffs cells) (2 * i + 1)
+                    if not (validAt cells i)
+                        then stToIO (appendNull b)
+                        else
+                            if s >= 0
+                                then stToIO (appendTextSliceFromPtr b (base `plusPtr` s) (e - s))
+                                else
+                                    let cell = cEsc cells IM.! e
+                                     in BSU.unsafeUseAsCStringLen cell $ \(q, l) ->
+                                            stToIO
+                                                (appendTextSliceFromPtr b (castPtr q) l)
+                    go (i + 1)
+        go 0
+    stToIO (freezeBuilder b)
+
+handleBSNo :: String -> Cells -> IO Column
+handleBSNo dfmt cells
+    | VU.all (== 0) (cValid cells) =
+        pure (fromVector (V.replicate (cLen cells) (Nothing :: Maybe T.Text)))
+    | Just (mbm, out) <-
+        parseUnboxedCells False (\i -> withCell cells i parseBoolFieldSlice) cells =
+        pure (UnboxedColumn mbm out)
+    | Just (mbm, out) <-
+        parseUnboxedCells 0 (\i -> withCell cells i parseIntFieldSlice) cells =
+        pure (UnboxedColumn mbm out)
+    | Just (mbm, out) <-
+        parseUnboxedCells 0 (\i -> withCell cells i parseDoubleFieldSlice) cells =
+        pure (UnboxedColumn mbm out)
+    | otherwise = maybe (textColumn cells) pure (handleBSDate dfmt cells)
+
+-- Boxed counterpart to 'parseUnboxedCells' for types that aren't
+-- 'VU.Unbox' (e.g. 'Day').  Same one-pass + early-bail shape.
+parseBoxedMaybeCells ::
+    (Int -> Maybe a) -> Cells -> Maybe (Bool, V.Vector (Maybe a))
+parseBoxedMaybeCells parser cells = runST $ do
+    let n = cLen cells
+    out <- VM.new n
+    let loop !i !anyNull
+            | i >= n = do
+                frozen <- V.unsafeFreeze out
+                return (Just (anyNull, frozen))
+            | not (validAt cells i) = do
+                VM.unsafeWrite out i Nothing
+                loop (i + 1) True
+            | otherwise = case parser i of
+                Just v -> do
+                    VM.unsafeWrite out i (Just v)
+                    loop (i + 1) anyNull
+                Nothing -> return Nothing
+    loop 0 False
+
+{- | One-pass fused parse into a typed unboxed column, reading
+nullability from the builder's validity vector. Returns @Just@ only
+when every non-null cell parses; the first unparseable cell
+short-circuits so the caller can fall back to the next assumption.
+-}
+parseUnboxedCells ::
+    forall a.
+    (VU.Unbox a) =>
+    a ->
+    (Int -> Maybe a) ->
+    Cells ->
+    Maybe (Maybe Bitmap, VU.Vector a)
+parseUnboxedCells nullValue parser cells = runST $ do
+    let n = cLen cells
+    values <- VUM.unsafeNew n
+    vmask <- VUM.unsafeNew n
+    let go !i !anyNull
+            | i >= n = finalizeParseResult values vmask anyNull
+            | not (validAt cells i) = do
+                VUM.unsafeWrite vmask i 0
+                VUM.unsafeWrite values i nullValue
+                go (i + 1) True
+            | otherwise = case parser i of
+                Just v -> do
+                    VUM.unsafeWrite vmask i 1
+                    VUM.unsafeWrite values i v
+                    go (i + 1) anyNull
+                Nothing -> return Nothing
+    go 0 False
+{-# INLINE parseUnboxedCells #-}
diff --git a/src/DataFrame/IO/CSV/Internal/Options.hs b/src/DataFrame/IO/CSV/Internal/Options.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Internal/Options.hs
@@ -0,0 +1,140 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+{- | Configuration types for the CSV readers. Re-exported through
+"DataFrame.IO.CSV"; shared by the default reader, @dataframe-fastcsv@
+and @dataframe-lazy@.
+-}
+module DataFrame.IO.CSV.Internal.Options (
+    HeaderSpec (..),
+    TypeSpec (..),
+    RaggedRowPolicy (..),
+    UnclosedQuotePolicy (..),
+    ReadOptions (..),
+    defaultReadOptions,
+    shouldInferFromSample,
+    schemaTypeMap,
+    typeInferenceSampleSize,
+) where
+
+import qualified Data.Map.Strict as M
+import qualified Data.Text as T
+
+import DataFrame.Internal.Schema (SchemaType)
+import DataFrame.Operations.Typing (SafeReadMode (..))
+
+data HeaderSpec = NoHeader | UseFirstRow | ProvideNames [T.Text]
+    deriving (Eq, Show)
+
+data TypeSpec
+    = InferFromSample Int
+    | SpecifyTypes [(T.Text, SchemaType)] TypeSpec
+    | NoInference
+
+{- | How the fast reader should treat a row whose field count does not
+match the header row.  Only consulted by @dataframe-fastcsv@; the pure
+Haskell reader always pads short rows with nulls and drops extras.
+-}
+data RaggedRowPolicy
+    = {- | Fill missing cells with nulls; silently drop extras.
+      Matches pandas / polars lenient defaults.
+      -}
+      PadWithNull
+    | {- | Fill missing cells with nulls; silently drop extras.  Alias
+      kept for ergonomic naming when the caller only cares about the
+      "don't raise, just forget the extras" half of 'PadWithNull'.
+      -}
+      Truncate
+    | {- | Raise a 'DataFrame.IO.CSV.Fast.CsvParseError' on any row whose
+      field count differs from the header.  Matches polars' strict
+      schema-bound mode.
+      -}
+      RaiseOnRagged
+    deriving (Eq, Show)
+
+-- | How the fast reader should treat an unclosed quoted field at EOF.
+data UnclosedQuotePolicy
+    = -- | Raise a 'DataFrame.IO.CSV.Fast.CsvParseError'.  Default.
+      RaiseOnUnclosedQuote
+    | {- | Return whatever rows were parsed before the stray quote; the
+      remainder is silently dropped.
+      -}
+      BestEffort
+    deriving (Eq, Show)
+
+-- | CSV read parameters.
+data ReadOptions = ReadOptions
+    { headerSpec :: HeaderSpec
+    -- ^ Where to get the headers from. (default: UseFirstRow)
+    , typeSpec :: TypeSpec
+    -- ^ Whether/how to infer types. (default: InferFromSample 100)
+    , safeRead :: SafeReadMode
+    {- ^ Default 'SafeReadMode' for columns without an entry in
+    'safeReadOverrides'. (default: 'NoSafeRead')
+    -}
+    , safeReadOverrides :: [(T.Text, SafeReadMode)]
+    -- ^ Per-column 'SafeReadMode' overrides; takes precedence over 'safeRead'.
+    , dateFormat :: String
+    {- ^ Format of date fields as recognized by the Data.Time.Format module.
+
+    __Examples:__
+
+    @
+    > parseTimeM True defaultTimeLocale "%Y/%-m/%-d" "2010/3/04" :: Maybe Day
+    Just 2010-03-04
+    > parseTimeM True defaultTimeLocale "%d/%-m/%-Y" "04/3/2010" :: Maybe Day
+    Just 2010-03-04
+    @
+    -}
+    , columnSeparator :: Char
+    -- ^ Character that separates column values.
+    , numColumns :: Maybe Int
+    {- ^ Maximum number of data rows to read ('Nothing' = all). Note: despite
+    the legacy name, this has always capped /rows/, not columns.
+    -}
+    , missingIndicators :: [T.Text]
+    -- ^ Values that should be read as `Nothing`.
+    , fastCsvOnRaggedRow :: RaggedRowPolicy
+    {- ^ @dataframe-fastcsv@: how to treat rows with a non-header field count.
+    (default: 'PadWithNull')
+    -}
+    , fastCsvOnUnclosedQuote :: UnclosedQuotePolicy
+    {- ^ @dataframe-fastcsv@: how to treat an unclosed quoted field at EOF.
+    (default: 'RaiseOnUnclosedQuote')
+    -}
+    , fastCsvTrimUnquoted :: Bool
+    {- ^ @dataframe-fastcsv@: if 'True', leading/trailing whitespace is
+    stripped from unquoted fields after decoding.  RFC 4180 preserves
+    this whitespace, and that is the default ('False').
+    -}
+    }
+
+shouldInferFromSample :: TypeSpec -> Bool
+shouldInferFromSample (InferFromSample _) = True
+shouldInferFromSample (SpecifyTypes _ fallback) = shouldInferFromSample fallback
+shouldInferFromSample _ = False
+
+schemaTypeMap :: TypeSpec -> M.Map T.Text SchemaType
+schemaTypeMap (SpecifyTypes xs _) = M.fromList xs
+schemaTypeMap _ = M.empty
+
+typeInferenceSampleSize :: TypeSpec -> Int
+typeInferenceSampleSize (InferFromSample n) = n
+typeInferenceSampleSize (SpecifyTypes _ fallback) = typeInferenceSampleSize fallback
+typeInferenceSampleSize _ = 0
+
+defaultReadOptions :: ReadOptions
+defaultReadOptions =
+    ReadOptions
+        { headerSpec = UseFirstRow
+        , typeSpec = InferFromSample 100
+        , safeRead = NoSafeRead
+        , safeReadOverrides = []
+        , dateFormat = "%Y-%m-%d"
+        , columnSeparator = ','
+        , numColumns = Nothing
+        , missingIndicators =
+            ["Nothing", "NULL", "", " ", "nan", "null", "N/A", "NaN", "NAN", "NA"]
+        , fastCsvOnRaggedRow = PadWithNull
+        , fastCsvOnUnclosedQuote = RaiseOnUnclosedQuote
+        , fastCsvTrimUnquoted = False
+        }
diff --git a/src/DataFrame/IO/CSV/Internal/Read.hs b/src/DataFrame/IO/CSV/Internal/Read.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Internal/Read.hs
@@ -0,0 +1,173 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+
+{- | Driver for the default CSV reader (Round-2 WS-D): a single strict
+scan feeding per-column sinks, with cassava-parity semantics pinned by
+@IO.CsvGolden@. Ragged rows pad trailing columns with null and drop
+extra fields (audit D6). The result is fully forced (audit D7).
+-}
+module DataFrame.IO.CSV.Internal.Read (decodeCsvStrict) where
+
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Unsafe as BSU
+import qualified Data.Map.Strict as M
+import qualified Data.Proxy as P
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import qualified Data.Vector as V
+
+import Control.Monad (when, zipWithM)
+import Control.Monad.ST (stToIO)
+import Data.Char (ord)
+import Data.Maybe (fromMaybe)
+import Data.Type.Equality (TestEquality (testEquality))
+import Data.Word (Word8)
+import DataFrame.IO.CSV.Internal.Infer
+import DataFrame.IO.CSV.Internal.Options
+import DataFrame.IO.CSV.Internal.Scanner
+import DataFrame.IO.CSV.Internal.Sink
+import DataFrame.Internal.Column (Column, ensureOptional)
+import DataFrame.Internal.ColumnBuilder
+import DataFrame.Internal.DataFrame (DataFrame (..), forceDataFrame)
+import DataFrame.Internal.Schema (SchemaType (..), schemaType)
+import DataFrame.Operations.Typing (
+    SafeReadMode (..),
+    effectiveSafeRead,
+    parseWithTypes,
+ )
+import Foreign.Ptr (castPtr)
+import Type.Reflection (typeRep)
+
+-- | Decode a strict CSV buffer into a fully forced DataFrame.
+decodeCsvStrict :: ReadOptions -> BS.ByteString -> IO DataFrame
+decodeCsvStrict opts bs = BSU.unsafeUseAsCStringLen bs $ \(cstr, _) -> do
+    let !len = BS.length bs
+        !sep = fromIntegral (ord (columnSeparator opts)) :: Word8
+
+        -- Position of the next non-blank record at/after @pos@ (cassava
+        -- drops single-empty-field records); @len@ when none remain.
+        findRecord !pos
+            | pos >= len = len
+            | otherwise = withField bs len sep pos $ \cs ce _ term next ->
+                if cs == ce && term /= termSep
+                    then if term == termEof then len else findRecord next
+                    else pos
+
+        -- Decode the record at @pos@ as header fields; returns the
+        -- position after the record.
+        headerFields !pos = go pos id
+          where
+            go !p acc = withField bs len sep p $ \cs ce unesc term next ->
+                let t =
+                        TE.decodeUtf8Lenient
+                            (if unesc then unescapeQuotes bs cs ce else sliceBS bs cs ce)
+                 in if term == termSep
+                        then go next (acc . (t :))
+                        else (acc [t], next)
+
+    let hdrPos = findRecord 0
+    when (hdrPos >= len) (error "Empty CSV file")
+    let (hdrFields, afterHdr) = headerFields hdrPos
+        positional n = map (T.pack . show) [0 .. n - 1 :: Int]
+        (names, dataPos0) = case headerSpec opts of
+            UseFirstRow -> (map T.strip hdrFields, afterHdr)
+            NoHeader -> (positional (length hdrFields), hdrPos)
+            ProvideNames ns ->
+                (ns ++ drop (length ns) (positional (length hdrFields)), hdrPos)
+        !ncols = length names
+        dataPos = findRecord dataPos0
+    when (dataPos >= len) (error "Empty CSV file")
+
+    let resolveMode = effectiveSafeRead (safeRead opts) (safeReadOverrides opts)
+        -- If ANY column is EitherRead we keep every raw cell (including
+        -- "N/A" etc.) verbatim; otherwise the missing-indicator list applies.
+        anyEither = any (\n -> resolveMode n == EitherRead) names
+        missing = if anyEither then [] else missingIndicators opts
+        missMode
+            | null missing = MissNone
+            | missing == missingIndicators defaultReadOptions = MissCanonical
+            | otherwise = MissCustom
+        env = Env bs (castPtr cstr) missMode missing
+        rowHint = len `div` max 1 (ncols * 8) + 16
+
+    sinks <- mapM (newSink opts resolveMode rowHint) names
+    let !sinksV = V.fromList sinks
+
+    let pad !row !col
+            | col >= ncols = pure ()
+            | otherwise = nullSink (V.unsafeIndex sinksV col) row >> pad row (col + 1)
+        dec b = if b > 0 then b - 1 else b
+        rowLoop !pos !row !budget
+            | budget == 0 || pos >= len = pure row
+            | otherwise = withField bs len sep pos $ \cs ce unesc term next ->
+                if cs == ce && term /= termSep
+                    then if term == termEof then pure row else rowLoop next row budget
+                    else do
+                        feedSink env (V.unsafeIndex sinksV 0) row cs ce unesc
+                        if term == termSep
+                            then fieldLoop next row 1 budget
+                            else pad row 1 >> rowLoop next (row + 1) (dec budget)
+        fieldLoop !pos !row !col !budget =
+            withField bs len sep pos $ \cs ce unesc term next -> do
+                when (col < ncols) $
+                    feedSink env (V.unsafeIndex sinksV col) row cs ce unesc
+                if term == termSep
+                    then fieldLoop next row (col + 1) budget
+                    else pad row (col + 1) >> rowLoop next (row + 1) (dec budget)
+
+    nrows <- rowLoop dataPos 0 (fromMaybe (-1) (numColumns opts))
+
+    cols <- zipWithM (finalizeSink bs opts resolveMode nrows) names sinks
+    let df =
+            DataFrame
+                (V.fromList cols)
+                (M.fromList (zip names [0 ..]))
+                (nrows, ncols)
+                M.empty
+    pure $!
+        forceDataFrame
+            (parseWithTypes resolveMode (schemaTypeMap (typeSpec opts)) df)
+
+{- | Resolve a column's sink: EitherRead and to-be-inferred columns
+collect raw slices; schema'd Int/Double parse straight into builders;
+every other schema type builds Text ('parseWithTypes' converts after).
+-}
+newSink ::
+    ReadOptions -> (T.Text -> SafeReadMode) -> Int -> T.Text -> IO Sink
+newSink opts resolveMode rowHint name
+    | resolveMode name == EitherRead = SinkBS <$> newSliceCol rowHint
+    | Nothing <- entry
+    , shouldInferFromSample (typeSpec opts) =
+        SinkBS <$> newSliceCol rowHint
+    | otherwise = case fromMaybe (schemaType @T.Text) entry of
+        SType (_ :: P.Proxy a) -> case testEquality (typeRep @a) (typeRep @Int) of
+            Just _ -> SinkInt <$> stToIO (newIntBuilder rowHint)
+            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of
+                Just _ -> SinkDouble <$> stToIO (newDoubleBuilder rowHint)
+                Nothing ->
+                    SinkText <$> stToIO (newTextBuilder rowHint (rowHint * 8))
+  where
+    entry = M.lookup name (schemaTypeMap (typeSpec opts))
+
+finalizeSink ::
+    BS.ByteString ->
+    ReadOptions ->
+    (T.Text -> SafeReadMode) ->
+    Int ->
+    T.Text ->
+    Sink ->
+    IO Column
+finalizeSink bs opts resolveMode nrows name sink = do
+    col <- case sink of
+        SinkInt b -> stToIO (freezeBuilder b)
+        SinkDouble b -> stToIO (freezeBuilder b)
+        SinkText b -> stToIO (freezeBuilder b)
+        SinkBS sc -> do
+            cells <- freezeCells bs sc nrows
+            inferColumnFromBS (resolveMode name) opts cells
+    pure $! case resolveMode name of
+        MaybeRead -> ensureOptional col
+        _ -> col
diff --git a/src/DataFrame/IO/CSV/Internal/Scanner.hs b/src/DataFrame/IO/CSV/Internal/Scanner.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Internal/Scanner.hs
@@ -0,0 +1,154 @@
+{-# LANGUAGE BangPatterns #-}
+
+{- | RFC 4180 field scanner for the default CSV reader (Round-2 WS-D).
+One pass over a strict 'BS.ByteString'; cassava-parity semantics pinned
+by the @IO.CsvGolden@ suite: quotes open fields only at the first byte,
+a double quote inside an unquoted field is an error, garbage after a
+closing quote is an error, an unclosed quote consumes to EOF, and rows
+end at @\\n@, @\\r\\n@ or a lone @\\r@.
+-}
+module DataFrame.IO.CSV.Internal.Scanner (
+    Term,
+    termSep,
+    termEol,
+    termEof,
+    withField,
+    withStripC8,
+    withStripAscii,
+    unescapeQuotes,
+    isBlankRecordAt,
+) where
+
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Internal as BSI
+import qualified Data.ByteString.Unsafe as BSU
+
+import Data.Word (Word8)
+import Foreign.Ptr (Ptr)
+import Foreign.Storable (pokeByteOff)
+
+-- | What ended a field: another field follows, the row ended, or EOF.
+type Term = Int
+
+termSep, termEol, termEof :: Term
+termSep = 0
+termEol = 1
+termEof = 2
+
+quote, nl, cr :: Word8
+quote = 34
+nl = 10
+cr = 13
+
+parseError :: Int -> String -> a
+parseError off what =
+    error ("CSV Parse Error: " ++ what ++ " at byte " ++ show off)
+
+{- | Scan one field starting at @pos@ and continue with
+@k contentStart contentEnd needsUnescape term nextPos@. Quoted fields
+yield the bytes between the outer quotes; @needsUnescape@ is set when
+the content contains doubled quotes ('unescapeQuotes' collapses them).
+-}
+{-# INLINE withField #-}
+withField ::
+    BS.ByteString ->
+    Int ->
+    Word8 ->
+    Int ->
+    (Int -> Int -> Bool -> Term -> Int -> r) ->
+    r
+withField bs !len !sep !pos k
+    | pos < len && BSU.unsafeIndex bs pos == quote = quoted (pos + 1) False
+    | otherwise = unquoted pos
+  where
+    -- Field terminator dispatch shared by both shapes.
+    {-# INLINE finish #-}
+    finish !cs !ce !unesc !i
+        | i >= len = k cs ce unesc termEof i
+        | otherwise = case BSU.unsafeIndex bs i of
+            b
+                | b == sep -> k cs ce unesc termSep (i + 1)
+                | b == nl -> k cs ce unesc termEol (i + 1)
+                | b == cr ->
+                    if i + 1 < len && BSU.unsafeIndex bs (i + 1) == nl
+                        then k cs ce unesc termEol (i + 2)
+                        else k cs ce unesc termEol (i + 1)
+                | otherwise ->
+                    parseError i "malformed quoted field (garbage after closing quote)"
+    unquoted !i
+        | i >= len = k pos i False termEof i
+        | otherwise = case BSU.unsafeIndex bs i of
+            b
+                | b == sep -> k pos i False termSep (i + 1)
+                | b == nl -> k pos i False termEol (i + 1)
+                | b == cr ->
+                    if i + 1 < len && BSU.unsafeIndex bs (i + 1) == nl
+                        then k pos i False termEol (i + 2)
+                        else k pos i False termEol (i + 1)
+                | b == quote -> parseError i "stray double quote in unquoted field"
+                | otherwise -> unquoted (i + 1)
+    quoted !i !unesc
+        | i >= len = k (pos + 1) len unesc termEof len -- unclosed: rest of input
+        | BSU.unsafeIndex bs i == quote =
+            if i + 1 < len && BSU.unsafeIndex bs (i + 1) == quote
+                then quoted (i + 2) True
+                else finish (pos + 1) i unesc (i + 1)
+        | otherwise = quoted (i + 1) unesc
+
+{- | Whether the record starting at @pos@ is a blank record (a single
+empty field, quoted or not, then end of row) — cassava drops these.
+Assumes @pos < len@. Errors propagate from 'withField'.
+-}
+{-# INLINE isBlankRecordAt #-}
+isBlankRecordAt :: BS.ByteString -> Int -> Word8 -> Int -> Bool
+isBlankRecordAt bs !len !sep !pos =
+    withField bs len sep pos (\cs ce _ term _ -> cs == ce && term /= termSep)
+
+{- | Collapse doubled quotes (@\"\"@ → @\"@) in @[s, e)@ into a fresh strict
+'BS.ByteString'. Every quote in the range is the first of a pair.
+-}
+unescapeQuotes :: BS.ByteString -> Int -> Int -> BS.ByteString
+unescapeQuotes bs !s !e = BSI.unsafeCreateUptoN (e - s) (go s 0)
+  where
+    go !i !j !dst
+        | i >= e = pure j
+        | otherwise = do
+            let w = BSU.unsafeIndex bs i
+            pokeByteOff (dst :: Ptr Word8) j w
+            if w == quote then go (i + 2) (j + 1) dst else go (i + 1) (j + 1) dst
+
+{- | Strip with @Data.ByteString.Char8.strip@ semantics (Latin-1
+whitespace: HT..CR, space and NBSP 0xA0) and continue with the
+stripped range.
+-}
+{-# INLINE withStripC8 #-}
+withStripC8 :: BS.ByteString -> Int -> Int -> (Int -> Int -> r) -> r
+withStripC8 bs !s0 !e0 k = k s e
+  where
+    isC8Space w = w == 32 || (w - 9) <= 4 || w == 160
+    s = skipF s0
+    e = skipB e0
+    skipF !i
+        | i < e0 && isC8Space (BSU.unsafeIndex bs i) = skipF (i + 1)
+        | otherwise = i
+    skipB !i
+        | i > s && isC8Space (BSU.unsafeIndex bs (i - 1)) = skipB (i - 1)
+        | otherwise = i
+
+{- | Strip ASCII whitespace (the ASCII subset of @Data.Text.strip@) and
+continue with the stripped range. Callers must fall back to a full
+'Data.Text.strip' when the stripped range still has non-ASCII edges.
+-}
+{-# INLINE withStripAscii #-}
+withStripAscii :: BS.ByteString -> Int -> Int -> (Int -> Int -> r) -> r
+withStripAscii bs !s0 !e0 k = k s e
+  where
+    isAsciiSpace w = w == 32 || (w - 9) <= 4
+    s = skipF s0
+    e = skipB e0
+    skipF !i
+        | i < e0 && isAsciiSpace (BSU.unsafeIndex bs i) = skipF (i + 1)
+        | otherwise = i
+    skipB !i
+        | i > s && isAsciiSpace (BSU.unsafeIndex bs (i - 1)) = skipB (i - 1)
+        | otherwise = i
diff --git a/src/DataFrame/IO/CSV/Internal/Sink.hs b/src/DataFrame/IO/CSV/Internal/Sink.hs
new file mode 100644
--- /dev/null
+++ b/src/DataFrame/IO/CSV/Internal/Sink.hs
@@ -0,0 +1,239 @@
+{-# LANGUAGE BangPatterns #-}
+
+{- | Per-column sinks for the default CSV reader. Typed columns (explicit
+schema) parse straight from the input slice into the WS-C builders;
+inferred and EitherRead columns record unboxed @(start, end)@ offsets
+into the input buffer (plus a cold side map for quote-unescaped cells)
+for the inference pass ("DataFrame.IO.CSV.Internal.Infer").
+-}
+module DataFrame.IO.CSV.Internal.Sink (
+    Env (..),
+    MissingMode (..),
+    Sink (..),
+    SliceCol,
+    Cells (..),
+    cellAt,
+    withCell,
+    newSliceCol,
+    feedSink,
+    nullSink,
+    freezeCells,
+    sliceBS,
+) where
+
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Unsafe as BSU
+import qualified Data.IntMap.Strict as IM
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import qualified Data.Vector.Unboxed as VU
+import qualified Data.Vector.Unboxed.Mutable as VUM
+
+import Control.Monad.ST (RealWorld, stToIO)
+import Data.IORef
+import Data.Word (Word8)
+import DataFrame.IO.CSV.Internal.Scanner
+import DataFrame.Internal.ColumnBuilder
+import DataFrame.Internal.Parsing.Fast (
+    isMissingFieldSlice,
+    parseDoubleFieldSlice,
+    parseIntFieldSlice,
+ )
+import Foreign.Ptr (Ptr, castPtr, plusPtr)
+
+{- | How missing tokens are detected (audit D2: canonical list goes
+through the WS-B memcmp fast path; custom lists decode and compare).
+-}
+data MissingMode = MissNone | MissCanonical | MissCustom
+
+-- | Read-time environment shared by every sink.
+data Env = Env
+    { envBS :: !BS.ByteString
+    -- ^ Whole input (BOM-stripped when the caller stripped it).
+    , envPtr :: !(Ptr Word8)
+    -- ^ Base pointer of 'envBS' (valid for the whole decode).
+    , envMode :: !MissingMode
+    , envTexts :: ![T.Text]
+    -- ^ The effective missing-indicator list ('Data.Text' level).
+    }
+
+data Sink
+    = SinkInt !(IntBuilder RealWorld)
+    | SinkDouble !(DoubleBuilder RealWorld)
+    | SinkText !(TextBuilder RealWorld)
+    | SinkBS !SliceCol
+
+{- | Growable cell store for inferred columns: per cell a @(start, end)@
+offset pair into the input (already Char8-stripped) and a 0\/1 validity
+byte. Quote-unescaped cells (cold) live in an 'IM.IntMap' keyed by row;
+their offset pair is @(-1, row)@.
+-}
+data SliceCol = SliceCol
+    { scOffs :: !(IORef (VUM.IOVector Int))
+    , scValid :: !(IORef (VUM.IOVector Word8))
+    , scEsc :: !(IORef (IM.IntMap BS.ByteString))
+    }
+
+-- | Frozen 'SliceCol' plus the buffer it points into.
+data Cells = Cells
+    { cBS :: !BS.ByteString
+    , cOffs :: !(VU.Vector Int)
+    , cEsc :: !(IM.IntMap BS.ByteString)
+    , cValid :: !(VU.Vector Word8)
+    , cLen :: !Int
+    }
+
+{- | Resolve cell @i@ as a slice and continue (no 'BS.ByteString' header
+is allocated for the common in-buffer case).
+-}
+withCell :: Cells -> Int -> (BS.ByteString -> Int -> Int -> r) -> r
+withCell c !i k =
+    let s = VU.unsafeIndex (cOffs c) (2 * i)
+        e = VU.unsafeIndex (cOffs c) (2 * i + 1)
+     in if s >= 0
+            then k (cBS c) s e
+            else let cell = cEsc c IM.! e in k cell 0 (BS.length cell)
+{-# INLINE withCell #-}
+
+-- | Cell @i@ as a (zero-copy) 'BS.ByteString'.
+cellAt :: Cells -> Int -> BS.ByteString
+cellAt c i = withCell c i sliceBS
+{-# INLINE cellAt #-}
+
+newSliceCol :: Int -> IO SliceCol
+newSliceCol hint = do
+    let cap = max 16 hint
+    offs <- VUM.unsafeNew (2 * cap)
+    val <- VUM.unsafeNew cap
+    SliceCol <$> newIORef offs <*> newIORef val <*> newIORef IM.empty
+
+appendSliceCol :: SliceCol -> Int -> Int -> Int -> Word8 -> IO ()
+appendSliceCol (SliceCol oRef vRef _) !row !s !e !ok = do
+    offs <- readIORef oRef
+    offs' <-
+        if 2 * row < VUM.length offs
+            then pure offs
+            else do
+                o <- VUM.unsafeGrow offs (VUM.length offs)
+                writeIORef oRef o
+                pure o
+    VUM.unsafeWrite offs' (2 * row) s
+    VUM.unsafeWrite offs' (2 * row + 1) e
+    val <- readIORef vRef
+    val' <-
+        if row < VUM.length val
+            then pure val
+            else do
+                v <- VUM.unsafeGrow val (VUM.length val)
+                writeIORef vRef v
+                pure v
+    VUM.unsafeWrite val' row ok
+{-# INLINE appendSliceCol #-}
+
+-- | Freeze the first @n@ cells (the kept row count).
+freezeCells :: BS.ByteString -> SliceCol -> Int -> IO Cells
+freezeCells bs (SliceCol oRef vRef eRef) n = do
+    offs <- readIORef oRef
+    val <- readIORef vRef
+    esc <- readIORef eRef
+    Cells bs
+        <$> VU.freeze (VUM.slice 0 (2 * n) offs)
+        <*> pure esc
+        <*> VU.freeze (VUM.slice 0 n val)
+        <*> pure n
+
+-- | O(1) sub-'BS.ByteString' of a buffer (no copy).
+sliceBS :: BS.ByteString -> Int -> Int -> BS.ByteString
+sliceBS bs s e = BSU.unsafeTake (e - s) (BSU.unsafeDrop s bs)
+{-# INLINE sliceBS #-}
+
+{- | Feed one field slice @[s, e)@ of the input into a sink. @unesc@
+fields (quoted with embedded @\"\"@) are unescaped into a fresh strict
+'BS.ByteString' first (cold path).
+-}
+feedSink :: Env -> Sink -> Int -> Int -> Int -> Bool -> IO ()
+feedSink env sink !row !s !e !unesc = case sink of
+    SinkBS sc
+        | unesc -> do
+            let fresh = unescapeQuotes (envBS env) s e
+                stripped = withStripC8 fresh 0 (BS.length fresh) (sliceBS fresh)
+            modifyIORef' (scEsc sc) (IM.insert row stripped)
+            appendSliceCol sc row (-1) row (missingOk env stripped)
+        | otherwise -> withStripC8 (envBS env) s e $ \s' e' ->
+            let ok = case envMode env of
+                    MissNone -> 1
+                    MissCanonical ->
+                        if isMissingFieldSlice (envBS env) s' e' then 0 else 1
+                    MissCustom -> missingOk env (sliceBS (envBS env) s' e')
+             in appendSliceCol sc row s' e' ok
+    _
+        | unesc ->
+            let fresh = unescapeQuotes (envBS env) s e
+             in BSU.unsafeUseAsCStringLen fresh $ \(p, l) ->
+                    feedTyped env sink fresh (castPtr p) 0 l
+        | otherwise -> feedTyped env sink (envBS env) (envPtr env) s e
+{-# INLINE feedSink #-}
+
+-- | Validity byte for an already-stripped cell.
+missingOk :: Env -> BS.ByteString -> Word8
+missingOk env cell = case envMode env of
+    MissNone -> 1
+    MissCanonical -> if isMissingFieldSlice cell 0 (BS.length cell) then 0 else 1
+    MissCustom -> if TE.decodeUtf8Lenient cell `elem` envTexts env then 0 else 1
+
+feedTyped ::
+    Env -> Sink -> BS.ByteString -> Ptr Word8 -> Int -> Int -> IO ()
+feedTyped env sink bs !ptr !s !e = case sink of
+    SinkInt b -> case parseIntFieldSlice bs s e of
+        Just v -> stToIO (appendInt b v)
+        Nothing -> stToIO (appendNull b)
+    SinkDouble b -> case parseDoubleFieldSlice bs s e of
+        Just v -> stToIO (appendDouble b v)
+        Nothing -> stToIO (appendNull b)
+    SinkText b -> appendTextCell env b bs ptr s e
+    SinkBS _ -> error "feedTyped: SinkBS handled by feedSink"
+{-# INLINE feedTyped #-}
+
+{- | Text cells reproduce @T.strip . decodeUtf8Lenient@ plus the
+missing-indicator test. Fast path: ASCII-edge slices append raw bytes
+(the shared-buffer freeze decodes leniently per field); anything with
+non-ASCII edges or a custom missing list goes through real 'T.strip'.
+-}
+appendTextCell ::
+    Env ->
+    TextBuilder RealWorld ->
+    BS.ByteString ->
+    Ptr Word8 ->
+    Int ->
+    Int ->
+    IO ()
+appendTextCell env b bs !ptr !s !e =
+    withStripAscii bs s e $ \s' e' ->
+        if s' < e'
+            && (BSU.unsafeIndex bs s' >= 0x80 || BSU.unsafeIndex bs (e' - 1) >= 0x80)
+            then slow
+            else case envMode env of
+                MissNone -> fast s' e'
+                MissCanonical ->
+                    if isMissingFieldSlice bs s' e'
+                        then stToIO (appendNull b)
+                        else fast s' e'
+                MissCustom -> slow
+  where
+    fast s' e' = stToIO (appendTextSliceFromPtr b (ptr `plusPtr` s') (e' - s'))
+    slow = do
+        let t = T.strip (TE.decodeUtf8Lenient (sliceBS bs s e))
+            missing = case envMode env of
+                MissNone -> False
+                _ -> t `elem` envTexts env
+        if missing then stToIO (appendNull b) else stToIO (appendText b t)
+{-# INLINE appendTextCell #-}
+
+-- | Append a null cell (pad-with-null for short rows, audit D6).
+nullSink :: Sink -> Int -> IO ()
+nullSink sink !row = case sink of
+    SinkInt b -> stToIO (appendNull b)
+    SinkDouble b -> stToIO (appendNull b)
+    SinkText b -> stToIO (appendNull b)
+    SinkBS sc -> appendSliceCol sc row 0 0 0
+{-# INLINE nullSink #-}