feat: add JSONL dataset store with on-the-fly tokenization

- Add JsonlStore registered under "jsonl" in astrai/dataset/storage.py
- Reuse PipelineConfig schema for JSONL dataset configuration
- Update detect_format to recognize JSONL directories and files
- Move save_h5/load_h5/save_bin/load_bin to astrai/serialization
- Split astrai/serialization.py into checkpoint/dataset submodules
- Add tests for JSONL detection, seq/SFT stores, and config roundtrip

.gitignore

@@ -5,8 +5,10 @@
 !*/
 
 # Allow specific file types and root files
-!*.py
+!astrai/**/*.py
-!*.sh
+!scripts/**/*.py
+!scripts/**/*.sh
+!tests/**/*.py
 
 # Allow GitHub files
 !/.github/**

assets/docs/architecture.md

@@ -21,6 +21,7 @@ classDiagram
 
         class BaseModelConfig {
             +Optional[str] model_type
+            +float neftune_alpha
             +from_file(config_path) Self
             +to_file(config_path)
         }
@@ -58,10 +59,12 @@ classDiagram
             +Optional[int] dim_ffn
             +Optional[int] max_len
             +Optional[float] rope_theta
+            +str attn_type
             +Optional[int] n_heads
             +Optional[int] n_kv_heads
             +Optional[bool] use_qk_norm
             +Optional[bool] use_gated_attention
+            +str ffn_type
             +Optional[dict] rope_scaling
             +Optional[str] pooling_type
             +Optional[bool] normalize_embeddings
@@ -118,7 +121,7 @@ classDiagram
             +float max_grad_norm
             +list gradient_checkpointing_modules
             +int start_epoch
-            +int start_batch
+            +int start_samples
             +str ckpt_dir
             +int ckpt_interval
             +str log_dir
@@ -136,7 +139,9 @@ classDiagram
             +str start_method
             +str device_type
             +Optional[Dataset] val_dataset
+            +Optional[float] val_split
             +int val_step
+            +float neftune_alpha
             +str parallel_mode
             +dict executor_kwargs
             +dict extra_kwargs
@@ -215,12 +220,13 @@ classDiagram
         class Checkpoint {
             +dict state_dict
             +int epoch
-            +int iteration
+            +int consumed_samples
             +dict extra
             +dict meta
             +dict config
             +save(save_dir)
             +load(save_dir, broadcast) Checkpoint
+            +load_any(save_dir, broadcast) Optional[Checkpoint]
         }
     }
 
@@ -350,7 +356,9 @@ classDiagram
 
         class Embedding {
             +Parameter weight
+            +float neftune_noise_alpha
             +forward(x) Tensor
+            +set_neftune_alpha(alpha)
         }
     }
 
@@ -407,7 +415,9 @@ classDiagram
             +Dict _entries
             +register(name) decorator
             +create(name, *args, **kwargs) T
+            +get_component_class(name) Type
             +list_registered() list
+            +is_registered(name) bool
         }
 
         class MaskBuilderFactory {
@@ -436,13 +446,15 @@ classDiagram
             +dict model_config
             +BaseExecutor executor
             +int epoch
-            +int iteration
+            +int consumed_samples
             +float loss
+            +float grad_norm
             +DataLoader val_dataloader
             +float val_loss
             +int world_size
             +int rank
             +dict kwargs
+            +optimizer_step() int
         }
 
         class TrainContextBuilder {
@@ -594,18 +606,6 @@ classDiagram
             +create(name, **kwargs) TrainCallback
         }
 
-        class Muon {
-            +float lr
-            +float momentum
-            +float weight_decay
-            +bool nesterov
-            +int ns_steps
-            +Optional[float] adamw_lr
-            +tuple adamw_betas
-            +float adamw_eps
-            +float adamw_wd
-            +step(closure) Optional[float]
-        }
     }
 
     namespace inference {
@@ -810,7 +810,9 @@ classDiagram
 
         class ChatMessage {
             +str role
-            +str content
+            +Optional[str] content
+            +Optional[List[Dict]] tool_calls
+            +Optional[str] tool_call_id
         }
 
         class ChatCompletionRequest {
@@ -827,6 +829,8 @@ classDiagram
             +Optional[float] frequency_penalty
             +Optional[Dict[int, float]] logit_bias
             +Optional[str] user
+            +Optional[List[ToolDef]] tools
+            +Optional[Union[str, Dict]] tool_choice
         }
 
         class AnthropicMessage {
@@ -850,7 +854,7 @@ classDiagram
             <<abstract>>
             +prepare(request, engine) Tuple[str, GenContext, List[str]]
             +format_stream_start(ctx) List[str]
-            +format_chunk(token) str
+            +format_chunk(token) List[str]
             +format_stream_end(ctx, stop) List[str]
             +format_response(ctx, content, stop) Dict
         }
@@ -858,7 +862,7 @@ classDiagram
         class OpenAIResponseBuilder {
             +prepare(request, engine) Tuple
             +format_stream_start(ctx) List[str]
-            +format_chunk(token) str
+            +format_chunk(token) List[str]
             +format_stream_end(ctx, stop) List[str]
             +format_response(ctx, content, stop) Dict
         }
@@ -866,7 +870,7 @@ classDiagram
         class AnthropicResponseBuilder {
             +prepare(request, engine) Tuple
             +format_stream_start(ctx) List[str]
-            +format_chunk(token) str
+            +format_chunk(token) List[str]
             +format_stream_end(ctx, stop) List[str]
             +format_response(ctx, content, stop) Dict
         }
@@ -1171,10 +1175,10 @@ classDiagram
 | **astrai.serialization** | Checkpoint | Model serialization |
 | **astrai.model** | AutoModel, AutoRegressiveLM, EmbeddingEncoder, DecoderBlock, GQA, MLA, MLP, DeepSeekMoE, AttnFactory, FFNFactory, RMSNorm, Linear, RotaryEmbedding, Embedding | Neural network model |
 | **astrai.tokenize** | AutoTokenizer, ChatTemplate | Tokenizer and chat template |
-| **astrai.trainer** | Trainer, TrainContext, TrainContextBuilder, BaseStrategy–GRPOStrategy, StrategyFactory, BaseScheduler–WSDScheduler, SchedulerFactory, TrainCallback(Protocol)–ValidationCallback, CallbackFactory, Muon | Training workflow |
+| **astrai.trainer** | Trainer, TrainContext, TrainContextBuilder, BaseStrategy–GRPOStrategy, StrategyFactory, BaseScheduler–WSDScheduler, SchedulerFactory, TrainCallback(Protocol)–ValidationCallback, CallbackFactory | Training workflow |
 | **astrai.inference** | InferenceEngine, InferenceScheduler, Executor, KVCache–KvcacheView, Allocator–Storage, Task, TaskManager, TaskStatus, GenerationRequest, GenerateResult, BaseSamplingStrategy–SamplingPipeline, ProtocolHandler, ResponseBuilder, OpenAIResponseBuilder, AnthropicResponseBuilder, StopChecker, GenContext, ChatMessage–MessagesRequest, app | Inference service |
 | **astrai.parallel** | spawn_parallel_fn, setup_parallel, get_rank/get_world_size/get_current_device, only_on_rank, BaseExecutor, ExecutorFactory, NoneExecutor, DDPExecutor, FSDPExecutor, GradientState, AccumOptimizer, AccumScheduler, ParallelModel, RowParallelLinear, ColumnParallelLinear | Distributed parallel & gradient accumulation |
-| **astrai.factory** | Registry, BaseFactory[T] | Component registration |
+| **astrai.factory** | BaseFactory | Component registration |
 | **astrai.protocols** | OptimizerProtocol, SchedulerProtocol | Structural subtyping for optimizer/scheduler wrappers |
 
 ## Design Patterns

assets/docs/dataflow.md

@@ -46,10 +46,10 @@ The output `meta.json` records the storage format, key names, dtype, total token
 
 ### Format Detection
 
-`detect_format(load_path)` inspects the directory:
+`detect_format(load_path)` inspects the path:
 
-- If `*.h5` files exist → `"h5"` (HDF5 backend)
+- If `load_path` is a file: checks suffix — `.h5`/`.hdf5` → `"h5"`, unknown suffix raises `ValueError`
-- If `*.bin` + `meta.json` files exist → `"bin"` (memory-mapped backend)
+- If `load_path` is a directory: recursively globs for `*.h5`/`*.hdf5` files → `"h5"`, or `*.bin` + `**/meta.json` → `"bin"`
 
 ### Store Backends
 
@@ -83,7 +83,7 @@ DatasetFactory.load(train_type, load_path, window_size, stride=None, storage_typ
     → detect_format(load_path)
     → StoreFactory.create(storage_type)
     → Store.load(load_path)
-      → H5Store._normalize() / MmapStore._normalize()
+      → _normalize(raw)  # base Store, shared by both backends
         → Store._data[Dict[str, List[Tensor]]] + _cum[Dict[str, List[int]]]
           → BaseDataset.__getitem__(idx)
             → get_index(idx) → [begin, end)

assets/docs/inference.md

@@ -23,7 +23,7 @@ RoPE is applied **before** KV cache write, not after — otherwise position enco
 
 ## KVCache System
 
-Six classes (plus two helpers) working together:
+Seven classes working together:
 
 ```
 KVCache (facade)
@@ -152,12 +152,13 @@ Supports `stop_sequences` and streaming via `event: content_block_delta`.
 data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":...,"model":"astrai",
        "choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}
 
-data: {"id":"chatcmpl-...","object":"chat.completion.chunk",...,
+data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":0,"model":"astrai",
        "choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}
 
-data: {"id":"chatcmpl-...","object":"chat.completion.chunk",...,
+data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":...,"model":"astrai",
-       "choices":[{"index":0,"delta":{},"finish_reason":"stop"}],
+       "choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
-       "usage":{"prompt_tokens":5,"completion_tokens":1,"total_tokens":6}}
+
+data: {"prompt_tokens":5,"completion_tokens":1,"total_tokens":6}
 
 data: [DONE]
 ```
@@ -167,7 +168,7 @@ data: [DONE]
 ```
 event: message_start
 data: {"type":"message_start","message":{"id":"msg_...","model":"astrai","role":"assistant",
-       "content":[],"stop_reason":null,...}}
+       "content":[],"usage":{"input_tokens":0}}}
 
 event: content_block_start
 data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
@@ -179,7 +180,7 @@ event: content_block_stop
 data: {"type":"content_block_stop","index":0}
 
 event: message_delta
-data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{...}}
+data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{...}}
 
 event: message_stop
 data: {"type":"message_stop"}
@@ -187,26 +188,20 @@ data: {"type":"message_stop"}
 
 ### Error Responses
 
-All endpoints use standard HTTP status codes:
+The server returns standard HTTP status codes. Pydantic validation errors (e.g. missing required fields)
+are handled automatically by FastAPI with 422 status. The only application-level error is engine initialization:
 
 | Status | Meaning |
 |--------|---------|
 | 200 | Success |
-| 400 | Invalid request (bad JSON, missing fields, validation error) |
-| 405 | Method not allowed |
 | 422 | Unprocessable entity (Pydantic validation) |
-| 500 | Internal server error (model crash, OOM, scheduler failure) |
 | 503 | Service unavailable (model not loaded, engine not ready) |
 
-Error response body:
+Error response body (503):
 
 ```json
 {
-    "error": {
+    "detail": "Engine not initialized"
-        "message": "Invalid request: max_tokens must be > 0",
-        "type": "invalid_request_error",
-        "code": 400
-    }
 }
 ```
 
@@ -220,16 +215,13 @@ Response:
 
 ```json
 {
-    "active_requests": 3,
+    "total_tasks": 128,
-    "waiting_requests": 2,
+    "total_tokens": 10240,
-    "total_requests": 128,
+    "active_tasks": 3,
-    "cache_usage": 0.45,
+    "waiting_queue": 2
-    "tokens_generated": 10240
 }
 ```
 
-`cache_usage` is the fraction of KV cache pages currently in use (0.0–1.0).
-
 ## Engine API
 
 ```python

assets/docs/params.md

@@ -53,7 +53,7 @@
 | `--ckpt_interval` | Iterations between checkpoints | 5000 |
 | `--ckpt_dir` | Checkpoint save directory | checkpoint |
 | `--start_epoch` | Resume from epoch (0 = from scratch) | 0 |
-| `--start_batch` | Resume from batch iteration | 0 |
+| `--start_samples` | Resume from sample count per rank | 0 |
 
 ### Validation
 
@@ -67,8 +67,8 @@
 | Parameter | Description | Default |
 |-----------|-------------|---------|
 | `--log_dir` | Directory for metric logs | checkpoint/logs |
-| `--log_interval` | Number of batch iterations between metric logs | 100 |
+| `--log_interval` | Number of optimizer steps between metric logs | 1 |
-| `--metrics` | Metrics to log (e.g. --metrics loss lr val_loss) | ["loss", "lr"] |
+| `--metrics` | Metrics to log (e.g. --metrics loss lr val_loss) | ["loss", "lr", "grad_norm"] |
 
 ### Gradient Checkpointing
 
@@ -100,6 +100,17 @@
 | `--grpo_sync_interval` | GRPO ref_model sync interval (steps) | 200 | `grpo` |
 | `--neftune_alpha` | NEFTune noise alpha (0=disabled, typical: 5.0) | 0.0 | `sft` |
 
+### Scheduler
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `--schedule_type` | LR scheduler type (`cosine`, `sgdr`, `wsd`) | cosine |
+| `--min_rate` | Minimum LR as fraction of base LR | None (scheduler default) |
+| `--cycle_length` | SGDR first cycle length in steps | None (total_steps - warmup_steps) |
+| `--t_mult` | SGDR cycle length multiplier per restart | 2 |
+| `--stable_steps` | WSD stable plateau steps | None (required for wsd) |
+| `--decay_steps` | WSD decay steps | None (total_steps - warmup_steps - stable_steps) |
+
 ### Usage Example
 
 ```bash
@@ -178,7 +189,7 @@ python scripts/tools/generate.py \
 | `input_files` | path(s) | required | Input JSONL file(s), supports glob (`data/*.jsonl`) |
 | `--output_dir`, `-o` | path | required | Output directory for processed data |
 | `--config`, `-c` | path | required | Preprocessing pipeline config (JSON) |
-| `--num_workers` | int | `4` | Number of parallel workers |
+| `--tokenizer_path` | str | `params` | Path to tokenizer directory |
 
 Usage:
 ```bash

assets/docs/preprocessing.md

@@ -26,8 +26,9 @@ A single config file captures the entire pipeline, reusable and version-controll
 
 ```json
 {
+  "version":       1,
   "input":         {},   // sections (single) or sources (multi)
-  "mask":          {},   // role → "train" | "mask"
+  "mask":          {},   // role -> "train" | "mask"
   "mask_default":  "mask",
   "preprocessing": {},
   "output":        {}
@@ -220,11 +221,12 @@ Config:
 }
 ```
 
-Output keys: `prompts`, `responses`, `masks`, `rewards` (float32)
+Output keys: `prompts`, `prompts_mask`, `responses`, `masks`, `rewards` (float32)
 
 - `action: "value"` — extract raw values from JSONL without tokenisation
 - `list_field: true` — tokenise each list element independently, then concatenate
 - `mask_key: "masks"` — rename the auto-generated mask key (default: `responses_mask`)
+- `prompts_mask` is auto-generated (all masked) and unused by GRPOStrategy
 
 ---
 
@@ -274,12 +276,11 @@ When `sources` is set, `sections` is ignored.
 
 ### Template mode (`template: true`)
 
-For each message in the field's array:
-
 1. Prepend BOS token (masked)
-2. Render through `chat_template` for that single message
+2. For each message in the field's array:
-3. Encode rendered text
+   1. Render through `chat_template` for that single message
-4. Apply mask rule for the message's role
+   2. Encode rendered text
+   3. Apply mask rule for the message's role
 
 ### Non-template mode
 
@@ -287,7 +288,7 @@ Encode the field value as text. Mask value is 1 (train) or 0 (mask) per the sect
 
 ### Text config detection
 
-When no section uses `template` and all sections have `action: "train"`, the builder skips mask generation entirely — all tokens are trained.
+When no section uses `template` and all sections have `action: "train"`, the builder omits `loss_mask` from the output — all tokens are trained.
 
 ---
 
@@ -298,10 +299,12 @@ When no section uses `template` and all sections have `action: "train"`, the bui
 ```
 output/
   __default__/
+    shard_0000/
       meta.json
       sequence.bin
       loss_mask.bin
   wiki/
+    shard_0000/
       meta.json
       sequence.bin
       loss_mask.bin
@@ -324,7 +327,7 @@ output/
       loss_mask.bin
 ```
 
-`MmapStore` discovers all shards under the domain directory via `rglob("meta.json")`.
+For `bin` format, `MmapStore` discovers all shards under the domain directory via `rglob("meta.json")`. For `h5` format, `H5Store` discovers `.h5`/`.hdf5` files via recursive glob.
 
 ---
 
@@ -349,7 +352,7 @@ python scripts/tools/preprocess.py data/grpo/*.jsonl -o output/grpo/ -c configs/
 from astrai.preprocessing.pipeline import Pipeline
 from astrai.config.preprocess_config import PipelineConfig
 
-config = PipelineConfig.from_json("sft.json")
+config = PipelineConfig.from_file("sft.json")
 Pipeline(
     config,
     ["data_part1.jsonl", "data_part2.jsonl"],

assets/docs/training.md

@@ -58,7 +58,9 @@ on_train_begin
         context.loss = loss.item()
         stand_loss = loss / executor.grad_accum_steps
         executor.backward(stand_loss)
-        context.iteration += 1
+        context.consumed_samples += (
+            context.config.batch_per_device * context.world_size
+        )
         on_batch_end
 
         if executor.sync_gradients:
@@ -78,13 +80,13 @@ on_train_end
 | `on_train_begin` | Before training starts | `GradientCheckpointingCallback` |
 | `on_epoch_begin` | Start of each epoch | `ProgressBarCallback` |
 | `on_batch_begin` | Every batch | — |
-| `on_optimizer_step` | Every accumulation window | `GradientClippingCallback`, `ValidationCallback` |
+| `on_optimizer_step` | Every accumulation window | `GradientClippingCallback`, `MetricLoggerCallback`, `ValidationCallback` |
 | `on_batch_end` | Every batch | `CheckpointCallback`, `MetricLoggerCallback`, `ProgressBarCallback` |
 | `on_epoch_end` | End of each epoch | `ProgressBarCallback` |
 | `on_error` | On exception during training | `CheckpointCallback`, `MetricLoggerCallback` |
 | `on_train_end` | Training ends (always via finally) | `CheckpointCallback`, `MetricLoggerCallback`, `GradientCheckpointingCallback` |
 
-Default callbacks (in order): `gradient_checkpointing` (activation checkpointing, optional), `checkpoint` (safetensors, rank-0), `metric_logger` (JSONL, rank-0), `progress_bar` (tqdm), `gradient_clipping`, `validation` (periodic validation on val_dataset).
+Default callbacks (in order): `gradient_checkpointing` (activation checkpointing, optional), `checkpoint` (safetensors, rank-0), `validation` (periodic validation on val_dataset), `metric_logger` (JSONL, rank-0), `progress_bar` (tqdm), `gradient_clipping`.
 
 ## Strategies
 
@@ -158,8 +160,8 @@ Callback wraps each `DecoderBlock.forward` with `torch.utils.checkpoint.checkpoi
 ## Checkpoint
 
 ```
-Checkpoint(state_dict, epoch, iteration, extra, meta, config)
+Checkpoint(state_dict, epoch, consumed_samples, extra, meta, config)
-  ├── save(save_dir)    rank-0 only: meta.json (epoch/iteration/timestamp) + config.json (model config) + model.safetensors + optional {key}.pt (optimizer.pt, scheduler.pt)
+  ├── save(save_dir)    rank-0 only: meta.json (epoch/consumed_samples/timestamp) + config.json (model config) + model.safetensors + optional {key}.pt (optimizer.pt, scheduler.pt)
   └── load(save_dir, broadcast=False)    loads from local disk; set broadcast=True to broadcast metadata from rank-0
 ```
 

astrai/config/train_config.py

@@ -47,14 +47,18 @@ class TrainConfig(BaseConfig):
 
     # checkpoint setting
     start_epoch: int = field(default=0, metadata={"help": "Start epoch for training."})
-    start_batch: int = field(
+    start_samples: int = field(
-        default=0, metadata={"help": "Start batch iteration for training."}
+        default=0,
+        metadata={
+            "help": "Start samples count (per rank). Superseded by checkpoint consumed_samples."
+        },
     )
     ckpt_dir: str = field(
         default="./checkpoint", metadata={"help": "Checkpoint directory."}
     )
     ckpt_interval: int = field(
-        default=5000, metadata={"help": "Number of iterations between checkpoints."}
+        default=5000,
+        metadata={"help": "Number of optimizer steps between checkpoints."},
     )
 
     # lora setting
@@ -67,12 +71,8 @@ class TrainConfig(BaseConfig):
     log_dir: str = field(
         default="./checkpoint/logs", metadata={"help": "Directory for metric logs."}
     )
-    log_interval: int = field(
-        default=100,
-        metadata={"help": "Number of batch iterations between metric logs."},
-    )
     metrics: List[str] = field(
-        default_factory=lambda: ["loss", "lr"],
+        default_factory=lambda: ["loss", "lr", "grad_norm"],
         metadata={"help": "Metrics to record during training."},
     )
 

astrai/dataset/__init__.py

@@ -5,10 +5,13 @@ from astrai.dataset.dataset import (
 from astrai.dataset.sampler import ResumableDistributedSampler
 from astrai.dataset.storage import (
     H5Store,
+    JsonlStore,
     MmapStore,
     Store,
     StoreFactory,
     detect_format,
+)
+from astrai.serialization import (
     load_bin,
     load_h5,
     save_bin,
@@ -22,6 +25,7 @@ __all__ = [
     "StoreFactory",
     "H5Store",
     "MmapStore",
+    "JsonlStore",
     "detect_format",
     "save_h5",
     "load_h5",

astrai/dataset/dataset.py

@@ -48,24 +48,26 @@ class BaseDataset(Dataset, ABC):
                 f"Missing: {missing}"
             )
 
-    def load(self, load_path: str, storage_type: Optional[str] = None):
+    def load(self, load_path: str, storage_type: Optional[str] = None, **kwargs):
         """Load dataset from the given path.
 
         Auto-detects the storage format if not specified.
 
         Args:
             load_path: Path to the data directory or file
-            storage_type: Force a specific storage type ("h5", "bin"),
+            storage_type: Force a specific storage type ("h5", "bin", "jsonl"),
                           or None for auto-detection
+            **kwargs: Extra arguments forwarded to the store constructor and
+                      to ``store.load()``.
 
         Raises:
             KeyError: If the loaded storage is missing required keys.
         """
         if storage_type is None:
             storage_type = detect_format(load_path)
-        self.storage = StoreFactory.create(storage_type)
+        self.storage = StoreFactory.create(storage_type, **kwargs)
         self._load_path = load_path
-        self.storage.load(load_path)
+        self.storage.load(load_path, **kwargs)
         self._validate_keys()
 
     @property
@@ -144,6 +146,7 @@ class DatasetFactory(BaseFactory["BaseDataset"]):
         window_size: int,
         stride: Optional[int] = None,
         storage_type: Optional[str] = None,
+        **kwargs,
     ) -> "BaseDataset":
         """Create and load a dataset in one step.
 
@@ -152,7 +155,8 @@ class DatasetFactory(BaseFactory["BaseDataset"]):
             load_path: Path to the data file
             window_size: Window size for data sampling
             stride: Stride between consecutive samples (default: same as window_size)
-            storage_type: Storage type ("h5", "bin") or None for auto-detection
+            storage_type: Storage type ("h5", "bin", "jsonl") or None for auto-detection
+            **kwargs: Extra arguments forwarded to ``dataset.load()``.
 
         Returns:
             Loaded dataset instance
@@ -161,7 +165,7 @@ class DatasetFactory(BaseFactory["BaseDataset"]):
             stride = window_size
 
         dataset = cls.create(train_type, window_size, stride)
-        dataset.load(load_path, storage_type=storage_type)
+        dataset.load(load_path, storage_type=storage_type, **kwargs)
 
         return dataset
 

astrai/dataset/sampler.py

@@ -74,6 +74,7 @@ class ResumableDistributedSampler(Sampler[int]):
 
         self.epoch += 1
         self._indices = None
+        self.iter = self.iter % self.num_samples_per_replica
 
     @property
     def _remaining(self):

astrai/dataset/storage.py

@@ -20,79 +20,25 @@ Key properties:
 import bisect
 import glob
 import json
-import os
+import logging
 from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Dict, List, Union
 
-import h5py
-import numpy as np
 import torch
 from torch import Tensor
 
+from astrai.config.preprocess_config import PipelineConfig
 from astrai.factory import BaseFactory
+from astrai.preprocessing.builder import MaskBuilderFactory
+from astrai.preprocessing.position_id import PositionIdStrategyFactory
+from astrai.serialization import (
+    load_bin,
+    load_h5,
+)
+from astrai.tokenize import AutoTokenizer
 
-
+logger = logging.getLogger(__name__)
-def save_h5(file_path: str, file_name: str, tensor_group: Dict[str, List[Tensor]]):
-    os.makedirs(file_path, exist_ok=True)
-    full_file_path = os.path.join(file_path, f"{file_name}.h5")
-    with h5py.File(full_file_path, "w") as f:
-        for key, tensors in tensor_group.items():
-            grp = f.create_group(key)
-            for idx, tensor in enumerate(tensors):
-                arr = tensor.cpu().numpy()
-                grp.create_dataset(f"data_{idx}", data=arr)
-
-
-def load_h5(file_path: str, share_memory=True) -> Dict[str, List[Tensor]]:
-    tensor_group: Dict[str, List[Tensor]] = {}
-
-    root_path = Path(file_path)
-    h5_files = list(root_path.rglob("*.h5")) + list(root_path.rglob("*.hdf5"))
-
-    for h5_file in h5_files:
-        with h5py.File(h5_file, "r") as f:
-            for key in f.keys():
-                grp = f[key]
-                dsets = []
-                for dset_name in grp.keys():
-                    dset = grp[dset_name]
-                    tensor = torch.from_numpy(dset[:])
-                    if share_memory:
-                        tensor = tensor.share_memory_()
-                    dsets.append(tensor)
-
-                if tensor_group.get(key) is None:
-                    tensor_group[key] = []
-                tensor_group[key].extend(dsets)
-
-    return tensor_group
-
-
-def save_bin(file_path: str, tensor_group: Dict[str, List[Tensor]]):
-    os.makedirs(file_path, exist_ok=True)
-    meta = {}
-    for key, tensors in tensor_group.items():
-        cat = torch.cat(tensors, dim=0)
-        meta[key] = {"shape": list(cat.shape), "dtype": str(cat.dtype).split(".")[-1]}
-        np.asarray(cat.cpu().numpy()).tofile(os.path.join(file_path, f"{key}.bin"))
-    with open(os.path.join(file_path, "meta.json"), "w") as f:
-        json.dump(meta, f)
-
-
-def load_bin(file_path: str) -> Dict[str, List[Tensor]]:
-    with open(os.path.join(file_path, "meta.json"), "r") as f:
-        meta = json.load(f)
-    segments: Dict[str, List[Tensor]] = {}
-    for key, info in meta.items():
-        arr = np.memmap(
-            os.path.join(file_path, f"{key}.bin"),
-            dtype=info["dtype"],
-            mode="r+",
-            shape=tuple(info["shape"]),
-        )
-        segments[key] = [torch.from_numpy(arr)]
-    return segments
 
 
 def detect_format(load_path: str) -> str:
@@ -102,7 +48,7 @@ def detect_format(load_path: str) -> str:
         load_path: Directory or file path
 
     Returns:
-        Format string ("h5" or "bin")
+        Format string ("h5", "bin", or "jsonl")
 
     Raises:
         FileNotFoundError: If no supported data files are found
@@ -112,6 +58,8 @@ def detect_format(load_path: str) -> str:
         suffix = root.suffix.lower()
         if suffix in (".h5", ".hdf5"):
             return "h5"
+        if suffix == ".jsonl":
+            return "jsonl"
         raise ValueError(f"Unsupported file format: {suffix}")
 
     h5_files = [
@@ -128,6 +76,11 @@ def detect_format(load_path: str) -> str:
         ) > 0
         if has_meta:
             return "bin"
+    jsonl_files = [
+        Path(p) for p in glob.glob(str(root / "**" / "*.jsonl"), recursive=True)
+    ]
+    if jsonl_files:
+        return "jsonl"
     raise FileNotFoundError(f"No supported data files found at {load_path}")
 
 
@@ -264,3 +217,96 @@ class MmapStore(Store):
         self._normalize(all_raw)
         for tensors in self._data.values():
             self._mmap_refs.extend(tensors)
+
+
+@StoreFactory.register("jsonl")
+class JsonlStore(Store):
+    """On-the-fly tokenization store for raw JSONL files.
+
+    A JSONL dataset directory contains ``*.jsonl`` files plus a
+    ``dataset_config.json`` file that follows the same schema as
+    :class:`PipelineConfig` with an additional ``tokenizer_path`` field.
+    Records are tokenized when the store is loaded and concatenated into
+    segmented tensors matching the key layout expected by the dataset
+    classes (``sequence``, ``loss_mask``, ``position_ids``, ...).
+    """
+
+    CONFIG_NAME = "dataset_config.json"
+
+    def load(self, path: str):
+        root = Path(path)
+        config_path = root / self.CONFIG_NAME
+        if not config_path.exists():
+            raise FileNotFoundError(
+                f"JSONL dataset config not found: {config_path}. "
+                f"Expected {self.CONFIG_NAME} alongside *.jsonl files."
+            )
+
+        with open(config_path, "r", encoding="utf-8") as f:
+            raw_config = json.load(f)
+
+        tokenizer_path = raw_config.pop("tokenizer_path", None)
+        if tokenizer_path is None:
+            raise ValueError(
+                f"JSONL dataset config must specify 'tokenizer_path': {config_path}"
+            )
+
+        self.config = PipelineConfig.from_dict(raw_config)
+        tokenizer = AutoTokenizer.from_pretrained(tokenizer_path)
+        mask_builder = MaskBuilderFactory.create("sectioned")
+        position_strategy = PositionIdStrategyFactory.create(
+            self.config.output.position_ids_mode
+        )
+
+        raw: Dict[str, List[Tensor]] = {}
+        doc_sequences: List[List[int]] = []
+
+        for jsonl_path in sorted(root.glob("*.jsonl")):
+            with open(jsonl_path, "r", encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        item = json.loads(line)
+                    except json.JSONDecodeError:
+                        logger.warning(
+                            "Failed to parse JSON line in %s, skipping", jsonl_path
+                        )
+                        continue
+
+                    result = mask_builder.build(item, self.config, tokenizer)
+                    if result is None:
+                        continue
+
+                    result.pop("domain", None)
+                    primary_ids = self._primary_ids(result)
+                    if not primary_ids:
+                        continue
+
+                    doc_sequences.append(primary_ids)
+                    for key, ids in result.items():
+                        if key not in raw:
+                            raw[key] = []
+                        raw[key].append(torch.tensor(ids, dtype=self._infer_dtype(ids)))
+
+        pos_ids = position_strategy.generate(doc_sequences)
+        if pos_ids:
+            raw["position_ids"] = [torch.tensor(pos_ids, dtype=torch.int32)]
+
+        self._normalize(raw)
+
+    @staticmethod
+    def _primary_ids(result: dict) -> List[int]:
+        """Return the first integer list in *result* as the primary id sequence."""
+        for val in result.values():
+            if isinstance(val, list) and val and isinstance(val[0], int):
+                return val
+        return []
+
+    @staticmethod
+    def _infer_dtype(ids: List) -> torch.dtype:
+        """Infer tensor dtype from the first element of a token/value list."""
+        if ids and isinstance(ids[0], float):
+            return torch.float32
+        return torch.int32

astrai/factory.py

@@ -37,7 +37,7 @@ def _resolve_type(
     ns = vars(mod)
 
     if isinstance(arg, ForwardRef):
-        return arg._evaluate(ns, None, frozenset(), recursive_guard=frozenset())
+        return arg._evaluate(ns, None, recursive_guard=frozenset())
 
     return ns.get(name)
 

astrai/parallel/executor.py

@@ -132,6 +132,12 @@ class BaseExecutor:
     def grad_accum_steps(self) -> int:
         return self.gradient_state.num_steps
 
+    def clip_grad_norm(self, model: nn.Module, max_norm: float) -> float:
+        total_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm)
+        if isinstance(total_norm, torch.Tensor):
+            return total_norm.item()
+        return total_norm
+
 
 class ExecutorFactory(BaseFactory[BaseExecutor]):
     pass
@@ -260,6 +266,14 @@ class FSDPExecutor(BaseExecutor):
             return model.no_sync()
         return contextlib.nullcontext()
 
+    def clip_grad_norm(self, model: nn.Module, max_norm: float) -> float:
+        if isinstance(model, FSDP) and self.use_distributed:
+            total_norm = model.clip_grad_norm_(max_norm)
+            if isinstance(total_norm, torch.Tensor):
+                return total_norm.item()
+            return total_norm
+        return super().clip_grad_norm(model, max_norm)
+
     def unwrap_model(self, model: nn.Module):
         if isinstance(model, FSDP) and self.use_distributed:
             with FSDP.state_dict_type(

astrai/preprocessing/writer.py

@@ -14,8 +14,8 @@ from typing import Dict, List
 
 import torch
 
-from astrai.dataset.storage import save_bin, save_h5
 from astrai.factory import BaseFactory
+from astrai.serialization import save_bin, save_h5
 
 logger = logging.getLogger(__name__)
 

astrai/serialization/__init__.py

@@ -0,0 +1,43 @@
+"""Serialization utilities for models and datasets.
+
+This package re-exports checkpoint helpers and dataset storage helpers so
+that existing imports from ``astrai.serialization`` continue to work.
+"""
+
+from astrai.serialization.checkpoint import (
+    Checkpoint,
+    load_json,
+    load_model_config,
+    load_model_weights,
+    load_safetensors,
+    load_state_dict,
+    load_torch,
+    save_json,
+    save_model,
+    save_safetensors,
+    save_torch,
+)
+from astrai.serialization.dataset import (
+    load_bin,
+    load_h5,
+    save_bin,
+    save_h5,
+)
+
+__all__ = [
+    "Checkpoint",
+    "load_json",
+    "load_model_config",
+    "load_model_weights",
+    "load_safetensors",
+    "load_state_dict",
+    "load_torch",
+    "save_json",
+    "save_model",
+    "save_safetensors",
+    "save_torch",
+    "load_bin",
+    "load_h5",
+    "save_bin",
+    "save_h5",
+]

astrai/serialization/checkpoint.py

@@ -1,5 +1,8 @@
+"""Model checkpoint serialization helpers."""
+
 import io
 import json
+import os
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -136,7 +139,7 @@ def load_state_dict(path: Union[str, Path], broadcast: bool = False) -> dict:
 class Checkpoint:
     state_dict: Dict[str, Any] = field(default_factory=dict)
     epoch: int = 0
-    iteration: int = 0
+    consumed_samples: int = 0
     extra: Dict[str, Any] = field(default_factory=dict)
     meta: Dict[str, Any] = field(default_factory=dict)
     config: Dict[str, Any] = field(default_factory=dict)
@@ -150,7 +153,7 @@ class Checkpoint:
 
         meta = {
             "epoch": self.epoch,
-            "iteration": self.iteration,
+            "consumed_samples": self.consumed_samples,
             "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
             **self.meta,
         }
@@ -176,7 +179,7 @@ class Checkpoint:
         return cls(
             state_dict=state_dict,
             epoch=meta.get("epoch", 0),
-            iteration=meta.get("iteration", 0),
+            consumed_samples=meta.get("consumed_samples", 0),
             extra=extra,
             config=config,
         )

astrai/serialization/dataset.py

@@ -0,0 +1,73 @@
+"""Dataset storage serialization helpers (HDF5 / memory-mapped binary)."""
+
+import json
+import os
+from pathlib import Path
+from typing import Dict, List
+
+import h5py
+import numpy as np
+import torch
+from torch import Tensor
+
+
+def save_h5(file_path: str, file_name: str, tensor_group: Dict[str, List[Tensor]]):
+    os.makedirs(file_path, exist_ok=True)
+    full_file_path = os.path.join(file_path, f"{file_name}.h5")
+    with h5py.File(full_file_path, "w") as f:
+        for key, tensors in tensor_group.items():
+            grp = f.create_group(key)
+            for idx, tensor in enumerate(tensors):
+                arr = tensor.cpu().numpy()
+                grp.create_dataset(f"data_{idx}", data=arr)
+
+
+def load_h5(file_path: str, share_memory=True) -> Dict[str, List[Tensor]]:
+    tensor_group: Dict[str, List[Tensor]] = {}
+
+    root_path = Path(file_path)
+    h5_files = list(root_path.rglob("*.h5")) + list(root_path.rglob("*.hdf5"))
+
+    for h5_file in h5_files:
+        with h5py.File(h5_file, "r") as f:
+            for key in f.keys():
+                grp = f[key]
+                dsets = []
+                for dset_name in grp.keys():
+                    dset = grp[dset_name]
+                    tensor = torch.from_numpy(dset[:])
+                    if share_memory:
+                        tensor = tensor.share_memory_()
+                    dsets.append(tensor)
+
+                if tensor_group.get(key) is None:
+                    tensor_group[key] = []
+                tensor_group[key].extend(dsets)
+
+    return tensor_group
+
+
+def save_bin(file_path: str, tensor_group: Dict[str, List[Tensor]]):
+    os.makedirs(file_path, exist_ok=True)
+    meta = {}
+    for key, tensors in tensor_group.items():
+        cat = torch.cat(tensors, dim=0)
+        meta[key] = {"shape": list(cat.shape), "dtype": str(cat.dtype).split(".")[-1]}
+        np.asarray(cat.cpu().numpy()).tofile(os.path.join(file_path, f"{key}.bin"))
+    with open(os.path.join(file_path, "meta.json"), "w") as f:
+        json.dump(meta, f)
+
+
+def load_bin(file_path: str) -> Dict[str, List[Tensor]]:
+    with open(os.path.join(file_path, "meta.json"), "r") as f:
+        meta = json.load(f)
+    segments: Dict[str, List[Tensor]] = {}
+    for key, info in meta.items():
+        arr = np.memmap(
+            os.path.join(file_path, f"{key}.bin"),
+            dtype=info["dtype"],
+            mode="r+",
+            shape=tuple(info["shape"]),
+        )
+        segments[key] = [torch.from_numpy(arr)]
+    return segments

astrai/trainer/metric_util.py

@@ -1,42 +1,25 @@
-from typing import Any, Callable, Dict
+from typing import Dict
 
 import torch
 import torch.nn as nn
 
 
-def _grad_stat(
+def grad_norm(model: nn.Module, per_param: bool = False) -> float | Dict[str, float]:
-    model: nn.Module, fn: Callable[[torch.Tensor], Any], default: Any
+    grads = [p.grad.detach() for p in model.parameters() if p.grad is not None]
-) -> dict:
+    if not grads:
-    results = {}
+        return 0.0
+
+    total_sq = torch.stack([g.pow(2).sum() for g in grads]).sum()
+    if per_param:
+        norms = {}
         for name, param in model.named_parameters():
-        results[name] = default
             if param.grad is not None:
-            results[name] = fn(param.grad.data)
+                norms[name] = param.grad.norm(2).item()
-    return results
+            else:
-
+                norms[name] = 0.0
-
+        norms["total"] = total_sq.sqrt().item()
-def grad_norm(model: nn.Module, norm_type: int = 2) -> Dict[str, float]:
+        return norms
-    return _grad_stat(model, lambda g: g.norm(norm_type).item(), 0.0)
+    return total_sq.sqrt().item()
-
-
-def grad_std(model: nn.Module) -> Dict[str, float]:
-    return _grad_stat(model, lambda g: g.std().item(), 0.0)
-
-
-def grad_max(model: nn.Module) -> Dict[str, float]:
-    return _grad_stat(model, lambda g: g.max().item(), -float("inf"))
-
-
-def grad_min(model: nn.Module) -> Dict[str, float]:
-    return _grad_stat(model, lambda g: g.min().item(), float("inf"))
-
-
-def grad_mean(model: nn.Module) -> Dict[str, float]:
-    return _grad_stat(model, lambda g: g.mean().item(), 0.0)
-
-
-def grad_nan_num(model: nn.Module) -> Dict[str, int]:
-    return _grad_stat(model, lambda g: g.isnan().sum().item(), 0)
 
 
 def ctx_get_loss(ctx):
@@ -52,24 +35,4 @@ def ctx_get_val_loss(ctx):
 
 
 def ctx_get_grad_norm(ctx):
-    return grad_norm(ctx.model)
+    return ctx.grad_norm
-
-
-def ctx_get_grad_std(ctx):
-    return grad_std(ctx.model)
-
-
-def ctx_get_grad_max(ctx):
-    return grad_max(ctx.model)
-
-
-def ctx_get_grad_min(ctx):
-    return grad_min(ctx.model)
-
-
-def ctx_get_grad_mean(ctx):
-    return grad_mean(ctx.model)
-
-
-def ctx_get_grad_nan_num(ctx):
-    return grad_nan_num(ctx.model)

astrai/trainer/schedule.py

@@ -53,7 +53,7 @@ class CosineScheduler(BaseScheduler):
         optimizer,
         warmup_steps: int,
         lr_decay_steps: int,
-        min_rate: float = 0.05,
+        min_rate: float = 0.01,
         last_epoch: int = -1,
     ):
         self.warmup_steps = warmup_steps
@@ -65,11 +65,15 @@ class CosineScheduler(BaseScheduler):
     def get_lr(self) -> List[float]:
         # warmup
         if self.last_epoch < self.warmup_steps:
-            warmup_factor = max(self.min_rate, self.last_epoch / self.warmup_steps)
+            warmup_factor = max(
+                self.min_rate, self.last_epoch / max(self.warmup_steps, 1)
+            )
             return [base_lr * warmup_factor for base_lr in self.base_lrs]
 
         # cosine decay
-        decay_progress = (self.last_epoch - self.warmup_steps) / self.lr_decay_steps
+        decay_progress = (self.last_epoch - self.warmup_steps) / max(
+            self.lr_decay_steps, 1
+        )
         decay_progress = min(decay_progress, 1.0)
         cosine_decay = 0.5 * (1.0 + math.cos(math.pi * decay_progress))
         decay_factor = max(self.min_rate, cosine_decay)
@@ -104,7 +108,7 @@ class SGDRScheduler(BaseScheduler):
         optimizer,
         warmup_steps: int,
         cycle_length: int,
-        min_rate: float = 0.05,
+        min_rate: float = 0.01,
         t_mult: int = 2,
         last_epoch: int = -1,
     ):
@@ -118,7 +122,9 @@ class SGDRScheduler(BaseScheduler):
     def get_lr(self):
         # warmup
         if self.last_epoch < self.warmup_steps:
-            warmup_factor = max(self.min_rate, self.last_epoch / self.warmup_steps)
+            warmup_factor = max(
+                self.min_rate, self.last_epoch / max(self.warmup_steps, 1)
+            )
             return [base_lr * warmup_factor for base_lr in self.base_lrs]
 
         # SGDR
@@ -182,7 +188,7 @@ class WSDScheduler(BaseScheduler):
         warmup_steps: int,
         stable_steps: int,
         decay_steps: int,
-        min_rate: float = 0.0,
+        min_rate: float = 0.01,
         last_epoch: int = -1,
     ):
         self.warmup_steps = warmup_steps
@@ -194,7 +200,7 @@ class WSDScheduler(BaseScheduler):
 
     def get_lr(self) -> List[float]:
         if self.last_epoch < self.warmup_steps:
-            factor = self.last_epoch / max(self.warmup_steps, 1)
+            factor = max(self.min_rate, self.last_epoch / max(self.warmup_steps, 1))
             return [base_lr * factor for base_lr in self.base_lrs]
 
         offset = self.last_epoch - self.warmup_steps

astrai/trainer/strategy.py

@@ -196,7 +196,7 @@ class SFTStrategy(BaseStrategy):
 
         ignore_index = -100
         input_mask = make_doc_boundary_mask(position_ids)
-        target_ids = target_ids.masked_fill(loss_mask == 0, ignore_index)
+        target_ids = target_ids.masked_fill(~loss_mask, ignore_index)
         logits = self.model(
             input_ids=input_ids, position_ids=position_ids, input_mask=input_mask
         )["logits"]

astrai/trainer/train_callback.py

@@ -9,7 +9,6 @@ from typing import IO, Callable, List, Optional, Protocol, runtime_checkable
 import torch
 import torch.distributed as dist
 import torch.nn as nn
-from torch.nn.utils import clip_grad_norm_
 from torch.utils.checkpoint import checkpoint as torch_checkpoint
 from tqdm import tqdm
 
@@ -18,12 +17,7 @@ from astrai.parallel import only_on_rank
 from astrai.parallel.setup import get_current_device, get_rank
 from astrai.serialization import Checkpoint
 from astrai.trainer.metric_util import (
-    ctx_get_grad_max,
-    ctx_get_grad_mean,
-    ctx_get_grad_min,
-    ctx_get_grad_nan_num,
     ctx_get_grad_norm,
-    ctx_get_grad_std,
     ctx_get_loss,
     ctx_get_lr,
     ctx_get_val_loss,
@@ -86,7 +80,9 @@ class GradientClippingCallback(TrainCallback):
         self.max_grad_norm = max_grad_norm
 
     def on_optimizer_step(self, context: TrainContext):
-        clip_grad_norm_(context.model.parameters(), self.max_grad_norm)
+        context.grad_norm = context.executor.clip_grad_norm(
+            context.model, self.max_grad_norm
+        )
 
 
 @CallbackFactory.register("gradient_checkpointing")
@@ -143,34 +139,35 @@ class CheckpointCallback(TrainCallback):
         self.interval = interval
         self.weight_only = weight_only
         self.save_extra_fn = save_extra_fn or CheckpointCallback.save_extra
-        self.last_ckpt_iter = 0
+        self.last_ckpt_step = 0
 
     def _save_checkpoint(self, context: TrainContext):
         state_dict = context.executor.unwrap_model(context.model)
-        self.last_ckpt_iter = context.iteration
+        self.last_ckpt_step = context.optimizer_step
 
         if get_rank() == 0:
             save_path = os.path.join(
-                self.save_dir, f"epoch_{context.epoch}_iter_{context.iteration}"
+                self.save_dir,
+                f"epoch_{context.epoch}_step_{context.optimizer_step}",
             )
             extra = self.save_extra_fn(context)
             meta = context.config.to_dict()
             context.checkpoint = Checkpoint(
                 state_dict=state_dict,
                 epoch=context.epoch,
-                iteration=context.iteration,
+                consumed_samples=context.consumed_samples,
+                config=context.model_config,
                 extra=extra,
                 meta=meta,
-                config=context.model_config,
             )
             context.checkpoint.save(save_path)
 
     def on_batch_end(self, context: TrainContext):
-        if context.iteration - self.last_ckpt_iter >= self.interval:
+        if context.optimizer_step - self.last_ckpt_step >= self.interval:
             self._save_checkpoint(context)
 
     def on_train_end(self, context: TrainContext):
-        if context.iteration != self.last_ckpt_iter:
+        if context.optimizer_step != self.last_ckpt_step:
             self._save_checkpoint(context)
 
     def on_error(self, context: TrainContext):
@@ -202,23 +199,27 @@ class ProgressBarCallback(TrainCallback):
 
     @only_on_rank(0)
     def on_epoch_begin(self, context: TrainContext):
+        total_steps = len(context.dataloader) // context.executor.grad_accum_steps
         self.progress_bar = tqdm(
-            context.dataloader,
+            total=total_steps,
             desc=f"Epoch {context.epoch + 1}/{self.num_epoch}",
             dynamic_ncols=True,
             file=self.file or sys.stdout,
         )
 
     @only_on_rank(0)
-    def on_batch_end(self, context: TrainContext):
+    def on_optimizer_step(self, context: TrainContext):
+        self.progress_bar.update(1)
         postfix = {
+            "step": context.optimizer_step,
             "loss": f"{context.loss:.4f}",
             "lr": f"{context.optimizer.param_groups[-1]['lr']:.2e}",
         }
+        if context.grad_norm is not None:
+            postfix["grad_norm"] = f"{context.grad_norm:.2f}"
         if context.val_loss is not None:
             postfix["val_loss"] = f"{context.val_loss:.4f}"
         self.progress_bar.set_postfix(postfix)
-        self.progress_bar.update(1)
 
     @only_on_rank(0)
     def on_epoch_end(self, context: TrainContext):
@@ -227,20 +228,20 @@ class ProgressBarCallback(TrainCallback):
             self.progress_bar.close()
 
 
-@CallbackFactory.register("metric_logger")
+@CallbackFactory.register("metric")
-class MetricLoggerCallback(TrainCallback):
+class MetricCallback(TrainCallback):
     def __init__(
         self,
         log_dir: str,
         save_interval: int,
-        log_interval: int = 10,
         metrics: List[str] = None,
+        val_step: int = 0,
     ):
-        self.last_log_iter = 0
+        self.last_log_flush_step = 0
-        self._last_val_loss = None
         self.save_interval = save_interval
-        self.log_interval = log_interval
         self.metrics = metrics or ["loss", "lr"]
+        self.val_step = val_step
+        self._next_val_step = 0
 
         self.log_dir = Path(log_dir) if log_dir else Path.cwd() / "logs"
         self.log_dir.mkdir(parents=True, exist_ok=True)
@@ -252,11 +253,6 @@ class MetricLoggerCallback(TrainCallback):
             "lr": ctx_get_lr,
             "val_loss": ctx_get_val_loss,
             "grad_norm": ctx_get_grad_norm,
-            "grad_std": ctx_get_grad_std,
-            "grad_max": ctx_get_grad_max,
-            "grad_min": ctx_get_grad_min,
-            "grad_mean": ctx_get_grad_mean,
-            "grad_nan_num": ctx_get_grad_nan_num,
         }
 
     def _metrics(self, context: TrainContext, names):
@@ -272,46 +268,13 @@ class MetricLoggerCallback(TrainCallback):
             "type": event_type,
             "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
             "epoch": context.epoch,
-            "iter": context.iteration,
+            "step": context.optimizer_step,
+            "consumed_samples": context.consumed_samples,
             **extra,
         }
         self.log_cache.append(entry)
 
-    @only_on_rank(0)
+    def _run_validation(self, context: TrainContext) -> float:
-    def _flush(self, epoch, iter):
-        log_file = self.log_dir / f"epoch_{epoch}_iter_{iter}_metric.jsonl"
-        log_file.parent.mkdir(parents=True, exist_ok=True)
-        with open(log_file, "w") as f:
-            for log in self.log_cache:
-                f.write(json.dumps(log) + "\n")
-
-    def on_batch_end(self, context):
-        if context.iteration % self.log_interval == 0:
-            step_metrics = [m for m in self.metrics if m != "val_loss"]
-            self._append("step", context, **self._metrics(context, step_metrics))
-        if context.iteration - self.last_log_iter >= self.save_interval:
-            self._flush(context.epoch, context.iteration)
-            self.last_log_iter = context.iteration
-
-    def on_optimizer_step(self, context):
-        if context.val_loss is not None and context.val_loss != self._last_val_loss:
-            self._append("validation", context, val_loss=context.val_loss)
-            self._last_val_loss = context.val_loss
-
-    def on_epoch_end(self, context):
-        self._append("epoch", context)
-
-    def on_train_end(self, context):
-        if context.iteration != self.last_log_iter:
-            self._flush(context.epoch, context.iteration)
-
-    def on_error(self, context):
-        self._flush(context.epoch, context.iteration)
-
-
-@CallbackFactory.register("validation")
-class ValidationCallback(TrainCallback):
-    def _run_validation(self, context: TrainContext):
         context.model.eval()
 
         total_loss = 0.0
@@ -323,27 +286,49 @@ class ValidationCallback(TrainCallback):
                 total_loss += loss.item()
                 num_batches += 1
 
+        if context.world_size > 1 and dist.is_initialized():
+            stats = torch.tensor(
+                [total_loss, float(num_batches)], device=get_current_device()
+            )
+            dist.all_reduce(stats, op=dist.ReduceOp.SUM)
+            avg_loss = (stats[0] / stats[1]).item()
+        else:
             avg_loss = total_loss / max(num_batches, 1)
 
-        if context.world_size > 1 and dist.is_initialized():
-            loss_tensor = torch.tensor([avg_loss], device=get_current_device())
-            dist.all_reduce(loss_tensor, op=dist.ReduceOp.AVG)
-            avg_loss = loss_tensor.item()
-
-        context.val_loss = avg_loss
         context.model.train()
+        return avg_loss
 
-        step_count = context.iteration // context.config.grad_accum_steps
+    @only_on_rank(0)
-        logger.info(
+    def _flush(self, epoch, step):
-            f"Epoch {context.epoch + 1}, Step {step_count}, Val Loss: {avg_loss:.4f}"
+        log_file = self.log_dir / f"epoch_{epoch}_step_{step}_metric.jsonl"
-        )
+        log_file.parent.mkdir(parents=True, exist_ok=True)
+        with open(log_file, "w") as f:
+            for log in self.log_cache:
+                f.write(json.dumps(log) + "\n")
 
-    def on_optimizer_step(self, context: TrainContext):
+    def on_optimizer_step(self, context):
-        if context.val_dataloader is None:
+        if (
-            return
+            context.val_dataloader is not None
-        cfg = context.config
+            and self.val_step > 0
-        if cfg.val_step <= 0:
+            and context.optimizer_step >= self._next_val_step
-            return
+        ):
-        step_count = context.iteration // cfg.grad_accum_steps
+            context.val_loss = self._run_validation(context)
-        if step_count % cfg.val_step == 0:
+            self._next_val_step = context.optimizer_step + self.val_step
-            self._run_validation(context)
+            self._append("validation", context, val_loss=context.val_loss)
+
+        step_metrics = [m for m in self.metrics if m != "val_loss"]
+        self._append("step", context, **self._metrics(context, step_metrics))
+
+        if context.optimizer_step - self.last_log_flush_step >= self.save_interval:
+            self._flush(context.epoch, context.optimizer_step)
+            self.last_log_flush_step = context.optimizer_step
+
+    def on_epoch_end(self, context):
+        self._append("epoch", context)
+
+    def on_train_end(self, context):
+        if context.optimizer_step != self.last_log_flush_step:
+            self._flush(context.epoch, context.optimizer_step)
+
+    def on_error(self, context):
+        self._flush(context.epoch, context.optimizer_step)

astrai/trainer/train_context.py

@@ -29,8 +29,9 @@ class TrainContext:
     executor: BaseExecutor = field(default=None)
 
     epoch: int = field(default=0)
-    iteration: int = field(default=0)
+    consumed_samples: int = field(default=0)
     loss: float = field(default=0.0)
+    grad_norm: Optional[float] = field(default=None)
     val_dataloader: Optional[DataLoader] = field(default=None)
     val_loss: Optional[float] = field(default=None)
 
@@ -38,6 +39,14 @@ class TrainContext:
     rank: int = field(default=0)
     kwargs: Dict[str, Any] = field(default_factory=dict)
 
+    @property
+    def optimizer_step(self) -> int:
+        return self.consumed_samples // (
+            self.config.batch_per_device
+            * self.world_size
+            * self.config.grad_accum_steps
+        )
+
 
 class TrainContextBuilder:
     def __init__(
@@ -89,7 +98,10 @@ class TrainContextBuilder:
                 if checkpoint.config:
                     context.model_config = checkpoint.config
                 context.epoch = checkpoint.epoch or cfg.start_epoch
-                context.iteration = checkpoint.iteration or cfg.start_batch
+                if checkpoint.consumed_samples > 0:
+                    context.consumed_samples = checkpoint.consumed_samples
+                else:
+                    context.consumed_samples = cfg.start_samples * context.world_size
                 context.checkpoint = checkpoint
 
         if cfg.lora is not None:
@@ -115,7 +127,7 @@ class TrainContextBuilder:
                 cfg.dataset, [n_train, n_val], generator=generator
             )
 
-        sampler_offset = context.iteration * cfg.batch_per_device
+        sampler_offset = context.consumed_samples // context.world_size
         sampler = ResumableDistributedSampler(
             data_source=train_dataset,
             start_epoch=context.epoch,

astrai/trainer/trainer.py

@@ -34,13 +34,12 @@ class Trainer:
                 cfg.ckpt_dir,
                 cfg.ckpt_interval,
             ),
-            CallbackFactory.create("validation"),
             CallbackFactory.create(
-                "metric_logger",
+                "metric",
                 log_dir=cfg.log_dir,
                 save_interval=cfg.ckpt_interval,
-                log_interval=cfg.log_interval,
                 metrics=cfg.metrics,
+                val_step=cfg.val_step,
             ),
             CallbackFactory.create("progress_bar", cfg.n_epoch),
             CallbackFactory.create("gradient_clipping", cfg.max_grad_norm),
@@ -74,7 +73,9 @@ class Trainer:
                         context.loss = loss.item()
                         stand_loss = loss / executor.grad_accum_steps
                         executor.backward(stand_loss)
-                        context.iteration += 1
+                        context.consumed_samples += (
+                            context.config.batch_per_device * context.world_size
+                        )
                         self._call_callbacks("on_batch_end", context)
 
                         if executor.sync_gradients:

scripts/demo/stream_chat.py

@@ -1,3 +1,4 @@
+from argparse import ArgumentParser
 from pathlib import Path
 
 import torch
@@ -7,42 +8,82 @@ from astrai.model import AutoModel
 from astrai.tokenize import AutoTokenizer
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
-PARAMETER_ROOT = Path(PROJECT_ROOT, "params")
+
+
+def parse_args():
+    parser = ArgumentParser(description="Interactive streaming chat")
+    parser.add_argument(
+        "--model_path",
+        type=Path,
+        default=PROJECT_ROOT / "params",
+        help="Path to model weights (params/ or checkpoint/epoch_N_step_M/)",
+    )
+    parser.add_argument(
+        "--temperature",
+        type=float,
+        default=0.8,
+        help="Sampling temperature (default: 0.8)",
+    )
+    parser.add_argument(
+        "--top_p",
+        type=float,
+        default=0.95,
+        help="Top-p sampling threshold",
+    )
+    parser.add_argument(
+        "--top_k",
+        type=int,
+        default=50,
+        help="Top-k sampling threshold",
+    )
+    parser.add_argument(
+        "--max_tokens",
+        type=int,
+        default=2048,
+        help="Maximum tokens to generate",
+    )
+    parser.add_argument(
+        "--system_prompt",
+        type=str,
+        default="You are a helpful assistant.",
+        help="Optional system prompt",
+    )
+    return parser.parse_args()
 
 
 def chat():
-    model = AutoModel.from_pretrained(PARAMETER_ROOT)
+    args = parse_args()
-    tokenizer = AutoTokenizer.from_pretrained(PARAMETER_ROOT)
+    model_path = args.model_path
-    model.to(device="cuda", dtype=torch.bfloat16)
 
-    messages = [{"role": "system", "content": "You are a helpful assistant."}]
+    model = AutoModel.from_pretrained(model_path)
+    tokenizer = AutoTokenizer.from_pretrained(model_path)
+    model.to(device="cuda", dtype=torch.bfloat16)
     engine = InferenceEngine(model=model, tokenizer=tokenizer)
 
+    messages = [{"role": "system", "content": args.system_prompt}]
+
     while True:
         query = input(">> ")
         if query == "!exit":
             break
 
-        # Add user message
         messages.append({"role": "user", "content": query})
 
-        # Generate response
         full_response = ""
         prompt = tokenizer.apply_chat_template(messages, tokenize=False)
 
         for token in engine.generate(
             prompt=prompt,
             stream=True,
-            max_tokens=2048,
+            max_tokens=args.max_tokens,
-            temperature=0.8,
+            temperature=args.temperature,
-            top_p=0.95,
+            top_p=args.top_p,
-            top_k=50,
+            top_k=args.top_k,
         ):
             print(token, end="", flush=True)
             full_response += token
 
         print()
-        # Add assistant response to messages
         messages.append({"role": "assistant", "content": full_response.strip()})
 
 

scripts/eval/analyze_weights.py

@@ -0,0 +1,307 @@
+"""SVD effective rank & weight statistics analysis for model checkpoints."""
+
+import argparse
+import json
+from pathlib import Path
+
+import safetensors.torch
+import torch
+
+
+def effective_rank_metrics(w: torch.Tensor) -> dict:
+    if w.ndim == 1:
+        return {"shape": tuple(w.shape), "is_1d": True}
+
+    w = w.float()
+    s = torch.linalg.svdvals(w)
+    s_sq = s**2
+    total = s_sq.sum()
+    cumsum = torch.cumsum(s_sq, dim=0) / total
+
+    min_dim = min(w.shape[0], w.shape[1])
+    er_90 = (cumsum < 0.90).sum().item() + 1
+    er_95 = (cumsum < 0.95).sum().item() + 1
+    er_99 = (cumsum < 0.99).sum().item() + 1
+
+    p = s_sq / total
+    p = p[p > 1e-30]
+    entropy = -(p * torch.log(p)).sum()
+    entropic_rank = torch.exp(entropy).item()
+
+    return {
+        "shape": tuple(w.shape),
+        "min_dim": min_dim,
+        "er_90": er_90,
+        "er_95": er_95,
+        "er_99": er_99,
+        "er_99_norm": er_99 / min_dim,
+        "er_95_norm": er_95 / min_dim,
+        "entropic_rank": entropic_rank,
+        "entropic_rank_norm": entropic_rank / min_dim,
+        "top1_ratio": s[0].item() / s.sum().item(),
+        "top5_ratio": s[:5].sum().item() / s.sum().item(),
+        "decay_ratio": s[-1].item() / s[0].item(),
+        "condition_number": s[0].item() / s[-1].item(),
+        "mean": w.mean().item(),
+        "std": w.std().item(),
+        "min": w.min().item(),
+        "max": w.max().item(),
+    }
+
+
+def format_header(headers: list[str], widths: list[int]) -> str:
+    return "".join(h.ljust(w) for h, w in zip(headers, widths))
+
+
+def format_row(values: list[str], widths: list[int]) -> str:
+    return "".join(v.ljust(w) for v, w in zip(values, widths))
+
+
+def group_by_component(results: dict[str, dict]) -> dict[str, list[dict]]:
+    groups: dict[str, list[dict]] = {}
+    for key, r in results.items():
+        parts = key.split(".")
+        if parts[0] == "layers" and len(parts) >= 3:
+            sub = parts[2:]
+            if sub[0] == "attention":
+                comp = f"attn.{sub[1]}"
+            elif sub[0] == "mlp":
+                comp = f"mlp.{sub[1]}"
+            elif sub[0] == "input_norm":
+                comp = "input_norm"
+            elif sub[0] == "post_attention_norm":
+                comp = "post_attn_norm"
+            else:
+                comp = ".".join(sub)
+        else:
+            comp = key
+        groups.setdefault(comp, []).append(r)
+    return groups
+
+
+def print_component_summary(results: dict[str, dict], title: str):
+    groups = group_by_component(results)
+    matrix_groups = {
+        k: [v for v in vs if not v.get("is_1d")]
+        for k, vs in groups.items()
+        if any(not v.get("is_1d") for v in vs)
+    }
+
+    widths = [20, 12, 12, 12, 12, 12]
+    print(f"\n{title}")
+    print(
+        format_header(
+            ["Component", "N", "ER@99%", "EntRank%", "Top1 σ(%)", "Cond. Num"], widths
+        )
+    )
+    print("-" * sum(widths))
+
+    for name in sorted(matrix_groups.keys()):
+        items = matrix_groups[name]
+        n = len(items)
+        print(
+            format_row(
+                [
+                    name,
+                    str(n),
+                    f"{sum(r['er_99_norm'] for r in items) / n:.4f}",
+                    f"{sum(r['entropic_rank_norm'] for r in items) / n:.4f}",
+                    f"{sum(r['top1_ratio'] for r in items) / n:.4f}",
+                    f"{sum(r['condition_number'] for r in items) / n:.1f}",
+                ],
+                widths,
+            )
+        )
+
+    all_er = [
+        r["er_99_norm"]
+        for vs in matrix_groups.values()
+        for r in vs
+        if "_norm" not in r or not r.get("is_1d")
+    ]
+    if all_er:
+        m = sum(all_er) / len(all_er)
+        print(f"\n  Overall Mean ER@99: {m:.4f} ({m * 100:.1f}% of dimension)")
+        if m > 0.85:
+            print("  → HIGH utilization: model near capacity → need more params")
+        elif m > 0.5:
+            print("  → MODERATE utilization: some headroom left")
+        else:
+            print("  → LOW utilization: significant unused capacity")
+
+
+def print_layer_grid(results: dict[str, dict]):
+    comps = [
+        "attn.q_proj",
+        "attn.k_proj",
+        "attn.v_proj",
+        "attn.o_proj",
+        "mlp.up",
+        "mlp.gate",
+        "mlp.down",
+    ]
+    widths = [6] + [10] * len(comps)
+    metric = "er_99_norm"
+
+    print(f"\n--- Per-Layer Effective Rank (99% energy) ---")
+    print(format_header(["Layer"] + comps, widths))
+    print("-" * sum(widths))
+
+    layer_data: dict[int, dict[str, dict]] = {}
+    for key, r in results.items():
+        parts = key.split(".")
+        if parts[0] != "layers":
+            continue
+        li = int(parts[1])
+        sub = parts[2:]
+        if sub[0] == "attention":
+            cname = f"attn.{sub[1]}"
+        elif sub[0] == "mlp":
+            cname = f"mlp.{sub[1]}"
+        else:
+            continue
+        layer_data.setdefault(li, {})[cname] = r
+
+    for li in sorted(layer_data):
+        values = [str(li)]
+        for c in comps:
+            v = layer_data[li].get(c, {}).get(metric, 0)
+            values.append(f"{v:.4f}")
+        print(format_row(values, widths))
+
+
+def print_weight_stats(results: dict[str, dict]):
+    groups = group_by_component(results)
+    widths = [20, 12, 12, 12, 12]
+    print(f"\n--- Weight Value Statistics ---")
+    print(format_header(["Component", "Mean", "Std", "Min", "Max"], widths))
+    print("-" * sum(widths))
+
+    for name in sorted(groups.keys()):
+        items = groups[name]
+        means = [r.get("mean", 0) for r in items]
+        stds = [r.get("std", 0) for r in items]
+        mins = [r.get("min", 0) for r in items]
+        maxs = [r.get("max", 0) for r in items]
+        g_mean = sum(means) / len(means)
+        g_std = sum(stds) / len(stds)
+        g_min = min(mins)
+        g_max = max(maxs)
+        print(
+            format_row(
+                [
+                    name,
+                    f"{g_mean:.6f}",
+                    f"{g_std:.6f}",
+                    f"{g_min:.6f}",
+                    f"{g_max:.6f}",
+                ],
+                widths,
+            )
+        )
+
+
+def print_params_summary(results: dict[str, dict]):
+    total_2d = sum(
+        r["shape"][0] * r["shape"][1] for r in results.values() if not r.get("is_1d")
+    )
+    total_1d = sum(r["shape"][0] for r in results.values() if r.get("is_1d"))
+    print(f"\n  Total 2D params: {total_2d:,}")
+    print(f"  Total 1D params: {total_1d:,}")
+    print(f"  Total params:     {total_2d + total_1d:,}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="SVD effective rank & weight statistics of a model checkpoint."
+    )
+    parser.add_argument(
+        "--ckpt_dir",
+        type=str,
+        required=True,
+        help="Path to checkpoint directory (containing model.safetensors + config.json).",
+    )
+    parser.add_argument(
+        "--compare",
+        type=str,
+        nargs="*",
+        help="Additional checkpoint directories to compare against.",
+    )
+    parser.add_argument(
+        "--no_svd",
+        action="store_true",
+        help="Skip SVD analysis, only show weight statistics (mean/std/min/max).",
+    )
+    args = parser.parse_args()
+
+    def analyze_one(ckpt_dir: str, label: str):
+        ckpt_dir = Path(ckpt_dir)
+        weights_path = ckpt_dir / "model.safetensors"
+        if not weights_path.exists():
+            print(f"ERROR: {weights_path} not found")
+            return {}
+
+        meta = {}
+        meta_path = ckpt_dir / "meta.json"
+        if meta_path.exists():
+            with open(meta_path) as f:
+                meta = json.load(f)
+
+        print(f"\n{'=' * 70}")
+        print(f"  {label}: {ckpt_dir}")
+        if meta:
+            print(
+                f"  Iteration: {meta.get('iteration', '?')}, "
+                f"Strategy: {meta.get('strategy', '?')}, "
+                f"nprocs={meta.get('nprocs', '?')}"
+            )
+        print(f"{'=' * 70}")
+
+        print(f"Loading weights...")
+        sd = safetensors.torch.load_file(str(weights_path))
+        print(f"  {len(sd)} keys loaded")
+
+        weight_keys = [
+            k
+            for k in sd
+            if ".weight" in k and "rotary_embedding" not in k and "freqs_cis" not in k
+        ]
+
+        results = {}
+        if not args.no_svd:
+            print(f"Computing SVD on {len(weight_keys)} tensors...")
+            for i, k in enumerate(sorted(weight_keys)):
+                print(f"  [{i + 1}/{len(weight_keys)}] {k:<60s}", end="\r")
+                results[k] = effective_rank_metrics(sd[k])
+            print()
+        else:
+            print(f"Computing stats on {len(weight_keys)} tensors (no SVD)...")
+            for i, k in enumerate(sorted(weight_keys)):
+                t = sd[k]
+                results[k] = {
+                    "shape": tuple(t.shape),
+                    "is_1d": t.ndim == 1,
+                    "mean": t.float().mean().item(),
+                    "std": t.float().std().item(),
+                    "min": t.float().min().item(),
+                    "max": t.float().max().item(),
+                }
+
+        print_params_summary(results)
+        if not args.no_svd:
+            print_component_summary(
+                results, "\n=== SVD Effective Rank by Component ==="
+            )
+            print_layer_grid(results)
+        print_weight_stats(results)
+        return results
+
+    analyze_one(args.ckpt_dir, "Primary")
+
+    if args.compare:
+        for cdir in args.compare:
+            analyze_one(cdir, "Compare")
+
+
+if __name__ == "__main__":
+    main()

scripts/eval/evaluate_ifd.py

@@ -52,8 +52,11 @@ def compute_ifd(
 
 
 def _compute_ifd_raw(model, tokenizer, instruction, response, device, max_len) -> dict:
-    instr_ids = tokenizer.encode(instruction)
+    instr_ids = tokenizer.encode(instruction, add_special_tokens=False)
-    resp_ids = tokenizer.encode(response)
+    resp_ids = tokenizer.encode(response, add_special_tokens=False)
+
+    if len(resp_ids) > max_len:
+        resp_ids = resp_ids[:max_len]
 
     if not resp_ids:
         return {
@@ -66,28 +69,39 @@ def _compute_ifd_raw(model, tokenizer, instruction, response, device, max_len) -
     qa_len = len(instr_ids) + len(resp_ids)
     if qa_len > max_len:
         overflow = qa_len - max_len
+        if overflow >= len(instr_ids):
+            resp_ids = resp_ids[:max_len]
+            instr_ids = []
+        else:
             instr_ids = instr_ids[overflow:]
 
+    if not instr_ids:
+        return {
+            "L_cond": None,
+            "L_uncond": None,
+            "ifd": None,
+            "error": "response too long for context",
+        }
+
     instr_len = len(instr_ids)
     resp_len = len(resp_ids)
 
     qa_ids = instr_ids + resp_ids
-    qa_tensor = torch.tensor([qa_ids], device=device, dtype=torch.long)
 
     with torch.inference_mode():
-        logits_qa = model(qa_tensor)["logits"][0]
+        logits_qa = model(torch.tensor([qa_ids], device=device, dtype=torch.long))[
+            "logits"
+        ][0]
+        logits_resp = model(torch.tensor([resp_ids], device=device, dtype=torch.long))[
+            "logits"
+        ][0]
 
     resp_logits = logits_qa[instr_len - 1 : -1]
-    resp_targets = torch.tensor(resp_ids, device=device, dtype=torch.long)
+    resp_targets = logits_resp.new_tensor(resp_ids, dtype=torch.long)
     L_cond = F.cross_entropy(resp_logits, resp_targets, reduction="mean").item()
 
-    resp_tensor = torch.tensor([resp_ids], device=device, dtype=torch.long)
-
-    with torch.inference_mode():
-        logits_resp = model(resp_tensor)["logits"][0]
-
     unp_logits = logits_resp[:-1]
-    unp_targets = resp_tensor[0, 1:]
+    unp_targets = logits_resp.new_tensor(resp_ids[1:], dtype=torch.long)
     L_uncond = F.cross_entropy(unp_logits, unp_targets, reduction="mean").item()
 
     ifd = L_cond / L_uncond if L_uncond > 0 else None
@@ -185,7 +199,7 @@ def process_file(
     output_file: str,
     instr_key: str,
     resp_key: str,
-    max_len: int,
+    max_len: int = 2048,
     use_chat_template: bool = False,
 ):
     device = "cuda" if torch.cuda.is_available() else "cpu"

scripts/tools/train.py

@@ -150,8 +150,8 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument(
         "--metrics",
         nargs="*",
-        default=["loss", "lr"],
+        default=["loss", "lr", "grad_norm"],
-        help="Metrics to log (e.g. --metrics loss lr val_loss). Default: loss lr.",
+        help="Metrics to log (e.g. --metrics loss lr val_loss). Default: loss lr grad_norm.",
     )
     parser.add_argument(
         "--log_dir",
@@ -159,12 +159,6 @@ def parse_args() -> argparse.Namespace:
         default="checkpoint/logs",
         help="Directory for metric logs.",
     )
-    parser.add_argument(
-        "--log_interval",
-        type=int,
-        default=100,
-        help="Number of batch iterations between metric logs.",
-    )
     parser.add_argument(
         "--grpo_sync_interval",
         type=int,
@@ -175,7 +169,10 @@ def parse_args() -> argparse.Namespace:
         "--start_epoch", type=int, default=0, help="Start epoch for training."
     )
     parser.add_argument(
-        "--start_batch", type=int, default=0, help="Start batch for training."
+        "--start_samples",
+        type=int,
+        default=0,
+        help="Start samples (per rank) for training.",
     )
 
     parser.add_argument(
@@ -269,7 +266,20 @@ def create_model(config):
 
 
 def create_optimizer(model, **kwargs) -> optim.Optimizer:
-    return optim.AdamW(model.parameters(), fused=True, **kwargs)
+    decay_params = []
+    no_decay_params = []
+    for name, param in model.named_parameters():
+        if not param.requires_grad:
+            continue
+        if param.dim() < 2 or "norm" in name or "bias" in name:
+            no_decay_params.append(param)
+        else:
+            decay_params.append(param)
+    param_groups = [
+        {"params": decay_params, "weight_decay": kwargs.pop("weight_decay", 0.01)},
+        {"params": no_decay_params, "weight_decay": 0.0},
+    ]
+    return optim.AdamW(param_groups, fused=True, **kwargs)
 
 
 def create_scheduler(
@@ -304,7 +314,7 @@ def train(
     n_epoch: int,
     batch_per_device: int,
     start_epoch: int,
-    start_batch: int,
+    start_samples: int,
     grad_accum_steps: int,
     warmup_ratio: float,
     ckpt_interval: int,
@@ -313,7 +323,6 @@ def train(
     val_step: int,
     metrics: list[str],
     log_dir: str,
-    log_interval: int,
     dpo_beta: float,
     grpo_clip_eps: float,
     grpo_kl_coef: float,
@@ -431,7 +440,7 @@ def train(
         n_epoch=n_epoch,
         batch_per_device=batch_per_device,
         start_epoch=start_epoch,
-        start_batch=start_batch,
+        start_samples=start_samples,
         ckpt_interval=ckpt_interval,
         grad_accum_steps=grad_accum_steps,
         max_grad_norm=max_grad_norm,
@@ -449,7 +458,6 @@ def train(
         val_step=val_step,
         metrics=metrics,
         log_dir=log_dir,
-        log_interval=log_interval,
         gradient_checkpointing_modules=grad_ckpt_modules,
         executor_kwargs=executor_kwargs,
         extra_kwargs=strategy_kwargs,

tests/conftest.py

@@ -75,7 +75,7 @@ class MultiTurnDataset(Dataset):
 
 
 class EarlyStoppingDataset(Dataset):
-    """Dataset that triggers early stopping after a specified number of iterations."""
+    """Dataset that triggers early stopping after consuming a specified number of samples."""
 
     def __init__(self, length=10, stop_after=5):
         self.length = length

tests/data/test_checkpoint.py

@@ -25,7 +25,9 @@ def test_single_process():
 
         scheduler.step()
 
-    checkpoint = Checkpoint(state_dict=model.state_dict(), epoch=3, iteration=30)
+    checkpoint = Checkpoint(
+        state_dict=model.state_dict(), epoch=3, consumed_samples=120
+    )
 
     with tempfile.TemporaryDirectory() as tmpdir:
         checkpoint.save(tmpdir)
@@ -33,7 +35,7 @@ def test_single_process():
         loaded_checkpoint = Checkpoint.load(tmpdir)
 
         assert loaded_checkpoint.epoch == 3
-        assert loaded_checkpoint.iteration == 30
+        assert loaded_checkpoint.consumed_samples == 120
 
 
 def test_checkpoint_with_extra():
@@ -46,7 +48,10 @@ def test_checkpoint_with_extra():
         "scheduler": {"last_epoch": 5},
     }
     checkpoint = Checkpoint(
-        state_dict=model.state_dict(), epoch=1, iteration=10, extra=extra
+        state_dict=model.state_dict(),
+        epoch=1,
+        consumed_samples=40,
+        extra=extra,
     )
 
     with tempfile.TemporaryDirectory() as tmpdir:
@@ -77,7 +82,7 @@ def simple_training():
     checkpoint = Checkpoint(
         state_dict=model.state_dict(),
         epoch=2,
-        iteration=10,
+        consumed_samples=40,
     )
 
     rank = get_rank()

tests/data/test_dataset.py

@@ -1,14 +1,18 @@
+import json
 import os
 
 import numpy as np
 import pytest
 import torch
 
+from astrai.config.preprocess_config import PipelineConfig
 from astrai.dataset.dataset import DatasetFactory, SEQDataset
 from astrai.dataset.storage import (
     H5Store,
     StoreFactory,
     detect_format,
+)
+from astrai.serialization import (
     load_bin,
     save_bin,
     save_h5,
@@ -19,6 +23,39 @@ def _rand_seq(length, vocab=1000):
     return torch.randint(0, vocab, (length,), dtype=torch.int64)
 
 
+def _save_test_tokenizer(test_dir, tokenizer):
+    tokenizer_path = os.path.join(test_dir, "tokenizer")
+    os.makedirs(tokenizer_path, exist_ok=True)
+    tokenizer.save_pretrained(tokenizer_path)
+    return tokenizer_path
+
+
+def _write_jsonl_dataset(test_dir, tokenizer_path, records, config_overrides=None):
+    data_dir = os.path.join(test_dir, "jsonl_data")
+    os.makedirs(data_dir, exist_ok=True)
+
+    with open(os.path.join(data_dir, "data.jsonl"), "w", encoding="utf-8") as f:
+        for record in records:
+            f.write(json.dumps(record, ensure_ascii=False) + "\n")
+
+    config = {
+        "tokenizer_path": tokenizer_path,
+        "version": 1,
+        "input": {"sections": [{"field": "text", "action": "train"}]},
+        "preprocessing": {"max_seq_len": 128},
+        "output": {"position_ids_mode": "continuous"},
+    }
+    if config_overrides:
+        config.update(config_overrides)
+
+    with open(
+        os.path.join(data_dir, "dataset_config.json"), "w", encoding="utf-8"
+    ) as f:
+        json.dump(config, f, ensure_ascii=False, indent=2)
+
+    return data_dir
+
+
 def _make_seq_dataset(
     test_dir, name="data", seq_length=200, train_type="seq", data=None, **load_kwargs
 ):
@@ -372,3 +409,106 @@ def test_dataset_load_explicit_storage_type(base_test_env):
     dataset = _make_seq_dataset(test_dir, "explicit", storage_type="h5")
     assert len(dataset) > 0
     assert dataset.count == 200
+
+
+def test_detect_format_jsonl_dir(base_test_env):
+    test_dir = base_test_env["test_dir"]
+    tokenizer_path = _save_test_tokenizer(test_dir, base_test_env["tokenizer"])
+    data_dir = _write_jsonl_dataset(
+        test_dir,
+        tokenizer_path,
+        [{"text": "hello world"}, {"text": "foo bar baz"}],
+    )
+    assert detect_format(data_dir) == "jsonl"
+
+
+def test_jsonl_store_seq(base_test_env):
+    test_dir = base_test_env["test_dir"]
+    tokenizer_path = _save_test_tokenizer(test_dir, base_test_env["tokenizer"])
+    data_dir = _write_jsonl_dataset(
+        test_dir,
+        tokenizer_path,
+        [{"text": "hello world"}, {"text": "foo bar baz qux"}],
+        config_overrides={"preprocessing": {"max_seq_len": 128, "min_chars": 0}},
+    )
+
+    store = StoreFactory.create("jsonl")
+    store.load(data_dir)
+    assert len(store) > 0
+    assert "sequence" in store.keys
+
+    dataset = DatasetFactory.load("seq", data_dir, window_size=8)
+    assert len(dataset) > 0
+    item = dataset[0]
+    assert "input_ids" in item
+    assert "target_ids" in item
+    assert item["input_ids"].dtype == torch.long
+
+
+def test_jsonl_store_sft(base_test_env):
+    test_dir = base_test_env["test_dir"]
+    tokenizer = base_test_env["tokenizer"]
+    tokenizer.set_chat_template(
+        "{% for message in messages %}{{ message['role'] }}:{{ message['content'] }}\n{% endfor %}"
+    )
+    tokenizer_path = _save_test_tokenizer(test_dir, tokenizer)
+    data_dir = _write_jsonl_dataset(
+        test_dir,
+        tokenizer_path,
+        [
+            {
+                "messages": [
+                    {"role": "system", "content": "sys"},
+                    {"role": "user", "content": "hi"},
+                    {"role": "assistant", "content": "hello"},
+                ]
+            }
+        ],
+        config_overrides={
+            "input": {
+                "sections": [{"field": "messages", "action": "$role", "template": True}]
+            },
+            "mask": {"system": "mask", "user": "mask", "assistant": "train"},
+            "mask_default": "mask",
+        },
+    )
+
+    store = StoreFactory.create("jsonl")
+    store.load(data_dir)
+    assert "sequence" in store.keys
+    assert "loss_mask" in store.keys
+    assert "position_ids" in store.keys
+
+    dataset = DatasetFactory.load("sft", data_dir, window_size=8)
+    item = dataset[0]
+    assert "input_ids" in item
+    assert "target_ids" in item
+    assert "loss_mask" in item
+    assert "position_ids" in item
+    assert item["loss_mask"].dtype == torch.bool
+
+
+def test_jsonl_store_pipeline_config_roundtrip(base_test_env):
+    test_dir = base_test_env["test_dir"]
+    config_path = os.path.join(test_dir, "dataset_config.json")
+    with open(config_path, "w", encoding="utf-8") as f:
+        json.dump(
+            {
+                "tokenizer_path": os.path.join(test_dir, "tokenizer"),
+                "version": 1,
+                "input": {"sections": [{"field": "text", "action": "train"}]},
+                "mask": {"assistant": "train"},
+                "preprocessing": {"max_seq_len": 64},
+                "output": {"position_ids_mode": "doc_reset"},
+            },
+            f,
+            ensure_ascii=False,
+            indent=2,
+        )
+
+    with open(config_path, "r", encoding="utf-8") as f:
+        raw = json.load(f)
+    raw.pop("tokenizer_path")
+    config = PipelineConfig.from_dict(raw)
+    assert config.output.position_ids_mode == "doc_reset"
+    assert config.preprocessing.max_seq_len == 64

tests/trainer/conftest.py

@@ -52,7 +52,7 @@ def create_train_config(
         batch_per_device: Batch size per device (default: 2)
         grad_accum_steps: Gradient accumulation steps (default: 1)
         max_grad_norm: Maximum gradient norm for clipping (default: 1.0)
-        ckpt_interval: Checkpoint save interval in iterations (default: 5)
+        ckpt_interval: Checkpoint save interval in optimizer steps (default: 5)
         random_seed: Random seed for reproducibility (default: 42)
         **kwargs: Additional arguments passed to TrainConfig
 

tests/trainer/test_early_stopping.py

@@ -44,14 +44,14 @@ def test_early_stopping_simulation(base_test_env, early_stopping_dataset):
         pass
 
     # Resume from latest checkpoint
-    load_dir = os.path.join(base_test_env["test_dir"], "epoch_0_iter_2")
+    load_dir = os.path.join(base_test_env["test_dir"], "epoch_0_step_1")
     trainer = Trainer(train_config)
     trainer.train(resume_dir=load_dir)
 
-    # Verify checkpoint was saved at expected iteration
+    # Verify checkpoint was saved at expected step
-    load_dir = os.path.join(base_test_env["test_dir"], "epoch_1_iter_10")
+    load_dir = os.path.join(base_test_env["test_dir"], "epoch_1_step_5")
     import json
 
     with open(os.path.join(load_dir, "meta.json")) as f:
         meta = json.load(f)
-    assert meta["iteration"] == 10
+    assert meta["consumed_samples"] == 20