feat(acp): register daemon-managed sessions in the session registry and let them send peer messages (#11488)

* feat(acp): register daemon-managed sessions in the session registry and let them send peer messages

Only the interactive UI registered, so a session `qwen serve` drives was
invisible to `qwen sessions ps`, absent from every peer's `list_agents`,
and unable to send at all: the send path reads this process's own record
to learn its reply address, and with no record it reported the feature
disabled. A user with a terminal open and a daemon session working in
the same repository could not have the two speak to each other.

The registry was keyed by PID, one record per process, which was right
while a process meant a session. The `qwen --acp` child a daemon spawns
hosts up to thirty-two at once, each with its own id, name and working
directory, and one record cannot describe them. So a process hosting
several writes one record per session, `<pid>-<8 hex>.json`. The suffix
is minted at registration and never moves — deliberately not derived
from the session id, because an id is swapped underneath a live session
by a `/clear` or a session load, and a record that renamed itself would
strand every reader holding the old name. The PID in front of the suffix
is still what liveness, the sweep and the namespace guards read, so a
minted record is reaped exactly like a shared one, and a reader too old
to know the name does not see it rather than mistaking it for litter.

Which record a caller means is a slot, threaded through patch,
unregister and the own-record read. A Config learns its slot from the
registration itself, so a session's `/clear`, its `/cd` and its exit
each touch its own record and no sibling's.

The process binds one inbox for all its sessions and writes that address
into every record. Senders already pin `toSessionId` on every frame, so
that is what tells the sessions apart; a frame naming a session this
process does not hold — or naming none, which used to be deliverable
because there was only one session it could have meant — is answered
`misaddressed`.

That sharing broke an assumption elsewhere. The send path and
`list_agents` excluded "this session" by matching either its id or its
reply address, the address being a proxy for the id that held while one
process bound one socket. Siblings share an inbox, so the proxy stopped
holding and every sibling of the sending session would have been
filtered out as if it were the sender. Identity is the session id alone
now; a twin under a second process is still excluded, because that is
what the id catches.

Inbound is refused rather than held. A hold is a question put to a
person, and nobody is watching a held-message list on a daemon-managed
session's behalf, so holding would leave every sender waiting out a
five-minute expiry to learn nothing. Where a held message should surface
for these sessions is a real question and separate work; the protocol
page says so rather than leaving a reader to find out.

Registration happens only while cross-session messaging is on. The
interactive UI registers unconditionally because its record also answers
"what is running right now"; a daemon-managed session's record exists to
be addressed, and one with no inbox behind it would put a name in every
peer's listing that can be addressed and never answered.

The inbox is bound by the first session that needs one, and the
transport is imported at that moment, so an ACP process that never
messages a peer never loads it. Closing is registered as exit cleanup
and also runs from both teardown methods, because a bare signal reaches
neither. Whether to bind at all is read from the settings the process
started with: the agent's `settings` field is re-pointed at whichever
session is being handled, and this is a question about the process.

Part of #10925
Unblocks #10118 (M3)

* fix(acp): import the peer-messaging transport statically so the fast-path bundle closure stays clean

The agent imported the transport and `registerSession` with dynamic
`import()`s, on the theory that a process with messaging off should not
load them. The theory bought nothing and cost something: every module
behind both imports already sits in the agent's static closure, and a
dynamic import of the core barrel turns that barrel into a code-splitting
entry. esbuild then re-partitions the shared chunks, and one of the
chunks the agent imports statically came to hold iconv-lite's encoding
tables — a module the ACP fast path must not load, which is exactly what
the startup bundle closure check exists to catch, and did.

Both imports are static again, the catch-up loop that existed only to
cover the import's latency is gone, and the design note says why the
transport is loaded the way it is.

Part of #10925

* fix(acp): judge hosted sessions by live ids and their own settings

Review follow-ups on daemon-managed session registration:

- ownsSessionId tested the frozen publication key, so a frame pinned to
  the id the record actually advertises after /clear was answered
  misaddressed; test each hosted Config's live id as well.
- Registration and the inbox bind followed the process's startup
  settings; each session's own settings decide instead.
- A failed inbox bind latched as started, so no session ever retried
  it; treat null/failure as not-started, and latch closed on teardown
  so a retry cannot resurrect the socket.
- disposeSessions awaited the inbox drain before aborting generations;
  abort first so a running turn cannot keep executing tools after the
  client is gone.
- The send-side mirror re-minted its bucket per addressed session id
  while the receiver meters per address, letting alternating siblings
  reset the allowance and duplicate window; keep one bucket per inbox
  address and stop the repeat refusal naming "that session".
- The self-exclusion could be beaten by a re-id patch landing between
  the own-record read and the directory read; re-check a peer on this
  session's own inbox against a fresh read, on the send path and in
  list_agents alike.
- The record filename check accepted a zero-padded pid on platforms
  without a namespace id; require the canonical decimal form.
- Drop the dead sessionId parameter from the misaddressed-frame
  re-assert, document the two record filename forms and the one-sender
  consequence, and add the Chinese design doc.

* fix(ipc): probe each inbox once, and keep a minted record nameable

Two behaviour fixes from review, plus the tests that would have caught
them and the cases the slot plumbing was missing.

A process hosting several sessions writes one record per session, all
pointing at the one inbox. The liveness pass dialled that socket once
per record, so a daemon with 32 hosted sessions opened 32 simultaneous
connections at one address every time a peer listed the directory —
enough to trip the inbox backlog. Probe each distinct address once and
apply the verdict to every record that names it.

Re-registration deleted the captured path of a minted record before it
knew whether the delete would happen. On a transient read error it
returned early with the capture already gone, and a minted suffix cannot
be derived a second time: the record stayed on disk under a name this
process could no longer patch or remove, outliving the session as a
directory entry pointing at a dead inbox. Keep the capture on that one
branch.

Tests: every registration fixture returned the shared slot, which is
also the parameter default, so nothing could tell "passes the slot it
registered under" from "passes the default" — config and send_message
now register a minted slot and name it in the assertion. Also covers
the unknown-slot path against a live shared record, a minted filename
surviving an id swap, near-miss record filenames, probe-once, the
ownsSessionId branches of the inbound gate, and an ACP host running two
sessions through one inbox.

* test(core): name the minted slot in the list_agents identity read

---------

Co-authored-by: qwen-code-dev-bot <[email protected]>
This commit is contained in:
qqqysandqwen-code-dev-bot authored and GitHub committed 2026-09-10 08:30:36 +00:00
1 parent 2488d12d8b
commit 0ef35351e9
26 files changed
+2189 -237

No files matched your search

@@ -0,0 +1,163 @@
# Daemon-managed sessions in the session registry
[English](2026-09-09-daemon-sessions-in-directory.md) | [简体中文](2026-09-09-daemon-sessions-in-directory.zh-CN.md)
Status: implemented alongside this document. Lets a session `qwen serve`
drives be discovered and addressed like any other, and lets it send.
## Problem
Only the interactive UI registered. A session the daemon drives was
invisible to `qwen sessions ps`, absent from every peer's `list_agents`,
and unable to send: `sendToPeer` reads this process's own registry record
to learn its reply address, and with no record it reported the feature
disabled. A user with a terminal open and a daemon session working in the
same repository could not have the two speak to each other at all.
Three things stood in the way.
### One process, several sessions — one record
The registry is keyed by PID, one `<pid>.json` per process. That was
exactly right while a process meant a session. The `qwen --acp` child the
daemon spawns hosts up to 32 sessions at once, each with its own id, name
and working directory, and one record cannot describe them.
### The reply address had stopped meaning "the same session"
Both the send path and `list_agents` excluded "this session" from the
peers they consider by matching either the session id **or** the reply
address. The address was a proxy for the id, and a correct one while a
process held one session and bound one socket. Sessions in one process
share an inbox, so the proxy stopped holding: every sibling of the
sending session would have been filtered out as if it were the sender
itself, leaving sessions in one process unable to see each other.
### An unpinned frame had no single answer
A frame that names no `toSessionId` used to be deliverable: with one
session, it could only have meant that one. Reaching a process with
several, it names nothing at all.
## Design
**One record per session, in a minted slot.** A process hosting several
sessions writes `<pid>-<8 hex>.json` per session. The suffix is minted at
registration and never moves. It is deliberately not derived from the
session id: an id can be swapped underneath a live session (`/clear`, a
session load), and a record that renamed itself would strand every reader
holding the old name — the record is patched in place instead, which is
what the registry already did for every other mid-session change.
Everything else about a record is unchanged, and that is the point: the
PID in front of the suffix is what the liveness check, the sweep and the
namespace guards read, so a minted record is reaped exactly like a shared
one, and a reader too old to know the new name simply does not see it
(its filename pattern does not match) rather than mistaking it for
litter to delete.
Which record a caller means is a `slot`, threaded through `patch`,
`unregister` and the own-record read. `Config` learns its slot from the
registration itself and passes it to every write it queues, so a
session's `/clear`, its `/cd` and its exit each touch its own record and
no sibling's.
**One inbox, sessions told apart by the frame.** The process binds one
socket and writes that address into every record it owns. A sender picks
a session by the `toSessionId` it already pins on every frame; a frame
naming a session this process does not hold, or naming none, is answered
`misaddressed`. Binding a socket per session would multiply file
descriptors by the session count and buy nothing, since the frame
already carries the discriminator.
**Exclude self by session id, re-read on a shared address.** The
address is no longer evidence of identity, so it is out of the filter.
A twin — the same session id under a second process, which
`qwen --resume` creates — is still excluded, because that is what the id
catches. One window remains: the id filter reads this session's record
before the directory, and a re-id patch (`/clear`, or the re-assert a
misaddressed frame triggers) can land in between, re-admitting this
session's own record under its new id. The address does not move on a
re-id, so a peer on this session's own inbox is checked against a fresh
read of the record before it is believed to be a sibling — on the send
path and in `list_agents` alike.
**Inbound is refused, not held.** A hold is a question put to a person,
and nobody is watching a held-message list on a daemon-managed session's
behalf: holding would leave every sender waiting out a five-minute expiry
to learn nothing. `refused` says so immediately. Where a held message
_should_ surface for these sessions — the ACP client, the daemon's own
API — is a real question, and answering it is separate work.
**Register when the session's own settings turn messaging on.** The
interactive UI registers unconditionally, because its record also
answers "what is running right now". A hosted session's record exists
to be addressed; with messaging off, registering one would put a name
in every peer's listing that can be addressed and never answered. The
decision is read from the session's own `LoadedSettings`, not the
process's startup settings: one process can host sessions from more
than one workspace, and each appears in the directory — or stays
invisible — by what its own settings say.
**Bind on the first session, close on every exit.** The inbox is bound by
the first session that needs one — an ACP process with no session has
nothing to advertise. A bind that fails is not "started": the next
hosted session retries, and a late success fans the address out to every
record the process already wrote. Closing is registered as exit cleanup
and also runs from the two teardown methods, because a bare signal
reaches neither; once closed it stays closed, so a retry cannot
resurrect the socket after teardown. In `disposeSessions` the close is
awaited after the generation aborts — a slow drain must not keep a
running turn executing tools once the client is gone — and before the
record teardown, because the address clear it fans out is a patch the
records must still be there for.
The transport is imported statically, on purpose. Its own imports all
sit inside the agent's existing static closure, so lazy-loading it saves
nothing; and a dynamic import of the core barrel makes that barrel a
code-splitting entry, which re-partitions the shared chunks and can land
modules the ACP fast path must not load (iconv-lite's encoding tables) in
a chunk the agent then imports statically — the startup bundle closure
check exists to catch exactly that.
**One bind per process, a decision per session.** The inbox is bound
once for the process and shared by every session it hosts — but whether
a session registers at all is that session's own call, read from the
settings its workspace loaded. `this.settings` on the agent is
re-pointed at whichever session is being handled, so the decision is
threaded through `createAndStoreSession` explicitly rather than read
from a process-wide capture.
## Trade-offs
**A record outlives a process that is killed outright.** The next listing
sweeps it, which is the same guarantee every other session has had; a
graceful shutdown removes it immediately.
**`qwen -p` still does not register.** A one-shot run measured in seconds
would mostly contribute stale records.
**Outbound only.** A hosted session can send and receives its own
receipts; what it cannot do is take a message in. That is the next step,
and the protocol page says so rather than leaving a reader to discover it.
**`kind` is still a self-report.** `serve` comes from the environment
marker the daemon sets on its children, which a process could set for
itself. It labels a listing; nothing reads it to decide anything.
## Files
- `packages/core/src/services/session-registry.ts` — slots, the minted
filename, and the per-slot path capture.
- `packages/core/src/config/config.ts` — remembering a session's slot and
passing it to every registry write.
- `packages/core/src/ipc/peer-send.ts` — reading the sending session's
own record, excluding self by id with a fresh-read recheck on a shared
address, and pacing sends per inbox address.
- `packages/core/src/tools/list-agents.ts`, `send-message.ts` — passing
the slot through.
- `packages/core/src/ipc/inbound-gate.ts`,
`packages/cli/src/peerMessaging/peer-messaging.ts` — `ownsSessionId`
for a process hosting several sessions.
- `packages/cli/src/acp-integration/acpAgent.ts` — binding the inbox,
registering and removing each hosted session's record.
@@ -0,0 +1,132 @@
# 会话注册表中的 daemon 托管会话
[English](2026-09-09-daemon-sessions-in-directory.md) | [简体中文](2026-09-09-daemon-sessions-in-directory.zh-CN.md)
状态:与本文档一同实现。让 `qwen serve` 驱动的会话可以像其他会话一样被发现和寻址,并且可以发送消息。
## 问题
此前只有交互式 UI 会注册。daemon 驱动的会话对 `qwen sessions ps`
不可见,缺席于每个对端的 `list_agents`,并且无法发送消息:
`sendToPeer` 要读取本进程自己的注册表记录来获取回执地址,没有记录时
它会报告该功能未启用。一个用户开着终端、同时有 daemon 会话在同一个
仓库里工作时,两者完全无法对话。
有三个障碍。
### 一个进程、多个会话——却只有一条记录
注册表以 PID 为键,每个进程一条 `<pid>.json`。在"一个进程即一个
会话"的时代这完全正确。daemon 派生的 `qwen --acp` 子进程同时托管
多达 32 个会话,每个会话有自己的 id、名称和工作目录,一条记录无法
描述它们。
### 回执地址不再意味着"同一个会话"
发送路径和 `list_agents` 此前通过匹配会话 id **或**回执地址来把
"本会话"从候选对端中排除。地址曾是 id 的代理——在一个进程只持有
一个会话、只绑定一个 socket 的时代,这个代理是成立的。同一进程里的
多个会话共享一个收件箱,代理随之失效:发送会话的每个兄弟会话都会被
当作发送者自己过滤掉,同一进程内的会话彼此完全不可见。
### 未钉选的帧没有唯一答案
不携带 `toSessionId` 的帧曾经是可投递的:只有一个会话时,它只可能
指向那一个。而当帧到达一个持有多个会话的进程时,它什么也没指向。
## 设计
**每个会话一条记录,放在铸造的槽位里。**托管多个会话的进程为每个
会话写一条 `<pid>-<8 hex>.json`。后缀在注册时铸造,永不改变。它刻意
不从会话 id 派生:id 可能在存活会话底下被换掉(`/clear`、会话加载),
而自己改名的记录会让所有持有旧名字的读者 stranded——记录改为原地
修补(patch),这正是注册表对其他一切会话中途变更已有的做法。
记录的其他方面完全不变,这正是要点:后缀前面的 PID 是存活性检查、
清扫和命名空间防护读取的东西,所以铸造的记录与共享记录被同样地回收;
而太老、不认识新名字的读者只是看不到它(其文件名模式不匹配),而不会
把它误认为待删除的垃圾。
调用方指的是哪条记录由一个 `slot` 表达,贯穿 `patch`、`unregister`
和自身记录读取。`Config` 从注册本身得知自己的槽位,并把它传给每一次
排队的写入,于是会话的 `/clear`、`/cd` 和退出各自只触碰自己的记录,
不碰兄弟会话的。
**一个收件箱,靠帧区分会话。**进程绑定一个 socket,并把该地址写进
它拥有的每条记录。发送方通过它本就在每帧上钉选的 `toSessionId` 来
选择会话;指向本进程不持有的会话、或没有指向任何会话的帧,会被应答
`misaddressed`。按会话绑定 socket 会让文件描述符数量乘以会话数,而且
毫无收益,因为帧本身已经携带了区分信息。
**按会话 id 排除自己,共享地址时重新读取。**地址不再是身份的证据,
所以它退出了过滤条件。孪生会话——同一会话 id 运行在第二个进程下,
即 `qwen --resume` 所产生的那种——仍然被排除,因为这正是 id 能捕获
的。还剩一个窗口:id 过滤先于目录读取本会话的记录,而一次重命名修补
(`/clear`,或由 misaddressed 帧触发的重断言)可能落在两次读取之间,
让本会话的记录以新 id 重新混入候选。重命名时地址不会变,所以在相信
某个对端是兄弟会话之前,凡落在本会话自己收件箱上的对端都要对照记录
的新读取再核对一次——发送路径和 `list_agents` 都如此。
**入站一律拒绝,而不是挂起。**挂起是向人提出的问题,而没有人替
daemon 托管的会话盯着挂起消息列表:挂起会让每个发送方干等五分钟的
过期却一无所获。`refused` 会立即说明。挂起的消息*应该*在这些会话的
哪里浮现——ACP 客户端、daemon 自己的 API——是个真实的问题,回答它
是另一项工作。
**会话自己的设置打开消息功能时才注册。**交互式 UI 无条件注册,因为
它的记录还要回答"现在有什么在运行"。托管会话的记录存在是为了被寻址;
消息功能关闭时注册它,会在每个对端的列表里放上一个可以寻址却永远
不应答的名字。这个决定读的是会话自己的 `LoadedSettings`,而不是进程
的启动设置:一个进程可以托管来自多个工作区的会话,每个会话是否出现
在目录里——或保持不可见——由它自己的设置决定。
**第一个会话绑回收件箱,每条退出路径都关闭。**收件箱由第一个需要它
的会话绑定——没有会话的 ACP 进程没有什么可广告的。失败的绑定不算
"已启动":下一个托管会话会重试,而迟来的成功会把地址扇出到进程已经
写出的每条记录。关闭同时注册为退出清理,并在两个拆解方法里运行,因为
裸信号两者都到不了;一旦关闭就保持关闭,重试不会在拆解之后复活
socket。在 `disposeSessions` 中,关闭在生成 abort 之后等待——缓慢的
排空不应让仍在运行的轮次在客户端已走后继续执行工具——又在记录拆解
之前,因为它扇出的地址清除是一次修补,记录必须还在原处。
传输层是刻意静态导入的。它自己的导入都位于 agent 现有静态闭包之内,
惰性加载省不了什么;而对 core 桶的动态导入会让该桶成为代码分割入口,
重新划分共享 chunk,可能把 ACP 快速路径不得加载的模块(iconv-lite 的
编码表)落到 agent 随后又静态导入的 chunk 里——启动 bundle 闭包检查
的存在正是为了抓住这种情况。
**每个进程一次绑定,每个会话各自决定。**收件箱为整个进程绑定一次,
由它托管的所有会话共享——但一个会话是否注册,是该会话自己的决定,
读的是其工作区加载的设置。agent 上的 `this.settings` 会被重新指向
正在处理的会话,所以这个决定通过 `createAndStoreSession` 显式穿线,
而不是从进程级快照读取。
## 取舍
**被直接杀掉的进程会留下记录。**下一次列表会清扫它,这与所有其他
会话一直以来的保证相同;优雅关闭会立即移除记录。
**`qwen -p` 仍然不注册。**以秒计的一次性运行多半只会贡献过期记录。
**仅出站。**托管会话可以发送,也能收到自己的回执;它不能做的是接收
消息。那是下一步,协议页面写明了这一点,而不是留给读者自己去发现。
**`kind` 仍是自我声明。**`serve` 来自 daemon 在其子进程上设置的环境
标记,进程自己也可以设置它。它只是列表的标签;没有任何东西依据它做
决定。
## 涉及文件
- `packages/core/src/services/session-registry.ts`——槽位、铸造的文件名,
以及按槽位的路径捕获。
- `packages/core/src/config/config.ts`——记住会话的槽位,并把它传给
每一次注册表写入。
- `packages/core/src/ipc/peer-send.ts`——读取发送会话自己的记录,按 id
排除自己并在共享地址时用新读取复核,以及按收件箱地址限速发送。
- `packages/core/src/tools/list-agents.ts`、`send-message.ts`——把槽位
传递下去。
- `packages/core/src/ipc/inbound-gate.ts`、
`packages/cli/src/peerMessaging/peer-messaging.ts`——为托管多个会话
的进程提供 `ownsSessionId`。
- `packages/cli/src/acp-integration/acpAgent.ts`——绑定收件箱,注册和
移除每个托管会话的记录。
+26 -1
View File
@@ -776,7 +776,10 @@ A table with columns: NAME, KIND, PID, AGE, DIRECTORY.
KIND says what registered the session — `tui` for someone at a terminal,
`external` for a program that is not a Qwen Code session at all (a voice
front-end, a relay), and `headless` or `serve` for a session another
program drives. It is a self-report, like NAME and DIRECTORY: every field
program drives. Several `serve` or `headless` rows can share one PID: a
`qwen --acp` child hosts all its sessions in one process — `serve` when
the daemon spawned it, `headless` when a client is driving it directly —
and each of them registers separately. It is a self-report, like NAME and DIRECTORY: every field
here was written by the process it describes, and nothing about what a
session is allowed to do depends on it. See
[Cross-Session Protocol](./cross-session-protocol.md) for the record
@@ -1027,6 +1030,28 @@ Anyone who holds the token can send as that controller, so treat it like
any other credential: give it to one program, keep it out of shared
config, and revoke it when that program is done.
### Sessions a program drives over ACP
Any `qwen --acp` child registers each session it hosts — as `serve` when
the daemon spawned the process, as `headless` when an editor or another
client is driving `qwen --acp` directly — and the session appears in
`qwen sessions ps` and in another session's `list_agents` like any
other. It can send: its model can call `send_message` to reach a terminal
you have open. Several of them share one process and one inbox, so a
sender has to name the session it means — every Qwen Code session does
that automatically.
Messages sent _to_ one are refused rather than held. Holding is a
question put to a person, and nobody is watching a held-message list on a
driven session's behalf; a sender is told at once instead of
waiting out an expiry. Where a held message should surface for those
sessions is not settled yet.
A session registers only while its own settings have
`agents.crossSessionMessaging` on. With it off it stays invisible,
because the only reason to list a session nobody can message would be to
advertise an address that never answers.
### Programs that are not Qwen Code sessions
Everything above works between sessions, but nothing in it is specific to
+40 -11
View File
@@ -17,12 +17,16 @@ shape, a value that does not is dropped, never rejected with an error.
A running session publishes one record:
```
$QWEN_HOME/sessions/<pid>.json (directory 0700, file 0600)
$QWEN_HOME/sessions/<pid>.json (directory 0700, file 0600)
$QWEN_HOME/sessions/<pid>-<8 hex>.json (a process hosting several sessions)
```
`$QWEN_HOME` defaults to `~/.qwen`. The file name is the writer's PID
and nothing else; a record whose `pid` field disagrees with its file
name is ignored.
`$QWEN_HOME` defaults to `~/.qwen`. The file name is keyed by the
writer's PID — either the bare PID, or the PID, a dash, and eight
lowercase hex characters minted at registration (see "Several records
from one process" below). A record whose `pid` field disagrees with the
PID prefix of its file name — compared in canonical decimal form, so a
zero-padded name agrees with nothing — is ignored.
```json
{
@@ -44,7 +48,7 @@ name is ignored.
| Field | Meaning |
| --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `schemaVersion` | Always `1`. A reader skips a record with a higher version and never deletes it. |
| `pid` | The writer's process id. Must equal the file name. |
| `pid` | The writer's process id. Must equal the PID the file name is keyed by: the whole name for the bare form, the digits before the `-<8 hex>` suffix for the minted one. |
| `procStart` | `<boot id>:<process start ticks>` on Linux (`/proc/sys/kernel/random/boot_id` and field 22 of `/proc/<pid>/stat`); `null` elsewhere. Guards against PID reuse, and against records written on another machine that shares this home directory. |
| `pidNs` | Inode number of `/proc/self/ns/pid` on Linux; `null` elsewhere. A reader only lists and sweeps records from its own namespace. |
| `sessionId` | The session's id. `/clear` and `/resume` swap it under the same PID, so re-read the record before each send. |
@@ -80,8 +84,9 @@ including tokens: being able to discover a session and being able to
authenticate to it are one capability by design. Do not print
`ipcToken` anywhere a model or a log can see it.
**Liveness.** A record is live when all of these hold: the file name
matches `pid`; `pidNs` equals the reader's; the boot id inside
**Liveness.** A record is live when all of these hold: the file name is
`<pid>.json` or `<pid>-<8 hex>.json` and its PID prefix equals `pid`;
`pidNs` equals the reader's; the boot id inside
`procStart` equals the reader's (or `procStart` is `null`); and the PID
is alive with the same start ticks. A live record with an `ipcPath`
still has to be dialed before it is advertised as reachable — a socket
@@ -92,6 +97,21 @@ sessions may share a `name`; the address grammar a sender types is
`name`, `name [ref]`, `[ref]` or the bare `ref`, and an ambiguous
`name` is an error rather than a guess.
**Several records from one process.** Any `qwen --acp` child — spawned
by the daemon, or driven directly by an editor or another client —
writes one record per session, named `<pid>-<8 hex>.json`, from its
first session on. The suffix is minted at
registration and never changes; a session id swapped underneath is a
patch to the record, not a rename of it. Every one of them carries the
same `ipcPath`, because the process binds one inbox for all its sessions
and tells them apart by the `toSessionId` on each frame — so **always
send `toSessionId`**: a frame without one that reaches such a process is
answered `misaddressed`, since there is no single session it could have
meant. Liveness, sweeping and the namespace and boot guards read the
record exactly as they do for the bare name; only the PID/filename
agreement check differs, and only in comparing `pid` against the digits
before the suffix rather than the whole name.
## 2. The inbox socket
One UNIX domain socket per session, at the first of these that binds:
@@ -285,7 +305,16 @@ that look like the envelope are defanged inside `content`.
renamed itself, are both still to come.
- **Same-name reporting.** `qwen sessions ps` and `list_agents` do not
flag records that still collide.
- **Daemon-managed sessions.** Only the interactive UI registers today, so
a session `qwen serve` drives is not in the registry, cannot be
addressed, and cannot send. The `serve` and `headless` kinds are
reserved for it.
- **Inbound messages to ACP-driven sessions.** A session a program
drives over ACP — daemon-spawned or not — registers and can send, but
answers `refused` to anything sent to it: a hold is a question put to
a person, and nobody is watching a hold list on its behalf. Where a
held message should surface for those sessions — its client, or the
daemon's own API — is still open.
- **Sessions behind one inbox are one sender to every peer.** A process
hosting several sessions sends with one `from` address, so a
receiver's per-sender budget and duplicate window (§6) are shared by
all of that process's sessions at once: a busy sibling can spend
another's allowance, and a body just sent to one cannot be repeated
to its sibling inside the window. Per-session accounting would have
to trust a frame-asserted field, which §3's trust model rules out.
@@ -34,8 +34,25 @@ import { getDefaultReasoningConfig } from './model-configuration.js';
import { getConversationDirectoryName } from '../utils/conversation-directory-identity.js';
// Mock cleanup module before importing anything else
const { mockRunExitCleanup } = vi.hoisted(() => ({
const { mockRunExitCleanup, mockRegisterCleanup } = vi.hoisted(() => ({
mockRunExitCleanup: vi.fn().mockResolvedValue(undefined),
mockRegisterCleanup: vi.fn(),
}));
const mockRegisterSession = vi.hoisted(() => {
let next = 0;
return vi.fn().mockImplementation(() => {
next += 1;
return Promise.resolve({
registered: true,
slot: next.toString(16).padStart(8, '0'),
});
});
});
const mockPeerMessagingStart = vi.hoisted(() => vi.fn());
vi.mock('../peerMessaging/peer-messaging.js', () => ({
PeerMessaging: {
start: (...args: unknown[]) => mockPeerMessagingStart(...args),
},
}));
const { mockStartNonInteractiveOpenAILogHousekeeping } = vi.hoisted(() => ({
mockStartNonInteractiveOpenAILogHousekeeping: vi.fn(),
@@ -52,6 +69,7 @@ const { mockMcpPoolDrainAll, mockMcpPoolGetSnapshot } = vi.hoisted(() => ({
}));
vi.mock('../utils/cleanup.js', () => ({
runExitCleanup: mockRunExitCleanup,
registerCleanup: mockRegisterCleanup,
}));
vi.mock('../services/housekeeping/scheduler.js', () => ({
startNonInteractiveOpenAILogHousekeeping:
@@ -240,6 +258,7 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => ({
}
},
INVOCATION_CONTEXT_META_KEY: 'qwen-code/invocation',
registerSession: mockRegisterSession,
PRIVATE_ACP_CAPABILITY_ENV: 'QWEN_CODE_PRIVATE_ACP_CAPABILITY',
PRIVATE_PARENT_CAPABILITY_META_KEY: 'qwen-code/private-parent-capability',
parseInvocationContext: vi.fn(
@@ -4463,6 +4482,14 @@ describe('QwenAgent MCP SSE/HTTP support', () => {
},
getProjectRoot: vi.fn().mockReturnValue('/tmp'),
getTargetDir: vi.fn().mockReturnValue('/tmp'),
// The session-registry seams. A hosted session registers a record of
// its own when cross-session messaging is on; with it off none of
// these is called, which is the state most of this suite runs in.
getCliVersion: vi.fn().mockReturnValue('9.9.9'),
trackSessionRegistration: vi.fn(),
updateSessionRegistryIpcPath: vi.fn().mockResolvedValue(undefined),
unregisterSessionRegistry: vi.fn().mockResolvedValue(undefined),
reassertSessionRegistryRecord: vi.fn().mockResolvedValue(undefined),
// Per-session folder-trust flag (security.folderTrust.enabled of the
// admission workspace), bound once at Config construction. Default off;
// bootRelocatableSession overrides it to reflect the fed workspace.
@@ -4883,6 +4910,406 @@ describe('QwenAgent MCP SSE/HTTP support', () => {
return { agent, agentPromise };
}
describe('daemon-managed sessions in the session registry', () => {
/** Settings with cross-session messaging on, which is off by default. */
function messagingOn() {
return makeSessionSettings({
mcpServers: {},
agents: { crossSessionMessaging: true },
});
}
beforeEach(() => {
mockRegisterSession.mockClear();
mockPeerMessagingStart.mockReset();
mockPeerMessagingStart.mockResolvedValue({
close: vi.fn().mockResolvedValue(undefined),
});
});
it('gives each hosted session a record of its own, and the shared inbox address', async () => {
const innerConfig = await setupSessionMocks('hosted-a');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
// The inbox is bound by the first session that needs one, so the
// address arrives after that session already has a record — the
// ordering production sees, and the one the fan-out exists for.
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
const options = mockPeerMessagingStart.mock.calls[0]![0] as {
updateSessionRegistryIpcPath: (
path: string,
token: string,
) => Promise<void>;
getPolicySetting: () => string;
ownsSessionId: (id: string) => boolean;
};
await options.updateSessionRegistryIpcPath('/tmp/acp.sock', 'tok');
expect(mockRegisterSession).toHaveBeenCalledWith(
expect.objectContaining({
sessionId: 'hosted-a',
cwd: '/tmp',
// One record per session, not one per process: they have their
// own ids, names and working directories.
slot: 'own',
kind: 'headless',
}),
);
expect(innerConfig.trackSessionRegistration).toHaveBeenCalledTimes(1);
// Every hosted session advertises the one inbox this process bound.
expect(innerConfig.updateSessionRegistryIpcPath).toHaveBeenCalledWith(
'/tmp/acp.sock',
'tok',
);
// Inbound is turned away rather than parked: nobody is watching a
// hold list on a daemon-managed session's behalf.
expect(options.getPolicySetting()).toBe('refuse');
expect(options.ownsSessionId('hosted-a')).toBe(true);
expect(options.ownsSessionId('someone-else')).toBe(false);
mockConnectionState.resolve();
await agentPromise;
});
it('records the daemon as the registrant when it spawned this process', async () => {
const previous = process.env['QWEN_CODE_SERVE'];
process.env['QWEN_CODE_SERVE'] = '1';
try {
await setupSessionMocks('hosted-serve');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() =>
expect(mockPeerMessagingStart).toHaveBeenCalled(),
);
expect(mockRegisterSession).toHaveBeenCalledWith(
expect.objectContaining({ kind: 'serve' }),
);
mockConnectionState.resolve();
await agentPromise;
} finally {
if (previous === undefined) delete process.env['QWEN_CODE_SERVE'];
else process.env['QWEN_CODE_SERVE'] = previous;
}
});
it('registers nothing while cross-session messaging is off', async () => {
// A record with no inbox behind it would advertise an address that
// never answers, which is worse than not being listed at all.
await setupSessionMocks('hosted-off');
const { agent, agentPromise } = await bootInitializedAcpAgent(
makeSessionSettings(),
);
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
expect(mockPeerMessagingStart).not.toHaveBeenCalled();
expect(mockRegisterSession).not.toHaveBeenCalled();
mockConnectionState.resolve();
await agentPromise;
});
it("removes a session's record when the session goes", async () => {
const innerConfig = await setupSessionMocks('hosted-gone');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
expect(innerConfig.unregisterSessionRegistry).not.toHaveBeenCalled();
mockConnectionState.resolve();
await agentPromise;
// Left behind, the record would keep advertising a session that is
// gone until this process exits.
expect(innerConfig.unregisterSessionRegistry).toHaveBeenCalled();
});
it('answers for the id the record advertises after a /clear swap', async () => {
// The sessions map is keyed by the id as it was at publication,
// while the record a sender reads follows the Config's live id —
// /clear swaps it under a running session. Ownership must be
// judged against both, or every frame pinned to the advertised id
// is answered misaddressed from then on.
const innerConfig = await setupSessionMocks('hosted-a');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
const options = mockPeerMessagingStart.mock.calls[0]![0] as {
ownsSessionId: (id: string) => boolean;
};
vi.mocked(innerConfig.getSessionId).mockReturnValue('hosted-clear-2');
expect(options.ownsSessionId('hosted-clear-2')).toBe(true);
// The frozen publication key still answers: a skipped /clear patch
// leaves the record advertising it.
expect(options.ownsSessionId('hosted-a')).toBe(true);
expect(options.ownsSessionId('someone-else')).toBe(false);
mockConnectionState.resolve();
await agentPromise;
});
it("registers nothing when the session's own settings turn messaging off, whatever the process started with", async () => {
// Startup settings on, session settings off: the per-setting
// contract the docs state is about the session, and a session that
// turned messaging off must not appear in any peer's listing.
await setupSessionMocks('hosted-off');
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
expect(mockPeerMessagingStart).not.toHaveBeenCalled();
expect(mockRegisterSession).not.toHaveBeenCalled();
mockConnectionState.resolve();
await agentPromise;
});
it('registers a session whose own settings turn messaging on, whatever the process started with', async () => {
// The mirror case: startup settings off, session settings on. The
// bind fires for the first session that needs one; it is not
// settled once for the process at boot.
await setupSessionMocks('hosted-on');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } = await bootInitializedAcpAgent(
makeSessionSettings(),
);
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
expect(mockRegisterSession).toHaveBeenCalledWith(
expect.objectContaining({ sessionId: 'hosted-on', slot: 'own' }),
);
mockConnectionState.resolve();
await agentPromise;
});
it('retries the inbox bind for the next session after a failed start', async () => {
mockPeerMessagingStart.mockResolvedValue(null);
await setupSessionMocks('hosted-retry-a');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() =>
expect(mockPeerMessagingStart).toHaveBeenCalledTimes(1),
);
// The failed bind is not "started": the next hosted session tries
// again rather than the process staying dark until exit.
await setupSessionMocks('hosted-retry-b');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() =>
expect(mockPeerMessagingStart).toHaveBeenCalledTimes(2),
);
mockConnectionState.resolve();
await agentPromise;
});
it('does not retry the bind once the inbox has been closed', async () => {
mockPeerMessagingStart.mockResolvedValue(null);
await setupSessionMocks('hosted-closed-a');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() =>
expect(mockPeerMessagingStart).toHaveBeenCalledTimes(1),
);
await (
agent as unknown as { closePeerMessaging(): Promise<void> }
).closePeerMessaging();
await setupSessionMocks('hosted-closed-b');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
// A retry must not resurrect an inbox after teardown ran.
expect(mockPeerMessagingStart).toHaveBeenCalledTimes(1);
mockConnectionState.resolve();
await agentPromise;
});
it('gives every hosted session the one inbox, however late it binds', async () => {
// The daemon's steady state is many sessions in one process, and
// the bind resolves while the first is still booting. Sessions
// published before the address arrives depend on the fan-out;
// sessions published after it depend on the catch-up. One test
// walks both, because in production almost every session is the
// second kind.
const first = await setupSessionMocks('hosted-1');
const second = makeInnerConfig();
second.getSessionId = vi.fn().mockReturnValue('hosted-2');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
vi.mocked(loadCliConfig)
.mockResolvedValueOnce(first as unknown as Config)
.mockResolvedValue(second as unknown as Config);
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp/one', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
const options = mockPeerMessagingStart.mock.calls[0]![0] as {
updateSessionRegistryIpcPath: (
path: string | undefined,
token?: string,
) => Promise<void>;
ownsSessionId: (id: string) => boolean;
};
// Fan-out: the session already published gets the address.
await options.updateSessionRegistryIpcPath('/tmp/acp.sock', 'tok');
expect(first.updateSessionRegistryIpcPath).toHaveBeenCalledWith(
'/tmp/acp.sock',
'tok',
);
// Catch-up: a session published afterwards gets it on registration,
// and the bind is not attempted a second time.
await agent.newSession({ cwd: '/tmp/two', mcpServers: [] });
expect(second.updateSessionRegistryIpcPath).toHaveBeenCalledWith(
'/tmp/acp.sock',
'tok',
);
expect(mockPeerMessagingStart).toHaveBeenCalledTimes(1);
// Two records, one per session, both addressable through the
// process's single inbox.
expect(mockRegisterSession).toHaveBeenCalledTimes(2);
expect(
mockRegisterSession.mock.calls.map(([fields]) => fields.sessionId),
).toEqual(['hosted-1', 'hosted-2']);
expect(
mockRegisterSession.mock.calls.every(
([fields]) => fields.slot === 'own',
),
).toBe(true);
expect(options.ownsSessionId('hosted-1')).toBe(true);
expect(options.ownsSessionId('hosted-2')).toBe(true);
mockConnectionState.resolve();
await agentPromise;
});
it('closes the inbox and stops advertising it when the sessions go', async () => {
const close = vi.fn().mockResolvedValue(undefined);
mockPeerMessagingStart.mockResolvedValue({ close });
const innerConfig = await setupSessionMocks('hosted-close');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
const options = mockPeerMessagingStart.mock.calls[0]![0] as {
updateSessionRegistryIpcPath: (
path: string | undefined,
token?: string,
) => Promise<void>;
};
await options.updateSessionRegistryIpcPath('/tmp/acp.sock', 'tok');
// The clearing half, driven through the callback the agent handed
// the transport — which is what `PeerMessaging.close()` calls when
// the socket goes. A record still naming a socket that is gone
// sends every peer at a dead address until the PID is swept.
await options.updateSessionRegistryIpcPath(undefined, undefined);
expect(innerConfig.updateSessionRegistryIpcPath).toHaveBeenLastCalledWith(
undefined,
undefined,
);
mockConnectionState.resolve();
await agentPromise;
// And the socket itself is closed on the way out, rather than left
// listening for the life of the process.
expect(close).toHaveBeenCalled();
});
it('arms the inbox close as exit cleanup, for the paths teardown never reaches', async () => {
// A bare SIGTERM to a `qwen --acp` child reaches neither
// disposeSessions nor finishManagedShutdown; runExitCleanup is
// what runs there, so the close has to be registered with it.
const close = vi.fn().mockResolvedValue(undefined);
mockPeerMessagingStart.mockResolvedValue({ close });
await setupSessionMocks('hosted-signal');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
expect(mockRegisterCleanup).toHaveBeenCalled();
const armed = mockRegisterCleanup.mock.calls.map(
([entry]) => entry as () => Promise<void> | void,
);
await Promise.all(armed.map((entry) => entry()));
expect(close).toHaveBeenCalled();
mockConnectionState.resolve();
await agentPromise;
});
it('aborts in-flight generations before waiting on the inbox drain', async () => {
// The close can hang on a slow drain; a running turn must not keep
// executing tools after the client is gone while it waits.
let releaseClose: () => void = () => {};
const close = vi.fn().mockImplementation(
() =>
new Promise<void>((resolve) => {
releaseClose = resolve;
}),
);
mockPeerMessagingStart.mockResolvedValue({ close });
const innerConfig = await setupSessionMocks('hosted-drain');
vi.mocked(loadSettings).mockReturnValue(messagingOn());
const { agent, agentPromise } =
await bootInitializedAcpAgent(messagingOn());
await agent.newSession({ cwd: '/tmp', mcpServers: [] });
await vi.waitFor(() => expect(mockPeerMessagingStart).toHaveBeenCalled());
const internals = agent as unknown as {
generationControllers: Map<
string,
{ sessionId: string; controller: AbortController }
>;
disposeSessions: () => Promise<void>;
};
const controller = new AbortController();
internals.generationControllers.set('req-1', {
sessionId: 'hosted-drain',
controller,
});
const disposed = internals.disposeSessions();
await vi.waitFor(() => expect(controller.signal.aborted).toBe(true));
expect(close).toHaveBeenCalled();
// The record teardown still waits for the drain: the address clear
// is a patch the records must still be there for.
expect(innerConfig.unregisterSessionRegistry).not.toHaveBeenCalled();
releaseClose();
await disposed;
expect(innerConfig.unregisterSessionRegistry).toHaveBeenCalled();
mockConnectionState.resolve();
await agentPromise;
});
});
async function withEmptyTrustedFolders<T>(
run: (directory: string) => Promise<T>,
): Promise<T> {
@@ -23922,6 +24349,16 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => {
resumedConversation?: { messages: unknown[] };
} = {},
) {
// Same registry seams `makeInnerConfig` carries: teardown removes a
// hosted session's record before its Config shuts down, and a Config
// that cannot answer would turn a clean shutdown into a failure.
const registrySeams = {
getCliVersion: vi.fn().mockReturnValue('9.9.9'),
trackSessionRegistration: vi.fn(),
updateSessionRegistryIpcPath: vi.fn().mockResolvedValue(undefined),
unregisterSessionRegistry: vi.fn().mockResolvedValue(undefined),
reassertSessionRegistryRecord: vi.fn().mockResolvedValue(undefined),
};
const recording = {
rebuildTurnBoundaries: vi.fn(),
flush: vi.fn().mockResolvedValue(undefined),
@@ -23934,6 +24371,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => {
),
};
return {
...registrySeams,
initialize: vi.fn().mockResolvedValue(undefined),
shutdown: vi.fn().mockResolvedValue(undefined),
closeSessionWriter: vi.fn().mockResolvedValue(undefined),
+204 -1
View File
@@ -148,6 +148,7 @@ import {
listWorkflowSnapshots,
type TurnResultRecordPayload,
sessionIdContext,
registerSession,
SessionSourceService,
SessionSourceError,
} from '@qwen-code/qwen-code-core';
@@ -329,7 +330,9 @@ import {
} from '../i18n/languageUtils.js';
import { runWithAcpRuntimeOutputDir } from './runtimeOutputDirContext.js';
import { ACP_ERROR_CODES } from './errorCodes.js';
import { runExitCleanup } from '../utils/cleanup.js';
import { registerCleanup, runExitCleanup } from '../utils/cleanup.js';
import { QWEN_CODE_SERVE_ENV } from '../config/acp-channel-fallback.js';
import { PeerMessaging } from '../peerMessaging/peer-messaging.js';
import { startNonInteractiveOpenAILogHousekeeping } from '../services/housekeeping/scheduler.js';
import { appEvents, AppEvent } from '../utils/events.js';
import {
@@ -2891,6 +2894,12 @@ export async function runAcpAgent(
return agentInstance;
}, stream);
markAcpStartup('transportSetupEnd');
// Read at call time rather than captured: the connection sets
// `agentInstance` when it builds the agent, and this runs first. The
// inbox is bound by the first hosted session, but it has to be closed
// on every path out — including a bare signal, which reaches neither
// `disposeSessions` nor `finishManagedShutdown`.
registerCleanup(() => agentInstance?.closePeerMessaging());
} catch (err) {
eventLoopMonitor.dispose();
throw err;
@@ -3479,6 +3488,30 @@ async function assertManagedConversationDirectoryIdentity(
class QwenAgent implements Agent {
private sessions: Map<string, Session> = new Map();
/**
* Cross-session messaging for every session this process hosts.
*
* One inbox, not one per session: the address is a socket, the sessions
* are told apart by the `toSessionId` every sender pins on its frame,
* and binding a socket per session would multiply file descriptors by
* the session count for no added reach.
*
* Outbound only for now. Inbound is refused rather than held, because a
* hold is a question put to a person and nobody is watching a hold list
* on a daemon-managed session's behalf; a sender is told so at once
* instead of waiting out an expiry.
*/
private peerMessagingStart: Promise<PeerMessaging | null> | null = null;
// Set by closePeerMessaging: a retry must not resurrect an inbox after
// teardown ran.
private peerMessagingClosed = false;
/**
* Sessions already given a record. A session is published once, but a
* reload can hand the same id back through the same path, and two
* records for one session would be two names for it in every listing.
*/
private readonly registeredSessions = new Set<string>();
private inboxAddress: { ipcPath: string; ipcToken: string } | null = null;
private modelProviderReloadRevision = 0;
private readonly historyMutationTails = new Map<string, Promise<void>>();
private readonly startingSessionIds = new Set<string>();
@@ -3743,6 +3776,10 @@ class QwenAgent implements Agent {
async finishManagedShutdown(configs: Config[]): Promise<void> {
const failures: unknown[] = [];
// Ahead of the session teardown below: the expiry receipts an inbox
// owes its senders travel over the socket, and its own record clear
// is a patch the records must still be there for.
await this.closePeerMessaging();
for (const [sessionId, session] of [...this.sessions]) {
await this.removeStoredSessionEntry(sessionId, session, failures, {
shutdownConfig: false,
@@ -4155,6 +4192,15 @@ class QwenAgent implements Agent {
} catch (error) {
cleanupErrors.push(error);
}
// Before the Config shuts down, which does not touch the registry:
// a record left behind advertises a session that is gone, and peers
// would keep addressing it until this process exits.
this.registeredSessions.delete(sessionId);
try {
await session.getConfig().unregisterSessionRegistry();
} catch (error) {
cleanupErrors.push(error);
}
if (options.shutdownConfig !== false) {
try {
await session.getConfig().shutdown({ shutdownTelemetry: false });
@@ -4512,6 +4558,156 @@ class QwenAgent implements Agent {
await this.closeStoredSession(sessionId, opts);
}
/**
* Bind the one inbox every session this process hosts shares.
*
* Not awaited: binding a socket must not delay the first prompt, and a
* session published before it resolves still registers — the address
* is patched into every record when it arrives (`publishInboxAddress`).
*/
private startPeerMessaging(): void {
if (this.peerMessagingStart) return;
// A closed inbox stays closed: teardown runs on every exit path, and
// a session registered after it must not resurrect the socket.
if (this.peerMessagingClosed) return;
// Imported statically on purpose. The transport's own imports are all
// inside this file's existing static closure, so lazy-loading it buys
// nothing — and a dynamic import of the core barrel turns it into a
// code-splitting entry, which re-partitions the shared chunks and can
// land modules the ACP fast path must not load (iconv-lite's tables)
// in a chunk this file then imports statically.
this.peerMessagingStart = (async () => {
try {
const messaging = await PeerMessaging.start({
// Inbound is refused outright, so neither the approval mode nor
// the parity rule it feeds is ever consulted. Stated rather than
// left to a default: what a daemon-managed session may be told
// is settled here and nowhere else.
getApprovalMode: () => null,
getPolicySetting: () => 'refuse',
updateSessionRegistryIpcPath: (ipcPath, ipcToken) =>
this.publishInboxAddress(ipcPath, ipcToken),
ownsSessionId: (id) => {
// The map key froze when the session was published, while
// the record a sender reads follows the Config's live id —
// /clear swaps the id under a running session. Test both, so
// a frame pinned to either spelling is answered by the
// session that holds it.
const wanted = normalizeSessionIdForLookup(id);
return (
this.sessions.has(wanted) ||
[...this.sessions.values()].some(
(session) =>
normalizeSessionIdForLookup(
session.getConfig().getSessionId(),
) === wanted,
)
);
},
});
// A bind that could not start is not "started": the next hosted
// session retries rather than the process staying dark until exit.
if (messaging === null) this.peerMessagingStart = null;
return messaging;
} catch (error) {
debugLogger.error(
'[ACP] cross-session messaging failed to start:',
error,
);
this.peerMessagingStart = null;
return null;
}
})();
}
/** Close the inbox and stop advertising it. Safe to call more than once. */
async closePeerMessaging(): Promise<void> {
this.peerMessagingClosed = true;
const pending = this.peerMessagingStart;
if (!pending) return;
this.peerMessagingStart = null;
try {
await (await pending)?.close();
} catch (error) {
debugLogger.debug(
`[ACP] closing cross-session messaging failed: ${
error instanceof Error ? error.message : String(error)
}`,
);
}
this.inboxAddress = null;
}
/**
* Give a newly published session a registry record of its own.
*
* Only when that session's own settings turn messaging on. A record
* with no inbox behind it would put a name in every peer's listing
* that can be addressed and never answered, which is worse than not
* appearing at all — the interactive UI registers unconditionally
* because its record also answers "what is running right now", a
* question nobody asks of a session a daemon is driving.
*/
private registerHostedSession(
sessionId: string,
config: Config,
settings: LoadedSettings,
): void {
// Each session's own settings decide: one process can host sessions
// from more than one workspace, and a record exists to be addressed,
// so it is written only when that session's settings turn messaging
// on. The process's startup settings answer for nobody else.
if (settings.merged.agents?.crossSessionMessaging !== true) return;
// Bound by the first session that needs it rather than at startup: an
// ACP process with no session has nothing to advertise and nobody to
// receive for, and this is also the first moment the agent exists.
this.startPeerMessaging();
if (this.registeredSessions.has(sessionId)) return;
this.registeredSessions.add(sessionId);
config.trackSessionRegistration(
registerSession({
sessionId,
cwd: config.getTargetDir(),
qwenVersion: config.getCliVersion() ?? null,
// What spawned this process, as far as it can tell: the daemon
// marks the children it starts, and anything else running
// `qwen --acp` is a client driving it directly.
kind: process.env[QWEN_CODE_SERVE_ENV] === '1' ? 'serve' : 'headless',
// One record per session rather than one per process: they have
// separate ids, names and working directories.
slot: 'own',
}),
);
if (this.inboxAddress) {
void config.updateSessionRegistryIpcPath(
this.inboxAddress.ipcPath,
this.inboxAddress.ipcToken,
);
}
}
/**
* Write the inbox address into every hosted session's record.
*
* They share one address, so this runs once per bind rather than once
* per session, and again with `undefined` at close so no record
* advertises a socket that is gone.
*/
private async publishInboxAddress(
ipcPath: string | undefined,
ipcToken?: string,
): Promise<void> {
this.inboxAddress =
ipcPath !== undefined && ipcToken !== undefined
? { ipcPath, ipcToken }
: null;
await Promise.allSettled(
[...this.sessions.values()].map((session) =>
session.getConfig().updateSessionRegistryIpcPath(ipcPath, ipcToken),
),
);
}
async disposeSessions(): Promise<void> {
this.activeWorkReporter?.dispose();
this.activeWorkReporter = undefined;
@@ -4523,6 +4719,12 @@ class QwenAgent implements Agent {
controller.abort();
}
this.workspaceGenerationControllers.clear();
// After the aborts, so a slow inbox drain cannot keep a running turn
// executing tools after the client is gone; ahead of the record
// teardown below, because the close's address clear is a patch the
// records must still be there for (the ordering
// finishManagedShutdown states).
await this.closePeerMessaging();
await Promise.allSettled(
[...this.sessions.entries()].map(([sessionId, session]) =>
this.discardStoredSessionIfCurrent(sessionId, session, {
@@ -14704,6 +14906,7 @@ class QwenAgent implements Agent {
);
}
this.sessions.set(sessionId, session);
this.registerHostedSession(sessionId, config, settings);
// The session boots converged on the mode its settings derived; later
// reloads track convergence from here. Restricted sessions derive
// DEFAULT, mirroring the fold the reload loop applies to them.
@@ -28,11 +28,13 @@ import {
// Hoisted mocks
// ---------------------------------------------------------------------------
const { mockRunExitCleanup } = vi.hoisted(() => ({
const { mockRunExitCleanup, mockRegisterCleanup } = vi.hoisted(() => ({
mockRunExitCleanup: vi.fn().mockResolvedValue(undefined),
mockRegisterCleanup: vi.fn(),
}));
vi.mock('../utils/cleanup.js', () => ({
runExitCleanup: mockRunExitCleanup,
registerCleanup: mockRegisterCleanup,
}));
const { mockConnectionState } = vi.hoisted(() => {
@@ -105,7 +107,14 @@ const { mockRestoreWorktreeContext, mockWithDaemonSpan } = vi.hoisted(() => {
};
});
// The agent imports the peer-messaging transport statically; its own core
// imports would reach past this suite's exhaustive core mock, and nothing
// here turns messaging on.
vi.mock('../peerMessaging/peer-messaging.js', () => ({
PeerMessaging: { start: vi.fn() },
}));
vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => ({
registerSession: vi.fn(),
createDebugLogger: () => ({
debug: vi.fn(),
error: vi.fn(),
@@ -210,6 +210,7 @@ async function start(
mode: ApprovalMode | null = ApprovalMode.DEFAULT,
extra: {
getSessionId?: () => string;
ownsSessionId?: (id: string) => boolean;
settleSentMessage?: (
msgId: string,
status: string,
@@ -502,6 +503,74 @@ describe.skipIf(isWindows)('PeerMessaging', () => {
});
});
it('delivers to a host of several sessions only when the frame names one it holds', async () => {
const sender = await startSenderInbox();
const hosted = new Set(['session-a', 'session-b']);
const { messaging: m, submitted } = await start(ApprovalMode.DEFAULT, {
ownsSessionId: (id) => hosted.has(id),
});
const mine = peerFrame({
content: 'for session-b',
from: sender.socketPath,
fromMode: 'prompting',
toSessionId: 'session-b',
});
await send(m.socketPath!, mine);
await settle();
expect(submitted).toHaveLength(1);
const stranger = peerFrame({
content: 'for nobody here',
from: sender.socketPath,
fromMode: 'prompting',
toSessionId: 'session-gone',
});
await send(m.socketPath!, stranger);
await settle();
expect(submitted).toHaveLength(1);
expect(receipts.at(-1)).toMatchObject({
status: 'misaddressed',
origMsgId: stranger.msgId,
});
});
it('refuses an unpinned frame at a host of several sessions', async () => {
// The one place the two rules differ. A process holding one session
// takes an unpinned frame — it could only have meant that one. A
// process hosting several has nothing to guess from, and the
// protocol page tells senders to always pin for exactly this reason.
const sender = await startSenderInbox();
const { messaging: m, submitted } = await start(ApprovalMode.DEFAULT, {
ownsSessionId: () => true,
});
const unpinned = peerFrame({
content: 'to whom it may concern',
from: sender.socketPath,
fromMode: 'prompting',
});
await send(m.socketPath!, unpinned);
await settle();
expect(submitted).toHaveLength(0);
expect(receipts.at(-1)).toMatchObject({
status: 'misaddressed',
origMsgId: unpinned.msgId,
});
});
it('refuses to be wired as both one session and a host of several', async () => {
// A configuration mistake, not a runtime condition: a process that
// answered both would judge pins by whichever check ran first.
await expect(
start(ApprovalMode.DEFAULT, {
getSessionId: () => 'session-now',
ownsSessionId: () => true,
}),
).rejects.toThrow('getSessionId or ownsSessionId');
});
it('admits a pinned message when it has no session id to judge it against', async () => {
const { messaging: m, submitted } = await start(ApprovalMode.DEFAULT);
await send(
@@ -189,6 +189,16 @@ export interface PeerMessagingOptions {
* session holds now. Absent means frames are never checked against it.
*/
getSessionId?: () => string;
/**
* For a process hosting several sessions: whether `id` is one of them.
*
* Wired instead of `getSessionId` — the two are mutually exclusive,
* because a process either has one session to name or a set to test
* against. With this set, a frame naming no session at all is
* misaddressed: an unpinned frame could have meant the one session a
* single-session process holds, and here it could mean any of several.
*/
ownsSessionId?: (id: string) => boolean;
socketPath?: string;
/**
* Overrides the generated inbox token. A test seam like `socketPath`:
@@ -241,6 +251,7 @@ export class PeerMessaging {
ipcToken?: string,
) => Promise<void> = async () => {};
private getSessionId: (() => string) | null = null;
private ownsSessionId: ((id: string) => boolean) | null = null;
private settleSentMessage: (
msgId: string,
status: PeerDeliveryStatus,
@@ -292,6 +303,15 @@ export class PeerMessaging {
static async start(
options: PeerMessagingOptions,
): Promise<PeerMessaging | null> {
if (options.getSessionId && options.ownsSessionId) {
// A configuration mistake rather than a runtime condition: one asks
// which session this process is, the other which sessions it hosts,
// and a process that answered both would judge pins against
// whichever happened to be checked first.
throw new Error(
'PeerMessaging: pass getSessionId or ownsSessionId, not both',
);
}
const messaging = new PeerMessaging();
const controllerRegistryPath =
options.controllerRegistryPath ?? getPeerControllerRegistryPath();
@@ -335,6 +355,9 @@ export class PeerMessaging {
isControllerValid: (id) => messaging.validControllerIds?.has(id) ?? true,
...(options.admission ? { admission: options.admission } : {}),
getSessionId: options.getSessionId,
...(options.ownsSessionId
? { ownsSessionId: options.ownsSessionId }
: {}),
deliver: (frame, origin) => messaging.deliver(frame, origin),
reportDropped: (frame, reason, origin) =>
dropReceipts.note(frame, origin ?? { selfSent: false }, reason),
@@ -364,6 +387,7 @@ export class PeerMessaging {
// session id and the send ledger this process holds, not against the
// nulls a later assignment would leave in place.
messaging.getSessionId = options.getSessionId ?? null;
messaging.ownsSessionId = options.ownsSessionId ?? null;
messaging.settleSentMessage =
options.settleSentMessage ?? settleSentPeerMessage;
messaging.reassertSessionRecord = options.reassertSessionRecord ?? null;
@@ -857,13 +881,19 @@ export class PeerMessaging {
// may be the stale side (a skipped /clear patch), so it is re-asserted
// too; otherwise every later send here would be refused the same way.
const ownSessionId = this.getSessionId?.();
if (
frame.toSessionId !== undefined &&
ownSessionId !== undefined &&
frame.toSessionId !== ownSessionId
) {
const ownsSessionId = this.ownsSessionId;
const misaddressed = ownsSessionId
? // Hosting several sessions: a frame has to say which, and name one
// this process still holds.
frame.toSessionId === undefined || !ownsSessionId(frame.toSessionId)
: frame.toSessionId !== undefined &&
ownSessionId !== undefined &&
frame.toSessionId !== ownSessionId;
if (misaddressed) {
debugLogger.debug(
`refusing peer message ${frame.msgId}: addressed to session ${frame.toSessionId}, this is ${ownSessionId}`,
`refusing peer message ${frame.msgId}: addressed to session ${
frame.toSessionId ?? '(unspecified)'
}, this process does not hold it`,
);
if (frame.from) {
void sendDeliveryStatus(
@@ -20,6 +20,12 @@ import { SettingScope, type LoadedSettings } from '../config/settings.js';
import type { InitializationResult } from '../core/initializer.js';
const registerSession = vi.hoisted(() => vi.fn());
/**
* What `registerSession` reports for the interactive UI: the record
* landed, in the one shared slot a single-session process owns.
*/
const REGISTERED = { registered: true, slot: 'shared' };
const registerCleanup = vi.hoisted(() => vi.fn());
const inkRender = vi.hoisted(() => vi.fn());
const lastPeerInboxFailure = vi.hoisted(() => ({ value: null as unknown }));
@@ -145,7 +151,7 @@ type TestConfig = Config & {
};
function makeConfig(): TestConfig {
const trackSessionRegistration = vi.fn((registration: Promise<boolean>) => {
const trackSessionRegistration = vi.fn((registration: Promise<unknown>) => {
void registration.catch(() => undefined);
});
return {
@@ -186,7 +192,7 @@ describe('startInteractiveUI session registration', () => {
});
it('registers the session with its id, target dir, CLI version, and kind', async () => {
registerSession.mockResolvedValue(true);
registerSession.mockResolvedValue(REGISTERED);
const config = makeConfig();
await start(config);
@@ -203,11 +209,11 @@ describe('startInteractiveUI session registration', () => {
expect(config.trackSessionRegistration).toHaveBeenCalledTimes(1);
await expect(
config.trackSessionRegistration.mock.calls[0]?.[0],
).resolves.toBe(true);
).resolves.toEqual(REGISTERED);
});
it('arms teardown before serialized registry cleanup', async () => {
registerSession.mockResolvedValue(true);
registerSession.mockResolvedValue(REGISTERED);
const config = makeConfig();
await start(config);
@@ -219,7 +225,7 @@ describe('startInteractiveUI session registration', () => {
});
it('does not await a stalled registration before returning startup', async () => {
registerSession.mockReturnValue(new Promise<boolean>(() => undefined));
registerSession.mockReturnValue(new Promise(() => undefined));
const config = makeConfig();
const result = await Promise.race([
@@ -254,7 +260,7 @@ describe('startInteractiveUI cross-session messaging', () => {
vi.clearAllMocks();
lastPeerInboxFailure.value = null;
observedPeerInboxFailure.value = null;
registerSession.mockResolvedValue(true);
registerSession.mockResolvedValue(REGISTERED);
peerMessagingStart.mockResolvedValue({
close: vi.fn().mockResolvedValue(undefined),
});
+121 -42
View File
@@ -109,6 +109,20 @@ import { syncTeamMemory } from '../memory/team-memory-sync.js';
import { getTeamMemoryShareabilityWarning } from '../memory/team-memory-git-status.js';
import * as runtimeStatus from '../utils/runtimeStatus.js';
import * as sessionRegistry from '../services/session-registry.js';
/**
* A settled registration for the shared record, the shape
* `registerSession` reports. Every test here models a process holding one
* session; the slot only differs for a process hosting several.
*/
function sharedRegistration(
registered = true,
): Promise<sessionRegistry.SessionRegistration> {
return Promise.resolve({
registered,
slot: sessionRegistry.SHARED_RECORD_SLOT,
});
}
import {
ExtensionManager,
type Extension,
@@ -8454,7 +8468,7 @@ describe('Server Config (config.ts)', () => {
it('relocateWorkingDirectory should refresh runtime status after moving session artifacts', async () => {
const config = new Config(baseParams);
config.markRuntimeStatusEnabled();
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
const sessionId = config.getSessionId();
const newDir = path.resolve('/path/to/other');
const oldStorage = new Storage(config.getTargetDir());
@@ -8503,10 +8517,13 @@ describe('Server Config (config.ts)', () => {
// sessions apart; the switch must reach it (and the directory-derived
// name) or `qwen sessions ps` keeps showing the folder that was left.
await vi.waitFor(() => {
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
});
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
},
sessionRegistry.SHARED_RECORD_SLOT,
);
expect(settled).toContain('patch');
});
expect(settled[0]).toBe('relocated');
@@ -8524,7 +8541,7 @@ describe('Server Config (config.ts)', () => {
// state is reachable and the /cd patch must survive it.
const config = new Config(baseParams);
// No markRuntimeStatusEnabled(): models the failed sidecar write.
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
const sessionId = config.getSessionId();
const newDir = path.resolve('/path/to/other');
const chdirSpy = vi.spyOn(process, 'chdir').mockImplementation(() => {
@@ -8545,10 +8562,13 @@ describe('Server Config (config.ts)', () => {
// The patch rides its own fire-and-forget chain; let it settle.
await vi.waitFor(() =>
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
}),
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
},
sessionRegistry.SHARED_RECORD_SLOT,
),
);
expect(writeRuntimeStatusSpy).not.toHaveBeenCalled();
@@ -8611,7 +8631,7 @@ describe('Server Config (config.ts)', () => {
// until process exit.
const config = new Config(baseParams);
config.markRuntimeStatusEnabled();
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
const writeRuntimeStatusSpy = vi
.spyOn(runtimeStatus, 'writeRuntimeStatus')
.mockRejectedValue(new Error('read-only project fs'));
@@ -8622,10 +8642,13 @@ describe('Server Config (config.ts)', () => {
const newSessionId = config.startNewSession('replacement-session');
await vi.waitFor(() =>
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
sessionId: newSessionId,
cwd: config.getTargetDir(),
}),
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
sessionId: newSessionId,
cwd: config.getTargetDir(),
},
sessionRegistry.SHARED_RECORD_SLOT,
),
);
writeRuntimeStatusSpy.mockRestore();
@@ -8635,9 +8658,12 @@ describe('Server Config (config.ts)', () => {
it('serializes pending registration, transitions, and unregister', async () => {
const config = new Config(baseParams);
let finishRegistration!: (registered: boolean) => void;
const registration = new Promise<boolean>((resolve) => {
finishRegistration = resolve;
});
const registration = new Promise<sessionRegistry.SessionRegistration>(
(resolve) => {
finishRegistration = (registered) =>
resolve({ registered, slot: sessionRegistry.SHARED_RECORD_SLOT });
},
);
let finishPatch!: () => void;
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8660,10 +8686,13 @@ describe('Server Config (config.ts)', () => {
finishRegistration(true);
await vi.waitFor(() => {
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
sessionId: newSessionId,
cwd: config.getTargetDir(),
});
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
sessionId: newSessionId,
cwd: config.getTargetDir(),
},
sessionRegistry.SHARED_RECORD_SLOT,
);
});
expect(unregisterSessionSpy).not.toHaveBeenCalled();
@@ -8677,7 +8706,7 @@ describe('Server Config (config.ts)', () => {
it('serializes the peer inbox address with session transitions', async () => {
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
let finishIpcPatch!: () => void;
@@ -8717,7 +8746,7 @@ describe('Server Config (config.ts)', () => {
// ipcPath, so a patch skipped on transient fd pressure must retry
// itself or the inbox stays undiscoverable until restart.
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8727,18 +8756,62 @@ describe('Server Config (config.ts)', () => {
await config.updateSessionRegistryIpcPath('/tmp/peer.sock');
expect(patchSessionRecordSpy).toHaveBeenCalledTimes(2);
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
ipcPath: '/tmp/peer.sock',
});
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
ipcPath: '/tmp/peer.sock',
},
sessionRegistry.SHARED_RECORD_SLOT,
);
patchSessionRecordSpy.mockRestore();
});
it('names the minted record a hosted session registered under, on every write', async () => {
// A process hosting several sessions owns one record each. Every
// patch and the final removal have to name the right one — with the
// default slot they would all resolve to a `<pid>.json` that such a
// process never wrote, so a hosted session's record would never be
// updated and never be removed.
const config = new Config(baseParams);
config.trackSessionRegistration(
Promise.resolve({ registered: true, slot: 'a1b2c3d4' }),
);
await expect(config.whenSessionRegistered()).resolves.toBe(true);
expect(config.getSessionRegistrySlot()).toBe('a1b2c3d4');
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
.mockResolvedValue(true);
const unregisterSessionSpy = vi
.spyOn(sessionRegistry, 'unregisterSession')
.mockResolvedValue(undefined);
await config.updateSessionRegistryIpcPath('/tmp/acp.sock', 'tok');
expect(patchSessionRecordSpy).toHaveBeenLastCalledWith(
{ ipcPath: '/tmp/acp.sock', ipcToken: 'tok' },
'a1b2c3d4',
);
config.startNewSession('replacement-session');
await vi.waitFor(() =>
expect(patchSessionRecordSpy).toHaveBeenLastCalledWith(
expect.objectContaining({ sessionId: 'replacement-session' }),
'a1b2c3d4',
),
);
await config.unregisterSessionRegistry();
expect(unregisterSessionSpy).toHaveBeenCalledWith('a1b2c3d4');
patchSessionRecordSpy.mockRestore();
unregisterSessionSpy.mockRestore();
});
it('re-asserts the registry record with the current session id, retrying a skipped patch', async () => {
// A peer message pinned to an id this process does not hold means the
// record may be the stale side (a /clear patch skipped under fd
// pressure); re-asserting is the fix, and it retries like the advertise.
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8748,16 +8821,19 @@ describe('Server Config (config.ts)', () => {
await config.reassertSessionRegistryRecord();
expect(patchSessionRecordSpy).toHaveBeenCalledTimes(2);
expect(patchSessionRecordSpy).toHaveBeenLastCalledWith({
sessionId: config.getSessionId(),
cwd: config.getTargetDir(),
});
expect(patchSessionRecordSpy).toHaveBeenLastCalledWith(
{
sessionId: config.getSessionId(),
cwd: config.getTargetDir(),
},
sessionRegistry.SHARED_RECORD_SLOT,
);
patchSessionRecordSpy.mockRestore();
});
it('bounds the re-assert retry and is a no-op with no registration', async () => {
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8776,7 +8852,7 @@ describe('Server Config (config.ts)', () => {
it('retries the /clear session-id patch when the registry skips it', async () => {
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8805,7 +8881,7 @@ describe('Server Config (config.ts)', () => {
// would publish an address peers cannot authenticate to — sends read as
// 'sent' and are silently dropped — with the whole suite still green.
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8824,7 +8900,7 @@ describe('Server Config (config.ts)', () => {
it('gives up on the peer inbox advertise after a bounded retry', async () => {
const config = new Config(baseParams);
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
await expect(config.whenSessionRegistered()).resolves.toBe(true);
const patchSessionRecordSpy = vi
.spyOn(sessionRegistry, 'patchSessionRecord')
@@ -8844,7 +8920,7 @@ describe('Server Config (config.ts)', () => {
.spyOn(sessionRegistry, 'unregisterSession')
.mockResolvedValue(undefined);
config.trackSessionRegistration(Promise.resolve(false));
config.trackSessionRegistration(sharedRegistration(false));
await config.unregisterSessionRegistry();
expect(unregisterSessionSpy).not.toHaveBeenCalled();
@@ -8857,7 +8933,7 @@ describe('Server Config (config.ts)', () => {
// surface through relocateWorkingDirectory either.
const config = new Config(baseParams);
config.markRuntimeStatusEnabled();
config.trackSessionRegistration(Promise.resolve(true));
config.trackSessionRegistration(sharedRegistration());
const sessionId = config.getSessionId();
const newDir = path.resolve('/path/to/other');
const chdirSpy = vi.spyOn(process, 'chdir').mockImplementation(() => {
@@ -8877,10 +8953,13 @@ describe('Server Config (config.ts)', () => {
await config.relocateWorkingDirectory(newDir);
await vi.waitFor(() =>
expect(patchSessionRecordSpy).toHaveBeenCalledWith({
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
}),
expect(patchSessionRecordSpy).toHaveBeenCalledWith(
{
cwd: newDir,
name: sessionRegistry.deriveSessionName(newDir, sessionId),
},
sessionRegistry.SHARED_RECORD_SLOT,
),
);
writeRuntimeStatusSpy.mockRestore();
+44 -11
View File
@@ -241,6 +241,9 @@ import {
import {
deriveSessionName,
patchSessionRecord,
SHARED_RECORD_SLOT,
type SessionRecordSlot,
type SessionRegistration,
unregisterSession,
} from '../services/session-registry.js';
import { delay } from '../utils/retry.js';
@@ -2424,6 +2427,15 @@ export class Config {
private runtimeStatusEnabled = false;
private sessionRegistryActive = false;
private sessionRegistered = false;
/**
* Which of this process's registry records belongs to this session.
*
* A process hosting one session owns the shared `<pid>.json`; one
* hosting several owns a minted record per session, and every patch and
* the final removal have to name the right one or they would rewrite a
* sibling's record. Learned from the registration itself.
*/
private sessionRegistrySlot: SessionRecordSlot = SHARED_RECORD_SLOT;
private readonly experimentalZedIntegration: boolean = false;
private readonly restoreAskUserQuestion: boolean = false;
/**
@@ -4704,10 +4716,11 @@ export class Config {
failureWarning: string,
): void {
this.queueSessionRegistryWrite(async () => {
let applied = await patchSessionRecord(patch);
const slot = this.sessionRegistrySlot;
let applied = await patchSessionRecord(patch, slot);
for (let attempt = 0; attempt < 2 && !applied; attempt += 1) {
await delay(250);
applied = await patchSessionRecord(patch);
applied = await patchSessionRecord(patch, slot);
}
if (!applied) {
this.debugLogger.warn(failureWarning);
@@ -4732,14 +4745,19 @@ export class Config {
* Serializes initial registration with mid-session patches and cleanup.
* The registration promise is deliberately not awaited by UI startup.
*/
trackSessionRegistration(registration: Promise<boolean>): void {
trackSessionRegistration(registration: Promise<SessionRegistration>): void {
this.sessionRegistryActive = true;
this.sessionRegistryWrite = this.sessionRegistryWrite
.catch(() => {
// Keep registration independent from an earlier best-effort write.
})
.then(async () => {
this.sessionRegistered = await registration;
// The slot is taken whether or not the write landed: it names the
// record this session would own, and every later patch is queued
// behind this step, so none of them can run against the wrong one.
const outcome = await registration;
this.sessionRegistrySlot = outcome.slot;
this.sessionRegistered = outcome.registered;
if (!this.sessionRegistered) this.sessionRegistryActive = false;
})
.catch(() => {
@@ -4748,6 +4766,17 @@ export class Config {
});
}
/**
* Which registry record describes this session.
*
* Read by the send path: a process hosting several sessions has a record
* each, and the reply address, name and id a message carries have to
* come from the sending session's own.
*/
getSessionRegistrySlot(): SessionRecordSlot {
return this.sessionRegistrySlot;
}
/**
* Resolves once initial registration has settled, reporting whether a
* record actually exists.
@@ -4773,7 +4802,8 @@ export class Config {
if (!this.sessionRegistryActive) return;
let applied = false;
this.queueSessionRegistryWrite(async () => {
applied = await patchSessionRecord({ ipcPath, ipcToken });
const slot = this.sessionRegistrySlot;
applied = await patchSessionRecord({ ipcPath, ipcToken }, slot);
if (ipcPath === undefined || applied) return;
// The advertise is one-shot: no later patch re-asserts ipcPath, and
// every skip is transient (the fd-pressure window on this process's
@@ -4782,7 +4812,7 @@ export class Config {
// session would keep a live inbox no peer can ever discover.
for (let attempt = 0; attempt < 2 && !applied; attempt += 1) {
await delay(250);
applied = await patchSessionRecord({ ipcPath, ipcToken });
applied = await patchSessionRecord({ ipcPath, ipcToken }, slot);
}
if (!applied) {
this.debugLogger.warn(
@@ -4803,7 +4833,7 @@ export class Config {
.then(async () => {
if (!this.sessionRegistered) return;
this.sessionRegistered = false;
await unregisterSession();
await unregisterSession(this.sessionRegistrySlot);
})
.catch(() => {
// ignored: registry cleanup must not disrupt process teardown.
@@ -4882,10 +4912,13 @@ export class Config {
// folder this session left. Unlike the /clear path, `name`
// follows: it is derived from the directory's basename, which is
// exactly what changed here.
await patchSessionRecord({
cwd: workDir,
name: deriveSessionName(workDir, sessionId),
});
await patchSessionRecord(
{
cwd: workDir,
name: deriveSessionName(workDir, sessionId),
},
this.sessionRegistrySlot,
);
});
}
await this.flushRuntimeStatusWrites();
@@ -821,6 +821,60 @@ describe('receipts', () => {
expect(statuses).toEqual(['held', 'misaddressed']);
});
it('judges a parked frame against the sessions a host still holds', () => {
// The multi-session shape: no single id to compare, so the release
// path asks whether the frame's addressee is still one of them. The
// ACP host that wires this today refuses everything on arrival, so
// nothing reaches here from it — the rule belongs to the gate all the
// same, and a caller that parks (an inbound policy that holds) must
// not have its pin judged by whichever check ran first.
const hosted = new Set(['session-a', 'session-b']);
const delivered: PeerUserFrame[] = [];
const statuses: string[] = [];
const gate = new InboundGate({
getApprovalMode: () => ApprovalMode.YOLO,
getPolicySetting: () => 'hold',
ownsSessionId: (id) => hosted.has(id),
deliver: (candidate) => delivered.push(candidate),
reportStatus: (_candidate, status) => statuses.push(status),
});
const forB = frame({ fromMode: 'prompting', toSessionId: 'session-b' });
expect(gate.admit(forB)).toBe('held');
expect(gate.decide(forB.msgId, 'approve')).toBe('done');
expect(delivered).toEqual([forB]);
// The addressee goes while its message waits: releasing it now would
// hand one session's message to whatever else the process hosts.
const forA = frame({ fromMode: 'prompting', toSessionId: 'session-a' });
expect(gate.admit(forA)).toBe('held');
hosted.delete('session-a');
expect(gate.decide(forA.msgId, 'approve')).toBe('gone');
expect(delivered).toEqual([forB]);
expect(statuses.at(-1)).toBe('misaddressed');
});
it('will not release a parked frame that named no session to a host of several', () => {
// With one session an unpinned frame could only have meant that one;
// with several there is nothing to guess from, so it is misaddressed
// on the release path exactly as it is on arrival.
const delivered: PeerUserFrame[] = [];
const statuses: string[] = [];
const gate = new InboundGate({
getApprovalMode: () => ApprovalMode.YOLO,
getPolicySetting: () => 'hold',
ownsSessionId: () => true,
deliver: (candidate) => delivered.push(candidate),
reportStatus: (_candidate, status) => statuses.push(status),
});
const unpinned = frame({ fromMode: 'prompting' });
expect(gate.admit(unpinned)).toBe('held');
expect(gate.decide(unpinned.msgId, 'approve')).toBe('gone');
expect(delivered).toEqual([]);
expect(statuses).toEqual(['held', 'misaddressed']);
});
it('tombstones a misaddressed drop so a re-send repeats the verdict', () => {
for (const path of ['decide', 'reevaluate'] as const) {
let currentSessionId = 'session-a';
+24
View File
@@ -346,6 +346,16 @@ export interface InboundGateOptions {
* into the session that replaced its addressee.
*/
getSessionId?: () => string | undefined;
/**
* For a process hosting several sessions: whether `id` is one of them.
*
* Wired instead of `getSessionId`, which asks "what is the one session
* here" — a question such a process has no single answer to. A frame
* that names no session at all is not for any of them: with one session
* an unpinned frame can only have meant that one, and with several
* there is nothing to guess from.
*/
ownsSessionId?: (id: string) => boolean;
/**
* How long a message may sit parked before it expires, in
* milliseconds, or null to keep it until the session ends. Read on
@@ -624,6 +634,12 @@ export class InboundGate {
frame: PeerUserFrame,
origin: PeerOrigin = { selfSent: false },
): GateDecision {
// Only a process holding one session can "change session" — that is
// what a `/clear` or a resume does, and the meter starts over because
// the conversation it was pacing is gone. A process hosting several
// wires `ownsSessionId` instead, has no single id to compare, and
// meters across all of them: its sessions come and go constantly, and
// resetting on each would hand a flooding peer a fresh allowance.
const sessionId = this.options.getSessionId?.();
if (
this.admissionSessionObserved &&
@@ -971,6 +987,14 @@ export class InboundGate {
* session holds now, not the one the frame saw on arrival.
*/
private pinStillValid(frame: PeerUserFrame): boolean {
const ownsSessionId = this.options.ownsSessionId;
if (ownsSessionId) {
// A process hosting several sessions cannot act on an unpinned
// frame: there is no one session it could have meant.
return (
frame.toSessionId !== undefined && ownsSessionId(frame.toSessionId)
);
}
if (frame.toSessionId === undefined) return true;
const ownSessionId = this.options.getSessionId?.();
return ownSessionId === undefined || frame.toSessionId === ownSessionId;
@@ -287,6 +287,50 @@ describe('listMessageablePeers', () => {
expect(peers.map((p) => p.sessionId)).toEqual(['s1']);
});
it('probes each inbox address once, not once per record', async () => {
// A process hosting several sessions advertises one inbox in every
// one of their records. Dialling per record would open a fistful of
// simultaneous connections to a single socket on every listing and
// every send, from every session on the machine — and the answer is
// a property of the address, so asking twice cannot differ.
listLiveSessions.mockResolvedValue([
record({ sessionId: 'a', cwd: '/w/a', ipcPath: '/tmp/hosted.sock' }),
record({ sessionId: 'b', cwd: '/w/b', ipcPath: '/tmp/hosted.sock' }),
record({ sessionId: 'c', cwd: '/w/c', ipcPath: '/tmp/hosted.sock' }),
record({ sessionId: 'd', cwd: '/w/d', ipcPath: '/tmp/other.sock' }),
]);
probePeerSocketVerdict.mockResolvedValue('alive');
const peers = await listMessageablePeers();
expect(probePeerSocketVerdict).toHaveBeenCalledTimes(2);
expect(
probePeerSocketVerdict.mock.calls.map(([path]) => path).sort(),
).toEqual(['/tmp/hosted.sock', '/tmp/other.sock']);
// Every record behind the one answer is still advertised.
expect(peers.map((peer) => peer.sessionId).sort()).toEqual([
'a',
'b',
'c',
'd',
]);
});
it('drops every record behind an address that does not answer', async () => {
listLiveSessions.mockResolvedValue([
record({ sessionId: 'a', cwd: '/w/a', ipcPath: '/tmp/hosted.sock' }),
record({ sessionId: 'b', cwd: '/w/b', ipcPath: '/tmp/hosted.sock' }),
record({ sessionId: 'd', cwd: '/w/d', ipcPath: '/tmp/other.sock' }),
]);
probePeerSocketVerdict.mockImplementation(async (path: string) =>
path === '/tmp/other.sock' ? 'alive' : 'dead',
);
expect(
(await listMessageablePeers()).map((peer) => peer.sessionId),
).toEqual(['d']);
});
it('probes concurrently rather than one at a time', async () => {
listLiveSessions.mockResolvedValue([
record({ sessionId: 's1', ipcPath: '/tmp/s1.sock' }),
+16 -4
View File
@@ -110,16 +110,28 @@ export async function listMessageablePeers(): Promise<PeerSessionInfo[]> {
.map(toPeerSessionInfo)
.filter((peer): peer is PeerSessionInfo => peer !== null);
const verdicts = await Promise.all(
candidates.map((peer) => probePeerSocketVerdict(peer.ipcPath)),
// One address per probe, not one record per probe. A process hosting
// several sessions advertises the same inbox in every one of their
// records, and dialling it once per record would mean a fistful of
// simultaneous connections to a single socket on every listing and
// every send — from every session on the machine. The answer is a
// property of the address, so asking once is also the honest shape.
const verdicts = new Map(
await Promise.all(
[...new Set(candidates.map((peer) => peer.ipcPath))].map(
async (ipcPath) =>
[ipcPath, await probePeerSocketVerdict(ipcPath)] as const,
),
),
);
// Only a definitive `alive` advertises a peer. An `unknown` verdict --
// local descriptor exhaustion, a permission error, the 250 ms deadline --
// establishes nothing about the peer, and admitting it here would let a
// newer, unreachable twin shadow the one that answers in the dedupe
// below, whose tie-break is `startedAt` alone.
const reachable = verdicts.map((verdict) => verdict === 'alive');
return dedupeSameNameTwins(candidates.filter((_, index) => reachable[index]));
return dedupeSameNameTwins(
candidates.filter((peer) => verdicts.get(peer.ipcPath) === 'alive'),
);
}
/**
+83 -2
View File
@@ -263,6 +263,49 @@ describe('sendToPeer', () => {
expect(sendPeerFrame).not.toHaveBeenCalled();
});
it("delivers to a sibling session that shares this session's inbox", async () => {
// A process hosting several sessions binds one inbox for all of
// them, so a sibling's reply address is this session's own. It is
// still a different session with its own id — excluding by address
// would hide every sibling of the sending session from it.
const sibling = { ...peer('sibling', 'app-ab'), ipcPath: '/tmp/self.sock' };
listMessageablePeers.mockResolvedValue([sibling]);
const outcome = await sendToPeer({
target: 'app-ab',
message: 'hi',
approvalMode: ApprovalMode.DEFAULT,
});
expect(outcome).toMatchObject({ kind: 'sent', address: 'app-ab' });
const [socketPath, frame] = sendPeerFrame.mock.calls[0];
expect(socketPath).toBe('/tmp/self.sock');
expect(frame.toSessionId).toBe('sibling');
});
it("re-reads a peer on this session's own inbox before believing it a sibling", async () => {
// A patch to this session's own record — a /clear re-id, or the
// re-assert a misaddressed inbound frame triggers — can land between
// the own-record read and the directory read, and the stale id
// filter then keeps this session's own record under its new id.
// The address does not move on a re-id, so the entry is checked
// against a fresh read before it is believed to be a sibling.
readOwnSessionRecord.mockResolvedValueOnce({ ...SELF, sessionId: 's1' });
readOwnSessionRecord.mockResolvedValue({ ...SELF, sessionId: 's2' });
listMessageablePeers.mockResolvedValue([
{ ...peer('s2', 'self-00', '/w/self'), ipcPath: '/tmp/self.sock' },
]);
const outcome = await sendToPeer({
target: 'self-00',
message: 'hi',
approvalMode: ApprovalMode.DEFAULT,
});
expect(outcome).toEqual({ kind: 'self', name: 'self-00' });
expect(sendPeerFrame).not.toHaveBeenCalled();
});
it('treats a differently named twin of this session as itself', async () => {
// `qwen --resume <id>` from another directory runs this very session
// id under a second process with another name. Its inbound gate would
@@ -1087,13 +1130,51 @@ describe('the mirror and the receiver disagreeing', () => {
}
});
it('starts a fresh mirror conversation when the session id changes', async () => {
it('keeps one mirror per inbox address when the session id changes', async () => {
// Several sessions can share one inbox, and the receiver meters by
// the address a frame arrives on — so a new session id under the
// same address neither refills the bucket nor forgets the bodies,
// or alternating between siblings would reset both on every send.
expect((await send('same body')).kind).toBe('sent');
listMessageablePeers.mockResolvedValue([
{ ...target, sessionId: 'p2', ref: peerRef('p2') },
]);
expect((await send('same body')).kind).toBe('sent');
const repeat = await send('same body');
expect(repeat.kind).toBe('failed');
expect(repeat.kind === 'failed' && repeat.reason).toContain(
'turns away a repeat',
);
});
it('paces two sessions behind one inbox as one destination', async () => {
const siblingA = { ...peer('sa', 'app-aa'), ipcPath: '/tmp/shared.sock' };
const siblingB = { ...peer('sb', 'app-bb'), ipcPath: '/tmp/shared.sock' };
listMessageablePeers.mockResolvedValue([siblingA, siblingB]);
const sendTo = (name: string, message: string) =>
sendToPeer({ target: name, message, approvalMode: ApprovalMode.DEFAULT });
// The duplicate window is shared: a body one sibling just received
// is a repeat when addressed to the other.
expect((await sendTo('app-aa', 'same body')).kind).toBe('sent');
const repeat = await sendTo('app-bb', 'same body');
expect(repeat.kind).toBe('failed');
expect(repeat.kind === 'failed' && repeat.reason).toContain(
'turns away a repeat',
);
// And the rate budget is shared: a burst exhausted against one
// sibling leaves nothing for the other.
for (
let index = 0;
index < PEER_ADMISSION_LIMITS.bucketCapacity - 1;
index++
) {
expect((await sendTo('app-aa', `burst ${index}`)).kind).toBe('sent');
}
const over = await sendTo('app-bb', 'one too many');
expect(over.kind).toBe('failed');
expect(over.kind === 'failed' && over.reason).toContain('rate limit');
});
it('does not un-drain the mirror with a refund that lands after the drain', async () => {
+52 -34
View File
@@ -12,7 +12,10 @@
*/
import type { ApprovalMode } from '../config/approval-mode.js';
import { readOwnSessionRecord } from '../services/session-registry.js';
import {
readOwnSessionRecord,
type SessionRecordSlot,
} from '../services/session-registry.js';
import { modeClass } from './inbound-gate.js';
import {
hasToken,
@@ -53,8 +56,10 @@ export interface OwnPeerIdentity {
ref: string;
}
export async function getOwnPeerIdentity(): Promise<OwnPeerIdentity | null> {
const record = await readOwnSessionRecord();
export async function getOwnPeerIdentity(
slot?: SessionRecordSlot,
): Promise<OwnPeerIdentity | null> {
const record = await readOwnSessionRecord(slot);
// The same projection peers see, so the name this session reports for
// itself is the flattened one they would type.
const self = record === null ? null : toPeerSessionInfo(record);
@@ -269,10 +274,12 @@ export const MAX_PACED_TARGETS = 256;
* fails, the tool result says to batch, and the receiver never spends a
* connection on a message it was going to drop.
*
* Keyed by socket path rather than by name, with the session id retained
* in the value. The path is what an inbox is and what the receiver meters
* by (`from` is the mirror image of this key), while the id detects a new
* session that reused the same path and therefore owns a fresh bucket.
* Keyed by socket path rather than by name, and by path alone. The path
* is what an inbox is and what the receiver meters by (`from` is the
* mirror image of this key), and several sessions can share one inbox:
* re-minting the bucket when the addressed session id changes would let
* a sender alternating between siblings reset its allowance on every
* send, and forget the bodies the receiver still remembers.
*
* A mirror can only ever be approximate: this session is not the only one
* sending, and the receiver's bucket is shared. It is deliberately no
@@ -290,7 +297,6 @@ interface PacedBody {
}
interface PacedTarget {
sessionId: string;
tokens: number;
lastRefill: number;
/** Sends in the current burst, for the message the refusal carries. */
@@ -315,19 +321,13 @@ interface PacedTarget {
const pacedTargets = new Map<string, PacedTarget>();
function pacedTargetFor(
ipcPath: string,
sessionId: string,
now: number,
): PacedTarget {
function pacedTargetFor(ipcPath: string, now: number): PacedTarget {
const existing = pacedTargets.get(ipcPath);
if (existing?.sessionId === sessionId) {
if (existing !== undefined) {
pacedTargets.delete(ipcPath);
pacedTargets.set(ipcPath, existing);
return existing;
}
const generation = (existing?.generation ?? -1) + 1;
pacedTargets.delete(ipcPath);
while (pacedTargets.size >= MAX_PACED_TARGETS) {
let victim: string | undefined;
for (const [candidate, target] of pacedTargets) {
@@ -348,13 +348,12 @@ function pacedTargetFor(
pacedTargets.delete(victim);
}
const fresh: PacedTarget = {
sessionId,
tokens: PEER_ADMISSION_LIMITS.bucketCapacity,
lastRefill: now,
sentInBurst: 0,
burstStartedAt: now,
bodies: [],
generation,
generation: 0,
};
pacedTargets.set(ipcPath, fresh);
return fresh;
@@ -386,7 +385,6 @@ export function setSendPacerClockForTest(
*/
function reservePacerToken(
ipcPath: string,
sessionId: string,
body: string,
messageId: string,
):
@@ -395,7 +393,7 @@ function reservePacerToken(
| { ok: false; repeat?: false; sentInBurst: number } {
const now = pacerNow();
const wallNow = pacerWallNow();
const target = pacedTargetFor(ipcPath, sessionId, now);
const target = pacedTargetFor(ipcPath, now);
target.tokens = refillBucket(
target.tokens,
target.lastRefill,
@@ -593,6 +591,13 @@ export interface SendToPeerOptions {
* the form that reaches it.
*/
isReserved?: (address: string) => boolean;
/**
* Which of this process's registry records describes the sending
* session. Omitted by a process holding one session, which is most of
* them; a process hosting several has a record per session, and the
* `from`, name and id on the frame have to come from the right one.
*/
slot?: SessionRecordSlot;
}
/**
@@ -605,21 +610,23 @@ export interface SendToPeerOptions {
export async function sendToPeer(
options: SendToPeerOptions,
): Promise<PeerSendOutcome> {
const own = await readOwnSessionRecord();
const own = await readOwnSessionRecord(options.slot);
const self = own === null ? null : toPeerSessionInfo(own);
if (!self) return { kind: 'disabled' };
const directory = await listMessageablePeers();
// Exclude every incarnation of this session, not just its own socket:
// the registry is keyed by PID, and `qwen --resume <id>` from a second
// pane runs the same session id under another process — differently
// named when resumed from another directory. The receiver's gate
// accepts a frame pinned to its own id, so such a twin would deliver a
// message right back to this session while the ledger reads delivered.
const peers = directory.filter(
(peer) =>
peer.ipcPath !== self.ipcPath && peer.sessionId !== self.sessionId,
);
// Exclude every incarnation of this session — `qwen --resume <id>` from
// a second pane runs the same session id under another process,
// differently named when resumed from another directory. The receiver's
// gate accepts a frame pinned to its own id, so such a twin would
// deliver a message right back to this session while the ledger reads
// delivered.
//
// By session id alone. The reply address used to join the test as a
// proxy for the same thing, and stopped being one when a process could
// host several sessions: they share one inbox, so excluding by address
// would hide every sibling of the sending session from it.
const peers = directory.filter((peer) => peer.sessionId !== self.sessionId);
const resolved = resolvePeerTarget(peers, options.target);
if (resolved.kind === 'none') {
@@ -668,6 +675,18 @@ export async function sendToPeer(
}
const peer = resolved.peer;
// This session's own record can be patched — a /clear re-id, or the
// re-assert a misaddressed inbound frame triggers — between the two
// reads above, and the stale id filter then keeps it under its new
// id. The reply address does not move on a re-id, so a peer on this
// session's own inbox is checked against a fresh read before it is
// believed to be a sibling.
if (peer.ipcPath === self.ipcPath) {
const fresh = await readOwnSessionRecord(options.slot);
if (fresh !== null && fresh.sessionId === peer.sessionId) {
return { kind: 'self', name: self.name };
}
}
// The address the ledger remembers is the one list_agents would print:
// a receipt that names an address which re-resolves ambiguous — or that
// the listing never showed — sends the model in circles.
@@ -716,7 +735,6 @@ export async function sendToPeer(
});
const reservation = reservePacerToken(
peer.ipcPath,
peer.sessionId,
options.message,
frame.msgId,
);
@@ -726,8 +744,8 @@ export async function sendToPeer(
peer,
address,
reason: reservation.repeat
? `that exact message went to that session within the last ${PEER_ADMISSION_LIMITS.dedupWindowMs / 1000} seconds, and its ` +
'inbox turns away a repeat before anyone reads it, so this one was not sent. ' +
? `that exact message went to that inbox within the last ${PEER_ADMISSION_LIMITS.dedupWindowMs / 1000} seconds, and it ` +
'turns away a repeat before anyone reads it, so this one was not sent. ' +
'Say something different, or wait for a reply rather than re-sending.'
: reservation.sentInBurst === 0
? 'that session inbox is still over its rate limit, so this message was not sent. Wait a little before sending more.'
@@ -242,11 +242,13 @@ describe('registerSession', () => {
it('writes a record for this process and lists it back', async () => {
const before = Date.now();
expect(
await registerSession({
sessionId: 's1',
cwd: '/w/app',
qwenVersion: '1.2.3',
}),
(
await registerSession({
sessionId: 's1',
cwd: '/w/app',
qwenVersion: '1.2.3',
})
).registered,
).toBe(true);
const after = Date.now();
@@ -399,9 +401,9 @@ describe('registerSession', () => {
await fs.writeFile(outside, 'untouched');
await fs.symlink(outside, getSessionRecordPath());
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(true);
expect(await fs.readFile(outside, 'utf8')).toBe('untouched');
expect((await fs.lstat(getSessionRecordPath())).isSymbolicLink()).toBe(
@@ -418,9 +420,9 @@ describe('registerSession', () => {
const foreign = liveBody({ pidNs: 1, sessionId: 'theirs' });
await writeRaw(`${process.pid}.json`, foreign);
expect(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).registered,
).toBe(false);
expect(
JSON.parse(await fs.readFile(getSessionRecordPath(), 'utf8')),
@@ -440,9 +442,10 @@ describe('registerSession', () => {
});
await writeRaw(`${process.pid}.json`, foreign);
expect(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 'mine', cwd: '/w/app' }))
.registered,
).toBe(false);
expect(
JSON.parse(await fs.readFile(getSessionRecordPath(), 'utf8')),
@@ -464,9 +467,10 @@ describe('registerSession', () => {
liveBody({ procStart: `${bootId}:1`, sessionId: 'stale' }),
);
expect(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 'mine', cwd: '/w/app' }))
.registered,
).toBe(true);
const raw = JSON.parse(
await fs.readFile(getSessionRecordPath(), 'utf8'),
@@ -487,9 +491,9 @@ describe('registerSession', () => {
// destroy ours through.
vi.spyOn(processLiveness, 'readProcStartToken').mockReturnValue(null);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(false);
await expect(fs.stat(getSessionRecordPath())).rejects.toThrow();
expect(await listLiveSessions()).toEqual([]);
@@ -503,9 +507,9 @@ describe('registerSession', () => {
// a transient fd-pressure moment; only a persistent outage refuses.
vi.spyOn(processLiveness, 'readProcStartToken').mockReturnValueOnce(null);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(true);
const raw = JSON.parse(
await fs.readFile(getSessionRecordPath(), 'utf8'),
@@ -524,9 +528,9 @@ describe('registerSession', () => {
// in place — poisoning the PID slot for the next session.
vi.spyOn(processLiveness, 'readPidNamespaceId').mockReturnValue(null);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(false);
await expect(fs.stat(getSessionRecordPath())).rejects.toThrow();
expect(await listLiveSessions()).toEqual([]);
@@ -540,9 +544,9 @@ describe('registerSession', () => {
// and only a persistent outage refuses.
vi.spyOn(processLiveness, 'readPidNamespaceId').mockReturnValueOnce(null);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(true);
const raw = JSON.parse(
await fs.readFile(getSessionRecordPath(), 'utf8'),
@@ -562,9 +566,9 @@ describe('registerSession', () => {
err.code = 'ESTALE';
vi.spyOn(fs, 'readFile').mockRejectedValueOnce(err);
expect(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).registered,
).toBe(false);
expect(
JSON.parse(await fs.readFile(getSessionRecordPath(), 'utf8')),
@@ -581,9 +585,9 @@ describe('registerSession', () => {
err.code = 'ENOTSUP';
vi.spyOn(fs, 'chmod').mockRejectedValue(err);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(true);
const raw = JSON.parse(await fs.readFile(getSessionRecordPath(), 'utf8'));
expect(raw.sessionId).toBe('s1');
@@ -596,9 +600,9 @@ describe('registerSession', () => {
err.code = 'EPERM';
vi.spyOn(fs, 'chmod').mockRejectedValue(err);
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(false);
await expect(fs.stat(getSessionRecordPath())).rejects.toThrow();
});
@@ -617,9 +621,9 @@ describe('registerSession', () => {
});
await writeRaw(`${process.pid}.json`, future);
expect(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).toBe(
false,
);
expect(
(await registerSession({ sessionId: 'mine', cwd: '/w/app' })).registered,
).toBe(false);
await patchSessionRecord({ sessionId: 'mine' });
await unregisterSession();
@@ -633,9 +637,9 @@ describe('registerSession', () => {
// call, and /cd changes the cwd mid-session: patch and unregister
// must keep hitting the directory registration wrote to, or the /cd
// patch silently no-ops and exit leaks the record.
expect(await registerSession({ sessionId: 's1', cwd: '/w/app' })).toBe(
true,
);
expect(
(await registerSession({ sessionId: 's1', cwd: '/w/app' })).registered,
).toBe(true);
const originalPath = getSessionRecordPath();
__setMockGlobalDir(path.join(tmpDir, 'moved-home'));
@@ -657,14 +661,229 @@ describe('registerSession', () => {
it('reports failure instead of throwing when the home dir is unwritable', async () => {
__setMockGlobalDir(path.join(tmpDir, 'nope', '\0invalid'));
expect(
await registerSession({
sessionId: 's1',
cwd: '/w/app',
}),
(
await registerSession({
sessionId: 's1',
cwd: '/w/app',
})
).registered,
).toBe(false);
});
});
describe('registerSession — one process, several records', () => {
it('writes a record of its own per session, keyed by a minted slot', async () => {
const first = await registerSession({
sessionId: 'a',
cwd: '/w/one',
kind: 'serve',
slot: 'own',
});
const second = await registerSession({
sessionId: 'b',
cwd: '/w/two',
kind: 'serve',
slot: 'own',
});
expect(first.registered && second.registered).toBe(true);
expect(first.slot).not.toBe(second.slot);
expect(first.slot).toMatch(/^[0-9a-f]{8}$/);
// Named by the writer's PID like every other record, so the sweep and
// the namespace guards judge them exactly as they judge a shared one.
const files = (await fs.readdir(getSessionRegistryDir())).sort();
expect(files).toEqual(
[
`${process.pid}-${first.slot}.json`,
`${process.pid}-${second.slot}.json`,
].sort(),
);
const live = await listLiveSessions();
expect(live.map((record) => record.sessionId).sort()).toEqual(['a', 'b']);
expect(live.every((record) => record.pid === process.pid)).toBe(true);
});
it('patches and removes one slot without touching its siblings', async () => {
const first = await registerSession({
sessionId: 'a',
cwd: '/w/one',
slot: 'own',
});
const second = await registerSession({
sessionId: 'b',
cwd: '/w/two',
slot: 'own',
});
expect(
await patchSessionRecord({ ipcPath: '/tmp/shared.sock' }, first.slot),
).toBe(true);
const afterPatch = await listLiveSessions();
expect(afterPatch.find((record) => record.sessionId === 'a')?.ipcPath).toBe(
'/tmp/shared.sock',
);
// The sibling is a separate file: a patch that resolved the path by
// PID alone would have rewritten whichever record it found first.
expect(
afterPatch.find((record) => record.sessionId === 'b')?.ipcPath,
).toBeUndefined();
await unregisterSession(second.slot);
const afterRemove = await listLiveSessions();
expect(afterRemove.map((record) => record.sessionId)).toEqual(['a']);
});
it('reads back the record of the slot it is asked for', async () => {
const first = await registerSession({
sessionId: 'a',
cwd: '/w/one',
slot: 'own',
});
const second = await registerSession({
sessionId: 'b',
cwd: '/w/two',
slot: 'own',
});
expect((await readOwnSessionRecord(first.slot))?.sessionId).toBe('a');
expect((await readOwnSessionRecord(second.slot))?.sessionId).toBe('b');
// A slot this process never registered names no record, and must not
// fall back to the PID-keyed path — that record belongs to whatever
// else is running under this PID, not to the caller.
expect(await readOwnSessionRecord('deadbeef')).toBeNull();
expect(await patchSessionRecord({ sessionId: 'x' }, 'deadbeef')).toBe(
false,
);
await expect(unregisterSession('deadbeef')).resolves.toBeUndefined();
expect((await listLiveSessions()).length).toBe(2);
});
it('never resolves an unknown slot onto a live shared record', async () => {
// The interesting arm of the rule above: with a shared record present
// at this PID's path, a stale or unknown minted slot must still name
// nothing. Falling back would let one session read, rewrite and
// finally unlink a record belonging to whatever else runs under this
// PID — the case the three assertions above cannot see, because they
// run with no shared record on disk at all.
await registerSession({ sessionId: 'plain', cwd: '/w/plain' });
const before = await listLiveSessions();
expect(before.map((record) => record.sessionId)).toEqual(['plain']);
expect(await readOwnSessionRecord('deadbeef')).toBeNull();
expect(await patchSessionRecord({ sessionId: 'stolen' }, 'deadbeef')).toBe(
false,
);
await unregisterSession('deadbeef');
const after = await listLiveSessions();
expect(after.map((record) => record.sessionId)).toEqual(['plain']);
});
it('keeps a minted record at its own filename when the session id is swapped', async () => {
// The invariant the whole design rests on: a `/clear` or a session
// load swaps the id underneath, and the record follows by patch. A
// filename derived from the id — or renamed on a patch — would
// strand every reader holding the old name.
const own = await registerSession({
sessionId: 'before',
cwd: '/w/hosted',
slot: 'own',
});
const fileName = `${process.pid}-${own.slot}.json`;
expect(await fs.readdir(getSessionRegistryDir())).toEqual([fileName]);
expect(await patchSessionRecord({ sessionId: 'after' }, own.slot)).toBe(
true,
);
expect(await fs.readdir(getSessionRegistryDir())).toEqual([fileName]);
expect((await readOwnSessionRecord(own.slot))?.sessionId).toBe('after');
expect((await listLiveSessions())[0]?.sessionId).toBe('after');
});
it('keeps the path of a record whose removal could not read it', async () => {
// A minted path cannot be derived a second time, so forgetting it on
// a transient read failure would leave a record this process can
// never name again — advertising a session that is gone until the
// PID dies. Every other exit has established the path is not ours.
const own = await registerSession({
sessionId: 'hosted',
cwd: '/w/hosted',
slot: 'own',
});
const failing = vi
.spyOn(fs, 'stat')
.mockRejectedValueOnce(
Object.assign(new Error('EIO'), { code: 'EIO' }) as never,
);
await unregisterSession(own.slot);
failing.mockRestore();
// Still there, and still addressable: the capture survived.
expect((await listLiveSessions()).map((r) => r.sessionId)).toEqual([
'hosted',
]);
await unregisterSession(own.slot);
expect(await listLiveSessions()).toEqual([]);
});
it('leaves the shared record alone, and is left alone by it', async () => {
const own = await registerSession({
sessionId: 'hosted',
cwd: '/w/hosted',
slot: 'own',
});
const plain = await registerSession({
sessionId: 'plain',
cwd: '/w/plain',
});
// Asserted, not assumed: any of registration's refusal branches
// hitting this call would leave the half of this test its name comes
// from checking nothing at all.
expect(plain.registered).toBe(true);
expect((await fs.readdir(getSessionRegistryDir())).sort()).toEqual(
[`${process.pid}.json`, `${process.pid}-${own.slot}.json`].sort(),
);
await unregisterSession();
const live = await listLiveSessions();
expect(live.map((record) => record.sessionId)).toEqual(['hosted']);
expect(live[0]?.name).toBe(deriveSessionName('/w/hosted', 'hosted'));
await unregisterSession(own.slot);
expect(await listLiveSessions()).toEqual([]);
});
it('sweeps a minted record whose process is gone, and its temp files', async () => {
// Same reaping the shared name gets: the filename says which PID to
// check, and a suffix does not change that answer.
const record = await writeRaw(
`${DEAD_PID}-a1b2c3d4.json`,
liveBody({ pid: DEAD_PID, startedAt: Date.now() }),
);
const temp = await writeRaw(
`${DEAD_PID}-a1b2c3d4.json.0123456789ab.tmp`,
'partial',
);
await fs.utimes(temp, new Date(0), new Date(0));
expect(await listLiveSessions()).toEqual([]);
await expect(fs.stat(record)).rejects.toThrow();
await expect(fs.stat(temp)).rejects.toThrow();
});
it('skips a minted record whose pid disagrees with its filename', async () => {
// The suffix is the only new thing in the name; the PID in front of it
// still has to match the record, or nothing can reason about it.
const filePath = await writeRaw(
`${process.pid}-a1b2c3d4.json`,
liveBody({ pid: process.pid + 1 }),
);
expect(await listLiveSessions()).toEqual([]);
await expect(fs.stat(filePath)).resolves.toBeDefined();
});
});
describe('never-throw guarantee', () => {
it('every entry point resolves when the home directory cannot be resolved', async () => {
// `os.homedir()` throws when HOME is unset and the passwd lookup
@@ -679,7 +898,7 @@ describe('never-throw guarantee', () => {
await expect(patchSessionRecord({ sessionId: 'new' })).resolves.toBe(false);
await expect(
registerSession({ sessionId: 's1', cwd: '/w/app' }),
).resolves.toBe(false);
).resolves.toMatchObject({ registered: false });
await expect(unregisterSession()).resolves.toBeUndefined();
});
});
@@ -1259,6 +1478,33 @@ describe('listLiveSessions', () => {
await expect(fs.stat(fresh)).resolves.toBeDefined();
});
it('ignores near-misses of the minted <pid>-<8 hex>.json shape', async () => {
// The suffix widened the grammar the sweep unlinks through, so the
// "matched exactly" strictness the filter claims needs near-misses of
// the NEW shape, not only of the old one. Each of these is a name a
// backup tool, another build, or a shared home could plausibly drop
// into the directory; every one of them names a dead PID, so a
// grammar that accepted it would both list a phantom and delete a
// file this code never wrote.
const nearMisses = [
`${DEAD_PID}-a1b2c3d.json`, // seven hex, not eight
`${DEAD_PID}-a1b2c3d4e.json`, // nine
`${DEAD_PID}-A1B2C3D4.json`, // uppercase
`${DEAD_PID}-a1b2c3g4.json`, // 'g' is not hex
`${DEAD_PID}-notes.json`, // words
`${DEAD_PID}-a1b2c3d4-e5f6a7b8.json`, // two suffixes
`${DEAD_PID}_a1b2c3d4.json`, // underscore, not dash
];
for (const name of nearMisses) {
await writeRaw(name, liveBody({ pid: DEAD_PID }));
}
expect(await listLiveSessions()).toEqual([]);
expect((await fs.readdir(getSessionRegistryDir())).sort()).toEqual(
[...nearMisses].sort(),
);
});
it('ignores files that are not <pid>.json', async () => {
await writeRaw('2026-planning-notes.json', { hello: 'world' });
await writeRaw('notes.txt', 'nope');
@@ -1281,6 +1527,30 @@ describe('listLiveSessions', () => {
]);
});
it('ignores a record whose zero-padded filename parses to its pid', async () => {
// The agreement check parses the pid out of the filename; without a
// canonical-form requirement `007` parses to 7 and a name this code
// never wrote passes a guard whose stated job is to reject it — and
// a passing name reaches the sweep's unlink. Linux masks this (its
// pidNs is never null), so the namespace read is spied to the value
// every other platform returns, and the filename grammar decides.
vi.spyOn(processLiveness, 'readPidNamespaceId').mockReturnValue(null);
// Live PID arm: the phantom is listed as a live session. Dead PID
// arm: the sweep unlinks a file this code never wrote.
const alive = await writeRaw(
`0${process.pid}.json`,
liveBody({ pidNs: null }),
);
const dead = await writeRaw(
`0${DEAD_PID}.json`,
liveBody({ pid: DEAD_PID, pidNs: null }),
);
expect(await listLiveSessions()).toEqual([]);
await expect(fs.stat(alive)).resolves.toBeDefined();
await expect(fs.stat(dead)).resolves.toBeDefined();
});
it('never opens a file that is not named <pid>.json', async () => {
await writeRaw('2026-planning-notes.json', { hello: 'world' });
await writeRaw('notes.txt', 'nope');
+148 -41
View File
@@ -12,6 +12,16 @@
* so that "who else is running" is one `readdir` plus a handful of small
* reads.
*
* A process that hosts several sessions at once — the `qwen --acp` child a
* daemon spawns — writes one record per session instead, named
* `<pid>-<8 hex>.json`. They are separate sessions with separate ids,
* names and working directories, and one record cannot describe them all.
* The suffix is minted at registration and never moves: a session id
* swapped underneath (a `/clear`, a session load) is a patch, because a
* record that renamed itself would strand every reader holding the old
* name. Which record a caller means is its `slot` — see
* {@link RegisterSessionFields.slot}.
*
* The index is scoped to one Qwen home: the global dir resolves
* `QWEN_HOME` on every call, so a session started under a redirected
* `QWEN_HOME` registers elsewhere and stays invisible to readers under
@@ -65,7 +75,7 @@
* undiscoverable until the file is removed by hand.
*/
import { createHash } from 'node:crypto';
import { createHash, randomBytes } from 'node:crypto';
import * as fs from 'node:fs/promises';
import * as path from 'node:path';
import { Storage } from '../config/storage.js';
@@ -97,20 +107,36 @@ const REGISTRY_FILE_MODE = 0o600;
const MAX_RECORD_BYTES = 64 * 1024;
/**
* Only `<digits>.json` is a candidate record.
* Only `<digits>.json`, or the `<digits>-<8 hex>.json` a process hosting
* several sessions writes, is a candidate record.
*
* This is deliberately strict. A lenient `parseInt` prefix match would
* read `2026-planning-notes.json` as PID 2026, fail its liveness check,
* and delete a file this code never wrote.
* and delete a file this code never wrote. The optional suffix is matched
* exactly, and is the same shape the peer socket uses when a PID
* collision forces it to a sibling name.
*/
const RECORD_FILENAME = /^\d+\.json$/;
const RECORD_FILENAME = /^\d+(-[0-9a-f]{8})?\.json$/;
/**
* Temp files left by `atomicWriteJSON` next to their target. Matched
* separately from `RECORD_FILENAME` so enumeration can reap the orphans
* a crashed registration leaves behind (see `sweepOrphanedTempFile`).
*/
const TEMP_FILENAME = /^\d+\.json\.[0-9a-f]{12}\.tmp$/;
const TEMP_FILENAME = /^\d+(-[0-9a-f]{8})?\.json\.[0-9a-f]{12}\.tmp$/;
/** The PID a record filename is keyed by, or null if it is not ours. */
function pidOfRecordFilename(name: string): number | null {
if (!RECORD_FILENAME.test(name)) return null;
const digits = name.split(/[-.]/)[0]!;
const pid = Number.parseInt(digits, 10);
// Canonical decimal form only: `007` parses to 7, but this code never
// writes a padded name, so one that passes here belongs to something
// else — and a passing name reaches the sweep's unlink.
return Number.isInteger(pid) && pid > 0 && String(pid) === digits
? pid
: null;
}
/** Younger temps may belong to a writer mid-rename — leave them alone. */
const TEMP_MAX_AGE_MS = 5 * 60 * 1000;
@@ -225,6 +251,41 @@ export interface RegisterSessionFields {
* derived name rather than recording a session nobody can address.
*/
name?: string;
/**
* Which record this registration owns.
*
* `shared` — the default, and what a process holding one session wants —
* is the single `<pid>.json` keyed by this PID. `own` mints a record of
* its own, `<pid>-<8 hex>.json`, for a process hosting several sessions
* at once: they are separate sessions with separate ids and working
* directories, and one record cannot describe them all.
*
* The suffix is minted rather than derived from the session id on
* purpose. An id can be swapped under a running session (`/clear`, a
* session load), and a record that renamed itself when that happened
* would strand every reader holding the old name; the record is patched
* in place instead, and its filename never moves.
*/
slot?: 'shared' | 'own';
}
/**
* Which record a registration owns, as its owner refers to it later.
*
* `SHARED_RECORD_SLOT` is the process-wide `<pid>.json`; anything else is
* the 8-hex suffix of a `<pid>-<8 hex>.json`. Patch, unregister and
* own-record reads all take one, so a process with several records says
* which of them it means.
*/
export type SessionRecordSlot = string;
/** The slot of the one record a single-session process writes. */
export const SHARED_RECORD_SLOT = 'shared';
/** What a registration reports back: whether it landed, and where. */
export interface SessionRegistration {
registered: boolean;
slot: SessionRecordSlot;
}
/**
@@ -272,27 +333,45 @@ export function getSessionRecordPath(): string {
}
/**
* The record path captured when registration succeeds. `getGlobalQwenDir()`
* resolves a relative `QWEN_HOME` against the CURRENT `process.cwd()` on
* every call, and `/cd` changes the cwd mid-session — so patch and
* unregister must keep operating on the directory registration wrote to
* rather than wherever the cwd moved afterwards. Null before a successful
* registration (both seams no-op then anyway) and consumed by unregister.
* The record paths captured when registrations succeed, by slot.
*
* `getGlobalQwenDir()` resolves a relative `QWEN_HOME` against the CURRENT
* `process.cwd()` on every call, and `/cd` changes the cwd mid-session —
* so patch and unregister must keep operating on the directory
* registration wrote to rather than wherever the cwd moved afterwards.
*
* A map rather than one path because a process can hold several records
* at once (see `RegisterSessionFields.slot`), and each of its sessions
* patches and removes its own.
*/
let registeredRecordPath: string | null = null;
const registeredRecordPaths = new Map<SessionRecordSlot, string>();
/**
* Test-only: clear the registration-path capture so a suite starts each
* Test-only: clear the registration-path captures so a suite starts each
* test from the unregistered state regardless of what an earlier test
* registered. Without a reset, tests that need the capture null only
* registered. Without a reset, tests that need the capture empty only
* pass by accident of test order.
*/
export function resetRegisteredRecordPathForTest(): void {
registeredRecordPath = null;
registeredRecordPaths.clear();
}
function thisProcessRecordPath(): string {
return registeredRecordPath ?? getSessionRecordPath();
/**
* Where a slot's record lives, or null when this process has none.
*
* The shared slot falls back to the PID-keyed path: a process that never
* registered still has one well-defined place its record would be, and
* the callers that read it are no-ops on a missing file anyway. A minted
* slot has no such fallback — its filename exists only because a
* registration returned it — so an unknown one is "no record", never a
* guessed path that could belong to another session.
*/
function thisProcessRecordPath(
slot: SessionRecordSlot = SHARED_RECORD_SLOT,
): string | null {
const captured = registeredRecordPaths.get(slot);
if (captured !== undefined) return captured;
return slot === SHARED_RECORD_SLOT ? getSessionRecordPath() : null;
}
/**
@@ -350,10 +429,19 @@ export function deriveSessionName(cwd: string, sessionId: string): string {
* swept by any healthy reader, so it would poison its PID slot until
* removed by hand. In all of these cases this session simply stays
* undiscoverable.
*
* Reports the slot alongside the verdict, because a process holding more
* than one record has to be told which one it just wrote in order to
* patch or remove it later.
*/
export async function registerSession(
fields: RegisterSessionFields,
): Promise<boolean> {
): Promise<SessionRegistration> {
// Minted before any of the refusals below so every exit reports the same
// slot, whether or not a record reached the disk. It is a name, not a
// reservation: nothing is written or held by minting one.
const slot =
fields.slot === 'own' ? randomBytes(4).toString('hex') : SHARED_RECORD_SLOT;
let procStart = readProcStartToken(process.pid);
if (process.platform === 'linux' && procStart === null) {
// Boot-id read failures are not cached, so a second attempt recovers
@@ -363,7 +451,7 @@ export async function registerSession(
debugLogger.debug(
'registerSession: start token unreadable; refusing to write an impersonable record',
);
return false;
return { registered: false, slot };
}
}
let pidNs = readPidNamespaceId();
@@ -380,7 +468,7 @@ export async function registerSession(
debugLogger.debug(
'registerSession: namespace id unreadable; refusing to write an unreclaimable record',
);
return false;
return { registered: false, slot };
}
}
const record: SessionRegistryRecord = {
@@ -398,7 +486,10 @@ export async function registerSession(
try {
const dir = getSessionRegistryDir();
const filePath = getSessionRecordPath();
const filePath =
slot === SHARED_RECORD_SLOT
? getSessionRecordPath()
: path.join(dir, `${process.pid}-${slot}.json`);
// A record already at our path may belong to a live session that
// collided with us on the PID number across a namespace or machine
// boundary (host + devcontainer, sibling containers, NFS-shared
@@ -413,19 +504,19 @@ export async function registerSession(
debugLogger.debug(
'registerSession: record path read failed transiently; refusing to overwrite an intact record',
);
return false;
return { registered: false, slot };
}
if (existing.status === 'unsupported-version') {
debugLogger.debug(
'registerSession: record path held by a newer-schema record',
);
return false;
return { registered: false, slot };
}
if (existing.status === 'ok' && !matchesLocalIdentity(existing.record)) {
debugLogger.debug(
'registerSession: record path held by a foreign-identity record',
);
return false;
return { registered: false, slot };
}
await fs.mkdir(dir, { recursive: true, mode: REGISTRY_DIR_MODE });
// mkdir's mode is masked by the umask, and does nothing at all when
@@ -448,11 +539,11 @@ export async function registerSession(
forceMode: true,
noFollow: true,
});
registeredRecordPath = filePath;
return true;
registeredRecordPaths.set(slot, filePath);
return { registered: true, slot };
} catch (error) {
debugLogger.debug(`registerSession failed: ${describe(error)}`);
return false;
return { registered: false, slot };
}
}
@@ -479,12 +570,16 @@ export async function patchSessionRecord(
patch: Partial<
Omit<SessionRegistryRecord, 'pid' | 'schemaVersion' | 'procStart' | 'pidNs'>
>,
slot: SessionRecordSlot = SHARED_RECORD_SLOT,
): Promise<boolean> {
try {
// Inside the try: the fallback `getGlobalQwenDir()` resolution reads
// the home directory and can throw, and this function promises never
// to reject.
const filePath = thisProcessRecordPath();
const filePath = thisProcessRecordPath(slot);
// A slot this process never registered has no record to merge into,
// and no path worth guessing — the same no-op a missing file gets.
if (filePath === null) return false;
const existing = await readRecord(filePath);
// Missing, unreadable, newer-schema, or not actually a record for
// this PID, namespace and boot: the path is keyed by PID alone, and
@@ -522,13 +617,16 @@ export async function patchSessionRecord(
}
}
/** Remove this process's record. Safe to call when none was written. */
export async function unregisterSession(): Promise<void> {
/**
* Remove one of this process's records. Safe to call when none was
* written, and for a slot that never registered.
*/
export async function unregisterSession(
slot: SessionRecordSlot = SHARED_RECORD_SLOT,
): Promise<void> {
try {
const filePath = thisProcessRecordPath();
// Consume the capture regardless of the outcome below: after this
// call returns, this process no longer holds a record at the path.
registeredRecordPath = null;
const filePath = thisProcessRecordPath(slot);
if (filePath === null) return;
// The path is keyed by PID alone, and PIDs collide across namespace
// and machine boundaries: the record sitting at our path may belong
// to a live session on the other side of a shared home. Unlink only
@@ -538,12 +636,17 @@ export async function unregisterSession(): Promise<void> {
// and the same holds for a read that failed transiently: the file
// is intact and may be a foreign live record.
const existing = await readRecord(filePath);
if (
existing.status === 'unsupported-version' ||
existing.status === 'read-error'
) {
if (existing.status === 'read-error') {
// The one branch that keeps the capture. The file is intact and
// may still be this session's own, and a minted path cannot be
// derived a second time — forgetting it here would leave a record
// this process can no longer name, patch or remove, advertising a
// session that is gone until the PID itself dies. Every other exit
// below has established that the path is not ours to hold.
return;
}
registeredRecordPaths.delete(slot);
if (existing.status === 'unsupported-version') return;
if (existing.status === 'ok' && !matchesLocalIdentity(existing.record)) {
return;
}
@@ -565,9 +668,13 @@ export async function unregisterSession(): Promise<void> {
* caller never reads back a record that a patch would have refused to
* touch. Never throws.
*/
export async function readOwnSessionRecord(): Promise<SessionRegistryRecord | null> {
export async function readOwnSessionRecord(
slot: SessionRecordSlot = SHARED_RECORD_SLOT,
): Promise<SessionRegistryRecord | null> {
try {
const existing = await readRecord(thisProcessRecordPath());
const filePath = thisProcessRecordPath(slot);
if (filePath === null) return null;
const existing = await readRecord(filePath);
if (existing.status !== 'ok' || !matchesLocalIdentity(existing.record)) {
return null;
}
@@ -627,7 +734,7 @@ export async function listLiveSessions(): Promise<SessionRegistryRecord[]> {
// A record whose filename disagrees with its contents was not
// written by this code (or was renamed by hand). Skip it, and
// never sweep it — we cannot reason about which PID it describes.
if (`${record.pid}.json` !== name) return;
if (pidOfRecordFilename(name) !== record.pid) return;
// A record from another PID namespace describes PIDs that do not
// resolve in ours: kill(pid, 0) reports ESRCH for a process that
+72 -2
View File
@@ -7,6 +7,7 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import type { Config } from '../config/config.js';
import { BackgroundTaskRegistry } from '../agents/background-tasks.js';
import { SHARED_RECORD_SLOT } from '../services/session-registry.js';
const getOwnPeerIdentity = vi.fn();
const listMessageablePeers = vi.fn();
@@ -54,6 +55,7 @@ describe('ListAgentsTool', () => {
tool = new ListAgentsTool({
getBackgroundTaskRegistry: () => registry,
getTeamManager: () => null,
getSessionRegistrySlot: () => SHARED_RECORD_SLOT,
} as unknown as Config);
});
@@ -72,6 +74,22 @@ describe('ListAgentsTool', () => {
);
});
it('reads its own identity from the record this session registered under', async () => {
// A daemon-hosted session owns a minted record; its process holds no
// `<pid>.json` at all. Reading the shared path instead would return
// null, and the listing would be silently empty for exactly the
// sessions this path exists to serve.
const hosted = new ListAgentsTool({
getBackgroundTaskRegistry: () => registry,
getTeamManager: () => null,
getSessionRegistrySlot: () => 'a1b2c3d4',
} as unknown as Config);
await hosted.validateBuildAndExecute({}, new AbortController().signal);
expect(getOwnPeerIdentity).toHaveBeenCalledWith('a1b2c3d4');
});
it('states the Agent Team boundary in the tool description', () => {
expect(tool.description).toContain(
'Named Agent Team teammates are NOT listed here',
@@ -156,6 +174,7 @@ describe('ListAgentsTool — peer sessions', () => {
) {
return new ListAgentsTool({
getBackgroundTaskRegistry: () => registry,
getSessionRegistrySlot: () => SHARED_RECORD_SLOT,
getTeamManager: () =>
teammates.length === 0
? null
@@ -324,7 +343,12 @@ describe('ListAgentsTool — peer sessions', () => {
it('excludes this session from its own listing', async () => {
listMessageablePeers.mockResolvedValue([
peerRow({ name: 'self-00', ipcPath: '/tmp/self.sock' }),
peerRow({
sessionId: 'self',
ref: 'se1f00',
name: 'self-00',
ipcPath: '/tmp/self.sock',
}),
peerRow({ sessionId: 's2', ref: 'bbb222' }),
]);
const parsed = JSON.parse(String((await run()).llmContent));
@@ -332,6 +356,47 @@ describe('ListAgentsTool — peer sessions', () => {
expect(parsed.sessions[0].name).toBe('docs-cd');
});
it("keeps a session that merely shares this one's inbox", async () => {
// A process hosting several sessions binds one inbox for all of them,
// so a sibling's reply address is this session's own. It is still a
// different session with its own id, its own work and its own model —
// hiding it would leave the sessions in one process unable to see
// each other at all.
listMessageablePeers.mockResolvedValue([
peerRow({
sessionId: 'sibling',
ref: 'bbb222',
name: 'app-11',
ipcPath: '/tmp/self.sock',
}),
]);
const parsed = JSON.parse(String((await run()).llmContent));
expect(parsed.sessions).toEqual([
expect.objectContaining({ name: 'app-11', ref: 'bbb222' }),
]);
});
it('excludes this session when its record carries a freshly patched id', async () => {
// A /clear re-id patches this session's record in place; landing
// between the two reads, the stale id filter no longer recognizes
// it. The reply address does not move on a re-id, so the entry is
// re-read before it is advertised as a sibling.
getOwnPeerIdentity
.mockResolvedValueOnce(SELF)
.mockResolvedValue({ ...SELF, sessionId: 's2' });
listMessageablePeers.mockResolvedValue([
peerRow({
sessionId: 's2',
ref: 'bbb222',
name: 'self-00',
ipcPath: '/tmp/self.sock',
}),
]);
// No peers left: the roster falls back to the plain-text empty case.
const result = await run();
expect(result.llmContent).toContain('no other Qwen Code session');
});
it('excludes a differently named twin of this session', async () => {
// `qwen --resume <id>` from another directory: same session id under
// a second process. sendToPeer refuses it as self, so advertising it
@@ -353,7 +418,12 @@ describe('ListAgentsTool — peer sessions', () => {
it("keeps a peer that merely shares this session's name", async () => {
listMessageablePeers.mockResolvedValue([
peerRow({ name: 'self-00', ipcPath: '/tmp/self.sock' }),
peerRow({
sessionId: 'self',
ref: 'se1f00',
name: 'self-00',
ipcPath: '/tmp/self.sock',
}),
peerRow({
sessionId: 's2',
ref: 'bbb222',
+28 -9
View File
@@ -60,18 +60,37 @@ class ListAgentsInvocation extends BaseToolInvocation<
// Peer sessions are only listed once this session has an inbox of its
// own: without one a message could be sent but never answered, and
// advertising a one-way address invites exactly that.
// Every incarnation of this session is dropped, not just its own
// socket: `qwen --resume <id>` in a second pane runs the same id under
// another process, and a twin advertised here would be sent to as a
// peer — sendToPeer excludes by the same rule, so the listing never
// shows an address the send path treats as self.
const self = await getOwnPeerIdentity();
const peers = self
// Every incarnation of this session is dropped: `qwen --resume <id>`
// in a second pane runs the same id under another process, and a twin
// advertised here would be sent to as a peer — sendToPeer excludes by
// the same rule, so the listing never shows an address the send path
// treats as self.
//
// By session id alone. The reply address used to join the test, and
// stopped meaning "the same session" once a process could host
// several: they share one inbox, so filtering on it would hide every
// sibling of this session from the listing.
const slot = this.config.getSessionRegistrySlot();
const self = await getOwnPeerIdentity(slot);
let peers = self
? (await listMessageablePeers()).filter(
(peer) =>
peer.ipcPath !== self.ipcPath && peer.sessionId !== self.sessionId,
(peer) => peer.sessionId !== self.sessionId,
)
: [];
if (self && peers.some((peer) => peer.ipcPath === self.ipcPath)) {
// The id filter above can read this session's record from before
// a re-id patch landed (/clear, or a peer-driven re-assert); the
// address does not move on a re-id, so an entry on this session's
// own inbox is re-read before it is advertised as a sibling.
// sendToPeer applies the same re-read before sending.
const fresh = await getOwnPeerIdentity(slot);
if (fresh) {
peers = peers.filter(
(peer) =>
peer.ipcPath !== self.ipcPath || peer.sessionId !== fresh.sessionId,
);
}
}
// send_message claims some addresses before it looks at peers — the
// broadcast keyword, and with a team active the leader handle, the
@@ -7,6 +7,7 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { SendMessageTool } from './send-message.js';
import { BackgroundTaskRegistry } from '../agents/background-tasks.js';
import { SHARED_RECORD_SLOT } from '../services/session-registry.js';
import { ToolErrorType } from './tool-error.js';
import type { ApprovalMode, Config } from '../config/config.js';
import { runWithTeammateIdentity } from '../agents/team/identity.js';
@@ -47,6 +48,7 @@ function makeTeamConfig(opts?: {
getBackgroundTaskRegistry: () =>
opts?.registry ?? new BackgroundTaskRegistry(),
getApprovalMode: () => opts?.approvalMode ?? DEFAULT_MODE,
getSessionRegistrySlot: () => SHARED_RECORD_SLOT,
} as unknown as Config;
}
@@ -752,10 +754,37 @@ describe('SendMessageTool — peer mode', () => {
target: 'docs-cd',
message: 'check the tests',
approvalMode: DEFAULT_MODE,
slot: SHARED_RECORD_SLOT,
}),
);
});
it('sends from the record this session registered under, not the default', async () => {
// A session hosted by a daemon owns a minted record; the send path
// reads its own identity from that record. Passing the default here
// would make it advertise a `<pid>.json` reply address the host
// never wrote — and exclude the wrong session from the directory.
sendToPeer.mockResolvedValue({
kind: 'sent',
address: 'docs-cd',
peer: { cwd: '/w/docs' },
});
const tool = new SendMessageTool({
getTeamManager: () => null,
getBackgroundTaskRegistry: () => new BackgroundTaskRegistry(),
getApprovalMode: () => DEFAULT_MODE,
getSessionRegistrySlot: () => 'a1b2c3d4',
} as unknown as Config);
await tool
.build({ to: 'docs-cd', message: 'hi' })
.execute(new AbortController().signal);
expect(sendToPeer).toHaveBeenCalledWith(
expect.objectContaining({ slot: 'a1b2c3d4' }),
);
});
it('asserts nothing about its mode when the mode is unreadable', async () => {
sendToPeer.mockResolvedValue({
kind: 'sent',
@@ -768,6 +797,7 @@ describe('SendMessageTool — peer mode', () => {
getApprovalMode: () => {
throw new Error('not yet');
},
getSessionRegistrySlot: () => SHARED_RECORD_SLOT,
} as unknown as Config);
await tool
+3
View File
@@ -135,6 +135,9 @@ class SendMessageInvocation extends BaseToolInvocation<
// Addresses this tool would keep in-process must never be handed
// back to the model as a peer address, bare.
isReserved: (address) => isInProcessRecipient(address, teamFile),
// Which of this process's records is the sender. Matters only for a
// process hosting several sessions; the default covers the rest.
slot: this.config.getSessionRegistrySlot(),
});
switch (outcome.kind) {
@@ -24,6 +24,7 @@ import * as os from 'node:os';
import * as path from 'node:path';
import { TeamCreateTool } from './team-create.js';
import { SHARED_RECORD_SLOT } from '../services/session-registry.js';
import { TeamDeleteTool } from './team-delete.js';
import { SendMessageTool } from './send-message.js';
import { TaskCreateTool } from './task-create.js';
@@ -93,6 +94,7 @@ function makeConfig(): Config {
getSubagentManager: () => null,
getAgentsSettings: () => ({}),
getSessionId: () => 'test-session-id',
getSessionRegistrySlot: () => SHARED_RECORD_SLOT,
setTeamManager: vi.fn((m: TeamManager | null) => {
teamManager = m;
}),
@@ -183,14 +183,15 @@ describe('Config.startNewSession session-registry patch', () => {
// registerSession never throws — it returns false on a failed write.
// Assert the seam itself before relying on anything downstream of
// it, or a write failure surfaces as a confusing patch-path error.
expect(
await registerSession({
sessionId: sessionA,
cwd: tmpDir,
qwenVersion: '0.0.0-test',
}),
).toBe(true);
config.trackSessionRegistration(Promise.resolve(true));
const registration = await registerSession({
sessionId: sessionA,
cwd: tmpDir,
qwenVersion: '0.0.0-test',
});
expect(registration.registered).toBe(true);
// The real outcome rather than a stand-in: it carries the slot the
// record was written under, which is what the patch below has to name.
config.trackSessionRegistration(Promise.resolve(registration));
const [before] = await listLiveSessions();
@@ -221,14 +222,15 @@ describe('Config.startNewSession session-registry patch', () => {
const sessionB = 'bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb';
const config = makeConfig(sessionA);
// No markRuntimeStatusEnabled(): models the failed sidecar write.
expect(
await registerSession({
sessionId: sessionA,
cwd: tmpDir,
qwenVersion: '0.0.0-test',
}),
).toBe(true);
config.trackSessionRegistration(Promise.resolve(true));
const registration = await registerSession({
sessionId: sessionA,
cwd: tmpDir,
qwenVersion: '0.0.0-test',
});
expect(registration.registered).toBe(true);
// The real outcome rather than a stand-in: it carries the slot the
// record was written under, which is what the patch below has to name.
config.trackSessionRegistration(Promise.resolve(registration));
config.startNewSession(sessionB);