Menu
popagent
publicLatest change fa391bb84299efc38430196d6f7156e2c978d1b7 - Respect the autonomy pause in follow-ups and refresh scheduled workspaces by AkurAI Build
import { describe, expect, test } from "bun:test";
import type { Agent, IterationCompleteContext } from "@mastra/core/agent";
import {
type AgentRuntimeSettings,
type AgentExecutionSource,
type AgentTask,
type AgentTaskProgress,
type MemorySettings,
} from "./api-types";
import { agentActivity, AgentExecutionRuntime, conversationWithoutUnresolvedTail, createAgentTaskExecutor, drainLiveActivity, iterationActivity, LIVE_ACTIVITY_INTERVAL_MS, observationalMemoryConfig } from "./agent-execution";
import { HookRuntime, type HookEvent } from "./hooks";
import type { SelfUpdateRun } from "./self-update-policy";
import { TaskAgentCommunicationSession } from "./task-agent-communication";
import { createAgentRequestContext } from "./agent-context";
import type { TaskAgentCommunicationMessage } from "./task-agent-communication";
const memoryConfiguration: MemorySettings = {
autoCompact: false,
observationTokens: 30_000,
reflectionTokens: 40_000,
recentMessagePercent: 20,
asyncBuffering: true,
bufferIntervalPercent: 20,
bufferOnIdle: true,
observationBlockPercent: 120,
reflectionBufferPercent: 50,
reflectionBlockPercent: 120,
optimizeObserverContext: true,
previousObserverTokens: 2_000,
retrievalEnabled: true,
retrievalScope: "resource",
temporalMarkers: false,
activateAfterIdle: "off",
activateOnProviderChange: false,
shareTokenBudget: false,
observeAttachments: "auto",
observationInstruction: "",
reflectionInstruction: "",
internalRecall: true,
internalRetention: true,
updatedAt: "",
};
const runtimeConfiguration: AgentRuntimeSettings = {
defaultModel: "test/default",
modelSource: "configured",
modelRouting: { enabled: false, strategy: "fallback", cooldownMs: 900_000, routes: [] },
supervisorMaxSteps: 9,
specialistMaxSteps: 7,
toolConcurrency: 2,
delegationContextMessages: 5,
delegationResultCharacters: 4_000,
maxProcessorRetries: 4,
finalResponseFeedback: "Return the configured final response",
delegationFailureFeedback: "Delegate {{agentId}} failed: {{error}}",
delegationResultTruncationMarker: "\n[truncated]",
taskConcurrency: 2,
taskPollIntervalMs: 1_000,
taskTimeoutMs: 30_000,
taskStaleAfterMs: 60_000,
updatedAt: "",
};
describe("AgentExecutionRuntime", () => {
test("maps persisted advanced observational-memory settings into Mastra config", () => {
const config = observationalMemoryConfig({
...memoryConfiguration,
autoCompact: true,
recentMessagePercent: 25,
bufferIntervalPercent: 30,
observationBlockPercent: 135,
reflectionBufferPercent: 60,
reflectionBlockPercent: 145,
previousObserverTokens: 3_500,
retrievalScope: "thread",
temporalMarkers: true,
activateAfterIdle: "auto",
activateOnProviderChange: true,
observeAttachments: "none",
observationInstruction: " Keep decisions. ",
reflectionInstruction: " Preserve blockers. ",
}, "test/default");
expect(config).toEqual({
observationalMemory: expect.objectContaining({
enabled: true,
scope: "thread",
activateAfterIdle: "auto",
activateOnProviderChange: true,
temporalMarkers: true,
retrieval: expect.objectContaining({ scope: "thread" }),
observation: {
messageTokens: 30_000,
bufferTokens: 0.3,
bufferActivation: 0.75,
bufferOnIdle: true,
blockAfter: 1.35,
previousObserverTokens: 3_500,
observeAttachments: false,
instruction: "Keep decisions.",
},
reflection: {
observationTokens: 40_000,
bufferActivation: 0.6,
blockAfter: 1.45,
instruction: "Preserve blockers.",
},
}),
});
});
test("disables async buffering when token budgets are shared", () => {
const config = observationalMemoryConfig({
...memoryConfiguration,
autoCompact: true,
shareTokenBudget: true,
}, "test/default");
expect(config.observationalMemory.observation).toEqual(expect.objectContaining({
bufferTokens: false,
bufferOnIdle: false,
}));
expect(config.observationalMemory.reflection).not.toHaveProperty("bufferActivation");
});
test("gives background tasks the chat execution context", async () => {
const events: HookEvent[] = [];
const episodes: Array<{ sessionId: string; userText: string; assistantText: string }> = [];
const progressUpdates: AgentTaskProgress[] = [];
let appliedAgent: Agent | undefined;
let streamedPrompt: string | undefined;
let streamedOptions: Record<string, unknown> | undefined;
const hooks = new HookRuntime({
config: {
version: 1,
hooks: {
UserPromptSubmit: [{ id: "normalize", type: "http", url: "https://hooks.example/normalize" }],
},
},
transport: async (_handler, event) => {
events.push(event);
return {
schemaVersion: 1,
outcome: "modify",
reason: "normalize",
replacement: "normalized task prompt",
additionalContext: "managed hook context",
};
},
});
const agent = {
stream: async (prompt: string, options: Record<string, unknown>) => {
streamedPrompt = prompt;
streamedOptions = options;
await (options.onIterationComplete as (
context: IterationCompleteContext,
) => Promise<unknown>)({
iteration: 2,
maxIterations: runtimeConfiguration.supervisorMaxSteps,
text: "",
toolCalls: [{ id: "tool-1", name: "webSearch", args: {} }],
toolResults: [],
isFinal: false,
finishReason: "tool-calls",
runId: "run-1",
agentId: "orchistrator",
agentName: "orchistrator",
messages: [],
});
const observeSpecialist = (
options.requestContext as { get: (key: string) => unknown }
).get("popagent.iterationObserver") as (context: IterationCompleteContext) => Promise<void>;
await observeSpecialist({
iteration: 4,
maxIterations: 12,
text: "",
toolCalls: [{ id: "tool-2", name: "webSearch", args: {} }],
toolResults: [],
isFinal: false,
finishReason: "tool-calls",
runId: "run-2",
agentId: "researcher",
agentName: "Researcher",
messages: [],
});
await (options.onFinish as (result: { text: string }) => Promise<void>)({ text: "task result" });
return { text: Promise.resolve("task result") };
},
} as unknown as Agent;
const execution = new AgentExecutionRuntime(
agent,
hooks,
{
formatRecall: async ({ resourceId, query }) => {
expect(resourceId).toBe("popagent-user");
expect(query).toBe("normalized task prompt");
return "recalled durable fact";
},
retainEpisode: async ({ sessionId, userText, assistantText }) => {
episodes.push({ sessionId, userText, assistantText });
},
},
{
apply: async (candidate) => { appliedAgent = candidate; },
},
{
get: async () => memoryConfiguration,
},
{
get: async () => runtimeConfiguration,
},
);
const task: AgentTask = {
id: "task-1",
sessionId: "session-1",
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "original task prompt",
model: runtimeConfiguration.defaultModel,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const controller = new AbortController();
const traceId = "a".repeat(32);
let communicationSession!: TaskAgentCommunicationSession;
const output = await createAgentTaskExecutor(agent, execution, undefined, (claimedTask, signal) => {
communicationSession = new TaskAgentCommunicationSession({
taskId: claimedTask.id,
workspaceId: claimedTask.workspaceId,
resourceId: "popagent-user",
participants: ["orchistrator", "researcher"],
signal,
});
return communicationSession;
})(
task,
controller.signal,
"turn-1",
traceId,
async (progress) => { progressUpdates.push(progress); },
);
expect(output).toBe("task result");
expect(progressUpdates).toEqual([
expect.objectContaining({
stepsCompleted: 2,
progress: "Using webSearch",
activity: expect.objectContaining({ agentName: "orchistrator", tools: ["webSearch"] }),
}),
expect.objectContaining({
stepsCompleted: 4,
progress: "Researcher: Using webSearch",
activity: expect.objectContaining({ agentName: "Researcher", tools: ["webSearch"] }),
}),
]);
expect(streamedPrompt).toBe("normalized task prompt");
expect(appliedAgent).toBe(agent);
expect(events).toEqual([expect.objectContaining({
eventName: "UserPromptSubmit",
sessionId: "session-1",
detail: { prompt: "original task prompt" },
})]);
expect(streamedOptions).toEqual(expect.objectContaining({
abortSignal: controller.signal,
memory: { thread: "session-1", resource: "popagent-user" },
hooks: expect.objectContaining({
beforeToolCall: expect.any(Function),
afterToolCall: expect.any(Function),
}),
delegation: expect.objectContaining({
onDelegationStart: expect.any(Function),
onDelegationComplete: expect.any(Function),
}),
outputProcessors: [expect.anything()],
errorProcessors: [expect.anything()],
memoryConfig: { observationalMemory: { enabled: false } },
maxSteps: runtimeConfiguration.supervisorMaxSteps,
toolCallConcurrency: { limit: runtimeConfiguration.toolConcurrency, strategy: "called" },
tracingOptions: expect.objectContaining({
traceId,
tags: ["task"],
metadata: expect.objectContaining({
sessionId: "session-1",
turnId: "turn-1",
workspaceId: "default",
executionSource: "task",
taskId: "task-1",
}),
}),
}));
const streamedRequestContext = streamedOptions?.requestContext;
if (
!streamedRequestContext
|| typeof streamedRequestContext !== "object"
|| !("get" in streamedRequestContext)
|| typeof streamedRequestContext.get !== "function"
) {
throw new Error("Expected a Mastra request context");
}
expect(streamedRequestContext.get("popagent.sessionId")).toBe("session-1");
expect(streamedRequestContext.get("popagent.model")).toBe(runtimeConfiguration.defaultModel);
expect(streamedRequestContext.get("popagent.executionSource")).toBe("task");
expect(streamedRequestContext.get("popagent.taskId")).toBe("task-1");
expect(streamedRequestContext.get("popagent.communicationSession")).toBe(communicationSession);
expect(await communicationSession.send({
from: "orchistrator",
to: "researcher",
body: "late",
})).toMatchObject({ outcome: "failed", error: expect.stringContaining("closed") });
expect(streamedOptions?.context).toEqual([
{ role: "user", content: "<memory-context>\nmanaged hook context\n</memory-context>" },
{ role: "user", content: "<memory-context>\nrecalled durable fact\n</memory-context>" },
]);
expect(episodes).toEqual([{
sessionId: "session-1",
userText: "normalized task prompt",
assistantText: "task result",
}]);
});
test("runs an idle addressed participant and records its identity-bound reply", async () => {
const persisted: TaskAgentCommunicationMessage[] = [];
const reviewerPrompts: string[] = [];
let communicationSession!: TaskAgentCommunicationSession;
const reviewer = {
__fork() { return this; },
setBrowser() {},
async stream(prompt: string) {
reviewerPrompts.push(prompt);
return { text: Promise.resolve("") };
},
} as unknown as Agent;
const main = {
async stream() {
const receipt = await communicationSession.send({
from: "implementer",
to: "reviewer",
body: "Review the completed change.",
});
expect(receipt.outcome).toBe("woken");
return { text: Promise.resolve("Main task complete.") };
},
} as unknown as Agent;
const execution = {
async prepare(input: {
communicationSession?: TaskAgentCommunicationSession;
executionSource?: AgentExecutionSource;
selfUpdateWorkspacePath?: string;
prompt: string;
}) {
input.communicationSession?.configureExecution({
executionSource: input.executionSource ?? "task",
...(input.selfUpdateWorkspacePath
? { selfUpdateWorkspacePath: input.selfUpdateWorkspacePath }
: {}),
});
return {
prompt: input.prompt,
options: {
requestContext: createAgentRequestContext({
resourceId: "popagent-user",
sessionId: "task:wake",
turnId: "turn-wake",
model: runtimeConfiguration.defaultModel,
workspaceId: "default",
executionSource: input.executionSource ?? "task",
runtimeSettings: runtimeConfiguration,
communicationSession: input.communicationSession,
}),
},
};
},
} as unknown as AgentExecutionRuntime;
const task: AgentTask = {
id: "task-wake",
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Implement and review",
model: runtimeConfiguration.defaultModel,
status: "running",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 1,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: "",
completedAt: null,
};
const execute = createAgentTaskExecutor(
main,
execution,
undefined,
(claimedTask, signal, wake) => {
communicationSession = new TaskAgentCommunicationSession({
taskId: claimedTask.id,
workspaceId: claimedTask.workspaceId,
resourceId: "popagent-user",
participants: ["implementer", "reviewer"],
signal,
wake,
persist: async (message) => { persisted.push(message); },
});
return communicationSession;
},
{ reviewer },
);
expect(await execute(
task,
new AbortController().signal,
"turn-wake",
"d".repeat(32),
async () => undefined,
)).toBe("Main task complete.");
expect(reviewerPrompts).toEqual([
expect.stringContaining('Reply as yourself to "implementer"'),
]);
expect(persisted).toEqual([
expect.objectContaining({ from: "implementer", to: "reviewer" }),
expect.objectContaining({
from: "reviewer",
to: "implementer",
body: "No additional findings.",
replyTo: persisted[0]?.id,
}),
]);
});
test("honors internal recall and retention settings for background tasks", async () => {
let recallCalls = 0;
const episodes: string[] = [];
const runtime = new AgentExecutionRuntime(
{ } as Agent,
{ dispatch: async () => ({ additionalContext: [] }) } as unknown as HookRuntime,
{
formatRecall: async () => { recallCalls++; return "unexpected"; },
retainEpisode: async () => { episodes.push("retained"); },
},
{ apply: async () => {} },
{ get: async () => ({ ...memoryConfiguration, internalRecall: false, internalRetention: false }) },
{ get: async () => runtimeConfiguration },
);
const prepared = await runtime.prepare({
sessionId: "internal-session",
turnId: "turn-1",
model: runtimeConfiguration.defaultModel,
workspaceId: "default",
executionSource: "task",
prompt: "background work",
traceId: "b".repeat(32),
});
await prepared.options.onFinish({ text: "done" });
expect(recallCalls).toBe(0);
expect(prepared.options.context).toEqual([]);
expect(episodes).toEqual([]);
});
test("injects deterministic repository evidence before the model runs", async () => {
const runtime = new AgentExecutionRuntime(
{ } as Agent,
{ dispatch: async () => ({ additionalContext: [] }) } as unknown as HookRuntime,
{ formatRecall: async () => "", retainEpisode: async () => {} },
{ apply: async () => {} },
{ get: async () => memoryConfiguration },
{ get: async () => runtimeConfiguration },
{
build: async () => ({
repository: { clean: true, changedFiles: [] },
candidates: [{ path: "src/payment.ts", score: 60, symbols: [] }],
tests: ["bun test src/payment.test.ts"],
readyForImplementation: true,
gaps: [],
}),
},
);
const prepared = await runtime.prepare({
sessionId: "session-1",
turnId: "turn-1",
model: runtimeConfiguration.defaultModel,
workspaceId: "default",
executionSource: "chat",
prompt: "Fix payment",
traceId: "c".repeat(32),
});
expect(prepared.options.context).toEqual(expect.arrayContaining([
expect.objectContaining({ content: expect.stringContaining("<repository-brief>") }),
]));
expect((prepared.options.requestContext as { get: (key: string) => unknown })
.get("popagent.repositoryBriefReady")).toBe(true);
});
test("refreshes the workspace checkout before a scheduled run reads it", async () => {
const order: string[] = [];
const runtime = new AgentExecutionRuntime(
{ } as Agent,
{ dispatch: async () => ({ additionalContext: [] }) } as unknown as HookRuntime,
{ formatRecall: async () => "", retainEpisode: async () => {} },
{ apply: async () => {} },
{ get: async () => memoryConfiguration },
{ get: async () => runtimeConfiguration },
{
build: async () => {
order.push("brief");
return { repository: { clean: true, changedFiles: [] }, candidates: [], tests: [], readyForImplementation: false, gaps: [] };
},
},
async (workspaceId: string) => { order.push(`sync:${workspaceId}`); },
);
const input = {
sessionId: "session-1",
turnId: "turn-1",
model: runtimeConfiguration.defaultModel,
workspaceId: "default",
prompt: "Prepare the daily report",
traceId: "d".repeat(32),
};
await runtime.prepare({ ...input, executionSource: "task", scheduleId: "schedule-1" });
expect(order).toEqual(["sync:default", "brief"]);
order.length = 0;
await runtime.prepare({ ...input, executionSource: "chat" });
expect(order).toEqual(["brief"]);
});
test("derives trusted self-update execution source before ordinary schedule lineage", async () => {
const prepared: Array<Record<string, unknown>> = [];
const accepted: string[] = [];
const run = {
taskId: "self-update-task",
workspaceId: "default",
repositoryPath: "/registered",
path: "/contained/self-update",
branch: "autonomous/self-update/test",
snapshot: { head: "a".repeat(40), branch: "main", remotes: "remote-hash" },
cloneRemotes: "clone-remote-hash",
delegations: new Set<string>(),
deploymentEnabled: false,
} satisfies SelfUpdateRun;
const execution = {
async prepare(input: Record<string, unknown>) {
prepared.push(input);
return { prompt: input.prompt as string, options: {} };
},
} as unknown as AgentExecutionRuntime;
const agent = {
__fork() { return this; },
setBrowser() {},
async stream() { return { text: Promise.resolve("done") }; },
} as unknown as Agent;
const policy = {
async begin() { return run; },
observeDelegation() {},
async accept(candidate: SelfUpdateRun) { accepted.push(candidate.taskId); },
async reject() {},
};
const baseTask: AgentTask = {
id: "scheduled-user-task",
sessionId: null,
scheduleId: "user-schedule",
source: "user",
workspaceId: "default",
prompt: "scheduled",
model: "test/default",
status: "running",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 1,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: "",
completedAt: null,
};
const executeTask = createAgentTaskExecutor(agent, execution, policy);
await executeTask(baseTask, new AbortController().signal, "turn-user", "b".repeat(32), async () => undefined);
await executeTask(
{ ...baseTask, id: run.taskId, source: "self-update" },
new AbortController().signal,
"turn-self-update",
"c".repeat(32),
async () => undefined,
);
expect(prepared.map((input) => input.executionSource)).toEqual(["schedule", "self-update"]);
expect(prepared[1]).toEqual(expect.objectContaining({
selfUpdateWorkspacePath: run.path,
scheduleId: "user-schedule",
}));
expect(accepted).toEqual([run.taskId]);
});
test("describes workspace file reads without exposing internal tool names", () => {
expect(iterationActivity({
iteration: 1,
maxIterations: 12,
text: "",
toolCalls: [{
id: "tool-1",
name: "mastra_workspace_read_file",
args: { path: "src/agent.ts" },
}],
toolResults: [],
isFinal: false,
finishReason: "tool-calls",
runId: "run-1",
agentId: "researcher",
agentName: "Researcher",
messages: [],
})).toBe("Researcher: Reading File: src/agent.ts");
});
test("serializes specialist iterations for live chat activity", () => {
const activity = agentActivity("turn-7", {
iteration: 3,
maxIterations: 12,
text: "Verified the repository contract.",
toolCalls: [
{ id: "tool-1", name: "read", args: { path: "src" } },
{ id: "tool-2", name: "read", args: { path: "test" } },
],
toolResults: [
{ id: "tool-1", name: "read", result: { files: ["src/agent.ts"] } },
{ id: "tool-2", name: "read", result: null, error: new Error("Path denied") },
],
isFinal: true,
finishReason: "stop",
runId: "run-7",
agentId: "reviewer",
agentName: "Reviewer",
messages: [],
});
expect(activity).toEqual({
turnId: "turn-7",
runId: "run-7",
agentId: "reviewer",
agentName: "Reviewer",
iteration: 3,
maxIterations: 12,
isFinal: true,
finishReason: "stop",
text: "Verified the repository contract.",
tools: ["read"],
toolCalls: [
{
id: "tool-1",
name: "read",
args: { path: "src" },
result: { files: ["src/agent.ts"] },
status: "complete",
},
{
id: "tool-2",
name: "read",
args: { path: "test" },
result: null,
error: "Path denied",
status: "error",
},
],
});
});
});
describe("conversationWithoutUnresolvedTail", () => {
test("drops the unresolved assistant tail left by an exhausted step budget", () => {
const messages = [
{ role: "user", content: "prepare the report" },
{ role: "assistant", content: [{ type: "tool-call", toolCallId: "call-1" }] },
{ role: "tool", content: [{ type: "tool-result", toolCallId: "call-1" }] },
{ role: "assistant", content: [{ type: "tool-call", toolCallId: "call-2" }] },
];
expect(conversationWithoutUnresolvedTail(messages)).toEqual(messages.slice(0, 3));
});
test("keeps a conversation that already ends on resolved evidence", () => {
const messages = [
{ role: "user", content: "prepare the report" },
{ role: "tool", content: [{ type: "tool-result", toolCallId: "call-1" }] },
];
expect(conversationWithoutUnresolvedTail(messages)).toEqual(messages);
});
});
describe("drainLiveActivity", () => {
const base = { turnId: "turn-1", agentId: "orchistrator", agentName: "Orchistrator", maxIterations: 12 };
async function* chunks(items: { type: string; payload?: unknown }[]) {
for (const item of items) yield item;
}
test("reports a tool call before the iteration finishes", async () => {
const seen: { iteration: number; tools: string[]; text: string }[] = [];
await drainLiveActivity(chunks([
{ type: "step-start", payload: {} },
{ type: "tool-call", payload: { toolName: "skill", toolCallId: "call-1" } },
{ type: "tool-call", payload: { toolName: "skill", toolCallId: "call-2" } },
{ type: "tool-call", payload: { toolName: "codeContext", toolCallId: "call-3" } },
]), {
...base,
now: () => 0,
report: (activity) => { seen.push({ iteration: activity.iteration, tools: activity.tools, text: activity.text }); },
});
expect(seen.map((entry) => entry.tools)).toEqual([["skill"], ["skill", "codeContext"]]);
expect(seen.every((entry) => entry.iteration === 1)).toBe(true);
});
test("throttles streamed text and resets state on the next step", async () => {
const seen: string[] = [];
let clock = 0;
await drainLiveActivity(chunks([
{ type: "step-start", payload: {} },
{ type: "text-delta", payload: { text: "Prep" } },
{ type: "text-delta", payload: { text: "aring" } },
{ type: "step-start", payload: {} },
{ type: "text-delta", payload: { text: "Second" } },
]), {
...base,
now: () => (clock += LIVE_ACTIVITY_INTERVAL_MS),
report: (activity) => { seen.push(activity.text); },
});
expect(seen).toEqual(["Prep", "Preparing", "Second"]);
});
});