From c1ad2c2d2bc6d30879cba0a86d5855297a564153 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Tue, 1 Sep 2026 16:00:28 +0100 Subject: [PATCH 1/2] fix(flows): keep gateway tokens fresh across flow execution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multi-step flows failed mid-run with `APIError: Unauthorized: invalid_token` from the internal provider gateway: the runner synced provider access exactly once before the flow started, gateway tokens carry a short TTL (900s default), and a flow could also inherit a partially aged token when it started shortly after interactive activity. `invalid_token` was not retryable, so expiry failed the run terminally. - Force a provider-access refresh at flow start (`force` flag on `ensureProviderAccessFreshForExecution`, threaded through `ensureWorkspaceRunningForExecution`) so flows start with a full-TTL token; the credential-hash check and active-run deferral still apply. - Refresh provider access before each flow node in `executeFlowNodes`. At a step boundary the flow's own message run is finalized, so this only defers when an unrelated run is active — the case where a concurrent sync (which disposes the instance) must not abort in-flight generation. A failed boundary refresh warns and continues. - Classify `invalid_token` as retryable so an expiry that survives the above costs one step re-run after backoff; the retry re-syncs and resumes from the failed node. No changes to token TTL, claims, or issuance. Mid-step refresh without dispose is deliberately left for a follow-up pending verification that OpenCode re-reads auth keys per request. Co-Authored-By: Claude Code --- .../lib/flows/__tests__/retry-policy.test.ts | 1 + .../src/lib/flows/__tests__/runner.test.ts | 47 +++++++++++++++++-- apps/web/src/lib/flows/retry-policy.ts | 3 ++ apps/web/src/lib/flows/runner.ts | 25 +++++++++- .../lib/opencode/__tests__/providers.test.ts | 36 ++++++++++++++ .../__tests__/session-execution.test.ts | 16 +++++++ apps/web/src/lib/opencode/providers.ts | 5 ++ .../web/src/lib/opencode/session-execution.ts | 10 +++- .../flow-gateway-token-refresh/.openspec.yaml | 2 + .../flow-gateway-token-refresh/proposal.md | 20 ++++++++ .../specs/flow-execution/spec.md | 42 +++++++++++++++++ .../flow-gateway-token-refresh/tasks.md | 13 +++++ 12 files changed, 213 insertions(+), 7 deletions(-) create mode 100644 openspec/changes/flow-gateway-token-refresh/.openspec.yaml create mode 100644 openspec/changes/flow-gateway-token-refresh/proposal.md create mode 100644 openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md create mode 100644 openspec/changes/flow-gateway-token-refresh/tasks.md diff --git a/apps/web/src/lib/flows/__tests__/retry-policy.test.ts b/apps/web/src/lib/flows/__tests__/retry-policy.test.ts index 36e71c90..4e9ab8a9 100644 --- a/apps/web/src/lib/flows/__tests__/retry-policy.test.ts +++ b/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) diff --git a/apps/web/src/lib/flows/__tests__/runner.test.ts b/apps/web/src/lib/flows/__tests__/runner.test.ts index 070ce08b..9689dc5c 100644 --- a/apps/web/src/lib/flows/__tests__/runner.test.ts +++ b/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,8 +353,43 @@ describe('triggerFlowNow', () => { await runClaimedFlow(createClaimedFlow(), FlowRunTrigger.manual) - expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1') + expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1', { forceProviderRefresh: true }) + expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledTimes(1) + expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' })) + }) + + it('refreshes provider access before each flow node', 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(2) + expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledWith({ slug: 'alice', userId: 'user-1' }) + 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 () => { + mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' }) + mocks.createRun.mockResolvedValue(createRunRecord()) + mocks.ensureProviderAccessFreshForExecution.mockRejectedValue(new Error('instance_unavailable')) + + await runClaimedFlow(createClaimedFlow(), FlowRunTrigger.manual) + + expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledTimes(1) 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 () => { @@ -383,7 +424,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 +946,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 () => { diff --git a/apps/web/src/lib/flows/retry-policy.ts b/apps/web/src/lib/flows/retry-policy.ts index 5506048c..09385f84 100644 --- a/apps/web/src/lib/flows/retry-policy.ts +++ b/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') || diff --git a/apps/web/src/lib/flows/runner.ts b/apps/web/src/lib/flows/runner.ts index a7c6911d..f5425a8e 100644 --- a/apps/web/src/lib/flows/runner.ts +++ b/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,23 @@ async function executeFlowNodes(params: { return { status: 'failed', error: 'flow_lease_lost' } } + // Gateway tokens carry a short TTL, so multi-step flows must not carry an + // aged token into the next step. 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. + try { + await ensureProviderAccessFreshForExecution({ + slug: params.slug, + userId: params.executionUserId, + }) + } 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 +426,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 +638,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') diff --git a/apps/web/src/lib/opencode/__tests__/providers.test.ts b/apps/web/src/lib/opencode/__tests__/providers.test.ts index 995e7b80..ecde8ec7 100644 --- a/apps/web/src/lib/opencode/__tests__/providers.test.ts +++ b/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 }], diff --git a/apps/web/src/lib/opencode/__tests__/session-execution.test.ts b/apps/web/src/lib/opencode/__tests__/session-execution.test.ts index bf6f2131..04f8ec88 100644 --- a/apps/web/src/lib/opencode/__tests__/session-execution.test.ts +++ b/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() diff --git a/apps/web/src/lib/opencode/providers.ts b/apps/web/src/lib/opencode/providers.ts index 1ba8ade8..f4857e12 100644 --- a/apps/web/src/lib/opencode/providers.ts +++ b/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 { 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, diff --git a/apps/web/src/lib/opencode/session-execution.ts b/apps/web/src/lib/opencode/session-execution.ts index c247a2a3..c525cc1e 100644 --- a/apps/web/src/lib/opencode/session-execution.ts +++ b/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 { +export async function ensureWorkspaceRunningForExecution( + slug: string, + userId: string, + options: { forceProviderRefresh?: boolean } = {}, +): Promise { 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 } diff --git a/openspec/changes/flow-gateway-token-refresh/.openspec.yaml b/openspec/changes/flow-gateway-token-refresh/.openspec.yaml new file mode 100644 index 00000000..b4b3ece7 --- /dev/null +++ b/openspec/changes/flow-gateway-token-refresh/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-09-01 diff --git a/openspec/changes/flow-gateway-token-refresh/proposal.md b/openspec/changes/flow-gateway-token-refresh/proposal.md new file mode 100644 index 00000000..d3bc8540 --- /dev/null +++ b/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. +- Refresh provider access at every flow step boundary in `executeFlowNodes`, before each node executes. 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. diff --git a/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md b/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md new file mode 100644 index 00000000..c8dab9ab --- /dev/null +++ b/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md @@ -0,0 +1,42 @@ +## Purpose + +Defines the behavioral contract for gateway-token freshness during flow execution: that a flow run starts with a full-TTL gateway token, that token freshness is maintained between flow steps without aborting concurrent runs, and that gateway authentication failures are recoverable through the standard flow retry policy. + +## ADDED Requirements + +### Requirement: Flow runs start with a full-TTL gateway token + +A flow run SHALL refresh provider access when it starts on an already-running workspace, regardless of how recently the workspace last synced providers, so the gateway token issued for the run carries the full configured TTL. The refresh SHALL still defer while another run is active in the workspace, and SHALL still proceed through the standard provider-sync lock and credential-hash comparison. + +#### Scenario: Flow starts long after the last interactive sync + +- **WHEN** a flow run starts on a workspace that is already running and whose provider sync is younger than the freshness threshold +- **THEN** provider access is refreshed anyway and the flow starts with a freshly issued gateway token + +#### Scenario: Another run is active when the flow starts + +- **WHEN** a flow run starts while an unrelated run is active in the same workspace +- **THEN** the forced refresh is deferred, the concurrent run is not interrupted, and the flow proceeds + +### Requirement: Provider access is refreshed at flow step boundaries + +A flow run SHALL attempt a provider-access refresh before each step executes, after the run's cancellation and lease checks for that step. Because a flow's own message run is finalized between steps, the refresh SHALL defer only when an unrelated run is active in the workspace. A failed step-boundary refresh SHALL NOT fail the flow run; authentication failures surface from the step execution itself. + +#### Scenario: Multi-step flow crosses the token TTL + +- **WHEN** a flow runs longer than the gateway token TTL and reaches a step boundary +- **THEN** provider access is refreshed before the next step starts and the next step's gateway calls succeed + +#### Scenario: Boundary refresh fails + +- **WHEN** the step-boundary provider-access refresh fails +- **THEN** the flow continues to the next step and the failure is logged as a warning + +### Requirement: Gateway authentication failures are retryable + +A flow step failure caused by an invalid or expired gateway token SHALL be classified as retryable by the flow retry policy. A retry SHALL refresh provider access before resuming and SHALL resume execution at the failed node. + +#### Scenario: Step fails with an expired gateway token + +- **WHEN** a flow step fails with an `invalid_token` error from the provider gateway +- **THEN** the run is scheduled for retry under the standard backoff policy instead of failing terminally diff --git a/openspec/changes/flow-gateway-token-refresh/tasks.md b/openspec/changes/flow-gateway-token-refresh/tasks.md new file mode 100644 index 00000000..b2415346 --- /dev/null +++ b/openspec/changes/flow-gateway-token-refresh/tasks.md @@ -0,0 +1,13 @@ +## 1. Forced refresh at flow start + +- [x] 1.1 Add `force?: boolean` to `ensureProviderAccessFreshForExecution` in `apps/web/src/lib/opencode/providers.ts`: when set, skip the freshness-age early return for a running instance while keeping the provider-sync lock, the credential-hash check, and the active-run deferral. Verify with `src/lib/opencode/__tests__/providers.test.ts` cases: forced refresh runs despite a fresh matching sync record; a forced refresh still defers while the workspace has active runs. +- [x] 1.2 Add `options: { forceProviderRefresh?: boolean }` to `ensureWorkspaceRunningForExecution` in `apps/web/src/lib/opencode/session-execution.ts` and honor it only on the already-running path (a freshly started workspace was just synced). Verify with a `src/lib/opencode/__tests__/session-execution.test.ts` case asserting the flag is threaded through as `force: true`. +- [x] 1.3 Pass `{ forceProviderRefresh: true }` from both flow entry points in `apps/web/src/lib/flows/runner.ts` (`executeClaimedFlowRun` and `resumeClaimedFlowRun`). Verify with updated `runner.test.ts` assertions for scheduled/manual and resumed runs. + +## 2. Step-boundary refresh + +- [x] 2.1 In `executeFlowNodes` in `apps/web/src/lib/flows/runner.ts`, call `ensureProviderAccessFreshForExecution` before each node executes, after the cancellation and lease checks; catch and warn on failure without failing the run. Verify with `runner.test.ts` cases: one refresh per node in a two-node flow, and the run still succeeds when the refresh rejects. + +## 3. Retryable gateway auth failures + +- [x] 3.1 Add `invalid_token` to `isRetryableFlowRunError` in `apps/web/src/lib/flows/retry-policy.ts`. Verify with a `retry-policy.test.ts` case covering the surfaced `APIError: Unauthorized: invalid_token` text. From 7541798539befa688b396839cd9b4387e053c85c Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 2 Sep 2026 12:56:41 +0100 Subject: [PATCH 2/2] fix(flows): force the between-steps token refresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The step-boundary refresh used the freshness-threshold check, which is tuned for interactive cadence: it only fires once the sync is older than TTL minus the 60s skew. Flow steps run for minutes, so boundaries rarely land in that window — an 11-minute step started with 10 minutes of token life left and still failed at expiry (2026-09-02 11:07 run), defeating the between-steps refresh for exactly the flows it was built for. Force the refresh before every step after the first, so each step starts with the full gateway-token TTL; mid-step expiry now requires a single step longer than the TTL, which the retryable classification bounds to one step re-run. The first iteration stays exempt because the run entry points already force a fresh sync before the loop. Co-Authored-By: Claude Code --- .../src/lib/flows/__tests__/runner.test.ts | 24 +++++++++--- apps/web/src/lib/flows/runner.ts | 37 +++++++++++-------- .../flow-gateway-token-refresh/proposal.md | 2 +- .../specs/flow-execution/spec.md | 4 +- .../flow-gateway-token-refresh/tasks.md | 1 + 5 files changed, 44 insertions(+), 24 deletions(-) diff --git a/apps/web/src/lib/flows/__tests__/runner.test.ts b/apps/web/src/lib/flows/__tests__/runner.test.ts index 9689dc5c..153f82cd 100644 --- a/apps/web/src/lib/flows/__tests__/runner.test.ts +++ b/apps/web/src/lib/flows/__tests__/runner.test.ts @@ -354,11 +354,13 @@ describe('triggerFlowNow', () => { await runClaimedFlow(createClaimedFlow(), FlowRunTrigger.manual) expect(mocks.ensureWorkspaceRunningForExecution).toHaveBeenCalledWith('alice', 'user-1', { forceProviderRefresh: true }) - expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledTimes(1) + // 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('refreshes provider access before each flow node', async () => { + 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' }], @@ -374,20 +376,30 @@ describe('triggerFlowNow', () => { await runClaimedFlow(flow, FlowRunTrigger.manual) - expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledTimes(2) - expect(mocks.ensureProviderAccessFreshForExecution).toHaveBeenCalledWith({ slug: 'alice', userId: 'user-1' }) + 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(createClaimedFlow(), FlowRunTrigger.manual) + await runClaimedFlow(flow, FlowRunTrigger.manual) - expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledTimes(1) + expect(mocks.runFlowPromptAndReadOutput).toHaveBeenCalledTimes(2) expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' })) expect(mocks.markRunFailed).not.toHaveBeenCalled() }) diff --git a/apps/web/src/lib/flows/runner.ts b/apps/web/src/lib/flows/runner.ts index f5425a8e..2c22d665 100644 --- a/apps/web/src/lib/flows/runner.ts +++ b/apps/web/src/lib/flows/runner.ts @@ -142,21 +142,28 @@ async function executeFlowNodes(params: { return { status: 'failed', error: 'flow_lease_lost' } } - // Gateway tokens carry a short TTL, so multi-step flows must not carry an - // aged token into the next step. 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. - try { - await ensureProviderAccessFreshForExecution({ - slug: params.slug, - userId: params.executionUserId, - }) - } 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, - }) + // 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) diff --git a/openspec/changes/flow-gateway-token-refresh/proposal.md b/openspec/changes/flow-gateway-token-refresh/proposal.md index d3bc8540..1af681e7 100644 --- a/openspec/changes/flow-gateway-token-refresh/proposal.md +++ b/openspec/changes/flow-gateway-token-refresh/proposal.md @@ -10,7 +10,7 @@ During flow execution tokens are never refreshed: the runner makes no further re ## 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. -- Refresh provider access at every flow step boundary in `executeFlowNodes`, before each node executes. 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. +- 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 diff --git a/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md b/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md index c8dab9ab..424fc7da 100644 --- a/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md +++ b/openspec/changes/flow-gateway-token-refresh/specs/flow-execution/spec.md @@ -20,12 +20,12 @@ A flow run SHALL refresh provider access when it starts on an already-running wo ### Requirement: Provider access is refreshed at flow step boundaries -A flow run SHALL attempt a provider-access refresh before each step executes, after the run's cancellation and lease checks for that step. Because a flow's own message run is finalized between steps, the refresh SHALL defer only when an unrelated run is active in the workspace. A failed step-boundary refresh SHALL NOT fail the flow run; authentication failures surface from the step execution itself. +A flow run SHALL force a provider-access refresh before each step after the first executes — after the run's cancellation and lease checks for that step — so every step begins with a gateway token carrying the full configured TTL. Flow steps run for minutes, so a freshness-threshold check at a boundary would skip until the token is nearly expired and hand the next step a token that dies mid-step. The first step of an execution is exempt because the run's entry refresh has just re-issued tokens. Because a flow's own message run is finalized between steps, the forced refresh SHALL defer only when an unrelated run is active in the workspace. A failed step-boundary refresh SHALL NOT fail the flow run; authentication failures surface from the step execution itself. #### Scenario: Multi-step flow crosses the token TTL - **WHEN** a flow runs longer than the gateway token TTL and reaches a step boundary -- **THEN** provider access is refreshed before the next step starts and the next step's gateway calls succeed +- **THEN** provider access is force-refreshed before the next step starts and the next step's gateway calls succeed #### Scenario: Boundary refresh fails diff --git a/openspec/changes/flow-gateway-token-refresh/tasks.md b/openspec/changes/flow-gateway-token-refresh/tasks.md index b2415346..a58a0e26 100644 --- a/openspec/changes/flow-gateway-token-refresh/tasks.md +++ b/openspec/changes/flow-gateway-token-refresh/tasks.md @@ -7,6 +7,7 @@ ## 2. Step-boundary refresh - [x] 2.1 In `executeFlowNodes` in `apps/web/src/lib/flows/runner.ts`, call `ensureProviderAccessFreshForExecution` before each node executes, after the cancellation and lease checks; catch and warn on failure without failing the run. Verify with `runner.test.ts` cases: one refresh per node in a two-node flow, and the run still succeeds when the refresh rejects. +- [x] 2.2 Force the boundary refresh (`force: true`) except on the first loop iteration, which the entry-point refresh already covers. A threshold-based boundary check skips while the sync is young and cannot keep multi-minute steps inside the token TTL (2026-09-02 incident: an 11-minute step failed at expiry after two skipped boundaries). Verify with updated `runner.test.ts` cases: the boundary call carries `force: true`, a single-node flow makes no boundary call, and the run still succeeds when the forced refresh rejects. ## 3. Retryable gateway auth failures