hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 06:40:45 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

Dedupe WebCRT command audit on double-fired Enter.

Browse source 
xterm and keydown can both emit carriage return for one keypress; skip duplicate Enter frames on the client and collapse identical audits within 450ms on the server.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-02 14:25:27 +08:00
parent 7ca57278fe
commit 52424dce37
3 changed files with 58 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

17
netx_api/webcrt_session_model.py
View file

@ -96,10 +96,25 @@ class WebcrtSession:
_password_mode: bool = field(default=False, repr=False)
_stdout_tail: str = field(default="", repr=False)
_last_prompt_line: str = field(default="", repr=False)
_last_audited_command: str = field(default="", repr=False)
_last_audited_at: float = field(default=0.0, repr=False)
def touch(self) -> None:
self.last_activity = time.time()
def _should_emit_command_audit(self, cmd: str, *, dedup_sec: float = 0.45) -> bool:
"""Drop duplicate audits from double-fired Enter (same command within dedup_sec)."""
norm = str(cmd or "").strip()
if not norm:
return False
now = time.time()
with self._cmd_buf_lock:
if norm == self._last_audited_command and (now - self._last_audited_at) < dedup_sec:
return False
self._last_audited_command = norm
self._last_audited_at = now
return True
def push_connect_echo(self, text: str) -> None:
"""Queue login transcript for the WS wait-loop (thread-safe)."""
chunk = str(text or "")
@ -398,6 +413,8 @@ class WebcrtSession:
for cmd in lines:
if not str(cmd).strip():
continue
if not self._should_emit_command_audit(cmd):
continue
try:
_audit(
"command",