Add TF009 Docker integration references and fixed runtime

pipelines/Tr006_V2/docker/segment-algorithm/inference.py

@@ -97,6 +97,7 @@ fully local and does not download model files.
 
 import json
 import logging
+import re
 import sys
 import time
 from pathlib import Path
@@ -175,6 +176,156 @@ SYSTEM_PROMPT = (
 )
 
 
+def _flatten_token_ids(value) -> list[int]:
+    """Flatten tokenizer/config token-id values while dropping invalid entries."""
+    if value is None or isinstance(value, bool):
+        return []
+    if isinstance(value, (int, np.integer)):
+        return [int(value)] if int(value) >= 0 else []
+    if hasattr(value, "detach"):
+        value = value.detach().cpu().tolist()
+    elif hasattr(value, "tolist") and not isinstance(value, (str, bytes)):
+        value = value.tolist()
+    if isinstance(value, (list, tuple, set)):
+        out = []
+        for item in value:
+            out.extend(_flatten_token_ids(item))
+        return out
+    return []
+
+
+def _ordered_unique(values: list[int]) -> list[int]:
+    return list(dict.fromkeys(int(value) for value in values if int(value) >= 0))
+
+
+def resolve_eos_token_ids(model, tokenizer) -> list[int]:
+    """Merge model and tokenizer EOS values, preserving order and removing duplicates."""
+    values = []
+    for owner in (
+        getattr(model, "generation_config", None),
+        getattr(model, "config", None),
+        tokenizer,
+    ):
+        values.extend(_flatten_token_ids(getattr(owner, "eos_token_id", None)))
+    return _ordered_unique(values)
+
+
+def resolve_im_start_token_ids(tokenizer) -> list[int]:
+    """Return the tokenizer's (possibly multi-token) ``<|im_start|>`` sequence."""
+    try:
+        values = _flatten_token_ids(tokenizer.encode("<|im_start|>", add_special_tokens=False))
+    except Exception:
+        values = []
+    if values:
+        return values
+    try:
+        value = tokenizer.convert_tokens_to_ids("<|im_start|>")
+    except Exception:
+        value = None
+    return _flatten_token_ids(value)
+
+
+def _token_rows(value) -> list[list[int]]:
+    if hasattr(value, "detach"):
+        value = value.detach().cpu().tolist()
+    elif hasattr(value, "tolist") and not isinstance(value, (str, bytes)):
+        value = value.tolist()
+    if not isinstance(value, list):
+        value = [value]
+    if value and not isinstance(value[0], list):
+        value = [value]
+    return [[int(token) for token in row] for row in value]
+
+
+def extract_new_token_ids(generated, input_ids, attention_mask=None) -> list[list[int]]:
+    """Extract only generated IDs, accounting for padded prompt layouts."""
+    outputs = _token_rows(generated)
+    prompts = _token_rows(input_ids)
+    masks = _token_rows(attention_mask) if attention_mask is not None else []
+    if not prompts:
+        return outputs
+    extracted = []
+    for index, output in enumerate(outputs):
+        prompt = prompts[min(index, len(prompts) - 1)]
+        mask = masks[min(index, len(masks) - 1)] if masks else [1] * len(prompt)
+        active_prompt = [token for token, flag in zip(prompt, mask) if flag]
+        if len(output) >= len(prompt) and output[: len(prompt)] == prompt:
+            offset = len(prompt)
+        elif active_prompt and len(output) >= len(active_prompt) and output[: len(active_prompt)] == active_prompt:
+            offset = len(active_prompt)
+        else:
+            offset = 0
+        extracted.append(output[offset:])
+    return extracted
+
+
+def _first_subsequence(values: list[int], needle: list[int]) -> int | None:
+    if not needle or len(needle) > len(values):
+        return None
+    width = len(needle)
+    for index in range(len(values) - width + 1):
+        if values[index : index + width] == needle:
+            return index
+    return None
+
+
+def trim_generated_token_ids(
+    raw_token_ids,
+    *,
+    eos_token_ids,
+    im_start_token_ids,
+    max_new_tokens: int,
+) -> dict:
+    """Stop at the earliest EOS or im_start sequence and retain a raw-token audit."""
+    raw = _flatten_token_ids(raw_token_ids)
+    eos_ids = set(_flatten_token_ids(eos_token_ids))
+    im_start = _flatten_token_ids(im_start_token_ids)
+    eos_positions = [index for index, token in enumerate(raw) if token in eos_ids]
+    eos_index = eos_positions[0] if eos_positions else None
+    im_start_index = _first_subsequence(raw, im_start)
+    candidates = [(index, "eos") for index in (eos_index,) if index is not None]
+    candidates.extend((index, "im_start") for index in (im_start_index,) if index is not None)
+    stop_index, stop_reason = min(candidates, default=(None, None), key=lambda item: (item[0], 0 if item[1] == "eos" else 1))
+    hit_max = stop_index is None and len(raw) >= int(max_new_tokens)
+    if stop_index is None:
+        stop_reason = "max_new_tokens" if hit_max else "completed"
+        token_ids = raw
+    else:
+        token_ids = raw[:stop_index]
+    return {
+        "raw_token_ids": raw,
+        "token_ids": token_ids,
+        "stop_index": stop_index,
+        "eos_position": eos_index,
+        "eos_positions": eos_positions,
+        "stop_reason": stop_reason,
+        "hit_max_new_tokens": hit_max,
+    }
+
+
+_LEADING_THINK = re.compile(r"\A\s*<think>.*?</think>\s*", re.IGNORECASE | re.DOTALL)
+_TRAILING_SPECIAL = re.compile(r"(?:<\|[^>\n]+\|>|</think>|<think>)", re.IGNORECASE)
+_ROLE_MARKER = re.compile(r"\A\s*(?:assistant|user|system)\s*(?::|\Z)", re.IGNORECASE)
+
+
+def clean_decoded_answer(text: str) -> str:
+    """Remove only a complete leading think block and trailing template pollution."""
+    value = str(text or "").strip()
+    leading = _LEADING_THINK.match(value)
+    if leading:
+        value = value[leading.end() :].strip()
+    elif value.lower().startswith("<think>"):
+        return value
+    value = value.splitlines()[0].strip() if value else ""
+    marker = _TRAILING_SPECIAL.search(value)
+    if marker:
+        value = value[: marker.start()].rstrip()
+    role = _ROLE_MARKER.match(value)
+    if role:
+        value = ""
+    return value
+
+
 def log_environment(device: torch.device) -> None:
     """Report the CUDA stack this container ended up with — once, at startup.
 
@@ -321,7 +472,7 @@ def your_preprocessing(video_path: Path, req: Request) -> tuple[np.ndarray, dict
     return frames, metadata
 
 
-def answer_one(model, processor, req: Request) -> str:
+def answer_one(model, processor, req: Request) -> tuple[str, dict]:
     """Run the merged Qwen3.5 checkpoint on one official request."""
     frames, metadata = your_preprocessing(clip_path_for(req), req)
     messages = [
@@ -340,18 +491,41 @@ def answer_one(model, processor, req: Request) -> str:
     )
     model_device = next(model.parameters()).device
     inputs = inputs.to(model_device)
+    tokenizer = getattr(processor, "tokenizer", processor)
+    eos_token_ids = resolve_eos_token_ids(model, tokenizer)
+    im_start_token_ids = resolve_im_start_token_ids(tokenizer)
+    generation_kwargs = {"max_new_tokens": MAX_NEW_TOKENS, "do_sample": False}
+    if eos_token_ids:
+        generation_kwargs["eos_token_id"] = eos_token_ids if len(eos_token_ids) > 1 else eos_token_ids[0]
     with torch.inference_mode():
-        generated = model.generate(**inputs, max_new_tokens=MAX_NEW_TOKENS, do_sample=False)
-    new_tokens = generated[:, inputs["input_ids"].shape[1] :]
-    answer = processor.batch_decode(
-        new_tokens, skip_special_tokens=True, clean_up_tokenization_spaces=False
-    )[0].strip()
-    # The official evaluator consumes the content verbatim; only remove the
-    # optional reasoning wrapper if a checkpoint emits it despite disabled
-    # thinking, preserving all answer text otherwise.
-    if "</think>" in answer:
-        answer = answer.split("</think>", 1)[1].strip()
-    return answer
+        generated = model.generate(**inputs, **generation_kwargs)
+    if hasattr(generated, "sequences"):
+        generated = generated.sequences
+    new_token_rows = extract_new_token_ids(
+        generated,
+        inputs["input_ids"],
+        inputs.get("attention_mask"),
+    )
+    trimmed = trim_generated_token_ids(
+        new_token_rows[0] if new_token_rows else [],
+        eos_token_ids=eos_token_ids,
+        im_start_token_ids=im_start_token_ids,
+        max_new_tokens=MAX_NEW_TOKENS,
+    )
+    decoded = processor.batch_decode(
+        [trimmed["token_ids"]], skip_special_tokens=False, clean_up_tokenization_spaces=False
+    )[0]
+    answer = clean_decoded_answer(decoded)
+    audit = {
+        "raw_token_ids": trimmed["raw_token_ids"],
+        "stop_index": trimmed["stop_index"],
+        "eos_position": trimmed["eos_position"],
+        "eos_positions": trimmed["eos_positions"],
+        "stop_reason": trimmed["stop_reason"],
+        "hit_max_new_tokens": trimmed["hit_max_new_tokens"],
+        "new_token_count": len(trimmed["raw_token_ids"]),
+    }
+    return answer, audit
 
 
 def run() -> int:
@@ -415,6 +589,7 @@ def run() -> int:
     # ── Inference — one pass per question ──────────────────────────────────────
     log.info("--- Running inference over %d question(s) ---", len(requests))
     responses = []
+    generation_audits = []
     n_failed = 0
     t_batch = time.monotonic()
 
@@ -425,19 +600,37 @@ def run() -> int:
         )
         t0 = time.monotonic()
         try:
-            answer = answer_one(model, processor, req)
+            answer, audit = answer_one(model, processor, req)
         except Exception:
             # Keep going: one failed question should not cost you the rest of the
             # batch. An empty answer is simply scored incorrect.
             n_failed += 1
             log.exception("[%d/%d] qID=%s failed; emitting empty answer", i, len(requests), req.qID)
             answer = ""
+            audit = {
+                "raw_token_ids": [],
+                "stop_index": None,
+                "eos_position": None,
+                "eos_positions": [],
+                "stop_reason": "error",
+                "hit_max_new_tokens": False,
+                "new_token_count": 0,
+            }
         latency = time.monotonic() - t0
 
         # Informational only. The platform times the run itself and scores against
         # that measurement — see "The latency budget" in the module docstring.
         responses.append(Response(qID=req.qID, content=answer, latency=latency))
-        log.info("[%d/%d] qID=%s answered in %.4f s: %r", i, len(requests), req.qID, latency, answer)
+        generation_audits.append({"qID": req.qID, **audit})
+        log.info(
+            "[%d/%d] qID=%s answered in %.4f s answer_len=%d stop_reason=%s",
+            i,
+            len(requests),
+            req.qID,
+            latency,
+            len(answer),
+            audit["stop_reason"],
+        )
 
     batch_seconds = time.monotonic() - t_batch
     log.info(
@@ -455,6 +648,9 @@ def run() -> int:
     output_path = OUTPUT_PATH / "answer.json"
     save_items(responses, output_path)
     log.info("Wrote %d response(s) to %s", len(responses), output_path)
+    audit_path = OUTPUT_PATH / "generation_audit.json"
+    audit_path.write_text(json.dumps(generation_audits, ensure_ascii=False, indent=2) + "\n")
+    log.info("Wrote generation audit for %d response(s) to %s", len(generation_audits), audit_path)
     log.info("=== inference done in %.2f s total ===", time.monotonic() - t_start)
     return 0
 

pipelines/Tr006_V2/tf009/test_cfff_docker_runtime.py

@@ -0,0 +1,221 @@
+"""Deterministic CPU regression gate for the CFFF Docker generation runtime."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+import types
+import unittest
+from pathlib import Path
+
+import torch
+
+ROOT = Path(__file__).resolve().parents[1]
+LOCAL_RUNTIME = (
+    ROOT
+    / "experiments/Qwen3.5-9B/debug/debug022_segment_v2_online_preprocess_hf_sync_20260902"
+    / "results/staging/pipelines/Tr006_V2/docker/segment-algorithm/inference.py"
+)
+HUB_RUNTIME = ROOT / "docker/segment-algorithm/inference.py"
+RUNTIME = Path(os.environ["CFFF_RUNTIME"]) if os.environ.get("CFFF_RUNTIME") else (
+    HUB_RUNTIME if HUB_RUNTIME.is_file() else LOCAL_RUNTIME
+)
+
+
+def _load_runtime():
+    # The package's ``focus`` dependency is supplied by the official image.  Stub
+    # its import-only symbols so this gate remains a seconds-level CPU test.
+    focus = types.ModuleType("focus")
+    focus.Request = object
+    focus.Response = object
+    focus.load_requests = lambda _: []
+    focus.save_items = lambda *_: None
+    old_focus = sys.modules.get("focus")
+    sys.modules["focus"] = focus
+    try:
+        spec = importlib.util.spec_from_file_location("cfff_runtime_under_test", RUNTIME)
+        module = importlib.util.module_from_spec(spec)
+        assert spec and spec.loader
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        if old_focus is None:
+            sys.modules.pop("focus", None)
+        else:
+            sys.modules["focus"] = old_focus
+
+
+class FakeTokenizer:
+    eos_token_id = 248046
+    all_special_ids = [248044, 248046, 151645, 777, 778]
+
+    def encode(self, value, add_special_tokens=False):
+        assert not add_special_tokens
+        if value == "<|im_start|>":
+            return [777, 778]
+        return []
+
+
+class SingleImStartTokenizer(FakeTokenizer):
+    def encode(self, value, add_special_tokens=False):
+        assert not add_special_tokens
+        return [777] if value == "<|im_start|>" else []
+
+
+class FakeModel:
+    class Config:
+        eos_token_id = 248044
+
+    class GenerationConfig:
+        eos_token_id = 248044
+
+    config = Config()
+    generation_config = GenerationConfig()
+
+    def __init__(self):
+        self.generate_kwargs = None
+
+    def parameters(self):
+        yield types.SimpleNamespace(device=torch.device("cpu"))
+
+    def generate(self, **kwargs):
+        self.generate_kwargs = kwargs
+        return [[9, 10, 30, 248046, 40]]
+
+
+class FakeInputs(dict):
+    def to(self, _device):
+        return self
+
+
+class FakeProcessor(FakeTokenizer):
+    def __init__(self):
+        self.tokenizer = self
+        self.decode_input = None
+
+    def apply_chat_template(self, *args, **kwargs):
+        return "prompt"
+
+    def __call__(self, **kwargs):
+        return FakeInputs(input_ids=[[9, 10]], attention_mask=[[1, 1]])
+
+    def batch_decode(self, token_ids, **kwargs):
+        self.decode_input = token_ids
+        return ["yes"]
+
+
+class RuntimeTests(unittest.TestCase):
+  def test_dynamic_eos_merge_and_im_start_pattern(self):
+    runtime = _load_runtime()
+    tokenizer = FakeTokenizer()
+    model = FakeModel()
+
+    assert runtime.resolve_eos_token_ids(model, tokenizer) == [248044, 248046]
+    assert runtime.resolve_im_start_token_ids(tokenizer) == [777, 778]
+    assert runtime.resolve_eos_token_ids(
+        type("M", (), {"generation_config": type("G", (), {"eos_token_id": [248044, None, 248044]})(), "config": type("C", (), {"eos_token_id": None})()})(),
+        type("T", (), {"eos_token_id": None})(),
+    ) == [248044]
+    assert runtime.resolve_im_start_token_ids(SingleImStartTokenizer()) == [777]
+
+
+  def test_only_new_tokens_with_padded_batch_layout(self):
+    runtime = _load_runtime()
+    generated = [
+        [9, 10, 11, 0, 40, 41],
+        [0, 20, 21, 22, 50, 51],
+    ]
+    input_ids = [[9, 10, 11, 0], [0, 20, 21, 22]]
+    attention_mask = [[1, 1, 1, 0], [0, 1, 1, 1]]
+    assert runtime.extract_new_token_ids(generated, input_ids, attention_mask) == [[40, 41], [50, 51]]
+
+
+  def test_token_stop_and_saturation_audit(self):
+    runtime = _load_runtime()
+    for raw, expected, reason, eos_positions in (
+        ([10, 248046, 11], [10], "eos", [1]),
+        ([10, 777, 778, 11], [10], "im_start", []),
+        ([10, 11, 12], [10, 11, 12], "max_new_tokens", []),
+    ):
+      with self.subTest(reason=reason):
+        audit = runtime.trim_generated_token_ids(
+            raw,
+            eos_token_ids=[248044, 248046],
+            im_start_token_ids=[777, 778],
+            max_new_tokens=3,
+        )
+        self.assertEqual(audit["raw_token_ids"], raw)
+        self.assertEqual(audit["token_ids"], expected)
+        self.assertEqual(audit["stop_reason"], reason)
+        self.assertEqual(audit["eos_positions"], eos_positions)
+        self.assertEqual(audit["hit_max_new_tokens"], reason == "max_new_tokens")
+    earlier_eos = runtime.trim_generated_token_ids(
+        [10, 248046, 777, 778, 11],
+        eos_token_ids=[248044, 248046],
+        im_start_token_ids=[777, 778],
+        max_new_tokens=8,
+    )
+    earlier_im_start = runtime.trim_generated_token_ids(
+        [10, 777, 778, 248046, 11],
+        eos_token_ids=[248044, 248046],
+        im_start_token_ids=[777, 778],
+        max_new_tokens=8,
+    )
+    self.assertEqual((earlier_eos["stop_index"], earlier_eos["stop_reason"]), (1, "eos"))
+    self.assertEqual((earlier_im_start["stop_index"], earlier_im_start["stop_reason"]), (1, "im_start"))
+    completed = runtime.trim_generated_token_ids(
+        [10, 11], eos_token_ids=[248044], im_start_token_ids=[777, 778], max_new_tokens=8
+    )
+    self.assertEqual(completed["stop_reason"], "completed")
+    self.assertFalse(completed["hit_max_new_tokens"])
+
+
+  def test_text_cleanup_only_leading_complete_think_and_trailing_pollution(self):
+    runtime = _load_runtime()
+    assert runtime.clean_decoded_answer("  <think>draft</think>\nyes\nassistant\nnoise") == "yes"
+    assert runtime.clean_decoded_answer("yes\n<think>late reasoning</think>\nassistant\nnoise") == "yes"
+    assert runtime.clean_decoded_answer("yes <|im_end|> assistant") == "yes"
+    assert runtime.clean_decoded_answer("yes\nsecond line") == "yes"
+    assert runtime.clean_decoded_answer("The assistant observes anatomy") == "The assistant observes anatomy"
+    assert runtime.clean_decoded_answer("<think>unfinished") == "<think>unfinished"
+
+
+  def test_public_gt_smoke_strict_matching_without_test_references(self):
+    runtime = _load_runtime()
+    cases = [
+        ([101, 248046, 202], "<think>draft</think>\nyes\nassistant", "yes"),
+        ([101, 777, 778, 202], "no <|im_end|>\nuser", "no"),
+        ([101, 202], "Clip\nassistant\nnoise", "Clip"),
+    ]
+    for raw, decoded, expected in cases:
+        audit = runtime.trim_generated_token_ids(
+            raw,
+            eos_token_ids=[248044, 248046],
+            im_start_token_ids=[777, 778],
+            max_new_tokens=8,
+        )
+        # This tiny public/synthetic batch checks exact decoded answer behavior;
+        # no held-out/test references are read by the gate.
+        text = runtime.clean_decoded_answer(decoded)
+        self.assertEqual(text, expected)
+        self.assertEqual(audit["raw_token_ids"], raw)
+
+  def test_answer_one_passes_eos_and_decodes_trimmed_new_tokens(self):
+    runtime = _load_runtime()
+    model = FakeModel()
+    processor = FakeProcessor()
+    runtime.clip_path_for = lambda _req: Path("unused.mp4")
+    runtime.your_preprocessing = lambda _path, _req: (None, {})
+    answer, audit = runtime.answer_one(model, processor, types.SimpleNamespace(question="Q"))
+    self.assertEqual(model.generate_kwargs["eos_token_id"], [248044, 248046])
+    self.assertEqual(processor.decode_input, [[30]])
+    self.assertEqual(answer, "yes")
+    self.assertEqual(audit["raw_token_ids"], [30, 248046, 40])
+    self.assertEqual(audit["eos_positions"], [1])
+    self.assertEqual(audit["stop_reason"], "eos")
+    self.assertFalse(audit["hit_max_new_tokens"])
+
+
+if __name__ == "__main__":
+    unittest.main()

pipelines/Tr006_V2/tf009/test_tf009_self_consistency.py

@@ -0,0 +1,151 @@
+import copy
+import sys
+import unittest
+from collections import Counter
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from tf009_self_consistency import (  # noqa: E402
+    FORMAT_ORDER,
+    MAX_NEW_TOKENS,
+    SEEDS,
+    TEMPERATURE,
+    TOP_P,
+    _prepare_request,
+    aggregate_numeric,
+    aggregate_prediction,
+    aggregate_time,
+    merge_predictions,
+    normalize_for_majority,
+    parse_strict_number,
+    parse_strict_time_list,
+    select_target_rows,
+)
+
+
+def row(fmt: str, rid: str, question: str = "Question") -> dict:
+    return {
+        "id": rid,
+        "answer_format": fmt,
+        "question": question,
+        "messages": [
+            {"role": "system", "content": "system"},
+            {"role": "user", "content": "<video>\n" + question},
+        ],
+    }
+
+
+class TF009SelfConsistencyTest(unittest.TestCase):
+    def test_prepared_request_packs_video_metadata(self):
+        frames = object()
+        metadata = {
+            "fps": 5.0,
+            "frames_indices": list(range(80)),
+            "video_backend": "precomputed_absolute_pts",
+            "total_num_frames": 80,
+        }
+        request_row = {
+            "id": "video-row",
+            "videos": ["clip.mp4"],
+            "messages": [{"role": "user", "content": "<video>"}],
+            "chat_template_kwargs": {"video_metadata": metadata},
+        }
+
+        class Tokenizer:
+            def apply_chat_template(self, *args, **kwargs):
+                return "prompt"
+
+        with patch(
+            "eval_segmentqa_vllm.fixed_absmeta_frames", return_value=(frames, metadata)
+        ), patch("eval_segmentqa_vllm.make_messages", return_value=[]):
+            request = _prepare_request(request_row, Tokenizer(), 80)
+
+        video = request["multi_modal_data"]["video"]
+        self.assertIsInstance(video, tuple)
+        self.assertIs(video[0], frames)
+        self.assertEqual(video[1], metadata)
+
+    def test_frozen_sampling_policy_and_format_stratified_smoke(self):
+        self.assertEqual(SEEDS, (42, 43, 44))
+        self.assertEqual(TEMPERATURE, 0.2)
+        self.assertEqual(TOP_P, 0.95)
+        self.assertEqual(MAX_NEW_TOKENS, 64)
+        rows = [row(fmt, str(index)) for index, fmt in enumerate(FORMAT_ORDER)]
+        rows.append(row("binary", "extra"))
+        selected = select_target_rows(rows, smoke=7)
+        self.assertEqual({item["answer_format"] for item in selected}, set(FORMAT_ORDER))
+        self.assertEqual(len(selected), 7)
+        with self.assertRaises(ValueError):
+            select_target_rows(rows, smoke=6)
+
+    def test_strict_numeric_and_percentage_parse(self):
+        self.assertEqual(parse_strict_number(" 12.50 "), 12.50)
+        self.assertEqual(parse_strict_number("42%", percentage=True), 42)
+        self.assertIsNone(parse_strict_number("12 items"))
+        self.assertIsNone(parse_strict_number("1,000"))
+        self.assertIsNone(parse_strict_number("nan"))
+        number = row("number", "n")
+        final, audit = aggregate_numeric(number, ["1", "3", "2"])
+        self.assertEqual(final, "2")
+        self.assertEqual(audit["aggregation"], "median")
+        pct = row("percentage", "p")
+        self.assertEqual(aggregate_prediction(pct, ["10%", "30", "20%"], "99")[0], "20")
+        fallback, fallback_audit = aggregate_prediction(number, ["1", "bad", "3"], "8")
+        self.assertEqual(fallback, "8")
+        self.assertTrue(fallback_audit["fallback"])
+
+    def test_time_requires_three_valid_equal_length_lists_and_medians(self):
+        self.assertEqual(parse_strict_time_list("1:02:03, 00:00:04"), [3723, 4])
+        self.assertIsNone(parse_strict_time_list("01:60:00"))
+        self.assertIsNone(parse_strict_time_list("00:00:01 because"))
+        final, audit = aggregate_time(["00:00:01, 00:00:10", "00:00:03, 00:00:08", "00:00:02, 00:00:12"])
+        self.assertEqual(final, "00:00:02, 00:00:10")
+        self.assertFalse(audit["fallback"])
+        mismatch, mismatch_audit = aggregate_time(["00:00:01", "00:00:02, 00:00:03", "00:00:02"])
+        self.assertIsNone(mismatch)
+        self.assertEqual(mismatch_audit["fallback_reason"], "time_point_count_mismatch")
+        time_row = row("time", "t")
+        self.assertEqual(aggregate_prediction(time_row, ["00:00:01", "bad", "00:00:03"], "00:00:09")[0], "00:00:09")
+
+    def test_majority_normalizes_non_numeric_formats_and_falls_back_on_tie(self):
+        binary = row("binary", "b")
+        self.assertEqual(normalize_for_majority(binary, "Final answer: YES."), "yes")
+        final, audit = aggregate_prediction(binary, ["yes", "no", "yes"], "no")
+        self.assertEqual(final, "yes")
+        self.assertEqual(audit["selected"], "sample_majority")
+        tie, tie_audit = aggregate_prediction(binary, ["yes", "no", "maybe"], "no")
+        self.assertEqual(tie, "no")
+        self.assertTrue(tie_audit["fallback"])
+
+        mc = row("multiple_choice", "mc", "Please select one answer: top/left; top/right; bottom/left; bottom/right")
+        self.assertEqual(normalize_for_majority(mc, "top/left"), "top/left")
+        self.assertEqual(aggregate_prediction(mc, ["top/left", "top/left", "bottom/right"], "bottom/right")[0], "top/left")
+        open_ended = row("open_ended", "o")
+        self.assertEqual(aggregate_prediction(open_ended, ["short", "short", "other"], "baseline")[0], "short")
+
+    def test_merge_is_complete_and_outside_target_is_frozen(self):
+        rows = [row(fmt, str(index)) for index, fmt in enumerate(FORMAT_ORDER)]
+        baseline = {item["id"]: {"id": item["id"], "prediction": "baseline", "latency": 0.1} for item in rows}
+        generated = {
+            "0": {
+                "id": "0",
+                "prediction": "yes",
+                "raw_predictions": ["yes", "yes", "no"],
+                "audit": {"views": {"A": {}, "B": {}, "C": {}}, "selected": "sample_majority"},
+            }
+        }
+        before = copy.deepcopy(baseline)
+        merged, audit = merge_predictions(rows, baseline, generated, {"0"})
+        self.assertEqual(len(merged), 7)
+        self.assertEqual(len(audit), 7)
+        self.assertEqual(merged[0]["prediction"], "yes")
+        self.assertEqual(merged[0]["generation_seeds"], list(SEEDS))
+        self.assertEqual(merged[1], before["1"])
+        self.assertEqual(sum(item["target"] for item in audit), 1)
+        self.assertEqual(Counter(item["id"] for item in audit), Counter(item["id"] for item in rows))
+
+
+if __name__ == "__main__":
+    unittest.main()

pipelines/Tr006_V2/tf009/tf009_self_consistency.py

@@ -0,0 +1,790 @@
+#!/usr/bin/env python3
+"""Reference-free TF009 self-consistency runner.
+
+Each selected fixed-80 request is generated three times from the same merged
+Tr006 checkpoint with seeds 42/43/44, temperature 0.2, top-p 0.95, and 64
+new-token budget.  A format-aware, reference-free reducer selects a result or
+keeps the frozen baseline.  Rows outside the selected (or smoke) target set
+are copied byte-for-byte from the baseline prediction file.
+"""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import json
+import math
+import os
+import platform
+import re
+import shlex
+import subprocess
+import sys
+import time
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor
+from decimal import Decimal, InvalidOperation
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+from training_free_postprocess import (
+    FOREIGN_OBJECT_CLASSES,
+    SUPPORTED_FORMATS,
+    normalize_prediction,
+)
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_NFRAMES = 80
+EXPECTED_OFFICIAL_COUNT = 6254
+FORMAT_ORDER = (
+    "binary",
+    "number",
+    "percentage",
+    "time",
+    "multiple_choice",
+    "fo_class",
+    "open_ended",
+)
+SEEDS = (42, 43, 44)
+TEMPERATURE = 0.2
+TOP_P = 0.95
+MAX_NEW_TOKENS = 64
+
+_THINK_RE = re.compile(r"<think>.*?</think>", re.I | re.S)
+_NUMBER_RE = re.compile(r"[+-]?(?:\d+(?:\.\d+)?|\.\d+)")
+_PERCENT_RE = re.compile(r"[+-]?(?:\d+(?:\.\d+)?|\.\d+)%")
+_HMS_RE = re.compile(r"\d{1,6}:\d{2}:\d{2}")
+_ANSWER_PREFIX_RE = re.compile(
+    r"^(?:(?:the\s+)?(?:final\s+)?(?:answer|response|result)\s*(?:(?:is|=)\s*|:\s*)|(?:final\s+)?answer\s*-\s*)",
+    re.I,
+)
+
+
+def read_jsonl(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        raise FileNotFoundError(path)
+    rows: list[dict[str, Any]] = []
+    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
+        if not line.strip():
+            continue
+        value = json.loads(line)
+        if not isinstance(value, dict):
+            raise ValueError(f"{path}:{line_number} is not a JSON object")
+        rows.append(value)
+    return rows
+
+
+def write_json(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def write_jsonl(path: Path, rows: Sequence[Mapping[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8")
+
+
+def resolve_path(value: str) -> Path:
+    path = Path(value)
+    return path if path.is_absolute() else PROJECT_ROOT / path
+
+
+def row_id(row: Mapping[str, Any]) -> str:
+    value = row.get("id")
+    if value is None:
+        nested = row.get("segment_metadata")
+        if isinstance(nested, Mapping):
+            value = nested.get("id")
+    if value is None:
+        raise ValueError("row is missing id and segment_metadata.id")
+    return str(value)
+
+
+def answer_format(row: Mapping[str, Any]) -> str | None:
+    value = row.get("answer_format")
+    if value is None:
+        nested = row.get("segment_metadata")
+        if isinstance(nested, Mapping):
+            value = nested.get("answer_format")
+    if value is None:
+        return None
+    value = str(value).strip().lower()
+    return value if value in SUPPORTED_FORMATS else value
+
+
+def select_target_rows(
+    rows: list[dict[str, Any]], smoke: int | None = None
+) -> list[dict[str, Any]]:
+    """Select all formats, or one deterministic row per format plus fill."""
+
+    candidates = [row for row in rows if answer_format(row) in FORMAT_ORDER]
+    if smoke is None:
+        return candidates
+    if smoke < 1:
+        raise ValueError("--smoke must be positive")
+    if smoke < len(FORMAT_ORDER):
+        raise ValueError(f"--smoke={smoke} is smaller than seven format strata")
+    if smoke > len(candidates):
+        raise ValueError(f"--smoke={smoke} exceeds candidate rows={len(candidates)}")
+    selected: list[dict[str, Any]] = []
+    selected_ids: set[str] = set()
+    for fmt in FORMAT_ORDER:
+        for row in candidates:
+            if answer_format(row) == fmt:
+                selected.append(row)
+                selected_ids.add(row_id(row))
+                break
+        else:
+            raise ValueError(f"smoke target coverage missing answer_format={fmt}")
+    for row in candidates:
+        if len(selected) >= smoke:
+            break
+        if row_id(row) not in selected_ids:
+            selected.append(row)
+            selected_ids.add(row_id(row))
+    return selected
+
+
+def _clean_text(value: Any, remove_prefix: bool = True) -> str:
+    text = _THINK_RE.sub("", str(value if value is not None else ""))
+    text = re.sub(r"\s+", " ", text).strip()
+    if remove_prefix:
+        for _ in range(3):
+            match = _ANSWER_PREFIX_RE.match(text)
+            if not match:
+                break
+            text = text[match.end() :].strip()
+    return text
+
+
+def _canonical_decimal(value: Decimal) -> str:
+    if not value.is_finite():
+        raise ValueError("non-finite number")
+    if value == 0:
+        return "0"
+    rendered = format(value, "f")
+    if "." in rendered:
+        rendered = rendered.rstrip("0").rstrip(".")
+    if rendered.startswith("+"):
+        rendered = rendered[1:]
+    return rendered
+
+
+def parse_strict_number(value: Any, percentage: bool = False) -> Decimal | None:
+    """Parse one complete numeric answer, rejecting prose and extra tokens."""
+
+    text = _clean_text(value)
+    if not text or " " in text or "," in text:
+        return None
+    if percentage:
+        if _PERCENT_RE.fullmatch(text):
+            text = text[:-1]
+        elif not _NUMBER_RE.fullmatch(text):
+            return None
+    elif not _NUMBER_RE.fullmatch(text) or "%" in text:
+        return None
+    try:
+        number = Decimal(text)
+    except (InvalidOperation, ValueError):
+        return None
+    return number if number.is_finite() else None
+
+
+def parse_strict_time_list(value: Any) -> list[int] | None:
+    """Parse an exact comma-separated HMS list into seconds."""
+
+    text = _clean_text(value)
+    if not text:
+        return None
+    parts = [part.strip() for part in text.split(",")]
+    if any(not part or _HMS_RE.fullmatch(part) is None for part in parts):
+        return None
+    seconds: list[int] = []
+    for part in parts:
+        match = _HMS_RE.fullmatch(part)
+        assert match is not None
+        hours, minutes, remainder = part.split(":")
+        minute, second = int(minutes), int(remainder)
+        if minute >= 60 or second >= 60:
+            return None
+        seconds.append(int(hours) * 3600 + minute * 60 + second)
+    return seconds
+
+
+def format_hms(seconds: int) -> str:
+    if seconds < 0:
+        raise ValueError("HMS seconds must be non-negative")
+    return f"{seconds // 3600:02d}:{(seconds % 3600) // 60:02d}:{seconds % 60:02d}"
+
+
+def normalize_for_majority(row: Mapping[str, Any], raw: Any) -> str | None:
+    """Normalize one sample without using references; ``None`` means invalid."""
+
+    fmt = answer_format(row)
+    if fmt not in FORMAT_ORDER:
+        return None
+    if fmt == "binary":
+        text = _clean_text(raw).lower().strip(" .!?;,:")
+        return text if text in {"yes", "no"} else None
+    if fmt in {"number", "percentage"}:
+        parsed = parse_strict_number(raw, percentage=fmt == "percentage")
+        return _canonical_decimal(parsed) if parsed is not None else None
+    if fmt == "time":
+        parsed = parse_strict_time_list(raw)
+        return ", ".join(format_hms(value) for value in parsed) if parsed is not None else None
+    normalized = normalize_prediction(row, raw)
+    text = _clean_text(normalized, remove_prefix=False)
+    if not text:
+        return None
+    if fmt == "fo_class":
+        allowed = {label.casefold(): label for label in FOREIGN_OBJECT_CLASSES}
+        if text.casefold() == "none":
+            return "none"
+        values = [part.strip() for part in text.split(",") if part.strip()]
+        canonical: list[str] = []
+        for value in values:
+            label = allowed.get(value.casefold())
+            if label is None or label in canonical:
+                return None
+            canonical.append(label)
+        return ", ".join(canonical) if canonical else None
+    return text
+
+
+def aggregate_numeric(
+    row: Mapping[str, Any], raw_predictions: Sequence[Any]
+) -> tuple[str | None, dict[str, Any]]:
+    fmt = answer_format(row)
+    if fmt not in {"number", "percentage"}:
+        raise ValueError(f"numeric aggregation requires number/percentage, got {fmt!r}")
+    parsed = [parse_strict_number(value, percentage=fmt == "percentage") for value in raw_predictions]
+    normalized = [None if value is None else _canonical_decimal(value) for value in parsed]
+    if len(parsed) != 3 or any(value is None for value in parsed):
+        return None, {"parse": normalized, "aggregation": None, "fallback": True, "fallback_reason": "strict_numeric_parse_failure"}
+    ordered = sorted(value for value in parsed if value is not None)
+    median = ordered[1]
+    return _canonical_decimal(median), {
+        "parse": normalized,
+        "aggregation": "median",
+        "aggregated_value": _canonical_decimal(median),
+        "fallback": False,
+    }
+
+
+def aggregate_time(
+    raw_predictions: Sequence[Any],
+) -> tuple[str | None, dict[str, Any]]:
+    parsed = [parse_strict_time_list(value) for value in raw_predictions]
+    normalized = [None if values is None else [format_hms(value) for value in values] for values in parsed]
+    if len(parsed) != 3 or any(values is None for values in parsed):
+        return None, {"parse": normalized, "aggregation": None, "fallback": True, "fallback_reason": "strict_time_parse_failure"}
+    lengths = {len(values) for values in parsed if values is not None}
+    if len(lengths) != 1:
+        return None, {"parse": normalized, "aggregation": None, "fallback": True, "fallback_reason": "time_point_count_mismatch"}
+    count = next(iter(lengths))
+    medians: list[str] = []
+    for index in range(count):
+        values = sorted(values[index] for values in parsed if values is not None)
+        medians.append(format_hms(values[1]))
+    return ", ".join(medians), {
+        "parse": normalized,
+        "aggregation": "per_position_median",
+        "aggregated_value": ", ".join(medians),
+        "fallback": False,
+    }
+
+
+def aggregate_majority(
+    row: Mapping[str, Any], raw_predictions: Sequence[Any], baseline_raw: Any
+) -> tuple[str, dict[str, Any]]:
+    normalized = [normalize_for_majority(row, value) for value in raw_predictions]
+    valid = [value for value in normalized if value is not None]
+    counts = Counter(valid)
+    winner: str | None = None
+    if counts:
+        candidate, count = counts.most_common(1)[0]
+        if count >= 2:
+            winner = candidate
+    baseline_normalized = normalize_for_majority(row, baseline_raw)
+    selected = winner if winner is not None else str(baseline_raw if baseline_raw is not None else "")
+    return selected, {
+        "normalized": normalized,
+        "baseline_normalized": baseline_normalized,
+        "votes": dict(counts),
+        "aggregation": "majority" if winner is not None else "baseline",
+        "selected": "sample_majority" if winner is not None else "baseline",
+        "fallback": winner is None,
+        "fallback_reason": None if winner is not None else "no_two_vote_majority",
+    }
+
+
+def aggregate_prediction(
+    row: Mapping[str, Any], raw_predictions: Sequence[Any], baseline_raw: Any
+) -> tuple[str, dict[str, Any]]:
+    """Reduce A/B/C samples according to the frozen format protocol."""
+
+    fmt = answer_format(row)
+    if fmt in {"number", "percentage"}:
+        final, audit = aggregate_numeric(row, raw_predictions)
+        if final is None:
+            final = str(baseline_raw if baseline_raw is not None else "")
+            audit.update(
+                {
+                    "selected": "baseline",
+                    "fallback": True,
+                    "fallback_reason": audit.get("fallback_reason", "numeric_fallback"),
+                }
+            )
+        else:
+            audit["selected"] = "median"
+        audit["raw"] = [str(value) for value in raw_predictions]
+        return final, audit
+    if fmt == "time":
+        final, audit = aggregate_time(raw_predictions)
+        if final is None:
+            final = str(baseline_raw if baseline_raw is not None else "")
+            audit.update({"selected": "baseline", "fallback": True})
+        else:
+            audit["selected"] = "per_position_median"
+        audit["raw"] = [str(value) for value in raw_predictions]
+        return final, audit
+    final, audit = aggregate_majority(row, raw_predictions, baseline_raw)
+    audit["raw"] = [str(value) for value in raw_predictions]
+    return final, audit
+
+
+def merge_predictions(
+    rows: list[dict[str, Any]],
+    baseline_by_id: Mapping[str, Mapping[str, Any]],
+    generated_by_id: Mapping[str, Mapping[str, Any]],
+    selected_ids: set[str] | None = None,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    selected = selected_ids if selected_ids is not None else {row_id(row) for row in rows}
+    merged: list[dict[str, Any]] = []
+    audit_rows: list[dict[str, Any]] = []
+    for row in rows:
+        rid = row_id(row)
+        if rid not in baseline_by_id:
+            raise ValueError(f"baseline missing id={rid}")
+        baseline = copy.deepcopy(dict(baseline_by_id[rid]))
+        generated = generated_by_id.get(rid) if rid in selected else None
+        output = copy.deepcopy(baseline)
+        audit: dict[str, Any] = {
+            "id": rid,
+            "target": rid in selected,
+            "source": "frozen_baseline",
+            "baseline_prediction": baseline.get("prediction", ""),
+            "changed_vs_baseline": False,
+            "error": None,
+        }
+        if rid in selected:
+            if generated is None or generated.get("error"):
+                audit["error"] = (generated or {}).get("error", "missing_generated_row")
+                if generated and isinstance(generated.get("audit"), Mapping):
+                    audit.update(generated["audit"])
+            else:
+                output.update(
+                    {
+                        "prediction": generated.get("prediction", ""),
+                        "tf_trick": "tf009",
+                        "tf_raw_predictions": generated.get("raw_predictions", []),
+                        "generation_seeds": list(SEEDS),
+                        "generation_temperature": TEMPERATURE,
+                        "generation_top_p": TOP_P,
+                        "generation_max_new_tokens": MAX_NEW_TOKENS,
+                    }
+                )
+                if "latency" in baseline:
+                    output["baseline_latency"] = baseline["latency"]
+                if generated.get("latency") is not None:
+                    output["generation_latency"] = generated["latency"]
+                if isinstance(generated.get("audit"), Mapping):
+                    audit.update(generated["audit"])
+                audit["source"] = "self_consistency"
+                audit["changed_vs_baseline"] = output.get("prediction") != baseline.get("prediction")
+        audit["final_prediction"] = output.get("prediction", "")
+        merged.append(output)
+        audit_rows.append(audit)
+    return merged, audit_rows
+
+
+class _RunLog:
+    def __init__(self, path: Path) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self.handle = path.open("a", encoding="utf-8")
+
+    def write(self, message: str) -> None:
+        line = message.rstrip("\n")
+        print(line, flush=True)
+        self.handle.write(line + "\n")
+        self.handle.flush()
+
+    def close(self) -> None:
+        self.handle.close()
+
+
+def _git_state() -> tuple[str, bool, list[str]]:
+    commit = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=PROJECT_ROOT, text=True, capture_output=True, check=False
+    ).stdout.strip() or "unknown"
+    status = subprocess.run(
+        ["git", "status", "--porcelain"], cwd=PROJECT_ROOT, text=True, capture_output=True, check=False
+    ).stdout
+    files = [line[3:] if len(line) >= 3 else line for line in status.splitlines() if line]
+    return commit, bool(files), files
+
+
+def _load_rows(path: Path, nframes: int) -> list[dict[str, Any]]:
+    from eval_segmentqa_vllm import load_fixed80_absmeta_requests
+
+    return load_fixed80_absmeta_requests(path, nframes)
+
+
+def _validate_model(model: Path) -> None:
+    if model.name != "merged_checkpoint-6013":
+        raise ValueError(f"TF009 requires merged Tr006 checkpoint-6013, got {model}")
+    if not model.is_dir():
+        raise FileNotFoundError(model)
+
+
+def _prepare_request(row: Mapping[str, Any], tokenizer: Any, nframes: int) -> dict[str, Any]:
+    from eval_segmentqa_vllm import fixed_absmeta_frames, make_messages
+
+    metadata = (row.get("chat_template_kwargs") or {}).get("video_metadata")
+    frames, metadata = fixed_absmeta_frames(Path(row["videos"][0]), metadata, nframes)
+    prompt = tokenizer.apply_chat_template(
+        make_messages(dict(row)), tokenize=False, add_generation_prompt=True, enable_thinking=False
+    )
+    return {"prompt": prompt, "multi_modal_data": {"video": (frames, metadata)}}
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--trick", choices=("tf009",), required=True)
+    parser.add_argument("--model", required=True, help="merged Tr006 checkpoint-6013")
+    parser.add_argument("--dataset-jsonl", required=True, help="fixed80 absmeta test_requests.jsonl")
+    parser.add_argument("--baseline-raw", required=True, help="frozen checkpoint-6013 raw_predictions.jsonl")
+    parser.add_argument("--output", required=True, help="RUN_DIR")
+    parser.add_argument("--results-subdir", action="store_true")
+    parser.add_argument("--run-id", default=None)
+    parser.add_argument("--nframes", type=int, default=DEFAULT_NFRAMES)
+    parser.add_argument("--tensor-parallel-size", type=int, default=2)
+    parser.add_argument("--batch-size", type=int, default=8)
+    parser.add_argument("--max-model-len", type=int, default=16384)
+    parser.add_argument("--max-new-tokens", type=int, default=MAX_NEW_TOKENS)
+    parser.add_argument("--gpu-memory-utilization", type=float, default=0.9)
+    parser.add_argument("--max-num-seqs", type=int, default=32)
+    parser.add_argument("--decode-workers", type=int, default=4)
+    parser.add_argument("--smoke", type=int, help="generate N format-stratified rows, then merge all baseline rows")
+    return parser
+
+
+def run(args: argparse.Namespace) -> int:
+    if args.nframes != DEFAULT_NFRAMES:
+        raise ValueError(f"TF009 fixed-frame protocol requires nframes={DEFAULT_NFRAMES}")
+    if args.max_new_tokens != MAX_NEW_TOKENS:
+        raise ValueError(f"TF009 fixed decode protocol requires max_new_tokens={MAX_NEW_TOKENS}")
+    out = resolve_path(args.output)
+    if any(part.lower().startswith("tf001") or "tf001_" in part.lower() for part in out.parts):
+        raise ValueError(f"refusing to write inside a TF001 output path: {out}")
+    results = out / "results" if args.results_subdir else out
+    out.mkdir(parents=True, exist_ok=True)
+    results.mkdir(parents=True, exist_ok=True)
+    logger = _RunLog(out / "run.log")
+    run_id = args.run_id or "TF009"
+    started = time.time()
+    commit, dirty, modified_files = _git_state()
+    runmeta_path = out / "run.json"
+    runmeta: dict[str, Any] = {}
+    try:
+        if runmeta_path.exists():
+            try:
+                runmeta = json.loads(runmeta_path.read_text(encoding="utf-8"))
+            except json.JSONDecodeError:
+                runmeta = {}
+        runmeta.update(
+            {
+                "run": {
+                    **runmeta.get("run", {}),
+                    "id": run_id,
+                    "name": out.name,
+                    "task_type": "debug" if args.smoke is not None else "test",
+                    "trick": "tf009",
+                    "status": "running",
+                },
+                "code": {
+                    **runmeta.get("code", {}),
+                    "commit": commit,
+                    "dirty": dirty,
+                    "modified_files": modified_files,
+                    "entrypoint": "scripts/tf009_self_consistency.py",
+                },
+                "runtime": {
+                    **runmeta.get("runtime", {}),
+                    "python": sys.executable,
+                    "host": platform.node(),
+                    "launcher": "caller-managed tmux/tee; no launcher in runner",
+                    "devices": ["GPU0", "GPU1"],
+                    "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES", "unknown"),
+                    "tensor_parallel_size": args.tensor_parallel_size,
+                    "precision": "bfloat16",
+                },
+                "execution": {
+                    **runmeta.get("execution", {}),
+                    "argv": sys.argv,
+                    "command": shlex.join([sys.executable, *sys.argv]),
+                    "seeds": list(SEEDS),
+                    "temperature": TEMPERATURE,
+                    "top_p": TOP_P,
+                    "max_new_tokens": MAX_NEW_TOKENS,
+                    "nframes": args.nframes,
+                    "batch_size": args.batch_size,
+                    "max_model_len": args.max_model_len,
+                    "max_num_seqs": args.max_num_seqs,
+                    "decode_workers": args.decode_workers,
+                    "only_variable": "stochastic sampling policy + format-aware aggregation",
+                    "reference_read_for_selection": False,
+                },
+                "tracking": {"backend": "none (analysis-only test-time run)", "location": None},
+                "time": {"started_at": started, "finished_at": None},
+                "issues": sorted(set(runmeta.get("issues", [])) | {"analysis-only", "test-reuse", "no-reference-selection"}),
+            }
+        )
+        write_json(runmeta_path, runmeta)
+        logger.write(
+            f"START trick=TF009 run_id={run_id} host={platform.node()} python={sys.executable} "
+            f"CUDA_VISIBLE_DEVICES={os.environ.get('CUDA_VISIBLE_DEVICES', 'unknown')}"
+        )
+        logger.write(f"model={args.model} dataset={args.dataset_jsonl} baseline={args.baseline_raw}")
+        logger.write(f"seeds={list(SEEDS)} temperature={TEMPERATURE} top_p={TOP_P} max_new_tokens={MAX_NEW_TOKENS}")
+
+        model = resolve_path(args.model)
+        _validate_model(model)
+        dataset_path = resolve_path(args.dataset_jsonl)
+        baseline_path = resolve_path(args.baseline_raw)
+        rows = _load_rows(dataset_path, args.nframes)
+        baseline_rows = read_jsonl(baseline_path)
+        baseline_by_id: dict[str, dict[str, Any]] = {}
+        for item in baseline_rows:
+            rid = row_id(item)
+            if rid in baseline_by_id:
+                raise ValueError(f"duplicate baseline id={rid}")
+            baseline_by_id[rid] = item
+        request_ids = [row_id(row) for row in rows]
+        if len(request_ids) != len(set(request_ids)):
+            raise ValueError("requests contain duplicate IDs")
+        missing = [rid for rid in request_ids if rid not in baseline_by_id]
+        if missing:
+            raise ValueError(f"baseline missing {len(missing)} request IDs")
+        target_rows = select_target_rows(rows, args.smoke)
+        target_ids = {row_id(row) for row in target_rows}
+        runmeta["inputs"] = {
+            "model": str(model),
+            "requests": str(dataset_path),
+            "baseline_raw_predictions": str(baseline_path),
+            "split": "test",
+            "count": len(rows),
+            "expected_official_count": EXPECTED_OFFICIAL_COUNT,
+            "target_count": len([row for row in rows if answer_format(row) in FORMAT_ORDER]),
+            "generated_target_count": len(target_rows),
+            "reference_read_for_selection": False,
+        }
+        write_json(runmeta_path, runmeta)
+        logger.write(f"loaded rows={len(rows)} baseline={len(baseline_rows)} selected_targets={len(target_rows)}")
+
+        from vllm import LLM, SamplingParams
+
+        llm = LLM(
+            model=str(model),
+            dtype="bfloat16",
+            tensor_parallel_size=args.tensor_parallel_size,
+            trust_remote_code=True,
+            limit_mm_per_prompt={"video": 1},
+            max_model_len=args.max_model_len,
+            max_num_seqs=args.max_num_seqs,
+            gpu_memory_utilization=args.gpu_memory_utilization,
+        )
+        tokenizer = llm.get_tokenizer()
+        rows_by_id = {row_id(row): row for row in target_rows}
+        prepared: dict[str, dict[str, Any]] = {}
+        decode_workers = max(1, args.decode_workers)
+        with ThreadPoolExecutor(max_workers=decode_workers) as pool:
+            futures = {rid: pool.submit(_prepare_request, row, tokenizer, args.nframes) for rid, row in rows_by_id.items()}
+            for rid, future in futures.items():
+                prepared[rid] = future.result()
+
+        raw_samples: dict[str, list[str]] = {rid: [] for rid in rows_by_id}
+        latencies: dict[str, list[float]] = {rid: [] for rid in rows_by_id}
+        generation_errors: dict[str, str] = {}
+        batch_size = max(1, args.batch_size)
+        for seed_index, seed in enumerate(SEEDS):
+            sampling = SamplingParams(
+                max_tokens=MAX_NEW_TOKENS,
+                temperature=TEMPERATURE,
+                top_p=TOP_P,
+                seed=seed,
+            )
+            for start in range(0, len(target_rows), batch_size):
+                batch = target_rows[start : start + batch_size]
+                requests = [prepared[row_id(row)] for row in batch]
+                try:
+                    generation_started = time.perf_counter()
+                    outputs = llm.generate(requests, sampling)
+                    elapsed = time.perf_counter() - generation_started
+                    if len(outputs) != len(batch):
+                        raise RuntimeError(f"vLLM returned {len(outputs)} outputs for {len(batch)} requests")
+                    for row, result in zip(batch, outputs):
+                        values = getattr(result, "outputs", None) or []
+                        if not values:
+                            raise RuntimeError(f"empty vLLM output for id={row_id(row)} seed={seed}")
+                        output = values[0]
+                        rid = row_id(row)
+                        raw_samples[rid].append(str(getattr(output, "text", "")))
+                        latencies[rid].append(elapsed / len(batch))
+                except Exception as exc:
+                    error = repr(exc)
+                    logger.write(f"generation_error seed={seed} batch_start={start} error={error}")
+                    for row in batch:
+                        rid = row_id(row)
+                        generation_errors[rid] = error
+                        raw_samples[rid].append("")
+                        latencies[rid].append(0.0)
+            logger.write(f"completed seed={seed} ({seed_index + 1}/{len(SEEDS)})")
+
+        generated_by_id: dict[str, dict[str, Any]] = {}
+        for rid, row in rows_by_id.items():
+            baseline_raw = baseline_by_id[rid].get("prediction", "")
+            if rid in generation_errors:
+                generated_by_id[rid] = {
+                    "id": rid,
+                    "error": generation_errors[rid],
+                    "audit": {
+                        "views": {label: {"seed": seed, "raw": raw, "normalized": None} for label, seed, raw in zip(("A", "B", "C"), SEEDS, raw_samples[rid])},
+                        "selected": "baseline",
+                        "fallback": True,
+                        "fallback_reason": "generation_error",
+                    },
+                }
+                continue
+            final, aggregation_audit = aggregate_prediction(row, raw_samples[rid], baseline_raw)
+            normalized = aggregation_audit.get("normalized")
+            if normalized is None:
+                normalized = [normalize_for_majority(row, raw) for raw in raw_samples[rid]]
+            views = {
+                label: {"seed": seed, "raw": raw, "normalized": normalized[index] if index < len(normalized) else None}
+                for index, (label, seed, raw) in enumerate(zip(("A", "B", "C"), SEEDS, raw_samples[rid]))
+            }
+            audit = {"views": views, **aggregation_audit}
+            generated_by_id[rid] = {
+                "id": rid,
+                "prediction": final,
+                "raw_predictions": raw_samples[rid],
+                "latency": sum(latencies[rid]) / len(latencies[rid]) if latencies[rid] else None,
+                "audit": audit,
+            }
+
+        merged, audit_rows = merge_predictions(rows, baseline_by_id, generated_by_id, target_ids)
+        raw_path = results / "raw_predictions.jsonl"
+        write_jsonl(raw_path, merged)
+        audit_path = results / "prepost_audit.jsonl"
+        write_jsonl(audit_path, audit_rows)
+        write_json(
+            results / "answer.json",
+            [{"qID": row_id(item), "content": _clean_text(item.get("prediction", ""), remove_prefix=False), "latency": item.get("latency")} for item in merged],
+        )
+        expected = set(request_ids)
+        actual_ids = [row_id(item) for item in merged]
+        actual = set(actual_ids)
+        errors = [item for item in audit_rows if item.get("error")]
+        changed = [item for item in audit_rows if item.get("changed_vs_baseline")]
+        target_audits = [item for item in audit_rows if item.get("target")]
+        valid_samples = 0
+        total_samples = 3 * len(target_audits)
+        for item in target_audits:
+            views = item.get("views") or {}
+            valid_samples += sum(views.get(label, {}).get("normalized") is not None for label in ("A", "B", "C"))
+        integrity = {
+            "predictions": len(merged),
+            "unique_ids": len(actual),
+            "missing": len(expected - actual),
+            "duplicate": len(actual_ids) - len(actual),
+            "unexpected": len(actual - expected),
+            "failed": len(errors),
+            "skipped": 0,
+            "empty_predictions": sum(not str(item.get("prediction", "")).strip() for item in merged),
+            "target_rows": len(target_audits),
+            "generated_target_rows": sum(item.get("source") == "self_consistency" for item in audit_rows),
+            "selected_smoke_target_rows": len(target_ids),
+            "baseline_reuse_rows": sum(not item.get("target") for item in audit_rows),
+            "changed_vs_baseline": len(changed),
+            "fallback_rows": sum(item.get("selected") == "baseline" for item in target_audits),
+            "switch_rate": len(changed) / len(target_audits) if target_audits else 0.0,
+            "valid_sample_count": valid_samples,
+            "sample_count": total_samples,
+            "parse_rate": valid_samples / total_samples if total_samples else 0.0,
+            "generation_errors": len(errors),
+        }
+        write_json(results / "integrity.json", integrity)
+        metrics = {
+            "protocol": "TF009 self-consistency; three seeded samples; no-reference selection",
+            "status": "passed"
+            if not errors
+            and integrity["missing"] == 0
+            and integrity["duplicate"] == 0
+            and integrity["unexpected"] == 0
+            and integrity["empty_predictions"] == 0
+            and integrity["generated_target_rows"] == integrity["selected_smoke_target_rows"]
+            else "partial",
+            "expected_official_count": EXPECTED_OFFICIAL_COUNT,
+            "seeds": list(SEEDS),
+            "temperature": TEMPERATURE,
+            "top_p": TOP_P,
+            "max_new_tokens": MAX_NEW_TOKENS,
+            "only_variable": "stochastic sampling policy + format-aware aggregation",
+            "reference_read_for_selection": False,
+            "format_counts": dict(Counter(answer_format(row) for row in rows)),
+            "target_rows": integrity["target_rows"],
+            "generated_target_rows": integrity["generated_target_rows"],
+            "baseline_reuse_rows": integrity["baseline_reuse_rows"],
+            "changed_vs_baseline": integrity["changed_vs_baseline"],
+            "switch_rate": integrity["switch_rate"],
+            "parse_rate": integrity["parse_rate"],
+            "fallback_rows": integrity["fallback_rows"],
+            "generation_errors": integrity["generation_errors"],
+        }
+        write_json(results / "metrics.json", metrics)
+        status = "completed" if metrics["status"] == "passed" else "failed"
+        runmeta["run"]["status"] = status
+        runmeta["time"]["finished_at"] = time.time()
+        runmeta["outputs"] = {
+            "run_dir": str(out),
+            "results_dir": str(results),
+            "raw_predictions": str(raw_path),
+            "answer": str(results / "answer.json"),
+            "prepost_audit": str(audit_path),
+            "integrity": str(results / "integrity.json"),
+            "metrics": str(results / "metrics.json"),
+        }
+        if errors:
+            runmeta.setdefault("issues", []).append("target-generation-error-fallback")
+        write_json(runmeta_path, runmeta)
+        logger.write(json.dumps({"status": status, "integrity": integrity}, ensure_ascii=False))
+        return 0 if status == "completed" else 2
+    except BaseException as exc:
+        runmeta.setdefault("run", {})["status"] = "failed"
+        runmeta.setdefault("time", {})["finished_at"] = time.time()
+        runmeta["error"] = repr(exc)
+        write_json(runmeta_path, runmeta)
+        logger.write(f"FAILED error={exc!r}")
+        raise
+    finally:
+        logger.close()
+
+
+def main() -> int:
+    return run(build_parser().parse_args())
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pipelines/Tr006_V2/tf009/training_free_postprocess.py

@@ -0,0 +1,406 @@
+#!/usr/bin/env python3
+"""Reference-free, deterministic prediction post-processing.
+
+The functions in this module deliberately do not inspect references or model
+scores.  They only use the request contract (primarily ``answer_format`` and
+the question text) and the raw model output.  A transformation is applied
+only when the answer can be parsed without guessing; otherwise the original
+prediction is returned unchanged.
+
+The module is intended to be called after generation, before writing the
+official ``answer.json``.  It is not a replacement for the official evaluator:
+in particular, it does not apply time tolerances or judge open-ended answers.
+"""
+
+from __future__ import annotations
+
+import re
+from decimal import Decimal, InvalidOperation
+from typing import Any, Mapping
+
+
+SUPPORTED_FORMATS = frozenset(
+    {"binary", "number", "percentage", "time", "multiple_choice", "fo_class", "open_ended"}
+)
+
+# This is the ontology used by the SegmentQA system prompt.  We keep the
+# spelling here so that a case-insensitive model response can be rendered in
+# the contract's canonical spelling without consulting test references.
+FOREIGN_OBJECT_CLASSES = (
+    "Sponge",
+    "Clip",
+    "Specimen Bag",
+    "Silicone Loop",
+    "External Drain",
+    "Needle",
+    "Gallstone",
+    "Specimen",
+    "Mesh",
+    "Absorbable Hemostatic Agent",
+)
+
+_THINK_RE = re.compile(r"<think>.*?</think>", re.IGNORECASE | re.DOTALL)
+_TIME_RE = re.compile(r"(?<![\d:])(?P<h>\d{1,6}):(?P<m>\d{2}):(?P<s>\d{2})(?![\d:])")
+_NUMBER_RE = re.compile(
+    r"(?<![\w.])"
+    r"(?P<value>[+-]?(?:(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?|\.\d+))"
+    r"(?![\w.])"
+)
+_ANSWER_PREFIX_RE = re.compile(
+    r"^(?:(?:the\s+)?(?:final\s+)?(?:answer|response|result)\s*(?:(?:is|=)\s*|:\s*)|"
+    r"(?:final\s+)?answer\s*-\s*)",
+    re.IGNORECASE,
+)
+_MULTI_CHOICE_RE = re.compile(
+    r"\b(?:one\s+or\s+multiple|one\s+or\s+more|multiple\s+answers?|"
+    r"all\s+that\s+apply|none\s*,\s*one)\b",
+    re.IGNORECASE,
+)
+_OPTION_MARKER_RE = re.compile(
+    r"(?:following\s+)?(?:answer\s+)?(?:options?|choices?)\s*(?::|are|include)?\s*(.*)$",
+    re.IGNORECASE | re.DOTALL,
+)
+_EXPLANATION_RE = re.compile(
+    r"^(?:because|as|since|which|that|where|when|visible|present|shown|"
+    r"indicating|indicates|the\s+answer|the\s+correct\s+(?:answer|choice)|"
+    r"is|was|are|were)\b",
+    re.IGNORECASE,
+)
+
+
+def _as_text(value: Any) -> str:
+    """Return a stable textual representation for a model prediction."""
+
+    return value if isinstance(value, str) else str(value if value is not None else "")
+
+
+def _parse_text(value: Any) -> str:
+    """Remove completed reasoning blocks and collapse whitespace for parsing."""
+
+    return re.sub(r"\s+", " ", _THINK_RE.sub("", _as_text(value))).strip()
+
+
+def _answer_format(request: Mapping[str, Any] | None) -> str | None:
+    """Read ``answer_format`` from the row or its common metadata wrappers."""
+
+    if not isinstance(request, Mapping):
+        return None
+    candidates: list[Any] = [request.get("answer_format"), request.get("format")]
+    for key in ("segment_metadata", "metadata", "request_metadata"):
+        nested = request.get(key)
+        if isinstance(nested, Mapping):
+            candidates.extend((nested.get("answer_format"), nested.get("format")))
+    for value in candidates:
+        if value is not None:
+            text = str(value).strip().lower()
+            if text in SUPPORTED_FORMATS:
+                return text
+    return None
+
+
+def _question_text(request: Mapping[str, Any] | None) -> str:
+    if not isinstance(request, Mapping):
+        return ""
+    values: list[Any] = [request.get("question"), request.get("prompt")]
+    for key in ("segment_metadata", "metadata", "request_metadata"):
+        nested = request.get(key)
+        if isinstance(nested, Mapping):
+            values.extend((nested.get("question"), nested.get("prompt")))
+    for value in values:
+        if isinstance(value, str) and value.strip():
+            return value
+    # A fixed-frame request may only retain the question in a user message.
+    messages = request.get("messages")
+    if isinstance(messages, list):
+        for message in reversed(messages):
+            if isinstance(message, Mapping) and message.get("role") == "user":
+                content = message.get("content")
+                if isinstance(content, str):
+                    return content
+                if isinstance(content, list):
+                    pieces = [item.get("text", "") for item in content if isinstance(item, Mapping)]
+                    text = " ".join(piece for piece in pieces if isinstance(piece, str))
+                    if text.strip():
+                        return text
+    return ""
+
+
+def _remove_prefix(text: str) -> str:
+    """Strip an explicit answer label, but never infer an answer from prose."""
+
+    value = text.strip()
+    # A model can repeat a label (``Final answer: Answer: yes``); removing at
+    # most a few labels is deterministic and still leaves ambiguous prose
+    # untouched.
+    for _ in range(3):
+        match = _ANSWER_PREFIX_RE.match(value)
+        if not match:
+            break
+        value = value[match.end() :].lstrip()
+    return value
+
+
+def _canonical_decimal(raw: str) -> str | None:
+    try:
+        value = Decimal(raw.replace(",", ""))
+    except (InvalidOperation, ValueError):
+        return None
+    if not value.is_finite():
+        return None
+    if value == 0:
+        return "0"
+    rendered = format(value, "f")
+    if "." in rendered:
+        rendered = rendered.rstrip("0").rstrip(".")
+    if rendered.startswith("+"):
+        rendered = rendered[1:]
+    return rendered
+
+
+def normalize_time_output(prediction: Any) -> str:
+    """Normalize one or more ``h:mm:ss`` values without applying an offset.
+
+    All valid time tokens are extracted in their original order, duplicate
+    tokens are removed while retaining the first occurrence, and the result is
+    rendered as ``HH:MM:SS`` comma-separated values.  If no unambiguous valid
+    token exists (including an invalid minute/second), the exact raw text is
+    returned.  This function intentionally does *not* add the clip start time.
+    """
+
+    raw = _as_text(prediction)
+    cleaned = _parse_text(raw)
+    matches = list(_TIME_RE.finditer(cleaned))
+    if not matches:
+        return raw
+
+    values: list[str] = []
+    seen: set[str] = set()
+    for match in matches:
+        hours = int(match.group("h"))
+        minutes = int(match.group("m"))
+        seconds = int(match.group("s"))
+        if minutes >= 60 or seconds >= 60:
+            return raw
+        value = f"{hours:02d}:{minutes:02d}:{seconds:02d}"
+        if value not in seen:
+            seen.add(value)
+            values.append(value)
+    return ", ".join(values) if values else raw
+
+
+def _normalize_number(prediction: Any, percentage: bool = False) -> str:
+    raw = _as_text(prediction)
+    cleaned = _remove_prefix(_parse_text(raw))
+    matches = list(_NUMBER_RE.finditer(cleaned))
+    if len(matches) != 1:
+        return raw
+    match = matches[0]
+    # Number answers should not silently turn a percentage into a count, and
+    # percentage answers reject the negative values that the official parser
+    # considers invalid.
+    if not percentage and ("%" in cleaned or re.search(r"\bpercent(?:age)?\b", cleaned, re.IGNORECASE)):
+        return raw
+    value = _canonical_decimal(match.group("value"))
+    if value is None:
+        return raw
+    if percentage and value.startswith("-"):
+        return raw
+    return value
+
+
+def _boundary(text: str, end: int) -> bool:
+    return end >= len(text) or text[end].isspace() or text[end] in ".,;:!?)]}\"'"
+
+
+def _match_option(text: str, options: tuple[str, ...], start: int = 0) -> tuple[str, int] | None:
+    remaining = text[start:]
+    lowered = remaining.casefold()
+    for option in sorted(options, key=lambda item: (-len(item), item.casefold())):
+        candidate = option.casefold()
+        if not lowered.startswith(candidate):
+            continue
+        end = start + len(option)
+        if _boundary(text, end):
+            return option, end
+    return None
+
+
+def _question_options(request: Mapping[str, Any] | None) -> tuple[str, ...]:
+    """Extract explicit semicolon/comma choices from a question when present."""
+
+    if not isinstance(request, Mapping):
+        return ()
+    for key in ("options", "choices"):
+        values = request.get(key)
+        if isinstance(values, (list, tuple)):
+            result = tuple(str(value).strip() for value in values if str(value).strip())
+            if len(result) >= 2:
+                return result
+
+    question = _question_text(request)
+    if not question:
+        return ()
+    marker = _OPTION_MARKER_RE.search(question.replace("\n", " "))
+    if marker:
+        tail = marker.group(1).strip()
+    else:
+        # Segment prompts commonly use ``... options: a; b; c`` without the
+        # word "options" immediately before the colon.  Only use a suffix
+        # containing a semicolon to avoid treating timestamps as choices.
+        tail = question.rsplit(":", 1)[-1].strip() if ";" in question.rsplit(":", 1)[-1] else ""
+    if not tail:
+        return ()
+
+    # Remove examples and sentence punctuation from the choice tail.  Values
+    # such as ``top/left`` intentionally retain their slash.
+    parts = [part.strip(" \t\r\n .!?\"'“”()[]") for part in re.split(r"\s*;\s*", tail)]
+    parts = [part for part in parts if part]
+    if len(parts) < 2:
+        return ()
+    # A choice list should not contain sentence-like fragments.  This check is
+    # intentionally conservative; callers fall back to raw output on failure.
+    return tuple(dict.fromkeys(parts))
+
+
+def _safe_tail(tail: str) -> bool:
+    tail = tail.strip()
+    if not tail:
+        return True
+    if tail[0] in ".,;:!?)]}\"'" or tail.startswith("("):
+        return True
+    if re.match(r"^(?:or|/|and)\b", tail, re.IGNORECASE):
+        return False
+    return bool(_EXPLANATION_RE.match(tail))
+
+
+def _normalize_choice(prediction: Any, request: Mapping[str, Any] | None) -> str:
+    raw = _as_text(prediction)
+    cleaned = _remove_prefix(_parse_text(raw))
+    options = _question_options(request)
+    if not options:
+        # When a prompt does not retain its explicit choices, a single leading
+        # letter is still safe to canonicalize; a word answer is not.
+        match = re.match(r"^\(?([A-Za-z])\)?(?=$|[.)\]:\-\s])", cleaned)
+        if match:
+            return match.group(1).upper()
+        return raw
+
+    allow_multiple = bool(_MULTI_CHOICE_RE.search(_question_text(request)))
+    # Include none only when the prompt explicitly allows it.  Do not infer it
+    # from arbitrary prose.
+    if allow_multiple and not any(option.casefold() == "none" for option in options):
+        options = (*options, "none")
+
+    first = _match_option(cleaned, options)
+    if first is None:
+        # Letter-labelled options (A) foo; B) bar) are accepted only when the
+        # labels are explicit in the question and map to exactly one choice.
+        labelled = re.findall(r"(?:^|[;\n])\s*([A-Za-z])\s*[.)\]:-]\s*([^;\n]+)", _question_text(request))
+        labels = {label.casefold(): value.strip(" .!?\"'") for label, value in labelled}
+        letter = re.match(r"^\(?([A-Za-z])\)?(?=$|[.)\]:\-\s])", cleaned)
+        if letter and letter.group(1).casefold() in labels:
+            return labels[letter.group(1).casefold()]
+        return raw
+
+    selected = [first[0]]
+    position = first[1]
+    if not allow_multiple:
+        return selected[0] if _safe_tail(cleaned[position:]) else raw
+
+    while True:
+        tail = cleaned[position:].lstrip()
+        separator = re.match(r"^(?:[,;]|\band\b)\s*", tail, re.IGNORECASE)
+        if not separator:
+            break
+        position += len(cleaned[position:]) - len(tail) + separator.end()
+        match = _match_option(cleaned, options, position)
+        if match is None:
+            return raw
+        if any(match[0].casefold() == value.casefold() for value in selected):
+            return raw
+        selected.append(match[0])
+        position = match[1]
+    return ", ".join(selected) if _safe_tail(cleaned[position:]) else raw
+
+
+def _normalize_fo_class(prediction: Any, request: Mapping[str, Any] | None) -> str:
+    raw = _as_text(prediction)
+    cleaned = _remove_prefix(_parse_text(raw))
+    options = tuple(FOREIGN_OBJECT_CLASSES)
+    explicit = _question_options(request)
+    if explicit:
+        # Prefer the known ontology but retain custom classes if the request
+        # explicitly supplies them.
+        options = tuple(dict.fromkeys((*explicit, *options)))
+    options = (*options, "none")
+
+    first = _match_option(cleaned, options)
+    if first is None:
+        return raw
+    selected = [first[0]]
+    position = first[1]
+    while True:
+        tail = cleaned[position:].lstrip()
+        separator = re.match(r"^(?:[,;]|\band\b)\s*", tail, re.IGNORECASE)
+        if not separator:
+            break
+        position += len(cleaned[position:]) - len(tail) + separator.end()
+        match = _match_option(cleaned, options, position)
+        if match is None or any(match[0].casefold() == value.casefold() for value in selected):
+            return raw
+        selected.append(match[0])
+        position = match[1]
+    if not _safe_tail(cleaned[position:]):
+        return raw
+    # ``none`` cannot be combined with a concrete class without guessing what
+    # the model intended.
+    if len(selected) > 1 and any(value.casefold() == "none" for value in selected):
+        return raw
+    return ", ".join(selected)
+
+
+def normalize_prediction(request: Mapping[str, Any] | None, prediction: Any) -> str:
+    """Apply the conservative TF011/TF012 transform for one prediction.
+
+    ``request`` may be a parquet-like row, a fixed-frame request, or a small
+    metadata dictionary.  ``open_ended`` and unknown formats are deliberately
+    unchanged.  No function in this module reads an answer/reference field.
+    """
+
+    answer_format = _answer_format(request)
+    if answer_format == "time":
+        return normalize_time_output(prediction)
+    if answer_format == "binary":
+        raw = _as_text(prediction)
+        cleaned = _remove_prefix(_parse_text(raw))
+        match = re.match(r"^(yes|no)(?=$|[\s.,;:!?)}])", cleaned, re.IGNORECASE)
+        if not match:
+            return raw
+        tail = cleaned[match.end() :]
+        if re.match(r"^\s*(?:or|/|and)\s+(?:yes|no)\b", tail, re.IGNORECASE):
+            return raw
+        return match.group(1).lower()
+    if answer_format == "number":
+        return _normalize_number(prediction)
+    if answer_format == "percentage":
+        return _normalize_number(prediction, percentage=True)
+    if answer_format == "multiple_choice":
+        return _normalize_choice(prediction, request)
+    if answer_format == "fo_class":
+        return _normalize_fo_class(prediction, request)
+    return _as_text(prediction)
+
+
+# Friendly aliases for callers that use “post-process” terminology.
+postprocess_prediction = normalize_prediction
+normalize_answer = normalize_prediction
+
+
+__all__ = [
+    "FOREIGN_OBJECT_CLASSES",
+    "SUPPORTED_FORMATS",
+    "normalize_answer",
+    "normalize_prediction",
+    "normalize_time_output",
+    "postprocess_prediction",
+]