test(tui): cover empty truncate guard on prompt.submit

Refuse ordinal-0 wipes without confirm_empty_truncate; allow the
opt-in path used by first-turn restore/regenerate.
This commit is contained in:
HexLab98 2026-07-24 13:40:25 +07:00 committed by kshitij
parent 819e01c134
commit 75f61182fc
3 changed files with 164 additions and 3 deletions

View file

@ -1450,7 +1450,8 @@ describe('usePromptActions restoreToMessage', () => {
{
session_id: RUNTIME_SESSION_ID,
text: 'first prompt',
truncate_before_user_ordinal: 0
truncate_before_user_ordinal: 0,
confirm_empty_truncate: true
},
1_800_000
)
@ -1517,7 +1518,8 @@ describe('usePromptActions restoreToMessage', () => {
{
session_id: RUNTIME_SESSION_ID,
text: 'first prompt',
truncate_before_user_ordinal: 0
truncate_before_user_ordinal: 0,
confirm_empty_truncate: true
},
1_800_000
)
@ -1562,7 +1564,8 @@ describe('usePromptActions restoreToMessage', () => {
{
session_id: RUNTIME_SESSION_ID,
text: 'first prompt',
truncate_before_user_ordinal: 0
truncate_before_user_ordinal: 0,
confirm_empty_truncate: true
},
1_800_000
)

View file

@ -0,0 +1,19 @@
import { describe, expect, it } from 'vitest'
import { truncateSubmitParams } from './rewind'
describe('truncateSubmitParams', () => {
it('omits truncation fields when no ordinal is set', () => {
expect(truncateSubmitParams(undefined)).toEqual({})
})
it('requires confirm_empty_truncate only for ordinal 0', () => {
expect(truncateSubmitParams(0)).toEqual({
truncate_before_user_ordinal: 0,
confirm_empty_truncate: true
})
expect(truncateSubmitParams(1)).toEqual({
truncate_before_user_ordinal: 1
})
})
})

View file

@ -3239,6 +3239,145 @@ def test_prompt_submit_rejects_negative_truncate_ordinal(monkeypatch):
server._sessions.pop("trunc-sid", None)
def test_prompt_submit_refuses_empty_truncation_without_confirm(monkeypatch):
"""Stale truncate_before_user_ordinal=0 must not wipe a non-empty transcript.
Desktop desync can attach ordinal 0 to an ordinary fresh submit. That cuts
at the first user message (history[:0] == []) and replace_messages() would
DELETE every durable row. Refuse unless confirm_empty_truncate is set.
"""
replaced = []
class _FakeDB:
def replace_messages(self, key, messages):
replaced.append((key, list(messages)))
history = [
{"role": "user", "content": "first"},
{"role": "assistant", "content": "ok"},
{"role": "user", "content": "second"},
{"role": "assistant", "content": "done"},
]
server._sessions["empty-trunc-sid"] = _session(history=list(history))
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
monkeypatch.setattr(
server, "_start_agent_build", lambda *a, **k: pytest.fail("must not start a turn")
)
monkeypatch.setattr(
server, "_start_inflight_turn", lambda *a, **k: pytest.fail("must not start a turn")
)
try:
# Missing confirm → refuse.
resp = server.handle_request(
{
"id": "1",
"method": "prompt.submit",
"params": {
"session_id": "empty-trunc-sid",
"text": "fresh typed message",
"truncate_before_user_ordinal": 0,
},
}
)
assert resp["error"]["code"] == 4025
assert "confirm_empty_truncate" in resp["error"]["message"]
# Explicit falsey values must not satisfy the opt-in either.
for falsey in (False, 0, "", "false", "no"):
resp = server.handle_request(
{
"id": "1",
"method": "prompt.submit",
"params": {
"session_id": "empty-trunc-sid",
"text": "fresh typed message",
"truncate_before_user_ordinal": 0,
"confirm_empty_truncate": falsey,
},
}
)
assert resp["error"]["code"] == 4025, falsey
assert server._sessions["empty-trunc-sid"]["history"] == history
assert server._sessions["empty-trunc-sid"]["running"] is False
assert server._sessions["empty-trunc-sid"]["history_version"] == 0
assert replaced == []
finally:
server._sessions.pop("empty-trunc-sid", None)
def test_prompt_submit_empty_truncation_allowed_with_confirm(monkeypatch):
"""Intentional restore/regenerate of the first user turn may wipe history."""
seen = {}
replaced = []
class _Agent:
def run_conversation(
self, prompt, conversation_history=None, stream_callback=None
):
seen["prompt"] = prompt
seen["history"] = conversation_history
return {
"final_response": "regenerated",
"messages": [
*(conversation_history or []),
{"role": "user", "content": prompt},
{"role": "assistant", "content": "regenerated"},
],
}
class _ImmediateThread:
def __init__(self, target=None, daemon=None):
self._target = target
def start(self):
self._target()
class _FakeDB:
def replace_messages(self, key, messages):
replaced.append((key, list(messages)))
history = [
{"role": "user", "content": "first"},
{"role": "assistant", "content": "ok"},
{"role": "user", "content": "second"},
{"role": "assistant", "content": "done"},
]
server._sessions["confirm-empty-sid"] = _session(
agent=_Agent(), history=list(history)
)
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, "_get_db", lambda: _FakeDB())
resp = server.handle_request(
{
"id": "1",
"method": "prompt.submit",
"params": {
"session_id": "confirm-empty-sid",
"text": "first",
"truncate_before_user_ordinal": 0,
"confirm_empty_truncate": True,
},
}
)
assert resp.get("result"), f"got error: {resp.get('error')}"
assert seen["prompt"] == "first"
assert seen["history"] == []
assert replaced == [("session-key", [])]
assert server._sessions["confirm-empty-sid"]["history"] == [
{"role": "user", "content": "first"},
{"role": "assistant", "content": "regenerated"},
]
finally:
server._sessions.pop("confirm-empty-sid", None)
class _StopAfterOneNotificationPoll:
def __init__(self):
self._checks = 0