feat(workflows): retry a step that a driver restart interrupted
recoverStale leaves an interrupted run waiting on 'recovery', and nothing but the store could resume it; the instance page showed the owner-decision card, whose actions do not apply to that wait. POST /api/workflow-instances/:id/resume (expectedLifecycleSeq) resumes only a run waiting after an interrupted step. The timeline marks such a wait as kind 'recovery' with no decision actions, and the page shows 'Interrupted by a restart' with 'Retry the step' (Cancel stays in the header). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
591af85dd3
commit
95d1511bda
7 files changed
+126
-2
No files matched your search
@@ -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,
|
||||
|
||||
@@ -268,6 +268,7 @@ export const DecisionContextResponse = z.object({
|
||||
export type DecisionContextResponse = z.infer<typeof DecisionContextResponse>;
|
||||
|
||||
export const PendingDecisionResponse = z.object({
|
||||
kind: z.enum(['owner', 'recovery']).default('owner'),
|
||||
signal: z.string(),
|
||||
nodeId: z.string().nullable(),
|
||||
prompt: z.string().nullable(),
|
||||
|
||||
@@ -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-<app> workflow) with their mode', async () => {
|
||||
const { app, workflowCatalog, workflowInstanceStore, privilegedActionStore } = await createTestServer();
|
||||
const now = '2026-09-21T12:00:00.000Z';
|
||||
|
||||
@@ -835,6 +835,30 @@ export async function buildServer(deps: ServerDeps): Promise<FastifyInstance> {
|
||||
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(),
|
||||
|
||||
@@ -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'),
|
||||
|
||||
@@ -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 ? <ErrorNote message={error} /> : <Loading />;
|
||||
|
||||
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 (
|
||||
<section class="instance-timeline-view">
|
||||
@@ -105,6 +120,20 @@ export function InstanceTimelineView({ instanceId }: { readonly instanceId: stri
|
||||
|
||||
{submitError && !isWaiting && <p class="error-note">{submitError}</p>}
|
||||
|
||||
{interrupted && (
|
||||
<div class="decision-card-interactive">
|
||||
<h3>Interrupted by a restart</h3>
|
||||
<p>
|
||||
The step <code>{timeline.currentNode ?? 'unknown'}</code> 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.
|
||||
</p>
|
||||
<button type="button" class="btn-primary" disabled={submitting} onClick={() => void handleRetry()}>
|
||||
{submitting ? 'Retrying…' : 'Retry the step'}
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{isWaiting && (
|
||||
<div class="decision-card-interactive">
|
||||
<h3>Owner Decision Required</h3>
|
||||
|
||||
@@ -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(<App />, 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(<App />, container);
|
||||
|
||||
Reference in new issue
Block a user