diff --git a/app/ui_layer/adapters/browser_adapter.py b/app/ui_layer/adapters/browser_adapter.py index ab357d7f..8b290f1d 100644 --- a/app/ui_layer/adapters/browser_adapter.py +++ b/app/ui_layer/adapters/browser_adapter.py @@ -1611,6 +1611,9 @@ async def _handle_ws_message(self, data: Dict[str, Any], ws=None) -> None: elif msg_type == "session_clear": await self._handle_session_clear(data) + elif msg_type == "session_seen": + await self._handle_session_seen(data) + elif msg_type == "generative_ui_close": await self._handle_generative_ui_close(data) @@ -4534,6 +4537,26 @@ async def _handle_generative_ui_close(self, data: Dict[str, Any]) -> None: } ) + async def _handle_session_seen(self, data: Dict[str, Any]) -> None: + """Acknowledge read status only after its SQLite transaction commits.""" + session_id = data.get("sessionId") + message_id = data.get("messageId") + if not isinstance(session_id, str) or not isinstance(message_id, str): + return + storage = self._chat._storage + if storage is None: + return + seen = await self._chat.run_storage( + storage.mark_session_seen, session_id, message_id + ) + if seen is not None: + await self._broadcast( + { + "type": "session_seen", + "data": {"sessionId": session_id, "messageId": seen}, + } + ) + async def _handle_session_list(self, ws=None) -> None: """Send the current session list.""" message = { @@ -9997,6 +10020,11 @@ def _get_initial_state(self) -> Dict[str, Any]: # older history through chat_history. ChatMessage.to_dict() always # carries sessionId. "messages": [m.to_dict() for m in self._chat.get_recent_messages()], + "lastSeenMessageIds": ( + self._chat._storage.get_last_seen_message_ids() + if self._chat._storage is not None + else {} + ), # Recent activity items (per-session inline feed); each carries sessionId. "actions": [ BrowserActionPanelComponent._item_payload(a) diff --git a/app/ui_layer/adapters/ws_lanes.py b/app/ui_layer/adapters/ws_lanes.py index 23fb6c1e..e491d5d5 100644 --- a/app/ui_layer/adapters/ws_lanes.py +++ b/app/ui_layer/adapters/ws_lanes.py @@ -26,6 +26,7 @@ "session_delete", "session_rename", "session_clear", + "session_seen", "chat_history", "chat_attachment_upload", "option_click", diff --git a/app/ui_layer/browser/frontend/src/components/layout/NavBar.tsx b/app/ui_layer/browser/frontend/src/components/layout/NavBar.tsx index 14d4b66f..b6950b97 100644 --- a/app/ui_layer/browser/frontend/src/components/layout/NavBar.tsx +++ b/app/ui_layer/browser/frontend/src/components/layout/NavBar.tsx @@ -184,10 +184,8 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { renameSession, clearSession, } = useWebSocket() - // Both lists follow the user's drag-and-drop order. Items created since the - // last reorder land where they always have: new apps last, new chats first. + // Apps retain their saved drag-and-drop order; chats follow latest activity. const [agentAppOrder, setAgentAppOrder] = usePersistedState(UI_STATE.nav.agentAppOrder) - const [chatOrder, setChatOrder] = usePersistedState(UI_STATE.nav.chatOrder) const unorderedAgentAppProjects = useAppSelector(selectAgentAppProjects) const agentAppProjects = useMemo( () => applySavedOrder(unorderedAgentAppProjects, agentAppOrder, p => p.id, 'last'), @@ -204,11 +202,7 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { const [agentDelete, setAgentDelete] = useState<{ id: string; name: string } | null>(null) const mainSession = useAppSelector(selectMainSession) - const unorderedChatSessions = useAppSelector(selectChatSessions) - const chatSessions = useMemo( - () => applySavedOrder(unorderedChatSessions, chatOrder, s => s.id, 'first'), - [unorderedChatSessions, chatOrder], - ) + const chatSessions = useAppSelector(selectChatSessions) const lastMessageIdBySession = useAppSelector(selectLastMessageIdBySession) const runStateBySession = useAppSelector(state => state.agent.runStateBySession) @@ -296,8 +290,8 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { : agentAppProjects.slice(0, GROUP_PREVIEW_COUNT) const visibleChatSessions = chatSessions.slice(0, chatVisibleCount) - // Drag-to-reorder within each group. The rendered rows are always a prefix - // of the full list ("Show more" / scroll pagination), so the reordered + // Drag-to-reorder apps. The rendered rows are always a prefix + // of the full list ("Show more"), so the reordered // prefix followed by the untouched tail is the new full order. const withTail = (prefix: string[], all: string[]) => { const moved = new Set(prefix) @@ -308,11 +302,6 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { onReorder: ids => setAgentAppOrder(withTail(ids, agentAppProjects.map(p => p.id))), scrollRef, }) - const chatSortable = useSortableList({ - ids: visibleChatSessions.map(s => s.id), - onReorder: ids => setChatOrder(withTail(ids, chatSessions.map(s => s.id))), - scrollRef, - }) const isActive = (path: string) => { if (path === '/') { @@ -567,16 +556,10 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { const path = sessionPath(session.id) const active = isActive(path) const renaming = renamingId === session.id - // Main stays pinned first; every other chat can be dragged. - const sortable = opts.isMain ? null : chatSortable - const dragging = chatSortable.draggingId === session.id - return (
@@ -598,7 +581,6 @@ export function NavBar({ collapsed = false, onToggleCollapsed }: NavBarProps) { className={styles.sessionRowButton} onClick={() => navigate(path)} title={opts.isMain ? t('nav:items.main') : session.title} - {...sortable?.handleProps(session.id)} > {opts.isMain ? : } diff --git a/app/ui_layer/browser/frontend/src/contexts/WebSocketContext.tsx b/app/ui_layer/browser/frontend/src/contexts/WebSocketContext.tsx index 7dbd7684..264344f5 100644 --- a/app/ui_layer/browser/frontend/src/contexts/WebSocketContext.tsx +++ b/app/ui_layer/browser/frontend/src/contexts/WebSocketContext.tsx @@ -438,8 +438,8 @@ export function WebSocketProvider({ children }: { children: ReactNode }) { })) }, [dispatch]) - // Mark a session's newest message as seen (unread-dot bookkeeping). Stored - // as persisted UI state, so it survives reloads and syncs across tabs. + // Update the dot immediately, then commit the marker on the backend so + // a browser crash cannot undo it before Chrome flushes localStorage. const markSessionSeen = useCallback((sessionId: string) => { const current = store.getState() const lastId = selectLastMessageIdBySession(current)[sessionId] @@ -447,6 +447,7 @@ export function WebSocketProvider({ children }: { children: ReactNode }) { const seen = selectUiState(current, UI_STATE.chat.lastSeenMessageIds) if (seen[sessionId] === lastId) return dispatch(setUiState(UI_STATE.chat.lastSeenMessageIds, { ...seen, [sessionId]: lastId })) + client.sendString(JSON.stringify({ type: 'session_seen', sessionId, messageId: lastId })) }, [dispatch, store]) const enhancePrompt = useCallback((content: string, sessionId: string) => { diff --git a/app/ui_layer/browser/frontend/src/store/index.ts b/app/ui_layer/browser/frontend/src/store/index.ts index 47d244a4..121c052e 100644 --- a/app/ui_layer/browser/frontend/src/store/index.ts +++ b/app/ui_layer/browser/frontend/src/store/index.ts @@ -25,6 +25,7 @@ import miniBrowserReducer from './slices/miniBrowserSlice' import uiReducer from './slices/uiSlice' import { socketMiddleware } from './socket/socketMiddleware' import './socket/versionWatch' +import './socket/readReceipts' import { createUiPersistenceMiddleware, loadPersistedUiState } from './uiState' // Redux Toolkit's development-only safety checks walk the state on every diff --git a/app/ui_layer/browser/frontend/src/store/selectors/sessions.ts b/app/ui_layer/browser/frontend/src/store/selectors/sessions.ts index 0077719c..947aaccd 100644 --- a/app/ui_layer/browser/frontend/src/store/selectors/sessions.ts +++ b/app/ui_layer/browser/frontend/src/store/selectors/sessions.ts @@ -10,15 +10,19 @@ export const selectMainSession = createSelector( (sessions): SessionInfo | undefined => sessions.find(s => s.type === 'main'), ) -// Plain chat sessions, newest first (createdAt is an ISO string, so a -// lexicographic descending sort is chronological). +// Plain chats follow their latest activity, including after reconnecting. +// Older session records can fall back to their creation timestamp. export const selectChatSessions = createSelector( selectSessions, (sessions): SessionInfo[] => sessions .filter(s => s.type === 'chat') .slice() - .sort((a, b) => (a.createdAt < b.createdAt ? 1 : a.createdAt > b.createdAt ? -1 : 0)), + .sort((a, b) => { + const aUpdated = a.lastActiveAt || a.createdAt + const bUpdated = b.lastActiveAt || b.createdAt + return aUpdated < bUpdated ? 1 : aUpdated > bUpdated ? -1 : 0 + }), ) export const selectSessionById = (state: RootState, sessionId: string): SessionInfo | undefined => diff --git a/app/ui_layer/browser/frontend/src/store/socket/readReceipts.ts b/app/ui_layer/browser/frontend/src/store/socket/readReceipts.ts new file mode 100644 index 00000000..7b21acc3 --- /dev/null +++ b/app/ui_layer/browser/frontend/src/store/socket/readReceipts.ts @@ -0,0 +1,32 @@ +import { register } from './messageRegistry' +import { getSocketClient } from './socketInstance' +import { setUiState } from '../slices/uiSlice' +import { selectUiState } from '../selectors/ui' +import { UI_STATE } from '../uiState/catalog' + +// Read receipts are durable on the backend. The local preference remains a +// cache and a migration source for messages read before this feature existed. +register('init', (data, dispatch, getState) => { + const durable = (data as { lastSeenMessageIds?: Record } | undefined)?.lastSeenMessageIds + if (!durable) return // Older backends still use browser-only read status. + const local = selectUiState(getState(), UI_STATE.chat.lastSeenMessageIds) + dispatch(setUiState(UI_STATE.chat.lastSeenMessageIds, { ...local, ...durable })) + // Import legacy/local receipts; the server validates ownership and only + // advances a marker, so a stale tab can never overwrite a newer receipt. + for (const [sessionId, messageId] of Object.entries(local)) { + if (messageId && messageId !== durable[sessionId]) { + getSocketClient().sendString(JSON.stringify({ type: 'session_seen', sessionId, messageId })) + } + } +}) + +register('session_seen', (data, dispatch, getState) => { + const receipt = data as { sessionId?: string; messageId?: string } | undefined + if (!receipt?.sessionId || !receipt.messageId) return + const seen = selectUiState(getState(), UI_STATE.chat.lastSeenMessageIds) + if (seen[receipt.sessionId] !== receipt.messageId) { + dispatch(setUiState(UI_STATE.chat.lastSeenMessageIds, { + ...seen, [receipt.sessionId]: receipt.messageId, + })) + } +}) diff --git a/app/ui_layer/browser/frontend/src/store/uiState/catalog.ts b/app/ui_layer/browser/frontend/src/store/uiState/catalog.ts index 97ff1829..24b936c9 100644 --- a/app/ui_layer/browser/frontend/src/store/uiState/catalog.ts +++ b/app/ui_layer/browser/frontend/src/store/uiState/catalog.ts @@ -58,8 +58,6 @@ export const UI_STATE = { agentAppExpanded: defineUiState('nav.agentAppExpanded', true, 'preference'), /** Agent App ids in drag-and-drop order; apps created since go last. */ agentAppOrder: defineUiState('nav.agentAppOrder', [], 'preference'), - /** Chat session ids in drag-and-drop order; chats created since go first. */ - chatOrder: defineUiState('nav.chatOrder', [], 'preference'), }, tour: { diff --git a/app/ui_layer/browser/frontend/src/store/uiState/persistence.ts b/app/ui_layer/browser/frontend/src/store/uiState/persistence.ts index baa32276..a2c9afcf 100644 --- a/app/ui_layer/browser/frontend/src/store/uiState/persistence.ts +++ b/app/ui_layer/browser/frontend/src/store/uiState/persistence.ts @@ -86,6 +86,9 @@ export function createUiPersistenceMiddleware(storages: UiStorages = uiStorages) if (uiStateChanged.match(action)) { const { key, lifetime, value } = action.payload persister.enqueue(key, lifetime, value) + // Read markers are discrete acknowledgements, not continuous UI + // changes. Do not leave them waiting in the coalescing timer. + if (key === 'chat.lastSeenMessageIds') persister.flush() } return result } diff --git a/app/ui_layer/browser/frontend/src/types/index.ts b/app/ui_layer/browser/frontend/src/types/index.ts index 52c9f2ad..26a4c0d0 100644 --- a/app/ui_layer/browser/frontend/src/types/index.ts +++ b/app/ui_layer/browser/frontend/src/types/index.ts @@ -177,6 +177,7 @@ export type WSMessageType = | 'session_updated' | 'session_deleted' | 'session_cleared' + | 'session_seen' | 'generative_ui_close' | 'generative_ui_closed' | 'session_busy' diff --git a/app/usage/chat_storage.py b/app/usage/chat_storage.py index e8efb9bb..903c75a8 100644 --- a/app/usage/chat_storage.py +++ b/app/usage/chat_storage.py @@ -209,7 +209,65 @@ def _init_db(self) -> None: ON chat_messages(session_id, timestamp) """) + # Read receipts must outlive a browser process crash. Keep them + # beside the messages rather than relying on browser disk flushes. + cursor.execute(""" + CREATE TABLE IF NOT EXISTS chat_session_reads ( + session_id TEXT PRIMARY KEY, + message_id TEXT NOT NULL, + message_timestamp REAL NOT NULL, + message_row_id INTEGER NOT NULL + ) + """) + + conn.commit() + + def get_last_seen_message_ids(self) -> Dict[str, str]: + """Durable read markers for this single-user CraftBot instance.""" + with sqlite3.connect(self._db_path) as conn: + return dict( + conn.execute(""" + SELECT r.session_id, r.message_id FROM chat_session_reads r + JOIN chat_messages m ON m.message_id = r.message_id + AND m.session_id = r.session_id + """) + ) + + def mark_session_seen(self, session_id: str, message_id: str) -> Optional[str]: + """Commit a read receipt, ignoring unknown IDs and older receipts. + + Return the committed marker so delayed requests from another window + cannot move the browser's read position backwards. + """ + with sqlite3.connect(self._db_path) as conn: + message = conn.execute( + "SELECT timestamp, id FROM chat_messages " + "WHERE session_id = ? AND message_id = ?", + (session_id, message_id), + ).fetchone() + if message is None: + return None + conn.execute( + """ + INSERT INTO chat_session_reads + (session_id, message_id, message_timestamp, message_row_id) + VALUES (?, ?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + message_id = excluded.message_id, + message_timestamp = excluded.message_timestamp, + message_row_id = excluded.message_row_id + WHERE excluded.message_timestamp > chat_session_reads.message_timestamp + OR (excluded.message_timestamp = chat_session_reads.message_timestamp + AND excluded.message_row_id > chat_session_reads.message_row_id) + """, + (session_id, message_id, *message), + ) + seen = conn.execute( + "SELECT message_id FROM chat_session_reads WHERE session_id = ?", + (session_id,), + ).fetchone()[0] conn.commit() + return seen def insert_message(self, message: StoredChatMessage) -> int: """ @@ -385,10 +443,14 @@ def clear_messages(self, session_id: Optional[str] = None) -> int: cursor.execute( "DELETE FROM chat_messages WHERE session_id = ?", (session_id,) ) + cursor.execute( + "DELETE FROM chat_session_reads WHERE session_id = ?", (session_id,) + ) else: cursor.execute("SELECT COUNT(*) FROM chat_messages") count = cursor.fetchone()[0] cursor.execute("DELETE FROM chat_messages") + cursor.execute("DELETE FROM chat_session_reads") conn.commit() return count diff --git a/tests/test_chat_read_receipts.py b/tests/test_chat_read_receipts.py new file mode 100644 index 00000000..ca25ebb6 --- /dev/null +++ b/tests/test_chat_read_receipts.py @@ -0,0 +1,93 @@ +"""Read receipts survive browser/backend crashes and never move backwards.""" + +import asyncio +from concurrent.futures import ThreadPoolExecutor + +from app.usage.chat_storage import ChatStorage, StoredChatMessage + + +def _message(mid, sid="a", ts=1.0): + return StoredChatMessage(mid, "agent", "reply", "agent", ts, session_id=sid) + + +def test_receipt_survives_reopen_without_marking_new_reply_read(tmp_path): + path = str(tmp_path / "chat.db") + storage = ChatStorage(path) + storage.insert_message(_message("a1")) + assert storage.mark_session_seen("a", "a1") == "a1" + storage.insert_message(_message("a2", ts=2.0)) + + restored = ChatStorage(path) + assert restored.get_last_seen_message_ids() == {"a": "a1"} + assert restored.get_recent_messages(session_id="a")[-1].message_id == "a2" + + +def test_delayed_or_duplicate_receipts_cannot_regress(tmp_path): + storage = ChatStorage(str(tmp_path / "chat.db")) + for mid, ts in (("a1", 1), ("a2", 2), ("a3", 2)): + storage.insert_message(_message(mid, ts=ts)) + assert storage.mark_session_seen("a", "a3") == "a3" + assert storage.mark_session_seen("a", "a1") == "a3" + assert storage.mark_session_seen("a", "a2") == "a3" + assert storage.mark_session_seen("a", "a3") == "a3" + + +def test_unknown_and_cross_session_ids_cannot_clear_an_unread_dot(tmp_path): + storage = ChatStorage(str(tmp_path / "chat.db")) + storage.insert_message(_message("b1", sid="b")) + assert storage.mark_session_seen("a", "missing") is None + assert storage.mark_session_seen("a", "b1") is None + assert storage.get_last_seen_message_ids() == {} + + +def test_clear_removes_receipts_for_only_the_cleared_session(tmp_path): + storage = ChatStorage(str(tmp_path / "chat.db")) + for sid in ("a", "b"): + storage.insert_message(_message(sid + "1", sid=sid)) + storage.mark_session_seen(sid, sid + "1") + storage.clear_messages("a") + assert storage.get_last_seen_message_ids() == {"b": "b1"} + storage.clear_messages() + assert storage.get_last_seen_message_ids() == {} + + +def test_socket_ack_is_broadcast_after_commit_and_contains_durable_marker(tmp_path): + from app.ui_layer.adapters.browser_adapter import ( + BrowserAdapter, + BrowserChatComponent, + ) + + path = str(tmp_path / "chat.db") + storage = ChatStorage(path) + storage.insert_message(_message("a1")) + storage.insert_message(_message("a2", ts=2)) + adapter = BrowserAdapter.__new__(BrowserAdapter) + worker = ThreadPoolExecutor(max_workers=1) + chat = BrowserChatComponent.__new__(BrowserChatComponent) + chat._db = worker + chat._storage = storage + adapter._chat = chat + recorded = [] + + async def broadcast(payload): + # Read through a separate SQLite connection at acknowledgement time. + assert ChatStorage(path).get_last_seen_message_ids() == {"a": "a2"} + recorded.append(payload) + + adapter._broadcast = broadcast + + async def run(): + await adapter._handle_ws_message( + {"type": "session_seen", "sessionId": "a", "messageId": "a2"} + ) + await adapter._handle_session_seen({"sessionId": "a", "messageId": "a1"}) + await adapter._handle_session_seen({"sessionId": [], "messageId": "a2"}) + + try: + asyncio.run(run()) + finally: + worker.shutdown(wait=True) + assert ( + recorded + == [{"type": "session_seen", "data": {"sessionId": "a", "messageId": "a2"}}] * 2 + )