From d64cda2913c3c2868f78e40d3dde5e3ad03bddb1 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Wed, 5 Aug 2026 21:54:27 -0700 Subject: [PATCH] feat(cloud): retract the stale cloud row of a superseded continuation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The continuation election demotes a compacted conversation's old sibling out of the roster, but its Team Sessions row lingered until the retention window aged it out — teammates kept seeing two rows for one conversation while the owner's sidebar showed one. The vanished sweep now runs a second reconcile: a push-marked id that left the roster because the imported cache reports it SUPERSEDED (row present, strictly newer sibling exists — never inferred from absence) is retracted only when the family's listable winner is itself replay-pushed to the same org, under the same two-strike deferral as the vanished path. Deliberate content tradeoff, stated for review: the demoted row is the only cloud replay of the pre-compact detail; the winner carries the compacted continuation. The source transcript stays on the owner's disk and can be re-shared at any time. A failed status lookup reads as unknown, never superseded, and ids absent from the cache stay on the vanished path's evidence rules. Pre-commit hook ran. Total eslint: 18, total circular: 0 --- .../src/sources/imported_history/cache.rs | 23 ++++ src-tauri/src/commands/handler_list.inc | 1 + src-tauri/src/orgtrack/history_commands.rs | 45 ++++++++ .../externalHistory/imported/cloudReplay.ts | 22 ++++ .../Org2Cloud/org2CloudSessionSync.state.ts | 7 ++ src/features/Org2Cloud/org2CloudSyncEngine.ts | 85 +++++++++++++- .../org2CloudSyncEngine.vanishedSessions.ts | 57 +++++++++ .../org2CloudSyncEngine.vanishedSweep.test.ts | 109 +++++++++++++++++- 8 files changed, 342 insertions(+), 7 deletions(-) diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/cache.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/cache.rs index 177c48a79..81ba17fb7 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/cache.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/cache.rs @@ -832,6 +832,29 @@ pub fn query_cached_session_by_session_id_including_superseded_from_conn( query_cached_session_by_session_id_impl(conn, session_id, true) } +/** + * Continuation-family status for one cached session id: its elected lineage + * (when stamped) and whether a strictly newer continuation sibling exists. + * `None` = the id is not in the imported cache at all — callers must treat + * that as "unknown", never as superseded (a rebuilding cache reads absent). + */ +pub fn cached_session_continuation_status_from_conn( + conn: &Connection, + session_id: &str, +) -> Result, bool)>, String> { + let Some((source, session)) = + query_cached_session_by_session_id_including_superseded_from_conn(conn, session_id)? + else { + return Ok(None); + }; + let lineage = session + .source_metadata_json + .as_deref() + .and_then(continuation_lineage_id_from_metadata_json); + let superseded = has_newer_continuation_sibling(conn, &source, &session)?; + Ok(Some((lineage, superseded))) +} + fn query_cached_session_by_session_id_impl( conn: &Connection, session_id: &str, diff --git a/src-tauri/src/commands/handler_list.inc b/src-tauri/src/commands/handler_list.inc index 4a62b153c..8b0c7e2c7 100644 --- a/src-tauri/src/commands/handler_list.inc +++ b/src-tauri/src/commands/handler_list.inc @@ -1036,6 +1036,7 @@ orgtrack::history_commands::imported_history_initial_window, orgtrack::history_commands::imported_history_turn_windows, orgtrack::history_commands::imported_history_cloud_turn_ids, orgtrack::history_commands::imported_history_cloud_turn_windows, +orgtrack::history_commands::imported_history_continuation_statuses, orgtrack::history_commands::codex_app_chunks, orgtrack::history_commands::codex_app_initial_window, orgtrack::history_commands::codex_app_turn_window, diff --git a/src-tauri/src/orgtrack/history_commands.rs b/src-tauri/src/orgtrack/history_commands.rs index e877cc183..4f123f098 100644 --- a/src-tauri/src/orgtrack/history_commands.rs +++ b/src-tauri/src/orgtrack/history_commands.rs @@ -957,6 +957,51 @@ pub async fn imported_history_cloud_turn_windows( .map_err(|err| format!("Task join error: {err}"))? } +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ImportedContinuationStatus { + pub session_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub lineage_id: Option, + pub superseded: bool, +} + +/// Continuation-family status for the cloud engine's superseded-row +/// reconciliation: which push-marked sessions the imported cache reports as +/// demoted, plus the lineage that identifies their listable winner. Ids not +/// present in the cache are OMITTED — absence means "unknown" (a rebuilding +/// cache reads empty), never "superseded". +#[tauri::command] +pub async fn imported_history_continuation_statuses( + session_ids: Vec, +) -> Result, String> { + if session_ids.len() > 200 { + return Err("At most 200 continuation statuses can be resolved at once".to_string()); + } + tokio::task::spawn_blocking(move || { + let conn = open_cache_conn()?; + let mut out = Vec::with_capacity(session_ids.len()); + for session_id in session_ids { + let Some((lineage_id, superseded)) = + orgtrack_core::sources::imported_history::cache::cached_session_continuation_status_from_conn( + &conn, + &session_id, + )? + else { + continue; + }; + out.push(ImportedContinuationStatus { + session_id, + lineage_id, + superseded, + }); + } + Ok(out) + }) + .await + .map_err(|err| format!("Task join error: {err}"))? +} + #[tauri::command] pub async fn codex_app_chunks( session_id: String, diff --git a/src/api/tauri/externalHistory/imported/cloudReplay.ts b/src/api/tauri/externalHistory/imported/cloudReplay.ts index 641dc64d7..ea4d01513 100644 --- a/src/api/tauri/externalHistory/imported/cloudReplay.ts +++ b/src/api/tauri/externalHistory/imported/cloudReplay.ts @@ -23,3 +23,25 @@ export async function importedHistoryCloudTurnWindows(args: { args ); } + +export interface ImportedContinuationStatus { + sessionId: string; + /** Elected continuation-family id; absent on pre-lineage cache rows. */ + lineageId?: string; + /** True when a strictly newer continuation sibling exists in the cache. */ + superseded: boolean; +} + +/** + * Continuation-family status for push-marked session ids. Ids not present + * in the imported cache are omitted — absence is "unknown" (a rebuilding + * cache reads empty), never "superseded". + */ +export async function importedHistoryContinuationStatuses( + sessionIds: string[] +): Promise { + return invoke( + "imported_history_continuation_statuses", + { sessionIds } + ); +} diff --git a/src/features/Org2Cloud/org2CloudSessionSync.state.ts b/src/features/Org2Cloud/org2CloudSessionSync.state.ts index 4e5c0a7fd..75964e5ff 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.state.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.state.ts @@ -223,6 +223,13 @@ export class Org2CloudSessionSyncState { })); } + /** True when this device holds a replay cursor covering pushed events — + * the winner-side guard for superseded-continuation retraction. */ + hasReplayPushed(orgId: string, sessionId: string): boolean { + const cursor = this.getCursor(orgId, sessionId); + return Boolean(cursor && cursor.pushedCount > 0); + } + protected async computeFrozenChainHash( perEventHashes: string[], frozenEventCount: number diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.ts b/src/features/Org2Cloud/org2CloudSyncEngine.ts index cf5401a7d..507bf9753 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.ts @@ -135,8 +135,11 @@ import { } from "./org2CloudSyncEngine.schemaGate"; import { Org2CloudSessionColdStart } from "./org2CloudSyncEngine.sessionColdStart"; import { + type ContinuationStatusResolver, type LocalSessionIdResolver, + findSupersededPushedSessions, findVanishedPushedSessionIds, + resolveContinuationStatusesViaCache, resolveLocalSessionIdsViaAggregateList, } from "./org2CloudSyncEngine.vanishedSessions"; import { @@ -196,19 +199,24 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle { /** Confirms vanished-session suspects against every local store; a * constructor seam so engine tests can fake local resolution. */ private readonly resolveLocalSessionIds: LocalSessionIdResolver; + /** Continuation status of push-marked suspects; same seam pattern. */ + private readonly resolveContinuationStatuses: ContinuationStatusResolver; /** Per-org timestamp of the last vanished-session GC sweep. */ private readonly lastVanishedSweepAtMs = new Map(); /** `${orgId}:${sessionId}` → consecutive sweeps confirmed absent. A * suspect retracts only at VANISHED_SESSION_RETRACT_CONFIRMATIONS, so one * empty lookup during a cache rebuild cannot mass-retract live rows. */ private readonly vanishedStrikes = new Map(); + /** Same two-strike discipline for superseded-continuation retracts. */ + private readonly supersededStrikes = new Map(); constructor( client: Org2CloudSyncClientDeps = org2CloudSyncClient, projectsClient: Org2CloudProjectsClientDeps = org2CloudProjectsClient, projectSyncBridge: ProjectSyncBridge = tauriProjectSyncBridge, probeSchemaVersion: Org2CloudSchemaVersionProbe = schemaVersion, - resolveLocalSessionIds: LocalSessionIdResolver = resolveLocalSessionIdsViaAggregateList + resolveLocalSessionIds: LocalSessionIdResolver = resolveLocalSessionIdsViaAggregateList, + resolveContinuationStatuses: ContinuationStatusResolver = resolveContinuationStatusesViaCache ) { super(); this.client = client; @@ -227,6 +235,7 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle { this.sessionColdStart = new Org2CloudSessionColdStart(client); this.schemaGate = new Org2CloudSchemaGate(probeSchemaVersion); this.resolveLocalSessionIds = resolveLocalSessionIds; + this.resolveContinuationStatuses = resolveContinuationStatuses; } override start(store: CloudStore): void { @@ -863,12 +872,15 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle { if (now - lastSweepAt < VANISHED_SESSION_SWEEP_INTERVAL_MS) return; this.lastVanishedSweepAtMs.set(orgId, now); + const markedSessionIds = this.sessionSync.markedSessionIds(orgId); + const liveSessions = store.get(sessionsAtom); + const liveSessionIds = new Set( + liveSessions.map((session) => session.session_id) + ); const vanishedIds = await findVanishedPushedSessionIds({ orgId, - markedSessionIds: this.sessionSync.markedSessionIds(orgId), - liveSessionIds: new Set( - store.get(sessionsAtom).map((session) => session.session_id) - ), + markedSessionIds, + liveSessionIds, resolveSessionIds: this.resolveLocalSessionIds, }); if (this.generation !== generation) return; @@ -911,6 +923,69 @@ export class Org2CloudSyncEngine extends Org2CloudSyncLifecycle { ); } } + + // Continuation-superseded reconcile: a compaction demotes the old + // sibling out of the roster while its Team Sessions row lingers as a + // stale duplicate of the family. Retract it ONLY when the family's + // listable winner is itself replay-pushed to this org — the conversation + // stays represented by exactly one live row. NOTE the deliberate content + // tradeoff: the demoted row is the only cloud replay of the pre-compact + // detail; the winner carries the compacted continuation. The source + // transcript stays on the owner's disk and can always be re-shared. + const superseded = await findSupersededPushedSessions({ + orgId, + markedSessionIds, + liveSessionIds, + resolveStatuses: this.resolveContinuationStatuses, + }); + if (this.generation !== generation) return; + const supersededNow = new Set(superseded.map((entry) => entry.sessionId)); + for (const key of this.supersededStrikes.keys()) { + if (!key.startsWith(`${orgId}:`)) continue; + if (!supersededNow.has(key.slice(orgId.length + 1))) { + this.supersededStrikes.delete(key); + } + } + for (const { sessionId, lineageId } of superseded) { + if (this.generation !== generation) return; + const winner = liveSessions.find( + (session) => + session.session_id !== sessionId && + session.continuationLineageId === lineageId && + this.sessionSync.hasReplayPushed(orgId, session.session_id) + ); + if (!winner) continue; + const strikeKey = `${orgId}:${sessionId}`; + const strikes = (this.supersededStrikes.get(strikeKey) ?? 0) + 1; + if (strikes < VANISHED_SESSION_RETRACT_CONFIRMATIONS) { + this.supersededStrikes.set(strikeKey, strikes); + log.info( + `superseded-continuation suspect ${sessionId} org ${orgId} ` + + `(winner ${winner.session_id}, ` + + `${strikes}/${VANISHED_SESSION_RETRACT_CONFIRMATIONS}); ` + + `deferring retract to the next sweep` + ); + continue; + } + try { + log.info( + `cloud retract [superseded continuation]: session ${sessionId} ` + + `org ${orgId} (winner ${winner.session_id})` + ); + await this.sessionSync.retractSession(fresh, orgId, sessionId); + this.supersededStrikes.delete(strikeKey); + } catch (error) { + if (this.generation !== generation) return; + if (isCloudSyncBackoffError(error)) { + this.orgBackoff.backOffOrg(orgId, error); + return; + } + log.warn( + `cloud retract failed for superseded continuation ${sessionId}:`, + error + ); + } + } } /** The engine singleton outlives individual memberships. Keep every diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSessions.ts b/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSessions.ts index 107312c40..98e99d47c 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSessions.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSessions.ts @@ -25,6 +25,10 @@ * suspect to be confirmed absent on two consecutive sweeps before * retracting. */ +import { + type ImportedContinuationStatus, + importedHistoryContinuationStatuses, +} from "@src/api/tauri/externalHistory/imported/cloudReplay"; import { sessionAggregateList } from "@src/api/tauri/session"; import { createLogger } from "@src/hooks/logger"; @@ -88,3 +92,56 @@ export async function findVanishedPushedSessionIds({ } return suspects.filter((sessionId) => !resolved.has(sessionId)); } + +/** Resolver for continuation statuses of push-marked suspects. */ +export type ContinuationStatusResolver = ( + sessionIds: readonly string[] +) => Promise; + +export const resolveContinuationStatusesViaCache: ContinuationStatusResolver = ( + sessionIds +) => importedHistoryContinuationStatuses([...sessionIds]); + +export interface SupersededPushedSession { + sessionId: string; + lineageId: string; +} + +/** + * Push-marked ids that left the roster because the continuation election + * DEMOTED them — the imported cache still holds the row and reports a + * strictly newer sibling. These are candidates for retracting the stale + * Team Sessions duplicate, but ONLY the caller can confirm the family's + * listable winner is itself pushed to the same org; without a lineage id + * the winner cannot be identified and the row is left alone. A failed + * lookup means "unknown", never "superseded". + */ +export async function findSupersededPushedSessions({ + orgId, + markedSessionIds, + liveSessionIds, + resolveStatuses, +}: { + orgId: string; + markedSessionIds: ReadonlySet; + liveSessionIds: ReadonlySet; + resolveStatuses: ContinuationStatusResolver; +}): Promise { + const suspects = [...markedSessionIds].filter( + (sessionId) => !liveSessionIds.has(sessionId) + ); + if (suspects.length === 0) return []; + let statuses: readonly ImportedContinuationStatus[]; + try { + statuses = await resolveStatuses(suspects); + } catch (error) { + log.warn(`continuation-status lookup failed for org ${orgId}:`, error); + return []; + } + return statuses + .filter((status) => status.superseded && status.lineageId) + .map((status) => ({ + sessionId: status.sessionId, + lineageId: status.lineageId as string, + })); +} diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSweep.test.ts b/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSweep.test.ts index fd7a81406..eb305b72b 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSweep.test.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.vanishedSweep.test.ts @@ -5,14 +5,19 @@ import { VANISHED_SESSION_SWEEP_INTERVAL_MS, } from "./org2CloudSyncEngine.constants"; import { + SESSION, cleanupEngineFixture, createEngineFixture, engineTestDeps, } from "./org2CloudSyncEngine.testUtils"; import type { EngineFixture } from "./org2CloudSyncEngine.testUtils"; -const { Org2CloudSyncEngine, org2CloudPushedMetadataAtom, sessionsAtom } = - engineTestDeps; +const { + Org2CloudSyncEngine, + org2CloudPushCursorsAtom, + org2CloudPushedMetadataAtom, + sessionsAtom, +} = engineTestDeps; describe("vanished-session sweep two-strike confirmation", () => { let fixture: EngineFixture; @@ -92,3 +97,103 @@ describe("vanished-session sweep two-strike confirmation", () => { expect(client.deleteSession).toHaveBeenCalledTimes(1); }); }); + +describe("superseded-continuation reconcile", () => { + let fixture: EngineFixture; + let store: EngineFixture["store"]; + let client: EngineFixture["client"]; + let engine: EngineFixture["engine"]; + let resolveLocalSessionIds: ReturnType; + let resolveContinuationStatuses: ReturnType; + + function startSweepEngine(): void { + fixture.engine.stop(); + engine = new Org2CloudSyncEngine( + client, + fixture.projectsClient, + fixture.bridge, + undefined, + resolveLocalSessionIds as never, + resolveContinuationStatuses as never + ); + engine.start(store); + } + + async function runSweepPass(): Promise { + vi.setSystemTime(Date.now() + VANISHED_SESSION_SWEEP_INTERVAL_MS + 1); + await engine.runSyncPass(); + } + + beforeEach(() => { + fixture = createEngineFixture(); + ({ store, client } = fixture); + engine = fixture.engine; + // old-sib was push-marked, then the continuation election demoted it out + // of the roster; its file (and cache row) still exist locally. The + // winner is an ordinary scoped session the engine pushes naturally — + // its cursor comes from that push, exactly like production. + store.set(org2CloudPushedMetadataAtom, { "corg-1:old-sib": true }); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: "winner", + continuationLineageId: "lin-1", + }, + ]); + resolveLocalSessionIds = vi.fn().mockResolvedValue(new Set(["old-sib"])); + resolveContinuationStatuses = vi + .fn() + .mockResolvedValue([ + { sessionId: "old-sib", lineageId: "lin-1", superseded: true }, + ]); + }); + + afterEach(() => { + cleanupEngineFixture(engine); + }); + + it("retracts the demoted sibling only when the winner is replay-pushed, after two strikes", async () => { + startSweepEngine(); + + // Pass 1 pushes the winner (its cursor now covers pushed events) and + // records the superseded suspect's first strike — never a retract. + await engine.runSyncPass(); + expect(client.deleteSession).not.toHaveBeenCalled(); + expect( + store.get(org2CloudPushCursorsAtom)["corg-1:winner"]?.pushedCount ?? 0 + ).toBeGreaterThan(0); + + await runSweepPass(); + expect(client.deleteSession).toHaveBeenCalledTimes(1); + expect(client.deleteSession).toHaveBeenCalledWith( + "jwt-1", + "corg-1", + "old-sib" + ); + // The retract dropped the marker; nothing is left to reconcile. + await runSweepPass(); + expect(client.deleteSession).toHaveBeenCalledTimes(1); + }); + + it("leaves the row alone while the family has no pushed winner", async () => { + // No session carries the suspect's lineage at all. + store.set(sessionsAtom, []); + startSweepEngine(); + + await engine.runSyncPass(); + await runSweepPass(); + await runSweepPass(); + expect(client.deleteSession).not.toHaveBeenCalled(); + }); + + it("treats a failed status lookup as unknown, never superseded", async () => { + resolveContinuationStatuses.mockRejectedValue(new Error("cache busy")); + startSweepEngine(); + + await engine.runSyncPass(); + await runSweepPass(); + await runSweepPass(); + const retractedIds = client.deleteSession.mock.calls.map((call) => call[2]); + expect(retractedIds).not.toContain("old-sib"); + }); +});