fix(workflows): reject divergent root history (#8)

Co-authored-by: Yuriy Butenko <yuriy@MacBook-Pro-Yuriy.local>

Author: WyvernMonarch

packages/workflows/src/context.ts

@@ -565,29 +565,20 @@ export class WorkflowContextImpl implements WorkflowContextInterface {
 	}
 
 	/**
-	 * Validate that all expected entries in the branch were visited.
-	 * Throws HistoryDivergedError if there are unvisited entries.
+	 * Validate that every direct entry in this scope was visited. Nested scopes
+	 * validate their own entries when they execute.
 	 */
 	validateComplete(): void {
-		const prefix = locationToKey(this.storage, this.currentLocation);
-
-		for (const key of this.storage.history.entries.keys()) {
-			// Check if this key is under our current location prefix
-			// Handle root prefix (empty string) specially - all keys are under root
-			const isUnderPrefix =
-				prefix === ""
-					? true // Root: all keys are children
-					: key.startsWith(`${prefix}/`) || key === prefix;
+		for (const [key, entry] of this.storage.history.entries) {
+			const isDirectChild =
+				entry.location.length === this.currentLocation.length + 1 &&
+				isLocationPrefix(this.currentLocation, entry.location);
 
-			if (isUnderPrefix) {
-				if (!this.visitedKeys.has(key)) {
-					// Entry exists in history but wasn't visited
-					// This means workflow code may have changed
-					throw new HistoryDivergedError(
-						`Entry "${key}" exists in history but was not visited. ` +
-							`Workflow code may have changed. Use ctx.removed() to handle migrations.`,
-					);
-				}
+			if (isDirectChild && !this.visitedKeys.has(key)) {
+				throw new HistoryDivergedError(
+					`Entry "${key}" exists in history but was not visited. ` +
+						"Workflow code may have changed. Use ctx.removed() to handle migrations.",
+				);
 			}
 		}
 	}
@@ -791,6 +782,13 @@ export class WorkflowContextImpl implements WorkflowContextInterface {
 
 		// Check for duplicate name in current execution
 		this.checkDuplicateName(config.name);
+		const parentKey = locationToKey(this.storage, this.currentLocation);
+		const candidateKey = parentKey
+			? `${parentKey}/${config.name}`
+			: config.name;
+		if (!this.storage.history.entries.has(candidateKey)) {
+			this.validateComplete();
+		}
 
 		const location = appendName(
 			this.storage,
@@ -2558,6 +2556,14 @@ export class WorkflowContextImpl implements WorkflowContextInterface {
 
 		// Mark this entry as visited for validateComplete
 		this.markVisited(key);
+		if (originalType === "message") {
+			const generatedKeyPrefix = `${key}:`;
+			for (const existingKey of this.storage.history.entries.keys()) {
+				if (existingKey.startsWith(generatedKeyPrefix)) {
+					this.markVisited(existingKey);
+				}
+			}
+		}
 
 		this.stopRollbackIfMissing(existing);
 

packages/workflows/src/index.ts

@@ -154,6 +154,7 @@ import {
 import {
 	CriticalError,
 	EvictedError,
+	HistoryDivergedError,
 	MessageWaitError,
 	RollbackCheckpointError,
 	RollbackError,
@@ -993,6 +994,7 @@ async function executeWorkflow<TInput, TOutput>(
 
 	try {
 		const output = await workflowFn(ctx, effectiveInput);
+		ctx.validateComplete();
 
 		storage.state = "completed";
 		storage.output = output;
@@ -1035,7 +1037,10 @@ async function executeWorkflow<TInput, TOutput>(
 			);
 		}
 
-		if (error instanceof RollbackCheckpointError) {
+		if (
+			error instanceof HistoryDivergedError ||
+			error instanceof RollbackCheckpointError
+		) {
 			await setFailedState(storage, driver, error, historyNotifier);
 			if (onError && !isErrorReported(error)) {
 				await notifyError(onError, logger, {

packages/workflows/tests/steps.test.ts

@@ -6,6 +6,7 @@ import {
 	EntryInProgressError,
 	HistoryDivergedError,
 	InMemoryDriver,
+	loadStorage,
 	RollbackError,
 	runWorkflow,
 	StepExhaustedError,
@@ -74,6 +75,28 @@ for (const mode of modes) {
 				.result;
 			expect(callCount).toBe(1);
 		});
+		it("should reject a renamed root step on replay", async () => {
+			const originalWorkflow = async (ctx: WorkflowContextInterface) => {
+				return await ctx.step("original-step-name", async () => "original");
+			};
+			const renamedWorkflow = async (ctx: WorkflowContextInterface) => {
+				return await ctx.step("renamed-step-name", async () => "changed");
+			};
+
+			await runWorkflow("wf-1", originalWorkflow, undefined, driver, { mode })
+				.result;
+
+			await expect(
+				runWorkflow("wf-1", renamedWorkflow, undefined, driver, { mode })
+					.result,
+			).rejects.toThrow(HistoryDivergedError);
+			const storage = await loadStorage(driver);
+			expect(storage.nameRegistry).toEqual(["original-step-name"]);
+			expect([...storage.history.entries.keys()]).toEqual([
+				"original-step-name",
+			]);
+			expect(storage.state).not.toBe("completed");
+		});
 
 		it("should replay void step on restart", async () => {
 			let callCount = 0;