diff --git a/apps/desktop/src/app/chat/composer/controls.tsx b/apps/desktop/src/app/chat/composer/controls.tsx index 609cf51a44..5444cce0ef 100644 --- a/apps/desktop/src/app/chat/composer/controls.tsx +++ b/apps/desktop/src/app/chat/composer/controls.tsx @@ -87,10 +87,11 @@ export function ComposerControls({ // Steer is just send: a payload keeps the Send affordance mid-turn (it answers // a waiting card). Stop only when the composer is empty and a turn is running. const showStop = busy && !hasComposerPayload - const showQueueButton = - busy && busyAction !== 'stop' && busyAction !== 'queue' && hasComposerPayload && !cardWaiting - const sendLabel = composerSendLabel(busy, busyAction, c) + const showQueueButton = queueButtonVisible(busy, busyAction, hasComposerPayload, cardWaiting) + + // A waiting card takes the draft as its answer through Send, whatever the busy action. + const sendLabel = composerSendLabel(busy && !cardWaiting, busyAction, c) // The HUD is a Spotlight bar a few hundred pixels wide, so the four separate // voice toggles fold into one menu there and leave the row to the input. A @@ -191,6 +192,17 @@ export function ComposerControls({ ) } +// Queue is the secondary mid-turn action: never while the primary button already queues or stops, and +// never while a card waits (its Send answers the card). +function queueButtonVisible( + busy: boolean, + busyAction: 'interrupt' | 'steer' | 'queue' | 'stop', + hasPayload: boolean, + cardWaiting: boolean +): boolean { + return busy && busyAction !== 'stop' && busyAction !== 'queue' && hasPayload && !cardWaiting +} + function composerSendLabel( busy: boolean, busyAction: 'interrupt' | 'steer' | 'queue' | 'stop', diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/input-requests.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/input-requests.ts index 9f453d6b29..2251858523 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/input-requests.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/input-requests.ts @@ -232,14 +232,16 @@ export function handleInputRequestEvent(ctx: GatewayEventContext): boolean { }) } } else { - cancelParkedCredentialRequest(key, sessionId, id) + cancelParkedCredentialRequest(ctx, key, id) } return true } -/** Withdraw the sudo / secret / vault card parked under `id`, first match wins. */ -function cancelParkedCredentialRequest(key: string, sessionId: GatewayEventContext['sessionId'], id: string): void { +/** Withdraw the sudo / secret / vault / setup card parked under `id`, first match wins. */ +function cancelParkedCredentialRequest(ctx: GatewayEventContext, key: string, id: string): void { + const { sessionId } = ctx + if ($sudoRequests.get()[key]?.requestId === id) { clearSudoRequest(sessionId, id) } else if ($sudoRequests.get()['']?.requestId === id) { diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts index 036c096a55..2deb0a9a6b 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts @@ -211,10 +211,7 @@ export function useSlashCommand(deps: SlashCommandDeps) { const compressInFlightRef = useRef(new Set()) return useCallback( - async ( - rawCommand: string, - options?: SubmitTextOptions & { recordInput?: boolean; typed?: boolean } - ) => { + async (rawCommand: string, options?: SubmitTextOptions & { recordInput?: boolean; typed?: boolean }) => { const initialRuntimeId = options?.sessionId ?? activeSessionIdRef.current const initialSelectedId = selectedStoredSessionIdRef.current const initialRoutedId = getRoutedStoredSessionId() @@ -238,16 +235,18 @@ export function useSlashCommand(deps: SlashCommandDeps) { // An explicit retry of this exact invocation reuses its retained expansion; the same // command typed again is a new invocation. const prepared = options?.submission_id - ? (await adoptPreparedSubmission( - preparedSubmissionKey( - resolveComposerSessionKey(initialStoredId ?? initialRuntimeId, $sessions.get()), - destination, - rawCommand, - options?.attachments ?? $composerAttachments.get(), - retryOptions - ), - options.submission_id - ))?.entry + ? ( + await adoptPreparedSubmission( + preparedSubmissionKey( + resolveComposerSessionKey(initialStoredId ?? initialRuntimeId, $sessions.get()), + destination, + rawCommand, + options?.attachments ?? $composerAttachments.get(), + retryOptions + ), + options.submission_id + ) + )?.entry : undefined if (prepared) { diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index 883c034b05..e3351708a4 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -2691,7 +2691,9 @@ export function useSessionActions({ // Stable per-attempt key: a lost-response retry of session.branch / // session.branch_whole returns the SAME child (#65410). idempotency_key: key, - ...(branchCount !== undefined ? { count: branchCount, through_message_id: branchThroughRowId(branchMessages) } : {}) + ...(branchCount !== undefined + ? { count: branchCount, through_message_id: branchThroughRowId(branchMessages) } + : {}) } const createParams = { diff --git a/apps/desktop/src/i18n/overlay-gaps.json b/apps/desktop/src/i18n/overlay-gaps.json index ecc9f0c3da..3ca326dfb1 100644 --- a/apps/desktop/src/i18n/overlay-gaps.json +++ b/apps/desktop/src/i18n/overlay-gaps.json @@ -320,7 +320,6 @@ "connectors.directory.allApps": ["ar","ja","ru","zh-hant"], "connectors.directory.receiptConnected": ["ar","ja","ru","zh-hant"], "connectors.directory.receiptNone": ["ar","ja","ru","zh-hant"], - "connectors.directory.receiptSaid": ["ar","ja","ru","zh-hant"], "connectors.directory.receiptTimedOut": ["ar","ja","ru","zh-hant"], "connectors.directory.suggested": ["ar","ja","ru","zh-hant"], "connectors.disabled": ["ar","ja","ru","zh-hant"], diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py index 0a234f542a..ce2032608b 100644 --- a/hermes_state_sessions.py +++ b/hermes_state_sessions.py @@ -783,32 +783,14 @@ class SessionSessionsMixin: def update_conversation_model_config_value(self, session_id: str, key: str, update: Callable[[Any], Any]) -> Optional[str]: - """Read a conversation-wide ``key`` at *session_id*'s compression tip and write ``update(value)`` - back, in one write transaction: a concurrent writer of the key cannot be overwritten with a stale - copy, and a rotation cannot publish between finding the tip and writing it (the child would keep - the old copy). ``update`` may run more than once (lock retry) and must be pure; returning None - removes the key. Returns the id written, None when the row does not exist.""" - def _do(conn) -> Optional[str]: - tip = self._compression_tip_on(conn, session_id) - row = conn.execute("SELECT model_config FROM sessions WHERE id = ?", (tip,)).fetchone() - if row is None: - return None - config = _parse_model_config(row[0]) - if (value := update(config.get(key))) is None: - config.pop(key, None) - else: - config[key] = value - conn.execute("UPDATE sessions SET model_config = ? WHERE id = ?", (json.dumps(config), tip)) - return tip - return self._execute_write(_do) + """Write ``update(value)`` of a conversation-wide key at the compression tip (one transaction).""" + from hermes_state_sessions_model_config import update_conversation_model_config_value + return update_conversation_model_config_value(self, session_id, key, update) def session_ids_with_model_config_key(self, key: str) -> list[str]: """Ids of the sessions whose model_config carries ``key``, newest first.""" - with self._read_ctx() as conn: - rows = conn.execute( - f"SELECT id FROM sessions WHERE {_sql_json_extract('model_config', '$.' + key)} IS NOT NULL " - "ORDER BY started_at DESC").fetchall() - return [row["id"] for row in rows] + from hermes_state_sessions_model_config import session_ids_with_model_config_key + return session_ids_with_model_config_key(self, key) def update_session_runtime_lock( self, session_id: str, *, model: Optional[str] = None, provider: Optional[str] = None, diff --git a/hermes_state_sessions_model_config.py b/hermes_state_sessions_model_config.py new file mode 100644 index 0000000000..6e92a60bee --- /dev/null +++ b/hermes_state_sessions_model_config.py @@ -0,0 +1,42 @@ +"""Conversation-wide ``sessions.model_config`` keys: read-modify-write at the compression tip and the +key scan. Bodies of :class:`hermes_state_sessions.SessionSessionsMixin`'s two methods of the same names.""" + +from __future__ import annotations + +import json +from typing import Any, Callable, Optional + +from hermes_state_common import _sql_json_extract + + +def update_conversation_model_config_value(db, session_id: str, key: str, + update: Callable[[Any], Any]) -> Optional[str]: + """Read a conversation-wide ``key`` at *session_id*'s compression tip and write ``update(value)`` + back, in one write transaction: a concurrent writer of the key cannot be overwritten with a stale + copy, and a rotation cannot publish between finding the tip and writing it (the child would keep + the old copy). ``update`` may run more than once (lock retry) and must be pure; returning None + removes the key. Returns the id written, None when the row does not exist.""" + from hermes_state_sessions import _parse_model_config + + def _do(conn) -> Optional[str]: + tip = db._compression_tip_on(conn, session_id) + row = conn.execute("SELECT model_config FROM sessions WHERE id = ?", (tip,)).fetchone() + if row is None: + return None + config = _parse_model_config(row[0]) + if (value := update(config.get(key))) is None: + config.pop(key, None) + else: + config[key] = value + conn.execute("UPDATE sessions SET model_config = ? WHERE id = ?", (json.dumps(config), tip)) + return tip + return db._execute_write(_do) + + +def session_ids_with_model_config_key(db, key: str) -> list[str]: + """Ids of the sessions whose model_config carries ``key``, newest first.""" + with db._read_ctx() as conn: + rows = conn.execute( + f"SELECT id FROM sessions WHERE {_sql_json_extract('model_config', '$.' + key)} IS NOT NULL " + "ORDER BY started_at DESC").fetchall() + return [row["id"] for row in rows] diff --git a/tests/e2e/test_relay_native_openai_stream.py b/tests/e2e/test_relay_native_openai_stream.py index d240008104..7d1d2794dc 100644 --- a/tests/e2e/test_relay_native_openai_stream.py +++ b/tests/e2e/test_relay_native_openai_stream.py @@ -63,15 +63,10 @@ def _stream_through_relay(tmp_path, monkeypatch, response_body: bytes, *, finali subscriber_name = "test.openai_stream" events = [] relay_finalizer_started = threading.Event() + allow_relay_finalizer = threading.Event() relay_finalizer_finished = threading.Event() run_relay_finalizer = relay_llm.ManagedLlmStream._relay_finalizer - # The finalizer is never gated. Relay runs it on the stream's own loop, which only the consumer - # thread drives (``__next__`` -> ``run_until_complete``), and Relay may reach EOF and finalize - # BEFORE it hands over the terminal chunk. A finalizer that waited for the consumer then parked the - # only thread that could release it: a 30 s deadlock, then "consumer did not release Relay's - # finalizer". Finalizing first already IS the ordering under test; the consumer-first ordering is - # forced below by pumping the loop until the finalizer has finished. def run_synchronized_relay_finalizer(managed_stream, attempt): relay_finalizer_started.set() try: @@ -100,6 +95,7 @@ def _stream_through_relay(tmp_path, monkeypatch, response_body: bytes, *, finali import asyncio stream = self.managed_stream_holder["stream"] + allow_relay_finalizer.set() # A schedule probe may hold the provider generator at this chunk until the consumer has # taken it (adverse consumer-first ordering); release it so the pump below can reach EOF. gate = getattr(stream, "_review_gate", None)