Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/reference/topology-yaml.md
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,9 @@ fan-out unchanged. The exception is `resume --retry-failed` for a source whose
verifier failed: it moves the published manifests to
`dynamic-expansion-history/` before the source runs again (and refuses if
another source published any of them), so the group expands again from the new
output.
output. The generated children's artifacts and the rendered prompts of later
nodes that wait on the group move with them, so those prompts render again from
the new expansion.

## Model Fan-Out

Expand Down
29 changes: 24 additions & 5 deletions packages/runtime/src/dynamic-expansion-retry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,8 @@ export function planDynamicExpansionRetryArchive(input: {
}
// Validate the attempt-owned entries and the run state now; the move and the
// prune themselves wait for the reset.
collectAttemptStateMoves(runRoot, manifests);
const runtimeBase = readDynamicRuntimeBase(runRoot);
collectAttemptStateMoves(runRoot, manifests, runtimeBase);
readArchivableRunState(runRoot);
return {
projectRoot,
Expand All @@ -136,7 +137,7 @@ export function planDynamicExpansionRetryArchive(input: {
manifestDirMode: manifestStat.mode & 0o777,
sourceNodeIds: [...new Set(input.sourceNodeIds)].sort(),
manifests,
runtimeBase: readDynamicRuntimeBase(runRoot)
runtimeBase
};
}

Expand All @@ -146,7 +147,9 @@ export function planDynamicExpansionRetryArchive(input: {
* The whole manifest directory is renamed in one step, so the old generation
* stays durable and no partially rewritten manifest set can be observed. The
* generation's attempt-owned artifacts move with it, so a regenerated attempt
* with a stable storage ID never meets a stale rendered prompt. The mutable
* with a stable storage ID never meets a stale rendered prompt. The published
* prompt of each planned task that waits on the generation moves too, so the
* next expansion renders it afresh instead of refusing it as changed. The mutable
* runtime graph and task plan are then re-derived from the sealed base with no
* ready group, exactly as the next render would publish them, so the control
* admission check re-derives cleanly before that render happens. Finally the
Expand All @@ -164,7 +167,7 @@ export function archiveDynamicExpansionsForRetry(plan: DynamicExpansionRetryPlan
fs.mkdirSync(archiveDir, { mode: 0o700 });
fs.renameSync(plan.manifestDir, path.join(archiveDir, "manifests"));
fs.mkdirSync(plan.manifestDir, { mode: plan.manifestDirMode });
const archivedAttemptPaths = collectAttemptStateMoves(plan.runRoot, plan.manifests).map((move) => {
const archivedAttemptPaths = collectAttemptStateMoves(plan.runRoot, plan.manifests, plan.runtimeBase).map((move) => {
const destination = path.join(archiveDir, move.relativePath);
fs.mkdirSync(path.dirname(destination), { recursive: true, mode: 0o700 });
fs.renameSync(move.source, destination);
Expand Down Expand Up @@ -235,7 +238,8 @@ function readArchivableRunState(runRoot: string): { statePath: string; document:

function collectAttemptStateMoves(
runRoot: string,
manifests: readonly DynamicExpansionManifest[]
manifests: readonly DynamicExpansionManifest[],
runtimeBase: DynamicRuntimeBase | undefined
): Array<{ source: string; relativePath: string }> {
const ownsAttempt = generationOwnsAttempt(manifests);
const moves: Array<{ source: string; relativePath: string }> = [];
Expand All @@ -262,6 +266,21 @@ function collectAttemptStateMoves(
moves.push({ source: path.join(sourceRoot, entry.name), relativePath: `${root.name}/${entry.name}` });
}
}
// A planned task whose prompt waits on a withdrawn group was rendered from this generation's
// children. Left in place, a prompt that names them no longer matches the render once the
// retried source plans other items, and every later render and admission check refuses it as
// changed. Moved, it renders again from the new expansion. Only a prompt that was never rendered
// is skipped: `lstat` counts a dangling symlink as present, so the file-type check refuses it.
const groupIds = new Set(manifests.map((manifest) => manifest.group_node_id));
for (const task of runtimeBase?.tasks ?? []) {
if (!(task.deferredPromptGroups ?? []).some((groupId) => groupIds.has(groupId))) continue;
const relativePath = `artifacts/${task.attemptId}/prompt.rendered.md`;
const source = path.join(runRoot, relativePath);
assertNoSymlinkComponents(runRoot, source, `dynamic retry prompt for ${task.attemptId}`);
if (fs.lstatSync(source, { throwIfNoEntry: false }) === undefined) continue;
Comment on lines +279 to +280

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Dangling attempt link gets skipped If artifacts/<attemptId> is a dangling symlink, the path check treats it as missing, and the prompt lookup skips it. The retry then resets Smithers and archives the manifests rather than refusing the invalid directory before reset. The next expansion cannot create the prompt there, leaving the run in need of manual repair. Check parent directories with lstat before treating the prompt as absent.

Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/runtime/src/dynamic-expansion-retry.ts
Line: 279-280

Comment:
**Dangling attempt link gets skipped** If `artifacts/<attemptId>` is a dangling symlink, the path check treats it as missing, and the prompt lookup skips it. The retry then resets Smithers and archives the manifests rather than refusing the invalid directory before reset. The next expansion cannot create the prompt there, leaving the run in need of manual repair. Check parent directories with `lstat` before treating the prompt as absent.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Fix in Claude Code

assertRegularFileInside(runRoot, source, `dynamic retry prompt for ${task.attemptId}`);
moves.push({ source, relativePath });
}
return moves;
}

Expand Down
108 changes: 104 additions & 4 deletions packages/runtime/test/dynamic-expansion.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,13 @@ import fs from "node:fs";
import path from "node:path";
import test from "node:test";

import { createInitialRunState, createNodeState, readRunState, writeRunState } from "@ultrafuzz/artifacts";
import {
ArtifactPathError,
createInitialRunState,
createNodeState,
readRunState,
writeRunState
} from "@ultrafuzz/artifacts";

import {
DynamicExpansionError,
Expand Down Expand Up @@ -81,12 +87,17 @@ function item(index: number): Record<string, unknown> {
};
}

/** A run whose sealed runtime controls fan `goals` from `planner` out into `fanout`, joined by `join`. */
/**
* A run whose sealed runtime controls fan `goals` from `planner` out into `fanout`, joined by `join`.
* Like the stock final report, the join's prompt waits on the group. Unlike it, the prompt names the
* generated children, as a custom join using `{{artifact_path:<group>}}` does.
*/
function sealedDynamicRun(runId: string, goals: Array<Record<string, unknown>>) {
const projectRoot = tempDirectory();
const runRoot = path.join(projectRoot, "runs", runId);
const sourceArtifactPath = path.join(runRoot, "artifacts", "planner", "plan.json");
const templatePath = path.join(runRoot, "templates", "worker.md");
const joinTemplatePath = path.join(runRoot, "templates", "join.md");
const graphPath = path.join(runRoot, "graph.json");
const tasksPath = path.join(runRoot, "smithers", "tasks.json");
const baseGraphPath = path.join(runRoot, "smithers", "runtime-base-graph.json");
Expand All @@ -96,8 +107,12 @@ function sealedDynamicRun(runId: string, goals: Array<Record<string, unknown>>)
fs.mkdirSync(path.dirname(tasksPath), { recursive: true });
fs.writeFileSync(sourceArtifactPath, `${JSON.stringify({ goals })}\n`, "utf8");
fs.writeFileSync(templatePath, "Investigate {{item.goal_prompt}}.\n", "utf8");
fs.writeFileSync(joinTemplatePath, "Join {{artifact_path:fanout}}.\n", "utf8");
const templateTask = compiledTask(projectRoot, runRoot, "fanout", "fanout", templatePath);
const joinTask = compiledTask(projectRoot, runRoot, "join", "join", undefined, ["fanout"]);
const joinTask = {
...compiledTask(projectRoot, runRoot, "join", "join", joinTemplatePath, ["fanout"]),
deferredPromptGroups: ["fanout"]
};
const group: CompiledSmithersDynamicGroup = {
groupNodeId: "fanout",
logicalNodeId: "fanout",
Expand Down Expand Up @@ -462,13 +477,16 @@ test("explicit source retry archives a complete expansion generation and rejects
});

test("explicit source retry re-derives the base runtime controls after archiving an expansion", () => {
const { controls } = sealedDynamicRun("retry-rematerialize", [item(0)]);
const { sourceArtifactPath, controls } = sealedDynamicRun("retry-rematerialize", [item(0)]);
const { runId, projectRoot, runRoot } = controls;
const expanded = materializeDynamicRuntime({ ...controls, readyGroupIds: ["fanout"] });
assert.deepEqual(expanded.expandedGroupIds, ["fanout"]);
const generated = expanded.tasks.find((task) => task.metadata.node.dynamic !== undefined);
assert.ok(generated?.renderedPromptPath);
assert.equal(fs.existsSync(generated.renderedPromptPath), true);
const joinPromptPath = path.join(runRoot, "artifacts", "join", "prompt.rendered.md");
const withdrawnJoinPrompt = fs.readFileSync(joinPromptPath, "utf8");
assert.ok(withdrawnJoinPrompt.includes(generated.artifactDir), withdrawnJoinPrompt);
assert.deepEqual(verifyDynamicRuntimeMaterialization(controls).expandedGroupIds, ["fanout"]);

// The synchronizer keys a generated graph node by its storage ID and each generated attempt by
Expand Down Expand Up @@ -499,6 +517,7 @@ test("explicit source retry re-derives the base runtime controls after archiving
assert.equal(prunedState.graph_fingerprint, "a".repeat(64));
assert.equal(prunedState.config_fingerprint, "b".repeat(64));
const retryRecord = JSON.parse(fs.readFileSync(path.join(archived.archive_path, "retry.json"), "utf8")) as {
archived_attempt_paths?: string[];
pruned_state_node_ids?: string[];
};
assert.deepEqual(retryRecord.pruned_state_node_ids, generationStateIds);
Expand All @@ -519,6 +538,87 @@ test("explicit source retry re-derives the base runtime controls after archiving
verified.graph.nodes.map((node) => node.id),
["planner", "fanout", "join"]
);

// The retried source plans another item and the group expands again. The join's prompt named the
// withdrawn child, so the retry withdrew it with the generation and the join renders afresh from
// the new one. Left in place, it would not match that render, which then throws `runtime rendered
// prompt changed for join`, and so would every later render and admission check of the run.
fs.writeFileSync(sourceArtifactPath, `${JSON.stringify({ goals: [item(1)] })}\n`, "utf8");
const reexpanded = materializeDynamicRuntime({ ...controls, readyGroupIds: ["fanout"] });
const regenerated = reexpanded.tasks.find((task) => task.metadata.node.dynamic !== undefined);
assert.ok(regenerated);
assert.notEqual(regenerated.artifactDir, generated.artifactDir);
const joinPrompt = fs.readFileSync(joinPromptPath, "utf8");
assert.ok(joinPrompt.includes(regenerated.artifactDir), joinPrompt);
assert.ok(!joinPrompt.includes(generated.artifactDir), joinPrompt);
assert.deepEqual(verifyDynamicRuntimeMaterialization(controls).expandedGroupIds, ["fanout"]);
// The withdrawn prompt is kept in the archive, and the retry record lists it beside the
// generation's own paths and nothing else.
assert.equal(
fs.readFileSync(path.join(archived.archive_path, "artifacts", "join", "prompt.rendered.md"), "utf8"),
withdrawnJoinPrompt
);
const archiveRelative = path.relative(runRoot, archived.archive_path).split(path.sep).join("/");
assert.deepEqual(retryRecord.archived_attempt_paths, [
`${archiveRelative}/artifacts/${generated.attemptId}`,
`${archiveRelative}/artifacts/join/prompt.rendered.md`
]);
});

test("explicit source retry refuses a withdrawn prompt that is not a regular file before anything moves", () => {
// Each case leaves the join's published prompt, which the retry would withdraw, as something other
// than a regular file inside the run root. Planning runs before the Smithers reset, so refusing
// there keeps the generation published and moves nothing into the history directory.
const promptOf = (attemptDir: string): string => path.join(attemptDir, "prompt.rendered.md");
const cases: Array<{ name: string; code: string; corrupt: (attemptDir: string, outside: string) => void }> = [
{
name: "symlink",
code: "symlink-escape",
corrupt: (attemptDir, outside) => {
fs.writeFileSync(outside, "outside\n", "utf8");
fs.rmSync(promptOf(attemptDir));
fs.symlinkSync(outside, promptOf(attemptDir));
}
},
{
// Refused like a live one, not skipped as a prompt that was never rendered.
name: "dangling-symlink",
code: "symlink-escape",
corrupt: (attemptDir, outside) => {
fs.rmSync(promptOf(attemptDir));
fs.symlinkSync(outside, promptOf(attemptDir));
}
},
{
name: "directory",
code: "not-file",
corrupt: (attemptDir) => {
fs.rmSync(promptOf(attemptDir));
fs.mkdirSync(promptOf(attemptDir));
}
},
{
name: "symlinked-attempt-directory",
code: "symlink-escape",
corrupt: (attemptDir, outside) => {
fs.renameSync(attemptDir, outside);
fs.symlinkSync(outside, attemptDir);
}
}
];
for (const { name, code, corrupt } of cases) {
const { controls } = sealedDynamicRun(`retry-prompt-${name}`, [item(0)]);
const { projectRoot, runRoot } = controls;
materializeDynamicRuntime({ ...controls, readyGroupIds: ["fanout"] });
corrupt(path.join(runRoot, "artifacts", "join"), path.join(projectRoot, "outside"));
assert.throws(
() => planDynamicExpansionRetryArchive({ projectRoot, runRoot, sourceNodeIds: ["node:planner"] }),
(error: unknown) => error instanceof ArtifactPathError && error.code === code,
name
);
assert.deepEqual(fs.readdirSync(path.join(runRoot, "dynamic-expansions")), ["fanout.json"], name);
assert.equal(fs.existsSync(path.join(runRoot, "dynamic-expansion-history")), false, name);
}
});

test("a lock file left by a killed materializer blocks neither expansion nor a source retry", () => {
Expand Down
12 changes: 12 additions & 0 deletions packages/runtime/test/dynamic-lifecycle.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1360,6 +1360,17 @@ test("explicit source retry prunes the withdrawn generation from run state and k
};
assert.deepEqual(retryRecord.group_node_ids, ["fanout"]);
assert.deepEqual(retryRecord.pruned_state_node_ids, generationStateIds);
// The join's prompt waits on the group, so it was rendered from the withdrawn generation and is
// withdrawn with it.
const joinPromptPath = fixture.joinTask.renderedPromptPath;
assert.ok(joinPromptPath);
assert.equal(fs.existsSync(joinPromptPath), false);
assert.equal(
fs.existsSync(
path.join(historyRoot, archiveName, "artifacts", fixture.joinTask.attemptId, path.basename(joinPromptPath))
),
true
);
const pruned = readState(fixture);
for (const nodeId of generationStateIds) assert.equal(pruned.nodes[nodeId], undefined, nodeId);
for (const nodeId of [fixture.plannerTask.attemptId, "fanout", fixture.joinTask.attemptId]) {
Expand All @@ -1381,6 +1392,7 @@ test("explicit source retry prunes the withdrawn generation from run state and k
// shadow the new one anywhere, and every observer stays admitted.
const regenerated = republishRetriedExpansion(fixture);
assert.equal(regenerated.length, 2);
assert.equal(fs.existsSync(joinPromptPath), true);
const regeneratedStorageId = regenerated[0]?.metadata.node.storageId;
assert.ok(regeneratedStorageId);
assert.notEqual(regeneratedStorageId, fixture.storageId);
Expand Down
Loading