weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agentctl): add long-lived claim namespace and CLI

Browse Source 
Introduce claim: resources with additive claims/claim_queue tables, atomic
acquire/release under IMMEDIATE transactions, and concurrency regressions.
This commit is contained in:
WeiCHEN
2026-09-23 15:02:00 +08:00
parent 7cd0f505b2
commit 6699c4a8aa
4 changed files with 394 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.doc/ops/CURRENT.md
+1
View File
@@ -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
+46
View File
@@ -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
+235 -3
View File
@@ -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
+112
View File
@@ -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