fix(chat): skip live-registry context RPCs until resume confirms runtime session id (#768)
Opening a session from history fired session.context_breakdown/session.usage with the storage session id before session.resume had registered it on the gateway. Those RPCs resolve against the LIVE runtime registry (_sess_nowait) and returned 4001 "session not found", which the reducer surfaced as a spurious snackbar even though the session loaded fine. - fetchContextUsage() now skips the two live-registry RPCs while runtimeSessionId is null (pre-resume window); the REST-based context fallbacks stay live. - runtimeSessionId is cleared on socket drop so a stale runtime id can't be reused after a gateway restart/reconnect (same 4001 family). - regression test: no context/usage RPC fires with the storage id during the pre-resume window; after the resume result the RPCs fire with the runtime id. Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
M57andM57 authored and GitHub committed
1 parent
5e1ee0b4e5
commit
c50a178845
2 files changed
+142
-32
No files matched your search
@@ -302,6 +302,13 @@ class ChatViewModel(
|
||||
status == ConnectionStatus.AUTH_EXPIRED
|
||||
) {
|
||||
_uiState.update { it.copy(isLoading = false) }
|
||||
// The runtime session id is only valid while the socket
|
||||
// owns it — a dropped connection may mean the gateway
|
||||
// closed/pruned the session (or restarted, wiping the
|
||||
// runtime registry). Clear it so session-scoped RPCs can't
|
||||
// fire with a stale id and 4001 "session not found";
|
||||
// handleGatewayReady rebinds it on the re-resume.
|
||||
runtimeSessionId = null
|
||||
// Fail any in-flight awaited RPCs so callers don't hang
|
||||
// across the disconnect (delegated to HermesWsClient, issue #526).
|
||||
wsClient.rejectAllPending()
|
||||
@@ -1853,42 +1860,52 @@ class ChatViewModel(
|
||||
// real current prompt size (drops after compression); `context_max`
|
||||
// is the compressor's actual window. Any failure keeps the last
|
||||
// known values — never blank the meter over a transient RPC error.
|
||||
val rpcSessionId = runtimeSessionId ?: sessionId
|
||||
try {
|
||||
val result =
|
||||
sendRpcAndAwait(
|
||||
WsMethods.SESSION_CONTEXT_BREAKDOWN,
|
||||
mapOf("session_id" to rpcSessionId),
|
||||
)
|
||||
val ctx = parseContextBreakdown(result)
|
||||
if (ctx != null) {
|
||||
_uiState.update { current ->
|
||||
current.copy(
|
||||
usedContextTokens =
|
||||
ctx.contextUsed?.takeIf { it > 0L } ?: current.usedContextTokens,
|
||||
fullContextTokens =
|
||||
ctx.contextMax?.takeIf { it > 0L } ?: current.fullContextTokens,
|
||||
//
|
||||
// These RPCs resolve the session against the gateway's LIVE runtime
|
||||
// registry (_sess_nowait) — the storage session id 4001s "session
|
||||
// not found" until session.resume has registered it. ChatScreen's
|
||||
// sync effect fires this immediately on session switch, before the
|
||||
// resume result lands, so skip the RPCs until resume confirms the
|
||||
// runtime id. The REST parts below stay live (they key on the
|
||||
// storage id).
|
||||
val rpcSessionId = runtimeSessionId
|
||||
if (rpcSessionId != null) {
|
||||
try {
|
||||
val result =
|
||||
sendRpcAndAwait(
|
||||
WsMethods.SESSION_CONTEXT_BREAKDOWN,
|
||||
mapOf("session_id" to rpcSessionId),
|
||||
)
|
||||
val ctx = parseContextBreakdown(result)
|
||||
if (ctx != null) {
|
||||
_uiState.update { current ->
|
||||
current.copy(
|
||||
usedContextTokens =
|
||||
ctx.contextUsed?.takeIf { it > 0L } ?: current.usedContextTokens,
|
||||
fullContextTokens =
|
||||
ctx.contextMax?.takeIf { it > 0L } ?: current.fullContextTokens,
|
||||
)
|
||||
}
|
||||
}
|
||||
} catch (_: Exception) {
|
||||
// Best-effort: RPC error/timeout/disconnect — keep last values.
|
||||
}
|
||||
} catch (_: Exception) {
|
||||
// Best-effort: RPC error/timeout/disconnect — keep last values.
|
||||
}
|
||||
// Compression count: how many times this session has been compacted
|
||||
// (session.usage → compressions). Feeds the "compressed ×N" badge —
|
||||
// the same usage snapshot the desktop status bar reads.
|
||||
try {
|
||||
val usage =
|
||||
sendRpcAndAwait(
|
||||
WsMethods.SESSION_USAGE,
|
||||
mapOf("session_id" to rpcSessionId),
|
||||
)
|
||||
val snapshot = parseUsageSnapshot(usage)
|
||||
if (snapshot != null && snapshot.compressions != null) {
|
||||
_uiState.update { it.copy(compressionCount = snapshot.compressions) }
|
||||
// Compression count: how many times this session has been compacted
|
||||
// (session.usage → compressions). Feeds the "compressed ×N" badge —
|
||||
// the same usage snapshot the desktop status bar reads.
|
||||
try {
|
||||
val usage =
|
||||
sendRpcAndAwait(
|
||||
WsMethods.SESSION_USAGE,
|
||||
mapOf("session_id" to rpcSessionId),
|
||||
)
|
||||
val snapshot = parseUsageSnapshot(usage)
|
||||
if (snapshot != null && snapshot.compressions != null) {
|
||||
_uiState.update { it.copy(compressionCount = snapshot.compressions) }
|
||||
}
|
||||
} catch (_: Exception) {
|
||||
// Best-effort: keep the last known badge value.
|
||||
}
|
||||
} catch (_: Exception) {
|
||||
// Best-effort: keep the last known badge value.
|
||||
}
|
||||
// Detail-sheet accounting (cumulative REST counters, informational).
|
||||
val usedResult =
|
||||
|
||||
@@ -1255,6 +1255,99 @@ class ChatViewModelTest {
|
||||
assertFalse(viewModel.uiState.value.isResumeRetrying)
|
||||
}
|
||||
|
||||
/**
|
||||
* Regression for the spurious "Error: JsonRpcError(code=4001, message=session
|
||||
* not found, data=null)" snackbar when opening a session from history: the
|
||||
* chat sync effect fires fetchContextUsage() immediately on session switch,
|
||||
* BEFORE session.resume has registered the session on the gateway. The
|
||||
* context/usage RPCs resolve against the gateway's LIVE runtime registry
|
||||
* (_sess_nowait) and would 4001 on the storage id — so they must be skipped
|
||||
* until the resume result confirms the runtime id (which is then used).
|
||||
*/
|
||||
@Test
|
||||
fun testSwitchSession_contextRpcSkippedUntilResumeConfirmsRuntimeId() =
|
||||
runTest {
|
||||
stubSession456Rests(success = true)
|
||||
|
||||
// Awaited live-registry RPCs (sendRpcAndAwait → HermesWsClient.request).
|
||||
// Completed deferred so the awaits resolve; capture the params. Installed
|
||||
// BEFORE createViewModelWithSession so the SESSION_CREATE-era
|
||||
// fetchContextUsage() call is stubbed too (otherwise its unstubbed
|
||||
// invocation is still recorded and pollutes the verify(exactly = 0)).
|
||||
val paramsSlot = slot<Map<String, Any>>()
|
||||
every {
|
||||
HermesWsClient.request(any(), capture(paramsSlot), any())
|
||||
} returns CompletableDeferred<Any?>(emptyMap<String, Any?>())
|
||||
|
||||
val (viewModel, _) = createViewModelWithSession()
|
||||
|
||||
// Capture the session.resume request id so its result can be delivered.
|
||||
var resumeRequestId: String? = null
|
||||
every { HermesWsClient.send(WsMethods.SESSION_RESUME, any(), any()) } answers {
|
||||
reqCount++
|
||||
val id = "resume-$reqCount"
|
||||
arg<((String) -> Unit)?>(2)?.invoke(id)
|
||||
resumeRequestId = id
|
||||
id
|
||||
}
|
||||
|
||||
// Switch to session-456: runtimeSessionId resets to null until the
|
||||
// resume result lands — the storage id is not (yet) live on the
|
||||
// gateway. ChatScreen's sync effect fires fetchContextUsage()
|
||||
// immediately on session switch — simulate it while the resume
|
||||
// result is still in flight. The live-registry RPCs must NOT fire
|
||||
// with the stale storage id (the gateway would 4001).
|
||||
viewModel.switchSession("session-456")
|
||||
advanceUntilIdle()
|
||||
viewModel.fetchContextUsage()
|
||||
advanceUntilIdle()
|
||||
|
||||
verify(exactly = 0) {
|
||||
HermesWsClient.request(
|
||||
WsMethods.SESSION_CONTEXT_BREAKDOWN,
|
||||
match { it["session_id"] == "session-456" },
|
||||
any(),
|
||||
)
|
||||
}
|
||||
verify(exactly = 0) {
|
||||
HermesWsClient.request(
|
||||
WsMethods.SESSION_USAGE,
|
||||
match { it["session_id"] == "session-456" },
|
||||
any(),
|
||||
)
|
||||
}
|
||||
|
||||
// Resume result lands → runtime id confirmed → the RPCs fire with it.
|
||||
val resumeId = checkNotNull(resumeRequestId)
|
||||
mockEventsFlow.emit(
|
||||
WsEvent.RpcResult(
|
||||
resumeId,
|
||||
mapOf(
|
||||
"session_id" to "runtime-456",
|
||||
"resumed" to "session-456",
|
||||
"info" to emptyMap<String, Any?>(),
|
||||
),
|
||||
),
|
||||
)
|
||||
advanceUntilIdle()
|
||||
|
||||
verify {
|
||||
HermesWsClient.request(
|
||||
WsMethods.SESSION_CONTEXT_BREAKDOWN,
|
||||
match { it["session_id"] == "runtime-456" },
|
||||
any(),
|
||||
)
|
||||
}
|
||||
verify {
|
||||
HermesWsClient.request(
|
||||
WsMethods.SESSION_USAGE,
|
||||
match { it["session_id"] == "runtime-456" },
|
||||
any(),
|
||||
)
|
||||
}
|
||||
assertEquals("runtime-456", paramsSlot.captured["session_id"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testSessionResume_boundedRetry_thenExplicitError() =
|
||||
runTest {
|
||||
|
||||
Reference in new issue
Block a user