root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-1218): harden mission autopilot with self-healing recovery

Browse Source 
- Add mission self-healing project settings for stale activation thresholds, task retry budgets, and mission health-check intervals
- Extend MissionAutopilot with failed-task retry handling, blocked feature escalation, stale activating mission recovery, and periodic feature/task consistency reconciliation
- Wire autopilot failure and recovery flows through Scheduler and InProcessRuntime, including startup recovery for watched missions after crashes
- Add blocked feature status support in core mission types and CLI mission status labels
- Expand mission-autopilot coverage for retry flows, stale/startup recovery, health checks, and regression scenarios, and include a patch changeset for @gsxdsm/fusion
This commit is contained in:
gsxdsm
2026-04-08 10:01:06 -07:00
parent cb90e33853
commit bfc3c80841
8 changed files with 904 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

456
packages/engine/src/mission-autopilot.test.ts
View File

@@ -92,6 +92,7 @@ function createMockMissionStore(missions: Mission[] = []) {
listSlices: vi.fn(),
getFeatureByTaskId: vi.fn(),
listFeatures: vi.fn(),
updateFeatureStatus: vi.fn(),
getMissionWithHierarchy: vi.fn(),
on: vi.fn(),
off: vi.fn(),
@@ -103,6 +104,14 @@ function createMockTaskStore() {
return {
on: vi.fn(),
off: vi.fn(),
getSettings: vi.fn().mockResolvedValue({
missionStaleThresholdMs: 600_000,
missionMaxTaskRetries: 3,
missionHealthCheckIntervalMs: 300_000,
}),
getTask: vi.fn().mockResolvedValue({ id: "FN-001", column: "in-progress" }),
moveTask: vi.fn().mockResolvedValue({ id: "FN-001", column: "todo" }),
updateTask: vi.fn().mockResolvedValue({}),
};
}
@@ -356,6 +365,106 @@ describe("MissionAutopilot", () => {
});
});
describe("handleTaskFailure", () => {
function wireMissionTask(taskId = "FN-001") {
const feature = createMockFeature({ id: "F-001", taskId, sliceId: "SL-001", status: "in-progress" });
const slice = createMockSlice({ id: "SL-001", milestoneId: "MS-001" });
const milestone = createMockMilestone({ id: "MS-001", missionId: "M-TEST1" });
missionStore.getFeatureByTaskId.mockReturnValue(feature);
missionStore.getSlice.mockReturnValue(slice);
missionStore.getMilestone.mockReturnValue(milestone);
return { feature, slice, milestone };
}
it("increments retries and requeues failed tasks", async () => {
wireMissionTask();
autopilot.watchMission("M-TEST1");
await autopilot.handleTaskFailure("FN-001");
expect(taskStore.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(taskStore.updateTask).toHaveBeenCalledWith(
"FN-001",
expect.objectContaining({ error: null, status: null, paused: false }),
);
expect(missionStore.updateFeatureStatus).not.toHaveBeenCalled();
});
it("marks feature blocked after max retries and does not retry again", async () => {
const { feature } = wireMissionTask();
autopilot.watchMission("M-TEST1");
taskStore.getSettings.mockResolvedValue({
missionStaleThresholdMs: 600_000,
missionMaxTaskRetries: 1,
missionHealthCheckIntervalMs: 300_000,
});
await autopilot.handleTaskFailure("FN-001");
await autopilot.handleTaskFailure("FN-001");
expect(missionStore.updateFeatureStatus).toHaveBeenCalledWith(feature.id, "blocked");
expect(taskStore.moveTask).toHaveBeenCalledTimes(1);
expect(taskStore.updateTask).toHaveBeenCalledWith(
"FN-001",
expect.objectContaining({ status: "failed", paused: true }),
);
});
it("clears retry budget for a task after successful completion", async () => {
wireMissionTask();
autopilot.watchMission("M-TEST1");
taskStore.getSettings.mockResolvedValue({
missionStaleThresholdMs: 600_000,
missionMaxTaskRetries: 1,
missionHealthCheckIntervalMs: 300_000,
});
await autopilot.handleTaskFailure("FN-001");
missionStore.listFeatures.mockReturnValue([
createMockFeature({ id: "F-001", taskId: "FN-001", sliceId: "SL-001", status: "in-progress" }),
]);
await autopilot.handleTaskCompletion("FN-001");
await autopilot.handleTaskFailure("FN-001");
expect(missionStore.updateFeatureStatus).not.toHaveBeenCalledWith("F-001", "blocked");
expect(taskStore.moveTask).toHaveBeenCalledTimes(2);
});
it("is a no-op when task is not linked to a feature", async () => {
missionStore.getFeatureByTaskId.mockReturnValue(undefined);
await autopilot.handleTaskFailure("FN-001");
expect(taskStore.moveTask).not.toHaveBeenCalled();
expect(taskStore.updateTask).not.toHaveBeenCalled();
});
it("is a no-op for missions that are not being watched", async () => {
wireMissionTask();
await autopilot.handleTaskFailure("FN-001");
expect(taskStore.moveTask).not.toHaveBeenCalled();
expect(taskStore.updateTask).not.toHaveBeenCalled();
expect(missionStore.updateFeatureStatus).not.toHaveBeenCalled();
});
it("clears task error when retrying", async () => {
wireMissionTask();
autopilot.watchMission("M-TEST1");
await autopilot.handleTaskFailure("FN-001");
expect(taskStore.updateTask).toHaveBeenCalledWith(
"FN-001",
expect.objectContaining({ error: null, status: null }),
);
});
});
// ── Advance to Next Slice ────────────────────────────────────────
describe("advanceToNextSlice", () => {
@@ -500,33 +609,364 @@ describe("MissionAutopilot", () => {
});
});
describe("health check", () => {
it("fixes feature status when linked task is done", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "triaged", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "done" });
await (autopilot as any).runHealthCheck();
expect(missionStore.updateFeatureStatus).toHaveBeenCalledWith("F-001", "done");
autopilot.stop();
});
it("fixes feature status when task is in-progress but feature is triaged", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "triaged", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "in-progress" });
await (autopilot as any).runHealthCheck();
expect(missionStore.updateFeatureStatus).toHaveBeenCalledWith("F-001", "in-progress");
autopilot.stop();
});
it("fixes feature status when task regresses to todo/triage", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "in-progress", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "todo" });
await (autopilot as any).runHealthCheck();
expect(missionStore.updateFeatureStatus).toHaveBeenCalledWith("F-001", "triaged");
autopilot.stop();
});
it("triggers failure recovery for failed in-progress tasks", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "in-progress", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "in-progress", status: "failed" });
const failureSpy = vi.spyOn(autopilot, "handleTaskFailure").mockResolvedValue();
await (autopilot as any).runHealthCheck();
expect(failureSpy).toHaveBeenCalledWith("FN-001");
autopilot.stop();
});
it("leaves consistent feature/task states unchanged", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "in-progress", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "in-progress" });
await (autopilot as any).runHealthCheck();
expect(missionStore.updateFeatureStatus).not.toHaveBeenCalled();
autopilot.stop();
});
it("skips features that do not have linked tasks", async () => {
autopilot.start();
autopilot.watchMission("M-TEST1");
missionStore.getMissionWithHierarchy.mockReturnValue({
...createMockMission(),
milestones: [{
...createMockMilestone(),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "in-progress", taskId: undefined })],
}],
}],
});
await (autopilot as any).runHealthCheck();
expect(taskStore.getTask).not.toHaveBeenCalled();
expect(missionStore.updateFeatureStatus).not.toHaveBeenCalled();
autopilot.stop();
});
it("does not create a health-check timer when disabled", async () => {
taskStore.getSettings.mockResolvedValue({
missionStaleThresholdMs: 600_000,
missionMaxTaskRetries: 3,
missionHealthCheckIntervalMs: 0,
});
autopilot.start();
await vi.runOnlyPendingTimersAsync();
expect((autopilot as any).healthCheckTimer).toBeNull();
});
it("uses default health-check interval when setting is undefined", async () => {
taskStore.getSettings.mockResolvedValue({
missionStaleThresholdMs: 600_000,
missionMaxTaskRetries: 3,
missionHealthCheckIntervalMs: undefined,
});
autopilot.start();
await vi.runOnlyPendingTimersAsync();
expect((autopilot as any).healthCheckTimer).not.toBeNull();
});
});
// ── Poll / stale detection ──────────────────────────────────────
describe("poll stale detection", () => {
it("logs warning events for stale watched missions", () => {
it("recovers stale activating missions back to watching and advances slices", async () => {
const staleMission = createMockMission({
lastAutopilotActivityAt: new Date(Date.now() - 10 * 60 * 1000).toISOString(),
autopilotState: "activating",
lastAutopilotActivityAt: new Date(Date.now() - 11 * 60 * 1000).toISOString(),
});
const store = createMockMissionStore([staleMission]);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler });
const localScheduler = createMockScheduler();
localScheduler.activateNextPendingSlice.mockResolvedValue(
createMockSlice({ id: "SL-002", status: "active" }),
);
store.getMissionWithHierarchy.mockReturnValue({
...staleMission,
milestones: [{
...createMockMilestone({ missionId: staleMission.id }),
slices: [{
...createMockSlice({ id: "SL-001", status: "active" }),
features: [createMockFeature({ status: "done" })],
}],
}],
});
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler: localScheduler });
ap.start();
ap.watchMission("M-TEST1");
store.updateMission("M-TEST1", {
autopilotState: "activating",
lastAutopilotActivityAt: staleMission.lastAutopilotActivityAt,
});
store.updateMission.mockClear();
store.logMissionEvent.mockClear();
await vi.advanceTimersByTimeAsync(60_000);
expect(store.updateMission).toHaveBeenCalledWith(
"M-TEST1",
expect.objectContaining({ autopilotState: "watching" }),
);
expect(localScheduler.activateNextPendingSlice).toHaveBeenCalledWith("M-TEST1");
expect(store.logMissionEvent).toHaveBeenCalledWith(
"M-TEST1",
"autopilot_stale",
expect.stringContaining("stale"),
expect.objectContaining({ staleThresholdMs: 600_000 }),
);
ap.stop();
});
it("does not recover missions that are not in activating state", async () => {
const staleWatchingMission = createMockMission({
autopilotState: "watching",
lastAutopilotActivityAt: new Date(Date.now() - 11 * 60 * 1000).toISOString(),
});
const store = createMockMissionStore([staleWatchingMission]);
const localScheduler = createMockScheduler();
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler: localScheduler });
ap.start();
ap.watchMission("M-TEST1");
store.logMissionEvent.mockClear();
vi.advanceTimersByTime(60_000);
await vi.advanceTimersByTimeAsync(60_000);
expect(store.logMissionEvent).toHaveBeenCalledWith(
expect(localScheduler.activateNextPendingSlice).not.toHaveBeenCalled();
expect(store.logMissionEvent).not.toHaveBeenCalledWith(
"M-TEST1",
"warning",
expect.stringContaining("stale"),
expect.objectContaining({ category: "autopilot_stale" }),
"autopilot_stale",
expect.any(String),
expect.anything(),
);
ap.stop();
});
});
describe("recoverStaleMission", () => {
it("activates pending work when active slice is complete", async () => {
const mission = createMockMission();
const store = createMockMissionStore([mission]);
const localScheduler = createMockScheduler();
localScheduler.activateNextPendingSlice.mockResolvedValue(
createMockSlice({ id: "SL-002", status: "active" }),
);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler: localScheduler });
store.getMissionWithHierarchy.mockReturnValue({
...mission,
milestones: [{
...createMockMilestone({ missionId: mission.id }),
slices: [{
...createMockSlice({ id: "SL-001", status: "active" }),
features: [createMockFeature({ status: "done" })],
}],
}],
});
ap.watchMission("M-TEST1");
await ap.recoverStaleMission("M-TEST1");
expect(localScheduler.activateNextPendingSlice).toHaveBeenCalledWith("M-TEST1");
});
it("handles mission not found gracefully", async () => {
missionStore.getMissionWithHierarchy.mockReturnValue(undefined);
await expect(autopilot.recoverStaleMission("M-TEST1")).resolves.toBeUndefined();
expect(scheduler.activateNextPendingSlice).not.toHaveBeenCalled();
});
});
describe("recoverMissions", () => {
it("watches eligible missions and skips complete/archived missions", async () => {
const missions = [
createMockMission({ id: "M-ONE", status: "active", autopilotEnabled: true }),
createMockMission({ id: "M-TWO", status: "complete", autopilotEnabled: true }),
createMockMission({ id: "M-THREE", status: "archived", autopilotEnabled: true }),
createMockMission({ id: "M-FOUR", status: "active", autopilotEnabled: false }),
];
const store = createMockMissionStore(missions);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler });
store.getMissionWithHierarchy.mockReturnValue(undefined);
await ap.recoverMissions(store as any);
expect(ap.isWatching("M-ONE")).toBe(true);
expect(ap.isWatching("M-TWO")).toBe(false);
expect(ap.isWatching("M-THREE")).toBe(false);
expect(ap.isWatching("M-FOUR")).toBe(false);
});
it("recovers missions stuck in activating state", async () => {
const mission = createMockMission({ autopilotState: "activating" });
const store = createMockMissionStore([mission]);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler });
const recoverStaleSpy = vi.spyOn(ap, "recoverStaleMission").mockResolvedValue();
store.getMissionWithHierarchy.mockReturnValue(undefined);
await ap.recoverMissions(store as any);
expect(recoverStaleSpy).toHaveBeenCalledWith(mission.id);
});
it("fixes feature/task inconsistencies during recovery", async () => {
const mission = createMockMission();
const store = createMockMissionStore([mission]);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler });
store.getMissionWithHierarchy.mockReturnValue({
...mission,
milestones: [{
...createMockMilestone({ missionId: mission.id }),
slices: [{
...createMockSlice({ status: "active" }),
features: [createMockFeature({ id: "F-001", status: "triaged", taskId: "FN-001" })],
}],
}],
});
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "done" });
await ap.recoverMissions(store as any);
expect(store.updateFeatureStatus).toHaveBeenCalledWith("F-001", "done");
});
it("advances slices when active slice features are already done", async () => {
const mission = createMockMission();
const store = createMockMissionStore([mission]);
const localScheduler = createMockScheduler();
localScheduler.activateNextPendingSlice.mockResolvedValue(
createMockSlice({ id: "SL-002", status: "active" }),
);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler: localScheduler });
const hierarchy = {
...mission,
milestones: [{
...createMockMilestone({ missionId: mission.id }),
slices: [{
...createMockSlice({ id: "SL-001", status: "active" }),
features: [createMockFeature({ id: "F-001", status: "done", taskId: "FN-001" })],
}],
}],
};
store.getMissionWithHierarchy.mockReturnValue(hierarchy);
taskStore.getTask.mockResolvedValue({ id: "FN-001", column: "done" });
await ap.recoverMissions(store as any);
expect(localScheduler.activateNextPendingSlice).toHaveBeenCalledWith(mission.id);
});
it("handles empty mission lists", async () => {
const store = createMockMissionStore([]);
const ap = new MissionAutopilot(taskStore as any, store as any, { scheduler });
await expect(ap.recoverMissions(store as any)).resolves.toBeUndefined();
});
});
// ── Stop cleanup ─────────────────────────────────────────────────
describe("stop cleanup", () => {

395
packages/engine/src/mission-autopilot.ts
View File

@@ -39,8 +39,14 @@ const RETRY_BASE_DELAY_MS = 1000;
/** Background poll interval for checking mission health (ms). */
const POLL_INTERVAL_MS = 60_000;
/** Time after which a mission is considered stale (5 minutes). */
const STALE_THRESHOLD_MS = 5 * 60 * 1000;
/** Default time after which a mission activation is considered stale (10 minutes). */
const DEFAULT_STALE_THRESHOLD_MS = 10 * 60 * 1000;
/** Default per-task retry budget before a feature is marked blocked. */
const DEFAULT_MAX_TASK_RETRIES = 3;
/** Default cadence for mission consistency sweeps (5 minutes). */
const DEFAULT_HEALTH_CHECK_INTERVAL_MS = 5 * 60 * 1000;
/** Per-mission tracking state. */
interface WatchedMissionState {
@@ -65,8 +71,10 @@ export interface MissionAutopilotOptions {
*/
export class MissionAutopilot {
private watchedMissions = new Map<string, WatchedMissionState>();
private perMissionTaskRetries = new Map<string, Map<string, number>>();
private running = false;
private pollTimer: ReturnType<typeof setInterval> | null = null;
private healthCheckTimer: ReturnType<typeof setInterval> | null = null;
private scheduler: MissionAutopilotOptions["scheduler"];
constructor(
@@ -95,7 +103,12 @@ export class MissionAutopilot {
start(): void {
if (this.running) return;
this.running = true;
this.pollTimer = setInterval(() => this.poll(), POLL_INTERVAL_MS);
this.pollTimer = setInterval(() => {
void this.poll().catch((err) => {
autopilotLog.error("Error during autopilot poll:", err);
});
}, POLL_INTERVAL_MS);
void this.startHealthCheck();
autopilotLog.log("Started");
}
@@ -111,6 +124,7 @@ export class MissionAutopilot {
clearInterval(this.pollTimer);
this.pollTimer = null;
}
this.stopHealthCheck();
// Unwatch all missions
for (const [missionId] of this.watchedMissions) {
@@ -121,6 +135,7 @@ export class MissionAutopilot {
}
}
this.watchedMissions.clear();
this.perMissionTaskRetries.clear();
autopilotLog.log("Stopped");
}
@@ -176,6 +191,7 @@ export class MissionAutopilot {
}
this.watchedMissions.delete(missionId);
this.perMissionTaskRetries.delete(missionId);
try {
this.setAutopilotState(missionId, "inactive");
} catch {
@@ -252,6 +268,9 @@ export class MissionAutopilot {
// Only proceed if we're watching this mission
if (!this.isWatching(missionId)) return;
// Successful completion resets retry budget for this specific task.
this.perMissionTaskRetries.get(missionId)?.delete(taskId);
// Check if all features in the slice are done
const features = this.missionStore.listFeatures(slice.id);
const allDone = features.length > 0 && features.every((f) => f.status === "done");
@@ -265,6 +284,72 @@ export class MissionAutopilot {
}
}
/**
* Called when a mission-linked task fails execution.
* Applies retry budgets per mission/task and blocks features that exceed the budget.
*/
async handleTaskFailure(taskId: string): Promise<void> {
try {
const feature = this.missionStore.getFeatureByTaskId(taskId);
if (!feature) {
return;
}
const slice = this.missionStore.getSlice(feature.sliceId);
if (!slice) {
autopilotLog.warn(`Task failure ${taskId}: slice ${feature.sliceId} not found`);
return;
}
const milestone = this.missionStore.getMilestone(slice.milestoneId);
if (!milestone) {
autopilotLog.warn(`Task failure ${taskId}: milestone ${slice.milestoneId} not found`);
return;
}
const missionId = milestone.missionId;
if (!this.isWatching(missionId)) {
return;
}
const settings = await this.taskStore.getSettings();
const maxRetries = settings.missionMaxTaskRetries ?? DEFAULT_MAX_TASK_RETRIES;
const missionRetries = this.perMissionTaskRetries.get(missionId) ?? new Map<string, number>();
this.perMissionTaskRetries.set(missionId, missionRetries);
const retryCount = (missionRetries.get(taskId) ?? 0) + 1;
missionRetries.set(taskId, retryCount);
if (retryCount > maxRetries) {
this.missionStore.updateFeatureStatus(feature.id, "blocked");
await this.taskStore.updateTask(taskId, { status: "failed", paused: true });
this.logMissionEventSafe(
missionId,
"error",
`Feature ${feature.id} blocked after max retries (${retryCount}/${maxRetries})`,
{ taskId, featureId: feature.id, retryCount, maxRetries },
);
return;
}
this.logMissionEventSafe(
missionId,
"autopilot_retry",
`Retrying failed mission task ${taskId} (${retryCount}/${maxRetries})`,
{ taskId, featureId: feature.id, retryCount, maxRetries },
);
const task = await this.taskStore.getTask(taskId);
if (task?.column !== "todo") {
await this.taskStore.moveTask(taskId, "todo");
}
await this.taskStore.updateTask(taskId, { error: null, status: null, paused: false });
} catch (err) {
autopilotLog.error(`Error handling task failure for ${taskId}:`, err);
}
}
/**
* Activate the next pending slice in a mission.
* Uses the scheduler's `activateNextPendingSlice()` method.
@@ -398,6 +483,7 @@ export class MissionAutopilot {
this.updateActivity(missionId);
this.setAutopilotState(missionId, "inactive");
this.watchedMissions.delete(missionId);
this.perMissionTaskRetries.delete(missionId);
return true;
}
@@ -410,9 +496,9 @@ export class MissionAutopilot {
* Periodic health check for watched missions.
* - Re-watches missions with `autopilotEnabled: true` that aren't being tracked
* - Starts missions in `planning` with autopilot enabled
* - Flags stale missions
* - Recovers stale missions stuck in `activating`
*/
private poll(): void {
private async poll(): Promise<void> {
if (!this.running) return;
try {
@@ -427,10 +513,13 @@ export class MissionAutopilot {
// Start planning missions with autopilot
if (mission.autopilotEnabled && mission.status === "planning" && this.isWatching(mission.id)) {
void this.checkAndStartMission(mission.id);
await this.checkAndStartMission(mission.id);
}
}
const settings = await this.taskStore.getSettings();
const staleThresholdMs = settings.missionStaleThresholdMs ?? DEFAULT_STALE_THRESHOLD_MS;
// Check for stale missions
const now = Date.now();
for (const [missionId, state] of this.watchedMissions) {
@@ -438,34 +527,273 @@ export class MissionAutopilot {
if (!mission) {
// Mission deleted — unwatch
this.watchedMissions.delete(missionId);
this.perMissionTaskRetries.delete(missionId);
continue;
}
if (mission.lastAutopilotActivityAt) {
const lastActivity = new Date(mission.lastAutopilotActivityAt).getTime();
if (now - lastActivity > STALE_THRESHOLD_MS) {
const staleMinutes = Math.round((now - lastActivity) / 60_000);
this.logMissionEventSafe(
missionId,
"warning",
`Mission autopilot appears stale (no activity for ${staleMinutes} minutes)`,
{
staleMinutes,
staleThresholdMs: STALE_THRESHOLD_MS,
lastActivityAt: mission.lastAutopilotActivityAt,
retryCount: state.retryCount,
category: "autopilot_stale",
},
);
autopilotLog.warn(`Mission ${missionId} is stale (no activity for ${staleMinutes} minutes)`);
}
if (!mission.lastAutopilotActivityAt || mission.autopilotState !== "activating") {
continue;
}
const lastActivity = new Date(mission.lastAutopilotActivityAt).getTime();
if (now - lastActivity <= staleThresholdMs) {
continue;
}
const staleMinutes = Math.round((now - lastActivity) / 60_000);
this.logMissionEventSafe(
missionId,
"autopilot_stale",
`Mission autopilot is stale and will be recovered (${staleMinutes} minutes inactive)` ,
{
staleMinutes,
staleThresholdMs,
lastActivityAt: mission.lastAutopilotActivityAt,
retryCount: state.retryCount,
previousState: mission.autopilotState,
},
);
autopilotLog.warn(`Mission ${missionId} stale while activating (inactive ${staleMinutes}m) — recovering`);
this.setAutopilotState(missionId, "watching");
state.retryCount = 0;
await this.recoverStaleMission(missionId);
this.updateActivity(missionId);
}
} catch (err) {
autopilotLog.error("Error during autopilot poll:", err);
}
}
/**
* Attempt to recover a mission that appears stalled in the activating state.
* Re-evaluates active/pending slices and advances when progression is possible.
*/
async recoverStaleMission(missionId: string): Promise<void> {
try {
const mission = this.missionStore.getMissionWithHierarchy(missionId);
if (!mission) {
autopilotLog.warn(`recoverStaleMission: mission ${missionId} not found`);
return;
}
const activeSlices = mission.milestones.flatMap((milestone) => milestone.slices)
.filter((slice) => slice.status === "active");
let advanced = false;
if (activeSlices.length > 0) {
const hasCompletedActiveSlice = activeSlices.some((slice) =>
slice.features.length > 0 && slice.features.every((feature) => feature.status === "done"),
);
if (hasCompletedActiveSlice) {
await this.advanceToNextSlice(missionId);
advanced = true;
}
} else {
const hasPendingSlice = mission.milestones.some((milestone) =>
milestone.slices.some((slice) => slice.status === "pending"),
);
if (hasPendingSlice) {
await this.advanceToNextSlice(missionId);
advanced = true;
}
}
this.logMissionEventSafe(
missionId,
"autopilot_stale",
advanced
? `Recovered stale mission ${missionId} and resumed slice progression`
: `Recovered stale mission ${missionId}; no immediate slice progression needed`,
{
source: "recoverStaleMission",
activeSliceCount: activeSlices.length,
advanced,
},
);
} catch (err) {
autopilotLog.error(`recoverStaleMission failed for ${missionId}:`, err);
}
}
private async startHealthCheck(): Promise<void> {
this.stopHealthCheck();
let intervalMs = DEFAULT_HEALTH_CHECK_INTERVAL_MS;
try {
const settings = await this.taskStore.getSettings();
intervalMs = settings.missionHealthCheckIntervalMs ?? DEFAULT_HEALTH_CHECK_INTERVAL_MS;
} catch (err) {
autopilotLog.warn("Failed to read mission health check settings; using defaults", err);
}
if (!this.running) {
return;
}
if (intervalMs <= 0) {
autopilotLog.log("Mission health checks disabled (missionHealthCheckIntervalMs=0)");
return;
}
this.healthCheckTimer = setInterval(() => {
void this.runHealthCheck();
}, intervalMs);
autopilotLog.log(`Mission health checks started (every ${intervalMs}ms)`);
}
private stopHealthCheck(): void {
if (!this.healthCheckTimer) {
return;
}
clearInterval(this.healthCheckTimer);
this.healthCheckTimer = null;
}
private async runHealthCheck(): Promise<void> {
if (!this.running || this.watchedMissions.size === 0) {
return;
}
try {
let fixedCount = 0;
for (const missionId of this.watchedMissions.keys()) {
const mission = this.missionStore.getMissionWithHierarchy(missionId);
if (!mission) {
continue;
}
fixedCount += await this.reconcileMissionConsistency(mission);
}
autopilotLog.log(`Mission health check complete: fixed ${fixedCount} inconsistenc${fixedCount === 1 ? "y" : "ies"}`);
} catch (err) {
autopilotLog.error("Mission health check failed:", err);
}
}
/**
* Recover autopilot state after process restart.
* Watches active missions and performs a one-time consistency sweep.
*/
async recoverMissions(missionStore: MissionStore): Promise<void> {
try {
const missions = missionStore.listMissions();
let watchedCount = 0;
let recoveredActivatingCount = 0;
let inconsistencyFixes = 0;
for (const mission of missions) {
if (!mission.autopilotEnabled || mission.status === "complete" || mission.status === "archived") {
continue;
}
if (!this.isWatching(mission.id)) {
this.watchMission(mission.id);
watchedCount++;
}
if (mission.autopilotState === "activating") {
await this.recoverStaleMission(mission.id);
recoveredActivatingCount++;
}
const hierarchy = missionStore.getMissionWithHierarchy(mission.id);
if (!hierarchy) {
continue;
}
inconsistencyFixes += await this.reconcileMissionConsistency(hierarchy);
const refreshedHierarchy = missionStore.getMissionWithHierarchy(mission.id);
if (!refreshedHierarchy) {
continue;
}
const hasCompletedActiveSlice = refreshedHierarchy.milestones
.flatMap((milestone) => milestone.slices)
.filter((slice) => slice.status === "active")
.some((slice) => slice.features.length > 0 && slice.features.every((feature) => feature.status === "done"));
if (hasCompletedActiveSlice) {
await this.advanceToNextSlice(mission.id);
}
}
autopilotLog.log(
`Mission recovery complete: watched ${watchedCount}, recovered ${recoveredActivatingCount} activating missions, fixed ${inconsistencyFixes} inconsistenc${inconsistencyFixes === 1 ? "y" : "ies"}`,
);
} catch (err) {
autopilotLog.error("Mission recovery failed:", err);
}
}
private async reconcileMissionConsistency(
mission: ReturnType<MissionStore["getMissionWithHierarchy"]>,
): Promise<number> {
if (!mission) {
return 0;
}
const activeSlices = mission.milestones
.flatMap((milestone) => milestone.slices)
.filter((slice) => slice.status === "active");
if (activeSlices.length === 0) {
return 0;
}
let fixedCount = 0;
for (const slice of activeSlices) {
for (const feature of slice.features) {
if (!feature.taskId) {
continue;
}
const task = await this.taskStore.getTask(feature.taskId);
if (!task) {
continue;
}
if (task.status === "failed" && feature.status === "in-progress") {
await this.handleTaskFailure(feature.taskId);
fixedCount++;
continue;
}
if (task.column === "done" && feature.status !== "done") {
this.missionStore.updateFeatureStatus(feature.id, "done");
fixedCount++;
continue;
}
if (
task.column === "in-progress"
&& (feature.status === "triaged" || feature.status === "defined")
) {
this.missionStore.updateFeatureStatus(feature.id, "in-progress");
fixedCount++;
continue;
}
if (
(task.column === "triage" || task.column === "todo")
&& feature.status === "in-progress"
) {
this.missionStore.updateFeatureStatus(feature.id, "triaged");
fixedCount++;
}
}
}
return fixedCount;
}
// ── Helpers ────────────────────────────────────────────────────────
/**
@@ -477,8 +805,25 @@ export class MissionAutopilot {
description: string,
metadata?: Record<string, unknown>,
): void {
const missionStoreWithEvents = this.missionStore as MissionStore & {
logMissionEvent?: (
missionId: string,
eventType: MissionEventType,
description: string,
metadata?: Record<string, unknown>,
) => unknown;
};
if (typeof missionStoreWithEvents.logMissionEvent !== "function") {
autopilotLog.warn(
`[${eventType}] ${missionId}: ${description}`,
metadata ?? {},
);
return;
}
try {
this.missionStore.logMissionEvent(missionId, eventType, description, metadata);
missionStoreWithEvents.logMissionEvent(missionId, eventType, description, metadata);
} catch (err) {
autopilotLog.error(
`Failed to persist mission event (${eventType}) for ${missionId}:`,

34
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -23,6 +23,7 @@ import { runtimeLog } from "../logger.js";
import type { StuckTaskDetector } from "../stuck-task-detector.js";
import type { UsageLimitPauser } from "../usage-limit-detector.js";
import { SelfHealingManager } from "../self-healing.js";
import { MissionAutopilot } from "../mission-autopilot.js";
/**
* InProcessRuntime runs a project within the main process.
@@ -137,11 +138,21 @@ export class InProcessRuntime
// 4. Initialize Scheduler
const missionStore = this.taskStore.getMissionStore();
const missionAutopilot = missionStore
? new MissionAutopilot(this.taskStore, missionStore)
: undefined;
this.scheduler = new Scheduler(this.taskStore, {
maxConcurrent: this.config.maxConcurrent,
maxWorktrees: this.config.maxWorktrees,
semaphore: this.globalSemaphore,
missionStore,
missionAutopilot,
onTaskFailed: (taskId) => {
if (missionAutopilot) {
void missionAutopilot.handleTaskFailure(taskId);
}
},
onSchedule: (task) => {
this.recordActivity();
runtimeLog.log(`Scheduled task ${task.id}`);
@@ -193,6 +204,22 @@ export class InProcessRuntime
this.recordActivity();
runtimeLog.error(`Task ${task.id} failed:`, error.message);
this.recordTaskCompletion(task.id, false);
// Mission-linked failures should be re-queued to todo so autopilot retry
// policies can decide whether to retry or block the feature.
if (task.sliceId) {
void (async () => {
try {
const latest = await this.taskStore.getTask(task.id);
if (latest?.column === "in-progress") {
await this.taskStore.moveTask(task.id, "todo");
}
} catch (moveErr) {
runtimeLog.warn(`Failed to requeue mission task ${task.id} after error:`, moveErr);
}
})();
}
// Update agent state to terminated (failed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
@@ -288,6 +315,13 @@ export class InProcessRuntime
// 10. Start scheduler
this.scheduler.start();
// Mission crash recovery: restore autopilot state for missions that were active before crash
const activeMissionStore = this.taskStore.getMissionStore();
const activeMissionAutopilot = this.scheduler.getMissionAutopilot?.();
if (activeMissionStore && activeMissionAutopilot) {
void activeMissionAutopilot.recoverMissions(activeMissionStore);
}
this.setStatus("active");
runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
} catch (error) {

27
packages/engine/src/scheduler.ts
View File

@@ -59,6 +59,8 @@ export interface SchedulerOptions {
onSchedule?: (task: Task) => void;
/** Called when a task is blocked by deps */
onBlocked?: (task: Task, blockedBy: string[]) => void;
/** Called when a mission-linked task fails and is queued for retry handling. */
onTaskFailed?: (taskId: string) => void | Promise<void>;
/** Optional PR monitor for tracking in-review PRs */
prMonitor?: PrMonitor;
/** Optional MissionStore for slice activation and auto-advance */
@@ -105,6 +107,8 @@ export class Scheduler {
private activePollMs: number | null = null;
/** Tracks which task IDs are currently paused, to detect unpause transitions. */
private pausedTaskIds = new Set<string>();
/** Tracks mission-linked tasks observed with status=failed before moveTask clears status/error. */
private failedTaskIds = new Set<string>();
constructor(
private store: TaskStore,
@@ -194,6 +198,17 @@ export class Scheduler {
void this.handleMissionTaskCompletion(task.id, task.sliceId);
}
// Mission failure tracking: status/error are cleared during moveTask(in-progress → todo),
// so we pair this with failedTaskIds captured from task:updated events.
if (task.sliceId && to === "todo" && this.options.onTaskFailed) {
if (task.status === "failed" || this.failedTaskIds.has(task.id)) {
this.failedTaskIds.delete(task.id);
void Promise.resolve(this.options.onTaskFailed(task.id)).catch((err) => {
schedulerLog.error(`Error in onTaskFailed for ${task.id}:`, err);
});
}
}
// Event-driven scheduling: when a task moves to "done" (completion) or "todo" (retry/manual move),
// trigger scheduling immediately so waiting tasks can start without waiting
// for the next poll interval (up to 15 seconds).
@@ -208,6 +223,13 @@ export class Scheduler {
* Also detects task-level unpause transitions and triggers immediate scheduling.
*/
this.store.on("task:updated", (task) => {
// Track mission failure signals before moveTask clears failure metadata.
if (task.sliceId && task.column === "in-progress" && task.status === "failed") {
this.failedTaskIds.add(task.id);
} else if (task.status !== "failed") {
this.failedTaskIds.delete(task.id);
}
// Track pause state transitions for event-driven scheduling on unpause.
// When a previously-paused task is unpaused in a schedulable column,
// trigger a scheduling pass immediately instead of waiting for the next
@@ -313,6 +335,7 @@ export class Scheduler {
if (this.options.missionAutopilot) {
this.options.missionAutopilot.stop();
}
this.failedTaskIds.clear();
schedulerLog.log("Stopped");
}
@@ -332,6 +355,10 @@ export class Scheduler {
schedulerLog.log(`Poll interval updated to ${newIntervalMs}ms`);
}
getMissionAutopilot(): import("./mission-autopilot.js").MissionAutopilot | undefined {
return this.options.missionAutopilot;
}
/**
* Resolve the base branch for a task being started.
*