diff --git a/frontends/conductor.py b/frontends/conductor.py
index 00681234..3471d4bf 100644
--- a/frontends/conductor.py
+++ b/frontends/conductor.py
@@ -1,13 +1,4 @@
-import os
-import sys
-import re
-import time
-import json
-import uuid
-import queue
-import asyncio
-import threading
-import builtins
+import os, sys, re, time, json, uuid, queue, asyncio, threading, builtins
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, List
@@ -37,21 +28,17 @@ HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "conductor.
app = FastAPI(title="Conductor")
-
class ChatIn(BaseModel):
msg: str
role: str = "conductor" # conductor | system | user
-
class StartSubagentIn(BaseModel):
prompt: str
-
class SubagentActionIn(BaseModel):
action: str = "intervene" # intervene | abort | kill
msg: str = ""
-
@dataclass
class SubAgentState:
id: str
@@ -64,32 +51,25 @@ class SubAgentState:
last_done: str = ""
monitor_threads: List[threading.Thread] = field(default_factory=list)
-
subagents: Dict[str, SubAgentState] = {}
sub_lock = threading.RLock()
ws_clients: set[WebSocket] = set()
main_loop: Optional[asyncio.AbstractEventLoop] = None
-
# conductor event queue: only user messages and subagent-done events enter here.
conductor_events: "queue.Queue[dict]" = queue.Queue()
conductor_agent: Optional[GenericAgent] = None
conductor_started = False
-
chat_messages: List[dict] = []
-
def now_ms() -> int:
return int(time.time() * 1000)
-
def short_id() -> str:
return uuid.uuid4().hex[:8]
-
_TURN_SPLIT_RE = re.compile(r'\**LLM Running \(Turn \d+\) \.\.\.\**')
_SUMMARY_RE = re.compile(r'(.*?)\s*', re.DOTALL)
-
def extract_last_summary(full: str) -> str:
"""Extract the latest content for in-progress display."""
matches = _SUMMARY_RE.findall(full or "")
@@ -98,7 +78,6 @@ def extract_last_summary(full: str) -> str:
s = matches[-1].strip()
return s[-1000:] if len(s) > 1000 else s
-
def extract_last_text_reply(full: str) -> str:
"""Extract only the last turn's text reply (like stapp.py fold_turns logic)."""
# Split by turn markers, take last segment
@@ -113,7 +92,6 @@ def extract_last_text_reply(full: str) -> str:
# Cap length
return last[-3000:] if len(last) > 3000 else last
-
def subagent_snapshot() -> list[dict]:
with sub_lock:
return [
@@ -126,14 +104,13 @@ def subagent_snapshot() -> list[dict]:
"updated_at": s.updated_at,
}
for s in subagents.values()
+ if s.status != "aborted"
]
-
def schedule_broadcast(payload: dict):
if main_loop and main_loop.is_running():
asyncio.run_coroutine_threadsafe(broadcast(payload), main_loop)
-
async def broadcast(payload: dict):
dead = []
for ws in list(ws_clients):
@@ -144,11 +121,9 @@ async def broadcast(payload: dict):
for ws in dead:
ws_clients.discard(ws)
-
def push_cards():
schedule_broadcast({"type": "subagents", "items": subagent_snapshot()})
-
def add_chat(msg: str, role: str = "conductor"):
item = {"id": short_id(), "role": role, "msg": msg, "ts": now_ms(), "read": role != "user"}
chat_messages.append(item)
@@ -157,13 +132,11 @@ def add_chat(msg: str, role: str = "conductor"):
schedule_broadcast({"type": "chat", "item": item})
return item
-
def start_agent_runner(agent: GenericAgent, name: str):
t = threading.Thread(target=agent.run, name=name, daemon=True)
t.start()
return t
-
def monitor_display_queue(agent_id: str, dq: "queue.Queue", trigger_when_done: bool):
"""Consume one GenericAgent display_queue.
@@ -199,7 +172,6 @@ def monitor_display_queue(agent_id: str, dq: "queue.Queue", trigger_when_done: b
conductor_events.put({"type": "subagent_done", "id": agent_id, "reply": done})
break
-
def start_subagent(prompt: str) -> dict:
sid = short_id()
agent = GenericAgent()
@@ -217,7 +189,6 @@ def start_subagent(prompt: str) -> dict:
push_cards()
return {"id": sid, "status": "running"}
-
def keyinfo_subagent(sid: str, msg: str) -> dict:
"""Inject into agent's working key_info; visible from next turn onward."""
with sub_lock:
@@ -229,7 +200,6 @@ def keyinfo_subagent(sid: str, msg: str) -> dict:
s.updated_at = time.time()
return {"id": sid, "status": "keyinfo_injected"}
-
def input_subagent(sid: str, msg: str) -> dict:
"""Start a new task round (used for input/reply when agent is stopped)."""
with sub_lock:
@@ -249,7 +219,6 @@ def input_subagent(sid: str, msg: str) -> dict:
push_cards()
return {"id": sid, "status": "running"}
-
def conductor_readme() -> str:
base = f"http://{HOST}:{PORT}"
return "\n".join([
@@ -267,8 +236,6 @@ def conductor_readme() -> str:
"触发时机: 用户新消息 | subagent done",
])
-
-
def conductor_prompt_from_events(events: list) -> str:
# 极简摘要：subagent数量和状态
with sub_lock:
@@ -278,46 +245,64 @@ def conductor_prompt_from_events(events: list) -> str:
done_count = sum(1 for e in events if e.get("type") == "subagent_done")
summary = f"subagents: {running} running, {stopped} stopped | {unread}条用户未读消息, {done_count}个subagent完成报告"
base = f"http://{HOST}:{PORT}"
- return f"""你是agent总管。用户只和你对话，你负责一切执行细节。
-API: {base}
+ return f"""你是agent总管。用户只和你对话，你负责调度、验收、交付，目标是降低用户管理多个agent的负担。
+API: {base}；先requests，GET /readme查用法，GET /chat读未读对话，GET /subagent看状态；POST /chat是唯一对用户说话方式。
铁律：
-- 你绝不亲自执行任何任务，一切工作必须通过POST /subagent分派。你只做调度、审查、沟通。
-- 分派完毕后立刻结束本轮回复（停下来），等待下次轮询唤醒你。绝不阻塞等待subagent结果。
-- 每次唤醒只做最小必要动作（发消息/开subagent/reply subagent），然后立刻停。
+- 绝不亲自执行任务/探测环境；一切执行交给subagent。你只分析、派遣、审查、沟通。
+- 每次唤醒只做最小必要动作（发消息/开subagent/reply/keyinfo/abort），做完立刻停，等待下次事件唤醒。
+- 改写prompt时严禁添加用户未提及的假设、工具、前提条件。只能精炼/结构化用户原意，不能脑补，只能做很小的改写
-职责：
-- 理解用户意图，从记忆和上下文揣摩真实需要，不反复确认显而易见的事
-- 涉及改源码、删数据、安全敏感操作，需要改写prompt让subagent先产生方案，然后二次确认执行
-- 不准自己探测环境！！！快速转派给subagent以处理用户下一个指令
+用户消息流程：
+1. 结合记忆、上下文和用户偏好判断真实需求；不清楚/不能代劳时，用精简checklist一次性问用户。
+2. 判断是新任务还是延续现有任务；优先复用已有stopped subagent（用input追加），只有确实无关的新任务才新建。
+3. 分派前必须POST /chat告知用户：改写后的prompt + 分派方案（新建/复用哪个subagent）。
+4. 执行分派，完成即停。危险操作（改源码/删数据/安全敏感）必须改成先让subagent出方案；你验收后POST /chat请用户确认，确认后才继续执行。
-- 判断用户意图是新任务还是对现有任务的延续，小心分派subagent
-- 必要时对subagent追加指令或keyinfo引导
-- 有无法确定/无法代劳的事项，列精简checklist一次性问用户（不要一个个问）
-- 任务完成后给用户简洁报告
+subagent完成流程：
+1. 读subagent输出；若最后一条不足以判断，GET /subagent或日志补足信息。
+2. 预测用户是否满意；不满意就reply/keyinfo要求返工、修改、优化，继续监督，不急着报告。
+3. 预计用户满意后，POST /chat给简洁交付报告。
原则：
-- 能自己决定的自己决定，只在真正需要用户判断时才打扰
-- 验收优先：subagent完成后，先自己审查结果，有问题直接reply让它继续修，反复直到质量过关再报告用户
-- POST /chat 是你和用户说话的唯一方式（必须调接口，不要只在回复里写）
-- GET /readme 查API用法，GET /chat 看对话历史，GET /subagent 看状态
+- 信任subagent足够聪明，不要写具体步骤和容易探测的信息；能自己判断的自己判断，只在真正需要用户决策时打扰。
{summary}"""
+def _auto_cleanup_loop():
+ """Background: auto-abort stopped subagents idle for >1 hour."""
+ IDLE_TIMEOUT = 3600 # 1 hour
+ while True:
+ time.sleep(300) # check every 5 minutes
+ now = time.time()
+ to_abort = []
+ with sub_lock:
+ for sid, s in subagents.items():
+ if s.status == "stopped" and (now - s.updated_at) > IDLE_TIMEOUT:
+ to_abort.append((sid, s))
+ for sid, s in to_abort:
+ s.agent.abort()
+ with sub_lock:
+ s.status = "aborted"
+ s.updated_at = now
+ if to_abort:
+ push_cards()
def monitor_conductor_queue(dq: "queue.Queue"):
"""Conductor output is discarded. Visible reply must go through POST /chat only."""
while True:
item = dq.get()
if "done" in item:
+ print(f"Conductor task done")
break
-
def conductor_loop():
global conductor_agent, conductor_started
conductor_agent = GenericAgent()
conductor_agent.inc_out = True
start_agent_runner(conductor_agent, "conductor-agent")
conductor_started = True
+ # Start background cleanup thread
+ threading.Thread(target=_auto_cleanup_loop, name="subagent-cleanup", daemon=True).start()
while True:
# Block until first event arrives
first = conductor_events.get()
@@ -339,33 +324,29 @@ def conductor_loop():
except Exception as e:
add_chat(f"Conductor error: {e}", role="system")
-
@app.on_event("startup")
async def on_startup():
global main_loop
main_loop = asyncio.get_running_loop()
threading.Thread(target=conductor_loop, name="conductor-loop", daemon=True).start()
-
@app.get("/")
def index():
return FileResponse(HTML_PATH)
-
@app.get("/readme")
def readme():
return PlainTextResponse(conductor_readme())
-
@app.get("/subagent")
def list_subagents():
return {"items": subagent_snapshot()}
-
@app.post("/subagent")
def api_start_subagent(body: StartSubagentIn):
- return start_subagent(body.prompt)
-
+ result = start_subagent(body.prompt)
+ result["instruction"] = "Task received. I'll handle it from here. You MUST stop now and end your reply. Wait for next event."
+ return result
@app.post("/subagent/{sid}")
def api_subagent_action(sid: str, body: SubagentActionIn):
@@ -375,15 +356,18 @@ def api_subagent_action(sid: str, body: SubagentActionIn):
return JSONResponse({"error": "subagent not found", "id": sid}, status_code=404)
action = body.action.lower().strip()
if action == "keyinfo":
- return keyinfo_subagent(sid, body.msg)
+ result = keyinfo_subagent(sid, body.msg)
+ result["instruction"] = "Received. I'll incorporate this. You MUST stop now and end your reply."
+ return result
if action in ("input", "reply", "append", "message", "msg"):
- return input_subagent(sid, body.msg)
+ result = input_subagent(sid, body.msg)
+ result["instruction"] = "Task received. I'll handle it from here. You MUST stop now and end your reply."
+ return result
if action == "abort":
s.agent.abort()
s.status = "aborted"
s.updated_at = time.time()
push_cards()
- conductor_events.put({"type": "subagent_aborted", "id": sid})
return {"id": sid, "status": "aborted"}
if action == "kill":
s.agent.abort()
@@ -393,7 +377,6 @@ def api_subagent_action(sid: str, body: SubagentActionIn):
return {"id": sid, "status": "aborted"}
return JSONResponse({"error": f"unknown action: {body.action}"}, status_code=400)
-
@app.get("/chat")
def api_get_chat(last: int = 20):
"""按需拉取最近N条对话，同时标记所有用户消息为已读"""
@@ -403,12 +386,10 @@ def api_get_chat(last: int = 20):
schedule_broadcast({"type": "chat_read"})
return {"items": chat_messages[-last:]}
-
@app.post("/chat")
def api_chat(body: ChatIn):
return add_chat(body.msg, role=body.role)
-
@app.websocket("/ws")
async def websocket(ws: WebSocket):
await ws.accept()
@@ -427,7 +408,6 @@ async def websocket(ws: WebSocket):
finally:
ws_clients.discard(ws)
-
if __name__ == "__main__":
import uvicorn, webbrowser, threading
threading.Timer(1.0, lambda: webbrowser.open(f"http://{HOST}:{PORT}")).start()
diff --git a/llmcore.py b/llmcore.py
index 0a84165f..47994f3e 100644
--- a/llmcore.py
+++ b/llmcore.py
@@ -638,9 +638,11 @@ class NativeClaudeSession(BaseSession):
self._device_id = uuid.uuid4().hex + uuid.uuid4().hex[:32]
self.tools = None
def raw_ask(self, messages):
- messages = _ensure_thinking_blocks(_drop_unsigned_thinking(_fix_messages(messages)), self.model)
if self.max_tokens is None: self.max_tokens = 8192
model = self.model
+ messages = _fix_messages(messages)
+ if 'claude' in model.lower(): messages = _drop_unsigned_thinking(messages)
+ messages = _ensure_thinking_blocks(messages, self.model)
beta_parts = ["claude-code-20250219", "interleaved-thinking-2025-05-14", "redact-thinking-2026-02-12", "prompt-caching-scope-2026-01-05"]
if "[1m]" in model.lower():
beta_parts.insert(1, "context-1m-2025-08-07"); model = model.replace("[1m]", "").replace("[1M]", "")
diff --git a/memory/autonomous_operation_sop/task_planning.md b/memory/autonomous_operation_sop/task_planning.md
index 54f8aa40..209d5184 100644
--- a/memory/autonomous_operation_sop/task_planning.md
+++ b/memory/autonomous_operation_sop/task_planning.md
@@ -23,7 +23,7 @@
## 目标排序（按价值递减）：
1. **实用产出与能力扩展**：写工具解决痛点，在已有能力上解锁新能力（能力树每多一个节点，可能性空间变大）
2. **环境发现**：扫描已有但未利用的工具/库/数据源/配置
-3. **小众工具挖掘**：在GitHub/V2EX/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案
+3. **小众工具挖掘**：在GitHub/吾爱破解/果核剥壳**等**找冷门实用工具，实测AI常推荐但有坑的方案
4. **了解用户与推荐**：分析老代码/PC文件/书签推断偏好，给出个性化推荐（游戏/视频/工具附理由）（低频）
5. **自身演进**：思考框架不足，提出改进方案
6. **记忆审查**：修正错误或过时记录