test(input): preservation regressions and prompt.submit boundary (#62557)

Add cases for mid-string markers, trailing punctuation, and insufficient
tail repeats; verify prompt.submit passes sanitized text to run_conversation.
fix/verification-admin-route-recovery
HexLab98 2026-07-12 06:54:55 +07:00 committed by Teknium
parent 1011cd24e2
commit 2388e0687b
3 changed files with 147 additions and 0 deletions

View File

@ -0,0 +1,51 @@
import { describe, expect, it } from 'vitest'
import {
collapseRepeatedInputArtifacts,
sanitizeComposerInput,
stripLeakedBracketedPasteWrappers
} from './composer-input-sanitize'
describe('stripLeakedBracketedPasteWrappers', () => {
it('leaves plain text unchanged', () => {
expect(stripLeakedBracketedPasteWrappers('hello world')).toBe('hello world')
})
it('strips canonical escape wrappers', () => {
expect(stripLeakedBracketedPasteWrappers('\x1b[200~hello\x1b[201~')).toBe('hello')
})
it('keeps embedded literal bracket forms', () => {
const text = 'literal[200~tag and literal[201~tag should stay'
expect(stripLeakedBracketedPasteWrappers(text)).toBe(text)
})
})
describe('collapseRepeatedInputArtifacts', () => {
it('removes the desktop corruption tail from #62557', () => {
const prefix = '需要时随时叫我。'
const tail = '[e~[[e' + '~[[e'.repeat(20)
expect(collapseRepeatedInputArtifacts(prefix + tail)).toBe(prefix)
})
it('preserves a mid-string marker followed by valid suffix', () => {
const text = 'notes ~[[e more text here'
expect(collapseRepeatedInputArtifacts(text)).toBe(text)
})
it('preserves trailing punctuation that is not the corruption signature', () => {
expect(collapseRepeatedInputArtifacts('wait....')).toBe('wait....')
})
it('does not strip when fewer than minRepeats markers appear at the tail', () => {
const text = 'hello~[[e~[[e'
expect(collapseRepeatedInputArtifacts(text)).toBe(text)
})
})
describe('sanitizeComposerInput', () => {
it('normalizes wrappers and repeated artifact tails together', () => {
const corrupted = 'hello[' + '~[[e'.repeat(8)
expect(sanitizeComposerInput(corrupted)).toBe('hello')
})
})

View File

@ -0,0 +1,51 @@
"""Tests for shared user prompt input sanitization."""
from hermes_cli.input_sanitize import (
collapse_repeated_input_artifacts,
sanitize_user_prompt_text,
strip_leaked_bracketed_paste_wrappers,
)
class TestStripLeakedBracketedPasteWrappers:
def test_plain_text_unchanged(self):
assert strip_leaked_bracketed_paste_wrappers("hello world") == "hello world"
def test_strips_canonical_escape_wrappers(self):
assert strip_leaked_bracketed_paste_wrappers("\x1b[200~hello\x1b[201~") == "hello"
def test_strips_visible_caret_escape_wrappers(self):
assert strip_leaked_bracketed_paste_wrappers("^[[200~hello^[[201~") == "hello"
def test_does_not_strip_non_wrapper_bracket_forms_in_normal_text(self):
text = "literal[200~tag and literal[201~tag should stay"
assert strip_leaked_bracketed_paste_wrappers(text) == text
class TestCollapseRepeatedInputArtifacts:
def test_issue_62557_corruption_tail(self):
prefix = "需要时随时叫我。"
tail = "[e~[[e" + "~[[e" * 20
assert collapse_repeated_input_artifacts(prefix + tail) == prefix
def test_plain_text_unchanged(self):
text = "build00~tag should stay"
assert collapse_repeated_input_artifacts(text) == text
def test_mid_string_marker_followed_by_suffix_preserved(self):
text = "notes ~[[e more text here"
assert collapse_repeated_input_artifacts(text) == text
def test_trailing_punctuation_preserved(self):
assert collapse_repeated_input_artifacts("wait....") == "wait...."
def test_insufficient_tail_repeats_preserved(self):
text = "hello~[[e~[[e"
assert collapse_repeated_input_artifacts(text) == text
class TestSanitizeUserPromptText:
def test_combines_wrapper_strip_and_tail_collapse(self):
prefix = "hello["
corrupted = prefix + "~[[e" * 8
assert sanitize_user_prompt_text(corrupted) == "hello"

View File

@ -5093,6 +5093,51 @@ def test_prompt_submit_history_version_mismatch_surfaces_warning(monkeypatch):
server._sessions.pop("sid", None)
def test_prompt_submit_sanitizes_bracketed_paste_before_agent(monkeypatch):
"""prompt.submit must sanitize corrupted user text before run_conversation."""
captured: dict[str, str] = {}
class _Agent:
def run_conversation(
self, prompt, conversation_history=None, stream_callback=None
):
captured["prompt"] = prompt
return {
"final_response": "ok",
"messages": [{"role": "assistant", "content": "ok"}],
}
class _ImmediateThread:
def __init__(self, target=None, daemon=None, **kw):
self._target = target
def start(self):
self._target()
corrupted = "hello[" + "~[[e" * 8
server._sessions["sid"] = _session(agent=_Agent())
try:
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
monkeypatch.setattr(server, "_get_usage", lambda _a: {})
monkeypatch.setattr(server, "render_message", lambda _t, _c: "")
monkeypatch.setattr(server, "_emit", lambda *a: None)
monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
monkeypatch.setattr(server, "_ensure_session_db_row", lambda *a, **k: None)
monkeypatch.setattr(server, "_persist_branch_seed", lambda *a, **k: None)
resp = server.handle_request(
{
"id": "1",
"method": "prompt.submit",
"params": {"session_id": "sid", "text": corrupted},
}
)
assert resp.get("result"), f"got error: {resp.get('error')}"
assert captured["prompt"] == "hello"
finally:
server._sessions.pop("sid", None)
def test_prompt_submit_history_version_match_persists_normally(monkeypatch):
"""Regression guard: the backstop does not affect the happy path."""