feat(web): float queued messages above composer until invocation (#542)

* refactor: add invoked_at column and propagate via messages-consumed - Bump hub schema to V8: add `invoked_at INTEGER` to messages table - Add `migrateFromV7ToV8` (idempotent ALTER TABLE ADD COLUMN) - Add migration chain entries for V4/V5/V6/V7 → V8 - Expose `StoredMessage.invokedAt: number | null` and `markMessagesInvoked` - Record server-side `Date.now()` in hub on `messages-consumed` socket event - Propagate `invokedAt` through SSE (`messages-consumed` payload) - Update `markMessagesConsumed` in web store to accept and store `invokedAt` - Preserve optimistic `invokedAt` in `mergeMessages` (server echo path) - Add migration unit tests (fresh V8, V7→V8 ALTER, markMessagesInvoked) * feat(web): float queued messages above composer until invocation Show queued (uninvoked) user messages in a dedicated floating bar above the composer instead of inline in the thread timeline. Once the CLI acks the batch via messages-consumed, the bar disappears and the messages appear in the thread at their invocation position (invokedAt ordering). - Add QueuedMessagesBar component: subscribes to message-window-store, filters user messages with invokedAt==null, shows clock icon + text preview; disappears when all messages are invoked - Filter queued messages from thread (visibleMessages), sort by invokedAt ?? createdAt so invoked messages land at the right position - Extend markMessagesConsumed to update server-loaded messages (status undefined) in addition to optimistic (status 'queued'), enabling multi-device and post-refresh scenarios - Remove opacity-60 from UserMessage: queued messages no longer appear in the thread so the dimming branch is unreachable - Include invokedAt in getMessagesPage/getMessagesAfter API responses so the web client can restore floating-bar state after page refresh - Add invokedAt field to DecryptedMessageSchema for shared protocol type * fix(hub,web): make sort use invokedAt and V8 backfill idempotent - compareMessages: prioritize invokedAt/createdAt over seq so invoked messages land at their invocation position rather than their send-time seq position - migrateFromV7ToV8: move backfill outside the ALTER guard so it re-runs if a previous attempt crashed between ALTER and UPDATE before the user_version bump (idempotent WHERE invoked_at IS NULL) * fix(hub,web): cover localId-less messages and live-ack invokedAt - addMessage: messages without a localId have no ack path (markMessagesInvoked matches by localId). Treat them as already-invoked at insert time so they land in the thread instead of sitting in the queued floating bar forever. - markMessagesConsumed: apply the ack even when the message is already 'sent' optimistically, so the live window receives invokedAt instead of waiting until a full refetch. * fix(hub): propagate invokedAt in live message-received SSE payload The SSE `message-received` event omitted `invokedAt` while REST pagination included it, so localId-less CLI/local user messages arrived on the live wire as queued (`invokedAt == null`) and stayed in the floating bar until a full refetch replaced them with the stored row. * fix(hub): propagate invokedAt in CLI socket message-received handler The CLI socket 'message' handler fans out to web via a separate `onWebappEvent` publisher; the previous fix only touched the `MessageService` publisher. Aligns the live SSE payload shape with the REST/page-load shape so localId-less CLI/local user messages with `invokedAt = createdAt` (set in addMessage) reach web filters with the field already populated, instead of being misclassified as queued until a full refetch. * fix(hub,web): add byPosition pagination to fix long-session queued message loss Pagination used seq-based windows, so queued messages with low seq but late invokedAt fell outside the visible window on refresh. Fix by adding a V8 byPosition mode that orders by COALESCE(invoked_at, created_at) DESC, seq DESC with a composite cursor, while keeping the V7 seq path fully intact for backward compatibility. - hub/store/index: add idx_messages_session_position (createSchema + V7→V8 migration) - hub/store/messages: add getMessagesByPosition with composite cursor SQL - hub/store/messageStore: delegate getMessagesByPosition - hub/sync/messageService: add getMessagesPageByPosition with nextBeforeAt response - hub/sync/syncEngine: expose getMessagesPageByPosition - hub/web/routes/messages: byPosition=1 query param dispatches to V8 path - web/types/api: MessagesResponse.page gains optional nextBeforeAt - web/api/client: getMessages gains byPosition + beforeAt options - web/lib/message-window-store: fetchLatestMessages/fetchOlderMessages use V8 composite cursor; fallback to seq cursor when hub returns no nextBeforeAt - hub/store/migration-v8.test: 7 new tests covering position sort, composite cursor pagination, long-session scenario, V7 compat, and index existence * fix(hub,web): re-sort on consume and use position cursor for next fetch - markMessagesConsumed: re-merge with empty list to re-sort by position key after invokedAt is set. A queued user message becomes visible with the consume event; without re-sort it stays at its send-time array slot until the next fetch overwrites it. - getMessagesPageByPosition: pick the cursor from stored[0] (oldest in position order) instead of scanning for minimum seq. With the page already in ascending position order, scanning for min seq could land on a low-seq, late-invoked row that is actually the newest in the page, causing the next older fetch to overlap. * fix(web): trust invokedAt as the only invocation signal and pin cursor pair - visibleMessages predicate (SessionChat + QueuedMessagesBar): drop the status === 'sent' check. status='sent' only means the REST write returned, not that the CLI consumed the message; an optimistic 'sent' with no invokedAt is still queued. invokedAt is the single source of truth for invocation. - byPosition cursor: track oldestPositionSeq alongside oldestPositionAt so the server's cursor pair travels through the next older fetch unchanged. Recomputing beforeSeq from the local window's minimum seq could combine it with a server beforeAt that referred to a different row, causing the SQL cursor to skip or overlap. * fix(hub): include uninvoked local messages in latest page Long sessions can push a queued user message (invokedAt = null, sort key = createdAt) outside the latest position-ordered page once the agent emits more than `limit` later rows. A refresh or secondary client then never receives the row, the floating bar stays empty, and the later `messages-consumed` event only carries localIds — there is no way to materialize the missing row at invocation time. Pin uninvoked local user messages to every latest-page response out-of-band. The pagination cursor still anchors to the position-ordered page rows, so older-page fetches are unaffected. * fix(web): preserve queued messages across trimVisible The visible-window trim drops the oldest entries beyond VISIBLE_WINDOW_SIZE, but a queued user message (invokedAt = null) sorts by send time and is the oldest item. Once a long agent stream pushes it past the window the row is gone from the client store, and the `messages-consumed` SSE carries only localIds — there is no way to restore or reposition the dropped row without a full refetch. Pull queued rows out before slicing the regular budget, then merge them back in. Queued rows are bounded by composer/CLI queue depth and do not meaningfully grow the window. * fix(web): use strict null for queued check and fall back invokedAt - Optimistic message sets invokedAt: null explicitly so the strict-null queued check matches the local opt-in. Pre-V8 hub responses that omit the field (`undefined`) are treated as already-invoked and stay in the thread instead of being misclassified as queued. - markMessagesConsumed: when the consume SyncEvent omits invokedAt (older hub) fall back to client time, otherwise a message that receives an ack with no server timestamp stays queued forever under the new strict-null filter. The persisted server value is still authoritative on next fetch. * fix: comprehensive invokedAt propagation hardening (review feedback batch) Bot review surfaced 11 propagation bugs incrementally; this batch fixes 9 additional adjacent issues found by hostile-review to break the incremental discovery cycle: - legacy DB (user_version=0 with HAPI tables): step ladder runs V1→V8 before createSchema so pre-existing tables get all later columns/indexes - step ladder includes V1/V2/V3 entries; previously V1-V3 DBs threw - mergeSessionMessages collision branch forces invoked_at = created_at so unmergeable rows can't strand in the floating bar - session-end auto-invokes still-queued user messages and broadcasts messages-consumed; the floating bar no longer pins ghost rows after the CLI is gone - trimPending preserves queued rows symmetrically with trimVisible - markMessagesInvoked is first-write-wins; duplicate acks are no-ops rather than re-stamping invoked_at and reordering the thread - markMessagesConsumed migrates just-acked pending entries into the visible thread so non-at-bottom users see their own messages without scrolling - mergeMessages dedup window compares by position key (invokedAt ?? createdAt) instead of createdAt only, so late-invoked optimistic copies don't duplicate the server echo - isQueuedForInvocation centralized in lib/messages.ts (single predicate used by SessionChat, QueuedMessagesBar, and the store) * fix(web): mirror hub's first-write-wins on markMessagesConsumed The hub's markMessagesInvoked is first-write-wins, but the web store was still overwriting any non-null invokedAt with the latest messages-consumed timestamp. A duplicate ack (CLI re-emit) would leave the SQLite row at the original timestamp while live clients moved the message to the duplicate ack time, diverging until refetch. Mirror the guard: only set invokedAt when it is null. * fix: in-scope hostile-review polish Web: - fetchLatestMessages: persist the V8 composite cursor pair on the non-at-bottom branch too. Without this, a refresh while scrolled up dropped the cursor and the next loadMore fell back to V7 seq mode against a V8 hub — same asymmetric class of bug commit 30df6b2 fixed for the at-bottom path. - markMessagesConsumed: tighten the loose-null check on invokedAt to strict null, consistent with isQueuedForInvocation and the rest of the file. The idSet filter already shields V7-stamped rows from this path, but the strict-null contract should not vary by call site. - messages: drop the upsertMessagesInCache export. It has no callers (verified with grep) and is the only user of the InfiniteData / MessagesResponse imports, so the imports go with it. Hub tests: - migration-v8.test.ts: add a session-end auto-invoke test (getUninvokedLocalMessages + markMessagesInvoked clears every queued row and stamps them all with the same invokedAt) and two byPosition union tests covering (1) a low-position queued row pushed out of the latest page is still surfaced via the uninvoked set, and (2) pageRows[0] is the oldest row in the page so the web client can safely anchor the next-older cursor on it. * fix(hub,web): bot-13 polish — atomic SSE on DB success and attachment chip text - sessionHandlers messages-consumed: emit messages-consumed only after markMessagesInvoked succeeds. Otherwise a transient SQLite failure would broadcast an invokedAt that was never persisted; live clients would hide the queued rows while a refresh / secondary client would see them as queued again, diverging the state. - QueuedMessagesBar: fall back to attachment filenames when the message text is empty. The composer / POST /messages allow attachment-only sends; without the fallback those queued messages rendered as blank chips until invocation.
2026-08-06 06:41:56 +00:00 · 2026-04-29 17:23:01 +08:00
parent e76738aa5a
commit 7d55bc1456
20 changed files with 1489 additions and 149 deletions
@@ -191,8 +191,22 @@ export class ApiClient {
        return await this.request<SessionResponse>(`/api/sessions/${encodeURIComponent(sessionId)}`)
    }

-    async getMessages(sessionId: string, options: { beforeSeq?: number | null; limit?: number }): Promise<MessagesResponse> {
+    async getMessages(
+        sessionId: string,
+        options: {
+            beforeSeq?: number | null
+            beforeAt?: number | null
+            byPosition?: boolean
+            limit?: number
+        }
+    ): Promise<MessagesResponse> {
        const params = new URLSearchParams()
+        if (options.byPosition || options.beforeAt !== undefined && options.beforeAt !== null) {
+            params.set('byPosition', '1')
+        }
+        if (options.beforeAt !== undefined && options.beforeAt !== null) {
+            params.set('beforeAt', `${options.beforeAt}`)
+        }
        if (options.beforeSeq !== undefined && options.beforeSeq !== null) {
            params.set('beforeSeq', `${options.beforeSeq}`)
        }
@@ -0,0 +1,109 @@
+import { useCallback, useSyncExternalStore } from 'react'
+import { getMessageWindowState, subscribeMessageWindow } from '@/lib/message-window-store'
+import { isQueuedForInvocation } from '@/lib/messages'
+import { EMPTY_STATE } from '@/hooks/queries/useMessages'
+import { normalizeDecryptedMessage } from '@/chat/normalize'
+import type { DecryptedMessage } from '@/types/api'
+
+function ClockIcon() {
+    return (
+        <svg
+            className="h-[14px] w-[14px] shrink-0"
+            viewBox="0 0 16 16"
+            fill="none"
+            aria-hidden="true"
+        >
+            <circle cx="8" cy="8" r="6" stroke="currentColor" strokeWidth="1.5" />
+            <path
+                d="M8 5v3.5l2.5 1.5"
+                stroke="currentColor"
+                strokeWidth="1.5"
+                strokeLinecap="round"
+                strokeLinejoin="round"
+            />
+        </svg>
+    )
+}
+
+/**
+ * Returns user messages that haven't been invoked yet (invokedAt == null and not sent/failed).
+ * Covers both optimistic (status='queued') and server-loaded (status=undefined, invokedAt=null) cases.
+ */
+function useQueuedMessages(sessionId: string): DecryptedMessage[] {
+    const state = useSyncExternalStore(
+        useCallback((listener) => subscribeMessageWindow(sessionId, listener), [sessionId]),
+        useCallback(() => getMessageWindowState(sessionId), [sessionId]),
+        () => EMPTY_STATE
+    )
+
+    // `invokedAt` is the source of truth for invocation; see isQueuedForInvocation
+    // (lib/messages) for the shared predicate used by the thread filter and the
+    // window store trim helpers.
+    const allMessages = [...state.messages, ...state.pending]
+    return allMessages.filter(isQueuedForInvocation)
+}
+
+function getTextFromMessage(msg: DecryptedMessage): string {
+    const normalized = normalizeDecryptedMessage(msg)
+    if (!normalized || normalized.role !== 'user') {
+        return ''
+    }
+    const text = (normalized.content.text ?? '').trim()
+    if (text) {
+        return text
+    }
+    // Attachment-only sends: the composer / POST /messages allow empty text
+    // when attachments are present. Fall back to the filenames so the chip
+    // is not blank.
+    const attachments = normalized.content.attachments ?? []
+    if (attachments.length === 0) {
+        return ''
+    }
+    return attachments.map((a) => a.filename ?? 'attachment').join(', ')
+}
+
+/**
+ * Floating bar above the composer showing queued (pending invocation) messages.
+ * Disappears automatically when all queued messages are invoked or consumed.
+ *
+ * TODO PR 2: add cancel/edit buttons per item.
+ */
+export function QueuedMessagesBar({ sessionId }: { sessionId: string }) {
+    const queued = useQueuedMessages(sessionId)
+
+    if (queued.length === 0) {
+        return null
+    }
+
+    return (
+        <div
+            role="status"
+            aria-label={`${queued.length} queued message${queued.length === 1 ? '' : 's'} pending invocation`}
+            className="mx-auto w-full max-w-content mb-1"
+        >
+            <div className="px-3 py-2 text-sm text-[var(--app-fg-muted)]">
+                <div className="flex items-center gap-1.5 mb-1.5 text-xs font-medium text-[var(--app-hint)]">
+                    <ClockIcon />
+                    <span>Queued</span>
+                </div>
+                <ul
+                    className="flex flex-col gap-1.5 max-h-32 sm:max-h-48 overflow-y-auto"
+                    aria-label="Queued messages"
+                >
+                    {queued.map((msg) => {
+                        const text = getTextFromMessage(msg)
+                        return (
+                            <li
+                                key={msg.localId ?? msg.id}
+                                className="flex items-start gap-2 min-w-0 rounded-lg bg-[var(--app-secondary-bg)] px-3 py-2 shadow-sm"
+                            >
+                                <span className="line-clamp-3 whitespace-pre-wrap break-words text-[var(--app-fg)]">{text}</span>
+                                {/* TODO PR 2: cancel/edit buttons */}
+                            </li>
+                        )
+                    })}
+                </ul>
+            </div>
+        </div>
+    )
+}
@@ -47,7 +47,7 @@ export function HappyUserMessage() {
    const canRetry = status === 'failed' && typeof localId === 'string' && Boolean(ctx.onRetryMessage)
    const onRetry = canRetry ? () => ctx.onRetryMessage!(localId) : undefined

-    const userBubbleClass = `w-fit min-w-0 max-w-[92%] ml-auto rounded-xl bg-[var(--app-secondary-bg)] px-3 py-2 text-[var(--app-fg)] shadow-sm${status === 'queued' ? ' opacity-60' : ''}`
+    const userBubbleClass = `w-fit min-w-0 max-w-[92%] ml-auto rounded-xl bg-[var(--app-secondary-bg)] px-3 py-2 text-[var(--app-fg)] shadow-sm`

    if (isCliOutput) {
        return (
@@ -16,8 +16,10 @@ import { normalizeDecryptedMessage } from '@/chat/normalize'
 import { reduceChatBlocks } from '@/chat/reducer'
 import { reconcileChatBlocks } from '@/chat/reconcile'
 import { buildConversationOutline } from '@/chat/outline'
+import { isQueuedForInvocation } from '@/lib/messages'
 import { HappyComposer } from '@/components/AssistantChat/HappyComposer'
 import { HappyThread } from '@/components/AssistantChat/HappyThread'
+import { QueuedMessagesBar } from '@/components/AssistantChat/QueuedMessagesBar'
 import { useHappyRuntime } from '@/lib/assistant-runtime'
 import { createAttachmentAdapter } from '@/lib/attachmentAdapter'
 import { useTranslation } from '@/lib/use-translation'
@@ -209,6 +211,15 @@ export function SessionChat(props: {
        setOutlineOpen(false)
    }, [props.session.id])

+    // Exclude user messages that haven't been invoked yet — those appear in the
+    // QueuedMessagesBar above the composer, not in the thread timeline. The
+    // `isQueuedForInvocation` predicate is shared with the window store and the
+    // floating bar so the three views never disagree about queued state.
+    const visibleMessages = useMemo(
+        () => props.messages.filter((m) => !isQueuedForInvocation(m)),
+        [props.messages]
+    )
+
    const normalizedMessages: NormalizedMessage[] = useMemo(() => {
        // Clear caches immediately when session changes (before useEffect runs)
        if (prevSessionIdRef.current !== null && prevSessionIdRef.current !== props.session.id) {
@@ -220,7 +231,7 @@ export function SessionChat(props: {
        const cache = normalizedCacheRef.current
        const normalized: NormalizedMessage[] = []
        const seen = new Set<string>()
-        for (const message of props.messages) {
+        for (const message of visibleMessages) {
            seen.add(message.id)
            const cached = cache.get(message.id)
            if (cached && cached.source === message) {
@@ -237,7 +248,7 @@ export function SessionChat(props: {
            }
        }
        return normalized
-    }, [props.messages])
+    }, [visibleMessages])

    const reduced = useMemo(
        () => reduceChatBlocks(normalizedMessages, props.session.agentState),
@@ -408,7 +419,7 @@ export function SessionChat(props: {
                        isLoadingMoreMessages={props.isLoadingMoreMessages}
                        onLoadMore={props.onLoadMore}
                        pendingCount={props.pendingCount}
-                        rawMessagesCount={props.messages.length}
+                        rawMessagesCount={visibleMessages.length}
                        normalizedMessagesCount={normalizedMessages.length}
                        messagesVersion={props.messagesVersion}
                        forceScrollToken={forceScrollToken}
@@ -426,6 +437,10 @@ export function SessionChat(props: {
                        </div>
                    ) : null}

+                    <div className="px-3">
+                        <QueuedMessagesBar sessionId={props.session.id} />
+                    </div>
+
                    <HappyComposer
                        key={props.session.id}
                        sessionId={props.session.id}
@@ -44,6 +44,10 @@ function createOptimisticMessage(input: SendMessageInput, status: 'queued' | 'se
            }
        },
        createdAt: input.createdAt,
+        // Explicit null so the strict-null queued check matches. A pre-V8 hub
+        // response that omits the field entirely (`undefined`) is treated as
+        // already-invoked and stays in the thread, not the floating bar.
+        invokedAt: null,
        status,
        originalText: input.text,
    }
@@ -12,7 +12,7 @@ import {
    type MessageWindowState,
 } from '@/lib/message-window-store'

-const EMPTY_STATE: MessageWindowState = {
+export const EMPTY_STATE: MessageWindowState = {
    sessionId: 'unknown',
    messages: [],
    pending: [],
@@ -498,7 +498,7 @@ export function useSSE(options: {
            }

            if (event.type === 'messages-consumed') {
-                markMessagesConsumed(event.sessionId, event.localIds)
+                markMessagesConsumed(event.sessionId, event.localIds, event.invokedAt)
            }

            if (event.type === 'message-received') {
@@ -1,7 +1,7 @@
 import type { ApiClient } from '@/api/client'
 import type { DecryptedMessage, MessageStatus } from '@/types/api'
 import { normalizeDecryptedMessage } from '@/chat/normalize'
-import { isUserMessage, mergeMessages } from '@/lib/messages'
+import { isQueuedForInvocation, isUserMessage, mergeMessages } from '@/lib/messages'

 export type MessageWindowState = {
    sessionId: string
@@ -27,6 +27,13 @@ type InternalState = MessageWindowState & {
    pendingOverflowCount: number
    pendingVisibleCount: number
    pendingOverflowVisibleCount: number
+    // V8 composite cursor: defined when hub responded with nextBeforeAt
+    oldestPositionAt: number | null
+    // Paired with oldestPositionAt — the server returns both as a cursor; keep them
+    // together so we don't accidentally combine `nextBeforeAt` from the server with
+    // a recomputed minimum `seq` from the local window (those can refer to
+    // different rows after a low-seq message is invoked late).
+    oldestPositionSeq: number | null
 }

 type PendingVisibilityCacheEntry = {
@@ -138,6 +145,8 @@ function createState(sessionId: string): InternalState {
        pendingOverflowVisibleCount: 0,
        hasMore: false,
        oldestSeq: null,
+        oldestPositionAt: null,
+        oldestPositionSeq: null,
        newestSeq: null,
        isLoading: false,
        isLoadingMore: false,
@@ -214,6 +223,8 @@ function buildState(
        pendingVisibleCount?: number
        pendingOverflowVisibleCount?: number
        hasMore?: boolean
+        oldestPositionAt?: number | null
+        oldestPositionSeq?: number | null
        isLoading?: boolean
        isLoadingMore?: boolean
        warning?: string | null
@@ -245,6 +256,8 @@ function buildState(
        pendingOverflowVisibleCount,
        pendingCount,
        oldestSeq,
+        oldestPositionAt: updates.oldestPositionAt !== undefined ? updates.oldestPositionAt : prev.oldestPositionAt,
+        oldestPositionSeq: updates.oldestPositionSeq !== undefined ? updates.oldestPositionSeq : prev.oldestPositionSeq,
        newestSeq,
        hasMore: updates.hasMore !== undefined ? updates.hasMore : prev.hasMore,
        isLoading: updates.isLoading !== undefined ? updates.isLoading : prev.isLoading,
@@ -255,14 +268,44 @@ function buildState(
    }
 }

+/** Trim `messages` down to `limit` while preserving every queued user message.
+ *  Queued rows must survive trimming on both windows: the `messages-consumed`
+ *  SSE only carries localIds, so a dropped queued row cannot be restored or
+ *  repositioned without a full refetch.  Returns the kept slice plus the list
+ *  of regular (non-queued) rows that were dropped, so the pending-overflow
+ *  warning counter can be advanced symmetrically. */
+function trimPreservingQueued(
+    messages: DecryptedMessage[],
+    limit: number,
+    mode: 'append' | 'prepend'
+): { kept: DecryptedMessage[]; dropped: DecryptedMessage[] } {
+    if (messages.length <= limit) {
+        return { kept: messages, dropped: [] }
+    }
+    const queued = messages.filter(isQueuedForInvocation)
+    if (queued.length === 0) {
+        const kept = mode === 'prepend'
+            ? messages.slice(0, limit)
+            : messages.slice(messages.length - limit)
+        const dropped = mode === 'prepend'
+            ? messages.slice(limit)
+            : messages.slice(0, messages.length - limit)
+        return { kept, dropped }
+    }
+    const queuedIds = new Set(queued.map((message) => message.id))
+    const regular = messages.filter((message) => !queuedIds.has(message.id))
+    const budget = Math.max(0, limit - queued.length)
+    const trimmedRegular = mode === 'prepend'
+        ? regular.slice(0, budget)
+        : regular.slice(Math.max(0, regular.length - budget))
+    const droppedRegular = mode === 'prepend'
+        ? regular.slice(budget)
+        : regular.slice(0, Math.max(0, regular.length - budget))
+    return { kept: mergeMessages(trimmedRegular, queued), dropped: droppedRegular }
+}
+
 function trimVisible(messages: DecryptedMessage[], mode: 'append' | 'prepend'): DecryptedMessage[] {
-    if (messages.length <= VISIBLE_WINDOW_SIZE) {
-        return messages
-    }
-    if (mode === 'prepend') {
-        return messages.slice(0, VISIBLE_WINDOW_SIZE)
-    }
-    return messages.slice(messages.length - VISIBLE_WINDOW_SIZE)
+    return trimPreservingQueued(messages, VISIBLE_WINDOW_SIZE, mode).kept
 }

 function trimPending(
@@ -272,11 +315,12 @@ function trimPending(
    if (messages.length <= PENDING_WINDOW_SIZE) {
        return { pending: messages, dropped: 0, droppedVisible: 0 }
    }
-    const cutoff = messages.length - PENDING_WINDOW_SIZE
-    const droppedMessages = messages.slice(0, cutoff)
-    const pending = messages.slice(cutoff)
-    const droppedVisible = countVisiblePendingMessages(sessionId, droppedMessages)
-    return { pending, dropped: droppedMessages.length, droppedVisible }
+    // Symmetric with trimVisible: agents that overflow the pending window
+    // (200) must not evict queued user messages — the floating bar holds the
+    // only client-visible reference to them until the CLI ack arrives.
+    const { kept, dropped } = trimPreservingQueued(messages, PENDING_WINDOW_SIZE, 'append')
+    const droppedVisible = countVisiblePendingMessages(sessionId, dropped)
+    return { pending: kept, dropped: dropped.length, droppedVisible }
 }

 function filterPendingAgainstVisible(pending: DecryptedMessage[], visible: DecryptedMessage[]): DecryptedMessage[] {
@@ -360,6 +404,8 @@ export function seedMessageWindowFromSession(fromSessionId: string, toSessionId:
        pendingOverflowCount: source.pendingOverflowCount,
        pendingOverflowVisibleCount: source.pendingOverflowVisibleCount,
        hasMore: source.hasMore,
+        oldestPositionAt: source.oldestPositionAt,
+        oldestPositionSeq: source.oldestPositionSeq,
        warning: source.warning,
        atBottom: source.atBottom,
        isLoading: false,
@@ -376,7 +422,17 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr
    updateState(sessionId, (prev) => buildState(prev, { isLoading: true, warning: null }))

    try {
-        const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: null })
+        // Always request byPosition mode (V8). If the hub is V7 it ignores byPosition and
+        // returns the standard seq-based response (no nextBeforeAt field) — we fall back
+        // to seq-cursor mode seamlessly.
+        const response = await api.getMessages(sessionId, { byPosition: true, limit: PAGE_SIZE })
+        // Derive composite cursor pair from server response. Both values come from
+        // the same row on the server; we keep them paired so the next older fetch
+        // doesn't mix `beforeAt` from the server with a recomputed minimum `seq`.
+        const nextBeforeAt = response.page.nextBeforeAt ?? null
+        const nextBeforeSeq = response.page.nextBeforeSeq ?? null
+        const isV8Cursor = nextBeforeAt !== null && nextBeforeSeq !== null
+
        updateState(sessionId, (prev) => {
            if (prev.atBottom) {
                const merged = mergeMessages(prev.messages, [...prev.pending, ...response.messages])
@@ -388,6 +444,8 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr
                    pendingVisibleCount: 0,
                    pendingOverflowVisibleCount: 0,
                    hasMore: response.page.hasMore,
+                    oldestPositionAt: isV8Cursor ? nextBeforeAt : null,
+                    oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null,
                    isLoading: false,
                    warning: null,
                })
@@ -398,6 +456,12 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr
                pendingVisibleCount: pendingResult.pendingVisibleCount,
                pendingOverflowCount: pendingResult.pendingOverflowCount,
                pendingOverflowVisibleCount: pendingResult.pendingOverflowVisibleCount,
+                // Persist the V8 cursor pair on the non-at-bottom path too. Without this
+                // a refresh while scrolled up dropped the composite cursor and the next
+                // loadMore fell back to V7 seq mode against a V8 hub — the same
+                // asymmetric class of bug the at-bottom branch already guards against.
+                oldestPositionAt: isV8Cursor ? nextBeforeAt : null,
+                oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null,
                isLoading: false,
                warning: pendingResult.warning,
            })
@@ -419,13 +483,31 @@ export async function fetchOlderMessages(api: ApiClient, sessionId: string): Pro
    updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: true }))

    try {
-        const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: initial.oldestSeq })
+        // V8 mode: use the server-provided cursor pair as-is. Mixing `beforeAt` from
+        // the server with a recomputed minimum `seq` from the local window can refer
+        // to different rows after a low-seq message is invoked late.
+        const useV8Cursor = initial.oldestPositionAt !== null && initial.oldestPositionSeq !== null
+        const response = useV8Cursor
+            ? await api.getMessages(sessionId, {
+                byPosition: true,
+                beforeAt: initial.oldestPositionAt!,
+                beforeSeq: initial.oldestPositionSeq!,
+                limit: PAGE_SIZE
+            })
+            : await api.getMessages(sessionId, { beforeSeq: initial.oldestSeq, limit: PAGE_SIZE })
+
+        const nextBeforeAt = response.page.nextBeforeAt ?? null
+        const nextBeforeSeq = response.page.nextBeforeSeq ?? null
+        const isV8Cursor = nextBeforeAt !== null && nextBeforeSeq !== null
+
        updateState(sessionId, (prev) => {
            const merged = mergeMessages(response.messages, prev.messages)
            const trimmed = trimVisible(merged, 'prepend')
            return buildState(prev, {
                messages: trimmed,
                hasMore: response.page.hasMore,
+                oldestPositionAt: isV8Cursor ? nextBeforeAt : null,
+                oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null,
                isLoadingMore: false,
            })
        })
@@ -538,24 +620,78 @@ export function updateMessageStatus(sessionId: string, localId: string, status:
    })
 }

-/** Transition the queued messages whose localIds match to 'sent'. Driven by the
- *  CLI ack (messages-consumed). Unmatched messages remain queued. */
-export function markMessagesConsumed(sessionId: string, localIds: string[]): void {
+/** Transition the queued messages whose localIds match to 'sent' and record invokedAt.
+ *  Driven by the CLI ack (messages-consumed). Unmatched messages remain queued.
+ *  Also handles server-loaded messages (status=undefined) that have a matching localId.
+ *  V7 hub compat: if `invokedAt` is undefined the SyncEvent had no server timestamp,
+ *  so we fall back to client time — without it the row would stay queued forever
+ *  under the strict-null filter. The fallback only affects display ordering on
+ *  this client; the persisted server value is the authoritative one when present. */
+export function markMessagesConsumed(sessionId: string, localIds: string[], invokedAt: number | undefined): void {
    if (localIds.length === 0) return
    const idSet = new Set(localIds)
+    const effectiveInvokedAt = invokedAt ?? Date.now()
    updateState(sessionId, (prev) => {
        let changed = false
        const updateList = (list: DecryptedMessage[]) => {
            return list.map((message) => {
-                if (message.status !== 'queued' || !message.localId || !idSet.has(message.localId)) {
+                if (!message.localId || !idSet.has(message.localId)) {
+                    return message
+                }
+                if (message.status === 'failed') {
+                    return message
+                }
+                // Apply the ack even if the message is already 'sent' (optimistic) — otherwise
+                // a message that flipped to 'sent' before the consume event arrives would
+                // never receive `invokedAt` and keep sorting by send time.
+                // First-write-wins on `invokedAt`: mirror the hub's UPDATE guard so a
+                // duplicate `messages-consumed` (e.g. CLI re-emit) doesn't restamp a
+                // message and shuffle its byPosition slot on live clients while the
+                // DB still holds the original timestamp.
+                const needsStatus = message.status !== 'sent'
+                // Strict null to stay consistent with isQueuedForInvocation and the rest
+                // of this file. The idSet filter already shields V7-stamped rows from
+                // this path, but the strict-null contract should not vary by call site.
+                const needsInvokedAt = message.invokedAt === null
+                if (!needsStatus && !needsInvokedAt) {
                    return message
                }
                changed = true
-                return { ...message, status: 'sent' as MessageStatus }
+                const update: Partial<DecryptedMessage> = {}
+                if (needsStatus) {
+                    update.status = 'sent' as MessageStatus
+                }
+                if (needsInvokedAt) {
+                    update.invokedAt = effectiveInvokedAt
+                }
+                return { ...message, ...update }
            })
        }
-        const messages = updateList(prev.messages)
-        const pending = updateList(prev.pending)
+        // Migrate just-acked pending entries into the visible thread. Without
+        // this step, an at-bottom=false user that is stuck in pending never
+        // sees their own message at the invocation slot — it stays in the
+        // pending bucket until they scroll, even though the floating bar
+        // already cleared.  Identifying the migrated rows by (localId,
+        // invokedAt = effectiveInvokedAt) ensures we only move rows whose
+        // ack just arrived, not unrelated pending entries.
+        const updatedPending = updateList(prev.pending)
+        const consumedFromPending: DecryptedMessage[] = []
+        const remainingPending = updatedPending.filter((message) => {
+            if (
+                message.localId &&
+                idSet.has(message.localId) &&
+                message.invokedAt === effectiveInvokedAt
+            ) {
+                consumedFromPending.push(message)
+                return false
+            }
+            return true
+        })
+        // After update, re-merge to re-sort by the position key (`invokedAt ?? createdAt`):
+        // a queued message that just received `invokedAt` should move to its invocation
+        // position, not stay at its original send-time slot until the next fetch.
+        const messages = mergeMessages(updateList(prev.messages), consumedFromPending)
+        const pending = mergeMessages([], remainingPending)
        if (!changed) {
            return prev
        }
@@ -1,5 +1,4 @@
-import type { InfiniteData } from '@tanstack/react-query'
-import type { DecryptedMessage, MessagesResponse } from '@/types/api'
+import type { DecryptedMessage } from '@/types/api'
 import { randomId } from '@/lib/randomId'

 export function makeClientSideId(prefix: string): string {
@@ -14,21 +13,33 @@ export function isUserMessage(msg: DecryptedMessage): boolean {
    return false
 }

+/** A user message that is still waiting for the CLI ack (messages-consumed).
+ *  Strict null on `invokedAt` so a pre-V8 hub response that omits the field
+ *  (`undefined`) is treated as already-invoked; only optimistic / V8-loaded
+ *  rows that explicitly carry `invokedAt: null` are queued. `failed` rows are
+ *  not queued either — they're surfaced as send errors, not pending work. */
+export function isQueuedForInvocation(msg: DecryptedMessage): boolean {
+    return isUserMessage(msg) && msg.invokedAt === null && msg.status !== 'failed'
+}
+
 function isOptimisticMessage(msg: DecryptedMessage): boolean {
    return Boolean(msg.localId && msg.id === msg.localId)
 }

 function compareMessages(a: DecryptedMessage, b: DecryptedMessage): number {
+    const aTime = a.invokedAt ?? a.createdAt
+    const bTime = b.invokedAt ?? b.createdAt
+
+    if (aTime !== bTime) {
+        return aTime - bTime
+    }
+
    const aSeq = typeof a.seq === 'number' ? a.seq : null
    const bSeq = typeof b.seq === 'number' ? b.seq : null

    if (aSeq !== null && bSeq !== null && aSeq !== bSeq) {
        return aSeq - bSeq
    }
-
-    if (a.createdAt !== b.createdAt) {
-        return a.createdAt - b.createdAt
-    }
    return a.id.localeCompare(b.id)
 }

@@ -58,12 +69,18 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM
    }

    // If we received stored messages with a localId, drop any optimistic bubbles with the same localId.
-    // Preserve client-side status (e.g. 'queued') on the replacing server message.
+    // Preserve client-side status (e.g. 'queued') and invokedAt on the replacing server message.
    if (incomingStoredLocalIds.size > 0) {
        const optimisticStatusByLocalId = new Map<string, DecryptedMessage['status']>()
+        const optimisticInvokedAtByLocalId = new Map<string, number | null | undefined>()
        for (const msg of merged) {
-            if (msg.localId && isOptimisticMessage(msg) && incomingStoredLocalIds.has(msg.localId) && msg.status) {
-                optimisticStatusByLocalId.set(msg.localId, msg.status)
+            if (msg.localId && isOptimisticMessage(msg) && incomingStoredLocalIds.has(msg.localId)) {
+                if (msg.status) {
+                    optimisticStatusByLocalId.set(msg.localId, msg.status)
+                }
+                if (msg.invokedAt !== undefined) {
+                    optimisticInvokedAtByLocalId.set(msg.localId, msg.invokedAt)
+                }
            }
        }
        merged = merged.filter((msg) => {
@@ -72,10 +89,21 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM
            }
            return !isOptimisticMessage(msg)
        })
-        if (optimisticStatusByLocalId.size > 0) {
+        if (optimisticStatusByLocalId.size > 0 || optimisticInvokedAtByLocalId.size > 0) {
            merged = merged.map((msg) => {
-                if (msg.localId && optimisticStatusByLocalId.has(msg.localId) && !msg.status) {
-                    return { ...msg, status: optimisticStatusByLocalId.get(msg.localId) }
+                if (!msg.localId) return msg
+                const update: Partial<DecryptedMessage> = {}
+                if (optimisticStatusByLocalId.has(msg.localId) && !msg.status) {
+                    update.status = optimisticStatusByLocalId.get(msg.localId)
+                }
+                if (optimisticInvokedAtByLocalId.has(msg.localId) && msg.invokedAt == null) {
+                    const optimisticInvokedAt = optimisticInvokedAtByLocalId.get(msg.localId)
+                    if (optimisticInvokedAt != null) {
+                        update.invokedAt = optimisticInvokedAt
+                    }
+                }
+                if (Object.keys(update).length > 0) {
+                    return { ...msg, ...update }
                }
                return msg
            })
@@ -90,9 +118,14 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM

    for (const optimistic of optimisticMessages) {
        if (optimistic.status === 'sent') {
+            // Compare by the position key (invokedAt ?? createdAt). A late ack can
+            // attach `invokedAt` long after `createdAt`, so the optimistic copy and
+            // the server echo end up at the same byPosition slot — using
+            // `createdAt` alone misses that match and renders both as duplicates.
+            const optimisticTime = optimistic.invokedAt ?? optimistic.createdAt
            const hasServerUserMessage = nonOptimisticMessages.some((m) =>
                isUserMessage(m) &&
-                Math.abs(m.createdAt - optimistic.createdAt) < 10_000
+                Math.abs((m.invokedAt ?? m.createdAt) - optimisticTime) < 10_000
            )
            if (hasServerUserMessage) {
                continue
@@ -104,39 +137,3 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM
    result.sort(compareMessages)
    return result
 }
-
-export function upsertMessagesInCache(
-    data: InfiniteData<MessagesResponse> | undefined,
-    incoming: DecryptedMessage[],
-): InfiniteData<MessagesResponse> {
-    const mergedIncoming = mergeMessages([], incoming)
-
-    if (!data || data.pages.length === 0) {
-        return {
-            pages: [
-                {
-                    messages: mergedIncoming,
-                    page: {
-                        limit: 50,
-                        beforeSeq: null,
-                        nextBeforeSeq: null,
-                        hasMore: false,
-                    },
-                },
-            ],
-            pageParams: [null],
-        }
-    }
-
-    const pages = data.pages.slice()
-    const first = pages[0]
-    pages[0] = {
-        ...first,
-        messages: mergeMessages(first.messages, mergedIncoming),
-    }
-
-    return {
-        ...data,
-        pages,
-    }
-}
@@ -40,6 +40,7 @@ export type MessageStatus = 'queued' | 'sending' | 'sent' | 'failed'
 export type DecryptedMessage = ProtocolDecryptedMessage & {
    status?: MessageStatus
    originalText?: string
+    invokedAt?: number | null
 }

 export type RunnerState = {
@@ -87,8 +88,9 @@ export type MessagesResponse = {
    messages: DecryptedMessage[]
    page: {
        limit: number
-        beforeSeq: number | null
+        beforeSeq?: number | null
        nextBeforeSeq: number | null
+        nextBeforeAt?: number | null
        hasMore: boolean
    }
 }