fix(gateway/desktop): durable row-id addressing for rewind truncation
Address rewinds/edits via SQLite messages.id (truncate_before_row_id) instead of shifting user ordinals. Resolve against in-memory stamps, then durable session history when live turns drop _row_id; refuse unknown durable targets with 4018 (no ordinal fallback) and 4030 on ordinal/row_id mismatch. Stamp _row_id on insert, load row ids on resume paths, send rowId from Desktop, filter renderer-synthetic ids, and stop silently resending failed targeted edits without truncation. Add production-shaped SessionDB tests for resolve and fail-closed paths. Fixes #82959
This commit is contained in:
StanleyStetson
kshitij
parent
9460cc11d4
commit
23da6d6fe2
|
|
@ -366,13 +366,28 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses
|
|||
// Rewind primitive (interrupt-first for live turns, busy-retry) — shared with
|
||||
// the primary chat so the two can't diverge.
|
||||
const submitRewind = useCallback(
|
||||
(text: string, truncateOrdinal: number | undefined, interruptFirst: boolean) =>
|
||||
runRewindSubmit(requestGateway, runtimeIdRef.current, text, truncateOrdinal, interruptFirst, {
|
||||
storedSessionId: storedIdRef.current,
|
||||
onSessionRecovered: recoveredId => {
|
||||
runtimeIdRef.current = recoveredId
|
||||
}
|
||||
}),
|
||||
(
|
||||
text: string,
|
||||
truncateOrdinal: number | undefined,
|
||||
interruptFirst: boolean,
|
||||
truncateMessageId?: string,
|
||||
truncateRowId?: number
|
||||
) =>
|
||||
runRewindSubmit(
|
||||
requestGateway,
|
||||
runtimeIdRef.current,
|
||||
text,
|
||||
truncateOrdinal,
|
||||
truncateMessageId,
|
||||
interruptFirst,
|
||||
{
|
||||
storedSessionId: storedIdRef.current,
|
||||
onSessionRecovered: recoveredId => {
|
||||
runtimeIdRef.current = recoveredId
|
||||
}
|
||||
},
|
||||
truncateRowId
|
||||
),
|
||||
[requestGateway]
|
||||
)
|
||||
|
||||
|
|
@ -398,7 +413,7 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses
|
|||
{
|
||||
session_id: runtimeIdRef.current,
|
||||
text: plan.text,
|
||||
...truncateSubmitParams(plan.truncateOrdinal)
|
||||
...truncateSubmitParams(plan.truncateOrdinal, plan.truncateMessageId, plan.truncateRowId)
|
||||
},
|
||||
PROMPT_SUBMIT_REQUEST_TIMEOUT_MS
|
||||
)
|
||||
|
|
@ -425,7 +440,7 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses
|
|||
update(state => applyRewindOptimistic(state, plan.sourceIndex))
|
||||
|
||||
try {
|
||||
await submitRewind(plan.text, plan.truncateOrdinal, wasBusy)
|
||||
await submitRewind(plan.text, plan.truncateOrdinal, wasBusy, plan.truncateMessageId, plan.truncateRowId)
|
||||
} catch (err) {
|
||||
update(state => ({ ...state, busy: false, awaitingResponse: false, messages }))
|
||||
throw err
|
||||
|
|
@ -454,7 +469,7 @@ export function useSessionTileActions({ runtimeId, scope, storedSessionId }: Ses
|
|||
update(state => applyRewindOptimistic(state, plan.sourceIndex, plan.editedMessage))
|
||||
|
||||
try {
|
||||
await submitRewind(plan.text, plan.truncateOrdinal, wasBusy)
|
||||
await submitRewind(plan.text, plan.truncateOrdinal, wasBusy, plan.truncateMessageId, plan.truncateRowId)
|
||||
} catch (err) {
|
||||
update(state => ({ ...state, busy: false, awaitingResponse: false, messages }))
|
||||
notifyError(err, copy.editFailed)
|
||||
|
|
|
|||
|
|
@ -2294,6 +2294,7 @@ describe('usePromptActions restoreToMessage', () => {
|
|||
text: 'first prompt',
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 0,
|
||||
truncate_before_message_id: 'u1',
|
||||
confirm_empty_truncate: true
|
||||
},
|
||||
1_800_000
|
||||
|
|
@ -2363,6 +2364,7 @@ describe('usePromptActions restoreToMessage', () => {
|
|||
text: 'first prompt',
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 0,
|
||||
truncate_before_message_id: 'u1',
|
||||
confirm_empty_truncate: true
|
||||
},
|
||||
1_800_000
|
||||
|
|
@ -2410,6 +2412,7 @@ describe('usePromptActions restoreToMessage', () => {
|
|||
text: 'first prompt',
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 0,
|
||||
truncate_before_message_id: 'u1',
|
||||
confirm_empty_truncate: true
|
||||
},
|
||||
1_800_000
|
||||
|
|
|
|||
|
|
@ -809,7 +809,7 @@ export function usePromptActions({
|
|||
{
|
||||
session_id: sessionId,
|
||||
text: plan.text,
|
||||
...truncateSubmitParams(plan.truncateOrdinal)
|
||||
...truncateSubmitParams(plan.truncateOrdinal, plan.truncateMessageId, plan.truncateRowId)
|
||||
},
|
||||
PROMPT_SUBMIT_REQUEST_TIMEOUT_MS
|
||||
)
|
||||
|
|
@ -835,14 +835,30 @@ export function usePromptActions({
|
|||
// fresh turn. Live/stuck turns interrupt first, and a raced "session busy"
|
||||
// response interrupts + retries through the shared busy gate.
|
||||
const submitRewindPrompt = useCallback(
|
||||
(sessionId: string, text: string, truncateOrdinal: number | undefined, interruptFirst: boolean) =>
|
||||
runRewindSubmit(requestGateway, sessionId, text, truncateOrdinal, interruptFirst, {
|
||||
storedSessionId: selectedStoredSessionIdRef.current,
|
||||
onSessionRecovered: recoveredId => {
|
||||
activeSessionIdRef.current = recoveredId
|
||||
setActiveSessionId(recoveredId)
|
||||
}
|
||||
}),
|
||||
(
|
||||
sessionId: string,
|
||||
text: string,
|
||||
truncateOrdinal: number | undefined,
|
||||
truncateMessageId: string | undefined,
|
||||
interruptFirst: boolean,
|
||||
truncateRowId?: number
|
||||
) =>
|
||||
runRewindSubmit(
|
||||
requestGateway,
|
||||
sessionId,
|
||||
text,
|
||||
truncateOrdinal,
|
||||
truncateMessageId,
|
||||
interruptFirst,
|
||||
{
|
||||
storedSessionId: selectedStoredSessionIdRef.current,
|
||||
onSessionRecovered: recoveredId => {
|
||||
activeSessionIdRef.current = recoveredId
|
||||
setActiveSessionId(recoveredId)
|
||||
}
|
||||
},
|
||||
truncateRowId
|
||||
),
|
||||
[activeSessionIdRef, requestGateway, selectedStoredSessionIdRef]
|
||||
)
|
||||
|
||||
|
|
@ -873,7 +889,14 @@ export function usePromptActions({
|
|||
updateSessionState(sessionId, state => applyRewindOptimistic(state, plan.sourceIndex))
|
||||
|
||||
try {
|
||||
await submitRewindPrompt(sessionId, plan.text, plan.truncateOrdinal, busyRef.current || $busy.get())
|
||||
await submitRewindPrompt(
|
||||
sessionId,
|
||||
plan.text,
|
||||
plan.truncateOrdinal,
|
||||
plan.truncateMessageId,
|
||||
busyRef.current || $busy.get(),
|
||||
plan.truncateRowId
|
||||
)
|
||||
} catch (err) {
|
||||
// The rewind never landed (e.g. the gateway stayed busy past the retry
|
||||
// deadline). Roll the optimistic truncation back to the full original
|
||||
|
|
@ -920,25 +943,16 @@ export function usePromptActions({
|
|||
setAwaitingResponse(true)
|
||||
updateSessionState(sessionId, state => applyRewindOptimistic(state, plan.sourceIndex, plan.editedMessage))
|
||||
|
||||
const isStaleTargetError = (err: unknown) =>
|
||||
/no longer in session history|not in session history/i.test(err instanceof Error ? err.message : String(err))
|
||||
|
||||
try {
|
||||
await submitRewindPrompt(sessionId, plan.text, plan.truncateOrdinal, busyRef.current || $busy.get())
|
||||
await submitRewindPrompt(
|
||||
sessionId,
|
||||
plan.text,
|
||||
plan.truncateOrdinal,
|
||||
plan.truncateMessageId,
|
||||
busyRef.current || $busy.get(),
|
||||
plan.truncateRowId
|
||||
)
|
||||
} catch (err) {
|
||||
let surfaced = err
|
||||
|
||||
if (!plan.isFailedTurn && isStaleTargetError(err)) {
|
||||
try {
|
||||
// Already interrupted on the first attempt — submit as a plain resend.
|
||||
await submitRewindPrompt(sessionId, plan.text, undefined, false)
|
||||
|
||||
return
|
||||
} catch (retryErr) {
|
||||
surfaced = retryErr
|
||||
}
|
||||
}
|
||||
|
||||
// Roll the optimistic edit/truncation back to the original history so the
|
||||
// UI stays in sync with what's persisted instead of stranding a partial
|
||||
// timeline.
|
||||
|
|
@ -946,7 +960,7 @@ export function usePromptActions({
|
|||
setBusy(false)
|
||||
setAwaitingResponse(false)
|
||||
updateSessionState(sessionId, state => ({ ...state, busy: false, awaitingResponse: false, messages }))
|
||||
notifyError(surfaced, copy.editFailed)
|
||||
notifyError(err, copy.editFailed)
|
||||
}
|
||||
},
|
||||
[activeSessionIdRef, busyRef, copy.editFailed, submitRewindPrompt, updateSessionState]
|
||||
|
|
|
|||
|
|
@ -30,4 +30,31 @@ describe('truncateSubmitParams', () => {
|
|||
expect(params.confirm_truncate).toBe(true)
|
||||
}
|
||||
})
|
||||
|
||||
it('includes truncate_before_row_id when passed', () => {
|
||||
expect(truncateSubmitParams(1, 'msg-123', 456)).toEqual({
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 1,
|
||||
truncate_before_message_id: 'msg-123',
|
||||
truncate_before_row_id: 456
|
||||
})
|
||||
expect(truncateSubmitParams(undefined, undefined, 456)).toEqual({
|
||||
confirm_truncate: true,
|
||||
truncate_before_row_id: 456
|
||||
})
|
||||
})
|
||||
|
||||
it('drops renderer-synthetic message ids but keeps durable row ids', () => {
|
||||
// chat-messages.ts: `${timestamp}-${index}-${role}`
|
||||
expect(truncateSubmitParams(1, '1723456789-0-user', 456)).toEqual({
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 1,
|
||||
truncate_before_row_id: 456
|
||||
})
|
||||
expect(truncateSubmitParams(0, 'user-1723456789-0', undefined)).toEqual({
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: 0,
|
||||
confirm_empty_truncate: true
|
||||
})
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -34,38 +34,52 @@ type RequestGateway = <T = unknown>(method: string, params?: Record<string, unkn
|
|||
* transcript (restore/regenerate the first user turn), which the gateway gates
|
||||
* behind `confirm_empty_truncate` on top of that.
|
||||
*/
|
||||
export function truncateSubmitParams(truncateOrdinal: number | undefined): Record<string, unknown> {
|
||||
if (truncateOrdinal === undefined) {
|
||||
export function truncateSubmitParams(
|
||||
truncateOrdinal: number | undefined,
|
||||
truncateMessageId?: string,
|
||||
truncateRowId?: number
|
||||
): Record<string, unknown> {
|
||||
const hasOrdinal = typeof truncateOrdinal === 'number' && Number.isInteger(truncateOrdinal) && truncateOrdinal >= 0;
|
||||
const hasRowId = typeof truncateRowId === 'number' && Number.isInteger(truncateRowId);
|
||||
// Renderer ids are ephemeral (`${timestamp}-${index}-${role}` from
|
||||
// chat-messages.ts, plus older `user-…` / `assistant-…` shapes). Gateway
|
||||
// history never carries them — only durable `row_id` / platform message_id.
|
||||
const isSyntheticId =
|
||||
typeof truncateMessageId === 'string' &&
|
||||
(truncateMessageId.startsWith('user-') ||
|
||||
truncateMessageId.startsWith('assistant-') ||
|
||||
truncateMessageId.includes('-synthetic-') ||
|
||||
/^\d+-\d+-(user|assistant|tools)\b/.test(truncateMessageId))
|
||||
const hasMessageId = typeof truncateMessageId === 'string' && truncateMessageId.length > 0 && !isSyntheticId;
|
||||
|
||||
if (!hasOrdinal && !hasMessageId && !hasRowId) {
|
||||
return {}
|
||||
}
|
||||
|
||||
return {
|
||||
confirm_truncate: true,
|
||||
truncate_before_user_ordinal: truncateOrdinal,
|
||||
...(hasOrdinal ? { truncate_before_user_ordinal: truncateOrdinal } : {}),
|
||||
...(hasMessageId ? { truncate_before_message_id: truncateMessageId } : {}),
|
||||
...(hasRowId ? { truncate_before_row_id: truncateRowId } : {}),
|
||||
...(truncateOrdinal === 0 ? { confirm_empty_truncate: true } : {})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Rewind a turn: `prompt.submit` with an optional `truncate_before_user_ordinal`
|
||||
* (drops that user turn + everything after). Idle rewinds submit directly
|
||||
* (interrupting an idle agent can leave a stale interrupt flag that cancels the
|
||||
* fresh turn); live/stuck turns interrupt first, and a raced "session busy"
|
||||
* response interrupts + retries through the shared busy gate.
|
||||
*
|
||||
* A rewind runs right after `cancelRun`, and interrupting can drop the
|
||||
* gateway's in-memory session — so the follow-up submit hits a dead runtime id
|
||||
* and used to surface a bare "session not found" ("Restore checkpoint" failing
|
||||
* after a stop). Pass `recovery` so a stale id re-registers and retries like
|
||||
* every other session-scoped RPC.
|
||||
* / `truncate_before_message_id` / `truncate_before_row_id` (drops that user turn + everything after).
|
||||
* Idle rewinds submit directly; live/stuck turns interrupt first, and a raced
|
||||
* "session busy" response interrupts + retries through the shared busy gate.
|
||||
*/
|
||||
export async function runRewindSubmit(
|
||||
requestGateway: RequestGateway,
|
||||
sessionId: string,
|
||||
text: string,
|
||||
truncateOrdinal: number | undefined,
|
||||
truncateMessageId: string | undefined,
|
||||
interruptFirst: boolean,
|
||||
recovery?: { storedSessionId?: null | string; onSessionRecovered?: (sessionId: string) => void }
|
||||
recovery?: { storedSessionId?: null | string; onSessionRecovered?: (sessionId: string) => void },
|
||||
truncateRowId?: number
|
||||
): Promise<void> {
|
||||
// Recovery may rebind the live id mid-flight; interrupt/submit must both
|
||||
// follow it rather than pinning the dead one.
|
||||
|
|
@ -85,7 +99,7 @@ export async function runRewindSubmit(
|
|||
{
|
||||
session_id: targetId,
|
||||
text,
|
||||
...truncateSubmitParams(truncateOrdinal)
|
||||
...truncateSubmitParams(truncateOrdinal, truncateMessageId, truncateRowId)
|
||||
},
|
||||
PROMPT_SUBMIT_REQUEST_TIMEOUT_MS
|
||||
)
|
||||
|
|
@ -133,6 +147,8 @@ export interface ReloadPlan {
|
|||
branchGroupId: string
|
||||
text: string
|
||||
truncateOrdinal: number
|
||||
truncateMessageId?: string
|
||||
truncateRowId?: number
|
||||
userIndex: number
|
||||
}
|
||||
|
||||
|
|
@ -164,6 +180,8 @@ export function planReload(messages: ChatMessage[], parentId: null | string): nu
|
|||
branchGroupId: targetAssistant?.branchGroupId ?? branchGroupForUser(userMessage),
|
||||
text,
|
||||
truncateOrdinal: visibleUserOrdinal(messages, userIndex),
|
||||
truncateMessageId: userMessage.id,
|
||||
truncateRowId: userMessage.rowId,
|
||||
userIndex
|
||||
}
|
||||
}
|
||||
|
|
@ -202,6 +220,8 @@ export interface RestorePlan {
|
|||
sourceIndex: number
|
||||
text: string
|
||||
truncateOrdinal: number
|
||||
truncateMessageId?: string
|
||||
truncateRowId?: number
|
||||
}
|
||||
|
||||
/** Resolve the user turn to rewind to; throws with a user-facing reason. */
|
||||
|
|
@ -231,7 +251,7 @@ export function planRestore(messages: ChatMessage[], messageId: string, target?:
|
|||
? visibleUserOrdinal(messages, sourceIndex)
|
||||
: target.userOrdinal
|
||||
|
||||
return { sourceIndex, text, truncateOrdinal }
|
||||
return { sourceIndex, text, truncateOrdinal, truncateMessageId: source.id, truncateRowId: source.rowId }
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
@ -244,6 +264,8 @@ export interface EditPlan {
|
|||
sourceIndex: number
|
||||
text: string
|
||||
truncateOrdinal: number | undefined
|
||||
truncateMessageId?: string
|
||||
truncateRowId?: number
|
||||
}
|
||||
|
||||
/** Resolve the edited user turn, or null when nothing changed / invalid. */
|
||||
|
|
@ -272,7 +294,9 @@ export function planEdit(messages: ChatMessage[], edited: AppendMessage): EditPl
|
|||
isFailedTurn,
|
||||
sourceIndex,
|
||||
text,
|
||||
truncateOrdinal: isFailedTurn ? undefined : visibleUserOrdinal(messages, sourceIndex)
|
||||
truncateOrdinal: isFailedTurn ? undefined : visibleUserOrdinal(messages, sourceIndex),
|
||||
truncateMessageId: isFailedTurn ? undefined : source.id,
|
||||
truncateRowId: isFailedTurn ? undefined : source.rowId
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -8494,7 +8494,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
|
|||
|
||||
api_content = msg.get("api_content")
|
||||
|
||||
conn.execute(
|
||||
cur = conn.execute(
|
||||
"""INSERT INTO messages (session_id, role, content, tool_call_id,
|
||||
tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
|
||||
reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
|
||||
|
|
@ -8524,6 +8524,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
|
|||
self._encode_display_metadata(msg.get("display_metadata")),
|
||||
),
|
||||
)
|
||||
if isinstance(msg, dict) and cur.lastrowid is not None:
|
||||
msg["_row_id"] = cur.lastrowid
|
||||
inserted += 1
|
||||
if tool_calls is not None:
|
||||
tool_calls_total += (
|
||||
|
|
|
|||
|
|
@ -678,7 +678,7 @@ class TestTodoSnapshotScaffoldingTails:
|
|||
_msgs(), "sys", approx_tokens=120_000
|
||||
)
|
||||
|
||||
assert compressed == expected
|
||||
assert [{k: v for k, v in m.items() if k != "_row_id"} for m in compressed] == expected
|
||||
assert not any(
|
||||
TODO_INJECTION_HEADER in str(message.get("content") or "")
|
||||
for message in compressed
|
||||
|
|
|
|||
|
|
@ -4542,6 +4542,336 @@ def test_prompt_submit_refuses_unconfirmed_nonempty_truncation(monkeypatch):
|
|||
server._sessions.pop("unconfirmed-trunc-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_truncates_by_message_id(monkeypatch):
|
||||
"""#82756: truncate_before_message_id resolves target message and cuts history accurately."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
history = [
|
||||
{"id": "msg-1", "role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
{"id": "msg-2", "role": "user", "content": "second"},
|
||||
{"role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
server._sessions["msg-id-trunc-sid"] = _session(history=list(history))
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(
|
||||
server, "_start_agent_build", lambda *a, **k: None
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
server, "_start_inflight_turn", lambda *a, **k: None
|
||||
)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "msg-id-trunc-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_message_id": "msg-2",
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("result") is not None
|
||||
assert len(replaced) == 1
|
||||
assert replaced[0][1] == history[:2]
|
||||
finally:
|
||||
server._sessions.pop("msg-id-trunc-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_refuses_ordinal_and_message_id_mismatch(monkeypatch):
|
||||
"""#82756: A mismatch between truncate_before_user_ordinal and truncate_before_message_id must return 4030."""
|
||||
history = [
|
||||
{"id": "msg-1", "role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
{"id": "msg-2", "role": "user", "content": "second"},
|
||||
{"role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
server._sessions["mismatch-trunc-sid"] = _session(history=list(history))
|
||||
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:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "mismatch-trunc-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_message_id": "msg-2", # ordinal index 1
|
||||
"truncate_before_user_ordinal": 0, # mismatch (stale 0)
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4030
|
||||
assert "does not match" in resp["error"]["message"]
|
||||
finally:
|
||||
server._sessions.pop("mismatch-trunc-sid", None)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def test_prompt_submit_truncates_by_row_id(monkeypatch):
|
||||
"""#82959: prompt.submit with truncate_before_row_id must cut at the target row id."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
history = [
|
||||
{"_row_id": 101, "role": "user", "content": "first"},
|
||||
{"_row_id": 102, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 103, "role": "user", "content": "second"},
|
||||
{"_row_id": 104, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
sess = _session(history=list(history))
|
||||
server._sessions["row-id-trunc-sid"] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
started = []
|
||||
monkeypatch.setattr(
|
||||
server, "_start_agent_build", lambda *a, **k: started.append(k)
|
||||
)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "row-id-trunc-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 103,
|
||||
"truncate_before_user_ordinal": 1,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is None
|
||||
assert len(sess["history"]) == 2
|
||||
assert sess["history"][-1]["content"] == "reply 1"
|
||||
assert len(replaced) == 1
|
||||
assert replaced[0][1] == history[:2]
|
||||
finally:
|
||||
server._sessions.pop("row-id-trunc-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_truncates_by_string_row_id(monkeypatch):
|
||||
"""#82959: String row IDs in history match correctly against integer truncate_before_row_id."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
history = [
|
||||
{"_row_id": "101", "role": "user", "content": "first"},
|
||||
{"_row_id": "102", "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": "103", "role": "user", "content": "second"},
|
||||
{"_row_id": "104", "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
sess = _session(history=list(history))
|
||||
server._sessions["str-row-id-trunc-sid"] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "str-row-id-trunc-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 103,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is None
|
||||
assert len(sess["history"]) == 2
|
||||
finally:
|
||||
server._sessions.pop("str-row-id-trunc-sid", None)
|
||||
|
||||
|
||||
def test_reproduce_row_id_truncation(monkeypatch):
|
||||
"""#82959: Reproduction test for durable row_id truncation and 4030 ordinal mismatch validation."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
history = [
|
||||
{"_row_id": 101, "role": "user", "content": "first"},
|
||||
{"_row_id": 102, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 103, "role": "user", "content": "second"},
|
||||
{"_row_id": 104, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
server._sessions["repro-sid"] = _session(history=list(history))
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
|
||||
|
||||
try:
|
||||
# 1. Target turn 2 via row_id 103
|
||||
resp = server.handle_request({
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "repro-sid",
|
||||
"text": "edited second turn",
|
||||
"truncate_before_row_id": 103,
|
||||
"confirm_truncate": True,
|
||||
}
|
||||
})
|
||||
assert resp.get("error") is None
|
||||
|
||||
# 2. Refuse ordinal & row_id mismatch with code 4030
|
||||
server._sessions["mismatch-sid"] = _session(history=list(history))
|
||||
resp_mismatch = server.handle_request({
|
||||
"id": "2",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "mismatch-sid",
|
||||
"text": "stale rewind",
|
||||
"truncate_before_row_id": 103,
|
||||
"truncate_before_user_ordinal": 0,
|
||||
"confirm_truncate": True,
|
||||
}
|
||||
})
|
||||
assert resp_mismatch["error"]["code"] == 4030
|
||||
finally:
|
||||
server._sessions.pop("repro-sid", None)
|
||||
server._sessions.pop("mismatch-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_refuses_ordinal_and_row_id_mismatch(monkeypatch):
|
||||
"""#82959: A mismatch between truncate_before_user_ordinal and truncate_before_row_id must return 4030."""
|
||||
history = [
|
||||
{"_row_id": 201, "role": "user", "content": "first"},
|
||||
{"_row_id": 202, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 203, "role": "user", "content": "second"},
|
||||
{"_row_id": 204, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
server._sessions["row-mismatch-sid"] = _session(history=list(history))
|
||||
monkeypatch.setattr(
|
||||
server, "_start_agent_build", lambda *a, **k: pytest.fail("must not start a turn")
|
||||
)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "row-mismatch-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 203, # user turn ordinal 1
|
||||
"truncate_before_user_ordinal": 0, # mismatch (stale 0)
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4030
|
||||
assert "does not match" in resp["error"]["message"]
|
||||
finally:
|
||||
server._sessions.pop("row-mismatch-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_refuses_boolean_row_id(monkeypatch):
|
||||
"""Boolean truncate_before_row_id must return 4004."""
|
||||
history = [
|
||||
{"_row_id": 301, "role": "user", "content": "first"},
|
||||
{"_row_id": 302, "role": "assistant", "content": "reply 1"},
|
||||
]
|
||||
server._sessions["bool-row-sid"] = _session(history=list(history))
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "bool-row-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": True,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4004
|
||||
assert "must be an integer" in resp["error"]["message"]
|
||||
finally:
|
||||
server._sessions.pop("bool-row-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_row_id_not_found(monkeypatch):
|
||||
"""Unknown truncate_before_row_id must return 4018."""
|
||||
history = [
|
||||
{"_row_id": 401, "role": "user", "content": "first"},
|
||||
]
|
||||
server._sessions["missing-row-sid"] = _session(history=list(history))
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "missing-row-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 999,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4018
|
||||
assert "no longer in session history" in resp["error"]["message"]
|
||||
finally:
|
||||
server._sessions.pop("missing-row-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_row_id_ignores_platform_id_fallback(monkeypatch):
|
||||
"""truncate_before_row_id must not match string platform IDs."""
|
||||
history = [
|
||||
{"id": "999", "role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
]
|
||||
server._sessions["string-id-sid"] = _session(history=list(history))
|
||||
try:
|
||||
resp = server.handle_request({
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "string-id-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 999,
|
||||
"confirm_truncate": True,
|
||||
}
|
||||
})
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4018
|
||||
finally:
|
||||
server._sessions.pop("string-id-sid", None)
|
||||
|
||||
|
||||
|
||||
|
||||
def test_prompt_submit_refuses_empty_truncation_without_confirm(monkeypatch):
|
||||
"""A confirmed rewind still must not wipe a non-empty transcript by accident.
|
||||
|
||||
|
|
@ -17061,3 +17391,288 @@ def test_prompt_submit_truncation_archives_instead_of_deleting(monkeypatch):
|
|||
assert captured.get("active_only") is True
|
||||
finally:
|
||||
server._sessions.pop("archive-trunc-sid", None)
|
||||
|
||||
|
||||
def test_insert_message_rows_sets_row_id_on_fresh_dicts(tmp_path):
|
||||
"""#82959: _insert_message_rows must assign _row_id on freshly inserted message dicts."""
|
||||
from hermes_state import SessionDB
|
||||
db = SessionDB(db_path=tmp_path / "state.db")
|
||||
db.create_session("fresh-msg-row-id-sid", "cli")
|
||||
msg = {"role": "user", "content": "fresh turn without pre-existing _row_id"}
|
||||
with db._lock:
|
||||
db._insert_message_rows(db._conn, "fresh-msg-row-id-sid", [msg])
|
||||
assert "_row_id" in msg, "New message dict did not receive _row_id"
|
||||
assert isinstance(msg["_row_id"], int) and msg["_row_id"] > 0
|
||||
|
||||
|
||||
def test_prompt_submit_unmatched_row_id_refuses_even_with_ordinal(monkeypatch):
|
||||
"""#82959: Unknown row_id must refuse (4018), not fall back to a client ordinal."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
history = [
|
||||
{"_row_id": 101, "role": "user", "content": "first"},
|
||||
{"_row_id": 102, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 103, "role": "user", "content": "second"},
|
||||
{"_row_id": 104, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
sess = _session(history=list(history))
|
||||
server._sessions["fallback-row-id-sid"] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(
|
||||
server, "_start_agent_build", lambda *a, **k: pytest.fail("must not start a turn")
|
||||
)
|
||||
|
||||
try:
|
||||
# Stale row_id 999 not in history — even with a valid ordinal, refuse.
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "fallback-row-id-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 999,
|
||||
"truncate_before_user_ordinal": 1,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4018
|
||||
assert replaced == []
|
||||
assert len(sess["history"]) == 4
|
||||
finally:
|
||||
server._sessions.pop("fallback-row-id-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_unmatched_message_id_refuses_even_with_ordinal(monkeypatch):
|
||||
"""#82959: Unknown message_id must refuse; no silent ordinal degradation."""
|
||||
replaced = []
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
# Production-shaped history: no renderer "id" keys on user dicts.
|
||||
history = [
|
||||
{"_row_id": 201, "role": "user", "content": "first"},
|
||||
{"_row_id": 202, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 203, "role": "user", "content": "second"},
|
||||
{"_row_id": 204, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
sess = _session(history=list(history))
|
||||
server._sessions["synthetic-msg-id-sid"] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(
|
||||
server, "_start_agent_build", lambda *a, **k: pytest.fail("must not start a turn")
|
||||
)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "synthetic-msg-id-sid",
|
||||
"text": "fresh start",
|
||||
"truncate_before_message_id": "user-1723456789-0",
|
||||
"truncate_before_user_ordinal": 0,
|
||||
"confirm_truncate": True,
|
||||
"confirm_empty_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4018
|
||||
assert replaced == []
|
||||
assert len(sess["history"]) == 4
|
||||
finally:
|
||||
server._sessions.pop("synthetic-msg-id-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_row_id_resolves_via_db_when_memory_lacks_stamps(monkeypatch):
|
||||
"""#82959: After turn rewrite strips _row_id, resolve against durable DB history."""
|
||||
replaced = []
|
||||
# Live memory after turn completion: provider-format, no _row_id stamps.
|
||||
live_history = [
|
||||
{"role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
{"role": "user", "content": "second"},
|
||||
{"role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
durable_history = [
|
||||
{"_row_id": 501, "role": "user", "content": "first"},
|
||||
{"_row_id": 502, "role": "assistant", "content": "reply 1"},
|
||||
{"_row_id": 503, "role": "user", "content": "second"},
|
||||
{"_row_id": 504, "role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
|
||||
class _FakeDB:
|
||||
def replace_messages(self, key, messages, active_only=False, archive_dropped=False):
|
||||
replaced.append((key, list(messages)))
|
||||
|
||||
def get_messages_as_conversation(self, key, repair_alternation=False, include_row_ids=False):
|
||||
assert include_row_ids is True
|
||||
return list(durable_history)
|
||||
|
||||
sess = _session(history=list(live_history), session_key="db-row-key")
|
||||
server._sessions["db-row-resolve-sid"] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
|
||||
monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
|
||||
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": "db-row-resolve-sid",
|
||||
"text": "new turn",
|
||||
"truncate_before_row_id": 503,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is None, resp
|
||||
assert len(sess["history"]) == 2
|
||||
assert sess["history"][-1]["content"] == "reply 1"
|
||||
assert len(replaced) == 1
|
||||
# Healing: live list should now carry stamps for subsequent rewinds.
|
||||
assert sess["history"][0].get("_row_id") == 501
|
||||
finally:
|
||||
server._sessions.pop("db-row-resolve-sid", None)
|
||||
|
||||
|
||||
def test_prompt_submit_row_id_real_sessiondb_resolve_without_memory_stamps(
|
||||
monkeypatch, tmp_path
|
||||
):
|
||||
"""#82959 production path: real SessionDB insert → live history without
|
||||
_row_id (turn rewrite) → truncate_before_row_id cuts durable + memory.
|
||||
|
||||
No hand-seeded ids and no MagicMock state manager — the contract that
|
||||
#82766 review said unit fixtures must exercise.
|
||||
"""
|
||||
from hermes_state import SessionDB
|
||||
|
||||
db = SessionDB(db_path=tmp_path / "rowid-trunc.db")
|
||||
session_key = "real-db-row-trunc"
|
||||
db.create_session(session_key, "cli")
|
||||
msgs = [
|
||||
{"role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
{"role": "user", "content": "second"},
|
||||
{"role": "assistant", "content": "reply 2"},
|
||||
{"role": "user", "content": "third"},
|
||||
{"role": "assistant", "content": "reply 3"},
|
||||
]
|
||||
with db._lock:
|
||||
db._insert_message_rows(db._conn, session_key, msgs)
|
||||
db._conn.commit()
|
||||
row_ids = [m["_row_id"] for m in msgs]
|
||||
assert all(isinstance(r, int) and r > 0 for r in row_ids)
|
||||
|
||||
# After turn completion gateway rewrites history as provider-format dicts
|
||||
# without _row_id — production-shaped live memory.
|
||||
live_history = [{"role": m["role"], "content": m["content"]} for m in msgs]
|
||||
assert all("_row_id" not in m for m in live_history)
|
||||
|
||||
sess = _session(history=list(live_history), session_key=session_key)
|
||||
sid = "real-db-row-trunc-sid"
|
||||
server._sessions[sid] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: db)
|
||||
monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
|
||||
monkeypatch.setattr(server, "_start_inflight_turn", lambda *a, **k: None)
|
||||
|
||||
try:
|
||||
# Cut before second user turn (row_ids[2]) — leave first exchange only.
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": sid,
|
||||
"text": "rewound second",
|
||||
"truncate_before_row_id": row_ids[2],
|
||||
"truncate_before_user_ordinal": 1,
|
||||
"confirm_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is None, resp
|
||||
assert len(sess["history"]) == 2
|
||||
assert sess["history"][0]["content"] == "first"
|
||||
assert sess["history"][1]["content"] == "reply 1"
|
||||
# Durable active transcript matches the cut (archive_dropped keeps
|
||||
# inactive rows; get_messages_as_conversation returns active only).
|
||||
active = db.get_messages_as_conversation(session_key)
|
||||
assert len(active) == 2
|
||||
assert active[0]["content"] == "first"
|
||||
assert active[1]["content"] == "reply 1"
|
||||
# Heal stamps for subsequent rewinds when memory lined up with DB.
|
||||
assert sess["history"][0].get("_row_id") is not None
|
||||
finally:
|
||||
server._sessions.pop(sid, None)
|
||||
|
||||
|
||||
def test_prompt_submit_row_id_real_sessiondb_unknown_refuses_despite_ordinal(
|
||||
monkeypatch, tmp_path
|
||||
):
|
||||
"""#82959 fail-closed: unknown row_id + valid ordinal must not truncate
|
||||
real SessionDB (the mass-delete class when durable id cannot resolve).
|
||||
"""
|
||||
from hermes_state import SessionDB
|
||||
|
||||
db = SessionDB(db_path=tmp_path / "rowid-refuse.db")
|
||||
session_key = "real-db-row-refuse"
|
||||
db.create_session(session_key, "cli")
|
||||
msgs = [
|
||||
{"role": "user", "content": "first"},
|
||||
{"role": "assistant", "content": "reply 1"},
|
||||
{"role": "user", "content": "second"},
|
||||
{"role": "assistant", "content": "reply 2"},
|
||||
]
|
||||
with db._lock:
|
||||
db._insert_message_rows(db._conn, session_key, msgs)
|
||||
db._conn.commit()
|
||||
|
||||
live_history = [{"role": m["role"], "content": m["content"]} for m in msgs]
|
||||
sess = _session(history=list(live_history), session_key=session_key)
|
||||
sid = "real-db-row-refuse-sid"
|
||||
server._sessions[sid] = sess
|
||||
monkeypatch.setattr(server, "_get_db", lambda: db)
|
||||
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")
|
||||
)
|
||||
|
||||
n_before = len(db.get_messages_as_conversation(session_key))
|
||||
try:
|
||||
resp = server.handle_request(
|
||||
{
|
||||
"id": "1",
|
||||
"method": "prompt.submit",
|
||||
"params": {
|
||||
"session_id": sid,
|
||||
"text": "stale durable id",
|
||||
"truncate_before_row_id": 999_999_999,
|
||||
"truncate_before_user_ordinal": 0,
|
||||
"confirm_truncate": True,
|
||||
"confirm_empty_truncate": True,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert resp.get("error") is not None
|
||||
assert resp["error"]["code"] == 4018
|
||||
assert len(sess["history"]) == 4
|
||||
assert len(db.get_messages_as_conversation(session_key)) == n_before
|
||||
finally:
|
||||
server._sessions.pop(sid, None)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -13,13 +13,107 @@ method = _registry.method
|
|||
_profile_scoped = _registry.profile_scoped
|
||||
|
||||
|
||||
def _history_user_indices(history: list) -> list:
|
||||
"""Indices of model-visible user turns (excludes display_kind timeline markers)."""
|
||||
return [
|
||||
i
|
||||
for i, m in enumerate(history)
|
||||
if m.get("role") == "user" and not m.get("display_kind")
|
||||
]
|
||||
|
||||
|
||||
def _message_row_id(msg: dict):
|
||||
"""Parse durable SQLite row id from a history entry, or None."""
|
||||
raw = msg.get("_row_id")
|
||||
if raw is None:
|
||||
raw = msg.get("row_id")
|
||||
if raw is None:
|
||||
return None
|
||||
try:
|
||||
return int(raw)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _find_user_turn_by_row_id(history: list, target_row_id: int):
|
||||
"""Return ``(user_ordinal, history_index)`` for ``target_row_id``, or None."""
|
||||
for u_ord, h_idx in enumerate(_history_user_indices(history)):
|
||||
if _message_row_id(history[h_idx]) == target_row_id:
|
||||
return u_ord, h_idx
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int):
|
||||
"""Resolve ``truncate_before_row_id`` to ``(user_ordinal, history_index)``.
|
||||
|
||||
Prefer in-memory ``_row_id`` / ``row_id`` stamps. When a live turn rewrote
|
||||
``session["history"]`` without stamps (provider-format messages), load the
|
||||
session's durable transcript with ``include_row_ids=True`` and map the
|
||||
matched user-turn ordinal onto the live list. Does **not** fall back to a
|
||||
client-supplied ordinal — unknown row ids must refuse (#82959).
|
||||
"""
|
||||
hit = _find_user_turn_by_row_id(history, target_row_id)
|
||||
if hit is not None:
|
||||
return hit
|
||||
|
||||
session_key = str(session.get("session_key") or "")
|
||||
if not session_key:
|
||||
return None
|
||||
|
||||
try:
|
||||
db = _get_db()
|
||||
except Exception:
|
||||
db = None
|
||||
if db is None:
|
||||
return None
|
||||
|
||||
get_conv = getattr(db, "get_messages_as_conversation", None)
|
||||
if not callable(get_conv):
|
||||
return None
|
||||
|
||||
try:
|
||||
db_history = get_conv(
|
||||
session_key, repair_alternation=True, include_row_ids=True
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"prompt.submit: failed loading DB history for row_id %s session %s",
|
||||
target_row_id,
|
||||
session_key,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
if not isinstance(db_history, list):
|
||||
return None
|
||||
|
||||
# Heal missing in-memory stamps when the live list still lines up 1:1 with
|
||||
# the durable transcript (common after turn-completion rewrites).
|
||||
if len(db_history) == len(history):
|
||||
for mem, db_msg in zip(history, db_history):
|
||||
db_rid = _message_row_id(db_msg) if isinstance(db_msg, dict) else None
|
||||
if db_rid is not None and _message_row_id(mem) is None:
|
||||
mem["_row_id"] = db_rid
|
||||
hit = _find_user_turn_by_row_id(history, target_row_id)
|
||||
if hit is not None:
|
||||
return hit
|
||||
|
||||
db_hit = _find_user_turn_by_row_id(db_history, target_row_id)
|
||||
if db_hit is None:
|
||||
return None
|
||||
db_ord, _ = db_hit
|
||||
mem_user_indices = _history_user_indices(history)
|
||||
if db_ord < 0 or db_ord >= len(mem_user_indices):
|
||||
return None
|
||||
return db_ord, mem_user_indices[db_ord]
|
||||
|
||||
|
||||
def _pending_reaction_notes(session: dict) -> str:
|
||||
"""Note block describing reactions the user added since the last turn, or "".
|
||||
|
||||
Applied to the MODEL INPUT only (``run_message``, beside the
|
||||
speech-interrupted note) — never to the text that gets persisted. Prefixing
|
||||
the persisted prompt bakes scaffolding into the transcript, which every
|
||||
surface then renders as a garbled user message on reload. Each reaction is
|
||||
the persisted prompt bakes scaffolding into the transcript. Each reaction is
|
||||
announced once — the row is stamped ``seen`` on read.
|
||||
"""
|
||||
session_key = str(session.get("session_key") or "")
|
||||
|
|
@ -118,7 +212,12 @@ def _(rid, params: dict) -> dict:
|
|||
# in turn: a stale "hud" would tell the model the user is still floating
|
||||
# over another app when they are back in Hermes.
|
||||
session["client_surface"] = "hud" if params.get("surface") == "hud" else ""
|
||||
if truncate_user_ordinal is not None and isinstance(text, str):
|
||||
has_truncation = (
|
||||
truncate_user_ordinal is not None
|
||||
or params.get("truncate_before_row_id") is not None
|
||||
or params.get("truncate_before_message_id") is not None
|
||||
)
|
||||
if has_truncation and isinstance(text, str):
|
||||
# A rewind/regenerate replays a turn from what the transcript shows. A
|
||||
# skill turn shows its invocation, so re-expand it here — otherwise
|
||||
# re-running `/work fix it` sends the agent nine literal characters
|
||||
|
|
@ -162,27 +261,189 @@ def _(rid, params: dict) -> dict:
|
|||
# the upgrade resumes the child's transcript as a normal conversation.
|
||||
if session.get("lazy") and _child_run_active(str(session.get("session_key") or "")):
|
||||
return _err(rid, 4009, "subagent still running — wait for it to finish")
|
||||
# confirm_truncate with no target is malformed: the flag is consent for
|
||||
# a specific cut, and a client that sends it bare has leaked rewind
|
||||
# state onto an ordinary submit (#82756). Fail fast instead of quietly
|
||||
# ignoring the flag so the broken client state is surfaced.
|
||||
if is_truthy_value(params.get("confirm_truncate")) and truncate_user_ordinal is None:
|
||||
truncate_message_id = params.get("truncate_before_message_id")
|
||||
truncate_row_id = params.get("truncate_before_row_id")
|
||||
if (
|
||||
is_truthy_value(params.get("confirm_truncate"))
|
||||
and truncate_user_ordinal is None
|
||||
and truncate_message_id is None
|
||||
and truncate_row_id is None
|
||||
):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"confirm_truncate requires truncate_before_user_ordinal",
|
||||
"confirm_truncate requires truncate_before_user_ordinal, truncate_before_message_id, or truncate_before_row_id",
|
||||
)
|
||||
if truncate_user_ordinal is not None:
|
||||
# bool is an int subclass: a JSON `true` would coerce via int() to
|
||||
# ordinal 1 and aim a confirmed rewind at the second user turn.
|
||||
if isinstance(truncate_user_ordinal, bool):
|
||||
return _err(rid, 4004, "truncate_before_user_ordinal must be an integer")
|
||||
try:
|
||||
ordinal = int(truncate_user_ordinal)
|
||||
except (TypeError, ValueError):
|
||||
return _err(rid, 4004, "truncate_before_user_ordinal must be an integer")
|
||||
if (
|
||||
truncate_user_ordinal is not None
|
||||
or truncate_message_id is not None
|
||||
or truncate_row_id is not None
|
||||
):
|
||||
history = session.get("history", [])
|
||||
# An ordinal alone is not consent. A client that carries a leftover
|
||||
user_indices = _history_user_indices(history)
|
||||
|
||||
target_idx = None
|
||||
ordinal = None
|
||||
|
||||
if truncate_row_id is not None:
|
||||
if isinstance(truncate_row_id, bool):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_row_id must be an integer",
|
||||
)
|
||||
try:
|
||||
target_row_id = int(truncate_row_id)
|
||||
except (TypeError, ValueError):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_row_id must be an integer",
|
||||
)
|
||||
|
||||
# Durable address first — never degrade a missing row_id into a
|
||||
# client ordinal cut (#82959 / #82766 review). Unknown id refuses
|
||||
# without touching data; stale ordinal with a *resolved* row_id
|
||||
# is a separate 4030 mismatch below.
|
||||
found_match = _resolve_truncate_row_id(
|
||||
session, history, target_row_id
|
||||
)
|
||||
# user_indices may have been healed with _row_id stamps
|
||||
user_indices = _history_user_indices(history)
|
||||
|
||||
if found_match is None:
|
||||
logger.warning(
|
||||
"prompt.submit: target row_id %d not found for session %s "
|
||||
"(in-memory + durable); refusing truncation without fallback",
|
||||
target_row_id,
|
||||
sid,
|
||||
)
|
||||
return _err(
|
||||
rid,
|
||||
4018,
|
||||
"target user message is no longer in session history",
|
||||
)
|
||||
|
||||
msg_ordinal, target_idx = found_match
|
||||
|
||||
if truncate_user_ordinal is not None:
|
||||
if isinstance(truncate_user_ordinal, bool):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
try:
|
||||
ordinal = int(truncate_user_ordinal)
|
||||
except (TypeError, ValueError):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
if ordinal != msg_ordinal:
|
||||
logger.warning(
|
||||
"prompt.submit: REFUSED truncation due to ordinal mismatch for session %s "
|
||||
"(ordinal=%d, row_id_ordinal=%d, row_id=%d). "
|
||||
"Stale truncate_before_user_ordinal detected.",
|
||||
sid,
|
||||
ordinal,
|
||||
msg_ordinal,
|
||||
target_row_id,
|
||||
)
|
||||
return _err(
|
||||
rid,
|
||||
4030,
|
||||
f"truncate_before_user_ordinal ({ordinal}) does not match "
|
||||
f"truncate_before_row_id target turn ({msg_ordinal})",
|
||||
)
|
||||
else:
|
||||
ordinal = msg_ordinal
|
||||
elif truncate_message_id is not None:
|
||||
msg_id_str = str(truncate_message_id)
|
||||
found_match = None
|
||||
for u_ord, h_idx in enumerate(user_indices):
|
||||
msg = history[h_idx]
|
||||
if msg.get("id") == msg_id_str or msg.get("message_id") == msg_id_str:
|
||||
found_match = (u_ord, h_idx)
|
||||
break
|
||||
|
||||
if found_match is None:
|
||||
# Fail closed: a supplied message_id that does not resolve
|
||||
# must not fall back to a (possibly stale) ordinal. Desktop
|
||||
# clients should send truncate_before_row_id instead.
|
||||
logger.warning(
|
||||
"prompt.submit: target message_id %s not found in history "
|
||||
"for session %s; refusing truncation without fallback",
|
||||
msg_id_str,
|
||||
sid,
|
||||
)
|
||||
return _err(
|
||||
rid,
|
||||
4018,
|
||||
"target user message is no longer in session history",
|
||||
)
|
||||
|
||||
msg_ordinal, target_idx = found_match
|
||||
|
||||
if truncate_user_ordinal is not None:
|
||||
if isinstance(truncate_user_ordinal, bool):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
try:
|
||||
ordinal = int(truncate_user_ordinal)
|
||||
except (TypeError, ValueError):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
if ordinal != msg_ordinal:
|
||||
logger.warning(
|
||||
"prompt.submit: REFUSED truncation due to ordinal mismatch for session %s "
|
||||
"(ordinal=%d, message_id_ordinal=%d, message_id=%s). "
|
||||
"Stale truncate_before_user_ordinal detected.",
|
||||
sid,
|
||||
ordinal,
|
||||
msg_ordinal,
|
||||
msg_id_str,
|
||||
)
|
||||
return _err(
|
||||
rid,
|
||||
4030,
|
||||
f"truncate_before_user_ordinal ({ordinal}) does not match "
|
||||
f"truncate_before_message_id target turn ({msg_ordinal})",
|
||||
)
|
||||
else:
|
||||
ordinal = msg_ordinal
|
||||
else:
|
||||
if isinstance(truncate_user_ordinal, bool):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
try:
|
||||
ordinal = int(truncate_user_ordinal)
|
||||
except (TypeError, ValueError):
|
||||
return _err(
|
||||
rid,
|
||||
4004,
|
||||
"truncate_before_user_ordinal must be an integer",
|
||||
)
|
||||
|
||||
if ordinal < 0 or ordinal >= len(user_indices):
|
||||
return _err(
|
||||
rid,
|
||||
4018,
|
||||
"target user message is no longer in session history",
|
||||
)
|
||||
target_idx = user_indices[ordinal]
|
||||
|
||||
# An ordinal/id alone is not consent. A client that carries a leftover
|
||||
# ordinal into an ORDINARY submit sends a request that is
|
||||
# indistinguishable, field by field, from a real rewind — same
|
||||
# method, same shape, an in-range target — and the cut it asks for
|
||||
|
|
@ -194,8 +455,8 @@ def _(rid, params: dict) -> dict:
|
|||
logger.warning(
|
||||
"prompt.submit: REFUSED unconfirmed truncation of session %s "
|
||||
"(%d messages held; ordinal=%d). The client attached "
|
||||
"truncate_before_user_ordinal without confirm_truncate — "
|
||||
"likely a stale ordinal on an ordinary submit.",
|
||||
"truncation parameter without confirm_truncate — "
|
||||
"likely stale truncation parameters on an ordinary submit.",
|
||||
sid,
|
||||
len(history),
|
||||
ordinal,
|
||||
|
|
@ -203,7 +464,7 @@ def _(rid, params: dict) -> dict:
|
|||
return _err(
|
||||
rid,
|
||||
4029,
|
||||
"truncate_before_user_ordinal requires confirm_truncate=true; "
|
||||
"truncation parameters require confirm_truncate=true; "
|
||||
"an ordinary prompt.submit must not drop session history "
|
||||
"(update your Hermes client if a rewind was intended)",
|
||||
)
|
||||
|
|
@ -1019,12 +1280,24 @@ def register(server) -> None:
|
|||
"""Bind this module's handlers onto ``server``'s globals and registry."""
|
||||
_registry.install(server)
|
||||
# Module-level helpers aren't @method handlers, so install() doesn't see
|
||||
# them — but server.py's run path calls this one (run_message enrichment,
|
||||
# beside the speech-interrupted note). Rebind and publish it the same way.
|
||||
server._pending_reaction_notes = types.FunctionType(
|
||||
_pending_reaction_notes.__code__,
|
||||
vars(server),
|
||||
_pending_reaction_notes.__name__,
|
||||
_pending_reaction_notes.__defaults__,
|
||||
_pending_reaction_notes.__closure__,
|
||||
)
|
||||
# them. Rebind onto server globals so handler bodies (and server.py call
|
||||
# sites) resolve the same free names after the split.
|
||||
g = vars(server)
|
||||
for helper in (
|
||||
_history_user_indices,
|
||||
_message_row_id,
|
||||
_find_user_turn_by_row_id,
|
||||
_resolve_truncate_row_id,
|
||||
_pending_reaction_notes,
|
||||
):
|
||||
setattr(
|
||||
server,
|
||||
helper.__name__,
|
||||
types.FunctionType(
|
||||
helper.__code__,
|
||||
g,
|
||||
helper.__name__,
|
||||
helper.__defaults__,
|
||||
helper.__closure__,
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -456,7 +456,9 @@ def _(rid, params: dict) -> dict:
|
|||
# history becomes the resumed session record's working conversation),
|
||||
# so heal a durable ``user;user`` violation once here instead of
|
||||
# re-firing the pre-request repair on every subsequent turn.
|
||||
history = db.get_messages_as_conversation(target, repair_alternation=True)
|
||||
history = db.get_messages_as_conversation(
|
||||
target, repair_alternation=True, include_row_ids=True
|
||||
)
|
||||
except Exception as e:
|
||||
if lease is not None:
|
||||
lease.release()
|
||||
|
|
|
|||
|
|
@ -64,10 +64,11 @@ A rewind / edit / regenerate is a `prompt.submit` that drops part of the stored
|
|||
| Parameter | Meaning |
|
||||
|-----------|---------|
|
||||
| `truncate_before_user_ordinal` | Zero-based index of the user turn to cut at. Everything from that turn onward is dropped. Display-only timeline rows (`display_kind`) are not counted. Must be a real integer — a JSON boolean is refused with code `4004`. |
|
||||
| `confirm_truncate` | Required whenever an ordinal is sent. Declares that this submit really is a rewind, not an ordinary send that happens to carry a leftover ordinal. Sending it without an ordinal is refused with code `4004` (leaked rewind state). |
|
||||
| `truncate_before_row_id` | Integer SQLite row ID (`messages.id` / `row_id`) of the target user turn to cut at. Preferred durable address. When both ordinal and row ID are provided, gateway verifies they match (returning `4030` on mismatch). An unknown/stale row ID is refused with `4018` — it does **not** fall back to the ordinal. |
|
||||
| `confirm_truncate` | Required whenever an ordinal, message ID, or row ID is sent. Declares that this submit really is a rewind, not an ordinary send that happens to carry leftover parameters. Sending it without a target is refused with code `4004`. |
|
||||
| `confirm_empty_truncate` | Additionally required when the cut would leave the transcript empty (ordinal `0`). |
|
||||
|
||||
An ordinal without `confirm_truncate` is refused with code `4029` and nothing is written. Hosts that implement rewind must set the flag at the moment the user asks for it, and must never keep the ordinal in state across ordinary submits.
|
||||
A truncation parameter without `confirm_truncate` is refused with code `4004` or `4029` and nothing is written. Hosts that implement rewind must set the flag at the moment the user asks for it, and must never keep truncation parameters in state across ordinary submits. Prefer `truncate_before_row_id` (from resume `row_id` / `_row_id`) over ordinals; keep the ordinal as a back-compat / optimistic-row path only when no durable id is available yet.
|
||||
|
||||
### Events streamed back
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue