From d04f91cd8c06cab46b6b478b5c12ce29c39cf6c0 Mon Sep 17 00:00:00 2001 From: anti Date: Sun, 3 May 2026 21:27:46 -0400 Subject: [PATCH] feat(profiler/behave_shell): emit motor.error_correction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BEHAVE-EXTRACTOR.md Phase B Step B.3. Replaces the prototype's two-line "0 vs >0 backspaces" placeholder with a backspace-timing classifier that honours the registry's full vocabulary. * SessionContext gains backspace_count, backspace_iats (IAT from each backspace back to the preceding non-backspace input event), and kill_line_count (^U / ^W). Built by _scan_correction_signals, which retains only counts and timing aggregates — no character data leaves the helper, in line with the BEHAVE PII discipline. * _features/motor.py:error_correction(ctx) emits one Observation in {immediate, deferred, absent, route_around}. - 0 backspaces + ≥1 ^U/^W → route_around (rewrite, not correct) - 0 backspaces + 0 kill-lines → absent - backspaces with median IAT ≤ 500 ms → immediate - slower → deferred Confidence 0.65 / 0.65 / 0.55 / 0.55. * < 3 inputs → skip emit. * Calibration grid widened to include motor.error_correction; green across all five shards. Tests cover all four buckets, the < 3 inputs skip, and the PII regression (raw command body never appears in the serialised observation). --- decnet/profiler/behave_shell/_ctx.py | 40 ++++++++++ .../behave_shell/_features/__init__.py | 2 + .../profiler/behave_shell/_features/motor.py | 39 +++++++++ decnet/profiler/behave_shell/_thresholds.py | 6 ++ .../behave_shell/test_calibration_grid.py | 1 + .../test_motor_error_correction.py | 79 +++++++++++++++++++ 6 files changed, 167 insertions(+) create mode 100644 tests/profiler/behave_shell/test_motor_error_correction.py diff --git a/decnet/profiler/behave_shell/_ctx.py b/decnet/profiler/behave_shell/_ctx.py index fcca6163..04ee4c13 100644 --- a/decnet/profiler/behave_shell/_ctx.py +++ b/decnet/profiler/behave_shell/_ctx.py @@ -51,6 +51,11 @@ class SessionContext: # Step B.1 derivations — typing bursts (IATs split at think-pauses) typing_bursts: tuple[tuple[float, ...], ...] = field(default_factory=tuple) + # Step B.3 derivations — error-correction signals + backspace_count: int = 0 + backspace_iats: tuple[float, ...] = field(default_factory=tuple) + kill_line_count: int = 0 + def _detect_paste_bursts( inputs: list[AsciinemaEvent], @@ -106,6 +111,37 @@ def _detect_paste_bursts( return tuple(bursts), paste_count +_BACKSPACE_CHARS = ("\x7f", "\x08") +_KILL_LINE_CHARS = ("\x15", "\x17") + + +def _scan_correction_signals( + inputs: list[AsciinemaEvent], +) -> tuple[int, tuple[float, ...], int]: + """Walk input events char-by-char, count backspaces / kill-lines / + timing IATs. + + PII discipline: only counts and IATs leave this function — no + character data is retained or returned. + """ + backspace_count = 0 + kill_line_count = 0 + iats: list[float] = [] + last_non_bs_t: float | None = None + for t, _kind, data in inputs: + for c in data: + if c in _BACKSPACE_CHARS: + backspace_count += 1 + if last_non_bs_t is not None: + iats.append(max(0.0, t - last_non_bs_t)) + elif c in _KILL_LINE_CHARS: + kill_line_count += 1 + last_non_bs_t = t + else: + last_non_bs_t = t + return backspace_count, tuple(iats), kill_line_count + + def _split_typing_bursts(iats: tuple[float, ...]) -> tuple[tuple[float, ...], ...]: """Split a flat IAT sequence at gaps > IKI_THINK_MAX_S. @@ -200,6 +236,7 @@ def build_session_context( ) paste_bursts, paste_count = _detect_paste_bursts(inputs) typing_bursts = _split_typing_bursts(iats) + backspace_count, backspace_iats, kill_line_count = _scan_correction_signals(inputs) commands = _segment_commands(inputs) inter_cmd_iats = tuple( max(0.0, commands[i + 1].start_ts - commands[i].end_ts) @@ -226,4 +263,7 @@ def build_session_context( inter_cmd_iats=inter_cmd_iats, output_per_cmd=output_per_cmd, typing_bursts=typing_bursts, + backspace_count=backspace_count, + backspace_iats=backspace_iats, + kill_line_count=kill_line_count, ) diff --git a/decnet/profiler/behave_shell/_features/__init__.py b/decnet/profiler/behave_shell/_features/__init__.py index 4df7e40d..188fd4b5 100644 --- a/decnet/profiler/behave_shell/_features/__init__.py +++ b/decnet/profiler/behave_shell/_features/__init__.py @@ -18,6 +18,7 @@ from decnet.profiler.behave_shell._features.cognitive import ( inter_command_latency_class, ) from decnet.profiler.behave_shell._features.motor import ( + error_correction, input_modality, keystroke_cadence, motor_stability, @@ -31,6 +32,7 @@ FEATURES: tuple[FeatureFn, ...] = ( paste_burst_rate, keystroke_cadence, motor_stability, + error_correction, 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 48890d03..982029b8 100644 --- a/decnet/profiler/behave_shell/_features/motor.py +++ b/decnet/profiler/behave_shell/_features/motor.py @@ -15,6 +15,7 @@ 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 ( + BACKSPACE_IMMEDIATE_MAX_S, CV_BURSTY_MAX, CV_MACHINE_MAX, CV_STEADY_MAX, @@ -166,3 +167,41 @@ def motor_stability(ctx: SessionContext) -> Iterator[Observation]: value=value, confidence=confidence, ) + + +def error_correction(ctx: SessionContext) -> Iterator[Observation]: + """Emit ``motor.error_correction`` ∈ {immediate, deferred, absent, route_around}. + + Backspace timing relative to the preceding non-backspace key: + + * 0 backspaces + ≥1 ^U/^W → ``route_around`` (operator killed + the line and rewrote rather than correcting in place). + * 0 backspaces + 0 ^U/^W → ``absent`` (no correction observed). + * Backspaces with median IAT ≤ ``BACKSPACE_IMMEDIATE_MAX_S`` + (500 ms) → ``immediate`` (caught the typo mid-keystroke). + * Slower → ``deferred`` (paused, noticed, then went back). + + < 3 input events → skip emission. + """ + if len(ctx.input_events) < 3: + return + if ctx.backspace_count == 0: + if ctx.kill_line_count > 0: + value, confidence = "route_around", 0.55 + else: + value, confidence = "absent", 0.65 + else: + if ctx.backspace_iats: + med = statistics.median(ctx.backspace_iats) + else: + med = float("inf") + if med <= BACKSPACE_IMMEDIATE_MAX_S: + value, confidence = "immediate", 0.65 + else: + value, confidence = "deferred", 0.55 + yield make_observation( + ctx, + primitive="motor.error_correction", + value=value, + confidence=confidence, + ) diff --git a/decnet/profiler/behave_shell/_thresholds.py b/decnet/profiler/behave_shell/_thresholds.py index 7f13e94a..24aa047f 100644 --- a/decnet/profiler/behave_shell/_thresholds.py +++ b/decnet/profiler/behave_shell/_thresholds.py @@ -98,3 +98,9 @@ MIN_INPUTS_FOR_CADENCE: int = 5 # of sub-floor IATs flags double-press / motor twitch / stuck-key. TREMOR_FAST_FLOOR_S: float = 0.030 TREMOR_RATE_MIN: float = 0.10 # ≥10% sub-floor → tremor + +# ── motor.error_correction (Step B.3) ─────────────────────────────────────── +# Backspace within this many seconds of the preceding key = "caught the +# typo mid-keystroke" (immediate). Beyond this = the operator paused, +# noticed, then went back (deferred). +BACKSPACE_IMMEDIATE_MAX_S: float = 0.50 diff --git a/tests/profiler/behave_shell/test_calibration_grid.py b/tests/profiler/behave_shell/test_calibration_grid.py index d6579c58..ed9353a3 100644 --- a/tests/profiler/behave_shell/test_calibration_grid.py +++ b/tests/profiler/behave_shell/test_calibration_grid.py @@ -42,6 +42,7 @@ PHASE_AB_PRIMITIVES: frozenset[str] = frozenset({ # Phase B — motor.* completion (lands one primitive per commit) "motor.keystroke_cadence", "motor.motor_stability", + "motor.error_correction", }) diff --git a/tests/profiler/behave_shell/test_motor_error_correction.py b/tests/profiler/behave_shell/test_motor_error_correction.py new file mode 100644 index 00000000..549d5b22 --- /dev/null +++ b/tests/profiler/behave_shell/test_motor_error_correction.py @@ -0,0 +1,79 @@ +"""Step B.3: ``motor.error_correction``.""" +from __future__ import annotations + +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 test_too_few_inputs_no_emission() -> None: + events: list[AsciinemaEvent] = [(0.0, "i", "a"), (0.1, "i", "b")] + out = list(extract_session(events, sid="ec-low")) + assert [o for o in out if o.primitive == "motor.error_correction"] == [] + + +def test_no_backspaces_no_kill_emits_absent() -> None: + events: list[AsciinemaEvent] = [(i * 0.1, "i", c) for i, c in enumerate("hello\r")] + out = list(extract_session(events, sid="ec-absent")) + obs = _of(out, "motor.error_correction") + assert obs.value == "absent" + assert obs.confidence == 0.65 + + +def test_kill_line_with_no_backspaces_emits_route_around() -> None: + events: list[AsciinemaEvent] = [ + (0.0, "i", "l"), + (0.1, "i", "s"), + (0.2, "i", "\x15"), # ^U — kill line + (0.3, "i", "p"), + (0.4, "i", "s"), + (0.5, "i", "\r"), + ] + out = list(extract_session(events, sid="ec-route")) + obs = _of(out, "motor.error_correction") + assert obs.value == "route_around" + + +def test_backspace_within_500ms_emits_immediate() -> None: + events: list[AsciinemaEvent] = [ + (0.0, "i", "h"), + (0.1, "i", "e"), + (0.2, "i", "y"), + (0.30, "i", "\x7f"), # backspace 100ms after 'y' — immediate + (0.4, "i", "l"), + (0.5, "i", "l"), + (0.6, "i", "o"), + (0.7, "i", "\r"), + ] + out = list(extract_session(events, sid="ec-immediate")) + obs = _of(out, "motor.error_correction") + assert obs.value == "immediate" + + +def test_backspace_after_long_pause_emits_deferred() -> None: + events: list[AsciinemaEvent] = [ + (0.0, "i", "h"), + (0.1, "i", "e"), + (0.2, "i", "y"), + # Backspace 2s after 'y' — deferred + (2.2, "i", "\x7f"), + (2.3, "i", "l"), + (2.4, "i", "l"), + (2.5, "i", "o"), + (2.6, "i", "\r"), + ] + out = list(extract_session(events, sid="ec-deferred")) + obs = _of(out, "motor.error_correction") + assert obs.value == "deferred" + + +def test_pii_no_command_bodies_in_observation() -> None: + events: list[AsciinemaEvent] = [(i * 0.1, "i", c) for i, c in enumerate("supersecret\r")] + out = list(extract_session(events, sid="ec-pii")) + obs = _of(out, "motor.error_correction") + assert "supersecret" not in obs.model_dump_json()