Skip to content

Commit ad7f569

Browse files
authored
🤖 refactor: extract the agent peer-messaging broker from TaskService (#4010)
## Summary Extracts the peer/family agent-messaging protocol's admission invariants out of `TaskService` into a new `AgentPeerMessageBroker`: sliding-window rate limits (per sender-target pair and per target), duplicate suppression, queued-peer-message capacity, consecutive-peer-wake caps, per-pair/per-target session budgets with idempotent refunds, per-target delivery locks, title capping, and peer envelope/trigger composition. `TaskService` keeps tree topology, stop-epoch/overlay orchestration, and dispatch, and delegates every admission decision directly (no pass-through wrappers). ## Background Refactor #4 from the 2026-08-29 architecture review (evidence at main @ f04e0f8): taskService.ts inlined the whole protocol across two regions (~5654-6736, ~9612-10097), and every protocol-invariant test had to spin up the 35-method `WorkspaceHost` mock harness. Protocol invariants now live in one module testable with a fake one-method host and a mock clock. Behavior-preserving by construction: check ordering, refusal codes, reason strings, `retryAfterMs` math, and budget charge arithmetic are byte-identical to the pre-refactor code. ## Net LOC delta | | added | removed | net | | ---------------------------------------------------------------- | ----- | ------- | -------- | | Production (`agentPeerMessageBroker.ts` + `taskService.ts`) | 271 | 286 | **-15** | | Tests (`agentPeerMessageBroker.test.ts` + `taskService.test.ts`) | 238 | 561 | **-323** | | Total | 509 | 847 | **-338** | Harness-heavy tests that only re-asserted pure invariants through the full TaskService stack were deleted and replaced by table-driven broker tests; one thin delegation test per route (peer rate-limit refusal, `task_message_parent` and sibling budget exhaustion) plus every test covering behavior the broker cannot see (refund-on-dispatch-failure, charge-at-admission ordering, stop races, envelope-row persistence, delivery serialization) were kept. The only irreducible addition is the broker's structural boilerplate (imports, class/constructor, the local one-method host interface), which is what makes the invariants independently testable. The broker declares its own minimal host interface (`countQueuedAgentPeerMessages`) instead of importing from `taskWorkspaceSeam.ts`, keeping this diff merge-independent from the parallel WorkspaceHost role-interface split. ## Validation - Remote dogfood UAT (Coder Agents): **PASS** on `ae9bf3e93` — live Xum UI multi-agent tree across descendant/sibling/ancestor relations, an independent 22-assertion edge-case probe (rate/dedupe/wake/budget boundaries at exact ceilings, refund idempotence), and a source-level cross-check of every refusal string and retry computation against f04e0f8. The one commit since (`daa28db4f`) is cosmetic (test-cast type reuse), revalidated locally. - `bun test src/node/services/agentPeerMessageBroker.test.ts`: 12/12. - `bun test src/node/services/taskService.test.ts`: 514 pass; the 2-3 sporadic failures (`terminal recovery ...` x2, `higher-ancestor waiters ...`) reproduce identically on pristine `origin/main` in this environment (config.json atomic-write chown/rename ENOENT under /tmp) and are unrelated to this diff. - `make static-check`: green. ## Risks Low. The moved logic is byte-identical and the delegation points are 1:1 at existing callsites; the riskiest surface (admission/charge ordering under concurrent sends) is covered by kept integration tests and the broker's check-order test (queued-count host query is only reached after in-memory checks pass). --- _Generated with `xum` • Model: `anthropic:claude-fable-5` • Thinking: `xhigh` • Cost: `$37.96`_ <!-- mux-attribution: model=anthropic:claude-fable-5 thinking=xhigh costs=37.96 --> ## Stack Layer 3/10 of the architecture refactor stack (net -5,101 LOC overall). This PR's diff is only this layer, against `mike/arch-git-patch-engine`.
1 parent aedd055 commit ad7f569

4 files changed

Lines changed: 509 additions & 847 deletions

File tree

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
import { describe, expect, mock, test } from "bun:test";
2+
3+
import {
4+
MAX_CONSECUTIVE_PEER_WAKES,
5+
MAX_QUEUED_PEER_MESSAGES_PER_TARGET,
6+
PEER_MESSAGE_DEDUPE_WINDOW_MS,
7+
PEER_MESSAGE_RATE_LIMIT_MAX,
8+
PEER_MESSAGE_RATE_WINDOW_MS,
9+
PEER_MESSAGE_TARGET_RATE_LIMIT_MAX,
10+
} from "@/constants/agentMessaging";
11+
import {
12+
TASK_FAMILY_MESSAGE_MAX_TITLE_CHARS,
13+
TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS,
14+
TASK_FAMILY_MESSAGE_MAX_TOTAL_MESSAGES,
15+
TASK_FAMILY_MESSAGE_TARGET_MAX_TOTAL_CHARS,
16+
TASK_FAMILY_MESSAGE_TARGET_MAX_TOTAL_MESSAGES,
17+
} from "@/constants/taskMessages";
18+
import { AgentPeerMessageBroker } from "@/node/services/agentPeerMessageBroker";
19+
20+
function createHarness(initialNow = 1_000) {
21+
let now = initialNow;
22+
let queuedCount = 0;
23+
const countQueuedAgentPeerMessages = mock(() => queuedCount);
24+
const broker = new AgentPeerMessageBroker({ countQueuedAgentPeerMessages }, () => now);
25+
return {
26+
broker,
27+
countQueuedAgentPeerMessages,
28+
setNow(value: number) {
29+
now = value;
30+
},
31+
setQueuedCount(value: number) {
32+
queuedCount = value;
33+
},
34+
};
35+
}
36+
37+
describe("AgentPeerMessageBroker", () => {
38+
test("enforces pair rate limits with exact retry boundaries", () => {
39+
const harness = createHarness(10_000);
40+
for (let i = 0; i < PEER_MESSAGE_RATE_LIMIT_MAX; i++) {
41+
expect(harness.broker.checkPeerAdmission("sender", "target", `message ${i}`)).toBeNull();
42+
harness.broker.recordPeerSend("sender", "target", `message ${i}`);
43+
}
44+
45+
harness.setNow(10_001);
46+
expect(harness.broker.checkPeerAdmission("sender", "target", "limited")).toEqual({
47+
code: "rate_limited",
48+
retryAfterMs: PEER_MESSAGE_RATE_WINDOW_MS - 1,
49+
});
50+
51+
harness.setNow(10_000 + PEER_MESSAGE_RATE_WINDOW_MS);
52+
expect(harness.broker.checkPeerAdmission("sender", "target", "boundary")).toBeNull();
53+
});
54+
55+
test("enforces the target-wide rate limit", () => {
56+
const { broker } = createHarness();
57+
for (let i = 0; i < PEER_MESSAGE_TARGET_RATE_LIMIT_MAX; i++) {
58+
const sender = `sender-${i}`;
59+
broker.recordPeerSend(sender, "target", `message-${i}`);
60+
}
61+
expect(broker.checkPeerAdmission("another-sender", "target", "next")).toEqual({
62+
code: "rate_limited",
63+
retryAfterMs: PEER_MESSAGE_RATE_WINDOW_MS,
64+
});
65+
});
66+
67+
test("suppresses duplicates until the dedupe entry expires", () => {
68+
const harness = createHarness();
69+
harness.broker.recordPeerSend("sender", "target", "same");
70+
expect(harness.broker.checkPeerAdmission("sender", "target", "same")).toEqual({
71+
code: "refused",
72+
reason: "Duplicate of an identical message recently sent to this target.",
73+
});
74+
75+
harness.setNow(1_000 + PEER_MESSAGE_DEDUPE_WINDOW_MS);
76+
expect(harness.broker.checkPeerAdmission("sender", "target", "same")).toBeNull();
77+
});
78+
79+
test("checks queued capacity after in-memory throttles", () => {
80+
const harness = createHarness();
81+
harness.setQueuedCount(MAX_QUEUED_PEER_MESSAGES_PER_TARGET);
82+
expect(harness.broker.checkPeerAdmission("sender", "target", "message")).toEqual({
83+
code: "refused",
84+
reason: "Target already has the maximum number of queued peer messages.",
85+
});
86+
87+
harness.broker.recordPeerSend("sender", "target", "duplicate");
88+
harness.countQueuedAgentPeerMessages.mockClear();
89+
expect(harness.broker.checkPeerAdmission("sender", "target", "duplicate")).toEqual({
90+
code: "refused",
91+
reason: "Duplicate of an identical message recently sent to this target.",
92+
});
93+
expect(harness.countQueuedAgentPeerMessages).not.toHaveBeenCalled();
94+
});
95+
96+
test("caps consecutive wakes until attention resets the target", () => {
97+
const { broker } = createHarness();
98+
for (let i = 0; i < MAX_CONSECUTIVE_PEER_WAKES; i++) {
99+
broker.chargeConsecutivePeerWake("target");
100+
}
101+
expect(broker.checkPeerAdmission("sender", "target", "message")).toEqual({
102+
code: "refused",
103+
reason: "Target reached its consecutive peer-wake limit and needs user or parent attention.",
104+
});
105+
broker.resetConsecutivePeerWakes("target");
106+
expect(broker.checkPeerAdmission("sender", "target", "message")).toBeNull();
107+
});
108+
109+
test.each([
110+
{
111+
name: "pair message count",
112+
fill: (broker: AgentPeerMessageBroker) => {
113+
for (let i = 0; i < TASK_FAMILY_MESSAGE_MAX_TOTAL_MESSAGES; i++) {
114+
expect(broker.reserveBudget("sender", "target", 1)).not.toBeNull();
115+
}
116+
return broker.reserveBudget("sender", "target", 1);
117+
},
118+
},
119+
{
120+
name: "pair character count",
121+
fill: (broker: AgentPeerMessageBroker) => {
122+
expect(
123+
broker.reserveBudget("sender", "target", TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS)
124+
).not.toBeNull();
125+
return broker.reserveBudget("sender", "target", 1);
126+
},
127+
},
128+
{
129+
name: "target message count",
130+
fill: (broker: AgentPeerMessageBroker) => {
131+
for (let i = 0; i < TASK_FAMILY_MESSAGE_TARGET_MAX_TOTAL_MESSAGES; i++) {
132+
expect(broker.reserveBudget(`sender-${i}`, "target", 1)).not.toBeNull();
133+
}
134+
return broker.reserveBudget("another-sender", "target", 1);
135+
},
136+
},
137+
{
138+
name: "target character count",
139+
fill: (broker: AgentPeerMessageBroker) => {
140+
const senderCount =
141+
TASK_FAMILY_MESSAGE_TARGET_MAX_TOTAL_CHARS / TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS;
142+
for (let i = 0; i < senderCount; i++) {
143+
expect(
144+
broker.reserveBudget(`sender-${i}`, "target", TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS)
145+
).not.toBeNull();
146+
}
147+
return broker.reserveBudget("another-sender", "target", 1);
148+
},
149+
},
150+
])("enforces the $name budget", ({ fill }) => {
151+
expect(fill(createHarness().broker)).toBeNull();
152+
});
153+
154+
test("refunds reservations idempotently", () => {
155+
const { broker } = createHarness();
156+
const refund = broker.reserveBudget("sender", "target", TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS);
157+
expect(refund).not.toBeNull();
158+
refund?.();
159+
refund?.();
160+
expect(
161+
broker.reserveBudget("sender", "target", TASK_FAMILY_MESSAGE_MAX_TOTAL_CHARS)
162+
).not.toBeNull();
163+
});
164+
165+
test("caps titles and composes the peer envelope and trigger", () => {
166+
const { broker } = createHarness();
167+
const title = "T".repeat(TASK_FAMILY_MESSAGE_MAX_TITLE_CHARS + 1);
168+
const prepared = broker.preparePeerMessage({
169+
senderWorkspaceId: "sender",
170+
senderTitle: title,
171+
relation: "target_ancestor",
172+
message: "hello",
173+
});
174+
expect(prepared.fromTitle).toBe(`${title.slice(0, TASK_FAMILY_MESSAGE_MAX_TITLE_CHARS)}…`);
175+
expect(prepared.relationship).toBe("descendant");
176+
expect(prepared.envelope).toContain("hello");
177+
expect(prepared.trigger).toContain(prepared.payloadMessageId);
178+
});
179+
180+
test("charges a queued trigger separator", () => {
181+
const { broker } = createHarness();
182+
expect(broker.triggerCharge("trigger")).toBe("trigger".length + "\n".length);
183+
});
184+
});

0 commit comments

Comments
 (0)