feat(core): run agentic loop on the evented workflow engine (#16796)

## Summary

- Migrates the agent's internal agentic loop onto the evented workflow
engine. Each LLM call and tool call now runs as a workflow step, sharing
suspend/resume and observability with the rest of the workflow system.
No user-visible behavior change yet — this is the structural
prerequisite for future durability and event-based observability
features.
- Fixes a nested-workflow bug where `resumeLabel` wasn't propagated up
through a parent's suspendedPaths, so resuming a nested workflow by
label silently failed.


🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
ELI5: This change moves the agent's internal loop to run as a series of
steps in the evented workflow engine so each model call and tool call is
tracked like a tiny workflow. That makes suspend/resume, observability,
and future durability much easier and also fixes a bug that broke
resuming nested workflows by label.

- What it does
- Migrates the agentic loop so each LLM and tool call executes as
evented workflow steps (uses createEventedWorkflow /
createEventedWorkflow factory).
- Agents can run without a user-provided Mastra by creating a
lazily-initialized ephemeral Mastra (InMemoryStore + EventEmitterPubSub)
and starting ephemeral workers on first use.
- Adds a per-run runScope (PrepareStreamRunScope) to thread
non-JSON-safe runtime state (MessageList, processorStates,
convertedTools, loopOptions) across prepare-stream workflow steps
instead of passing those objects through step inputs/outputs.
- Introduces runId-scoped internal workflow registration/unregistration
to avoid collisions for concurrent/nested runs.

- Bug fixes / behavioral corrections
- Fixes nested-workflow resume-by-label: resumeLabel is propagated
through parent suspendedPaths and nested suspend metadata so resuming a
nested workflow by label works reliably.
- Preserves and aggregates suspendPayloads (not just resumeLabels)
across suspended foreach iterations and when storage persistence is
skipped/absent.
- Ensures persisted snapshots are retained for pending/paused/suspended
states (not only suspended).
- Tightens internal workflow lookup to prefer runId-scoped internal
workflow instances, preventing wrong closure/registration resolution in
nested/concurrent cases.

- Implementation details / notable API surface changes
- New helper: createEventedWorkflow(...) exported from workflows
entrypoint (throws if evented factory not loaded).
  - Mastra changes:
- Defaults to InMemoryStore when storage not provided (with warning) and
augments the store.
- Internal workflow registry supports runId-scoped keys; added
__unregisterInternalWorkflow and updated __registerInternalWorkflow /
__hasInternalWorkflow / __getInternalWorkflow signatures.
  - Evented processor and execution engine:
- Nested workflow resolution now passes runId to getNestedWorkflow and
resolveWorkflow.
- Workflow event processor rebuilt resumeLabels and suspend payload
propagation; adjusted foreach aggregation and merge behavior when
storage updates are absent.
- In-memory workflow storage replaced JSON-serialize deep-copy with a
cycle-safe deep clone preserving Dates, Errors, Maps/Sets, prototypes,
typed arrays, etc.

- Tests / tooling
- Added suspend/resume test to validate resuming a suspended nested
workflow step by label and updated many tests to run against an evented
Mastra (InMemoryStore + EventEmitterPubSub) with worker lifecycle
management helpers (createTestMastra, start/stop workers).
- Multiple test updates to use real InMemoryStore + evented pubsub
instead of mocked storage or missing worker lifecycles.

- Minor / UX
- Updated background-task generated system prompt to clarify _background
behavior and placeholder result guidance.
- stream/loop runtime: distinguishes "pre-run" queued signals from
normal "pending" signals and exposes drainPendingSignals(runId, scope?)
with scope 'pending'|'pre-run'; stream lifecycle registers/unregisters
run-scoped internal workflows to prevent leakage on suspend.

- Risk / review notes
- Structural and high-impact refactor touching core agent loop, workflow
engine integration, storage cloning, and Mastra internals—review effort:
high in multiple areas (agent, workflows, mastra, storage, evented
processor).

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/mastra-ai/mastra/pull/16796?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Mastra Code (anthropic/claude-opus-4-6) <noreply@mastra.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Abhi Aiyer <abhi@mastra.ai>
Co-authored-by: Abhi Aiyer <abhiaiyer91@gmail.com>

Author: 6 people

.changeset/good-parents-help.md

@@ -0,0 +1,7 @@
+---
+'@mastra/core': patch
+---
+
+- Use evented-workflow engine in the internal agentic loop.
+- Fix nested workflow resume with resumeLabel in evented-workflow engine.
+- Default to an in-memory store when no `storage` is configured on `Mastra`, and warn that it is not durable and a persistent storage adapter should be used in production.

.changeset/silly-insects-do.md

@@ -0,0 +1,5 @@
+---
+'@mastra/pg': patch
+---
+
+Make atomic db updates better

.changeset/some-bees-work.md

@@ -0,0 +1,9 @@
+---
+'@mastra/libsql': patch
+---
+
+Fixed concurrent writes silently disappearing when using LibSQL with a local (`file:`) database.
+
+LibSQL backs a local database with a single connection. When one operation held an interactive write transaction (for example, persisting workflow snapshots) and another operation wrote at the same time (for example, creating a dataset experiment), the second write could be swept into the open transaction and rolled back — so it appeared to succeed but never persisted. This surfaced as concurrent agent/workflow runs losing unrelated records.
+
+Writes on a LibSQL client are now serialized, so a write issued during an in-flight transaction no longer interleaves with it.

e2e-tests/client-js/_test-utils/src/server-setup.ts

@@ -104,10 +104,16 @@ export function createTestServerSetup(config: TestServerSetupConfig) {
     const port = await getPort();
     const baseUrl = `http://localhost:${port}`;
 
-    // Create storage
+    // Create storage. LibSQL's `:memory:` URL gives each pooled connection its own
+    // private in-memory database, so a table created by one connection is invisible
+    // to the next — which the evented agentic-loop tickles by spreading workflow
+    // snapshot reads/writes across multiple operations. Use a per-process temp file
+    // so all pooled connections open the same on-disk database (mirrors the vector
+    // store treatment already in place below).
+    const libSqlDbPath = `file:${join(tmpdir(), `mastra-libsql-${randomUUID()}.db`)}`;
     const libSqlStore = new LibSQLStore({
       id: storageId,
-      url: ':memory:',
+      url: libSqlDbPath,
     });
 
     const inMemoryStore = new InMemoryStore({

e2e-tests/monorepo/monorepo.test.ts

@@ -126,6 +126,10 @@ describe.for([['pnpm'] as const])(`%s monorepo`, ([pkgManager]) => {
             // Ignore punycode warning
             return;
           }
+          if (errMsg && errMsg.includes('falling back to an in-memory store')) {
+            // Ignore in-memory storage fallback warning (no storage configured in fixture)
+            return;
+          }
           reject(new Error('failed to start dev: ' + errMsg));
         });
         proc!.stdout?.on('data', data => {
@@ -181,6 +185,10 @@ describe.for([['pnpm'] as const])(`%s monorepo`, ([pkgManager]) => {
             // Ignore punycode warning
             return;
           }
+          if (errMsg && errMsg.includes('falling back to an in-memory store')) {
+            // Ignore in-memory storage fallback warning (no storage configured in fixture)
+            return;
+          }
 
           reject(new Error('failed to start: ' + errMsg));
         });

integrations/brightdata/src/__tests__/client.test.ts

@@ -16,8 +16,7 @@ describe('getBrightDataClient', () => {
     delete process.env.BRIGHTDATA_SERP_ZONE;
     delete process.env.BRIGHTDATA_WEB_UNLOCKER_ZONE;
     fetchMock.mockImplementation(async (_url, init: RequestInit) => {
-      const body =
-        typeof init.body === 'string' ? (JSON.parse(init.body) as { format?: string }) : {};
+      const body = typeof init.body === 'string' ? (JSON.parse(init.body) as { format?: string }) : {};
 
       return body.format === 'json' ? Response.json({}) : new Response('ok');
     });

integrations/brightdata/src/__tests__/fetch.test.ts

@@ -58,9 +58,7 @@ describe('createBrightDataFetchTool', () => {
 
     const tool = createBrightDataFetchTool({ apiKey: 'test-key' });
 
-    await expect(tool.execute!({ url: 'https://example.com' }, {} as any)).rejects.toThrow(
-      'Network unreachable',
-    );
+    await expect(tool.execute!({ url: 'https://example.com' }, {} as any)).rejects.toThrow('Network unreachable');
   });
 
   it('should make one Bright Data request after a successful execute', async () => {

integrations/brightdata/src/__tests__/search.test.ts

@@ -127,37 +127,29 @@ describe('createBrightDataSearchTool', () => {
     const tool = createBrightDataSearchTool({ apiKey: 'test-key' });
     const result = (await tool.execute!({ query: 'test' }, {} as any)) as any;
 
-    expect(result.results).toEqual([
-      { link: 'https://ok.example', title: 'Has both', description: 'ok' },
-    ]);
+    expect(result.results).toEqual([{ link: 'https://ok.example', title: 'Has both', description: 'ok' }]);
   });
 
   it('should let errors propagate', async () => {
     fetchMock.mockRejectedValue(new Error('API rate limit exceeded'));
 
     const tool = createBrightDataSearchTool({ apiKey: 'test-key' });
 
-    await expect(tool.execute!({ query: 'test' }, {} as any)).rejects.toThrow(
-      'API rate limit exceeded',
-    );
+    await expect(tool.execute!({ query: 'test' }, {} as any)).rejects.toThrow('API rate limit exceeded');
   });
 
   it('should parse string responses (SDK returns JSON-encoded text)', async () => {
     fetchMock.mockResolvedValue(
       Response.json({
-        organic: [
-          { link: 'https://from.string', title: 'Stringified', description: 'ok' },
-        ],
+        organic: [{ link: 'https://from.string', title: 'Stringified', description: 'ok' }],
         current_page: 3,
       }),
     );
 
     const tool = createBrightDataSearchTool({ apiKey: 'test-key' });
     const result = (await tool.execute!({ query: 'test' }, {} as any)) as any;
 
-    expect(result.results).toEqual([
-      { link: 'https://from.string', title: 'Stringified', description: 'ok' },
-    ]);
+    expect(result.results).toEqual([{ link: 'https://from.string', title: 'Stringified', description: 'ok' }]);
     expect(result.currentPage).toBe(3);
   });
 

integrations/brightdata/src/client.ts

@@ -135,9 +135,7 @@ function toRequestBody(url: string, zone: string, options: RequestOptions = {}):
 export function getBrightDataClient(config?: BrightDataClientOptions): BrightDataClient {
   const apiKey = config?.apiKey ?? process.env.BRIGHTDATA_API_TOKEN;
   if (!apiKey) {
-    throw new Error(
-      'Bright Data API token is required. Pass { apiKey } or set BRIGHTDATA_API_TOKEN env var.',
-    );
+    throw new Error('Bright Data API token is required. Pass { apiKey } or set BRIGHTDATA_API_TOKEN env var.');
   }
 
   const timeout = config?.timeout;
@@ -152,9 +150,7 @@ export function getBrightDataClient(config?: BrightDataClientOptions): BrightDat
           throw new Error('language must be a two-letter code (e.g. "en", "es")');
         }
 
-        const normalizedOptions = options.language
-          ? { ...options, language: options.language.toLowerCase() }
-          : options;
+        const normalizedOptions = options.language ? { ...options, language: options.language.toLowerCase() } : options;
 
         const url = buildGoogleSearchUrl(query, normalizedOptions);
         return requestBrightData(

mastracode/src/headless-integration.test.ts

@@ -843,6 +843,8 @@ describe('headless mode — thread control', () => {
     const thread = await harness.createThread({ title: 'target-thread' });
     const updatedAtBefore = thread.updatedAt.getTime();
 
+    await new Promise(resolve => setTimeout(resolve, 300));
+
     const exitCode = await runHeadless(harness, {
       prompt: 'Hello',
       format: 'default',
@@ -853,6 +855,9 @@ describe('headless mode — thread control', () => {
 
     expect(exitCode).toBe(0);
 
+    // Allow fire-and-forget persistTokenUsage to flush
+    await new Promise(resolve => setTimeout(resolve, 300));
+
     // Verify the targeted thread was actually used (updatedAt advanced)
     const threads = await harness.listThreads();
     const targeted = threads.find(t => t.id === thread.id);
@@ -869,6 +874,8 @@ describe('headless mode — thread control', () => {
     const thread = await harness.createThread({ title: 'my-feature' });
     const updatedAtBefore = thread.updatedAt.getTime();
 
+    await new Promise(resolve => setTimeout(resolve, 300));
+
     const exitCode = await runHeadless(harness, {
       prompt: 'Hello',
       format: 'default',
@@ -879,6 +886,9 @@ describe('headless mode — thread control', () => {
 
     expect(exitCode).toBe(0);
 
+    // Allow fire-and-forget persistTokenUsage to flush
+    await new Promise(resolve => setTimeout(resolve, 300));
+
     // Verify the titled thread was actually used
     const threads = await harness.listThreads();
     const targeted = threads.find(t => t.id === thread.id);