feat: 命令台子代理链路(主任务/模型子代理/委派子任务 + 过程调用实时同步)

host:链路条目带 thread 归属;delegate 向父链路文件写 agent-start/done 标记。服务端:GET /tasks/{id}/agents 聚合父链路 + 子链路文件成代理链(状态口径:委派以结束标记为准、无标记即进行中;模型子代理无结束信号时父任务运行中为进行中、否则未知)。命令台:新增子代理链路区,随 SSE 流防抖刷新;扁平链路也渲染 agent 标记。
This commit is contained in:
host committed 2026-10-27 11:51:11 +08:00
1 parent 34cd0eeba7
commit c4d3ec9bff
6 files changed
+366 -14

No files matched your search

+144
View File
@@ -184,6 +184,127 @@ def _read_trace_entries(path):
except OSError:
return []
return entries
AGENT_CALL_LIMIT = 1000
def _read_agent_entries(path):
"""读链路文件供子代理聚合:按 (thread, id) 去重,后到的完成态
覆盖先到的开始态,保首次顺序;坏行与无 id 行跳过;文件不
存在为空。同一 item id 在不同线程出现时不互相覆盖(模型内
子代理与主任务各有自己的线程)。"""
entries = []
index = {}
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
except ValueError:
continue
if not isinstance(entry, dict) or "id" not in entry:
continue
key = (entry.get("thread") or "", entry["id"])
if key in index:
entries[index[key]] = entry
else:
index[key] = len(entries)
entries.append(entry)
except OSError:
return []
return entries
def _latest_ts(values):
vals = [v for v in values if isinstance(v, (int, float))]
return max(vals) if vals else None
def _agent_status_from_task(status):
return {"queued": "running", "running": "running",
"completed": "completed", "merged": "completed",
"failed": "failed", "merge-failed": "failed",
"interrupted": "interrupted"}.get(status, "unknown")
def _build_agent_chain(task, trace_dir):
"""任务 → 子代理链:主任务 + 模型内子代理(同文件多线程)+
委派子任务(父链路里的 agent-start/done 标记 + 子链路文件)。
状态诚实口径:委派子任务以 agent-done 标记为准,无标记即
进行中——委派进程与父任务相互独立,父任务结束不代表子任务
停;模型内子代理无明确结束信号,父任务运行中即进行中,
否则未知。无 thread 的旧条目一律归主任务。
"""
parent_path = os.path.join(
trace_dir, _trace_file_name(task["id"]) + ".jsonl")
entries = _read_agent_entries(parent_path)
calls = [e for e in entries
if e.get("kind") not in ("agent-start", "agent-done")]
main_thread = None
for e in calls:
if e.get("thread"):
main_thread = e["thread"]
break
root_calls = [e for e in calls
if not e.get("thread") or e.get("thread") == main_thread]
agents = [{
"id": "main", "name": "主任务", "kind": "main", "parent": None,
"status": _agent_status_from_task(task.get("status")),
"prompt": None, "trace": None,
"calls": root_calls[:AGENT_CALL_LIMIT],
"updated_ts": _latest_ts([e.get("ts") for e in root_calls]),
}]
task_running = task.get("status") in ("queued", "running")
threads = {}
for e in calls:
t = e.get("thread")
if t and t != main_thread:
threads.setdefault(t, []).append(e)
for t, es in threads.items():
agents.append({
"id": t, "name": "模型子代理 " + t[:12], "kind": "model",
"parent": "main",
"status": "running" if task_running else "unknown",
"prompt": None, "trace": None,
"calls": es[:AGENT_CALL_LIMIT],
"updated_ts": _latest_ts([e.get("ts") for e in es]),
})
starts = {}
dones = {}
for e in entries:
if e.get("kind") == "agent-start" and e.get("name"):
starts[e["name"]] = e
elif e.get("kind") == "agent-done" and e.get("name"):
dones[e["name"]] = e
for name, s in starts.items():
trace_name = s.get("trace") or dones.get(name, {}).get("trace")
child_calls = []
if trace_name:
child_path = os.path.join(
trace_dir, os.path.basename(trace_name))
child_calls = [e for e in _read_agent_entries(child_path)
if e.get("kind") not in
("agent-start", "agent-done")]
done = dones.get(name)
status = "running"
if done and done.get("status") in (
"completed", "failed", "interrupted"):
status = done["status"]
agents.append({
"id": "delegate:" + name, "name": name, "kind": "delegate",
"parent": "main", "status": status,
"prompt": s.get("prompt"), "trace": trace_name,
"calls": child_calls[:AGENT_CALL_LIMIT],
"updated_ts": _latest_ts(
[e.get("ts") for e in child_calls] +
[s.get("ts"), done.get("ts") if done else None]),
})
return agents
DEFAULT_SWEEP_THRESHOLD_SECONDS = 300
# 限流默认值:每 IP / 每 API key 每分钟 120 次;环境变量
# CLOUD_RATE_LIMIT_PER_MINUTE / CLOUD_RATE_LIMIT_WINDOW_SECONDS 可调。
@@ -1262,6 +1383,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, rest[: -len("/trace/stream")])
elif rest.endswith("/trace") and method == "GET":
self._task_trace(user_id, rest[: -len("/trace")])
elif rest.endswith("/agents") and method == "GET":
self._task_agents(user_id, rest[: -len("/agents")])
elif rest.endswith("/cancel") and method == "POST":
self._cancel_task(user_id, rest[: -len("/cancel")])
elif rest.endswith("/merge") and method == "POST":
@@ -2471,6 +2594,27 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(200, {"task_id": task_id, "status": task["status"],
"entries": entries, "note": note})
def _task_agents(self, user_id, task_id):
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
with _auth_store(self.server.db_path) as store:
if not store.check_permission(user_id, task["project_id"], "read"):
self._send_json(403, {"error": "forbidden"})
return
worktree_host = task.get("worktree_host")
if worktree_host and worktree_host != socket.gethostname():
self._send_json(200, {"task_id": task_id, "status": task["status"],
"agents": [],
"note": ("链路文件在 worktree 主机 %s 上,本机读不到"
% worktree_host)})
return
agents = _build_agent_chain(task, _trace_dir_for(self.server))
self._send_json(200, {"task_id": task_id, "status": task["status"],
"agents": agents, "note": ""})
def _task_trace_stream(self, user_id, task_id):
row = self._fetch_task(task_id)
if row is None: