{"record":{"id":"25b7e141149d181a","repo":"mastra-ai/mastra","slug":"cannot-register-run-output-runid-thread-is-alr","errorCode":null,"errorMessage":"Cannot register run ${output.runId}: thread is already active with run ${activeRunId}","messagePattern":"Cannot register run (.+?): thread is already active with run (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"packages/core/src/agent/thread-stream-runtime.ts","lineNumber":1327,"sourceCode":"\n  async #registerRunStrict<OUTPUT>(\n    agent: Agent<any, any, any, any>,\n    output: MastraModelOutput<OUTPUT>,\n    streamOptions: AgentExecutionOptions<OUTPUT>,\n    pubsub: PubSub | undefined,\n    threadId: string,\n    resourceId: string | undefined,\n    registrationOptions: AgentThreadStrictRegistrationOptions,\n  ): Promise<AgentThreadRunRegistration> {\n    await registrationOptions.validate?.();\n\n    const state = this.#getState(pubsub);\n    this.#sweepStaleSuspendedRecords(state, pubsub);\n    const key = this.#threadKey(resourceId, threadId);\n    const activeRunId = state.activeThreadRunIds.get(key);\n    const activeRecord = activeRunId ? state.threadRunsById.get(activeRunId) : undefined;\n    if (activeRecord && activeRecord.runId !== output.runId && this.#isThreadBlockingRun(state, activeRecord)) {\n      throw new Error(`Cannot register run ${output.runId}: thread is already active with run ${activeRunId}`);\n    }\n    const resolvedPubSub = this.#getPubSub(pubsub);\n    const leaseProvider = this.#getLeaseProvider(resolvedPubSub);\n    const lease = await leaseProvider.acquireLease(key, output.runId, AGENT_THREAD_LEASE_TTL_MS);\n    if (!lease.acquired) {\n      throw new AgentThreadLeaseConflictError(output.runId, lease.owner ?? 'another owner');\n    }\n\n    // A failed external-ownership validation means another recovery attempt may\n    // already own this same runId. Do not release its indistinguishable thread\n    // lease; its TTL or the successor registration will take over renewal.\n    await registrationOptions.validate?.();\n\n    this.#startLeaseRenewal(resolvedPubSub, key, output.runId);\n    const { streamId, streamSeq } = this.#nextStreamIdentity(state, output.runId);\n    const {\n      output: outputForSubscribers,\n      createSubscriberStream,","sourceCodeStart":1309,"sourceCodeEnd":1345,"githubUrl":"https://github.com/mastra-ai/mastra/blob/75dd419e613fe9c39f846ffc500716141b74fda6/packages/core/src/agent/thread-stream-runtime.ts#L1309-L1345","documentation":"The thread stream runtime allows only one blocking run per (resourceId, threadId) at a time. When registering a new run, if the thread already has an active run (different runId) that is still blocking, registration fails with this error to prevent two runs from concurrently driving the same thread's stream/history.","triggerScenarios":"Starting a second agent stream/generation on the same thread while a previous blocking run is still active — e.g. double-submitting a chat prompt without cancelling the first request, or a retried request whose original run hasn't been reaped by #sweepStaleSuspendedRecords.","commonSituations":"UI allows parallel sends; server retries duplicate an in-flight request; a crashed run left a stale active record that hasn't been swept yet; multi-instance deployments sharing state with a slow/stuck first run.","solutions":["Wait for or cancel/complete the existing run before starting a new one on the same thread.","Abort the first run's AbortController so its registration is cleaned up, then retry the new stream.","Fix client-side double submits (disable send button while streaming, debounce, dedupe by request id).","If the active record is genuinely stale, ensure the sweep logic/lease TTL runs or clear stale thread-run state in your storage.","Use separate threads for concurrent generations instead of sharing one thread."],"exampleFix":"// before\nagent.stream(...) // while previous run on same threadId still active\n// after\nawait previousRun; // or previousRun.abort()\nawait agent.stream({ ..., threadId });","handlingStrategy":"try-catch","validationCode":"// before starting a run, check locally whether the thread is already streaming\nif (activeStreamsByThread.has(threadId)) {\n  throw new Error(`thread ${threadId} already has an active run`);\n}","typeGuard":null,"tryCatchPattern":"try {\n  runtime.registerRun(output);\n} catch (e) {\n  if (e instanceof Error && e.message.startsWith('Cannot register run')) {\n    // wait/cancel the prior run, then retry once\n    await cancelActiveRun(threadId);\n    await runtime.registerRun(output);\n  } else throw e;\n}","preventionTips":["Disable/queue chat sends while a run on the same thread is streaming","Always abort runs via AbortController when the user navigates away or cancels","Dedupe retries with idempotency/request ids","Use distinct threads for concurrent generations","Monitor for stale active-run records and ensure sweeps/leases expire"],"tags":["concurrency","threads","streaming","conflict"],"backgroundTag":"concurrent-run-conflict","analyzedSha":"75dd419e613fe9c39f846ffc500716141b74fda6","analyzedAt":"2026-08-30T00:15:31.844Z","schemaVersion":2},"datasetVersion":"2026-08-30T08:17:16.595Z"}