fix(cron): normalize skill-filter snapshots and split isolated run helpers

This commit is contained in:
Peter Steinberger
2026-02-16 04:26:30 +01:00
parent 6754a926ee
commit aef1d55300
10 changed files with 360 additions and 191 deletions

View File

@@ -7,6 +7,7 @@ import {
parseAgentSessionKey, parseAgentSessionKey,
} from "../routing/session-key.js"; } from "../routing/session-key.js";
import { resolveUserPath } from "../utils.js"; import { resolveUserPath } from "../utils.js";
import { normalizeSkillFilter } from "./skills/filter.js";
import { resolveDefaultAgentWorkspaceDir } from "./workspace.js"; import { resolveDefaultAgentWorkspaceDir } from "./workspace.js";
export { resolveAgentIdFromSessionKey } from "../routing/session-key.js"; export { resolveAgentIdFromSessionKey } from "../routing/session-key.js";
@@ -128,12 +129,7 @@ export function resolveAgentSkillsFilter(
cfg: OpenClawConfig, cfg: OpenClawConfig,
agentId: string, agentId: string,
): string[] | undefined { ): string[] | undefined {
const raw = resolveAgentConfig(cfg, agentId)?.skills; return normalizeSkillFilter(resolveAgentConfig(cfg, agentId)?.skills);
if (!raw) {
return undefined;
}
const normalized = raw.map((entry) => String(entry).trim()).filter(Boolean);
return normalized.length > 0 ? normalized : [];
} }
export function resolveAgentModelPrimary(cfg: OpenClawConfig, agentId: string): string | undefined { export function resolveAgentModelPrimary(cfg: OpenClawConfig, agentId: string): string | undefined {

View File

@@ -0,0 +1,35 @@
import { describe, expect, it } from "vitest";
import {
matchesSkillFilter,
normalizeSkillFilter,
normalizeSkillFilterForComparison,
} from "./filter.js";
describe("skills/filter", () => {
it("normalizes configured filters with trimming", () => {
expect(normalizeSkillFilter([" weather ", "", "meme-factory"])).toEqual([
"weather",
"meme-factory",
]);
});
it("preserves explicit empty list as []", () => {
expect(normalizeSkillFilter([])).toEqual([]);
expect(normalizeSkillFilter(undefined)).toBeUndefined();
});
it("normalizes for comparison with dedupe + ordering", () => {
expect(normalizeSkillFilterForComparison(["weather", "meme-factory", "weather"])).toEqual([
"meme-factory",
"weather",
]);
});
it("matches equivalent filters after normalization", () => {
expect(matchesSkillFilter(["weather", "meme-factory"], [" meme-factory ", "weather"])).toBe(
true,
);
expect(matchesSkillFilter(undefined, undefined)).toBe(true);
expect(matchesSkillFilter([], undefined)).toBe(false);
});
});

View File

@@ -0,0 +1,31 @@
export function normalizeSkillFilter(skillFilter?: ReadonlyArray<unknown>): string[] | undefined {
if (skillFilter === undefined) {
return undefined;
}
return skillFilter.map((entry) => String(entry).trim()).filter(Boolean);
}
export function normalizeSkillFilterForComparison(
skillFilter?: ReadonlyArray<unknown>,
): string[] | undefined {
const normalized = normalizeSkillFilter(skillFilter);
if (normalized === undefined) {
return undefined;
}
return Array.from(new Set(normalized)).toSorted();
}
export function matchesSkillFilter(
cached?: ReadonlyArray<unknown>,
next?: ReadonlyArray<unknown>,
): boolean {
const cachedNormalized = normalizeSkillFilterForComparison(cached);
const nextNormalized = normalizeSkillFilterForComparison(next);
if (cachedNormalized === undefined || nextNormalized === undefined) {
return cachedNormalized === nextNormalized;
}
if (cachedNormalized.length !== nextNormalized.length) {
return false;
}
return cachedNormalized.every((entry, index) => entry === nextNormalized[index]);
}

View File

@@ -82,6 +82,8 @@ export type SkillEligibilityContext = {
export type SkillSnapshot = { export type SkillSnapshot = {
prompt: string; prompt: string;
skills: Array<{ name: string; primaryEnv?: string }>; skills: Array<{ name: string; primaryEnv?: string }>;
/** Normalized agent-level filter used to build this snapshot; undefined means unrestricted. */
skillFilter?: string[];
resolvedSkills?: Skill[]; resolvedSkills?: Skill[];
version?: number; version?: number;
}; };

View File

@@ -19,6 +19,7 @@ import { CONFIG_DIR, resolveUserPath } from "../../utils.js";
import { resolveSandboxPath } from "../sandbox-paths.js"; import { resolveSandboxPath } from "../sandbox-paths.js";
import { resolveBundledSkillsDir } from "./bundled-dir.js"; import { resolveBundledSkillsDir } from "./bundled-dir.js";
import { shouldIncludeSkill } from "./config.js"; import { shouldIncludeSkill } from "./config.js";
import { normalizeSkillFilter } from "./filter.js";
import { import {
parseFrontmatter, parseFrontmatter,
resolveOpenClawMetadata, resolveOpenClawMetadata,
@@ -52,14 +53,16 @@ function filterSkillEntries(
let filtered = entries.filter((entry) => shouldIncludeSkill({ entry, config, eligibility })); let filtered = entries.filter((entry) => shouldIncludeSkill({ entry, config, eligibility }));
// If skillFilter is provided, only include skills in the filter list. // If skillFilter is provided, only include skills in the filter list.
if (skillFilter !== undefined) { if (skillFilter !== undefined) {
const normalized = skillFilter.map((entry) => String(entry).trim()).filter(Boolean); const normalized = normalizeSkillFilter(skillFilter) ?? [];
const label = normalized.length > 0 ? normalized.join(", ") : "(none)"; const label = normalized.length > 0 ? normalized.join(", ") : "(none)";
console.log(`[skills] Applying skill filter: ${label}`); skillsLogger.debug(`Applying skill filter: ${label}`);
filtered = filtered =
normalized.length > 0 normalized.length > 0
? filtered.filter((entry) => normalized.includes(entry.skill.name)) ? filtered.filter((entry) => normalized.includes(entry.skill.name))
: []; : [];
console.log(`[skills] After filter: ${filtered.map((entry) => entry.skill.name).join(", ")}`); skillsLogger.debug(
`After skill filter: ${filtered.map((entry) => entry.skill.name).join(", ") || "(none)"}`,
);
} }
return filtered; return filtered;
} }
@@ -232,12 +235,14 @@ export function buildWorkspaceSkillSnapshot(
const resolvedSkills = promptEntries.map((entry) => entry.skill); const resolvedSkills = promptEntries.map((entry) => entry.skill);
const remoteNote = opts?.eligibility?.remote?.note?.trim(); const remoteNote = opts?.eligibility?.remote?.note?.trim();
const prompt = [remoteNote, formatSkillsForPrompt(resolvedSkills)].filter(Boolean).join("\n"); const prompt = [remoteNote, formatSkillsForPrompt(resolvedSkills)].filter(Boolean).join("\n");
const skillFilter = normalizeSkillFilter(opts?.skillFilter);
return { return {
prompt, prompt,
skills: eligible.map((entry) => ({ skills: eligible.map((entry) => ({
name: entry.skill.name, name: entry.skill.name,
primaryEnv: entry.metadata?.primaryEnv, primaryEnv: entry.metadata?.primaryEnv,
})), })),
...(skillFilter === undefined ? {} : { skillFilter }),
resolvedSkills, resolvedSkills,
version: opts?.snapshotVersion, version: opts?.snapshotVersion,
}; };

View File

@@ -144,6 +144,8 @@ export type GroupKeyResolution = {
export type SessionSkillSnapshot = { export type SessionSkillSnapshot = {
prompt: string; prompt: string;
skills: Array<{ name: string; primaryEnv?: string }>; skills: Array<{ name: string; primaryEnv?: string }>;
/** Normalized agent-level filter used to build this snapshot; undefined means unrestricted. */
skillFilter?: string[];
resolvedSkills?: Skill[]; resolvedSkills?: Skill[];
version?: number; version?: number;
}; };

View File

@@ -194,7 +194,6 @@ function makeParams(overrides?: Record<string, unknown>) {
describe("runCronIsolatedAgentTurn — skill filter", () => { describe("runCronIsolatedAgentTurn — skill filter", () => {
let previousFastTestEnv: string | undefined; let previousFastTestEnv: string | undefined;
beforeEach(() => { beforeEach(() => {
vi.clearAllMocks(); vi.clearAllMocks();
previousFastTestEnv = process.env.OPENCLAW_TEST_FAST; previousFastTestEnv = process.env.OPENCLAW_TEST_FAST;
@@ -283,4 +282,72 @@ describe("runCronIsolatedAgentTurn — skill filter", () => {
// Explicit empty skills list should forward [] to filter out all skills // Explicit empty skills list should forward [] to filter out all skills
expect(buildWorkspaceSkillSnapshotMock.mock.calls[0][1]).toHaveProperty("skillFilter", []); expect(buildWorkspaceSkillSnapshotMock.mock.calls[0][1]).toHaveProperty("skillFilter", []);
}); });
it("refreshes cached snapshot when skillFilter changes without version bump", async () => {
resolveAgentSkillsFilterMock.mockReturnValue(["weather"]);
resolveCronSessionMock.mockReturnValue({
storePath: "/tmp/store.json",
store: {},
sessionEntry: {
sessionId: "test-session-id",
updatedAt: 0,
systemSent: false,
skillsSnapshot: {
prompt: "<available_skills><skill>meme-factory</skill></available_skills>",
skills: [{ name: "meme-factory" }],
version: 42,
},
},
systemSent: false,
isNewSession: true,
});
const { runCronIsolatedAgentTurn } = await import("./run.js");
const result = await runCronIsolatedAgentTurn(
makeParams({
cfg: { agents: { list: [{ id: "weather-bot", skills: ["weather"] }] } },
agentId: "weather-bot",
}),
);
expect(result.status).toBe("ok");
expect(buildWorkspaceSkillSnapshotMock).toHaveBeenCalledOnce();
expect(buildWorkspaceSkillSnapshotMock.mock.calls[0][1]).toHaveProperty("skillFilter", [
"weather",
]);
});
it("reuses cached snapshot when version and normalized skillFilter are unchanged", async () => {
resolveAgentSkillsFilterMock.mockReturnValue([" weather ", "meme-factory", "weather"]);
resolveCronSessionMock.mockReturnValue({
storePath: "/tmp/store.json",
store: {},
sessionEntry: {
sessionId: "test-session-id",
updatedAt: 0,
systemSent: false,
skillsSnapshot: {
prompt: "<available_skills><skill>weather</skill></available_skills>",
skills: [{ name: "weather" }],
skillFilter: ["meme-factory", "weather"],
version: 42,
},
},
systemSent: false,
isNewSession: true,
});
const { runCronIsolatedAgentTurn } = await import("./run.js");
const result = await runCronIsolatedAgentTurn(
makeParams({
cfg: { agents: { list: [{ id: "weather-bot", skills: ["weather", "meme-factory"] }] } },
agentId: "weather-bot",
}),
);
expect(result.status).toBe("ok");
expect(buildWorkspaceSkillSnapshotMock).not.toHaveBeenCalled();
});
}); });

View File

@@ -6,7 +6,6 @@ import {
resolveAgentConfig, resolveAgentConfig,
resolveAgentDir, resolveAgentDir,
resolveAgentModelFallbacksOverride, resolveAgentModelFallbacksOverride,
resolveAgentSkillsFilter,
resolveAgentWorkspaceDir, resolveAgentWorkspaceDir,
resolveDefaultAgentId, resolveDefaultAgentId,
} from "../../agents/agent-scope.js"; } from "../../agents/agent-scope.js";
@@ -26,15 +25,9 @@ import {
resolveThinkingDefault, resolveThinkingDefault,
} from "../../agents/model-selection.js"; } from "../../agents/model-selection.js";
import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js"; import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js";
import { buildWorkspaceSkillSnapshot } from "../../agents/skills.js";
import { getSkillsSnapshotVersion } from "../../agents/skills/refresh.js";
import { runSubagentAnnounceFlow } from "../../agents/subagent-announce.js"; import { runSubagentAnnounceFlow } from "../../agents/subagent-announce.js";
import { import { countActiveDescendantRuns } from "../../agents/subagent-registry.js";
countActiveDescendantRuns,
listDescendantRunsForRequester,
} from "../../agents/subagent-registry.js";
import { resolveAgentTimeoutMs } from "../../agents/timeout.js"; import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
import { readLatestAssistantReply } from "../../agents/tools/agent-step.js";
import { deriveSessionTotalTokens, hasNonzeroUsage } from "../../agents/usage.js"; import { deriveSessionTotalTokens, hasNonzeroUsage } from "../../agents/usage.js";
import { ensureAgentWorkspace } from "../../agents/workspace.js"; import { ensureAgentWorkspace } from "../../agents/workspace.js";
import { import {
@@ -52,7 +45,6 @@ import {
import { registerAgentRunContext } from "../../infra/agent-events.js"; import { registerAgentRunContext } from "../../infra/agent-events.js";
import { deliverOutboundPayloads } from "../../infra/outbound/deliver.js"; import { deliverOutboundPayloads } from "../../infra/outbound/deliver.js";
import { resolveAgentOutboundIdentity } from "../../infra/outbound/identity.js"; import { resolveAgentOutboundIdentity } from "../../infra/outbound/identity.js";
import { getRemoteSkillEligibility } from "../../infra/skills-remote.js";
import { logWarn } from "../../logger.js"; import { logWarn } from "../../logger.js";
import { buildAgentMainSessionKey, normalizeAgentId } from "../../routing/session-key.js"; import { buildAgentMainSessionKey, normalizeAgentId } from "../../routing/session-key.js";
import { import {
@@ -72,6 +64,13 @@ import {
resolveHeartbeatAckMaxChars, resolveHeartbeatAckMaxChars,
} from "./helpers.js"; } from "./helpers.js";
import { resolveCronSession } from "./session.js"; import { resolveCronSession } from "./session.js";
import { resolveCronSkillsSnapshot } from "./skills-snapshot.js";
import {
expectsSubagentFollowup,
isLikelyInterimCronMessage,
readDescendantSubagentFallbackReply,
waitForDescendantSubagentSummary,
} from "./subagent-followup.js";
function matchesMessagingToolDeliveryTarget( function matchesMessagingToolDeliveryTarget(
target: MessagingToolSend, target: MessagingToolSend,
@@ -101,152 +100,6 @@ function resolveCronDeliveryBestEffort(job: CronJob): boolean {
return false; return false;
} }
const CRON_SUBAGENT_WAIT_POLL_MS = 500;
const CRON_SUBAGENT_WAIT_MIN_MS = 30_000;
const CRON_SUBAGENT_FINAL_REPLY_GRACE_MS = 5_000;
function isLikelyInterimCronMessage(value: string): boolean {
const text = value.trim();
if (!text) {
return true;
}
const normalized = text.toLowerCase().replace(/\s+/g, " ");
const words = normalized.split(" ").filter(Boolean).length;
const interimHints = [
"on it",
"pulling everything together",
"give me a few",
"give me a few min",
"few minutes",
"let me compile",
"i'll gather",
"i will gather",
"working on it",
"retrying now",
"should be about",
"should have your summary",
"subagent spawned",
"spawned a subagent",
"it'll auto-announce when done",
"it will auto-announce when done",
"auto-announce when done",
"both subagents are running",
"wait for them to report back",
];
return words <= 45 && interimHints.some((hint) => normalized.includes(hint));
}
function expectsSubagentFollowup(value: string): boolean {
const normalized = value.trim().toLowerCase().replace(/\s+/g, " ");
if (!normalized) {
return false;
}
const hints = [
"subagent spawned",
"spawned a subagent",
"auto-announce when done",
"both subagents are running",
"wait for them to report back",
];
return hints.some((hint) => normalized.includes(hint));
}
async function readDescendantSubagentFallbackReply(params: {
sessionKey: string;
runStartedAt: number;
}): Promise<string | undefined> {
const descendants = listDescendantRunsForRequester(params.sessionKey)
.filter(
(entry) =>
typeof entry.endedAt === "number" &&
entry.endedAt >= params.runStartedAt &&
entry.childSessionKey.trim().length > 0,
)
.toSorted((a, b) => (a.endedAt ?? 0) - (b.endedAt ?? 0));
if (descendants.length === 0) {
return undefined;
}
const latestByChild = new Map<string, (typeof descendants)[number]>();
for (const entry of descendants) {
const childKey = entry.childSessionKey.trim();
if (!childKey) {
continue;
}
const current = latestByChild.get(childKey);
if (!current || (entry.endedAt ?? 0) >= (current.endedAt ?? 0)) {
latestByChild.set(childKey, entry);
}
}
const replies: string[] = [];
const latestRuns = [...latestByChild.values()]
.toSorted((a, b) => (a.endedAt ?? 0) - (b.endedAt ?? 0))
.slice(-4);
for (const entry of latestRuns) {
const reply = (await readLatestAssistantReply({ sessionKey: entry.childSessionKey }))?.trim();
if (!reply || reply.toUpperCase() === SILENT_REPLY_TOKEN.toUpperCase()) {
continue;
}
replies.push(reply);
}
if (replies.length === 0) {
return undefined;
}
if (replies.length === 1) {
return replies[0];
}
return replies.join("\n\n");
}
async function waitForDescendantSubagentSummary(params: {
sessionKey: string;
initialReply?: string;
timeoutMs: number;
observedActiveDescendants?: boolean;
}): Promise<string | undefined> {
const initialReply = params.initialReply?.trim();
const deadline = Date.now() + Math.max(CRON_SUBAGENT_WAIT_MIN_MS, Math.floor(params.timeoutMs));
let sawActiveDescendants = params.observedActiveDescendants === true;
let drainedAtMs: number | undefined;
while (Date.now() < deadline) {
const activeDescendants = countActiveDescendantRuns(params.sessionKey);
if (activeDescendants > 0) {
sawActiveDescendants = true;
drainedAtMs = undefined;
await new Promise((resolve) => setTimeout(resolve, CRON_SUBAGENT_WAIT_POLL_MS));
continue;
}
if (!sawActiveDescendants) {
return initialReply;
}
if (!drainedAtMs) {
drainedAtMs = Date.now();
}
const latest = (await readLatestAssistantReply({ sessionKey: params.sessionKey }))?.trim();
if (
latest &&
latest.toUpperCase() !== SILENT_REPLY_TOKEN.toUpperCase() &&
(latest !== initialReply || !isLikelyInterimCronMessage(latest))
) {
return latest;
}
if (Date.now() - drainedAtMs >= CRON_SUBAGENT_FINAL_REPLY_GRACE_MS) {
return undefined;
}
await new Promise((resolve) => setTimeout(resolve, CRON_SUBAGENT_WAIT_POLL_MS));
}
const latest = (await readLatestAssistantReply({ sessionKey: params.sessionKey }))?.trim();
if (
latest &&
latest.toUpperCase() !== SILENT_REPLY_TOKEN.toUpperCase() &&
(latest !== initialReply || !isLikelyInterimCronMessage(latest))
) {
return latest;
}
return undefined;
}
export type RunCronAgentTurnResult = { export type RunCronAgentTurnResult = {
status: "ok" | "error" | "skipped"; status: "ok" | "error" | "skipped";
summary?: string; summary?: string;
@@ -521,32 +374,21 @@ export async function runCronIsolatedAgentTurn(params: {
`${commandBody}\n\nReturn your summary as plain text; it will be delivered automatically. If the task explicitly calls for messaging a specific external recipient, note who/where it should go instead of sending it yourself.`.trim(); `${commandBody}\n\nReturn your summary as plain text; it will be delivered automatically. If the task explicitly calls for messaging a specific external recipient, note who/where it should go instead of sending it yourself.`.trim();
} }
let skillsSnapshot = cronSession.sessionEntry.skillsSnapshot; const existingSkillsSnapshot = cronSession.sessionEntry.skillsSnapshot;
if (isFastTestEnv) { const skillsSnapshot = resolveCronSkillsSnapshot({
// Fast unit-test mode: avoid scanning the workspace and writing session stores. workspaceDir,
skillsSnapshot = skillsSnapshot ?? { prompt: "", skills: [] }; config: cfgWithAgentDefaults,
} else { agentId,
const existingSnapshot = cronSession.sessionEntry.skillsSnapshot; existingSnapshot: existingSkillsSnapshot,
const skillsSnapshotVersion = getSkillsSnapshotVersion(workspaceDir); isFastTestEnv,
const needsSkillsSnapshot = });
!existingSnapshot || existingSnapshot.version !== skillsSnapshotVersion; if (!isFastTestEnv && skillsSnapshot !== existingSkillsSnapshot) {
const skillFilter = resolveAgentSkillsFilter(cfgWithAgentDefaults, agentId); cronSession.sessionEntry = {
if (needsSkillsSnapshot) { ...cronSession.sessionEntry,
skillsSnapshot = buildWorkspaceSkillSnapshot(workspaceDir, { updatedAt: Date.now(),
config: cfgWithAgentDefaults, skillsSnapshot,
skillFilter, };
eligibility: { remote: getRemoteSkillEligibility() }, await persistSessionEntry();
snapshotVersion: skillsSnapshotVersion,
});
if (skillsSnapshot) {
cronSession.sessionEntry = {
...cronSession.sessionEntry,
updatedAt: Date.now(),
skillsSnapshot,
};
await persistSessionEntry();
}
}
} }
// Persist systemSent before the run, mirroring the inbound auto-reply behavior. // Persist systemSent before the run, mirroring the inbound auto-reply behavior.

View File

@@ -0,0 +1,37 @@
import type { OpenClawConfig } from "../../config/config.js";
import { resolveAgentSkillsFilter } from "../../agents/agent-scope.js";
import { buildWorkspaceSkillSnapshot, type SkillSnapshot } from "../../agents/skills.js";
import { matchesSkillFilter } from "../../agents/skills/filter.js";
import { getSkillsSnapshotVersion } from "../../agents/skills/refresh.js";
import { getRemoteSkillEligibility } from "../../infra/skills-remote.js";
export function resolveCronSkillsSnapshot(params: {
workspaceDir: string;
config: OpenClawConfig;
agentId: string;
existingSnapshot?: SkillSnapshot;
isFastTestEnv: boolean;
}): SkillSnapshot {
if (params.isFastTestEnv) {
// Fast unit-test mode skips filesystem scans and snapshot refresh writes.
return params.existingSnapshot ?? { prompt: "", skills: [] };
}
const snapshotVersion = getSkillsSnapshotVersion(params.workspaceDir);
const skillFilter = resolveAgentSkillsFilter(params.config, params.agentId);
const existingSnapshot = params.existingSnapshot;
const shouldRefresh =
!existingSnapshot ||
existingSnapshot.version !== snapshotVersion ||
!matchesSkillFilter(existingSnapshot.skillFilter, skillFilter);
if (!shouldRefresh) {
return existingSnapshot;
}
return buildWorkspaceSkillSnapshot(params.workspaceDir, {
config: params.config,
skillFilter,
eligibility: { remote: getRemoteSkillEligibility() },
snapshotVersion,
});
}

View File

@@ -0,0 +1,152 @@
import {
countActiveDescendantRuns,
listDescendantRunsForRequester,
} from "../../agents/subagent-registry.js";
import { readLatestAssistantReply } from "../../agents/tools/agent-step.js";
import { SILENT_REPLY_TOKEN } from "../../auto-reply/tokens.js";
const CRON_SUBAGENT_WAIT_POLL_MS = 500;
const CRON_SUBAGENT_WAIT_MIN_MS = 30_000;
const CRON_SUBAGENT_FINAL_REPLY_GRACE_MS = 5_000;
export function isLikelyInterimCronMessage(value: string): boolean {
const text = value.trim();
if (!text) {
return true;
}
const normalized = text.toLowerCase().replace(/\s+/g, " ");
const words = normalized.split(" ").filter(Boolean).length;
const interimHints = [
"on it",
"pulling everything together",
"give me a few",
"give me a few min",
"few minutes",
"let me compile",
"i'll gather",
"i will gather",
"working on it",
"retrying now",
"should be about",
"should have your summary",
"subagent spawned",
"spawned a subagent",
"it'll auto-announce when done",
"it will auto-announce when done",
"auto-announce when done",
"both subagents are running",
"wait for them to report back",
];
return words <= 45 && interimHints.some((hint) => normalized.includes(hint));
}
export function expectsSubagentFollowup(value: string): boolean {
const normalized = value.trim().toLowerCase().replace(/\s+/g, " ");
if (!normalized) {
return false;
}
const hints = [
"subagent spawned",
"spawned a subagent",
"auto-announce when done",
"both subagents are running",
"wait for them to report back",
];
return hints.some((hint) => normalized.includes(hint));
}
export async function readDescendantSubagentFallbackReply(params: {
sessionKey: string;
runStartedAt: number;
}): Promise<string | undefined> {
const descendants = listDescendantRunsForRequester(params.sessionKey)
.filter(
(entry) =>
typeof entry.endedAt === "number" &&
entry.endedAt >= params.runStartedAt &&
entry.childSessionKey.trim().length > 0,
)
.toSorted((a, b) => (a.endedAt ?? 0) - (b.endedAt ?? 0));
if (descendants.length === 0) {
return undefined;
}
const latestByChild = new Map<string, (typeof descendants)[number]>();
for (const entry of descendants) {
const childKey = entry.childSessionKey.trim();
if (!childKey) {
continue;
}
const current = latestByChild.get(childKey);
if (!current || (entry.endedAt ?? 0) >= (current.endedAt ?? 0)) {
latestByChild.set(childKey, entry);
}
}
const replies: string[] = [];
const latestRuns = [...latestByChild.values()]
.toSorted((a, b) => (a.endedAt ?? 0) - (b.endedAt ?? 0))
.slice(-4);
for (const entry of latestRuns) {
const reply = (await readLatestAssistantReply({ sessionKey: entry.childSessionKey }))?.trim();
if (!reply || reply.toUpperCase() === SILENT_REPLY_TOKEN.toUpperCase()) {
continue;
}
replies.push(reply);
}
if (replies.length === 0) {
return undefined;
}
if (replies.length === 1) {
return replies[0];
}
return replies.join("\n\n");
}
export async function waitForDescendantSubagentSummary(params: {
sessionKey: string;
initialReply?: string;
timeoutMs: number;
observedActiveDescendants?: boolean;
}): Promise<string | undefined> {
const initialReply = params.initialReply?.trim();
const deadline = Date.now() + Math.max(CRON_SUBAGENT_WAIT_MIN_MS, Math.floor(params.timeoutMs));
let sawActiveDescendants = params.observedActiveDescendants === true;
let drainedAtMs: number | undefined;
while (Date.now() < deadline) {
const activeDescendants = countActiveDescendantRuns(params.sessionKey);
if (activeDescendants > 0) {
sawActiveDescendants = true;
drainedAtMs = undefined;
await new Promise((resolve) => setTimeout(resolve, CRON_SUBAGENT_WAIT_POLL_MS));
continue;
}
if (!sawActiveDescendants) {
return initialReply;
}
if (!drainedAtMs) {
drainedAtMs = Date.now();
}
const latest = (await readLatestAssistantReply({ sessionKey: params.sessionKey }))?.trim();
if (
latest &&
latest.toUpperCase() !== SILENT_REPLY_TOKEN.toUpperCase() &&
(latest !== initialReply || !isLikelyInterimCronMessage(latest))
) {
return latest;
}
if (Date.now() - drainedAtMs >= CRON_SUBAGENT_FINAL_REPLY_GRACE_MS) {
return undefined;
}
await new Promise((resolve) => setTimeout(resolve, CRON_SUBAGENT_WAIT_POLL_MS));
}
const latest = (await readLatestAssistantReply({ sessionKey: params.sessionKey }))?.trim();
if (
latest &&
latest.toUpperCase() !== SILENT_REPLY_TOKEN.toUpperCase() &&
(latest !== initialReply || !isLikelyInterimCronMessage(latest))
) {
return latest;
}
return undefined;
}