fix(connectors): make the sync-log sweep aware of a live run's heartbeat

Review round 6 on #6909, plus a consistency pass over the whole branch.

The sweep keys on the log row's startedAt, and the heartbeat added earlier in
this PR refreshes the connector's updatedAt — the log table has no equivalent,
so nothing refreshed what the sweep reads. A legitimately long in-process sync
kept its connector lock exactly as designed while its log row was closed as
failed at the TTL, and completeSyncLog's started guard then no-opped when the
run finished. A successful sync was recorded permanently as a failure and its
counters were lost to the listing-safety check.

Neither round was wrong alone; the combination was. The sweep now spares a row
whose id is still the connector's lock token, reusing the ownership mechanism
rather than adding another. Every orphan still drains: a reclaimed run's token
is cleared, a replaced run's token belongs to its successor, and rows predating
the column have none.

Five documentation claims that later rounds falsified are corrected, including
the sweep's own rationale, which still argued the platform kills every run at
the duration ceiling — the reasoning the heartbeat exists because it does not
hold for the in-process path.

applySupersededOutcome's boolean parameter was vestigial: both call sites passed
false and a test asserted the dead branch. Simplified.

Author: waleedlatif1

apps/sim/app/api/knowledge/connectors/sync/route.test.ts

@@ -9,6 +9,7 @@
  */
 import {
   dbChainMockFns,
+  flattenMockConditions,
   hasMockCondition,
   type MockCondition,
   resetDbChainMock,
@@ -213,6 +214,29 @@ describe('connector sync scheduler stale-lock reaper', () => {
     ).toBe(true)
   })
 
+  it('spares the log row of a run that still holds its connector lock', async () => {
+    await runTickRecovering(['connector-1'])
+
+    /**
+     * The sweep keys on `startedAt`, which no heartbeat refreshes, so age alone
+     * would close a legitimately long in-process run's row and record a
+     * successful sync as failed.
+     */
+    const where = dbChainMockFns.where.mock.calls[1][0]
+    const liveness = flattenMockConditions(where).find(
+      (node: MockCondition) => typeof node.toSQL === 'function'
+    )
+    expect(liveness).toBeDefined()
+
+    const rendered = (liveness as unknown as MockSqlFragment).toSQL().sql
+    expect(rendered).toContain('NOT EXISTS')
+    expect(rendered).toContain("'syncing'")
+
+    const bound = (liveness as unknown as MockSqlFragment).values
+    expect(bound).toContain(schemaMock.knowledgeConnector.syncLockToken)
+    expect(bound).toContain(schemaMock.knowledgeConnectorSyncLog.id)
+  })
+
   it('closes stale sync-log rows even when no connector was reclaimed this tick', async () => {
     /**
      * The self-healing assertion. A row orphaned before this sweep existed —

apps/sim/app/api/knowledge/connectors/sync/route.ts

@@ -31,6 +31,29 @@ const DISPATCH_CONCURRENCY = 10
 
 const STALE_LOCK_ERROR_MESSAGE = 'Sync timed out (stale lock recovered)'
 
+/**
+ * Excludes a sync-log row whose run still demonstrably holds its connector's
+ * lock.
+ *
+ * The sweep keys on `startedAt`, and nothing refreshes that — the heartbeat
+ * renews `knowledge_connector.updatedAt`, and the log table has no equivalent
+ * column. So a legitimately long in-process run keeps its connector lock but
+ * would still have its log row closed as `failed` at the TTL, recording a
+ * successful sync as a failure and losing its counters to
+ * `loadPreviousListingObservation`. Matching the connector's `syncLockToken`
+ * against the row's own id is exactly "this run is still the lock holder", so a
+ * live run is spared while every orphan — reclaimed, replaced, or predating the
+ * token column, where the token is NULL — is still swept.
+ */
+function runNoLongerHoldsItsLock(): SQL {
+  return sql`NOT EXISTS (
+    SELECT 1 FROM ${knowledgeConnector}
+    WHERE ${knowledgeConnector.id} = ${knowledgeConnectorSyncLog.connectorId}
+      AND ${knowledgeConnector.syncLockToken} = ${knowledgeConnectorSyncLog.id}
+      AND ${knowledgeConnector.status} = 'syncing'
+  )`
+}
+
 /**
  * The reclaimed connector's new consecutive-failure count.
  *
@@ -117,12 +140,13 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
      * stay stranded forever. Keying off the row's own `startedAt` instead makes
      * the sweep self-healing and lets it drain the existing backlog.
      *
-     * Safe on liveness: the run ceiling is
-     * {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS} and this TTL is twice that, so
-     * a row older than the cutoff belongs to a run the platform has already
-     * killed and which cannot still be writing. The predicate is per-row on
-     * `startedAt`, so a fresh run's log row can never be caught by it — even on
-     * a connector whose previous run is being reclaimed in this same tick.
+     * Age alone does not prove a run is dead: the in-process fallback path has
+     * no duration cap, so a large self-hosted sync can genuinely still be
+     * working past the TTL. `runNoLongerHoldsItsLock` is what makes this safe —
+     * a run still holding its connector's lock is spared regardless of age.
+     * The age predicate is also per-row on `startedAt`, so a fresh run's log row
+     * can never be caught by it, even on a connector whose previous run is being
+     * reclaimed in this same tick.
      */
     const closedSyncLogs = await db
       .update(knowledgeConnectorSyncLog)
@@ -134,7 +158,8 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       .where(
         and(
           eq(knowledgeConnectorSyncLog.status, 'started'),
-          lte(knowledgeConnectorSyncLog.startedAt, staleCutoff)
+          lte(knowledgeConnectorSyncLog.startedAt, staleCutoff),
+          runNoLongerHoldsItsLock()
         )
       )
       .returning({ id: knowledgeConnectorSyncLog.id })