Add encoder-only Ego4D comparison workflow

README.md

@@ -300,6 +300,13 @@ extraction, training and inference; no symlink index is needed.
 GPU workers, with per-GPU logs, validated resume, and foreground supervision.
 See the guide's L20 commands for the `/data1/yfzx/wangx` paths.
 
+For **encoder-only A / V / AV**, without Thinker states, see
+[the encoder-only comparison guide](docs/ego4d_encoder_only.md).
+`prepare_encoder_only_features.py` packs complete local vectors and validity
+bits on CPU; `run_ego4d_encoder_only.py` trains/evaluates four-epoch heads with
+no dev set. This native-concatenation baseline is not capacity- or
+context-matched to the Thinker baseline.
+
 ## Current Ego4D context-length results
 
 The following single-seed snapshot was recorded on 2026-08-22. Both variants

docs/ego4d_encoder_only.md

@@ -0,0 +1,95 @@
+# Ego4D：不输入 Thinker 状态的 encoder-only 对照
+
+本实验复用已经提取完的 `local_qwen_1s/<uid>.pt`，仅将选定的 Qwen 编码器向量和
+有效性标志作为 anticipation MLP 的输入。**不读取全局 Thinker `.pt`，不运行 Qwen，
+不重新处理视频，不使用软链接。** 原来的 `run_ego4d_local_baselines.py` 跑的是
+G / G+A / G+V / G+A+V，并不是本实验的 encoder-only。
+
+沿用原 EgoSpeak manifest 的 train/test、5Hz 行索引、十个预测 horizon、LSTR-MCE，
+固定 4 epochs、无 dev set，保存并评测最后一轮，不按 test 选择 epoch。
+
+## 1. 一次性整理输入（CPU）
+
+必须先完成全部 manifest clips 的局部特征提取；不要用 preview 或 partial smoke 文件。
+下面的默认路径就是 L20 上的 `/data1/yfzx/wangx`：
+
+```bash
+cd /data1/yfzx/wangx/Code/icassp2027-experiment
+
+python prepare_encoder_only_features.py \
+  --data-root /data1/yfzx/wangx/Datasets/ego4d \
+  --local-feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/local_qwen_1s \
+  --modalities av \
+  --dry-run
+
+# 校验通过后，去掉 --dry-run 写出实体文件。
+python prepare_encoder_only_features.py \
+  --data-root /data1/yfzx/wangx/Datasets/ego4d \
+  --local-feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/local_qwen_1s \
+  --modalities av
+```
+
+默认输出为 `omni_features/encoder_only_av_1s`。换成 `--modalities audio` 或 `visual`
+会分别写入 `encoder_only_audio_1s` / `encoder_only_visual_1s`。
+可显式指定 `--manifest`、`--target-root`、`--output-root`。
+局部输入支持平铺目录和 `train/test` 子目录，不改变 manifest 中的 split。
+
+整理前逐条检查全部 train/test 的完整性、时间/索引、选定向量/有效性掩码、编码器配置一致性
+及 target 长度，再创建新目录。整理时逐条重新校验并原子保存；不保留全数据集张量。
+同名目录已存在时拒绝覆盖。中断或磁盘写入失败的目录不会有 `complete.txt`，不能拿来训练；
+修复原因后用新的 `--output-root`，不要手工补写完成标记。
+
+每个输出文件明确标记 `mode=encoder-only-av/audio/visual`、`uses_thinker_states=False`，
+保存原始 times/label_indices 和编码器配置。`export_manifest.json` 记录 clip 行数、总量、
+路径和输入维度；`complete.txt` 最后写出。更换 seed 只重新训练，不重复整理。
+此前内联 Python 命令完整生成的文件和 `complete.txt` 也兼容下面的训练入口。
+
+## 2. 训练和评测（单张空闲 GPU、前台运行）
+
+```bash
+CUDA_VISIBLE_DEVICES=0 \
+OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4 \
+python run_ego4d_encoder_only.py \
+  --data-root /data1/yfzx/wangx/Datasets/ego4d \
+  --feature-root /data1/yfzx/wangx/Datasets/ego4d/omni_features/encoder_only_av_1s \
+  --modalities av \
+  --seeds 0 1 2
+```
+
+可先加 `--dry-run`：只检查全部 train/test 输入和已有输出，打印命令，不训练或写结果。
+默认结果为 `results/ego4d_encoder_only_1s/encoder_av_seed0` 等，checkpoint 位于
+`checkpoints/ego4d_encoder_only_1s`。已有 checkpoint、训练摘要或结果目录时拒绝覆盖；
+补跑时选择未完成的 seeds，或指定新的 `--checkpoint-root` / `--result-root`。
+
+换模态时同时修改 `--feature-root` 和 `--modalities`，不能只改其中一个。
+不传 `--local-feature-root`、`--layer-fusion` 或 `--dev-sessions`；原训练/推理文件无需修改。
+训练超参是 batch=512、lr=7e-5、weight_decay=5e-5、warmup_epochs=1.6、
+warmup_factor=0.3、dropout=0.2、transition_objective=none。
+训练前仍逐条检查所有导出输入，不能仅靠完成标记绕过缺文件或模式检查。
+
+入口结束后自动汇总，也可单独打印：
+
+```bash
+python summarize_local_baselines.py \
+  --root results/ego4d_encoder_only_1s --dataset ego4d
+```
+
+## 如何解释比较
+
+以两路原生向量均为 2048D 为例，audio/visual 单模态输入各 2049D（一个有效标志），
+AV 为 4098D，顺序是 `[visual, visual_valid, audio, audio_valid]`。
+不做原 B1 的 trainable 256D 分支投影；仍用原 MLP 类，但输入宽度决定其隐藏层宽度，
+即单模态 1024、AV 2049。因此应报告参数量，**不是相同 head 容量的 Thinker 消融**。
+
+窗口和视觉视野也不同：这些 sidecars 是已保存的近期音频／人物局部 crop pooling，
+不是把 Thinker 的相同输入 token 绕过 decoder。目录名中的 `1s` 只是默认实验命名，
+真实窗口以 `config.encoder_contract.config` 为准。它回答“这些编码器 readout 本身能预测多好”，
+不能单独证明 Thinker 有用或无用。严格归因仍需统一窗口、视野、pooling 与 head 容量。
+5Hz 时间及 horizon 语义沿用原实验，不因重新打包而改变。
+
+```bash
+python -m unittest -q test_encoder_only_workflow
+```
+
+CPU 测试覆盖三种模态的精确输入构造、缺失/不完整/混合配置拒绝、无 Thinker 依赖、
+输出防覆盖与完成标记，以及三组各 4 epochs 的真实小规模训练/推理。L20 真实数据效果需另跑。

docs/ego4d_local_baselines.md

@@ -1,5 +1,8 @@
 # Ego4D：近期音频／逐人视觉增强的四组对照
 
+本页四组均包含全局 Thinker 输入 G。若要完全不输入 Thinker 状态，使用
+[encoder-only A / V / AV 对照](ego4d_encoder_only.md)，不要把本页的 audio/visual/av 当作纯编码器组。
+
 目标是检验 EasyCom 上的局部特征收益能否迁移到 Ego4D，而不是预先承诺提分。
 保留现有 Ego4D 全局 Omni 特征 G、EgoSpeak manifest 和 target；不重提 Thinker，
 不重新编码视频，不重划 train/test。新增的两路侧路特征一次提取、三组复用：

prepare_encoder_only_features.py

@@ -0,0 +1,137 @@
+"""Pack Ego4D local Qwen encoder vectors as standalone head inputs on CPU.
+
+No model, raw media, Thinker states or symlink index is loaded. All manifest
+train/test sidecars are validated before creating a fresh output directory.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import tempfile
+
+import numpy as np
+import torch
+from tqdm.auto import tqdm
+
+from easycom_head import load_ego4d_manifest_split, resolve_feature_path
+from local_turn_features import MODALITIES, load_local_features
+
+
+DEFAULT_DATA_ROOT = Path("/data1/yfzx/wangx/Datasets/ego4d")
+PACKING = "native-vectors-plus-validity-bits"
+
+
+def encoder_only_mode(modalities):
+    if modalities not in MODALITIES:
+        raise ValueError(f"Unknown encoder modalities: {modalities}")
+    return f"encoder-only-{modalities}"
+
+
+def parse_args(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT)
+    parser.add_argument("--manifest", type=Path)
+    parser.add_argument("--local-feature-root", type=Path)
+    parser.add_argument("--target-root", type=Path)
+    parser.add_argument("--output-root", type=Path)
+    parser.add_argument("--modalities", choices=tuple(MODALITIES), default="av")
+    parser.add_argument("--dry-run", action="store_true", help="Validate all inputs; no output files or directory")
+    parser.add_argument("--no-progress", action="store_true")
+    args = parser.parse_args(argv)
+    args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv"
+    args.local_feature_root = args.local_feature_root or args.data_root / "omni_features/local_qwen_1s"
+    args.target_root = args.target_root or args.data_root / "ego4d_target_frame"
+    args.output_root = args.output_root or args.data_root / f"omni_features/encoder_only_{args.modalities}_1s"
+    for key in ("data_root", "manifest", "local_feature_root", "target_root", "output_root"):
+        setattr(args, key, getattr(args, key).expanduser().resolve())
+    if args.output_root == args.local_feature_root:
+        parser.error("Encoder-only output must be separate from local source features")
+    return args
+
+
+def pack_sample(path, target_path, sample_id, modalities, expected_contract=None):
+    """Read only selected encoder vectors; even a sidecar's reference path is unused."""
+    source = torch.load(path, map_location="cpu", weights_only=False)
+    if not isinstance(source, dict):
+        raise ValueError(f"Invalid local sidecar: {path}")
+    anchor = MODALITIES[modalities][0]
+    reference = {
+        "states": source.get(anchor), "times": source.get("times"),
+        "label_indices": source.get("label_indices"), "start": 0.,
+    }
+    states, contract = load_local_features(
+        path, reference, sample_id=sample_id, dataset="ego4d", modalities=modalities,
+        expected_contract=expected_contract,
+    )
+    target = np.load(target_path, allow_pickle=False, mmap_mode="r")
+    if target.shape != (len(states), 3):
+        raise ValueError(f"{sample_id}: feature/target mismatch: {tuple(states.shape)} vs {target.shape}")
+    if len(states) <= 10:
+        raise ValueError(f"{sample_id}: no examples with all ten future horizons")
+    states = states.to(torch.bfloat16)
+    if not torch.isfinite(states).all():
+        raise ValueError(f"{sample_id}: encoder vectors overflow bfloat16 storage")
+    return {
+        "format": "qwen-encoder-only-v1", "dataset": "ego4d", "sample_id": sample_id,
+        "states": states, "times": source["times"], "label_indices": source["label_indices"],
+        "mode": encoder_only_mode(modalities), "start": 0., "end": len(states) / 5,
+        "num_steps": len(states), "source_local_features": str(path),
+        "config": {
+            "input_source": "qwen-encoders-only", "uses_thinker_states": False,
+            "packing": PACKING, "encoder_contract": contract,
+        },
+    }, contract
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    names = load_ego4d_manifest_split(args.manifest, "all")
+    if args.output_root.exists() and not args.dry_run:
+        raise FileExistsError(f"Refusing to overwrite {args.output_root}; use a fresh output directory")
+    paths = {name: resolve_feature_path(args.local_feature_root, name, dataset="ego4d") for name in names}
+    targets = {name: args.target_root / f"{name}.npy" for name in names}
+    for path in targets.values():
+        if not path.is_file():
+            raise FileNotFoundError(path)
+    contract, total, width = None, 0, None
+    # Two passes keep memory bounded and prevent bad/partial test inputs from
+    # leaving an apparently ready export. No full-dataset tensors are retained.
+    for name in tqdm(names, desc="encoder-only preflight", unit="clip", disable=args.no_progress):
+        packed, contract = pack_sample(paths[name], targets[name], name, args.modalities, contract)
+        total += len(packed["states"])
+        width = packed["states"].shape[-1]
+        del packed
+    print(f"Validated {len(names)} clips / {total} states; {encoder_only_mode(args.modalities)}; "
+          f"input width={width}; no Thinker features", flush=True)
+    if args.dry_run:
+        return
+    args.output_root.mkdir(parents=True, exist_ok=False)
+    rows = {}
+    for name in tqdm(names, desc="pack encoder-only", unit="clip", disable=args.no_progress):
+        packed, _ = pack_sample(paths[name], targets[name], name, args.modalities, contract)
+        destination = args.output_root / f"{name}.pt"
+        with tempfile.NamedTemporaryFile(dir=args.output_root, prefix=f".{name}.", suffix=".tmp", delete=False) as handle:
+            temporary = Path(handle.name)
+        try:
+            torch.save(packed, temporary)
+            os.replace(temporary, destination)
+        finally:
+            temporary.unlink(missing_ok=True)
+        rows[name] = len(packed["states"])
+        del packed
+    (args.output_root / "export_manifest.json").write_text(json.dumps({
+        "manifest": str(args.manifest), "mode": encoder_only_mode(args.modalities),
+        "local_feature_root": str(args.local_feature_root), "target_root": str(args.target_root),
+        "encoder_contract": contract, "input_width": width, "num_states": total, "clips": rows,
+    }, indent=2) + "\n")
+    # Written last. The runner also validates every train/test input instead of
+    # trusting this marker alone. Compatible with the original inline command.
+    (args.output_root / "complete.txt").write_text(f"{len(names)} clips\n")
+    print(f"Complete: {args.output_root}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

run_ego4d_encoder_only.py

@@ -0,0 +1,133 @@
+"""Run raw encoder-only Ego4D heads: four epochs, no dev, manifest splits.
+
+This is not a capacity/context-matched Thinker ablation: native encoder vectors
+and validity flags are concatenated, so the existing MLP adapts to input width.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+
+from tqdm.auto import tqdm
+
+from easycom_head import load_ego4d_manifest_split, load_session_data
+from local_turn_features import LOCAL_FORMAT, MODALITIES
+from prepare_encoder_only_features import DEFAULT_DATA_ROOT, PACKING, encoder_only_mode
+
+
+REPO = Path(__file__).resolve().parent
+
+
+def parse_args(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT)
+    parser.add_argument("--manifest", type=Path)
+    parser.add_argument("--target-root", type=Path)
+    parser.add_argument("--feature-root", type=Path, required=True, help="Prepared encoder-only inputs, not Thinker/local sidecars")
+    parser.add_argument("--modalities", choices=tuple(MODALITIES), default="av")
+    parser.add_argument("--checkpoint-root", type=Path, default=Path("checkpoints/ego4d_encoder_only_1s"))
+    parser.add_argument("--result-root", type=Path, default=Path("results/ego4d_encoder_only_1s"))
+    parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2])
+    parser.add_argument("--device", default="cuda:0")
+    parser.add_argument("--dry-run", action="store_true", help="Validate all train/test inputs and print commands without training")
+    parser.add_argument("--no-progress", action="store_true")
+    args = parser.parse_args(argv)
+    args.manifest = args.manifest or args.data_root / "ego4d_egospeak.tsv"
+    args.target_root = args.target_root or args.data_root / "ego4d_target_frame"
+    for key in ("data_root", "manifest", "target_root", "feature_root", "checkpoint_root", "result_root"):
+        setattr(args, key, getattr(args, key).expanduser().resolve())
+    if len(set(args.seeds)) != len(args.seeds) or any(seed < 0 for seed in args.seeds):
+        parser.error("Seeds must be unique and nonnegative")
+    return args
+
+
+def run_paths(args, seed):
+    name = f"encoder_{args.modalities}_seed{seed}"
+    return args.checkpoint_root / f"{name}.pth", args.result_root / name
+
+
+def preflight(args):
+    names = load_ego4d_manifest_split(args.manifest, "all")
+    train = load_ego4d_manifest_split(args.manifest, "train")
+    test = load_ego4d_manifest_split(args.manifest, "test")
+    marker = args.feature_root / "complete.txt"
+    if not marker.is_file() or marker.read_text().strip() != f"{len(names)} clips":
+        raise ValueError(f"Missing/incomplete encoder-only export: {marker}")
+    for seed in args.seeds:
+        checkpoint, result = run_paths(args, seed)
+        if checkpoint.exists() or checkpoint.with_suffix(".pth.json").exists() or result.exists():
+            raise FileExistsError(f"Refusing to overwrite {checkpoint} or {result}; choose unfinished seeds/new output roots")
+    expected, total, width = None, 0, None
+    for name in tqdm(names, desc="encoder-only inputs", unit="clip", disable=args.no_progress):
+        data = load_session_data(args.feature_root, args.target_root, name, dataset="ego4d",
+                                 expected_mode=encoder_only_mode(args.modalities))
+        config = data.feature_metadata.get("config") or {}
+        contract = config.get("encoder_contract") or {}
+        if (config.get("input_source") != "qwen-encoders-only" or config.get("uses_thinker_states") is not False
+                or config.get("packing") != PACKING or contract.get("format") != LOCAL_FORMAT
+                or contract.get("modalities") != args.modalities):
+            raise ValueError(f"{name}: not a matching encoder-only export")
+        encoder_config = contract.get("config") or {}
+        if (encoder_config.get("causal_policy") != "half-open-trailing-window"
+                or encoder_config.get("uses_speech_labels") is not False):
+            raise ValueError(f"{name}: missing causal/no-speech-label encoder contract")
+        dimensions = [encoder_config.get(f"{part}_dim") for part in MODALITIES[args.modalities]]
+        if any(not isinstance(d, int) or d <= 0 for d in dimensions):
+            raise ValueError(f"{name}: invalid encoder dimensions")
+        width = sum(d + 1 for d in dimensions)
+        if data.states.ndim != 2 or data.hidden_size != width or len(data.states) <= 10:
+            raise ValueError(f"{name}: invalid native encoder-vector/validity-bit shape")
+        if expected is not None and config != expected:
+            raise ValueError(f"{name}: encoder export contract differs across clips")
+        expected = config
+        total += len(data.states)
+        del data
+    print(f"Encoder-only {args.modalities}: {len(train)} train / {len(test)} test; "
+          f"{total} states, input width={width}, MLP hidden width={width // 2}; epochs=4, dev=none", flush=True)
+    print("No Thinker features or local-fusion branch. Native concatenation changes head capacity; "
+          "context remains the saved encoder windows.", flush=True)
+
+
+def experiment_commands(args):
+    common = ["--dataset", "ego4d", "--manifest", str(args.manifest),
+              "--feature-root", str(args.feature_root), "--target-root", str(args.target_root),
+              "--expected-mode", encoder_only_mode(args.modalities), "--device", args.device]
+    if args.no_progress:
+        common.append("--no-progress")
+    for seed in args.seeds:
+        checkpoint, result = run_paths(args, seed)
+        train = [sys.executable, str(REPO / "train_turn_head.py"), *common,
+                 "--output", str(checkpoint), "--epochs", "4", "--batch-size", "512",
+                 "--learning-rate", "7e-5", "--weight-decay", "5e-5", "--warmup-epochs", "1.6",
+                 "--warmup-factor", "0.3", "--dropout", "0.2", "--loss", "lstr-mce",
+                 "--transition-objective", "none", "--seed", str(seed)]
+        infer = [sys.executable, str(REPO / "infer_turn_head.py"), *common,
+                 "--checkpoint", str(checkpoint), "--output-dir", str(result)]
+        yield train, infer
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    preflight(args)
+    env = os.environ.copy()
+    for name in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS"):
+        env.setdefault(name, "4")
+    for commands in experiment_commands(args):
+        for command in commands:
+            print(shlex.join(command), flush=True)
+            if not args.dry_run:
+                subprocess.run(command, cwd=REPO, env=env, check=True)
+    summary = [sys.executable, str(REPO / "summarize_local_baselines.py"),
+               "--root", str(args.result_root), "--dataset", "ego4d"]
+    print(shlex.join(summary), flush=True)
+    if not args.dry_run:
+        subprocess.run(summary, cwd=REPO, env=env, check=True)
+
+
+if __name__ == "__main__":
+    main()

test_encoder_only_workflow.py

@@ -0,0 +1,233 @@
+"""CPU checks for encoder-only packing and the unchanged four-epoch head."""
+
+from __future__ import annotations
+
+import contextlib
+import copy
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import numpy as np
+import torch
+
+from easycom_head import load_session_data
+from local_turn_features import MODALITIES
+import prepare_encoder_only_features as prepare
+import run_ego4d_encoder_only as runner
+from test_local_turn_features import fixture
+
+
+class EncoderOnlyWorkflowTests(unittest.TestCase):
+    def setUp(self):
+        torch.set_num_threads(1)
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.names = [f"00000000-0000-0000-0000-{i:012d}" for i in (4, 1, 5)]
+        self.splits = ("train", "test", "train")
+        self.source = self.root / "local"
+        self.targets = self.root / "ego4d_target_frame"
+        self.targets.mkdir()
+        self.manifest = self.root / "ego4d_egospeak.tsv"
+        self.manifest.write_text("split\tuid\n" + "".join(
+            f"{split}\t{name}\n" for split, name in zip(self.splits, self.names)))
+        self.paths, self.payloads = {}, {}
+        for name, split in zip(self.names, self.splits):
+            # No global feature directory, raw video, annotation or model exists.
+            _, side, labels = fixture(name=name, dataset="ego4d")
+            side["reference_features"] = "/must-not-be-read/thinker.pt"
+            side["states"] = "POISON: never use this field as the encoder input"
+            path = self.source / split / f"{name}.pt"
+            path.parent.mkdir(parents=True, exist_ok=True)
+            torch.save(side, path)
+            np.save(self.targets / f"{name}.npy", labels)
+            self.paths[name], self.payloads[name] = path, side
+
+    def output_root(self, modality="av"):
+        return self.root / f"encoder_{modality}"
+
+    def prepare_args(self, modality="av", extra=()):
+        return ["--data-root", str(self.root), "--local-feature-root", str(self.source),
+                "--output-root", str(self.output_root(modality)), "--modalities", modality,
+                "--no-progress", *extra]
+
+    def runner_args(self, modality="av", extra=()):
+        return ["--data-root", str(self.root), "--feature-root", str(self.output_root(modality)),
+                "--modalities", modality, "--checkpoint-root", str(self.root / "checkpoints"),
+                "--result-root", str(self.root / "results"), "--seeds", "0", "--device", "cpu",
+                "--no-progress", *extra]
+
+    def export(self, modality="av"):
+        with contextlib.redirect_stdout(io.StringIO()):
+            prepare.main(self.prepare_args(modality))
+
+    def test_pack_all_modalities_uses_only_native_vectors_and_validity(self):
+        name = self.names[0]
+        for modality, parts in MODALITIES.items():
+            with self.subTest(modality=modality):
+                result, contract = prepare.pack_sample(
+                    self.paths[name], self.targets / f"{name}.npy", name, modality)
+                original = self.payloads[name]
+                expected = torch.cat([tensor for part in parts for tensor in (
+                    original[part].float(), original[f"{part}_valid"][:, None].float(),
+                )], dim=-1).to(torch.bfloat16)
+                self.assertTrue(torch.equal(result["states"], expected))
+                self.assertTrue(torch.equal(result["times"], original["times"]))
+                self.assertTrue(torch.equal(result["label_indices"], original["label_indices"]))
+                self.assertEqual(result["mode"], f"encoder-only-{modality}")
+                self.assertIs(result["config"]["uses_thinker_states"], False)
+                self.assertEqual(contract["modalities"], modality)
+
+    def test_export_real_files_full_timeline_and_refuse_overwrite(self):
+        self.export()
+        output = self.output_root()
+        self.assertEqual((output / "complete.txt").read_text(), "3 clips\n")
+        manifest = json.loads((output / "export_manifest.json").read_text())
+        self.assertEqual(manifest["num_states"], 60)
+        self.assertEqual(manifest["input_width"], 13)
+        self.assertEqual(manifest["clips"], dict.fromkeys(self.names, 20))
+        for name in self.names:
+            path = output / f"{name}.pt"
+            self.assertFalse(path.is_symlink())
+            data = load_session_data(output, self.targets, name, dataset="ego4d", expected_mode="encoder-only-av")
+            self.assertEqual(data.states.shape, (20, 13))
+            self.assertIsNone(data.local_features)
+        before = (output / f"{self.names[0]}.pt").read_bytes()
+        with self.assertRaises(FileExistsError):
+            prepare.main(self.prepare_args())
+        self.assertEqual((output / f"{self.names[0]}.pt").read_bytes(), before)
+
+    def test_dry_run_does_not_write(self):
+        with patch.object(prepare.torch, "save") as save, contextlib.redirect_stdout(io.StringIO()):
+            prepare.main(self.prepare_args(extra=["--dry-run"]))
+            save.assert_not_called()
+        self.assertFalse(self.output_root().exists())
+
+    def test_corrupt_test_sidecars_rejected_before_output_creation(self):
+        name = self.names[1]
+        for issue in ("partial", "identity", "times", "indices", "contract", "nan", "invalid_mask"):
+            side = copy.deepcopy(self.payloads[name])
+            if issue == "partial":
+                side["partial"] = True
+            elif issue == "identity":
+                side["sample_id"] = self.names[0]
+            elif issue == "times":
+                side["times"][0] = .4
+            elif issue == "indices":
+                side["label_indices"][0] = 1
+            elif issue == "contract":
+                side["config"]["encoder_id"] = "different-encoder"
+            elif issue == "nan":
+                side["audio"][0, 0] = float("nan")
+            else:
+                side["visual"][0, 0] = 1  # Invalid visual rows must remain zero.
+            torch.save(side, self.paths[name])
+            with self.subTest(issue=issue), contextlib.redirect_stdout(io.StringIO()), self.assertRaises(ValueError):
+                prepare.main(self.prepare_args())
+            self.assertFalse(self.output_root().exists())
+
+    def test_target_mismatch_and_missing_sidecar_fail_before_writes(self):
+        name = self.names[1]
+        np.save(self.targets / f"{name}.npy", np.zeros((19, 3)))
+        with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ValueError, "feature/target mismatch"):
+            prepare.main(self.prepare_args())
+        self.assertFalse(self.output_root().exists())
+        self.paths[name].unlink()
+        with self.assertRaises(FileNotFoundError):
+            prepare.main(self.prepare_args())
+
+    def test_interrupted_export_has_no_complete_marker_or_temporary_file(self):
+        with patch.object(prepare.torch, "save", side_effect=RuntimeError("test write failure")), \
+             contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(RuntimeError, "write failure"):
+            prepare.main(self.prepare_args())
+        self.assertFalse((self.output_root() / "complete.txt").exists())
+        self.assertEqual(list(self.output_root().glob("*.tmp")), [])
+
+    def test_runner_commands_and_dry_run_never_select_thinker_or_dev(self):
+        self.export()
+        args = runner.parse_args(self.runner_args(extra=["--seeds", "0", "1", "2"]))
+        commands = list(runner.experiment_commands(args))
+        self.assertEqual(len(commands), 3)
+        for seed, (train, infer) in enumerate(commands):
+            self.assertEqual(train[train.index("--epochs") + 1], "4")
+            self.assertEqual(train[train.index("--seed") + 1], str(seed))
+            self.assertEqual(train[train.index("--warmup-epochs") + 1], "1.6")
+            for command in (train, infer):
+                for forbidden in ("--local-feature-root", "--layer-fusion", "--dev-sessions", "--sessions"):
+                    self.assertNotIn(forbidden, command)
+                self.assertEqual(command[command.index("--expected-mode") + 1], "encoder-only-av")
+        with patch.object(runner.subprocess, "run") as run, contextlib.redirect_stdout(io.StringIO()):
+            runner.main(self.runner_args(extra=["--dry-run"]))
+            run.assert_not_called()
+        self.assertFalse((self.root / "checkpoints").exists())
+
+    def test_runner_rejects_missing_test_features_marker_and_wrong_contract(self):
+        self.export()
+        args = runner.parse_args(self.runner_args())
+        marker = self.output_root() / "complete.txt"
+        marker.write_text("2 clips\n")
+        with self.assertRaisesRegex(ValueError, "incomplete"):
+            runner.preflight(args)
+        marker.write_text("3 clips\n")
+        path = self.output_root() / f"{self.names[1]}.pt"
+        payload = torch.load(path, weights_only=False)
+        payload["config"]["uses_thinker_states"] = True
+        torch.save(payload, path)
+        with self.assertRaisesRegex(ValueError, "encoder-only export"):
+            runner.preflight(args)
+        path.unlink()
+        with self.assertRaises(FileNotFoundError):
+            runner.preflight(args)
+
+    def test_invalid_cli_and_existing_outputs_are_rejected(self):
+        for extra in (["--seeds", "0", "0"], ["--seeds", "-1"], ["--modalities", "thinker"]):
+            with self.subTest(extra=extra), contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+                runner.parse_args(self.runner_args(extra=extra))
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
+            prepare.parse_args(self.prepare_args(extra=["--output-root", str(self.source)]))
+        self.export()
+        args = runner.parse_args(self.runner_args(extra=["--seeds", "0", "1"]))
+        checkpoint, _ = runner.run_paths(args, 1)
+        checkpoint.parent.mkdir()
+        checkpoint.write_bytes(b"preserve")
+        with self.assertRaises(FileExistsError):
+            runner.preflight(args)
+        self.assertEqual(checkpoint.read_bytes(), b"preserve")
+
+    def test_cpu_round_trip_all_three_modalities_four_epochs_without_dev(self):
+        env = os.environ.copy()
+        env.update(OMP_NUM_THREADS="1", MKL_NUM_THREADS="1", OPENBLAS_NUM_THREADS="1")
+        for modality, parts in MODALITIES.items():
+            with self.subTest(modality=modality):
+                self.export(modality)
+                command = [sys.executable, str(Path(runner.__file__).resolve()), *self.runner_args(modality)]
+                result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=120)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                args = runner.parse_args(self.runner_args(modality))
+                checkpoint, output = runner.run_paths(args, 0)
+                saved = torch.load(checkpoint, map_location="cpu", weights_only=False)
+                self.assertEqual(saved["epoch"], 4)
+                self.assertEqual(saved["dev_sessions"], [])
+                self.assertEqual(set(saved["train_sessions"]), {self.names[0], self.names[2]})
+                self.assertIsNone(saved["local_contract"])
+                width = sum(self.payloads[self.names[0]]["config"][f"{part}_dim"] + 1 for part in parts)
+                self.assertEqual(saved["head_config"]["hidden_size"], width)
+                self.assertEqual(saved["head_config"]["local_visual_dim"], 0)
+                self.assertEqual(saved["head_config"]["local_audio_dim"], 0)
+                self.assertEqual(saved["head_config"]["layer_fusion"], "single")
+                self.assertFalse(any("local_projections" in key for key in saved["head"]))
+                metrics = json.loads((output / "metrics.json").read_text())
+                self.assertEqual(metrics["sessions"], [self.names[1]])
+                self.assertEqual(metrics["checkpoint_epoch"], 4)
+                self.assertEqual(np.load(output / "pred" / f"{self.names[1]}.npy").shape, (20, 3, 10))
+
+
+if __name__ == "__main__":
+    unittest.main()