Skip to content

release: merge develop → main (persist user state fix) #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
atomize merged 2 commits into from
Mar 29, 2026
Merged

release: merge develop → main (persist user state fix) #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6b82518
fix: persist alerts, analyses, and chat history across page refresh
atomize Mar 29, 2026
75a5730
Merge pull request #2 from atomize/fix/persist-user-state-on-refresh
atomize Mar 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 86 additions & 5 deletions packages/monitor/src/broadcaster.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,11 @@ import { config } from './config.js'
import { log } from './logger.js'
import { isEncryptionEnabled, encrypt, maskApiKey } from './crypto.js'
import { login, register, verifyToken } from './auth.js'
import { getAgentConfig, upsertAgentConfig, findUserById, upsertScheduleConfig, getScheduleConfig, getReportsForDate } from './db.js'
import {
getAgentConfig, upsertAgentConfig, findUserById, upsertScheduleConfig, getScheduleConfig, getReportsForDate,
appendAlertLogForAllUsers, getRecentAlerts, appendAnalysisLogForAllUsers, getRecentAnalyses,
getChatHistory,
} from './db.js'
import type { JwtPayload } from './auth.js'
import { handleOAuthRoute, getEnabledOAuthProviders } from './oauth.js'
import { getUserWatchlistsWithItems, saveWatchlist, removeWatchlistItem, syncFromTastytrade, searchSymbols } from './watchlistService.js'
Expand Down Expand Up @@ -54,7 +58,16 @@ export function onAlertForOrchestrator(handler: OrchestratorHandler): void {
export function broadcastToAll(msg: WsMessage | Record<string, unknown>): void {
const typed = msg as Record<string, unknown>
if (typed.type === 'agent_analysis' && typed.data) {
pushAnalysis(typed.data as AgentAnalysis)
const analysis = typed.data as AgentAnalysis
pushAnalysis(analysis)
if (isMultiTenant()) {
try {
const analysisId = (analysis as Record<string, unknown>).id as string ?? `analysis-${Date.now()}`
appendAnalysisLogForAllUsers(analysisId, JSON.stringify(analysis))
} catch (err) {
log.warn(`Failed to persist analysis: ${(err as Error).message}`)
}
}

@cursor cursor Bot Mar 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Alert/analysis persistence logic duplicated across four locations

Medium Severity

The same try/catch pattern for persisting alerts and analyses (check isMultiTenant(), extract ID with fallback, call appendAlertLogForAllUsers/appendAnalysisLogForAllUsers, catch and warn) is copy-pasted in four separate locations in broadcaster.ts: broadcastToAll, the onAlert callback, and the WS agent_analysis and alert route handlers. Any future change to the persistence logic (e.g., deduplication, retention) would require updating all four spots.

Additional Locations (2)
Fix in Cursor Fix in Web

}
if (typed.type === 'agent_status' && typed.data) {
agentStatus = typed.data as AgentStatus
Expand Down Expand Up @@ -189,6 +202,14 @@ export function startBroadcaster(): void {
onAlert((alert: OptionsAlert) => {
pushAlert(alert)
broadcast({ type: 'alert', data: alert })
if (isMultiTenant()) {
try {
const alertId = (alert as Record<string, unknown>).id as string ?? `alert-${Date.now()}`
appendAlertLogForAllUsers(alertId, JSON.stringify(alert))
} catch (err) {
log.warn(`Failed to persist alert: ${(err as Error).message}`)
}
}
if (orchestratorHandler) {
Promise.resolve(orchestratorHandler(alert)).catch(err => {
log.error('Orchestrator dispatch error:', err)
Expand Down Expand Up @@ -255,9 +276,52 @@ function send(ws: WebSocket, msg: WsMessage | Record<string, unknown>): void {
function sendFullState(ws: WebSocket): void {
send(ws, { type: 'snapshot', data: getAllSnapshots() })
send(ws, { type: 'account', data: getAccountContext() })
if (recentAlerts.length > 0) send(ws, { type: 'alert_history', data: recentAlerts })
if (recentAnalyses.length > 0) send(ws, { type: 'analysis_history', data: recentAnalyses })
if (agentStatus) send(ws, { type: 'agent_status', data: agentStatus })

if (isMultiTenant()) {
sendPerUserState(ws)
} else {
if (recentAlerts.length > 0) send(ws, { type: 'alert_history', data: recentAlerts })
if (recentAnalyses.length > 0) send(ws, { type: 'analysis_history', data: recentAnalyses })
}
}

function sendPerUserState(ws: WebSocket): void {
const auth = clientAuth.get(ws)
if (!auth) return

try {
const alertPayloads = getRecentAlerts(auth.userId)
if (alertPayloads.length > 0) {
const alerts = alertPayloads.map(p => JSON.parse(p))
send(ws, { type: 'alert_history', data: alerts })
}

const analysisPayloads = getRecentAnalyses(auth.userId)
if (analysisPayloads.length > 0) {
const analyses = analysisPayloads.map(p => JSON.parse(p))
send(ws, { type: 'analysis_history', data: analyses })
}

const chatRows = getChatHistory(auth.userId)
if (chatRows.length > 0) {
const messages = chatRows.map(r => ({
id: `db-${r.id}`,
role: r.role,
content: r.content,
timestamp: r.created_at,
}))
send(ws, { type: 'chat_history', data: messages })
}

handleRequestAgentConfig(ws)
handleRequestWatchlist(ws)
handleRequestScheduleConfig(ws)
handleRequestBudgetStatus(ws)
handleRequestReports(ws, undefined)
} catch (err) {
log.warn(`Failed to send per-user state for ${auth.email}: ${(err as Error).message}`)
}

@cursor cursor Bot Mar 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Broad try-catch causes total state loss on single failure

Medium Severity

The sendPerUserState function wraps all state-sending operations (alerts, analyses, chat, agent config, watchlists, schedule config, budget, reports) in a single try-catch. A JSON.parse failure on one corrupt alert or analysis payload will abort the entire function, silently preventing the user from receiving their chat history, watchlists, config, and reports on reconnect. This undermines the core goal of the PR — persisting state across page refresh — because one bad row in alerts_log can block all other state delivery.

Fix in Cursor Fix in Web

}

async function handleAuth(ws: WebSocket, msg: Record<string, unknown>): Promise<void> {
Expand Down Expand Up @@ -643,7 +707,16 @@ const authenticatedWsRoutes: Record<string, AuthenticatedWsRoute> = {
if (!data || typeof data !== 'object') return
const d = data as Record<string, unknown>
log.info(`Agent analysis received for ${d.ticker ?? 'unknown'} (model: ${d.model ?? 'unknown'})`)
pushAnalysis(data as unknown as AgentAnalysis)
const analysis = data as unknown as AgentAnalysis
pushAnalysis(analysis)
if (isMultiTenant()) {
try {
const analysisId = d.id as string ?? `analysis-${Date.now()}`
appendAnalysisLogForAllUsers(analysisId, JSON.stringify(analysis))
} catch (err) {
log.warn(`Failed to persist analysis: ${(err as Error).message}`)
}
}
broadcastRaw(raw)
},

Expand All @@ -654,6 +727,14 @@ const authenticatedWsRoutes: Record<string, AuthenticatedWsRoute> = {
log.info(`Injected test alert for ${payload.trigger?.ticker ?? 'unknown'}`)
pushAlert(payload)
broadcast({ type: 'alert', data: payload })
if (isMultiTenant()) {
try {
const alertId = (data as Record<string, unknown>).id as string ?? `alert-${Date.now()}`
appendAlertLogForAllUsers(alertId, JSON.stringify(payload))
} catch (err) {
log.warn(`Failed to persist alert: ${(err as Error).message}`)
}
}
if (orchestratorHandler) {
Promise.resolve(orchestratorHandler(payload)).catch(err => {
log.error('Orchestrator dispatch error:', err)
Expand Down
42 changes: 37 additions & 5 deletions packages/monitor/src/chatHandler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ import { getAllSnapshots } from './state.js'
import { getAccountContext } from './account.js'
import { getUserWatchlistsWithItems } from './watchlistService.js'
import { checkBudget, trackUsage } from './budgetTracker.js'
import { getAgentConfig } from './db.js'
import { decrypt } from './crypto.js'
import { getAgentConfig, appendChatMessage, getChatHistory as getChatHistoryDb, clearChatHistoryDb } from './db.js'
import { decrypt, isEncryptionEnabled } from './crypto.js'
import { log } from './logger.js'

const MAX_HISTORY = 20
Expand Down Expand Up @@ -60,11 +60,20 @@ export async function handleChatMessage(userId: number, message: string): Promis
}
}

const history = conversations.get(userId) ?? []
let history = conversations.get(userId)
if (!history && isEncryptionEnabled()) {
const rows = getChatHistoryDb(userId)
history = rows.map(r => ({ role: r.role, content: r.content }))
}
if (!history) history = []
history.push({ role: 'user', content: message })
if (history.length > MAX_HISTORY) history.splice(0, history.length - MAX_HISTORY)
conversations.set(userId, history)

if (isEncryptionEnabled()) {
try { appendChatMessage(userId, 'user', message) } catch { /* best-effort */ }
}

try {
const apiKey = decrypt(agentCfg.encrypted_api_key)
const systemPrompt = buildChatSystemPrompt(userId)
Expand All @@ -83,6 +92,10 @@ export async function handleChatMessage(userId: number, message: string): Promis
if (history.length > MAX_HISTORY) history.splice(0, history.length - MAX_HISTORY)
conversations.set(userId, history)

if (isEncryptionEnabled()) {
try { appendChatMessage(userId, 'assistant', result.text) } catch { /* best-effort */ }
}

log.info(`Chat: user ${userId} response (${result.outputTokens} tokens, $${costEstimate.toFixed(4)})`)

return {
Expand All @@ -95,10 +108,14 @@ export async function handleChatMessage(userId: number, message: string): Promis
} catch (err) {
const errMsg = (err as Error).message
log.error(`Chat failed for user ${userId}: ${errMsg}`)
const errorMsg = `Error: ${errMsg}`
if (isEncryptionEnabled()) {
try { appendChatMessage(userId, 'assistant', errorMsg) } catch { /* best-effort */ }
}
return {
id: randomUUID(),
role: 'assistant',
content: `Error: ${errMsg}`,
content: errorMsg,
timestamp: new Date().toISOString(),
costUsd: 0,
}
Expand All @@ -107,10 +124,25 @@ export async function handleChatMessage(userId: number, message: string): Promis

export function clearChatHistory(userId: number): void {
conversations.delete(userId)
if (isEncryptionEnabled()) {
try { clearChatHistoryDb(userId) } catch { /* best-effort */ }
}
log.info(`Chat: cleared history for user ${userId}`)
}

export function getChatHistory(userId: number): ChatMessage[] {
export function getChatHistoryForUser(userId: number): ChatMessage[] {
if (isEncryptionEnabled()) {
const rows = getChatHistoryDb(userId)
if (rows.length > 0 && !conversations.has(userId)) {
conversations.set(userId, rows.map(r => ({ role: r.role, content: r.content })))
}
return rows.map(r => ({
id: `db-${r.id}`,
role: r.role,
content: r.content,
timestamp: r.created_at,
}))
}

@cursor cursor Bot Mar 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exported getChatHistoryForUser is never called anywhere

Low Severity

getChatHistoryForUser (renamed from the previous getChatHistory) is exported but never imported or called anywhere in the codebase. The broadcaster.ts imports getChatHistory directly from db.ts instead and does its own formatting in sendPerUserState. This function also hydrates the in-memory conversations cache, which suggests it was intended to replace the direct DB call in the broadcaster — but the integration was missed, leaving this as dead code.

Fix in Cursor Fix in Web

const history = conversations.get(userId) ?? []
return history.map((h, i) => ({
id: `hist-${i}`,
Expand Down
107 changes: 107 additions & 0 deletions packages/monitor/src/db.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,39 @@ function migrateV2Tables(db: Database.Database): void {

CREATE INDEX IF NOT EXISTS idx_analysis_reports_user
ON analysis_reports(user_id, created_at);

CREATE TABLE IF NOT EXISTS chat_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
role TEXT NOT NULL CHECK(role IN ('user','assistant')),
content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_chat_history_user
ON chat_history(user_id, created_at);

CREATE TABLE IF NOT EXISTS alerts_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
alert_id TEXT NOT NULL,
payload TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_alerts_log_user
ON alerts_log(user_id, created_at);

CREATE TABLE IF NOT EXISTS analyses_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
analysis_id TEXT NOT NULL,
payload TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_analyses_log_user
ON analyses_log(user_id, created_at);
`)
}

Expand Down Expand Up @@ -463,3 +496,77 @@ export function getLatestReport(userId: number): AnalysisReportRow | undefined {
'SELECT * FROM analysis_reports WHERE user_id = ? ORDER BY created_at DESC LIMIT 1'
).get(userId) as AnalysisReportRow | undefined
}

// ─── Chat History ─────────────────────────────────────────────────

export interface ChatHistoryRow {
id: number
user_id: number
role: 'user' | 'assistant'
content: string
created_at: string
}

export function appendChatMessage(userId: number, role: 'user' | 'assistant', content: string): ChatHistoryRow {
return getDb().prepare(
"INSERT INTO chat_history (user_id, role, content) VALUES (?, ?, ?) RETURNING *"
).get(userId, role, content) as ChatHistoryRow
}

export function getChatHistory(userId: number, limit = 50): ChatHistoryRow[] {
return getDb().prepare(
'SELECT * FROM (SELECT * FROM chat_history WHERE user_id = ? ORDER BY created_at DESC LIMIT ?) ORDER BY created_at ASC'
).all(userId, limit) as ChatHistoryRow[]
}

export function clearChatHistoryDb(userId: number): void {
getDb().prepare('DELETE FROM chat_history WHERE user_id = ?').run(userId)
}

// ─── Alerts Log ───────────────────────────────────────────────────

export function appendAlertLog(userId: number, alertId: string, payload: string): void {
getDb().prepare(
"INSERT INTO alerts_log (user_id, alert_id, payload) VALUES (?, ?, ?)"
).run(userId, alertId, payload)

@cursor cursor Bot Mar 29, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exported appendAlertLog and appendAnalysisLog never called

Low Severity

appendAlertLog and appendAnalysisLog (per-user variants) are exported but never imported or called anywhere. All call sites use the ForAllUsers variants (appendAlertLogForAllUsers, appendAnalysisLogForAllUsers) instead. These are dead exports that add confusion about which function to use.

Additional Locations (1)
Fix in Cursor Fix in Web

}

export function getRecentAlerts(userId: number, limit = 200): string[] {
const rows = getDb().prepare(
'SELECT payload FROM alerts_log WHERE user_id = ? ORDER BY created_at DESC LIMIT ?'
).all(userId, limit) as { payload: string }[]
return rows.reverse().map(r => r.payload)
}

export function appendAlertLogForAllUsers(alertId: string, payload: string): void {
const users = getDb().prepare('SELECT id FROM users').all() as { id: number }[]
const stmt = getDb().prepare("INSERT INTO alerts_log (user_id, alert_id, payload) VALUES (?, ?, ?)")
const tx = getDb().transaction(() => {
for (const u of users) stmt.run(u.id, alertId, payload)
})
tx()
}

// ─── Analyses Log ─────────────────────────────────────────────────

export function appendAnalysisLog(userId: number, analysisId: string, payload: string): void {
getDb().prepare(
"INSERT INTO analyses_log (user_id, analysis_id, payload) VALUES (?, ?, ?)"
).run(userId, analysisId, payload)
}

export function getRecentAnalyses(userId: number, limit = 50): string[] {
const rows = getDb().prepare(
'SELECT payload FROM analyses_log WHERE user_id = ? ORDER BY created_at DESC LIMIT ?'
).all(userId, limit) as { payload: string }[]
return rows.reverse().map(r => r.payload)
}

export function appendAnalysisLogForAllUsers(analysisId: string, payload: string): void {
const users = getDb().prepare('SELECT id FROM users').all() as { id: number }[]
const stmt = getDb().prepare("INSERT INTO analyses_log (user_id, analysis_id, payload) VALUES (?, ?, ?)")
const tx = getDb().transaction(() => {
for (const u of users) stmt.run(u.id, analysisId, payload)
})
tx()
}
Loading