Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions docs/subagent.md
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,12 @@

## 结果回注与 Mailbox

子智能体结果**不通过回调透传**,而是走**会话级 Mailbox**(`session/mailbox.ts`):
子智能体结果**不通过回调透传**,而是走 **「先写父转录、再入队列」**:

- Mailbox 是一个全局单例的易失入站队列,按收件人 `sessionId` 分区;子代理终态进的是**父会话**的队列(嵌套委派下 B 的终态进 `mailbox[A]`)。
- `SubagentRunRegistry.drainRun` 消费完子帧流后,把结果渲染成一条 `subagent_result` 条目 `offer` 给父会话。
- `SubagentRunRegistry.drainRun` 消费完子帧流后,把结果组装成一条带 `uuid` 的 `subagent_result` 事件,**先** `session.appendEvent` 追加到父会话自己的转录,**再** `mailbox.offer` 通知父会话。
- 顺序不可颠倒:先入队后落盘会在「已 absorb 但未落盘」的崩溃窗口里丢结果;先落盘后入队最坏只丢「通知」,事实仍在盘上,换回合重建上下文时会读回。
- Mailbox(`session/mailbox.ts`)因此只是**本回合的唤醒通道**:全局单例、按收件人 `sessionId` 分区、进程退出即清空都不影响结果可达性。嵌套委派下 B 的终态进 `mailbox[A]`。
- 父会话自己的转录位置由 `SpawnOptions.parentChain` 推出(链首位即父,其余是父的祖先),见 `subagent/registry.ts` 的 `parentTranscriptOf`。**子代理之间不逐层嵌套**:转录目录一律由链末位(顶层会话)定位,所以父本身是子智能体时与它的父平级——顶层父落在 `sessions/<parent>.jsonl`,其余落在 `sessions/<顶层会话>/subagents/<parent>.jsonl`。
- 条目内容 `renderResult()` 组装为:

```
Expand All @@ -81,8 +83,10 @@ Payload:

### 渲染与终态来源

- **终态由子会话自己写**:子回合的 `finish` 通过 `turn.transition` 落终态;父侧只负责派发与投递。
- **投递信号**:`drainRun` 回注 mailbox 后调用 `turn.markDelivered(childSessionId)`,用于唤醒 `wait_agent`。
- **终态由子会话自己写**:子回合的 `finish` 通过 `turn.transition` 落终态;父侧只负责派发、落盘与投递。
- **投递信号**:`drainRun` 落盘并 `offer` 之后调用 `turn.markDelivered(childSessionId)`,用于唤醒 `wait_agent`。

`send_message` 的来信走同一条契约:先追加到收件人转录(只追加到已存在的转录,避免为已删除的祖先建幽灵文件),再 `offer` 给对方 mailbox。

---

Expand All @@ -97,7 +101,9 @@ Payload:
3. 调 LLM、执行工具
```

顺序上 steer(用户中途输入)优先于子代理汇报,避免"用户指令"排在"子代理汇报"之后。每条 mailbox 条目都经 `session.recordSubagentResult` 落盘为 `subagent_result` 事件,再 `context.absorb` 进入上下文。首轮(`step === 0`)不 drain mailbox——此时子代理刚派发,尚无结果。
顺序上 steer(用户中途输入)优先于子代理汇报,避免"用户指令"排在"子代理汇报"之后。drain 到的条目**已经在盘上**,父回合只做 `context.absorb`,不再重复写盘。

跨回合才被 drain 到的条目(父回合提前结束、结果等到下一回合才被取走)不会重复注入:`absorb` 按事件 `uuid` 判重,读盘时已进上下文的事件直接被跳过。首轮(`step === 0`)不 drain mailbox——此时子代理刚派发,尚无结果。

---

Expand All @@ -120,7 +126,7 @@ Payload:

- **不要在不需要结果时等待**:子智能体最终输出会自动追加到父会话,`wait_agent` 不用于取文本。
- **只在结果阻塞下一步时等**:且给一个宽松的超时,不要用短超时轮询。
- **等待到"已投递"为止**:委派记录的 wait 会先等终态(`ended`),再等结果进入父会话 mailbox(`delivered`),所以返回 `completed` 时结果已可被父回合 drain 到。
- **等待到"已投递"为止**:委派记录的 wait 会先等终态(`ended`),再等结果写进父转录并进入父会话 mailbox(`delivered`),所以返回 `completed` 时结果已可被父回合 drain 到——即使父回合不再 drain,下一回合重建上下文也能读到。

---

Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "coding-agent",
"version": "0.1.0",
"version": "0.2.0",
"packageManager": "[email protected]",
"type": "module",
"workspaces": [
Expand Down
2 changes: 1 addition & 1 deletion packages/codingcode/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "@codingcode/core",
"version": "0.1.0",
"version": "0.2.0",
"type": "module",
"main": "./src/layer.ts",
"scripts": {
Expand Down
12 changes: 5 additions & 7 deletions packages/codingcode/src/agent/agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -394,13 +394,11 @@ export const AgentLayer = Layer.effect(
let drainedCount = yield* absorbPendingInputs();

if (step > 0) {
for (const item of yield* mailbox.drain(state.sessionId)) {
const ev =
item.type === 'subagent_message'
? yield* session.recordSubagentMessage(state, item)
: yield* session.recordSubagentResult(state, item);
yield* context.absorb(sessionRef, [ev]);
drainedCount++;
// 条目在投递点已写进本会话转录,这里只把它们并进本回合上下文
const items = yield* mailbox.drain(state.sessionId);
if (items.length > 0) {
yield* context.absorb(sessionRef, items);
drainedCount += items.length;
}
}

Expand Down
13 changes: 12 additions & 1 deletion packages/codingcode/src/context/context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -453,12 +453,23 @@ export const ContextLayer = Layer.effect(
);
});

/** 投递点已落盘,跨回合才 drain 到的条目其事件早被重建时的读盘读进来了 */
const isBuffered = (buf: ContextBuffer, ev: SessionEvent): boolean => {
const uuid = (ev as { uuid?: string }).uuid;
return (
typeof uuid === 'string' && buf.events.some((e) => (e as { uuid?: string }).uuid === uuid)
);
};

const absorb = (ref: SessionRef, events: readonly SessionEvent[]): Effect.Effect<void> =>
Effect.sync(() => {
const buf = buffers.get(ref.sessionId);
// 未建(事件已在盘上,重建时会读到)或已换回合 ⇒ no-op
if (!buf || buf.turnId !== ref.currentTurnId) return;
for (const ev of events) if (passesContextFilter(ev)) buf.events.push(ev);
for (const ev of events) {
if (!passesContextFilter(ev) || isBuffered(buf, ev)) continue;
buf.events.push(ev);
}
});

const compact = (
Expand Down
15 changes: 12 additions & 3 deletions packages/codingcode/src/layer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,20 @@ const AgentWithDeps = AgentLayer.pipe(
// subagent runner (depends on agent)
const SubagentWithDeps = SubagentRunnerLayer.pipe(Layer.provide(AgentWithDeps));

// 运行注册表:要 runner 起子代理、要 mailbox 投递终态、要 sink 发 subagent_event 帧、
// 要 hooks 在子代理终态时触发 agent.subagent.complete。
// 运行注册表:要 runner 起子代理、要 session 把终态写进父转录、要 mailbox 投递终态、
// 要 sink 发 subagent_event 帧、要 hooks 在子代理终态时触发 agent.subagent.complete。
// 终态归属已并入状态表(turn 已在 AgentWithDeps 链路里,Layer 按引用去重 ⇒ 全进程一份 records)。
const SubagentRunRegistryWithDeps = SubagentRunRegistryLayer.pipe(
Layer.provide(Layer.mergeAll(SubagentWithDeps, MailboxLayer, EventSinkLayer, HookLayer, TurnWithDeps))
Layer.provide(
Layer.mergeAll(
SubagentWithDeps,
SessionLayer,
MailboxLayer,
EventSinkLayer,
HookLayer,
TurnWithDeps
)
)
);

export const AppLayer = Layer.mergeAll(
Expand Down
6 changes: 5 additions & 1 deletion packages/codingcode/src/llm/providers/shared.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,11 @@ export function convertMessages(messages: ResolvedMessage[]): ModelMessage[] {
} as unknown as ModelMessage;
}

if (m.role === 'user' || m.role === 'system') {
if (m.role === 'system') {
return { role: 'system', content: textOf(m.content) };
}

if (m.role === 'user') {
const parts: any[] = [];
for (const p of m.content) {
if (p.type === 'text') {
Expand Down
5 changes: 3 additions & 2 deletions packages/codingcode/src/session/mailbox.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import { Chunk, Context, Effect, Layer, Queue } from 'effect';
import type { SubagentMessageEvent, SubagentResultEvent } from './types.js';

/** 入站暂存条目:子代理终态结果,或其它智能体主动投递的来信 */
/** 入站暂存条目:子代理终态结果,或其它智能体主动投递的来信;都是**已落盘**的事件 */
export type MailboxItem = SubagentResultEvent | SubagentMessageEvent;

export interface MailboxShape {
Expand All @@ -16,7 +16,8 @@ export interface MailboxShape {
export class MailboxService extends Context.Tag('Mailbox')<MailboxService, MailboxShape>() {}

/**
* 会话的易失入站队列,与 transcript(持久出站)对称:同一把键 sessionId。
* 会话的入站通知队列:条目在 offer 之前已写进收件人转录,队列只负责「本回合叫醒收件人」。
* 因此队列易失不丢事实 —— 进程退出或收件人回合提前结束,重建上下文仍会从盘上读到。
* 全局单例,内部按收件人分区 —— 嵌套委派下 B 的终态进 mailbox[A]。
*/
export const MailboxLayer = Layer.scoped(
Expand Down
10 changes: 0 additions & 10 deletions packages/codingcode/src/session/port.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,6 @@ import type {
AssistantEvent,
RollbackEvent,
SessionEvent,
SubagentMessageEvent,
SubagentResultEvent,
SummaryEvent,
ToolResultEvent,
UserEvent,
Expand Down Expand Up @@ -73,14 +71,6 @@ export interface SessionShape {
toolCallId: string,
output: string
): Effect.Effect<ToolResultEvent, AgentError>;
recordSubagentResult(
state: SessionStoreState,
result: { sessionId: string; agentName: string; content: string }
): Effect.Effect<SubagentResultEvent, AgentError>;
recordSubagentMessage(
state: SessionStoreState,
message: { sessionId: string; agentName: string; content: string }
): Effect.Effect<SubagentMessageEvent, AgentError>;
appendSummary(
state: SessionStoreState,
summaryText: string,
Expand Down
49 changes: 0 additions & 49 deletions packages/codingcode/src/session/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,6 @@
UserEvent,
AssistantEvent,
ToolResultEvent,
SubagentResultEvent,
SubagentMessageEvent,
SummaryEvent,
RollbackEvent,
SessionEvent,
Expand Down Expand Up @@ -188,7 +186,7 @@

export const SessionLayer = Layer.effect(
SessionService,
Effect.gen(function* () {

Check warning on line 189 in packages/codingcode/src/session/session.ts

View workflow job for this annotation

GitHub Actions / lint

This generator function does not have 'yield'
const assetCache = new Map<string, string>();

const create = (
Expand Down Expand Up @@ -476,51 +474,6 @@
: new AgentError('SESSION_IO_ERROR', `Session write failed: ${String(e)}`, e),
});

// 由父回合循环在 drain 点调用:终态先入 mailbox,到这里才落盘。
// 不写 state.usage —— 子代理的用量不算进父会话。
const recordSubagentResult = (
state: SessionStoreState,
result: { sessionId: string; agentName: string; content: string }
): Effect.Effect<SubagentResultEvent, AgentError> =>
Effect.try({
try: () => {
const event: SubagentResultEvent = {
type: 'subagent_result',
sessionId: result.sessionId,
agentName: result.agentName,
content: result.content,
};
appendLine(pathsFromState(state).transcriptPath, event);
return event;
},
catch: (e) =>
e instanceof AgentError
? e
: new AgentError('SESSION_IO_ERROR', `Session write failed: ${String(e)}`, e),
});

// 与 recordSubagentResult 同构:另一个智能体主动投递的来信,落盘后由收件人回合吸收
const recordSubagentMessage = (
state: SessionStoreState,
message: { sessionId: string; agentName: string; content: string }
): Effect.Effect<SubagentMessageEvent, AgentError> =>
Effect.try({
try: () => {
const event: SubagentMessageEvent = {
type: 'subagent_message',
sessionId: message.sessionId,
agentName: message.agentName,
content: message.content,
};
appendLine(pathsFromState(state).transcriptPath, event);
return event;
},
catch: (e) =>
e instanceof AgentError
? e
: new AgentError('SESSION_IO_ERROR', `Session write failed: ${String(e)}`, e),
});

const appendSummary = (
state: SessionStoreState,
summaryText: string,
Expand Down Expand Up @@ -643,8 +596,6 @@
recordSystem,
recordAssistant,
recordToolResult,
recordSubagentResult,
recordSubagentMessage,
appendSummary,
rollbackToTurn,

Expand Down
4 changes: 4 additions & 0 deletions packages/codingcode/src/session/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,8 @@ export const SUBAGENT_RESULT_PREFIX = 'Message Type: FINAL_ANSWER';

export interface SubagentResultEvent {
type: 'subagent_result';
/** 事件身份:投递点已落盘,上下文重建后又 drain 到同一条时按它去重 */
uuid: string;
sessionId: string;
agentName: string;
content: string;
Expand All @@ -97,6 +99,8 @@ export const SUBAGENT_MESSAGE_PREFIX = 'Message Type: INTER_AGENT_MESSAGE';

export interface SubagentMessageEvent {
type: 'subagent_message';
/** 同 subagent_result:投递即落盘,重建后按它去重 */
uuid: string;
sessionId: string;
agentName: string;
content: string;
Expand Down
55 changes: 42 additions & 13 deletions packages/codingcode/src/subagent/registry.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,18 @@
import { Context, Effect, Layer } from 'effect';
import { randomUUID } from 'crypto';
import { AgentError } from '../util/error.js';
import { isTurnEnd } from '../sink/types.js';
import type { EndTransition, FrameBody } from '../sink/types.js';
import { BYPASS_PERMISSION_MODE, type ProfileName } from '../util/enums.js';
import { SUBAGENT_RESULT_PREFIX } from '../session/types.js';
import type { SubagentResultEvent } from '../session/types.js';
import { transcriptPathOf } from '../session/paths.js';
import { estimateTokensForContent } from '../context/tokens.js';
import { textPart } from '../llm/types.js';
import { loadConfig } from '../infra/config.js';

import { MailboxService } from '../session/mailbox.js';
import { SessionService } from '../session/port.js';
import { EventSinkService } from '../sink/port.js';
import { HookService } from '../hooks/port.js';
import { TurnRegistryService } from '../turn/port.js';
Expand All @@ -27,11 +31,14 @@ export interface SpawnOptions {
resumeSessionId?: string;
/** 首次派发的用途描述;续跑不传(meta 已有) */
description?: string;
/** 祖先会话 id,由近及远;子会话回合的工具上下文由此而来 */
parentChain?: readonly string[];
/**
* 祖先会话 id,由近及远,首位即 `parentSessionId`;子会话回合的工具上下文由此而来,
* 父会话自己的转录位置也由其余位推出(见 parentTranscriptOf)
*/
parentChain: readonly string[];
}

/** 终态由子会话自己的 finish 写入;本表只负责派发与投递 */
/** 终态由子会话自己的 finish 写入;本表只负责派发、落盘与投递 */
export interface SubagentRunRegistryShape {
spawn(opts: SpawnOptions): Effect.Effect<{ sessionId: string; agentName: string }, AgentError>;
}
Expand All @@ -46,8 +53,24 @@ interface SubagentRun {
readonly parentSessionId: string;
readonly parentCwd: string;
readonly agentName: string;
/** 父会话自己的转录:结果在投递点就写进这里,不等父回合 drain */
readonly parentTranscriptPath: string;
}

/**
* 父会话自己的转录位置。链首位就是父,故父的「父 id」与祖链都是本链去掉首位。
* 转录目录由链末位(顶层会话)定位,子代理之间不逐层嵌套,所以父本身是子智能体时
* 与它的父平级:顶层父(链长 1)⇒ `sessions/<parent>.jsonl`,
* 否则 ⇒ `sessions/<链末位>/subagents/<parent>.jsonl`。
*/
const parentTranscriptOf = (opts: SpawnOptions): string =>
transcriptPathOf(
opts.parentCwd,
opts.parentSessionId,
opts.parentChain[1],
opts.parentChain.slice(1)
);

const BODY_TOKEN_BUDGET = 900;

/** 二分收敛到不超过预算的最长前缀,避免逐字扫描 */
Expand Down Expand Up @@ -123,6 +146,7 @@ export const SubagentRunRegistryLayer = Layer.scoped(
Effect.gen(function* () {
const mailbox = yield* MailboxService;
const runner = yield* SubagentRunnerService;
const session = yield* SessionService;
const sink = yield* EventSinkService;
const hooks = yield* HookService;
const turn = yield* TurnRegistryService;
Expand All @@ -142,7 +166,8 @@ export const SubagentRunRegistryLayer = Layer.scoped(
event: { type: 'subagent_event', sessionId, agentName, status },
});

// 只入队,不写盘:写盘由父回合在自己的 drain 点做(父回合循环手里才有 state)
// 先落盘再入队:盘是事实源,队列只是本回合的唤醒通道。
// 父回合提前结束、进程退出都不会丢结果 —— 换回合重建上下文时从父转录读回。
const drainRun = (run: SubagentRun, stream: AsyncGenerator<FrameBody, unknown, unknown>) =>
Effect.gen(function* () {
const settled = yield* Effect.either(consume(stream));
Expand All @@ -151,17 +176,20 @@ export const SubagentRunRegistryLayer = Layer.scoped(
? settled.right
: { end: failedEnd(settled.left.message), content: settled.left.message };

yield* mailbox
.offer(run.parentSessionId, {
type: 'subagent_result',
sessionId: run.sessionId,
agentName: run.agentName,
content: renderResult(run, outcome),
})
.pipe(Effect.ignore);
const event: SubagentResultEvent = {
type: 'subagent_result',
uuid: randomUUID(),
sessionId: run.sessionId,
agentName: run.agentName,
content: renderResult(run, outcome),
};

// 落盘失败仍投递:退化成本回合可见、崩溃不可恢复,不影响父回合继续
yield* session.appendEvent(run.parentTranscriptPath, event).pipe(Effect.ignore);
yield* mailbox.offer(run.parentSessionId, event).pipe(Effect.ignore);

// 终态由子会话自己的 finish 写入(turn.transition);这里只补投递信号,
// 让 wait 在"结果已进父会话 mailbox"之后才返回
// 让 wait 在"结果已落盘且已进父会话 mailbox"之后才返回
yield* turn.markDelivered(run.sessionId);
yield* emitSubagent(
run.parentSessionId,
Expand Down Expand Up @@ -206,6 +234,7 @@ export const SubagentRunRegistryLayer = Layer.scoped(
parentSessionId: opts.parentSessionId,
parentCwd: opts.parentCwd,
agentName: opts.agentName,
parentTranscriptPath: parentTranscriptOf(opts),
};
parents.add(opts.parentSessionId);
yield* emitSubagent(opts.parentSessionId, sessionId, opts.agentName, 'spawned');
Expand Down
Loading
Loading