feat(cloud): retract the stale cloud row of a superseded continuation

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<Option<(Option<String>, 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,

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,

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<String>,
+    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<String>,
+) -> Result<Vec<ImportedContinuationStatus>, 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,

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<ImportedContinuationStatus[]> {
+  return invoke<ImportedContinuationStatus[]>(
+    "imported_history_continuation_statuses",
+    { sessionIds }
+  );
+}

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

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<string, number>();
   /** `${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<string, number>();
+  /** Same two-strike discipline for superseded-continuation retracts. */
+  private readonly supersededStrikes = new Map<string, number>();
 
   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

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<readonly ImportedContinuationStatus[]>;
+
+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<string>;
+  liveSessionIds: ReadonlySet<string>;
+  resolveStatuses: ContinuationStatusResolver;
+}): Promise<SupersededPushedSession[]> {
+  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,
+    }));
+}