fix(flows): keep gateway tokens fresh across flow execution

apps/web/src/lib/flows/__tests__/retry-policy.test.ts

@@ -48,6 +48,7 @@ describe('flow retry policy', () => {
     expect(isRetryableFlowRunError('healthcheck timeout while starting')).toBe(true)
     expect(isRetryableFlowRunError('INSTANCE_UNAVAILABLE')).toBe(true)
     expect(isRetryableFlowRunError('flow_mcp_connector_unavailable:Mixpanel')).toBe(true)
+    expect(isRetryableFlowRunError('APIError: Unauthorized: invalid_token')).toBe(true)
     expect(isRetryableFlowRunError('fetch failed: ECONNREFUSED')).toBe(true)
     expect(isRetryableFlowRunError('UND_ERR_CONNECT_TIMEOUT')).toBe(true)
     expect(isRetryableFlowRunError('container name /arche is already in use')).toBe(true)

apps/web/src/lib/flows/__tests__/runner.test.ts

@@ -13,6 +13,7 @@ const mocks = vi.hoisted(() => ({
   createInstanceClient: vi.fn(),
   createRun: vi.fn(),
   extendFlowLease: vi.fn(),
+  ensureProviderAccessFreshForExecution: vi.fn(),
   ensureWorkspaceRunningForExecution: vi.fn(),
   findFlowByIdForScope: vi.fn(),
   findRunByIdForScope: vi.fn(),
@@ -48,6 +49,10 @@ vi.mock('@/lib/opencode/client', () => ({
   createInstanceClient: mocks.createInstanceClient,
 }))
 
+vi.mock('@/lib/opencode/providers', () => ({
+  ensureProviderAccessFreshForExecution: mocks.ensureProviderAccessFreshForExecution,
+}))
+
 vi.mock('@/lib/opencode/session-execution', () => ({
   ensureWorkspaceRunningForExecution: mocks.ensureWorkspaceRunningForExecution,
 }))
@@ -214,6 +219,7 @@ describe('triggerFlowNow', () => {
     mocks.findRunByIdForScope.mockResolvedValue(null)
     mocks.findRunStatusById.mockResolvedValue({ status: FlowRunStatus.running })
     mocks.extendFlowLease.mockResolvedValue({ count: 1 })
+    mocks.ensureProviderAccessFreshForExecution.mockResolvedValue(undefined)
     mocks.ensureWorkspaceRunningForExecution.mockResolvedValue(undefined)
     mocks.markRunFailed.mockResolvedValue({ count: 1 })
     mocks.markRunRunning.mockResolvedValue(undefined)
@@ -347,10 +353,57 @@ describe('triggerFlowNow', () => {
 
     await runClaimedFlow(createClaimedFlow(), FlowRunTrigger.manual)
 
-    expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1')
+    expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1', { forceProviderRefresh: true })
+    // Single-node flow: the first iteration is exempt because the forced
+    // entry-point sync above already re-issued tokens.
+    expect(mocks.ensureProviderAccessFreshForExecution).not.toHaveBeenCalled()
+    expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' }))
+  })
+
+  it('forces a provider access refresh between flow nodes', async () => {
+    const flow = createClaimedFlow()
+    flow.definition = {
+      edges: [{ id: 'edge-1', sourceNodeId: 'agent-1', targetNodeId: 'agent-2' }],
+      nodes: [
+        { compactOutput: false, id: 'agent-1', name: 'First', promptTemplate: 'First', targetAgentId: null, type: 'agent' },
+        { compactOutput: false, id: 'agent-2', name: 'Second', promptTemplate: 'Second', targetAgentId: null, type: 'agent' },
+      ],
+      startNodeId: 'agent-1',
+      version: 1,
+    }
+    mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' })
+    mocks.createRun.mockResolvedValue(createRunRecord())
+
+    await runClaimedFlow(flow, FlowRunTrigger.manual)
+
+    expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledTimes(1)
+    expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledWith({ slug: 'alice', userId: 'user-1', force: true })
+    expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledTimes(2)
     expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' }))
   })
 
+  it('continues the flow when the between-step provider refresh fails', async () => {
+    const flow = createClaimedFlow()
+    flow.definition = {
+      edges: [{ id: 'edge-1', sourceNodeId: 'agent-1', targetNodeId: 'agent-2' }],
+      nodes: [
+        { compactOutput: false, id: 'agent-1', name: 'First', promptTemplate: 'First', targetAgentId: null, type: 'agent' },
+        { compactOutput: false, id: 'agent-2', name: 'Second', promptTemplate: 'Second', targetAgentId: null, type: 'agent' },
+      ],
+      startNodeId: 'agent-1',
+      version: 1,
+    }
+    mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' })
+    mocks.createRun.mockResolvedValue(createRunRecord())
+    mocks.ensureProviderAccessFreshForExecution.mockRejectedValue(new Error('instance_unavailable'))
+
+    await runClaimedFlow(flow, FlowRunTrigger.manual)
+
+    expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledTimes(2)
+    expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' }))
+    expect(mocks.markRunFailed).not.toHaveBeenCalled()
+  })
+
   it('keeps the flow run and lease active when runtime termination is unconfirmed', async () => {
     mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' })
     mocks.runFlowPromptAndReadOutput.mockResolvedValue({
@@ -383,7 +436,7 @@ describe('triggerFlowNow', () => {
       scheduledFor: now,
       trigger: FlowRunTrigger.manual,
     })
-    await vi.waitFor(() => expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('bob', 'user-2'))
+    await vi.waitFor(() => expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('bob', 'user-2', { forceProviderRefresh: true }))
     await vi.waitFor(() => expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledWith(expect.objectContaining({
       slug: 'bob',
       userId: 'user-2',
@@ -905,7 +958,7 @@ describe('triggerFlowNow', () => {
       .resolves.toMatchObject({ ok: true, run: { id: 'run-1' } })
 
     await vi.waitFor(() => expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' })))
-    expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1')
+    expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1', { forceProviderRefresh: true })
   })
 
   it('keeps a resumed flow run and lease active when termination is unconfirmed', async () => {

apps/web/src/lib/flows/retry-policy.ts

@@ -60,6 +60,9 @@ export function isRetryableFlowRunError(error: string): boolean {
     normalized.includes('instance_start_timeout') ||
     normalized.includes('instance_unavailable') ||
     normalized.includes('flow_mcp_connector_unavailable') ||
+    // Gateway token expiry surfaces as invalid_token mid-run; the retry
+    // re-syncs provider access at startup and re-runs only the failed step.
+    normalized.includes('invalid_token') ||
     normalized.includes('kb_unavailable') ||
     normalized.includes('user_data_unavailable') ||
     normalized.includes('fetch failed') ||

apps/web/src/lib/flows/runner.ts

@@ -19,6 +19,7 @@ import {
 import type { FlowDefinition } from '@/lib/flows/types'
 import { validateFlowDefinition } from '@/lib/flows/validation'
 import { createInstanceClient } from '@/lib/opencode/client'
+import { ensureProviderAccessFreshForExecution } from '@/lib/opencode/providers'
 import {
   ensureWorkspaceRunningForExecution,
   type SessionExecutionClient,
@@ -141,6 +142,30 @@ async function executeFlowNodes(params: {
       return { status: 'failed', error: 'flow_lease_lost' }
     }
 
+    // Force a token refresh between steps so each step after the first starts
+    // with the full gateway-token TTL: flow steps run for minutes, so a
+    // freshness-threshold check here would skip until the token is nearly
+    // expired and hand the next step a token that dies mid-step. The first
+    // iteration is exempt — the run entry points already force a fresh sync
+    // before the loop starts. The flow's own message run is finalized between
+    // steps, so this only defers when an unrelated run is active — the case
+    // where a concurrent sync (which disposes the instance) must not run.
+    if (visitedNodeIds.size > 0) {
+      try {
+        await ensureProviderAccessFreshForExecution({
+          slug: params.slug,
+          userId: params.executionUserId,
+          force: true,
+        })
+      } catch (error) {
+        console.warn('[flows] Failed to refresh provider access between steps', {
+          error: error instanceof Error ? error.message : String(error),
+          flowId: params.flow.id,
+          runId: params.run.id,
+        })
+      }
+    }
+
     visitedNodeIds.add(currentNodeId)
     const node = getFlowNodeById(params.definition, currentNodeId)
     if (!node) {
@@ -408,7 +433,10 @@ async function executeClaimedFlowRun(
       return
     }
 
-    await ensureWorkspaceRunningForExecution(slug, executionUserId)
+    // Force a token refresh at flow start: a scheduled flow can begin long
+    // after the last interactive sync, and would otherwise inherit a gateway
+    // token with too little TTL left to cover the first step.
+    await ensureWorkspaceRunningForExecution(slug, executionUserId, { forceProviderRefresh: true })
     await instanceService.touchActivity(slug).catch(() => undefined)
 
     const client = await createInstanceClient(slug)
@@ -617,7 +645,7 @@ async function resumeClaimedFlowRun(params: {
       return
     }
 
-    await ensureWorkspaceRunningForExecution(slug, executionUserId)
+    await ensureWorkspaceRunningForExecution(slug, executionUserId, { forceProviderRefresh: true })
     const client = await createInstanceClient(slug)
     if (!client) throw new Error('instance_unavailable')
 

apps/web/src/lib/opencode/__tests__/providers.test.ts

@@ -270,6 +270,42 @@ describe('syncProviderAccessForInstance', () => {
     expect(mockGetInstanceBasicAuth).not.toHaveBeenCalled()
   })
 
+  it('forces a refresh even when the running instance matches the expected hash', async () => {
+    mockGetEnabledCredentials.mockResolvedValue(enabledCredentials([
+      ['openai', { credentialId: 'org-1', source: 'organization', version: 3 }],
+    ]))
+
+    mockInstanceService.findProviderSyncBySlug.mockResolvedValue({
+      providerSyncHash: await getProviderSyncHashForUser('user-1'),
+      providerSyncedAt: new Date(),
+      status: 'running',
+    })
+
+    await ensureProviderAccessFreshForExecution({ slug: 'alice', userId: 'user-1', force: true })
+
+    expect(mockGetInstanceBasicAuth).toHaveBeenCalledWith('alice')
+  })
+
+  it('still defers a forced refresh while the workspace has active runs', async () => {
+    const consoleWarnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    mockGetEnabledCredentials.mockResolvedValue(enabledCredentials([
+      ['openai', { credentialId: 'org-1', source: 'organization', version: 3 }],
+    ]))
+    mockInstanceService.findProviderSyncBySlug.mockResolvedValue({
+      providerSyncHash: await getProviderSyncHashForUser('user-1'),
+      providerSyncedAt: new Date(),
+      status: 'running',
+    })
+    mockMessageRunService.hasActiveRunForSlug.mockResolvedValue(true)
+
+    await ensureProviderAccessFreshForExecution({ slug: 'alice', userId: 'user-1', force: true })
+
+    expect(mockGetInstanceBasicAuth).not.toHaveBeenCalled()
+    expect(mockInstanceService.setProviderSyncState).not.toHaveBeenCalled()
+
+    consoleWarnSpy.mockRestore()
+  })
+
   it('refreshes provider access when the sync record is stale by age', async () => {
     mockGetEnabledCredentials.mockResolvedValue(enabledCredentials([
       ['openai', { credentialId: 'org-1', source: 'organization', version: 3 }],

apps/web/src/lib/opencode/__tests__/session-execution.test.ts

@@ -576,6 +576,22 @@ describe('session execution helpers', () => {
     })
   })
 
+  it('forces a provider refresh on a running workspace when requested', async () => {
+    const { ensureProviderAccessFreshForExecution } = await import('@/lib/opencode/providers')
+    const { getWorkspaceStatus } = await import('@/lib/runtime/workspace-host')
+
+    vi.mocked(getWorkspaceStatus).mockResolvedValue({ status: 'running' } as never)
+
+    const { ensureWorkspaceRunningForExecution } = await import('../session-execution')
+    await ensureWorkspaceRunningForExecution('slack-bot', 'user-1', { forceProviderRefresh: true })
+
+    expect(ensureProviderAccessFreshForExecution).toHaveBeenCalledWith({
+      slug: 'slack-bot',
+      userId: 'user-1',
+      force: true,
+    })
+  })
+
   it('refreshes provider access after a workspace finishes starting', async () => {
     vi.useFakeTimers()
 

apps/web/src/lib/opencode/providers.ts

@@ -86,12 +86,17 @@ export async function ensureProviderAccessFreshForExecution(args: {
   slug: string
   userId: string
   ignoreActiveRunId?: string
+  // Flow runs force a refresh at start so they never inherit a partially aged
+  // gateway token from earlier interactive activity. The active-run deferral
+  // below still applies, so a concurrent generation is never aborted.
+  force?: boolean
 }): Promise<void> {
   await withProviderSyncLock(args.slug, async () => {
     const expectedHash = await getProviderSyncHashForUser(args.userId)
     const current = await instanceService.findProviderSyncBySlug(args.slug)
 
     if (
+      !args.force &&
       current?.status === 'running' &&
       !shouldRefreshProviderAccess({
         expectedHash,

apps/web/src/lib/opencode/session-execution.ts

@@ -302,10 +302,16 @@ function inspectSessionOutcome(
   return null
 }
 
-export async function ensureWorkspaceRunningForExecution(slug: string, userId: string): Promise<void> {
+export async function ensureWorkspaceRunningForExecution(
+  slug: string,
+  userId: string,
+  options: { forceProviderRefresh?: boolean } = {},
+): Promise<void> {
   const current = await getWorkspaceStatus(slug)
   if (current?.status === 'running') {
-    await ensureProviderAccessFreshForExecution({ slug, userId })
+    // Only the already-running path honors the flag: a freshly started
+    // workspace was just synced, so it has no aged token to refresh.
+    await ensureProviderAccessFreshForExecution({ slug, userId, force: options.forceProviderRefresh })
     return
   }
 

openspec/changes/flow-gateway-token-refresh/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-09-01

openspec/changes/flow-gateway-token-refresh/proposal.md

@@ -0,0 +1,20 @@
+## Why
+
+Multi-step flows fail mid-run with `APIError: Unauthorized: invalid_token` from the internal provider gateway. The flow runner syncs provider access exactly once, before the flow starts (`ensureWorkspaceRunningForExecution` in `apps/web/src/lib/flows/runner.ts`), and gateway tokens carry a short TTL (`ARCHE_GATEWAY_TOKEN_TTL_SECONDS`, default 900s). Two runs of the same Codebase Hunter flow on 2026-08-31 failed this way:
+
+- A 17-minute run whose token was issued at flow start expired between steps; the next step's first gateway call was rejected.
+- A run that started 7 minutes after an interactive workspace visit: `shouldRefreshProviderAccess` saw a "fresh" sync (under the TTL-minus-skew threshold), skipped re-syncing, and the flow inherited a partially aged token that expired mid-run.
+
+During flow execution tokens are never refreshed: the runner makes no further refresh calls, and `ensureProviderAccessFreshForExecution` defers while any run is active — including the flow's own runs, so a naive between-step refresh from inside a step would defer forever (and disposing the instance mid-generation aborts it). Worse, `invalid_token` is absent from `isRetryableFlowRunError`, so an expiry is a terminal run failure even though the retry machinery resumes from the failed node and would re-sync on the retry.
+
+## What Changes
+
+- Force a provider-access refresh at flow start: `ensureProviderAccessFreshForExecution` gains a `force` flag that bypasses the freshness-age skip (credential-hash check and active-run deferral still apply), and the flow runner requests it via `ensureWorkspaceRunningForExecution(slug, userId, { forceProviderRefresh: true })` on both entry points (claimed runs and resumes). A flow therefore starts with a token valid for the full TTL window.
+- Force a provider-access refresh before every step after the first in `executeFlowNodes`; the first step is covered by the forced entry-point refresh. Forcing is required rather than a freshness-threshold check: flow steps run for minutes, so a threshold tuned to the token TTL would skip boundaries until the token is nearly expired and hand the next step a token that dies mid-step (observed on 2026-09-02: an 11-minute step started with 10 minutes of token life left and failed at expiry). At a boundary the flow's own message run is finalized, so the active-run deferral only fires for unrelated runs — exactly the case where a concurrent sync (which disposes the OpenCode instance) must not abort in-flight generation. A failed boundary refresh logs a warning and does not fail the run; genuine auth failures still surface from the step itself.
+- Classify gateway token expiry (`invalid_token`) as retryable in `isRetryableFlowRunError`, so an expiry that survives the above costs one step re-run after backoff instead of failing the run; the retry re-syncs provider access and resumes from `run.currentNodeId`.
+
+## Non-goals
+
+- No mid-flight (mid-step) token refresh. `syncProviderAccessForInstance` already supports `disposeInstance: false`, but this only helps if the OpenCode instance re-reads auth keys per request instead of caching them at provider creation. Follow-up change once verified against a live instance; would hook into the existing session-executor pulse.
+- No change to token TTL, token claims, or token issuance. Tokens stay short-lived and scoped per user, workspace, provider, and credential version.
+- No per-provider partial syncs; refresh keeps the existing sync-all-providers behavior.