From 6699c4a8aa5556494f979dfb33e7131194072ef5 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Wed, 23 Sep 2026 15:02:00 +0800 Subject: [PATCH] feat(agentctl): add long-lived claim namespace and CLI Introduce claim: resources with additive claims/claim_queue tables, atomic acquire/release under IMMEDIATE transactions, and concurrency regressions. --- .doc/ops/CURRENT.md | 1 + .../short-term/2026-09-23-agentctl-claim.md | 46 ++++ scripts/agentctl.py | 238 +++++++++++++++++- scripts/tests/test_agentctl.py | 112 +++++++++ 4 files changed, 394 insertions(+), 3 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-23-agentctl-claim.md diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index b862568f..f6e882bf 100644 --- a/.doc/ops/CURRENT.md +++ b/.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 | diff --git a/.doc/ops/short-term/2026-09-23-agentctl-claim.md b/.doc/ops/short-term/2026-09-23-agentctl-claim.md new file mode 100644 index 00000000..ba64f5eb --- /dev/null +++ b/.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) diff --git a/scripts/agentctl.py b/scripts/agentctl.py index 3268d539..749b91d2 100755 --- a/scripts/agentctl.py +++ b/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 diff --git a/scripts/tests/test_agentctl.py b/scripts/tests/test_agentctl.py index 930cdbf5..0133f8e5 100644 --- a/scripts/tests/test_agentctl.py +++ b/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