feat(profiler/behave_shell): emit motor.error_correction

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

@@ -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,
     )

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,

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,
+    )

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

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",
 })
 
 

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()