Menu
popagent
publicLatest change 324ce6d82a1e3e0f842da61c5c6109199a026104 - Control internal agent memory lifecycle by Ólafur Búi Ólafsson
import { describe, expect, test } from "bun:test";
import type { Agent, IterationCompleteContext } from "@mastra/core/agent";
import {
type AgentRuntimeSettings,
type AgentTask,
type AgentTaskProgress,
type MemorySettings,
} from "./api-types";
import { agentActivity, AgentExecutionRuntime, createAgentTaskExecutor, observationalMemoryConfig } from "./agent-execution";
import { HookRuntime, type HookEvent } from "./hooks";
import type { SelfUpdateRun } from "./self-update-policy";
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",
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: "popagent",
agentName: "popagent",
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);
const output = await createAgentTaskExecutor(agent, execution)(
task,
controller.signal,
"turn-1",
traceId,
async (progress) => { progressUpdates.push(progress); },
);
expect(output).toBe("task result");
expect(progressUpdates).toEqual([
{ stepsCompleted: 2, progress: "Using webSearch" },
{ stepsCompleted: 4, progress: "Researcher: Using 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",
}),
}),
}));
expect((streamedOptions?.requestContext as { get: (key: string) => unknown }).get("popagent.sessionId"))
.toBe("session-1");
expect((streamedOptions?.requestContext as { get: (key: string) => unknown }).get("popagent.model"))
.toBe(runtimeConfiguration.defaultModel);
expect((streamedOptions?.requestContext as { get: (key: string) => unknown }).get("popagent.executionSource"))
.toBe("task");
expect((streamedOptions?.requestContext as { get: (key: string) => unknown }).get("popagent.taskId"))
.toBe("task-1");
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("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("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>(),
} 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("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: [],
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"],
});
});
});