feat(profiler/behave_shell): emit environmental.terminal_multiplexer
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.
This commit is contained in:
@@ -26,6 +26,7 @@ from decnet.profiler.behave_shell._features.cognitive import (
|
|||||||
)
|
)
|
||||||
from decnet.profiler.behave_shell._features.environmental import (
|
from decnet.profiler.behave_shell._features.environmental import (
|
||||||
shell_type,
|
shell_type,
|
||||||
|
terminal_multiplexer,
|
||||||
)
|
)
|
||||||
from decnet.profiler.behave_shell._features.temporal import (
|
from decnet.profiler.behave_shell._features.temporal import (
|
||||||
escalation_pattern,
|
escalation_pattern,
|
||||||
@@ -71,4 +72,5 @@ FEATURES: tuple[FeatureFn, ...] = (
|
|||||||
escalation_pattern,
|
escalation_pattern,
|
||||||
landing_ritual,
|
landing_ritual,
|
||||||
shell_type,
|
shell_type,
|
||||||
|
terminal_multiplexer,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -6,12 +6,31 @@ prompt-line detector. F.0 itself emits no primitive — it populates
|
|||||||
which F.1 / F.3 / E.4 read.
|
which F.1 / F.3 / E.4 read.
|
||||||
|
|
||||||
Step F.1: ``environmental.shell_type``.
|
Step F.1: ``environmental.shell_type``.
|
||||||
|
Step F.2: ``environmental.terminal_multiplexer``.
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import collections
|
import collections
|
||||||
from typing import Iterator
|
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_behave_core.spec.envelope import Observation
|
||||||
|
|
||||||
from decnet.profiler.behave_shell._ctx import SessionContext
|
from decnet.profiler.behave_shell._ctx import SessionContext
|
||||||
@@ -73,3 +92,50 @@ def shell_type(ctx: SessionContext) -> Iterator[Observation]:
|
|||||||
value=value,
|
value=value,
|
||||||
confidence=confidence,
|
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,
|
||||||
|
)
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in New Issue
Block a user