DurableStep

KIT · interface · Type contract

One step of a durable task.

Import

TypeScript

import type { DurableStep } from '@jangts/agent-core/kits/durable';

JavaScript

/** @typedef {import('@jangts/agent-core/kits/durable').DurableStep} DurableStep */

Other public entry points:@jangts/agent-core/types, @jangts/agent-core/structs, @jangts/agent-core/kits, @jangts/agent-core/kits/durable.

Definition

interface DurableStep<State, App>

Type parameters

Type parameter	Constraint	Default type
State	—	—
App	—	—

Field structure

Field	Type	Required	Description
id	string	Required	Read-only. Unique step id used in errors and diagnostics.
replayPolicy	"safe" | "reconcile"	Required	Read-only. Resume behaviour when a previous attempt was interrupted mid-step.
run	(state: Readonly<State>, context: DurableStepContext<App>) => Promise<State>	Required	Executes the step and returns the complete next state.
reconcile	((state: Readonly<State>, context: DurableStepContext<App>) => Promise<State>) | undefined	Optional	Required when replayPolicy is reconcile; repairs state after an interrupted attempt.

Member index

Member	Purpose
run	Executes the step and returns the complete next state.

Member details

run

Executes the step and returns the complete next state.

DurableStep.run(state: Readonly<State>, context: DurableStepContext<App>): Promise<State>

Executes the step and returns the complete next state.

Parameter	Type	Passing	Description
state	Readonly<State>	Required	—
context	DurableStepContext<App>	Required	—

Fields of context

Field	Type	Required	Description
attemptCount	number	Required	Read-only. One-based attempt number currently executing (already counted and persisted).
maxModelCalls	number | undefined	Optional	Read-only. App binds its real Runner to this reserved per-attempt maximum, keeping request limits.
app	unknown extends App ? App : Readonly<App>	Required	Read-only. See the type definition and usage context.
logger	SubFlowLog | undefined	Optional	Read-only. Explicit optional recorder for this operation; never inferred from diagnostics.
signal	AbortSignal	Required	Read-only. Non-optional cancellation signal; never undefined once execution has started.
diagnostics	DiagnosticContext | undefined	Optional	Read-only. Diagnostic scope of the enclosing call, when recording is enabled.

Returns: Promise<State>.

Awaited value: Awaited<State>.

Examples

• Durable: explicit checkpoints and controlled recovery: The App provides real checkpoint storage and an exclusive lock; a task runs or resumes explicitly along the same definition and accumulated budget.

First understand one execution with an isolated in-memory checkpoint

Beginner · Node.js

This is an isolated demonstration of the minimal storage contract; real cross-process or cross-request recovery must use an App store with exclusivity and durability guarantees.

TypeScript

import { Durable } from '@jangts/agent-core/kits/durable';
import type { DurableCheckpoint, DurableStore } from '@jangts/agent-core/kits/durable';
type State = { count: number };
const records = new Map<string, DurableCheckpoint<State>>([
  ['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store: DurableStore<State> = {
  exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => {}, () => {}); return result; },
  async load(id) { const record = records.get(id); return record && structuredClone(record); },
  async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value: unknown): value is State => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
const task = Durable.defineDurableTask<State, undefined>({
  id: 'count', revision: '1', isState,
  steps: [{ id: 'increment', replayPolicy: 'safe', async run(state) { return { count: state.count + 1 }; } }],
});
const context = { app: undefined, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(records.get('demo-task')?.attemptCount);

JavaScript

import { Durable } from '@jangts/agent-core/kits/durable';
const records = new Map([
    ['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store = {
    exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => { }, () => { }); return result; },
    async load(id) { const record = records.get(id); return record && structuredClone(record); },
    async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value) => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
const task = Durable.defineDurableTask({
    id: 'count', revision: '1', isState,
    steps: [{ id: 'increment', replayPolicy: 'safe', async run(state) { return { count: state.count + 1 }; } }],
});
const context = { app: undefined, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(records.get('demo-task')?.attemptCount);

Add a real reconciliation policy for interruptions to the same run

Configuration · Node.js

The example does not presume to interpret "cannot confirm whether it committed" as failure or re-execution; production reconciliation must query the real business system.

TypeScript

import { Durable } from '@jangts/agent-core/kits/durable';
import type { DurableCheckpoint, DurableStore } from '@jangts/agent-core/kits/durable';
type State = { count: number };
const records = new Map<string, DurableCheckpoint<State>>([
  ['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store: DurableStore<State> = {
  exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => {}, () => {}); return result; },
  async load(id) { const record = records.get(id); return record && structuredClone(record); },
  async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value: unknown): value is State => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
// 假设业务服务以 taskId 为幂等键保存真实提交结果；这里用内存 Map 演示。
const committed = new Map<string, number>();
const task = Durable.defineDurableTask<State, { key: string }>({
  id: 'count', revision: '1', isState,
  steps: [{ id: 'commit', replayPolicy: 'reconcile',
    async run(state, { app }) {
      const value = committed.get(app.key) ?? state.count + 1;
      committed.set(app.key, value);
      return { count: value };
    },
    async reconcile(state, { app }) {
      const value = committed.get(app.key);
      if (value === undefined) throw new Error('没有可核实的提交结果，需要 App 决定后续动作');
      return { count: value };
    },
  }],
});
const context = { app: { key: 'demo-task' }, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(await task.run({ taskId: 'demo-task', store }, context)); // 已完成记录不会再次执行步骤
console.log(committed.size);

JavaScript

import { Durable } from '@jangts/agent-core/kits/durable';
const records = new Map([
    ['demo-task', { version: 2, definitionId: 'count', revision: '1', nextStepIndex: 0, attemptCount: 0, maxAttempts: 3, deadlineAtMs: Date.now() + 60_000, status: 'ready', state: { count: 0 } }],
]);
let tail = Promise.resolve();
// 隔离示例：进程内串行锁和内存记录，不是生产持久化实现。
const store = {
    exclusive(_id, operation) { const result = tail.then(operation); tail = result.then(() => { }, () => { }); return result; },
    async load(id) { const record = records.get(id); return record && structuredClone(record); },
    async save(id, record) { records.set(id, structuredClone(record)); },
};
const isState = (value) => !!value && typeof value === 'object' && 'count' in value && typeof value.count === 'number';
// 假设业务服务以 taskId 为幂等键保存真实提交结果；这里用内存 Map 演示。
const committed = new Map();
const task = Durable.defineDurableTask({
    id: 'count', revision: '1', isState,
    steps: [{ id: 'commit', replayPolicy: 'reconcile',
            async run(state, { app }) {
                const value = committed.get(app.key) ?? state.count + 1;
                committed.set(app.key, value);
                return { count: value };
            },
            async reconcile(state, { app }) {
                const value = committed.get(app.key);
                if (value === undefined)
                    throw new Error('没有可核实的提交结果，需要 App 决定后续动作');
                return { count: value };
            },
        }],
});
const context = { app: { key: 'demo-task' }, signal: new AbortController().signal };
console.log(await task.run({ taskId: 'demo-task', store }, context));
console.log(await task.run({ taskId: 'demo-task', store }, context)); // 已完成记录不会再次执行步骤
console.log(committed.size);

Related policy source

• State, accumulated budget, deadline, and reconciliation order
• Current status and replay policy enumerations

Related definitions

• durableReplayPolicies
• DurableStepContext