diff --git a/decnet/profiler/behave_shell/_features/__init__.py b/decnet/profiler/behave_shell/_features/__init__.py index 13cd111e..1f9d8e24 100644 --- a/decnet/profiler/behave_shell/_features/__init__.py +++ b/decnet/profiler/behave_shell/_features/__init__.py @@ -26,6 +26,7 @@ from decnet.profiler.behave_shell._features.cognitive import ( ) from decnet.profiler.behave_shell._features.emotional_valence import ( arousal, + stress_response, valence, ) from decnet.profiler.behave_shell._features.environmental import ( @@ -97,4 +98,5 @@ FEATURES: tuple[FeatureFn, ...] = ( multi_actor_indicators, valence, arousal, + stress_response, ) diff --git a/decnet/profiler/behave_shell/_features/emotional_valence.py b/decnet/profiler/behave_shell/_features/emotional_valence.py index b255c88a..10fe8c9f 100644 --- a/decnet/profiler/behave_shell/_features/emotional_valence.py +++ b/decnet/profiler/behave_shell/_features/emotional_valence.py @@ -26,6 +26,9 @@ from decnet.profiler.behave_shell._thresholds import ( AROUSAL_FAST_IAT_S, AROUSAL_MIN_IATS, EMOTIONAL_VALENCE_CONFIDENCE_CAP, + STRESS_DISTRESS_RATIO_MIN, + STRESS_EUSTRESS_RATIO_MIN, + STRESS_MIN_ERRORED_WITH_IATS, VALENCE_FULL_CONFIDENCE_MIN, VALENCE_MIN_HITS, VALENCE_MIN_TYPED_CHARS, @@ -117,3 +120,69 @@ def arousal(ctx: SessionContext) -> Iterator[Observation]: value=value, confidence=_cap_soft(raw), ) + + +def stress_response(ctx: SessionContext) -> Iterator[Observation]: + """Emit ``emotional_valence.stress_response`` ∈ {none, + eustress_positive, distress_negative}. + + Compare typing speed *after* an errored command vs the session + baseline: + + * For each errored command at index ``i``, gather + ``ctx.intra_command_iats[i+1]`` — the response command's intra- + command IATs. + * Baseline: median of all intra-command IATs from commands NOT + immediately following an errored command. + + Verdict by ratio of post-error / baseline: + + * ratio ≥ ``STRESS_EUSTRESS_RATIO_MIN`` (1.20) → ``eustress_positive`` + (slowed down — recovered, deliberate). + * ratio ≤ ``1 / STRESS_DISTRESS_RATIO_MIN`` → ``distress_negative`` + (sped up — anxious, mashing keys). + * otherwise → ``none``. + + Skip emission when no commands. Confidence hard-capped at 0.50; + 0.30 below ``STRESS_MIN_ERRORED_WITH_IATS`` (2) errored commands + with non-empty post-error IAT data. + """ + if not ctx.commands: + return + post_error_iats: list[float] = [] + baseline_iats: list[float] = [] + n = len(ctx.commands) + qualifying_errored = 0 + for i, cmd in enumerate(ctx.commands): + is_post_error = i > 0 and ctx.commands[i - 1].errored + iats = list(ctx.intra_command_iats[i]) if i < len(ctx.intra_command_iats) else [] + if is_post_error: + if iats: + qualifying_errored += 1 + post_error_iats.extend(iats) + else: + baseline_iats.extend(iats) + # mypy: silence unused-var on n / cmd (kept for clarity) + _ = (n, cmd) + if not post_error_iats or not baseline_iats: + value = "none" + else: + med_post = statistics.median(post_error_iats) + med_base = statistics.median(baseline_iats) + if med_base <= 0.0: + value = "none" + else: + ratio = med_post / med_base + if ratio >= STRESS_EUSTRESS_RATIO_MIN: + value = "eustress_positive" + elif ratio <= 1.0 / STRESS_DISTRESS_RATIO_MIN: + value = "distress_negative" + else: + value = "none" + raw = 0.50 if qualifying_errored >= STRESS_MIN_ERRORED_WITH_IATS else 0.30 + yield make_observation( + ctx, + primitive="emotional_valence.stress_response", + value=value, + confidence=_cap_soft(raw), + ) diff --git a/tests/profiler/behave_shell/test_emotional_valence_stress_response.py b/tests/profiler/behave_shell/test_emotional_valence_stress_response.py new file mode 100644 index 00000000..04f04d5c --- /dev/null +++ b/tests/profiler/behave_shell/test_emotional_valence_stress_response.py @@ -0,0 +1,89 @@ +"""Step G.7: ``emotional_valence.stress_response`` ∈ {none, +eustress_positive, distress_negative}. + +Hard 0.5 confidence cap. +""" +from __future__ import annotations + +from decnet.profiler.behave_shell import extract_session +from decnet.profiler.behave_shell._parse import AsciinemaEvent + + +PRIMITIVE = "emotional_valence.stress_response" + + +def _of(observations: list, primitive: str): + obs = [o for o in observations if o.primitive == primitive] + assert len(obs) == 1, f"expected exactly one {primitive}, got {len(obs)}" + return obs[0] + + +def _typed(text: str, t0: float, dt: float = 0.05) -> list[AsciinemaEvent]: + return [(t0 + i * dt, "i", c) for i, c in enumerate(text)] + + +def _cmd_with_error(token: str, t0: float, dt: float = 0.05) -> list[AsciinemaEvent]: + """Emit one command followed by an error-fingerprint output line.""" + events = _typed(f"{token}\r", t0=t0, dt=dt) + cmd_end = t0 + len(token) * dt + events.append((cmd_end + 0.10, "o", "bash: command not found\nanti@host:~$ ")) + return events + + +def _cmd_ok(token: str, t0: float, dt: float = 0.05) -> list[AsciinemaEvent]: + events = _typed(f"{token}\r", t0=t0, dt=dt) + cmd_end = t0 + len(token) * dt + events.append((cmd_end + 0.10, "o", "out\nanti@host:~$ ")) + return events + + +def test_no_commands_no_emission() -> None: + out = list(extract_session([(0.0, "i", "x")], sid="g7-empty")) + assert [o for o in out if o.primitive == PRIMITIVE] == [] + + +def test_no_errors_emits_none() -> None: + events = _cmd_ok("hostname", t0=0.0) + _cmd_ok("date", t0=2.0) + obs = _of(list(extract_session(events, sid="g7-noerr")), PRIMITIVE) + assert obs.value == "none" + + +def test_distress_post_error_speed_up() -> None: + """After an error, the operator types the next command faster.""" + events = ( + _cmd_ok("hostname", t0=0.0, dt=0.20) + + _cmd_ok("date", t0=2.0, dt=0.20) + + _cmd_with_error("foobar", t0=4.0, dt=0.20) + + _cmd_ok("hostname", t0=6.0, dt=0.04) # post-error: fast + + _cmd_with_error("baz", t0=8.0, dt=0.20) + + _cmd_ok("date", t0=10.0, dt=0.04) # post-error: fast + + _cmd_ok("hostname", t0=12.0, dt=0.20) + ) + obs = _of(list(extract_session(events, sid="g7-dist")), PRIMITIVE) + assert obs.value == "distress_negative" + + +def test_eustress_post_error_slow_down() -> None: + """After an error, the operator slows down — deliberate recovery.""" + events = ( + _cmd_ok("hostname", t0=0.0, dt=0.04) + + _cmd_ok("date", t0=2.0, dt=0.04) + + _cmd_with_error("foobar", t0=4.0, dt=0.04) + + _cmd_ok("hostname", t0=6.0, dt=0.20) # post-error: slow + + _cmd_with_error("baz", t0=8.0, dt=0.04) + + _cmd_ok("date", t0=10.0, dt=0.20) # post-error: slow + + _cmd_ok("hostname", t0=12.0, dt=0.04) + ) + obs = _of(list(extract_session(events, sid="g7-eu")), PRIMITIVE) + assert obs.value == "eustress_positive" + + +def test_confidence_capped_at_05() -> None: + events = ( + _cmd_with_error("foobar", t0=0.0, dt=0.05) + + _cmd_ok("hostname", t0=2.0, dt=0.20) + + _cmd_with_error("baz", t0=4.0, dt=0.05) + + _cmd_ok("date", t0=6.0, dt=0.20) + ) + obs = _of(list(extract_session(events, sid="g7-cap")), PRIMITIVE) + assert obs.confidence <= 0.50