moral-maps/src/tinymfv/guided.py

"""Guided rollout: think + flat prefix+suffix scoring for forced-choice probes.

Public API: `guided_rollout_forced_choice` (K-way moral-foundation probe with
two-pass enum-reversal position-bias debias).

Core: `_rollout_kv_fork` does Phase-1 batched think-gen + Phase-2a prefix
forward (for free per-row prompt-NLL) + Phase-2b per-slot flat prefix+suffix
forward. Reads logits at the prefill's last position, gathers logprobs at the
foundation first-tokens.

Cost: 1 prefix forward + N_slots full prefix+suffix forwards (suffix <=14
tokens). We re-encode the prefix per slot to stay correct on hybrid attention
(sliding-window / SSM) where KV-fork loses cached context past the window.
Function is still named `_rollout_kv_fork` for caller-API stability.

Why turn-boundary close+nudge: matches what a chat UI emits when a human
interrupts a partial assistant turn. On-policy in chat-tuned data, where the
prior `\\nI should answer now.</think>` mid-turn splice was OOD.
"""
from __future__ import annotations

from dataclasses import dataclass

import torch
import torch.nn.functional as F
from loguru import logger

_CLOSE_MARKER: str = "</think>"
_ASSISTANT_SENTINEL: str = "ZZUNIQ_ASSISTANT_SENTINEL_ZZ"


def _assistant_close(tok) -> str:
    """Probe chat template for the assistant-turn close marker (e.g. `<|im_end|>\\n`
    on Qwen/ChatML, `<|eot_id|>` on Llama3). Tokenizer-agnostic: mimics what a chat
    UI emits when a human stops a partial assistant turn before sending a new user
    message. Sentinel-diff because Qwen3 auto-injects empty `<think></think>` in
    non-generation-prompt mode, breaking opened-vs-closed prefix-diff."""
    closed = tok.apply_chat_template(
        [{"role": "user", "content": "_"},
         {"role": "assistant", "content": _ASSISTANT_SENTINEL}],
        tokenize=False,
    )
    assert _ASSISTANT_SENTINEL in closed, f"sentinel not in template output: {closed!r}"
    return closed.split(_ASSISTANT_SENTINEL, 1)[1]


def _split_choice_ids(choice_token_ids: list) -> tuple[list[int], list[int]]:
    if len(choice_token_ids) == 2 and all(isinstance(x, (list, tuple)) for x in choice_token_ids):
        return list(choice_token_ids[0]), list(choice_token_ids[1])
    return list(choice_token_ids), []




@torch.no_grad()
def _rollout_kv_fork(
    model, tok,
    user_prompts: list[str],
    schema_hint: str,
    max_think_tokens: int,
    scoring_slots: list[tuple[str, str]],   # (nudge_user_text, prefill) per slot
    choice_token_ids: list,                 # [a_ids, b_ids]
    verbose: bool = False,
    gather_token_ids: list[int] | None = None,
) -> tuple[list[tuple[str, int, bool]], list[list[dict]], list[float]]:
    """Returns (thinks, slots, nll_prompts).
    thinks[i] = (think_text, n_think_tokens, emitted_close)
    slots[i][j] = {pmass_format, logratio, p_true, [lp_gather]}
    nll_prompts[i] = mean teacher-forcing NLL (nats/token) on the user-side chat
        tokens of row i (excluding the very first chat token, which has no
        in-prompt context). Computed from the Phase-2a prefix forward, so it's
        free vs the existing rollout. Use as a coherence/degradation probe:
        a steered or perturbed model that breaks normal text representation
        will see nll_prompt rise (the model is "more surprised" by ordinary
        prompt text).

    If `gather_token_ids` is provided, slot dict also has `lp_gather`: list[float]
    of log-probs at last suffix position for those token ids (in the same order).
    Used by forced-choice (K-way) probing where a 2-way logratio is insufficient.
    """
    if tok.padding_side != "left":
        raise ValueError("tok.padding_side must be 'left'")
    device = next(model.parameters()).device
    pad_id = tok.pad_token_id if tok.pad_token_id is not None else tok.eos_token_id
    close = _assistant_close(tok)

    # === Phase 1: think generation (batched) ===
    chats = [
        tok.apply_chat_template(
            [{"role": "user", "content": f"{up}\n\n{schema_hint}" if schema_hint else up}],
            tokenize=False, add_generation_prompt=True,
        ) + "<think>\n"
        for up in user_prompts
    ]

    think_end_id = tok.convert_tokens_to_ids("</think>")
    if think_end_id in (None, getattr(tok, "unk_token_id", None)):
        think_end_id = tok.eos_token_id

    enc = tok(chats, return_tensors="pt", padding=True).to(device)
    prompt_len = enc.input_ids.shape[1]
    phase1 = model.generate(
        **enc, max_new_tokens=max_think_tokens, do_sample=False,
        eos_token_id=think_end_id, pad_token_id=pad_id,
    )

    # === Build per-row scoring prefix: chat + think + </think> ===
    # Also retokenise the chat alone so we can locate the user-prompt span
    # inside sp_ids for free prompt-NLL below. We assert prefix-equality so
    # tokenizer boundary merges (rare, but possible across `<think>\n` →
    # think-text) crash loudly rather than silently misaligning the NLL window.
    sp_per_row: list[str] = []
    sp_ids_per_row: list[list[int]] = []
    chat_ids_per_row: list[list[int]] = [
        tok(c, add_special_tokens=False)["input_ids"] for c in chats
    ]
    thinks: list[tuple[str, int, bool]] = []
    for i, p in enumerate(chats):
        gen_ids = phase1[i, prompt_len:]
        keep = gen_ids != pad_id
        gen_ids = gen_ids[keep] if keep.any() else gen_ids[:0]
        gen_text = tok.decode(gen_ids, skip_special_tokens=True)
        n_think = int(gen_ids.shape[0])
        emitted_close = _CLOSE_MARKER in gen_text
        think_text = gen_text.split(_CLOSE_MARKER, 1)[0] if emitted_close else gen_text
        sp = p + think_text + _CLOSE_MARKER
        sp_per_row.append(sp)
        sp_ids_per_row.append(tok(sp, add_special_tokens=False)["input_ids"])
        thinks.append((think_text, n_think, emitted_close))

    B = len(sp_per_row)
    P_max = max(len(s) for s in sp_ids_per_row)

    # === Phase 2a: batched prefix forward (left-padded), cache ===
    pref_input = torch.full((B, P_max), pad_id, dtype=torch.long, device=device)
    pref_attn = torch.zeros((B, P_max), dtype=torch.long, device=device)
    pref_real = torch.zeros(B, dtype=torch.long, device=device)
    for i, sp_ids in enumerate(sp_ids_per_row):
        L = len(sp_ids)
        pref_input[i, P_max - L:] = torch.tensor(sp_ids, device=device)
        pref_attn[i, P_max - L:] = 1
        pref_real[i] = L
    pref_out = model(input_ids=pref_input, attention_mask=pref_attn, use_cache=False)

    # === Per-row prompt NLL (free; reuses pref_out.logits) ===
    # Teacher-forcing on the chat tokens (= rendered user turn + generation
    # prompt). We skip the first chat token because position-(start-1) is a
    # left-pad slot, so its prediction is meaningless. Per-row slicing keeps
    # peak memory low (avoids a full [B, P_max, V] log_softmax).
    nll_prompts: list[float] = []
    for i in range(B):
        L_pref = int(pref_real[i].item())
        chat_ids_i = chat_ids_per_row[i]
        chat_len = len(chat_ids_i)
        assert sp_ids_per_row[i][:chat_len] == chat_ids_i, (
            f"chat retokenisation diverged from prefix at row {i}: "
            f"boundary tokens shifted between chat and chat+think+close. "
            f"prompt-NLL window cannot be located safely."
        )
        if chat_len < 2:
            nll_prompts.append(float("nan"))
            continue
        start = P_max - L_pref
        logits_slice = pref_out.logits[i, start : start + chat_len - 1].float()
        targets = pref_input[i, start + 1 : start + chat_len]
        logp = F.log_softmax(logits_slice, dim=-1)
        nll = -logp.gather(-1, targets[:, None]).squeeze(-1).mean()
        nll_prompts.append(float(nll.item()))

    # === Phase 2b: per-slot flat prefix+suffix forward ===
    # We re-encode the prefix per slot rather than re-using a KV cache. Slower
    # (no cache reuse, ~7× per-slot cost on full-attention models) but correct
    # for both full-attention AND sliding-window/SSM models. KV-fork was
    # incompatible with hybrid attention (Gemma-2/3) where the cache drops
    # context past the window.
    a_ids, b_ids = _split_choice_ids(choice_token_ids)
    a_t = torch.tensor(a_ids, device=device, dtype=torch.long) if a_ids else None
    b_t = torch.tensor(b_ids, device=device, dtype=torch.long) if b_ids else None
    all_ids = torch.tensor(a_ids + b_ids, device=device, dtype=torch.long)

    def suf_ids_for(nudge: str, prefill: str) -> list[list[int]]:
        # close-turn + apply_chat_template([user(nudge), assistant(prefill)], continue_final_message=True)
        # gives the on-policy interrupt-then-renudge structure.
        interrupt = tok.apply_chat_template(
            [{"role": "user", "content": nudge},
             {"role": "assistant", "content": prefill}],
            tokenize=False, continue_final_message=True,
        )
        suf_text = close + interrupt
        return [tok(sp + suf_text, add_special_tokens=False)["input_ids"][len(sp_ids):]
                for sp, sp_ids in zip(sp_per_row, sp_ids_per_row)]

    def fork(suffixes: list[list[int]]) -> torch.Tensor:
        """Flat prefix+suffix forward. Returns [B, V] logp at last real suffix token."""
        J_max = max(len(s) for s in suffixes)
        suf_input = torch.full((B, J_max), pad_id, dtype=torch.long, device=device)
        suf_mask = torch.zeros((B, J_max), dtype=torch.long, device=device)
        last_pos = torch.zeros(B, dtype=torch.long, device=device)
        for i, s in enumerate(suffixes):
            L = len(s)
            suf_input[i, :L] = torch.tensor(s, device=device)
            suf_mask[i, :L] = 1
            last_pos[i] = L - 1
        full_input = torch.cat([pref_input, suf_input], dim=1)
        full_attn = torch.cat([pref_attn, suf_mask], dim=1)
        out = model(input_ids=full_input, attention_mask=full_attn, use_cache=False)
        logp = F.log_softmax(out.logits.float(), dim=-1)
        # Suffix's last real token is at absolute position P_max + last_pos[i].
        absolute_last = P_max + last_pos
        return logp[torch.arange(B, device=device), absolute_last]

    slots: list[list[dict]] = [[] for _ in range(B)]
    for j, (nudge, prefill) in enumerate(scoring_slots):
        suf_ids = suf_ids_for(nudge, prefill)
        if verbose:
            full_text = sp_per_row[0] + tok.decode(suf_ids[0], skip_special_tokens=False)
            # Free-form generate to see what the model actually says after the prefill.
            full_ids = torch.tensor(
                [sp_ids_per_row[0] + suf_ids[0]], device=device, dtype=torch.long,
            )
            gen = model.generate(full_ids, max_new_tokens=64, do_sample=False, pad_token_id=pad_id)
            free = tok.decode(gen[0, full_ids.shape[1]:], skip_special_tokens=False)
            logger.info(
                f"--- slot {j} (nudge={nudge!r}, prefill={prefill!r}) ---\n"
                f"{full_text}<<<MODEL CONTINUES>>>{free}\n--- end slot {j} ---"
            )
        lp_last = fork(suf_ids)
        pmass = lp_last[:, all_ids].exp().sum(-1)
        if a_t is not None and b_t is not None:
            la = torch.logsumexp(lp_last[:, a_t], dim=-1)
            lb = torch.logsumexp(lp_last[:, b_t], dim=-1)
            logratio = la - lb
            p_true = torch.softmax(torch.stack([la, lb], dim=-1), dim=-1)[:, 0]
        else:
            logratio = torch.full((B,), float("nan"), device=device)
            p_true = torch.full((B,), float("nan"), device=device)
        for i in range(B):
            top5 = lp_last[i].topk(5)
            top5_str = " ".join(
                f"{tok.decode([int(idx)])!r}:{float(prob.exp()):.3f}"
                for idx, prob in zip(top5.indices, top5.values)
            )
            d = {
                "pmass_format": float(pmass[i].item()),
                "logratio": float(logratio[i].item()),
                "p_true": float(p_true[i].item()),
                "top5_str": top5_str,
            }
            if gather_token_ids is not None:
                gid_t = torch.tensor(gather_token_ids, device=device, dtype=torch.long)
                d["lp_gather"] = lp_last[i, gid_t].cpu().tolist()
            slots[i].append(d)

    return thinks, slots, nll_prompts


# ===== Forced-choice (K-way primary foundation) =====

# Foundation set + descriptions adapted from the response options in
# Clifford et al. (2015) Study 1. Their "social" option = "not morally wrong"
# (the social-norms control set). Descriptions are parallel and short
# (no repetitive "violates norms of" prefix) so no option stands out
# structurally.
_DEFAULT_FORCED_FOUNDATIONS: tuple[str, ...] = (
    "care", "fairness", "loyalty", "authority", "sanctity", "liberty", "social",
)

_FORCED_FOUNDATION_DESCS: dict[str, str] = {
    "care":      "harm or unkindness, causing pain to another",
    "fairness":  "cheating or reducing equality",
    "loyalty":   "betrayal of a group",
    "authority": "subversion or lack of respect for tradition",
    "sanctity":  "purity, degrading or disgusting acts",
    "liberty":   "bullying or dominating",
    "social":    "weird or unusual behaviour, but not morally wrong",
}


def _format_options(foundations: list[str]) -> str:
    """JSON-schema-style enum lines with `# description` comments per option."""
    return "\n".join(
        f'          "{f}", # {_FORCED_FOUNDATION_DESCS.get(f, f)}'
        for f in foundations
    )


# Schema hint (shown ONCE in the user-side prompt that precedes the think).
# JSON-pseudo-schema with enum + per-option comments puts the description
# right at the option, not in separate prose. The prefill
# `This is wrong because {"violation": "` then forces a single-token answer.
def _make_forced_hint(foundations: list[str]) -> str:
    return (
        "This is wrong because:\n"
        "  {\n"
        '    "properties": {\n'
        '      "violation": {\n'
        '        "enum": [\n'
        f"{_format_options(foundations)}\n"
        "        ]\n"
        "      }\n"
        "    }\n"
        "  }"
    )


_DEFAULT_FORCED_HINT: str = _make_forced_hint(list(_DEFAULT_FORCED_FOUNDATIONS))


@dataclass
class ForcedChoiceResult:
    user_prompt: str
    # Two thinks: one per enum-ordering frame. think_fwd uses the forward enum
    # order, think_rev uses the reversed enum order. These cancel position bias
    # when the resulting logprobs are averaged.
    think_text: str       # forward-frame think (for backward compatibility)
    think_text_rev: str   # reversed-frame think
    # Per-frame raw logprobs (unnormalised) at the prefill position.
    lp_fwd: dict[str, float]   # enum listed [care, ..., social]
    lp_rev: dict[str, float]   # enum listed [social, ..., care]
    # Debiased score: average of lp_fwd and lp_rev. Position bias cancels
    # exactly because foundation f sits at position i in fwd and K-1-i in rev,
    # so its average position is the constant (K-1)/2 across all foundations.
    score: dict[str, float]
    p: dict[str, float]    # softmax over the K options of `score`
    top1: str
    margin: float          # score[top1] - score[top2], in nats
    think_tokens: int
    emitted_close: bool
    # Mean teacher-forcing NLL (nats/token) on the user-side chat tokens,
    # averaged across the fwd and rev framings. Free coherence/degradation
    # signal: rises when the model is perturbed (steering, ablation, etc.)
    # to a state where ordinary prompt text becomes "surprising".
    nll_prompt: float


def _resolve_first_token_ids(tok, words: list[str]) -> tuple[list[int], dict[str, int]]:
    """Return (ids_in_order, word->id). Each id is the first token of the word
    when it appears immediately after a `"` in JSON, i.e. with no leading space.
    Asserts the K first-tokens are distinct.

    The forced-choice prefill is `... "violates": "` so the model's
    next token is the first BPE piece of the foundation word with no space prefix."""
    ids: list[int] = []
    mapping: dict[str, int] = {}
    for w in words:
        toks = tok.encode(w, add_special_tokens=False)
        assert toks, f"tokenizer returned empty for {w!r}"
        ids.append(toks[0])
        mapping[w] = toks[0]
    assert len(set(ids)) == len(ids), (
        f"first-token collision among forced-choice words: "
        f"{[(w, i, tok.decode([i])) for w, i in mapping.items()]}"
    )
    return ids, mapping


def guided_rollout_forced_choice(
    model, tok,
    user_prompts: list[str],
    foundations: list[str] | None = None,
    *,
    max_think_tokens: int = 256,
    schema_hint: str | None = None,
    verbose: bool = False,
) -> list[ForcedChoiceResult]:
    """K-way forced-choice over MFT foundations with two-pass position-bias debias.

    Why K-way: pairwise yes/no probes (multibool) leak a global "badness" prior
    across foundations, collapsing per-foundation discrimination. A K-way
    softmax over foundation-name first-tokens forces exactly one and sidesteps
    that prior.

    Why two passes (forward + reversed enum):
    - Position bias is real: models often favour the first option listed.
    - If foundation `f` is at position `i` in the forward enum, it's at position
      `K-1-i` in the reversed enum. Averaging logprobs across the two frames
      makes the position contribution `(P[i] + P[K-1-i]) / 2` -- a constant
      independent of `f`, so it falls out of the softmax over foundations.
    - Two SEPARATE thinks (one per ordering) keep the framing internally
      consistent. We don't try to share KV across orderings since the schema
      text itself differs.

    Format (one user turn per frame):
        <user>
        This is wrong because of which moral foundation
        > {scenario}
        {JSON-pseudo-schema with enum + # comments}
        </user>
        <assistant>
        <think>...</think>
        </assistant>
        <user>Just answer</user>
        <assistant>This is wrong because {"violation": "<scored token>
    """
    if foundations is None:
        foundations = list(_DEFAULT_FORCED_FOUNDATIONS)
    K = len(foundations)
    assert K >= 2, "need >=2 foundations for forced choice"

    foundations_rev = list(reversed(foundations))
    schema_fwd = schema_hint if schema_hint is not None else _make_forced_hint(foundations)
    schema_rev = _make_forced_hint(foundations_rev)

    first_ids, _ = _resolve_first_token_ids(tok, foundations)

    nudge = "Just answer"
    prefill = 'This is wrong because {"violation": "'
    scoring_slot = [(nudge, prefill)]

    # Frame A: forward enum order.
    thinks_fwd, slots_fwd, nll_fwd = _rollout_kv_fork(
        model, tok, user_prompts, schema_fwd, max_think_tokens,
        scoring_slots=scoring_slot,
        choice_token_ids=[[first_ids[0]]],  # unused; satisfies API
        verbose=verbose,
        gather_token_ids=first_ids,
    )
    # Frame B: reversed enum order. Same gather order (by foundation name) so
    # lp_rev[f] is comparable to lp_fwd[f].
    thinks_rev, slots_rev, nll_rev = _rollout_kv_fork(
        model, tok, user_prompts, schema_rev, max_think_tokens,
        scoring_slots=scoring_slot,
        choice_token_ids=[[first_ids[0]]],
        verbose=verbose,
        gather_token_ids=first_ids,
    )

    results: list[ForcedChoiceResult] = []
    import math
    for i in range(len(user_prompts)):
        think_fwd, n_fwd, close_fwd = thinks_fwd[i]
        think_rev, _, _ = thinks_rev[i]
        lp_f = slots_fwd[i][0]["lp_gather"]
        lp_r = slots_rev[i][0]["lp_gather"]
        score = [(lp_f[k] + lp_r[k]) / 2.0 for k in range(K)]

        m = max(score)
        exps = [math.exp(x - m) for x in score]
        Z = sum(exps)
        p = {foundations[k]: exps[k] / Z for k in range(K)}
        order_sorted = sorted(range(K), key=lambda k: -score[k])
        top1 = foundations[order_sorted[0]]
        margin = score[order_sorted[0]] - score[order_sorted[1]]
        # Average prompt NLL across the two framings (schema_hint differs in
        # enum order but the user vignette is identical).
        nll_p = 0.5 * (nll_fwd[i] + nll_rev[i])
        results.append(ForcedChoiceResult(
            user_prompt=user_prompts[i],
            think_text=think_fwd,
            think_text_rev=think_rev,
            lp_fwd={foundations[k]: lp_f[k] for k in range(K)},
            lp_rev={foundations[k]: lp_r[k] for k in range(K)},
            score={foundations[k]: score[k] for k in range(K)},
            p=p,
            top1=top1,
            margin=float(margin),
            think_tokens=n_fwd,
            emitted_close=close_fwd,
            nll_prompt=float(nll_p),
        ))

    return results