【2026 最新】Agent Task Lease 是什麼？用 SQLite 防止平行 Agent 重複領單與過期回寫

你讓兩個 AI 助手同時整理工作，任務卡明明只有一張，兩邊卻都說「我接到了」。Agent Task Lease（任務租約）就是替認領加上有效期限與驗證規則：誰能做、何時能接手、誰還有資格交卷，都交給程式檢查。

這篇專為剛接觸平行 Agent 的讀者寫。先用借書證理解原理，再用 Python 與 SQLite 跑兩個 Mock Worker（模擬工作者）；不需要模型 API、付費帳戶或真的發送任何訊息。看完你能讀懂一條任務從認領、續租、失聯到接手的完整路徑。

先說結論：租約＝限時借書證＋每次換手的新版本

Agent Task Lease＝有效持有人＋到期時間＋遞增版本＋寫入端檢查。借書證有期限；換人借，就換一個新版本。舊證即使還握在手裡，交卷時也必須被擋下。

這套機制保護的是「哪個結果被接受」。它不會讓失聯的程式憑空停止，也不會替另一個網站撤銷已寄出的郵件。任務所有權與外部操作冪等性，要分兩層設計。

一、Agent Task Lease 解決什麼問題？

假設你要整理一份報告。A 看見任務還沒人接，B 也看見；A 把狀態改成處理中，B 接著覆蓋它。最後資料庫只記得 B，兩個程式卻都開始寫報告。錯誤出在「讀取是否空閒」與「寫入我的名字」分成兩次獨立操作，中間留出了插隊空間，這叫競態條件。

白話說，看板能顯示進度，但認領協議才決定誰有權工作。若你正在建立整體執行層，可以先讀 Agent Harness 原理；本文只補上任務所有權這個零件。

也別把它理解成某個產品專屬的新功能。截至 2026 年 11 月 2 日，Claude Code Agent Teams 文件已明載以檔案鎖處理同時認領。這裡的 SQLite 實驗，是讓你看懂自己串接多種 Worker 時，如何設計到期接手與回寫檢查；不需要替既有產品再造一套任務看板。

二、拆開 Agent Task Lease 的 5 個零件

① 原子認領：「拿號碼牌與登記，一次完成」

如何避免 A、B 都覺得自己搶到？把資格判斷與狀態更新放進同一個短交易。本例使用 BEGIN IMMEDIATE 取得寫入交易，再執行附條件的 UPDATE。只把更新一列的人算成成功，沒有更新就回傳空值。

SQLite 官方交易文件說明，同一個資料庫同時只有一個寫入交易；BEGIN IMMEDIATE 也可能遇到 SQLITE_BUSY。所以交易鎖保護的是短暫的登記過程：登記完就提交，模型思考、網路請求放在交易外。

② owner：「這張證發給誰」

如何辨識領單者？保存持有人識別碼。實際系統可用每次啟動新產生的 Worker UUID；本例用 A、B、C 方便看懂。回寫時除了任務 ID，也必須核對 owner。不要讓所有 Worker 都共用同一個固定名字。

③ expires：「到鐘就失效」

A 斷線後，B 憑什麼接手？保存到期時間。本例規則是 expires <= now 可接手、expires > now 才能續租或完成；到期那一刻算失效。認領時設定期限，正常工作時定期 heartbeat（心跳續租），失聯後讓期限自然耗盡。

④ version：「每次換手，都換證號」

如果 A 名字重複，舊請求會混進來嗎？每次成功認領，把版本加一。A 的版本 1 到期後，C 拿到版本 2。版本應持續保存在任務列裡，不能刪掉再從零開始；若要重建任務，要換新的任務 ID，避免舊證號復活。

⑤ 寫入端驗證：「收卷的人真的查證」

有版本號就安全了嗎？接收結果的地方要原子核對它。本例完成時同時檢查狀態、owner、version 與期限，再更新結果。這是本地結果儲存的版本柵欄；把版本印在日誌裡沒有保護效果。若共用另一個資源，還要依那個資源的範圍設計可比較的 token，不能拿不同任務各自從 1 開始的版本直接比較。

Hazelcast FencedLock 文件用 fencing token 說明同類問題：接收端必須辨識過期持有人。外部服務若只比較已見過的最大 token，舊請求在新 token 尚未抵達前仍有其邊界；本文的同資料庫寫入，直接核對當前租約與期限。

三、複製即可跑：兩個 Worker 搶一張卡

先在終端機執行 python3 -c "import sqlite3; print(sqlite3.sqlite_version)"，確認現有 Python 能載入 sqlite3。接著把下列完整程式存成 lease_demo.py，執行 python3 lease_demo.py。它會在暫存資料夾建立自己的資料庫，不讀寫你的工作看板。

程式刻意使用共用邏輯時鐘：從 100、103 前進到 110、113、123、133，代表故障測試的時間點，並非真的等待十秒。所有 Worker 讀同一個值，只在測例之間調整它。正式服務應由單一租約權威提供時間，在取得交易後才讀取，監控時鐘跳動；不要信任各台 Worker 自報的時間。

import sqlite3
import tempfile
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from contextlib import closing

# Logical time: all workers see the same clock; tests advance it explicitly.
now = 100
root = tempfile.TemporaryDirectory()
db = str(Path(root.name) / 'tasks.db')

def connect():
    return sqlite3.connect(db, isolation_level=None, timeout=5)

with closing(connect()) as c:
    c.executescript('''
    CREATE TABLE tasks (
      id TEXT PRIMARY KEY, state TEXT NOT NULL DEFAULT 'ready',
      owner TEXT, expires INTEGER NOT NULL DEFAULT 0,
      version INTEGER NOT NULL DEFAULT 0, result TEXT);
    CREATE TABLE receipts (key TEXT PRIMARY KEY, payload TEXT NOT NULL);
    INSERT INTO tasks(id) VALUES ('race'), ('lease'), ('retry'), ('aba');
    ''')

# Intentionally wrong: separate SELECT and unconditional UPDATE.
barrier = Barrier(2)
def naive(owner):
    c = connect()
    ready = c.execute("SELECT state FROM tasks WHERE id='race'").fetchone()[0]
    barrier.wait()
    if ready == 'ready':
        c.execute("UPDATE tasks SET state='running',owner=? WHERE id='race'", (owner,))
    c.close()
    return ready == 'ready'

with ThreadPoolExecutor(2) as pool:
    assert sum(pool.map(naive, ['A', 'B'])) == 2
print('naive: both workers believed they claimed the task')

def claim(task, owner, ttl=10):
    c = connect()
    try:
        c.execute('BEGIN IMMEDIATE')
        t = now  # production: sample authority clock AFTER acquiring tx
        n = c.execute('''UPDATE tasks
          SET state='running',owner=?,expires=?,version=version+1
          WHERE id=? AND (state='ready' OR
            (state='running' AND expires<=?))''',
          (owner, t+ttl, task, t)).rowcount
        token = c.execute('SELECT version FROM tasks WHERE id=?', (task,)).fetchone()[0] if n else None
        c.execute('COMMIT')
        return token
    except BaseException:
        if c.in_transaction:
            c.execute('ROLLBACK')
        raise
    finally:
        c.close()

def guarded(task, owner, token, result=None, ttl=10):
    c = connect()
    try:
        c.execute('BEGIN IMMEDIATE')
        t = now
        if result is None:  # heartbeat
            sql = 'UPDATE tasks SET expires=?'
            values = (t+ttl,)
        else:  # commit result to this SAME protected database
            sql = "UPDATE tasks SET state='done',result=?"
            values = (result,)
        n = c.execute(sql+''' WHERE id=? AND state='running'
          AND owner=? AND version=? AND expires>?''',
          values+(task,owner,token,t)).rowcount
        c.execute('COMMIT')
        return n == 1
    except BaseException:
        if c.in_transaction:
            c.execute('ROLLBACK')
        raise
    finally:
        c.close()

# Mock service: receipt and effect represented by ONE unique-key row.
# Separate transaction from task completion, to model a lost response.
def effect(key, payload):
    c = connect()
    try:
        c.execute('INSERT OR IGNORE INTO receipts VALUES (?,?)', (key,payload))
        saved = c.execute('SELECT payload FROM receipts WHERE key=?', (key,)).fetchone()[0]
        if saved != payload:
            raise ValueError('same key, different intent')
        return saved
    finally:
        c.close()

barrier = Barrier(2)
def safe(owner):
    barrier.wait()
    return claim('lease', owner)
with ThreadPoolExecutor(2) as pool:
    tokens = list(pool.map(safe, ['A','B']))
assert sum(t is not None for t in tokens) == 1
owner = ['A','B'][0 if tokens[0] is not None else 1]
a = next(t for t in tokens if t is not None)
assert not guarded('lease','not-the-owner',a)
now = 103
assert guarded('lease',owner,a)  # expiry extended from 110 to 113
now = 110
assert claim('lease','C') is None  # renewal really extended ownership
now = 113  # expires == now means expired
assert not guarded('lease',owner,a)
assert not guarded('lease',owner,a,'expired')
b = claim('lease','C')
assert b == a+1
assert not guarded('lease',owner,a,'stale')
assert guarded('lease','C',b,'fresh')
assert not guarded('lease','C',b,'duplicate')
print('lease: 1 winner; expired renewal/stale/duplicate writes rejected')

x = claim('retry','D')
effect('retry:publish:v1','article')  # success, response lost; no finish
now = 123  # D disappears; new connection/worker takes over
x2 = claim('retry','E')
assert x2 == x+1
effect('retry:publish:v1','article')  # stable business key, NOT lease token
assert guarded('retry','E',x2,'receipt-ok')
with closing(connect()) as c:
    assert c.execute('SELECT count(*) FROM receipts').fetchone()[0] == 1
print('retry: mock service has 1 receipt after replay with the same key')
# Same owner name reused: only version distinguishes old and new leases.
z = claim('aba','R')
now = 133
z2 = claim('aba','R')
assert z2 == z+1
assert not guarded('aba','R',z,'old-generation')
assert guarded('aba','R',z2,'new-generation')
print('version: old token rejected even when owner name is reused')
print('PASS')
root.cleanup()

這段使用 Python 的明確交易控制與參數綁定，沒有額外套件；可對照 Python sqlite3 文件閱讀。timeout=5 是鎖等待上限設定，不是租約期限。若爭用超過等待時間，程式會報錯而不是假裝領到單；正式派工端應對暫時性鎖衝突做有上限的退避重試。

四、沿著輸出追一次：哪些事情真的被驗證？

本次 AlphaLab 在本機執行上列程式，保留程式與輸出紀錄；它是功能測例，並非吞吐量或可靠性基準。第一行顯示兩個錯誤 Worker 都自認成功：Barrier（集合點）讓雙方都讀完 ready，才一起繼續，確實重現先讀後寫的競態。

• 同時認領：第二組 Worker 各開自己的資料庫連線；成功拿到 token 的數量必須等於 1。誰先到不重要，唯一有效持有人才是驗收條件。
• 到期續租：先在 103 續租，把期限從 110 延長到 113；110 時接手仍失敗。時鐘前進到 113，期限正好相等，原持有人續租必須失敗；若只檢查名字，這項很容易漏掉。
• 接手與晚到結果：C 接手，版本從 1 變成 2；原持有人提交 stale 必須失敗，C 提交 fresh 成功，再提交一次也失敗。另有 R 名稱重用的測例：owner 相同且新租約有效時，舊 token 仍被拒絕，單獨驗證版本檢查。
• 失聯與重送：D 呼叫模擬服務成功後，刻意不寫完成；期限過後 E 用新連線接手，以同一業務鍵重送，服務紀錄仍只有 1 列。

全部斷言通過，最後才會印出 PASS。失聯是透過「停止續租與不回報完成」模擬，新連線代表重新連入；這個腳本沒有真的切斷網路、殺掉作業系統程序，也沒有測試真實郵件服務。下一步測程序重啟時，保留同一份資料庫，重建 Worker 身分，仍要重跑相同的所有權斷言。

五、租約不是外部操作恰好一次：再補一把業務鑰匙

最危險的縫隙是：外部動作已完成，但成功回應遺失。D 寄出郵件後斷線，資料庫仍顯示處理中；E 接手再寄，兩封郵件就出去了。只把本地結果寫一次，並不能撤銷這個外部重複。

解法是穩定的冪等鍵，也就是重試相同業務動作時，仍使用同一把鑰匙。例如 retry:publish:v1 表示某一版內容的一次發布意圖。不要把新租約版本當成這把鑰匙，否則每次接手都像新動作。AWS 冪等 API 設計文章說明了以呼叫端請求識別碼表達同一意圖，以及服務端把去重紀錄與動作一起可靠保存的必要性。

腳本的 receipts 是模擬接收端：唯一鍵那一列本身就是副作用，所以插入與去重在同一個資料庫操作完成；相同鍵卻傳不同內容會報錯。它不是「先寫本地收據，再呼叫外部 API」的萬用方案。真正接第三方服務，要查明它的冪等鍵作用範圍、保存期限與參數衝突規則；若沒有合適契約，就把結果標成待對帳，避免盲目重送。

類似地，SQS visibility timeout會讓領走的訊息暫時不可見，逾時可重新處理；Standard queue 的文件仍明確保留重複投遞可能。訊息暫時隱藏、任務所有權、外部動作去重，是三個不同的承諾。

六、上線前先踩煞車的 4 個地方

• 把資料庫交易鎖整段工作：只鎖認領與提交；耗時運算在交易外。否則一個慢 Worker 就讓其他寫入排隊。
• 續租失敗卻繼續當老闆：心跳更新不到一列，就停止發起後續受保護操作；晚到結果仍必須由收件端拒絕。
• 把本地檔案丟上共用磁碟：本例限定同機本地資料庫。多機直接開同一份網路檔案，應先重新設計儲存與鎖定架構。
• 接手無限循環：正式任務加上嘗試次數、失敗原因、退避時間與人工處理狀態；必須把永久錯誤和暫時故障分開。

SQLite 網路使用指引提醒直接透過網路檔案系統共用資料庫的鎖定風險。多台 Agent 經由同一個服務 API，由該服務管理本地 SQLite，是另一種架構；不要和多台電腦直接開同一個檔案混為一談。

租約長度則從工作耗時與恢復目標決定：太短容易誤判，太長拖慢接手。先記錄任務耗時、心跳延遲、接手次數與拒絕舊 token 次數，再調整。這份示例刻意不做分散式時鐘、認證授權、佇列公平性或正式重試器；需要的是擴充這些具體部件，而不是把十個時間單位直接照搬上線。

常見問題 FAQ

1. 有租約就只會有一個程式在跑嗎？

不一定。舊 Worker 暫停後恢復，新 Worker 可能已接手，兩者會同時運算；驗證點保證的是符合條件的寫入才被接受。

2. 到期了，原 Worker 可以自行續租嗎？

本例不行。到期必須重新認領並取得新版本；這避免過期請求復活。

3. 只核對 owner 可以嗎？

本例仍核對版本與期限。即使名稱重用，舊版本也不能通過；owner 是身分，version 是這一次持有的世代。

4. fencing token 就是冪等鍵嗎？

不是。token 處理誰的權限已過期；冪等鍵處理同一業務動作的重試。接手換 token，重試同一動作保留業務鍵。

5. 需要真的呼叫 AI 模型嗎？

不需要。先驗證派工狀態機，再接模型；模型回應內容不會改變認領交易的規則。

6. 一定要用 SQLite 嗎？

不一定。這裡選它是方便在同機跑實驗；多機、高寫入競爭或已有成熟佇列時，依部署條件選資料庫或訊息服務，保留相同驗收情境。

7. 這是在測網路斷線嗎？

不是實際網路測試。腳本用不續租、不完成、再開連線與重送來控制故障時序，讓斷言可重複；真實網路與程序故障需另外注入。

8. 下一步應該先改什麼？

先把 claim、renew、finish 接到你的一種測試任務，保留 owner、到期與版本檢查，再針對外部動作加入經查證的冪等或對帳流程。

給新手的 3 個重點

第一，認領是一個原子決定，不是先看再寫。第二，每次接手都換版本，且在寫入端檢查當前資格。第三，外部動作要有自己的去重契約；本地任務顯示完成，與外部只發生一次，不能畫上等號。