Only Dad's Voice Cuts Her Off, When the Ear Can Tell

CUT_IN_TIMEOUT_S = 2.0
# A cut-in is about a second of Dad's speech, too little for the bands a whole
# turn is ruled by: on 132 past web cut-ins (§7.5) his voice scored 0.20 or
# more on 91% of them with a second heard after the trip, while other voices
# topped out near 0.24 on two-second chunks. Missing Dad costs more than a
# rare stranger cutting her off, as every loud run did before.
CUT_IN_DAD_LINE = 0.20


def cut_in_decision(heard: Mapping[str, Any] | None) -> tuple[bool, str]:
    """Whether a loud run on the open mic while she spoke may cut her off
    (Dad, 2026-10-08: noise was cutting her off). Only Dad's voice cuts:
    named his, sure or unsure, or scoring at least ``CUT_IN_DAD_LINE``
    against his voice. A stranger, Mom, her own voice coming back and a room
    with no speech in it leave her reading. Without Dad's enrolled voice the
    ear cannot tell, and every trip cuts, as before."""
    if not heard or DAD_PROFILE not in (heard.get("profiles") or []):
        return True, "unchecked"
    voices = [v for v in heard.get("voices") or [] if isinstance(v, Mapping)]
    if not voices:
        return False, "no_speech"
    if any(_is_dad(v) or _leans_dad(v) for v in voices):
        return True, "dad"
    return False, "not_dad"


def _leans_dad(voice: Mapping[str, Any]) -> bool:
    """Dad's score reaches the cut-in line and no other enrolled voice scores
    higher. Mom's own recording scored 0.20 against Dad's voice (2026-10-08),
    so the line alone would let her cut her daughter off."""
    scores = voice.get("scores")
    if not isinstance(scores, Mapping):
        return False
    dad = scores.get(DAD_PROFILE)
    if not isinstance(dad, int | float) or dad < CUT_IN_DAD_LINE:
        return False
    return all(
        not isinstance(score, int | float) or score <= dad
        for profile, score in scores.items()
        if profile != DAD_PROFILE
    )

/** Measured on 132 web cut-ins since 2026-09-27 (docs/PIPPA-SPEAKER-SEPARATION.md
 *  §7.5): with half a second after the trip the ear caught 68% of Dad's real
 *  cut-ins on the first ask, with a full second 91%. A miss is asked again
 *  when he keeps talking. */
export const CUT_IN_HOLD = {
  /** Chunks heard after the trip before the ear is asked: 1 s at 128 ms. */
  afterChunks: 8,
  /** What the ear is sent: the latest chunks, the run among them, ~2.3 s. */
  sendChunks: 18,
  /** An ear slower than this cuts. */
  timeoutMs: 2500,
  /** After a "not Dad", trips are ignored this long. */
  cooldownMs: 1000,
} as const;

# A barge-in trip held and asked, in miniature. The ruling is cwkPippa's
# cut_in_decision; the hold follows the web's cutInHold timings.
DAD_PROFILE = "dad"
CUT_IN_DAD_LINE = 0.20
TIMEOUT_S = 2.5        # the client cuts if the ear is slower than this
COOLDOWN_S = 1.0       # after "not Dad", trips are ignored this long


def is_dad(v):
    return v.get("profile_id") == DAD_PROFILE and v.get("band") in ("match", "uncertain")


def leans_dad(v):
    scores = v.get("scores") or {}
    dad = scores.get(DAD_PROFILE)
    if dad is None or dad < CUT_IN_DAD_LINE:
        return False
    return all(s <= dad for p, s in scores.items() if p != DAD_PROFILE)


def cut_in_decision(heard):
    if not heard or DAD_PROFILE not in heard.get("profiles", []):
        return True, "unchecked"
    voices = heard.get("voices") or []
    if not voices:
        return False, "no_speech"
    if any(is_dad(v) or leans_dad(v) for v in voices):
        return True, "dad"
    return False, "not_dad"


def hold(trip_at, answer, answer_after_s, quiet_until=0.0):
    """One trip: duck to a quarter, hear one more second, ask, settle."""
    if trip_at < quiet_until:
        return "ignored (cooling down)", quiet_until
    if answer_after_s > TIMEOUT_S:
        cut, reason = True, "timeout"
    else:
        cut, reason = cut_in_decision(answer)
    settled = trip_at + 1.0 + min(answer_after_s, TIMEOUT_S)
    if cut:
        return f"cut at {settled:.1f}s ({reason})", 0.0
    return f"reads on at full volume at {settled:.1f}s ({reason})", settled + COOLDOWN_S


profiles = ["dad", "mom"]
cases = [
    ("Dad, a short unsure cut-in", {"profiles": profiles, "voices": [{"profile_id": None, "band": "none", "scores": {"dad": 0.23, "mom": 0.08}}]}, 0.3),
    ("Mom, near Dad's line", {"profiles": profiles, "voices": [{"profile_id": None, "band": "none", "scores": {"dad": 0.20, "mom": 0.24}}]}, 0.3),
    ("a door slams", {"profiles": profiles, "voices": []}, 0.2),
    ("the ear is slow", {"profiles": profiles, "voices": []}, 4.0),
    ("no Dad voice enrolled", {"profiles": ["mom"], "voices": []}, 0.2),
]
for name, answer, after in cases:
    outcome, _ = hold(10.0, answer, after)
    print(f"{name:<28} -> {outcome}")

# Dad keeps talking after a "not Dad": the next trip inside the second waits.
_, quiet = hold(10.0, cases[2][1], 0.2)
print(f"{'a trip 0.5 s after it':<28} -> {hold(quiet - 0.5, cases[0][1], 0.3, quiet)[0]}")
print(f"{'a trip 1.5 s after it':<28} -> {hold(quiet + 0.5, cases[0][1], 0.3, quiet)[0]}")

# Output:
# Dad, a short unsure cut-in   -> cut at 11.3s (dad)
# Mom, near Dad's line         -> reads on at full volume at 11.3s (not_dad)
# a door slams                 -> reads on at full volume at 11.2s (no_speech)
# the ear is slow              -> cut at 13.5s (timeout)
# no Dad voice enrolled        -> cut at 11.2s (unchecked)
# a trip 0.5 s after it        -> ignored (cooling down)
# a trip 1.5 s after it        -> cut at 14.0s (dad)