fix(harness,libsql): keep agent loop alive after tool suspension (#17990)

Author: abhiaiyer91

.changeset/keep-agent-loop-alive-after-suspension.md

@@ -0,0 +1,11 @@
+---
+"@mastra/core": patch
+"@mastra/libsql": patch
+---
+
+Keep the agent loop alive after a tool suspension (e.g. `ask_user`, `request_access`).
+
+Two related fixes that stop the agent from appearing to quit mid-task right after a suspended tool resumes:
+
+- **Harness resume step budget**: `resumeStream()` was called without a step budget, so a resumed run merged over the agent's small default `maxSteps` and stopped after a single step. The harness now threads a shared set of run options (`maxSteps`, `savePerStep`, `modelSettings`, etc.) into both the initial stream and the resume via one helper, so the two paths can no longer drift.
+- **LibSQL `busy_timeout` survival**: bump `@libsql/client` to `^0.17.4` and add a configurable connection timeout so `PRAGMA busy_timeout` survives connections created after `transaction()` (libsql-client-ts#288/#345). This reduces `SQLITE_BUSY` retry-exhaustion stalls under the evented engine's concurrent writes.

packages/core/src/harness/__tests__/harness-tool-suspension.test.ts

@@ -348,4 +348,72 @@ describe('Harness: tool suspension and resumption', () => {
     // Yolo mode should disable tool approval gating on resume, matching sendMessage's behavior
     expect(resumeOptions.requireToolApproval).toBe(false);
   });
+
+  it('should forward the full run budget (maxSteps) to resumeStream so the resumed run does not stop mid-task', async () => {
+    // Regression: resumeStream previously omitted maxSteps, so the resumed run
+    // merged over the agent's small default budget and ended with reason
+    // "complete" after a few steps — the agent stopped mid-task after ask_user.
+    const confirmTool = createTool({
+      id: 'confirm-action',
+      description: 'Confirms an action with the user',
+      inputSchema: z.object({ action: z.string() }),
+      execute: async (input: { action: string }, context?: any) => {
+        // Resume-aware: continue instead of re-suspending once resumeData arrives,
+        // so the resumed run can actually complete.
+        const resumeData = context?.agent?.resumeData ?? context?.workflow?.resumeData ?? context?.resumeData;
+        if (resumeData) {
+          return { result: `Action "${input.action}" confirmed`, resumed: resumeData };
+        }
+        const suspend = context?.suspend ?? context?.agent?.suspend;
+        if (!suspend) throw new Error('suspend not available in context');
+        await suspend({ action: input.action });
+        return { result: `Action "${input.action}" confirmed` };
+      },
+    });
+
+    const agent = new Agent({
+      id: 'test-agent-budget-resume',
+      name: 'Test Agent Budget Resume',
+      instructions: 'You confirm actions.',
+      model: new MastraLanguageModelV2Mock({
+        doStream: (() => {
+          let callCount = 0;
+          return async () => {
+            callCount++;
+            return { stream: callCount === 1 ? createToolCallStream() : createTextStream() };
+          };
+        })(),
+      }),
+      tools: { confirmAction: confirmTool },
+    });
+
+    const storage = new InMemoryStore();
+    const mastra = new Mastra({
+      agents: { 'test-agent-budget-resume': agent },
+      logger: false,
+      storage,
+    });
+
+    const registeredAgent = mastra.getAgent('test-agent-budget-resume');
+    const resumeStreamSpy = vi.spyOn(registeredAgent, 'resumeStream');
+
+    const harness = new Harness({
+      id: 'test-harness-budget-resume',
+      storage,
+      modes: [{ id: 'default', name: 'Default', default: true, agent: registeredAgent }],
+      initialState: { yolo: true } as any,
+    });
+
+    await harness.init();
+    await harness.createThread();
+
+    await harness.sendMessage({ content: 'Deploy to production' });
+    await harness.respondToToolSuspension({ resumeData: { confirmed: true } });
+
+    expect(resumeStreamSpy).toHaveBeenCalled();
+    const [, resumeOptions] = resumeStreamSpy.mock.calls[0] as [any, any];
+    // Must match the budget used for the initial stream, not the agent default.
+    expect(resumeOptions.maxSteps).toBe(1000);
+    expect(resumeOptions.savePerStep).toBe(false);
+  });
 });

packages/core/src/harness/harness.ts

@@ -153,6 +153,18 @@ export function describeNonSuccessFinishReason(reason: string, providerMetadata:
  */
 const FABLE_FALLBACK_MODEL = 'claude-opus-4-8';
 
+/**
+ * Step budget applied to every harness-driven agent run.
+ *
+ * This MUST be passed to both the initial stream and `resumeStream`: when a run
+ * suspends on an interactive tool (e.g. `ask_user`) and then resumes, the
+ * resumed call merges over the agent's *default* options, whose `maxSteps` is
+ * small (~5). Without re-supplying this budget the resumed run is silently
+ * capped and ends with `reason:"complete"` after a few steps — the agent stops
+ * mid-task even though it promised to continue. See {@link buildSharedRunOptions}.
+ */
+const HARNESS_MAX_STEPS = 1000;
+
 /**
  * Returns Anthropic `providerOptions` that enable a server-side fallback to
  * {@link FABLE_FALLBACK_MODEL} when the active model is `claude-fable-5`, and
@@ -1881,33 +1893,42 @@ export class Harness<TState = {}> {
     this.abortRequested = false;
     this.abortController ??= new AbortController();
     const requestContext = await this.buildRequestContext(requestContextInput);
-    const isYolo = (this.state as Record<string, unknown>).yolo === true;
     const streamOptions: Record<string, unknown> = {
+      ...this.buildSharedRunOptions(),
       memory: { thread: this.currentThreadId, resource: this.resourceId },
       abortSignal: this.abortController.signal,
       requestContext,
-      maxSteps: 1000,
-      savePerStep: false,
-      requireToolApproval: !isYolo,
-      modelSettings: { temperature: 1 },
       ...(tracingContext && { tracingContext }),
       ...(tracingOptions && { tracingOptions }),
     };
     streamOptions.toolsets = await this.buildToolsets(requestContext);
 
+    return streamOptions;
+  }
+
+  /**
+   * Options that every harness-driven agent run must carry — the initial stream
+   * AND every `resumeStream`. Centralized so the two paths can't drift: a
+   * missing `maxSteps` on resume silently caps the resumed run at the agent's
+   * small default and ends it mid-task (see {@link HARNESS_MAX_STEPS}).
+   */
+  private buildSharedRunOptions(): Record<string, unknown> {
+    const isYolo = (this.state as Record<string, unknown>).yolo === true;
+    const shared: Record<string, unknown> = {
+      maxSteps: HARNESS_MAX_STEPS,
+      savePerStep: false,
+      requireToolApproval: !isYolo,
+      modelSettings: { temperature: 1 },
+    };
+
     // Auto-enable Anthropic server-side fallbacks for fable-5 so a classifier
     // block is transparently retried on the fallback model instead of failing.
     const fableFallback = buildFableFallbackProviderOptions(this.getCurrentModelId());
     if (fableFallback) {
-      const existing = streamOptions.providerOptions as Record<string, unknown> | undefined;
-      const existingAnthropic = (existing?.anthropic as Record<string, unknown> | undefined) ?? {};
-      streamOptions.providerOptions = {
-        ...existing,
-        anthropic: { ...existingAnthropic, ...fableFallback.anthropic },
-      };
+      shared.providerOptions = { anthropic: { ...fableFallback.anthropic } };
     }
 
-    return streamOptions;
+    return shared;
   }
 
   private async drainFollowUpQueue(options?: {
@@ -3683,16 +3704,19 @@ export class Harness<TState = {}> {
     this.displayState.pendingSuspensions.delete(toolCallId);
 
     const requestContext = await this.buildRequestContext(requestContextInput);
-    const isYolo = (this.state as Record<string, unknown>).yolo === true;
+
     const output = await agent.resumeStream(resumeData, {
+      // Re-supply the shared run budget (maxSteps, etc). Without it the resumed
+      // run merges over the agent's small default maxSteps and stops mid-task.
+      ...this.buildSharedRunOptions(),
       runId: suspension.runId,
       toolCallId,
-      requireToolApproval: !isYolo,
       memory: this.currentThreadId ? { thread: this.currentThreadId, resource: this.resourceId } : undefined,
       abortSignal: this.abortController.signal,
       requestContext,
       toolsets: await this.buildToolsets(requestContext),
     });
+
     await this.processStream(output, requestContext);
   }
 

pnpm-lock.yaml

@@ -80,6 +80,11 @@ minimumReleaseAgeExclude:
   - 'mastracode'
   - '@earendil-works/pi-tui'
   - railway@3.3.1
+  # Turso-recommended busy_timeout fix (libsql-client-ts#288/#345). Adopted
+  # immediately on their guidance; pinned so the release-age gate still applies
+  # to any later @libsql/client versions.
+  - '@libsql/client@0.17.4'
+  - '@libsql/core@0.17.4'
 trustPolicyExclude:
   # Security fix for GHSA middleware-based route protection bypass (CVE).
   # 3.47.4 is published with provenance attestation rather than trusted publisher,

pnpm-workspace.yaml

@@ -27,7 +27,7 @@
   },
   "license": "Apache-2.0",
   "dependencies": {
-    "@libsql/client": "^0.17.3"
+    "@libsql/client": "^0.17.4"
   },
   "devDependencies": {
     "@internal/lint": "workspace:*",

stores/libsql/package.json

@@ -0,0 +1,81 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import type { createClient } from '@libsql/client';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+
+import { resolveClient, DEFAULT_CONNECTION_TIMEOUT_MS } from './db';
+import { LibSQLStore } from './index';
+
+type TestClient = ReturnType<typeof createClient>;
+
+const getClient = (store: LibSQLStore): TestClient => (store as unknown as { client: TestClient }).client;
+
+const readBusyTimeout = async (client: TestClient): Promise<number> => {
+  const rs = await client.execute('PRAGMA busy_timeout');
+  return Number(rs.rows[0]!['timeout']);
+};
+
+describe('LibSQL busy_timeout configuration', () => {
+  let tmpDir: string;
+
+  beforeEach(() => {
+    tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'libsql-busy-timeout-test-'));
+  });
+
+  afterEach(() => {
+    fs.rmSync(tmpDir, { recursive: true, force: true });
+  });
+
+  it('applies the default busy_timeout to a local file store', async () => {
+    const store = new LibSQLStore({ id: 'busy-default', url: `file:${path.join(tmpDir, 'default.db')}` });
+    try {
+      await store.init();
+
+      expect(await readBusyTimeout(getClient(store))).toBe(DEFAULT_CONNECTION_TIMEOUT_MS);
+    } finally {
+      await store.close();
+    }
+  });
+
+  it('honors a custom connectionTimeoutMs', async () => {
+    const store = new LibSQLStore({
+      id: 'busy-custom',
+      url: `file:${path.join(tmpDir, 'custom.db')}`,
+      connectionTimeoutMs: 1234,
+    });
+    try {
+      await store.init();
+
+      expect(await readBusyTimeout(getClient(store))).toBe(1234);
+    } finally {
+      await store.close();
+    }
+  });
+
+  it('keeps the busy_timeout after a transaction() opens a new connection (libsql-client-ts#288)', async () => {
+    const client = resolveClient({ url: `file:${path.join(tmpDir, 'txn.db')}`, connectionTimeoutMs: 4321 });
+
+    try {
+      const txn = await client.transaction('write');
+      await txn.execute('CREATE TABLE t (a)');
+      await txn.commit();
+
+      // transaction() hands the client's connection to the transaction and lazily
+      // opens a new one; the timeout must apply to that new connection too.
+      expect(await readBusyTimeout(client)).toBe(4321);
+    } finally {
+      client.close();
+    }
+  });
+
+  it('resolveClient defaults the busy_timeout for local urls', async () => {
+    const client = resolveClient({ url: `file:${path.join(tmpDir, 'resolve.db')}` });
+
+    try {
+      expect(await readBusyTimeout(client)).toBe(DEFAULT_CONNECTION_TIMEOUT_MS);
+    } finally {
+      client.close();
+    }
+  });
+});

stores/libsql/src/storage/busy-timeout.test.ts

@@ -35,8 +35,24 @@ export type LibSQLDomainBaseConfig = {
    * @default 100
    */
   initialBackoffMs?: number;
+  /**
+   * SQLite `busy_timeout` (in milliseconds) applied to the underlying connection
+   * for local (`file:`/`:memory:`) databases. Lets a write wait for a lock to
+   * clear instead of failing immediately with `SQLITE_BUSY`. Requires
+   * `@libsql/client` >= 0.17.4 (see libsql-client-ts#288/#345). Ignored when an
+   * existing `client` is supplied.
+   * @default 5000
+   */
+  connectionTimeoutMs?: number;
 };
 
+/**
+ * Default SQLite `busy_timeout` (ms) for local LibSQL connections. Chosen to
+ * comfortably exceed the write-retry backoff window so contended writes block
+ * briefly rather than surfacing as `SQLITE_BUSY` errors.
+ */
+export const DEFAULT_CONNECTION_TIMEOUT_MS = 5000;
+
 /**
  * Configuration for LibSQL domains - accepts either credentials or an existing client
  */
@@ -63,9 +79,14 @@ export function resolveClient(config: LibSQLDomainConfig): Client {
   if ('client' in config) {
     return config.client;
   }
+  const isLocal = config.url.startsWith('file:') || config.url.includes(':memory:');
+  const timeout = config.connectionTimeoutMs ?? DEFAULT_CONNECTION_TIMEOUT_MS;
   return createClient({
     url: config.url,
     ...(config.authToken ? { authToken: config.authToken } : {}),
+    // Only local sqlite3 connections honor `busy_timeout`; remote contention is
+    // resolved server-side, so passing it there is meaningless.
+    ...(isLocal ? { timeout } : {}),
   });
 }