|
|
@@ -81,6 +81,14 @@ const ACTIVITY_TIME_KEYS = ['last_active', 'last_active_at', 'last_activity_at',
|
|
|
|
|
|
function mergeSessionMetadata(previous: Record<string, unknown>, incoming: Record<string, unknown>): Record<string, unknown> {
|
|
|
const summary = { ...previous, ...incoming }
|
|
|
+ for (const flag of ['pinned', 'archived'] as const) {
|
|
|
+ const camel = `is${flag[0]!.toUpperCase()}${flag.slice(1)}`
|
|
|
+ const value = incoming[`is_${flag}`] ?? incoming[camel] ?? incoming[flag]
|
|
|
+ if (value === undefined) continue
|
|
|
+ summary[flag] = bool(value)
|
|
|
+ delete summary[`is_${flag}`]
|
|
|
+ delete summary[camel]
|
|
|
+ }
|
|
|
// A messages response can carry title/model metadata, but its envelope is
|
|
|
// not a replacement for the session detail.
|
|
|
delete summary.messages
|
|
|
@@ -377,7 +385,7 @@ export class ChatCacheStore {
|
|
|
|
|
|
applySync(owner: string, profile: string, sessionID: string, revision: number,
|
|
|
pages: Array<{ key: string; kind: string; response: UpstreamResponse; startedAt: number }>, force: boolean): boolean {
|
|
|
- if (this.revision(owner, profile, sessionID) !== revision) return false
|
|
|
+ if (!this.ownsSession(owner, profile, sessionID) || this.revision(owner, profile, sessionID) !== revision) return false
|
|
|
this.db.exec('SAVEPOINT history_sync')
|
|
|
try {
|
|
|
if (force) this.db.prepare('DELETE FROM chat_messages WHERE owner=? AND profile=? AND session_id=?').run(owner, profile, sessionID)
|
|
|
@@ -513,6 +521,23 @@ export class ChatCacheStore {
|
|
|
WHERE owner=? AND (session_id=? OR kind='list')`).run(owner, sessionID)
|
|
|
}
|
|
|
|
|
|
+ recordSessionPatch(owner: string, profile: string, sessionID: string, patch: Record<string, unknown>): void {
|
|
|
+ const current = this.localDetail(owner, profile, sessionID)
|
|
|
+ if (!current) return
|
|
|
+ const summary = mergeSessionMetadata(parseResponse(current.response) ?? {}, patch)
|
|
|
+ this.db.prepare(`UPDATE chat_sessions SET data=?,sync_revision=sync_revision+1
|
|
|
+ WHERE owner=? AND profile=? AND session_id=?`).run(JSON.stringify(summary), owner, profile, sessionID)
|
|
|
+ if (typeof patch.title === 'string') this.recordAuthoritativeTitle(owner, profile, sessionID, patch.title)
|
|
|
+ this.markListsStale(owner)
|
|
|
+ }
|
|
|
+
|
|
|
+ validationCandidates(owner: string, profile: string): Array<{ profile: string; session_id: string; sync_revision: number }> {
|
|
|
+ // Newly created runtime sessions may not have been persisted by Hermes yet.
|
|
|
+ return this.db.prepare(`SELECT profile,session_id,sync_revision FROM chat_sessions
|
|
|
+ WHERE owner=? AND owned_at IS NOT NULL AND owned_at<? AND (?='' OR profile=?)`)
|
|
|
+ .all(owner, Date.now() - 60_000, profile, profile) as Array<{ profile: string; session_id: string; sync_revision: number }>
|
|
|
+ }
|
|
|
+
|
|
|
deleteSession(owner: string, profile: string, sessionID: string): void {
|
|
|
const assets = this.db.prepare(`SELECT DISTINCT local_path FROM chat_attachments
|
|
|
WHERE owner=? AND profile=? AND session_id=? AND local_path IS NOT NULL`)
|
|
|
@@ -919,15 +944,62 @@ export class ChatCacheStore {
|
|
|
|
|
|
export class ChatCacheCoordinator {
|
|
|
private reconciling = new Map<string, Promise<void>>()
|
|
|
+ private validations = new Map<string, number>()
|
|
|
+ private validating = new Set<string>()
|
|
|
+ private stopped = false
|
|
|
onSynchronized: (owner: string, profile: string, sessionID: string) => void = () => {}
|
|
|
constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
|
|
|
readonly mode: ChatCacheMode = 'prefer-local') {}
|
|
|
|
|
|
+ close(): void { this.stopped = true }
|
|
|
+
|
|
|
+ private validateList(owner: string, profile: string, load: () => Promise<UpstreamResponse>): void {
|
|
|
+ const key = JSON.stringify([owner, profile])
|
|
|
+ if (this.stopped || this.validating.has(key) || Date.now() - (this.validations.get(key) ?? 0) < 30_000) return
|
|
|
+ const candidates = this.store.validationCandidates(owner, profile)
|
|
|
+ if (!candidates.length) return
|
|
|
+ this.validations.set(key, Date.now())
|
|
|
+ this.validating.add(key)
|
|
|
+ void (async () => {
|
|
|
+ const response = await load()
|
|
|
+ if (this.stopped) return
|
|
|
+ const payload = parseResponse(response)
|
|
|
+ const items = payload?.sessions ?? payload?.items
|
|
|
+ if (!Array.isArray(items)) return
|
|
|
+ const existing = new Set(items.flatMap(value => {
|
|
|
+ const session = object(value)
|
|
|
+ return session ? [`${profileOf(session, profile) || 'default'}\u0000${sessionIDOf(session)}`] : []
|
|
|
+ }))
|
|
|
+ // A filtered/paginated list cannot prove deletion. Confirm each absent
|
|
|
+ // candidate against its own profile; outages must retain local history.
|
|
|
+ for (const candidate of candidates) {
|
|
|
+ if (this.stopped) return
|
|
|
+ if (existing.has(`${candidate.profile}\u0000${candidate.session_id}`)) continue
|
|
|
+ try {
|
|
|
+ const detail = await this.upstream.request(`/api/sessions/${encodeURIComponent(candidate.session_id)}`, {
|
|
|
+ search: new URLSearchParams({ profile: candidate.profile }), cache: 'reload',
|
|
|
+ })
|
|
|
+ if (this.stopped) return
|
|
|
+ if (detail.status === 404 && this.store.ownsSession(owner, candidate.profile, candidate.session_id)
|
|
|
+ && this.store.revision(owner, candidate.profile, candidate.session_id) === candidate.sync_revision) {
|
|
|
+ this.store.deleteSession(owner, candidate.profile, candidate.session_id)
|
|
|
+ this.onSynchronized(owner, candidate.profile, candidate.session_id)
|
|
|
+ }
|
|
|
+ } catch { /* Retain the session and retry on a later list refresh. */ }
|
|
|
+ }
|
|
|
+ })().catch(() => {}).finally(() => this.validating.delete(key))
|
|
|
+ }
|
|
|
+
|
|
|
async read(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
|
|
|
load: () => Promise<UpstreamResponse>, ownedList = false,
|
|
|
listPage: { offset?: number; limit?: number; archived?: string } = {}): Promise<{ response: UpstreamResponse; source: ChatCacheSource; state: ChatCacheState }> {
|
|
|
+ if (kind === 'list' && ownedList) this.validateList(owner, profile, load)
|
|
|
if (this.mode === 'upstream-only') {
|
|
|
const upstreamResponse = await load()
|
|
|
+ if (sessionID && upstreamResponse.status === 404) {
|
|
|
+ this.store.deleteSession(owner, profile, sessionID)
|
|
|
+ this.onSynchronized(owner, profile, sessionID)
|
|
|
+ }
|
|
|
return {
|
|
|
response: kind === 'list' && ownedList
|
|
|
? this.store.mergeOwnedSessionsIntoList(owner, profile, upstreamResponse, listPage)
|
|
|
@@ -954,6 +1026,10 @@ export class ChatCacheCoordinator {
|
|
|
try {
|
|
|
const requestStartedAt = Date.now()
|
|
|
const upstreamResponse = await load()
|
|
|
+ if (sessionID && upstreamResponse.status === 404 && this.store.ownsSession(owner, profile, sessionID)) {
|
|
|
+ this.store.deleteSession(owner, profile, sessionID)
|
|
|
+ this.onSynchronized(owner, profile, sessionID)
|
|
|
+ }
|
|
|
if (this.mode === 'prefer-local' && local && upstreamResponse.status >= 500) {
|
|
|
return { response: projectedLocal!.response, source: 'local', state: 'stale' }
|
|
|
}
|
|
|
@@ -1023,6 +1099,11 @@ export class ChatCacheCoordinator {
|
|
|
const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
|
|
|
const detailStartedAt = Date.now()
|
|
|
const detail = await this.upstream.request(detailPath, { search: new URLSearchParams({ profile }), cache: 'reload' })
|
|
|
+ if (this.stopped) return
|
|
|
+ if (detail.status === 404 && this.store.revision(owner, profile, sessionID) === revision) {
|
|
|
+ this.store.deleteSession(owner, profile, sessionID)
|
|
|
+ this.onSynchronized(owner, profile, sessionID)
|
|
|
+ }
|
|
|
if (detail.status !== 200) throw new Error('Unable to synchronize conversation metadata')
|
|
|
const detailPayload = parseResponse(detail) ?? {}
|
|
|
const detailSummary = object(detailPayload.session) ?? detailPayload
|
|
|
@@ -1038,6 +1119,7 @@ export class ChatCacheCoordinator {
|
|
|
const search = new URLSearchParams({ profile, offset: String(offset), limit: '100', order: 'latest', include_compacted: 'true' })
|
|
|
const startedAt = Date.now()
|
|
|
const response = await this.upstream.request(`${detailPath}/messages`, { search, cache: 'reload' })
|
|
|
+ if (this.stopped) return
|
|
|
const payload = parseResponse(response)
|
|
|
if (response.status !== 200 || !Array.isArray(payload?.messages)) throw new Error('Unable to synchronize conversation messages')
|
|
|
const messages = payload.messages.flatMap(value => object(value) ? [object(value)!] : [])
|