Skip to content

Commit c45f1ac

Browse files
steipetesteipete
committed
perf(agents): isolate subagent announce origin helper
1 parent 3ee823b commit c45f1ac

3 files changed

Lines changed: 186 additions & 80 deletions

File tree

‎src/agents/subagent-announce-delivery.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { describe, expect, it } from "vitest";
2-
import { resolveAnnounceOrigin } from "./subagent-announce-delivery.js";
2+
import { resolveAnnounceOrigin } from "./subagent-announce-origin.js";
33

44
describe("resolveAnnounceOrigin telegram forum topics", () => {
55
it("preserves stored forum topic thread ids when requester origin omits one for the same chat", () => {

‎src/agents/subagent-announce-delivery.ts‎

Lines changed: 2 additions & 79 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,8 @@
1-
import { getChannelPlugin } from "../channels/plugins/index.js";
21
import type { ConversationRef } from "../infra/outbound/session-binding-service.js";
32
import { normalizeAccountId, normalizeMainKey } from "../routing/session-key.js";
43
import { defaultRuntime } from "../runtime.js";
54
import { isCronSessionKey } from "../sessions/session-key-utils.js";
6-
import {
7-
type DeliveryContext,
8-
deliveryContextFromSession,
9-
mergeDeliveryContext,
10-
normalizeDeliveryContext,
11-
resolveConversationDeliveryTarget,
12-
} from "../utils/delivery-context.js";
5+
import { resolveConversationDeliveryTarget } from "../utils/delivery-context.js";
136
import {
147
INTERNAL_MESSAGE_CHANNEL,
158
isGatewayMessageChannel,
@@ -37,6 +30,7 @@ import {
3730
runSubagentAnnounceDispatch,
3831
type SubagentAnnounceDeliveryResult,
3932
} from "./subagent-announce-dispatch.js";
33+
import { resolveAnnounceOrigin, type DeliveryContext } from "./subagent-announce-origin.js";
4034
import { type AnnounceQueueItem, enqueueAnnounce } from "./subagent-announce-queue.js";
4135
import { getSubagentDepthFromSessionStore } from "./subagent-depth.js";
4236
import type { SpawnSubagentMode } from "./subagent-spawn.js";
@@ -63,8 +57,6 @@ function resolveDirectAnnounceTransientRetryDelaysMs() {
6357
: ([5_000, 10_000, 20_000] as const);
6458
}
6559

66-
type DeliveryContextSource = Parameters<typeof deliveryContextFromSession>[0];
67-
6860
export function resolveSubagentAnnounceTimeoutMs(cfg: ReturnType<typeof loadConfig>): number {
6961
const configured = cfg.agents?.defaults?.subagents?.announceTimeoutMs;
7062
if (typeof configured !== "number" || !Number.isFinite(configured)) {
@@ -94,50 +86,6 @@ function summarizeDeliveryError(error: unknown): string {
9486
}
9587
}
9688

97-
function normalizeTelegramAnnounceTarget(target: string | undefined): string | undefined {
98-
const trimmed = target?.trim();
99-
if (!trimmed) {
100-
return undefined;
101-
}
102-
if (trimmed.startsWith("group:")) {
103-
return `telegram:${trimmed.slice("group:".length)}`;
104-
}
105-
if (!trimmed.startsWith("telegram:")) {
106-
return undefined;
107-
}
108-
const raw = trimmed.slice("telegram:".length);
109-
const topicMatch = /^(.*):topic:[^:]+$/u.exec(raw);
110-
return `telegram:${topicMatch?.[1] ?? raw}`;
111-
}
112-
113-
function shouldStripThreadFromAnnounceEntry(
114-
normalizedRequester?: DeliveryContext,
115-
normalizedEntry?: DeliveryContext,
116-
): boolean {
117-
if (
118-
!normalizedRequester?.to ||
119-
normalizedRequester.threadId != null ||
120-
normalizedEntry?.threadId == null
121-
) {
122-
return false;
123-
}
124-
const requesterChannel = normalizedRequester.channel?.trim().toLowerCase();
125-
if (requesterChannel === "telegram") {
126-
const requesterTarget = normalizeTelegramAnnounceTarget(normalizedRequester.to);
127-
const entryTarget = normalizeTelegramAnnounceTarget(normalizedEntry?.to);
128-
if (requesterTarget && entryTarget) {
129-
return requesterTarget !== entryTarget;
130-
}
131-
}
132-
const plugin = requesterChannel ? getChannelPlugin(requesterChannel) : undefined;
133-
return Boolean(
134-
plugin?.conversationBindings?.shouldStripThreadFromAnnounceOrigin?.({
135-
requester: normalizedRequester,
136-
entry: normalizedEntry,
137-
}),
138-
);
139-
}
140-
14189
const TRANSIENT_ANNOUNCE_DELIVERY_ERROR_PATTERNS: readonly RegExp[] = [
14290
/\berrorcode=unavailable\b/i,
14391
/\bstatus\s*[:=]\s*"?unavailable\b/i,
@@ -226,31 +174,6 @@ export async function runAnnounceDeliveryWithRetry<T>(params: {
226174
}
227175
}
228176

229-
export function resolveAnnounceOrigin(
230-
entry?: DeliveryContextSource,
231-
requesterOrigin?: DeliveryContext,
232-
): DeliveryContext | undefined {
233-
const normalizedRequester = normalizeDeliveryContext(requesterOrigin);
234-
const normalizedEntry = deliveryContextFromSession(entry);
235-
if (normalizedRequester?.channel && isInternalMessageChannel(normalizedRequester.channel)) {
236-
return mergeDeliveryContext(
237-
{
238-
accountId: normalizedRequester.accountId,
239-
threadId: normalizedRequester.threadId,
240-
},
241-
normalizedEntry,
242-
);
243-
}
244-
const entryForMerge =
245-
normalizedEntry && shouldStripThreadFromAnnounceEntry(normalizedRequester, normalizedEntry)
246-
? (() => {
247-
const { threadId: _ignore, ...rest } = normalizedEntry;
248-
return rest;
249-
})()
250-
: normalizedEntry;
251-
return mergeDeliveryContext(normalizedRequester, entryForMerge);
252-
}
253-
254177
export async function resolveSubagentCompletionOrigin(params: {
255178
childSessionKey: string;
256179
requesterSessionKey: string;

‎src/agents/subagent-announce-origin.ts‎

Lines changed: 183 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,183 @@
1+
import { getChannelPlugin } from "../channels/plugins/index.js";
2+
3+
export type DeliveryContext = {
4+
channel?: string;
5+
to?: string;
6+
accountId?: string;
7+
threadId?: string | number;
8+
};
9+
10+
type DeliveryContextSource = {
11+
channel?: string;
12+
lastChannel?: string;
13+
lastTo?: string;
14+
lastAccountId?: string;
15+
lastThreadId?: string | number;
16+
origin?: {
17+
provider?: string;
18+
accountId?: string;
19+
threadId?: string | number;
20+
};
21+
deliveryContext?: DeliveryContext;
22+
};
23+
24+
function normalizeChannel(raw?: string): string | undefined {
25+
const value = raw?.trim().toLowerCase();
26+
return value || undefined;
27+
}
28+
29+
function normalizeText(raw?: string): string | undefined {
30+
const value = raw?.trim();
31+
return value || undefined;
32+
}
33+
34+
function normalizeThreadId(raw?: string | number): string | number | undefined {
35+
if (typeof raw === "number" && Number.isFinite(raw)) {
36+
return Math.trunc(raw);
37+
}
38+
if (typeof raw === "string") {
39+
const value = raw.trim();
40+
return value || undefined;
41+
}
42+
return undefined;
43+
}
44+
45+
function normalizeDeliveryContext(context?: DeliveryContext): DeliveryContext | undefined {
46+
if (!context) {
47+
return undefined;
48+
}
49+
const normalized: DeliveryContext = {
50+
channel: normalizeChannel(context.channel),
51+
to: normalizeText(context.to),
52+
accountId: normalizeText(context.accountId),
53+
};
54+
const threadId = normalizeThreadId(context.threadId);
55+
if (threadId != null) {
56+
normalized.threadId = threadId;
57+
}
58+
if (
59+
!normalized.channel &&
60+
!normalized.to &&
61+
!normalized.accountId &&
62+
normalized.threadId == null
63+
) {
64+
return undefined;
65+
}
66+
return normalized;
67+
}
68+
69+
function mergeDeliveryContext(
70+
primary?: DeliveryContext,
71+
fallback?: DeliveryContext,
72+
): DeliveryContext | undefined {
73+
const normalizedPrimary = normalizeDeliveryContext(primary);
74+
const normalizedFallback = normalizeDeliveryContext(fallback);
75+
if (!normalizedPrimary && !normalizedFallback) {
76+
return undefined;
77+
}
78+
const channelsConflict =
79+
normalizedPrimary?.channel &&
80+
normalizedFallback?.channel &&
81+
normalizedPrimary.channel !== normalizedFallback.channel;
82+
return normalizeDeliveryContext({
83+
channel: normalizedPrimary?.channel ?? normalizedFallback?.channel,
84+
to: channelsConflict
85+
? normalizedPrimary?.to
86+
: (normalizedPrimary?.to ?? normalizedFallback?.to),
87+
accountId: channelsConflict
88+
? normalizedPrimary?.accountId
89+
: (normalizedPrimary?.accountId ?? normalizedFallback?.accountId),
90+
threadId: channelsConflict
91+
? normalizedPrimary?.threadId
92+
: (normalizedPrimary?.threadId ?? normalizedFallback?.threadId),
93+
});
94+
}
95+
96+
function deliveryContextFromSession(entry?: DeliveryContextSource): DeliveryContext | undefined {
97+
if (!entry) {
98+
return undefined;
99+
}
100+
return normalizeDeliveryContext({
101+
channel:
102+
entry.deliveryContext?.channel ??
103+
entry.lastChannel ??
104+
entry.channel ??
105+
entry.origin?.provider,
106+
to: entry.deliveryContext?.to ?? entry.lastTo,
107+
accountId: entry.deliveryContext?.accountId ?? entry.lastAccountId ?? entry.origin?.accountId,
108+
threadId: entry.deliveryContext?.threadId ?? entry.lastThreadId ?? entry.origin?.threadId,
109+
});
110+
}
111+
112+
function isInternalMessageChannel(raw?: string): boolean {
113+
return normalizeChannel(raw) === "webchat";
114+
}
115+
116+
function normalizeTelegramAnnounceTarget(target: string | undefined): string | undefined {
117+
const trimmed = target?.trim();
118+
if (!trimmed) {
119+
return undefined;
120+
}
121+
if (trimmed.startsWith("group:")) {
122+
return `telegram:${trimmed.slice("group:".length)}`;
123+
}
124+
if (!trimmed.startsWith("telegram:")) {
125+
return undefined;
126+
}
127+
const raw = trimmed.slice("telegram:".length);
128+
const topicMatch = /^(.*):topic:[^:]+$/u.exec(raw);
129+
return `telegram:${topicMatch?.[1] ?? raw}`;
130+
}
131+
132+
function shouldStripThreadFromAnnounceEntry(
133+
normalizedRequester?: DeliveryContext,
134+
normalizedEntry?: DeliveryContext,
135+
): boolean {
136+
if (
137+
!normalizedRequester?.to ||
138+
normalizedRequester.threadId != null ||
139+
normalizedEntry?.threadId == null
140+
) {
141+
return false;
142+
}
143+
const requesterChannel = normalizeChannel(normalizedRequester.channel);
144+
if (requesterChannel === "telegram") {
145+
const requesterTarget = normalizeTelegramAnnounceTarget(normalizedRequester.to);
146+
const entryTarget = normalizeTelegramAnnounceTarget(normalizedEntry?.to);
147+
if (requesterTarget && entryTarget) {
148+
return requesterTarget !== entryTarget;
149+
}
150+
}
151+
const plugin = requesterChannel ? getChannelPlugin(requesterChannel) : undefined;
152+
return Boolean(
153+
plugin?.conversationBindings?.shouldStripThreadFromAnnounceOrigin?.({
154+
requester: normalizedRequester,
155+
entry: normalizedEntry,
156+
}),
157+
);
158+
}
159+
160+
export function resolveAnnounceOrigin(
161+
entry?: DeliveryContextSource,
162+
requesterOrigin?: DeliveryContext,
163+
): DeliveryContext | undefined {
164+
const normalizedRequester = normalizeDeliveryContext(requesterOrigin);
165+
const normalizedEntry = deliveryContextFromSession(entry);
166+
if (normalizedRequester?.channel && isInternalMessageChannel(normalizedRequester.channel)) {
167+
return mergeDeliveryContext(
168+
{
169+
accountId: normalizedRequester.accountId,
170+
threadId: normalizedRequester.threadId,
171+
},
172+
normalizedEntry,
173+
);
174+
}
175+
const entryForMerge =
176+
normalizedEntry && shouldStripThreadFromAnnounceEntry(normalizedRequester, normalizedEntry)
177+
? (() => {
178+
const { threadId: _ignore, ...rest } = normalizedEntry;
179+
return rest;
180+
})()
181+
: normalizedEntry;
182+
return mergeDeliveryContext(normalizedRequester, entryForMerge);
183+
}

0 commit comments

Comments
 (0)