From d90c8b70ce8c4ded4a41cfa84d98d2a210d3b1f0 Mon Sep 17 00:00:00 2001 From: anti Date: Sun, 3 May 2026 21:24:13 -0400 Subject: [PATCH] feat(profiler/behave_shell): emit motor.keystroke_cadence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BEHAVE-EXTRACTOR.md Phase B Step B.1. * SessionContext gains typing_bursts: tuple[tuple[float, ...], ...] built by _split_typing_bursts(iats) — splits at gaps > IKI_THINK_MAX_S (1.5s) and drops bursts of fewer than 3 IATs. Mirrors prototype's _split_into_bursts at BEHAVE/prototype_extractors/shell/extract.py:275. * _features/motor.py:keystroke_cadence(ctx) emits one Observation in {steady, bursty, hunt_and_peck, machine}. Median CV across typing bursts; mean IKI < IKI_MACHINE_MAX_S paired with CV < CV_MACHINE_MAX → machine. Confidence 0.85/0.70/0.65/0.60 per the prototype's calibration history. * < MIN_INPUTS_FOR_CADENCE inputs or zero typing bursts → skip emission. v0.1 emits only the burst-CV variant; the prototype's NAIVE session-CV variant is parked for v0.2. * Calibration grid widened (PHASE_A_PRIMITIVES → PHASE_AB_PRIMITIVES) to include motor.keystroke_cadence. Grid green across all five shards. Tests: too-few-inputs → no emit, all-think-pauses → no burst → no emit, uniform IATs → steady, sub-5ms → machine, mixed-pace → bursty, extreme bimodal → hunt_and_peck. --- decnet/profiler/behave_shell/_ctx.py | 22 +++++ .../behave_shell/_features/__init__.py | 2 + .../profiler/behave_shell/_features/motor.py | 50 +++++++++++ decnet/profiler/behave_shell/_thresholds.py | 15 ++++ .../behave_shell/test_calibration_grid.py | 9 +- .../test_motor_keystroke_cadence.py | 85 +++++++++++++++++++ 6 files changed, 180 insertions(+), 3 deletions(-) create mode 100644 tests/profiler/behave_shell/test_motor_keystroke_cadence.py diff --git a/decnet/profiler/behave_shell/_ctx.py b/decnet/profiler/behave_shell/_ctx.py index 93c35a2d..fcca6163 100644 --- a/decnet/profiler/behave_shell/_ctx.py +++ b/decnet/profiler/behave_shell/_ctx.py @@ -20,6 +20,7 @@ from decnet.profiler.behave_shell._parse import ( hash_token, ) from decnet.profiler.behave_shell._thresholds import ( + IKI_THINK_MAX_S, PASTE_BURST_MAX_IAT_S, PASTE_MIN_CHARS_PER_EVENT, ) @@ -47,6 +48,9 @@ class SessionContext: inter_cmd_iats: tuple[float, ...] = field(default_factory=tuple) output_per_cmd: tuple[int, ...] = field(default_factory=tuple) + # Step B.1 derivations — typing bursts (IATs split at think-pauses) + typing_bursts: tuple[tuple[float, ...], ...] = field(default_factory=tuple) + def _detect_paste_bursts( inputs: list[AsciinemaEvent], @@ -102,6 +106,22 @@ def _detect_paste_bursts( return tuple(bursts), paste_count +def _split_typing_bursts(iats: tuple[float, ...]) -> tuple[tuple[float, ...], ...]: + """Split a flat IAT sequence at gaps > IKI_THINK_MAX_S. + + Drops bursts of fewer than 3 IATs — too short to compute a stable + CV. Mirrors BEHAVE prototype's ``_split_into_bursts``. + """ + bursts: list[list[float]] = [[]] + for x in iats: + if x > IKI_THINK_MAX_S: + if bursts[-1]: + bursts.append([]) + else: + bursts[-1].append(x) + return tuple(tuple(b) for b in bursts if len(b) >= 3) + + def _segment_commands(inputs: list[AsciinemaEvent]) -> tuple[Command, ...]: """Walk input events, splitting on ``\\r`` / ``\\n`` into commands. @@ -179,6 +199,7 @@ def build_session_context( max(0.0, inputs[i][0] - inputs[i - 1][0]) for i in range(1, len(inputs)) ) paste_bursts, paste_count = _detect_paste_bursts(inputs) + typing_bursts = _split_typing_bursts(iats) commands = _segment_commands(inputs) inter_cmd_iats = tuple( max(0.0, commands[i + 1].start_ts - commands[i].end_ts) @@ -204,4 +225,5 @@ def build_session_context( commands=commands, inter_cmd_iats=inter_cmd_iats, output_per_cmd=output_per_cmd, + typing_bursts=typing_bursts, ) diff --git a/decnet/profiler/behave_shell/_features/__init__.py b/decnet/profiler/behave_shell/_features/__init__.py index 2c7d07f4..94cec4ab 100644 --- a/decnet/profiler/behave_shell/_features/__init__.py +++ b/decnet/profiler/behave_shell/_features/__init__.py @@ -19,6 +19,7 @@ from decnet.profiler.behave_shell._features.cognitive import ( ) from decnet.profiler.behave_shell._features.motor import ( input_modality, + keystroke_cadence, paste_burst_rate, ) @@ -27,6 +28,7 @@ FeatureFn = Callable[[SessionContext], Iterable[Observation]] FEATURES: tuple[FeatureFn, ...] = ( input_modality, paste_burst_rate, + keystroke_cadence, inter_command_latency_class, command_branch_diversity, feedback_loop_engagement, diff --git a/decnet/profiler/behave_shell/_features/motor.py b/decnet/profiler/behave_shell/_features/motor.py index f808c895..36cc55e2 100644 --- a/decnet/profiler/behave_shell/_features/motor.py +++ b/decnet/profiler/behave_shell/_features/motor.py @@ -2,9 +2,12 @@ Step 2: ``motor.input_modality`` — typed / pasted / mixed. Step 3: ``motor.paste_burst_rate`` — none / occasional / habitual. +Step B.1: ``motor.keystroke_cadence`` — steady / bursty / hunt_and_peck / machine. """ from __future__ import annotations +import statistics +from itertools import chain from typing import Iterator from decnet_behave_core.spec.envelope import Observation @@ -12,6 +15,11 @@ from decnet_behave_core.spec.envelope import Observation from decnet.profiler.behave_shell._ctx import SessionContext from decnet.profiler.behave_shell._features._emit import make_observation from decnet.profiler.behave_shell._thresholds import ( + CV_BURSTY_MAX, + CV_MACHINE_MAX, + CV_STEADY_MAX, + IKI_MACHINE_MAX_S, + MIN_INPUTS_FOR_CADENCE, MODALITY_PASTED_MIN, MODALITY_TYPED_MAX, PASTE_RATE_HABITUAL_MIN, @@ -76,3 +84,45 @@ def paste_burst_rate(ctx: SessionContext) -> Iterator[Observation]: value=level, confidence=confidence, ) + + +def keystroke_cadence(ctx: SessionContext) -> Iterator[Observation]: + """Emit ``motor.keystroke_cadence`` ∈ {steady, bursty, hunt_and_peck, machine}. + + Median CV of within-typing-burst IATs (bursts split at gaps > + ``IKI_THINK_MAX_S`` so think-pauses between commands don't + inflate the variance). Pasted-only sessions and sessions below + ``MIN_INPUTS_FOR_CADENCE`` skip emission — no honest cadence + available. + + v0.1 emits only the burst-CV variant. The prototype's NAIVE + session-CV variant (lower confidence, second emission per + primitive) is parked for v0.2. + """ + if len(ctx.input_events) < MIN_INPUTS_FOR_CADENCE: + return + if not ctx.typing_bursts: + return + burst_cvs: list[float] = [] + for b in ctx.typing_bursts: + m = statistics.fmean(b) + if m > 0: + burst_cvs.append(statistics.pstdev(b) / m) + if not burst_cvs: + return + cv = statistics.median(burst_cvs) + mean_iki = statistics.fmean(chain.from_iterable(ctx.typing_bursts)) + if mean_iki < IKI_MACHINE_MAX_S and cv < CV_MACHINE_MAX: + value, confidence = "machine", 0.85 + elif cv < CV_STEADY_MAX: + value, confidence = "steady", 0.70 + elif cv < CV_BURSTY_MAX: + value, confidence = "bursty", 0.65 + else: + value, confidence = "hunt_and_peck", 0.60 + yield make_observation( + ctx, + primitive="motor.keystroke_cadence", + value=value, + confidence=confidence, + ) diff --git a/decnet/profiler/behave_shell/_thresholds.py b/decnet/profiler/behave_shell/_thresholds.py index b9281640..9a5a7e2f 100644 --- a/decnet/profiler/behave_shell/_thresholds.py +++ b/decnet/profiler/behave_shell/_thresholds.py @@ -75,3 +75,18 @@ FEEDBACK_MIN_PAIRS: int = 5 # via Hartigan dip is filed for v0.2). PAUSE_CV_METRONOMIC_MAX: float = 0.40 PAUSE_CV_BIMODAL_MIN: float = 1.50 + +# ── motor.keystroke_cadence (Step B.1) ────────────────────────────────────── +# Typing bursts split at gaps > IKI_THINK_MAX_S so think-pauses between +# commands don't inflate the within-burst CV. Mirrors the prototype's +# _split_into_bursts (BEHAVE/prototype_extractors/shell/extract.py:275-286). +IKI_THINK_MAX_S: float = 1.50 +# Sub-human floor for the "machine" classification — only paired with a +# pathologically uniform CV, since real humans never produce sub-5ms IATs +# in a sustained burst. +IKI_MACHINE_MAX_S: float = 0.005 +CV_MACHINE_MAX: float = 0.05 +CV_STEADY_MAX: float = 0.50 +CV_BURSTY_MAX: float = 1.50 +# Need this many input events before we'll claim a cadence at all. +MIN_INPUTS_FOR_CADENCE: int = 5 diff --git a/tests/profiler/behave_shell/test_calibration_grid.py b/tests/profiler/behave_shell/test_calibration_grid.py index 7e1e16b0..5355e991 100644 --- a/tests/profiler/behave_shell/test_calibration_grid.py +++ b/tests/profiler/behave_shell/test_calibration_grid.py @@ -31,13 +31,16 @@ from decnet.profiler.behave_shell import extract_session from decnet.profiler.behave_shell._parse import parse_shard_line -PHASE_A_PRIMITIVES: frozenset[str] = frozenset({ +PHASE_AB_PRIMITIVES: frozenset[str] = frozenset({ + # Phase A — calibration floor "motor.input_modality", "motor.paste_burst_rate", "cognitive.inter_command_latency_class", "cognitive.command_branch_diversity", "cognitive.feedback_loop_engagement", "cognitive.inter_command_consistency", + # Phase B — motor.* completion (lands one primitive per commit) + "motor.keystroke_cadence", }) @@ -105,7 +108,7 @@ def test_shard_emits_all_phase_a_primitives( obs = _all_observations(path) assert obs, f"{class_label}: extractor produced zero observations" seen = {o.primitive for o in obs} - missing = PHASE_A_PRIMITIVES - seen + missing = PHASE_AB_PRIMITIVES - seen assert not missing, ( f"{class_label} ({shard_file}) missing primitives: " f"{sorted(missing)}" @@ -142,7 +145,7 @@ def test_shards_are_discriminative_across_classes( # At least one primitive should produce different majority values # across the present classes. discriminative_primitives: list[str] = [] - for prim in PHASE_A_PRIMITIVES: + for prim in PHASE_AB_PRIMITIVES: values = {by_class[c].get(prim) for c in by_class if prim in by_class[c]} if len(values) >= 2: discriminative_primitives.append(prim) diff --git a/tests/profiler/behave_shell/test_motor_keystroke_cadence.py b/tests/profiler/behave_shell/test_motor_keystroke_cadence.py new file mode 100644 index 00000000..96a71776 --- /dev/null +++ b/tests/profiler/behave_shell/test_motor_keystroke_cadence.py @@ -0,0 +1,85 @@ +"""Step B.1: ``motor.keystroke_cadence``.""" +from __future__ import annotations + +import random + +from decnet.profiler.behave_shell import extract_session +from decnet.profiler.behave_shell._parse import AsciinemaEvent + + +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_events(iats: list[float], terminator: bool = True) -> list[AsciinemaEvent]: + """Build a typed input stream where consecutive single-char events are + separated by ``iats``.""" + events: list[AsciinemaEvent] = [] + t = 0.0 + events.append((t, "i", "a")) + for x in iats: + t += x + events.append((t, "i", "b")) + if terminator: + events.append((t + 0.1, "i", "\r")) + return events + + +def test_too_few_inputs_no_emission() -> None: + out = list(extract_session(_typed_events([0.1, 0.1]), sid="cad-low")) + assert [o for o in out if o.primitive == "motor.keystroke_cadence"] == [] + + +def test_huge_think_pauses_yield_no_typing_bursts() -> None: + # Two events 5s apart → no IAT under IKI_THINK_MAX_S, and only 1 + # IAT total — below the 3-IAT-per-burst minimum. No burst, no emit. + events: list[AsciinemaEvent] = [ + (0.0, "i", "a"), + (5.0, "i", "b"), + (10.0, "i", "c"), + (15.0, "i", "d"), + (20.0, "i", "e"), + ] + out = list(extract_session(events, sid="cad-no-bursts")) + assert [o for o in out if o.primitive == "motor.keystroke_cadence"] == [] + + +def test_uniform_iats_emit_steady() -> None: + iats = [0.15] * 12 + out = list(extract_session(_typed_events(iats), sid="cad-steady")) + obs = _of(out, "motor.keystroke_cadence") + assert obs.value == "steady" + assert obs.confidence == 0.70 + + +def test_machine_iats_emit_machine() -> None: + # Sub-5ms IATs with near-zero CV — no terminator IAT to inflate the + # variance away from machine + iats = [0.002] * 20 + out = list(extract_session(_typed_events(iats, terminator=False), sid="cad-machine")) + obs = _of(out, "motor.keystroke_cadence") + assert obs.value == "machine" + assert obs.confidence == 0.85 + + +def test_bursty_iats_emit_bursty() -> None: + # Mean ~0.15 with moderate variance, CV between 0.5 and 1.5 + rng = random.Random(42) + iats = [] + for _ in range(20): + # Mostly fast, occasionally slow → CV in the bursty band + iats.append(rng.choice([0.05, 0.05, 0.05, 0.30, 0.50])) + out = list(extract_session(_typed_events(iats), sid="cad-bursty")) + obs = _of(out, "motor.keystroke_cadence") + assert obs.value == "bursty" + + +def test_hunt_and_peck_iats_emit_hunt_and_peck() -> None: + # CV >= 1.5: extreme bimodal (very-fast + very-slow within burst). + # Most IATs are tiny; a few are ~10x the mean — drives stdev/mean above 1.5. + iats = [0.01] * 15 + [1.4] * 5 + out = list(extract_session(_typed_events(iats, terminator=False), sid="cad-hp")) + obs = _of(out, "motor.keystroke_cadence") + assert obs.value == "hunt_and_peck"