Skip to content
This repository was archived by the owner on Jun 8, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/desktop/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "desktop",
"version": "0.15.115",
"version": "0.15.116",
"description": "ClosedLoop Desktop",
"author": "ClosedLoop AI <support@closedloop.ai>",
"private": true,
Expand Down
89 changes: 68 additions & 21 deletions apps/desktop/src/main/collectors/import-session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = ?",
);
Expand Down Expand Up @@ -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 &&
Comment thread
mikeangstadt marked this conversation as resolved.
Number.isFinite(sourceUpdatedAtMs) &&
sourceUpdatedAtMs <= nowMs &&
nowMs - sourceUpdatedAtMs < RECENT_ACTIVITY_MS
);
}

Expand Down Expand Up @@ -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 };
Expand All @@ -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);

Expand All @@ -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),
Expand All @@ -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,
);
Expand All @@ -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;
Expand All @@ -261,6 +298,7 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
}

let inserted = 0;
let namelessEventCounter = 0;
const addEvent = (
eventType: string,
agentId: string,
Expand All @@ -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,
Expand All @@ -280,7 +323,7 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
toolName,
summary,
data,
ts,
eventTimestamp,
);
inserted++;
};
Expand Down Expand Up @@ -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));
Expand All @@ -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");
Expand Down
Loading
Loading