Skip to content

refactor(deliver): one delivery function per host #74

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
leeguooooo merged 1 commit into from
Nov 6, 2026
Merged

refactor(deliver): one delivery function per host #74

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
57 changes: 16 additions & 41 deletions src/cli.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import {
runClaudeChannelServer,
stripSessionEnv,
} from "./claude-channel.ts";
import { trafficStats, writeTraffic, startDeliveryObservation, type TrafficEvent, type TrafficOutcome } from "./traffic.ts";
import { trafficStats, writeTraffic, type TrafficEvent, type TrafficOutcome } from "./traffic.ts";
import { entryCapabilities } from "./capabilities.ts";
import { findAgyWorker, stopAgyWorker } from "./agy.ts";
import { grokSessionIdFromTarget, listGrokSessions } from "./grok.ts";
Expand All @@ -39,10 +39,13 @@ import {
listCodexSessions,
} from "./codex-sessions.ts";
import {
bufferedSink,
collectingSink,
deliverDm,
deliverToClaude,
deliverToCodexTask,
reportTrackedWake,
deliverToHermes,
deliverToPi,
trackDelivery,
type DeliverySink,
type StoredDeliveryFailure,
Expand All @@ -66,14 +69,13 @@ import {
piExtensionPath,
piSkillPath,
} from "./pi-extension.ts";
import { listPiSessions, piSessionIdFromTarget, piTargetName, wakePiSession } from "./pi-sessions.ts";
import { listPiSessions, piSessionIdFromTarget, piTargetName } from "./pi-sessions.ts";
import {
hermesIdentity,
hermesSessionKeyFromTarget,
hermesTargetName,
listHermesSessions,
selfHermesSessionKey,
wakeHermesSession,
} from "./hermes.ts";
import {
createIdleSubscription,
Expand Down Expand Up @@ -134,11 +136,10 @@ import {
parentPidOf,
selectWakeTargets,
splitWakeMentions,
wakeNote,
} from "./wake.ts";
import { runWakeHelper } from "./wake-helper.ts";
import { CODEX_WAKE_WATCH_COMMAND, runCodexWakeWatch, setCodexWakeWatcherCommand } from "./codex-defer.ts";
import { setWakeHelperCommand, WAKE_HELPER_COMMAND, wakeClaudeTracked, type NoticeSender } from "./wake-receipt.ts";
import { setWakeHelperCommand, WAKE_HELPER_COMMAND, type NoticeSender } from "./wake-receipt.ts";
import {
checkUpgrade,
OCS_INSTALL_PS1_URL,
Expand Down Expand Up @@ -694,20 +695,9 @@ async function cmdSend(parsed: Parsed): Promise<void> {
markStoredDeliveryFailure("failed");
continue;
}
const finishDelivery = startDeliveryObservation(channel, message.seq);
const result = await wakePiSession(
resolved.piSession,
wakeNote({ ...wakeInput, receiver: resolved.name, implicitReceiver: true }),
);
finishDelivery(result.ok ? "accepted" : result.reason === "unknown-outcome" ? "unknown" : "failed");
if (result.ok) console.log(M.piWakeAccepted(target));
else if (result.reason === "unknown-outcome") {
console.log(M.piWakeUnknownOutcome(target, result.detail ?? ""));
markStoredDeliveryFailure("unknown");
} else {
console.log(M.piWakeFailed(target, result.reason, result.detail ?? ""));
markStoredDeliveryFailure("failed");
}
const piSession = resolved.piSession;
await trackDelivery(channel, message.seq, CLI_SINK, (sink) =>
deliverToPi(piSession, { receiver: resolved.name, label: target }, wakeInput, M, sink));
}

// Hermes 侧：连宿主 WebSocket，排在对方当前这一轮后面（queued），不打断。
Expand All @@ -717,17 +707,8 @@ async function cmdSend(parsed: Parsed): Promise<void> {
continue;
}
const key = hermesSessionKeyFromTarget(target)!;
const finishDelivery = startDeliveryObservation(channel, message.seq);
const result = await wakeHermesSession(key, wakeNote({ ...wakeInput, receiver: target, implicitReceiver: true }));
finishDelivery(result.ok ? "accepted" : result.reason === "unknown-outcome" ? "unknown" : "failed");
if (result.ok) console.log(result.delivery === "started" ? M.hermesWakeStarted(target) : M.hermesWakeQueued(target));
else if (result.reason === "unknown-outcome") {
console.log(M.hermesWakeUnknownOutcome(target, result.detail ?? ""));
markStoredDeliveryFailure("unknown");
} else {
console.log(M.hermesWakeFailed(target, result.reason, result.detail ?? ""));
markStoredDeliveryFailure("failed");
}
await trackDelivery(channel, message.seq, CLI_SINK, (sink) =>
deliverToHermes(key, { receiver: target, label: target }, wakeInput, M, sink));
}

const openCodeMentions = claudeNames.filter((name) => opencodeSessionIdFromTarget(name) !== null);
Expand Down Expand Up @@ -782,16 +763,10 @@ async function cmdSend(parsed: Parsed): Promise<void> {
// 铁律 4：帧进了收件箱 ≠ 进了对话；回执能把「被扣 / 被拒」说出来，说不出「已读」。
const sender = noticeSender();
const claudeWokeAt = Date.now();
const wakes = await Promise.all(
selection.targets.map((session) => trackDelivery(channel, message.seq, collectingSink(), async (sink) => {
const wake = await wakeClaudeTracked(session, wakeInput, { sender });
reportTrackedWake(`${session.name ?? "?"}(pid ${session.pid})`, wake, M, sink, { followUp: sender !== null });
return wake;
})),
);
selection.targets.forEach((session, index) => {
reportTrackedWake(`${session.name ?? "?"}(pid ${session.pid})`, wakes[index]!, M, CLI_SINK, { followUp: sender !== null });
});
const buffers = selection.targets.map(() => bufferedSink());
await Promise.all(selection.targets.map((session, index) => trackDelivery(channel, message.seq, buffers[index]!, (sink) =>
deliverToClaude(session, wakeInput, M, sink, { receipts: { sender, followUp: sender !== null } }))));
for (const buffer of buffers) buffer.flush(CLI_SINK);
if (idleSubscriber !== null) subscribeIdle(idleSubscriber, selection.targets, claudeWokeAt);
}

Expand Down
165 changes: 119 additions & 46 deletions src/deliver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import { steerCodexTurn } from "./codex-steer.ts";
import { codexHosts, codexThreadLivePid, psTable, queueCodexThread } from "./codex-queue.ts";
import type { messages } from "./i18n.ts";
import { hermesTargetName, wakeHermesSession, type HermesWakeResult } from "./hermes.ts";
import { listPiSessions, wakePiSession } from "./pi-sessions.ts";
import { listPiSessions, wakePiSession, type PiSessionRegistration } from "./pi-sessions.ts";
import {
CODEX_THREAD_ID_ENV,
selfNameOwner,
Expand Down Expand Up @@ -72,6 +72,107 @@ export function collectingSink(): DeliverySink & { lines: string[]; outcome(): "
};
}

/**
* 先收着、按顺序回放：并发投递多个目标时，每个目标的输出仍按目标顺序成段打印。
*/
export function bufferedSink(): DeliverySink & { flush(into: DeliverySink): void } {
const events: Array<(into: DeliverySink) => void> = [];
return {
log: (line) => events.push((into) => into.log(line)),
fail: (outcome) => events.push((into) => into.fail(outcome)),
skipped: () => events.push((into) => into.skipped?.()),
pending: (state) => events.push((into) => into.pending?.(state)),
flush: (into) => { for (const event of events) event(into); },
};
}

// ───────────────────────── 各宿主的投递（载体） ─────────────────────────
// send、dm、ocs 自己的通知、局域网守护进程都经这里：入口只负责把地址解析成目标，
// 投递、措辞、退出码、「结果未知不重投」只写在这一处。Codex 见下方 deliverToCodexTask。

export interface ClaudeDeliveryOptions {
/** 带回执的唤醒（docs/wake-protocol.md §6）；不传 = 无回执路径（ocs 自己的通知必须如此）。 */
receipts?: { sender: NoticeSender | null; followUp: boolean };
/** 输出里怎么称呼目标；缺省 `<name>(pid N)`。 */
label?: string;
}

/** Claude：channel（`ocs claude` 启动的）> 回执 helper > 收件箱 socket，见 wake.ts / wake-receipt.ts。 */
export async function deliverToClaude(
session: NativeClaudeSession,
input: WakeInput,
M: Catalog,
sink: DeliverySink,
options: ClaudeDeliveryOptions = {},
): Promise<void> {
const label = options.label ?? `${session.name ?? "?"}(pid ${session.pid})`;
if (options.receipts !== undefined) {
const wake = await wakeClaudeTracked(session, input, { sender: options.receipts.sender, env: input.env ?? process.env });
reportTrackedWake(label, wake, M, sink, { followUp: options.receipts.followUp });
return;
}
const [outcome] = await wakeSessions([session], input);
reportPlainWake(label, outcome!.result, M, sink);
}

/** Pi：官方扩展 + 私有 socket，忙时排在本轮之后（follow-up）。 */
export async function deliverToPi(
session: PiSessionRegistration,
names: { receiver: string; label: string },
wakeInput: Omit<WakeNoteInput, "receiver">,
M: Catalog,
sink: DeliverySink,
): Promise<void> {
const result = await wakePiSession(session, wakeNote({ ...wakeInput, receiver: names.receiver, implicitReceiver: true }));
if (result.ok) sink.log(M.piWakeAccepted(names.label));
else if (result.reason === "unknown-outcome") {
sink.log(M.piWakeUnknownOutcome(names.label, result.detail ?? ""));
sink.fail("unknown");
} else {
sink.log(M.piWakeFailed(names.label, result.reason, result.detail ?? ""));
sink.fail("failed");
}
}

/** Hermes：宿主 WebSocket，queued:true 排在当前这一轮后面。 */
export async function deliverToHermes(
sessionKey: string,
names: { receiver: string; label: string },
wakeInput: Omit<WakeNoteInput, "receiver">,
M: Catalog,
sink: DeliverySink,
env: NodeJS.ProcessEnv = process.env,
): Promise<void> {
const note = wakeNote({ ...wakeInput, receiver: names.receiver, implicitReceiver: true });
reportHermesWake(names.label, await wakeHermesSession(sessionKey, note, { env }), M, sink);
}

/** OpenCode 2.2.1+：本地插件，原生持久队列 `delivery: "queue"`。 */
export async function deliverToOpenCode(
session: Parameters<typeof wakeOpenCodeSession>[0],
names: { receiver: string; label: string },
wakeInput: Omit<WakeNoteInput, "receiver">,
M: Catalog,
sink: DeliverySink,
): Promise<void> {
const result = await wakeOpenCodeSession(session, wakeNote({ ...wakeInput, receiver: names.receiver, implicitReceiver: true }));
sink.log(M.openCodeWake(names.label, result.ok ? M.openCodeAccepted : result.reason + (result.detail ? ": " + result.detail : "")));
if (!result.ok) sink.fail(result.reason === "unknown-outcome" ? "unknown" : "failed");
}

/** Grok Build：已有 shared leader 的原生 IPC，sendNow:false 排队。 */
export async function deliverToGrok(
sessionId: string,
names: { receiver: string; label: string },
wakeInput: Omit<WakeNoteInput, "receiver">,
M: Catalog,
sink: DeliverySink,
env: NodeJS.ProcessEnv = process.env,
): Promise<void> {
const note = wakeNote({ ...wakeInput, receiver: names.receiver, implicitReceiver: true });
reportGrokWake(names.label, await wakeGrokSession(sessionId, note, { env }), M, sink);
}

/**
* #30：Desktop 明确没有投递时，若同一个 task 仍运行在唯一可验证的 cmux Codex surface，
* 用同一条已落盘消息的 channel/seq 唤醒它。unknown-outcome 绝不能走这里，避免重复投递。
Expand Down Expand Up @@ -378,30 +479,29 @@ export async function deliverNotice(
sink.fail("failed");
return;
}
const [outcome] = await wakeSessions([resolved.session], { ...input, env });
reportPlainWake(target.name, outcome!.result, M, sink);
await deliverToClaude(resolved.session, { ...input, env }, M, sink, { label: target.name });
return;
}
const clean: NodeJS.ProcessEnv = { ...env };
for (const key of SELF_ENV_KEYS) delete clean[key];
if (target.kind === "opencode") {
const session = listOpenCodeSessions(clean).find((s) => s.session_id === target.sessionId);
if (session === undefined) { sink.fail("failed"); return; }
const result = await wakeOpenCodeSession(session, note);
sink.log(M.openCodeWake(target.sessionId, result.ok ? M.openCodeAccepted : result.reason));
if (!result.ok) sink.fail(result.reason === "unknown-outcome" ? "unknown" : "failed");
await deliverToOpenCode(session, { receiver: target.sessionId, label: target.sessionId }, input, M, sink);
return;
}
if (target.kind === "codex") {
await deliverToCodexTask(target.threadId, input, M, sink, undefined, clean);
return;
}
if (target.kind === "hermes") {
reportHermesWake(hermesTargetName(target.sessionKey), await wakeHermesSession(target.sessionKey, note, { env: clean }), M, sink);
const name = hermesTargetName(target.sessionKey);
await deliverToHermes(target.sessionKey, { receiver: name, label: name }, input, M, sink, clean);
return;
}
if (target.kind === "grok") {
reportGrokWake(grokTargetName(target.sessionId), await wakeGrokSession(target.sessionId, note, { env: clean }), M, sink);
const name = grokTargetName(target.sessionId);
await deliverToGrok(target.sessionId, { receiver: name, label: name }, input, M, sink, clean);
return;
}
const session = listPiSessions(clean).find((candidate) => candidate.session_id.toLowerCase() === target.sessionId.toLowerCase());
Expand All @@ -410,12 +510,7 @@ export async function deliverNotice(
sink.fail("failed");
return;
}
const result = await wakePiSession(session, note);
if (result.ok) sink.log(M.piWakeAccepted(session.target));
else {
sink.log(M.piWakeFailed(session.target, result.reason, result.detail ?? ""));
sink.fail(result.reason === "unknown-outcome" ? "unknown" : "failed");
}
await deliverToPi(session, { receiver: session.target, label: session.target }, input, M, sink);
}

export interface DmDeliveryInput {
Expand Down Expand Up @@ -467,10 +562,8 @@ async function deliverDmImpl(
sink.log(M.openCodeParked(resolved.name, channel));
sink.pending?.("parked"); return null;
}
const result = await wakeOpenCodeSession(resolved.openCodeSession,
wakeNote({ ...input.wakeInput, ...anyReply, receiver: resolved.name, implicitReceiver: true }));
sink.log(M.openCodeWake(resolved.name, result.ok ? M.openCodeAccepted : result.reason + (result.detail ? ": " + result.detail : "")));
if (!result.ok) sink.fail(result.reason === "unknown-outcome" ? "unknown" : "failed");
await deliverToOpenCode(resolved.openCodeSession, { receiver: resolved.name, label: resolved.name },
{ ...input.wakeInput, ...anyReply }, M, sink);
return null;
}
if (resolved.kind === "agy") {
Expand Down Expand Up @@ -514,14 +607,8 @@ async function deliverDmImpl(
...(input.dmReplyTarget !== null ? { dmReplyTarget: input.dmReplyTarget } : {}),
env,
};
const label = `${resolved.claude.name ?? "?"}(pid ${resolved.claude.pid})`;
if (input.receipts !== undefined) {
const wake = await wakeClaudeTracked(resolved.claude, claudeInput, { sender: input.receipts.sender, env });
reportTrackedWake(label, wake, M, sink, { followUp: input.receipts.followUp });
return resolved.claude;
}
const [outcome] = await wakeSessions([resolved.claude], claudeInput);
reportPlainWake(label, outcome!.result, M, sink);
await deliverToClaude(resolved.claude, claudeInput, M, sink,
input.receipts === undefined ? {} : { receipts: input.receipts });
return resolved.claude;
}
if (resolved.kind === "codex-task" && resolved.threadId !== undefined) {
Expand All @@ -534,37 +621,23 @@ async function deliverDmImpl(
sink.pending?.("parked");
return null;
}
const result = await wakePiSession(
resolved.piSession,
wakeNote({ ...input.wakeInput, ...anyReply, receiver: resolved.name, implicitReceiver: true }),
);
if (result.ok) sink.log(M.piWakeAccepted(resolved.name));
else if (result.reason === "unknown-outcome") {
sink.log(M.piWakeUnknownOutcome(resolved.name, result.detail ?? ""));
sink.fail("unknown");
} else {
sink.log(M.piWakeFailed(resolved.name, result.reason, result.detail ?? ""));
sink.fail("failed");
}
await deliverToPi(resolved.piSession, { receiver: resolved.name, label: resolved.name },
{ ...input.wakeInput, ...anyReply }, M, sink);
return null;
}
if (resolved.kind === "hermes" && resolved.hermesSessionKey !== undefined) {
// 不在线（宿主没开、会话没打开）→ wakeHermesSession 报 unavailable / not-open：消息已在频道里。
const result = await wakeHermesSession(
resolved.hermesSessionKey,
wakeNote({ ...input.wakeInput, ...anyReply, receiver: resolved.name, implicitReceiver: true }),
{ env },
);
reportHermesWake(resolved.name, result, M, sink);
await deliverToHermes(resolved.hermesSessionKey, { receiver: resolved.name, label: resolved.name },
{ ...input.wakeInput, ...anyReply }, M, sink, env);
return null;
}
if (resolved.kind === "grok" && resolved.grokSessionId !== undefined) {
if (selfGrokSessionId(env) === resolved.grokSessionId || input.wakeInput.from === resolved.name) {
sink.log(M.grokWakeSelfSkipped(resolved.name));
return null;
}
reportGrokWake(resolved.name, await wakeGrokSession(resolved.grokSessionId,
wakeNote({ ...input.wakeInput, ...anyReply, receiver: resolved.name, implicitReceiver: true }), { env }), M, sink);
await deliverToGrok(resolved.grokSessionId, { receiver: resolved.name, label: resolved.name },
{ ...input.wakeInput, ...anyReply }, M, sink, env);
return null;
}
if (resolved.kind === "cmux" && resolved.cmuxRef !== undefined) {
Expand Down
21 changes: 21 additions & 0 deletions test/deliver.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import { describe, expect, test } from "bun:test";
import { bufferedSink, collectingSink } from "../src/deliver.ts";

describe("bufferedSink", () => {
test("replays lines and outcomes in order, so parallel Claude wakes still print per target", () => {
const first = bufferedSink();
const second = bufferedSink();
// 第二个目标先完成
second.log("wake → b");
second.fail("unknown");
first.log("wake → a");
first.pending?.("deferred");
const into = collectingSink();
const pending: string[] = [];
first.flush({ ...into, pending: (state) => pending.push(state) });
second.flush(into);
expect(into.lines).toEqual(["wake → a", "wake → b"]);
expect(into.outcome()).toBe("unknown");
expect(pending).toEqual(["deferred"]);
});
});
Loading