State, accumulated budget, deadline, and reconciliation order
The policy implementation follows.
ts
export function defineDurableTask<State, App>(
definition: DurableTaskDefinition<State, App>,
options: DurableTaskOptions<State> = {},
): DurableTask<State, App> {
if (!definition || typeof definition !== 'object')
throw new TypeError('Use isState and replayPolicy');
const { id: definitionId, revision, steps: definitions, isState } = definition;
const clock = options.now;
if (
typeof definitionId !== 'string' ||
!definitionId.trim() ||
typeof revision !== 'string' ||
!revision.trim() ||
typeof isState !== 'function' ||
!Array.isArray(definitions) ||
!definitions.length ||
(clock !== undefined && typeof clock !== 'function')
)
throw new TypeError('Invalid durable task');
const now = clock?.bind(options) ?? Date.now,
validate = isState.bind(definition);
const authorize = definition.authorize?.bind(definition);
const boundStore = options.store && {
exclusive: options.store.exclusive.bind(options.store),
load: options.store.load.bind(options.store),
save: options.store.save.bind(options.store),
};
const steps = Object.freeze(
definitions.map((step: DurableStep<State, App>) => {
if (!step || typeof step !== 'object')
throw new TypeError('Use replayPolicy and OperationContext');
const { id, replayPolicy, run, reconcile } = step;
if (
typeof id !== 'string' ||
!id.trim() ||
!durableReplayPolicies.includes(replayPolicy) ||
typeof run !== 'function' ||
(reconcile !== undefined && typeof reconcile !== 'function')
)
throw new TypeError('Invalid durable step');
return Object.freeze({
id,
replayPolicy,
run: run.bind(step),
reconcile: reconcile?.bind(step),
});
}),
);
if (new Set(steps.map((step) => step.id)).size !== steps.length)
throw new TypeError('Duplicate durable step id');
return Object.freeze({
/**
* Resume or start one durable task.
*
* @param request - Task id plus an optional per-call store that overrides the bound one.
* @param context - Operation context supplying app, signal and diagnostics.
* @returns The final state, cloned so the caller cannot mutate the stored checkpoint.
* @throws As described for `defineDurableTask`; an already-completed task returns its state
* without running any step.
*/
async run(
request: { readonly taskId: string; readonly store?: DurableStore<State> },
context: OperationContext<App>,
): Promise<State> {
if (
!request ||
typeof request !== 'object' ||
typeof request.taskId !== 'string' ||
!request.taskId.trim()
)
throw new TypeError('Use a named durable task request');
if (!context?.signal) throw new TypeError('Durable tasks require OperationContext');
const { taskId } = request,
store = request.store ?? boundStore,
{ app, signal, diagnostics } = context;
signal.throwIfAborted();
if (
!store ||
typeof store.exclusive !== 'function' ||
typeof store.load !== 'function' ||
typeof store.save !== 'function'
)
throw new TypeError('Invalid durable store');
const exclusive = store.exclusive.bind(store),
load = store.load.bind(store),
save = store.save.bind(store);
signal.throwIfAborted();
return exclusive(taskId, async () => {
signal.throwIfAborted();
const loaded = await awaitWithSignal(load(taskId), signal);
signal.throwIfAborted();
if (!loaded) throw new TypeError('Invalid durable checkpoint');
let record = snapshotCheckpoint(loaded);
if (
record.definitionId !== definitionId ||
record.revision !== revision ||
!booleanResult(
validate(record.state),
'Durable state predicates must return boolean synchronously',
) ||
!Number.isSafeInteger(record.nextStepIndex) ||
record.nextStepIndex < 0 ||
record.nextStepIndex > steps.length ||
!Number.isSafeInteger(record.attemptCount) ||
record.attemptCount < 0 ||
!Number.isSafeInteger(record.maxAttempts) ||
record.maxAttempts < 1 ||
record.attemptCount > record.maxAttempts ||
record.attemptCount < record.nextStepIndex ||
!Number.isFinite(record.deadlineAtMs) ||
!durableStatuses.includes(record.status) ||
(record.status === 'completed') !== (record.nextStepIndex === steps.length)
)
throw new TypeError('Invalid durable checkpoint');
const deadline = record.deadlineAtMs;
assertActive(signal, deadline, now);
if (record.modelBudget) {
const { maxCalls, reservedCalls, maxCallsPerAttempt } = record.modelBudget;
if (
!Number.isSafeInteger(maxCalls) ||
maxCalls < 1 ||
!Number.isSafeInteger(reservedCalls) ||
reservedCalls < 0 ||
reservedCalls > maxCalls ||
!Number.isSafeInteger(maxCallsPerAttempt) ||
maxCallsPerAttempt < 1 ||
maxCallsPerAttempt > maxCalls
)
throw new TypeError('Invalid durable model budget');
}
const lifetime = createDeadlineSignal(deadline, signal, now),
activeSignal = lifetime.signal;
let saving: Promise<void> | undefined;
const persist = async () => {
saving = save(taskId, record);
await saving;
};
try {
if (authorize)
await awaitAssertion(
() =>
authorize(snapshotCheckpoint(record), { app, signal: activeSignal, diagnostics }),
activeSignal,
'Durable authorization must return void',
);
assertActive(activeSignal, deadline, now);
if (record.status === 'completed') {
const result = structuredClone(record.state);
assertActive(activeSignal, deadline, now);
return result;
}
if (record.status === 'cancelled') throw new Error('Durable task cancelled');
const engine = new AgentCoreEngine<State>({
maxRounds: Math.max(1, record.maxAttempts - record.attemptCount),
execution: { signal: activeSignal, diagnostics },
});
return await engine.run(async (_index, execution) => {
assertActive(activeSignal, deadline, now);
if (record.attemptCount >= record.maxAttempts)
throw new Error('Durable task budget exhausted');
const step = steps[record.nextStepIndex],
interrupted = record.status === 'running';
if (interrupted && step.replayPolicy === 'reconcile' && !step.reconcile)
throw new Error('Durable task reconciliation required');
const modelBudget = record.modelBudget;
if (
modelBudget &&
modelBudget.maxCalls - modelBudget.reservedCalls < modelBudget.maxCallsPerAttempt
)
throw new Error('Durable model budget exhausted');
record = {
...record,
attemptCount: record.attemptCount + 1,
status: 'running',
...(modelBudget
? {
modelBudget: {
...modelBudget,
reservedCalls: modelBudget.reservedCalls + modelBudget.maxCallsPerAttempt,
},
}
: {}),
};
// Await every save under the lock; cancellation cannot undo an in-flight commit.
await persist();
assertActive(activeSignal, deadline, now);
const local = structuredClone(record.state);
assertActive(activeSignal, deadline, now);
const handle =
interrupted && step.replayPolicy === 'reconcile' ? step.reconcile! : step.run;
const nextState = await awaitWithSignal(
handle(local, {
app,
signal: execution.signal,
diagnostics: execution.diagnostics,
attemptCount: record.attemptCount,
maxModelCalls: record.modelBudget?.maxCallsPerAttempt,
}),
activeSignal,
);
assertActive(activeSignal, deadline, now);
if (
!booleanResult(
validate(nextState),
'Durable state predicates must return boolean synchronously',
)
)
throw new TypeError('Invalid durable step state');
const state = structuredClone(nextState);
assertActive(activeSignal, deadline, now);
record = {
...record,
state,
nextStepIndex: record.nextStepIndex + 1,
status: record.nextStepIndex + 1 === steps.length ? 'completed' : 'ready',
};
await persist();
assertActive(activeSignal, deadline, now);
if (record.status === 'completed') {
const value = structuredClone(record.state);
assertActive(activeSignal, deadline, now);
return { type: 'complete' as const, value };
}
return { type: 'continue' as const };
});
} finally {
// Engine may stop waiting immediately on cancellation. Keep the store
// lock until an already-started atomic save has acknowledged completion.
try {
await saving?.catch(() => {});
} finally {
lifetime.dispose();
}
}
});
},
});
}
function assertActive(signal: AbortSignal, deadlineAtMs: number, now: () => number): void {
signal.throwIfAborted();
if (deadlineAtMs <= now()) throw new Error('Durable task expired');
signal.throwIfAborted();
}
function snapshotCheckpoint<State>(record: DurableCheckpoint<State>): DurableCheckpoint<State> {
const { version } = record;
// Only checkpoint layout 2 is readable; older layouts are not upgraded in place.
if (version !== 2) throw new TypeError('Unsupported durable checkpoint version');
const {
definitionId,
revision,
nextStepIndex,
attemptCount,
maxAttempts,
deadlineAtMs,
status,
state,
modelBudget,
} = record;
if (
modelBudget !== undefined &&
(!modelBudget || typeof modelBudget !== 'object' || Array.isArray(modelBudget))
)
throw new TypeError('Invalid durable model budget');
return {
version,
definitionId,
revision,
nextStepIndex,
attemptCount,
maxAttempts,
deadlineAtMs,
status,
state: structuredClone(state),
...(modelBudget !== undefined
? {
modelBudget: {
maxCalls: modelBudget.maxCalls,
reservedCalls: modelBudget.reservedCalls,
maxCallsPerAttempt: modelBudget.maxCallsPerAttempt,
},
}
: {}),
};
}