engine/mapping.py

"""Step 3 of the living loop: deterministic signal -> delta mapping (F2).

Pure, documented, no model calls. Takes the appraisal dict produced by
engine/appraise.py and returns a list of (field, delta, reason) tuples to be
passed to engine/spec_bridge.mutate(). The spec engine (clamp + governance +
audit) has the final say - this module only PROPOSES deltas.

Mapping table (each rule capped to keep per-turn movement small and stable,
per MASTER_CHECKLIST F2 "Estabilidad: limitar magnitud de deltas por turno"):

| signal                       | field(s)                              | delta formula                     | max  |
|-------------------------------|-----------------------------------------|--------------------------------------|------|
| sentiment in [-1, 1]          | mood.tone, affect.valence                | sentiment * SENTIMENT_SCALE           | 0.05 |
| engagement in [0, 1]          | traits.extraversion                      | (engagement - 0.5) * EXTRA_SCALE      | 0.04 |
| engagement in [0, 1]          | traits.openness                          | (engagement - 0.5) * OPEN_SCALE       | 0.03 |
| correction & target != "none" | mapped trait, see TARGET_FIELD           | direction * CORRECTION_SCALE          | 0.08 |
"""

from __future__ import annotations

SENTIMENT_SCALE = 0.05
EXTRA_SCALE = 0.04
OPEN_SCALE = 0.03
CORRECTION_SCALE = 0.08

TARGET_FIELD = {
    "tone": "mood.tone",
    "openness": "traits.openness",
    "extraversion": "traits.extraversion",
    "agreeableness": "traits.agreeableness",
}

_MIN_DELTA = 1e-3


def signals_to_deltas(signals: dict) -> list[tuple[str, float, str]]:
    """Turn an appraisal dict into a list of (field, delta, reason).

    Deltas with magnitude below _MIN_DELTA are dropped so a neutral turn
    produces no audit-log noise.
    """
    deltas: list[tuple[str, float, str]] = []
    reason_suffix = str(signals.get("reason", "")).strip()

    sentiment = float(signals.get("sentiment", 0.0))
    if abs(sentiment) >= _MIN_DELTA:
        reason = f"sentiment={sentiment:+.2f}" + (f" ({reason_suffix})" if reason_suffix else "")
        deltas.append(("mood.tone", sentiment * SENTIMENT_SCALE, reason))
        deltas.append(("affect.valence", sentiment * SENTIMENT_SCALE, reason))

    engagement = float(signals.get("engagement", 0.0))
    centered = engagement - 0.5
    if abs(centered) >= _MIN_DELTA:
        reason = f"engagement={engagement:.2f}" + (f" ({reason_suffix})" if reason_suffix else "")
        deltas.append(("traits.extraversion", centered * EXTRA_SCALE, reason))
        deltas.append(("traits.openness", centered * OPEN_SCALE, reason))

    if signals.get("correction") and signals.get("direction", 0) != 0:
        field = TARGET_FIELD.get(signals.get("target", "none"))
        if field is not None:
            direction = signals["direction"]
            reason = f"user correction: {reason_suffix or 'requested change'}"
            deltas.append((field, direction * CORRECTION_SCALE, reason))

    return deltas