diff --git a/apps/tray/src/rules.rs b/apps/tray/src/rules.rs index 8e5ac9f9..70b098b7 100644 --- a/apps/tray/src/rules.rs +++ b/apps/tray/src/rules.rs @@ -20,6 +20,7 @@ const SHARED: &str = "## AgentNave 编排偏好 - 探索、规划派 Explorer，实现派 Worker。 - 同一组文件同一时间只有一个 Worker 在写；同一职责续接只传 `session_id`，其余由服务继承；换职责时传新的 `role` 或 `seat` 并给自足简报。 - 本偏好不决定何时评审，用户要求或其规则要求时才评审。评审时有 agentnave-review Skill 就按它定级、选工位，不另派 Explorer；没有时派 Explorer，不交给写这段代码的 Worker，尽量换模型。 +- 结果不会主动推送：派发后结束本轮前用 `wait` 收齐全部终态；任务在跑时要先问用户，也先 `wait` 或 `cancel`。 - `succeeded` 不等于完成，boss 核对产物后验收；失败或取消时先核对现状，再决定续接、重派还是收尾。"; const IMAGE_GENERATION: &str = "\n- 生成图片派生图工位，`seat` 用 `image-generator`，prompt 写清图像需求和产物路径，由它的对话模型自行调用生图服务。"; diff --git a/docs/adr/0007-wait-any-ready-subset.md b/docs/adr/0007-wait-any-ready-subset.md index eaf8dff3..8954953c 100644 --- a/docs/adr/0007-wait-any-ready-subset.md +++ b/docs/adr/0007-wait-any-ready-subset.md @@ -8,6 +8,7 @@ - 轮次 `seat_lost` 就绪条件由 [ADR-0039](0039-retry-lost-seat-within-round.md) 补充 - 服务主动终止情形与终态句柄保留七天的边界由 [ADR-0049](0049-service-initiated-termination.md) 补充 - 由 [ADR-0057](0057-boss-answers-native-prompts.md) 补充：待答弹框是新的就绪原因，收件箱同样返回。 +- 由 [ADR-0064](0064-dispatch-receipts-steer-to-wait.md) 修订：派发与等待回执在仍有待收结果时带 `next` 引导，派发调用可选顺带等一次。 ## 背景 diff --git a/docs/adr/0064-dispatch-receipts-steer-to-wait.md b/docs/adr/0064-dispatch-receipts-steer-to-wait.md new file mode 100644 index 00000000..60f16d51 --- /dev/null +++ b/docs/adr/0064-dispatch-receipts-steer-to-wait.md @@ -0,0 +1,15 @@ +# ADR-0064：派发回执引导收取，可选派发后先等 + +- 状态：已接受 +- 日期：2026-11-07 +- 修订 ADR-0007：等待仍由 boss 主动发起，派发调用可以顺带等一次。 + +结束结果只能由 `wait` 收取（ADR-0007），boss 漏掉 `wait` 就会在任务还在跑时结束回合。真实使用中多个模型出现过这种情况：派发回执只写 `running`，模型把它当成会自动回报的后台任务；`unreceived` 只附在下一次派发回执上，模型不再调用时没有机会出现。同类应用（Paseo、t3code、Cindy、Lody）托管 boss 会话，能在子任务结束后把结果注入 boss 并唤醒它；AgentNave 挂在用户自己的宿主上，没有这条通道，各宿主的推送能力也不一致，所以不依赖推送。 + +派发回执在仍有待收结果时带 `next`，写明结束回合前先 `wait`；`wait` 返回时若仍有待收句柄，同样带 `next` 要求再次等待。这是模型做决定时一定会读到的位置，比只写在 Skill 中可靠。 + +`start_agent` 和 `run_round` 提供可选的 `wait`：为真时派发后按 `wait` 的默认时长等待本次句柄，并在回执中返回那次等待的结果，收取语义与单独调用 `wait` 相同。它默认关闭，派发仍立即返回；只用于本回合需要结果的单次派发，并联派发仍应全部派出后统一等待，否则会变成串行。 + +App 写入宿主全局规则的编排段落加入同样的收取要求（ADR-0037），让每轮都可见；自定义规则不改写。 + +不做：托管 boss 会话或依赖某个宿主的推送通道；在托盘提示未收取结果。宿主 Stop 钩子能拦住结束回合，但各宿主的信任和入参不同，另行决定。 diff --git a/docs/adr/README.md b/docs/adr/README.md index a2957354..56b2550a 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -67,3 +67,4 @@ | 0061 | [宿主配置指向固定启动器，升级不再改写宿主配置](0061-fixed-service-launcher.md) | 已接受 | — | | 0062 | [服务统一生成成员称呼](0062-service-owned-member-names.md) | 已接受 | — | | 0063 | [CLI 共用模型池，编排替代普通任务目录](0063-cli-model-pool-and-orchestration-precedence.md) | 已接受 | 修订 ADR-0038、ADR-0053 | +| 0064 | [派发回执引导收取，可选派发后先等](0064-dispatch-receipts-steer-to-wait.md) | 已接受 | 修订 ADR-0007 | diff --git a/skills/agentnave-manager/SKILL.md b/skills/agentnave-manager/SKILL.md index d189a401..34ca8e9f 100644 --- a/skills/agentnave-manager/SKILL.md +++ b/skills/agentnave-manager/SKILL.md @@ -153,7 +153,7 @@ description: 用户要求用 Claude Code、CodeBuddy Code、Codex CLI、Grok CLI ## 等待、验收与取消 1. `start_agent` 保存 `invocation_id`，`run_round` 保存 `round_id`；持续等到终态或显式取消，不另加总时限。 - 终态和阻塞不会主动推送，只能由 `wait` 收取：手头没有不依赖结果的工作时，派发后立刻单独调用 `wait`；有就先完成当前步骤再调用，并在结束本轮前收齐全部终态。常规省略 `wait_seconds`，`wait_elapsed` 后直接再等。调用已被宿主转入后台的，结束时按宿主提醒或句柄收取。回执里的 `unreceived` 是还没取走的结果，见到就先用 `wait` 收取。 + 终态和阻塞不会主动推送，只能由 `wait` 收取：手头没有不依赖结果的工作时，派发后立刻单独调用 `wait`；有就先完成当前步骤再调用，并在结束本轮前收齐全部终态。常规省略 `wait_seconds`，`wait_elapsed` 后直接再等。调用已被宿主转入后台的，结束时按宿主提醒或句柄收取。回执里的 `unreceived` 是还没取走的结果，见到就先用 `wait` 收取；回执带 `next` 时照做，任务在跑时要先问用户，也先 `wait` 或 `cancel`。只派一个任务且本轮就要结果时，可传 `wait=true`，派发后先等一次，结果在回执的 `wait` 里；并联派发不要逐个传，全部派出后再统一等。 2. 常规用收件箱：省略 `ids`、带上派发时相同的 `manager_session_id`，每次重复同样的调用；无显式或宿主注入身份时会报错。定向等待用显式 `ids`，下次传返回的 `pending_ids`；保留已终态句柄会重读结果。任一就绪即返回，先处理阻塞和待审，再继续等到 `all_finished=true`；它只表示所选范围无待收终态，不代表全部成功。遇到 `review_needed` 先用 `decide_answers` 裁决，否则继续 `wait` 会立即返回。仅在用户询问进展或现有状态不足以判断是否介入时，对相关 `ids` 调用 `wait(wait_seconds=0)`；它仍会收取就绪答案或阻塞，不是只读查询。在 Codex 用 `functions.exec` 调 `wait` 时，首行写 `// @exec: {"yield_time_ms": 250000}`；返回 `Script running with cell ID ...` 后，只用 `functions.wait(cell_id, yield_time_ms=250000)` 等该脚本完成。让出间隔要长过 `wait` 的 240 秒，否则同一结果会被反复轮询，可能被宿主或模型代理当成死循环而中断本轮；外层让出控制权不代表超时。 3. `status=running` 尚未结束。`execution_blocked` 按 `error.code` 决定继续等或取消：阻塞不停止 CLI，同类阻塞每个任务只提醒一次；非交互模式下单个工具调用被拒绝不算阻塞，CLI 常会换工具继续，终态再按结果核对；普通工具失败、重试、沉默不证明任务失败。`seat_lost` 表示某席已失败或受阻而其他席仍在跑，每席提醒一次，只给状态和错误，不交付答案；按 `error` 决定补跑或继续等。 4. `wait(0)` 返回的运行中 `output` 只是最新公开正文尾部，不含思考或工具参数；材料不足就报告未知，不凭此判定卡死。普通会议中，单席成功不单独唤醒等待，未就绪快照也不交付该席完整答案。 diff --git a/src/agentnave/mcp_payloads.py b/src/agentnave/mcp_payloads.py index ef7ece31..d619df65 100644 --- a/src/agentnave/mcp_payloads.py +++ b/src/agentnave/mcp_payloads.py @@ -70,6 +70,8 @@ class TaskStartPayload(InvocationPayload): conversation_id: str access: NotRequired[str | None] # the declared tier; recorded only, never enforced unreceived: NotRequired[list[UnreceivedPayload]] + next: NotRequired[str] # what to call before ending the turn while ends wait (ADR-0064) + wait: NotRequired[WaitPayload] # the wait this dispatch ran when asked to class InvocationListItem(TypedDict): @@ -154,6 +156,7 @@ class WaitPayload(TypedDict): all_finished: bool updates: list[InvocationPayload] rounds: list[RoundPayload] + next: NotRequired[str] class CancelPayload(TypedDict): @@ -168,6 +171,8 @@ class ShowRunPayload(TypedDict): round: NotRequired[RoundPayload] unreceived: NotRequired[list[UnreceivedPayload]] warnings: NotRequired[list[str]] + next: NotRequired[str] + wait: NotRequired[WaitPayload] class ConversationSummary(TypedDict): diff --git a/src/agentnave/mcp_rounds.py b/src/agentnave/mcp_rounds.py index bccf2789..9a1a722a 100644 --- a/src/agentnave/mcp_rounds.py +++ b/src/agentnave/mcp_rounds.py @@ -24,6 +24,7 @@ ShowRunPayload, ) from agentnave.mcp_runtime import ( + DISPATCH_WAIT, GroupSeatSpec, ManagerModel, ManagerSessionId, @@ -33,9 +34,9 @@ _rejected, _review_map, _short_title, - _unreceived, round_payload, ) +from agentnave.mcp_tasks import waited_dispatch from agentnave.member_names import seat_identity from agentnave.members import ( CapacityError, @@ -55,7 +56,6 @@ def _add_started( payload: ShowRunPayload, conversations: Conversations, round_: Round, - manager_session_id: str | None, ) -> None: payload["round"] = round_payload( round_, @@ -75,9 +75,6 @@ def _add_started( warnings = model_warnings(conversations.host, message) if warnings: payload["warnings"] = warnings - unreceived = _unreceived(conversations, manager_session_id) - if unreceived: - payload["unreceived"] = unreceived def _retry_seats( @@ -85,7 +82,7 @@ def _retry_seats( group_id: str, seat_ids: list[str], manager_session_id: str | None, -) -> ShowRunPayload: +) -> tuple[ShowRunPayload, str]: try: round_ = conversations.retry(group_id, seat_ids, manager_session_id=manager_session_id) except KeyError as exc: @@ -95,8 +92,8 @@ def _retry_seats( raise ToolError(str(exc)) from exc raise _rejected(exc) from exc payload: ShowRunPayload = {"group_id": group_id} - _add_started(payload, conversations, round_, manager_session_id) - return payload + _add_started(payload, conversations, round_) + return payload, round_.id def _open_show( @@ -246,6 +243,7 @@ async def run_round( ] = None, manager_model: ManagerModel = None, manager_session_id: ManagerSessionId = None, + wait: Annotated[bool, Field(description=DISPATCH_WAIT)] = False, *, ctx: Context[Runtime], ) -> ShowRunPayload: @@ -261,7 +259,8 @@ async def run_round( extras = (prompt, seats, instructions, forward, title, members, cwd, review, isolated) if group_id is None or any(extras): raise ToolError("retry_seats takes only group_id. Correct the arguments and retry.") - return _retry_seats(conversations, group_id, retry_seats, manager_session_id) + payload, round_id = _retry_seats(conversations, group_id, retry_seats, manager_session_id) + return await waited_dispatch(payload, round_id, wait, manager_session_id, ctx) new = group_id is None if new: if not title or not title.strip() or not members: @@ -278,6 +277,8 @@ async def run_round( ) if not seats and forward: raise ToolError("forward needs seats to answer this round. Correct and retry.") + if not seats and wait: + raise ToolError("wait needs seats to run a round. Correct the arguments and retry.") if prompt is None and (seats or not new): raise ToolError("prompt is required unless only opening a new meeting. Correct and retry.") opened: list[GroupSeatPayload] | None = None @@ -337,8 +338,8 @@ async def run_round( if isinstance(exc, (SettingsFileError, CapacityError)): raise ToolError(str(exc)) from exc raise _rejected(exc) from exc - _add_started(payload, conversations, round_, manager_session_id) - return payload + _add_started(payload, conversations, round_) + return await waited_dispatch(payload, round_.id, wait, manager_session_id, ctx) class ReviewReceipt(TypedDict): diff --git a/src/agentnave/mcp_runtime.py b/src/agentnave/mcp_runtime.py index 836e09e7..dd7c25ac 100644 --- a/src/agentnave/mcp_runtime.py +++ b/src/agentnave/mcp_runtime.py @@ -117,6 +117,16 @@ def _read_excluded_providers() -> frozenset[str]: WAIT_SECONDS = 240 +# Ends are never pushed (ADR-0007), so receipts say what to call before the turn ends (ADR-0064). +WAIT_NEXT = ( + "Ends are not pushed: before ending your turn, call wait (omit ids, same " + "manager_session_id) until all_finished." +) +WAIT_AGAIN = ( + "Still pending: handle any prompt_pending or review_needed first, then call wait again " + "before ending your turn." +) + EXCLUDED_PROVIDERS = _read_excluded_providers() PROVIDER_SELECTION = ( "Providers permitted by this host configuration: " @@ -131,6 +141,10 @@ def _read_excluded_providers() -> frozenset[str]: + ". start_agent rejects excluded providers; meetings follow discussion member limits without host exclusions." ) ACCESS = "Declared read/write access; see MCP instructions." +DISPATCH_WAIT = ( + "true waits up to 240s like wait(ids=[this handle]) and returns it in wait; only for a single " + "dispatch whose result this turn needs, not parallel dispatches." +) PROVIDER_OPTIONS: dict[ProviderName, str] = { provider: get_adapter(provider).options_description for provider in get_args(ProviderName.__value__) @@ -657,10 +671,13 @@ def _review_map(conversations: Conversations, round_: Round) -> dict[str, Review "TITLE_WIDTH", "_TITLE_WIDE_RANGES", "HOST", + "WAIT_AGAIN", + "WAIT_NEXT", "WAIT_SECONDS", "EXCLUDED_PROVIDERS", "PROVIDER_SELECTION", "ACCESS", + "DISPATCH_WAIT", "_HIDDEN_TOOLS", "_CHAT_TOOLS", "mcp", diff --git a/src/agentnave/mcp_tasks.py b/src/agentnave/mcp_tasks.py index 2280f147..a5961714 100644 --- a/src/agentnave/mcp_tasks.py +++ b/src/agentnave/mcp_tasks.py @@ -26,11 +26,13 @@ InvocationListPayload, InvocationPayload, RoundPayload, + ShowRunPayload, TaskStartPayload, WaitPayload, ) from agentnave.mcp_runtime import ( ACCESS, + DISPATCH_WAIT, ManagerModel, ManagerSessionId, Runtime, @@ -149,6 +151,7 @@ async def start_agent( ] = None, manager_model: ManagerModel = None, manager_session_id: ManagerSessionId = None, + wait: Annotated[bool, Field(description=DISPATCH_WAIT)] = False, *, ctx: Context[Runtime], ) -> TaskStartPayload: @@ -231,9 +234,29 @@ async def start_agent( add_write_warning(warnings, manager, request, started.invocation_id, access) if warnings: payload["warnings"] = warnings - unreceived = _unreceived(_conversations(ctx), manager_session_id) - if unreceived: + return await waited_dispatch(payload, started.invocation_id, wait, manager_session_id, ctx) + + +async def waited_dispatch[P: (TaskStartPayload, ShowRunPayload)]( + payload: P, + handle: str, + wait_now: bool, + manager_session_id: str | None, + ctx: Context[Runtime], +) -> P: + """Point a dispatch at wait, or run that wait now when asked (ADR-0064).""" + conversations = _conversations(ctx) + if not wait_now: + if unreceived := _unreceived(conversations, manager_session_id): + payload["unreceived"] = unreceived + payload["next"] = runtime.WAIT_NEXT + return payload + waited = await receive(ids=[handle], manager_session_id=manager_session_id, ctx=ctx) + payload["wait"] = waited + if unreceived := _unreceived(conversations, manager_session_id): payload["unreceived"] = unreceived + if not waited["all_finished"] or any(conversations.inbox(manager_session_id)): + payload["next"] = runtime.WAIT_AGAIN return payload @@ -429,6 +452,9 @@ async def wait( ) +receive = wait + + @dataclass class _WaitResults: invocation_ids: list[str] @@ -501,13 +527,16 @@ def _wait_payload( else: delivered = conversations.delivered_tasks | conversations.delivered_rounds pending_ids = [handle for handle in ids if handle not in delivered] - return { + payload: WaitPayload = { "reason": "wait_elapsed" if elapsed else "ready", "pending_ids": pending_ids, "all_finished": not pending_ids, "updates": updates, "rounds": rounds, } + if pending_ids: + payload["next"] = runtime.WAIT_AGAIN + return payload async def cancel( diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 324c962b..6e9f0aa1 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -427,6 +427,10 @@ async def test_wait_returns_whichever_task_or_round_is_ready(tmp_path: Path) -> "run_round", {"group_id": show["group_id"], "title": "again", "prompt": "x"} ) assert rejected.is_error and "only open a new meeting" in str(rejected.content) + rejected = await client.call_tool( + "run_round", {"group_id": show["group_id"], "prompt": "remark", "wait": True} + ) + assert rejected.is_error and "wait needs seats" in str(rejected.content) sleeping = await _call( client, "start_agent", @@ -436,15 +440,21 @@ async def test_wait_returns_whichever_task_or_round_is_ready(tmp_path: Path) -> "cwd": str(tmp_path), }, ) - round_ = await _run( - client, {"group_id": show["group_id"], "seats": ["a"], "prompt": "quick"} + started = await _call( + client, "run_round", {"group_id": show["group_id"], "seats": ["a"], "prompt": "quick"} ) + round_ = _data(started["round"]) ready = await _call( client, "wait", {"ids": [sleeping["invocation_id"], round_["round_id"]]} ) assert ready["reason"] == "ready" and ready["updates"] == [] assert ready["pending_ids"] == [sleeping["invocation_id"]] assert ready["all_finished"] is False + # Ends are not pushed, so every reply that leaves work pending points at wait. + assert ( + sleeping["next"] == mcp_runtime.WAIT_NEXT and started["next"] == mcp_runtime.WAIT_NEXT + ) + assert ready["next"] == mcp_runtime.WAIT_AGAIN (finished,) = cast(list[dict[str, object]], ready["rounds"]) assert finished["round_id"] == round_["round_id"] and finished["status"] == "finished" # The inbox skips the round already delivered and keeps the running task. @@ -468,10 +478,38 @@ async def test_wait_returns_whichever_task_or_round_is_ready(tmp_path: Path) -> theirs = next(item for item in both if item["reason"] == "ready") updates = cast(list[dict[str, object]], theirs["updates"]) assert [update["invocation_id"] for update in updates] == [other["invocation_id"]] + waited = await _call( + client, + "start_agent", + { + "provider": "claude", + "prompt": "hi", + "cwd": str(tmp_path), + "manager_session_id": "c", + "wait": True, + }, + ) + received = cast(dict[str, object], waited["wait"]) + assert received["all_finished"] is True and "next" not in waited + updates = cast(list[dict[str, object]], received["updates"]) + assert [update["status"] for update in updates] == ["succeeded"] + assert (await _call(client, "wait", {"manager_session_id": "c"}))[ + "reason" + ] == "nothing_pending" + # Same chat still has sleeping pending, so wait=True still steers back to wait. + still_pending = await _call( + client, + "start_agent", + {"provider": "claude", "prompt": "hi", "cwd": str(tmp_path), "wait": True}, + ) + assert ( + cast(dict[str, object], still_pending["wait"])["all_finished"] is True + and still_pending["next"] == mcp_runtime.WAIT_AGAIN + ) cancelled = await _cancel(client, sleeping["invocation_id"]) assert cancelled["status"] == "cancelled" empty = await _call(client, "wait", {}) - assert empty["reason"] == "nothing_pending" + assert empty["reason"] == "nothing_pending" and "next" not in empty assert empty["pending_ids"] == [] and empty["all_finished"] is True