diff --git a/src/domain/workflow-timeline.ts b/src/domain/workflow-timeline.ts index 4de0ed8..9240a86 100644 --- a/src/domain/workflow-timeline.ts +++ b/src/domain/workflow-timeline.ts @@ -19,6 +19,8 @@ export interface WorkflowTimelineItem { import { OWNER_DECISION_SIGNAL } from './owner-decision.ts'; export interface PendingDecision { + /** `recovery`: a restart interrupted the step at nodeId; the owner retries it or cancels the run. */ + readonly kind: 'owner' | 'recovery'; readonly signal: string; readonly nodeId: string | null; readonly prompt: string | null; @@ -220,12 +222,14 @@ export function buildWorkflowTimeline(input: { ? (record(resolver['config'])['actions'] as unknown[]).map((action) => record(action)['action']).filter((action): action is string => typeof action === 'string') : []; const frame = input.position?.frames.findLast((item) => item.kind === 'workflow'); + const recovery = signal === 'recovery'; pendingDecision = { + kind: recovery ? 'recovery' : 'owner', signal, nodeId: waitNodeId, prompt, // The gate's resolver says what it accepts; the generic list only for a wait without one. - availableActions: actions.length > 0 ? actions : ['proceed', 'skip', 'rollback', 'note'], + availableActions: recovery ? [] : actions.length > 0 ? actions : ['proceed', 'skip', 'rollback', 'note'], context: frame?.kind === 'workflow' ? decisionContext(frame.context) : null, expectedLifecycleSeq: input.instance.projection.seq, expectedExecutionSeq: input.replayStatus.replaySeq, diff --git a/src/http/contract.ts b/src/http/contract.ts index 9cec479..29bd1bf 100644 --- a/src/http/contract.ts +++ b/src/http/contract.ts @@ -268,6 +268,7 @@ export const DecisionContextResponse = z.object({ export type DecisionContextResponse = z.infer; export const PendingDecisionResponse = z.object({ + kind: z.enum(['owner', 'recovery']).default('owner'), signal: z.string(), nodeId: z.string().nullable(), prompt: z.string().nullable(), diff --git a/src/http/server-workflows.test.ts b/src/http/server-workflows.test.ts index 7409a35..926673d 100644 --- a/src/http/server-workflows.test.ts +++ b/src/http/server-workflows.test.ts @@ -386,6 +386,41 @@ describe('HTTP Server Workflow, Function, Layout, and Timeline Endpoints', () => expect((await app.inject({ method: 'POST', url: '/api/workflow-instances/missing/cancel', payload: { expectedLifecycleSeq: 0 } })).statusCode).toBe(404); }); + it('retries the step a restart interrupted, and only an interrupted run', async () => { + const { app, workflowCatalog, workflowInstanceStore, privilegedActionStore } = await createTestServer(); + const now = '2026-09-21T12:00:00.000Z'; + const doc: WorkflowDocumentV2 = { schemaVersion: 2, entryNodeId: 'work', nodes: [ + { nodeId: 'work', type: 'emit', event: 'go', routes: { go: 'end' } }, { nodeId: 'end', type: 'succeed' }, + ] }; + const policy = { schemaVersion: 1 as const, policyVersion: 'privileged-action-policy.v1' as const, revision: 1, rules: [] }; + privilegedActionStore.putPolicy({ policy, createdAt: now }); + workflowCatalog.createDraft({ workflowId: 'wf-restart', name: 'wf-restart', document: doc, now }); + workflowCatalog.recordValidation({ workflowId: 'wf-restart', expectedRevision: 1, document: doc, validation: { diagnostics: [] } }); + const pub = workflowCatalog.publish({ workflowId: 'wf-restart', expectedRevision: 1, now }); + if (!pub.ok) throw new Error('publish failed'); + const version = PublishedWorkflowVersionV2.parse(pub.published); + const snapshot: WorkflowSnapshotV2 = { schemaVersion: 2, capturedAt: now, rootWorkflow: version, workflowDocuments: [version], appConfigs: [], skills: [], modelProfiles: [], functionVersions: [], privilegedActionPolicy: policy }; + workflowInstanceStore.create({ id: 'run-restart', workflowId: 'wf-restart', workflowVersion: 1, snapshot, dueAt: now, now }); + workflowInstanceStore.initializeExecution({ instanceId: 'run-restart', input: {}, now }); + const url = '/api/workflow-instances/run-restart/resume'; + + workflowInstanceStore.claimNext(now); + expect((await app.inject({ method: 'POST', url, payload: { expectedLifecycleSeq: 1 } })).statusCode).toBe(409); + + // The driver restarts: the running step is interrupted and the run waits. + expect(workflowInstanceStore.recoverStale('2026-09-21T12:05:00.000Z', '2026-09-21T12:01:00.000Z')).toBe(1); + const timeline = asJson<{ lifecycleSeq: number; pendingDecision: { kind: string; nodeId: string; availableActions: string[] } }>( + await app.inject({ method: 'GET', url: '/api/workflow-instances/run-restart/timeline' }), + ); + expect(timeline.pendingDecision).toMatchObject({ kind: 'recovery', nodeId: 'work', availableActions: [] }); + + expect((await app.inject({ method: 'POST', url, payload: {} })).statusCode).toBe(400); + expect((await app.inject({ method: 'POST', url, payload: { expectedLifecycleSeq: timeline.lifecycleSeq - 1 } })).statusCode).toBe(409); + expect((await app.inject({ method: 'POST', url, payload: { expectedLifecycleSeq: timeline.lifecycleSeq } })).statusCode).toBe(200); + expect(workflowInstanceStore.get('run-restart')?.projection.status).toBe('queued'); + expect(workflowInstanceStore.getReplayStatus('run-restart').status).toBe('ok'); + }); + it('lists an app\'s runs (its update- workflow) with their mode', async () => { const { app, workflowCatalog, workflowInstanceStore, privilegedActionStore } = await createTestServer(); const now = '2026-09-21T12:00:00.000Z'; diff --git a/src/http/server.ts b/src/http/server.ts index 22cc453..939d414 100644 --- a/src/http/server.ts +++ b/src/http/server.ts @@ -835,6 +835,30 @@ export async function buildServer(deps: ServerDeps): Promise { return reply.send({ ok: true, instance: res.instance, projection: res.projection }); }); + const ResumeBody = z.object({ expectedLifecycleSeq: z.number().int().nonnegative() }); + + // A restart interrupted the running step (recoverStale): the owner has it run again. Any other wait is not resumed here. + app.post('/api/workflow-instances/:instanceId/resume', (request, reply) => { + if (!deps.workflowInstanceStore) return fail(reply, 503, 'unavailable', 'instance store unavailable'); + const params = InstanceIdParam.safeParse(request.params); + if (!params.success) return fail(reply, 400, 'invalid_request', issues(params.error)); + const body = ResumeBody.safeParse(request.body); + if (!body.success) return fail(reply, 400, 'invalid_request', issues(body.error)); + if (deps.workflowInstanceStore.get(params.data.instanceId) === null) return fail(reply, 404, 'not_found', 'workflow instance not found'); + if (deps.workflowInstanceStore.getExecutionPosition(params.data.instanceId)?.waitingFor !== 'recovery') { + return fail(reply, 409, 'not_interrupted', 'the run is not waiting after an interrupted step'); + } + const res = deps.workflowInstanceStore.append({ + instanceId: params.data.instanceId, + expectedSeq: body.data.expectedLifecycleSeq, + event: { type: 'RESUME', data: { signal: 'recovery' } }, + now: clock.now().toISOString(), + }); + if (!res.ok) return fail(reply, 409, res.reason, 'cannot resume workflow instance'); + hub.notify(); + return reply.send({ ok: true, instance: res.instance }); + }); + const CancelBody = z.object({ expectedLifecycleSeq: z.number().int().nonnegative(), note: z.string().min(1).max(500).nullable().optional(), diff --git a/src/web/api.ts b/src/web/api.ts index 3550aab..018ed91 100644 --- a/src/web/api.ts +++ b/src/web/api.ts @@ -110,6 +110,8 @@ export const api = { workflowTimeline: (id: string) => call(WorkflowTimelineResponse, `/api/workflow-instances/${encodeURIComponent(id)}/timeline`), submitWorkflowDecision: (id: string, body: unknown) => call(z.object({ ok: z.boolean(), instance: z.unknown(), projection: z.unknown() }), `/api/workflow-instances/${encodeURIComponent(id)}/decisions`, body), + resumeWorkflowInstance: (id: string, expectedLifecycleSeq: number) => + call(z.object({ ok: z.boolean(), instance: z.unknown() }), `/api/workflow-instances/${encodeURIComponent(id)}/resume`, { expectedLifecycleSeq }), cancelWorkflowInstance: (id: string, expectedLifecycleSeq: number) => call(z.object({ ok: z.boolean(), instance: z.unknown() }), `/api/workflow-instances/${encodeURIComponent(id)}/cancel`, { expectedLifecycleSeq }), modelProfiles: () => call(ModelProfileListResponse, '/api/model-profiles'), diff --git a/src/web/views/instance-timeline.tsx b/src/web/views/instance-timeline.tsx index f9c9f62..baf9db1 100644 --- a/src/web/views/instance-timeline.tsx +++ b/src/web/views/instance-timeline.tsx @@ -42,6 +42,20 @@ export function InstanceTimelineView({ instanceId }: { readonly instanceId: stri } }; + const handleRetry = async () => { + if (!timeline) return; + setSubmitting(true); + setSubmitError(null); + try { + await api.resumeWorkflowInstance(instanceId, timeline.lifecycleSeq); + refresh(); + } catch (err) { + setSubmitError(err instanceof Error ? err.message : 'Failed to retry'); + } finally { + setSubmitting(false); + } + }; + const handleCancel = async () => { if (!timeline || !window.confirm(`Cancel ${timeline.instanceId}? It cannot be resumed.`)) return; setSubmitting(true); @@ -58,7 +72,8 @@ export function InstanceTimelineView({ instanceId }: { readonly instanceId: stri if (!timeline) return error ? : ; - const isWaiting = timeline.status === 'waiting_owner' || timeline.waitingFor !== null; + const interrupted = timeline.pendingDecision?.kind === 'recovery'; + const isWaiting = !interrupted && (timeline.status === 'waiting_owner' || timeline.waitingFor !== null); return (
@@ -105,6 +120,20 @@ export function InstanceTimelineView({ instanceId }: { readonly instanceId: stri {submitError && !isWaiting &&

{submitError}

} + {interrupted && ( +
+

Interrupted by a restart

+

+ The step {timeline.currentNode ?? 'unknown'} was running when the driver restarted, so its result is unknown. + Retrying runs the step again; for a step that changes the app (backup, apply) the executor refuses to apply over a + manifest that has changed since. +

+ +
+ )} + {isWaiting && (

Owner Decision Required

diff --git a/src/web/views/workflow-views.test.tsx b/src/web/views/workflow-views.test.tsx index fb8f4b0..2b59c72 100644 --- a/src/web/views/workflow-views.test.tsx +++ b/src/web/views/workflow-views.test.tsx @@ -403,6 +403,35 @@ describe('Web UI Stage 7 Views', () => { expect(container.querySelector('.graph-fallback-list li[data-run="waiting"]')?.textContent).toBe('gate (wait)'); }); + it('offers to retry a step a restart interrupted', async () => { + const now = '2026-09-21T12:00:00.000Z'; + const wfDoc: WorkflowDocumentV2 = { schemaVersion: 2, entryNodeId: 'work', nodes: [ + { nodeId: 'work', type: 'emit', event: 'go', routes: { go: 'end' } }, { nodeId: 'end', type: 'succeed' }, + ] }; + workflowCatalog.createDraft({ workflowId: 'wf-restart', name: 'Restart WF', document: wfDoc, now }); + workflowCatalog.recordValidation({ workflowId: 'wf-restart', expectedRevision: 1, document: wfDoc, validation: { diagnostics: [] } }); + const pub = workflowCatalog.publish({ workflowId: 'wf-restart', expectedRevision: 1, now }); + if (!pub.ok) throw new Error('publish failed'); + const pubV2 = PublishedWorkflowVersionV2.parse(pub.published); + const policy = { schemaVersion: 1 as const, policyVersion: 'privileged-action-policy.v1' as const, revision: 1, rules: [] }; + privilegedActionStore.putPolicy({ policy, createdAt: now }); + workflowInstanceStore.create({ + id: 'inst-restart', workflowId: 'wf-restart', workflowVersion: 1, dueAt: now, now, + snapshot: { schemaVersion: 2, capturedAt: now, rootWorkflow: pubV2, workflowDocuments: [pubV2], appConfigs: [], skills: [], modelProfiles: [], functionVersions: [], privilegedActionPolicy: policy }, + }); + workflowInstanceStore.initializeExecution({ instanceId: 'inst-restart', input: {}, now }); + workflowInstanceStore.claimNext(now); + workflowInstanceStore.recoverStale('2026-09-21T12:05:00.000Z', '2026-09-21T12:01:00.000Z'); + + window.location.hash = '#/instances/inst-restart'; + render(, container); + await waitForText('Interrupted by a restart'); + expect(container.textContent).not.toContain('Owner Decision Required'); + [...container.querySelectorAll('button')].find((button) => button.textContent === 'Retry the step')!.click(); + for (let i = 0; i < 50 && workflowInstanceStore.get('inst-restart')?.projection.status !== 'queued'; i++) await flush(); + expect(workflowInstanceStore.get('inst-restart')?.projection.status).toBe('queued'); + }); + it('renders model profiles view', async () => { window.location.hash = '#/models'; render(, container);