diff --git a/README.md b/README.md
index fc64af16f..728c549b3 100644
--- a/README.md
+++ b/README.md
@@ -406,6 +406,7 @@ foundation for building customizable, enterprise-grade training services.
 | Component Type | Component Link                                                                                           | Component Function                                                                      | Author              |
 | -------------- | -------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------- | ------------------- |
 | Patch          | [qwen3_moe_transformers4_patch](https://www.modelscope.cn/models/twinkle-kit/qwen3_moe_transformers4_patch) | Fixes Qwen3 MoE model hang issue during FSDP2 training, effective for transformers==4.x | ModelScope Official |
+| Training recipe | [Intern-Decision on Ascend](cookbook/transformers/intern_decision_npu/README.md) | Text-only masked decision SFT, joint-field supervision and FSDP2 checkpoint recovery | — |
 
 ## Contributions
 
diff --git a/cookbook/transformers/intern_decision_npu/.gitignore b/cookbook/transformers/intern_decision_npu/.gitignore
new file mode 100644
index 000000000..ee6321c46
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/.gitignore
@@ -0,0 +1,9 @@
+/raw/
+/prepared/
+/decision_data/
+/sources/
+/results/
+/outputs/
+/data-audit.json
+__pycache__/
+*.log
diff --git a/cookbook/transformers/intern_decision_npu/JOINT_REPRODUCTION.md b/cookbook/transformers/intern_decision_npu/JOINT_REPRODUCTION.md
new file mode 100644
index 000000000..09ace6108
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/JOINT_REPRODUCTION.md
@@ -0,0 +1,61 @@
+# Joint decision training with Twinkle
+
+## Objective and scope
+
+Start from Qwen3.5-4B instruction weights, train all language parameters and freeze vision/projector parameters. Input messages retain the entire joint JSON skeleton. Gold answer symbols are supplied separately as `decision_targets`, one single-token symbol per marker, in field order. The logit immediately before a marker predicts its answer. The existing `TransformersModel` and `InputProcessor` extension points are reused. `twinkle_adapter` shifts labels once for the external cross-entropy and selects supervised positions without restricting the training vocabulary. No duplicate Qwen model implementation is introduced.
+
+Images, 35B models, sequence parallelism and cross-example packing are outside the validated recipe. Preserve question and option order; do not truncate evidence.
+
+## Environment
+
+Historical four-NPU validation used Ascend 910B, Python 3.12, CANN 9.1.0, Torch 2.10.0+cpu, torch_npu 2.10.0.post2, Transformers 5.15.1 and Accelerate 1.14.0. Twinkle additionally used PEFT 0.19.0. Use the repository installation instructions for the selected platform.
+
+The historical base image is `quay.nju.edu.cn/ascend/ms-swift@sha256:d1b56f2d77882edb92615c45641556c8d5adaf8ed360be73f4f0978f70fa01c1`. The optional `create_container.sh` reproduces its original fixed four-device layout (devices 0–3); it is not a portable scheduler or a complete dependency lock. Inspect device assignment before using that helper. No image export is required.
+
+The historical NPU image used NumPy 1.26.4 and an explicit `--no-deps` package installation. That is a reproduction-only dependency exception, outside the current upstream NumPy >=2.0 requirement. This PR leaves global dependency metadata unchanged. A dependency-compliant NPU environment still needs device validation; do not interpret the historical image as an upstream-supported dependency combination.
+
+## Prepare inputs
+
+From this example directory, supply already prepared case-level splits and the reference Intern-Decision schema:
+
+```bash
+python prepare_joint.py --prepared /path/to/case-splits \
+  --output /path/to/new-joint-data \
+  --official-schema /path/to/Intern-Decision/src/inputs/schema.py
+```
+
+The preparation tool checks compiled messages and targets against the supplied reference, records its SHA256, rejects duplicate/cross-split inputs and refuses an existing output directory. The historical schema hash was `d2fce8ffb8af19dca08a04eb01f4daae70ed43311d73c052ddcab797346d79bd`.
+
+The historical split contained 960 train, 120 validation, 120 calibration and 400 test cases, with five fields per case. Test and calibration records never enter training or checkpoint selection. The unpublished original training corpus is not reconstructed.
+
+## Launch
+
+The following variables override machine-specific paths and device assignment:
+
+```bash
+export PYTHON_BIN=/path/to/training-venv/bin/python
+export MODEL_PATH=/path/to/Qwen3.5-4B
+export DATA_ROOT=/path/to/new-joint-data
+export OUTPUT_ROOT=/path/to/new-training-run
+export ASCEND_RT_VISIBLE_DEVICES='<FOUR_ASSIGNED_NPU_IDS>'
+# Run sequentially, from this directory:
+bash run_joint.sh smoke
+bash run_joint.sh resume
+bash run_joint.sh train
+```
+
+The fixed four-rank recipe uses global batch 16, AdamW, learning rate 2e-6, cosine minimum 2e-7, warmup 0.03, weight decay 0.01 and seed 42. With the historical 960-case data, two epochs give 120 optimizer updates. Do not reuse that step count as an epoch claim for a different dataset.
+
+Smoke runs two steps and saves full state; resume verifies a third step in a fresh directory. Formal training saves a final model-only checkpoint. Model-only export does not support full optimizer/scheduler/RNG recovery. CPU checkpoint fences and Twinkle DCP hooks are scoped to this opt-in recipe; they are not general fixes for every backend.
+
+## Independent evaluation
+
+Use a separate MS-SWIFT HF/NPU inference environment to load either framework's exported weights:
+
+```bash
+ASCEND_RT_VISIBLE_DEVICES='<ONE_ASSIGNED_NPU_ID>' /path/to/eval-venv/bin/python evaluate_joint.py \
+  --checkpoint /path/to/final-hf-checkpoint \
+  --data /path/to/case-splits/test.jsonl --output /path/to/new-evaluation
+```
+
+Only validation loss may influence checkpoint selection. Official seven-suite evaluation is a separate protocol, not this held-out business subset. Historical full training and evaluation completed on the reproduction branch, but aggregate quality remained below the official target. Current PR-head device validation and upstream CI are pending. Keep measured metrics and raw predictions local.
diff --git a/cookbook/transformers/intern_decision_npu/README.md b/cookbook/transformers/intern_decision_npu/README.md
new file mode 100644
index 000000000..8cb0a0784
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/README.md
@@ -0,0 +1,31 @@
+# Intern-Decision text training on Ascend
+
+This example trains the language backbone of Qwen3.5-4B with joint decision supervision in Twinkle, while freezing the vision tower and projector. Each assistant input contains a complete JSON skeleton with `<decision>` markers. Answer symbols appear only in labels. All supervised fields use full-vocabulary causal cross-entropy in the same forward pass.
+
+The existing `TransformersModel` and `InputProcessor` extension points are reused. `twinkle_adapter` shifts labels once for the external cross-entropy and selects supervised positions without restricting the training vocabulary. No duplicate Qwen model implementation is introduced.
+
+## Run the joint-field recipe
+
+Use [JOINT_REPRODUCTION.md](JOINT_REPRODUCTION.md) for environment requirements, input preparation, smoke training, checkpoint recovery and the full run. The example currently requires four Ascend NPUs, text-only inputs and no sequence parallelism or cross-example packing.
+
+```bash
+# From the repository root, after installing its dependencies:
+python -m pytest tests/processor/test_intern_decision.py -q
+```
+
+These synthetic CPU regressions require no checkpoint or benchmark download. They cover causal loss and gradient equivalence; real-tokenizer and distributed checkpoint audits remain separate checks.
+
+## Validation scope
+
+The pinned historical reproduction completed four-rank short training and full-state recovery, 120 optimizer steps, final model export/reload and independent evaluation. The final PR revision still needs device validation and upstream CI. Historical training evidence must not be attributed to a different source revision.
+
+The original Intern-Decision training corpus is not public. This recipe uses public Typed Decisions data and reproduces the supervision method, not the official training mixture or published model quality. It is not a StartLux or 35B training recipe. Keep weights, predictions and measured results outside the source tree.
+
+## Supporting tools
+
+- `prepare_joint.py`: compile all fields of a case together and compare them with a supplied official schema; reject split overlap and answer leakage.
+- `check_checkpoint.py`: check exported model keys, frozen vision weights, sampled language updates and checkpoint artifacts.
+- `evaluate_joint.py`: evaluate a saved model on held-out joint cases using the separately installed MS-SWIFT HF/NPU backend.
+- `run.sh`, `prepare_inputs.py` (where present), `evaluate_decision.py` and `evaluate_laya_suite.py`: earlier single-question experiment utilities. They are not the default joint recipe and their budgets and metrics must not be mixed with it.
+
+See [THIRD_PARTY.md](THIRD_PARTY.md) for schema attribution. Dataset access and licenses remain the caller's responsibility.
diff --git a/cookbook/transformers/intern_decision_npu/THIRD_PARTY.md b/cookbook/transformers/intern_decision_npu/THIRD_PARTY.md
new file mode 100644
index 000000000..2f538fc58
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/THIRD_PARTY.md
@@ -0,0 +1,6 @@
+# Source attribution
+
+`decision_schema.py` is derived from the Apache-2.0 licensed
+[InternLM/Intern-Decision](https://github.com/InternLM/Intern-Decision) input schema.
+Public training inputs come from `LocalLLaMA/typed-decisions`; data is downloaded separately.
+No dataset contents or experimental results are included in this example.
diff --git a/cookbook/transformers/intern_decision_npu/check_checkpoint.py b/cookbook/transformers/intern_decision_npu/check_checkpoint.py
new file mode 100644
index 000000000..7c6957b80
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/check_checkpoint.py
@@ -0,0 +1,88 @@
+"""Audit saved model changes and full training-state files after a short run."""
+import argparse
+import json
+from pathlib import Path
+
+import torch
+from safetensors import safe_open
+from transformers import AutoTokenizer
+
+
+def weight_map(root):
+    manifest = root / 'model.safetensors.index.json'
+    if manifest.exists():
+        return json.loads(manifest.read_text())['weight_map']
+    with safe_open(root / 'model.safetensors', framework='pt', device='cpu') as handle:
+        return {key: 'model.safetensors' for key in handle.keys()}
+
+
+def tensor(root, mapping, key):
+    with safe_open(root / mapping[key], framework='pt', device='cpu') as handle:
+        return handle.get_tensor(key)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('checkpoint', type=Path)
+    parser.add_argument('--base', type=Path, default=Path('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/models/Qwen3.5-4B'))
+    parser.add_argument('--framework', choices=['swift', 'twinkle'], required=True)
+    parser.add_argument('--expected-step', type=int, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    args = parser.parse_args()
+    before, after = weight_map(args.base), weight_map(args.checkpoint)
+    # HF Qwen3_5ForConditionalGeneration does not load the base MTP predictor.
+    # The training checkpoint may explicitly serialize the tied output head.
+    ignored_mtp = {key for key in before if key.startswith('mtp.')}
+    expected = set(before) - ignored_mtp
+    extra = set(after) - expected
+    assert expected <= set(after), 'A runtime parameter is missing'
+    assert extra <= {'lm_head.weight'}, 'Unexpected exported parameter'
+    config = json.loads((args.checkpoint / 'config.json').read_text())
+    assert config['architectures'] == ['Qwen3_5ForConditionalGeneration']
+    if 'lm_head.weight' in extra:
+        assert config.get('tie_word_embeddings') is True
+        head = tensor(args.checkpoint, after, 'lm_head.weight')
+        embedding = tensor(args.checkpoint, after, 'model.language_model.embed_tokens.weight')
+        torch.testing.assert_close(head, embedding, rtol=0, atol=0)
+        del head, embedding
+    assert all((args.checkpoint / name).is_file() for name in set(after.values()))
+    vision = [key for key in before if '.visual.' in key]
+    language = [key for key in before if '.language_model.layers.0.' in key and key.endswith('weight')][:5]
+    assert vision and language
+    report = {
+        'checkpoint': str(args.checkpoint), 'frozen_tensors_verified': len(vision),
+        'unloaded_base_mtp_tensors': sorted(ignored_mtp), 'language_sample': {}}
+    for key in vision:
+        old = tensor(args.base, before, key).float()
+        new = tensor(args.checkpoint, after, key).float()
+        torch.testing.assert_close(old, new, rtol=0, atol=0)
+    for key in language:
+        old = tensor(args.base, before, key).float()
+        new = tensor(args.checkpoint, after, key).float()
+        report['language_sample'][key] = {
+            'max_abs_delta': float((old - new).abs().max()),
+            'changed_elements': int((old != new).sum())}
+    assert any(row['changed_elements'] for row in report['language_sample'].values())
+    marker = AutoTokenizer.from_pretrained(args.checkpoint).encode('<decision>', add_special_tokens=False)
+    assert len(marker) == 1
+    report['marker_id'] = marker[0]
+    state = json.loads((args.checkpoint / 'trainer_state.json').read_text())
+    report['global_step'] = state['global_step' if args.framework == 'swift' else 'cur_step']
+    assert report['global_step'] == args.expected_step
+    assert any((args.checkpoint / name).exists() for name in ['optimizer.pt', 'optimizer.bin'])
+    optimizer_dir = args.checkpoint / 'optimizer.pt'
+    if optimizer_dir.is_dir():
+        assert (optimizer_dir / '.metadata').is_file()
+        assert len(list(optimizer_dir.glob('*.distcp'))) == 4
+    assert (args.checkpoint / 'scheduler.pt').exists()
+    pattern = 'rng_state*.pth' if args.framework == 'swift' else 'rng_state_rank*.pt'
+    assert len(list(args.checkpoint.glob(pattern))) == 4
+    report['status'] = 'passed'
+    report['scope'] = 'all frozen vision tensors, sampled language updates, saved state; resume validated separately'
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(report, indent=2))
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/cookbook/transformers/intern_decision_npu/create_container.sh b/cookbook/transformers/intern_decision_npu/create_container.sh
new file mode 100755
index 000000000..f6a8fca37
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/create_container.sh
@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+# Create a fresh, isolated four-card workspace on an Ascend host.
+set -euo pipefail
+framework=${1:?swift or twinkle}
+container=${2:?new container name}
+runtime=$(realpath "${3:?new runtime directory}")
+source_dir=$(realpath "${4:?framework checkout directory}")
+model=$(realpath "${5:?base checkpoint directory}")
+data=$(realpath "${6:?data directory containing joint and prepared}")
+case "$framework" in swift|twinkle) ;; *) exit 2 ;; esac
+image=quay.nju.edu.cn/ascend/ms-swift@sha256:d1b56f2d77882edb92615c45641556c8d5adaf8ed360be73f4f0978f70fa01c1
+test -z "$(ls -A "$runtime")"
+docker run -d --name "$container" --privileged --network host --ipc host --shm-size 32g \
+  -v "$runtime:/workspace" -v "$source_dir:/workspace/framework:ro" \
+  -v "$model:/models/Qwen3.5-4B:ro" -v "$data:/data:ro" \
+  -v /usr/local/Ascend/driver:/usr/local/Ascend/driver:ro \
+  -v /usr/local/dcmi:/usr/local/dcmi:ro \
+  -v /usr/local/bin/npu-smi:/usr/local/bin/npu-smi:ro \
+  -v /etc/ascend_install.info:/etc/ascend_install.info:ro \
+  "$image" bash -c 'rm -f /dev/davinci4 /dev/davinci5 /dev/davinci6 /dev/davinci7; exec sleep infinity'
+docker exec "$container" python -m venv --system-site-packages /workspace/.venv
+if [[ "$framework" == twinkle ]]; then
+  docker exec "$container" /workspace/.venv/bin/python -m pip install --no-deps peft==0.19.0
+  # PYTHONPATH in run_joint.sh selects this exact checkout; distribution metadata
+  # is installed without copying or modifying the read-only checkout.
+  docker exec "$container" bash -c 'cp -a /workspace/framework /workspace/package-build; /workspace/.venv/bin/python -m pip install --no-deps /workspace/package-build'
+fi
diff --git a/cookbook/transformers/intern_decision_npu/decision_schema.py b/cookbook/transformers/intern_decision_npu/decision_schema.py
new file mode 100644
index 000000000..87e194fdf
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/decision_schema.py
@@ -0,0 +1,163 @@
+"""Dependency-free decision schema and prompt compiler shared by training and inference."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+DECISION_TOKEN = '<decision>'
+ANSWER_SYMBOLS = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
+SYSTEM_PROMPT = (
+    'You are a careful decision assistant. Use the state and decision schema in '
+    'the user message to make the requested decisions. For every field, choose '
+    'exactly one answer symbol (e.g. A, B, C, ...) from its listed options and '
+    'return one valid JSON object mapping each field name to its chosen symbol. '
+    'Use the field names and symbols exactly as given. Do not include '
+    'explanations, Markdown, or extra text.'
+)
+
+
+@dataclass(frozen=True)
+class CompiledExample:
+    messages: list[dict[str, Any]]
+    fields: tuple[str, ...]
+    symbols: dict[str, tuple[str, ...]]
+    targets: dict[str, str] | None
+
+
+def _json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=False)
+
+
+def _options(question: Mapping[str, Any]) -> list[tuple[str, str]]:
+    kind = question.get('type')
+    criteria = question.get('criteria')
+    if kind == 'choice':
+        if not isinstance(criteria, Mapping):
+            raise ValueError('choice criteria must be an object')
+        return [(str(key), str(value)) for key, value in criteria.items()]
+    if kind == 'score':
+        if isinstance(criteria, list):
+            return [(str(index), str(value)) for index, value in enumerate(criteria)]
+        if isinstance(criteria, Mapping):
+            return [(str(key), str(value)) for key, value in criteria.items()]
+        raise ValueError('score criteria must be a list or object')
+    if kind == 'noul':
+        descriptions = criteria if isinstance(criteria, Mapping) else {}
+        yes = next(
+            (str(descriptions[k]) for k in descriptions if str(k).lower() in {'yes', 'true', '1'}),
+            'The answer is yes (affirmative, or align with the claim).',
+        )
+        no = next(
+            (str(descriptions[k]) for k in descriptions if str(k).lower() in {'no', 'false', '0'}),
+            'The answer is no (negative, or disagree with the claim).',
+        )
+        return [('no', no), ('yes', yes)]
+    raise ValueError(f"unsupported question type: {kind!r}")
+
+
+def _answer_value(question: Mapping[str, Any], target: Mapping[str, Any] | None = None) -> str | None:
+    answer = question.get('answer')
+    if not isinstance(answer, Mapping):
+        answer = target
+    if not isinstance(answer, Mapping):
+        return None
+    kind = question.get('type')
+    if 'label' in answer:
+        value = answer['label']
+        if kind == 'noul':
+            if str(value).lower() not in {'yes', 'true', '1', 'no', 'false', '0'}:
+                raise ValueError(f"Invalid noul label: {value!r}")
+            return 'yes' if str(value).lower() in {'yes', 'true', '1'} else 'no'
+        return str(value)
+    if kind == 'choice':
+        value = answer.get('choice')
+        return None if value is None else str(value)
+    if kind == 'score':
+        value = answer.get('score')
+        return None if value is None else str(int(value) if isinstance(value, float) and value.is_integer() else value)
+    if kind == 'noul':
+        value = answer.get('noul')
+        return None if value is None else ('yes' if float(value) >= 0.5 else 'no')
+    return None
+
+
+def _message_content(row: Mapping[str, Any], user_text: str) -> str | list[dict[str, Any]]:
+    images = row.get('images') or []
+    if not images:
+        return user_text
+    content: list[dict[str, Any]] = []
+    for image in images:
+        if isinstance(image, Mapping):
+            url = image.get('url') or image.get('path')
+            image_wh = image.get('image_wh')
+        else:
+            url, image_wh = str(image), None
+        image_url: dict[str, Any] = {'url': url}
+        if image_wh is not None:
+            image_url['image_wh'] = image_wh
+        content.append({'type': 'image_url', 'image_url': image_url})
+    content.append({'type': 'text', 'text': user_text})
+    return content
+
+
+def compile_row(row: Mapping[str, Any], include_targets: bool = True) -> CompiledExample:
+    questions = row.get('questions')
+    if not isinstance(questions, Mapping) or not questions:
+        raise ValueError(f"{row.get('id', '<missing id>')}: questions must be a non-empty object")
+
+    fields: list[str] = []
+    symbols: dict[str, tuple[str, ...]] = {}
+    schema_lines: list[str] = []
+    targets: dict[str, str] = {}
+    row_targets = row.get('targets')
+    if not isinstance(row_targets, Mapping):
+        row_targets = {}
+    for field, question in questions.items():
+        if not isinstance(question, Mapping):
+            raise ValueError(f"{row.get('id', '<missing id>')}: question {field!r} is not an object")
+        options = _options(question)
+        if not options:
+            raise ValueError(f"{row.get('id', '<missing id>')}: question {field!r} has no options")
+        if len(options) > len(ANSWER_SYMBOLS):
+            raise ValueError(f"At most {len(ANSWER_SYMBOLS)} single-token answer symbols are supported")
+        field_name = str(field)
+        fields.append(field_name)
+        field_symbols = tuple(ANSWER_SYMBOLS[: len(options)])
+        symbols[field_name] = field_symbols
+        schema_lines.append(f"{field_name}: {question.get('instructions', '')}")
+        for symbol, (value, description) in zip(field_symbols, options):
+            schema_lines.append(f"    {symbol} = {value}: {description}")
+        if include_targets:
+            answer = _answer_value(question, row_targets.get(field))
+            if answer is not None:
+                values = [value for value, _ in options]
+                if answer not in values:
+                    raise ValueError(
+                        f"{row.get('id', '<missing id>')}: answer {answer!r} is not an option for {field_name!r}"
+                    )
+                targets[field_name] = field_symbols[values.index(answer)]
+
+    state = _json(row.get('state'))
+    user_text = (
+        'Return one answer for every field using the supplied answer symbols.\n\n'
+        '## State\n'
+        f"{state}\n"
+        '## Decision schema\n' + '\n'.join(schema_lines)
+    )
+    if DECISION_TOKEN in user_text:
+        raise ValueError('Reserved decision marker appears in input evidence')
+    skeleton = json.dumps(dict.fromkeys(fields, DECISION_TOKEN), ensure_ascii=False, indent=4)
+    messages = [
+        {'role': 'system', 'content': SYSTEM_PROMPT},
+        {'role': 'user', 'content': _message_content(row, user_text)},
+        {'role': 'assistant', 'content': skeleton},
+    ]
+    return CompiledExample(
+        messages=messages,
+        fields=tuple(fields),
+        symbols=symbols,
+        targets=targets if include_targets else None,
+    )
diff --git a/cookbook/transformers/intern_decision_npu/download_data.py b/cookbook/transformers/intern_decision_npu/download_data.py
new file mode 100644
index 000000000..29d5e06ee
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/download_data.py
@@ -0,0 +1,18 @@
+"""Download the public, pinned typed-decisions source files."""
+import json
+from pathlib import Path
+import shutil
+from huggingface_hub import HfApi, hf_hub_download
+
+root = Path(__file__).resolve().parent
+revision = 'd0e2f0c42fef86cc15d1688d25a19f5ba7c85b18'
+repo = 'LocalLLaMA/typed-decisions'
+(root / 'raw').mkdir(exist_ok=True)
+(root / 'sources').mkdir(exist_ok=True)
+info = HfApi().dataset_info(repo, revision=revision)
+assert info.sha == revision
+(root / 'sources/typed-meta.json').write_text(json.dumps({'sha': info.sha}))
+for split in ['train', 'test']:
+    source = hf_hub_download(repo, 'all/' + split + '-00000-of-00001.parquet',
+                             repo_type='dataset', revision=revision)
+    shutil.copy2(source, root / 'raw' / (split + '.parquet'))
diff --git a/cookbook/transformers/intern_decision_npu/evaluate_joint.py b/cookbook/transformers/intern_decision_npu/evaluate_joint.py
new file mode 100644
index 000000000..a5dabe60a
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/evaluate_joint.py
@@ -0,0 +1,75 @@
+"""Independent joint-field candidate evaluation, one complete case per forward."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import time
+
+import torch
+import torch_npu  # noqa: F401
+from swift.model import get_model_processor
+from decision_schema import compile_row
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--checkpoint', required=True)
+    parser.add_argument('--data', type=Path, required=True,
+                        help='Original prepared case JSONL, not tokenized training rows')
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--limit', type=int, default=0, help='Nonzero is smoke only')
+    args = parser.parse_args()
+    args.output.mkdir(parents=True, exist_ok=False)
+    cases = [json.loads(line) for line in args.data.read_text().splitlines()]
+    if args.limit:
+        cases = cases[:args.limit]
+    assert cases and len({case['id'] for case in cases}) == len(cases)
+    torch.manual_seed(42)
+    torch.npu.set_device(0)
+    model, processor = get_model_processor(
+        args.checkpoint, model_type='qwen3_5', torch_dtype=torch.bfloat16,
+        device_map='npu:0', attn_impl='sdpa', new_special_tokens=['<decision>'])
+    model.eval()
+    tokenizer = processor.tokenizer
+    marker = tokenizer.convert_tokens_to_ids('<decision>')
+    correct = total = 0
+    started = time.monotonic()
+    with (args.output / 'predictions.jsonl').open('x') as stream, torch.inference_mode():
+        for case in cases:
+            compiled = compile_row(case, include_targets=False)
+            gold = compile_row(case, include_targets=True).targets
+            assert compiled.targets is None
+            text = tokenizer.apply_chat_template(compiled.messages, tokenize=False,
+                                                 add_generation_prompt=False, enable_thinking=False)
+            inputs = tokenizer(text, add_special_tokens=False, return_tensors='pt').to('npu:0')
+            assert inputs['input_ids'].shape[1] <= 8192
+            positions = (inputs['input_ids'][0] == marker).nonzero().flatten()
+            assert len(positions) == len(compiled.fields) and bool((positions > 0).all())
+            scores = model(**inputs, use_cache=False, logits_to_keep=positions - 1).logits[0]
+            for index, field in enumerate(compiled.fields):
+                symbols = compiled.symbols[field]
+                ids = [tokenizer.encode(symbol, add_special_tokens=False) for symbol in symbols]
+                assert all(len(token_ids) == 1 for token_ids in ids)
+                logits = scores[index, [token_ids[0] for token_ids in ids]].float().cpu()
+                assert bool(torch.isfinite(logits).all())
+                prediction = int(logits.argmax())
+                expected = symbols.index(gold[field])
+                record = {'case_id': case['id'], 'field': field, 'gold_index': expected,
+                          'prediction_index': prediction, 'symbols': symbols,
+                          'probabilities': logits.softmax(-1).tolist()}
+                stream.write(json.dumps(record) + '\n')
+                correct += prediction == expected
+                total += 1
+            stream.flush()
+    assert total == sum(len(case['questions']) for case in cases)
+    result = {'cases': len(cases), 'decisions': total, 'correct': correct, 'accuracy': correct / total,
+              'protocol': 'joint fields, original order, one complete case per forward, no padding or truncation',
+              'checkpoint': args.checkpoint, 'data_sha256': hashlib.sha256(args.data.read_bytes()).hexdigest(),
+              'backend': 'common MS-SWIFT HF NPU BF16 SDPA', 'temperature': 1.0,
+              'elapsed_seconds': time.monotonic() - started, 'smoke_limit': args.limit}
+    (args.output / 'metrics.json').write_text(json.dumps(result, indent=2))
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/cookbook/transformers/intern_decision_npu/prepare_data.py b/cookbook/transformers/intern_decision_npu/prepare_data.py
new file mode 100644
index 000000000..57446a014
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/prepare_data.py
@@ -0,0 +1,101 @@
+"""Prepare a case-disjoint decision-training pilot; never tokenize test labels as input."""
+import hashlib
+import json
+import math
+from collections import Counter, defaultdict
+from pathlib import Path
+
+import pyarrow.parquet as pq
+
+ROOT = Path(__file__).resolve().parent
+OUT = ROOT / 'prepared'
+OUT.mkdir(exist_ok=True)
+
+
+def sha(value):
+    return hashlib.sha256(value).hexdigest()
+
+
+def canonical(value):
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)
+
+
+def convert(raw):
+    state, questions, gold = [json.loads(raw[k]) for k in ('state', 'questions', 'gold')]
+    assert len(questions) == raw['n_questions'] == 5
+    targets = {}
+    for field, q in questions.items():
+        assert set(q) <= {'type', 'instructions', 'criteria'}
+        if q['type'] == 'choice':
+            keys = list(q['criteria'])
+            output_keys = keys
+        elif q['type'] == 'score':
+            keys = [str(i) for i in range(len(q['criteria']))]
+            output_keys = keys
+        else:
+            assert q['type'] == 'noul'
+            keys, output_keys = ['false', 'true'], ['no', 'yes']
+        p = [float(gold[field]['probabilities'][k]) for k in keys]
+        assert all(math.isfinite(v) and 0 <= v <= 1 for v in p)
+        assert abs(sum(p)-1) <= 1e-4
+        p = [v/sum(p) for v in p]
+        # Rounded probabilities can tie. Preserve the published hard label instead
+        # of silently selecting a different first-argmax answer.
+        label_index = keys.index(str(gold[field]['label']).lower() if q['type'] == 'noul' else str(gold[field]['label']))
+        assert max(p)-p[label_index] <= 2e-6
+        targets[field] = {'label': output_keys[label_index],
+                          'soft_probabilities': dict(zip(output_keys, p))}
+    return {'id': raw['id'], 'workflow': raw['workflow'], 'state': state,
+            'questions': questions, 'images': [], 'targets': targets}
+
+
+raw = {s: pq.read_table(ROOT/'raw'/f'{s}.parquet').to_pylist() for s in ('train', 'test')}
+assert len(raw['train']) == 1200 and len(raw['test']) == 400
+converted = {s: [convert(r) for r in rr] for s, rr in raw.items()}
+by_workflow = defaultdict(list)
+for row in converted['train']:
+    by_workflow[row['workflow']].append(row)
+splits = {'train': [], 'validation': [], 'calibration': [], 'test': converted['test']}
+for workflow, rows in sorted(by_workflow.items()):
+    assert len(rows) == 300
+    ordered = sorted(rows, key=lambda r: sha(('20261003:'+workflow+':'+r['id']).encode()))
+    splits['train'] += ordered[:240]
+    splits['validation'] += ordered[240:270]
+    splits['calibration'] += ordered[270:]
+seen_ids, seen_states, seen_inputs = set(), set(), set()
+audit = {'dataset_revision': json.loads((ROOT/'sources/typed-meta.json').read_text())['sha'],
+         'source_license': 'apache-2.0 (dataset card)',
+         'split_policy': 'per workflow 240 train / 30 validation / 30 calibration cases; official test unchanged',
+         'seed_key': '20261003', 'supervision': 'normalized teacher probabilities and original published hard labels; not human gold',
+         'model_input_allowlist': ['state', 'questions', 'images'],
+         'excluded_input_fields': ['id', 'workflow', 'targets', 'gold', 'factors', 'label_agreement'],
+         'tokenization': 'not yet performed with target model tokenizer', 'splits': {}}
+for split, rows in splits.items():
+    ids = {r['id'] for r in rows}
+    states = {sha(canonical(r['state']).encode()) for r in rows}
+    inputs = {sha(canonical({'state': r['state'], 'questions': r['questions']}).encode()) for r in rows}
+    assert len(ids) == len(rows)
+    assert not ids & seen_ids and not states & seen_states and not inputs & seen_inputs
+    seen_ids |= ids; seen_states |= states; seen_inputs |= inputs
+    path = OUT/f'{split}.jsonl'
+    payload = ''.join(json.dumps(r, ensure_ascii=False, allow_nan=False)+'\n' for r in rows)
+    path.write_text(payload)
+    audit['splits'][split] = {'cases': len(rows), 'decisions': sum(len(r['questions']) for r in rows),
+                              'workflows': dict(Counter(r['workflow'] for r in rows)),
+                              'question_types': dict(Counter(q['type'] for r in rows for q in r['questions'].values())),
+                              'sha256': sha(path.read_bytes()), 'ids': sorted(ids)}
+# Confirm these are the same test cases previously evaluated, even if IDs differ.
+prior = json.loads((ROOT.parent/'new-model-20261001/laya-comparison-20261003/laya_suites_multi.json').read_text())['suites']['typed_decisions']['cases']
+assert len(prior) == len(converted['test'])
+for a, b in zip(prior, converted['test']):
+    assert a['state'] == b['state'] and a['questions'] == b['questions']
+    for field, g in a['gold'].items():
+        keys = list(b['targets'][field]['soft_probabilities'])
+        assert keys[g['idx']] == b['targets'][field]['label']
+audit['previous_test_verified'] = {'cases': 400, 'decisions': 2000, 'state_question_label_match': True}
+audit['exact_cross_split_overlap'] = {'id': 0, 'state': 0, 'state_questions': 0}
+audit['semantic_near_duplicate_audit'] = 'not performed; exact matching does not prove semantic independence'
+audit['raw_sha256'] = {s: sha((ROOT/'raw'/f'{s}.parquet').read_bytes()) for s in raw}
+(ROOT/'data-audit.json').write_text(json.dumps(audit, ensure_ascii=False, indent=2))
+print(json.dumps({k: {x:v[x] for x in ['cases','decisions','workflows']} for k,v in audit['splits'].items()}, ensure_ascii=False, indent=2))
+print('Exact split audits and prior test identity checks passed.')
diff --git a/cookbook/transformers/intern_decision_npu/prepare_inputs.py b/cookbook/transformers/intern_decision_npu/prepare_inputs.py
new file mode 100644
index 000000000..e8f0f8151
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/prepare_inputs.py
@@ -0,0 +1,34 @@
+"""Compile per-question training inputs; keep gold symbols outside messages."""
+import hashlib
+import json
+from pathlib import Path
+
+from decision_schema import compile_row
+
+
+def main():
+    root = Path(__file__).parent
+    out = root / 'decision_data'
+    out.mkdir(exist_ok=True)
+    counts = {}
+    for split in ('train', 'validation', 'calibration', 'test'):
+        rows = []
+        for line in (root / 'prepared' / f'{split}.jsonl').read_text().splitlines():
+            case = json.loads(line)
+            for field, question in case['questions'].items():
+                row = {'state': case['state'], 'questions': {field: question},
+                       'targets': {field: case['targets'][field]}}
+                compiled = compile_row(row)
+                assert compiled.messages == compile_row(row, include_targets=False).messages
+                rows.append({'messages': compiled.messages,
+                             'decision_targets': [compiled.targets[field]],
+                             'case_id': case['id'], 'field': field})
+        path = out / f'{split}.jsonl'
+        path.write_text(''.join(json.dumps(r, ensure_ascii=False) + '\n' for r in rows))
+        counts[split] = {'decisions': len(rows), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()}
+    (out / 'manifest.json').write_text(json.dumps(counts, indent=2))
+    print(json.dumps(counts, indent=2))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/cookbook/transformers/intern_decision_npu/prepare_joint.py b/cookbook/transformers/intern_decision_npu/prepare_joint.py
new file mode 100644
index 000000000..af2a9d54f
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/prepare_joint.py
@@ -0,0 +1,76 @@
+"""Compile joint decisions and check the pinned official input contract."""
+import argparse
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import sys
+
+from decision_schema import compile_row
+
+
+def prepare(prepared, output, official_schema):
+    spec = importlib.util.spec_from_file_location('official_decision_schema', official_schema)
+    official = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = official
+    spec.loader.exec_module(official)
+    if output.exists():
+        raise FileExistsError('Use a fresh output directory; do not overwrite frozen data')
+    splits = {}
+    seen_ids, seen_inputs = set(), set()
+    manifest = {
+        'protocol': 'joint fields; full-vocabulary causal CE; targets excluded from messages',
+        'official_schema_sha256': hashlib.sha256(official_schema.read_bytes()).hexdigest(),
+        'training_source': 'public Typed Decisions training split, not the unpublished official training corpus',
+        'splits': {},
+    }
+    for split in ('train', 'validation', 'calibration', 'test'):
+        source = prepared / f"{split}.jsonl"
+        cases = [json.loads(line) for line in source.read_text().splitlines()]
+        rows, ids, inputs = [], set(), set()
+        for case in cases:
+            compiled = compile_row(case)
+            reference = official.compile_row(case)
+            if compiled.messages != reference.messages or compiled.targets != reference.targets:
+                raise ValueError(f"Official compiler mismatch in {split}")
+            if compiled.messages != compile_row(case, include_targets=False).messages:
+                raise ValueError('Targets leaked into input messages')
+            if len(compiled.targets) != len(compiled.fields):
+                raise ValueError('All fields must have one supervised answer')
+            identity = hashlib.sha256(json.dumps(compiled.messages, ensure_ascii=False).encode()).hexdigest()
+            if case['id'] in ids or identity in inputs:
+                raise ValueError(f"Duplicate case in {split}")
+            ids.add(case['id'])
+            inputs.add(identity)
+            rows.append({'messages': compiled.messages,
+                         'decision_targets': [compiled.targets[f] for f in compiled.fields],
+                         'case_id': case['id'], 'fields': list(compiled.fields)})
+        if ids & seen_ids or inputs & seen_inputs:
+            raise ValueError('Cross-split input or case overlap')
+        seen_ids |= ids
+        seen_inputs |= inputs
+        payload = ''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in rows)
+        splits[split] = payload
+        manifest['splits'][split] = {
+            'cases': len(rows), 'decisions': sum(len(r['decision_targets']) for r in rows),
+            'source_sha256': hashlib.sha256(source.read_bytes()).hexdigest(),
+            'sha256': hashlib.sha256(payload.encode()).hexdigest(),
+        }
+    output.mkdir(parents=True)
+    for split, payload in splits.items():
+        (output / f"{split}.jsonl").write_text(payload)
+    (output / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
+    return manifest
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--prepared', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--official-schema', type=Path, required=True)
+    args = parser.parse_args()
+    print(json.dumps(prepare(args.prepared, args.output, args.official_schema), indent=2))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/cookbook/transformers/intern_decision_npu/run.sh b/cookbook/transformers/intern_decision_npu/run.sh
new file mode 100755
index 000000000..e3b0fefe7
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/run.sh
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd /workspace
+export ASCEND_RT_VISIBLE_DEVICES=${ASCEND_RT_VISIBLE_DEVICES:-0,1,2,3}
+export OMP_NUM_THREADS=${OMP_NUM_THREADS:-4}
+export TOKENIZERS_PARALLELISM=false
+export HCCL_CONNECT_TIMEOUT=300 HCCL_EXEC_TIMEOUT=600
+export PYTORCH_NPU_ALLOC_CONF=expandable_segments:True
+export PYTHONPATH=/workspace:${PYTHONPATH:-}
+torchrun --nproc_per_node=4 --master_port=29641 -m twinkle_adapter.train "$@"
diff --git a/cookbook/transformers/intern_decision_npu/run_joint.sh b/cookbook/transformers/intern_decision_npu/run_joint.sh
new file mode 100644
index 000000000..8b9fc2bf0
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/run_joint.sh
@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")"
+python_bin=${PYTHON_BIN:-python}
+model=${MODEL_PATH:?Set MODEL_PATH to the local Qwen3.5-4B base checkpoint}
+data=${DATA_ROOT:?Set DATA_ROOT to the prepared joint JSONL directory}
+source_root=$(cd ../../.. && pwd)
+export PYTHONPATH="$PWD:$source_root/src:${PYTHONPATH:-}"
+export ASCEND_RT_VISIBLE_DEVICES=${ASCEND_RT_VISIBLE_DEVICES:?Assign four NPU devices}
+export OMP_NUM_THREADS=${OMP_NUM_THREADS:-4}
+export TOKENIZERS_PARALLELISM=false HF_HUB_OFFLINE=1 HF_DATASETS_OFFLINE=1
+export HCCL_CONNECT_TIMEOUT=300 HCCL_EXEC_TIMEOUT=600
+export PYTORCH_NPU_ALLOC_CONF=expandable_segments:True
+stage=${1:-smoke}
+output=${OUTPUT_ROOT:?Set OUTPUT_ROOT to a new experiment directory}
+case "$stage" in
+  smoke) extra=(--steps 2 --save-every 2 --output "$output/smoke") ;;
+  resume) extra=(--steps 3 --save-every 3 --resume "$output/smoke/checkpoint-2" --output "$output/resume") ;;
+  train) extra=(--steps 120 --save-every 120 --model-only --validation "$data/validation.jsonl" --output "$output/train") ;;
+  *) exit 2 ;;
+esac
+"$python_bin" -m torch.distributed.run --nproc-per-node 4 \
+  --master-addr "${MASTER_ADDR:-127.0.0.1}" --master-port "${MASTER_PORT:-29661}" -m twinkle_adapter.train \
+  --model "$model" --data "$data/train.jsonl" \
+  --schedule-steps 120 --global-batch 16 "${extra[@]}"
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/__init__.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/__init__.py
new file mode 100644
index 000000000..b1205f32c
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/__init__.py
@@ -0,0 +1 @@
+"""Text decision training example for Ascend NPU."""
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_checkpoint_io.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_checkpoint_io.py
new file mode 100644
index 000000000..c4b7a2781
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_checkpoint_io.py
@@ -0,0 +1,60 @@
+"""Four-rank NPU check: sharded Adam moments and step survive DCP save/reload."""
+from datetime import timedelta
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import torch
+import torch_npu  # noqa: F401
+import torch.distributed as dist
+from torch.distributed.device_mesh import init_device_mesh
+from torch.distributed.fsdp import fully_shard, MixedPrecisionPolicy
+from .checkpoint import install_sharded_optimizer_io, reshard_for_checkpoint
+
+rank = int(os.environ['RANK'])
+torch.npu.set_device(int(os.environ['LOCAL_RANK']))
+dist.init_process_group('hccl')
+mesh = init_device_mesh('npu', (dist.get_world_size(),))
+cpu_group = dist.new_group(backend='gloo', timeout=timedelta(minutes=5))
+def fence():
+    dist.monitored_barrier(group=cpu_group, timeout=timedelta(minutes=5))
+torch.manual_seed(42)
+model = torch.nn.Linear(16, 16).to('npu')
+fully_shard(model, mesh=mesh, mp_policy=MixedPrecisionPolicy(param_dtype=torch.bfloat16))
+optimizer = torch.optim.AdamW(model.parameters(), lr=2e-6)
+x = torch.randn(4, 16, device='npu', dtype=torch.bfloat16)
+model(x).float().square().mean().backward()
+optimizer.step()
+optimizer.zero_grad()
+def local(t):
+    return t.to_local() if hasattr(t, 'to_local') else t
+expected = [{k: local(v).detach().cpu().clone() for k, v in s.items()}
+            for s in optimizer.state.values()]
+model.eval()
+with torch.no_grad():
+    model(x)
+optimizer_ids = {id(p) for g in optimizer.param_groups for p in g['params']}
+before_reshard_match = optimizer_ids.issubset({id(p) for p in model.parameters()})
+reshard_for_checkpoint(model)
+assert optimizer_ids.issubset({id(p) for p in model.parameters()})
+strategy = SimpleNamespace()
+install_sharded_optimizer_io(strategy, cpu_group, fence)
+path = '/workspace/results/twinkle-dcp-eval-io-check'
+strategy.save_optimizer_checkpoint(model, optimizer, path)
+new_optimizer = torch.optim.AdamW(model.parameters(), lr=2e-6)
+strategy.load_optimizer_checkpoint(model, new_optimizer, path)
+assert len(expected) == len(new_optimizer.state)
+for reference, actual in zip(expected, new_optimizer.state.values()):
+    assert reference.keys() == actual.keys()
+    for key in reference:
+        torch.testing.assert_close(local(actual[key]).cpu(), reference[key], rtol=0, atol=0)
+fence()
+if rank == 0:
+    result = {'status': 'passed', 'ranks': dist.get_world_size(),
+              'optimizer_parameter_identity_match_before_reshard': before_reshard_match,
+              'optimizer_parameter_identity_match_after_reshard': True,
+              'optimizer_moments_and_step_exactly_restored': True}
+    Path('/workspace/results/twinkle-dcp-eval-io-check.json').write_text(json.dumps(result, indent=2))
+    print(json.dumps(result), flush=True)
+dist.destroy_process_group()
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_contract.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_contract.py
new file mode 100644
index 000000000..ea98c0b17
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/check_contract.py
@@ -0,0 +1,83 @@
+"""CPU checks against Twinkle's actual processor and CE, with real tokenizer."""
+import argparse
+import json
+from copy import deepcopy
+from pathlib import Path
+from unittest.mock import patch
+import torch
+import torch.nn.functional as F
+from transformers import AutoTokenizer
+from twinkle.loss import CrossEntropyLoss
+from .data import count_supervised_tokens, encode_decision
+from .processor import DecisionProcessor
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--output', type=Path, default=Path('/workspace/results/twinkle-cpu-contract.json'))
+    args = parser.parse_args()
+    raw = [{'labels': [-100, 2, -100]}, {'labels': [3, -100, 5, -100]}]
+    tensors = [{'labels': torch.tensor(row['labels']).unsqueeze(0)} for row in raw]
+    assert count_supervised_tokens(raw) == count_supervised_tokens(tensors) == 3
+    tokenizer = AutoTokenizer.from_pretrained('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/models/Qwen3.5-4B')
+    tokenizer.add_special_tokens({'additional_special_tokens': ['<decision>']})
+    processor = DecisionProcessor(pad_token_id=tokenizer.pad_token_id)
+    count = 0
+    decisions = 0
+    sample_features = []
+    for split in ['train', 'validation', 'calibration', 'test']:
+        for line in Path('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/data/joint', split + '.jsonl').read_text().splitlines():
+            record = json.loads(line)
+            feature = encode_decision(tokenizer, record)
+            assert sum(x != -100 for x in feature['labels']) == len(record['decision_targets'])
+            changed = deepcopy(record)
+            changed['decision_targets'] = ['Z'] * len(record['decision_targets'])
+            assert encode_decision(tokenizer, changed)['input_ids'] == feature['input_ids']
+            if len(sample_features) < 4:
+                sample_features.append(feature)
+            count += 1
+            decisions += len(record['decision_targets'])
+    # Call the real processor on CPU in a container with no accelerator devices.
+    # Twinkle's platform defaults to CUDA when neither accelerator exists.
+    # Override device placement ONLY in this CPU numerical test.
+    with patch('twinkle.processor.base.Platform.get_local_device', return_value='cpu'):
+        batch = processor(deepcopy(sample_features))
+    assert batch['input_ids'].shape[-1] % 128 == 0
+    assert batch['use_cache'] is False
+    assert batch['labels'].shape[-1] == len(batch['logits_to_keep'])
+    assert int((batch['labels'] != -100).sum()) == sum(sum(x != -100 for x in f['labels']) for f in sample_features)
+    # Smaller synthetic vocabulary gives an exact numerical/gradient comparison
+    # without allocating a full sequence x Qwen vocabulary tensor on CPU.
+    torch.manual_seed(42)
+    b, t, v = 3, 12, 17
+    unshifted = torch.full((b, t), -100, dtype=torch.long)
+    unshifted[0, 4], unshifted[1, 7], unshifted[2, 3] = 2, 5, 9
+    unshifted[2, 8] = 11  # Also cover multi-marker supervision.
+    shifted = torch.full_like(unshifted, -100)
+    shifted[:, :-1] = unshifted[:, 1:]
+    logits = torch.randn(b, t, v, dtype=torch.float64, requires_grad=True)
+    reference = F.cross_entropy(logits[:, :-1].reshape(-1, v),
+                                unshifted[:, 1:].reshape(-1), ignore_index=-100)
+    selected = processor.select_decision_logits({
+        'input_ids': torch.ones(b, t, dtype=torch.long),
+        'attention_mask': torch.ones(b, t, dtype=torch.long), 'labels': shifted})
+    sparse = logits.index_select(1, selected['logits_to_keep'])
+    labels = selected['labels']
+    logps = sparse.log_softmax(-1).gather(-1, labels.clamp_min(0).unsqueeze(-1)).squeeze(-1)
+    actual = CrossEntropyLoss(reduction='mean')(selected, {'logps': logps})['loss']
+    torch.testing.assert_close(actual, reference, rtol=1e-12, atol=1e-12)
+    a = torch.autograd.grad(actual, logits, retain_graph=True)[0]
+    r = torch.autograd.grad(reference, logits)[0]
+    torch.testing.assert_close(a, r, rtol=1e-12, atol=1e-12)
+    result = {'status': 'passed', 'encoded_records': count, 'encoded_decisions': decisions,
+              'label_shift': 'exactly_once_before_Twinkle_CE', 'gold_not_in_input': True,
+              'real_processor': True, 'test_device_override': 'cpu', 'loss_and_gradients_match_causal_CE': True,
+              'npu_training_tested': False}
+    output = args.output
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(result, indent=2))
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/checkpoint.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/checkpoint.py
new file mode 100644
index 000000000..3f9da04a1
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/checkpoint.py
@@ -0,0 +1,47 @@
+"""FSDP2 sharded optimizer I/O via PyTorch DCP; HF model export stays separate."""
+from pathlib import Path
+
+import torch
+import torch.distributed.checkpoint as dcp
+from torch.distributed.fsdp import FSDPModule
+from torch.distributed.checkpoint.state_dict import (
+    StateDictOptions, get_optimizer_state_dict, set_optimizer_state_dict,
+)
+
+
+def reshard_for_checkpoint(model):
+    """Restore sharded parameter identities after inference without backward."""
+    for module in model.modules():
+        if isinstance(module, FSDPModule):
+            module.reshard()
+
+
+def install_sharded_optimizer_io(strategy, cpu_group, fence):
+    def save(model, optimizer, output_path, *, param_name_mapping=None):
+        if param_name_mapping:
+            raise ValueError('Only full-parameter optimizer names are supported')
+        inner = getattr(optimizer, 'optimizer', optimizer)
+        reshard_for_checkpoint(model)
+        torch.npu.synchronize()
+        fence()
+        state = {'optimizer': get_optimizer_state_dict(
+            model, inner, options=StateDictOptions(full_state_dict=False, cpu_offload=False))}
+        dcp.save(state, checkpoint_id=output_path, process_group=cpu_group)
+        fence()
+
+    def load(model, optimizer, input_path, *, param_name_mapping=None):
+        if param_name_mapping:
+            raise ValueError('Only full-parameter optimizer names are supported')
+        if not (Path(input_path) / '.metadata').is_file():
+            raise ValueError('Missing completed distributed optimizer checkpoint')
+        inner = getattr(optimizer, 'optimizer', optimizer)
+        reshard_for_checkpoint(model)
+        options = StateDictOptions(full_state_dict=False, cpu_offload=False)
+        state = {'optimizer': get_optimizer_state_dict(model, inner, options=options)}
+        dcp.load(state, checkpoint_id=input_path, process_group=cpu_group)
+        set_optimizer_state_dict(model, inner, state['optimizer'], options=options)
+        torch.npu.synchronize()
+        fence()
+
+    strategy.save_optimizer_checkpoint = save
+    strategy.load_optimizer_checkpoint = load
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/data.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/data.py
new file mode 100644
index 000000000..21f3fc550
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/data.py
@@ -0,0 +1,38 @@
+"""Text decision encoding; Twinkle consumes labels already shifted for causal CE."""
+from collections.abc import Mapping
+from copy import deepcopy
+
+import torch
+
+
+def count_supervised_tokens(features):
+    """Count labels in raw rows and in the dataloader's singleton tensor rows."""
+    return sum(int(torch.as_tensor(row['labels']).ne(-100).sum().item()) for row in features)
+
+
+def encode_decision(tokenizer, record, max_length=8192):
+    messages = deepcopy(record['messages'])
+    if any(not isinstance(m['content'], str) for m in messages):
+        raise ValueError('Only text decisions are supported by this training adapter')
+    ids = tokenizer.apply_chat_template(messages, tokenize=True,
+                                        add_generation_prompt=False, enable_thinking=False)
+    if isinstance(ids, Mapping):
+        ids = ids['input_ids']
+    ids = list(ids)
+    if len(ids) > max_length:
+        raise ValueError('Refusing to truncate decision evidence')
+    marker = tokenizer.encode('<decision>', add_special_tokens=False)
+    if len(marker) != 1:
+        raise ValueError('Register <decision> as one special token first')
+    positions = [i for i, token in enumerate(ids) if token == marker[0]]
+    targets = record['decision_targets']
+    if not positions or len(positions) != len(targets):
+        raise ValueError('Decision marker/target mismatch')
+    labels = [-100] * len(ids)
+    for position, symbol in zip(positions, targets):
+        target = tokenizer.encode(symbol, add_special_tokens=False)
+        if position == 0 or len(target) != 1:
+            raise ValueError('Decision requires one answer token and a preceding position')
+        # Unlike HF causal loss, Twinkle's external CE does NOT shift labels.
+        labels[position - 1] = target[0]
+    return {'input_ids': ids, 'attention_mask': [1] * len(ids), 'labels': labels}
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/processor.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/processor.py
new file mode 100644
index 000000000..195196bdf
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/processor.py
@@ -0,0 +1,35 @@
+"""Keep only supervised positions in lm_head while preserving full-vocabulary CE."""
+import torch
+import torch.nn.functional as F
+from twinkle.processor import InputProcessor
+
+
+class DecisionProcessor(InputProcessor):
+    def __init__(self, *, pad_token_id, pad_multiple=128, **kwargs):
+        super().__init__(padding_free=False, **kwargs)
+        self.pad_token_id = pad_token_id
+        self.pad_multiple = pad_multiple
+        # Append after the base whitelist, which would discard logits_to_keep.
+        self.process_pipeline.append(self.select_decision_logits)
+
+    def select_decision_logits(self, inputs, **kwargs):
+        if self.framework != 'transformers':
+            raise ValueError('This adapter is for TransformersModel without sequence parallelism')
+        if kwargs.get('enable_sp'):
+            raise ValueError('Sequence parallelism has not been validated for sparse decision logits')
+        labels = inputs['labels']
+        if labels.ndim != 2 or not torch.all((labels != -100).sum(-1) > 0):
+            raise ValueError('Every decision example must contain supervision')
+        length = inputs['input_ids'].shape[-1]
+        multiple = self.pad_multiple
+        padding = (-length) % multiple if multiple else 0
+        for key, value in [('input_ids', self.pad_token_id), ('attention_mask', 0)]:
+            if padding:
+                inputs[key] = F.pad(inputs[key], (0, padding), value=value)
+        # Positions derive from labels already shifted by encode_decision.
+        selected = (labels != -100).any(dim=0).nonzero(as_tuple=True)[0]
+        inputs['labels'] = labels.index_select(-1, selected)
+        inputs['logits_to_keep'] = selected
+        # Recomputing a checkpointed layer must not append to a mutable KV cache.
+        inputs['use_cache'] = False
+        return inputs
diff --git a/cookbook/transformers/intern_decision_npu/twinkle_adapter/train.py b/cookbook/transformers/intern_decision_npu/twinkle_adapter/train.py
new file mode 100644
index 000000000..517cdac58
--- /dev/null
+++ b/cookbook/transformers/intern_decision_npu/twinkle_adapter/train.py
@@ -0,0 +1,221 @@
+"""Full language-parameter decision SFT with Twinkle; initial NPU smoke entry."""
+import argparse
+from datetime import timedelta
+import json
+import math
+import os
+from pathlib import Path
+import shutil
+import time
+
+import torch
+import torch.distributed as dist
+import torch_npu  # noqa: F401
+from transformers import AutoTokenizer, get_cosine_with_min_lr_schedule_with_warmup
+import twinkle
+from twinkle import DeviceMesh
+from twinkle.dataloader import DataLoader
+from twinkle.dataset import Dataset, DatasetMeta
+from twinkle.model import TransformersModel
+from twinkle.kernel.ops.fla.npu import apply_qwen3_5_fla
+from .data import count_supervised_tokens, encode_decision
+from .processor import DecisionProcessor
+from .checkpoint import install_sharded_optimizer_io, reshard_for_checkpoint
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--model', default='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/models/Qwen3.5-4B')
+    parser.add_argument('--data', default='/data/joint/train.jsonl')
+    parser.add_argument('--output', required=True)
+    parser.add_argument('--steps', type=int, default=3)
+    parser.add_argument('--schedule-steps', type=int, default=120)
+    parser.add_argument('--model-only', action='store_true', help='Final weights only; no optimizer resume')
+    parser.add_argument('--global-batch', type=int, default=16)
+    parser.add_argument('--save-every', type=int, default=150)
+    parser.add_argument('--resume')
+    parser.add_argument('--validation')
+    args = parser.parse_args()
+    world = int(os.environ['WORLD_SIZE'])
+    rank = int(os.environ['RANK'])
+    local_rank = int(os.environ['LOCAL_RANK'])
+    if world != 4 or args.global_batch % world:
+        raise ValueError('This entry was configured for four NPU workers')
+    if args.steps > args.schedule_steps:
+        raise ValueError('Requested steps exceed the fixed LR schedule')
+    torch.npu.set_device(local_rank)
+    mesh = DeviceMesh.from_sizes(fsdp_size=world, dp_size=1)
+    twinkle.initialize(mode='local', nproc_per_node=world, global_device_mesh=mesh, seed=42)
+    tokenizer = AutoTokenizer.from_pretrained(args.resume or args.model)
+    tokenizer.add_special_tokens({'additional_special_tokens': ['<decision>']})
+    records = [json.loads(s) for s in Path(args.data).read_text().splitlines()]
+    features = [encode_decision(tokenizer, r) for r in records]
+    if len(features) % args.global_batch:
+        raise ValueError('Use complete global training batches for this baseline')
+    steps_per_epoch = len(features) // args.global_batch
+    # An explicit deterministic permutation for each epoch is shared by all ranks;
+    # Twinkle's sampler then slices each GLOBAL batch exactly once across the mesh.
+    generator = torch.Generator().manual_seed(42)
+    ordered = []
+    for _ in range(math.ceil(args.schedule_steps / steps_per_epoch)):
+        ordered.extend(features[i] for i in torch.randperm(len(features), generator=generator).tolist())
+    dataset = Dataset(dataset_meta=DatasetMeta(data=ordered))
+    loader = DataLoader(dataset=dataset, device_mesh=mesh, batch_size=args.global_batch,
+                        shuffle=False, num_workers=0, drop_last=True)
+    validation_features = None
+    if args.validation:
+        validation_features = [encode_decision(tokenizer, json.loads(s))
+                               for s in Path(args.validation).read_text().splitlines()]
+    apply_qwen3_5_fla()
+    model = TransformersModel(
+        model_id=args.resume or args.model, device_mesh=mesh,
+        dtype=torch.float32, attn_implementation='sdpa', mixed_precision='bf16',
+        memory_efficient_init=True, strategy='accelerate',
+        fsdp_config={'reshard_after_forward': True, 'activation_checkpointing': True,
+                     'cpu_ram_efficient_loading': True, 'state_dict_type': 'FULL_STATE_DICT'})
+    model.model.gradient_checkpointing_disable()
+    model.model.config.use_cache = False
+    model.model.config.text_config.use_cache = False
+    apply_qwen3_5_fla(model)
+    model.model._no_split_modules = sorted({type(m).__name__ for m in model.model.modules()
+                                            if type(m).__name__.endswith('DecoderLayer')})
+    frozen = 0
+    trainable = 0
+    for name, param in model.model.named_parameters():
+        is_vision = name.startswith('visual.') or '.visual.' in name
+        param.requires_grad_(not is_vision)
+        if is_vision:
+            frozen += param.numel()
+        else:
+            trainable += param.numel()
+    if frozen == 0 or trainable == 0:
+        raise ValueError('Vision/language parameter partition failed')
+    if max(tokenizer.encode('<decision>', add_special_tokens=False)) >= model.model.get_input_embeddings().num_embeddings:
+        raise ValueError('Marker is outside the model vocabulary')
+    group = model.optimizer_group['']
+    # Twinkle registers a forward hook even for pre-encoded text batches.
+    # Retain its template; input_ids bypass batch_encode, so labels are not shifted again.
+    if hasattr(group.template.processor, 'tokenizer'):
+        group.template.processor.tokenizer = tokenizer
+    else:
+        group.template.processor = tokenizer
+    model._default_tokenizer = tokenizer
+    model.set_processor(DecisionProcessor, pad_token_id=tokenizer.pad_token_id, pad_multiple=128)
+    model.set_loss('CrossEntropyLoss', reduction='sum')
+    model.set_optimizer('AdamW', lr=2e-6, weight_decay=0.01, betas=(0.9, 0.999), eps=1e-8)
+    group.lr_scheduler = get_cosine_with_min_lr_schedule_with_warmup(
+        group.optimizer, num_warmup_steps=math.ceil(args.schedule_steps * 0.03),
+        num_training_steps=args.schedule_steps, min_lr=2e-7)
+    # Prevent early HCCL collectives from overlapping rank-zero CPU offload.
+    cpu_group = dist.new_group(backend='gloo', timeout=timedelta(minutes=20))
+    def fence():
+        dist.monitored_barrier(group=cpu_group, timeout=timedelta(minutes=20))
+    for method in ['get_full_state_dict']:
+        original = getattr(model.strategy, method)
+        def guarded(*a, _original=original, **kw):
+            reshard_for_checkpoint(model.model)
+            result = _original(*a, **kw)
+            # Accelerate activation checkpointing introduces a wrapper namespace.
+            # Export the original HF names so loading cannot silently miss layers.
+            normalized = {}
+            for key, value in result.items():
+                canonical = key.replace('._checkpoint_wrapped_module.', '.')
+                if canonical in normalized:
+                    raise RuntimeError(f'Duplicate export key: {canonical}')
+                normalized[canonical] = value
+            fence()
+            return normalized
+        setattr(model.strategy, method, guarded)
+    install_sharded_optimizer_io(model.strategy, cpu_group, fence)
+    if args.resume:
+        progress = model.resume_from_checkpoint(args.resume)
+        loader.resume_from_checkpoint(progress['consumed_train_samples'])
+        model._load_rng_state(str(Path(args.resume) / f'rng_state_rank{rank}.pt'))
+    output = Path(args.output)
+    if rank == 0:
+        output.mkdir(parents=True, exist_ok=False)
+    fence()
+    best_path = output / 'best-selection.json'
+    best = json.loads(best_path.read_text()) if best_path.exists() else {'loss': float('inf')}
+
+    def evaluate():
+        validation_loader = DataLoader(
+            dataset=Dataset(dataset_meta=DatasetMeta(data=validation_features)),
+            device_mesh=mesh, batch_size=args.global_batch, shuffle=False,
+            num_workers=0, drop_last=False)
+        loss_sum, decisions, cases = 0.0, 0, 0
+        for validation_batch in validation_loader:
+            decisions += count_supervised_tokens(validation_batch)
+            cases += len(validation_batch)
+            model.forward_only(inputs=validation_batch)
+            loss_sum += model.calculate_loss()
+        model.calculate_metric(is_training=False)
+        totals = torch.tensor([loss_sum, decisions, cases], dtype=torch.float64)
+        dist.all_reduce(totals, group=cpu_group)
+        expected_decisions = count_supervised_tokens(validation_features)
+        if int(totals[1]) != expected_decisions or int(totals[2]) != len(validation_features):
+            raise RuntimeError('Validation sampler did not cover each decision exactly once')
+        loss = float(totals[0] / totals[1])
+        if not math.isfinite(loss):
+            raise RuntimeError('Non-finite validation loss')
+        return loss
+    if rank == 0:
+        (output / 'run-config.json').write_text(json.dumps({
+            **vars(args), 'trainable_parameters': trainable, 'frozen_parameters': frozen,
+            'twinkle_revision': 'eff13b93a8576f773f11c85adccffdd3eb22fab3',
+            'status': 'running', 'precision_acceptance': 'not_yet_evaluated'}, indent=2))
+    started = time.monotonic()
+    for batch in loader:
+        result = model.forward_backward(inputs=batch)
+        if not math.isfinite(float(result['loss'])):
+            raise RuntimeError('Non-finite training loss')
+        model.clip_grad_and_step(max_grad_norm=1.0)
+        if not math.isfinite(float(group._last_grad_norm)):
+            raise RuntimeError('Non-finite gradient norm; stopping before checkpoint save')
+        step = group.cur_step
+        metrics = model.calculate_metric(is_training=True)
+        if rank == 0:
+            record = {'step': step, 'elapsed_s': time.monotonic() - started, 'metrics': metrics,
+                      'peak_memory_gib': torch.npu.max_memory_allocated() / 1024**3}
+            with (output / 'logging.jsonl').open('a') as handle:
+                handle.write(json.dumps(record, default=str) + '\n')
+            print(json.dumps(record, default=str), flush=True)
+        if step % args.save_every == 0 or step >= args.steps:
+            fence()
+            name = f'checkpoint-{step}'
+            model.save(name, output_dir=str(output), save_optimizer=not args.model_only,
+                       consumed_train_samples=loader.get_state()['consumed_train_samples'])
+            fence()
+            validation_loss = evaluate() if validation_features else None
+            torch.save(model._get_training_rng_state(), output / name / f'rng_state_rank{rank}.pt')
+            fence()
+            if rank == 0:
+                for config in Path(args.model).glob('*processor*.json'):
+                    shutil.copy2(config, output / name / config.name)
+                if validation_loss is not None:
+                    with (output / 'validation.jsonl').open('a') as handle:
+                        handle.write(json.dumps({'step': step, 'loss': validation_loss}) + '\n')
+                    if validation_loss < best['loss']:
+                        best = {'loss': validation_loss, 'step': step,
+                                'checkpoint': str(output / name)}
+                        best_path.write_text(json.dumps(best, indent=2))
+                    # Retain the latest recoverable checkpoint plus validation-best.
+                    keep = {str(output / name), best.get('checkpoint')}
+                    for old in output.glob('checkpoint-*'):
+                        if old.is_dir() and str(old) not in keep:
+                            shutil.rmtree(old)
+            fence()
+        if step >= args.steps:
+            break
+    if group.cur_step != args.steps:
+        raise RuntimeError(f'Incomplete training: {group.cur_step} steps, expected {args.steps}')
+    if rank == 0:
+        (output / 'training-complete.json').write_text(json.dumps({'step': group.cur_step,
+            'checkpoint': str(output / f'checkpoint-{group.cur_step}'),
+            'best_model_checkpoint': best.get('checkpoint'),
+            'accuracy_evaluated': False, 'optimizer_saved': not args.model_only}, indent=2))
+    dist.destroy_process_group()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/processor/test_intern_decision.py b/tests/processor/test_intern_decision.py
new file mode 100644
index 000000000..1bef5f9cb
--- /dev/null
+++ b/tests/processor/test_intern_decision.py
@@ -0,0 +1,86 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+"""CPU decision encoding and full-vocabulary loss/gradient regressions."""
+import importlib.util
+from copy import deepcopy
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+import torch.nn.functional as F
+
+from twinkle.loss import CrossEntropyLoss
+
+
+@pytest.fixture
+def adapter():
+    root = Path(__file__).resolve().parents[2] / 'cookbook/transformers/intern_decision_npu/twinkle_adapter'
+    modules = {}
+    for name in ('data', 'processor'):
+        spec = importlib.util.spec_from_file_location(f'decision_test_{name}', root / f'{name}.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        modules[name] = module
+    return SimpleNamespace(**modules)
+
+
+class Tokenizer:
+    def apply_chat_template(self, messages, **kwargs):
+        return [1, 2, 9, 3, 9, 4]
+
+    def encode(self, text, **kwargs):
+        return {'<decision>': [9], 'A': [5], 'B': [6], 'invalid': [7, 8]}[text]
+
+
+def test_single_shift_and_no_answer_leakage(adapter):
+    record = {'messages': [{'role': 'assistant', 'content': 'skeleton'}], 'decision_targets': ['A', 'B']}
+    feature = adapter.data.encode_decision(Tokenizer(), record)
+    assert feature['labels'] == [-100, 5, -100, 6, -100, -100]
+    changed = deepcopy(record)
+    changed['decision_targets'] = ['B', 'A']
+    assert adapter.data.encode_decision(Tokenizer(), changed)['input_ids'] == feature['input_ids']
+    assert adapter.data.count_supervised_tokens([feature]) == 2
+    tensor_row = {'labels': torch.tensor(feature['labels']).unsqueeze(0)}
+    assert adapter.data.count_supervised_tokens([tensor_row]) == 2
+
+
+@pytest.mark.parametrize('targets', [[], ['A'], ['A', 'invalid']])
+def test_invalid_targets(adapter, targets):
+    with pytest.raises(ValueError):
+        adapter.data.encode_decision(Tokenizer(), {'messages': [], 'decision_targets': targets})
+
+
+def test_no_truncation(adapter):
+    with pytest.raises(ValueError, match='truncate'):
+        adapter.data.encode_decision(Tokenizer(), {'messages': [], 'decision_targets': ['A', 'B']}, max_length=5)
+
+
+def test_selected_loss_and_gradient(adapter):
+    processor = adapter.processor.DecisionProcessor(pad_token_id=0)
+    b, t, v = 3, 12, 17
+    unshifted = torch.full((b, t), -100, dtype=torch.long)
+    unshifted[0, 4], unshifted[1, 7], unshifted[2, 3], unshifted[2, 8] = 2, 5, 9, 11
+    shifted = torch.full_like(unshifted, -100)
+    shifted[:, :-1] = unshifted[:, 1:]
+    logits = torch.randn(b, t, v, generator=torch.Generator().manual_seed(42),
+                         dtype=torch.float64, requires_grad=True)
+    reference = F.cross_entropy(logits[:, :-1].reshape(-1, v), unshifted[:, 1:].reshape(-1))
+    selected = processor.select_decision_logits({'input_ids': torch.ones(b, t, dtype=torch.long),
+                                                 'attention_mask': torch.ones(b, t, dtype=torch.long),
+                                                 'labels': shifted})
+    assert selected['input_ids'].shape[-1] == 128
+    assert not selected['use_cache']
+    sparse = logits.index_select(1, selected['logits_to_keep'])
+    labels = selected['labels']
+    logps = sparse.log_softmax(-1).gather(-1, labels.clamp_min(0).unsqueeze(-1)).squeeze(-1)
+    actual = CrossEntropyLoss(reduction='mean')(selected, {'logps': logps})['loss']
+    torch.testing.assert_close(actual, reference, rtol=1e-12, atol=1e-12)
+    a = torch.autograd.grad(actual, logits, retain_graph=True)[0]
+    r = torch.autograd.grad(reference, logits)[0]
+    torch.testing.assert_close(a, r, rtol=1e-12, atol=1e-12)
+
+
+def test_missing_supervision_rejected(adapter):
+    processor = adapter.processor.DecisionProcessor(pad_token_id=0)
+    with pytest.raises(ValueError, match='supervision'):
+        processor.select_decision_logits({'labels': torch.full((1, 5), -100)})