diff --git a/apps/sim/app/api/chat/[identifier]/route.test.ts b/apps/sim/app/api/chat/[identifier]/route.test.ts
index 5f1bd146087..d6aa709cd63 100644
--- a/apps/sim/app/api/chat/[identifier]/route.test.ts
+++ b/apps/sim/app/api/chat/[identifier]/route.test.ts
@@ -534,6 +534,38 @@ describe('Chat Identifier API Route', () => {
       )
     }, 10000)
 
+    it('projects the internal completion envelope to the public streaming callback', async () => {
+      const response = await POST(createMockNextRequest('POST', { input: 'Hello' }), {
+        params: Promise.resolve({ identifier: 'test-chat' }),
+      })
+      expect(response.status).toBe(200)
+      const onBlockComplete = vi.fn()
+      await vi.mocked(createStreamingResponse).mock.calls[0][0].executeFn({
+        onStream: vi.fn(),
+        onBlockComplete,
+        abortSignal: new AbortController().signal,
+      })
+      await vi.mocked(executeWorkflow).mock.calls[0][4]?.onBlockComplete?.('block-1', {
+        output: { value: 'public output' },
+        outputBlockId: 'child:block-1',
+        resolvedSecretTraceProvenance: {
+          version: 1,
+          complete: true,
+          entries: [{ encryptedValue: 'private-ciphertext' }],
+          scope: { userId: 'user-1', workspaceId: 'workspace-1' },
+        },
+        executionTime: 1,
+        executionOrder: 0,
+        startedAt: '2026-01-01T00:00:00Z',
+        endedAt: '2026-01-01T00:00:01Z',
+      })
+      expect(onBlockComplete).toHaveBeenCalledExactlyOnceWith(
+        'block-1',
+        { value: 'public output' },
+        'child:block-1'
+      )
+    })
+
     it('executes with the email proven by the chat authentication gate', async () => {
       mockValidateChatAuth.mockResolvedValueOnce({
         authorized: true,
diff --git a/apps/sim/app/api/chat/[identifier]/route.ts b/apps/sim/app/api/chat/[identifier]/route.ts
index 7cc25ed7d18..ec5abdec4cc 100644
--- a/apps/sim/app/api/chat/[identifier]/route.ts
+++ b/apps/sim/app/api/chat/[identifier]/route.ts
@@ -414,7 +414,8 @@ export const POST = withRouteHandler(
                 isSecureMode: true,
                 workflowTriggerType: 'chat',
                 onStream,
-                onBlockComplete,
+                onBlockComplete: (blockId, data) =>
+                  onBlockComplete(blockId, data.output, data.outputBlockId),
                 skipLoggingComplete: true,
                 abortSignal,
                 executionMode: 'stream',
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts
index 2d99b7a7a7f..bbd7632ffc1 100644
--- a/apps/sim/app/api/workflows/[id]/execute/route.ts
+++ b/apps/sim/app/api/workflows/[id]/execute/route.ts
@@ -1718,7 +1718,8 @@ async function handleExecutePost(
               isSecureMode: false,
               workflowTriggerType: triggerType === 'chat' ? 'chat' : 'api',
               onStream,
-              onBlockComplete,
+              onBlockComplete: (blockId, data) =>
+                onBlockComplete(blockId, data.output, data.outputBlockId),
               skipLoggingComplete: true,
               includeFileBase64,
               base64MaxBytes,
diff --git a/apps/sim/background/resume-execution.ts b/apps/sim/background/resume-execution.ts
index 9b2f7d4f584..0d763e391ca 100644
--- a/apps/sim/background/resume-execution.ts
+++ b/apps/sim/background/resume-execution.ts
@@ -14,6 +14,7 @@ import {
   getTimeoutErrorMessage,
 } from '@/lib/core/execution-limits'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
+import type { WorkflowCellProgressWriter } from '@/lib/table/cell-write'
 import { isExecCancelled } from '@/lib/table/deps'
 import type { RowExecutionMetadata } from '@/lib/table/types'
 import { classifyWorkflowCellTerminalResult } from '@/lib/table/workflow-cell-result'
@@ -247,7 +248,7 @@ function throwIfResumeAttemptTimedOut(
 }
 
 type CellWriters = {
-  cellOnBlockComplete: (blockId: string, output: unknown) => Promise<void>
+  cellOnBlockComplete: WorkflowCellProgressWriter['onBlockComplete']
   writeCellTerminal: (
     status: 'completed' | 'error' | 'cancelled' | 'paused',
     error: string | null
diff --git a/apps/sim/lib/copilot/async-runs/errors.ts b/apps/sim/lib/copilot/async-runs/errors.ts
new file mode 100644
index 00000000000..a9e30c3a8b6
--- /dev/null
+++ b/apps/sim/lib/copilot/async-runs/errors.ts
@@ -0,0 +1,7 @@
+/** A durable tool identity must never be reused by a different run. */
+export class AsyncToolCallOwnershipError extends Error {
+  constructor() {
+    super('Async tool call belongs to another run')
+    this.name = 'AsyncToolCallOwnershipError'
+  }
+}
diff --git a/apps/sim/lib/copilot/async-runs/repository.test.ts b/apps/sim/lib/copilot/async-runs/repository.test.ts
index c150413ad14..0c035a2b537 100644
--- a/apps/sim/lib/copilot/async-runs/repository.test.ts
+++ b/apps/sim/lib/copilot/async-runs/repository.test.ts
@@ -406,4 +406,28 @@ describe('async tool repository single-row semantics', () => {
       expect(dbChainMockFns.values).not.toHaveBeenCalled()
     }
   )
+
+  it('refuses a provider-ID collision with an existing row in another run', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { runId: 'old-run', toolCallId: 'provider-shared-call', toolName: 'browser_close_tab' },
+    ])
+    await expect(
+      upsertAsyncToolCall({
+        runId: 'current-run',
+        toolCallId: 'provider-shared-call',
+        toolName: 'glob',
+      })
+    ).rejects.toThrow('Async tool call belongs to another run')
+    expect(dbChainMockFns.values).not.toHaveBeenCalled()
+  })
+
+  it('refuses a foreign row that wins the insert race', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([{ runId: 'other-run', toolCallId: 'call-race' }])
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+    await expect(
+      upsertAsyncToolCall({ runId: 'current-run', toolCallId: 'call-race', toolName: 'glob' })
+    ).rejects.toThrow('Async tool call belongs to another run')
+  })
 })
diff --git a/apps/sim/lib/copilot/async-runs/repository.ts b/apps/sim/lib/copilot/async-runs/repository.ts
index 8d3da676cc6..e00d5474ae6 100644
--- a/apps/sim/lib/copilot/async-runs/repository.ts
+++ b/apps/sim/lib/copilot/async-runs/repository.ts
@@ -11,6 +11,7 @@ import { createLogger } from '@sim/logger'
 import { filterUndefined } from '@sim/utils/object'
 import { sanitizeValueForJsonb } from '@sim/utils/string'
 import { and, desc, eq, inArray, isNull, or, sql } from 'drizzle-orm'
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import { TraceAttr } from '@/lib/copilot/generated/trace-attributes-v1'
 import { TraceSpan } from '@/lib/copilot/generated/trace-spans-v1'
 import { markSpanForError } from '@/lib/copilot/request/otel'
@@ -219,7 +220,12 @@ export async function upsertAsyncToolCall(input: {
     },
     async () => {
       const existing = await getAsyncToolCall(input.toolCallId)
-      if (existing) return existing
+      if (existing) {
+        if (input.runId && existing.runId !== input.runId) {
+          throw new AsyncToolCallOwnershipError()
+        }
+        return existing
+      }
 
       const incomingStatus = input.status ?? 'pending'
       const effectiveRunId = input.runId ?? null
@@ -250,7 +256,11 @@ export async function upsertAsyncToolCall(input: {
         .onConflictDoNothing()
         .returning()
 
-      return row ?? getAsyncToolCall(input.toolCallId)
+      const persisted = row ?? (await getAsyncToolCall(input.toolCallId))
+      if (persisted && persisted.runId !== effectiveRunId) {
+        throw new AsyncToolCallOwnershipError()
+      }
+      return persisted
     }
   )
 }
diff --git a/apps/sim/lib/copilot/async-runs/tool-identity.postgres.test.ts b/apps/sim/lib/copilot/async-runs/tool-identity.postgres.test.ts
new file mode 100644
index 00000000000..dcec96fb2da
--- /dev/null
+++ b/apps/sim/lib/copilot/async-runs/tool-identity.postgres.test.ts
@@ -0,0 +1,323 @@
+/**
+ * @vitest-environment node
+ *
+ * Exercises real repository queries against temporary PostgreSQL tables and,
+ * when configured, the production confirmation/permission boundary over Redis.
+ * Set COPILOT_IDENTITY_TEST_DATABASE_URL to a local PostgreSQL database and
+ * COPILOT_IDENTITY_TEST_REDIS_URL to an isolated local Redis service.
+ */
+import { generateId } from '@sim/utils/id'
+import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
+import Redis from 'ioredis'
+import postgres from 'postgres'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { database, redisState, databaseUrl, redisUrl } = vi.hoisted(() => {
+  const databaseUrl = process.env.COPILOT_IDENTITY_TEST_DATABASE_URL
+  const redisUrl = process.env.COPILOT_IDENTITY_TEST_REDIS_URL
+  for (const value of [databaseUrl, redisUrl]) {
+    if (value && !['localhost', '127.0.0.1', '[::1]'].includes(new URL(value).hostname)) {
+      throw new Error('Copilot identity integration tests require local services')
+    }
+  }
+  const channels = globalThis as typeof globalThis & {
+    _toolConfirmationChannel?: { dispose(): void }
+    _toolPermissionChannel?: { dispose(): void }
+  }
+  channels._toolConfirmationChannel?.dispose()
+  channels._toolPermissionChannel?.dispose()
+  channels._toolConfirmationChannel = undefined
+  channels._toolPermissionChannel = undefined
+  return {
+    databaseUrl,
+    redisUrl: databaseUrl ? redisUrl : undefined,
+    database: { current: undefined as PostgresJsDatabase | undefined },
+    redisState: { current: undefined as Redis | undefined },
+  }
+})
+
+vi.unmock('@sim/db/schema')
+vi.unmock('drizzle-orm')
+vi.unmock('ioredis')
+vi.unmock('@/lib/events/pubsub')
+vi.mock('@sim/db', () => ({
+  db: {
+    select: (...args: unknown[]) => {
+      if (!database.current) throw new Error('PostgreSQL test database is not initialized')
+      return Reflect.apply(database.current.select, database.current, args)
+    },
+    insert: (...args: unknown[]) => {
+      if (!database.current) throw new Error('PostgreSQL test database is not initialized')
+      return Reflect.apply(database.current.insert, database.current, args)
+    },
+    update: (...args: unknown[]) => {
+      if (!database.current) throw new Error('PostgreSQL test database is not initialized')
+      return Reflect.apply(database.current.update, database.current, args)
+    },
+  },
+}))
+vi.mock('@/lib/copilot/request/otel', () => ({ markSpanForError: vi.fn() }))
+vi.mock('@/lib/core/config/redis', () => ({
+  getConfiguredRedisUrl: () => redisUrl,
+  getRedisConnectionDefaults: () => ({}),
+  getRedisClient: () => redisState.current,
+}))
+
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
+import * as asyncRepository from '@/lib/copilot/async-runs/repository'
+import {
+  completeAsyncToolCall,
+  getAsyncToolCall,
+  recordToolPermissionDecision,
+  replaceTerminalAsyncToolCallResult,
+  upsertAsyncToolCall,
+} from '@/lib/copilot/async-runs/repository'
+import {
+  publishToolConfirmation,
+  waitForToolConfirmation,
+} from '@/lib/copilot/persistence/tool-confirm'
+import {
+  publishToolPermissionDecision,
+  waitForToolPermissionDecision,
+} from '@/lib/copilot/persistence/tool-permission'
+import {
+  createProviderToolCallIdentity,
+  scopeProviderToolCallId,
+} from '@/lib/copilot/request/go/tool-call-identity'
+
+const connection = databaseUrl ? postgres(databaseUrl, { max: 1 }) : undefined
+const providerId = 'call_reused_fixture'
+const legacyRunId = generateId()
+const firstRunId = generateId()
+const secondRunId = generateId()
+const firstIdentity = createProviderToolCallIdentity(firstRunId)
+const secondIdentity = createProviderToolCallIdentity(secondRunId)
+const firstId = scopeProviderToolCallId(providerId, firstIdentity)
+const secondId = scopeProviderToolCallId(providerId, secondIdentity)
+const publishedIds = new Set<string>()
+
+async function createCurrentCalls() {
+  return Promise.all([
+    upsertAsyncToolCall({ runId: firstRunId, toolCallId: firstId, toolName: 'glob' }),
+    upsertAsyncToolCall({ runId: secondRunId, toolCallId: secondId, toolName: 'glob' }),
+  ])
+}
+
+function publishCompletion(toolCallId: string, status: 'success' | 'error' | 'background') {
+  publishedIds.add(toolCallId)
+  publishToolConfirmation({ toolCallId, status })
+}
+
+afterAll(async () => {
+  const channels = globalThis as typeof globalThis & {
+    _toolConfirmationChannel?: { dispose(): void }
+    _toolPermissionChannel?: { dispose(): void }
+  }
+  channels._toolConfirmationChannel?.dispose()
+  channels._toolPermissionChannel?.dispose()
+  channels._toolConfirmationChannel = undefined
+  channels._toolPermissionChannel = undefined
+  if (redisState.current) {
+    for (const id of publishedIds) {
+      await redisState.current.del(`copilot:tool-confirmation:${id}`)
+    }
+    await redisState.current.quit()
+  }
+  await connection?.end()
+})
+
+describe.skipIf(!databaseUrl)('Copilot tool identity with PostgreSQL', () => {
+  beforeAll(async () => {
+    if (!connection) throw new Error('PostgreSQL test database is not initialized')
+    database.current = drizzle(connection)
+    await connection.unsafe(`
+      CREATE TEMP TABLE copilot_async_tool_calls (
+        id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+        run_id uuid NOT NULL, checkpoint_id uuid,
+        tool_call_id text NOT NULL UNIQUE, tool_name text NOT NULL,
+        args jsonb NOT NULL DEFAULT '{}', status text NOT NULL DEFAULT 'pending',
+        result jsonb, error text, permission_decision text, permission_decided_at timestamp,
+        claimed_at timestamp, claimed_by text, completed_at timestamp,
+        created_at timestamp NOT NULL DEFAULT now(), updated_at timestamp NOT NULL DEFAULT now()
+      )
+    `)
+    if (redisUrl) {
+      redisState.current = new Redis(redisUrl, { maxRetriesPerRequest: 1 })
+      await redisState.current.ping()
+      await vi.waitFor(async () => {
+        const counts = await redisState.current?.pubsub(
+          'NUMSUB',
+          'copilot:tool-confirmation',
+          'copilot:tool-permission'
+        )
+        expect(Array.isArray(counts) ? [Number(counts[1]), Number(counts[3])] : []).toEqual([1, 1])
+      })
+    }
+  })
+
+  beforeEach(async () => {
+    vi.restoreAllMocks()
+    if (!connection) throw new Error('PostgreSQL test database is not initialized')
+    await connection.unsafe('TRUNCATE pg_temp.copilot_async_tool_calls')
+    await connection`
+      INSERT INTO copilot_async_tool_calls
+        (run_id, tool_call_id, tool_name, status, result, created_at, updated_at)
+      VALUES (${legacyRunId}, ${providerId}, 'browser_close_tab', 'completed',
+        '{"legacy":true}'::jsonb, '2025-01-02 00:00:00', '2025-01-02 00:00:00')
+    `
+  })
+
+  it('keeps repeated provider IDs independent and retries one run idempotently', async () => {
+    const legacy = await getAsyncToolCall(providerId)
+    const [first, second, ...retries] = await Promise.all([
+      upsertAsyncToolCall({ runId: firstRunId, toolCallId: firstId, toolName: 'glob' }),
+      upsertAsyncToolCall({ runId: secondRunId, toolCallId: secondId, toolName: 'glob' }),
+      ...Array.from({ length: 4 }, () =>
+        upsertAsyncToolCall({
+          runId: firstRunId,
+          toolCallId: scopeProviderToolCallId(providerId, firstIdentity),
+          toolName: 'glob',
+        })
+      ),
+    ])
+
+    expect(first?.id).not.toBe(second?.id)
+    expect(first?.runId).toBe(firstRunId)
+    expect(second?.runId).toBe(secondRunId)
+    expect(retries.every((row) => row?.id === first?.id)).toBe(true)
+    expect(await getAsyncToolCall(providerId)).toEqual(legacy)
+    if (!connection) throw new Error('PostgreSQL test database is not initialized')
+    const counts = await connection`SELECT count(*)::integer AS count FROM copilot_async_tool_calls`
+    expect(counts[0].count).toBe(3)
+  })
+
+  it('refuses a conflicting owner without changing the existing legacy row', async () => {
+    const legacy = await getAsyncToolCall(providerId)
+    await expect(
+      upsertAsyncToolCall({ runId: firstRunId, toolCallId: providerId, toolName: 'glob' })
+    ).rejects.toBeInstanceOf(AsyncToolCallOwnershipError)
+    expect(await getAsyncToolCall(providerId)).toEqual(legacy)
+  })
+
+  it('settles and replaces only the intended current run, preserving the legacy completion', async () => {
+    const legacy = await getAsyncToolCall(providerId)
+    await createCurrentCalls()
+    await Promise.all([
+      completeAsyncToolCall({ toolCallId: firstId, status: 'completed', result: { run: 'first' } }),
+      completeAsyncToolCall({ toolCallId: secondId, status: 'failed', error: 'second failed' }),
+    ])
+    const second = await getAsyncToolCall(secondId)
+    await replaceTerminalAsyncToolCallResult({
+      toolCallId: firstId,
+      status: 'completed',
+      result: { projected: 'first' },
+      error: null,
+    })
+    expect((await getAsyncToolCall(firstId))?.result).toEqual({ projected: 'first' })
+    expect(await getAsyncToolCall(secondId)).toEqual(second)
+    expect(await getAsyncToolCall(providerId)).toEqual(legacy)
+    await expect(
+      completeAsyncToolCall({ toolCallId: firstId, status: 'failed', error: 'late failure' })
+    ).resolves.toBeNull()
+  })
+
+  describe.skipIf(!redisUrl)('with actual Redis confirmation and permission channels', () => {
+    it('wakes only the matching waiter and reads its durable terminal result', async () => {
+      await createCurrentCalls()
+      const durableReads = vi.spyOn(asyncRepository, 'getAsyncToolCalls')
+      const abort = new AbortController()
+      let firstSettled = false
+      const acceptStatus = (status: string) => status === 'success' || status === 'error'
+      const first = waitForToolConfirmation(firstId, 5_000, abort.signal, { acceptStatus }).then(
+        (result) => {
+          firstSettled = true
+          return result
+        }
+      )
+      const second = waitForToolConfirmation(secondId, 5_000, abort.signal, { acceptStatus })
+      try {
+        expect(durableReads).toHaveBeenCalledTimes(2)
+        await Promise.all(durableReads.mock.results.map((result) => result.value))
+        await completeAsyncToolCall({
+          toolCallId: secondId,
+          status: 'completed',
+          result: { run: 'second' },
+        })
+        publishCompletion(secondId, 'success')
+        expect(await second).toMatchObject({ status: 'success', data: { run: 'second' } })
+        expect(firstSettled).toBe(false)
+        expect((await getAsyncToolCall(firstId))?.status).toBe('pending')
+
+        await completeAsyncToolCall({
+          toolCallId: firstId,
+          status: 'completed',
+          result: { run: 'first' },
+        })
+        publishCompletion(firstId, 'success')
+        expect(await first).toMatchObject({ status: 'success', data: { run: 'first' } })
+        await expect
+          .poll(() => redisState.current?.get(`copilot:tool-confirmation:${firstId}`))
+          .toBe(JSON.stringify({ toolCallId: firstId, status: 'success' }))
+      } finally {
+        abort.abort()
+        await Promise.all([first, second])
+      }
+    })
+
+    it('does not detach another run when a background signal arrives', async () => {
+      await createCurrentCalls()
+      const durableReads = vi.spyOn(asyncRepository, 'getAsyncToolCalls')
+      const abort = new AbortController()
+      let firstSettled = false
+      const acceptStatus = (status: string) => status === 'background'
+      const first = waitForToolConfirmation(firstId, 5_000, abort.signal, { acceptStatus }).then(
+        (result) => {
+          firstSettled = true
+          return result
+        }
+      )
+      const second = waitForToolConfirmation(secondId, 5_000, abort.signal, { acceptStatus })
+      try {
+        expect(durableReads).toHaveBeenCalledTimes(2)
+        await Promise.all(durableReads.mock.results.map((result) => result.value))
+        publishCompletion(secondId, 'background')
+        expect(await second).toMatchObject({ status: 'background' })
+        expect(firstSettled).toBe(false)
+        expect((await getAsyncToolCall(firstId))?.status).toBe('pending')
+      } finally {
+        abort.abort()
+        await Promise.all([first, second])
+      }
+    })
+
+    it('does not authorize another run when a permission decision arrives', async () => {
+      await createCurrentCalls()
+      const durableReads = vi.spyOn(asyncRepository, 'getAsyncToolCall')
+      const abort = new AbortController()
+      let firstSettled = false
+      const first = waitForToolPermissionDecision(firstId, 5_000, abort.signal).then((result) => {
+        firstSettled = true
+        return result
+      })
+      const second = waitForToolPermissionDecision(secondId, 5_000, abort.signal)
+      try {
+        expect(durableReads).toHaveBeenCalledTimes(2)
+        await Promise.all(durableReads.mock.results.map((result) => result.value))
+        await recordToolPermissionDecision(secondId, 'skip')
+        publishToolPermissionDecision({ toolCallId: secondId, decision: 'skip' })
+        expect(await second).toMatchObject({ toolCallId: secondId, decision: 'skip' })
+        expect(firstSettled).toBe(false)
+        expect((await getAsyncToolCall(firstId))?.permissionDecision).toBeNull()
+
+        await recordToolPermissionDecision(firstId, 'allow')
+        publishToolPermissionDecision({ toolCallId: firstId, decision: 'allow' })
+        expect(await first).toMatchObject({ toolCallId: firstId, decision: 'allow' })
+        expect((await getAsyncToolCall(secondId))?.permissionDecision).toBe('skip')
+        expect((await getAsyncToolCall(providerId))?.permissionDecision).toBeNull()
+      } finally {
+        abort.abort()
+        await Promise.all([first, second])
+      }
+    })
+  })
+})
diff --git a/apps/sim/lib/copilot/request/go/stream.test.ts b/apps/sim/lib/copilot/request/go/stream.test.ts
index c1e7081cb8c..9dc388f3157 100644
--- a/apps/sim/lib/copilot/request/go/stream.test.ts
+++ b/apps/sim/lib/copilot/request/go/stream.test.ts
@@ -86,6 +86,11 @@ import {
   STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE,
   StreamEndedWithoutTerminalError,
 } from '@/lib/copilot/request/go/stream'
+import {
+  createProviderToolCallIdentity,
+  PROVIDER_TOOL_CALL_IDENTITY_LIMITS,
+  scopeProviderToolCallId,
+} from '@/lib/copilot/request/go/tool-call-identity'
 import { AbortReason, createEvent, hasAbortMarker } from '@/lib/copilot/request/session'
 import { RequestTraceV1Outcome, TraceCollector } from '@/lib/copilot/request/trace'
 import type { ExecutionContext, StreamingContext } from '@/lib/copilot/request/types'
@@ -182,6 +187,112 @@ describe('copilot go stream helpers', () => {
     vi.unstubAllGlobals()
   })
 
+  it('terminates the stream on an exhausted identity budget before forwarding later events', async () => {
+    const identity = createProviderToolCallIdentity('exhausted-identity-run')
+    identity.retainedBytes = PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxRetainedBytes
+    const context = createStreamingContext()
+    context.providerToolCallIdentity = identity
+    const onEvent = vi.fn()
+    vi.mocked(fetch).mockResolvedValueOnce(
+      createSseResponse([
+        createEvent({
+          streamId: 'identity-budget-stream',
+          cursor: '1',
+          requestId: 'identity-budget-request',
+          seq: 1,
+          type: 'tool',
+          payload: {
+            phase: 'call',
+            toolCallId: 'new-call-over-budget',
+            toolName: 'glob',
+            executor: 'client',
+            mode: 'async',
+            arguments: { path: 'files' },
+          },
+        }),
+        createEvent({
+          streamId: 'identity-budget-stream',
+          cursor: '2',
+          requestId: 'identity-budget-request',
+          seq: 2,
+          type: 'complete',
+          payload: { status: 'complete' },
+        }),
+      ])
+    )
+
+    await expect(
+      runStreamLoop('https://example.com/api/mothership', {}, context, turnScopedExecContext(), {
+        timeout: 1000,
+        onEvent,
+      })
+    ).rejects.toThrow('Provider tool call identity budget exceeded')
+    expect(onEvent).not.toHaveBeenCalled()
+    expect(context.completionStatus).toBeUndefined()
+    expect(context.errors).toContain('Provider tool call identity budget exceeded')
+  })
+
+  it('namespaces repeated provider IDs before forwarding and checkpoint handling', async () => {
+    for (const runId of ['stream-identity-run-1', 'stream-identity-run-2']) {
+      const identity = createProviderToolCallIdentity(runId)
+      const context = createStreamingContext()
+      context.providerToolCallIdentity = identity
+      const onEvent = vi.fn()
+      vi.mocked(fetch).mockResolvedValueOnce(
+        createSseResponse([
+          createEvent({
+            streamId: 'identity-stream',
+            cursor: '1',
+            requestId: 'identity-request',
+            seq: 1,
+            type: 'tool',
+            payload: {
+              phase: 'call',
+              toolCallId: 'shared-provider-call',
+              toolName: 'glob',
+              executor: 'client',
+              mode: 'async',
+              arguments: { path: 'files', toolCallId: 'unchanged-user-argument' },
+            },
+          }),
+          createEvent({
+            streamId: 'identity-stream',
+            cursor: '2',
+            requestId: 'identity-request',
+            seq: 2,
+            type: 'run',
+            payload: {
+              kind: 'checkpoint_pause',
+              checkpointId: 'identity-checkpoint',
+              executionId: 'identity-execution',
+              runId: 'provider-run',
+              pendingToolCallIds: ['shared-provider-call'],
+            },
+          }),
+        ])
+      )
+      await runStreamLoop(
+        'https://example.com/api/mothership',
+        {},
+        context,
+        turnScopedExecContext(),
+        { timeout: 1000, onEvent, onBeforeDispatch: (event) => event.type === 'tool' }
+      )
+
+      const canonicalId = scopeProviderToolCallId('shared-provider-call', identity)
+      expect(onEvent).toHaveBeenCalledWith(
+        expect.objectContaining({
+          type: 'tool',
+          payload: expect.objectContaining({
+            toolCallId: canonicalId,
+            arguments: { path: 'files', toolCallId: 'unchanged-user-argument' },
+          }),
+        })
+      )
+      expect(context.awaitingAsyncContinuation?.pendingToolCallIds).toEqual([canonicalId])
+    }
+  })
+
   it('decodes complete escapes and stops at incomplete unicode escapes', () => {
     expect(decodeJsonStringPrefix('hello\\nworld')).toBe('hello\nworld')
     expect(decodeJsonStringPrefix('emoji \\u263A')).toBe('emoji ☺')
diff --git a/apps/sim/lib/copilot/request/go/stream.ts b/apps/sim/lib/copilot/request/go/stream.ts
index 9608589cb07..e1effef338e 100644
--- a/apps/sim/lib/copilot/request/go/stream.ts
+++ b/apps/sim/lib/copilot/request/go/stream.ts
@@ -20,6 +20,7 @@ import {
   processFilePreviewStreamEvent,
 } from '@/lib/copilot/request/go/file-preview-adapter'
 import { FatalSseEventError, processSSEStream } from '@/lib/copilot/request/go/parser'
+import { scopeProviderToolCallEvent } from '@/lib/copilot/request/go/tool-call-identity'
 import {
   handleSubagentRouting,
   prePersistClientExecutableToolCall,
@@ -326,7 +327,15 @@ export async function runStreamLoop(
         }
 
         const envelope = parsedEvent.event
-        const streamEvent = eventToStreamEvent(envelope)
+        let streamEvent: ReturnType<typeof eventToStreamEvent>
+        try {
+          streamEvent = scopeProviderToolCallEvent(
+            eventToStreamEvent(envelope),
+            context.providerToolCallIdentity
+          )
+        } catch (error) {
+          throw new FatalSseEventError(getErrorMessage(error))
+        }
         if (envelope.trace?.requestId) {
           const goTraceId = envelope.trace.goTraceId || envelope.trace.requestId
           context.trace.setGoTraceId(goTraceId)
diff --git a/apps/sim/lib/copilot/request/go/tool-call-identity.test.ts b/apps/sim/lib/copilot/request/go/tool-call-identity.test.ts
new file mode 100644
index 00000000000..f4d0e138a09
--- /dev/null
+++ b/apps/sim/lib/copilot/request/go/tool-call-identity.test.ts
@@ -0,0 +1,202 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import {
+  createProviderToolCallIdentity,
+  PROVIDER_TOOL_CALL_IDENTITY_LIMITS,
+  restoreProviderToolCallId,
+  scopeProviderToolCallEvent,
+  scopeProviderToolCallId,
+} from '@/lib/copilot/request/go/tool-call-identity'
+import {
+  markToolResultSeen,
+  shouldSkipToolCallEvent,
+  shouldSkipToolResultEvent,
+} from '@/lib/copilot/request/sse-utils'
+import type { StreamEvent } from '@/lib/copilot/request/types'
+
+function toolCall(toolCallId: string): StreamEvent {
+  return {
+    type: 'tool',
+    payload: {
+      phase: 'call',
+      toolCallId,
+      toolName: 'glob',
+      executor: 'client',
+      mode: 'async',
+      arguments: { toolCallId, nested: { tool_call_id: toolCallId } },
+    },
+  }
+}
+
+describe('provider tool-call identity boundary', () => {
+  it('bounds retained entries while allowing replay at the limit', () => {
+    const identity = createProviderToolCallIdentity('bounded-run')
+    const first = scopeProviderToolCallId('0', identity)
+    for (let index = 1; index < PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxEntries; index++) {
+      scopeProviderToolCallId(String(index), identity)
+    }
+    const bytes = identity.retainedBytes
+    expect(scopeProviderToolCallId('0', identity)).toBe(first)
+    expect(identity.retainedBytes).toBe(bytes)
+    expect(() => scopeProviderToolCallId('one-more', identity)).toThrow('identity budget exceeded')
+    expect(identity.providerIds.size).toBe(PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxEntries)
+  })
+
+  it('bounds provider IDs and cumulative string bytes without evicting pending mappings', () => {
+    const identity = createProviderToolCallIdentity('byte-bounded-run')
+    const { maxIdChars, maxRetainedBytes } = PROVIDER_TOOL_CALL_IDENTITY_LIMITS
+    expect(() => scopeProviderToolCallId('x'.repeat(maxIdChars + 1), identity)).toThrow(
+      'supported length'
+    )
+    expect(identity.providerIds.size).toBe(0)
+    const raw = 'x'.repeat(maxIdChars)
+    const first = scopeProviderToolCallId(raw, identity)
+    for (
+      let index = 1;
+      identity.retainedBytes + 2 * (maxIdChars + 73) <= maxRetainedBytes;
+      index++
+    ) {
+      scopeProviderToolCallId(String(index).padEnd(maxIdChars, 'x'), identity)
+    }
+    expect(() => scopeProviderToolCallId('y'.repeat(maxIdChars), identity)).toThrow(
+      'identity budget exceeded'
+    )
+    expect(identity.retainedBytes).toBeLessThanOrEqual(maxRetainedBytes)
+    expect(restoreProviderToolCallId(first, identity)).toBe(raw)
+  })
+
+  it('isolates runs, is stable on replay, and keeps long provider IDs within native limits', () => {
+    const first = createProviderToolCallIdentity('run-1')
+    const second = createProviderToolCallIdentity('run-2')
+    const rawId = 'provider-'.repeat(1000)
+    const firstId = scopeProviderToolCallId(rawId, first)
+    const secondId = scopeProviderToolCallId(rawId, second)
+
+    expect(firstId).not.toBe(secondId)
+    expect(firstId.length).toBeLessThanOrEqual(256)
+    expect(scopeProviderToolCallId(rawId, first)).toBe(firstId)
+    expect(restoreProviderToolCallId(firstId, first)).toBe(rawId)
+    expect(first.providerIds.size).toBe(1)
+  })
+
+  it('treats a provider ID that resembles a Sim ID as a new opaque input', () => {
+    const identity = createProviderToolCallIdentity('prefix-run')
+    const rawId = `sim_tool_${'a'.repeat(64)}`
+    const canonicalId = scopeProviderToolCallId(rawId, identity)
+    expect(canonicalId).not.toBe(rawId)
+    expect(restoreProviderToolCallId(canonicalId, identity)).toBe(rawId)
+    expect(restoreProviderToolCallId(rawId)).toBe(rawId)
+  })
+
+  it('refuses to send a canonical ID to Go after losing its reverse mapping', () => {
+    const canonicalId = scopeProviderToolCallId('call-1', createProviderToolCallIdentity('run'))
+    expect(() =>
+      restoreProviderToolCallId(canonicalId, createProviderToolCallIdentity('run'))
+    ).toThrow('Provider tool call identity is missing')
+    expect(restoreProviderToolCallId('legacy-call-1')).toBe('legacy-call-1')
+  })
+
+  it('normalizes before global call/result dedupe so one run cannot suppress another', () => {
+    const first = createProviderToolCallIdentity('dedupe-run-1')
+    const second = createProviderToolCallIdentity('dedupe-run-2')
+    const firstEvent = scopeProviderToolCallEvent(toolCall('provider-shared-call'), first)
+    const secondEvent = scopeProviderToolCallEvent(toolCall('provider-shared-call'), second)
+    expect(shouldSkipToolCallEvent(firstEvent)).toBe(false)
+    expect(shouldSkipToolCallEvent(firstEvent)).toBe(true)
+    markToolResultSeen(scopeProviderToolCallId('provider-shared-call', first))
+    expect(shouldSkipToolCallEvent(secondEvent)).toBe(false)
+    expect(
+      shouldSkipToolResultEvent(
+        scopeProviderToolCallEvent(
+          {
+            type: 'tool',
+            payload: {
+              phase: 'result',
+              toolCallId: 'provider-shared-call',
+              toolName: 'glob',
+              executor: 'client',
+              mode: 'async',
+              success: true,
+            },
+          },
+          second
+        )
+      )
+    ).toBe(false)
+  })
+
+  it('maps checkpoint-only IDs and all parent references consistently without touching payloads', () => {
+    const identity = createProviderToolCallIdentity('checkpoint-run')
+    const checkpoint: StreamEvent = {
+      type: 'run',
+      scope: { lane: 'subagent', parentToolCallId: 'parent' },
+      payload: {
+        kind: 'checkpoint_pause',
+        checkpointId: 'checkpoint',
+        executionId: 'execution',
+        runId: 'provider-run',
+        pendingToolCallIds: ['pending-only'],
+        frames: [
+          {
+            parentToolCallId: 'parent',
+            parentToolName: 'files',
+            pendingToolIds: ['pending-only'],
+            checkpointId: 'child-checkpoint',
+          },
+        ],
+      },
+    }
+    const mapped = scopeProviderToolCallEvent(checkpoint, identity)
+    const parentId = scopeProviderToolCallId('parent', identity)
+    const pendingId = scopeProviderToolCallId('pending-only', identity)
+    expect(mapped).toEqual({
+      ...checkpoint,
+      scope: { lane: 'subagent', parentToolCallId: parentId },
+      payload: {
+        ...checkpoint.payload,
+        pendingToolCallIds: [pendingId],
+        frames: [
+          {
+            parentToolCallId: parentId,
+            parentToolName: 'files',
+            pendingToolIds: [pendingId],
+            checkpointId: 'child-checkpoint',
+          },
+        ],
+      },
+    })
+    expect(restoreProviderToolCallId(pendingId, identity)).toBe('pending-only')
+    expect(checkpoint.payload.pendingToolCallIds).toEqual(['pending-only'])
+
+    const call = toolCall('pending-only')
+    expect(scopeProviderToolCallEvent(call, identity)).toEqual({
+      ...call,
+      payload: { ...call.payload, toolCallId: pendingId },
+    })
+    expect(scopeProviderToolCallEvent(call)).toBe(call)
+  })
+
+  it('maps both subagent span ID spellings and leaves structured result data opaque', () => {
+    const identity = createProviderToolCallIdentity('span-run')
+    const data = { tool_call_id: 'parent', toolCallId: 'parent', nested: { toolCallId: 'parent' } }
+    const span: StreamEvent = {
+      type: 'span',
+      payload: { kind: 'subagent', event: 'start', data },
+    }
+    const parentId = scopeProviderToolCallId('parent', identity)
+    expect(scopeProviderToolCallEvent(span, identity)).toEqual({
+      ...span,
+      payload: {
+        ...span.payload,
+        data: { ...data, tool_call_id: parentId, toolCallId: parentId },
+      },
+    })
+    const result: StreamEvent = {
+      type: 'span',
+      payload: { kind: 'structured_result', data },
+    }
+    expect(scopeProviderToolCallEvent(result, identity)).toBe(result)
+  })
+})
diff --git a/apps/sim/lib/copilot/request/go/tool-call-identity.ts b/apps/sim/lib/copilot/request/go/tool-call-identity.ts
new file mode 100644
index 00000000000..8fa70259a2e
--- /dev/null
+++ b/apps/sim/lib/copilot/request/go/tool-call-identity.ts
@@ -0,0 +1,119 @@
+import { createHash } from 'node:crypto'
+import { isRecordLike } from '@sim/utils/object'
+import type { StreamEvent } from '@/lib/copilot/request/types'
+
+export const PROVIDER_TOOL_CALL_IDENTITY_LIMITS = {
+  maxEntries: 10_000,
+  maxIdChars: 16_384,
+  maxRetainedBytes: 4 * 1024 * 1024,
+} as const
+
+export interface ProviderToolCallIdentity {
+  namespace: string
+  providerIds: Map<string, string>
+  retainedBytes: number
+}
+
+export function createProviderToolCallIdentity(namespace: string): ProviderToolCallIdentity {
+  return { namespace, providerIds: new Map(), retainedBytes: 0 }
+}
+
+/** Provider IDs are only unique within a run; Sim persists and publishes globally unique IDs. */
+export function scopeProviderToolCallId(
+  providerId: string,
+  identity: ProviderToolCallIdentity
+): string {
+  if (providerId.length > PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxIdChars) {
+    throw new Error('Provider tool call ID exceeds the supported length')
+  }
+  const digest = createHash('sha256')
+    .update(JSON.stringify([identity.namespace, providerId]))
+    .digest('hex')
+  const toolCallId = `sim_tool_${digest}`
+  if (identity.providerIds.has(toolCallId)) return toolCallId
+  /** Two bytes per UTF-16 code unit conservatively bounds retained string storage. */
+  const retainedBytes = identity.retainedBytes + 2 * (toolCallId.length + providerId.length)
+  if (
+    identity.providerIds.size >= PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxEntries ||
+    retainedBytes > PROVIDER_TOOL_CALL_IDENTITY_LIMITS.maxRetainedBytes
+  ) {
+    throw new Error('Provider tool call identity budget exceeded')
+  }
+  identity.providerIds.set(toolCallId, providerId)
+  identity.retainedBytes = retainedBytes
+  return toolCallId
+}
+
+/** Only runs that opted into namespacing decode IDs on the return trip to Go. */
+export function restoreProviderToolCallId(
+  toolCallId: string,
+  identity?: ProviderToolCallIdentity
+): string {
+  if (!identity) return toolCallId
+  const providerId = identity.providerIds.get(toolCallId)
+  if (providerId !== undefined) return providerId
+  if (/^sim_tool_[a-f0-9]{64}$/.test(toolCallId)) {
+    throw new Error('Provider tool call identity is missing from the active run')
+  }
+  return toolCallId
+}
+
+/** Translate protocol references without changing opaque tool arguments, results, or resources. */
+export function scopeProviderToolCallEvent(
+  event: StreamEvent,
+  identity?: ProviderToolCallIdentity
+): StreamEvent {
+  if (!identity) return event
+  const scopeId = (id: string) => scopeProviderToolCallId(id, identity)
+  const scoped = event.scope?.parentToolCallId
+    ? {
+        ...event,
+        scope: { ...event.scope, parentToolCallId: scopeId(event.scope.parentToolCallId) },
+      }
+    : event
+
+  if (scoped.type === 'tool') {
+    const mapped = { ...scoped }
+    mapped.payload = { ...scoped.payload, toolCallId: scopeId(scoped.payload.toolCallId) }
+    return mapped
+  }
+  if (scoped.type === 'run' && scoped.payload.kind === 'checkpoint_pause') {
+    return {
+      ...scoped,
+      payload: {
+        ...scoped.payload,
+        pendingToolCallIds: scoped.payload.pendingToolCallIds.map(scopeId),
+        ...(scoped.payload.frames
+          ? {
+              frames: scoped.payload.frames.map((frame) => ({
+                ...frame,
+                parentToolCallId: scopeId(frame.parentToolCallId),
+                pendingToolIds: frame.pendingToolIds.map(scopeId),
+              })),
+            }
+          : {}),
+      },
+    }
+  }
+  if (
+    scoped.type === 'span' &&
+    scoped.payload.kind === 'subagent' &&
+    isRecordLike(scoped.payload.data)
+  ) {
+    const mapped = { ...scoped }
+    mapped.payload = {
+      ...scoped.payload,
+      data: {
+        ...scoped.payload.data,
+        ...(typeof scoped.payload.data.tool_call_id === 'string'
+          ? { tool_call_id: scopeId(scoped.payload.data.tool_call_id) }
+          : {}),
+        ...(typeof scoped.payload.data.toolCallId === 'string'
+          ? { toolCallId: scopeId(scoped.payload.data.toolCallId) }
+          : {}),
+      },
+    }
+    return mapped
+  }
+  return scoped
+}
diff --git a/apps/sim/lib/copilot/request/handlers/handlers.test.ts b/apps/sim/lib/copilot/request/handlers/handlers.test.ts
index 5e3b77b2427..33b1e718d59 100644
--- a/apps/sim/lib/copilot/request/handlers/handlers.test.ts
+++ b/apps/sim/lib/copilot/request/handlers/handlers.test.ts
@@ -4,6 +4,7 @@
 
 import { sleep } from '@sim/utils/helpers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import { TraceCollector } from '@/lib/copilot/request/trace'
 
 const { isSimExecuted, executeTool, ensureHandlersRegistered, toolRequiresApproval } = vi.hoisted(
@@ -150,6 +151,26 @@ describe('sse-handlers tool lifecycle', () => {
     }
   })
 
+  it('propagates an ownership conflict before a client call can be forwarded', async () => {
+    isSimExecuted.mockReturnValue(false)
+    context.runId = 'current-run'
+    upsertAsyncToolCall.mockRejectedValueOnce(new AsyncToolCallOwnershipError())
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'colliding-call',
+        toolName: 'run_workflow',
+        arguments: {},
+        executor: 'client',
+        mode: 'async',
+        phase: 'call',
+      },
+    }
+    await expect(
+      prePersistClientExecutableToolCall(event, context, {}, execContext)
+    ).rejects.toBeInstanceOf(AsyncToolCallOwnershipError)
+  })
+
   it('pins the workflow target into the args it persists and forwards', async () => {
     // The browser resolved its own target from the open tab while the server
     // resolved the run's workflow; in a workspace chat those disagreed and every
diff --git a/apps/sim/lib/copilot/request/handlers/tool.ts b/apps/sim/lib/copilot/request/handlers/tool.ts
index 1a21880d70a..cc88538a292 100644
--- a/apps/sim/lib/copilot/request/handlers/tool.ts
+++ b/apps/sim/lib/copilot/request/handlers/tool.ts
@@ -2,6 +2,7 @@ import { isCurrentBrowserToolName } from '@sim/browser-protocol'
 import { createLogger } from '@sim/logger'
 import { isTerminalToolName } from '@sim/terminal-protocol'
 import { getErrorMessage, toError } from '@sim/utils/errors'
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import type {
   AsyncCompletionSignal,
   AsyncTerminalCompletionSnapshot,
@@ -277,6 +278,7 @@ export async function prePersistClientExecutableToolCall(
         ? MothershipStreamV1AsyncToolRecordStatus.pending
         : MothershipStreamV1AsyncToolRecordStatus.running,
   }).catch((err) => {
+    if (err instanceof AsyncToolCallOwnershipError) throw err
     logger.warn('Failed to pre-persist async tool row before forwarding call frame', {
       toolCallId: data.toolCallId,
       toolName: data.toolName,
diff --git a/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts b/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts
index 8ba55a24485..02fa8bb049d 100644
--- a/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts
+++ b/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts
@@ -1,6 +1,11 @@
 import { describe, expect, it, vi } from 'vitest'
 import { MothershipStreamV1CompletionStatus } from '@/lib/copilot/generated/mothership-stream-v1'
 import { createStreamingContext } from '@/lib/copilot/request/context/request-context'
+import {
+  createProviderToolCallIdentity,
+  restoreProviderToolCallId,
+  scopeProviderToolCallId,
+} from '@/lib/copilot/request/go/tool-call-identity'
 
 /** Table side effects are not exercised here, and the real module loads the table application layer. */
 vi.mock('@/lib/copilot/request/tools/tables', () => ({
@@ -21,6 +26,22 @@ import { makeResumeLegContext, mergeResumeLegOutputs } from '@/lib/copilot/reque
 // reset per leg but folded back only for a turn-level abort, because a fanout
 // cancelling its own lanes must not mark the shared turn aborted.
 describe('resume leg context isolate/merge contract', () => {
+  it('shares provider identity mappings across sibling legs and retries', () => {
+    const identity = createProviderToolCallIdentity('run-1')
+    const base = createStreamingContext({ providerToolCallIdentity: identity })
+    const first = makeResumeLegContext(base)
+    const second = makeResumeLegContext(base)
+    const canonicalId = scopeProviderToolCallId('child-call', identity)
+
+    expect(first.providerToolCallIdentity).toBe(identity)
+    expect(second.providerToolCallIdentity).toBe(identity)
+    expect(restoreProviderToolCallId(canonicalId, second.providerToolCallIdentity)).toBe(
+      'child-call'
+    )
+    expect(scopeProviderToolCallId('child-call', identity)).toBe(canonicalId)
+    expect(makeResumeLegContext(createStreamingContext()).providerToolCallIdentity).toBeUndefined()
+  })
+
   it('isolates the per-leg scalars while sharing the heavy accumulators by reference', () => {
     const base = createStreamingContext({
       accumulatedContent: 'PRE',
diff --git a/apps/sim/lib/copilot/request/lifecycle/run.test.ts b/apps/sim/lib/copilot/request/lifecycle/run.test.ts
index 99ffba1adec..dfa163a5a4b 100644
--- a/apps/sim/lib/copilot/request/lifecycle/run.test.ts
+++ b/apps/sim/lib/copilot/request/lifecycle/run.test.ts
@@ -4,6 +4,7 @@
 
 import { resetEnvFlagsMock, resetEnvironmentUtilsMock, setEnvFlags } from '@sim/testing'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import { scopeProviderToolCallId } from '@/lib/copilot/request/go/tool-call-identity'
 import type { ExecutionContext, StreamingContext } from '@/lib/copilot/request/types'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 
@@ -1052,6 +1053,108 @@ describe('runCopilotLifecycle', () => {
     expect(mockRunStreamLoop).toHaveBeenCalledOnce()
   })
 
+  it('isolates independent headless starts that reuse a client-supplied message ID', async () => {
+    const namespaces: string[] = []
+    const canonicalIds: string[] = []
+    mockRunStreamLoop.mockImplementation(
+      async (_url: string, _options: RequestInit, context: StreamingContext) => {
+        if (!context.providerToolCallIdentity) throw new Error('Identity map not initialized')
+        namespaces.push(context.providerToolCallIdentity.namespace)
+        canonicalIds.push(
+          scopeProviderToolCallId('shared-provider-id', context.providerToolCallIdentity)
+        )
+      }
+    )
+
+    for (const userId of ['first-user', 'second-user']) {
+      const result = await runCopilotLifecycle(
+        { message: 'find files', messageId: 'reused-client-message-id' },
+        {
+          userId,
+          executionContext: { userId, workflowId: '' },
+          interactive: false,
+        }
+      )
+      expect(result.success).toBe(true)
+    }
+
+    expect(namespaces).toHaveLength(2)
+    expect(namespaces[0]).not.toBe(namespaces[1])
+    expect(namespaces).not.toContain('reused-client-message-id')
+    expect(canonicalIds[0]).not.toBe(canonicalIds[1])
+  })
+
+  it('returns the original provider call ID at the shared Go resume boundary', async () => {
+    let canonicalId = ''
+    mockRunStreamLoop.mockImplementationOnce(
+      async (_url: string, _options: RequestInit, context: StreamingContext) => {
+        expect(context.providerToolCallIdentity).toBeDefined()
+        if (!context.providerToolCallIdentity) throw new Error('Identity map not initialized')
+        canonicalId = scopeProviderToolCallId(
+          'provider-shared-call',
+          context.providerToolCallIdentity
+        )
+        context.toolCalls.set(canonicalId, {
+          id: canonicalId,
+          name: 'glob',
+          status: 'success',
+          result: { success: true, output: { files: [] } },
+        })
+        context.awaitingAsyncContinuation = {
+          checkpointId: 'identity-checkpoint',
+          pendingToolCallIds: [canonicalId],
+        }
+      }
+    )
+    mockRunStreamLoop.mockImplementationOnce(async () => {})
+
+    const result = await runCopilotLifecycle(
+      { message: 'find files', messageId: 'identity-stream' },
+      {
+        userId: 'user-1',
+        runId: 'identity-run',
+        executionContext: { userId: 'user-1', workflowId: '' },
+      }
+    )
+
+    expect(result.success).toBe(true)
+    expect(canonicalId).not.toBe('provider-shared-call')
+    const resumeBody = JSON.parse(String(mockRunStreamLoop.mock.calls[1]?.[1].body))
+    expect(resumeBody.results).toEqual([
+      { callId: 'provider-shared-call', name: 'glob', data: { files: [] }, success: true },
+    ])
+  })
+
+  it('does not resume Go with a canonical ID after its reverse mapping is lost', async () => {
+    mockRunStreamLoop.mockImplementationOnce(
+      async (_url: string, _options: RequestInit, context: StreamingContext) => {
+        if (!context.providerToolCallIdentity) throw new Error('Identity map not initialized')
+        const canonicalId = scopeProviderToolCallId('call-lost', context.providerToolCallIdentity)
+        context.providerToolCallIdentity.providerIds.clear()
+        context.toolCalls.set(canonicalId, {
+          id: canonicalId,
+          name: 'glob',
+          status: 'success',
+          result: { success: true, output: { files: [] } },
+        })
+        context.awaitingAsyncContinuation = {
+          checkpointId: 'lost-identity-checkpoint',
+          pendingToolCallIds: [canonicalId],
+        }
+      }
+    )
+    const result = await runCopilotLifecycle(
+      { message: 'find files', messageId: 'lost-identity-stream' },
+      {
+        userId: 'user-1',
+        runId: 'lost-identity-run',
+        executionContext: { userId: 'user-1', workflowId: '' },
+      }
+    )
+    expect(result.success).toBe(false)
+    expect(mockRunStreamLoop).toHaveBeenCalledOnce()
+  })
+
   describe('tool permission feature flag', () => {
     const runMothershipTurn = () =>
       runCopilotLifecycle(
@@ -2073,13 +2176,13 @@ describe('runCopilotLifecycle', () => {
       // Mirror the real helper: settle the tool call into a terminal error
       // state so the resume loop can serialize an error result for it.
       mockForceFailHungToolCall.mockImplementation(
-        async (toolCallId: string, context: StreamingContext, message: string) => {
+        async (toolCallId: string, context: StreamingContext) => {
           const tool = context.toolCalls.get(toolCallId)
           if (!tool) return
           tool.status = MothershipStreamV1ToolOutcome.error
           tool.endTime = Date.now()
           tool.result = { success: false }
-          tool.error = message
+          tool.error = 'Tool execution hung'
         }
       )
 
@@ -2138,7 +2241,7 @@ describe('runCopilotLifecycle', () => {
       expect(mockForceFailHungToolCall).toHaveBeenCalledWith(
         'tool-hung',
         expect.anything(),
-        expect.stringContaining('hung')
+        expect.objectContaining({ userId: 'user-1' })
       )
       expect(fetchUrls[1]).toBe('http://mothership.test/api/tools/resume')
       expect(bodies[1].results).toEqual([
@@ -2239,13 +2342,13 @@ describe('runCopilotLifecycle', () => {
         toolCall?.status === 'awaiting_approval' ? 3_600_000 : 60_000
       )
       mockForceFailHungToolCall.mockImplementation(
-        async (toolCallId: string, context: StreamingContext, message: string) => {
+        async (toolCallId: string, context: StreamingContext) => {
           const tool = context.toolCalls.get(toolCallId)
           if (!tool) return
           tool.status = MothershipStreamV1ToolOutcome.error
           tool.endTime = Date.now()
           tool.result = { success: false }
-          tool.error = message
+          tool.error = 'Tool execution hung'
         }
       )
 
@@ -2311,7 +2414,7 @@ describe('runCopilotLifecycle', () => {
       expect(mockForceFailHungToolCall).toHaveBeenCalledWith(
         'tool-hung',
         expect.anything(),
-        expect.stringContaining('hung')
+        expect.objectContaining({ userId: 'user-1' })
       )
       expect(lifecycleSettled).toBe(false)
       expect(fetchUrls).toEqual(['http://mothership.test/api/copilot'])
@@ -2431,13 +2534,13 @@ describe('runCopilotLifecycle', () => {
         toolCall?.status === 'awaiting_approval' ? 3_600_000 : 60_000
       )
       mockForceFailHungToolCall.mockImplementation(
-        async (toolCallId: string, context: StreamingContext, message: string) => {
+        async (toolCallId: string, context: StreamingContext) => {
           const tool = context.toolCalls.get(toolCallId)
           if (!tool) return
           tool.status = MothershipStreamV1ToolOutcome.error
           tool.endTime = Date.now()
           tool.result = { success: false }
-          tool.error = message
+          tool.error = 'Tool execution hung'
         }
       )
 
@@ -2512,7 +2615,7 @@ describe('runCopilotLifecycle', () => {
       expect(mockForceFailHungToolCall).toHaveBeenCalledWith(
         'tool-replaced',
         expect.anything(),
-        expect.stringContaining('hung')
+        expect.objectContaining({ userId: 'user-1' })
       )
       expect(lifecycleSettled).toBe(false)
       expect(fetchUrls).toEqual(['http://mothership.test/api/copilot'])
diff --git a/apps/sim/lib/copilot/request/lifecycle/run.ts b/apps/sim/lib/copilot/request/lifecycle/run.ts
index e532d15ba1c..13ba64e020a 100644
--- a/apps/sim/lib/copilot/request/lifecycle/run.ts
+++ b/apps/sim/lib/copilot/request/lifecycle/run.ts
@@ -36,6 +36,10 @@ import {
   runStreamLoop,
   StreamEndedWithoutTerminalError,
 } from '@/lib/copilot/request/go/stream'
+import {
+  createProviderToolCallIdentity,
+  restoreProviderToolCallId,
+} from '@/lib/copilot/request/go/tool-call-identity'
 import { recordDegraded } from '@/lib/copilot/request/metrics'
 import { AbortReason } from '@/lib/copilot/request/session/abort-reason'
 import {
@@ -373,6 +377,10 @@ export async function runCopilotLifecycle(
     executionId: resolvedExecutionId,
     runId: resolvedRunId,
     messageId: payloadMsgId,
+    providerToolCallIdentity:
+      goRoute === '/api/tools/resume'
+        ? undefined
+        : createProviderToolCallIdentity(resolvedRunId ?? generateId()),
     toolPermissions: await resolveToolPermissions(lifecycleOptions),
     ...(lifecycleOptions.trace ? { trace: lifecycleOptions.trace } : {}),
   })
@@ -653,6 +661,7 @@ function buildResumeToolResult(
   checkpointId: string | undefined
 ): ResumeToolResult {
   const tool = context.toolCalls.get(toolCallId)
+  const providerToolCallId = restoreProviderToolCallId(toolCallId, context.providerToolCallIdentity)
   if (!tool || !tool.result) {
     recordDegraded(CopilotDegradedReason.MissingToolResult)
     logger.error('Missing tool result for pending tool call; synthesizing a failure', {
@@ -664,14 +673,14 @@ function buildResumeToolResult(
       hasPendingPromise: context.pendingToolPromises.has(toolCallId),
     })
     return {
-      callId: toolCallId,
+      callId: providerToolCallId,
       name: tool?.name || '',
       data: { error: `no result was returned for tool call ${toolCallId}` },
       success: false,
     }
   }
   return {
-    callId: toolCallId,
+    callId: providerToolCallId,
     name: tool.name || '',
     data: getToolCallTerminalData(tool),
     success: requireToolCallStateResult(tool).success,
@@ -1202,11 +1211,7 @@ async function runCheckpointLoop(
                   waitBudgetMs: watchdog.waitBudgetMs,
                 }
               )
-              await forceFailHungToolCall(
-                toolCallId,
-                context,
-                'Tool execution hung on the Sim executor and was abandoned so the conversation could continue.'
-              )
+              await forceFailHungToolCall(toolCallId, context, execContext)
               if (context.pendingToolPromises.get(toolCallId) === watchdog.promise) {
                 context.pendingToolPromises.delete(toolCallId)
               }
diff --git a/apps/sim/lib/copilot/request/tools/executor.test.ts b/apps/sim/lib/copilot/request/tools/executor.test.ts
index a378f1b290e..c30ca83db03 100644
--- a/apps/sim/lib/copilot/request/tools/executor.test.ts
+++ b/apps/sim/lib/copilot/request/tools/executor.test.ts
@@ -1,3 +1,6 @@
+/**
+ * @vitest-environment node
+ */
 import '@sim/testing/mocks/executor'
 
 import { beforeEach, describe, expect, it, vi } from 'vitest'
@@ -11,10 +14,22 @@ const {
   recordSimToolMetric,
   setAttribute,
   withCopilotToolSpan,
+  encryptSecret,
+  decryptSecret,
+  publishToolConfirmation,
+  waitForToolConfirmation,
+  replaceTerminalAsyncToolCallResult,
+  mockError,
 } = vi.hoisted(() => {
   const setAttribute = vi.fn()
   return {
     executeTool: vi.fn(),
+    encryptSecret: vi.fn(),
+    decryptSecret: vi.fn(),
+    publishToolConfirmation: vi.fn(),
+    waitForToolConfirmation: vi.fn(),
+    replaceTerminalAsyncToolCallResult: vi.fn(),
+    mockError: vi.fn(),
     completeAsyncToolCall: vi.fn(),
     markAsyncToolRunning: vi.fn(),
     upsertAsyncToolCall: vi.fn(),
@@ -28,6 +43,16 @@ const {
   }
 })
 
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({ error: mockError, warn: vi.fn(), info: vi.fn(), debug: vi.fn() }),
+}))
+
+vi.mock('@/lib/core/security/encryption', () => ({ encryptSecret, decryptSecret }))
+
+vi.mock('@/lib/workflows/executor/execution-state', () => ({
+  getTrustedWorkflowToolExecution: vi.fn(),
+}))
+
 vi.mock('@/lib/copilot/tool-executor', () => ({
   ensureHandlersRegistered: vi.fn(),
   executeTool,
@@ -37,10 +62,12 @@ vi.mock('@/lib/copilot/async-runs/repository', () => ({
   completeAsyncToolCall,
   markAsyncToolRunning,
   upsertAsyncToolCall,
+  replaceTerminalAsyncToolCallResult,
 }))
 
 vi.mock('@/lib/copilot/persistence/tool-confirm', () => ({
-  publishToolConfirmation: vi.fn(),
+  publishToolConfirmation,
+  waitForToolConfirmation,
 }))
 
 vi.mock('@/lib/copilot/request/metrics', () => ({
@@ -72,6 +99,7 @@ vi.mock('@/lib/copilot/request/tools/workflow-context', () => ({
   applyCreateWorkflowOutputToContext: vi.fn(),
 }))
 
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import { TOOL_WATCHDOG_DEFAULT_MS, TOOL_WATCHDOG_LONG_RUNNING_MS } from '@/lib/copilot/constants'
 import {
   MothershipStreamV1EventType,
@@ -80,12 +108,24 @@ import {
 } from '@/lib/copilot/generated/mothership-stream-v1'
 import { GenerateApiKey } from '@/lib/copilot/generated/tool-catalog-v1'
 import { createStreamingContext } from '@/lib/copilot/request/context/request-context'
+import { handleClientCompletion } from '@/lib/copilot/request/handlers/types'
+import { waitForClientToolCompletion } from '@/lib/copilot/request/tools/client'
+import {
+  sealClientToolCompletion,
+  sealClientToolContext,
+} from '@/lib/copilot/request/tools/client-completion-seal.server'
 import {
   buildToolExecutionContext,
   executeToolAndReport,
+  forceFailHungToolCall,
   pendingToolWaitBudgetMs,
   toolWatchdogTimeoutMs,
 } from '@/lib/copilot/request/tools/executor'
+import { maybeWriteOutputToFile } from '@/lib/copilot/request/tools/files'
+import {
+  maybeWriteOutputToTable,
+  maybeWriteReadCsvToTable,
+} from '@/lib/copilot/request/tools/tables'
 import type { ExecutionContext, ToolCallState } from '@/lib/copilot/request/types'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 
@@ -233,6 +273,19 @@ describe('executeToolAndReport provenance isolation', () => {
     upsertAsyncToolCall.mockResolvedValue(null)
   })
 
+  it('does not execute or mutate a tool row owned by another run', async () => {
+    const toolCall = buildPendingToolCall()
+    const conflict = new AsyncToolCallOwnershipError()
+    upsertAsyncToolCall.mockRejectedValueOnce(conflict)
+
+    await expect(
+      executeToolAndReport(toolCall.id, buildStreamingContext(toolCall), { userId: 'user-1' })
+    ).rejects.toBe(conflict)
+
+    expect(markAsyncToolRunning).not.toHaveBeenCalled()
+    expect(executeTool).not.toHaveBeenCalled()
+  })
+
   it('merges a complete child only after its projected result is safe', async () => {
     const registry = new ResolvedSecretTraceRegistry([
       { name: 'TOKEN', plaintext: 'secret-value', encryptedValue: 'ciphertext' },
@@ -492,3 +545,339 @@ describe('executeToolAndReport metrics', () => {
     }
   )
 })
+
+describe('watchdog completion provenance', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    encryptSecret.mockImplementation(async (plaintext: string) => ({ encrypted: plaintext }))
+    decryptSecret.mockImplementation(async (encrypted: string) => ({ decrypted: encrypted }))
+    completeAsyncToolCall.mockImplementation(async (input) => ({ ...input }))
+    replaceTerminalAsyncToolCallResult.mockImplementation(async (input) => ({ ...input }))
+  })
+
+  function createHungClient() {
+    const registry = new ResolvedSecretTraceRegistry([
+      { name: 'TOKEN', plaintext: 'private-command-token', encryptedValue: 'ciphertext' },
+    ])
+    registry.recordResolved('TOKEN', 'private-command-token')
+    const toolCall: ToolCallState = {
+      id: 'terminal-call',
+      name: 'terminal_run',
+      status: 'executing',
+      params: { command: 'private-command-token' },
+    }
+    const context = buildStreamingContext(toolCall)
+    const execContext: ExecutionContext = {
+      userId: 'user-1',
+      resolvedSecretTraceRegistry: registry,
+    }
+    return { registry, toolCall, context, execContext }
+  }
+
+  it('restores a trusted timeout through the real sealed client completion reader', async () => {
+    const { registry, toolCall, context, execContext } = createHungClient()
+    const finishSiblingActivation = registry.beginPendingActivation()
+
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+    const persisted = completeAsyncToolCall.mock.calls[0][0]
+    expect(persisted.result).toEqual({
+      __sealedClientToolCompletionV1: expect.any(String),
+      __sealedClientToolContextV1: expect.any(String),
+    })
+    waitForToolConfirmation.mockResolvedValue({
+      status: 'error',
+      message: persisted.error,
+      data: persisted.result,
+    })
+
+    const completion = await waitForClientToolCompletion({
+      toolCallId: toolCall.id,
+      runId: context.runId,
+      userId: execContext.userId,
+      timeoutMs: 1,
+      registry,
+    })
+    finishSiblingActivation()
+
+    expect(completion).toEqual({
+      status: 'error',
+      message: expect.stringContaining('outcome is unknown'),
+      data: {
+        error: expect.stringContaining('hung'),
+        outcomeUnknown: true,
+        doNotRetry: true,
+      },
+    })
+    expect(registry.isComplete()).toBe(true)
+    expect(mockError).not.toHaveBeenCalledWith(
+      'Client tool provenance could not be restored',
+      expect.anything()
+    )
+    expect(
+      JSON.stringify([persisted, publishToolConfirmation.mock.calls, completion])
+    ).not.toContain('private-command-token')
+    expect(publishToolConfirmation).toHaveBeenCalledWith(
+      expect.objectContaining({ data: persisted.result })
+    )
+  })
+
+  it('preserves an actual completion that settles while encryption is pending', async () => {
+    const { toolCall, context, execContext } = createHungClient()
+    let finishEncryption: (value: { encrypted: string }) => void = () => {}
+    encryptSecret.mockImplementationOnce(
+      () =>
+        new Promise<{ encrypted: string }>((resolve) => {
+          finishEncryption = resolve
+        })
+    )
+    const settlement = forceFailHungToolCall(toolCall.id, context, execContext)
+    toolCall.status = 'success'
+    toolCall.endTime = Date.now()
+    toolCall.result = { success: true, output: 'actual completion' }
+    finishEncryption({ encrypted: 'unused' })
+    await settlement
+
+    expect(toolCall.result).toEqual({ success: true, output: 'actual completion' })
+    expect(completeAsyncToolCall).not.toHaveBeenCalled()
+    expect(publishToolConfirmation).not.toHaveBeenCalled()
+  })
+
+  it('does not publish or overwrite an actual completion that wins the durable race', async () => {
+    const { toolCall, context, execContext } = createHungClient()
+    completeAsyncToolCall.mockImplementationOnce(async () => {
+      toolCall.status = 'success'
+      toolCall.endTime = Date.now()
+      toolCall.result = { success: true, output: 'actual completion' }
+      return null
+    })
+
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+
+    expect(toolCall.status).toBe('success')
+    expect(toolCall.result).toEqual({ success: true, output: 'actual completion' })
+    expect(publishToolConfirmation).not.toHaveBeenCalled()
+  })
+
+  it('reports unavailable output locally when the durable winner has no settled live result', async () => {
+    const { registry, toolCall, context, execContext } = createHungClient()
+    completeAsyncToolCall.mockResolvedValueOnce(null)
+
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+
+    expect(toolCall.status).toBe('error')
+    expect(toolCall.error).toContain('result could not be restored')
+    expect(toolCall.result).toEqual({
+      success: false,
+      output: { error: toolCall.error, outcomeUnknown: true, doNotRetry: true },
+    })
+    expect(publishToolConfirmation).not.toHaveBeenCalled()
+    expect(registry.isComplete()).toBe(true)
+  })
+
+  it('allows the valid winning client completion to replace a local unavailable-result fallback', async () => {
+    const { registry, toolCall, context, execContext } = createHungClient()
+    completeAsyncToolCall.mockResolvedValueOnce(null)
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+    expect(toolCall.status).toBe('error')
+    const binding = { toolCallId: toolCall.id, runId: 'run-1', userId: execContext.userId }
+    waitForToolConfirmation.mockResolvedValueOnce({
+      status: 'success',
+      data: {
+        ...(await sealClientToolContext({ ...binding, registry, toolInput: undefined })),
+        ...(await sealClientToolCompletion({
+          ...binding,
+          message: 'Tool completed',
+          data: { exitCode: 0, output: 'successful output' },
+        })),
+      },
+    })
+    const completion = await waitForClientToolCompletion({
+      ...binding,
+      registry,
+      timeoutMs: 1,
+    })
+    handleClientCompletion(toolCall, toolCall.id, completion)
+
+    expect(toolCall.status).toBe('success')
+    expect(toolCall.result).toEqual({
+      success: true,
+      output: { exitCode: 0, output: 'successful output' },
+    })
+    expect(registry.isComplete()).toBe(true)
+    expect(publishToolConfirmation).not.toHaveBeenCalled()
+    expect(mockError).not.toHaveBeenCalledWith(
+      'Client tool provenance could not be restored',
+      expect.anything()
+    )
+  })
+
+  it('never falls back to an unsealed durable result when sealing fails', async () => {
+    const { registry, toolCall, context, execContext } = createHungClient()
+    encryptSecret.mockRejectedValueOnce(new Error('encryption unavailable'))
+
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+
+    expect(completeAsyncToolCall).not.toHaveBeenCalled()
+    expect(publishToolConfirmation).not.toHaveBeenCalled()
+    expect(toolCall.status).toBe('error')
+    expect(toolCall.error).toContain('outcome is unknown')
+    expect(registry.isComplete()).toBe(true)
+  })
+
+  it('retains structural failure compatibility when no provenance registry exists', async () => {
+    const { toolCall, context } = createHungClient()
+
+    await forceFailHungToolCall(toolCall.id, context, { userId: 'user-1' })
+
+    expect(toolCall.status).toBe('error')
+    expect(completeAsyncToolCall).toHaveBeenCalledWith(
+      expect.objectContaining({
+        result: { error: expect.any(String), outcomeUnknown: true, doNotRetry: true },
+      })
+    )
+    expect(encryptSecret).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    ['file output', maybeWriteOutputToFile],
+    ['table output', maybeWriteOutputToTable],
+    ['CSV output', maybeWriteReadCsvToTable],
+  ] as const)(
+    'ignores a late %s completion after watchdog settlement',
+    async (_name, postprocess) => {
+      const toolCall = buildPendingToolCall()
+      const context = buildStreamingContext(toolCall)
+      const execContext: ExecutionContext = {
+        userId: 'user-1',
+        resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+      }
+      const result = { success: true, output: 'late output' }
+      executeTool.mockResolvedValueOnce(result)
+      let finishPostprocessing: (value: typeof result) => void = () => {}
+      let startedPostprocessing: () => void = () => {}
+      const started = new Promise<void>((resolve) => {
+        startedPostprocessing = resolve
+      })
+      vi.mocked(postprocess).mockImplementationOnce(async () => {
+        startedPostprocessing()
+        return await new Promise<typeof result>((resolve) => {
+          finishPostprocessing = resolve
+        })
+      })
+      const execution = executeToolAndReport(toolCall.id, context, execContext)
+      await started
+      await forceFailHungToolCall(toolCall.id, context, execContext)
+      finishPostprocessing(result)
+      const completion = await execution
+
+      expect(completion.status).toBe('error')
+      expect(completion.message).toContain('hung')
+      expect(toolCall.status).toBe('error')
+      expect(completeAsyncToolCall).toHaveBeenCalledTimes(1)
+      expect(publishToolConfirmation).toHaveBeenCalledTimes(1)
+      expect(JSON.stringify(completion)).not.toContain('late output')
+    }
+  )
+
+  it.each([false, true])(
+    'does not publish a stale completion when the watchdog wins during persistence (aborted: %s)',
+    async (aborted) => {
+      const toolCall = buildPendingToolCall()
+      const context = buildStreamingContext(toolCall)
+      const execContext: ExecutionContext = {
+        userId: 'user-1',
+        resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+      }
+      const result = { success: true, output: 'late output' }
+      executeTool.mockResolvedValueOnce(result)
+      let finishPostprocessing: (value: typeof result) => void = () => {}
+      let startedPostprocessing: () => void = () => {}
+      const started = new Promise<void>((resolve) => {
+        startedPostprocessing = resolve
+      })
+      vi.mocked(maybeWriteOutputToFile).mockImplementationOnce(async () => {
+        startedPostprocessing()
+        return await new Promise<typeof result>((resolve) => {
+          finishPostprocessing = resolve
+        })
+      })
+      let finishWatchdogPersistence: (value: object) => void = () => {}
+      let startedWatchdogPersistence: () => void = () => {}
+      const watchdogWriting = new Promise<void>((resolve) => {
+        startedWatchdogPersistence = resolve
+      })
+      completeAsyncToolCall.mockImplementationOnce(async () => {
+        startedWatchdogPersistence()
+        return await new Promise<object>((resolve) => {
+          finishWatchdogPersistence = resolve
+        })
+      })
+      let finishToolPersistence: (value: null) => void = () => {}
+      let startedToolPersistence: () => void = () => {}
+      const toolWriting = new Promise<void>((resolve) => {
+        startedToolPersistence = resolve
+      })
+      completeAsyncToolCall.mockImplementationOnce(async () => {
+        startedToolPersistence()
+        return await new Promise<null>((resolve) => {
+          finishToolPersistence = resolve
+        })
+      })
+
+      const controller = new AbortController()
+      const execution = executeToolAndReport(toolCall.id, context, execContext, {
+        onEvent,
+        abortSignal: controller.signal,
+      })
+      await started
+      const watchdog = forceFailHungToolCall(toolCall.id, context, execContext)
+      await watchdogWriting
+      if (aborted) controller.abort()
+      finishPostprocessing(result)
+      await toolWriting
+      finishWatchdogPersistence({ status: 'failed' })
+      await watchdog
+      finishToolPersistence(null)
+      const completion = await execution
+
+      expect(completion.status).toBe('error')
+      expect(completion.message).toContain('hung')
+      expect(toolCall.status).toBe('error')
+      expect(publishToolConfirmation).toHaveBeenCalledTimes(1)
+      expect(onEvent).not.toHaveBeenCalled()
+      expect(JSON.stringify(completion)).not.toContain('late output')
+    }
+  )
+
+  it('ignores a late postprocessing rejection after watchdog settlement', async () => {
+    const toolCall = buildPendingToolCall()
+    const context = buildStreamingContext(toolCall)
+    const execContext: ExecutionContext = {
+      userId: 'user-1',
+      resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(),
+    }
+    executeTool.mockResolvedValueOnce({ success: true, output: 'late output' })
+    let rejectPostprocessing: (error: Error) => void = () => {}
+    let startedPostprocessing: () => void = () => {}
+    const started = new Promise<void>((resolve) => {
+      startedPostprocessing = resolve
+    })
+    vi.mocked(maybeWriteOutputToFile).mockImplementationOnce(async () => {
+      startedPostprocessing()
+      return await new Promise<never>((_resolve, reject) => {
+        rejectPostprocessing = reject
+      })
+    })
+    const execution = executeToolAndReport(toolCall.id, context, execContext)
+    await started
+    await forceFailHungToolCall(toolCall.id, context, execContext)
+    rejectPostprocessing(new Error('late rejected secret output'))
+    const completion = await execution
+
+    expect(completion.status).toBe('error')
+    expect(completion.message).toContain('hung')
+    expect(completeAsyncToolCall).toHaveBeenCalledTimes(1)
+    expect(publishToolConfirmation).toHaveBeenCalledTimes(1)
+    expect(JSON.stringify(completion)).not.toContain('late rejected secret output')
+  })
+})
diff --git a/apps/sim/lib/copilot/request/tools/executor.ts b/apps/sim/lib/copilot/request/tools/executor.ts
index 02a364eef68..d9c6b523e9d 100644
--- a/apps/sim/lib/copilot/request/tools/executor.ts
+++ b/apps/sim/lib/copilot/request/tools/executor.ts
@@ -2,6 +2,7 @@ import { browserToolRendererTimeoutMs, isCurrentBrowserToolName } from '@sim/bro
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import type {
   AsyncCompletionEnvelope,
   AsyncCompletionSignal,
@@ -61,6 +62,10 @@ import {
   requireToolCallError,
   setTerminalToolCallState,
 } from '@/lib/copilot/request/tool-call-state'
+import {
+  sealClientToolCompletion,
+  sealClientToolContext,
+} from '@/lib/copilot/request/tools/client-completion-seal.server'
 import { maybeWriteOutputToFile } from '@/lib/copilot/request/tools/files'
 import {
   describeWithholdingCause,
@@ -319,46 +324,85 @@ async function executeToolWithWatchdog(toolCall: ToolCallState, toolContext: Exe
   }
 }
 
+const HUNG_TOOL_MESSAGE =
+  'Tool execution hung and was abandoned so the conversation could continue. Its outcome is unknown; do not retry it automatically.'
+const UNAVAILABLE_TOOL_SETTLEMENT_MESSAGE =
+  'The tool result could not be restored before the conversation resumed. Its outcome is unknown; do not retry it automatically.'
+
 /**
- * Last-resort settlement for a tool whose promise never settled (a hang the
- * per-tool watchdog could not see, e.g. in post-processing or persistence).
- * Records a terminal error state + failed async row so the checkpoint loop
- * can resume Go with an error result instead of waiting forever.
+ * Settles an abandoned tool with a fixed server-owned failure. Client waiters consume the same
+ * sealed transport as ordinary client completions; no abandoned tool content is certified.
  */
 export async function forceFailHungToolCall(
   toolCallId: string,
   context: StreamingContext,
-  message: string
+  execContext: ExecutionContext
 ): Promise<void> {
   const toolCall = context.toolCalls.get(toolCallId)
   if (!toolCall || toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) return
+
+  const failure = { error: HUNG_TOOL_MESSAGE, outcomeUnknown: true, doNotRetry: true }
+  let durableData: unknown = failure
+  let completed = false
+  let lostSettlementRace = false
+  try {
+    if (context.runId && execContext.resolvedSecretTraceRegistry) {
+      const binding = { toolCallId, runId: context.runId, userId: execContext.userId }
+      const [completion, provenance] = await Promise.all([
+        sealClientToolCompletion({ ...binding, message: HUNG_TOOL_MESSAGE, data: failure }),
+        sealClientToolContext({
+          ...binding,
+          registry: execContext.resolvedSecretTraceRegistry,
+          /** The fixed failure contains no output or arguments from the abandoned tool. */
+          toolInput: undefined,
+        }),
+      ])
+      durableData = { ...completion, ...provenance }
+    }
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) return
+
+    completed = Boolean(
+      await completeAsyncToolCall({
+        toolCallId,
+        status: MothershipStreamV1AsyncToolRecordStatus.failed,
+        result: durableData,
+        error: HUNG_TOOL_MESSAGE,
+      })
+    )
+    if (!completed) {
+      if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) return
+      lostSettlementRace = true
+    }
+  } catch (error) {
+    logger.warn('Failed to persist force-failed async tool status', {
+      toolCallId,
+      error: toError(error).message,
+    })
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) return
+  }
+
+  /** A durable winner whose waiter is still hung must not become a fabricated local success. */
+  const message = lostSettlementRace ? UNAVAILABLE_TOOL_SETTLEMENT_MESSAGE : HUNG_TOOL_MESSAGE
   setTerminalToolCallState(toolCall, {
     status: MothershipStreamV1ToolOutcome.error,
+    output: { ...failure, error: message },
     error: message,
   })
   logger.error('Force-failed hung tool call', {
     toolCallId,
     toolName: toolCall.name,
-    message,
+    persisted: completed,
+    lostSettlementRace,
   })
   markToolResultSeen(toolCallId)
-  await completeAsyncToolCall({
-    toolCallId,
-    status: MothershipStreamV1AsyncToolRecordStatus.failed,
-    result: { error: message },
-    error: message,
-  }).catch((err) => {
-    logger.warn('Failed to persist force-failed async tool status', {
+  if (completed) {
+    publishTerminalToolConfirmation({
       toolCallId,
-      error: toError(err).message,
+      status: MothershipStreamV1ToolOutcome.error,
+      message: HUNG_TOOL_MESSAGE,
+      data: durableData,
     })
-  })
-  publishTerminalToolConfirmation({
-    toolCallId,
-    status: MothershipStreamV1ToolOutcome.error,
-    message,
-    data: { error: message },
-  })
+  }
 }
 
 function cancelledCompletion(message: string): AsyncToolCompletion {
@@ -503,6 +547,7 @@ async function executeToolAndReportInner(
   await ensureHandlersRegistered()
   if (abortRequested(context, execContext, options)) {
     markToolCallCancelled('Request aborted before tool execution')
+    const cancellationResult = toolCall.result
     markToolResultSeen(toolCall.id)
     await completeAsyncToolCall({
       toolCallId: toolCall.id,
@@ -515,6 +560,9 @@ async function executeToolAndReportInner(
         error: toError(err).message,
       })
     })
+    if (toolCall.result !== cancellationResult) {
+      return terminalCompletionFromToolCall(toolCall)
+    }
     publishTerminalToolConfirmation({
       toolCallId: toolCall.id,
       status: MothershipStreamV1ToolOutcome.cancelled,
@@ -531,6 +579,7 @@ async function executeToolAndReportInner(
     toolName: toolCall.name,
     args: toolCall.params,
   }).catch((err) => {
+    if (err instanceof AsyncToolCallOwnershipError) throw err
     logger.warn('Failed to persist async tool row before execution', {
       toolCallId: toolCall.id,
       error: toError(err).message,
@@ -616,6 +665,7 @@ async function executeToolAndReportInner(
         toolCall.name
       ).result
       markToolCallCancelled('Request aborted during tool execution')
+      const cancellationResult = toolCall.result
       markToolResultSeen(toolCall.id)
       await completeAsyncToolCall({
         toolCallId: toolCall.id,
@@ -628,6 +678,9 @@ async function executeToolAndReportInner(
           error: toError(err).message,
         })
       })
+      if (toolCall.result !== cancellationResult) {
+        return terminalCompletionFromToolCall(toolCall)
+      }
       publishTerminalToolConfirmation({
         toolCallId: toolCall.id,
         status: MothershipStreamV1ToolOutcome.cancelled,
@@ -646,8 +699,13 @@ async function executeToolAndReportInner(
       result,
       toolExecutionContext
     )
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     if (abortRequested(context, execContext, options)) {
       markToolCallCancelled('Request aborted during tool post-processing')
+      const cancellationResult = toolCall.result
       markToolResultSeen(toolCall.id)
       await completeAsyncToolCall({
         toolCallId: toolCall.id,
@@ -660,6 +718,9 @@ async function executeToolAndReportInner(
           error: toError(err).message,
         })
       })
+      if (toolCall.result !== cancellationResult) {
+        return terminalCompletionFromToolCall(toolCall)
+      }
       publishTerminalToolConfirmation({
         toolCallId: toolCall.id,
         status: MothershipStreamV1ToolOutcome.cancelled,
@@ -675,8 +736,13 @@ async function executeToolAndReportInner(
       result,
       toolExecutionContext
     )
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     if (abortRequested(context, execContext, options)) {
       markToolCallCancelled('Request aborted during tool post-processing')
+      const cancellationResult = toolCall.result
       markToolResultSeen(toolCall.id)
       await completeAsyncToolCall({
         toolCallId: toolCall.id,
@@ -689,6 +755,9 @@ async function executeToolAndReportInner(
           error: toError(err).message,
         })
       })
+      if (toolCall.result !== cancellationResult) {
+        return terminalCompletionFromToolCall(toolCall)
+      }
       publishTerminalToolConfirmation({
         toolCallId: toolCall.id,
         status: MothershipStreamV1ToolOutcome.cancelled,
@@ -704,8 +773,13 @@ async function executeToolAndReportInner(
       result,
       toolExecutionContext
     )
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     if (abortRequested(context, execContext, options)) {
       markToolCallCancelled('Request aborted during tool post-processing')
+      const cancellationResult = toolCall.result
       markToolResultSeen(toolCall.id)
       await completeAsyncToolCall({
         toolCallId: toolCall.id,
@@ -718,6 +792,9 @@ async function executeToolAndReportInner(
           error: toError(err).message,
         })
       })
+      if (toolCall.result !== cancellationResult) {
+        return terminalCompletionFromToolCall(toolCall)
+      }
       publishTerminalToolConfirmation({
         toolCallId: toolCall.id,
         status: MothershipStreamV1ToolOutcome.cancelled,
@@ -799,6 +876,7 @@ async function executeToolAndReportInner(
       : MothershipStreamV1ToolOutcome.error
     const terminalMessage = modelSucceeded ? 'Tool completed' : requireToolCallError(toolCall)
     const terminalData = getToolCallTerminalData(toolCall)
+    const terminalResult = toolCall.result
 
     markToolResultSeen(toolCall.id)
     await completeAsyncToolCall({
@@ -814,6 +892,10 @@ async function executeToolAndReportInner(
         error: toError(err).message,
       })
     })
+    if (toolCall.result !== terminalResult) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     publishTerminalToolConfirmation({
       toolCallId: toolCall.id,
       status: terminalStatus,
@@ -877,6 +959,10 @@ async function executeToolAndReportInner(
       ...(terminalData !== undefined ? { data: terminalData } : {}),
     })
   } catch (error) {
+    if (toolCall.endTime || isTerminalToolCallStatus(toolCall.status)) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     const thrownMessage = toError(error).message
     const projection = inspectToolResultForCopilot(
       { success: false, error: thrownMessage },
@@ -888,6 +974,7 @@ async function executeToolAndReportInner(
     const safeThrownMessage = copilotError.error || 'Tool failed'
     if (abortRequested(context, execContext, options)) {
       markToolCallCancelled('Request aborted during tool execution')
+      const cancellationResult = toolCall.result
       markToolResultSeen(toolCall.id)
       await completeAsyncToolCall({
         toolCallId: toolCall.id,
@@ -900,6 +987,9 @@ async function executeToolAndReportInner(
           error: toError(err).message,
         })
       })
+      if (toolCall.result !== cancellationResult) {
+        return terminalCompletionFromToolCall(toolCall)
+      }
       publishTerminalToolConfirmation({
         toolCallId: toolCall.id,
         status: MothershipStreamV1ToolOutcome.cancelled,
@@ -917,6 +1007,7 @@ async function executeToolAndReportInner(
       error: safeThrownMessage,
     })
 
+    const terminalErrorResult = toolCall.result
     logger.error('Tool execution threw', {
       toolCallId: toolCall.id,
       toolName: toolCall.name,
@@ -936,6 +1027,10 @@ async function executeToolAndReportInner(
         error: toError(err).message,
       })
     })
+    if (toolCall.result !== terminalErrorResult) {
+      endToolSpanFromTerminalState()
+      return terminalCompletionFromToolCall(toolCall)
+    }
     publishTerminalToolConfirmation({
       toolCallId: toolCall.id,
       status: MothershipStreamV1ToolOutcome.error,
diff --git a/apps/sim/lib/copilot/request/types.ts b/apps/sim/lib/copilot/request/types.ts
index 62f8c4852f3..5a720b5bb99 100644
--- a/apps/sim/lib/copilot/request/types.ts
+++ b/apps/sim/lib/copilot/request/types.ts
@@ -4,6 +4,7 @@ import {
   MothershipStreamV1ToolOutcome,
 } from '@/lib/copilot/generated/mothership-stream-v1'
 import type { RequestTraceV1Span } from '@/lib/copilot/generated/request-trace-v1'
+import type { ProviderToolCallIdentity } from '@/lib/copilot/request/go/tool-call-identity'
 import type { StreamEvent } from '@/lib/copilot/request/session'
 import type { TraceCollector } from '@/lib/copilot/request/trace'
 import type { ToolExecutionContext, ToolExecutionResult } from '@/lib/copilot/tool-executor/types'
@@ -133,6 +134,12 @@ export interface StreamingContext {
   executionId?: string
   runId?: string
   messageId: string
+  /**
+   * Shared by all live resume legs. Reconnects replay events without resuming Go; any future
+   * durable lifecycle takeover must persist and restore this map alongside its checkpoints.
+   * Absent on legacy contexts, whose tool IDs retain their original meaning.
+   */
+  providerToolCallIdentity?: ProviderToolCallIdentity
   accumulatedContent: string
   finalAssistantContent: string
   sawMainToolCall: boolean
diff --git a/apps/sim/lib/table/backfill-runner.ts b/apps/sim/lib/table/backfill-runner.ts
index 582add45b7c..0ddd65d8e09 100644
--- a/apps/sim/lib/table/backfill-runner.ts
+++ b/apps/sim/lib/table/backfill-runner.ts
@@ -3,7 +3,7 @@ import { tableRowExecutions, userTableRows, workflowExecutionLogs } from '@sim/d
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
-import { isRecordLike } from '@sim/utils/object'
+import { getValueAtPath, isRecordLike } from '@sim/utils/object'
 import { and, asc, count, eq, gt, inArray } from 'drizzle-orm'
 import { isTriggerDevEnabled } from '@/lib/core/config/env-flags'
 import { runDetached } from '@/lib/core/utils/background'
@@ -20,7 +20,6 @@ import {
   markTableJobRunning,
   updateJobProgress,
 } from '@/lib/table/jobs/service'
-import { pluckByPath } from '@/lib/table/pluck'
 import { createTableRowSecretProvenanceFromRegistry } from '@/lib/table/rows/secret-provenance'
 import { batchUpdateRows } from '@/lib/table/rows/service'
 import { getTableById } from '@/lib/table/service'
@@ -231,7 +230,7 @@ async function processBackfillPage(opts: {
       if (!overwrite && (r.data as RowData)[out.columnName] !== undefined) continue
       const functionalOutput = getFunctionalBlockOutput(log.data, out.blockId)
       if (functionalOutput === undefined) continue
-      const picked = pluckByPath(functionalOutput, out.path)
+      const picked = getValueAtPath(functionalOutput, out.path)
       if (picked === undefined) continue
       dataPatch[out.columnName] = picked as RowData[string]
       mutated = true
diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts
index 72b5f7968bd..1acba3c595e 100644
--- a/apps/sim/lib/table/cell-write.ts
+++ b/apps/sim/lib/table/cell-write.ts
@@ -12,9 +12,9 @@
 import { db } from '@sim/db'
 import { userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { getValueAtPath } from '@sim/utils/object'
 import { and, eq } from 'drizzle-orm'
 import { appendTableEvent } from '@/lib/table/events'
-import { pluckByPath } from '@/lib/table/pluck'
 import { TableRowNotFoundError } from '@/lib/table/rows/errors'
 import { writeExecutionsPatch } from '@/lib/table/rows/executions'
 import {
@@ -29,6 +29,7 @@ import type {
   WorkflowGroup,
 } from '@/lib/table/types'
 import { coerceRowValues } from '@/lib/table/validation'
+import type { BlockCompletionCallbackData } from '@/executor/execution/types'
 import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry'
 
 const logger = createLogger('WorkflowCellWrite')
@@ -184,7 +185,10 @@ interface CreateWorkflowCellProgressWriterOptions {
 
 export interface WorkflowCellProgressWriter {
   onBlockStart: (blockId: string) => Promise<void>
-  onBlockComplete: (blockId: string, output: unknown) => Promise<void>
+  onBlockComplete: (
+    blockId: string,
+    data: Pick<BlockCompletionCallbackData, 'output' | 'resolvedSecretTraceProvenance'>
+  ) => Promise<void>
   waitForPendingWrites: () => Promise<void>
   finish: () => Promise<void>
   getEventOutputs: () => RowData
@@ -284,19 +288,12 @@ export function createWorkflowCellProgressWriter(
     scheduleWrite(undefined)
   }
 
-  const onBlockComplete = async (blockId: string, output: unknown): Promise<void> => {
+  const onBlockComplete: WorkflowCellProgressWriter['onBlockComplete'] = async (blockId, data) => {
     const work = completionChain.then(async () => {
       const outputs = outputsByBlockId.get(blockId)
       if (!outputs) return
 
-      const callbackData =
-        output && typeof output === 'object' && 'output' in output
-          ? (output as {
-              output: unknown
-              resolvedSecretTraceProvenance?: unknown
-            })
-          : undefined
-      const blockResult = callbackData ? callbackData.output : output
+      const blockResult = data.output
       const blockErrorMessage =
         blockResult &&
         typeof blockResult === 'object' &&
@@ -309,7 +306,7 @@ export function createWorkflowCellProgressWriter(
         blockErrors[blockId] = blockErrorMessage
       } else {
         for (const outputMapping of outputs) {
-          const value = pluckByPath(blockResult, outputMapping.path)
+          const value = getValueAtPath(blockResult, outputMapping.path)
           if (value === undefined) continue
           changedData[outputMapping.columnName] = value as RowData[string]
           eventOutputs[outputMapping.columnName] = value as RowData[string]
@@ -318,7 +315,7 @@ export function createWorkflowCellProgressWriter(
         if (Object.keys(changedData).length > 0) {
           const provenance = await createTableRowSecretProvenanceFromEncryptedExecution(
             changedData,
-            callbackData?.resolvedSecretTraceProvenance
+            data.resolvedSecretTraceProvenance
           )
           for (const columnId of Object.keys(changedData)) {
             pendingSecretProvenance[columnId] = provenance.complete
diff --git a/apps/sim/lib/table/pluck.ts b/apps/sim/lib/table/pluck.ts
deleted file mode 100644
index a3cd7413ad2..00000000000
--- a/apps/sim/lib/table/pluck.ts
+++ /dev/null
@@ -1,26 +0,0 @@
-/**
- * Pure utility for plucking a dot-and-bracket path from a value.
- *
- * Lives in its own leaf file (no server-only imports) so client components
- * can import it without dragging in the rest of `lib/table` (which transitively
- * pulls `@sim/db` and `next/headers`).
- */
-
-/**
- * Walk a dot-and-bracket path into a value (e.g. `a.b[0].c` or `result.items.0`).
- * Returns undefined for any missing segment.
- */
-export function pluckByPath(source: unknown, path: string): unknown {
-  if (source === null || source === undefined || !path) return source
-  const segments = path
-    .replace(/\[(\w+)\]/g, '.$1')
-    .split('.')
-    .filter(Boolean)
-  let cursor: unknown = source
-  for (const seg of segments) {
-    if (cursor === null || cursor === undefined) return undefined
-    if (typeof cursor !== 'object') return undefined
-    cursor = (cursor as Record<string, unknown>)[seg]
-  }
-  return cursor
-}
diff --git a/apps/sim/lib/table/rows/secret-provenance.postgres.test.ts b/apps/sim/lib/table/rows/secret-provenance.postgres.test.ts
index c21e5ef0272..d247ad5d73a 100644
--- a/apps/sim/lib/table/rows/secret-provenance.postgres.test.ts
+++ b/apps/sim/lib/table/rows/secret-provenance.postgres.test.ts
@@ -9,11 +9,13 @@
  * checks flag policy, stale-snapshot reporting, and write-event attribution without a database.
  */
 import { userTableRows } from '@sim/db/schema'
+import { loggingSessionMock } from '@sim/testing'
 import { eq, sql } from 'drizzle-orm'
 import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
 import postgres from 'postgres'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { PROVENANCE_MAX_SERIALIZED_BYTES } from '@/lib/execution/provenance-limits'
+import { createWorkflowCellProgressWriter } from '@/lib/table/cell-write'
 import type { DbTransaction } from '@/lib/table/planner'
 import {
   getTableSnapshotModelMountSafety,
@@ -22,13 +24,18 @@ import {
   updateTableRowsWithDerivedSecretProvenance,
 } from '@/lib/table/rows/secret-provenance'
 import type { RowData, TableRowSecretProvenanceWrite } from '@/lib/table/types'
+import { executeWorkflow } from '@/lib/workflows/executor/execute-workflow'
+import type { ExecutionCallbacks } from '@/executor/execution/types'
 
-const { database, mockIsEnforced, mockReport, mockError } = vi.hoisted(() => ({
-  database: { current: undefined as PostgresJsDatabase | undefined },
-  mockIsEnforced: vi.fn(() => false),
-  mockReport: vi.fn(),
-  mockError: vi.fn(),
-}))
+const { database, mockIsEnforced, mockReport, mockError, mockExecuteWorkflowCore } = vi.hoisted(
+  () => ({
+    database: { current: undefined as PostgresJsDatabase | undefined },
+    mockIsEnforced: vi.fn(() => false),
+    mockReport: vi.fn(),
+    mockError: vi.fn(),
+    mockExecuteWorkflowCore: vi.fn(),
+  })
+)
 
 vi.unmock('@sim/db/schema')
 vi.unmock('drizzle-orm')
@@ -41,7 +48,19 @@ vi.mock('@sim/db', () => ({
   },
 }))
 vi.mock('@sim/logger', () => ({
-  createLogger: () => ({ error: mockError, warn: vi.fn() }),
+  createLogger: () => ({ error: mockError, warn: vi.fn(), info: vi.fn(), debug: vi.fn() }),
+}))
+vi.mock('@/lib/core/security/encryption', () => ({
+  decryptSecret: vi.fn(async () => ({ decrypted: 'secret-value' })),
+}))
+vi.mock('@/lib/table/events', () => ({ appendTableEvent: vi.fn() }))
+vi.mock('@/lib/logs/execution/logging-session', () => loggingSessionMock)
+vi.mock('@/lib/posthog/server', () => ({ captureServerEvent: vi.fn() }))
+vi.mock('@/lib/workflows/executor/execution-core', () => ({
+  executeWorkflowCore: mockExecuteWorkflowCore,
+}))
+vi.mock('@/lib/workflows/executor/pause-persistence', () => ({
+  handlePostExecutionPauseState: vi.fn(),
 }))
 vi.mock('@/lib/execution/durable-secret-provenance-enforcement', () => ({
   isDurableSecretProvenanceEnforced: mockIsEnforced,
@@ -170,6 +189,137 @@ describe.skipIf(!databaseUrl)('table provenance in PostgreSQL', () => {
     await connection?.end()
   })
 
+  it.each(['exact', 'unknown', 'legacy'] as const)(
+    'persists executor callback provenance across a retried partial write over a %s row',
+    async (baseStatus) => {
+      if (!connection || !database.current) throw new Error('PostgreSQL fixture unavailable')
+      await insertRow({
+        version: baseStatus === 'legacy' ? null : 1,
+        ...(baseStatus === 'legacy' ? {} : { status: baseStatus }),
+      })
+      const onWriteError = vi.fn()
+      let rejectFirstWrite = true
+      const writer = createWorkflowCellProgressWriter({
+        group: {
+          id: 'group-1',
+          workflowId: 'workflow-1',
+          outputs: [
+            { blockId: 'secret', path: 'output.value', columnName: 'derived' },
+            { blockId: 'public', path: 'value', columnName: 'public' },
+          ],
+        },
+        onWriteError,
+        writeProgress: async ({ dataPatch, secretProvenance }) => {
+          if (!dataPatch || !secretProvenance || !database.current)
+            throw new Error('Missing progress data')
+          if (rejectFirstWrite) {
+            rejectFirstWrite = false
+            throw new Error('Retryable write failure')
+          }
+          await database.current.transaction(async (tx) => {
+            await mutateTableRowsWithSecretProvenance(tx as DbTransaction, {
+              rows: [{ rowId: 'row-1', provenance: secretProvenance }],
+              rowState: 'existing',
+              mode: 'merge',
+              mutate: async () => {
+                await tx.execute(
+                  sql`UPDATE user_table_rows SET data = data || ${JSON.stringify(dataPatch)}::jsonb WHERE id = 'row-1'`
+                )
+                return { value: undefined, affectedRowIds: ['row-1'] }
+              },
+            })
+          })
+          return 'wrote'
+        },
+      })
+      mockExecuteWorkflowCore.mockImplementationOnce(
+        async ({ callbacks }: { callbacks: ExecutionCallbacks }) => {
+          for (const blockId of ['secret', 'public']) {
+            await callbacks.onBlockComplete?.(blockId, blockId, 'function', {
+              output:
+                blockId === 'secret'
+                  ? { output: { value: 'secret-value' } }
+                  : { value: 'public-value' },
+              resolvedSecretTraceProvenance: {
+                version: 1,
+                complete: true,
+                entries:
+                  blockId === 'secret'
+                    ? [{ encryptedValue: 'encrypted-secret', name: 'SECRET' }]
+                    : [],
+                scope: { userId: 'user-1', workspaceId: 'workspace-1' },
+              },
+              executionTime: 1,
+              executionOrder: 0,
+              startedAt: updatedAt.toISOString(),
+              endedAt: updatedAt.toISOString(),
+            })
+            await writer.waitForPendingWrites()
+          }
+          return {
+            success: true,
+            output: {},
+            logs: [],
+            status: 'completed',
+            metadata: { duration: 1 },
+          }
+        }
+      )
+      await executeWorkflow(
+        { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1' },
+        'request-1',
+        {},
+        'user-1',
+        {
+          enabled: true,
+          principal: { kind: 'session', userId: 'user-1', sessionId: 'session-1' },
+          billingAttribution: {
+            actorUserId: 'user-1',
+            workspaceId: 'workspace-1',
+            organizationId: null,
+            billedAccountUserId: 'user-1',
+            billingEntity: { type: 'user', id: 'user-1' },
+            billingPeriod: { start: '2026-01-01T00:00:00.000Z', end: '2026-02-01T00:00:00.000Z' },
+            payerSubscription: null,
+          },
+          onBlockComplete: writer.onBlockComplete,
+        }
+      )
+      await writer.finish()
+      expect(onWriteError).toHaveBeenCalledOnce()
+      const rows = await connection`
+        SELECT r.data, r.secret_provenance_version AS version, p.status, p.entries,
+          p.content_updated_at = r.updated_at AS current
+        FROM user_table_rows r JOIN user_table_row_secret_provenance p ON p.row_id = r.id WHERE r.id = 'row-1'
+      `
+      expect(rows).toEqual([
+        {
+          data: {
+            retained: 'value',
+            removed: 'other',
+            derived: 'secret-value',
+            public: 'public-value',
+          },
+          version: 1,
+          status: baseStatus === 'unknown' ? 'unknown' : 'exact',
+          entries:
+            baseStatus === 'unknown'
+              ? []
+              : [
+                  {
+                    columnId: 'derived',
+                    encryptedValue: 'encrypted-secret',
+                    name: 'SECRET',
+                    sourceUserId: 'user-1',
+                    sourceWorkspaceId: 'workspace-1',
+                  },
+                ],
+          current: true,
+        },
+      ])
+    }
+  )
+
   it.each([
     { name: 'missing sidecar', fixture: {}, unrecorded: true },
     { name: 'stored unknown', fixture: { status: 'unknown' }, unrecorded: true },
diff --git a/apps/sim/lib/webhooks/slack-execution-stream.ts b/apps/sim/lib/webhooks/slack-execution-stream.ts
index 3a9f43b3dbe..888996145f3 100644
--- a/apps/sim/lib/webhooks/slack-execution-stream.ts
+++ b/apps/sim/lib/webhooks/slack-execution-stream.ts
@@ -1,10 +1,9 @@
 import { getErrorMessage } from '@sim/utils/errors'
-import { isRecordLike } from '@sim/utils/object'
+import { getValueAtPath, isRecordLike } from '@sim/utils/object'
 import { truncate } from '@sim/utils/string'
 import { getToolDisplayTitle } from '@/lib/copilot/tools/tool-display'
 import type { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { getSlackBotCredential } from '@/lib/oauth/credential-service'
-import { pluckByPath } from '@/lib/table/pluck'
 import {
   appendSlackAgentStream,
   formatSlackApiFailure,
@@ -457,7 +456,7 @@ export class SlackExecutionStreamController {
       )
       if (!Object.hasOwn(display, 'output')) return
       const values = selected.flatMap((selection) => {
-        const value = pluckByPath(display.output, selection.path)
+        const value = getValueAtPath(display.output, selection.path)
         return value === undefined ? [] : [{ path: selection.path, value }]
       })
       if (values.length === 0) return
diff --git a/apps/sim/lib/workflows/executor/execute-service.ts b/apps/sim/lib/workflows/executor/execute-service.ts
index be5d94c4fae..a5071f9a737 100644
--- a/apps/sim/lib/workflows/executor/execute-service.ts
+++ b/apps/sim/lib/workflows/executor/execute-service.ts
@@ -604,7 +604,8 @@ export async function executeWorkflowService(
               useDraftState,
               runFromBlock,
               onStream,
-              onBlockComplete,
+              onBlockComplete: (blockId, data) =>
+                onBlockComplete(blockId, data.output, data.outputBlockId),
               skipLoggingComplete: true,
               includeFileBase64,
               base64MaxBytes,
diff --git a/apps/sim/lib/workflows/executor/execute-workflow.test.ts b/apps/sim/lib/workflows/executor/execute-workflow.test.ts
index 9f6b48317fe..ec4b094fa07 100644
--- a/apps/sim/lib/workflows/executor/execute-workflow.test.ts
+++ b/apps/sim/lib/workflows/executor/execute-workflow.test.ts
@@ -5,6 +5,8 @@ import { loggerMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution'
 import type { ExecutionSnapshot } from '@/executor/execution/snapshot'
+import type { ExecutionCallbacks } from '@/executor/execution/types'
+import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry'
 
 const {
   captureServerEventMock,
@@ -55,6 +57,15 @@ vi.mock('@/lib/workflows/executor/pause-persistence', () => ({
   handlePostExecutionPauseState: handlePostExecutionPauseStateMock,
 }))
 
+vi.mock('@/lib/table/events', () => ({ appendTableEvent: vi.fn() }))
+vi.mock('@/lib/core/security/encryption', () => ({
+  decryptSecret: vi.fn(async (value: string) => {
+    if (value !== 'encrypted-secret') throw new Error('Invalid ciphertext')
+    return { decrypted: 'secret-value' }
+  }),
+}))
+
+import { createWorkflowCellProgressWriter } from '@/lib/table/cell-write'
 import { executeWorkflow } from '@/lib/workflows/executor/execute-workflow'
 import { hasExecutionResult } from '@/executor/utils/errors'
 
@@ -235,6 +246,72 @@ describe('executeWorkflow', () => {
     )
   })
 
+  it.each([
+    ['secret-bearing', { complete: true, entries: [{ encryptedValue: 'encrypted-secret' }] }, true],
+    ['exact-empty', { complete: true, entries: [] }, true],
+    ['incomplete', { complete: false, entries: [] }, false],
+    ['undecryptable', { complete: true, entries: [{ encryptedValue: 'invalid' }] }, false],
+    ['legacy', undefined, false],
+  ] as const)(
+    'preserves %s provenance through the executor callback into table cells',
+    async (_kind, source, complete) => {
+      const provenance: ResolvedSecretTraceProvenanceV1 | undefined = source && {
+        version: 1,
+        complete: source.complete,
+        entries: [...source.entries],
+        scope: { userId: 'actor-1', workspaceId: 'workspace-1' },
+      }
+      const writeProgress = vi.fn().mockResolvedValue('wrote')
+      const writer = createWorkflowCellProgressWriter({
+        group: {
+          id: 'group-1',
+          workflowId: workflow.id,
+          outputs: [{ blockId: 'block-1', path: 'output.value', columnName: 'column-1' }],
+        },
+        writeProgress,
+        onWriteError: (error) => {
+          throw error
+        },
+      })
+      executeWorkflowCoreMock.mockImplementationOnce(
+        async ({ callbacks }: { callbacks: ExecutionCallbacks }) => {
+          await callbacks.onBlockComplete?.('block-1', 'Block', 'function', {
+            output: { output: { value: 'secret-value' } },
+            resolvedSecretTraceProvenance: provenance,
+            executionTime: 1,
+            startedAt: '2026-01-01T00:00:00Z',
+            endedAt: '2026-01-01T00:00:01Z',
+            executionOrder: 0,
+          })
+          return {
+            success: true,
+            output: {},
+            logs: [],
+            status: 'completed',
+            metadata: { duration: 1 },
+          }
+        }
+      )
+
+      await executeWorkflow(workflow, 'request-1', {}, 'actor-1', {
+        enabled: true,
+        principal,
+        billingAttribution,
+        onBlockComplete: writer.onBlockComplete,
+      })
+      await writer.finish()
+
+      expect(writer.getEventOutputs()).toEqual({ 'column-1': 'secret-value' })
+      expect(writer.getPendingDataPatch()).toEqual({})
+      expect(writeProgress).toHaveBeenCalledWith(
+        expect.objectContaining({
+          dataPatch: { 'column-1': 'secret-value' },
+          secretProvenance: { complete, columns: complete ? { 'column-1': provenance } : {} },
+        })
+      )
+    }
+  )
+
   it('forwards a trusted immutable workflow state to the execution snapshot', async () => {
     const workflowStateOverride = {
       blocks: { 'block-1': { id: 'block-1', type: 'start_trigger' } },
diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts
index 241e649be37..f5682b78025 100644
--- a/apps/sim/lib/workflows/executor/execute-workflow.ts
+++ b/apps/sim/lib/workflows/executor/execute-workflow.ts
@@ -44,7 +44,7 @@ export interface ExecuteWorkflowOptions {
     blockType: string,
     executionOrder: number
   ) => Promise<void>
-  onBlockComplete?: (blockId: string, output: unknown, outputBlockId?: string) => Promise<void>
+  onBlockComplete?: (blockId: string, data: BlockCompletionCallbackData) => Promise<void>
   /** Transfers post-execution logging ownership to the streaming caller after execution succeeds. */
   skipLoggingComplete?: boolean
   includeFileBase64?: boolean
@@ -212,7 +212,7 @@ export async function executeWorkflow(
               _blockType: string,
               data: BlockCompletionCallbackData
             ) => {
-              await streamConfig.onBlockComplete!(blockId, data.output, data.outputBlockId)
+              await streamConfig.onBlockComplete!(blockId, data)
             }
           : undefined,
       },
diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts
index 4cf96093dc2..bca73899221 100644
--- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts
+++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts
@@ -418,7 +418,7 @@ interface StartResumeExecutionArgs {
   userId: string
   sendEvent?: (event: ExecutionEvent) => void
   onStream?: (streamingExec: StreamingExecution) => Promise<void>
-  onBlockComplete?: (blockId: string, output: unknown) => Promise<void>
+  onBlockComplete?: (blockId: string, data: BlockCompletionCallbackData) => Promise<void>
   abortSignal?: AbortSignal
 }
 
@@ -1071,7 +1071,7 @@ export class PauseResumeManager {
     userId: string
     sendEvent?: (event: ExecutionEvent) => void
     onStream?: (streamingExec: StreamingExecution) => Promise<void>
-    onBlockComplete?: (blockId: string, output: unknown) => Promise<void>
+    onBlockComplete?: (blockId: string, data: BlockCompletionCallbackData) => Promise<void>
     abortSignal?: AbortSignal
   }): Promise<ExecutionResult> {
     const {
@@ -1727,7 +1727,7 @@ export class PauseResumeManager {
         } as ExecutionEvent)
 
         if (externalOnBlockComplete) {
-          await externalOnBlockComplete(blockId, callbackData.output)
+          await externalOnBlockComplete(blockId, callbackData)
         }
       },
       onChildWorkflowInstanceReady: async (
diff --git a/apps/sim/lib/workflows/executor/resume-execution.ts b/apps/sim/lib/workflows/executor/resume-execution.ts
index b30ef77cc55..2f54fc33dd4 100644
--- a/apps/sim/lib/workflows/executor/resume-execution.ts
+++ b/apps/sim/lib/workflows/executor/resume-execution.ts
@@ -289,7 +289,8 @@ export async function executeResumeWorkflow({
           PauseResumeManager.startResumeExecution({
             ...resumeArgs,
             onStream,
-            onBlockComplete,
+            onBlockComplete: (blockId, data) =>
+              onBlockComplete(blockId, data.output, data.outputBlockId),
             abortSignal,
           }),
       })
diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts
index 586128fc509..c486a88bd8e 100644
--- a/packages/utils/src/index.ts
+++ b/packages/utils/src/index.ts
@@ -16,6 +16,7 @@ export type { EmbedInfo } from './media-embed'
 export { getEmbedInfo } from './media-embed'
 export {
   filterUndefined,
+  getValueAtPath,
   isPlainRecord,
   isRecordLike,
   omit,
diff --git a/packages/utils/src/object.test.ts b/packages/utils/src/object.test.ts
index 78c3a900d87..4818071fe43 100644
--- a/packages/utils/src/object.test.ts
+++ b/packages/utils/src/object.test.ts
@@ -3,6 +3,7 @@
  */
 import { describe, expect, it } from 'vitest'
 import {
+  getValueAtPath,
   isPlainRecord,
   isRecordLike,
   sortObjectKeysDeep,
@@ -14,6 +15,27 @@ class Sample {
   value = 1
 }
 
+describe('getValueAtPath', () => {
+  const source = { items: [{ name: 'first', active: false, count: 0 }], empty: null }
+
+  it.each([
+    ['items[0].name', 'first'],
+    ['items.0.active', false],
+    ['items[0].count', 0],
+    ['items[1].name', undefined],
+    ['items[0].name.missing', undefined],
+    ['empty.missing', undefined],
+  ])('reads %s without confusing missing and falsy values', (path, expected) => {
+    expect(getValueAtPath(source, path)).toBe(expected)
+  })
+
+  it('preserves an empty path and nullish roots', () => {
+    expect(getValueAtPath(source, '')).toBe(source)
+    expect(getValueAtPath(null, 'items')).toBeNull()
+    expect(getValueAtPath(undefined, 'items')).toBeUndefined()
+  })
+})
+
 describe('isRecordLike', () => {
   it('returns true for plain objects, Date, and class instances', () => {
     expect(isRecordLike({})).toBe(true)
diff --git a/packages/utils/src/object.ts b/packages/utils/src/object.ts
index 6ff480711a9..4c722bb1b49 100644
--- a/packages/utils/src/object.ts
+++ b/packages/utils/src/object.ts
@@ -91,3 +91,18 @@ export function sortObjectKeysDeep(value: unknown): unknown {
   }
   return value
 }
+
+/** Reads a dot-and-bracket path such as `items[0].name`; missing segments return undefined. */
+export function getValueAtPath(source: unknown, path: string): unknown {
+  if (source === null || source === undefined || !path) return source
+  const segments = path
+    .replace(/\[(\w+)\]/g, '.$1')
+    .split('.')
+    .filter(Boolean)
+  let cursor: unknown = source
+  for (const segment of segments) {
+    if (cursor === null || typeof cursor !== 'object') return undefined
+    cursor = (cursor as Record<string, unknown>)[segment]
+  }
+  return cursor
+}