Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions apps/tray/src/rules.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 写清图像需求和产物路径,由它的对话模型自行调用生图服务。";
Expand Down
1 change: 1 addition & 0 deletions docs/adr/0007-wait-any-ready-subset.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` 引导,派发调用可选顺带等一次。

## 背景

Expand Down
15 changes: 15 additions & 0 deletions docs/adr/0064-dispatch-receipts-steer-to-wait.md
Original file line number Diff line number Diff line change
@@ -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 钩子能拦住结束回合,但各宿主的信任和入参不同,另行决定。
1 change: 1 addition & 0 deletions docs/adr/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
2 changes: 1 addition & 1 deletion skills/agentnave-manager/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` 只是最新公开正文尾部,不含思考或工具参数;材料不足就报告未知,不凭此判定卡死。普通会议中,单席成功不单独唤醒等待,未就绪快照也不交付该席完整答案。
Expand Down
5 changes: 5 additions & 0 deletions src/agentnave/mcp_payloads.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -154,6 +156,7 @@ class WaitPayload(TypedDict):
all_finished: bool
updates: list[InvocationPayload]
rounds: list[RoundPayload]
next: NotRequired[str]


class CancelPayload(TypedDict):
Expand All @@ -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):
Expand Down
23 changes: 12 additions & 11 deletions src/agentnave/mcp_rounds.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
ShowRunPayload,
)
from agentnave.mcp_runtime import (
DISPATCH_WAIT,
GroupSeatSpec,
ManagerModel,
ManagerSessionId,
Expand All @@ -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,
Expand All @@ -55,7 +56,6 @@ def _add_started(
payload: ShowRunPayload,
conversations: Conversations,
round_: Round,
manager_session_id: str | None,
) -> None:
payload["round"] = round_payload(
round_,
Expand All @@ -75,17 +75,14 @@ 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(
conversations: Conversations,
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:
Expand All @@ -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(
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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):
Expand Down
17 changes: 17 additions & 0 deletions src/agentnave/mcp_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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: "
Expand All @@ -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__)
Expand Down Expand Up @@ -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",
Expand Down
35 changes: 32 additions & 3 deletions src/agentnave/mcp_tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,13 @@
InvocationListPayload,
InvocationPayload,
RoundPayload,
ShowRunPayload,
TaskStartPayload,
WaitPayload,
)
from agentnave.mcp_runtime import (
ACCESS,
DISPATCH_WAIT,
ManagerModel,
ManagerSessionId,
Runtime,
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -429,6 +452,9 @@ async def wait(
)


receive = wait


@dataclass
class _WaitResults:
invocation_ids: list[str]
Expand Down Expand Up @@ -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(
Expand Down
44 changes: 41 additions & 3 deletions tests/test_conversations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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.
Expand All @@ -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


Expand Down
Loading