From 8161c67ec56cf45f61a1cdb6f5fd3dbcc0a04e55 Mon Sep 17 00:00:00 2001 From: anti Date: Sun, 3 May 2026 21:29:31 -0400 Subject: [PATCH] feat(profiler/behave_shell): emit motor.command_chunking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BEHAVE-EXTRACTOR.md Phase B Step B.4. First implementation — prototype doesn't ship this primitive. * SessionContext gains intra_command_iats: per-command tuple of IATs between consecutive input events whose timestamps fall inside [cmd.start_ts, cmd.end_ts). Excludes the terminator IAT. Built by _per_command_iats. * _features/motor.py:command_chunking(ctx) emits one Observation in {fluent, fragmented, single_command}. - 0 commands → skip emit - 1 command → single_command (registry-allowed point) - ≥2 commands → median CV across per-command typed-IATs; < CMD_CHUNKING_FLUENT_CV_MAX (0.50) → fluent, else fragmented - paste-only sessions (no command has ≥3 typed IATs) → skip emit (no honest within-command rhythm to measure) Confidence 0.80 / 0.65 / 0.60. * Calibration grid widened to include motor.command_chunking; green across all five shards. Phase B primitive set complete. Tests: no commands → skip, 1 command → single_command, uniform typing → fluent, alternating fast/slow → fragmented, paste-only multi-command → skip emit. --- decnet/profiler/behave_shell/_ctx.py | 29 ++++++++ .../behave_shell/_features/__init__.py | 2 + .../profiler/behave_shell/_features/motor.py | 47 ++++++++++++ decnet/profiler/behave_shell/_thresholds.py | 5 ++ .../behave_shell/test_calibration_grid.py | 1 + .../test_motor_command_chunking.py | 74 +++++++++++++++++++ 6 files changed, 158 insertions(+) create mode 100644 tests/profiler/behave_shell/test_motor_command_chunking.py diff --git a/decnet/profiler/behave_shell/_ctx.py b/decnet/profiler/behave_shell/_ctx.py index 04ee4c13..1e002384 100644 --- a/decnet/profiler/behave_shell/_ctx.py +++ b/decnet/profiler/behave_shell/_ctx.py @@ -56,6 +56,9 @@ class SessionContext: backspace_iats: tuple[float, ...] = field(default_factory=tuple) kill_line_count: int = 0 + # Step B.4 derivations — per-command intra-typing IATs + intra_command_iats: tuple[tuple[float, ...], ...] = field(default_factory=tuple) + def _detect_paste_bursts( inputs: list[AsciinemaEvent], @@ -191,6 +194,30 @@ def _segment_commands(inputs: list[AsciinemaEvent]) -> tuple[Command, ...]: return tuple(cmds) +def _per_command_iats( + commands: tuple[Command, ...], + inputs: list[AsciinemaEvent], +) -> tuple[tuple[float, ...], ...]: + """Per-command IATs between consecutive input events whose + timestamps fall in ``[cmd.start_ts, cmd.end_ts)``. + + Excludes the terminator IAT (the last event at ``cmd.end_ts`` is + the ``\\r``/``\\n`` itself). Returns one tuple per command. + """ + out: list[tuple[float, ...]] = [] + for cmd in commands: + prev_t: float | None = None + cmd_iats: list[float] = [] + for t, _kind, _data in inputs: + if t < cmd.start_ts or t >= cmd.end_ts: + continue + if prev_t is not None: + cmd_iats.append(max(0.0, t - prev_t)) + prev_t = t + out.append(tuple(cmd_iats)) + return tuple(out) + + def _output_bytes_between( outputs: list[AsciinemaEvent], start: float, @@ -246,6 +273,7 @@ def build_session_context( _output_bytes_between(outputs, commands[i].end_ts, commands[i + 1].start_ts) for i in range(len(commands) - 1) ) + intra_command_iats = _per_command_iats(commands, inputs) return SessionContext( sid=sid, @@ -266,4 +294,5 @@ def build_session_context( backspace_count=backspace_count, backspace_iats=backspace_iats, kill_line_count=kill_line_count, + intra_command_iats=intra_command_iats, ) diff --git a/decnet/profiler/behave_shell/_features/__init__.py b/decnet/profiler/behave_shell/_features/__init__.py index 188fd4b5..a8a0b693 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 ( + command_chunking, error_correction, input_modality, keystroke_cadence, @@ -33,6 +34,7 @@ FEATURES: tuple[FeatureFn, ...] = ( keystroke_cadence, motor_stability, error_correction, + command_chunking, 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 982029b8..31a6dd57 100644 --- a/decnet/profiler/behave_shell/_features/motor.py +++ b/decnet/profiler/behave_shell/_features/motor.py @@ -16,6 +16,7 @@ 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, + CMD_CHUNKING_FLUENT_CV_MAX, CV_BURSTY_MAX, CV_MACHINE_MAX, CV_STEADY_MAX, @@ -205,3 +206,49 @@ def error_correction(ctx: SessionContext) -> Iterator[Observation]: value=value, confidence=confidence, ) + + +def command_chunking(ctx: SessionContext) -> Iterator[Observation]: + """Emit ``motor.command_chunking`` ∈ {fluent, fragmented, single_command}. + + * 0 commands → skip (no honest answer). + * 1 command → ``single_command`` (registry-allowed, distinct from + the fluent/fragmented continuum that needs multiple commands). + * ≥2 commands → median CV across per-command intra-typing IATs; + below ``CMD_CHUNKING_FLUENT_CV_MAX`` → fluent, else fragmented. + + Skips emission if no command has ≥3 typed IATs to compute a CV + over (paste-driven sessions where every command arrived as one + bulk write — no honest within-command rhythm to measure). + """ + n = len(ctx.commands) + if n == 0: + return + if n == 1: + yield make_observation( + ctx, + primitive="motor.command_chunking", + value="single_command", + confidence=0.80, + ) + return + cvs: list[float] = [] + for iats in ctx.intra_command_iats: + if len(iats) < 3: + continue + m = statistics.fmean(iats) + if m > 0: + cvs.append(statistics.pstdev(iats) / m) + if not cvs: + return + cv = statistics.median(cvs) + if cv < CMD_CHUNKING_FLUENT_CV_MAX: + value, confidence = "fluent", 0.65 + else: + value, confidence = "fragmented", 0.60 + yield make_observation( + ctx, + primitive="motor.command_chunking", + value=value, + confidence=confidence, + ) diff --git a/decnet/profiler/behave_shell/_thresholds.py b/decnet/profiler/behave_shell/_thresholds.py index 24aa047f..860aea72 100644 --- a/decnet/profiler/behave_shell/_thresholds.py +++ b/decnet/profiler/behave_shell/_thresholds.py @@ -104,3 +104,8 @@ TREMOR_RATE_MIN: float = 0.10 # ≥10% sub-floor → tremor # typo mid-keystroke" (immediate). Beyond this = the operator paused, # noticed, then went back (deferred). BACKSPACE_IMMEDIATE_MAX_S: float = 0.50 + +# ── motor.command_chunking (Step B.4) ─────────────────────────────────────── +# Median CV of within-command IATs. Below this → fluent (steady within +# each command); above → fragmented (operator pauses mid-command). +CMD_CHUNKING_FLUENT_CV_MAX: float = 0.50 diff --git a/tests/profiler/behave_shell/test_calibration_grid.py b/tests/profiler/behave_shell/test_calibration_grid.py index ed9353a3..2f3443bf 100644 --- a/tests/profiler/behave_shell/test_calibration_grid.py +++ b/tests/profiler/behave_shell/test_calibration_grid.py @@ -43,6 +43,7 @@ PHASE_AB_PRIMITIVES: frozenset[str] = frozenset({ "motor.keystroke_cadence", "motor.motor_stability", "motor.error_correction", + "motor.command_chunking", }) diff --git a/tests/profiler/behave_shell/test_motor_command_chunking.py b/tests/profiler/behave_shell/test_motor_command_chunking.py new file mode 100644 index 00000000..3d9549c7 --- /dev/null +++ b/tests/profiler/behave_shell/test_motor_command_chunking.py @@ -0,0 +1,74 @@ +"""Step B.4: ``motor.command_chunking``.""" +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 _typed_command(start_ts: float, chars: str, iat: float) -> list[AsciinemaEvent]: + """Build a typed command starting at ``start_ts`` with uniform IAT + between chars. Terminates with ``\\r``.""" + events: list[AsciinemaEvent] = [] + t = start_ts + for c in chars: + events.append((t, "i", c)) + t += iat + events.append((t, "i", "\r")) + return events + + +def test_no_commands_no_emission() -> None: + out = list(extract_session([(0.0, "i", "ls")], sid="cc-empty")) + assert [o for o in out if o.primitive == "motor.command_chunking"] == [] + + +def test_single_command_emits_single_command() -> None: + events = _typed_command(0.0, "ls -la", 0.1) + out = list(extract_session(events, sid="cc-single")) + obs = _of(out, "motor.command_chunking") + assert obs.value == "single_command" + assert obs.confidence == 0.80 + + +def test_uniform_intra_command_typing_emits_fluent() -> None: + events = [] + for i in range(4): + events += _typed_command(i * 5.0, "ls -la", 0.1) + out = list(extract_session(events, sid="cc-fluent")) + obs = _of(out, "motor.command_chunking") + assert obs.value == "fluent" + + +def test_high_intra_command_variance_emits_fragmented() -> None: + # Per-command IATs drawn so within-command CV >= 0.50: alternating + # very-fast and very-slow keystrokes. + events: list[AsciinemaEvent] = [] + base = 0.0 + for cmd_idx in range(3): + t = base + for j, c in enumerate("hello"): + events.append((t, "i", c)) + # Alternate fast/slow → high CV inside each command + t += 0.05 if j % 2 == 0 else 0.50 + events.append((t, "i", "\r")) + base += 5.0 + out = list(extract_session(events, sid="cc-fragmented")) + obs = _of(out, "motor.command_chunking") + assert obs.value == "fragmented" + + +def test_paste_only_multi_command_no_emission() -> None: + # Each command arrives as one paste event — no within-command IATs + events: list[AsciinemaEvent] = [ + (0.0, "i", "echo aaaa\r"), + (1.0, "i", "echo bbbb\r"), + (2.0, "i", "echo cccc\r"), + ] + out = list(extract_session(events, sid="cc-paste-only")) + assert [o for o in out if o.primitive == "motor.command_chunking"] == []