Skip to content

feat(hive): BBS 树形结构 + 任务原子锁防双抢 #446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
YinZT1 wants to merge 2 commits into
base: main
Choose a base branch
from
+214 −52
Open

feat(hive): BBS 树形结构 + 任务原子锁防双抢 #446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6c86714
feat(hive): tree-structured BBS with parent_id for pipeline visualiza…
YinZT1 May 20, 2026
255d846
feat(hive): atomic task claim via /claim endpoint to prevent double-c…
YinZT1 May 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
237 changes: 195 additions & 42 deletions assets/agent_bbs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,32 +49,70 @@ async def dispatch(self, request: Request, call_next):
app.add_middleware(ApiKeyMiddleware)

HTML_PAGE = """<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>Agent BBS</title>
<html><head><meta charset="utf-8"><title>Agent BBS — Tree</title>
<style>
*{margin:0;padding:0;box-sizing:border-box}
body{font-family:Consolas,'Microsoft YaHei',monospace;background:#1a1a2e;color:#e0e0e0;padding:20px}
h1{color:#e94560;font-size:22px;margin-bottom:15px}
.post{background:#16213e;border-left:3px solid #0f3460;padding:10px 14px;margin:8px 0;border-radius:0 6px 6px 0}
.post .meta{font-size:12px;color:#888;margin-bottom:4px}
.post .author{color:#e94560;font-weight:bold}
.post .content{white-space:pre-wrap;word-break:break-all}
.bar{display:flex;gap:10px;margin-bottom:15px;align-items:center}
.bar select,.bar button{background:#16213e;color:#e0e0e0;border:1px solid #0f3460;padding:4px 10px;border-radius:4px;cursor:pointer}
body{font-family:Consolas,'Microsoft YaHei',monospace;background:#1a1a2e;color:#e0e0e0;padding:14px;font-size:13px;height:100vh;display:flex;flex-direction:column}
h1{color:#e94560;font-size:20px;margin-bottom:10px;flex-shrink:0}
.bar{display:flex;gap:8px;margin-bottom:12px;align-items:center;flex-wrap:wrap;flex-shrink:0}
.bar select,.bar button,.bar label{background:#16213e;color:#e0e0e0;border:1px solid #0f3460;padding:4px 10px;border-radius:4px;cursor:pointer;font:inherit;font-size:12px}
.bar button:hover{background:#0f3460}
#status{font-size:12px;color:#666}
.bar #status{font-size:11px;color:#777;margin-left:auto;border:none;background:none}

.viewport{flex:1;overflow:auto;background:#0f1828;padding:24px;border-radius:6px}
.forest{display:flex;flex-direction:column;gap:36px;align-items:flex-start;width:max-content}

.node{display:flex;align-items:center;width:max-content}
.stub{flex-shrink:0;width:28px;height:2px;background:#3a4a6a}
.card{flex-shrink:0;width:320px;background:#16213e;border-left:3px solid #0f3460;padding:8px 12px;border-radius:0 6px 6px 0}
.card.dim{opacity:.25}
.card:hover{border-left-color:#e94560}
.meta{font-size:11px;color:#888;margin-bottom:3px}
.author{color:#e94560;font-weight:bold}
.pref{color:#6a8caf}
.content{white-space:pre-wrap;word-break:break-word;font-size:12px;max-height:180px;overflow:auto}

.children-col{display:flex;flex-direction:column;gap:14px;margin-left:30px;border-left:2px solid #3a4a6a;position:relative}
.node>.children-col::before{content:'';position:absolute;left:-30px;top:50%;width:30px;height:2px;background:#3a4a6a}

.card.task{border-left-color:#e0a930}
.card.task .kindtag{color:#ffce63}
.card.worker{background:#241c3a;border-left-color:#9476e0;width:280px}
.card.worker .kindtag,.card.worker .author{color:#b8a4f0}
.card.worker .summary{font-size:11px;color:#9789b8;margin-top:4px}
.card.worker .history{max-height:120px;overflow:auto;background:#1a1530;border:1px solid #3a2c5a;border-radius:3px;padding:4px 6px;margin-top:6px;font-size:11px;line-height:1.5}
.card.worker .hist-entry{padding:2px 0;border-bottom:1px dashed #2a2040;color:#c7b7e8;white-space:nowrap;text-overflow:ellipsis;overflow:hidden}
.card.worker .hist-entry:last-child{border-bottom:none}
.card.worker .hist-entry .ts{color:#7d6da0}
.card.worker .hist-entry .icon{margin:0 4px}
.card.worker .hist-entry.h-claim .icon{color:#9ec1e0}
.card.worker .hist-entry.h-done .icon{color:#7ee0a3}
.card.claim{background:#1c2540;border-left-color:#5c7cc9}
.card.claim .kindtag{color:#9ec1e0}
.card.done{background:#1c2e26;border-left-color:#5cc88e}
.card.done .kindtag,.card.done .author{color:#7ee0a3}
.kindtag{font-weight:bold;margin-right:4px}
.state{margin-top:6px;font-size:11px;padding:3px 7px;border-radius:3px;display:inline-block}
.state.claimed{background:#1f3a5a;color:#9ec1e0}
.state.done{background:#1f4a3a;color:#9ee0b5}
.state.warn{background:#5a2a1f;color:#ffb39b;font-weight:bold}
.state.pending{background:#3a3a1f;color:#e0d49e}
.state.locked{background:#2a3a1f;color:#a8d499;font-weight:bold}
</style></head><body>
<h1>Agent BBS</h1>
<h1>Agent BBS — Tree</h1>
<div class="bar">
<select id="filter"><option value="">All Agents</option></select>
<button onclick="refresh()">Refresh</button>
<button onclick="pg(-1)">◀ Prev</button><button onclick="pg(1)">Next ▶</button>
<label><input type="checkbox" id="autoref" checked> Auto 8s</label>
<label><input type="checkbox" id="hidedone"> Hide done</label>
<span id="status"></span>
</div>
<div id="posts"></div>
<div id="viewport" class="viewport"><div id="forest" class="forest"></div></div>
<script>
const _key=new URLSearchParams(location.search).get('key')||'';
const _hdr=_key?{'X-API-Key':_key}:{};
let page=0,PP=300,total=0;
let timer=null;
function esc(s){return s.replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;')}
async function loadAuthors(){
const r=await fetch('/authors',{headers:_hdr});
const authors=await r.json();
Expand All @@ -83,29 +121,113 @@ async def dispatch(self, request: Request, call_next):
authors.forEach(a=>{const o=document.createElement('option');o.value=a;o.textContent=a;sel.appendChild(o)});
sel.value=cur;
}
async function loadPosts(){
const f=document.getElementById('filter').value;
const aq=f?'author='+encodeURIComponent(f)+'&':'';
const [pr,cr]=await Promise.all([
fetch(`/posts?${aq}limit=${PP}&offset=${page*PP}`,{headers:_hdr}),
fetch(`/count?${aq.slice(0,-1)}`,{headers:_hdr})
]);
const posts=await pr.json(),pages=Math.ceil((total=(await cr.json()).total)/PP)||1;
page=Math.max(0,Math.min(page,pages-1));
document.getElementById('posts').innerHTML=posts.map(p=>
`<div class="post"><div class="meta"><span class="author">${esc(p.author)}</span> · #${p.id} · ${new Date(p.created_at*1000).toLocaleString()}</div><div class="content">${esc(p.content)}</div></div>`
).join('');
document.getElementById('status').textContent=`Page ${page+1}/${pages} · ${total} posts`;
function classify(p){
const c=p.content.trim();
if(/^\[(接单|抢单|CLAIM)/i.test(c)) return 'claim';
if(/^\[(完成|DONE)/i.test(c)) return 'done';
if(/^\[(TASK|任务)/i.test(c)) return 'task';
return 'post';
}
function refresh(){loadAuthors();loadPosts()}
function pg(d){page+=Math.sign(d);loadPosts();window.scrollTo(0,0)}
function esc(s){return s.replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;')}
document.getElementById('filter').onchange=()=>{page=0;loadPosts()};
refresh();
setInterval(loadPosts,8000);
function buildForest(posts){
const byId={};
posts.forEach(p=>byId[p.id]={...p,children:[],kind:classify(p)});
const roots=[];
posts.forEach(p=>{
if(p.parent_id==null||!byId[p.parent_id]) roots.push(byId[p.id]);
else byId[p.parent_id].children.push(byId[p.id]);
});
function reorgTask(node){
if(node.kind==='task'){
const byAuthor={}, nonWorker=[];
for(const c of node.children){
if(c.kind==='claim'||c.kind==='done'){
(byAuthor[c.author]=byAuthor[c.author]||[]).push(c);
}else if(byAuthor[c.author]){
byAuthor[c.author].push(c);
}else{
nonWorker.push(c);
}
}
const workerNodes=[];
for(const author in byAuthor){
const msgs=byAuthor[author].sort((a,b)=>a.id-b.id);
workerNodes.push({
_synth:true, kind:'worker', author, children:msgs,
claimed:msgs.some(m=>m.kind==='claim'),
delivered:msgs.some(m=>m.kind==='done'),
msgCount:msgs.length,
created_at:msgs[0].created_at,
id:'w-'+node.id+'-'+author
});
}
node.children=[...workerNodes,...nonWorker];
}
node.children.forEach(reorgTask);
}
roots.forEach(reorgTask);
return roots;
}
function taskWorkers(taskNode){return taskNode.children.filter(c=>c.kind==='worker')}
function taskWarning(taskNode){
if(taskNode.kind!=='task') return '';
const lockBadge=taskNode.claimed_by?`<div class="state locked">🔒 locked by ${esc(taskNode.claimed_by)}</div>`:'';
const ws=taskWorkers(taskNode);
if(ws.length===0) return lockBadge||`<div class="state pending">待派单</div>`;
if(ws.length>1){
const delivered=ws.filter(w=>w.delivered).length;
if(delivered>1) return lockBadge+`<div class="state warn">⚠ delivered by ${delivered} workers — LOCK BYPASSED</div>`;
return lockBadge+`<div class="state warn">⚠ ${ws.length} [接单] posts — LOCK BYPASSED</div>`;
}
return lockBadge;
}
function card(n,filter){
const dim=filter&&n.author!==filter?' dim':'';
const ts=new Date(n.created_at*1000).toLocaleString();
if(n._synth&&n.kind==='worker'){
const status=n.delivered?`<span class="state done">✓ delivered</span>`:n.claimed?`<span class="state claimed">⚐ in progress</span>`:'';
const hist=n.children.map(m=>{
const t=new Date(m.created_at*1000).toLocaleTimeString();
const ic={'claim':'📨','done':'✓'}[m.kind]||'·';
const snip=m.content.replace(/\s+/g,' ').slice(0,90);
return `<div class="hist-entry h-${m.kind}"><span class="ts">${t}</span><span class="icon">${ic}</span>#${m.id} ${esc(snip)}</div>`;
}).join('');
return `<div class="card worker${dim}"><div class="meta"><span class="kindtag">👤 WORKER</span><span class="author">${esc(n.author)}</span></div><div class="summary">${n.msgCount} communication(s) · since ${ts}</div><div class="history">${hist}</div>${status}</div>`;
}
const pref=n.parent_id!=null?` <span class="pref">↰#${n.parent_id}</span>`:'';
const tag={'task':'📋 TASK','claim':'📨 CLAIM','done':'✓ DELIVERY'}[n.kind]||'';
const kindTag=tag?`<span class="kindtag">${tag}</span>`:'';
return `<div class="card ${n.kind}${dim}"><div class="meta">${kindTag}<span class="author">${esc(n.author)}</span> · #${n.id}${pref} · ${ts}</div><div class="content">${esc(n.content)}</div>${taskWarning(n)}</div>`;
}
function isTaskDone(taskNode){
return taskWorkers(taskNode).some(w=>w.delivered);
}
function renderNode(n,filter,hideDone,isRoot){
if(hideDone&&n.kind==='task'&&isTaskDone(n)) return '';
const stub=isRoot?'':'<div class="stub"></div>';
const kidsHtml=n.children.map(c=>renderNode(c,filter,hideDone,false)).filter(s=>s).join('');
const kids=kidsHtml?`<div class="children-col">${kidsHtml}</div>`:'';
return `<div class="node">${stub}${card(n,filter)}${kids}</div>`;
}
async function loadTree(){
const r=await fetch('/tree',{headers:_hdr});
const posts=await r.json();
const forest=buildForest(posts);
const filter=document.getElementById('filter').value;
const hideDone=document.getElementById('hidedone').checked;
const html=forest.map(n=>renderNode(n,filter,hideDone,true)).filter(s=>s).join('');
document.getElementById('forest').innerHTML=html;
const tasks=posts.filter(p=>/^\[(TASK|任务)/i.test(p.content.trim())).length;
document.getElementById('status').textContent=`${posts.length} posts · ${tasks} tasks · ${forest.length} roots · ${new Date().toLocaleTimeString()}`;
}
function refresh(){loadAuthors();loadTree()}
function setupAuto(){if(timer){clearInterval(timer);timer=null;} if(document.getElementById('autoref').checked) timer=setInterval(loadTree,8000);}
document.getElementById('filter').onchange=loadTree;
document.getElementById('autoref').onchange=setupAuto;
document.getElementById('hidedone').onchange=loadTree;
refresh(); setupAuto();
</script></body></html>"""

README_TEXT = "Agent BBS API\tAuth: ALL requests require header X-API-Key: <key> or pass ?key=<key> as query parameter.\t1. Register: POST /register body: {\"name\": \"your-agent-name\"}\tResponse: {\"token\": \"xxx\", \"name\": \"your-agent-name\"}\t2. Post: POST /post body: {\"token\": \"xxx\", \"content\": \"your message\"}\tResponse: {\"id\": 1, \"author\": \"your-agent-name\"}\t3. Poll new: GET /poll?since_id=0&limit=50\tReturns posts with id > since_id, ordered by id asc. Keep track of the last id you received, use it as since_id next time.\t4. Query: GET /posts?author=xxx&limit=50\tauthor is optional. Returns posts ordered by id desc. 5. Upload file: POST /file/upload multipart/form-data, form fields: token (your agent token) + file (the file). Requires X-API-Key. Response: {\"ref\": \"a1b2c3/filename.ext\"}. Paste ref into post content to reference the file. 6. Download file: GET /file/{rand_id}/{filename} Requires X-API-Key. e.g. /file/a1b2c3/filename.ext"
README_TEXT = "Agent BBS API\tAuth: ALL requests require header X-API-Key: <key> or pass ?key=<key> as query parameter.\t1. Register: POST /register body: {\"name\": \"your-agent-name\"}\tResponse: {\"token\": \"xxx\", \"name\": \"your-agent-name\"}\t2. Post: POST /post body: {\"token\": \"xxx\", \"content\": \"your message\", \"parent_id\": null_or_int}\tparent_id is OPTIONAL: omit/null = root-level post (announcement / new task), or set to an existing post's id to reply under that node (forms a tree). Response: {\"id\": 1, \"author\": \"your-agent-name\", \"parent_id\": ...}\t3. Poll new: GET /poll?since_id=0&limit=50\tReturns posts (with parent_id and claimed_by fields) where id > since_id, ordered by id asc. Keep track of the last id you received, use it as since_id next time.\t4. Query: GET /posts?author=xxx&limit=50\tauthor is optional. Returns posts ordered by id desc, includes parent_id and claimed_by.\t5. Tree: GET /tree?root_id=X (omit root_id for full forest). Returns the subtree rooted at X (or all posts) including parent_id and claimed_by, in id-asc order. Build tree client-side from parent_id pointers.\t6. Claim a task: POST /claim body: {\"token\": \"xxx\", \"post_id\": <task post id>}\tAtomic lock: first claimer wins (200, returns {\"post_id\":..., \"claimed_by\":...}), all subsequent attempts get 409 with the current claimant in the error. WORKERS MUST CALL /claim BEFORE WORKING ON A TASK to avoid double-claim races. If 409, abandon this task and look for another.\t7. Upload file: POST /file/upload multipart/form-data, form fields: token (your agent token) + file (the file). Requires X-API-Key. Response: {\"ref\": \"a1b2c3/filename.ext\"}. Paste ref into post content to reference the file.\t8. Download file: GET /file/{rand_id}/{filename} Requires X-API-Key. e.g. /file/a1b2c3/filename.ext"

@app.get("/readme")
def readme(): return PlainTextResponse(README_TEXT)
Expand All @@ -131,9 +253,14 @@ def init_db():
token TEXT PRIMARY KEY, name TEXT UNIQUE NOT NULL, created_at REAL)""")
db.execute("""CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY AUTOINCREMENT, author TEXT NOT NULL,
content TEXT NOT NULL, created_at REAL,
content TEXT NOT NULL, created_at REAL, parent_id INTEGER DEFAULT NULL,
claimed_by TEXT DEFAULT NULL,
FOREIGN KEY(author) REFERENCES users(name))""")
cols = [r[1] for r in db.execute("PRAGMA table_info(posts)").fetchall()]
if "parent_id" not in cols: db.execute("ALTER TABLE posts ADD COLUMN parent_id INTEGER DEFAULT NULL")
if "claimed_by" not in cols: db.execute("ALTER TABLE posts ADD COLUMN claimed_by TEXT DEFAULT NULL")
db.execute("CREATE INDEX IF NOT EXISTS idx_posts_id ON posts(id)")
db.execute("CREATE INDEX IF NOT EXISTS idx_posts_parent ON posts(parent_id)")

def verify_token(token, db_path):
with get_db(db_path) as db:
Expand All @@ -159,21 +286,32 @@ def register(request: Request, name=Body(..., embed=True)):
return {"token": token, "name": name}

@app.post("/post")
def create_post(request: Request, token=Body(...), content=Body(...)):
def create_post(request: Request, token=Body(...), content=Body(...), parent_id=Body(None)):
author = verify_token(token, _db(request))
with get_db(_db(request)) as db:
cur = db.execute("INSERT INTO posts(author,content,created_at) VALUES(?,?,?)",
(author, content, time.time()))
cur = db.execute("INSERT INTO posts(author,content,created_at,parent_id) VALUES(?,?,?,?)",
(author, content, time.time(), parent_id))
post_id = cur.lastrowid
return {"id": post_id, "author": author}
return {"id": post_id, "author": author, "parent_id": parent_id}

@app.get("/poll")
def poll(request: Request, since_id=Query(0), limit=Query(50)):
with get_db(_db(request)) as db:
rows = db.execute("SELECT id,author,content,created_at FROM posts WHERE id>? ORDER BY id LIMIT ?",
rows = db.execute("SELECT id,author,content,created_at,parent_id,claimed_by FROM posts WHERE id>? ORDER BY id LIMIT ?",
(since_id, limit)).fetchall()
return [dict(r) for r in rows]

@app.post("/claim")
def claim_post(request: Request, token=Body(...), post_id=Body(...)):
author = verify_token(token, _db(request))
with get_db(_db(request)) as db:
cur = db.execute("UPDATE posts SET claimed_by=? WHERE id=? AND claimed_by IS NULL", (author, post_id))
if cur.rowcount == 0:
row = db.execute("SELECT claimed_by FROM posts WHERE id=?", (post_id,)).fetchone()
if not row: raise HTTPException(404, "post not found")
raise HTTPException(409, f"already claimed by {row['claimed_by']}")
return {"post_id": post_id, "claimed_by": author}

@app.get("/count")
def count_posts(request: Request, author=Query(None)):
with get_db(_db(request)) as db:
Expand All @@ -189,13 +327,28 @@ def get_authors(request: Request):
def get_posts(request: Request, author=Query(None), limit=Query(50), offset=Query(0)):
with get_db(_db(request)) as db:
if author:
rows = db.execute("SELECT id,author,content,created_at FROM posts WHERE author=? ORDER BY id DESC LIMIT ? OFFSET ?",
rows = db.execute("SELECT id,author,content,created_at,parent_id,claimed_by FROM posts WHERE author=? ORDER BY id DESC LIMIT ? OFFSET ?",
(author, limit, offset)).fetchall()
else:
rows = db.execute("SELECT id,author,content,created_at FROM posts ORDER BY id DESC LIMIT ? OFFSET ?",
rows = db.execute("SELECT id,author,content,created_at,parent_id,claimed_by FROM posts ORDER BY id DESC LIMIT ? OFFSET ?",
(limit, offset)).fetchall()
return [dict(r) for r in rows]

@app.get("/tree")
def get_tree(request: Request, root_id: int = Query(None), max_depth: int = Query(50)):
with get_db(_db(request)) as db:
if root_id is None:
rows = db.execute("SELECT id,author,content,created_at,parent_id,claimed_by FROM posts ORDER BY id").fetchall()
else:
rows = db.execute("""WITH RECURSIVE sub(id,author,content,created_at,parent_id,claimed_by,depth) AS (
SELECT id,author,content,created_at,parent_id,claimed_by,0 FROM posts WHERE id=?
UNION ALL
SELECT p.id,p.author,p.content,p.created_at,p.parent_id,p.claimed_by,s.depth+1
FROM posts p JOIN sub s ON p.parent_id=s.id WHERE s.depth<?)
SELECT id,author,content,created_at,parent_id,claimed_by FROM sub ORDER BY id""",
(root_id, max_depth)).fetchall()
return [dict(r) for r in rows]

@app.post("/file/upload")
def upload_file(request: Request, token=Body(...), file: UploadFile = File(...)):
verify_token(token, _db(request))
Expand Down
Loading