smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge main into refactor/agent-hub-fullscreen

Browse Source
This commit is contained in:
Kyle McCleary
2026-08-04 18:15:42 -07:00
parent 5b68e99796 b361bbde20
commit 5cc4f5c93a
193 changed files with 8225 additions and 5332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/coding-agent/src/task/agents.ts
+1 -1
View File
@@ -85,7 +85,7 @@ export class AgentParsingError extends Error {
this.name = "AgentParsingError";
}
toString(): string {
override toString(): string {
const details: string[] = [this.message];
if (this.source !== undefined) {
details.push(`Source: ${JSON.stringify(this.source)}`);
packages/coding-agent/src/task/executor.ts
+105 -29
View File
@@ -55,6 +55,7 @@ import { normalizeSchema } from "../tools/jtd-to-json-schema";
import { buildOutputValidator, summarizeValidationFailure } from "../tools/output-schema-validator";
import { ToolAbortError } from "../tools/tool-errors";
import type { EventBus } from "../utils/event-bus";
import { trackLateCleanup } from "../utils/late-cleanup";
import { buildNamedToolChoice } from "../utils/tool-choice";
import type { WorkspaceTree } from "../workspace-tree";
import { generateTaskLabel } from "./label";
@@ -81,6 +82,7 @@ import { arrayValuedLabels, assembleYieldResult } from "./yield-assembly";
export type { YieldItem } from "./types";
const MCP_CALL_TIMEOUT_MS = 60_000;
const TASK_ABORT_CLEANUP_GRACE_MS = 10_000;
/**
* Soft per-agent request budgets (assistant requests per run). Crossing the
@@ -168,9 +170,11 @@ function resolveSubagentRetryFallbackCandidates(
): SubagentRetryFallbackCandidate[] {
const candidates: SubagentRetryFallbackCandidate[] = [];
const seen = new Set<string>();
const disabledProviders = new Set(settings.get("disabledProviders"));
for (const pattern of modelPatterns) {
const resolved = resolveModelOverride([pattern], modelRegistry, settings);
if (!resolved.model) continue;
if (disabledProviders.has(resolved.model.provider)) continue;
const selector = resolved.explicitThinkingLevel
? formatModelSelectorValue(formatModelStringWithRouting(resolved.model), resolved.thinkingLevel)
: formatModelStringWithRouting(resolved.model);
@@ -181,7 +185,10 @@ function resolveSubagentRetryFallbackCandidates(
return candidates;
}
function resolveSubagentDefaultRetryFallbackChain(settings: Settings): string[] | undefined {
function resolveSubagentDefaultRetryFallbackChain(
settings: Settings,
modelRegistry: ModelRegistry,
): string[] | undefined {
const fallbackChain = settings.get("retry.fallbackChains")?.default;
if (
!Array.isArray(fallbackChain) ||
@@ -190,7 +197,11 @@ function resolveSubagentDefaultRetryFallbackChain(settings: Settings): string[]
) {
return undefined;
}
return fallbackChain;
const disabledProviders = new Set(settings.get("disabledProviders"));
return fallbackChain.filter(entry => {
const resolved = resolveModelOverride([entry], modelRegistry, settings);
return !resolved.model || !disabledProviders.has(resolved.model.provider);
});
}
function installSubagentRetryFallbackChain(args: {
@@ -459,6 +470,8 @@ export interface ExecutorOptions {
* set this false so disposal unregisters them instead of leaving idle peers.
*/
keepAlive?: boolean;
/** Internal ownership handoff for cleanup that outlives the visible Task result. */
onCleanupDeferred?: (completion: Promise<void>) => void;
}
function parseStringifiedJson(value: unknown): unknown {
@@ -1971,9 +1984,7 @@ async function driveSessionToYield(
// yield: the next iteration's ladder demands a fresh one.
}
if (monitor.yieldCalled()) {
await session.waitForIdle();
} else {
if (!monitor.yieldCalled()) {
await awaitAbortable(session.waitForIdle());
}
@@ -2149,14 +2160,17 @@ async function finalizeRunResult(args: FinalizeRunArgs): Promise<SingleResult> {
exitCode = 1;
}
const wasAborted =
runtimeLimitExceeded || abortedViaYield || (!hasYield && (done.aborted || signal?.aborted || false));
runtimeLimitExceeded || Boolean(done.aborted) || abortedViaYield || (!hasYield && Boolean(signal?.aborted));
const finalAbortReason = wasAborted
? runtimeLimitExceeded
? monitor.resolveAbortReasonText()
: abortedViaYield
? yieldAbortReason
: (done.abortReason ??
(signal?.aborted ? monitor.resolveSignalAbortReason() : monitor.resolveAbortReasonText()))
: done.aborted
? (done.abortReason ?? monitor.resolveAbortReasonText())
: abortedViaYield
? yieldAbortReason
: signal?.aborted
? monitor.resolveSignalAbortReason()
: monitor.resolveAbortReasonText()
: undefined;
progress.status = wasAborted ? "aborted" : exitCode === 0 ? "completed" : "failed";
monitor.scheduleProgress(true);
@@ -2360,15 +2374,27 @@ export async function finalizeSubagentLifecycle(args: {
isolated: boolean;
agentIdleTtlMs: number;
reviveSession: AgentReviver | null;
cleanupDeadlineAt?: number;
onCleanupDeferred?: (completion: Promise<void>) => void;
}): Promise<void> {
const registry = AgentRegistry.global();
const ref = registry.get(args.id);
const ownsRef = Boolean(ref && ref.session === args.session);
const cleanupDeadlineAt = args.cleanupDeadlineAt ?? Date.now() + 5000;
const disposeSession = async (): Promise<void> => {
const disposal = args.session.dispose();
const remainingMs = Math.max(0, cleanupDeadlineAt - Date.now());
try {
await untilAborted(AbortSignal.timeout(5000), () => args.session.dispose());
} catch {
// Ignore cleanup errors
await untilAborted(AbortSignal.timeout(remainingMs), () => disposal);
} catch (error) {
if (Date.now() >= cleanupDeadlineAt) {
args.onCleanupDeferred?.(disposal);
return;
}
logger.warn("Subagent session cleanup failed", {
id: args.id,
error: error instanceof Error ? error.message : String(error),
});
}
};
@@ -2794,7 +2820,7 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
const configuredModelPatterns = resolveConfiguredModelPatterns(modelPatterns, settings);
const defaultRetryFallbackChain =
configuredModelPatterns.length === 1
? resolveSubagentDefaultRetryFallbackChain(subagentSettings)
? resolveSubagentDefaultRetryFallbackChain(subagentSettings, modelRegistry)
: undefined;
const {
model,
@@ -2828,7 +2854,7 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
const retryFallbackRole = installSubagentRetryFallbackChain({
settings: subagentSettings,
id,
candidates: resolveSubagentRetryFallbackCandidates(modelPatterns, modelRegistry, settings),
candidates: resolveSubagentRetryFallbackCandidates(modelPatterns, modelRegistry, subagentSettings),
defaultFallbackChain: defaultRetryFallbackChain,
model,
authFallbackUsed,
@@ -3218,6 +3244,20 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
error = err instanceof Error ? err.stack || err.message : String(err);
}
} finally {
const cleanupDeadlineAt = Date.now() + TASK_ABORT_CLEANUP_GRACE_MS;
const cleanupChangeStatus =
worktree === undefined
? "This task was not isolated, so its changes may remain in the working directory."
: "No isolated changes were applied.";
const lateCleanups: Promise<void>[] = [];
let deferredSessionShutdown: Promise<void> | undefined;
const deferCleanup = (completion: Promise<void>): void => {
lateCleanups.push(completion);
exitCode = 1;
aborted = true;
abortReasonText = `cleanup exceeded ${TASK_ABORT_CLEANUP_GRACE_MS} ms`;
error ??= `Task aborted. Cleanup did not finish within ${TASK_ABORT_CLEANUP_GRACE_MS} ms. ${cleanupChangeStatus}`;
};
if (abortSignal.aborted) {
aborted = monitor.isAbortedRun();
if (aborted) {
@@ -3226,10 +3266,21 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
if (exitCode === 0) exitCode = 1;
}
sessionAbortController.abort();
const activeSessionAbort = monitor.waitForActiveSessionAbort();
try {
await untilAborted(AbortSignal.timeout(5000), () => monitor.waitForActiveSessionAbort());
} catch {
// Ignore abort cleanup timeouts/errors; terminal disposal below is still best-effort.
await untilAborted(
AbortSignal.timeout(Math.max(0, cleanupDeadlineAt - Date.now())),
() => activeSessionAbort,
);
} catch (cleanupError) {
if (Date.now() >= cleanupDeadlineAt) {
deferCleanup(activeSessionAbort);
} else {
logger.warn("Subagent abort cleanup failed", {
id,
error: cleanupError instanceof Error ? cleanupError.message : String(cleanupError),
});
}
}
if (unsubscribe) {
try {
@@ -3239,6 +3290,17 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
}
unsubscribe = null;
}
const jobManager = AsyncJobManager.instance();
if (jobManager) {
const reap = await jobManager.cancelAndReapOwnerJobs(id, cleanupDeadlineAt);
if (!reap.settled) {
deferCleanup(reap.completion);
logger.warn("Subagent async job cleanup exceeded its deadline", {
id,
pendingJobIds: reap.pendingJobIds,
});
}
}
const session = monitor.takeActiveSession();
if (session) {
monitor.captureSalvage(session);
@@ -3254,22 +3316,36 @@ export async function runSubprocess(options: ExecutorOptions): Promise<SingleRes
isolated: worktree !== undefined,
agentIdleTtlMs,
reviveSession,
cleanupDeadlineAt,
onCleanupDeferred: completion => {
deferredSessionShutdown = completion;
deferCleanup(completion);
},
});
}
// Structured-concurrency reap: cancel and await ALL surviving owner
// jobs (abort paths; suppressed/watched jobs the model left behind)
// so isolation capture/cleanup never races a live process writing
// into the worktree. This never proceeds while an owner process is
// live: cancellation SIGKILL-escalates, so settlement is expected
// within one interval — an unkillable process blocks here visibly
// (with periodic warnings) instead of silently racing teardown.
const jobManager = AsyncJobManager.instance();
if (jobManager) {
jobManager.cancelAll({ ownerId: id });
while (!(await jobManager.waitForOwnerJobs(id, { timeoutMs: 10_000 }))) {
logger.warn("Subagent async jobs still settling; delaying teardown until process exit", { id });
if (deferredSessionShutdown) {
const finalReap = Promise.allSettled([deferredSessionShutdown]).then(async () => {
const reap = await jobManager.cancelAndReapOwnerJobs(id, Date.now());
await reap.completion;
});
lateCleanups.push(finalReap);
} else {
const reap = await jobManager.cancelAndReapOwnerJobs(id, cleanupDeadlineAt);
if (!reap.settled) {
deferCleanup(reap.completion);
logger.warn("Subagent async job cleanup exceeded its deadline after session shutdown", {
id,
pendingJobIds: reap.pendingJobIds,
});
}
}
}
if (lateCleanups.length > 0) {
const completion = Promise.allSettled(lateCleanups).then(() => {});
trackLateCleanup(completion, { id, resource: "subagent" });
options.onCleanupDeferred?.(completion);
}
}
// Launch-latency breakdown (subagent invocation → first chat dispatch).
packages/coding-agent/src/task/isolation-runner.ts
+19 -1
View File
@@ -24,6 +24,7 @@ import { AgentRegistry } from "../registry/agent-registry";
import type { ToolSession } from "../tools";
import { generateCommitMessage } from "../utils/commit-message-generator";
import * as git from "../utils/git";
import { trackLateCleanup } from "../utils/late-cleanup";
import type { ExecutorOptions } from "./executor";
import { runSubprocess } from "./executor";
import type { SingleResult } from "./types";
@@ -156,6 +157,7 @@ async function writeIsolationPatch(
*/
export async function runIsolatedSubprocess(opts: IsolatedRunOptions): Promise<SingleResult> {
let handle: IsolationHandle | undefined;
let deferredCleanup: Promise<void> | undefined;
try {
const taskBaseline = structuredClone(opts.context.baseline);
handle = await ensureIsolation(opts.context.repoRoot, opts.agentId, opts.preferredBackend);
@@ -165,7 +167,12 @@ export async function runIsolatedSubprocess(opts: IsolatedRunOptions): Promise<S
worktree: isolationDir,
preloadedExtensionPaths: undefined,
preloadedCustomToolPaths: undefined,
onCleanupDeferred: completion => {
deferredCleanup = completion;
opts.baseOptions.onCleanupDeferred?.(completion);
},
});
if (deferredCleanup) return result;
if (opts.mergeMode === "branch" && result.exitCode === 0) {
try {
const commitResult = await commitToBranch(
@@ -226,7 +233,18 @@ export async function runIsolatedSubprocess(opts: IsolatedRunOptions): Promise<S
return rememberAgentArtifacts(opts.buildFailureResult(err));
} finally {
if (handle) {
await cleanupIsolation(handle);
const isolationHandle = handle;
if (deferredCleanup) {
trackLateCleanup(
deferredCleanup.then(() => cleanupIsolation(isolationHandle)),
{
agentId: opts.agentId,
resource: "isolation",
},
);
} else {
await cleanupIsolation(isolationHandle);
}
}
}
}
packages/coding-agent/src/task/structured-subagent.ts
+17 -2
View File
@@ -20,6 +20,7 @@ import type { TaskEffort } from "../thinking";
import type { ToolSession } from "../tools";
import { isIrcEnabled } from "../tools/hub";
import { buildOutputValidator } from "../tools/output-schema-validator";
import { trackLateCleanup } from "../utils/late-cleanup";
import { type DiscoveryResult, discoverAgents, getAgent } from "./discovery";
import { type ExecutorOptions, runSubprocess } from "./executor";
import {
@@ -550,12 +551,16 @@ export async function runStructuredSubagent(request: StructuredSubagentRequest):
let mergeSummary = "";
let requiresRecoveryArtifacts = false;
let completedSuccessfully = false;
let deferredCleanup: Promise<void> | undefined;
try {
const id = await reserveStructuredSubagentId(request.session, {
...request.identity,
label: request.identity?.label ?? (request.invocationKind === "eval" ? "EvalAgent" : undefined),
});
const baseOptions = buildExecutorOptions(request, policy, lease, id);
baseOptions.onCleanupDeferred = completion => {
deferredCleanup = completion;
};
baseOptions.planReference = await loadPlanReference(request, policy);
let isolationContext: IsolationContext | null = null;
if (policy.isIsolated) {
@@ -649,8 +654,18 @@ export async function runStructuredSubagent(request: StructuredSubagentRequest):
(policy.isIsolated && (!policy.applyChanges || changesApplied === false || requiresRecoveryArtifacts));
const shouldCleanup = lease.temporary && !shouldRetainArtifacts;
if (shouldCleanup) {
await fs.rm(lease.artifactsDir, { recursive: true, force: true });
lease.unregister?.();
const cleanupArtifacts = async (): Promise<void> => {
await fs.rm(lease.artifactsDir, { recursive: true, force: true });
lease.unregister?.();
};
if (deferredCleanup) {
trackLateCleanup(deferredCleanup.then(cleanupArtifacts), {
resource: "artifacts",
artifactsDir: lease.artifactsDir,
});
} else {
await cleanupArtifacts();
}
}
}
}