From 4257f7b6e207430b85ed1584cd9116676be34000 Mon Sep 17 00:00:00 2001 From: anti Date: Mon, 4 May 2026 00:33:44 -0400 Subject: [PATCH] feat(profiler/behave_shell): emit environmental.terminal_multiplexer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scans RAW output (multiplexer escapes are themselves ANSI; never strip first) for tmux markers (DCS passthrough, focus-reporting, window-title with tmux marker) and screen markers (DCS, screen-OSC). Detected → tmux/screen at 0.85; otherwise → none at 0.55. Skips emission entirely when no commands — silence on a pure-echo or empty session, per the smoke gates. When both detected (nested mux), prefer tmux. --- .../behave_shell/_features/__init__.py | 2 + .../behave_shell/_features/environmental.py | 66 +++++++++++++++++ ...test_environmental_terminal_multiplexer.py | 74 +++++++++++++++++++ 3 files changed, 142 insertions(+) create mode 100644 tests/profiler/behave_shell/test_environmental_terminal_multiplexer.py diff --git a/decnet/profiler/behave_shell/_features/__init__.py b/decnet/profiler/behave_shell/_features/__init__.py index c2ba2c49..7cba8bc2 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.environmental import ( shell_type, + terminal_multiplexer, ) from decnet.profiler.behave_shell._features.temporal import ( escalation_pattern, @@ -71,4 +72,5 @@ FEATURES: tuple[FeatureFn, ...] = ( escalation_pattern, landing_ritual, shell_type, + terminal_multiplexer, ) diff --git a/decnet/profiler/behave_shell/_features/environmental.py b/decnet/profiler/behave_shell/_features/environmental.py index 9583b071..a96d815b 100644 --- a/decnet/profiler/behave_shell/_features/environmental.py +++ b/decnet/profiler/behave_shell/_features/environmental.py @@ -6,12 +6,31 @@ prompt-line detector. F.0 itself emits no primitive — it populates which F.1 / F.3 / E.4 read. Step F.1: ``environmental.shell_type``. +Step F.2: ``environmental.terminal_multiplexer``. """ from __future__ import annotations import collections from typing import Iterator +# Multiplexer fingerprints scanned over RAW output (multiplexer escapes +# ARE ANSI sequences, so we must NOT strip-ANSI before searching). +# Sources: +# tmux DCS passthrough: ESC P tmux ; +# tmux focus reporting: ESC [ ? 1004 (set/reset) +# tmux window-title with explicit tmux marker +# screen DCS: ESC P = +# screen-specific OSC: ESC ] 83 ; +_TMUX_MARKERS: tuple[str, ...] = ( + "\x1bPtmux;", + "\x1b[?1004", + "\x1b]2;tmux", +) +_SCREEN_MARKERS: tuple[str, ...] = ( + "\x1bP=", + "\x1b]83;", +) + from decnet_behave_core.spec.envelope import Observation from decnet.profiler.behave_shell._ctx import SessionContext @@ -73,3 +92,50 @@ def shell_type(ctx: SessionContext) -> Iterator[Observation]: value=value, confidence=confidence, ) + + +def terminal_multiplexer(ctx: SessionContext) -> Iterator[Observation]: + """Emit ``environmental.terminal_multiplexer`` ∈ {none, tmux, screen}. + + Scans raw output (NOT ANSI-stripped — multiplexer escapes ARE ANSI + sequences) for tmux/screen-specific fingerprints. If both detected, + prefer tmux (more common in 2026 nested-mux setups). Even one + escape is conclusive — no sample-size floor. + + Confidence 0.85 when a fingerprint matches; 0.55 for ``none`` (a + bare PTY genuinely has no multiplexer, but a hidden multiplexer + that suppresses its escapes would also yield ``none``). + + Skip emission when the session has no commands — without operator + interaction the engine should not emit operator-derived primitives. + The smoke gates (``test_extract_session_empty_stream_yields_no_observations``, + ``test_extract_session_zero_inputs_yields_nothing``) bind this: + no commands, no observations. + """ + if not ctx.commands: + return + has_tmux = False + has_screen = False + for _t, _k, data in ctx.output_events: + if not has_tmux and any(m in data for m in _TMUX_MARKERS): + has_tmux = True + if not has_screen and any(m in data for m in _SCREEN_MARKERS): + has_screen = True + if has_tmux and has_screen: + break + + if has_tmux: + value = "tmux" + confidence = 0.85 + elif has_screen: + value = "screen" + confidence = 0.85 + else: + value = "none" + confidence = 0.55 + yield make_observation( + ctx, + primitive="environmental.terminal_multiplexer", + value=value, + confidence=confidence, + ) diff --git a/tests/profiler/behave_shell/test_environmental_terminal_multiplexer.py b/tests/profiler/behave_shell/test_environmental_terminal_multiplexer.py new file mode 100644 index 00000000..20fc983b --- /dev/null +++ b/tests/profiler/behave_shell/test_environmental_terminal_multiplexer.py @@ -0,0 +1,74 @@ +"""Step F.2: ``environmental.terminal_multiplexer``.""" +from __future__ import annotations + +from decnet.profiler.behave_shell import extract_session +from decnet.profiler.behave_shell._parse import AsciinemaEvent + + +PRIMITIVE = "environmental.terminal_multiplexer" + + +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 = 0.0, dt: float = 0.05) -> list[AsciinemaEvent]: + return [(t0 + i * dt, "i", c) for i, c in enumerate(text)] + + +def test_clean_pty_emits_none() -> None: + events: list[AsciinemaEvent] = [ + *_typed("ls\r"), + (0.20, "o", "file1\nfile2\n"), + ] + out = list(extract_session(events, sid="mux-clean")) + obs = _of(out, PRIMITIVE) + assert obs.value == "none" + + +def test_tmux_dcs_passthrough_detected() -> None: + events: list[AsciinemaEvent] = [ + *_typed("ls\r"), + (0.20, "o", "\x1bPtmux;passthrough_payload\x1b\\"), + ] + obs = _of(list(extract_session(events, sid="mux-tmux-dcs")), PRIMITIVE) + assert obs.value == "tmux" + + +def test_tmux_focus_reporting_detected() -> None: + events: list[AsciinemaEvent] = [ + *_typed("ls\r"), + (0.20, "o", "\x1b[?1004h"), # set focus reporting + ] + obs = _of(list(extract_session(events, sid="mux-tmux-focus")), PRIMITIVE) + assert obs.value == "tmux" + + +def test_screen_dcs_detected() -> None: + events: list[AsciinemaEvent] = [ + *_typed("ls\r"), + (0.20, "o", "\x1bP=value\x1b\\"), + ] + obs = _of(list(extract_session(events, sid="mux-screen")), PRIMITIVE) + assert obs.value == "screen" + + +def test_both_present_prefers_tmux() -> None: + """Nested mux setup — prefer tmux (more common).""" + events: list[AsciinemaEvent] = [ + *_typed("ls\r"), + (0.20, "o", "\x1bPtmux;\x1b\\\x1bP=screen\x1b\\"), + ] + obs = _of(list(extract_session(events, sid="mux-both")), PRIMITIVE) + assert obs.value == "tmux" + + +def test_none_has_lower_confidence_than_detected() -> None: + """``none`` could be a hidden multiplexer; confidence reflects that.""" + none_events: list[AsciinemaEvent] = _typed("ls\r") + [(0.20, "o", "file1\n")] + tmux_events: list[AsciinemaEvent] = _typed("ls\r") + [(0.20, "o", "\x1bPtmux;\x1b\\")] + n = _of(list(extract_session(none_events, sid="mux-n")), PRIMITIVE) + t = _of(list(extract_session(tmux_events, sid="mux-t")), PRIMITIVE) + assert n.confidence < t.confidence