diff --git a/services/apps/connectors_worker/package.json b/services/apps/connectors_worker/package.json
index 6fb4787107..905a3271a1 100644
--- a/services/apps/connectors_worker/package.json
+++ b/services/apps/connectors_worker/package.json
@@ -10,7 +10,8 @@
     "script:seed-github-sync-units": "tsx src/bin/seed-github-sync-units.ts",
     "script:audit-github-app-budget": "tsx src/bin/audit-github-app-budget.ts",
     "script:diff-nango-shadow": "tsx src/bin/diff-nango-shadow.ts",
-    "script:diff-shadow-unit-detail": "tsx src/bin/diff-shadow-unit-detail.ts"
+    "script:diff-shadow-unit-detail": "tsx src/bin/diff-shadow-unit-detail.ts",
+    "script:backfill-octolens-mentions": "tsx src/bin/backfill-octolens-mentions.ts"
   },
   "dependencies": {
     "@crowd/archetype-standard": "workspace:*",
diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts
new file mode 100644
index 0000000000..cae9bba76b
--- /dev/null
+++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.test.ts
@@ -0,0 +1,86 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { deferUnit, startRun } from './dispatcherActivities'
+
+const SHORT_DEFER_AT = new Date('2026-10-16T12:01:00.000Z')
+
+const mocks = vi.hoisted(() => ({
+  start: vi.fn(),
+  renewRunningUnitLease: vi.fn(),
+  rescheduleUnit: vi.fn(),
+}))
+
+vi.mock('../main', () => ({
+  svc: {
+    postgres: { writer: {} },
+    log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+    temporal: {
+      workflow: { start: mocks.start },
+    },
+  },
+}))
+
+vi.mock('@crowd/data-access-layer/src/connectors', () => ({
+  claimDueUnits: vi.fn(),
+  renewRunningUnitLease: mocks.renewRunningUnitLease,
+  rescheduleUnit: mocks.rescheduleUnit,
+}))
+
+vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({
+  dbStoreQx: vi.fn(() => ({})),
+}))
+
+vi.mock('../scheduling', () => ({
+  shortDeferRunAt: () => SHORT_DEFER_AT,
+}))
+
+const UNIT = {
+  id: 'unit-1',
+  integrationId: 'integration-1',
+  platform: 'github',
+  syncName: 'issues',
+  channelId: 'repo',
+  channelName: 'https://github.com/org/repo',
+  leaseLapsed: false,
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  mocks.renewRunningUnitLease.mockResolvedValue(1)
+})
+
+describe('startRun', () => {
+  it('starts a workflow without touching the claim', async () => {
+    mocks.start.mockResolvedValue(undefined)
+
+    expect(await startRun(UNIT)).toBe('started')
+    expect(mocks.renewRunningUnitLease).not.toHaveBeenCalled()
+    expect(mocks.rescheduleUnit).not.toHaveBeenCalled()
+  })
+
+  it('keeps a still-running unit claimed instead of releasing it', async () => {
+    const alreadyStarted = new Error('already started')
+    alreadyStarted.name = 'WorkflowExecutionAlreadyStartedError'
+    mocks.start.mockRejectedValue(alreadyStarted)
+
+    expect(await startRun(UNIT)).toBe('alreadyRunning')
+    expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id)
+    expect(mocks.rescheduleUnit).not.toHaveBeenCalled()
+  })
+})
+
+describe('deferUnit', () => {
+  it('releases a freshly claimed unit', async () => {
+    await deferUnit(UNIT.id, false)
+
+    expect(mocks.rescheduleUnit).toHaveBeenCalledWith({}, UNIT.id, SHORT_DEFER_AT)
+    expect(mocks.renewRunningUnitLease).not.toHaveBeenCalled()
+  })
+
+  it('keeps the claim on a unit whose previous lease lapsed, as its run may still be going', async () => {
+    await deferUnit(UNIT.id, true)
+
+    expect(mocks.renewRunningUnitLease).toHaveBeenCalledWith({}, UNIT.id)
+    expect(mocks.rescheduleUnit).not.toHaveBeenCalled()
+  })
+})
diff --git a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts
index 2a563e2dd0..0a467a6d43 100644
--- a/services/apps/connectors_worker/src/activities/dispatcherActivities.ts
+++ b/services/apps/connectors_worker/src/activities/dispatcherActivities.ts
@@ -1,12 +1,16 @@
 import { createTokenPool, findManifest, mapWithConcurrency } from '@crowd/connectors'
-import { claimDueUnits, rescheduleUnit } from '@crowd/data-access-layer/src/connectors'
+import {
+  claimDueUnits,
+  renewRunningUnitLease,
+  rescheduleUnit,
+} from '@crowd/data-access-layer/src/connectors'
 import type { IClaimedUnit } from '@crowd/data-access-layer/src/connectors'
 import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { RedisCache } from '@crowd/redis'
 import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
 
 import { svc } from '../main'
-import { runningProbeRunAt, shortDeferRunAt } from '../scheduling'
+import { shortDeferRunAt } from '../scheduling'
 import type { IAdmissionResult, IDispatchCounts, StartRunResult } from '../types'
 
 const TASK_QUEUE = 'connectors'
@@ -40,8 +44,14 @@ export async function admitByBudget(units: IClaimedUnit[]): Promise<IAdmissionRe
   return result
 }
 
-export async function deferUnit(unitId: string): Promise<void> {
-  await rescheduleUnit(dbStoreQx(svc.postgres.writer), unitId, shortDeferRunAt())
+// A lapsed lease may belong to a run still in flight, so that claim is kept instead of released.
+export async function deferUnit(unitId: string, leaseLapsed = false): Promise<void> {
+  const qx = dbStoreQx(svc.postgres.writer)
+  if (leaseLapsed) {
+    await renewRunningUnitLease(qx, unitId)
+  } else {
+    await rescheduleUnit(qx, unitId, shortDeferRunAt())
+  }
 }
 
 function channelLabel(channelName: string): string {
@@ -86,18 +96,11 @@ export async function startRun(unit: IClaimedUnit): Promise<StartRunResult> {
 }
 
 async function backOffRunningUnit(unit: IClaimedUnit): Promise<void> {
-  const nextRunAt = await runningSince(unit)
-    .then(runningProbeRunAt)
-    .catch(() => shortDeferRunAt())
-
-  await rescheduleUnit(dbStoreQx(svc.postgres.writer), unit.id, nextRunAt)
-  svc.log.info({ unitId: unit.id, nextRunAt }, 'unit already running, rescheduled probe')
-}
-
-async function runningSince(unit: IClaimedUnit): Promise<Date> {
-  const description = await svc.temporal.workflow.getHandle(syncRunWorkflowId(unit)).describe()
-
-  return description.startTime
+  const renewed = await renewRunningUnitLease(dbStoreQx(svc.postgres.writer), unit.id)
+  svc.log.info(
+    { unitId: unit.id, renewed: renewed > 0 },
+    'unit already running, lease renewed until it lapses',
+  )
 }
 
 export async function logDispatchSummary(counts: IDispatchCounts): Promise<void> {
diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts
new file mode 100644
index 0000000000..ea6a531fe7
--- /dev/null
+++ b/services/apps/connectors_worker/src/activities/syncRunActivities.test.ts
@@ -0,0 +1,137 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { SyncContext, SyncOutcome } from '@crowd/connectors'
+
+import { executeSync } from './syncRunActivities'
+
+const CADENCE_RUN_AT = new Date('2026-10-16T12:05:00.000Z')
+const SHORT_DEFER_RUN_AT = new Date('2026-10-16T12:01:00.000Z')
+
+const mocks = vi.hoisted(() => ({
+  run: vi.fn(),
+  getUnitById: vi.fn(),
+  reassertUnitClaim: vi.fn(),
+  recordRunSuccess: vi.fn(),
+}))
+
+vi.mock('@temporalio/activity', () => ({
+  Context: {
+    current: () => ({
+      info: { workflowExecution: { workflowId: 'wf', runId: 'run' } },
+      heartbeat: vi.fn(),
+    }),
+  },
+}))
+
+vi.mock('../main', () => ({
+  svc: {
+    postgres: { writer: {} },
+    redis: {},
+    dataSinkWorkerEmitter: {},
+    log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+  },
+}))
+
+vi.mock('@crowd/logging', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@crowd/logging')>()),
+  getChildLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }),
+}))
+
+vi.mock('@crowd/connectors', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@crowd/connectors')>()),
+  getManifest: () => ({}),
+  getSync: () => ({ cadenceMinutes: 5, schema: {}, run: mocks.run }),
+  findSync: () => ({ cadenceMinutes: 5 }),
+  createTokenPool: () => ({ acquire: vi.fn(), park: vi.fn(), invalidate: vi.fn() }),
+  createHttpClient: () => ({ requestCount: () => 0 }),
+  createEmit: () => ({ emit: vi.fn(), emittedCount: () => 3 }),
+}))
+
+vi.mock('@crowd/data-access-layer/src/connectors', () => ({
+  getUnitById: mocks.getUnitById,
+  reassertUnitClaim: mocks.reassertUnitClaim,
+  recordRunSuccess: mocks.recordRunSuccess,
+  recordRunPartial: vi.fn(),
+  recordRunFailure: vi.fn(),
+  parkUnit: vi.fn(),
+  recordShadowRecords: vi.fn(),
+}))
+
+vi.mock('@crowd/data-access-layer/src/integrations', () => ({
+  fetchIntegrationById: async () => ({ segmentId: 'segment-1' }),
+}))
+
+vi.mock(
+  '@crowd/data-access-layer/src/old/apps/integration_stream_worker/integrationStream.repo',
+  () => ({
+    default: class {
+      publishExternalResult = vi.fn()
+    },
+  }),
+)
+
+vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({
+  dbStoreQx: vi.fn(() => ({})),
+}))
+
+vi.mock('../scheduling', () => ({
+  cadenceRunAt: () => CADENCE_RUN_AT,
+  shortDeferRunAt: () => SHORT_DEFER_RUN_AT,
+  failureRunAt: () => SHORT_DEFER_RUN_AT,
+}))
+
+const UNIT = {
+  id: 'unit-1',
+  integrationId: 'integration-1',
+  platform: 'octolens',
+  syncName: 'mentionPoll',
+  channelId: 'channel',
+  channelName: 'channel',
+  watermark: { cursor: 'before' },
+  emitEnabled: true,
+  consecutiveFailures: 0,
+}
+
+function runReturning(outcome: SyncOutcome) {
+  mocks.run.mockImplementation(async (ctx: SyncContext<unknown>) => {
+    await ctx.commitWatermark({ cursor: 'after' })
+    return outcome
+  })
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  mocks.getUnitById.mockResolvedValue(UNIT)
+})
+
+describe('executeSync outcome scheduling', () => {
+  it.each([
+    [{ complete: true }, true, CADENCE_RUN_AT],
+    [{ complete: false }, false, SHORT_DEFER_RUN_AT],
+    [{ complete: false, resumeAtCadence: true }, false, CADENCE_RUN_AT],
+  ])('records %j as complete=%s, next run at %s', async (outcome, complete, nextRunAt) => {
+    runReturning(outcome)
+
+    await executeSync(UNIT.id)
+
+    expect(mocks.recordRunSuccess).toHaveBeenCalledWith(
+      {},
+      UNIT.id,
+      { watermark: { cursor: 'after' }, emittedCount: 3, complete },
+      nextRunAt,
+    )
+  })
+})
+
+describe('executeSync claim', () => {
+  it('re-locks the unit before reading it, so the whole run is covered by lockedAt', async () => {
+    runReturning({ complete: true })
+
+    await executeSync(UNIT.id)
+
+    expect(mocks.reassertUnitClaim).toHaveBeenCalledWith({}, UNIT.id)
+    expect(mocks.reassertUnitClaim.mock.invocationCallOrder[0]).toBeLessThan(
+      mocks.getUnitById.mock.invocationCallOrder[0],
+    )
+  })
+})
diff --git a/services/apps/connectors_worker/src/activities/syncRunActivities.ts b/services/apps/connectors_worker/src/activities/syncRunActivities.ts
index ca1f265594..1fa0d37c37 100644
--- a/services/apps/connectors_worker/src/activities/syncRunActivities.ts
+++ b/services/apps/connectors_worker/src/activities/syncRunActivities.ts
@@ -14,6 +14,7 @@ import type { ConnectorHttp, Emitter, SyncContext } from '@crowd/connectors'
 import {
   getUnitById,
   parkUnit,
+  reassertUnitClaim,
   recordRunFailure,
   recordRunPartial,
   recordRunSuccess,
@@ -39,6 +40,7 @@ const PARKED_ERROR_CLASSES = ['provider.rate_limit', 'provider.unavailable'] as
 export async function executeSync(unitId: string): Promise<void> {
   const qx = dbStoreQx(svc.postgres.writer)
 
+  await reassertUnitClaim(qx, unitId)
   const unit = await getUnitById(qx, unitId)
   if (!unit) {
     throw new Error(`sync unit ${unitId} not found`)
@@ -156,7 +158,10 @@ export async function executeSync(unitId: string): Promise<void> {
     }
 
     const outcome = await sync.run(ctx)
-    const nextRunAt = outcome.complete ? cadenceRunAt(sync.cadenceMinutes) : shortDeferRunAt()
+    const nextRunAt =
+      outcome.complete || outcome.resumeAtCadence
+        ? cadenceRunAt(sync.cadenceMinutes)
+        : shortDeferRunAt()
 
     await recordRunSuccess(
       qx,
diff --git a/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts
new file mode 100644
index 0000000000..32b3300ddd
--- /dev/null
+++ b/services/apps/connectors_worker/src/bin/backfill-octolens-mentions.ts
@@ -0,0 +1,136 @@
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
+import {
+  listActiveOctolensKeywordMappings,
+  listOctolensMentionPollUnits,
+  queueOctolensKeywordBackfills,
+  rewindOctolensMentionPollStartDate,
+} from '@crowd/data-access-layer/src/octolens'
+import type { IOctolensMentionPollWriteResult } from '@crowd/data-access-layer/src/octolens'
+import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceLogger } from '@crowd/logging'
+
+import {
+  OCTOLENS_BACKFILL_USAGE,
+  parseOctolensBackfillArgs,
+  planOctolensBackfill,
+} from '../octolensBackfill'
+import type { OctolensBackfillArgs, OctolensBackfillPlan } from '../octolensBackfill'
+
+const log = getServiceLogger()
+
+const RETRY_HINT = 'retry in a few minutes, once the current mention poll run has finished'
+
+function parseArgsOrExit(): OctolensBackfillArgs {
+  try {
+    return parseOctolensBackfillArgs(process.argv.slice(2), new Date())
+  } catch (err) {
+    log.error((err as Error).message)
+    log.error(OCTOLENS_BACKFILL_USAGE)
+    process.exit(1)
+  }
+}
+
+async function loadPlan(qx: QueryExecutor, args: OctolensBackfillArgs) {
+  const [units, mappings] = await Promise.all([
+    listOctolensMentionPollUnits(qx),
+    listActiveOctolensKeywordMappings(qx),
+  ])
+  const activeMappingCounts = new Map<number, number>()
+  for (const { keywordId } of mappings) {
+    activeMappingCounts.set(keywordId, (activeMappingCounts.get(keywordId) ?? 0) + 1)
+  }
+  return planOctolensBackfill(args, { units, activeMappingCounts })
+}
+
+function describePlan(plan: OctolensBackfillPlan): Record<string, unknown> {
+  const base = { unitId: plan.unit.id, integrationId: plan.unit.integrationId }
+  if (plan.mode === 'keywords') {
+    return {
+      ...base,
+      keywordIdsToQueue: plan.keywordIdsToQueue,
+      alreadyQueued: plan.alreadyQueued,
+      currentBackfill: plan.unit.watermark?.backfill ?? [],
+    }
+  }
+  return {
+    ...base,
+    previousStartDate: plan.previousStartDate,
+    startDate: plan.startDate,
+    droppedCursor: plan.unit.watermark?.cursor ?? null,
+    droppedPendingMaxTimestamp: plan.unit.watermark?.pendingMaxTimestamp ?? null,
+  }
+}
+
+function applyPlan(
+  qx: QueryExecutor,
+  plan: OctolensBackfillPlan,
+): Promise<IOctolensMentionPollWriteResult> {
+  if (plan.mode === 'keywords') {
+    return queueOctolensKeywordBackfills(qx, plan.unit.id, plan.keywordIdsToQueue)
+  }
+  return rewindOctolensMentionPollStartDate(
+    qx,
+    plan.unit.id,
+    plan.startDate,
+    plan.previousStartDate,
+  )
+}
+
+setImmediate(async () => {
+  const args = parseArgsOrExit()
+  try {
+    const qx = pgpQx(await getDbConnection(WRITE_DB_CONFIG()))
+    const plan = await loadPlan(qx, args)
+    const summary = describePlan(plan)
+
+    if (plan.mode === 'keywords' && plan.keywordIdsToQueue.length === 0) {
+      log.info(summary, 'nothing to do: every keyword is already queued for backfill')
+      process.exit(0)
+    }
+
+    const busy = plan.unit.lockedAt !== null
+
+    if (args.dryRun) {
+      log.info({ ...summary, busy }, 'dry run: would update the octolens mention poll watermark')
+      if (busy) {
+        log.warn(`a mention poll run is in progress, a real run would refuse now: ${RETRY_HINT}`)
+      }
+      process.exit(0)
+    }
+
+    if (busy) {
+      log.error(summary, `refused: a mention poll run is in progress; ${RETRY_HINT}`)
+      process.exit(2)
+    }
+
+    const result = await applyPlan(qx, plan)
+    if (result.outcome === 'locked') {
+      log.error(summary, `refused: a mention poll run claimed the unit just now; ${RETRY_HINT}`)
+      process.exit(2)
+    }
+    if (result.outcome === 'changed') {
+      log.error(
+        { ...summary, watermark: result.watermark },
+        'refused: the start date changed since the plan was checked (another rewind or the first poll run); re-run to check against the new start date',
+      )
+      process.exit(2)
+    }
+    if (result.outcome !== 'applied') {
+      log.error(
+        { ...summary, outcome: result.outcome },
+        'refused: the sync unit is no longer active or was removed',
+      )
+      process.exit(1)
+    }
+
+    log.info(
+      { ...summary, watermark: result.watermark },
+      'octolens mention poll watermark updated; the next poll runs pick it up',
+    )
+    process.exit(0)
+  } catch (err) {
+    log.error(err, 'octolens mention backfill failed')
+    process.exit(1)
+  }
+})
diff --git a/services/apps/connectors_worker/src/octolensBackfill.test.ts b/services/apps/connectors_worker/src/octolensBackfill.test.ts
new file mode 100644
index 0000000000..494a704155
--- /dev/null
+++ b/services/apps/connectors_worker/src/octolensBackfill.test.ts
@@ -0,0 +1,166 @@
+import { describe, expect, it } from 'vitest'
+
+import type { IOctolensMentionPollUnit } from '@crowd/data-access-layer/src/octolens'
+
+import { parseOctolensBackfillArgs, planOctolensBackfill } from './octolensBackfill'
+import type { IOctolensBackfillState, OctolensBackfillArgs } from './octolensBackfill'
+
+const NOW = new Date('2026-10-16T12:00:00.000Z')
+
+function parse(...argv: string[]): OctolensBackfillArgs {
+  return parseOctolensBackfillArgs(argv, NOW)
+}
+
+function unit(overrides: Partial<IOctolensMentionPollUnit> = {}): IOctolensMentionPollUnit {
+  return {
+    id: 'unit-1',
+    integrationId: 'anchor-1',
+    status: 'active',
+    lockedAt: null,
+    watermark: {
+      startDate: '2026-10-16T11:55:00.000Z',
+      cursor: null,
+      pendingMaxTimestamp: null,
+      backfill: [{ keywordId: 7, cursor: 'k7-page2' }],
+    },
+    ...overrides,
+  }
+}
+
+function state(overrides: Partial<IOctolensBackfillState> = {}): IOctolensBackfillState {
+  return {
+    units: [unit()],
+    activeMappingCounts: new Map([
+      [7, 1],
+      [8, 1],
+      [9, 1],
+    ]),
+    ...overrides,
+  }
+}
+
+describe('parseOctolensBackfillArgs', () => {
+  it('parses keyword ids, dropping the pnpm separator and repeated ids', () => {
+    expect(parse('--', '--keywords', '9,8, 9', '--dry-run')).toEqual({
+      mode: 'keywords',
+      keywordIds: [9, 8],
+      dryRun: true,
+    })
+  })
+
+  it('normalizes --from to an ISO timestamp', () => {
+    expect(parse('--from', '2026-09-01')).toEqual({
+      mode: 'from',
+      startDate: '2026-09-01T00:00:00.000Z',
+      dryRun: false,
+    })
+    expect(parse('--from', '2026-09-01T10:30:00+02:00')).toMatchObject({
+      startDate: '2026-09-01T08:30:00.000Z',
+    })
+  })
+
+  it.each([
+    [[], 'exactly one'],
+    [['--keywords', '1', '--from', '2026-09-01'], 'exactly one'],
+    [['--keywords'], 'needs a value'],
+    [['--keywords', '--dry-run'], 'needs a value'],
+    [['--keywords', '1', '--keywords', '2'], 'more than once'],
+    [['--keywords', '1', '--dry-run', '--dry-run'], 'more than once'],
+    [['--keywords', '1', '--force'], 'unknown argument'],
+    [['--keywords', '0'], 'positive integer'],
+    [['--keywords', '-3'], 'positive integer'],
+    [['--keywords', '1.5'], 'positive integer'],
+    [['--keywords', '1,,2'], 'positive integer'],
+    [['--keywords', 'abc'], 'positive integer'],
+    [['--keywords', '2147483648'], 'positive integer'],
+    [['--from', 'yesterday'], 'expected YYYY-MM-DD'],
+    [['--from', '2026-09-01T10:00:00'], 'expected YYYY-MM-DD'],
+    [['--from', '2026-02-30'], 'not a real date'],
+    [['--from', '2026-13-01'], 'not a real date'],
+    [['--from', '2026-10-17'], 'in the future'],
+    [['--from', '2025-10-16T11:59:59Z'], 'more than 1 year back'],
+    [['--from', '2026-09-01T24:00Z'], 'expected YYYY-MM-DD'],
+    [['--from', '2026-09-01T10:60Z'], 'expected YYYY-MM-DD'],
+  ])('rejects %j', (argv, message) => {
+    expect(() => parse(...argv)).toThrow(message)
+  })
+})
+
+describe('planOctolensBackfill', () => {
+  it('queues only keywords that are not already in the backfill queue', () => {
+    const plan = planOctolensBackfill(parse('--keywords', '7,9'), state())
+
+    expect(plan).toMatchObject({ mode: 'keywords', keywordIdsToQueue: [9], alreadyQueued: [7] })
+  })
+
+  it('queues keywords whose only queue entry is one the poll would drop', () => {
+    const watermark = {
+      startDate: '2026-10-16T11:55:00.000Z',
+      backfill: [{ keywordId: 7 }, { keywordId: 8, cursor: 42 }, { keywordId: 9, cursor: null }],
+    }
+    const plan = planOctolensBackfill(
+      parse('--keywords', '7,8,9'),
+      state({ units: [unit({ watermark })] }),
+    )
+
+    expect(plan).toMatchObject({ keywordIdsToQueue: [7, 8], alreadyQueued: [9] })
+  })
+
+  it('accepts --from exactly one year back', () => {
+    expect(parse('--from', '2025-10-16T12:00:00Z')).toMatchObject({
+      startDate: '2025-10-16T12:00:00.000Z',
+    })
+  })
+
+  it('refuses a keyword mapped to more than one subproject', () => {
+    const counts = new Map([
+      [9, 1],
+      [8, 2],
+    ])
+
+    expect(() =>
+      planOctolensBackfill(parse('--keywords', '9,8'), state({ activeMappingCounts: counts })),
+    ).toThrow('keyword 8 is mapped to more than one subproject')
+  })
+
+  it('refuses keywords without an active subproject mapping', () => {
+    expect(() => planOctolensBackfill(parse('--keywords', '9,42,43'), state())).toThrow(
+      'keywords without an active subproject mapping: 42, 43',
+    )
+  })
+
+  it('rewinds to a date earlier than the current start date', () => {
+    const plan = planOctolensBackfill(parse('--from', '2026-09-01'), state())
+
+    expect(plan).toMatchObject({
+      mode: 'from',
+      startDate: '2026-09-01T00:00:00.000Z',
+      previousStartDate: '2026-10-16T11:55:00.000Z',
+    })
+  })
+
+  it('refuses a --from that would move the start date forward', () => {
+    const current = unit({ watermark: { startDate: '2026-09-01T00:00:00.000Z' } })
+
+    expect(() =>
+      planOctolensBackfill(parse('--from', '2026-09-02'), state({ units: [current] })),
+    ).toThrow('not earlier than the current start date')
+  })
+
+  it('accepts any past --from on a unit that has never run', () => {
+    const fresh = unit({ watermark: null })
+
+    expect(
+      planOctolensBackfill(parse('--from', '2026-09-02'), state({ units: [fresh] })),
+    ).toMatchObject({ previousStartDate: null })
+  })
+
+  it.each([
+    [{ units: [] }, 'exactly one octolens mentionPoll sync unit, found 0'],
+    [{ units: [unit(), unit({ id: 'unit-2' })] }, 'found 2'],
+    [{ units: [unit({ status: 'dead_letter' })] }, 'is dead_letter, not active'],
+    [{ units: [unit({ status: 'paused' })] }, 'is paused, not active'],
+  ])('refuses state %j', (overrides, message) => {
+    expect(() => planOctolensBackfill(parse('--keywords', '9'), state(overrides))).toThrow(message)
+  })
+})
diff --git a/services/apps/connectors_worker/src/octolensBackfill.ts b/services/apps/connectors_worker/src/octolensBackfill.ts
new file mode 100644
index 0000000000..58cb9a1447
--- /dev/null
+++ b/services/apps/connectors_worker/src/octolensBackfill.ts
@@ -0,0 +1,171 @@
+import type { IOctolensMentionPollUnit } from '@crowd/data-access-layer/src/octolens'
+
+const MAX_POSTGRES_INT = 2_147_483_647
+const MAX_REWIND_YEARS = 1
+const ISO_DATE =
+  /^(\d{4})-(\d{2})-(\d{2})(?:T(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d{1,3})?)?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d))?$/
+
+export type OctolensBackfillArgs =
+  | { mode: 'keywords'; keywordIds: number[]; dryRun: boolean }
+  | { mode: 'from'; startDate: string; dryRun: boolean }
+
+export type OctolensBackfillPlan =
+  | {
+      mode: 'keywords'
+      unit: IOctolensMentionPollUnit
+      keywordIdsToQueue: number[]
+      alreadyQueued: number[]
+    }
+  | {
+      mode: 'from'
+      unit: IOctolensMentionPollUnit
+      startDate: string
+      previousStartDate: string | null
+    }
+
+export interface IOctolensBackfillState {
+  units: IOctolensMentionPollUnit[]
+  activeMappingCounts: Map<number, number>
+}
+
+export const OCTOLENS_BACKFILL_USAGE = [
+  'Usage: backfill-octolens-mentions (--keywords <id,id,...> | --from <ISO date>) [--dry-run]',
+  '  --from takes YYYY-MM-DD (UTC midnight) or an ISO timestamp with a zone, at most 1 year back.',
+  '  Exit codes: 0 done or nothing to do, 1 invalid input or state, 2 a poll run is in progress (retry).',
+].join('\n')
+
+export function parseOctolensBackfillArgs(argv: string[], now: Date): OctolensBackfillArgs {
+  const values = new Map<string, string>()
+  let dryRun = false
+
+  const args = argv.filter((arg) => arg !== '--')
+  for (let i = 0; i < args.length; i++) {
+    const arg = args[i]
+    if (arg === '--dry-run') {
+      if (dryRun) throw new Error('--dry-run given more than once')
+      dryRun = true
+      continue
+    }
+    if (arg !== '--keywords' && arg !== '--from') {
+      throw new Error(`unknown argument "${arg}"`)
+    }
+    if (values.has(arg)) throw new Error(`${arg} given more than once`)
+    const value = args[i + 1]
+    if (value === undefined || value.startsWith('--')) throw new Error(`${arg} needs a value`)
+    values.set(arg, value)
+    i++
+  }
+
+  const keywords = values.get('--keywords')
+  const from = values.get('--from')
+  if ((keywords === undefined) === (from === undefined)) {
+    throw new Error('pass exactly one of --keywords or --from')
+  }
+
+  if (keywords !== undefined) {
+    return { mode: 'keywords', keywordIds: parseKeywordIds(keywords), dryRun }
+  }
+  return { mode: 'from', startDate: parseStartDate(from as string, now), dryRun }
+}
+
+// Repeats collapse to one entry so the plan and its log list each keyword once.
+function parseKeywordIds(raw: string): number[] {
+  const ids = raw.split(',').map((part) => {
+    const trimmed = part.trim()
+    const id = Number(trimmed)
+    if (!/^\d+$/.test(trimmed) || id < 1 || id > MAX_POSTGRES_INT) {
+      throw new Error(`invalid keyword id "${part}": expected a positive integer`)
+    }
+    return id
+  })
+  return [...new Set(ids)]
+}
+
+function parseStartDate(raw: string, now: Date): string {
+  const match = ISO_DATE.exec(raw)
+  if (!match) {
+    throw new Error(`invalid --from "${raw}": expected YYYY-MM-DD or an ISO timestamp with a zone`)
+  }
+  const [year, month, day] = match.slice(1, 4).map(Number)
+  const calendarDay = new Date(Date.UTC(year, month - 1, day))
+  const date = new Date(raw)
+  if (
+    calendarDay.getUTCFullYear() !== year ||
+    calendarDay.getUTCMonth() !== month - 1 ||
+    calendarDay.getUTCDate() !== day ||
+    Number.isNaN(date.getTime())
+  ) {
+    throw new Error(`invalid --from "${raw}": not a real date`)
+  }
+  if (date.getTime() > now.getTime()) {
+    throw new Error(`invalid --from "${raw}": in the future`)
+  }
+  const earliest = new Date(now)
+  earliest.setUTCFullYear(earliest.getUTCFullYear() - MAX_REWIND_YEARS)
+  if (date.getTime() < earliest.getTime()) {
+    throw new Error(
+      `invalid --from "${raw}": more than ${MAX_REWIND_YEARS} year back (earliest ${earliest.toISOString()})`,
+    )
+  }
+  return date.toISOString()
+}
+
+export function planOctolensBackfill(
+  args: OctolensBackfillArgs,
+  state: IOctolensBackfillState,
+): OctolensBackfillPlan {
+  if (state.units.length !== 1) {
+    throw new Error(
+      `expected exactly one octolens mentionPoll sync unit, found ${state.units.length}`,
+    )
+  }
+  const unit = state.units[0]
+  if (unit.status !== 'active') {
+    throw new Error(
+      `octolens mentionPoll sync unit ${unit.id} is ${unit.status}, not active: fix the unit first`,
+    )
+  }
+
+  if (args.mode === 'keywords') {
+    const unmapped = args.keywordIds.filter((id) => !state.activeMappingCounts.has(id))
+    if (unmapped.length > 0) {
+      throw new Error(`keywords without an active subproject mapping: ${unmapped.join(', ')}`)
+    }
+    const ambiguous = args.keywordIds.find((id) => (state.activeMappingCounts.get(id) ?? 0) > 1)
+    if (ambiguous !== undefined) {
+      throw new Error(`keyword ${ambiguous} is mapped to more than one subproject`)
+    }
+    const queued = new Set(queuedKeywordIds(unit.watermark))
+    return {
+      mode: 'keywords',
+      unit,
+      keywordIdsToQueue: args.keywordIds.filter((id) => !queued.has(id)),
+      alreadyQueued: args.keywordIds.filter((id) => queued.has(id)),
+    }
+  }
+
+  const previousStartDate =
+    typeof unit.watermark?.startDate === 'string' ? unit.watermark.startDate : null
+  if (previousStartDate !== null) {
+    const previous = new Date(previousStartDate).getTime()
+    if (!Number.isNaN(previous) && new Date(args.startDate).getTime() >= previous) {
+      throw new Error(
+        `--from ${args.startDate} is not earlier than the current start date ${previousStartDate}`,
+      )
+    }
+  }
+  return { mode: 'from', unit, startDate: args.startDate, previousStartDate }
+}
+
+function queuedKeywordIds(watermark: Record<string, unknown> | null): number[] {
+  const backfill = watermark?.backfill
+  if (!Array.isArray(backfill)) return []
+  // Matches the poll, which drops entries whose cursor is missing or not a string.
+  return backfill
+    .filter((item) => {
+      const cursor = (item as { cursor?: unknown } | null)?.cursor
+      return cursor === null || typeof cursor === 'string'
+    })
+    .map((item) => (item as { keywordId?: unknown }).keywordId)
+    .filter((id): id is number => Number.isInteger(id))
+}
diff --git a/services/apps/connectors_worker/src/scheduling.ts b/services/apps/connectors_worker/src/scheduling.ts
index ee5c886e07..6d742d16ae 100644
--- a/services/apps/connectors_worker/src/scheduling.ts
+++ b/services/apps/connectors_worker/src/scheduling.ts
@@ -2,7 +2,6 @@ const JITTER_RATIO = 0.1
 const SHORT_DEFER_MIN_MS = 30_000
 const SHORT_DEFER_JITTER_MS = 60_000
 const FAILURE_BACKOFF_BASE_MS = 60_000
-const RUNNING_PROBE_CAP_MS = 5 * 60_000
 const FALLBACK_FAILURE_CAP_MS = 60 * 60_000
 
 function withJitter(delayMs: number): number {
@@ -17,12 +16,6 @@ export function shortDeferRunAt(): Date {
   return new Date(Date.now() + SHORT_DEFER_MIN_MS + Math.random() * SHORT_DEFER_JITTER_MS)
 }
 
-export function runningProbeRunAt(startedAt: Date): Date {
-  const elapsedMs = Math.max(SHORT_DEFER_MIN_MS, Date.now() - startedAt.getTime())
-
-  return new Date(Date.now() + withJitter(Math.min(elapsedMs, RUNNING_PROBE_CAP_MS)))
-}
-
 export function failureRunAt(consecutiveFailures: number, cadenceMinutes: number | null): Date {
   const capMs = cadenceMinutes === null ? FALLBACK_FAILURE_CAP_MS : cadenceMinutes * 60_000
   const backoffMs = FAILURE_BACKOFF_BASE_MS * 2 ** (Math.max(1, consecutiveFailures) - 1)
diff --git a/services/apps/connectors_worker/src/workflows/dispatcher.ts b/services/apps/connectors_worker/src/workflows/dispatcher.ts
index 7cd25d05f2..c13ce7a96c 100644
--- a/services/apps/connectors_worker/src/workflows/dispatcher.ts
+++ b/services/apps/connectors_worker/src/workflows/dispatcher.ts
@@ -36,7 +36,7 @@ export async function dispatcher(): Promise<void> {
   }
 
   for (const unit of deferred) {
-    await activity.deferUnit(unit.id)
+    await activity.deferUnit(unit.id, unit.leaseLapsed)
   }
 
   await activity.logDispatchSummary({
diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts
index 64cfc9654e..35d4a02031 100644
--- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts
+++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts
@@ -729,3 +729,69 @@ describe('runMentionPoll keyword backfill (queued manually)', () => {
     expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] })
   })
 })
+
+describe('runMentionPoll page cap', () => {
+  const START = '2026-10-01T00:00:00.000Z'
+  const QUEUED = [{ keywordId: 7, cursor: null }]
+
+  it('stops the regular poll at 20 pages and keeps the cycle, without starting the backfill', async () => {
+    for (let i = 1; i <= 25; i++) {
+      mockedFetchMentionsPage.mockResolvedValueOnce(page([mention()], `cursor-${i}`))
+    }
+    const { ctx, watermarks } = makeCtx({ watermark: { startDate: START, backfill: QUEUED } })
+
+    expect(await runMentionPoll(ctx)).toEqual({ complete: false, resumeAtCadence: true })
+    expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(20)
+    expect(mockedFetchMentionsPage).not.toHaveBeenCalledWith('test-token', {
+      keywordIds: [7],
+      cursor: undefined,
+    })
+    expect(watermarks[watermarks.length - 1]).toEqual({
+      startDate: START,
+      cursor: 'cursor-20',
+      pendingMaxTimestamp: '2026-10-12T08:00:00.000Z',
+      backfill: QUEUED,
+    })
+  })
+
+  it('resumes the capped cycle from its cursor on the next run, then runs the backfill', async () => {
+    mockedFetchMentionsPage
+      .mockResolvedValueOnce(page([mention()], null))
+      .mockResolvedValueOnce(page([]))
+    const { ctx, watermarks } = makeCtx({
+      watermark: {
+        startDate: START,
+        cursor: 'cursor-20',
+        pendingMaxTimestamp: '2026-10-12T08:00:00.000Z',
+        backfill: QUEUED,
+      },
+    })
+
+    expect(await runMentionPoll(ctx)).toEqual({ complete: true })
+    expect(mockedFetchMentionsPage.mock.calls.map(([, params]) => params)).toEqual([
+      { startDate: START, cursor: 'cursor-20' },
+      { keywordIds: [7], cursor: undefined },
+    ])
+    expect(watermarks[watermarks.length - 1]).toEqual({
+      startDate: '2026-10-12T08:00:00.000Z',
+      cursor: null,
+      pendingMaxTimestamp: null,
+      backfill: [],
+    })
+  })
+
+  it('finishes a cycle that drains exactly on the 20th page', async () => {
+    for (let i = 1; i < 20; i++) {
+      mockedFetchMentionsPage.mockResolvedValueOnce(page([], `cursor-${i}`))
+    }
+    mockedFetchMentionsPage.mockResolvedValueOnce(page([mention()], null))
+    const { ctx, watermarks } = makeCtx({ watermark: { startDate: START } })
+
+    expect(await runMentionPoll(ctx)).toEqual({ complete: true })
+    expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(20)
+    expect(watermarks[watermarks.length - 1]).toMatchObject({
+      startDate: '2026-10-12T08:00:00.000Z',
+      cursor: null,
+    })
+  })
+})
diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts
index c5b6e61831..e6d0516cc0 100644
--- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts
+++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts
@@ -6,8 +6,9 @@ import type { MappedOctolensMention } from '../mapper'
 import { fetchMentionsPage } from '../mentionsClient'
 import type { OctolensMention } from '../mentionsClient'
 
-// Sized for the worst case, every page retried (4 attempts), at the 5-minute cadence: 8 × 4 × 12 =
-// 384 requests/hour, leaving the shared 500/hour budget room for the regular poll and discovery.
+// Worst case per run is 20 + 8 = 28 pages, 336/hour at the 5-minute cadence before retries, against
+// Octolens's shared 500/hour budget; a regular cycle longer than 20 pages resumes next cadence.
+const MAX_POLL_PAGES_PER_RUN = 20
 const MAX_BACKFILL_PAGES_PER_RUN = 8
 
 interface KeywordBackfill {
@@ -148,7 +149,7 @@ export async function runMentionPoll(ctx: SyncContext<OctolensDeps>): Promise<Sy
   }
 
   let polled = false
-  while (ctx.hasRunBudget()) {
+  while (pagesFetched < MAX_POLL_PAGES_PER_RUN && ctx.hasRunBudget()) {
     const page = await fetchMentionsPage(token, { startDate, cursor })
     pagesFetched += 1
 
@@ -176,6 +177,13 @@ export async function runMentionPoll(ctx: SyncContext<OctolensDeps>): Promise<Sy
     })
   }
 
+  if (!polled && pagesFetched >= MAX_POLL_PAGES_PER_RUN) {
+    ctx.log.info(
+      { pagesFetched },
+      'octolens mention poll paused at the page cap until next cadence',
+    )
+    return { complete: false, resumeAtCadence: true }
+  }
   if (!polled) {
     ctx.log.info({ pagesFetched }, 'octolens mention poll deferred: run budget exhausted')
     return { complete: false }
diff --git a/services/libs/connectors/src/types.ts b/services/libs/connectors/src/types.ts
index bcd7b05751..2803342bc5 100644
--- a/services/libs/connectors/src/types.ts
+++ b/services/libs/connectors/src/types.ts
@@ -51,6 +51,8 @@ export interface SyncContext<TDeps = undefined> {
 
 export interface SyncOutcome {
   complete: boolean
+  // Only meaningful with complete: false; the next run then waits for the sync's normal cadence.
+  resumeAtCadence?: boolean
 }
 
 export interface SyncDefinition<TDeps = undefined> {
diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts
index 655797dca7..e8140c33c3 100644
--- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts
+++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts
@@ -4,7 +4,14 @@ import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common'
 import type { QueryExecutor } from '@crowd/database'
 import { withQx } from '@crowd/test-kit/db'
 
-import { listShadowDiffUnits, reviveDeadLetteredSyncUnits, upsertSyncUnits } from './syncUnits'
+import {
+  claimDueUnits,
+  listShadowDiffUnits,
+  reassertUnitClaim,
+  renewRunningUnitLease,
+  reviveDeadLetteredSyncUnits,
+  upsertSyncUnits,
+} from './syncUnits'
 
 const test = withQx(base)
 
@@ -291,3 +298,105 @@ describe('reviveDeadLetteredSyncUnits', () => {
     expect(await reviveDeadLetteredSyncUnits(qx, integrationId, 'mentionPoll')).toBe(0)
   })
 })
+
+async function setClaimState(
+  qx: QueryExecutor,
+  id: string,
+  state: { lockedAgo: string | null; nextRunAgo: string },
+): Promise<void> {
+  await qx.result(
+    `UPDATE integration.sync_units
+     SET "lockedAt" = now() - $(lockedAgo)::interval,
+         "nextRunAt" = now() - $(nextRunAgo)::interval
+     WHERE id = $(id)`,
+    { id, ...state },
+  )
+}
+
+async function readClaimState(
+  qx: QueryExecutor,
+  id: string,
+): Promise<{ lockedAgoSeconds: number | null; nextRunAt: string }> {
+  return qx.selectOne(
+    `SELECT EXTRACT(EPOCH FROM (now() - "lockedAt"))::float8 AS "lockedAgoSeconds", "nextRunAt"
+     FROM integration.sync_units WHERE id = $(id)`,
+    { id },
+  )
+}
+
+describe('renewRunningUnitLease', () => {
+  test('renews the claim and moves the next run to the end of the new lease', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+    const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' })
+    await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '10 minutes' })
+
+    expect(await renewRunningUnitLease(qx, id)).toBe(1)
+
+    const lease: { lockedAgoSeconds: number; leaseSeconds: number } = await qx.selectOne(
+      `SELECT EXTRACT(EPOCH FROM (now() - "lockedAt"))::float8 AS "lockedAgoSeconds",
+              EXTRACT(EPOCH FROM ("nextRunAt" - "lockedAt"))::float8 AS "leaseSeconds"
+       FROM integration.sync_units WHERE id = $(id)`,
+      { id },
+    )
+    expect(lease.lockedAgoSeconds).toBeLessThan(5)
+    expect(lease.leaseSeconds).toBe(300)
+  })
+
+  test('leaves a unit alone once its run has recorded its end', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+    const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' })
+    await setClaimState(qx, id, { lockedAgo: null, nextRunAgo: '-5 minutes' })
+    const before = await readClaimState(qx, id)
+
+    expect(await renewRunningUnitLease(qx, id)).toBe(0)
+
+    expect(await readClaimState(qx, id)).toEqual(before)
+  })
+})
+
+describe('claimDueUnits', () => {
+  test('reports whether the claim took over a lapsed lease', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+    const freeId = await createSyncUnit(qx, integrationId, { channelName: 'a', syncName: 'issues' })
+    const lapsedId = await createSyncUnit(qx, integrationId, {
+      channelName: 'b',
+      syncName: 'issues',
+    })
+    await setClaimState(qx, freeId, { lockedAgo: null, nextRunAgo: '1 minute' })
+    await setClaimState(qx, lapsedId, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' })
+
+    const claimed = await claimDueUnits(qx, 100)
+
+    expect(claimed.find((u) => u.id === freeId)?.leaseLapsed).toBe(false)
+    expect(claimed.find((u) => u.id === lapsedId)?.leaseLapsed).toBe(true)
+    expect((await readClaimState(qx, lapsedId)).lockedAgoSeconds).toBeLessThan(5)
+  })
+
+  test('skips a due unit whose running lease was renewed, and reclaims it once the lease lapses', async ({
+    qx,
+  }) => {
+    const integrationId = await createIntegration(qx)
+    const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' })
+    await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' })
+    await renewRunningUnitLease(qx, id)
+    await setClaimState(qx, id, { lockedAgo: '0 minutes', nextRunAgo: '1 minute' })
+
+    expect((await claimDueUnits(qx, 100)).map((u) => u.id)).not.toContain(id)
+
+    await setClaimState(qx, id, { lockedAgo: '6 minutes', nextRunAgo: '1 minute' })
+    expect((await claimDueUnits(qx, 100)).map((u) => u.id)).toContain(id)
+  })
+})
+
+describe('reassertUnitClaim', () => {
+  test('locks a unit whose lock a previous run cleared', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+    const id = await createSyncUnit(qx, integrationId, { channelName: 'repo', syncName: 'issues' })
+    await setClaimState(qx, id, { lockedAgo: null, nextRunAgo: '1 minute' })
+
+    await reassertUnitClaim(qx, id)
+
+    expect((await readClaimState(qx, id)).lockedAgoSeconds).toBeLessThan(5)
+    expect((await claimDueUnits(qx, 100)).map((u) => u.id)).not.toContain(id)
+  })
+})
diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts
index 3cf5485b6b..207508d81f 100644
--- a/services/libs/data-access-layer/src/connectors/syncUnits.ts
+++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts
@@ -60,8 +60,8 @@ export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise<I
   return qx.select(
     `UPDATE integration.sync_units su
      SET "lockedAt" = now(), "updatedAt" = now()
-     WHERE su.id IN (
-       SELECT su2.id
+     FROM (
+       SELECT su2.id, su2."lockedAt" IS NOT NULL AS "leaseLapsed"
        FROM integration.sync_units su2
        WHERE su2.status = 'active'
          AND su2."nextRunAt" <= now()
@@ -74,8 +74,10 @@ export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise<I
        ORDER BY su2."nextRunAt"
        LIMIT $(limit)
        FOR UPDATE SKIP LOCKED
-     )
-     RETURNING su.id, su."integrationId", su.platform, su."syncName", su."channelId", su."channelName"`,
+     ) due
+     WHERE su.id = due.id
+     RETURNING su.id, su."integrationId", su.platform, su."syncName", su."channelId", su."channelName",
+               due."leaseLapsed"`,
     { limit, leaseMinutes: CLAIM_LEASE_MINUTES },
   )
 }
@@ -93,6 +95,30 @@ export async function rescheduleUnit(
   )
 }
 
+// Keeps a still-running unit claimed so lockedAt stays set for the whole run. A NULL lockedAt
+// means the run already recorded its end after this claim, so its own nextRunAt is kept.
+export async function renewRunningUnitLease(qx: QueryExecutor, id: string): Promise<number> {
+  return qx.result(
+    `UPDATE integration.sync_units
+     SET "lockedAt" = now(),
+         "nextRunAt" = now() + $(leaseMinutes) * interval '1 minute',
+         "updatedAt" = now()
+     WHERE id = $(id) AND "lockedAt" IS NOT NULL`,
+    { id, leaseMinutes: CLAIM_LEASE_MINUTES },
+  )
+}
+
+// A previous run's end write can clear the lock between this run's claim and its start, so
+// every run re-locks before reading its unit and stays locked until its own end write.
+export async function reassertUnitClaim(qx: QueryExecutor, id: string): Promise<void> {
+  await qx.result(
+    `UPDATE integration.sync_units
+     SET "lockedAt" = now(), "updatedAt" = now()
+     WHERE id = $(id)`,
+    { id },
+  )
+}
+
 export async function recordRunSuccess(
   qx: QueryExecutor,
   id: string,
diff --git a/services/libs/data-access-layer/src/connectors/types.ts b/services/libs/data-access-layer/src/connectors/types.ts
index ba841af3bc..6cba5cadac 100644
--- a/services/libs/data-access-layer/src/connectors/types.ts
+++ b/services/libs/data-access-layer/src/connectors/types.ts
@@ -43,7 +43,7 @@ export type SyncUnitUpsert = Pick<
 export type IClaimedUnit = Pick<
   ISyncUnit,
   'id' | 'integrationId' | 'platform' | 'syncName' | 'channelId' | 'channelName'
->
+> & { leaseLapsed: boolean }
 
 export type IShadowDiffUnit = Pick<ISyncUnit, 'id' | 'integrationId' | 'channelName' | 'syncName'>
 
diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts
index 496d1a9284..0ca1652dcc 100644
--- a/services/libs/data-access-layer/src/octolens/index.ts
+++ b/services/libs/data-access-layer/src/octolens/index.ts
@@ -273,3 +273,148 @@ export async function reconcileOctolensKeywordMappings(
     return { deactivated, created, restored, softDeleted }
   })
 }
+
+const MENTION_POLL_SYNC_NAME = 'mentionPoll'
+
+export interface IOctolensMentionPollUnit {
+  id: string
+  integrationId: string
+  status: string
+  lockedAt: Date | null
+  watermark: Record<string, unknown> | null
+}
+
+// 'changed': the unit was idle but its watermark no longer matched what the write was planned on.
+export type OctolensMentionPollWriteOutcome =
+  | 'applied'
+  | 'locked'
+  | 'changed'
+  | 'inactive'
+  | 'missing'
+
+export interface IOctolensMentionPollWriteResult {
+  outcome: OctolensMentionPollWriteOutcome
+  watermark: Record<string, unknown> | null
+}
+
+export async function listOctolensMentionPollUnits(
+  qx: QueryExecutor,
+): Promise<IOctolensMentionPollUnit[]> {
+  return qx.select(
+    `
+      SELECT su.id, su."integrationId", su.status, su."lockedAt", su.watermark
+      FROM integration.sync_units su
+      JOIN public.integrations i ON i.id = su."integrationId"
+      WHERE su.platform = $(platform)
+        AND su."syncName" = $(syncName)
+        AND i.platform = $(platform)
+        AND i."segmentId" IS NULL
+        AND i."deletedAt" IS NULL
+      ORDER BY su.id
+    `,
+    { platform: PlatformType.OCTOLENS, syncName: MENTION_POLL_SYNC_NAME },
+  )
+}
+
+const STORED_WATERMARK = `(CASE WHEN jsonb_typeof(su.watermark) = 'object' THEN su.watermark ELSE '{}'::jsonb END)`
+
+const STORED_BACKFILL = `(CASE WHEN jsonb_typeof(su.watermark->'backfill') = 'array' THEN su.watermark->'backfill' ELSE '[]'::jsonb END)`
+
+// A run keeps the watermark in memory and overwrites it when it ends, so this only writes while
+// the unit is unclaimed; the dispatcher keeps lockedAt set until the run records its end.
+async function updateIdleMentionPollWatermark(
+  qx: QueryExecutor,
+  unitId: string,
+  watermarkSql: string,
+  params: Record<string, unknown>,
+  guardSql = 'true',
+): Promise<IOctolensMentionPollWriteResult> {
+  const filter = { unitId, platform: PlatformType.OCTOLENS, syncName: MENTION_POLL_SYNC_NAME }
+  const updated: { watermark: Record<string, unknown> }[] = await qx.select(
+    `
+      UPDATE integration.sync_units su
+      SET watermark = ${watermarkSql}, "updatedAt" = now()
+      WHERE su.id = $(unitId)
+        AND su.platform = $(platform)
+        AND su."syncName" = $(syncName)
+        AND su.status = 'active'
+        AND su."lockedAt" IS NULL
+        AND ${guardSql}
+      RETURNING su.watermark
+    `,
+    { ...params, ...filter },
+  )
+  if (updated.length > 0) {
+    return { outcome: 'applied', watermark: updated[0].watermark }
+  }
+
+  const unit: {
+    status: string
+    lockedAt: Date | null
+    watermark: Record<string, unknown> | null
+  } | null = await qx.selectOneOrNone(
+    `
+        SELECT su.status, su."lockedAt", su.watermark
+        FROM integration.sync_units su
+        WHERE su.id = $(unitId) AND su.platform = $(platform) AND su."syncName" = $(syncName)
+      `,
+    filter,
+  )
+  if (!unit) {
+    return { outcome: 'missing', watermark: null }
+  }
+  if (unit.status !== 'active') {
+    return { outcome: 'inactive', watermark: unit.watermark }
+  }
+  return { outcome: unit.lockedAt ? 'locked' : 'changed', watermark: unit.watermark }
+}
+
+export async function queueOctolensKeywordBackfills(
+  qx: QueryExecutor,
+  unitId: string,
+  keywordIds: number[],
+): Promise<IOctolensMentionPollWriteResult> {
+  return updateIdleMentionPollWatermark(
+    qx,
+    unitId,
+    `jsonb_set(
+      ${STORED_WATERMARK},
+      '{backfill}',
+      ${STORED_BACKFILL} || COALESCE(
+        (
+          SELECT jsonb_agg(jsonb_build_object('keywordId', k.id, 'cursor', NULL) ORDER BY k.ord)
+          FROM unnest($(keywordIds)::int[]) WITH ORDINALITY AS k(id, ord)
+          WHERE NOT EXISTS (
+            SELECT 1 FROM jsonb_array_elements(${STORED_BACKFILL}) e
+            WHERE e->'keywordId' = to_jsonb(k.id)
+              -- The poll drops entries whose cursor is missing or not a string, so they don't count.
+              AND jsonb_typeof(e->'cursor') IN ('null', 'string')
+          )
+        ),
+        '[]'::jsonb
+      )
+    )`,
+    { keywordIds: [...new Set(keywordIds)] },
+  )
+}
+
+export async function rewindOctolensMentionPollStartDate(
+  qx: QueryExecutor,
+  unitId: string,
+  startDate: string,
+  expectedStartDate: string | null,
+): Promise<IOctolensMentionPollWriteResult> {
+  return updateIdleMentionPollWatermark(
+    qx,
+    unitId,
+    `${STORED_WATERMARK} || jsonb_build_object(
+      'startDate', $(startDate)::text,
+      'cursor', NULL,
+      'pendingMaxTimestamp', NULL
+    )`,
+    { startDate, expectedStartDate },
+    // Only over the start date the rewind was validated against, so a concurrent rewind or the
+    // first poll run can't be silently overwritten.
+    `(${STORED_WATERMARK}->>'startDate') IS NOT DISTINCT FROM $(expectedStartDate)`,
+  )
+}
diff --git a/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts
new file mode 100644
index 0000000000..066a1f79d8
--- /dev/null
+++ b/services/libs/data-access-layer/src/octolens/mentionPollWatermark.test.ts
@@ -0,0 +1,289 @@
+import { test as base, describe, expect } from 'vitest'
+
+import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common'
+import type { QueryExecutor } from '@crowd/database'
+import { withQx } from '@crowd/test-kit/db'
+
+import {
+  listOctolensMentionPollUnits,
+  queueOctolensKeywordBackfills,
+  rewindOctolensMentionPollStartDate,
+} from './index'
+
+const test = withQx(base)
+
+async function createAnchorIntegration(
+  qx: QueryExecutor,
+  overrides: { deletedAt?: string } = {},
+): Promise<string> {
+  const id = generateUUIDv1()
+  await qx.result(
+    `
+    INSERT INTO public.integrations (id, platform, status, "tenantId", "createdAt", "updatedAt", "deletedAt")
+    VALUES ($(id), 'octolens', 'done', $(tenantId), NOW(), NOW(), $(deletedAt)::timestamptz)
+    `,
+    { id, tenantId: DEFAULT_TENANT_ID, deletedAt: overrides.deletedAt ?? null },
+  )
+  return id
+}
+
+async function createMentionPollUnit(
+  qx: QueryExecutor,
+  integrationId: string,
+  overrides: {
+    status?: string
+    locked?: boolean
+    watermark?: Record<string, unknown> | null
+  } = {},
+): Promise<string> {
+  const id = generateUUIDv1()
+  await qx.result(
+    `
+    INSERT INTO integration.sync_units
+      (id, "integrationId", platform, "channelId", "channelName", "syncName", status, "emitEnabled", watermark, "lockedAt")
+    VALUES
+      ($(id), $(integrationId), 'octolens', 'octolens-mentions', 'octolens-mentions', 'mentionPoll',
+       $(status), true, $(watermark)::jsonb, CASE WHEN $(locked) THEN now() ELSE NULL END)
+    `,
+    {
+      id,
+      integrationId,
+      status: overrides.status ?? 'active',
+      locked: overrides.locked ?? false,
+      watermark: overrides.watermark == null ? null : JSON.stringify(overrides.watermark),
+    },
+  )
+  return id
+}
+
+async function readWatermark(qx: QueryExecutor, unitId: string): Promise<unknown> {
+  const row: { watermark: unknown } = await qx.selectOne(
+    `SELECT watermark FROM integration.sync_units WHERE id = $(unitId)`,
+    { unitId },
+  )
+  return row.watermark
+}
+
+const MID_CYCLE = {
+  startDate: '2026-10-16T10:00:00.000Z',
+  cursor: 'c-2',
+  pendingMaxTimestamp: '2026-10-16T10:04:00.000Z',
+  backfill: [{ keywordId: 7, cursor: 'k7-page3' }],
+}
+
+describe('listOctolensMentionPollUnits', () => {
+  test('returns only mention-poll units of live segment-less octolens integrations', async ({
+    qx,
+  }) => {
+    const anchorId = await createAnchorIntegration(qx)
+    const unitId = await createMentionPollUnit(qx, anchorId, { status: 'dead_letter' })
+    const deletedAnchorId = await createAnchorIntegration(qx, { deletedAt: '2026-01-01' })
+    await createMentionPollUnit(qx, deletedAnchorId)
+
+    const units = await listOctolensMentionPollUnits(qx)
+    expect(units.map((u) => [u.id, u.integrationId, u.status])).toEqual([
+      [unitId, anchorId, 'dead_letter'],
+    ])
+  })
+})
+
+describe('queueOctolensKeywordBackfills', () => {
+  test('appends new keywords after the existing queue and skips already-queued ones', async ({
+    qx,
+  }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: MID_CYCLE,
+    })
+
+    const result = await queueOctolensKeywordBackfills(qx, unitId, [9, 7, 8, 9])
+
+    expect(result.outcome).toBe('applied')
+    expect(await readWatermark(qx, unitId)).toEqual({
+      ...MID_CYCLE,
+      backfill: [
+        { keywordId: 7, cursor: 'k7-page3' },
+        { keywordId: 9, cursor: null },
+        { keywordId: 8, cursor: null },
+      ],
+    })
+  })
+
+  test('queues a keyword whose only entry is one the poll would drop', async ({ qx }) => {
+    const malformed = [{ keywordId: 7 }, { keywordId: 8, cursor: 42 }]
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: { ...MID_CYCLE, backfill: malformed },
+    })
+
+    const result = await queueOctolensKeywordBackfills(qx, unitId, [7, 8])
+
+    expect(result.outcome).toBe('applied')
+    expect(await readWatermark(qx, unitId)).toMatchObject({
+      backfill: [...malformed, { keywordId: 7, cursor: null }, { keywordId: 8, cursor: null }],
+    })
+  })
+
+  test('creates the queue on a unit that has never run', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx))
+
+    await queueOctolensKeywordBackfills(qx, unitId, [5])
+
+    expect(await readWatermark(qx, unitId)).toEqual({ backfill: [{ keywordId: 5, cursor: null }] })
+  })
+
+  test('adds the queue to a stored watermark without a backfill key', async ({ qx }) => {
+    const poll = {
+      startDate: MID_CYCLE.startDate,
+      cursor: MID_CYCLE.cursor,
+      pendingMaxTimestamp: MID_CYCLE.pendingMaxTimestamp,
+    }
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: poll,
+    })
+
+    await queueOctolensKeywordBackfills(qx, unitId, [5])
+
+    expect(await readWatermark(qx, unitId)).toEqual({
+      ...poll,
+      backfill: [{ keywordId: 5, cursor: null }],
+    })
+  })
+
+  test('replaces a backfill value that is not an array', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: { ...MID_CYCLE, backfill: { keywordId: 7 } },
+    })
+
+    await queueOctolensKeywordBackfills(qx, unitId, [5])
+
+    expect(await readWatermark(qx, unitId)).toEqual({
+      ...MID_CYCLE,
+      backfill: [{ keywordId: 5, cursor: null }],
+    })
+  })
+
+  test('refuses and writes nothing while the unit is claimed', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      locked: true,
+      watermark: MID_CYCLE,
+    })
+
+    const result = await queueOctolensKeywordBackfills(qx, unitId, [9])
+
+    expect(result.outcome).toBe('locked')
+    expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE)
+  })
+
+  test('refuses a unit that is not active', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      status: 'dead_letter',
+      watermark: MID_CYCLE,
+    })
+
+    const result = await queueOctolensKeywordBackfills(qx, unitId, [9])
+
+    expect(result.outcome).toBe('inactive')
+    expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE)
+  })
+
+  test('reports a unit id that is not a mention-poll unit as missing', async ({ qx }) => {
+    const result = await queueOctolensKeywordBackfills(qx, generateUUIDv1(), [9])
+
+    expect(result.outcome).toBe('missing')
+  })
+})
+
+describe('rewindOctolensMentionPollStartDate', () => {
+  test('sets the start date, clears the cycle progress and keeps the backfill queue', async ({
+    qx,
+  }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: MID_CYCLE,
+    })
+
+    const result = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-09-01T00:00:00.000Z',
+      MID_CYCLE.startDate,
+    )
+
+    expect(result.outcome).toBe('applied')
+    expect(await readWatermark(qx, unitId)).toEqual({
+      startDate: '2026-09-01T00:00:00.000Z',
+      cursor: null,
+      pendingMaxTimestamp: null,
+      backfill: MID_CYCLE.backfill,
+    })
+  })
+
+  test('refuses when the start date changed since the rewind was planned', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      watermark: MID_CYCLE,
+    })
+    const first = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-08-01T00:00:00.000Z',
+      MID_CYCLE.startDate,
+    )
+
+    const second = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-09-01T00:00:00.000Z',
+      MID_CYCLE.startDate,
+    )
+
+    expect(first.outcome).toBe('applied')
+    expect(second.outcome).toBe('changed')
+    expect(await readWatermark(qx, unitId)).toMatchObject({ startDate: '2026-08-01T00:00:00.000Z' })
+  })
+
+  test('rewinds a unit that has never run when it still has no start date', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx))
+
+    const result = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-09-01T00:00:00.000Z',
+      null,
+    )
+
+    expect(result.outcome).toBe('applied')
+    expect(await readWatermark(qx, unitId)).toMatchObject({ startDate: '2026-09-01T00:00:00.000Z' })
+  })
+
+  test('refuses and writes nothing while the unit is claimed', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      locked: true,
+      watermark: MID_CYCLE,
+    })
+
+    const result = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-09-01T00:00:00.000Z',
+      MID_CYCLE.startDate,
+    )
+
+    expect(result.outcome).toBe('locked')
+    expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE)
+  })
+
+  test('refuses a unit that is not active', async ({ qx }) => {
+    const unitId = await createMentionPollUnit(qx, await createAnchorIntegration(qx), {
+      status: 'paused',
+      watermark: MID_CYCLE,
+    })
+
+    const result = await rewindOctolensMentionPollStartDate(
+      qx,
+      unitId,
+      '2026-09-01T00:00:00.000Z',
+      MID_CYCLE.startDate,
+    )
+
+    expect(result.outcome).toBe('inactive')
+    expect(await readWatermark(qx, unitId)).toEqual(MID_CYCLE)
+  })
+})