diff --git a/README.md b/README.md
index 1067b38..f02d8ba 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,15 @@ the QR URL as `source/<name>` (passed through as-is, not defaulted).
 
 `PolicyConfig`: `requireMention` · `dmMode` (`'open' \| 'allowlist' \| 'pair' \| 'disabled'`) · `dmAllowlist` · `groupAllowlist` · `respondToMentionAll` · `botLoopGuard` (see [Bot-at-bot](#bot-at-bot)). `dmAllowlist` takes **sender ids** (`ou_…` / user_id / union_id), `groupAllowlist` takes **chat ids** (`oc_…`) — an app id (`cli_…`) belongs in neither and is warned about.
 
-`SafetyConfig`: `dedup` (`ttl`/`maxEntries`/`sweepIntervalMs`) · `chatQueue` (`enabled`, `mergeWhileBusy`) · `batch.text` / `batch.media` · `staleMessageWindowMs`.
+`SafetyConfig`: `dedup` (`ttl`/`maxEntries`/`sweepIntervalMs`) · `processingLock` (`ttlMs`/`renewIntervalMs`) · `chatQueue` (`enabled`, `mergeWhileBusy`) · `batch.text` / `batch.media` · `staleMessageWindowMs`.
+
+`processingLock` defaults to a 300,000 ms TTL and a 60,000 ms renewal interval. Both
+values must be integer milliseconds from 1 through 2,147,483,647, and
+`renewIntervalMs` must be less than `ttlMs`. If only `ttlMs` is overridden, the
+renewal interval is derived as the smaller of 60,000 ms and one third of the TTL
+(rounded down, with a 1 ms minimum). Lease ownership is token-bound: an active or
+finalizing handler cannot be displaced merely because wall-clock time has passed
+its TTL.
 
 ### Lifecycle
 
@@ -568,4 +576,3 @@ guarantee.
 ## License
 
 MIT
-
diff --git a/README.zh.md b/README.zh.md
index 446691f..597636b 100644
--- a/README.zh.md
+++ b/README.zh.md
@@ -103,7 +103,13 @@ const channel = createLarkChannel({ appId: client_id, appSecret: client_secret }
 
 `PolicyConfig`：`requireMention` · `dmMode`（`'open' \| 'allowlist' \| 'pair' \| 'disabled'`）· `dmAllowlist` · `groupAllowlist` · `respondToMentionAll` · `botLoopGuard`（见 [Bot-at-bot](#bot-at-bot)）。`dmAllowlist` 填**发送方 id**（`ou_…` / user_id / union_id），`groupAllowlist` 填**群 id**（`oc_…`）——应用 id（`cli_…`）两者都不属于，填了会告警。
 
-`SafetyConfig`：`dedup`（`ttl`/`maxEntries`/`sweepIntervalMs`）· `chatQueue`（`enabled`、`mergeWhileBusy`）· `batch.text` / `batch.media` · `staleMessageWindowMs`。
+`SafetyConfig`：`dedup`（`ttl`/`maxEntries`/`sweepIntervalMs`）· `processingLock`（`ttlMs`/`renewIntervalMs`）· `chatQueue`（`enabled`、`mergeWhileBusy`）· `batch.text` / `batch.media` · `staleMessageWindowMs`。
+
+`processingLock` 默认 TTL 为 300,000 ms，续租间隔为 60,000 ms。两者都必须是
+1 到 2,147,483,647 范围内的整数毫秒，并且 `renewIntervalMs` 必须小于
+`ttlMs`。如果只覆盖 `ttlMs`，续租间隔取 60,000 ms 与 TTL 三分之一向下取整
+后的较小值（最小 1 ms）。lease owner 由 token 绑定：active 或 finalizing
+handler 不会仅因墙上时钟越过 TTL 而被其他请求抢占。
 
 ### 生命周期
 
diff --git a/package.json b/package.json
index fa210a6..c381bfb 100644
--- a/package.json
+++ b/package.json
@@ -49,6 +49,7 @@
     "lint": "biome check .",
     "format": "biome check --write .",
     "example": "tsx",
+    "prepare": "npm run build",
     "prepublishOnly": "pnpm typecheck && pnpm test && pnpm build"
   },
   "dependencies": {
diff --git a/src/__tests__/l1l3-sink.test.ts b/src/__tests__/l1l3-sink.test.ts
index 0eb5058..3ca5a73 100644
--- a/src/__tests__/l1l3-sink.test.ts
+++ b/src/__tests__/l1l3-sink.test.ts
@@ -11,6 +11,7 @@
 import type { Logger, WSConnectionStatus } from '../internal';
 import { startKeepalive } from '../keepalive';
 import { ChatPipeline } from '../safety/chat-pipeline';
+import type { ProcessingLease } from '../safety/processing-lock';
 import type { BatchConfig, BatchedDispatch } from '../safety/types';
 import type { NormalizedMessage } from '../types';
 
@@ -22,6 +23,10 @@ const silent: Logger = {
   trace: () => {},
 };
 
+function lease(id: string): ProcessingLease {
+  return Object.freeze({ id, ownerToken: Symbol(id) });
+}
+
 function msg(id: string, content = id): NormalizedMessage {
   return {
     messageId: id,
@@ -65,19 +70,19 @@ describe('ChatPipeline mergeWhileBusy', () => {
       return Promise.resolve();
     };
 
-    p.push(msg('a'), handler); // flush #1 starts, pipeline now busy
-    p.push(msg('b'), handler); // accumulate while busy
-    p.push(msg('c'), handler); // accumulate while busy
+    p.push(msg('a'), lease('a'), handler); // flush #1 starts, pipeline now busy
+    p.push(msg('b'), lease('b'), handler); // accumulate while busy
+    p.push(msg('c'), lease('c'), handler); // accumulate while busy
     await Promise.resolve();
     // Only the first batch has dispatched so far.
     expect(batches).toHaveLength(1);
-    expect(batches[0].sourceIds).toEqual(['a']);
+    expect(batches[0].sources.map((source) => source.messageId)).toEqual(['a']);
 
     releaseFirst(); // first handler resolves → settle hook flushes b+c
     await p.flushNow();
 
     expect(batches).toHaveLength(2);
-    expect(batches[1].sourceIds).toEqual(['b', 'c']);
+    expect(batches[1].sources.map((source) => source.messageId)).toEqual(['b', 'c']);
     expect(batches[1].message.content).toBe('b\n\nc');
   });
 
@@ -88,10 +93,10 @@ describe('ChatPipeline mergeWhileBusy', () => {
       batches.push(d);
       return Promise.resolve();
     };
-    p.push(msg('a'), handler);
-    p.push(msg('b'), handler);
+    p.push(msg('a'), lease('a'), handler);
+    p.push(msg('b'), lease('b'), handler);
     await p.flushNow();
-    expect(batches.map((b) => b.sourceIds)).toEqual([['a'], ['b']]);
+    expect(batches.map((b) => b.sources.map((source) => source.messageId))).toEqual([['a'], ['b']]);
   });
 });
 
diff --git a/src/__tests__/message-handler-error.test.ts b/src/__tests__/message-handler-error.test.ts
new file mode 100644
index 0000000..41208d6
--- /dev/null
+++ b/src/__tests__/message-handler-error.test.ts
@@ -0,0 +1,147 @@
+import {
+  createTestChannel,
+  dispatchEvent,
+  flushMicrotasks,
+  markConnected,
+} from '../meeting/__tests__/fixtures';
+
+function directMessage(messageId: string): unknown {
+  return {
+    sender: { sender_id: { open_id: 'ou_sender' }, sender_type: 'user' },
+    message: {
+      message_id: messageId,
+      chat_id: 'oc_dm',
+      chat_type: 'p2p',
+      message_type: 'text',
+      content: '{"text":"hello"}',
+      create_time: String(Date.now()),
+    },
+  };
+}
+
+async function flushEventLoop(): Promise<void> {
+  await flushMicrotasks();
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  await flushMicrotasks();
+}
+
+describe.each([
+  ['queue disabled', false],
+  ['queue enabled', true],
+])('message handler errors with %s', (_label, queueEnabled) => {
+  test("surface through channel.on('error') exactly once", async () => {
+    const { ch } = createTestChannel({
+      safety: {
+        chatQueue: { enabled: queueEnabled },
+        batch: { text: { delayMs: 0 } },
+      },
+    });
+    const errors: unknown[] = [];
+    markConnected(ch);
+    ch.on('error', (error: unknown) => errors.push(error));
+    ch.on('message', async () => {
+      throw new Error('message handler exploded');
+    });
+
+    await dispatchEvent(ch, 'im.message.receive_v1', directMessage(`om_error_${queueEnabled}`));
+    await flushMicrotasks();
+
+    expect(errors).toHaveLength(1);
+    expect((errors[0] as Error).message).toContain('message handler exploded');
+  });
+});
+
+test('a throwing error observer cannot break handler cleanup', async () => {
+  const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } });
+  let handlerCalls = 0;
+  markConnected(ch);
+  ch.on('error', () => {
+    throw new Error('observer exploded');
+  });
+  ch.on('message', async () => {
+    handlerCalls++;
+    throw new Error('message handler exploded');
+  });
+
+  const raw = directMessage('om_observer_throw');
+  await dispatchEvent(ch, 'im.message.receive_v1', raw);
+  await flushMicrotasks();
+  await dispatchEvent(ch, 'im.message.receive_v1', raw);
+  await flushMicrotasks();
+
+  expect(handlerCalls).toBe(1);
+  expect(
+    logger.error.mock.calls.some(
+      ([entry]) =>
+        Array.isArray(entry) &&
+        entry[0] === 'channel: error handler threw' &&
+        entry[1]?.message === 'observer exploded',
+    ),
+  ).toBe(true);
+});
+
+test('an async rejecting error observer is consumed without unhandledRejection', async () => {
+  const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } });
+  const unhandled: unknown[] = [];
+  const onUnhandled = (reason: unknown) => unhandled.push(reason);
+  let handlerCalls = 0;
+  let observerCalls = 0;
+  process.on('unhandledRejection', onUnhandled);
+
+  try {
+    markConnected(ch);
+    ch.on('error', async () => {
+      observerCalls++;
+      throw new Error('async observer exploded');
+    });
+    ch.on('message', async () => {
+      handlerCalls++;
+      if (handlerCalls === 1) throw new Error('first message exploded');
+    });
+
+    await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_async_observer_1'));
+    await flushEventLoop();
+    await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_async_observer_2'));
+    await flushEventLoop();
+
+    expect(observerCalls).toBe(1);
+    expect(handlerCalls).toBe(2);
+    expect(unhandled).toEqual([]);
+    expect(
+      logger.error.mock.calls.some(
+        ([entry]) =>
+          Array.isArray(entry) &&
+          entry[0] === 'channel: error handler threw' &&
+          entry[1]?.message === 'async observer exploded',
+      ),
+    ).toBe(true);
+  } finally {
+    process.off('unhandledRejection', onUnhandled);
+  }
+});
+
+test('a throwing thenable returned by the error observer is isolated', async () => {
+  const { ch, logger } = createTestChannel({ safety: { chatQueue: { enabled: false } } });
+  markConnected(ch);
+  ch.on('error', (() =>
+    Object.defineProperty({}, 'then', {
+      get() {
+        throw new Error('then getter exploded');
+      },
+    })) as never);
+  ch.on('message', async () => {
+    throw new Error('message handler exploded');
+  });
+
+  await dispatchEvent(ch, 'im.message.receive_v1', directMessage('om_thenable_observer'));
+  await flushEventLoop();
+
+  expect(
+    logger.error.mock.calls.some(
+      ([entry]) =>
+        Array.isArray(entry) &&
+        entry[0] === 'channel: error handler threw' &&
+        entry[1]?.message === 'then getter exploded',
+    ),
+  ).toBe(true);
+});
diff --git a/src/channel.ts b/src/channel.ts
index 7eb2893..cf0c596 100644
--- a/src/channel.ts
+++ b/src/channel.ts
@@ -208,6 +208,7 @@ export class LarkChannel {
         const handler = this.handlers.message;
         if (handler) await handler(merged);
       },
+      onError: (error) => this.emitError(error),
     });
   }
 
@@ -1349,8 +1350,25 @@ export class LarkChannel {
             cause: e,
           });
     const handler = this.handlers.error;
-    if (handler) handler(err);
-    else this.logger.error?.('channel: unhandled error', err);
+    if (!handler) {
+      this.logger.error?.('channel: unhandled error', err);
+      return;
+    }
+    try {
+      void Promise.resolve(handler(err)).catch((observerError) => {
+        this.logErrorObserverFailure(observerError);
+      });
+    } catch (observerError) {
+      this.logErrorObserverFailure(observerError);
+    }
+  }
+
+  private logErrorObserverFailure(observerError: unknown): void {
+    try {
+      this.logger.error?.('channel: error handler threw', observerError);
+    } catch {
+      /* an observer failure must never escape through its logger */
+    }
   }
 }
 
diff --git a/src/safety/__tests__/chat-pipeline.test.ts b/src/safety/__tests__/chat-pipeline.test.ts
index 3e7d04a..9fd0043 100644
--- a/src/safety/__tests__/chat-pipeline.test.ts
+++ b/src/safety/__tests__/chat-pipeline.test.ts
@@ -1,7 +1,12 @@
 import type { NormalizedMessage } from '../../types';
 import { ChatPipeline, ChatPipelineManager } from '../chat-pipeline';
+import type { ProcessingLease } from '../processing-lock';
 import { DEFAULT_BATCH } from '../types';
 
+function lease(id: string): ProcessingLease {
+  return Object.freeze({ id, ownerToken: Symbol(id) });
+}
+
 function makeMsg(id: string, content: string, chatId = 'oc_test'): NormalizedMessage {
   return {
     messageId: id,
@@ -24,42 +29,45 @@ async function flushTimers(ms: number): Promise<void> {
 
 describe('ChatPipeline — push (batching)', () => {
   test('single message flushes after debounce delay', async () => {
-    const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = [];
+    const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> =
+      [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 50 }, false);
-    p.push(makeMsg('m1', 'hello'), async (b) => {
+    p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => {
       flushes.push(b);
     });
     await flushTimers(80);
     expect(flushes).toHaveLength(1);
-    expect(flushes[0].sourceIds).toEqual(['m1']);
+    expect(flushes[0].sources.map((source) => source.messageId)).toEqual(['m1']);
     expect(flushes[0].message.content).toBe('hello');
   });
 
   test('rapid messages within window merge into one batch', async () => {
-    const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = [];
+    const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> =
+      [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 50 }, false);
-    p.push(makeMsg('m1', 'hello'), async (b) => {
+    p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => {
       flushes.push(b);
     });
-    p.push(makeMsg('m2', 'world'), async (b) => {
+    p.push(makeMsg('m2', 'world'), lease('m2'), async (b) => {
       flushes.push(b);
     });
-    p.push(makeMsg('m3', 'foo'), async (b) => {
+    p.push(makeMsg('m3', 'foo'), lease('m3'), async (b) => {
       flushes.push(b);
     });
     await flushTimers(100);
     expect(flushes).toHaveLength(1);
-    expect(flushes[0].sourceIds).toEqual(['m1', 'm2', 'm3']);
+    expect(flushes[0].sources.map((source) => source.messageId)).toEqual(['m1', 'm2', 'm3']);
     expect(flushes[0].message.content).toBe('hello\n\nworld\n\nfoo');
   });
 
   test('maxMessages forces flush', async () => {
-    const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = [];
+    const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> =
+      [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 10_000, maxMessages: 2 }, false);
-    p.push(makeMsg('m1', 'a'), async (b) => {
+    p.push(makeMsg('m1', 'a'), lease('m1'), async (b) => {
       flushes.push(b);
     });
-    p.push(makeMsg('m2', 'b'), async (b) => {
+    p.push(makeMsg('m2', 'b'), lease('m2'), async (b) => {
       flushes.push(b);
     });
     await flushTimers(50);
@@ -67,9 +75,10 @@ describe('ChatPipeline — push (batching)', () => {
   });
 
   test('maxChars forces flush', async () => {
-    const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = [];
+    const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> =
+      [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 10_000, maxChars: 5 }, false);
-    p.push(makeMsg('m1', 'hello'), async (b) => {
+    p.push(makeMsg('m1', 'hello'), lease('m1'), async (b) => {
       flushes.push(b);
     });
     await flushTimers(50);
@@ -77,9 +86,10 @@ describe('ChatPipeline — push (batching)', () => {
   });
 
   test('serial-only mode flushes immediately', async () => {
-    const flushes: { message: NormalizedMessage; sourceIds: string[] }[] = [];
+    const flushes: Array<{ message: NormalizedMessage; sources: Array<{ messageId: string }> }> =
+      [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 1000 }, true);
-    p.push(makeMsg('m1', 'a'), async (b) => {
+    p.push(makeMsg('m1', 'a'), lease('m1'), async (b) => {
       flushes.push(b);
     });
     await flushTimers(20);
@@ -98,9 +108,9 @@ describe('ChatPipeline — serialization', () => {
       order.push(`${id}-end`);
     };
 
-    p.push(makeMsg('m1', 'a'), slow('m1', 40));
+    p.push(makeMsg('m1', 'a'), lease('m1'), slow('m1', 40));
     await flushTimers(30);
-    p.push(makeMsg('m2', 'b'), slow('m2', 20));
+    p.push(makeMsg('m2', 'b'), lease('m2'), slow('m2', 20));
     await flushTimers(200);
 
     // m1 must complete before m2 starts
@@ -115,7 +125,7 @@ describe('ChatPipeline — serialization', () => {
     const order: string[] = [];
     const p = new ChatPipeline({ ...DEFAULT_BATCH, delayMs: 30 }, false);
 
-    p.push(makeMsg('m1', 'a'), async () => {
+    p.push(makeMsg('m1', 'a'), lease('m1'), async () => {
       await flushTimers(50);
       order.push('batch');
     });
@@ -147,7 +157,7 @@ describe('ChatPipelineManager', () => {
     const mgr = new ChatPipelineManager({ ...DEFAULT_BATCH, delayMs: 10 });
     const order: string[] = [];
 
-    mgr.push('A', makeMsg('m1', 'a'), async () => {
+    mgr.push('A', makeMsg('m1', 'a'), lease('m1'), async () => {
       await flushTimers(30);
       order.push('push');
     });
diff --git a/src/safety/__tests__/processing-lock.test.ts b/src/safety/__tests__/processing-lock.test.ts
index c74319b..538d172 100644
--- a/src/safety/__tests__/processing-lock.test.ts
+++ b/src/safety/__tests__/processing-lock.test.ts
@@ -1,35 +1,149 @@
-import { ProcessingLock } from '../processing-lock';
+import { type ProcessingLease, ProcessingLock } from '../processing-lock';
+
+function expectLease(lease: ProcessingLease | undefined): ProcessingLease {
+  expect(lease).toBeDefined();
+  return lease as ProcessingLease;
+}
 
 describe('ProcessingLock', () => {
   let lock: ProcessingLock;
-  afterEach(() => lock?.dispose());
+  afterEach(() => {
+    lock?.dispose();
+    vi.useRealTimers();
+  });
 
-  test('first acquire succeeds, second fails until release', () => {
+  test('first acquire returns a lease and competing acquire fails until exact release', () => {
     lock = new ProcessingLock();
-    expect(lock.acquire('m1')).toBe(true);
-    expect(lock.acquire('m1')).toBe(false);
-    lock.release('m1');
-    expect(lock.acquire('m1')).toBe(true);
+    const first = expectLease(lock.acquire('m1'));
+    expect(first.id).toBe('m1');
+    expect(typeof first.ownerToken).toBe('symbol');
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(first);
+    expect(lock.acquire('m1')).toBeDefined();
   });
 
   test('different ids are independent', () => {
     lock = new ProcessingLock();
-    expect(lock.acquire('m1')).toBe(true);
-    expect(lock.acquire('m2')).toBe(true);
-    expect(lock.acquire('m1')).toBe(false);
-    expect(lock.acquire('m2')).toBe(false);
+    expect(lock.acquire('m1')).toBeDefined();
+    expect(lock.acquire('m2')).toBeDefined();
+    expect(lock.acquire('m1')).toBeUndefined();
+    expect(lock.acquire('m2')).toBeUndefined();
+  });
+
+  test('renews an acquired lease across multiple original ttl periods', async () => {
+    vi.useFakeTimers();
+    lock = new ProcessingLock(50, 10);
+    const lease = expectLease(lock.acquire('m1'));
+    await vi.advanceTimersByTimeAsync(120);
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.stopRenewal(lease);
+    await vi.advanceTimersByTimeAsync(51);
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(lease);
+    expect(lock.acquire('m1')).toBeDefined();
+  });
+
+  test('stale owner release cannot delete a replacement lease', () => {
+    lock = new ProcessingLock();
+    const stale = expectLease(lock.acquire('m1'));
+    lock.release(stale);
+    const current = expectLease(lock.acquire('m1'));
+
+    lock.release(stale);
+
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(current);
+  });
+
+  test('stale owner stopRenewal cannot stop a replacement lease', () => {
+    vi.useFakeTimers();
+    lock = new ProcessingLock(50, 10);
+    const stale = expectLease(lock.acquire('m1'));
+    lock.release(stale);
+    const current = expectLease(lock.acquire('m1'));
+
+    lock.stopRenewal(stale);
+
+    expect((lock as any).locks.get('m1').state).toBe('active');
+    expect(vi.getTimerCount()).toBe(1);
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(current);
   });
 
-  test('expires after ttl', async () => {
-    lock = new ProcessingLock(50);
-    expect(lock.acquire('m1')).toBe(true);
-    expect(lock.acquire('m1')).toBe(false);
-    await new Promise((r) => setTimeout(r, 80));
-    expect(lock.acquire('m1')).toBe(true);
+  test('active owner remains exclusive after fake clock passes ttl', () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000);
+    lock = new ProcessingLock(50, 10);
+    const lease = expectLease(lock.acquire('m1'));
+
+    vi.setSystemTime(2_000);
+
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(lease);
   });
 
-  test('release of non-held id is a no-op', () => {
+  test('multiple active ids renew independently', async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000);
+    lock = new ProcessingLock(50, 10);
+    const first = expectLease(lock.acquire('m1'));
+    const second = expectLease(lock.acquire('m2'));
+    await vi.advanceTimersByTimeAsync(10);
+    const firstExpiry = (lock as any).locks.get('m1').expiresAt;
+    const secondExpiry = (lock as any).locks.get('m2').expiresAt;
+
+    lock.stopRenewal(first);
+    await vi.advanceTimersByTimeAsync(20);
+
+    expect((lock as any).locks.get('m1').expiresAt).toBe(firstExpiry);
+    expect((lock as any).locks.get('m2').expiresAt).toBeGreaterThan(secondExpiry);
+    expect(lock.acquire('m1')).toBeUndefined();
+    expect(lock.acquire('m2')).toBeUndefined();
+    lock.release(first);
+    lock.release(second);
+  });
+
+  test('finalizing lease remains exclusive until exact release', () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000);
+    lock = new ProcessingLock(50, 10);
+    const lease = expectLease(lock.acquire('m1'));
+    lock.stopRenewal(lease);
+    vi.setSystemTime(2_000);
+
+    expect(lock.acquire('m1')).toBeUndefined();
+    lock.release(lease);
+    expect(lock.acquire('m1')).toBeDefined();
+  });
+
+  test('release of a non-current lease is a no-op', () => {
     lock = new ProcessingLock();
-    expect(() => lock.release('unknown')).not.toThrow();
+    const unknown = Object.freeze({ id: 'unknown', ownerToken: Symbol('unknown') });
+    expect(() => lock.release(unknown)).not.toThrow();
+  });
+
+  test('dispose clears the renewal timer', () => {
+    vi.useFakeTimers();
+    lock = new ProcessingLock(50, 10);
+    lock.acquire('m1');
+    expect(vi.getTimerCount()).toBe(1);
+    lock.dispose();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  test.each([
+    [0, 1],
+    [-1, 1],
+    [Number.NaN, 1],
+    [Number.POSITIVE_INFINITY, 1],
+    [50.5, 10],
+    [50, 10.5],
+    [2_147_483_648, 10],
+    [2_147_483_647, 2_147_483_648],
+    [50, 0],
+    [50, 50],
+    [50, 60],
+  ])('invalid ttl/renew config fails loudly (%s, %s)', (ttlMs, renewIntervalMs) => {
+    expect(() => new ProcessingLock(ttlMs, renewIntervalMs)).toThrow(RangeError);
   });
 });
diff --git a/src/safety/__tests__/renewable-lock-pipeline.test.ts b/src/safety/__tests__/renewable-lock-pipeline.test.ts
new file mode 100644
index 0000000..bf03309
--- /dev/null
+++ b/src/safety/__tests__/renewable-lock-pipeline.test.ts
@@ -0,0 +1,260 @@
+import type { NormalizedMessage } from '../../types';
+import { SafetyPipeline } from '../index';
+
+const logger = {
+  info: vi.fn(),
+  warn: vi.fn(),
+  error: vi.fn(),
+  debug: vi.fn(),
+  trace: vi.fn(),
+} as any;
+
+afterEach(() => vi.useRealTimers());
+
+function deferred(): { promise: Promise<void>; resolve: () => void } {
+  let resolve!: () => void;
+  const promise = new Promise<void>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+function message(messageId: string, chatId = 'oc_lock'): NormalizedMessage {
+  return {
+    messageId,
+    chatId,
+    chatType: 'p2p',
+    senderId: 'ou_sender',
+    content: messageId,
+    rawContentType: 'text',
+    resources: [],
+    mentions: [],
+    mentionAll: false,
+    mentionedBot: false,
+    createTime: Date.now(),
+  };
+}
+
+function makeCache() {
+  const values = new Map<string, string>();
+  return {
+    values,
+    cache: {
+      get: async (key: string) => values.get(key),
+      set: async (key: string, value: string) => {
+        values.set(key, value);
+        return true;
+      },
+    } as any,
+  };
+}
+
+async function wait(ms: number): Promise<void> {
+  await new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+test('handler pending across two ttl periods retains lock, then marks seen and releases', async () => {
+  const gate = deferred();
+  const { cache, values } = makeCache();
+  let handlerCalls = 0;
+  const pipeline = new SafetyPipeline({
+    cache,
+    logger,
+    onReject: () => {},
+    onMessage: async () => {
+      handlerCalls++;
+      await gate.promise;
+    },
+    config: {
+      chatQueue: { enabled: false },
+      processingLock: { ttlMs: 20, renewIntervalMs: 5 },
+    },
+  });
+
+  await pipeline.pushMessage(message('om_renew'));
+  await wait(50);
+  await pipeline.pushMessage(message('om_renew'));
+  expect(handlerCalls).toBe(1);
+
+  gate.resolve();
+  await wait(0);
+  expect(values.has('om_renew')).toBe(true);
+  const reacquired = (pipeline as any).lock.acquire('om_renew');
+  expect(reacquired).toBeDefined();
+  (pipeline as any).lock.release(reacquired);
+
+  await pipeline.pushMessage(message('om_renew'));
+  await wait(0);
+  expect(handlerCalls).toBe(1);
+  await pipeline.dispose();
+});
+
+test('event-loop stall beyond ttl cannot create a second handler lease', async () => {
+  vi.useFakeTimers();
+  vi.setSystemTime(1_000);
+  const gate = deferred();
+  const { cache } = makeCache();
+  let handlerCalls = 0;
+  const pipeline = new SafetyPipeline({
+    cache,
+    logger,
+    onReject: () => {},
+    onMessage: async () => {
+      handlerCalls++;
+      await gate.promise;
+    },
+    config: {
+      chatQueue: { enabled: false },
+      processingLock: { ttlMs: 20, renewIntervalMs: 5 },
+    },
+  });
+
+  await pipeline.pushMessage(message('om_stalled'));
+  await Promise.resolve();
+  expect(handlerCalls).toBe(1);
+
+  // Move wall clock beyond the lease TTL without executing any timer callback.
+  vi.setSystemTime(2_000);
+  await pipeline.pushMessage(message('om_stalled'));
+  expect(handlerCalls).toBe(1);
+
+  gate.resolve();
+  await Promise.resolve();
+  await Promise.resolve();
+  await pipeline.dispose();
+  vi.useRealTimers();
+});
+
+test('batching carries each source exact lease through cleanup', async () => {
+  const { cache } = makeCache();
+  const markSeen = vi.spyOn(cache, 'set');
+  const pipeline = new SafetyPipeline({
+    cache,
+    logger,
+    onReject: () => {},
+    onMessage: async () => {},
+    config: {
+      chatQueue: { enabled: true },
+      batch: { text: { delayMs: 10_000, maxMessages: 2 } },
+      processingLock: { ttlMs: 50, renewIntervalMs: 10 },
+    },
+  });
+  const lock = (pipeline as any).lock;
+  const acquire = vi.spyOn(lock, 'acquire');
+  const stopRenewal = vi.spyOn(lock, 'stopRenewal');
+  const release = vi.spyOn(lock, 'release');
+
+  await pipeline.pushMessage(message('om_batch_1'));
+  await pipeline.pushMessage(message('om_batch_2'));
+  await (pipeline as any).manager.flushAll();
+
+  const firstLease = acquire.mock.results[0].value;
+  const secondLease = acquire.mock.results[1].value;
+  expect(stopRenewal.mock.calls[0][0]).toBe(firstLease);
+  expect(stopRenewal.mock.calls[1][0]).toBe(secondLease);
+  expect(release.mock.calls[0][0]).toBe(firstLease);
+  expect(release.mock.calls[1][0]).toBe(secondLease);
+  expect(stopRenewal.mock.invocationCallOrder[0]).toBeLessThan(
+    markSeen.mock.invocationCallOrder[0],
+  );
+  expect(markSeen.mock.invocationCallOrder[0]).toBeLessThan(release.mock.invocationCallOrder[0]);
+  expect(stopRenewal.mock.invocationCallOrder[1]).toBeLessThan(
+    markSeen.mock.invocationCallOrder[1],
+  );
+  expect(markSeen.mock.invocationCallOrder[1]).toBeLessThan(release.mock.invocationCallOrder[1]);
+  await pipeline.dispose();
+});
+
+test('pushAction finalizes and releases the exact acquired lease', async () => {
+  const { cache } = makeCache();
+  const markSeen = vi.spyOn(cache, 'set');
+  const pipeline = new SafetyPipeline({
+    cache,
+    logger,
+    onReject: () => {},
+    onMessage: async () => {},
+    config: {
+      chatQueue: { enabled: false },
+      processingLock: { ttlMs: 50, renewIntervalMs: 10 },
+    },
+  });
+  const lock = (pipeline as any).lock;
+  const acquire = vi.spyOn(lock, 'acquire');
+  const stopRenewal = vi.spyOn(lock, 'stopRenewal');
+  const release = vi.spyOn(lock, 'release');
+
+  await pipeline.pushAction('action_exact', 'oc_action', async () => 'ok');
+
+  const lease = acquire.mock.results[0].value;
+  expect(stopRenewal.mock.calls[0][0]).toBe(lease);
+  expect(release.mock.calls[0][0]).toBe(lease);
+  expect(stopRenewal.mock.invocationCallOrder[0]).toBeLessThan(
+    markSeen.mock.invocationCallOrder[0],
+  );
+  expect(markSeen.mock.invocationCallOrder[0]).toBeLessThan(release.mock.invocationCallOrder[0]);
+  await pipeline.dispose();
+});
+
+test('lease stays renewable while a queued message waits behind another handler', async () => {
+  const firstGate = deferred();
+  const { cache } = makeCache();
+  const handled: string[] = [];
+  const pipeline = new SafetyPipeline({
+    cache,
+    logger,
+    onReject: () => {},
+    onMessage: async (msg) => {
+      handled.push(msg.messageId);
+      if (msg.messageId === 'om_first') await firstGate.promise;
+    },
+    config: {
+      chatQueue: { enabled: true },
+      batch: { text: { delayMs: 0, maxMessages: 1 } },
+      processingLock: { ttlMs: 20, renewIntervalMs: 5 },
+    },
+  });
+  const lock = (pipeline as any).lock;
+  const acquire = vi.spyOn(lock, 'acquire');
+  const stopRenewal = vi.spyOn(lock, 'stopRenewal');
+  const release = vi.spyOn(lock, 'release');
+
+  await pipeline.pushMessage(message('om_first'));
+  await pipeline.pushMessage(message('om_waiting'));
+  await wait(50);
+  await pipeline.pushMessage(message('om_waiting'));
+  expect(handled).toEqual(['om_first']);
+
+  firstGate.resolve();
+  await (pipeline as any).manager.flushAll();
+  expect(handled).toEqual(['om_first', 'om_waiting']);
+  const firstLease = acquire.mock.results[0].value;
+  const waitingLease = acquire.mock.results[1].value;
+  expect(stopRenewal.mock.calls[0][0]).toBe(firstLease);
+  expect(stopRenewal.mock.calls[1][0]).toBe(waitingLease);
+  expect(release.mock.calls[0][0]).toBe(firstLease);
+  expect(release.mock.calls[1][0]).toBe(waitingLease);
+  await pipeline.dispose();
+});
+
+test.each([
+  { ttlMs: 0, renewIntervalMs: 1 },
+  { ttlMs: 20, renewIntervalMs: 0 },
+  { ttlMs: 20, renewIntervalMs: 20 },
+  { ttlMs: 20.5, renewIntervalMs: 5 },
+  { ttlMs: 20, renewIntervalMs: 5.5 },
+  { ttlMs: Number.POSITIVE_INFINITY, renewIntervalMs: 5 },
+  { ttlMs: 2_147_483_648, renewIntervalMs: 5 },
+  { ttlMs: 2_147_483_647, renewIntervalMs: 2_147_483_648 },
+])('SafetyConfig rejects invalid processing lock config: %o', (processingLock) => {
+  const { cache } = makeCache();
+  expect(
+    () =>
+      new SafetyPipeline({
+        cache,
+        logger,
+        onReject: () => {},
+        onMessage: async () => {},
+        config: { processingLock },
+      }),
+  ).toThrow(RangeError);
+});
diff --git a/src/safety/chat-pipeline.ts b/src/safety/chat-pipeline.ts
index ea881d6..c90aacf 100644
--- a/src/safety/chat-pipeline.ts
+++ b/src/safety/chat-pipeline.ts
@@ -1,4 +1,5 @@
 import type { MentionInfo, NormalizedMessage, ResourceDescriptor } from '../types';
+import type { ProcessingLease } from './processing-lock';
 import type { BatchConfig, BatchedDispatch } from './types';
 
 type FlushHandler = (batch: BatchedDispatch) => Promise<void>;
@@ -15,7 +16,7 @@ type FlushHandler = (batch: BatchedDispatch) => Promise<void>;
  *                          pending batch and previous tasks
  */
 export class ChatPipeline {
-  private buffer: NormalizedMessage[] = [];
+  private buffer: Array<{ message: NormalizedMessage; lease: ProcessingLease }> = [];
   private bufferChars = 0;
   private timer?: NodeJS.Timeout;
   private tail: Promise<void> = Promise.resolve();
@@ -28,8 +29,8 @@ export class ChatPipeline {
     private serialOnly: boolean,
   ) {}
 
-  push(msg: NormalizedMessage, handler: FlushHandler): void {
-    this.buffer.push(msg);
+  push(msg: NormalizedMessage, lease: ProcessingLease, handler: FlushHandler): void {
+    this.buffer.push({ message: msg, lease });
     this.bufferChars += msg.content.length;
     this.pendingHandler ??= handler;
 
@@ -117,8 +118,8 @@ export class ChatPipeline {
     if (!handler) return;
 
     const dispatch: BatchedDispatch = {
-      message: mergeBatch(batch),
-      sourceIds: batch.map((m) => m.messageId),
+      message: mergeBatch(batch.map((source) => source.message)),
+      sources: batch.map(({ message, lease }) => ({ messageId: message.messageId, lease })),
     };
 
     this.busy = true;
@@ -149,8 +150,8 @@ export class ChatPipelineManager {
 
   constructor(private config: BatchConfig) {}
 
-  push(scope: string, msg: NormalizedMessage, handler: FlushHandler): void {
-    this.getOrCreate(scope, false).push(msg, handler);
+  push(scope: string, msg: NormalizedMessage, lease: ProcessingLease, handler: FlushHandler): void {
+    this.getOrCreate(scope, false).push(msg, lease, handler);
   }
 
   run<T>(scope: string, task: () => Promise<T>): Promise<T> {
diff --git a/src/safety/index.ts b/src/safety/index.ts
index 339d65d..8fb7aa7 100644
--- a/src/safety/index.ts
+++ b/src/safety/index.ts
@@ -12,7 +12,7 @@ import { ChatPipelineManager } from './chat-pipeline';
 import { SeenCache } from './dedup-cache';
 import { LoopGuard } from './loop-guard';
 import { PolicyGate } from './policy-gate';
-import { ProcessingLock } from './processing-lock';
+import { type ProcessingLease, ProcessingLock } from './processing-lock';
 import { isStale } from './stale-detector';
 import {
   DEFAULT_STALE_MS,
@@ -26,6 +26,7 @@ export { SeenCache } from './dedup-cache';
 export { LoopGuard } from './loop-guard';
 export type { PolicyDecision } from './policy-gate';
 export { PolicyGate } from './policy-gate';
+export type { ProcessingLease } from './processing-lock';
 export { ProcessingLock } from './processing-lock';
 export { isStale } from './stale-detector';
 
@@ -37,6 +38,7 @@ export interface SafetyPipelineOptions {
   logger: Logger;
   onReject: OnReject;
   onMessage: OnMessageDispatch;
+  onError?: (error: unknown) => void;
 }
 
 /**
@@ -59,21 +61,26 @@ export class SafetyPipeline {
   private readonly logger: Logger;
   private readonly onReject: OnReject;
   private readonly onMessage: OnMessageDispatch;
+  private readonly onError?: (error: unknown) => void;
 
   constructor(opts: SafetyPipelineOptions) {
     this.logger = opts.logger;
     this.onReject = opts.onReject;
     this.onMessage = opts.onMessage;
+    this.onError = opts.onError;
 
     this.staleWindow = opts.config?.staleMessageWindowMs ?? DEFAULT_STALE_MS;
     this.queueEnabled = opts.config?.chatQueue?.enabled ?? true;
 
+    this.lock = new ProcessingLock(
+      opts.config?.processingLock?.ttlMs,
+      opts.config?.processingLock?.renewIntervalMs,
+    );
     this.seenCache = new SeenCache(opts.cache, {
       ttlMs: opts.config?.dedup?.ttl,
       maxMemEntries: opts.config?.dedup?.maxEntries,
       sweepMs: opts.config?.dedup?.sweepIntervalMs,
     });
-    this.lock = new ProcessingLock();
     this.policy = new PolicyGate(opts.policy, opts.botIdentity, opts.logger);
     this.loopGuard = new LoopGuard(opts.policy?.botLoopGuard, opts.logger);
     this.manager = new ChatPipelineManager(resolveBatchConfig(opts.config));
@@ -118,33 +125,43 @@ export class SafetyPipeline {
       return;
     }
 
-    if (!this.lock.acquire(msg.messageId)) {
+    const lease = this.lock.acquire(msg.messageId);
+    if (!lease) {
       this.logger.debug?.(`safety: drop in-flight message ${msg.messageId}`);
       return;
     }
 
-    const dispatchHandler = async (batch: { message: NormalizedMessage; sourceIds: string[] }) => {
+    const dispatchHandler = async (batch: {
+      message: NormalizedMessage;
+      sources: Array<{ messageId: string; lease: ProcessingLease }>;
+    }) => {
       try {
         await this.onMessage(batch.message);
       } catch (e) {
         this.logger.error?.(`safety: message handler threw`, e);
+        try {
+          this.onError?.(e);
+        } catch (observerError) {
+          this.logger.error?.(`safety: error observer threw`, observerError);
+        }
       } finally {
-        for (const id of batch.sourceIds) {
+        for (const source of batch.sources) {
+          this.lock.stopRenewal(source.lease);
           try {
-            await this.seenCache.add(id);
+            await this.seenCache.add(source.messageId);
           } catch {
             /* best effort */
           }
-          this.lock.release(id);
+          this.lock.release(source.lease);
         }
       }
     };
 
     if (this.queueEnabled) {
-      this.manager.push(msg.chatId, msg, dispatchHandler);
+      this.manager.push(msg.chatId, msg, lease, dispatchHandler);
     } else {
       // queueing disabled: fire-and-forget, no batch either
-      void dispatchHandler({ message: msg, sourceIds: [msg.messageId] });
+      void dispatchHandler({ message: msg, sources: [{ messageId: msg.messageId, lease }] });
     }
   }
 
@@ -159,7 +176,8 @@ export class SafetyPipeline {
       this.logger.debug?.(`safety: drop duplicate action ${eventId}`);
       return undefined;
     }
-    if (!this.lock.acquire(eventId)) {
+    const lease = this.lock.acquire(eventId);
+    if (!lease) {
       this.logger.debug?.(`safety: drop in-flight action ${eventId}`);
       return undefined;
     }
@@ -175,12 +193,13 @@ export class SafetyPipeline {
         this.logger.error?.(`safety: action handler threw`, e);
         return undefined;
       } finally {
+        this.lock.stopRenewal(lease);
         try {
           await this.seenCache.add(eventId);
         } catch {
           /* best effort */
         }
-        this.lock.release(eventId);
+        this.lock.release(lease);
       }
     };
 
diff --git a/src/safety/processing-lock.ts b/src/safety/processing-lock.ts
index e308c7d..da011f2 100644
--- a/src/safety/processing-lock.ts
+++ b/src/safety/processing-lock.ts
@@ -1,44 +1,111 @@
-import { DEFAULT_LOCK_TTL_MS } from './types';
+import { DEFAULT_LOCK_RENEW_INTERVAL_MS, DEFAULT_LOCK_TTL_MS } from './types';
+
+export interface ProcessingLease {
+  readonly id: string;
+  readonly ownerToken: symbol;
+}
+
+interface LockEntry {
+  lease: ProcessingLease;
+  expiresAt: number;
+  state: 'active' | 'finalizing';
+}
+
+const MAX_TIMER_DELAY_MS = 2_147_483_647;
 
 /**
  * Short-TTL in-memory lock to prevent concurrent processing of the same
  * event — complements SeenCache by covering the "currently in flight"
- * window, during which the event is not yet committed to SeenCache.
+ * window, during which the event is not yet committed to SeenCache. TTL is a
+ * renewal deadline, never authority to steal an active or finalizing owner.
  */
 export class ProcessingLock {
-  private locks = new Map<string, number>(); // id → expireAt (ms)
-  private sweeper: NodeJS.Timeout;
+  private readonly locks = new Map<string, LockEntry>();
+  private renewalTimer?: NodeJS.Timeout;
+  private readonly ttlMs: number;
+  private readonly renewIntervalMs: number;
 
   constructor(
-    private ttlMs: number = DEFAULT_LOCK_TTL_MS,
-    sweepMs: number = 60_000,
+    ttlMs: number = DEFAULT_LOCK_TTL_MS,
+    renewIntervalMs: number = defaultRenewInterval(ttlMs),
   ) {
-    this.sweeper = setInterval(() => this.sweep(), sweepMs);
-    this.sweeper.unref?.();
+    assertDuration('processingLock.ttlMs', ttlMs);
+    assertDuration('processingLock.renewIntervalMs', renewIntervalMs);
+    if (renewIntervalMs >= ttlMs) {
+      throw new RangeError('processingLock.renewIntervalMs must be less than processingLock.ttlMs');
+    }
+    this.ttlMs = ttlMs;
+    this.renewIntervalMs = renewIntervalMs;
   }
 
-  /** Returns true if the lock is acquired; false if already held. */
-  acquire(id: string): boolean {
+  /**
+   * Acquire a renewable lease. The lease remains live until `stopRenewal` is
+   * called, even when its handler is waiting in a queue or batch.
+   */
+  acquire(id: string): ProcessingLease | undefined {
     const now = Date.now();
-    const exp = this.locks.get(id);
-    if (exp && exp > now) return false;
-    this.locks.set(id, now + this.ttlMs);
-    return true;
+    if (this.locks.has(id)) return undefined;
+    const lease = Object.freeze({ id, ownerToken: Symbol(id) });
+    this.locks.set(id, { lease, expiresAt: now + this.ttlMs, state: 'active' });
+    this.ensureRenewalTimer();
+    return lease;
+  }
+
+  /** Stop extending a lease while leaving it held until explicit release. */
+  stopRenewal(lease: ProcessingLease): void {
+    const entry = this.currentEntry(lease);
+    if (entry) entry.state = 'finalizing';
+    this.stopTimerIfIdle();
+  }
+
+  release(lease: ProcessingLease): void {
+    if (this.currentEntry(lease)) this.locks.delete(lease.id);
+    this.stopTimerIfIdle();
   }
 
-  release(id: string): void {
-    this.locks.delete(id);
+  private currentEntry(lease: ProcessingLease): LockEntry | undefined {
+    const entry = this.locks.get(lease.id);
+    return entry?.lease.ownerToken === lease.ownerToken ? entry : undefined;
   }
 
-  private sweep(): void {
+  private ensureRenewalTimer(): void {
+    if (this.renewalTimer) return;
+    this.renewalTimer = setInterval(() => this.renew(), this.renewIntervalMs);
+    this.renewalTimer.unref?.();
+  }
+
+  private renew(): void {
     const now = Date.now();
-    for (const [k, v] of this.locks) {
-      if (v <= now) this.locks.delete(k);
+    for (const entry of this.locks.values()) {
+      if (entry.state === 'active') entry.expiresAt = now + this.ttlMs;
+    }
+    this.stopTimerIfIdle();
+  }
+
+  private stopTimerIfIdle(): void {
+    if (!this.renewalTimer) return;
+    for (const entry of this.locks.values()) {
+      if (entry.state === 'active') return;
     }
+    clearInterval(this.renewalTimer);
+    this.renewalTimer = undefined;
   }
 
   dispose(): void {
-    clearInterval(this.sweeper);
+    if (this.renewalTimer) clearInterval(this.renewalTimer);
+    this.renewalTimer = undefined;
     this.locks.clear();
   }
 }
+
+function defaultRenewInterval(ttlMs: number): number {
+  return Math.min(DEFAULT_LOCK_RENEW_INTERVAL_MS, Math.max(1, Math.floor(ttlMs / 3)));
+}
+
+function assertDuration(name: string, value: number): void {
+  if (!Number.isSafeInteger(value) || value < 1 || value > MAX_TIMER_DELAY_MS) {
+    throw new RangeError(
+      `${name} must be a safe integer between 1 and ${MAX_TIMER_DELAY_MS} milliseconds`,
+    );
+  }
+}
diff --git a/src/safety/types.ts b/src/safety/types.ts
index 94ac294..6c35ed1 100644
--- a/src/safety/types.ts
+++ b/src/safety/types.ts
@@ -1,4 +1,5 @@
 import type { NormalizedMessage, RejectEvent, SafetyConfig } from '../types';
+import type { ProcessingLease } from './processing-lock';
 
 export interface BatchConfig {
   delayMs: number;
@@ -29,10 +30,11 @@ export const DEFAULT_DEDUP = {
 
 export const DEFAULT_STALE_MS = 30 * 60_000;
 export const DEFAULT_LOCK_TTL_MS = 5 * 60_000;
+export const DEFAULT_LOCK_RENEW_INTERVAL_MS = 60_000;
 
 export interface BatchedDispatch {
   message: NormalizedMessage;
-  sourceIds: string[];
+  sources: Array<{ messageId: string; lease: ProcessingLease }>;
 }
 
 export type OnReject = (evt: RejectEvent) => void;
diff --git a/src/types.ts b/src/types.ts
index 3930f1c..6983c66 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -171,7 +171,7 @@ export interface EventMap {
   reaction: (evt: ReactionEvent) => void;
   botAdded: (evt: BotAddedEvent) => void;
   comment: (evt: CommentEvent) => void | Promise<void>;
-  error: (err: LarkChannelError) => void;
+  error: (err: LarkChannelError) => void | Promise<void>;
   reconnecting: () => void;
   reconnected: () => void;
 }
@@ -450,6 +450,19 @@ export interface SafetyConfig {
     maxEntries?: number;
     sweepIntervalMs?: number;
   };
+  processingLock?: {
+    /**
+     * Lease TTL in integer milliseconds (1..2,147,483,647). Defaults to
+     * 300,000. Wall-clock expiry never lets a competing owner steal an active
+     * or finalizing lease.
+     */
+    ttlMs?: number;
+    /**
+     * Renewal interval in integer milliseconds (1..2,147,483,647). Defaults
+     * to 60,000 with the default TTL and must be less than `ttlMs`.
+     */
+    renewIntervalMs?: number;
+  };
   chatQueue?: {
     enabled?: boolean;
     /**