diff --git a/App/memmy-agent/src/core/agent-runtime/loop.ts b/App/memmy-agent/src/core/agent-runtime/loop.ts index 1262a0e72..d007ebf28 100644 --- a/App/memmy-agent/src/core/agent-runtime/loop.ts +++ b/App/memmy-agent/src/core/agent-runtime/loop.ts @@ -1,3 +1,4 @@ +import { OutputLocationPolicy } from "./tools/output-location.js"; import { initializeDesktopScreenCapture } from '../../tools/computer-use/desktop-screen-capture.js'; import { closeNativeSurfacesForTurn, registerComputerUseTurnInterrupt } from '../../tools/computer-use/surface-preview.js'; import fs from "node:fs"; @@ -754,6 +755,9 @@ export class AgentLoop { sessions: SessionManager; cronService: CronService; execSessionManager: ExecSessionManager; + private readonly outputTasks = new Map(); + private readonly committedOutputResponses = new WeakSet(); + private readonly pendingOutputConversions = new Map>(); fileStateStore: FileStateStore; browserSessionManager: BrowserSessionManager; subagents: SubagentManager; @@ -2635,6 +2639,8 @@ export class AgentLoop { } if (waits.length) await Promise.allSettled(waits); + const outputTask = this.outputTasks.get(key); + if (outputTask) { await outputTask.cancel(); this.outputTasks.delete(key); } const subagents = this.subagents as any; const subCancelled = subagents?.cancelBySession ? await subagents.cancelBySession(key) : subagents?.cancel_by_session ? await subagents.cancel_by_session(key) : 0; return cancelled + Number(subCancelled || 0); @@ -2660,6 +2666,8 @@ export class AgentLoop { if (outcome !== "stopped") return outcome; if (!task || task.done() || task.cancel() === false) return "already_finished"; await Promise.resolve(task).catch(() => undefined); + const outputTask = this.outputTasks.get(sessionKey); + if (outputTask) { await outputTask.cancel(); this.outputTasks.delete(sessionKey); } return "stopped"; } @@ -2699,6 +2707,8 @@ export class AgentLoop { return { role: "user", content: this.context.buildUserContent(content, media), + ...(!msg.internal && msg.channel !== "system" && !msg.metadata?.injectedEvent + ? { output_user_instruction: msg.content } : {}), ...(clientRequestId ? { client_request_id: clientRequestId } : {}), @@ -2748,6 +2758,18 @@ export class AgentLoop { return true; } + private async convertPendingOutputInput(msg: InboundMessage): Promise | null> { + const key = this.effectiveSessionKey(msg); + const converting = this.pendingOutputConversions.get(key) ?? new Set(); + converting.add(msg); + this.pendingOutputConversions.set(key, converting); + try { return await this.pendingToUserMessage(msg); } + finally { + converting.delete(msg); + if (!converting.size) this.pendingOutputConversions.delete(key); + } + } + private async waitForPendingMessage(queue: AsyncQueue, timeoutMs: number): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { @@ -2764,7 +2786,7 @@ export class AgentLoop { const injections: Record[] = []; const appendPendingMessage = async (msg: InboundMessage): Promise => { try { - const userMessage = await this.pendingToUserMessage(msg); + const userMessage = await this.convertPendingOutputInput(msg); if (userMessage) { injections.push(userMessage); onUserMessage?.(msg); @@ -3657,6 +3679,63 @@ export class AgentLoop { return result; } + private hasPendingOutputInput(key: string, queue?: AsyncQueue | null): boolean { + return Boolean(queue?.size || this.pendingOutputConversions.get(key)?.size + || [...(this.queueSteerAdmissionsBySession.get(key)?.values() ?? [])] + .some(admission => admission.phase === "pending")); + } + + /** Last barrier: no await between the final input check and first visibility. */ + private async commitOutputResponse( + key: string, response: OutboundMessage | null, stopReason: string, + queue: AsyncQueue | null | undefined, publish: boolean, + session: Session | null, ctx?: TurnContext, abortSignal?: AbortSignal | null, + ): Promise { + const policy = this.outputTasks.get(key); + if (!response || !policy) return response; + if (!policy.hasWork) { + policy.retire(); + this.outputTasks.delete(key); + return response; + } + const pendingGenerators = policy.pendingProducers(); + const invalid = pendingGenerators.length ? [] : await policy.checkDelivery(); + return this.queueMutationLockFor(key).runExclusive(async () => { + if (ctx?.abortSignal?.aborted || abortSignal?.aborted) throw createTaskCancelledError(); + if (this.hasPendingOutputInput(key, queue)) { + if (ctx?.slot) ctx.slot.stopReason = "maxIterations"; + if (session?.messages.at(-1)?.role === "assistant") session.messages.pop(); + if (session) this.sessions.save(session); + return null; + } + if (pendingGenerators.length || invalid.length) { + response.content = pendingGenerators.length ? policy.pendingMessage() : policy.rejection(invalid); + delete response.metadata.streamed; + if (ctx) { ctx.finalContent = response.content; ctx.stopReason = pendingGenerators.length ? "deliveryPending" : "toolError"; } + if (session?.messages.at(-1)?.role === "assistant") session.messages.at(-1)!.content = response.content; + if (session) this.sessions.save(session); + stopReason = pendingGenerators.length ? "deliveryPending" : "toolError"; + } + if (ctx?.slot) { + ctx.slot.acceptingSteer = false; + ctx.slot.state = "settling"; + } + if (ctx?.mirrorTurn) { + this.guiTranscriptMirror?.final(ctx.mirrorTurn, response.content, ctx.turnLatencyMs, null, ctx.errorCategory); + } + if (publish) { + this.bus.outbound.put(response); + this.committedOutputResponses.add(response); + } + if (!pendingGenerators.length && stopReason !== "deliveryPending") { + policy.retire(); + this.outputTasks.delete(key); + if (session) this.sessions.save(session); + } + return response; + }); + } + async runAgentLoop( initialMessages: Record[], { @@ -3683,6 +3762,7 @@ export class AgentLoop { midTurnContext = null, consolidator = null, onTokenCompactionEvent = null, + outputInstruction = null, }: { onProgress?: any; onStream?: any; @@ -3707,6 +3787,7 @@ export class AgentLoop { midTurnContext?: MidTurnContextDescriptor | null; consolidator?: Consolidator | null; onTokenCompactionEvent?: TokenCompactionEventCallback | null; + outputInstruction?: string | null; } = {}, ): Promise { if (!modelSelection) this.refreshProviderSnapshot(); @@ -3740,6 +3821,12 @@ export class AgentLoop { }, }); const hook = this.extraHooks.length ? new CompositeAgentHook([loopHook, ...this.extraHooks]) : loopHook; + let outputLocation = activeSessionKey ? this.outputTasks.get(activeSessionKey) : undefined; + if (!outputLocation?.active || outputLocation.workspace !== path.resolve(sessionWorkspace)) { + outputLocation = new OutputLocationPolicy(sessionWorkspace, session?.metadata); + if (activeSessionKey) this.outputTasks.set(activeSessionKey, outputLocation); + } + if (outputInstruction) outputLocation.addUserRequest(outputInstruction); const result = await this.runner.run( new AgentRunSpec({ messages: initialMessages, @@ -3754,6 +3841,7 @@ export class AgentLoop { maxToolResultChars: this.maxToolResultChars, toolResultMaxCharsByName: SESSION_TOOL_RESULT_MAX_CHARS_BY_NAME, workspace: sessionWorkspace, + outputLocation, sessionKey: activeSessionKey, hostProjectId, contextWindowTokens: activeContextWindowTokens, @@ -3771,6 +3859,7 @@ export class AgentLoop { abortSignal, hook, concurrentTools: true, + hasPendingInput: () => activeSessionKey !== null && this.hasPendingOutputInput(activeSessionKey, pendingQueue), injectionCallback: ({ limit = 3 } = {}) => this.drainPendingQueue(pendingQueue, limit, session?.key ?? sessionKey, msg => { const request = computerUseRequestContext(msg, turnId); // A real user steer replaces the blocked/approved message scope. @@ -4257,6 +4346,7 @@ export class AgentLoop { failedProvider, failedModel, ] = await this.runAgentLoop(ctx.initialMessages, { + outputInstruction: ctx.msg.internal || ctx.msg.channel === "system" || ctx.msg.metadata?.injectedEvent ? null : ctx.msg.content, onProgress: ctx.onProgress, onStream: ctx.onStream, onStreamEnd: ctx.onStreamEnd, @@ -4288,7 +4378,7 @@ export class AgentLoop { }); ctx.stopReason = stopReason; if (ctx.slot) ctx.slot.stopReason = stopReason; - settleSlot(); + if (!this.outputTasks.get(ctx.sessionKey)?.hasWork) settleSlot(); ctx.errorCategory = errorCategory; ctx.failedProvider = failedProvider; ctx.failedModel = failedModel; @@ -4388,7 +4478,7 @@ export class AgentLoop { } else { this.sessions.save(ctx.session!); } - if (ctx.mirrorTurn && ctx.finalContent) { + if (ctx.mirrorTurn && ctx.finalContent && !this.outputTasks.get(ctx.sessionKey)?.hasWork) { this.guiTranscriptMirror?.final( ctx.mirrorTurn, ctx.finalContent, @@ -4456,6 +4546,7 @@ export class AgentLoop { abortSignal, turnId, sessionBindingOverride, + publishOutput = false, }: { onProgress?: (...args: any[]) => Promise | void; onStream?: (delta: string) => Promise | void; @@ -4465,6 +4556,7 @@ export class AgentLoop { turnId?: string | null; boundary?: TurnCancellationBoundary | null; sessionBindingOverride?: WebuiSessionBinding | null; + publishOutput?: boolean; } = {}, ): Promise { turnId ??= firstString(msg.metadata?.turn_id, msg.metadata?.turnId) ?? cryptoRandomId(); @@ -4473,6 +4565,12 @@ export class AgentLoop { const channel = separator >= 0 ? rawChatId.slice(0, separator) : "cli"; const chatId = separator >= 0 ? rawChatId.slice(separator + 1) : rawChatId; const key = sessionKey ?? msg.sessionKeyOverride ?? `${channel}:${chatId}`; + const originalTaskId = msg.metadata?.outputTaskId; + if (originalTaskId && this.outputTasks.get(key)?.taskId !== originalTaskId) { + if (this.outputTasks.has(key)) return null; + return new OutboundMessage({ channel, chatId, + content: "原文件任务的交付记录已结束或不可用，未确认本次后台结果可交付。继续该文件任务前需重新登记并核验最终产物。" }); + } const existingSession = this.sessions.get(key); const hasRequestedPreset = Object.prototype.hasOwnProperty.call( msg.metadata ?? {}, @@ -4646,6 +4744,7 @@ export class AgentLoop { failedProvider, failedModel, ] = await this.runAgentLoop(messages, { + outputInstruction: isSubagent || msg.internal ? null : msg.content, onProgress, onStream, onStreamEnd, @@ -4717,12 +4816,12 @@ export class AgentLoop { failedModel, }); } - return new OutboundMessage({ + return this.commitOutputResponse(key, new OutboundMessage({ channel, chatId, content: finalContent?.trim() || (stopReason === "error" ? EMPTY_FINAL_RESPONSE_MESSAGE : "Background task completed."), metadata, - }); + }), stopReason, pendingQueue, publishOutput, session, undefined, abortSignal); } async processMessageInternal( @@ -4739,6 +4838,7 @@ export class AgentLoop { messageSendCallback, sessionBindingOverride, slot, + publishOutput = false, }: { onProgress?: (...args: any[]) => Promise | void; onStream?: (delta: string) => Promise | void; @@ -4750,6 +4850,7 @@ export class AgentLoop { messageSendCallback?: MessageSendCallback | null; sessionBindingOverride?: WebuiSessionBinding | null; slot?: TurnSlot | null; + publishOutput?: boolean; } = {}, ): Promise { if (!this.modelSelectionResolver) this.refreshProviderSnapshot(); @@ -4761,6 +4862,7 @@ export class AgentLoop { pendingQueue, abortSignal, sessionBindingOverride, + publishOutput, }); } const key = sessionKey ?? this.sessionKey(message); @@ -4816,7 +4918,7 @@ export class AgentLoop { await this.stateRun(ctx); await this.stateSave(ctx); await this.stateRespond(ctx); - return ctx.outbound; + return await this.commitOutputResponse(key, ctx.outbound, ctx.stopReason, ctx.pendingQueue, publishOutput, ctx.session, ctx); } catch (error) { if (isTaskCancelledError(error)) { try { @@ -4918,6 +5020,8 @@ export class AgentLoop { throw new Error("session_deletion_in_progress"); } const deferred: InboundMessage[] = []; + const outputTask = this.outputTasks.get(sessionKey); + if (outputTask) { await outputTask.cancel(); this.outputTasks.delete(sessionKey); } this.goalRuntime.beginSessionDeletion(sessionKey); this.sessionDeletionQueues.set(sessionKey, deferred); try { @@ -5235,6 +5339,7 @@ export class AgentLoop { turnId, boundary, slot, + publishOutput: true, }); if ( typeof response?.metadata?.goalId === "string" @@ -5245,7 +5350,7 @@ export class AgentLoop { goalOutcome: response.metadata.goalOutcome as GoalStatus, }; } - if (!isCancelled() && response) await this.bus.publishOutbound(response); + if (!isCancelled() && response && !this.committedOutputResponses.has(response)) await this.bus.publishOutbound(response); if (!isCancelled() && effectiveMsg.channel === "cli") { await this.bus.publishOutbound( new OutboundMessage({ diff --git a/App/memmy-agent/src/core/agent-runtime/runner.ts b/App/memmy-agent/src/core/agent-runtime/runner.ts index 4f7f63c15..1e383ea4e 100644 --- a/App/memmy-agent/src/core/agent-runtime/runner.ts +++ b/App/memmy-agent/src/core/agent-runtime/runner.ts @@ -49,6 +49,8 @@ import type { TurnCancellationBoundary } from "./turn-cancellation-boundary.js"; import { resolveToolResultMaxChars, type ToolResultMaxCharsByName } from "./tool-result-budget.js"; import { convertMcpToolContent } from "./tools/mcp.js"; +import type { OutputLocationPolicy } from "./tools/output-location.js"; + const DEFAULT_ERROR_MESSAGE = "Sorry, I encountered an error calling the AI model."; const PERSISTED_MODEL_ERROR_PLACEHOLDER = "[Assistant reply unavailable due to model error.]"; export const MAX_EMPTY_RETRIES = 2; @@ -156,6 +158,7 @@ export class AgentRunSpec { errorMessage: string | null; maxIterationsMessage: string | null; maxIterationsFinalPrompt: string | null; + outputLocation?: OutputLocationPolicy; workspace?: string | null; sessionKey?: string | null; hostProjectId?: string | null; @@ -166,6 +169,7 @@ export class AgentRunSpec { streamProgressDeltas: boolean; retryWaitCallback?: any; checkpointCallback?: ((payload: Record) => Promise | void) | null; + hasPendingInput?: (() => boolean) | null; injectionCallback?: ((opts?: { limit?: number }) => Promise | any[]) | null; llmTimeoutS?: number | null; abortSignal?: AbortSignal | null; @@ -195,6 +199,7 @@ export class AgentRunSpec { errorMessage?: string | null; maxIterationsMessage?: string | null; maxIterationsFinalPrompt?: string | null; + outputLocation?: OutputLocationPolicy; workspace?: string | null; sessionKey?: string | null; hostProjectId?: string | null; @@ -205,6 +210,7 @@ export class AgentRunSpec { streamProgressDeltas?: boolean; retryWaitCallback?: any; checkpointCallback?: ((payload: Record) => Promise | void) | null; + hasPendingInput?: (() => boolean) | null; injectionCallback?: ((opts?: { limit?: number }) => Promise | any[]) | null; llmTimeoutS?: number | null; abortSignal?: AbortSignal | null; @@ -233,6 +239,7 @@ export class AgentRunSpec { this.maxIterationsMessage = init.maxIterationsMessage ?? null; this.maxIterationsFinalPrompt = init.maxIterationsFinalPrompt ?? null; this.workspace = init.workspace ?? null; + this.outputLocation = init.outputLocation; this.sessionKey = init.sessionKey ?? null; this.hostProjectId = init.hostProjectId ?? null; this.contextWindowTokens = init.contextWindowTokens ?? null; @@ -243,6 +250,7 @@ export class AgentRunSpec { this.retryWaitCallback = init.retryWaitCallback ?? null; this.checkpointCallback = init.checkpointCallback ?? null; this.injectionCallback = init.injectionCallback ?? null; + this.hasPendingInput = init.hasPendingInput ?? null; this.llmTimeoutS = init.llmTimeoutS ?? null; this.abortSignal = init.abortSignal ?? null; this.turnId = init.turnId ?? null; @@ -359,6 +367,8 @@ export class AgentRunner { messages.push({ role: "user", content, + ...(typeof item.output_user_instruction === "string" + ? { output_user_instruction: item.output_user_instruction } : {}), ...(typeof item.client_request_id === "string" ? { client_request_id: item.client_request_id } : {}), @@ -391,7 +401,7 @@ export class AgentRunner { ): Promise<[boolean, number]> { let injections: Record[] = []; let realInjection = false; - if (injectionCycles < MAX_INJECTION_CYCLES) { + if (injectionCycles < MAX_INJECTION_CYCLES && (opts.iteration == null || opts.iteration + 1 < spec.maxIterations)) { injections = await this.drainInjections(spec); realInjection = injections.length > 0; } @@ -410,6 +420,10 @@ export class AgentRunner { }); } } + for (const injection of injections) { + if (typeof injection.output_user_instruction === "string") spec.outputLocation?.addUserRequest(injection.output_user_instruction); + delete injection.output_user_instruction; + } AgentRunner.appendInjectedMessages(messages, injections); return [true, injectionCycles]; } @@ -425,6 +439,7 @@ export class AgentRunner { delete providerMessage.turn_source; delete providerMessage.webui_queue_steer_origin; delete providerMessage.webui_queue_steer_recovery; + delete providerMessage.output_user_instruction; return providerMessage; }), tools, @@ -1126,6 +1141,7 @@ export class AgentRunner { const fileMutationOutcomes = new Map(); await spec.hook?.beforeToolCall(new AgentHookContext({ spec, toolCalls: [call] }), call); const toolContext: ToolExecutionContext = { + outputLocation: spec.outputLocation, abortSignal: spec.abortSignal ?? null, toolName: call.name, callId: call.id ?? null, @@ -1571,6 +1587,15 @@ export class AgentRunner { return batches; } + private withOutputInstruction(spec: AgentRunSpec, messages: Record[]): Record[] { + if (!spec.outputLocation) return messages; + const instruction = spec.outputLocation.instruction(); + const index = messages.findIndex(message => message.role === "system" && typeof message.content === "string"); + if (index < 0) return [{ role: "system", content: instruction }, ...messages]; + return messages.map((message, position) => position === index + ? { ...message, content: `${message.content}\n\n${instruction}` } : message); + } + async run(spec: AgentRunSpec): Promise { spec.turnId ??= crypto.randomUUID(); spec.provider ??= this.provider; @@ -1590,6 +1615,7 @@ export class AgentRunner { let error: string | null = null; let emptyContentRetries = 0; let lengthRecoveries = 0; + let outputLocationRetries = 0; let injectionCycles = 0; let hadInjections = false; const imageTextState: TurnImageTextState = { @@ -1630,7 +1656,7 @@ export class AgentRunner { break; } } - let messagesForModel = this.modelContextMessages(messages, modelContextProjection); + let messagesForModel = this.withOutputInstruction(spec, this.modelContextMessages(messages, modelContextProjection)); messagesForModel = this.prepareMessagesForModel(spec, messagesForModel, { toolsForRequest, reservedPromptTokens: 0, @@ -1645,6 +1671,7 @@ export class AgentRunner { hook, context, imageTextState, + { forceNonStreaming: Boolean(spec.outputLocation?.hasWork) }, ); const iterationUsage = this.usageDict(response.usage); this.accumulateUsage(usage, iterationUsage); @@ -1756,7 +1783,7 @@ export class AgentRunner { context.error = error; context.stopReason = stopReason; await hook.afterIteration(context); - const [shouldContinue, cycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after tool error" }); + const [shouldContinue, cycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after tool error", iteration }); injectionCycles = cycles; if (shouldContinue) { hadInjections = true; @@ -1774,7 +1801,7 @@ export class AgentRunner { }); emptyContentRetries = 0; lengthRecoveries = 0; - const [drained, cycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after tool execution" }); + const [drained, cycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after tool execution", iteration }); injectionCycles = cycles; if (drained) hadInjections = true; await hook.afterIteration(context); @@ -1831,7 +1858,7 @@ export class AgentRunner { const terminalModelError = AgentRunner.terminalImageError(response) || response.errorCode === "context_length_exceeded"; let shouldContinue = false; if (!terminalModelError) { - const [drained, cycles] = await this.tryDrainInjections(spec, messages, assistant, injectionCycles, { + const [drained, cycles] = await this.tryDrainInjections(spec, messages, spec.outputLocation?.hasWork ? null : assistant, injectionCycles, { phase: "after final response", iteration, }); @@ -1857,7 +1884,7 @@ export class AgentRunner { context.stopReason = stopReason; await hook.afterIteration(context); if (terminalModelError) break; - const [drained, nextCycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after LLM error" }); + const [drained, nextCycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after LLM error", iteration }); injectionCycles = nextCycles; if (drained) { hadInjections = true; @@ -1875,7 +1902,7 @@ export class AgentRunner { context.error = error; context.stopReason = stopReason; await hook.afterIteration(context); - const [drained, nextCycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after empty response" }); + const [drained, nextCycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after empty response", iteration }); injectionCycles = nextCycles; if (drained) { hadInjections = true; @@ -1884,6 +1911,40 @@ export class AgentRunner { break; } + context.finalContent = clean; + context.stopReason = stopReason; + await hook.afterIteration(context); + const pendingGenerators = spec.outputLocation?.pendingProducers() ?? []; + if (pendingGenerators.length) { + const feedback = spec.outputLocation!.pendingMessage(); + if (outputLocationRetries++ < 2 && iteration + 1 < spec.maxIterations) { + messages.push({ role: "system", content: feedback }); + continue; + } + finalContent = feedback; + stopReason = "deliveryPending"; + AgentRunner.appendFinalMessage(messages, finalContent); + break; + } + const invalidPaths = await spec.outputLocation?.checkDelivery() ?? []; + if (invalidPaths.length) { + const feedback = spec.outputLocation!.rejection(invalidPaths); + if (outputLocationRetries++ < 2 && iteration + 1 < spec.maxIterations) { + messages.push({ role: "system", content: feedback }); + continue; + } + finalContent = feedback; + stopReason = "toolError"; + error = "Output location verification failed"; + AgentRunner.appendFinalMessage(messages, finalContent); + context.finalContent = finalContent; + context.stopReason = stopReason; + context.error = error; + break; + } + const [lateInput, lateCycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { iteration }); + injectionCycles = lateCycles; + if (lateInput) { hadInjections = true; continue; } messages.push(assistant ?? buildAssistantMessage(clean)); await this.emitCheckpoint(spec, { phase: "finalResponse", @@ -1897,7 +1958,6 @@ export class AgentRunner { finalContentStreamed = Boolean(context.streamedContent); context.finalContent = finalContent; context.stopReason = stopReason; - await hook.afterIteration(context); break; } @@ -1933,7 +1993,7 @@ export class AgentRunner { stopReason = "cancelled"; error = finalContent; } else { - const finalizationMessages = this.modelContextMessages(messages, modelContextProjection); + const finalizationMessages = this.withOutputInstruction(spec, this.modelContextMessages(messages, modelContextProjection)); const preparedMessages = this.prepareMessagesForModel(spec, finalizationMessages, { toolsForRequest: null, reservedPromptTokens, @@ -2007,7 +2067,8 @@ export class AgentRunner { } else { finalContent = fallback; AgentRunner.appendFinalMessage(messages, finalContent); - const [drained, cycles] = await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after maxIterations" }); + const [drained, cycles] = spec.outputLocation?.hasWork ? [false, injectionCycles] as const + : await this.tryDrainInjections(spec, messages, null, injectionCycles, { phase: "after maxIterations" }); injectionCycles = cycles; if (drained) hadInjections = true; } @@ -2028,6 +2089,28 @@ export class AgentRunner { hadInjections, }); await hook.afterRun(runCtx, result); + if (spec.outputLocation?.hasWork && !["cancelled", "error", "toolError"].includes(result.stopReason)) { + const pending = spec.outputLocation.pendingProducers(); + const invalid = pending.length ? [] : await spec.outputLocation.checkDelivery(); + let replacement: string | null = null; + if (pending.length) { + replacement = spec.outputLocation.pendingMessage(); + result.stopReason = "deliveryPending"; + } else if (invalid.length) { + replacement = spec.outputLocation.rejection(invalid); + result.stopReason = "toolError"; + result.error = "Output location verification failed"; + } else if (spec.hasPendingInput?.()) { + replacement = "收到新的要求，旧交付回复已停止；将先处理新要求，再重新核验完整产物清单。"; + result.stopReason = "maxIterations"; + } + if (replacement) { + if (result.messages.at(-1)?.role === "assistant") result.messages.pop(); + result.finalContent = result.content = replacement; + result.finalContentStreamed = false; + AgentRunner.appendFinalMessage(result.messages, replacement); + } + } return result; } } diff --git a/App/memmy-agent/src/core/agent-runtime/subagent.ts b/App/memmy-agent/src/core/agent-runtime/subagent.ts index cda229d24..8fa72d68d 100644 --- a/App/memmy-agent/src/core/agent-runtime/subagent.ts +++ b/App/memmy-agent/src/core/agent-runtime/subagent.ts @@ -1,3 +1,4 @@ +import { OutputLocationPolicy } from "./tools/output-location.js"; import path from "node:path"; import { performance } from "node:perf_hooks"; import { randomUUID } from "node:crypto"; @@ -260,6 +261,7 @@ export class SubagentManager { temperature?: number | null; workspace?: string | null; readonlySkillRoots?: readonly string[]; + outputLocation?: OutputLocationPolicy; provider?: any; model?: string | null; contextWindowTokens?: number; @@ -284,6 +286,7 @@ export class SubagentManager { contextWindowTokens: args.contextWindowTokens ?? this.contextWindowTokens, modelPreset: args.modelPreset ?? null, modelProvider: args.modelProvider ?? null, + outputTaskId: args.outputLocation?.taskId ?? null, actualModelContext: args.actualModelContext == null ? null : Object.freeze({ @@ -303,9 +306,12 @@ export class SubagentManager { this.taskOrigins.set(taskId, origin); const computerUseContext = resolveSubagentComputerUseAuthorization(args.computerUseAuthorization); if (computerUseContext) this.taskComputerUseContexts.set(taskId, computerUseContext); + let producerDone = false; + try { await this.emitSubagentStart(taskId, task, displayLabel, origin, status); } catch (error) { + producerDone = true; this.taskStatuses.delete(taskId); this.tasks.delete(taskId); this.taskOrigins.delete(taskId); @@ -324,10 +330,11 @@ export class SubagentManager { status, args.originMessageId ?? originMessageId, args.temperature ?? temperature, - { signal: controller.signal }, + { signal: controller.signal, outputLocation: args.outputLocation, onGenerationEnded: () => { producerDone = true; } }, ) .finally(() => { settled = true; + producerDone = true; this.runningTasks.delete(taskId); this.taskStatuses.delete(taskId); this.tasks.delete(taskId); @@ -352,6 +359,7 @@ export class SubagentManager { running.cancelled = () => cancelled || controller.signal.aborted; running.signal = controller.signal; this.runningTasks.set(taskId, running); + args.outputLocation?.trackDelegation(`subagent:${taskId}`, () => producerDone, async () => { abort(); await running; }); if (origin.sessionKey) { const ids = this.sessionTasks.get(origin.sessionKey) ?? new Set(); ids.add(taskId); @@ -368,7 +376,7 @@ export class SubagentManager { status: SubagentStatus, originMessageId: string | null = null, temperature: number | null = null, - { signal = null }: { signal?: AbortSignal | null } = {}, + { signal = null, outputLocation, onGenerationEnded }: { signal?: AbortSignal | null; outputLocation?: OutputLocationPolicy; onGenerationEnded?: () => void } = {}, ): Promise { let finalStatus = "error"; let resultText: string | null = null; @@ -430,6 +438,7 @@ export class SubagentManager { }, sessionKey: origin.sessionKey ?? null, workspace, + outputLocation: outputLocation ?? new OutputLocationPolicy(workspace), llmTimeoutS: this.llmWallTimeoutForSession?.(sessKey) ?? null, abortSignal: signal, turnId, @@ -448,17 +457,14 @@ export class SubagentManager { status.toolEvents = [...(result.toolEvents ?? [])]; finalStatus = "error"; resultText = SubagentManager.formatPartialProgress(result); - await this.announceResult(taskId, label, task, resultText, origin, "error", originMessageId); } else if (status.stopReason === "error") { finalStatus = "error"; const errorText = (result as any).error ?? "Error: subagent execution failed."; resultText = errorText; stopError = errorText; - await this.announceResult(taskId, label, task, errorText, origin, "error", originMessageId); } else { - finalStatus = "ok"; + finalStatus = status.stopReason === "deliveryPending" ? "pending" : "ok"; resultText = result.finalContent || "Task completed but no final response was generated."; - await this.announceResult(taskId, label, task, resultText, origin, "ok", originMessageId); } } catch (error) { if (isCancelled()) { @@ -471,7 +477,6 @@ export class SubagentManager { finalStatus = "error"; stopError = (error as Error).message; resultText = `Error: ${(error as Error).message}`; - await this.announceResult(taskId, label, task, resultText, origin, "error", originMessageId); } finally { releaseInterrupt?.(); boundary.close(isCancelled() ? 'aborted' : 'ended'); @@ -482,6 +487,10 @@ export class SubagentManager { finalStatus, error: stopError, }); + onGenerationEnded?.(); + if (!isCancelled() && resultText != null) { + await this.announceResult(taskId, label, task, resultText, origin, finalStatus, originMessageId); + } } } @@ -494,7 +503,7 @@ export class SubagentManager { status: string, originMessageId: string | null = null, ): Promise { - const statusText = status === "ok" ? "completed successfully" : "failed"; + const statusText = status === "ok" ? "completed successfully" : status === "pending" ? "still in progress" : "failed"; const template = readTemplate("agent/subagent-announce.md"); const content = renderInlineTemplate(template, { label, @@ -514,6 +523,7 @@ export class SubagentManager { } : {}), }; + if (origin.outputTaskId) metadata.outputTaskId = origin.outputTaskId; if (originMessageId) metadata.originMessageId = originMessageId; const msg = new InboundMessage({ channel: "system", diff --git a/App/memmy-agent/src/core/agent-runtime/tools/apply-patch.ts b/App/memmy-agent/src/core/agent-runtime/tools/apply-patch.ts index cea810679..3678b0129 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/apply-patch.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/apply-patch.ts @@ -1,3 +1,4 @@ +import { OUTPUT_PATHS_SCHEMA } from "./output-location.js"; import fs from "node:fs/promises"; import fsSync from "node:fs"; import path from "node:path"; @@ -292,6 +293,7 @@ export class ApplyPatchTool extends Tool { type: "object", additionalProperties: false, properties: { + output_paths: OUTPUT_PATHS_SCHEMA, input: { type: "string", minLength: 1, @@ -525,6 +527,10 @@ Paths must be workspace-relative. Use at most 20 file operations in one patch.`, return "No changes made by patch:\n" + plan.summaries.map(formatSummary).join("\n"); } + for (const change of actionable) { + const locationError = await context?.outputLocation?.checkWrite(change.target, context); + if (locationError) return locationError; + } const attemptedPaths: string[] = []; const attempted = new Set(); const markAttempted = (target: string): void => { @@ -577,6 +583,9 @@ Paths must be workspace-relative. Use at most 20 file operations in one patch.`, throwIfAborted(signal); const lintResults = await lintFiles(lintRequests, { abortSignal: signal }); throwIfAborted(signal); + for (const change of actionable) { + if (change.kind !== "delete") context?.outputLocation?.recordFile(change.target); + } const success = "Patch applied:\n" + plan.summaries.map(formatSummary).join("\n"); const result = appendFileLintResults(success, lintResults); for (const change of unchanged) { @@ -602,8 +611,8 @@ Paths must be workspace-relative. Use at most 20 file operations in one patch.`, ) { return "Error applying patch: input must be a non-empty string"; } - if (Object.keys(params).length !== 1 || !("input" in params)) { - return "Error applying patch: apply_patch accepts only the input parameter"; + if (Object.keys(params).some(key => key !== "input" && key !== "output_paths")) { + return "Error applying patch: apply_patch accepts only input and optional output_paths"; } const signal = context?.abortSignal ?? null; @@ -617,6 +626,8 @@ Paths must be workspace-relative. Use at most 20 file operations in one patch.`, if (isAbortError(error)) throw error; return "Error applying patch: " + errorMessage(error); } + const declarationError = await context?.outputLocation?.declareOutputs(params.output_paths, this.workspace, context); + if (declarationError) return declarationError; return this.applyChanges(plan, context); } } diff --git a/App/memmy-agent/src/core/agent-runtime/tools/base.ts b/App/memmy-agent/src/core/agent-runtime/tools/base.ts index 684eba28e..61d0705bc 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/base.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/base.ts @@ -1,3 +1,5 @@ +import type { OutputLocationPolicy } from "./output-location.js"; + export type JsonSchema = Record; export type FileMutationOutcome = { @@ -6,6 +8,7 @@ export type FileMutationOutcome = { }; export type ToolExecutionContext = { + outputLocation?: OutputLocationPolicy; abortSignal?: AbortSignal | null; toolName?: string; callId?: string | null; diff --git a/App/memmy-agent/src/core/agent-runtime/tools/exec-session.ts b/App/memmy-agent/src/core/agent-runtime/tools/exec-session.ts index 96724ad3c..d6530e403 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/exec-session.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/exec-session.ts @@ -315,6 +315,11 @@ export class ExecSession { }); } + get completion(): { done: boolean; exitCode: number | null; timedOut: boolean; lifecycleError: string | null } { + return { done: this.streamsClosed, exitCode: this.process.exitCode, + timedOut: this.timedOut, lifecycleError: this.lifecycleError }; + } + private isDone(): boolean { return processExited(this.process); } @@ -450,8 +455,10 @@ export class ExecSessionManager { decoderOptions = {}, spawnProcess = spawn, abortSignal = null, + onStart, }: { command: string; + onStart?: (id: string, session: ExecSession) => void; cwd?: string; env?: NodeJS.ProcessEnv; timeout?: number | null; @@ -469,6 +476,7 @@ export class ExecSessionManager { const sessionId = randomUUID().replace(/-/g, "").slice(0, 12); const session = new ExecSession({ command, cwd, env, timeoutS: timeout, shellProgram, login, decoderOptions, spawnProcess }); this.sessions.set(sessionId, session); + onStart?.(sessionId, session); try { const poll = await session.poll(yieldTimeMs, maxOutputChars, {}, abortSignal); if (poll.done) this.sessions.delete(sessionId); diff --git a/App/memmy-agent/src/core/agent-runtime/tools/filesystem.ts b/App/memmy-agent/src/core/agent-runtime/tools/filesystem.ts index 73a8d980f..309627f31 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/filesystem.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/filesystem.ts @@ -10,6 +10,8 @@ import { extractText } from "../../../utils/document.js"; import { getMediaDir } from "../../../config/paths.js"; import { BUILTIN_SKILLS_DIR } from "../skills.js"; +import { OUTPUT_PATHS_SCHEMA } from "./output-location.js"; + const IGNORED_DIRS = new Set([".git", "node_modules"]); const IMAGE_EXTS = new Set([".png", ".jpg", ".jpeg", ".gif", ".webp"]); const DOCUMENT_EXTS = new Set([".pdf", ".docx", ".xlsx", ".pptx"]); @@ -742,7 +744,7 @@ export class WriteFileTool extends Tool { get parameters() { return { type: "object", - properties: { path: { type: "string" }, content: { type: "string" } }, + properties: { path: { type: "string" }, content: { type: "string" }, output_paths: OUTPUT_PATHS_SCHEMA }, required: ["path", "content"], }; } @@ -755,10 +757,14 @@ export class WriteFileTool extends Tool { return resolveFileStates(this.fileStateOwner, this.fallbackFileStates, this.requestContext); } - async execute(params: { path?: string; content?: string } = {}, context?: ToolExecutionContext): Promise { + async execute(params: { path?: string; content?: string; output_paths?: string[] } = {}, context?: ToolExecutionContext): Promise { if (!params.path) return "Error writing file: Unknown path"; if (params.content == null) return "Error writing file: Unknown content"; const target = resolveWorkspacePath(this.workspace, params.path); + const declarationError = await context?.outputLocation?.declareOutputs(params.output_paths, this.workspace, context); + if (declarationError) return declarationError; + const locationError = await context?.outputLocation?.checkWrite(target, context); + if (locationError) return locationError; if (!isToolPathAllowed(target, this.allowedDir, this.extraAllowedDirs)) return `Error writing file: path outside workspace: ${params.path}`; const signal = context?.abortSignal ?? null; let writeStarted = false; @@ -771,6 +777,7 @@ export class WriteFileTool extends Tool { const beforeBytes = await fs.readFile(target, { signal: signal ?? undefined }); throwIfAborted(signal); if (beforeBytes.equals(finalBytes)) { + context?.outputLocation?.recordFile(target); context?.reportFileMutation?.({ path: target, changed: false }); return `No changes made to ${target}: existing content is identical.`; } @@ -784,6 +791,7 @@ export class WriteFileTool extends Tool { writeStarted = true; await fs.writeFile(target, params.content, { encoding: "utf8", signal: signal ?? undefined }); this.fileStates().recordWrite(target); + context?.outputLocation?.recordFile(target); if (!this.postWriteValidation) return `Successfully wrote ${target}`; const content = await readBackWrittenFile(target, params.content, signal); const lint = await lintFile({ path: target, content }, { abortSignal: signal }); @@ -800,6 +808,7 @@ export class WriteFileTool extends Tool { interface EditFileParams { path?: string; + output_paths?: string[]; old_text?: string; oldText?: string; new_text?: string; @@ -899,6 +908,7 @@ export class EditFileTool extends Tool { type: "object", properties: { path: { type: "string", description: "File path, absolute or relative to the workspace." }, + output_paths: OUTPUT_PATHS_SCHEMA, old_text: { type: "string", description: "Required text to replace, copied from read_file. Empty string selects create mode: create a missing file or fill an empty/whitespace-only file, without selectors." }, oldText: { type: "string", description: "Alias of old_text; prefer old_text. The snake_case value takes precedence when both are supplied." }, new_text: { type: "string", description: "Required replacement text; empty string deletes the selected text." }, @@ -939,6 +949,10 @@ export class EditFileTool extends Tool { const parameterError = editOptionsError(options); if (parameterError) return `Error: ${parameterError}`; const target = resolveWorkspacePath(this.workspace, params.path); + const declarationError = await context?.outputLocation?.declareOutputs(params.output_paths, this.workspace, context); + if (declarationError) return declarationError; + const locationError = await context?.outputLocation?.checkWrite(target, context); + if (locationError) return locationError; if (!isToolPathAllowed(target, this.allowedDir, this.extraAllowedDirs)) return `Error editing file: path outside workspace: ${params.path}`; const signal = context?.abortSignal ?? null; let writeStarted = false; @@ -954,6 +968,7 @@ export class EditFileTool extends Tool { if (existing.trim()) return `Error editing file: cannot create file; ${params.path} already exists and is not empty.`; throwIfAborted(signal); if (existingBytes.equals(Buffer.from(newText, "utf8"))) { + context?.outputLocation?.recordFile(target); context?.reportFileMutation?.({ path: target, changed: false }); return `No changes made to ${target}: replacement produced identical content.`; } @@ -964,6 +979,7 @@ export class EditFileTool extends Tool { writeStarted = true; await fs.writeFile(target, newText, { encoding: "utf8", signal: signal ?? undefined }); this.fileStates().recordWrite(target); + context?.outputLocation?.recordFile(target); const success = exists ? "Successfully edited file" : `Successfully created ${target}`; if (!this.postWriteValidation) return success; const content = await readBackWrittenFile(target, newText, signal); @@ -1025,6 +1041,7 @@ export class EditFileTool extends Tool { updated = updated.replace(/\n/g, lineEnding); throwIfAborted(signal); if (raw.equals(Buffer.from(updated, "utf8"))) { + context?.outputLocation?.recordFile(target); context?.reportFileMutation?.({ path: target, changed: false }); const noChange = `No changes made to ${target}: replacement produced identical content.`; return warning ? `${warning}\n${noChange}` : noChange; @@ -1032,6 +1049,7 @@ export class EditFileTool extends Tool { writeStarted = true; await fs.writeFile(target, updated, { encoding: "utf8", signal: signal ?? undefined }); this.fileStates().recordWrite(target); + context?.outputLocation?.recordFile(target); const success = warning ? `${warning}\nSuccessfully edited file` : "Successfully edited file"; if (!this.postWriteValidation) return success; const content = await readBackWrittenFile(target, updated, signal); diff --git a/App/memmy-agent/src/core/agent-runtime/tools/loader.ts b/App/memmy-agent/src/core/agent-runtime/tools/loader.ts index 6004d7bfb..0109c05c9 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/loader.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/loader.ts @@ -21,6 +21,7 @@ import { CreateGoalTool, GetGoalTool, UpdateGoalTool } from "./goal.js"; import { MessageTool } from "./message.js"; import { FindFilesTool, GrepTool } from "./search.js"; import { ExecTool } from "./shell.js"; +import { SetOutputDirectoryTool } from "./output-location.js"; import { SpawnTool } from "./spawn.js"; import { WebFetchTool, WebSearchTool } from "./web.js"; import { BROWSER_TOOL_CLASSES } from "./browser.js"; @@ -67,6 +68,7 @@ const BUILTIN_TOOL_CLASSES: ToolClass[] = [ GetGoalTool, MessageTool, ReadFileTool, + SetOutputDirectoryTool, SpawnTool, WebFetchTool, WebSearchTool, diff --git a/App/memmy-agent/src/core/agent-runtime/tools/output-location.ts b/App/memmy-agent/src/core/agent-runtime/tools/output-location.ts new file mode 100644 index 000000000..db74678cd --- /dev/null +++ b/App/memmy-agent/src/core/agent-runtime/tools/output-location.ts @@ -0,0 +1,232 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import os from "node:os"; +import { randomUUID } from "node:crypto"; +import { Tool, type ToolExecutionContext } from "./base.js"; +import { getMediaDir } from "../../../config/paths.js"; +import { isPathInside } from "./path-utils.js"; + +const STATE_KEY = "outputLocation"; +const PENDING_KEY = "pendingOutputDelivery"; +type Destination = { workspace: string; directory: string; canonical: string }; +type DeliveryEntry = { path: string; owners: Set }; +type Producer = { owner: string; done: () => boolean; cancel?: () => Promise }; +type DeliveryState = { + id: string; + destination?: Destination; + requests: string[]; + entries: Map; + producers: Map; + active: boolean; + needsRegistration: boolean; +}; + +export const OUTPUT_PATHS_SCHEMA = { + type: "array", items: { type: "string", minLength: 1 }, + description: "Complete or additional final files promised to the user, relative to the tool workspace or absolute. Accepted declarations accumulate even if generation fails. Omit or use [] for working files; this never clears prior declarations.", +}; + +/** Resolve existing parents as well as symlinks of not-yet-created files. */ +export async function canonicalOutputPath(value: string): Promise { + let current = path.resolve(value); + const tail: string[] = []; + for (;;) { + try { return path.join(await fs.realpath(current), ...tail.reverse()); } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + try { + if ((await fs.lstat(current)).isSymbolicLink()) throw new Error("Dangling symbolic link: " + current); + } catch (statError) { + if ((statError as NodeJS.ErrnoException).code !== "ENOENT") throw statError; + } + const parent = path.dirname(current); + if (parent === current) throw error; + tail.push(path.basename(current)); current = parent; + } + } +} + +/** Output routing only: this is not an OS sandbox or a permission grant. */ +export class OutputLocationPolicy { + private readonly shared: DeliveryState; + + constructor(readonly workspace: string, private readonly metadata: Record = {}, + private readonly intermediateRoots = [os.tmpdir(), getMediaDir()], + shared?: DeliveryState, private readonly owner = "parent") { + this.workspace = path.resolve(workspace); + const saved = metadata[STATE_KEY] as Destination | undefined; + this.shared = shared ?? { + id: randomUUID(), destination: saved?.workspace === this.workspace ? saved : undefined, + requests: [], entries: new Map(), producers: new Map(), active: true, + needsRegistration: Boolean(metadata[PENDING_KEY]), + }; + } + + get taskId(): string { return this.shared.id; } + get active(): boolean { return this.shared.active; } + get directory(): string { return this.shared.destination?.directory ?? this.workspace; } + get hasOutputs(): boolean { return this.shared.entries.size > 0; } + get hasWork(): boolean { return this.shared.needsRegistration || this.hasOutputs || this.shared.producers.size > 0; } + + /** Only live human input is supplied here, never a summary or recalled memory. */ + addUserRequest(text: string): void { if (text.trim()) this.shared.requests.push(text); } + + fork(): OutputLocationPolicy { + return new OutputLocationPolicy(this.workspace, this.metadata, this.intermediateRoots, this.shared, randomUUID()); + } + + /** Preflight the whole declaration before accepting any obligations. */ + async declareOutputs(values: unknown, cwd = this.workspace, context?: ToolExecutionContext): Promise { + if (values === undefined) return null; + if (!Array.isArray(values) || values.some(value => typeof value !== "string" || !value.trim())) { + return "Error: output_paths must be an array of non-empty file paths."; + } + const targets = values.map(value => path.resolve(cwd, value)); + for (const target of targets) { + const error = await this.checkWrite(target, context, true); + if (error) return error; + } + for (const target of targets) this.recordFile(target, true); + return null; + } + + trackProducer(id: string, done: () => boolean, cancel?: () => Promise): void { + if (this.active) this.shared.producers.set(id, { owner: this.owner, done, cancel }); + } + + trackDelegation(id: string, done: () => boolean, cancel: () => Promise): void { + // A child checks its own commands, but must not wait on its own delegation. + if (this.active) { + this.metadata[PENDING_KEY] = true; + this.shared.producers.set(id, { owner: "parent", done, cancel }); + } + } + + pendingProducers(): string[] { + if (this.shared.needsRegistration) return ["原交付清单不可用，请先通过 output_paths 重新登记完整最终产物清单"]; + return [...this.shared.producers].filter(([, producer]) => { + if (this.owner !== "parent" && producer.owner !== this.owner) return false; + try { return !producer.done(); } catch { return true; } + }).map(([id]) => id); + } + + async cancel(): Promise { + this.shared.active = false; + delete this.metadata[PENDING_KEY]; + await Promise.allSettled([...this.shared.producers.values()].map(producer => producer.cancel?.())); + this.shared.producers.clear(); + this.shared.entries.clear(); + } + + retire(): void { + this.shared.active = false; + delete this.metadata[PENDING_KEY]; + this.shared.producers.clear(); + this.shared.entries.clear(); + } + + pendingMessage(): string { + return `相关生成任务尚未确认结束，尚未确认交付：${this.pendingProducers().join(", ")}。请继续查询或等待受管理任务结束，再核验完整产物清单。`; + } + + async selectDirectory(value: string, userInstruction: string): Promise { + if (!this.active) return "Error: this delivery task has ended; its output directory cannot change."; + if (!userInstruction.trim() || !this.shared.requests.some(request => request.includes(userInstruction))) { + return "Error: output directory must come from the current user's destination instruction, not recalled history. " + this.recovery(); + } + const expanded = value === "~" ? os.homedir() : value.startsWith("~/") ? path.join(os.homedir(), value.slice(2)) : value; + const directory = path.resolve(this.workspace, expanded); + // Natural-language interpretation remains the model's job. Bind its choice to + // literal live input (or a standard named folder), without another model call. + const names: [string, string[]][] = [ + [this.workspace, ["workspace", "工作区"]], + [path.join(os.homedir(), "Desktop"), ["desktop", "桌面"]], + [path.join(os.homedir(), "Downloads"), ["downloads", "下载"]], + [path.join(os.homedir(), "Documents"), ["documents", "文稿"]], + ]; + const literal = userInstruction.includes(value) && value.trim().length > 1; + const named = names.some(([folder, aliases]) => directory === folder && aliases.some(alias => userInstruction.toLowerCase().includes(alias))); + if (!literal && !named) return "Error: destination does not match the quoted user instruction. " + this.recovery(); + const destination = { workspace: this.workspace, directory, canonical: await canonicalOutputPath(directory) }; + const previous = this.directory; + // Changing obligations does not copy files or rewrite already running commands. + if (directory !== previous) { + for (const entry of this.shared.entries.values()) { + entry.path = path.join(directory, path.relative(previous, entry.path)); + } + } + this.shared.destination = destination; + this.metadata[STATE_KEY] = destination; + return `Output directory: ${directory}. No additional confirmation is required. Save deliverables here; source locations do not change this choice.`; + } + + private recovery(): string { return `Save outputs in ${this.directory}. Do not retry the rejected source-directory path; only change the destination when the current user explicitly requests it.`; } + + private async destinationCanonical(): Promise { + const destination = this.shared.destination; + const canonical = await canonicalOutputPath(this.directory); + if (destination && canonical !== destination.canonical) throw new Error("Output directory symlink destination changed"); + return canonical; + } + + async checkWrite(value: string, context?: ToolExecutionContext, deliverable = false): Promise { + if (context?.abortSignal?.aborted) throw Object.assign(new Error("task cancelled"), { name: "AbortError" }); + const target = path.resolve(value); + if (!this.active) return "Error: this delivery task has ended; late writes cannot reactivate it."; + deliverable ||= [...this.shared.entries.values()].some(entry => entry.path === target); + try { + const canonical = await canonicalOutputPath(target); + const roots = deliverable ? [await this.destinationCanonical()] + : [await canonicalOutputPath(this.workspace), await this.destinationCanonical(), + ...await Promise.all(this.intermediateRoots.map(root => canonicalOutputPath(root)))]; + if (roots.some(root => isPathInside(canonical, root))) return null; + } catch (error) { return `Error: output path could not be resolved: ${(error as Error).message}. ${this.recovery()}`; } + return `Error: output path outside allowed directory: ${target}. ${this.recovery()}`; + } + + recordFile(value: string, deliverable = false): void { + if (!deliverable || !this.active) return; + this.shared.needsRegistration = false; + this.metadata[PENDING_KEY] = true; + const target = path.resolve(value); + const existing = [...this.shared.entries.values()].find(entry => entry.path === target); + if (existing) existing.owners.add(this.owner); + else this.shared.entries.set(target, { path: target, owners: new Set([this.owner]) }); + } + + instruction(): string { + return `Output location: ${this.directory}. Default: current workspace (${this.workspace}). Reading external files is allowed and never changes the output destination. If the current user explicitly requests another output directory, call set_output_directory with that directory and an exact quote of the live user instruction; it takes effect directly without a permission question. Declare all planned final files in output_paths before generation, including files not created yet. exec requires [] for commands without deliverables; write_file, edit_file and apply_patch accept optional output_paths. A later declaration adds obligations and never removes previous ones. Do not declare internal scripts or working README files; do declare README files intended for the user. Working directory and reference-file paths are not output destinations. Write intermediates in the workspace, temporary or media directories. Deliver files from the selected output directory. On a rejected path, use the output directory stated in the error instead of repeatedly retrying. Do not present an input/reference link as a newly generated output.`; + } + + async checkDelivery(_content?: string): Promise { + void _content; // Backwards-compatible argument; never used to select paths. + const invalid: string[] = []; + // Check literal registered paths. Final prose and Markdown never select a subset. + for (const entry of this.shared.entries.values()) { + if (this.owner !== "parent" && !entry.owners.has(this.owner)) continue; + const target = entry.path; + try { + if (!isPathInside(await canonicalOutputPath(target), await this.destinationCanonical()) || !(await fs.stat(target)).isFile()) invalid.push(target); + } catch { invalid.push(target); } + } + return invalid; + } + + rejection(paths: string[]): string { + return `输出文件位置核验未通过，尚未确认交付：\n${paths.join("\n")}\n当前输出目录：${this.directory}。请在此目录生成文件后再交付。`; + } +} + +export class SetOutputDirectoryTool extends Tool { + static scopes = new Set(["core", "subagent"]); + get name(): string { return "set_output_directory"; } + get description(): string { return "Use the current user's explicitly requested output directory, without an extra confirmation. Quote the user's destination instruction, not a remembered task or a source path."; } + get exclusive(): boolean { return true; } + get parameters() { return { type: "object", properties: { + directory: { type: "string", minLength: 1 }, user_instruction: { type: "string", minLength: 1 }, + }, required: ["directory", "user_instruction"] }; } + async execute(params: Record, context?: ToolExecutionContext): Promise { + if (!context?.outputLocation) return "Error: output location is unavailable."; + return context.outputLocation.selectDirectory(params.directory, params.user_instruction); + } +} diff --git a/App/memmy-agent/src/core/agent-runtime/tools/shell.ts b/App/memmy-agent/src/core/agent-runtime/tools/shell.ts index e2131983b..f5223c0ce 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/shell.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/shell.ts @@ -290,6 +290,7 @@ export class ExecTool extends Tool { yield_time_ms: { type: "integer", minimum: 0, maximum: MAX_YIELD_MS }, max_output_chars: { type: "integer", minimum: 1000, maximum: MAX_OUTPUT_CHARS }, max_output_tokens: { type: "integer", minimum: 1000, maximum: MAX_OUTPUT_CHARS }, + output_paths: { type: "array", items: { type: "string", minLength: 1 }, description: "All files this command will create/modify and deliver, relative to cwd or absolute. Use [] when there are no deliverables. Required when an output location policy is active." }, }, }; } @@ -512,6 +513,7 @@ export class ExecTool extends Tool { maxOutputChars?: number | null; max_output_tokens?: number | null; maxOutputTokens?: number | null; + output_paths?: string[]; } = {}, context?: ToolExecutionContext): Promise { const command = firstNonEmptyString(params.command, params.cmd); if (!command) return "Error: Missing command. Provide command or cmd."; @@ -525,6 +527,11 @@ export class ExecTool extends Tool { params.login, ); if (typeof prepared === "string") return prepared; + if (context?.outputLocation) { + if (!Array.isArray(params.output_paths)) return "Error: declare output_paths before executing (use [] for commands without deliverables). " + context.outputLocation.instruction(); + const declarationError = await context.outputLocation.declareOutputs(params.output_paths, prepared.cwd, context); + if (declarationError) return declarationError; + } const maxOutputChars = clampSessionInt(maxOutputParam, ExecTool.DEFAULT_OUTPUT_CHARS, 1000, MAX_OUTPUT_CHARS); if (params.yield_time_ms != null || params.yieldTimeMs != null) { @@ -532,6 +539,13 @@ export class ExecTool extends Tool { throwIfAborted(signal); const [sessionId, poll] = await this.sessionManager.start({ command: prepared.command, + // Install before the first poll can finish and remove the manager entry. + onStart: (id, session) => { + if (params.output_paths?.length) { + context?.outputLocation?.trackProducer(`exec:${id}`, () => session.completion.done, + () => this.sessionManager.stop(id)); + } + }, cwd: prepared.cwd, env: prepared.env, timeout: prepared.timeout, diff --git a/App/memmy-agent/src/core/agent-runtime/tools/spawn.ts b/App/memmy-agent/src/core/agent-runtime/tools/spawn.ts index d8f5cdff7..ac981d0be 100644 --- a/App/memmy-agent/src/core/agent-runtime/tools/spawn.ts +++ b/App/memmy-agent/src/core/agent-runtime/tools/spawn.ts @@ -1,3 +1,4 @@ +import type { ToolExecutionContext } from "./base.js"; import { Tool } from "./base.js"; import { RequestContextStore } from "./context.js"; import { delegateSubagentComputerUse } from "../subagent-computer-use-context.js"; @@ -63,7 +64,7 @@ export class SpawnTool extends Tool { this.workspace = ctx.workspace ?? null; } - async execute(params: { task?: string; label?: string; temperature?: number } = {}): Promise { + async execute(params: { task?: string; label?: string; temperature?: number } = {}, context?: ToolExecutionContext): Promise { if (!params.task) return "Error: missing task"; if (!this.manager || typeof this.manager.spawn !== "function") { return "Error: subagent manager is unavailable; cannot spawn subagent."; @@ -96,6 +97,7 @@ export class SpawnTool extends Tool { workspace, readonlySkillRoots: this.readonlySkillRoots, computerUseAuthorization: delegateSubagentComputerUse(requestContext), + outputLocation: context?.outputLocation?.fork(), }), ); } diff --git a/App/memmy-agent/tests/core/agent-runtime/delivery-boundary.test.ts b/App/memmy-agent/tests/core/agent-runtime/delivery-boundary.test.ts new file mode 100644 index 000000000..6d21af391 --- /dev/null +++ b/App/memmy-agent/tests/core/agent-runtime/delivery-boundary.test.ts @@ -0,0 +1,220 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { OutputLocationPolicy } from "../../../src/core/agent-runtime/tools/output-location.js"; +import { ExecTool } from "../../../src/core/agent-runtime/tools/shell.js"; +import { ExecSessionManager, WriteStdinTool } from "../../../src/core/agent-runtime/tools/exec-session.js"; +import { WriteFileTool } from "../../../src/core/agent-runtime/tools/filesystem.js"; +import { SetOutputDirectoryTool } from "../../../src/core/agent-runtime/tools/output-location.js"; +import { AgentRunner, AgentRunSpec } from "../../../src/core/agent-runtime/runner.js"; +import { AgentHook } from "../../../src/core/agent-runtime/hook.js"; +import { ToolRegistry } from "../../../src/core/agent-runtime/tools/registry.js"; +import { LLMProvider, LLMResponse, ToolCallRequest } from "../../../src/providers/base.js"; +import { SubagentManager } from "../../../src/core/agent-runtime/subagent.js"; +import { AgentLoop } from "../../../src/core/agent-runtime/loop.js"; +import { Config } from "../../../src/config/schema.js"; +import { AsyncQueue, InboundMessage, MessageBus } from "../../../src/core/runtime-messages/index.js"; +vi.mock("../../../src/tools/computer-use/desktop-screen-capture.js", async (importOriginal) => ({ + ...await importOriginal(), + initializeDesktopScreenCapture: vi.fn().mockResolvedValue(undefined), +})); +let root: string, workspace: string, policy: OutputLocationPolicy; +let manager: ExecSessionManager; +beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "memmy-delivery-boundary-")); + workspace = path.join(root, "workspace"); await fs.mkdir(workspace); + policy = new OutputLocationPolicy(workspace, {}, []); + manager = new ExecSessionManager(); +}); +afterEach(async () => { + for (const id of [...manager.sessions.keys()]) await manager.stop(id); + vi.restoreAllMocks(); await fs.rm(root, { recursive: true, force: true }); +}); +class Provider extends LLMProvider { + requests: any[] = []; + constructor(private responses: LLMResponse[]) { super(); } + getDefaultModel() { return "test-model"; } + async chat(args: any) { return this.chatWithRetry(args); } + async chatWithRetry(args: any) { + this.requests.push(structuredClone({ messages: args.messages })); + const response = this.responses.shift(); if (!response) throw new Error("Unexpected model request"); return response; + } +} +let sequence = 0; +const reply = (content: string) => new LLMResponse({ content }); +const call = (name: string, args: Record) => new LLMResponse({ content: null, + toolCalls: [new ToolCallRequest({ id: `delivery-${sequence++}`, name, arguments: args })], finishReason: "tool_calls" }); +function tools() { + const registry = new ToolRegistry(); + registry.register(new WriteFileTool({ workspace, postWriteValidation: false })); + registry.register(new SetOutputDirectoryTool()); return registry; +} +async function run(responses: LLMResponse[], options: Partial = {}) { + const provider = new Provider(responses); + return new AgentRunner(provider).run(new AgentRunSpec({ provider, outputLocation: policy, tools: tools(), + workspace, messages: [{ role: "user", content: "generate files" }], ...options })); +} +function makeLoop(responses: LLMResponse[], maxIterations = 4) { + return new AgentLoop({ bus: new MessageBus(), provider: new Provider(responses), workspace, + model: "test-model", maxIterations, sessionDir: path.join(root, "sessions"), + config: new Config({ fileMemory: { enabled: false }, memmyMemory: { enabled: false } }) }); +} +async function waitUntilFile(target: string) { + for (let attempt = 0; attempt < 200; attempt++) { + try { await fs.stat(target); return; } catch { await new Promise(resolve => setTimeout(resolve, 5)); } + } + throw new Error("background test file did not appear"); +} + +describe("managed generation completion", () => { + it("does not certify an existing file while the managed writer is still running", async () => { + const target = path.join(workspace, "report.md"); + const output = await new ExecTool({ workspace, sessionManager: manager }).execute({ + command: "printf partial > report.md; while [ ! -f release ]; do sleep 0.01; done; printf final > report.md", + output_paths: ["report.md"], yield_time_ms: 0, + }, { outputLocation: policy }); + const id = /session_id: (\S+)/.exec(output)![1]; + await waitUntilFile(target); + expect(await policy.checkDelivery()).toEqual([]); + expect(policy.pendingProducers()).toEqual([`exec:${id}`]); + const result = await run([reply("all done")], { maxIterations: 1 }); + expect(result.stopReason).toBe("deliveryPending"); expect(result.finalContent).not.toBe("all done"); + await fs.writeFile(path.join(workspace, "release"), "go"); + await new WriteStdinTool({ manager }).execute({ session_id: id, yield_time_ms: 1000 }); + expect(manager.get(id)).toBeUndefined(); + expect(policy.pendingProducers()).toEqual([]); + expect(await fs.readFile(target, "utf8")).toBe("final"); + expect((await run([reply("now complete")])).stopReason).toBe("completed"); + }); + + it("waits for every writer of the same artifact and does not track read-only exec declarations", async () => { + const exec = new ExecTool({ workspace, sessionManager: manager }); + await exec.execute({ command: "printf work > report.md; while [ ! -f release ]; do sleep 0.01; done", + output_paths: ["report.md"], yield_time_ms: 0 }, { outputLocation: policy }); + await exec.execute({ command: "printf second > report.md", output_paths: ["report.md"], yield_time_ms: 1000 }, { outputLocation: policy }); + expect(policy.pendingProducers()).toHaveLength(1); + const readPolicy = new OutputLocationPolicy(workspace, {}, []); + await exec.execute({ command: "pwd", output_paths: [], yield_time_ms: 1000 }, { outputLocation: readPolicy }); + expect(readPolicy.hasWork).toBe(false); + }); +}); + +describe("fresh checks and publication", () => { + it("runs write-capable finalization hooks before verification", async () => { + const target = path.join(workspace, "report.md"); policy.recordFile(target, true); await fs.writeFile(target, "report"); + class Hook extends AgentHook { override async afterIteration() { await fs.rm(target, { force: true }); } } + const result = await run([reply("complete"), reply("complete"), reply("complete")], { hook: new Hook() }); + expect(result.stopReason).toBe("toolError"); expect(result.finalContent).toContain(target); + expect(result.messages.some(message => message.role === "assistant" && message.content === "complete")).toBe(false); + }); + + it("does not reuse a pass after an end-of-run hook changes the file", async () => { + const target = path.join(workspace, "report.md"); policy.recordFile(target, true); await fs.writeFile(target, "report"); + class Hook extends AgentHook { override async afterRun() { await fs.rm(target); } } + const result = await run([reply("complete")], { hook: new Hook() }); + expect(result.stopReason).toBe("toolError"); expect(result.finalContent).toContain(target); + }); + + it("processes input arriving during verification and checks the complete relocated manifest", async () => { + const target = path.join(workspace, "report.md"), destination = path.join(root, "new"); + policy.recordFile(target, true); await fs.writeFile(target, "report"); + const quote = `改放 ${destination}`; let pending = false; let injected = false; + const original = policy.checkDelivery.bind(policy); + vi.spyOn(policy, "checkDelivery").mockImplementation(async () => { + const result = await original(); if (!injected) pending = true; return result; + }); + const result = await run([reply("old complete"), call("set_output_directory", { directory: destination, user_instruction: quote }), + call("write_file", { path: path.join(destination, "report.md"), content: "updated", output_paths: [path.join(destination, "report.md")] }), reply("new complete")], { + hasPendingInput: () => pending, + injectionCallback: () => { if (!pending) return []; pending = false; injected = true; + return [{ role: "user", content: quote, output_user_instruction: quote }]; }, + }); + expect(result.finalContent).toBe("new complete"); + expect(result.messages.some(message => message.role === "assistant" && message.content === "old complete")).toBe(false); + expect(await original()).toEqual([]); expect(policy.directory).toBe(destination); + }); + + it("does not publish an old completion when the iteration limit leaves input pending", async () => { + const target = path.join(workspace, "report.md"); policy.recordFile(target, true); await fs.writeFile(target, "report"); + const result = await run([reply("old complete")], { maxIterations: 1, hasPendingInput: () => true }); + expect(result.stopReason).toBe("maxIterations"); expect(result.finalContent).not.toBe("old complete"); + }); + + it("checks the actual GUI and bus publication boundary, preserves obligations, then handles the follow-up", async () => { + const target = path.join(workspace, "report.md"); + const agent = makeLoop([call("write_file", { path: target, content: "report", output_paths: [target] }), reply("old complete"), + call("write_file", { path: target, content: "updated", output_paths: [target] }), reply("new complete")]); + const queue = new AsyncQueue(); + agent.unifiedSession = false; + const mirror = { user: vi.fn(), prepareSession: vi.fn().mockReturnValue(null), sessionKeyForMessage: vi.fn().mockReturnValue(null), progress: vi.fn(), final: vi.fn(), ended: vi.fn(), running: vi.fn(), turn: vi.fn().mockReturnValue({}) }; + agent.guiTranscriptMirror = mirror as any; + const originalSave = agent.stateSave.bind(agent); let first = true; + vi.spyOn(agent, "stateSave").mockImplementation(async context => { + await originalSave(context); + if (first) { first = false; queue.put(new InboundMessage({ channel: "telegram", chatId: "test", senderId: "human", content: "revise report" })); } + return "ok"; + }); + await agent.dispatchMessage(new InboundMessage({ channel: "telegram", chatId: "test", senderId: "human", content: "generate report" }), () => false, null, queue); + const messages: string[] = []; + while (agent.bus.outboundSize) messages.push((await agent.bus.nextOutbound()).content); + expect(messages).not.toContain("old complete"); expect(mirror.final).not.toHaveBeenCalled(); + const task = (agent as any).outputTasks.get("telegram:test") as OutputLocationPolicy; + expect(task.hasOutputs).toBe(true); + await agent.dispatchMessage(await agent.bus.nextInbound()); + while (agent.bus.outboundSize) messages.push((await agent.bus.nextOutbound()).content); + expect(messages.filter(text => text === "new complete")).toHaveLength(1); + expect(mirror.final).toHaveBeenCalledTimes(1); + expect((agent as any).outputTasks.has("telegram:test")).toBe(false); + }); + + it("reattaches a cross-round child result and rejects missing child artifacts", async () => { + const agent = makeLoop([reply("child says complete"), reply("child says complete"), reply("child says complete")], 3); + const key = "cli:child-result"; + const child = policy.fork(), missing = path.join(workspace, "summary.xlsx"); child.recordFile(missing, true); + (agent as any).outputTasks.set(key, policy); + const result = await agent.processMessage(new InboundMessage({ channel: "system", senderId: "subagent", chatId: key, + content: "child complete", sessionKeyOverride: key, metadata: { injectedEvent: "subagentResult", outputTaskId: policy.taskId, subagentTaskId: "child-1" } }), key); + expect(result?.content).toContain(missing); expect(result?.content).toContain("核验未通过"); + }); + + it("does not attach a late child result to a different task", async () => { + const agent = makeLoop([]), key = "cli:late"; + (agent as any).outputTasks.set(key, new OutputLocationPolicy(workspace)); + const result = await agent.processMessage(new InboundMessage({ channel: "system", senderId: "subagent", chatId: key, + content: "old complete", metadata: { outputTaskId: policy.taskId } }), key); + expect(result).toBeNull(); + }); +}); + + +describe("delegation and cancellation", () => { + it("returns child records with the original task id and never waits on itself or siblings", async () => { + const missing = path.join(workspace, "parent.pdf"), childFile = path.join(workspace, "child.md"); + policy.recordFile(missing, true); + policy.trackDelegation("subagent:sibling", () => false, async () => undefined); + const bus = new MessageBus(); + const subagents = new SubagentManager({ bus, workspace, maxIterations: 3, provider: new Provider([ + call("write_file", { path: childFile, content: "child report", output_paths: [childFile] }), reply("child done"), + ]) }); + await subagents.spawn({ task: "generate child report", workspace, sessionKey: "cli:parent", outputLocation: policy.fork() }); + await Promise.all([...subagents.runningTasks.values()]); + const result = await bus.nextInbound(); + expect(result.metadata.outputTaskId).toBe(policy.taskId); + expect(result.content).toContain("child done"); + expect(await policy.checkDelivery()).toEqual([missing]); + expect(policy.pendingProducers()).toEqual(["subagent:sibling"]); + }); + + it("does not commit file success after cancellation during the save phase", async () => { + const target = path.join(workspace, "report.md"); + const agent = makeLoop([call("write_file", { path: target, content: "report", output_paths: [target] }), reply("complete")]); + const controller = new AbortController(); + const save = agent.stateSave.bind(agent); + vi.spyOn(agent, "stateSave").mockImplementation(async context => { await save(context); controller.abort(); return "ok"; }); + await agent.dispatchMessage(new InboundMessage({ channel: "telegram", chatId: "cancel", senderId: "human", content: "generate" }), () => controller.signal.aborted, controller.signal); + const output: string[] = []; + while (agent.bus.outboundSize) output.push((await agent.bus.nextOutbound()).content); + expect(output).not.toContain("complete"); + }); +}); diff --git a/App/memmy-agent/tests/core/agent-runtime/loop-progress.test.ts b/App/memmy-agent/tests/core/agent-runtime/loop-progress.test.ts index 112dde80b..f4b696bab 100644 --- a/App/memmy-agent/tests/core/agent-runtime/loop-progress.test.ts +++ b/App/memmy-agent/tests/core/agent-runtime/loop-progress.test.ts @@ -342,7 +342,7 @@ describe("AgentLoop progress integration", () => { fs.writeFileSync(path.join(root, "file.txt"), "fixture\n"); const loop = makeLoop(root); loop.bus = bus; - const toolCall = new ToolCallRequest({ id: "tc1", name: "exec", arguments: { command: "ls file.txt" } }); + const toolCall = new ToolCallRequest({ id: "tc1", name: "exec", arguments: { command: "ls file.txt", output_paths: [] } }); const responses = [new LLMResponse({ content: "", toolCalls: [toolCall] }), new LLMResponse({ content: "Done", toolCalls: [] })]; (loop.provider as any).chatWithRetry = vi.fn(async () => responses.shift()); diff --git a/App/memmy-agent/tests/core/agent-runtime/runner-injections.test.ts b/App/memmy-agent/tests/core/agent-runtime/runner-injections.test.ts index 8446fb7e6..c3c0750c7 100644 --- a/App/memmy-agent/tests/core/agent-runtime/runner-injections.test.ts +++ b/App/memmy-agent/tests/core/agent-runtime/runner-injections.test.ts @@ -369,6 +369,8 @@ describe("AgentLoop pending queues", () => { const bus = new MessageBus(); const loop = new AgentLoop({ bus, provider: makeProvider(async () => new LLMResponse({ content: "done" })), workspace: root, model: "test-model", config: isolatedLoopConfig() }); loop.unifiedSession = true; + // Queue routing is under test; desktop/browser startup is an external dependency. + vi.spyOn(loop, "initializeRuntimeTools").mockResolvedValue(undefined); loop.processMessageInternal = vi.fn(async () => null) as any; const pending = new AsyncQueue(); loop.pendingQueues.set(UNIFIED_SESSION_KEY, pending); @@ -415,6 +417,8 @@ describe("AgentLoop pending queues", () => { const root = tmpRoot(); const bus = new MessageBus(); const loop = new AgentLoop({ bus, provider: makeProvider(async () => new LLMResponse({ content: "done" })), workspace: root, model: "test-model", config: isolatedLoopConfig() }); + // Queue routing does not require desktop/browser startup. + vi.spyOn(loop, "initializeRuntimeTools").mockResolvedValue(undefined); loop.processMessageInternal = vi.fn(async () => null) as any; loop.pendingQueues.set("cli:c", { put: () => { throw new Error("full"); } } as any); diff --git a/App/memmy-agent/tests/core/agent-runtime/task-cancel.test.ts b/App/memmy-agent/tests/core/agent-runtime/task-cancel.test.ts index 899e2758b..dc0de000a 100644 --- a/App/memmy-agent/tests/core/agent-runtime/task-cancel.test.ts +++ b/App/memmy-agent/tests/core/agent-runtime/task-cancel.test.ts @@ -544,6 +544,6 @@ describe("task cancellation", () => { expect(out.content).toBe("done"); expect(loop.runner.run).toHaveBeenCalledTimes(1); - expect(injections).toEqual([{ role: "user", content: "second" }]); + expect(injections).toEqual([{ role: "user", content: "second", output_user_instruction: "second" }]); }); }); diff --git a/App/memmy-agent/tests/core/agent-runtime/tools/apply-patch-tool.test.ts b/App/memmy-agent/tests/core/agent-runtime/tools/apply-patch-tool.test.ts index 547cb136f..8d9a1b9be 100644 --- a/App/memmy-agent/tests/core/agent-runtime/tools/apply-patch-tool.test.ts +++ b/App/memmy-agent/tests/core/agent-runtime/tools/apply-patch-tool.test.ts @@ -37,7 +37,7 @@ describe("ApplyPatchTool contract", () => { required: ["input"], properties: { input: { type: "string", minLength: 1 } }, }); - expect(Object.keys(tool.parameters.properties)).toEqual(["input"]); + expect(Object.keys(tool.parameters.properties)).toEqual(["output_paths", "input"]); expect(tool.parameters.properties.input.description).toContain("*** Begin Patch"); expect(tool.parameters.properties.input.description).toContain("*** Add File:"); expect(tool.parameters.properties.input.description).toContain("*** Update File:"); @@ -76,7 +76,7 @@ describe("ApplyPatchTool contract", () => { ])("rejects extra parameters %#", async (extra) => { const tool = new ApplyPatchTool({ workspace: workspace() }); const result = await tool.execute({ ...input("*** Add File: a", "+x"), ...extra }); - expect(result).toBe("Error applying patch: apply_patch accepts only the input parameter"); + expect(result).toBe("Error applying patch: apply_patch accepts only input and optional output_paths"); }); it("lets the normal registry validation reject the removed edits-only shape", () => { diff --git a/App/memmy-agent/tests/core/agent-runtime/tools/delivery-manifest.test.ts b/App/memmy-agent/tests/core/agent-runtime/tools/delivery-manifest.test.ts new file mode 100644 index 000000000..94db89ba6 --- /dev/null +++ b/App/memmy-agent/tests/core/agent-runtime/tools/delivery-manifest.test.ts @@ -0,0 +1,111 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { OutputLocationPolicy } from "../../../../src/core/agent-runtime/tools/output-location.js"; +import { EditFileTool, WriteFileTool } from "../../../../src/core/agent-runtime/tools/filesystem.js"; +let root: string, workspace: string, policy: OutputLocationPolicy; +beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "memmy-delivery-manifest-")); + workspace = path.join(root, "workspace"); await fs.mkdir(workspace); + policy = new OutputLocationPolicy(workspace, {}, []); +}); +afterEach(async () => { await fs.rm(root, { recursive: true, force: true }); }); +describe("complete delivery manifest", () => { + it("checks a missing Excel even when the reply links only the existing PDF", async () => { + const pdf = path.join(workspace, "report.pdf"), excel = path.join(workspace, "summary.xlsx"); + policy.recordFile(pdf, true); policy.recordFile(excel, true); await fs.writeFile(pdf, "pdf"); + expect(await policy.checkDelivery(`[report](${pdf})`)).toEqual([excel]); + }); + it("checks literal parentheses without parsing Markdown", async () => { + const target = path.join(workspace, "report(终版).pdf"), existing = path.join(workspace, "summary.csv"); + policy.recordFile(target, true); policy.recordFile(existing, true); await fs.writeFile(existing, "summary"); + expect(await policy.checkDelivery(`[report](${target}) [summary](${existing})`)).toEqual([target]); + await fs.writeFile(target, "pdf"); expect(await policy.checkDelivery("done")).toEqual([]); + }); + it("registers file-tool declarations before generating one of two promised files", async () => { + const pdf = path.join(workspace, "report.md"), excel = path.join(workspace, "summary.csv"); + await new WriteFileTool({ workspace, postWriteValidation: false }).execute({ path: pdf, content: "report", output_paths: [pdf, excel] } as any, { outputLocation: policy }); + expect(await policy.checkDelivery(`[report](${pdf})`)).toEqual([excel]); + }); + it("permits disposable working files without making them delivery obligations", async () => { + const scratch = path.join(workspace, "README.md"); + await new WriteFileTool({ workspace, postWriteValidation: false }).execute({ path: scratch, content: "internal" }, { outputLocation: policy }); + await fs.rm(scratch); expect(await policy.checkDelivery(`[internal](${scratch})`)).toEqual([]); + }); + it("keeps the complete manifest when a validated edit selects one occurrence", async () => { + const report = path.join(workspace, "report.md"), summary = path.join(workspace, "summary.csv"); + await fs.writeFile(report, "draft draft"); + await new EditFileTool({ workspace, postWriteValidation: false }).execute({ + path: report, old_text: "draft", new_text: "final", occurrence: 2, expected_replacements: 1, + output_paths: [report, summary], + }, { outputLocation: policy }); + expect(await fs.readFile(report, "utf8")).toBe("draft final"); + expect(await policy.checkDelivery(`[report](${report})`)).toEqual([summary]); + }); + it("rejects incompatible edit selectors before registering outputs or changing files", async () => { + const report = path.join(workspace, "report.md"), summary = path.join(workspace, "summary.csv"); + await fs.writeFile(report, "draft"); + const result = await new EditFileTool({ workspace, postWriteValidation: false }).execute({ + path: report, old_text: "draft", new_text: "final", occurrence: 1, replace_all: true, + output_paths: [report, summary], + }, { outputLocation: policy }); + expect(result).toContain("occurrence cannot be used with replace_all=true"); + expect(await fs.readFile(report, "utf8")).toBe("draft"); + expect(policy.hasOutputs).toBe(false); + }); + it("merges child obligations into the parent while children check their own subset", async () => { + const parentFile = path.join(workspace, "report.pdf"), childFile = path.join(workspace, "summary.xlsx"); + policy.recordFile(parentFile, true); const child = policy.fork(); child.recordFile(childFile, true); + await fs.writeFile(childFile, "excel"); expect(await child.checkDelivery("done")).toEqual([]); + expect(await policy.checkDelivery("done")).toEqual([parentFile]); + await fs.rm(childFile); expect(await policy.checkDelivery("done")).toEqual([parentFile, childFile]); + }); + it("moves pending obligations by relative path when the destination changes", async () => { + const old = path.join(workspace, "nested", "report.pdf"); policy.recordFile(old, true); + await fs.mkdir(path.dirname(old)); await fs.writeFile(old, "pdf"); + const destination = path.join(root, "delivery"), request = `保存到 ${destination}`; + policy.addUserRequest(request); await policy.selectDirectory(destination, request); + expect(await policy.checkDelivery("done")).toEqual([path.join(destination, "nested", "report.pdf")]); + await fs.mkdir(path.join(destination, "nested"), { recursive: true }); + await fs.writeFile(path.join(destination, "nested", "report.pdf"), "pdf"); expect(await policy.checkDelivery("done")).toEqual([]); + }); +}); + +describe("declaration lifetime", () => { + it("keeps accepted obligations when a file write fails and merges later partial declarations", async () => { + const directory = path.join(workspace, "report.md"), missing = path.join(workspace, "summary.csv"); + await fs.mkdir(directory); + await expect(new WriteFileTool({ workspace, postWriteValidation: false }).execute({ path: directory, + content: "report", output_paths: [directory, missing] }, { outputLocation: policy })).rejects.toThrow(); + await policy.declareOutputs([missing]); await policy.declareOutputs([]); + expect(await policy.checkDelivery()).toEqual([directory, missing]); + }); + it("rejects the whole declaration before recording any invalid destination", async () => { + const allowed = path.join(workspace, "report.md"), wrong = path.join(root, "source", "summary.csv"); + expect(await policy.declareOutputs([allowed, wrong])).toContain("outside allowed directory"); + expect(policy.hasOutputs).toBe(false); + }); + it("does not downgrade an already-declared final file when an optional declaration is omitted", async () => { + const target = path.join(workspace, "report.md"), outside = path.join(root, "outside"); + await policy.declareOutputs([target]); await fs.mkdir(outside); await fs.writeFile(path.join(outside, "report.md"), "old"); await fs.symlink(path.join(outside, "report.md"), target); + expect(await new WriteFileTool({ workspace, postWriteValidation: false }).execute({ path: target, content: "new" }, { outputLocation: policy })).toContain("outside allowed directory"); + }); + it("cannot reactivate a retired task from an old child", async () => { + const child = policy.fork(), target = path.join(workspace, "report.md"); policy.retire(); + child.recordFile(target, true); child.addUserRequest(`保存到 ${root}`); + expect(await child.selectDirectory(root, `保存到 ${root}`)).toContain("task has ended"); + expect(await child.checkWrite(target)).toContain("task has ended"); expect(policy.hasOutputs).toBe(false); + }); + it("requires redeclaration when an unfinished task loses its in-memory manifest", async () => { + const metadata = {}; + const original = new OutputLocationPolicy(workspace, metadata, []); + await original.declareOutputs(["report.md"]); + const recovered = new OutputLocationPolicy(workspace, metadata, []); + expect(recovered.pendingProducers()).toHaveLength(1); + await recovered.declareOutputs([]); expect(recovered.pendingProducers()).toHaveLength(1); + await recovered.declareOutputs(["report.md"]); expect(recovered.pendingProducers()).toEqual([]); + expect(await recovered.checkDelivery()).toEqual([path.join(workspace, "report.md")]); + recovered.retire(); expect(new OutputLocationPolicy(workspace, metadata, []).hasWork).toBe(false); + }); +}); diff --git a/App/memmy-agent/tests/core/agent-runtime/tools/output-location.test.ts b/App/memmy-agent/tests/core/agent-runtime/tools/output-location.test.ts new file mode 100644 index 000000000..12957fc05 --- /dev/null +++ b/App/memmy-agent/tests/core/agent-runtime/tools/output-location.test.ts @@ -0,0 +1,264 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { OutputLocationPolicy, SetOutputDirectoryTool } from "../../../../src/core/agent-runtime/tools/output-location.js"; +import { ReadFileTool, WriteFileTool, EditFileTool } from "../../../../src/core/agent-runtime/tools/filesystem.js"; +import { ApplyPatchTool } from "../../../../src/core/agent-runtime/tools/apply-patch.js"; +import { ExecTool } from "../../../../src/core/agent-runtime/tools/shell.js"; +import { SpawnTool } from "../../../../src/core/agent-runtime/tools/spawn.js"; +import { AgentRunner, AgentRunSpec } from "../../../../src/core/agent-runtime/runner.js"; +import { ToolRegistry } from "../../../../src/core/agent-runtime/tools/registry.js"; +import { LLMProvider, LLMResponse, ToolCallRequest } from "../../../../src/providers/base.js"; +import { AgentLoop } from "../../../../src/core/agent-runtime/loop.js"; +import { Config } from "../../../../src/config/schema.js"; +import { InboundMessage, MessageBus } from "../../../../src/core/runtime-messages/index.js"; + +vi.mock("../../../../src/tools/computer-use/desktop-screen-capture.js", async (importOriginal) => ({ + ...await importOriginal(), + initializeDesktopScreenCapture: vi.fn().mockResolvedValue(undefined), +})); + +let root: string, a: string, b: string; +let metadata: Record, policy: OutputLocationPolicy; +beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "memmy-output-location-")); + a = path.join(root, "A"); b = path.join(root, "B"); + await fs.mkdir(a); await fs.mkdir(b); + metadata = {}; policy = new OutputLocationPolicy(a, metadata, []); +}); +afterEach(async () => { vi.restoreAllMocks(); await fs.rm(root, { recursive: true, force: true }); }); +const ctx = () => ({ outputLocation: policy }); +async function select(directory = b) { + const quote = `请保存到 ${directory}`; + policy.addUserRequest(quote); + return new SetOutputDirectoryTool().execute({ directory, user_instruction: quote }, ctx()); +} + +describe("output location routing", () => { + it("writes to the workspace and reads external sources without confirmation", async () => { + await fs.writeFile(path.join(b, "old.txt"), "old report"); + expect(await new ReadFileTool({ workspace: a }).execute({ path: path.join(b, "old.txt") })).toContain("old report"); + expect(await new WriteFileTool({ workspace: a }).execute({ path: "nested/new.txt", content: "new" }, ctx())).toContain("Successfully"); + expect(await policy.checkDelivery(`[report](${a}/nested/new.txt) [reference](${b}/old.txt)`)).toEqual([]); + }); + it("rejects an external write before creating its parent and gives a usable destination", async () => { + const target = path.join(b, "nested", "new.txt"); + const stopTurn = vi.fn(); + const result = await new WriteFileTool({ workspace: a }).execute({ path: target, content: "new" }, { ...ctx(), stopTurn }); + expect(result).toContain("outside allowed directory"); expect(result).toContain(a); + expect(stopTurn).not.toHaveBeenCalled(); + await expect(fs.stat(path.dirname(target))).rejects.toThrow(); + }); + it("does not modify an external reference through edit_file", async () => { + const target = path.join(b, "old.txt"); await fs.writeFile(target, "old"); + expect(await new EditFileTool({ workspace: a }).execute({ path: target, old_text: "old", new_text: "new" }, ctx())).toContain("outside allowed directory"); + expect(await fs.readFile(target, "utf8")).toBe("old"); + }); + it("uses a user's external destination immediately and persists it only in this session", async () => { + const directory = path.join(b, "new folder"); + expect(await select(directory)).toContain("No additional confirmation"); + const target = path.join(directory, "new.txt"); + expect(await new WriteFileTool({ workspace: a }).execute({ path: target, content: "new" }, ctx())).toContain("Successfully"); + expect(await policy.checkDelivery(`[file](<${target}>)`)).toEqual([]); + expect(new OutputLocationPolicy(a, metadata, []).directory).toBe(directory); + expect(new OutputLocationPolicy(a, {}, []).directory).toBe(a); + expect(new OutputLocationPolicy(b, metadata, []).directory).toBe(b); + }); + it("rejects a remembered destination quote or a destination absent from the current input", async () => { + policy.addUserRequest("重新写报告"); + expect(await policy.selectDirectory(b, `请保存到 ${b}`)).toContain("current user's"); + expect(await policy.selectDirectory(b, "重新写报告")).toContain("does not match"); + expect(policy.directory).toBe(a); + }); + it("supports the user's named desktop folder without another question", async () => { + policy.addUserRequest("保存到桌面"); + expect(await policy.selectDirectory(path.join(os.homedir(), "Desktop"), "保存到桌面")).toContain("Output directory"); + }); + it("requires final files at the user-selected location, while allowing workspace intermediates", async () => { + await select(); + expect(await new WriteFileTool({ workspace: a }).execute({ path: "report.txt", content: "new", output_paths: ["report.txt"] }, ctx())).toContain("outside allowed directory"); + expect(await policy.checkDelivery("done")).toEqual([]); + await new WriteFileTool({ workspace: a }).execute({ path: path.join(b, "report.txt"), content: "new" }, ctx()); + expect(await policy.checkDelivery(`[report](${b}/report.txt)`)).toEqual([]); + }); + it("allows temporary intermediates, but rejects declaring them as final outputs", async () => { + const temporary = new OutputLocationPolicy(a, {}, [b]); + const target = path.join(b, "scratch.txt"); + expect(await temporary.checkWrite(target)).toBeNull(); + expect(await temporary.checkWrite(target, undefined, true)).toContain("outside allowed directory"); + expect(await temporary.checkDelivery(`[reference](${b}/unrelated.txt)`)).toEqual([]); + }); + it("catches a symlink escape, including a missing output below the link", async () => { + await fs.symlink(b, path.join(a, "outside")); + expect(await policy.checkWrite(path.join(a, "outside", "new", "file.txt"))).toContain("outside allowed directory"); + }); + it("rechecks the actual destination at delivery if a directory link was replaced", async () => { + const destination = path.join(root, "destination"); await fs.symlink(b, destination); + await select(destination); + const target = path.join(destination, "file.txt"); await fs.writeFile(target, "new"); policy.recordFile(target, true); + await fs.unlink(destination); await fs.symlink(a, destination); + expect(await policy.checkDelivery(`[file](${target})`)).toEqual([target]); + }); + it("rejects missing declared artifacts even if the final response omits links", async () => { + const target = path.join(a, "missing.txt"); policy.recordFile(target, true); + expect(await policy.checkDelivery("完成了")).toEqual([target]); + }); + it("accepts encoded file URLs and does not scan reference links inside document contents", async () => { + const target = path.join(a, "new report.md"); + await new WriteFileTool({ workspace: a }).execute({ path: target, content: `[reference](${b}/old.pdf)` }, ctx()); + expect(await policy.checkDelivery(`[file](${pathToFileURL(target).href})`)).toEqual([]); + expect(await policy.checkDelivery(`[file](<${target}>) [source](${b}/old.pdf)`)).toEqual([]); + }); + it("inherits the selected location and merges delegated final artifact lists", async () => { + await select(); const child = policy.fork(); + expect(child.directory).toBe(b); + child.recordFile(path.join(b, "missing.txt"), true); + expect(await policy.checkDelivery("done")).toEqual([path.join(b, "missing.txt")]); + }); + it("passes the output policy to spawned tasks", async () => { + await select(); const spawn = vi.fn().mockResolvedValue("task-id"); + const tool = new SpawnTool({ manager: { spawn } }); + await tool.execute({ task: "generate output" }, ctx()); + expect(spawn.mock.calls[0][0].outputLocation.directory).toBe(b); + }); + it("cancels before a write", async () => { + const controller = new AbortController(); controller.abort(); + await expect(new WriteFileTool({ workspace: a }).execute({ path: "file.txt", content: "new" }, { ...ctx(), abortSignal: controller.signal })).rejects.toMatchObject({ name: "AbortError" }); + await expect(fs.stat(path.join(a, "file.txt"))).rejects.toThrow(); + }); + it("tracks patch outputs for the same delivery check", async () => { + const tool = new ApplyPatchTool({ workspace: a }); + expect(await tool.execute({ input: "*** Begin Patch\n*** Add File: file.txt\n+new\n*** End Patch", output_paths: ["file.txt"] }, ctx())).toContain("Patch applied"); + await select(); + expect(await policy.checkDelivery(`[file](${a}/file.txt)`)).toEqual([path.join(b, "file.txt")]); + }); + it("tracks workspace-relative patch names with spaces as absolute delivery paths", async () => { + const target = path.join(a, "new report.txt"); + const tool = new ApplyPatchTool({ workspace: a }); + expect(await tool.execute({ input: "*** Begin Patch\n*** Add File: new report.txt\n+new\n*** End Patch", output_paths: [target] }, ctx())).toContain("Patch applied"); + expect(await fs.readFile(target, "utf8")).toBe("new\n"); + expect(await policy.checkDelivery(`[file](<${target}>)`)).toEqual([]); + await select(); + expect(await policy.checkDelivery(`[file](<${target}>)`)).toEqual([path.join(b, "new report.txt")]); + }); +}); + +describe("command preflight", () => { + it("rejects undeclared command outputs without starting a process", async () => { + const spawn = vi.spyOn(ExecTool, "spawnProcess"); + expect(await new ExecTool({ workspace: a }).execute({ command: "echo test" }, ctx())).toContain("declare output_paths"); + expect(spawn).not.toHaveBeenCalled(); + }); + it("rejects an external command output before execution", async () => { + const spawn = vi.spyOn(ExecTool, "spawnProcess"); + const target = path.join(b, "file.txt"); + expect(await new ExecTool({ workspace: a }).execute({ command: `echo test > '${target}'`, output_paths: [target] }, ctx())).toContain("outside allowed directory"); + expect(spawn).not.toHaveBeenCalled(); await expect(fs.stat(target)).rejects.toThrow(); + }); + it("runs a real workspace command and verifies its output", async () => { + expect(await new ExecTool({ workspace: a }).execute({ command: "echo new > file.txt", output_paths: ["file.txt"] }, ctx())).toContain("Exit code: 0"); + expect(await policy.checkDelivery(`[file](${a}/file.txt)`)).toEqual([]); + }); + it("uses a user-selected destination with a different command cwd", async () => { + await select(); const target = path.join(b, "file.txt"); + expect(await new ExecTool({ workspace: a }).execute({ command: "echo new > file.txt", cwd: b, output_paths: ["file.txt"] }, ctx())).toContain("Exit code: 0"); + expect(await fs.readFile(target, "utf8")).toContain("new"); + expect(await policy.checkDelivery(`[file](${target})`)).toEqual([]); + }); + it("checks declared artifacts after a managed command", async () => { + const tool = new ExecTool({ workspace: a }); + await tool.execute({ command: "echo new > file.txt", output_paths: ["file.txt"], yield_time_ms: 1000 }, ctx()); + expect(await policy.checkDelivery(`[file](${a}/file.txt)`)).toEqual([]); + }); +}); + +class ScriptedProvider extends LLMProvider { + calls: any[] = []; + constructor(private responses: LLMResponse[]) { super(); } + getDefaultModel() { return "test-model"; } + async chat(args: any) { return this.chatWithRetry(args); } + async chatWithRetry(args: any) { this.calls.push(args); const response = this.responses.shift(); if (!response) throw new Error("Unexpected model request"); return response; } +} +let sequence = 0; +const call = (name: string, args: Record) => new LLMResponse({ content: null, toolCalls: [new ToolCallRequest({ id: `call-${sequence++}`, name, arguments: args })], finishReason: "tool_calls" }); +const reply = (content: string) => new LLMResponse({ content }); +function registry() { const tools = new ToolRegistry(); tools.register(new WriteFileTool({ workspace: a })); tools.register(new SetOutputDirectoryTool()); return tools; } +async function run(responses: LLMResponse[], options: Record = {}) { + const provider = new ScriptedProvider(responses); + const result = await new AgentRunner(provider).run(new AgentRunSpec({ workspace: a, outputLocation: policy, tools: registry(), messages: [{ role: "user", content: "write a file" }], ...options })); + return { provider, result }; +} + +describe("Runner delivery boundary", () => { + it("recovers from an external write using the workspace without a permission exchange", async () => { + const { result } = await run([call("write_file", { path: path.join(b, "file.txt"), content: "new" }), call("write_file", { path: "file.txt", content: "new" }), reply(`[file](${a}/file.txt)`)]); + expect(result.stopReason).toBe("completed"); expect(await fs.readFile(path.join(a, "file.txt"), "utf8")).toBe("new"); + await expect(fs.stat(path.join(b, "file.txt"))).rejects.toThrow(); + }); + it("takes a requested directory and delivers there in the same turn", async () => { + const quote = `保存到 ${b}`; policy.addUserRequest(quote); + const { result } = await run([call("set_output_directory", { directory: b, user_instruction: quote }), call("write_file", { path: path.join(b, "file.txt"), content: "new" }), reply(`[file](${b}/file.txt)`)]); + expect(result.stopReason).toBe("completed"); expect(result.error).toBeNull(); + }); + it("returns a bounded error for repeated invalid delivery, without persisting false success", async () => { + const missing = path.join(a, "missing.txt"); policy.recordFile(missing, true); + const { result, provider } = await run([reply("完成"), reply("完成"), reply("完成")]); + expect(result.stopReason).toBe("toolError"); expect(provider.calls).toHaveLength(3); + expect(result.messages.some(message => message.role === "assistant" && message.content === "完成")).toBe(false); + }); + it("lets the model correct a delivery after verification fails", async () => { + const missing = path.join(a, "missing.txt"); policy.recordFile(missing, true); + const { result } = await run([reply(`[file](${missing})`), call("write_file", { path: "missing.txt", content: "new" }), reply(`[file](${missing})`)]); + expect(result.stopReason).toBe("completed"); expect(result.finalContent).toContain(missing); + }); + it("also verifies the max-iteration finalization response", async () => { + const missing = path.join(a, "missing.txt"); policy.recordFile(missing, true); + const { result } = await run([call("write_file", { path: "scratch.txt", content: "intermediate" }), reply(`[file](${missing})`)], { maxIterations: 1, maxIterationsFinalPrompt: "give a final response" }); + expect(result.stopReason).toBe("toolError"); expect(result.finalContent).toContain("核验未通过"); + expect(result.messages.some(message => message.role === "assistant" && message.content === `[file](${missing})`)).toBe(false); + }); + it("accepts a live steering request to change the directory", async () => { + const quote = `保存到 ${b}`; let pending = true; + const { result } = await run([call("write_file", { path: "scratch.txt", content: "scratch" }), call("set_output_directory", { directory: b, user_instruction: quote }), call("write_file", { path: path.join(b, "file.txt"), content: "new" }), reply(`[file](${b}/file.txt)`)], { injectionCallback: () => { if (!pending) return []; pending = false; return [{ role: "user", content: quote, output_user_instruction: quote }]; } }); + expect(result.stopReason).toBe("completed"); expect(policy.directory).toBe(b); + }); + it("does not treat a generated follow-up as a user destination request", async () => { + const quote = `保存到 ${b}`; let pending = true; + const { result, provider } = await run([call("write_file", { path: "scratch.txt", content: "scratch" }), call("set_output_directory", { directory: b, user_instruction: quote }), reply("已参考旧任务，当前输出目录保持工作区")], { injectionCallback: () => { if (!pending) return []; pending = false; return [{ role: "user", content: quote }]; } }); + expect(result.stopReason).toBe("completed"); expect(policy.directory).toBe(a); + expect(provider.calls.at(-1).messages.some((message: any) => message.role === "tool" && message.content.includes("current user's destination"))).toBe(true); + }); +}); + +describe("AgentLoop output routing", () => { + function loop(responses: LLMResponse[]) { + return new AgentLoop({ bus: new MessageBus(), provider: new ScriptedProvider(responses), + config: new Config({ fileMemory: { enabled: false }, memmyMemory: { enabled: false } }), + workspace: a, model: "test-model", sessionDir: path.join(root, "sessions"), maxIterations: 8 }); + } + it("uses live user input for an external destination and saves it for the next turn", async () => { + const quote = `请保存到 ${b}`; + const target = path.join(b, "report.txt"); + const agent = loop([call("set_output_directory", { directory: b, user_instruction: quote }), + call("write_file", { path: target, content: "new" }), reply(`[report](${target})`), + call("write_file", { path: target, content: "revised" }), reply(`[report](${target})`)]); + const first = await agent.processDirect(quote, { sessionKey: "cli:output-test" }); + expect(first?.content).toContain(target); + expect(await fs.readFile(target, "utf8")).toBe("new"); + const second = await agent.processDirect("修改报告", { sessionKey: "cli:output-test" }); + expect(second?.content).toContain(target); + expect(await fs.readFile(target, "utf8")).toBe("revised"); + }); + it("routes a bus user request through the same policy without a confirmation turn", async () => { + const quote = `请保存到 ${b}`; const target = path.join(b, "report.txt"); + const agent = loop([call("set_output_directory", { directory: b, user_instruction: quote }), + call("write_file", { path: target, content: "new" }), reply(`[report](${target})`)]); + await agent.dispatchMessage(new InboundMessage({ channel: "telegram", chatId: "output-test", senderId: "test-user", content: quote })); + expect(await fs.readFile(target, "utf8")).toBe("new"); + const outbound = []; + while (agent.bus.outboundSize) outbound.push(await agent.bus.consumeOutbound()); + expect(outbound.some(message => message.content.includes("需要外部写入权限"))).toBe(false); + }); +});