fix(coding-agent): serialized isolated-task merges and fixed async batch accounting
stash/cherry-pick merge sequence runs under the repo lock (lost-uncommitted-changes race); stash-pop failure no longer mislabels merged branches; async batches cannot stick at running forever; queued tasks stop counting against the global job cap; aborted session startup disposes the late session; fail-fast propagates the worker signal; command expansion treats user input dollar-patterns literally; progress snapshots stop structured-cloning tool payloads.
This commit is contained in:
@@ -120,7 +120,8 @@ export function getCommand(commands: WorkflowCommand[], name: string): WorkflowC
|
||||
* Replaces $@ with the provided input.
|
||||
*/
|
||||
export function expandCommand(command: WorkflowCommand, input: string): string {
|
||||
return command.instructions.replace(/\$@/g, input);
|
||||
// Function replacement so `$`-patterns in user input ($$, $&, ...) stay literal.
|
||||
return command.instructions.replace(/\$@/g, () => input);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1285,59 +1285,67 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
|
||||
|
||||
const { normalized: normalizedOutputSchema } = normalizeSchema(outputSchema);
|
||||
|
||||
const { session } = await awaitAbortable(
|
||||
createAgentSession({
|
||||
cwd: worktree ?? cwd,
|
||||
authStorage,
|
||||
modelRegistry,
|
||||
settings: subagentSettings,
|
||||
model,
|
||||
thinkingLevel: effectiveThinkingLevel,
|
||||
toolNames,
|
||||
outputSchema,
|
||||
requireYieldTool: true,
|
||||
contextFiles: options.contextFiles,
|
||||
skills: options.skills,
|
||||
promptTemplates: options.promptTemplates,
|
||||
workspaceTree: options.workspaceTree,
|
||||
rules: options.rules,
|
||||
preloadedExtensionPaths: options.preloadedExtensionPaths,
|
||||
preloadedCustomToolPaths: options.preloadedCustomToolPaths,
|
||||
systemPrompt: defaultPrompt => {
|
||||
const subagentPrompt = prompt.render(subagentSystemPromptTemplate, {
|
||||
agent: agent.systemPrompt,
|
||||
context: options.context?.trim() ?? "",
|
||||
planReference: options.planReference?.content ?? "",
|
||||
planReferencePath: options.planReference?.path ?? "",
|
||||
worktree: worktree ?? "",
|
||||
outputSchema: normalizedOutputSchema,
|
||||
contextFile: contextFileForPrompt,
|
||||
ircPeers: ircEnabled ? renderIrcPeerRoster(id) : "",
|
||||
ircSelfId: ircEnabled ? id : "",
|
||||
});
|
||||
return defaultPrompt.length === 0
|
||||
? [subagentPrompt]
|
||||
: [...defaultPrompt.slice(0, -1), subagentPrompt, defaultPrompt[defaultPrompt.length - 1]];
|
||||
},
|
||||
sessionManager,
|
||||
hasUI: false,
|
||||
spawns: spawnsEnv,
|
||||
taskDepth: childDepth,
|
||||
parentHindsightSessionState: options.parentHindsightSessionState,
|
||||
parentMnemopiSessionState: options.parentMnemopiSessionState,
|
||||
parentTaskPrefix: id,
|
||||
agentId: id,
|
||||
agentDisplayName: agent.name,
|
||||
enableLsp: lspEnabled,
|
||||
skipPythonPreflight,
|
||||
enableMCP,
|
||||
mcpManager: options.mcpManager,
|
||||
customTools: mcpProxyTools.length > 0 ? mcpProxyTools : undefined,
|
||||
localProtocolOptions: options.localProtocolOptions,
|
||||
telemetry: subagentTelemetry,
|
||||
parentEvalSessionId: options.parentEvalSessionId,
|
||||
}),
|
||||
);
|
||||
const sessionPromise = createAgentSession({
|
||||
cwd: worktree ?? cwd,
|
||||
authStorage,
|
||||
modelRegistry,
|
||||
settings: subagentSettings,
|
||||
model,
|
||||
thinkingLevel: effectiveThinkingLevel,
|
||||
toolNames,
|
||||
outputSchema,
|
||||
requireYieldTool: true,
|
||||
contextFiles: options.contextFiles,
|
||||
skills: options.skills,
|
||||
promptTemplates: options.promptTemplates,
|
||||
workspaceTree: options.workspaceTree,
|
||||
rules: options.rules,
|
||||
preloadedExtensionPaths: options.preloadedExtensionPaths,
|
||||
preloadedCustomToolPaths: options.preloadedCustomToolPaths,
|
||||
systemPrompt: defaultPrompt => {
|
||||
const subagentPrompt = prompt.render(subagentSystemPromptTemplate, {
|
||||
agent: agent.systemPrompt,
|
||||
context: options.context?.trim() ?? "",
|
||||
planReference: options.planReference?.content ?? "",
|
||||
planReferencePath: options.planReference?.path ?? "",
|
||||
worktree: worktree ?? "",
|
||||
outputSchema: normalizedOutputSchema,
|
||||
contextFile: contextFileForPrompt,
|
||||
ircPeers: ircEnabled ? renderIrcPeerRoster(id) : "",
|
||||
ircSelfId: ircEnabled ? id : "",
|
||||
});
|
||||
return defaultPrompt.length === 0
|
||||
? [subagentPrompt]
|
||||
: [...defaultPrompt.slice(0, -1), subagentPrompt, defaultPrompt[defaultPrompt.length - 1]];
|
||||
},
|
||||
sessionManager,
|
||||
hasUI: false,
|
||||
spawns: spawnsEnv,
|
||||
taskDepth: childDepth,
|
||||
parentHindsightSessionState: options.parentHindsightSessionState,
|
||||
parentMnemopiSessionState: options.parentMnemopiSessionState,
|
||||
parentTaskPrefix: id,
|
||||
agentId: id,
|
||||
agentDisplayName: agent.name,
|
||||
enableLsp: lspEnabled,
|
||||
skipPythonPreflight,
|
||||
enableMCP,
|
||||
mcpManager: options.mcpManager,
|
||||
customTools: mcpProxyTools.length > 0 ? mcpProxyTools : undefined,
|
||||
localProtocolOptions: options.localProtocolOptions,
|
||||
telemetry: subagentTelemetry,
|
||||
parentEvalSessionId: options.parentEvalSessionId,
|
||||
});
|
||||
let session: AgentSession;
|
||||
try {
|
||||
({ session } = await awaitAbortable(sessionPromise));
|
||||
} catch (err) {
|
||||
// Abort raced session startup. The session may still resolve later
|
||||
// holding live LSP/MCP child processes — dispose it when it does so
|
||||
// a cancelled subagent cannot leak them.
|
||||
void sessionPromise.then(created => created.session.dispose()).catch(() => {});
|
||||
throw err;
|
||||
}
|
||||
|
||||
activeSession = session;
|
||||
|
||||
|
||||
@@ -242,6 +242,57 @@ function validateTaskModeParams(simpleMode: TaskSimpleMode, params: TaskParams):
|
||||
return "task.simple is set to independent, so the task tool does not accept `context` or `schema`. Put all required background and output expectations inside each task assignment or the selected agent definition.";
|
||||
}
|
||||
|
||||
/** Sentinel for async jobs whose subagent finished with a failing result; batch counters are already updated. */
|
||||
class TaskJobError extends Error {}
|
||||
|
||||
/**
|
||||
* Validate task ids: every task needs a non-empty id and ids must be unique
|
||||
* (case-insensitive). Returns a problem description, or undefined when valid.
|
||||
*/
|
||||
function validateTaskIds(tasks: TaskParams["tasks"]): string | undefined {
|
||||
const missingTaskIndexes: number[] = [];
|
||||
const idIndexes = new Map<string, number[]>();
|
||||
|
||||
for (let i = 0; i < tasks.length; i++) {
|
||||
const id = tasks[i]?.id;
|
||||
if (typeof id !== "string" || id.trim() === "") {
|
||||
missingTaskIndexes.push(i);
|
||||
continue;
|
||||
}
|
||||
const normalizedId = id.toLowerCase();
|
||||
const indexes = idIndexes.get(normalizedId);
|
||||
if (indexes) {
|
||||
indexes.push(i);
|
||||
} else {
|
||||
idIndexes.set(normalizedId, [i]);
|
||||
}
|
||||
}
|
||||
|
||||
const duplicateIds: Array<{ id: string; indexes: number[] }> = [];
|
||||
for (const [normalizedId, indexes] of idIndexes.entries()) {
|
||||
if (indexes.length > 1) {
|
||||
duplicateIds.push({
|
||||
id: tasks[indexes[0]]?.id ?? normalizedId,
|
||||
indexes,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (missingTaskIndexes.length === 0 && duplicateIds.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const problems: string[] = [];
|
||||
if (missingTaskIndexes.length > 0) {
|
||||
problems.push(`Missing task ids at indexes: ${missingTaskIndexes.join(", ")}`);
|
||||
}
|
||||
if (duplicateIds.length > 0) {
|
||||
const details = duplicateIds.map(entry => `${entry.id} (indexes ${entry.indexes.join(", ")})`).join("; ");
|
||||
problems.push(`Duplicate task ids detected (case-insensitive): ${details}`);
|
||||
}
|
||||
return `Invalid tasks: ${problems.join(". ")}`;
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// Tool Class
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
@@ -363,6 +414,11 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
return this.#executeSync(_toolCallId, params, signal, onUpdate);
|
||||
}
|
||||
|
||||
const taskIdProblem = validateTaskIds(taskItems);
|
||||
if (taskIdProblem) {
|
||||
return createTaskModeError(taskIdProblem);
|
||||
}
|
||||
|
||||
const outputManager =
|
||||
this.session.agentOutputManager ?? new AgentOutputManager(this.session.getArtifactsDir ?? (() => null));
|
||||
const uniqueIds = await outputManager.allocateBatch(taskItems.map(t => t.id));
|
||||
@@ -396,9 +452,13 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
let failedJobs = 0;
|
||||
|
||||
const getProgressSnapshot = (): AgentProgress[] => {
|
||||
// Shallow copies: top-level fields are reassigned (never mutated in
|
||||
// place) and the large nested payloads (extractedToolData) are
|
||||
// immutable once attached — structuredClone here cost O(batch × payload)
|
||||
// per progress event.
|
||||
return Array.from(progressByTaskId.values())
|
||||
.sort((a, b) => a.index - b.index)
|
||||
.map(progress => structuredClone(progress));
|
||||
.map(progress => ({ ...progress }));
|
||||
};
|
||||
|
||||
const buildAsyncDetails = (state: "running" | "completed" | "failed", jobId: string): TaskToolDetails => ({
|
||||
@@ -424,6 +484,7 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
const taskItem = taskItems[i];
|
||||
if (signal?.aborted) {
|
||||
failedSchedules.push(`${taskItem.id}: cancelled before scheduling`);
|
||||
completedJobs += 1;
|
||||
const progress = progressByTaskId.get(taskItem.id);
|
||||
if (progress) {
|
||||
progress.status = "aborted";
|
||||
@@ -438,7 +499,7 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
const jobId = manager.register(
|
||||
"task",
|
||||
label,
|
||||
async ({ signal: runSignal, reportProgress }) => {
|
||||
async ({ signal: runSignal, reportProgress, markRunning }) => {
|
||||
const startedAt = Date.now();
|
||||
const progress = progressByTaskId.get(taskItem.id);
|
||||
await semaphore.acquire();
|
||||
@@ -447,8 +508,11 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
if (progress) {
|
||||
progress.status = "aborted";
|
||||
}
|
||||
completedJobs += 1;
|
||||
failedJobs += 1;
|
||||
throw new Error("Aborted before execution");
|
||||
}
|
||||
markRunning();
|
||||
if (progress) {
|
||||
progress.status = "running";
|
||||
}
|
||||
@@ -462,12 +526,12 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
]);
|
||||
const finalText = result.content.find(part => part.type === "text")?.text ?? "(no output)";
|
||||
const singleResult = result.details?.results[0];
|
||||
// A missing per-task result means #executeSync failed at the
|
||||
// tool level (results: []) — treat it as a failure, not success.
|
||||
const resultFailed =
|
||||
!singleResult || (singleResult.aborted ?? false) || singleResult.exitCode !== 0;
|
||||
if (progress) {
|
||||
progress.status = singleResult?.aborted
|
||||
? "aborted"
|
||||
: (singleResult?.exitCode ?? 0) === 0
|
||||
? "completed"
|
||||
: "failed";
|
||||
progress.status = singleResult?.aborted ? "aborted" : resultFailed ? "failed" : "completed";
|
||||
progress.durationMs = singleResult?.durationMs ?? Math.max(0, Date.now() - startedAt);
|
||||
progress.tokens = singleResult?.tokens ?? 0;
|
||||
progress.contextTokens = singleResult?.contextTokens;
|
||||
@@ -478,7 +542,7 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
progress.retryState = undefined;
|
||||
}
|
||||
completedJobs += 1;
|
||||
if (singleResult && ((singleResult.aborted ?? false) || singleResult.exitCode !== 0)) {
|
||||
if (resultFailed) {
|
||||
failedJobs += 1;
|
||||
}
|
||||
const remaining = taskItems.length - completedJobs;
|
||||
@@ -498,8 +562,15 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
`Background task batch complete: ${completedJobs}/${taskItems.length} finished.`,
|
||||
);
|
||||
}
|
||||
if (resultFailed) {
|
||||
// Mark the job itself failed; counters above are already updated.
|
||||
throw new TaskJobError(finalText);
|
||||
}
|
||||
return finalText;
|
||||
} catch (error) {
|
||||
if (error instanceof TaskJobError) {
|
||||
throw error;
|
||||
}
|
||||
if (progress) {
|
||||
progress.status = "failed";
|
||||
progress.durationMs = Math.max(0, Date.now() - startedAt);
|
||||
@@ -530,6 +601,7 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
},
|
||||
{
|
||||
id: label,
|
||||
queued: true,
|
||||
ownerId: this.session.getAgentId?.() ?? undefined,
|
||||
onProgress: (text, details) => {
|
||||
const progressDetails =
|
||||
@@ -543,6 +615,7 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
failedSchedules.push(`${taskItem.id}: ${message}`);
|
||||
completedJobs += 1;
|
||||
const progress = progressByTaskId.get(taskItem.id);
|
||||
if (progress) {
|
||||
progress.status = "failed";
|
||||
@@ -734,45 +807,10 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
}
|
||||
|
||||
const tasks = params.tasks;
|
||||
const missingTaskIndexes: number[] = [];
|
||||
const idIndexes = new Map<string, number[]>();
|
||||
|
||||
for (let i = 0; i < tasks.length; i++) {
|
||||
const id = tasks[i]?.id;
|
||||
if (typeof id !== "string" || id.trim() === "") {
|
||||
missingTaskIndexes.push(i);
|
||||
continue;
|
||||
}
|
||||
const normalizedId = id.toLowerCase();
|
||||
const indexes = idIndexes.get(normalizedId);
|
||||
if (indexes) {
|
||||
indexes.push(i);
|
||||
} else {
|
||||
idIndexes.set(normalizedId, [i]);
|
||||
}
|
||||
}
|
||||
|
||||
const duplicateIds: Array<{ id: string; indexes: number[] }> = [];
|
||||
for (const [normalizedId, indexes] of idIndexes.entries()) {
|
||||
if (indexes.length > 1) {
|
||||
duplicateIds.push({
|
||||
id: tasks[indexes[0]]?.id ?? normalizedId,
|
||||
indexes,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (missingTaskIndexes.length > 0 || duplicateIds.length > 0) {
|
||||
const problems: string[] = [];
|
||||
if (missingTaskIndexes.length > 0) {
|
||||
problems.push(`Missing task ids at indexes: ${missingTaskIndexes.join(", ")}`);
|
||||
}
|
||||
if (duplicateIds.length > 0) {
|
||||
const details = duplicateIds.map(entry => `${entry.id} (indexes ${entry.indexes.join(", ")})`).join("; ");
|
||||
problems.push(`Duplicate task ids detected (case-insensitive): ${details}`);
|
||||
}
|
||||
const taskIdProblem = validateTaskIds(tasks);
|
||||
if (taskIdProblem) {
|
||||
return {
|
||||
content: [{ type: "text", text: `Invalid tasks: ${problems.join(". ")}` }],
|
||||
content: [{ type: "text", text: taskIdProblem }],
|
||||
details: {
|
||||
projectAgentsDir,
|
||||
results: [],
|
||||
@@ -951,7 +989,11 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
}
|
||||
emitProgress();
|
||||
|
||||
const runTask = async (task: (typeof tasksWithUniqueIds)[number], index: number) => {
|
||||
const runTask = async (
|
||||
task: (typeof tasksWithUniqueIds)[number],
|
||||
index: number,
|
||||
workerSignal?: AbortSignal,
|
||||
) => {
|
||||
if (!isIsolated) {
|
||||
return runSubprocess({
|
||||
cwd: this.session.cwd,
|
||||
@@ -973,12 +1015,13 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
artifactsDir: effectiveArtifactsDir,
|
||||
contextFile: contextFilePath,
|
||||
enableLsp: subagentLspEnabled,
|
||||
signal,
|
||||
signal: workerSignal ?? signal,
|
||||
eventBus: this.session.eventBus,
|
||||
onProgress: progress => {
|
||||
progressMap.set(index, {
|
||||
...structuredClone(progress),
|
||||
});
|
||||
// Shallow snapshot; recentTools is mutated in place by the
|
||||
// executor, the rest is reassigned or immutable. A deep clone
|
||||
// here cost O(extractedToolData) per progress event.
|
||||
progressMap.set(index, { ...progress, recentTools: progress.recentTools.slice() });
|
||||
emitProgress();
|
||||
},
|
||||
authStorage: this.session.authStorage,
|
||||
@@ -1034,12 +1077,10 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
artifactsDir: effectiveArtifactsDir,
|
||||
contextFile: contextFilePath,
|
||||
enableLsp: subagentLspEnabled,
|
||||
signal,
|
||||
signal: workerSignal ?? signal,
|
||||
eventBus: this.session.eventBus,
|
||||
onProgress: progress => {
|
||||
progressMap.set(index, {
|
||||
...structuredClone(progress),
|
||||
});
|
||||
progressMap.set(index, { ...progress, recentTools: progress.recentTools.slice() });
|
||||
emitProgress();
|
||||
},
|
||||
authStorage: this.session.authStorage,
|
||||
@@ -1226,6 +1267,9 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
const conflictPart = mergeResult.conflict ? `\nConflict: ${mergeResult.conflict}` : "";
|
||||
mergeSummary = `\n\n<system-notification>Branch merge failed. ${mergedPart}${failedPart}${conflictPart}\nUnmerged branches remain for manual resolution.</system-notification>`;
|
||||
}
|
||||
if (mergeResult.stashConflict) {
|
||||
mergeSummary += `\n\n<system-notification>${mergeResult.stashConflict}</system-notification>`;
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up merged branches (keep failed ones for manual resolution)
|
||||
@@ -1234,9 +1278,11 @@ export class TaskTool implements AgentTool<TaskToolSchemaInstance, TaskToolDetai
|
||||
await cleanupTaskBranches(repoRoot, allBranches);
|
||||
}
|
||||
} else {
|
||||
// Patch mode: combine and apply patches
|
||||
const patchesInOrder = results.map(result => result.patchPath).filter(Boolean) as string[];
|
||||
const missingPatch = results.some(result => !result.patchPath);
|
||||
// Patch mode: apply patches from successful tasks. Failed or
|
||||
// aborted siblings must not block completed work from landing.
|
||||
const successfulResults = results.filter(r => r.exitCode === 0 && !r.error && !r.aborted);
|
||||
const patchesInOrder = successfulResults.map(result => result.patchPath).filter(Boolean) as string[];
|
||||
const missingPatch = successfulResults.some(result => !result.patchPath);
|
||||
if (missingPatch) {
|
||||
changesApplied = false;
|
||||
hadAnyChanges = false;
|
||||
|
||||
@@ -20,13 +20,13 @@ export interface ParallelResult<R> {
|
||||
*
|
||||
* @param items - Items to process
|
||||
* @param concurrency - Maximum concurrent operations
|
||||
* @param fn - Async function to execute for each item
|
||||
* @param fn - Async function to execute for each item; receives a worker signal that fires on abort or fail-fast so in-flight siblings can cancel
|
||||
* @param signal - Optional abort signal to stop scheduling new work
|
||||
*/
|
||||
export async function mapWithConcurrencyLimit<T, R>(
|
||||
items: T[],
|
||||
concurrency: number,
|
||||
fn: (item: T, index: number) => Promise<R>,
|
||||
fn: (item: T, index: number, signal: AbortSignal) => Promise<R>,
|
||||
signal?: AbortSignal,
|
||||
): Promise<ParallelResult<R>> {
|
||||
const normalizedConcurrency = Number.isFinite(concurrency) ? Math.floor(concurrency) : items.length;
|
||||
@@ -52,7 +52,7 @@ export async function mapWithConcurrencyLimit<T, R>(
|
||||
const index = nextIndex++;
|
||||
if (index >= items.length) return;
|
||||
try {
|
||||
results[index] = await fn(items[index], index);
|
||||
results[index] = await fn(items[index], index, workerSignal);
|
||||
} catch (error) {
|
||||
// On abort, the fn itself handles it and returns a result
|
||||
// Only propagate non-abort errors
|
||||
|
||||
@@ -5,6 +5,7 @@ import * as path from "node:path";
|
||||
import * as natives from "@oh-my-pi/pi-natives";
|
||||
import { getWorktreeDir, hashPath, logger, Snowflake } from "@oh-my-pi/pi-utils";
|
||||
import * as git from "../utils/git";
|
||||
import { mapWithConcurrencyLimit } from "./parallel";
|
||||
|
||||
const { IsoBackendKind } = natives;
|
||||
type IsoBackendKind = natives.IsoBackendKind;
|
||||
@@ -82,16 +83,16 @@ async function discoverNestedRepos(repoRoot: string): Promise<string[]> {
|
||||
async function captureUntrackedPatch(repoRoot: string, untracked: readonly string[]): Promise<string> {
|
||||
if (untracked.length === 0) return "";
|
||||
const nullPath = getGitNoIndexNullPath();
|
||||
const untrackedDiffs = await Promise.all(
|
||||
untracked.map(entry =>
|
||||
git.diff(repoRoot, {
|
||||
allowFailure: true,
|
||||
binary: true,
|
||||
noIndex: { left: nullPath, right: entry },
|
||||
}),
|
||||
),
|
||||
// Bound concurrent git spawns; large untracked sets would otherwise fork one
|
||||
// process per file at once.
|
||||
const { results: untrackedDiffs } = await mapWithConcurrencyLimit([...untracked], 8, entry =>
|
||||
git.diff(repoRoot, {
|
||||
allowFailure: true,
|
||||
binary: true,
|
||||
noIndex: { left: nullPath, right: entry },
|
||||
}),
|
||||
);
|
||||
return untrackedDiffs.filter(diff => diff.trim()).join("\n");
|
||||
return untrackedDiffs.filter((diff): diff is string => !!diff?.trim()).join("\n");
|
||||
}
|
||||
|
||||
async function captureRepoBaseline(repoRoot: string): Promise<RepoBaseline> {
|
||||
@@ -427,6 +428,8 @@ export interface MergeBranchResult {
|
||||
merged: string[];
|
||||
failed: string[];
|
||||
conflict?: string;
|
||||
/** Set when cherry-picks landed on HEAD but restoring the stashed working tree failed. */
|
||||
stashConflict?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -438,64 +441,69 @@ export async function mergeTaskBranches(
|
||||
repoRoot: string,
|
||||
branches: Array<{ branchName: string; taskId: string; description?: string }>,
|
||||
): Promise<MergeBranchResult> {
|
||||
const merged: string[] = [];
|
||||
const failed: string[] = [];
|
||||
// Serialize against other in-process git mutations on this repo: concurrent
|
||||
// background merges interleaving stash push/pop + cherry-pick would corrupt
|
||||
// the working tree (lost uncommitted changes, mixed-up stash entries).
|
||||
return git.withRepoLock(repoRoot, async () => {
|
||||
const merged: string[] = [];
|
||||
const failed: string[] = [];
|
||||
|
||||
// Stash dirty working tree so cherry-pick can operate on a clean HEAD.
|
||||
// Without this, cherry-pick refuses to run when uncommitted changes exist.
|
||||
const didStash = await git.stash.push(repoRoot, "omp-task-merge");
|
||||
// Stash dirty working tree so cherry-pick can operate on a clean HEAD.
|
||||
// Without this, cherry-pick refuses to run when uncommitted changes exist.
|
||||
const didStash = await git.stash.push(repoRoot, "omp-task-merge");
|
||||
|
||||
let conflictResult: MergeBranchResult | undefined;
|
||||
let conflictResult: MergeBranchResult | undefined;
|
||||
|
||||
try {
|
||||
for (const { branchName } of branches) {
|
||||
try {
|
||||
await git.cherryPick(repoRoot, branchName);
|
||||
} catch (err) {
|
||||
try {
|
||||
for (const { branchName } of branches) {
|
||||
try {
|
||||
await git.cherryPick.abort(repoRoot);
|
||||
} catch {
|
||||
/* no state to abort */
|
||||
}
|
||||
const stderr =
|
||||
err instanceof git.GitCommandError
|
||||
? err.result.stderr.trim()
|
||||
: err instanceof Error
|
||||
? err.message
|
||||
: String(err);
|
||||
failed.push(branchName);
|
||||
conflictResult = {
|
||||
merged,
|
||||
failed: [...failed, ...branches.slice(merged.length + failed.length).map(b => b.branchName)],
|
||||
conflict: `${branchName}: ${stderr}`,
|
||||
};
|
||||
break;
|
||||
}
|
||||
|
||||
merged.push(branchName);
|
||||
}
|
||||
} finally {
|
||||
if (didStash) {
|
||||
try {
|
||||
await git.stash.pop(repoRoot, { index: true });
|
||||
} catch {
|
||||
// Stash-pop conflicts mean the replayed changes clash with the user's
|
||||
// uncommitted edits. Treat this as a merge failure so the caller preserves
|
||||
// recovery branches instead of reporting success and deleting them.
|
||||
logger.warn("Failed to restore stashed changes after task merge; stash entry preserved");
|
||||
if (!conflictResult) {
|
||||
await git.cherryPick(repoRoot, branchName);
|
||||
} catch (err) {
|
||||
try {
|
||||
await git.cherryPick.abort(repoRoot);
|
||||
} catch {
|
||||
/* no state to abort */
|
||||
}
|
||||
const stderr =
|
||||
err instanceof git.GitCommandError
|
||||
? err.result.stderr.trim()
|
||||
: err instanceof Error
|
||||
? err.message
|
||||
: String(err);
|
||||
failed.push(branchName);
|
||||
conflictResult = {
|
||||
merged,
|
||||
failed: merged,
|
||||
conflict:
|
||||
"stash pop: cherry-picked changes conflict with uncommitted edits. Run `git stash pop` and resolve manually.",
|
||||
failed: [...failed, ...branches.slice(merged.length + failed.length).map(b => b.branchName)],
|
||||
conflict: `${branchName}: ${stderr}`,
|
||||
};
|
||||
break;
|
||||
}
|
||||
|
||||
merged.push(branchName);
|
||||
}
|
||||
} finally {
|
||||
if (didStash) {
|
||||
try {
|
||||
await git.stash.pop(repoRoot, { index: true });
|
||||
} catch {
|
||||
// Stash-pop conflicts mean the replayed changes clash with the user's
|
||||
// uncommitted edits. The cherry-picked commits are already on HEAD, so
|
||||
// the merged branches DID land — report them as merged and surface the
|
||||
// stash conflict separately instead of claiming they are unmerged.
|
||||
logger.warn("Failed to restore stashed changes after task merge; stash entry preserved");
|
||||
const stashConflict =
|
||||
"stash pop: cherry-picked changes conflict with uncommitted edits. The merged commits are on HEAD; run `git stash pop` and resolve manually.";
|
||||
if (conflictResult) {
|
||||
conflictResult.stashConflict = stashConflict;
|
||||
} else {
|
||||
conflictResult = { merged, failed: [], stashConflict };
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return conflictResult ?? { merged, failed };
|
||||
return conflictResult ?? { merged, failed };
|
||||
});
|
||||
}
|
||||
|
||||
/** Clean up temporary task branches. */
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { expandCommand, type WorkflowCommand } from "@oh-my-pi/pi-coding-agent/task/commands";
|
||||
|
||||
function makeCommand(instructions: string): WorkflowCommand {
|
||||
return { name: "test", description: "test", instructions, source: "project", filePath: "test.md" };
|
||||
}
|
||||
|
||||
describe("expandCommand", () => {
|
||||
it("substitutes $@ with the input", () => {
|
||||
expect(expandCommand(makeCommand("Do: $@ and again $@"), "fix the bug")).toBe(
|
||||
"Do: fix the bug and again fix the bug",
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps $-patterns in user input literal", () => {
|
||||
expect(expandCommand(makeCommand("Run $@"), "echo $$ $& $' $` $@")).toBe("Run echo $$ $& $' $` $@");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user