merge: feature/agentctl-claim-2026-09-23 (long-lived claim CLI)

.doc/ops/CURRENT.md

@@ -18,6 +18,7 @@
 
 | owner | task | branch | worktree | mutex | status |
 |---|---|---|---|---|---|
+| Cursor | agentctl-claim-2026-09-23 | feature/agentctl-claim-2026-09-23 | .worktrees/agentctl-claim-2026-09-23 | agentctl-claim | in_progress |
 | Cursor | agent-governance-config-2026-09-23 | feature/agent-governance-config-2026-09-23 | .worktrees/agent-governance-config-2026-09-23 | agent-governance-config | in_progress |
 | Cursor | coordination-db-daemon-impl-plan-2026-09-23 | feature/coordination-db-daemon-impl-plan-2026-09-23 | .worktrees/coordination-db-daemon-impl-plan-2026-09-23 | coordination-db-daemon-impl-plan | done |
 | Cursor | coordination-db-daemon-2026-09-23 | feature/coordination-db-daemon-2026-09-23 | .worktrees/coordination-db-daemon-2026-09-23 | coordination-db-daemon | done |

.doc/ops/short-term/2026-09-23-agentctl-claim.md

@@ -0,0 +1,46 @@
+---
+title: "W1-U3：显式长期认领 claim"
+task_id: agentctl-claim-2026-09-23
+status: in_progress
+created: 2026-09-23
+updated: 2026-09-23
+owner: Cursor
+task_type: implementation
+task_tier: single
+policy_revision: "agent-governance-2026-09-21-r1 @ 7cd0f505b225b0196eaab3fe0afe3e196c7ca5f7"
+branch: feature/agentctl-claim-2026-09-23
+worktree: .worktrees/agentctl-claim-2026-09-23
+base_dev_sha: "7cd0f505b225b0196eaab3fe0afe3e196c7ca5f7"
+write_scope: ["scripts", ".doc"]
+allowed_environments: []
+mutex: agentctl-claim
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: "独立维护者"
+cleanup_trigger: "合入 local_dev 且无进程占用后复查"
+remote_sync: not_requested
+---
+
+# W1-U3：显式长期认领 claim
+
+## 用户目标与授权
+
+- 用户目标：`claim:` 命名空间 + claims/claim_queue 表 + acquire/release/status/list/cancel CLI；原子认领事务。
+- 授权：W1 波次 U3；依赖 U2 已合入；不 push。
+
+## 验收
+
+1. 20 并发恰好 1 held
+2. 不相关资源并行成功
+3. 父子重叠拒绝；兄弟成功
+4. 冻结拒绝（含持有者）
+5. request_id 幂等 / 换载荷拒绝
+6. task close 拒绝 held claim
+7. unittest discover 全绿
+
+## 相关链接
+
+- 前序 U2 tip：`7cd0f505`
+- [实施承接计划](../../remediation/2026-09-23-coordination-db-daemon-implementation-undertaking-plan.md)

scripts/agentctl.py

@@ -74,6 +74,30 @@ CREATE TABLE IF NOT EXISTS queue_items(
 );
 CREATE INDEX IF NOT EXISTS queue_pending ON queue_items(state, priority DESC, created);
 '''
+CLAIM_SCHEMA = '''
+CREATE TABLE IF NOT EXISTS claims(
+  id TEXT PRIMARY KEY,
+  resource TEXT NOT NULL,
+  mode TEXT NOT NULL,
+  task TEXT NOT NULL,
+  owner TEXT NOT NULL,
+  request_id TEXT,
+  state TEXT NOT NULL,
+  reason TEXT,
+  created REAL NOT NULL,
+  updated REAL NOT NULL,
+  UNIQUE(resource, task)
+);
+CREATE INDEX IF NOT EXISTS claims_active ON claims(state, resource);
+CREATE TABLE IF NOT EXISTS claim_queue(
+  id TEXT PRIMARY KEY,
+  resource TEXT NOT NULL,
+  task TEXT NOT NULL,
+  request_id TEXT,
+  state TEXT NOT NULL,
+  created REAL NOT NULL
+);
+'''
 
 Lookup = Literal['ok', 'gone', 'unreadable']
 AliveStatus = Literal['alive', 'dead', 'unknown']
@@ -425,6 +449,8 @@ def covers(parent, child):
 def resource(value):
     if value.startswith('path:'):
         return 'path:' + relative(value[5:])
+    if value.startswith('claim:'):
+        return 'claim:' + relative(value[6:])
     if value in ('docs:publish', 'git:worktrees', 'git:index') or re.fullmatch(r'(env|git):(dev|test|main|shared)', value):
         return value
     if value.startswith('tree:'):
@@ -437,6 +463,8 @@ def overlaps(a, b):
         return True
     if a.startswith('path:') and b.startswith('path:'):
         return covers(a[5:], b[5:]) or covers(b[5:], a[5:])
+    if a.startswith('claim:') and b.startswith('claim:'):
+        return covers(a[6:], b[6:]) or covers(b[6:], a[6:])
     # Publishing protected docs conflicts with direct managed path writers, too.
     if a == 'docs:publish' and b.startswith('path:'):
         return any(covers(b[5:], p) or covers(p, b[5:]) for p in SHARED_DOCS)
@@ -655,9 +683,11 @@ def initialize(repo):
 
 
 def ensure_schema(db):
-    """Additive tables for durable queue + W4 deploy/projection; write txn only."""
+    """Additive tables for durable queue + W4 deploy/projection + claims; write txn only."""
     for statement in filter(None, (part.strip() for part in QUEUE_SCHEMA.split(';'))):
         db.execute(statement)
+    for statement in filter(None, (part.strip() for part in CLAIM_SCHEMA.split(';'))):
+        db.execute(statement)
     w4.ensure_schema(db)
 
 
@@ -683,6 +713,12 @@ def check_resources(repo, task, resources):
             safe_path(Path(task['worktree']), p)
             if not any(covers(s, p) for s in scopes):
                 deny('PATH_OUT_OF_SCOPE', 'Resource is outside the registered write scope', resource=value)
+        elif value.startswith('claim:'):
+            p = value[6:]
+            if p != '.' and secret_path(p):
+                deny('PATH_OUT_OF_SCOPE', 'Runtime and Git administration paths are not claimable')
+            if not any(covers(s, p) for s in scopes):
+                deny('PATH_OUT_OF_SCOPE', 'Claim resource is outside the registered write scope', resource=value)
         elif value.startswith('tree:'):
             if value != 'tree:' + task['worktree']:
                 deny('PATH_OUT_OF_SCOPE', 'Task cannot acquire another worktree index')
@@ -1714,6 +1750,185 @@ def cleanup_status(repo, args):
     return 0
 
 
+def _claim_fingerprint(task_id, resources, mode='X'):
+    return digest(json.dumps({'task': task_id, 'resources': sorted(resources), 'mode': mode}, sort_keys=True))
+
+
+def claim_acquire(repo, args):
+    task = task_row(repo, args.task)
+    resources = sorted({resource(item) for item in args.resource})
+    if not resources or any(not item.startswith('claim:') for item in resources):
+        deny('INVALID_RESOURCE', 'claim acquire requires one or more claim: resources')
+    check_resources(repo, task, resources)
+    request_id = args.request_id or str(uuid.uuid4())
+    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', request_id):
+        deny('INVALID_REQUEST', 'Invalid request id')
+    fingerprint = _claim_fingerprint(task['id'], resources)
+    wait = max(0.0, min(float(args.wait or 0), 60.0))
+    deadline = time.monotonic() + wait
+    while True:
+        busy = None
+        with repo.db(True) as db:
+            prior = [dict(r) for r in db.execute(
+                'SELECT * FROM claims WHERE request_id=?', (request_id,)).fetchall()]
+            if prior:
+                held = [row for row in prior if row['state'] == 'held']
+                payload = sorted(row['resource'] for row in held) if held else sorted(
+                    row['resource'] for row in prior)
+                same_task = all(row['task'] == task['id'] for row in prior)
+                same_fp = _claim_fingerprint(task['id'], payload) == fingerprint and same_task
+                if held and same_fp and set(payload) == set(resources):
+                    emit({'status': 'held', 'replayed': True, 'request_id': request_id,
+                          'claims': held})
+                    return 0
+                deny('REQUEST_ID_REUSED', 'Claim request id belongs to a different payload',
+                     request_id=request_id)
+
+            frozen = []
+            for row in db.execute('SELECT * FROM freezes'):
+                if any(overlaps(item, row['resource']) for item in resources):
+                    frozen.append(dict(row))
+            if frozen:
+                deny('FROZEN', 'Resource is frozen', resource=frozen[0]['resource'],
+                     holder=frozen[0]['owner'])
+
+            holders = []
+            for row in db.execute("SELECT * FROM claims WHERE state='held'"):
+                if any(overlaps(item, row['resource']) for item in resources):
+                    # Self re-acquire of identical resource is covered by UNIQUE(resource,task)
+                    if row['task'] == task['id'] and row['resource'] in resources:
+                        continue
+                    holders.append(dict(row))
+            if holders:
+                tickets = []
+                now = time.time()
+                for item in resources:
+                    ticket = str(uuid.uuid4())
+                    db.execute(
+                        'INSERT INTO claim_queue(id,resource,task,request_id,state,created) VALUES(?,?,?,?,?,?)',
+                        (ticket, item, task['id'], request_id, 'queued', now))
+                    tickets.append(ticket)
+                    event(db, ticket, {'event': 'claim_queued', 'resource': item,
+                                       'task': task['id'], 'holder': holders[0]['task']})
+                busy = {
+                    'holder': holders[0]['task'],
+                    'ticket': tickets[0],
+                    'tickets': tickets,
+                    'retry_after': 1,
+                    'resources': resources,
+                }
+            else:
+                now = time.time()
+                claims = []
+                for item in resources:
+                    existing = db.execute(
+                        'SELECT * FROM claims WHERE resource=? AND task=?',
+                        (item, task['id'])).fetchone()
+                    if existing and existing['state'] == 'held':
+                        claims.append(dict(existing))
+                        continue
+                    claim_id = request_id if len(resources) == 1 else f'{request_id}:{digest(item)[:12]}'
+                    if existing:
+                        db.execute(
+                            '''UPDATE claims SET id=?, mode=?, owner=?, request_id=?, state='held',
+                               reason=?, updated=? WHERE resource=? AND task=?''',
+                            (claim_id, 'X', task['owner'], request_id, fingerprint, now,
+                             item, task['id']))
+                    else:
+                        db.execute(
+                            '''INSERT INTO claims(id,resource,mode,task,owner,request_id,state,reason,created,updated)
+                               VALUES(?,?,?,?,?,?,'held',?,?,?)''',
+                            (claim_id, item, 'X', task['id'], task['owner'], request_id,
+                             fingerprint, now, now))
+                    row = dict(db.execute(
+                        'SELECT * FROM claims WHERE resource=? AND task=?',
+                        (item, task['id'])).fetchone())
+                    claims.append(row)
+                    event(db, claim_id, {'event': 'claim_acquired', 'resource': item,
+                                         'task': task['id'], 'request_id': request_id})
+                emit({'status': 'held', 'request_id': request_id, 'claims': claims})
+                return 0
+        if busy:
+            if time.monotonic() >= deadline:
+                deny('LOCK_BUSY', 'Claim resource is held; retry with ticket or do independent work',
+                     **busy)
+            time.sleep(min(0.05, max(0.0, deadline - time.monotonic())))
+            continue
+        deny('RECOVERY_REQUIRED', 'Claim acquire exited without result')
+
+
+def claim_release(repo, args):
+    task = task_row(repo, args.task)
+    target = resource(args.resource)
+    if not target.startswith('claim:'):
+        deny('INVALID_RESOURCE', 'claim release requires a claim: resource')
+    with repo.db(True) as db:
+        row = db.execute(
+            "SELECT * FROM claims WHERE resource=? AND task=? AND state='held'",
+            (target, task['id'])).fetchone()
+        if not row:
+            deny('UNKNOWN_CLAIM', 'No held claim for this task/resource')
+        now = time.time()
+        db.execute("UPDATE claims SET state='released', updated=?, reason=? WHERE id=?",
+                   (now, args.reason or 'released', row['id']))
+        event(db, row['id'], {'event': 'claim_released', 'resource': target, 'task': task['id']})
+        # Promote earliest queued waiter to visibility only; acquire still races fairly.
+        waiting = db.execute(
+            "SELECT * FROM claim_queue WHERE resource=? AND state='queued' ORDER BY created LIMIT 1",
+            (target,)).fetchone()
+        emit({'status': 'released', 'claim': dict(row), 'next_ticket': waiting['id'] if waiting else None})
+    return 0
+
+
+def claim_status(repo, args):
+    target = resource(args.resource)
+    if not target.startswith('claim:'):
+        deny('INVALID_RESOURCE', 'claim status requires a claim: resource')
+    with repo.db() as db:
+        try:
+            held = [dict(r) for r in db.execute(
+                "SELECT * FROM claims WHERE state='held'").fetchall()
+                    if overlaps(target, r['resource'])]
+            queued = [dict(r) for r in db.execute(
+                "SELECT * FROM claim_queue WHERE state='queued' ORDER BY created").fetchall()
+                     if overlaps(target, r['resource'])]
+        except sqlite3.Error:
+            held, queued = [], []
+    emit({'status': 'ok', 'resource': target, 'held': held, 'queued': queued})
+    return 0
+
+
+def claim_list(repo, args):
+    with repo.db() as db:
+        try:
+            if args.all:
+                rows = [dict(r) for r in db.execute(
+                    'SELECT * FROM claims ORDER BY updated DESC').fetchall()]
+            else:
+                rows = [dict(r) for r in db.execute(
+                    "SELECT * FROM claims WHERE state='held' ORDER BY updated DESC").fetchall()]
+        except sqlite3.Error:
+            rows = []
+    emit({'status': 'ok', 'claims': rows})
+    return 0
+
+
+def claim_cancel(repo, args):
+    task = task_row(repo, args.task)
+    with repo.db(True) as db:
+        row = db.execute('SELECT * FROM claim_queue WHERE id=?', (args.id,)).fetchone()
+        if not row:
+            deny('UNKNOWN_TICKET', 'Unknown claim queue ticket')
+        if row['task'] != task['id']:
+            deny('PATH_OUT_OF_SCOPE', 'Only the ticket owner can cancel')
+        if row['state'] != 'queued':
+            deny('INVALID_REQUEST', f'Ticket is already {row["state"]}')
+        db.execute("UPDATE claim_queue SET state='cancelled' WHERE id=?", (args.id,))
+        event(db, args.id, {'event': 'claim_cancelled', 'reason': args.reason, 'task': task['id']})
+    emit({'status': 'cancelled', 'ticket': args.id, 'reason': args.reason})
+    return 0
+
+
 def parser():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('--repo',default='.')
     sub=p.add_subparsers(dest='action',required=True)
@@ -1760,6 +1975,13 @@ def parser():
     dr.add_argument('command',nargs=argparse.REMAINDER)
     ds=dep.add_parser('status');ds.add_argument('--id',required=True)
     drec=dep.add_parser('recover');drec.add_argument('--id',required=True);drec.add_argument('--reason',required=True)
+    cl=sub.add_parser('claim').add_subparsers(dest='claim_action',required=True)
+    ca=cl.add_parser('acquire');ca.add_argument('--task',required=True);ca.add_argument('--resource',action='append',required=True)
+    ca.add_argument('--request-id');ca.add_argument('--wait',type=float,default=0)
+    crl=cl.add_parser('release');crl.add_argument('--task',required=True);crl.add_argument('--resource',required=True);crl.add_argument('--reason',default='')
+    cst=cl.add_parser('status');cst.add_argument('--resource',required=True);cst.add_argument('--json',action='store_true')
+    clist=cl.add_parser('list');clist.add_argument('--all',action='store_true')
+    cc=cl.add_parser('cancel');cc.add_argument('--task',required=True);cc.add_argument('--id',required=True);cc.add_argument('--reason',required=True)
     return p
 
 
@@ -1767,7 +1989,7 @@ def main():
     args=parser().parse_args();repo=Repo(args.repo)
     write_actions = {
         'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue',
-        'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup',
+        'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup', 'claim',
     }
     # status/doctor remain readable without a policy file and never create one.
     if args.action in write_actions:
@@ -1775,8 +1997,10 @@ def main():
             pass  # snapshot is read-only CAS helper
         elif args.action == 'queue' and getattr(args, 'queue_action', None) == 'list':
             pass
+        elif args.action == 'claim' and getattr(args, 'claim_action', None) in ('status', 'list'):
+            require_policy(repo)  # status/list need schema constants but are non-mutating
         elif args.action == 'cleanup':
-            require_policy(repo)  # status report still needs contract constants
+            require_policy(repo)
         else:
             require_policy(repo)
     if args.action=='init':initialize(repo)
@@ -1828,6 +2052,8 @@ def main():
             task_row(repo,args.id)
             with repo.db(True) as db:
                 if db.execute("SELECT id FROM runs WHERE task=? AND state IN ('running','queued','recovery_required')",(args.id,)).fetchone():deny('LOCK_BUSY','Task still has an active or unresolved operation')
+                if db.execute("SELECT id FROM claims WHERE task=? AND state='held'",(args.id,)).fetchone():
+                    deny('LOCK_BUSY','Task still holds claims; release them before close')
                 db.execute('UPDATE tasks SET state=? WHERE id=?',(args.status,args.id))
             emit({'status':args.status,'task':args.id})
     elif args.action=='with':return execute(repo,args)
@@ -1873,6 +2099,12 @@ def main():
     elif args.action=='validate':return validate(repo,args)
     elif args.action=='integrate':return integrate(repo,args)
     elif args.action=='deploy':return deploy_cmd(repo,args)
+    elif args.action=='claim':
+        if args.claim_action=='acquire':return claim_acquire(repo,args)
+        if args.claim_action=='release':return claim_release(repo,args)
+        if args.claim_action=='status':return claim_status(repo,args)
+        if args.claim_action=='list':return claim_list(repo,args)
+        if args.claim_action=='cancel':return claim_cancel(repo,args)
     return 0
 
 

scripts/tests/test_agentctl.py

@@ -789,6 +789,118 @@ class AgentctlTests(unittest.TestCase):
         self.assertNotEqual(denied.returncode, 0)
         self.assertIn("POLICY_MISMATCH", denied.stderr)
 
+    def _register_claim_task(self, name, scopes=("src",)):
+        path = self.root / ".worktrees" / name
+        if not path.exists():
+            self.git("worktree", "add", "-b", "feature/" + name, str(path), "dev")
+        args = ["task", "register", "--id", name, "--owner", "ClaimAgent"]
+        for scope in scopes:
+            args.extend(("--scope", scope))
+        self.cli(*args, cwd=path)
+        return path
+
+    def test_claim_concurrent_twenty_exactly_one_held(self):
+        import concurrent.futures
+        self.cli("init")
+        (self.root / "src").mkdir(exist_ok=True)
+        tasks = []
+        for idx in range(20):
+            name = f"c{idx:02d}"
+            self._register_claim_task(name, scopes=("src",))
+            tasks.append(name)
+
+        def attempt(name):
+            result = subprocess.run(
+                [sys.executable, str(self.root / "scripts/agentctl.py"),
+                 "claim", "acquire", "--task", name, "--resource", "claim:src/shared",
+                 "--request-id", f"req-{name}"],
+                cwd=self.root / ".worktrees" / name, text=True, capture_output=True, timeout=30)
+            return name, result.returncode, result.stdout, result.stderr
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
+            outcomes = list(pool.map(attempt, tasks))
+        held = [item for item in outcomes if item[1] == 0]
+        busy = [item for item in outcomes if item[1] != 0]
+        self.assertEqual(len(held), 1, outcomes)
+        self.assertEqual(len(busy), 19)
+        for _, code, _out, err in busy:
+            self.assertEqual(code, 75)
+            payload = json.loads(err)
+            self.assertEqual(payload["code"], "LOCK_BUSY")
+            self.assertIn("holder", payload)
+            self.assertIn("ticket", payload)
+            self.assertIn("retry_after", payload)
+        listed = json.loads(self.cli("claim", "list").stdout)
+        held_rows = [row for row in listed["claims"] if row["state"] == "held"]
+        self.assertEqual(len(held_rows), 1)
+        self.assertEqual(held_rows[0]["resource"], "claim:src/shared")
+
+    def test_claim_unrelated_resources_parallel_ok(self):
+        alpha = self.initialized_task("alpha", scopes=("src/a", "src/b"))
+        beta = self.initialized_task("beta", scopes=("src/a", "src/b"))
+        a = json.loads(self.cli(
+            "claim", "acquire", "--task", "alpha", "--resource", "claim:src/a",
+            "--request-id", "par-a", cwd=alpha).stdout)
+        b = json.loads(self.cli(
+            "claim", "acquire", "--task", "beta", "--resource", "claim:src/b",
+            "--request-id", "par-b", cwd=beta).stdout)
+        self.assertEqual(a["status"], "held")
+        self.assertEqual(b["status"], "held")
+
+    def test_claim_parent_child_overlap_and_sibling_ok(self):
+        alpha = self.initialized_task("alpha", scopes=("src", "src2"))
+        beta = self.initialized_task("beta", scopes=("src", "src2"))
+        self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src",
+                 "--request-id", "parent-1", cwd=alpha)
+        child = self.cli(
+            "claim", "acquire", "--task", "beta", "--resource", "claim:src/a.py",
+            "--request-id", "child-1", cwd=beta, check=False)
+        self.assertEqual(child.returncode, 75)
+        self.assertIn("LOCK_BUSY", child.stderr)
+        sibling = json.loads(self.cli(
+            "claim", "acquire", "--task", "beta", "--resource", "claim:src2",
+            "--request-id", "sib-1", cwd=beta).stdout)
+        self.assertEqual(sibling["status"], "held")
+
+    def test_claim_frozen_rejects_even_holder(self):
+        alpha = self.initialized_task("alpha", scopes=("src",))
+        self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src/locked",
+                 "--request-id", "fz-1", cwd=alpha)
+        self.cli("freeze", "--task", "alpha", "--resource", "claim:src/locked",
+                 "--reason", "maintenance", cwd=alpha)
+        denied = self.cli(
+            "claim", "acquire", "--task", "alpha", "--resource", "claim:src/locked",
+            "--request-id", "fz-2", cwd=alpha, check=False)
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("FROZEN", denied.stderr)
+
+    def test_claim_request_id_idempotent_and_reuse_guard(self):
+        alpha = self.initialized_task("alpha", scopes=("src",))
+        first = json.loads(self.cli(
+            "claim", "acquire", "--task", "alpha", "--resource", "claim:src/idem",
+            "--request-id", "same-req", cwd=alpha).stdout)
+        self.assertEqual(first["status"], "held")
+        replay = json.loads(self.cli(
+            "claim", "acquire", "--task", "alpha", "--resource", "claim:src/idem",
+            "--request-id", "same-req", cwd=alpha).stdout)
+        self.assertTrue(replay.get("replayed"))
+        reused = self.cli(
+            "claim", "acquire", "--task", "alpha", "--resource", "claim:src/other",
+            "--request-id", "same-req", cwd=alpha, check=False)
+        self.assertNotEqual(reused.returncode, 0)
+        self.assertIn("REQUEST_ID_REUSED", reused.stderr)
+
+    def test_claim_blocks_task_close_until_released(self):
+        alpha = self.initialized_task("alpha", scopes=("src",))
+        self.cli("claim", "acquire", "--task", "alpha", "--resource", "claim:src/close",
+                 "--request-id", "close-1", cwd=alpha)
+        denied = self.cli("task", "close", "--id", "alpha", check=False)
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("LOCK_BUSY", denied.stderr)
+        self.cli("claim", "release", "--task", "alpha", "--resource", "claim:src/close", cwd=alpha)
+        closed = json.loads(self.cli("task", "close", "--id", "alpha").stdout)
+        self.assertEqual(closed["status"], "done")
+
 
 class ProcessPlatformTests(unittest.TestCase):
     @classmethod