From 2de5e72f0ad04bbdbde470ea9eb57907ec80d14d Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Tue, 15 Sep 2026 16:39:22 +0300 Subject: [PATCH] fix(coding-agent): pin the child cwd, harvest silent children, backpressure fan-in (R4, R11, R10, R9) R4 - children defaulted to whatever the host happened to pass, so recursive searches walked multi-TB caches and froze kernels for 21 and 92 minutes (01a09d7f, 01a09bf6, 01a09351). `CreateRlmSubagentRuntimeOptions` now carries an explicit `cwd`, the parent always sets it to its own session directory, and both runtime hosts and the inline path use it instead of reaching into the parent session. Proof: test/suite/subagent-cwd-propagation.test.ts. R11 - 11 consecutive children exited without replying (01a09782) while parents waited on fan-in that never came. A child that finishes with zero `agent_message.send` calls now hands its last assistant text to the parent as `[auto-harvested fallback from ]`, tagged as unreviewed, with an empty child still getting the plain no-reply notice. The reply obligation also moved into the task envelope and the child doctrine, where a child cannot miss it. Proof: test/suite/subagent-reply-harvesting.test.ts. R10 - fan-in hit a hard 20-pending ceiling and threw at the exact moment results arrived (01a09d35). The ceiling is now soft (200) and admission never fails on it: the message buffers in memory, and the daemon yields up to 30s for a saturated recipient to drain before delivering anyway. The throwing `assertAgentMessageQueueCapacity` helper is gone; the two tests that asserted the old fatal cap now assert the resilient behaviour. Proof: test/suite/agent-message-backpressure.test.ts (35 concurrent senders, zero rejections). R9 (host half) - the worker forwarded PATH alone to a session it created, so `cargo`/`nix` were missing under a GUI- or launchd-started daemon (01a09746 exit 127). The worker now forwards the launching shell's environment, which the supervisor already seeded it with. --- ...gent-cwd-reply-and-message-backpressure.md | 3 + .../coding-agent/src/core/agent-messages.ts | 20 ++-- .../src/core/agent-session-runtime.ts | 5 +- .../coding-agent/src/core/agent-session.ts | 48 ++++++---- packages/coding-agent/src/core/messages.ts | 20 +++- .../coding-agent/src/core/prompts/index.ts | 1 + packages/coding-agent/src/core/prompts/rlm.ts | 14 ++- packages/coding-agent/src/core/rlm-runtime.ts | 8 ++ .../src/modes/daemon/daemon-mode.ts | 25 ++++- .../test/agent-session-bus.test.ts | 11 ++- .../suite/agent-message-backpressure.test.ts | 87 +++++++++++++++++ .../suite/subagent-cwd-propagation.test.ts | 76 +++++++++++++++ .../suite/subagent-reply-harvesting.test.ts | 96 +++++++++++++++++++ 13 files changed, 372 insertions(+), 42 deletions(-) create mode 100644 packages/coding-agent/.changes/subagent-cwd-reply-and-message-backpressure.md create mode 100644 packages/coding-agent/test/suite/agent-message-backpressure.test.ts create mode 100644 packages/coding-agent/test/suite/subagent-cwd-propagation.test.ts create mode 100644 packages/coding-agent/test/suite/subagent-reply-harvesting.test.ts diff --git a/packages/coding-agent/.changes/subagent-cwd-reply-and-message-backpressure.md b/packages/coding-agent/.changes/subagent-cwd-reply-and-message-backpressure.md new file mode 100644 index 000000000..67d41d0cb --- /dev/null +++ b/packages/coding-agent/.changes/subagent-cwd-reply-and-message-backpressure.md @@ -0,0 +1,3 @@ +- Fixed a child subagent's working directory to come from the spawning session, with the option carried explicitly to every runtime host instead of each host guessing. +- Fixed agent messaging to stop failing when a recipient already has a long queue: the ceiling is soft, the sender yields while the target drains, and the message is buffered rather than rejected. +- Added automatic harvesting of a child's last assistant text when it finishes without replying, tagged as unreviewed, and stated the reply obligation in the task envelope. diff --git a/packages/coding-agent/src/core/agent-messages.ts b/packages/coding-agent/src/core/agent-messages.ts index 908840d95..6e2260cc5 100644 --- a/packages/coding-agent/src/core/agent-messages.ts +++ b/packages/coding-agent/src/core/agent-messages.ts @@ -16,7 +16,14 @@ export const AGENT_MESSAGE_SOURCE = "agent_message"; export const AGENT_MESSAGE_ID_PREFIX = "agentmsg_"; export const AGENT_MESSAGE_RECEIVED_PREVIEW_LABEL = "Agent message received"; export const DEFAULT_AGENT_MESSAGE_MAX_CHARS = 16_384; -export const DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION = 20; +/** + * Soft ceiling for one recipient's unfinished turns. Past it the sender yields while the + * target drains, and the message is still admitted (buffered in memory) rather than + * failing: hitting this number used to throw at the exact moment fan-in results arrived. + */ +export const DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION = 200; +/** How long a sender yields for a saturated recipient before delivering anyway. */ +export const AGENT_MESSAGE_BACKPRESSURE_WAIT_MS = 30_000; export const DEFAULT_AGENT_MESSAGE_RATE_LIMIT_CAPACITY = 3; export const DEFAULT_AGENT_MESSAGE_RATE_LIMIT_REFILL_MS = 1000; @@ -350,17 +357,6 @@ export function assertDirectAgentMessageTarget(target: string): string { return normalized; } -export function assertAgentMessageQueueCapacity( - unfinishedActionCount: number, - maxPending = DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, -): void { - if (unfinishedActionCount >= maxPending) { - throw new Error( - `Target session has too many pending messages: ${unfinishedActionCount} unfinished, limit is ${maxPending}`, - ); - } -} - /** * Parses the message id out of the pre-bracket-grammar header that persisted * transcripts still contain; current prompts keep the id in details only. diff --git a/packages/coding-agent/src/core/agent-session-runtime.ts b/packages/coding-agent/src/core/agent-session-runtime.ts index 113f6ed92..1bab48ec2 100644 --- a/packages/coding-agent/src/core/agent-session-runtime.ts +++ b/packages/coding-agent/src/core/agent-session-runtime.ts @@ -314,7 +314,10 @@ export class AgentSessionRuntime implements SubagentRuntimeHost { } async createRlmSubagentRuntime(options: CreateRlmSubagentRuntimeOptions): Promise { - const sessionManager = SessionManager.create(options.parentSession.sessionManager.getCwd(), options.sessionDir); + const sessionManager = SessionManager.create( + options.cwd ?? options.parentSession.sessionManager.getCwd(), + options.sessionDir, + ); if (options.parentSession.sessionFile) { sessionManager.newSession({ parentSession: options.parentSession.sessionFile, diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 16dff4e4e..2e802691a 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -49,11 +49,9 @@ import { type AgentSessionMessageListResult, type AgentSessionMessageReceipt, agentFamilyMemberName, - assertAgentMessageQueueCapacity, assertAgentSessionNameAvailable, assertDirectAgentMessageTarget, createAgentMessageHostHandlers, - DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, formatAgentSessionNameUnavailable, isAgentSessionMessage, isAgentSessionMessageId, @@ -213,6 +211,7 @@ import { findExactModelReferenceMatch } from "./model-resolver.js"; import { throwIfPromptAdmissionCancelled } from "./prompt-admission.js"; import { evaluateActiveCacheTtl, extractSessionCacheState, type PromptCacheSnapshot } from "./prompt-cache.js"; import { expandPromptTemplate, type PromptTemplate, parseCommandArgs } from "./prompt-templates.js"; +import { buildRlmChildTaskEnvelope } from "./prompts/index.js"; import { isAgentLifecycleFailure, isFauxProviderQueueExhausted, @@ -5504,7 +5503,7 @@ export class AgentSession { if (message.customType === RLM_CHILD_FAILURE_CUSTOM_TYPE) return true; if (message.customType !== RLM_CHILD_TERMINAL_NOTICE_CUSTOM_TYPE) return false; const details = message.details as RlmChildTerminalNoticeDetails | undefined; - return details?.kind === "completed_without_reply"; + return details?.kind === "completed_without_reply" || details?.kind === "harvested_fallback"; } private _enqueueRlmTerminalNoticeAction(message: CustomMessage): void { @@ -6498,16 +6497,9 @@ export class AgentSession { "Cannot admit a session action while session input admission is paused.", ); } - if ( - options.restore !== true && - action.payload.kind === "turn" && - isAgentSessionMessage(primaryDeliveryRecord(action).message) - ) { - assertAgentMessageQueueCapacity( - this._actionStore.unfinishedActions().length, - DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, - ); - } + // A message from another agent is admitted even when this session already carries a + // long queue: the ceiling is enforced by sender-side backpressure, and refusing here + // failed exactly when fan-in results were arriving. const coalescedOwner = options.restore ? undefined : this._coalescedFollowUpOwner(action); if (coalescedOwner) { if (action.agentMessageId !== coalescedOwner.agentMessageId) { @@ -10840,6 +10832,9 @@ export class AgentSession { rlmParentNodeId: options.id, spawnedByRequestId: options.spawnedByRequestId, remoteHost: options.remoteHost, + // The parent session's own directory, not the process cwd: a child must start + // where its parent works, never in whatever directory the daemon was launched. + cwd: this.sessionManager.getCwd(), }; } @@ -10852,7 +10847,7 @@ export class AgentSession { } private _createInlineRlmSubagentRuntime(options: CreateRlmSubagentRuntimeOptions): RlmSubagentRuntime { - const childSessionManager = SessionManager.create(this._cwd, options.sessionDir); + const childSessionManager = SessionManager.create(options.cwd ?? this._cwd, options.sessionDir); if (options.parentSession.sessionFile) { childSessionManager.newSession({ parentSession: options.parentSession.sessionFile, @@ -12336,7 +12331,9 @@ export class AgentSession { } }); run.unsubscribe = unsubscribeChildEvents; - const content = `[task from parent]\n\n${prompt}`; + // R11: the reply obligation rides the task envelope, so a child cannot miss it + // the way it can miss a line buried in the system prompt. + const content = buildRlmChildTaskEnvelope(prompt); const spawnMessage: AgentSessionMessage = { role: "custom", customType: AGENT_MESSAGE_CUSTOM_TYPE, @@ -12378,13 +12375,22 @@ export class AgentSession { child._parentReplyCount === parentReplyCountBeforeRun ) { const lastAssistantText = child.getLastAssistantText(); + // R11: a child that finished without replying must still hand over what it + // produced, tagged so the parent can tell a harvest from an answer. An + // empty child still gets the plain no-reply notice. await deliverTerminalMessageToParent( - createRlmChildTerminalNoticeMessage({ - kind: "completed_without_reply", - childId: run.id, - sessionName, - lastAssistantTextPreview: lastAssistantText ? compactRlmText(lastAssistantText) : undefined, - }), + lastAssistantText && lastAssistantText.trim().length > 0 + ? createRlmChildTerminalNoticeMessage({ + kind: "harvested_fallback", + childId: run.id, + sessionName, + text: lastAssistantText, + }) + : createRlmChildTerminalNoticeMessage({ + kind: "completed_without_reply", + childId: run.id, + sessionName, + }), ); } if (!this.registerRlmChildSession(run.id, child) && !run.detachedDeletion) { diff --git a/packages/coding-agent/src/core/messages.ts b/packages/coding-agent/src/core/messages.ts index 8bca84694..a5b55c209 100644 --- a/packages/coding-agent/src/core/messages.ts +++ b/packages/coding-agent/src/core/messages.ts @@ -258,6 +258,18 @@ export type RlmChildTerminalNoticeDetails = childId: string; sessionName: string; lastAssistantTextPreview?: string; + } + | { + /** + * A child that finished without calling `agent_message.send`, so its last + * assistant text is handed to the parent instead of the parent waiting for a + * reply that will never come. The tag keeps the parent from mistaking scratch + * for an answer. + */ + kind: "harvested_fallback"; + childId: string; + sessionName: string; + text: string; }; export interface AsyncBashCompletionDetails { @@ -333,7 +345,13 @@ export function createRlmChildTerminalNoticeMessage( const content = details.kind === "cancelled" ? `[child-exited: cancelled child:${childName}]${details.reason ? `\n\n${details.reason}` : ""}` - : `[child-exited: no-reply child:${childName}]${details.lastAssistantTextPreview ? `\n\nLast assistant text: ${details.lastAssistantTextPreview}` : ""}`; + : details.kind === "harvested_fallback" + ? `[auto-harvested fallback from ${childName}] + +This child finished without replying through agent_message, so its last assistant text is forwarded as-is. Treat it as unreviewed and not as an answer. + +${details.text}` + : `[child-exited: no-reply child:${childName}]${details.lastAssistantTextPreview ? `\n\nLast assistant text: ${details.lastAssistantTextPreview}` : ""}`; return { role: "custom", customType: RLM_CHILD_TERMINAL_NOTICE_CUSTOM_TYPE, diff --git a/packages/coding-agent/src/core/prompts/index.ts b/packages/coding-agent/src/core/prompts/index.ts index ecf670098..8c0e17853 100644 --- a/packages/coding-agent/src/core/prompts/index.ts +++ b/packages/coding-agent/src/core/prompts/index.ts @@ -1,5 +1,6 @@ export { buildChildAgentDoctrine, + buildRlmChildTaskEnvelope, buildRlmPrompt, buildSubagentGuidance, type ChildAgentDoctrineOptions, diff --git a/packages/coding-agent/src/core/prompts/rlm.ts b/packages/coding-agent/src/core/prompts/rlm.ts index 480fdaead..8c152651c 100644 --- a/packages/coding-agent/src/core/prompts/rlm.ts +++ b/packages/coding-agent/src/core/prompts/rlm.ts @@ -62,6 +62,18 @@ export interface ChildAgentDoctrineOptions { activeTools?: string[]; } +/** + * The envelope a parent's task arrives in. The reply obligation rides here rather than only + * in the system prompt, because a child cannot miss the message that carries its task. + */ +export function buildRlmChildTaskEnvelope(prompt: string): string { + return `[task from parent] + +${prompt} + +When this task is done, send its result to your parent with \`await agent_message.send(result, receiver_role="parent")\`. Finishing without a reply forwards your last assistant text to the parent, tagged as unreviewed.`; +} + export function buildChildAgentDoctrine(options: ChildAgentDoctrineOptions): string | undefined { const depth = options.depth ?? 0; const hasIpython = options.activeTools === undefined || options.activeTools.includes("ipython"); @@ -73,7 +85,7 @@ export function buildChildAgentDoctrine(options: ChildAgentDoctrineOptions): str ]; if (hasAgentMessage && hasIpython) { lines.push( - 'When a task calls for an answer, reply explicitly with `await agent_message.send(message, receiver_role="parent")`. Not every message or task needs a reply; continue cleanup after sending and go idle normally.', + 'Reply explicitly with `await agent_message.send(message, receiver_role="parent")` when a task calls for an answer. A task that produced a result is such a task: finishing without a reply does not keep the result private, it forwards your last assistant text to the parent tagged as unreviewed. Not every message needs a reply; continue cleanup after sending and go idle normally.', ); } if (hasIpython) { diff --git a/packages/coding-agent/src/core/rlm-runtime.ts b/packages/coding-agent/src/core/rlm-runtime.ts index 33cd67503..a8bcc44e4 100644 --- a/packages/coding-agent/src/core/rlm-runtime.ts +++ b/packages/coding-agent/src/core/rlm-runtime.ts @@ -483,6 +483,14 @@ export interface CreateRlmSubagentRuntimeOptions { onSessionPublished?: (session: AgentSession) => void; /** Remote host destination if this subagent's kernel should execute on a remote host over SSH. */ remoteHost?: string; + /** + * Working directory the child session, its kernel, and its `bash()` calls start in. + * The parent always sets this to its own session cwd, so a child cannot inherit an + * unrelated directory and walk a multi-TB home. Optional only so an older host + * implementation that reads the parent session directly still type-checks; a host + * that receives it must use it. + */ + cwd?: string; } export interface CreateRlmRootSessionOptions { diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 9a1d3c2c3..c53dc11cb 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -24,6 +24,7 @@ import { } from "../../config.js"; import { AGENT_FAMILY_REACH_ERROR, + AGENT_MESSAGE_BACKPRESSURE_WAIT_MS, AGENT_MESSAGE_SOURCE, type AgentFamilyCatalogEntry, type AgentFamilyMember, @@ -2824,7 +2825,12 @@ export class AgentDaemon { (runtimeConfig?.provider ?? parentState.runtime.session.model?.provider) === options.model.provider ? runtimeConfig?.apiKey : undefined; - const launchEnv = collectDaemonLaunchEnv({ PATH: process.env.PATH }); + // The worker already carries the launching shell's environment (the supervisor + // seeds it with the client's launchEnv). Forwarding PATH alone starved the child + // session of CARGO_HOME/RUSTUP_HOME/NIX_*/SSH_AUTH_SOCK, which is how a swarm + // died with exit 127 `cargo: command not found` under a parent shell that had + // the toolchain. + const launchEnv = collectDaemonLaunchEnv(); const envKey = authSource === "environment" ? findEnvKeys(options.model.provider)?.[0] : undefined; if (envKey && process.env[envKey]) launchEnv[envKey] = process.env[envKey]; if (options.model.provider === "prime-inference" && process.env.PRIME_TEAM_ID !== undefined) { @@ -2892,7 +2898,10 @@ export class AgentDaemon { parentState: ActiveSessionState, options: CreateRlmSubagentRuntimeOptions, ): Promise { - const sessionManager = SessionManager.create(options.parentSession.sessionManager.getCwd(), options.sessionDir); + const sessionManager = SessionManager.create( + options.cwd ?? options.parentSession.sessionManager.getCwd(), + options.sessionDir, + ); sessionManager.newSession({ parentSession: options.parentSession.sessionFile, rlmDepth: options.rlmDepth, @@ -6568,6 +6577,10 @@ export class AgentDaemon { fromRelationship: this.agentMessageRelationship(options.fromState, targetState), target: this.createAgentSessionMessageEndpoint(targetState), }; + // Bounded backpressure: a saturated recipient is a reason to yield, not to fail. The + // sender waits while the target drains, then delivers anyway (the message stays + // buffered in memory), so fan-in cannot be lost at the moment results arrive. + await this.waitForAgentMessageCapacity(targetState); try { const { status } = await this.acceptAgentSessionMessage(targetState, payload); return createAgentSessionMessageReceipt(payload, status); @@ -6577,6 +6590,14 @@ export class AgentDaemon { } } + private async waitForAgentMessageCapacity(targetState: ActiveSessionState): Promise { + const deadline = Date.now() + AGENT_MESSAGE_BACKPRESSURE_WAIT_MS; + while (Date.now() < deadline && !this.shuttingDown) { + if (targetState.runtime.session.unfinishedActionCount < DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION) return; + await new Promise((resolveDelay) => setTimeout(resolveDelay, 250)); + } + } + private async sendRemoteAgentSessionMessage( fromState: ActiveSessionState, targetSelector: string, diff --git a/packages/coding-agent/test/agent-session-bus.test.ts b/packages/coding-agent/test/agent-session-bus.test.ts index fb6a0ff9b..bf3c28586 100644 --- a/packages/coding-agent/test/agent-session-bus.test.ts +++ b/packages/coding-agent/test/agent-session-bus.test.ts @@ -1,14 +1,15 @@ import { describe, expect, it, vi } from "vitest"; import { + AGENT_MESSAGE_BACKPRESSURE_WAIT_MS, AGENT_MESSAGE_SOURCE, AgentSessionMessageRateLimiter, assertAgentFamilyReach, - assertAgentMessageQueueCapacity, assertAgentSessionNameAvailable, assertDirectAgentMessageTarget, createAgentMessageHostHandlers, createAgentSessionMessagePrompt, createAgentSessionMessageReceipt, + DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, normalizeAgentSessionMessage, parseAgentSessionMessagePromptId, selectAgentFamily, @@ -121,12 +122,14 @@ describe("agent session bus", () => { expect(() => normalizeAgentSessionMessage("abcd", 3)).toThrow("Agent session message is too long"); }); - it("rejects broadcast-style targets and full target queues", () => { + it("rejects broadcast-style targets and keeps a soft queue ceiling", () => { expect(assertDirectAgentMessageTarget(" worker ")).toBe("worker"); expect(() => assertDirectAgentMessageTarget("*")).toThrow("Broadcast agent messaging is not supported"); expect(() => assertDirectAgentMessageTarget("all")).toThrow("Broadcast agent messaging is not supported"); - expect(() => assertAgentMessageQueueCapacity(20, 20)).toThrow("Target session has too many pending messages"); - expect(() => assertAgentMessageQueueCapacity(19, 20)).not.toThrow(); + // The ceiling is a soft one: admission never fails on it (sender-side backpressure + // yields instead), so the constant only has to stay above the old hard cap of 20. + expect(DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION).toBeGreaterThanOrEqual(200); + expect(AGENT_MESSAGE_BACKPRESSURE_WAIT_MS).toBeGreaterThan(0); }); it("resolves role sends and scopes all broadcasts to the family roster", async () => { diff --git a/packages/coding-agent/test/suite/agent-message-backpressure.test.ts b/packages/coding-agent/test/suite/agent-message-backpressure.test.ts new file mode 100644 index 000000000..d57ba3851 --- /dev/null +++ b/packages/coding-agent/test/suite/agent-message-backpressure.test.ts @@ -0,0 +1,87 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import type { AgentSessionMessagePayload } from "../../src/core/agent-messages.js"; +import { + createAgentSessionMessage, + DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION, +} from "../../src/core/agent-messages.js"; +import { createHarness, getAssistantTexts, type Harness } from "./harness.js"; + +/** Well past the old hard ceiling of 20 pending messages per session. */ +const SENDERS = 35; + +function payload(index: number, target: Harness): AgentSessionMessagePayload { + return { + id: `agentmsg_backpressure_${index}`, + source: "agent_message", + message: `result ${index}`, + from: { sessionId: "sender-session", sessionName: `sender-${index}`, activeSessionId: "sender-active" }, + fromRelationship: "child", + target: { + sessionId: target.session.sessionId, + sessionName: target.session.sessionName ?? undefined, + activeSessionId: "target-active", + }, + }; +} + +describe("agent message backpressure", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + }); + + it("admits a burst past the old ceiling without throwing", async () => { + const target = await createHarness({ models: [{ id: "faux-target", reasoning: true }] }); + harnesses.push(target); + // One response per sender so every queued turn can complete. + target.setResponses(Array.from({ length: SENDERS }, (_, index) => fauxAssistantMessage(`ack ${index}`))); + + const results = await Promise.allSettled( + Array.from({ length: SENDERS }, (_, index) => { + const message = createAgentSessionMessage(payload(index, target)); + return target.session.acceptAgentMessagePrompt(message.content as string, { + streamingBehavior: "steer", + queueIfBusy: true, + customMessage: message, + }); + }), + ); + + const rejected = results.filter((result) => result.status === "rejected"); + expect(rejected.map((result) => (result as PromiseRejectedResult).reason?.message)).toEqual([]); + expect(SENDERS).toBeGreaterThan(20); + expect(DEFAULT_AGENT_MESSAGE_MAX_PENDING_PER_SESSION).toBeGreaterThanOrEqual(200); + + // Every message reached the target: none was dropped by the ceiling. + await expect + .poll( + () => + target.session.messages.filter( + (message) => message.role === "custom" && message.customType === "agent_message", + ).length, + ) + .toBe(SENDERS); + expect(getAssistantTexts(target).length).toBeGreaterThan(0); + }); + + it("keeps the queue visible and drained for a sequential sender", async () => { + const target = await createHarness({ models: [{ id: "faux-target", reasoning: true }] }); + harnesses.push(target); + target.setResponses(Array.from({ length: 3 }, (_, index) => fauxAssistantMessage(`ack ${index}`))); + + for (let index = 0; index < 3; index++) { + const message = createAgentSessionMessage(payload(index, target)); + await target.session.acceptAgentMessagePrompt(message.content as string, { + streamingBehavior: "steer", + queueIfBusy: true, + customMessage: message, + }); + } + + await expect.poll(() => target.session.unfinishedActionCount).toBe(0); + }); +}); diff --git a/packages/coding-agent/test/suite/subagent-cwd-propagation.test.ts b/packages/coding-agent/test/suite/subagent-cwd-propagation.test.ts new file mode 100644 index 000000000..5bf58adad --- /dev/null +++ b/packages/coding-agent/test/suite/subagent-cwd-propagation.test.ts @@ -0,0 +1,76 @@ +import { type FauxModelDefinition, fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import type { CreateRlmSubagentRuntimeOptions } from "../../src/core/rlm-runtime.js"; +import { createHarness, type Harness } from "./harness.js"; + +const PARENT_MODELS: FauxModelDefinition[] = [{ id: "faux-parent", reasoning: true }]; + +describe("subagent cwd propagation", () => { + let parent: Harness | undefined; + let child: Harness | undefined; + + afterEach(() => { + child?.cleanup(); + parent?.cleanup(); + child = undefined; + parent = undefined; + }); + + it("hands the child runtime the parent's working directory", async () => { + const seen: CreateRlmSubagentRuntimeOptions[] = []; + child = await createHarness(); + child.setResponses([fauxAssistantMessage("child answer")]); + parent = await createHarness({ + models: PARENT_MODELS, + rlmDepth: 0, + rlmMaxDepth: 1, + subagentRuntimeHost: { + createRlmSubagentRuntime: async (options) => { + seen.push(options); + return { session: child!.session }; + }, + deleteRlmSubagentRuntime: async () => {}, + }, + }); + parent.setResponses([fauxAssistantMessage("parent read it")]); + parent.session.hasCalledRlmModels = true; + + await parent.session.runRlmChild("work", { + model: `${parent.session.model!.provider}/${parent.session.model!.id}`, + name: "cwd-worker", + }); + + const parentCwd = parent.session.sessionManager.getCwd(); + expect(parentCwd).toBeTruthy(); + expect(seen).toHaveLength(1); + // The host must not have to ask the parent for a directory: the option carries it. + expect(seen[0]!.cwd).toBe(parentCwd as string); + }); + + it("keeps the child cwd stable across sibling spawns", async () => { + const seen: Array = []; + child = await createHarness(); + child.setResponses([fauxAssistantMessage("child answer")]); + parent = await createHarness({ + models: PARENT_MODELS, + rlmDepth: 0, + rlmMaxDepth: 1, + subagentRuntimeHost: { + createRlmSubagentRuntime: async (options) => { + seen.push(options.cwd); + return { session: child!.session }; + }, + deleteRlmSubagentRuntime: async () => {}, + }, + }); + parent.setResponses([fauxAssistantMessage("parent read them")]); + parent.session.hasCalledRlmModels = true; + const selector = `${parent.session.model!.provider}/${parent.session.model!.id}`; + + await parent.session.runRlmChild("first", { model: selector, name: "cwd-a" }); + await parent.session.runRlmChild("second", { model: selector, name: "cwd-b" }); + + expect(new Set(seen).size).toBe(1); + expect(seen[0]).toBe(parent.session.sessionManager.getCwd()); + }); +}); diff --git a/packages/coding-agent/test/suite/subagent-reply-harvesting.test.ts b/packages/coding-agent/test/suite/subagent-reply-harvesting.test.ts new file mode 100644 index 000000000..89ad2b215 --- /dev/null +++ b/packages/coding-agent/test/suite/subagent-reply-harvesting.test.ts @@ -0,0 +1,96 @@ +import { type FauxModelDefinition, fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import type { CustomMessage } from "../../src/core/messages.js"; +import { buildChildAgentDoctrine, buildRlmChildTaskEnvelope } from "../../src/core/prompts/index.js"; +import { createHarness, type Harness } from "./harness.js"; + +const PARENT_MODELS: FauxModelDefinition[] = [{ id: "faux-parent", reasoning: true }]; + +function notices(harness: Harness, kind?: string): CustomMessage[] { + return harness.session.messages.filter((message): message is CustomMessage => { + if (message.role !== "custom" || message.customType !== "rlm_child_terminal_notice") return false; + const details = message.details as { kind?: string } | undefined; + return kind === undefined || details?.kind === kind; + }); +} + +describe("subagent reply harvesting", () => { + let parent: Harness | undefined; + let child: Harness | undefined; + + afterEach(() => { + child?.cleanup(); + parent?.cleanup(); + child = undefined; + parent = undefined; + }); + + it("forwards a silent child's last assistant text, tagged as unreviewed", async () => { + child = await createHarness(); + child.setResponses([fauxAssistantMessage("harvested: here is the result")]); + parent = await createHarness({ + models: PARENT_MODELS, + rlmDepth: 0, + rlmMaxDepth: 1, + subagentRuntimeHost: { + createRlmSubagentRuntime: async () => ({ session: child!.session }), + deleteRlmSubagentRuntime: async () => {}, + }, + }); + parent.setResponses([fauxAssistantMessage("parent read the harvest")]); + parent.session.hasCalledRlmModels = true; + + await parent.session.runRlmChild("produce a result", { + model: `${parent.session.model!.provider}/${parent.session.model!.id}`, + name: "silent-worker", + }); + + await expect.poll(() => notices(parent!, "harvested_fallback").length).toBe(1); + const harvested = notices(parent!, "harvested_fallback")[0]!; + expect(harvested.content).toContain("[auto-harvested fallback from silent-worker]"); + expect(harvested.content).toContain("harvested: here is the result"); + expect(harvested.content).toContain("not as an answer"); + // The plain no-reply notice is replaced by the harvest, never sent alongside it. + expect(notices(parent!, "completed_without_reply")).toHaveLength(0); + }); + + it("keeps the plain no-reply notice when the child produced no text", async () => { + child = await createHarness(); + child.setResponses([fauxAssistantMessage("")]); + parent = await createHarness({ + models: PARENT_MODELS, + rlmDepth: 0, + rlmMaxDepth: 1, + subagentRuntimeHost: { + createRlmSubagentRuntime: async () => ({ session: child!.session }), + deleteRlmSubagentRuntime: async () => {}, + }, + }); + parent.setResponses([fauxAssistantMessage("parent read the notice")]); + parent.session.hasCalledRlmModels = true; + + await parent.session.runRlmChild("produce nothing", { + model: `${parent.session.model!.provider}/${parent.session.model!.id}`, + name: "empty-worker", + }); + + await expect.poll(() => notices(parent!, "completed_without_reply").length).toBe(1); + expect(notices(parent!, "harvested_fallback")).toHaveLength(0); + }); + + it("states the reply obligation in the envelope that carries the task", () => { + const envelope = buildRlmChildTaskEnvelope("do the work"); + expect(envelope.startsWith("[task from parent]\n\ndo the work")).toBe(true); + expect(envelope).toContain('receiver_role="parent"'); + expect(envelope).toContain("tagged as unreviewed"); + + const doctrine = buildChildAgentDoctrine({ + depth: 1, + installedSkills: ["agent_message", "ipython"], + }); + expect(doctrine).toContain('receiver_role="parent"'); + expect(doctrine).toContain("tagged as unreviewed"); + // A root agent gets no child doctrine at all. + expect(buildChildAgentDoctrine({ depth: 0, installedSkills: ["agent_message", "ipython"] })).toBeUndefined(); + }); +}); -- 2.51.2