diff --git a/apps/desktop/package.json b/apps/desktop/package.json index e49a6dc3..a6462fd1 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "desktop", - "version": "0.15.115", + "version": "0.15.116", "description": "ClosedLoop Desktop", "author": "ClosedLoop AI ", "private": true, diff --git a/apps/desktop/src/main/collectors/import-session.ts b/apps/desktop/src/main/collectors/import-session.ts index ed729669..09fa314c 100644 --- a/apps/desktop/src/main/collectors/import-session.ts +++ b/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"); diff --git a/apps/desktop/test/collectors-import.test.ts b/apps/desktop/test/collectors-import.test.ts index 74b54e7f..ae50f3c2 100644 --- a/apps/desktop/test/collectors-import.test.ts +++ b/apps/desktop/test/collectors-import.test.ts @@ -148,6 +148,231 @@ test("a new event with a later timestamp backfills without duplicating prior eve } }); +test("backfill source-dates analytics while preserving the sync mutation cursor", () => { + const { db, cleanup } = openTempDb(); + try { + const importer = createImporter(db.connection, { + tokenUsage: db.tokenUsage, + detectBillingMode: () => "api", + now: () => "2026-06-06T12:00:00.000Z", + }); + + importer.importSession( + makeSession({ + sessionId: "old-session", + startedAt: "2026-04-01T10:00:00.000Z", + endedAt: "2026-04-01T10:05:00.000Z", + fileModifiedAt: Date.parse("2026-06-06T11:59:00.000Z"), + messageTimestamps: [ + "2026-04-01T10:01:00.000Z", + "2026-04-01T10:02:00.000Z", + ], + toolUses: [ + { name: "Read", timestamp: "2026-04-01T10:01:30.000Z", input: { file: "x" } }, + ], + }), + "codex", + ); + + const session = db.sessions.getById("old-session"); + assert.ok(session); + assert.equal(session.status, "completed"); + assert.equal(session.startedAt, "2026-04-01T10:00:00.000Z"); + assert.equal(session.endedAt, "2026-04-01T10:05:00.000Z"); + assert.equal(session.updatedAt, "2026-06-06T12:00:00.000Z"); + + const agent = db.agents.getBySession("old-session")[0]; + assert.equal(agent.updatedAt, "2026-06-06T12:00:00.000Z"); + + const events = db.events.getBySession("old-session"); + assert.equal(events.length, 3); + assert.deepEqual( + events.map((event) => event.createdAt).sort(), + [ + "2026-04-01T10:01:00.000Z", + "2026-04-01T10:01:30.000Z", + "2026-04-01T10:02:00.000Z", + ], + ); + const recentRow = db.connection.prepare(` + SELECT COUNT(*) AS count + FROM events + WHERE session_id = ? + AND created_at >= datetime(?, '-30 days') + `).get("old-session", "2026-06-06T12:00:00.000Z") as { count: number }; + assert.equal( + recentRow.count, + 0, + "historical backfill must not appear in recent event windows", + ); + + const tokenRow = db.connection.prepare(` + SELECT created_at AS createdAt, updated_at AS updatedAt + FROM token_usage + WHERE session_id = ? AND model = ? + `).get("old-session", "gpt-5") as { createdAt: string; updatedAt: string }; + assert.equal(tokenRow.createdAt, "2026-04-01T10:05:00.000Z"); + assert.equal(tokenRow.updatedAt, "2026-04-01T10:05:00.000Z"); + } finally { + cleanup(); + } +}); + +test("backfill falls back missing event timestamps to the source session date", () => { + const { db, cleanup } = openTempDb(); + try { + const importer = createImporter(db.connection, { + tokenUsage: db.tokenUsage, + detectBillingMode: () => "api", + now: () => "2026-06-06T12:00:00.000Z", + }); + + importer.importSession( + makeSession({ + sessionId: "missing-event-ts", + startedAt: "2026-04-02T10:00:00.000Z", + endedAt: "2026-04-02T10:05:00.000Z", + messageTimestamps: [], + toolUses: [{ name: "Read", timestamp: null, input: { file: "x" } }], + }), + "codex", + ); + + const events = db.events.getBySession("missing-event-ts"); + assert.equal(events.length, 1); + assert.equal(events[0].createdAt, "2026-04-02T10:05:00.000Z"); + } finally { + cleanup(); + } +}); + +test("backfill appends new missing-timestamp events without high-water collision", () => { + const { db, cleanup } = openTempDb(); + try { + const importer = createImporter(db.connection, { + tokenUsage: db.tokenUsage, + detectBillingMode: () => "api", + now: () => "2026-06-06T12:00:00.000Z", + }); + + importer.importSession( + makeSession({ + sessionId: "missing-event-append", + startedAt: "2026-04-02T10:00:00.000Z", + endedAt: "2026-04-02T10:05:00.000Z", + messageTimestamps: [], + toolUses: [{ name: "Read", timestamp: null, input: { file: "a" } }], + }), + "codex", + ); + + importer.importSession( + makeSession({ + sessionId: "missing-event-append", + startedAt: "2026-04-02T10:00:00.000Z", + endedAt: "2026-04-02T10:05:00.000Z", + messageTimestamps: [], + toolUses: [ + { name: "Read", timestamp: null, input: { file: "a" } }, + { name: "Read", timestamp: null, input: { file: "b" } }, + ], + }), + "codex", + ); + + const events = db.events.getBySession("missing-event-append"); + assert.equal(events.length, 2); + assert.deepEqual( + events.map((event) => event.createdAt), + [ + "2026-04-02T10:05:00.000Z", + "2026-04-02T10:05:00.001Z", + ], + ); + } finally { + cleanup(); + } +}); + +test("historical backfill imported after sync cursor remains visible to incremental sync", () => { + const { db, cleanup } = openTempDb(); + try { + db.connection.prepare(` + INSERT INTO sessions (id, name, status, started_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `).run( + "cursor-sentinel", + "Cursor sentinel", + "completed", + "2026-06-06T11:59:00.000Z", + "2026-06-06T12:00:00.000Z", + ); + + const importer = createImporter(db.connection, { + tokenUsage: db.tokenUsage, + detectBillingMode: () => "api", + now: () => "2026-06-06T12:01:00.000Z", + }); + + importer.importSession( + makeSession({ + sessionId: "old-after-cursor", + startedAt: "2026-04-01T10:00:00.000Z", + endedAt: "2026-04-01T10:05:00.000Z", + messageTimestamps: [], + toolUses: [], + }), + "codex", + ); + + const rows = db.connection.prepare(` + SELECT id + FROM sessions + WHERE updated_at >= ? + ORDER BY updated_at DESC, id DESC + `).all("2026-06-06T12:00:00.000Z") as Array<{ id: string }>; + assert.ok( + rows.some((row) => row.id === "old-after-cursor"), + "new historical imports must remain visible to updated_at cursor sync", + ); + const session = db.sessions.getById("old-after-cursor"); + assert.equal(session?.startedAt, "2026-04-01T10:00:00.000Z"); + assert.equal(session?.updatedAt, "2026-06-06T12:01:00.000Z"); + } finally { + cleanup(); + } +}); + +test("future-dated source activity does not mark a backfilled session active", () => { + const { db, cleanup } = openTempDb(); + try { + const importer = createImporter(db.connection, { + tokenUsage: db.tokenUsage, + detectBillingMode: () => "api", + now: () => "2026-06-06T12:00:00.000Z", + }); + + importer.importSession( + makeSession({ + sessionId: "future-source", + startedAt: "2026-06-06T12:30:00.000Z", + endedAt: "2026-06-06T12:35:00.000Z", + fileModifiedAt: Date.parse("2026-06-06T11:59:00.000Z"), + messageTimestamps: [], + toolUses: [], + }), + "codex", + ); + + const session = db.sessions.getById("future-source"); + assert.ok(session); + assert.equal(session.status, "completed"); + assert.equal(session.updatedAt, "2026-06-06T12:00:00.000Z"); + } finally { + cleanup(); + } +}); + test("Agent/Task tool use creates an idempotent subagent row", () => { const { db, cleanup } = openTempDb(); try {