Merge pull request #277 from closedloop-ai/fix-backfill-source-timestamps

FEA-1564: Backdate backfill timestamps

Author: mikeangstadt

apps/desktop/src/main/collectors/import-session.ts

@@ -11,11 +11,13 @@ import type { Harness, NormalizedSession, NormalizedToolUse } from "./types.js";
  *
  * Idempotency (FEA-1503 AC): re-import adds nothing new.
  *  - session row: COALESCE-fill on conflict, never clobbers a live row.
+ *    `updated_at` stays an ingest-time mutation cursor for cloud sync; source
+ *    dates live on `started_at`, `ended_at`, events, and token usage analytics.
  *  - events: per-(session, event_type) high-water-mark on `created_at` — only
- *    events with a transcript timestamp strictly greater than the stored max are
- *    inserted (the exact vendor mechanism). Hook-written events carry
- *    `created_at ≈ now`, so file events with past transcript timestamps fall under
- *    the high-water-mark and are never double-counted against the live hook path.
+ *    events with a source timestamp strictly greater than the stored max are
+ *    inserted. Backfill never stamps events with importer runtime `now`; when an
+ *    individual source event has no timestamp, it falls back to the session's
+ *    source timestamp so date windows remain tied to when work occurred.
  *  - tokens: `tokenUsage.replace` nets zero when re-applying equal cumulatives.
  *
  * Each session is applied in one `BEGIN IMMEDIATE` transaction (mirrors
@@ -81,10 +83,12 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
       cwd = COALESCE(cwd, ?),
       harness = CASE WHEN COALESCE(harness, '') = '' THEN ? ELSE harness END,
       billing_mode = CASE WHEN COALESCE(billing_mode, '') IN ('', 'unknown') THEN ? ELSE billing_mode END,
-      metadata = ?,
-      updated_at = ?
+      metadata = ?
     WHERE id = ?
   `);
+  const touchSessionStmt = db.prepare(
+    "UPDATE sessions SET updated_at = CASE WHEN updated_at IS NULL OR updated_at < ? THEN ? ELSE updated_at END WHERE id = ?",
+  );
   const reactivateSessionStmt = db.prepare(
     "UPDATE sessions SET status = 'active', ended_at = NULL, updated_at = ? WHERE id = ?",
   );
@@ -114,11 +118,19 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
     return `${sessionId}-main`;
   }
 
-  function isRecentlyActive(session: NormalizedSession, nowMs: number): boolean {
+  function isRecentlyActive(
+    session: NormalizedSession,
+    nowMs: number,
+    sourceUpdatedAt: string,
+  ): boolean {
+    const sourceUpdatedAtMs = Date.parse(sourceUpdatedAt);
     return (
       session.fileModifiedAt != null &&
       Number.isFinite(session.fileModifiedAt) &&
-      nowMs - session.fileModifiedAt < RECENT_ACTIVITY_MS
+      nowMs - session.fileModifiedAt < RECENT_ACTIVITY_MS &&
+      Number.isFinite(sourceUpdatedAtMs) &&
+      sourceUpdatedAtMs <= nowMs &&
+      nowMs - sourceUpdatedAtMs < RECENT_ACTIVITY_MS
     );
   }
 
@@ -173,6 +185,29 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
     );
   }
 
+  function sessionSourceUpdatedAt(session: NormalizedSession, startedAt: string): string {
+    let latest = startedAt;
+    let latestMs = Date.parse(startedAt);
+    const consider = (value: string | null | undefined): void => {
+      if (!value) return;
+      const ms = Date.parse(value);
+      if (!Number.isFinite(ms)) return;
+      if (!Number.isFinite(latestMs) || ms > latestMs) {
+        latest = value;
+        latestMs = ms;
+      }
+    };
+
+    consider(session.endedAt);
+    for (const ts of session.messageTimestamps ?? []) consider(ts);
+    for (const toolUse of session.toolUses ?? []) consider(toolUse.timestamp);
+    for (const duration of session.turnDurations ?? []) consider(duration.timestamp);
+    for (const error of session.apiErrors ?? []) consider(error.timestamp);
+    for (const error of session.toolResultErrors ?? []) consider(error.timestamp);
+
+    return latest;
+  }
+
   function importSession(session: NormalizedSession, harness: Harness): ImportResult {
     if (typeof session.sessionId !== "string" || session.sessionId.length === 0) {
       return { skipped: true, reactivated: false };
@@ -181,11 +216,14 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
       return { skipped: true, reactivated: false };
     }
 
+    const startedAt = session.startedAt;
     const now = nowFn();
+    const sourceUpdatedAt = sessionSourceUpdatedAt(session, startedAt);
     const nowMs = Date.parse(now);
     const recentlyActive = isRecentlyActive(
       session,
       Number.isNaN(nowMs) ? Date.now() : nowMs,
+      sourceUpdatedAt,
     );
     const mainId = mainAgentId(session.sessionId);
 
@@ -205,9 +243,9 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
           status,
           session.cwd ?? null,
           session.model ?? null,
-          session.startedAt,
-          session.endedAt ?? session.startedAt,
-          status === "completed" ? session.endedAt ?? null : null,
+          startedAt,
+          now,
+          status === "completed" ? sourceUpdatedAt : null,
           harness,
           billingMode,
           buildMetadata(session, harness),
@@ -223,9 +261,9 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
           status === "completed" ? "completed" : "waiting",
           null,
           null,
-          session.startedAt,
+          startedAt,
           now,
-          status === "completed" ? session.endedAt ?? now : null,
+          status === "completed" ? sourceUpdatedAt : null,
           null,
           null,
         );
@@ -238,7 +276,6 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
           harness,
           billingMode,
           buildMetadata(session, harness),
-          now,
           session.sessionId,
         );
         const isLive = existing.status === "active" && existing.ended_at == null;
@@ -261,6 +298,7 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
       }
 
       let inserted = 0;
+      let namelessEventCounter = 0;
       const addEvent = (
         eventType: string,
         agentId: string,
@@ -269,9 +307,14 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
         summary: string | null,
         data: string | null,
       ): void => {
-        if (!ts) return;
+        const eventTimestamp = ts ?? (() => {
+          // Synthetic increment to distinguish no-timestamp events in the same
+          // batch so they don't all collide on the high-water-mark dedup.
+          const base = Date.parse(sourceUpdatedAt);
+          return new Date(base + namelessEventCounter++).toISOString();
+        })();
         const prev = highWater.get(eventType);
-        if (prev != null && ts <= prev) return;
+        if (prev != null && eventTimestamp <= prev) return;
         insertEventStmt.run(
           randomUUID(),
           session.sessionId,
@@ -280,7 +323,7 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
           toolName,
           summary,
           data,
-          ts,
+          eventTimestamp,
         );
         inserted++;
       };
@@ -312,9 +355,9 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
             subagentName(tu),
             strOf(input.subagent_type) ?? null,
             prompt ? prompt.slice(0, 500) : null,
-            tu.timestamp ?? session.startedAt,
-            now,
-            tu.timestamp ?? session.endedAt ?? now,
+            tu.timestamp ?? startedAt,
+            tu.timestamp ?? now,
+            tu.timestamp ?? sourceUpdatedAt,
             mainId,
           );
           addEvent("PreToolUse", subId, tu.timestamp, tu.name, "Spawned subagent", eventData(enrichedData));
@@ -335,7 +378,11 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
 
       // ── Tokens (store reconciles raw/effective; idempotent on equal counts) ───
       for (const [model, counts] of Object.entries(session.tokensByModel ?? {})) {
-        deps.tokenUsage.replace(session.sessionId, model, counts, now);
+        deps.tokenUsage.replace(session.sessionId, model, counts, sourceUpdatedAt);
+      }
+
+      if (existing != null && inserted > 0 && !reactivated) {
+        touchSessionStmt.run(now, now, session.sessionId);
       }
 
       db.exec("COMMIT");